diff --git a/PROTOCOL.md b/PROTOCOL.md index ff279c5..df75007 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -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 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 | diff --git a/daemon/api.go b/daemon/api.go index 188ef84..d5f9111 100644 --- a/daemon/api.go +++ b/daemon/api.go @@ -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//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()) } diff --git a/daemon/api_extra_test.go b/daemon/api_extra_test.go index b8fd2ad..0140c65 100644 --- a/daemon/api_extra_test.go +++ b/daemon/api_extra_test.go @@ -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") + } +} diff --git a/daemon/docker.go b/daemon/docker.go index 3db0278..5c05585 100644 --- a/daemon/docker.go +++ b/daemon/docker.go @@ -132,11 +132,12 @@ type Spawner struct { cli *client.Client baseURL string // gitlab base, for clone URLs - reposDir string - dockerfile string - buildContext string - containerURL string - network string + reposDir string + dockerfile string + controlDockerfile string + buildContext string + containerURL string + network string mu sync.Mutex jobs map[string]*SpawnJob // keyed by sessionId @@ -151,18 +152,19 @@ func NewSpawner(ctx context.Context, store *Store, hub *Hub, baseURL string) (*S } dockerfile := envOr(envWorkerDockerfile, defaultDockerfile) return &Spawner{ - ctx: ctx, - store: store, - hub: hub, - cli: cli, - baseURL: strings.TrimRight(baseURL, "/"), - reposDir: envOr(envRepoDir, defaultRepoDir), - dockerfile: dockerfile, - buildContext: envOr(envWorkerContext, filepath.Dir(filepath.Dir(dockerfile))), - containerURL: envOr(envContainerLVMHURL, defaultContainerURL), - network: envOr(envContainerNetwork, defaultNetwork), - jobs: make(map[string]*SpawnJob), - slugLocks: make(map[string]*sync.Mutex), + ctx: ctx, + store: store, + hub: hub, + cli: cli, + 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), + jobs: make(map[string]*SpawnJob), + slugLocks: make(map[string]*sync.Mutex), }, nil } @@ -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, diff --git a/daemon/docker_test.go b/daemon/docker_test.go index 915d9ae..e3e371e 100644 --- a/daemon/docker_test.go +++ b/daemon/docker_test.go @@ -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"` @@ -46,12 +47,16 @@ type recordedCreate struct { type fakeDocker struct { mu sync.Mutex - calls []dockerCall - images int // entries served by GET /images/json - volume map[string]bool // existing volumes - nextID int - create []recordedCreate - archives []int // sizes of accepted CopyToContainer tar streams + calls []dockerCall + 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 @@ -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() diff --git a/daemon/main.go b/daemon/main.go index cf912cc..9caef5a 100644 --- a/daemon/main.go +++ b/daemon/main.go @@ -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{ diff --git a/daemon/ops.go b/daemon/ops.go new file mode 100644 index 0000000..83a76d8 --- /dev/null +++ b/daemon/ops.go @@ -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 +} diff --git a/daemon/ops_test.go b/daemon/ops_test.go new file mode 100644 index 0000000..3f47469 --- /dev/null +++ b/daemon/ops_test.go @@ -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") + } +} diff --git a/daemon/spawner_test.go b/daemon/spawner_test.go index 0d031ec..1d9c39c 100644 --- a/daemon/spawner_test.go +++ b/daemon/spawner_test.go @@ -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) + } +} diff --git a/daemon/store.go b/daemon/store.go index e327be7..3991440 100644 --- a/daemon/store.go +++ b/daemon/store.go @@ -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() +} diff --git a/daemon/store_extra_test.go b/daemon/store_extra_test.go index d1333f3..7a1972f 100644 --- a/daemon/store_extra_test.go +++ b/daemon/store_extra_test.go @@ -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) { diff --git a/daemon/store_test.go b/daemon/store_test.go index 22c05c4..167aa36 100644 --- a/daemon/store_test.go +++ b/daemon/store_test.go @@ -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) + } +} diff --git a/docker/ops-context/entrypoint.sh b/docker/ops-context/entrypoint.sh index ed59a05..d3132b8 100644 --- a/docker/ops-context/entrypoint.sh +++ b/docker/ops-context/entrypoint.sh @@ -3,7 +3,7 @@ # never overwrite user/agent modifications after the first time). set -eu if [ ! -f /ops/AGENTS.md ]; then - mkdir -p /ops - cp /ops-seed/AGENTS.md /ops/AGENTS.md + mkdir -p /ops + cp /ops-seed/AGENTS.md /ops/AGENTS.md fi exec "$@" diff --git a/e2e/driver.mjs b/e2e/driver.mjs index 51cb910..ee50aac 100644 --- a/e2e/driver.mjs +++ b/e2e/driver.mjs @@ -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 ]; diff --git a/e2e/scenarios/repo-images.mjs b/e2e/scenarios/repo-images.mjs new file mode 100644 index 0000000..84d0545 --- /dev/null +++ b/e2e/scenarios/repo-images.mjs @@ -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}`, + ); +}