daemon: agent-driven workspace images — repo→image registry (GET/PUT/DELETE /api/repos), spawn resolves custom image, self-healing lvmh-ops control container with docker.sock; 132 tests, e2e 87/87

This commit is contained in:
Raphael Westphal
2026-08-18 23:54:09 +02:00
parent 8fcaeb7c03
commit 7287b831e4
15 changed files with 1216 additions and 38 deletions
+26
View File
@@ -170,6 +170,9 @@ Errors: `{"error": "message"}` with appropriate status.
| `POST /api/spawn` | `{repo: "group/project", branch?: "main"}``{sessionId, containerId}` | clone (per-repo volume) → run container |
| `GET /api/spawn/status` | → `[{repo, state, containerId, sessionId?}]` | in-flight clone/spawn jobs |
| `GET /api/repos` | → `[{repo, image}]` | repo→custom image registrations, sorted by repo |
| `PUT /api/repos/{repo}/image` | `{image}``{ok}` | register custom image; repo must pass repo validation, image must match `lvmh-worker-*` |
| `DELETE /api/repos/{repo}/image` | → `{ok}` | drop registration (idempotent); repo spawns fall back to `lvmh-worker:latest` |
`POST /api/spawn` semantics:
@@ -187,6 +190,26 @@ Errors: `{"error": "message"}` with appropriate status.
4. Response returns immediately after container create; status via
`/api/spawn/status` and session list.
**Repo→image resolution:** before creating the container the daemon looks up
`repo` in the repo-images registry (`repo_images` table, managed by the
`/api/repos` routes). A registered image replaces `lvmh-worker:latest` in
the created container; the daemon does NOT build custom images — if the
registered image is missing on the docker host the job errors with
`custom image <name> not built (ask the ops agent to build it)`. Image names
are pinned to the `lvmh-worker-*` namespace by validation. Without a
registration (or after DELETE) spawns use the default worker image.
**Ops container.** The daemon keeps one long-lived control container,
`lvmh-ops` (image `lvmh-ops:latest`, built from `docker/control.Dockerfile`
when missing): an agent pi with the docker socket mounted, its own
`lvmh-ops-work:/ops` workspace volume and the shared session/pi-cache
volumes, on the daemon network with a fixed session id `lvmh-ops-control`.
It prepares repos, builds per-repo worker images (docker CLI via the socket)
and registers them through `/api/repos`. A loop re-checks every 30s: stopped
→ started, missing → built+created+started; `DELETE
/api/sessions/lvmh-ops-control/container` removes it and the loop recreates
it on the next tick with a fresh session.
### GitLab
| `GET /api/gitlab/status` | → `{connected: bool, baseUrl, username?}` | |
@@ -244,6 +267,7 @@ CREATE TABLE events(sessionId TEXT NOT NULL, seq INTEGER NOT NULL,
CREATE TABLE sessions(id TEXT PRIMARY KEY, info TEXT NOT NULL, -- session JSON
lastSeq INTEGER NOT NULL DEFAULT 0, lastEventAt INTEGER, online INTEGER DEFAULT 0);
CREATE TABLE settings(key TEXT PRIMARY KEY, value TEXT NOT NULL); -- gitlab PAT etc
CREATE TABLE repo_images(repo TEXT PRIMARY KEY, image TEXT NOT NULL); -- repo→custom worker image
```
File `lvmh.db` on volume `lvmh-data`.
@@ -256,3 +280,5 @@ File `lvmh.db` on volume `lvmh-data`.
| `LVMH_TOKEN` | plugin, daemon, web UI storage | shared secret |
| `ZAI_RENAUD_API_KEY` | daemon, containers | LLM provider key |
| `LVMH_CONTAINER_DOCKER_SOCK` | daemon container | `/var/run/docker.sock` |
| `LVMH_GITEA_TOKEN` | ops container | PAT passed through from the daemon store, for repo clone/push |
| `LVMH_API` | ops container | `http://lvmh:8686` — daemon REST base for curl |
+78 -1
View File
@@ -12,6 +12,7 @@ import (
"net/http"
"os"
"path/filepath"
"regexp"
"strconv"
"strings"
)
@@ -25,8 +26,15 @@ const (
maxEventsLimit int = 10000
maxBodyBytes int64 = 1 << 20
webIndexFallback string = "index.html"
reposPathPrefix string = "/api/repos/"
reposPathSuffix string = "/image"
)
// imageNameRe pins registered images to the lvmh-worker- namespace so agents
// can only point spawns at images we built.
var imageNameRe = regexp.MustCompile(`^lvmh-worker-[a-z0-9][a-z0-9.-]*:?[a-zA-Z0-9._-]*$`)
//go:embed webdist
var embeddedWeb embed.FS
@@ -60,6 +68,11 @@ func (s *Server) Routes(webdist string) http.Handler {
api.HandleFunc("DELETE /api/sessions/{id}/container", s.handleDeleteContainer)
api.HandleFunc("POST /api/spawn", s.handleSpawn)
api.HandleFunc("GET /api/spawn/status", s.handleSpawnStatus)
api.HandleFunc("GET /api/repos", s.handleRepoImages)
// repo paths contain "/" (group/project), which a single {repo} wildcard
// segment cannot match — subtree routes with manual path parsing instead.
api.HandleFunc("PUT /api/repos/", s.handleSetRepoImage)
api.HandleFunc("DELETE /api/repos/", s.handleDeleteRepoImage)
api.HandleFunc("GET /api/gitlab/status", s.handleGitLabStatus)
api.HandleFunc("POST /api/gitlab/connect", s.handleGitLabConnect)
api.HandleFunc("DELETE /api/gitlab/connect", s.handleGitLabDisconnect)
@@ -257,7 +270,14 @@ func (s *Server) handleAbort(w http.ResponseWriter, r *http.Request) {
func (s *Server) handleDeleteContainer(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
if err := s.spawn.RemoveSession(r.Context(), id); err != nil {
var err error
if id == opsSessionID {
// the ops container has no containers-table row; remove by name
err = s.spawn.RemoveOps(r.Context())
} else {
err = s.spawn.RemoveSession(r.Context(), id)
}
if err != nil {
if errors.Is(err, errNoContainer) {
writeError(w, http.StatusNotFound, "no container for session")
return
@@ -296,6 +316,63 @@ func (s *Server) handleSpawnStatus(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.spawn.JobsSnapshot())
}
// repoFromImagePath extracts the repo path from /api/repos/<repo>/image.
func repoFromImagePath(path string) (string, bool) {
rest, ok := strings.CutSuffix(strings.TrimPrefix(path, reposPathPrefix), reposPathSuffix)
if !ok || !validRepoPath(rest) {
return "", false
}
return rest, true
}
func (s *Server) handleRepoImages(w http.ResponseWriter, r *http.Request) {
rows, err := s.store.ListRepoImages()
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if rows == nil {
rows = []RepoImageRow{}
}
writeJSON(w, http.StatusOK, rows)
}
func (s *Server) handleSetRepoImage(w http.ResponseWriter, r *http.Request) {
repo, ok := repoFromImagePath(r.URL.Path)
if !ok {
writeError(w, http.StatusBadRequest, "repo must look like group/project")
return
}
var body struct {
Image string `json:"image"`
}
if !decodeBody(w, r, &body) {
return
}
if !imageNameRe.MatchString(body.Image) {
writeError(w, http.StatusBadRequest, "invalid image name")
return
}
if err := s.store.SetRepoImage(repo, body.Image); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleDeleteRepoImage(w http.ResponseWriter, r *http.Request) {
repo, ok := repoFromImagePath(r.URL.Path)
if !ok {
writeError(w, http.StatusBadRequest, "repo must look like group/project")
return
}
if err := s.store.DeleteRepoImage(repo); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) handleGitLabStatus(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.gitlab.Status())
}
+134
View File
@@ -426,3 +426,137 @@ func TestAPIWebHandlerEmbedded(t *testing.T) {
}
}
}
func TestAPIRepoImagesCRUD(t *testing.T) {
ts, store := newTestServer(t)
auth := testToken
// auth required on the new namespace
if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/repos", "", ""); code != http.StatusUnauthorized {
t.Fatal("GET /api/repos must require auth")
}
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/image", "", `{"image":"lvmh-worker-x"}`); code != http.StatusUnauthorized {
t.Fatal("PUT image must require auth")
}
// empty listing
code, body := apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, "")
if code != http.StatusOK || body != "[]\n" {
t.Fatalf("empty repos = %d %q, want 200 []", code, body)
}
// valid registration
code, body = apiReq(t, http.MethodPut, ts.URL+"/api/repos/group/project/image", auth, `{"image":"lvmh-worker-group--project-ab12cd"}`)
if code != http.StatusOK || !strings.Contains(body, `"ok":true`) {
t.Fatalf("put image = %d %s", code, body)
}
// a second repo, then listing is sorted by repo
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/alpha/repo/image", auth, `{"image":"lvmh-worker-alpha:1.2.3"}`); code != http.StatusOK {
t.Fatalf("put tagged image = %d", code)
}
code, body = apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, "")
if code != http.StatusOK || body != `[{"repo":"alpha/repo","image":"lvmh-worker-alpha:1.2.3"},{"repo":"group/project","image":"lvmh-worker-group--project-ab12cd"}]`+"\n" {
t.Fatalf("repos = %d %s", code, body)
}
// upsert via PUT
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/group/project/image", auth, `{"image":"lvmh-worker-other"}`); code != http.StatusOK {
t.Fatalf("upsert = %d", code)
}
if img, ok, _ := store.GetRepoImage("group/project"); !ok || img != "lvmh-worker-other" {
t.Fatalf("after upsert = %q %v", img, ok)
}
// delete → gone; idempotent delete
if code, _ := apiReq(t, http.MethodDelete, ts.URL+"/api/repos/group/project/image", auth, ""); code != http.StatusOK {
t.Fatal("delete image failed")
}
if code, _ := apiReq(t, http.MethodDelete, ts.URL+"/api/repos/group/project/image", auth, ""); code != http.StatusOK {
t.Fatal("delete absent image must stay 200")
}
code, body = apiReq(t, http.MethodGet, ts.URL+"/api/repos", auth, "")
if code != http.StatusOK || strings.Contains(body, "group/project") {
t.Fatalf("after delete = %d %s", code, body)
}
}
func TestAPIRepoImageValidation(t *testing.T) {
ts, _ := newTestServer(t)
auth := testToken
valid := []string{
"lvmh-worker-x",
"lvmh-worker-group--project-ab12cd",
"lvmh-worker-x:latest",
"lvmh-worker-x:1.2.3",
"lvmh-worker-x-y.z:tag-1_2",
}
for _, img := range valid {
if code, body := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/image", auth, `{"image":"`+img+`"}`); code != http.StatusOK {
t.Fatalf("image %q = %d %s, want 200", img, code, body)
}
}
invalid := []string{
"", // empty
"lvmh-worker-", // no name after the prefix
"lvmh-worker-:tag", // tag without a name
"lvmh-worker-X", // uppercase name part
"evil", // foreign namespace
"lvmh-worker-x;rm -rf /", // shell metachars
"lvmh-worker-x/y", // path separator
"lvmh-worker-x:tag with space", // space in tag
}
for _, img := range invalid {
code, body := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/image", auth, `{"image":"`+img+`"}`)
if code != http.StatusBadRequest || !strings.Contains(body, "invalid image name") {
t.Fatalf("image %q = %d %s, want 400 invalid image name", img, code, body)
}
}
// bad repo shapes (single-slash-safe: ServeMux collapses "//" via
// redirect before our handler sees the path)
for _, repo := range []string{
"noslash",
"group/proj!bad",
} {
code, body := apiReq(t, http.MethodPut, ts.URL+"/api/repos/"+repo+"/image", auth, `{"image":"lvmh-worker-x"}`)
if code != http.StatusBadRequest {
t.Fatalf("repo %q = %d %s, want 400", repo, code, body)
}
code, _ = apiReq(t, http.MethodDelete, ts.URL+"/api/repos/"+repo+"/image", auth, "")
if code != http.StatusBadRequest {
t.Fatalf("delete repo %q = %d, want 400", repo, code)
}
}
// not an image path at all
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/notimage", auth, `{"image":"lvmh-worker-x"}`); code != http.StatusBadRequest {
t.Fatal("non-image path under /api/repos must 400")
}
// malformed body
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/image", auth, `not json`); code != http.StatusBadRequest {
t.Fatal("bad body must 400")
}
// oversized body
big := `{"image":"` + strings.Repeat("x", 1<<20) + `"}`
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/image", auth, big); code != http.StatusBadRequest {
t.Fatal("oversized body must 400")
}
}
func TestAPIRepoImagesStoreFailure(t *testing.T) {
ts, store := newTestServer(t)
if err := store.Close(); err != nil {
t.Fatalf("close store: %v", err)
}
if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/repos", testToken, ""); code != http.StatusInternalServerError {
t.Fatal("GET /api/repos with broken store must 500")
}
if code, _ := apiReq(t, http.MethodPut, ts.URL+"/api/repos/g/p/image", testToken, `{"image":"lvmh-worker-x"}`); code != http.StatusInternalServerError {
t.Fatal("PUT with broken store must 500")
}
if code, _ := apiReq(t, http.MethodDelete, ts.URL+"/api/repos/g/p/image", testToken, ""); code != http.StatusInternalServerError {
t.Fatal("DELETE with broken store must 500")
}
}
+33 -6
View File
@@ -134,6 +134,7 @@ type Spawner struct {
reposDir string
dockerfile string
controlDockerfile string
buildContext string
containerURL string
network string
@@ -158,6 +159,7 @@ func NewSpawner(ctx context.Context, store *Store, hub *Hub, baseURL string) (*S
baseURL: strings.TrimRight(baseURL, "/"),
reposDir: envOr(envRepoDir, defaultRepoDir),
dockerfile: dockerfile,
controlDockerfile: envOr(envControlDockerfile, defaultControlDockerfile),
buildContext: envOr(envWorkerContext, filepath.Dir(filepath.Dir(dockerfile))),
containerURL: envOr(envContainerLVMHURL, defaultContainerURL),
network: envOr(envContainerNetwork, defaultNetwork),
@@ -244,7 +246,12 @@ func (s *Spawner) Start(ctx context.Context, repo, branch string) (SpawnResult,
}
func (s *Spawner) imageExists(ctx context.Context) (bool, error) {
args := filters.NewArgs(filters.Arg("reference", imageRefWorker))
return s.imageRefExists(ctx, imageRefWorker)
}
// imageRefExists reports whether ref is present in the local docker store.
func (s *Spawner) imageRefExists(ctx context.Context, ref string) (bool, error) {
args := filters.NewArgs(filters.Arg("reference", ref))
summaries, err := s.cli.ImageList(ctx, image.ListOptions{Filters: args})
if err != nil {
return false, err
@@ -391,19 +398,26 @@ func (s *Spawner) ensureImage(ctx context.Context) error {
if _, err := os.Stat(s.dockerfile); err != nil {
return fmt.Errorf("worker Dockerfile missing at %s: %w", s.dockerfile, err)
}
relDockerfile, err := filepath.Rel(s.buildContext, s.dockerfile)
return s.buildImage(ctx, s.buildContext, s.dockerfile, imageRefWorker)
}
// buildImage streams a tar of buildContext into docker ImageBuild, building
// tag from dockerfile (relative to buildContext). Shared by the worker and
// ops image paths.
func (s *Spawner) buildImage(ctx context.Context, buildContext, dockerfile, tag string) error {
relDockerfile, err := filepath.Rel(buildContext, dockerfile)
if err != nil {
return err
}
pr, pw := io.Pipe()
tarDone := make(chan error, 1)
go func() {
err := tarDir(pw, s.buildContext)
err := tarDir(pw, buildContext)
_ = pw.CloseWithError(err)
tarDone <- err
}()
resp, buildErr := s.cli.ImageBuild(ctx, pr, build.ImageBuildOptions{
Tags: []string{imageRefWorker},
Tags: []string{tag},
Dockerfile: relDockerfile,
Remove: true,
})
@@ -414,7 +428,7 @@ func (s *Spawner) ensureImage(ctx context.Context) error {
if resp.Body != nil {
_ = resp.Body.Close()
}
return fmt.Errorf("build context %s: %w", s.buildContext, tarErr)
return fmt.Errorf("build context %s: %w", buildContext, tarErr)
}
if buildErr != nil {
return fmt.Errorf("docker build: %w", buildErr)
@@ -450,7 +464,20 @@ func extractBuildError(body []byte) string {
}
// createAndStart provisions volumes, creates and starts the worker container.
// A repo-registered custom image (see /api/repos) overrides the default
// worker image; the ops agent builds and registers those.
func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID string) (string, error) {
image := imageRefWorker
if custom, ok, _ := s.store.GetRepoImage(repo); ok && custom != "" {
exists, err := s.imageRefExists(ctx, custom)
if err != nil {
return "", fmt.Errorf("docker unavailable: %w", err)
}
if !exists {
return "", fmt.Errorf("custom image %s not built (ask the ops agent to build it)", custom)
}
image = custom
}
repoVolume := volumeRepoPrefix + slug
fresh, err := s.ensureRepoVolume(ctx, repoVolume)
if err != nil {
@@ -479,7 +506,7 @@ func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID stri
volumePiCache + ":" + cacheMount,
}
cfg := &container.Config{
Image: imageRefWorker,
Image: image,
Env: []string{
envProviderAPIKey + "=" + os.Getenv(envProviderAPIKey),
envToken + "=" + daemonToken,
+97 -7
View File
@@ -36,6 +36,7 @@ type recordedCreate struct {
Env []string `json:"Env"`
Labels map[string]string `json:"Labels"`
Cmd []string `json:"Cmd"`
WorkingDir string `json:"WorkingDir"`
HostConfig struct {
Binds []string `json:"Binds"`
NetworkMode string `json:"NetworkMode"`
@@ -47,12 +48,16 @@ type fakeDocker struct {
mu sync.Mutex
calls []dockerCall
images int // entries served by GET /images/json
images int // legacy: >0 serves the default worker image
imageTags map[string]bool // when set, /images/json serves exactly these
volume map[string]bool // existing volumes
nextID int
create []recordedCreate
archives []int // sizes of accepted CopyToContainer tar streams
containers map[string]string // name-or-id → container id
running map[string]bool // container id → running
failBuild bool
failBuildHTTP bool
failCreate bool
@@ -70,7 +75,23 @@ type fakeDocker struct {
}
func newFakeDocker() *fakeDocker {
return &fakeDocker{images: 1, volume: map[string]bool{}}
return &fakeDocker{images: 1, volume: map[string]bool{},
containers: map[string]string{}, running: map[string]bool{}}
}
// containerID resolves a name-or-id path token to the tracked container id.
func (f *fakeDocker) containerID(token string) (string, bool) {
f.mu.Lock()
defer f.mu.Unlock()
id, ok := f.containers[token]
return id, ok
}
// isRunning reports the running flag of a tracked container id.
func (f *fakeDocker) isRunning(id string) bool {
f.mu.Lock()
defer f.mu.Unlock()
return f.running[id]
}
func (f *fakeDocker) server(t *testing.T) *httptest.Server {
@@ -82,6 +103,43 @@ func (f *fakeDocker) server(t *testing.T) *httptest.Server {
// record appends a call; the /vX.Y version prefix negotiated by the SDK is
// stripped so assertions are version-agnostic.
// imageListJSON builds the GET /images/json entry list (joined with commas)
// honoring the reference filter. Legacy mode (imageTags nil): images>0 means
// exactly the default worker image is present.
func (f *fakeDocker) imageListJSON(filtersJSON string) string {
matches := func(ref string) bool {
if f.imageTags == nil {
return f.images > 0 && ref == imageRefWorker
}
return f.imageTags[ref]
}
var refs []string
if filtersJSON != "" {
var flt struct {
Reference map[string]bool `json:"reference"`
}
if err := json.Unmarshal([]byte(filtersJSON), &flt); err != nil {
return ""
}
for ref, want := range flt.Reference {
if want {
refs = append(refs, ref)
}
}
} else if f.imageTags != nil {
for ref := range f.imageTags {
refs = append(refs, ref)
}
}
entries := make([]string, 0, len(refs))
for _, ref := range refs {
if matches(ref) {
entries = append(entries, fmt.Sprintf(`{"Id":"sha256:abc","RepoTags":[%q]}`, ref))
}
}
return strings.Join(entries, ",")
}
func (f *fakeDocker) record(r *http.Request, body string) dockerCall {
path := r.URL.Path
if strings.HasPrefix(path, "/v") {
@@ -171,11 +229,7 @@ func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Api-Version", "1.44")
w.WriteHeader(http.StatusOK)
case call.Method == http.MethodGet && call.Path == "/images/json":
list := "[]"
if f.images > 0 {
list = `[{"Id":"sha256:abc","RepoTags":["lvmh-worker:latest"]}]`
}
writeJSONNow(w, http.StatusOK, list)
writeJSONNow(w, http.StatusOK, "["+f.imageListJSON(r.URL.Query().Get("filters"))+"]")
case call.Method == http.MethodPost && call.Path == "/build":
if f.failBuildHTTP {
writeJSONNow(w, http.StatusInternalServerError, `{"message":"build endpoint broken"}`)
@@ -238,19 +292,47 @@ func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
f.nextID++
id := fmt.Sprintf("cid-%d", f.nextID)
f.create = append(f.create, rc)
f.containers[id] = id
if rc.Name != "" {
f.containers[rc.Name] = id
}
f.running[id] = false
f.mu.Unlock()
writeJSONNow(w, http.StatusCreated, fmt.Sprintf(`{"Id":%q,"Warnings":null}`, id))
case call.Method == http.MethodGet && strings.HasPrefix(call.Path, "/containers/") && strings.HasSuffix(call.Path, "/json"):
token := strings.TrimSuffix(strings.TrimPrefix(call.Path, "/containers/"), "/json")
f.mu.Lock()
id, ok := f.containers[token]
running := ok && f.running[id]
f.mu.Unlock()
if !ok {
writeJSONNow(w, http.StatusNotFound, fmt.Sprintf(`{"message":"No such container: %s"}`, token))
return
}
writeJSONNow(w, http.StatusOK, fmt.Sprintf(`{"Id":%q,"State":{"Running":%t}}`, id, running))
case call.Method == http.MethodPost && strings.HasSuffix(call.Path, "/start"):
if f.failStart {
writeJSONNow(w, http.StatusInternalServerError, `{"message":"start failed"}`)
return
}
token := strings.TrimSuffix(strings.TrimPrefix(call.Path, "/containers/"), "/start")
f.mu.Lock()
if id, ok := f.containers[token]; ok {
f.running[id] = true
}
f.mu.Unlock()
w.WriteHeader(http.StatusNoContent)
case call.Method == http.MethodPost && strings.HasSuffix(call.Path, "/stop"):
if f.failStop {
writeJSONNow(w, http.StatusInternalServerError, `{"message":"stop failed"}`)
return
}
token := strings.TrimSuffix(strings.TrimPrefix(call.Path, "/containers/"), "/stop")
f.mu.Lock()
if id, ok := f.containers[token]; ok {
f.running[id] = false
}
f.mu.Unlock()
w.WriteHeader(http.StatusNoContent)
case call.Method == http.MethodPost && strings.HasSuffix(call.Path, "/wait"):
if f.failWait {
@@ -277,6 +359,14 @@ func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
f.mu.Unlock()
w.WriteHeader(http.StatusOK)
case call.Method == http.MethodDelete && strings.HasPrefix(call.Path, "/containers/"):
token := strings.TrimPrefix(call.Path, "/containers/")
f.mu.Lock()
if id, ok := f.containers[token]; ok {
delete(f.containers, token)
delete(f.containers, id)
delete(f.running, id)
}
f.mu.Unlock()
w.WriteHeader(http.StatusNoContent)
default:
f.mu.Lock()
+1
View File
@@ -73,6 +73,7 @@ func run(args []string) error {
if err != nil {
return fmt.Errorf("docker client: %w", err)
}
spawner.StartOpsLoop(spawnerCtx)
srv := NewServer(store, hub, spawner, gitlab)
httpServer := &http.Server{
+155
View File
@@ -0,0 +1,155 @@
package main
// ops.go — the lvmh-ops control container: an agent pi with docker access
// (docker.sock bind) that prepares repos, builds per-repo worker images and
// registers them via /api/repos. Kept alive by StartOpsLoop.
import (
"context"
"fmt"
"log"
"os"
"path/filepath"
"time"
cerrdefs "github.com/containerd/errdefs"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/volume"
)
// Ops container wiring constants.
const (
opsImageRef string = "lvmh-ops:latest"
opsContainerName string = "lvmh-ops"
labelOps string = "lvmh.ops"
opsSessionID string = "lvmh-ops-control"
envControlDockerfile string = "LVMH_CONTROL_DOCKERFILE"
defaultControlDockerfile string = "/app/build/docker/control.Dockerfile"
opsRecheckInterval time.Duration = 30 * time.Second
volumeOpsWork string = "lvmh-ops-work"
opsMount string = "/ops"
opsAPIBase string = "http://lvmh:8686"
dockerSock string = "/var/run/docker.sock"
)
// EnsureOps guarantees the ops control container is running: a running
// container is a no-op, a stopped one is started, a missing one is built
// (image) + created + started. Errors are returned, never fatal — the ops
// loop logs them and retries on the next tick (docker may be down at boot).
func (s *Spawner) EnsureOps(ctx context.Context) error {
inspect, err := s.cli.ContainerInspect(ctx, opsContainerName)
if err == nil {
if inspect.State != nil && inspect.State.Running {
return nil
}
return s.cli.ContainerStart(ctx, opsContainerName, container.StartOptions{})
}
if !cerrdefs.IsNotFound(err) {
return fmt.Errorf("inspect %s: %w", opsContainerName, err)
}
if err := s.ensureOpsImage(ctx); err != nil {
return err
}
return s.createOps(ctx)
}
// ensureOpsImage builds the ops image when missing, from the control
// Dockerfile (build context derived like the worker's: dir above the
// dockerfile's dir).
func (s *Spawner) ensureOpsImage(ctx context.Context) error {
exists, err := s.imageRefExists(ctx, opsImageRef)
if err != nil {
return err
}
if exists {
return nil
}
if _, err := os.Stat(s.controlDockerfile); err != nil {
return fmt.Errorf("control Dockerfile missing at %s: %w", s.controlDockerfile, err)
}
return s.buildImage(ctx, filepath.Dir(filepath.Dir(s.controlDockerfile)), s.controlDockerfile, opsImageRef)
}
// createOps creates and starts the ops container with its own workspace
// volume, the shared session/pi caches, and the docker socket. Never
// auto-removed: the loop restarts it in place.
func (s *Spawner) createOps(ctx context.Context) error {
for _, vol := range []string{volumeOpsWork, volumeSessions, volumePiCache} {
if _, err := s.cli.VolumeCreate(ctx, volume.CreateOptions{Name: vol}); err != nil {
return fmt.Errorf("volume %s: %w", vol, err)
}
}
env := []string{
envProviderAPIKey + "=" + os.Getenv(envProviderAPIKey),
envToken + "=" + daemonToken,
"LVMH_URL=" + s.containerURL,
envLVMHSessionID + "=" + opsSessionID,
envLVMHAgent + "=1",
"LVMH_API=" + opsAPIBase,
}
if pat, ok, _ := s.store.GetSetting(settingGitLabToken); ok && pat != "" {
env = append(env, "LVMH_GITEA_TOKEN="+pat)
}
cfg := &container.Config{
Image: opsImageRef,
Env: env,
WorkingDir: opsMount,
Labels: map[string]string{labelOps: "true", labelSession: opsSessionID},
}
hostCfg := &container.HostConfig{
Binds: []string{
volumeOpsWork + ":" + opsMount,
volumeSessions + ":" + sessionsMount,
volumePiCache + ":" + cacheMount,
dockerSock + ":" + dockerSock,
},
NetworkMode: container.NetworkMode(s.network),
AutoRemove: false,
Init: &[]bool{true}[0],
}
if _, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, opsContainerName); err != nil {
return fmt.Errorf("docker create: %w", err)
}
if err := s.cli.ContainerStart(ctx, opsContainerName, container.StartOptions{}); err != nil {
s.removeContainer(context.Background(), opsContainerName)
return fmt.Errorf("docker start: %w", err)
}
return nil
}
// StartOpsLoop keeps the ops container alive, re-checking every
// opsRecheckInterval. Ensure failures are logged, never fatal.
func (s *Spawner) StartOpsLoop(ctx context.Context) {
go func() {
if err := s.EnsureOps(ctx); err != nil {
log.Printf("ops: ensure: %v", err)
}
ticker := time.NewTicker(opsRecheckInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := s.EnsureOps(ctx); err != nil {
log.Printf("ops: ensure: %v", err)
}
}
}
}()
}
// RemoveOps stops and removes the ops container by NAME (it has no
// containers-table row, so RemoveSession cannot handle it). The ops loop
// recreates it on the next tick.
func (s *Spawner) RemoveOps(_ context.Context) error {
stopCtx, cancel := context.WithTimeout(context.Background(), time.Duration(stopTimeoutSeconds)*time.Second)
if err := s.cli.ContainerStop(stopCtx, opsContainerName, container.StopOptions{}); err != nil && !cerrdefs.IsNotFound(err) {
cancel()
return fmt.Errorf("docker stop: %w", err)
}
cancel()
s.removeContainer(context.Background(), opsContainerName)
return nil
}
+328
View File
@@ -0,0 +1,328 @@
package main
// ops_test.go — lvmh-ops control container: ensure/start/create paths and
// the delete-container special case, against the fake docker API.
import (
"context"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
"github.com/docker/docker/api/types/container"
)
// newOpsSpawner wires a spawner whose control Dockerfile exists on disk.
func newOpsSpawner(t *testing.T, f *fakeDocker) (*Spawner, *Store) {
t.Helper()
buildCtx := t.TempDir()
control := filepath.Join(buildCtx, "docker", "control.Dockerfile")
if err := os.MkdirAll(filepath.Dir(control), 0o755); err != nil {
t.Fatalf("mkdir control dir: %v", err)
}
if err := os.WriteFile(control, []byte("FROM node:24\n"), 0o644); err != nil {
t.Fatalf("write control dockerfile: %v", err)
}
t.Setenv(envControlDockerfile, control)
return newTestSpawner(t, f)
}
func TestEnsureOpsCreatesAndStarts(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{imageRefWorker: true, opsImageRef: true} // no build
sp, _ := newOpsSpawner(t, f)
if err := sp.EnsureOps(context.Background()); err != nil {
t.Fatalf("EnsureOps: %v", err)
}
creates := f.createsByName(opsContainerName)
if len(creates) != 1 {
t.Fatalf("ops creates = %+v, want 1", creates)
}
c := creates[0]
if c.Image != opsImageRef {
t.Fatalf("image = %q, want %q", c.Image, opsImageRef)
}
if c.WorkingDir != opsMount {
t.Fatalf("workdir = %q, want %q", c.WorkingDir, opsMount)
}
wantLabels := map[string]string{labelOps: "true", labelSession: opsSessionID}
if !reflect.DeepEqual(c.Labels, wantLabels) {
t.Fatalf("labels = %v, want %v", c.Labels, wantLabels)
}
wantEnv := map[string]string{
envProviderAPIKey: "key-123",
envToken: testToken,
"LVMH_URL": defaultContainerURL,
envLVMHSessionID: opsSessionID,
envLVMHAgent: "1",
"LVMH_API": opsAPIBase,
}
for _, e := range c.Env {
for k, want := range wantEnv {
if e == k+"="+want {
delete(wantEnv, k)
}
}
if e == "LVMH_GITEA_TOKEN=" {
t.Fatal("LVMH_GITEA_TOKEN present without stored PAT")
}
}
if len(wantEnv) != 0 {
t.Fatalf("missing env %v in %v", wantEnv, c.Env)
}
wantBinds := []string{
volumeOpsWork + ":" + opsMount,
volumeSessions + ":" + sessionsMount,
volumePiCache + ":" + cacheMount,
dockerSock + ":" + dockerSock,
}
if !reflect.DeepEqual(c.HostConfig.Binds, wantBinds) {
t.Fatalf("binds = %v, want %v", c.HostConfig.Binds, wantBinds)
}
if c.HostConfig.NetworkMode != defaultNetwork {
t.Fatalf("network = %q", c.HostConfig.NetworkMode)
}
if c.HostConfig.Init == nil || !*c.HostConfig.Init {
t.Fatalf("Init = %v, want true", c.HostConfig.Init)
}
if !f.hasCallSuffix(http.MethodPost, "/containers/"+opsContainerName+"/start") {
t.Fatal("ops container not started")
}
for _, vol := range []string{volumeOpsWork, volumeSessions, volumePiCache} {
if !f.volumeExists(vol) {
t.Fatalf("volume %q missing", vol)
}
}
f.assertNoUnknown(t)
}
func TestEnsureOpsEnvCarriesPAT(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{opsImageRef: true}
sp, store := newOpsSpawner(t, f)
if err := store.SetSetting(settingGitLabToken, "pat-ops"); err != nil {
t.Fatalf("set PAT: %v", err)
}
if err := sp.EnsureOps(context.Background()); err != nil {
t.Fatalf("EnsureOps: %v", err)
}
creates := f.createsByName(opsContainerName)
found := false
for _, e := range creates[0].Env {
if e == "LVMH_GITEA_TOKEN=pat-ops" {
found = true
}
}
if !found {
t.Fatalf("LVMH_GITEA_TOKEN missing from %v", creates[0].Env)
}
f.assertNoUnknown(t)
}
func TestEnsureOpsSkipsWhenRunning(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{opsImageRef: true}
sp, _ := newOpsSpawner(t, f)
ctx := context.Background()
if err := sp.EnsureOps(ctx); err != nil {
t.Fatalf("first EnsureOps: %v", err)
}
if err := sp.EnsureOps(ctx); err != nil {
t.Fatalf("second EnsureOps: %v", err)
}
if got := len(f.createsByName(opsContainerName)); got != 1 {
t.Fatalf("ops creates = %d, want 1 (running is a no-op)", got)
}
f.assertNoUnknown(t)
}
func TestEnsureOpsStartsStopped(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{opsImageRef: true}
sp, _ := newOpsSpawner(t, f)
ctx := context.Background()
if err := sp.EnsureOps(ctx); err != nil {
t.Fatalf("create: %v", err)
}
stopCtx, cancel := context.WithTimeout(ctx, time.Second)
defer cancel()
if err := sp.cli.ContainerStop(stopCtx, opsContainerName, container.StopOptions{}); err != nil {
t.Fatalf("stop: %v", err)
}
if err := sp.EnsureOps(ctx); err != nil {
t.Fatalf("EnsureOps stopped: %v", err)
}
if got := len(f.createsByName(opsContainerName)); got != 1 {
t.Fatalf("ops creates = %d, want 1 (start only)", got)
}
if !f.isRunning(f.containers[opsContainerName]) {
t.Fatal("ops container must be running again")
}
f.assertNoUnknown(t)
}
func TestEnsureOpsBuildsImageWhenMissing(t *testing.T) {
f := newFakeDocker() // legacy mode: only lvmh-worker present → ops missing
sp, _ := newOpsSpawner(t, f)
if err := sp.EnsureOps(context.Background()); err != nil {
t.Fatalf("EnsureOps: %v", err)
}
if n := f.countCalls(http.MethodPost, "/build"); n != 1 {
t.Fatalf("build calls = %d, want 1", n)
}
if got := len(f.createsByName(opsContainerName)); got != 1 {
t.Fatalf("ops creates = %d, want 1", got)
}
f.assertNoUnknown(t)
}
func TestEnsureOpsErrors(t *testing.T) {
// control Dockerfile missing → clear error
f := newFakeDocker()
sp, _ := newOpsSpawner(t, f)
sp.controlDockerfile = filepath.Join(t.TempDir(), "gone.Dockerfile")
if err := sp.EnsureOps(context.Background()); err == nil || !strings.Contains(err.Error(), "control Dockerfile missing") {
t.Fatalf("EnsureOps without dockerfile = %v", err)
}
// docker dead → error returned (caller logs; daemon keeps running)
sp2, _ := newOpsSpawner(t, newFakeDocker())
sp2.images0AndDead(t)
if err := sp2.EnsureOps(context.Background()); err == nil {
t.Fatal("EnsureOps with dead docker must error")
}
}
func TestEnsureOpsStartFailureCleansUp(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{opsImageRef: true}
f.failStart = true
sp, _ := newOpsSpawner(t, f)
if err := sp.EnsureOps(context.Background()); err == nil || !strings.Contains(err.Error(), "docker start") {
t.Fatalf("EnsureOps with failing start = %v", err)
}
if !f.hasCall(http.MethodDelete, "/containers/"+opsContainerName) {
t.Fatal("failed ops start must remove the created container")
}
}
func TestStartOpsLoopCreatesAndExits(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{opsImageRef: true}
sp, _ := newOpsSpawner(t, f)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
sp.StartOpsLoop(ctx)
waitFor(t, 5*time.Second, func() bool { return len(f.createsByName(opsContainerName)) == 1 })
cancel()
}
// newOpsAPIServer builds a full Server over a spawner with a control
// Dockerfile on disk; returns the spawner for EnsureOps calls.
func newOpsAPIServer(t *testing.T) (*httptest.Server, *fakeDocker, *Spawner) {
t.Helper()
useFakeGit(t, fakeGitModeOK)
f := newFakeDocker()
ts := f.server(t)
t.Setenv("DOCKER_HOST", "tcp://"+ts.Listener.Addr().String())
buildCtx := t.TempDir()
worker := filepath.Join(buildCtx, "docker", "worker.Dockerfile")
if err := os.MkdirAll(filepath.Dir(worker), 0o755); err != nil {
t.Fatalf("mkdir: %v", err)
}
if err := os.WriteFile(worker, []byte("FROM scratch\n"), 0o644); err != nil {
t.Fatalf("write: %v", err)
}
t.Setenv(envWorkerDockerfile, worker)
t.Setenv(envWorkerContext, buildCtx)
t.Setenv(envRepoDir, t.TempDir())
control := filepath.Join(buildCtx, "docker", "control.Dockerfile")
if err := os.WriteFile(control, []byte("FROM node:24\n"), 0o644); err != nil {
t.Fatalf("write control: %v", err)
}
t.Setenv(envControlDockerfile, control)
daemonToken = testToken
store := openTestStore(t)
hub := NewHub(store)
sp, err := NewSpawner(context.Background(), store, hub, "https://gitlab.example")
if err != nil {
t.Fatalf("NewSpawner: %v", err)
}
api := httptest.NewServer(NewServer(store, hub, sp, NewGitLab(store, "https://gitlab.example")).Routes(""))
t.Cleanup(api.Close)
return api, f, sp
}
// TestAPIDeleteOpsContainer: DELETE /api/sessions/lvmh-ops-control/container
// stops+removes the ops container by NAME (it has no containers-table row)
// and returns {ok:true}.
func TestAPIDeleteOpsContainer(t *testing.T) {
ts, f, sp := newOpsAPIServer(t)
f.imageTags = map[string]bool{opsImageRef: true}
if err := sp.EnsureOps(context.Background()); err != nil {
t.Fatalf("EnsureOps: %v", err)
}
if len(f.createsByName(opsContainerName)) != 1 {
t.Fatal("ops container missing before delete")
}
code, body := apiReq(t, http.MethodDelete, ts.URL+"/api/sessions/"+opsSessionID+"/container", testToken, "")
if code != http.StatusOK || !strings.Contains(body, `"ok":true`) {
t.Fatalf("delete ops = %d %s", code, body)
}
if !f.hasCall(http.MethodPost, "/containers/"+opsContainerName+"/stop") {
t.Fatal("ops must be stopped by name")
}
if !f.hasCall(http.MethodDelete, "/containers/"+opsContainerName) {
t.Fatal("ops must be removed by name")
}
if _, ok := f.containerID(opsContainerName); ok {
t.Fatal("ops container must be gone from the fake registry")
}
// the loop recreates it on the next tick (manual call here)
if err := sp.EnsureOps(context.Background()); err != nil {
t.Fatalf("recreate: %v", err)
}
if got := len(f.createsByName(opsContainerName)); got != 2 {
t.Fatalf("ops creates = %d, want 2 after recreate", got)
}
f.assertNoUnknown(t)
}
func TestRemoveOpsStopFailureAndAbsence(t *testing.T) {
f := newFakeDocker()
sp, _ := newOpsSpawner(t, f)
// absent container: stop+remove are tolerated, still ok
if err := sp.RemoveOps(context.Background()); err != nil {
t.Fatalf("RemoveOps absent = %v, want nil", err)
}
// stop failure surfaces
f.failStop = true
if err := sp.RemoveOps(context.Background()); err == nil || !strings.Contains(err.Error(), "docker stop") {
t.Fatalf("RemoveOps with stop failure = %v, want docker stop error", err)
}
}
func TestEnsureOpsVolumeCreateFailure(t *testing.T) {
f := newFakeDocker()
f.imageTags = map[string]bool{opsImageRef: true}
f.failVolumeCreate = true
sp, _ := newOpsSpawner(t, f)
if err := sp.EnsureOps(context.Background()); err == nil || !strings.Contains(err.Error(), volumeOpsWork) {
t.Fatalf("EnsureOps with volume failure = %v, want %q error", err, volumeOpsWork)
}
if len(f.createsByName(opsContainerName)) != 0 {
t.Fatal("no ops container may be created when volumes fail")
}
}
+96
View File
@@ -1187,3 +1187,99 @@ func TestSpawnerStartDedupesImageList(t *testing.T) {
}
f.assertNoUnknown(t)
}
// TestSpawnerCustomImageUsed: a repo-registered image replaces the default
// worker image in the created container.
func TestSpawnerCustomImageUsed(t *testing.T) {
useFakeGit(t, fakeGitModeOK)
f := newFakeDocker()
custom := "lvmh-worker-group--project-ab12cd"
f.imageTags = map[string]bool{imageRefWorker: true, custom: true}
sp, store := newTestSpawner(t, f)
if err := store.SetRepoImage("group/project", custom); err != nil {
t.Fatalf("set repo image: %v", err)
}
res, err := sp.Start(context.Background(), "group/project", "")
if err != nil {
t.Fatalf("Start: %v", err)
}
waitJobState(t, sp, res.SessionID, stateRunning)
creates := f.createsByName("lvmh-agent-")
if len(creates) != 1 || creates[0].Image != custom {
t.Fatalf("agent create image = %+v, want %q", creates[0], custom)
}
// seed container still uses the default worker image
seed := f.createsByName("lvmh-seed-")
if len(seed) != 1 || seed[0].Image != imageRefWorker {
t.Fatalf("seed create image = %+v, want default %q", seed[0], imageRefWorker)
}
f.assertNoUnknown(t)
}
// TestSpawnerCustomImageMissing: registering an image that was never built
// fails the job with the ask-ops message instead of a raw docker error.
func TestSpawnerCustomImageMissing(t *testing.T) {
useFakeGit(t, fakeGitModeOK)
f := newFakeDocker()
f.imageTags = map[string]bool{imageRefWorker: true} // custom absent
sp, store := newTestSpawner(t, f)
custom := "lvmh-worker-group--project-ab12cd"
if err := store.SetRepoImage("group/project", custom); err != nil {
t.Fatalf("set repo image: %v", err)
}
res, err := sp.Start(context.Background(), "group/project", "")
if err != nil {
t.Fatalf("Start: %v", err)
}
job := waitJobState(t, sp, res.SessionID, stateError)
want := "custom image " + custom + " not built (ask the ops agent to build it)"
if job.Message != want {
t.Fatalf("job message = %q, want %q", job.Message, want)
}
if len(f.createsByName("lvmh-agent-")) != 0 {
t.Fatal("no agent container may be created for a missing image")
}
f.assertNoUnknown(t)
}
// TestSpawnerCustomImageDeleteFallsBack: deleting the registration returns
// the repo to the default worker image.
func TestSpawnerCustomImageDeleteFallsBack(t *testing.T) {
useFakeGit(t, fakeGitModeOK)
f := newFakeDocker()
sp, store := newTestSpawner(t, f)
if err := store.SetRepoImage("group/project", "lvmh-worker-x"); err != nil {
t.Fatalf("set repo image: %v", err)
}
if err := store.DeleteRepoImage("group/project"); err != nil {
t.Fatalf("delete repo image: %v", err)
}
res, err := sp.Start(context.Background(), "group/project", "")
if err != nil {
t.Fatalf("Start: %v", err)
}
waitJobState(t, sp, res.SessionID, stateRunning)
creates := f.createsByName("lvmh-agent-")
if len(creates) != 1 || creates[0].Image != imageRefWorker {
t.Fatalf("create after deregister = %+v, want default %q", creates[0], imageRefWorker)
}
f.assertNoUnknown(t)
}
// TestSpawnerCustomImageCheckDockerDead: the custom-image presence check
// wraps docker unavailability like Start does.
func TestSpawnerCustomImageCheckDockerDead(t *testing.T) {
f := newFakeDocker()
sp, store := newTestSpawner(t, f)
if err := store.SetRepoImage("group/project", "lvmh-worker-x"); err != nil {
t.Fatalf("set repo image: %v", err)
}
sp.images0AndDead(t)
_, err := sp.createAndStart(context.Background(), "group/project", repoSlug("group/project"), "s1")
if err == nil || !strings.Contains(err.Error(), "docker unavailable") {
t.Fatalf("createAndStart with dead docker = %v, want docker unavailable", err)
}
}
+52
View File
@@ -47,6 +47,12 @@ type ContainerRow struct {
Repo string
}
// RepoImageRow maps a repo path to its registered custom worker image.
type RepoImageRow struct {
Repo string `json:"repo"`
Image string `json:"image"`
}
const dsnParams string = "?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)"
const schema string = `
@@ -59,6 +65,8 @@ CREATE TABLE IF NOT EXISTS sessions(
CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS containers(
sessionId TEXT PRIMARY KEY, containerId TEXT NOT NULL, repo TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS repo_images(
repo TEXT PRIMARY KEY, image TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS idx_events_session ON events(sessionId, seq);
`
@@ -278,3 +286,47 @@ func (s *Store) DeleteContainer(sessionID string) error {
_, err := s.db.Exec(`DELETE FROM containers WHERE sessionId=?`, sessionID)
return err
}
// GetRepoImage returns the custom image registered for a repo, if any.
func (s *Store) GetRepoImage(repo string) (string, bool, error) {
var image string
err := s.db.QueryRow(`SELECT image FROM repo_images WHERE repo=?`, repo).Scan(&image)
if err == sql.ErrNoRows {
return "", false, nil
}
if err != nil {
return "", false, err
}
return image, true, nil
}
// SetRepoImage upserts the custom image for a repo.
func (s *Store) SetRepoImage(repo, image string) error {
_, err := s.db.Exec(`INSERT INTO repo_images(repo, image) VALUES(?,?)
ON CONFLICT(repo) DO UPDATE SET image=excluded.image`, repo, image)
return err
}
// DeleteRepoImage removes a repo's custom image registration (idempotent).
func (s *Store) DeleteRepoImage(repo string) error {
_, err := s.db.Exec(`DELETE FROM repo_images WHERE repo=?`, repo)
return err
}
// ListRepoImages returns every repo→image registration, sorted by repo.
func (s *Store) ListRepoImages() ([]RepoImageRow, error) {
rows, err := s.db.Query(`SELECT repo, image FROM repo_images ORDER BY repo`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []RepoImageRow
for rows.Next() {
var row RepoImageRow
if err := rows.Scan(&row.Repo, &row.Image); err != nil {
return nil, err
}
out = append(out, row)
}
return out, rows.Err()
}
+12
View File
@@ -64,6 +64,18 @@ func TestStoreClosedHandleErrors(t *testing.T) {
if err := store.DeleteContainer("s"); errStr(err) == "" {
t.Fatal("DeleteContainer on closed store must error")
}
if _, _, err := store.GetRepoImage("r"); errStr(err) == "" {
t.Fatal("GetRepoImage on closed store must error")
}
if err := store.SetRepoImage("r", "lvmh-worker-r"); errStr(err) == "" {
t.Fatal("SetRepoImage on closed store must error")
}
if err := store.DeleteRepoImage("r"); errStr(err) == "" {
t.Fatal("DeleteRepoImage on closed store must error")
}
if _, err := store.ListRepoImages(); errStr(err) == "" {
t.Fatal("ListRepoImages on closed store must error")
}
}
func TestStoreCorruptRows(t *testing.T) {
+53
View File
@@ -4,6 +4,7 @@ package main
import (
"path/filepath"
"reflect"
"testing"
)
@@ -132,3 +133,55 @@ func TestStoreSettingsAndContainers(t *testing.T) {
t.Fatal("container row should be gone")
}
}
func TestStoreRepoImages(t *testing.T) {
store := openTestStore(t)
if _, ok, err := store.GetRepoImage("group/proj"); err != nil || ok {
t.Fatalf("missing repo image = %v %v, want absent no-error", ok, err)
}
if rows, err := store.ListRepoImages(); err != nil || rows != nil {
t.Fatalf("empty list = %v %v, want nil no-error", rows, err)
}
if err := store.SetRepoImage("group/proj", "lvmh-worker-g--p-ab12cd"); err != nil {
t.Fatalf("set: %v", err)
}
if img, ok, _ := store.GetRepoImage("group/proj"); !ok || img != "lvmh-worker-g--p-ab12cd" {
t.Fatalf("get = %q %v", img, ok)
}
// upsert replaces
if err := store.SetRepoImage("group/proj", "lvmh-worker-g--p-ff99ee"); err != nil {
t.Fatalf("upsert: %v", err)
}
if img, _, _ := store.GetRepoImage("group/proj"); img != "lvmh-worker-g--p-ff99ee" {
t.Fatalf("after upsert = %q", img)
}
if err := store.SetRepoImage("other/repo", "lvmh-worker-o--r-11aabb"); err != nil {
t.Fatalf("set second: %v", err)
}
rows, err := store.ListRepoImages()
if err != nil {
t.Fatalf("list: %v", err)
}
want := []RepoImageRow{
{Repo: "group/proj", Image: "lvmh-worker-g--p-ff99ee"},
{Repo: "other/repo", Image: "lvmh-worker-o--r-11aabb"},
}
if !reflect.DeepEqual(rows, want) {
t.Fatalf("list = %+v, want %+v (sorted by repo)", rows, want)
}
if err := store.DeleteRepoImage("group/proj"); err != nil {
t.Fatalf("delete: %v", err)
}
if _, ok, _ := store.GetRepoImage("group/proj"); ok {
t.Fatal("repo image should be gone")
}
// idempotent delete
if err := store.DeleteRepoImage("group/proj"); err != nil {
t.Fatalf("delete absent: %v", err)
}
}
+2
View File
@@ -37,6 +37,7 @@ import { run as promptRouting } from "./scenarios/prompt-routing.mjs";
import { run as sessionList } from "./scenarios/session-list.mjs";
import { run as gitlab } from "./scenarios/gitlab.mjs";
import { run as spawnValidation } from "./scenarios/spawn-validation.mjs";
import { run as repoImages } from "./scenarios/repo-images.mjs";
import { run as pluginContract } from "./scenarios/plugin-contract.mjs";
import { run as resilience } from "./scenarios/resilience.mjs";
@@ -49,6 +50,7 @@ const SCENARIOS = [
["session-list", sessionList],
["gitlab", gitlab],
["spawn-validation", spawnValidation],
["repo-images", repoImages],
["plugin-contract", pluginContract],
["resilience", resilience], // last: SIGKILLs and reboots the daemon
];
+125
View File
@@ -0,0 +1,125 @@
// scenarios/repo-images.mjs — repo→custom worker image registry (REST CRUD).
// Spawn itself is NOT exercised here: it needs Docker, which the harness must
// not require (see spawn-validation.mjs).
import { rest } from "../lib.mjs";
export async function run(ctx) {
const { r, base, token } = ctx;
const unauth = await rest(base, null, "/api/repos");
r.check(
"GET /api/repos without token → 401",
unauth.status === 401,
`got ${unauth.status}`,
);
const empty = await rest(base, token, "/api/repos");
r.check(
"GET /api/repos on fresh daemon → 200 []",
empty.status === 200 &&
Array.isArray(empty.json) &&
empty.json.length === 0,
`got ${empty.status} ${empty.text}`,
);
const put = await rest(base, token, "/api/repos/lvmh/e2e/image", {
method: "PUT",
body: { image: "lvmh-worker-lvmh--e2e-ab12cd" },
});
r.check(
"PUT valid repo/image → 200 {ok:true}",
put.status === 200 && put.json?.ok === true,
`got ${put.status} ${put.text}`,
);
const putTagged = await rest(base, token, "/api/repos/lvmh/e2e2/image", {
method: "PUT",
body: { image: "lvmh-worker-lvmh--e2e2-ab12cd:1.0" },
});
r.check(
"PUT tagged image → 200",
putTagged.status === 200,
`got ${putTagged.status}`,
);
const list = await rest(base, token, "/api/repos");
const registrations = Array.isArray(list.json)
? list.json.filter((x) => x.repo === "lvmh/e2e" || x.repo === "lvmh/e2e2")
: [];
r.check(
"GET /api/repos lists both registrations",
list.status === 200 &&
registrations.length === 2 &&
registrations[0].repo === "lvmh/e2e" &&
registrations[0].image === "lvmh-worker-lvmh--e2e-ab12cd" &&
registrations[1].repo === "lvmh/e2e2" &&
registrations[1].image === "lvmh-worker-lvmh--e2e2-ab12cd:1.0",
`got ${list.status} ${list.text}`,
);
for (const image of ["evil-image", "", "lvmh-worker-", "LVMH-worker-x"]) {
const bad = await rest(base, token, "/api/repos/lvmh/e2e/image", {
method: "PUT",
body: { image },
});
r.check(
`PUT invalid image ${JSON.stringify(image)} → 400`,
bad.status === 400 && typeof bad.json?.error === "string",
`got ${bad.status} ${bad.text}`,
);
}
const badRepo = await rest(base, token, "/api/repos/noslash/image", {
method: "PUT",
body: { image: "lvmh-worker-x" },
});
r.check(
"PUT invalid repo → 400",
badRepo.status === 400,
`got ${badRepo.status}`,
);
const malformed = await fetch(`${base}/api/repos/lvmh/e2e/image`, {
method: "PUT",
headers: {
Authorization: `Bearer ${token}`,
"Content-Type": "application/json",
},
body: "{not json",
});
r.check(
"PUT malformed JSON → 400",
malformed.status === 400,
`got ${malformed.status}`,
);
await malformed.text();
const del = await rest(base, token, "/api/repos/lvmh/e2e/image", {
method: "DELETE",
});
r.check(
"DELETE image → 200 {ok:true}",
del.status === 200 && del.json?.ok === true,
`got ${del.status}`,
);
const delAgain = await rest(base, token, "/api/repos/lvmh/e2e/image", {
method: "DELETE",
});
r.check(
"DELETE absent image → 200 (idempotent)",
delAgain.status === 200,
`got ${delAgain.status}`,
);
const after = await rest(base, token, "/api/repos");
const remaining = Array.isArray(after.json)
? after.json.filter((x) => x.repo === "lvmh/e2e")
: null;
r.check(
"GET after delete no longer lists lvmh/e2e",
after.status === 200 && remaining?.length === 0,
`got ${after.status} ${after.text}`,
);
}