From 6aac763563d6143dcd3695a37d73d8f08c564bed Mon Sep 17 00:00:00 2001 From: Raphael Westphal Date: Tue, 18 Aug 2026 18:49:52 +0200 Subject: [PATCH] fix(review round 1): daemon PAT-header auth, job pruning, session-split batches, streaming tars, seed cleanup, ping/pong, byte caps, branch switch, slug --, ctx cancel, Init:true, pagination, latest/before events; web newest-window history + load-older, offline busy gate, staleness guards, memo store, auth probe, scroll key, focus-visible, draft restore, poll dedup; 106+181 tests, coverage 96.2%/95.1%+ --- daemon/api.go | 24 +- daemon/api_extra_test.go | 49 ++- daemon/docker.go | 140 ++++-- daemon/docker_test.go | 20 +- daemon/gitlab.go | 43 +- daemon/gitlab_extra_test.go | 62 +++ daemon/hub.go | 134 ++++-- daemon/hub_extra_test.go | 223 ++++++++++ daemon/main.go | 6 +- daemon/spawner_test.go | 240 +++++++++- daemon/store.go | 43 +- daemon/store_extra_test.go | 54 +++ plugin/lvmh-agent.ts | 5 +- web/src/App.tsx | 167 +++---- web/src/ChatStream.test.tsx | 745 ++++++++++++++++++------------- web/src/ChatStream.tsx | 26 +- web/src/ChatView.test.tsx | 342 ++++++++++++++- web/src/ChatView.tsx | 60 ++- web/src/SessionsView.test.tsx | 58 ++- web/src/SessionsView.tsx | 6 +- web/src/SpawnView.test.tsx | 794 ++++++++++++++++------------------ web/src/SpawnView.tsx | 22 +- web/src/index.css | 13 + web/src/store.test.tsx | 193 ++++++++- web/src/store.ts | 54 ++- 25 files changed, 2564 insertions(+), 959 deletions(-) diff --git a/daemon/api.go b/daemon/api.go index 4b6ae8a..0428b0f 100644 --- a/daemon/api.go +++ b/daemon/api.go @@ -169,7 +169,27 @@ func (s *Server) handleSessionEvents(w http.ResponseWriter, r *http.Request) { if limit > maxEventsLimit { limit = maxEventsLimit } - events, err := s.store.EventsAfter(id, after, limit) + var ( + events []Event + err error + ) + switch { + case r.URL.Query().Get("latest") != "": + if r.URL.Query().Get("latest") != "1" { + writeError(w, http.StatusBadRequest, "invalid latest") + return + } + events, err = s.store.EventsLatest(id, limit) + case r.URL.Query().Get("before") != "": + before, perr := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64) + if perr != nil || before < 0 { + writeError(w, http.StatusBadRequest, "invalid before") + return + } + events, err = s.store.EventsBefore(id, before, limit) + default: + events, err = s.store.EventsAfter(id, after, limit) + } if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) return @@ -360,7 +380,5 @@ func (s *Server) webHandler(webdist string) http.Handler { }) } -var _ = daemonVersion - // daemonToken is set by main from LVMH_TOKEN (single process, read-only). var daemonToken string diff --git a/daemon/api_extra_test.go b/daemon/api_extra_test.go index 66a83ce..b8fd2ad 100644 --- a/daemon/api_extra_test.go +++ b/daemon/api_extra_test.go @@ -4,6 +4,7 @@ package main // serving, body/token validation edges. import ( + "context" "encoding/json" "io" "net/http" @@ -37,7 +38,7 @@ func newSpawnAPIServer(t *testing.T) (*httptest.Server, *fakeDocker) { daemonToken = testToken store := openTestStore(t) hub := NewHub(store) - sp, err := NewSpawner(store, hub, "https://gitlab.example") + sp, err := NewSpawner(context.Background(), store, hub, "https://gitlab.example") if err != nil { t.Fatalf("NewSpawner: %v", err) } @@ -139,7 +140,7 @@ func TestAPISpawnFailures(t *testing.T) { daemonToken = testToken store := openTestStore(t) hub := NewHub(store) - sp, err := NewSpawner(store, hub, "https://gitlab.example") + sp, err := NewSpawner(context.Background(), store, hub, "https://gitlab.example") if err != nil { t.Fatalf("NewSpawner: %v", err) } @@ -279,6 +280,50 @@ func TestAPIPromptBodyValidation(t *testing.T) { } } +func TestAPIEventsLatestAndBeforeParams(t *testing.T) { + ts, store := newTestServer(t) + auth := testToken + for seq := int64(1); seq <= 5; seq++ { + if err := store.AppendEvent(Event{SessionID: "s1", Seq: seq, TS: seq, Type: evAgentSettled, Payload: []byte(`{}`)}); err != nil { + t.Fatalf("append %d: %v", seq, err) + } + } + + getSeqs := func(query string) []int64 { + t.Helper() + code, body := apiReq(t, http.MethodGet, ts.URL+"/api/sessions/s1/events"+query, auth, "") + if code != http.StatusOK { + t.Fatalf("%s → %d %s, want 200", query, code, body) + } + var frames []map[string]any + if err := json.Unmarshal([]byte(body), &frames); err != nil { + t.Fatalf("decode %s: %v", query, err) + } + seqs := make([]int64, 0, len(frames)) + for _, f := range frames { + seqs = append(seqs, int64(f["seq"].(float64))) + } + return seqs + } + + if got := getSeqs("?latest=1&limit=2"); len(got) != 2 || got[0] != 4 || got[1] != 5 { + t.Fatalf("latest=1&limit=2 = %v, want [4 5] ascending", got) + } + if got := getSeqs("?before=4&limit=2"); len(got) != 2 || got[0] != 2 || got[1] != 3 { + t.Fatalf("before=4&limit=2 = %v, want [2 3] ascending", got) + } + // default behavior unchanged + if got := getSeqs("?after=0&limit=10"); len(got) != 5 || got[0] != 1 { + t.Fatalf("after=0 = %v, want full ascending replay", got) + } + + for _, q := range []string{"?before=abc", "?before=-1", "?latest=2"} { + if code, _ := apiReq(t, http.MethodGet, ts.URL+"/api/sessions/s1/events"+q, auth, ""); code != http.StatusBadRequest { + t.Fatalf("%s → %d, want 400", q, code) + } + } +} + func TestAPIEventsEdgeCases(t *testing.T) { ts, store := newTestServer(t) auth := testToken diff --git a/daemon/docker.go b/daemon/docker.go index a8686b6..34fb638 100644 --- a/daemon/docker.go +++ b/daemon/docker.go @@ -12,6 +12,7 @@ import ( "errors" "fmt" "io" + "log" "net/url" "os" "os/exec" @@ -61,6 +62,7 @@ const ( envLVMHRepo string = "LVMH_REPO" stopTimeoutSeconds int = 10 buildContextReadLimit int64 = 64 << 20 + maxSpawnJobs int = 50 // jobs map pruned to this many entries ) var errNoContainer = errors.New("no container for session") @@ -70,7 +72,7 @@ var repoPathRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+(/[A-Za-z0-9_.-]+)+$`) // validRepoPath accepts "group/project" style paths (at least two segments). func validRepoPath(repo string) bool { return repoPathRe.MatchString(repo) } -func repoSlug(repo string) string { return strings.ReplaceAll(repo, "/", "-") } +func repoSlug(repo string) string { return strings.ReplaceAll(repo, "/", "--") } // newUUID returns a random RFC 4122 v4 UUID string. func newUUID() string { @@ -112,6 +114,7 @@ type SpawnResult struct { // Spawner owns the docker client, per-repo clone serialization and job state. type Spawner struct { + ctx context.Context // base ctx for async jobs (cancelled on shutdown) store *Store hub *Hub cli *client.Client @@ -125,16 +128,18 @@ type Spawner struct { mu sync.Mutex jobs map[string]*SpawnJob // keyed by sessionId + jobOrder []string // insertion order of jobs, for pruning slugLocks map[string]*sync.Mutex } -func NewSpawner(store *Store, hub *Hub, baseURL string) (*Spawner, error) { +func NewSpawner(ctx context.Context, store *Store, hub *Hub, baseURL string) (*Spawner, error) { cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) if err != nil { return nil, fmt.Errorf("docker client: %w", err) } dockerfile := envOr(envWorkerDockerfile, defaultDockerfile) return &Spawner{ + ctx: ctx, store: store, hub: hub, cli: cli, @@ -173,6 +178,13 @@ func (s *Spawner) setJob(sessionID, repo, state, containerID, message string) { if !ok { j = &SpawnJob{SessionID: sessionID, Repo: repo} s.jobs[sessionID] = j + s.jobOrder = append(s.jobOrder, sessionID) + if len(s.jobOrder) > maxSpawnJobs { + oldest := s.jobOrder[0] + copy(s.jobOrder, s.jobOrder[1:]) + s.jobOrder = s.jobOrder[:len(s.jobOrder)-1] + delete(s.jobs, oldest) + } } j.Repo = repo j.State = state @@ -183,6 +195,23 @@ func (s *Spawner) setJob(sessionID, repo, state, containerID, message string) { s.hub.BroadcastSpawnStatus() } +// deleteJob drops a session's job from the map and the insertion order. +func (s *Spawner) deleteJob(sessionID string) { + s.mu.Lock() + if _, ok := s.jobs[sessionID]; !ok { + s.mu.Unlock() + return + } + delete(s.jobs, sessionID) + for i, id := range s.jobOrder { + if id == sessionID { + s.jobOrder = append(s.jobOrder[:i], s.jobOrder[i+1:]...) + break + } + } + s.mu.Unlock() +} + // Start launches the async spawn pipeline and returns the new sessionId. func (s *Spawner) Start(ctx context.Context, repo, branch string) (SpawnResult, error) { if _, err := s.imageExists(ctx); err != nil { @@ -234,12 +263,12 @@ func (s *Spawner) runJob(repo, branch, sessionID string) { return } s.setJob(sessionID, repo, stateBuilding, "", "") - if err := s.ensureImage(context.Background()); err != nil { + if err := s.ensureImage(s.ctx); err != nil { s.setJob(sessionID, repo, stateError, "", err.Error()) return } s.setJob(sessionID, repo, stateCreating, "", "") - containerID, err := s.createAndStart(context.Background(), repo, slug, sessionID) + containerID, err := s.createAndStart(s.ctx, repo, slug, sessionID) if err != nil { s.setJob(sessionID, repo, stateError, "", err.Error()) return @@ -252,6 +281,8 @@ func (s *Spawner) runJob(repo, branch, sessionID string) { } // cloneOrUpdate clones the repo into reposDir/ or fast-forwards it. +// Credentials never appear in the clone URL (which git persists into +// .git/config); auth is passed per-invocation via http.extraHeader. func (s *Spawner) cloneOrUpdate(repo, branch, slug string) error { dir := filepath.Join(s.reposDir, slug) cloneURL, err := s.cloneURL(repo) @@ -259,7 +290,16 @@ func (s *Spawner) cloneOrUpdate(repo, branch, slug string) error { return err } if st, err := os.Stat(filepath.Join(dir, ".git")); err == nil && st.IsDir() { - if err := gitRun(dir, "pull", "--ff-only"); err != nil { + auth := s.gitAuthArgs() + if branch != "" && gitBranch(dir) != branch { + if err := gitRun(dir, append(append([]string{}, auth...), "fetch", "origin", branch)...); err != nil { + return fmt.Errorf("git fetch %s %s: %w", repo, branch, err) + } + if err := gitRun(dir, "checkout", branch); err != nil { + return fmt.Errorf("git checkout %s: %w", repo, err) + } + } + if err := gitRun(dir, append(append([]string{}, auth...), "pull", "--ff-only")...); err != nil { return fmt.Errorf("git pull %s: %w", repo, err) } return nil @@ -267,7 +307,7 @@ func (s *Spawner) cloneOrUpdate(repo, branch, slug string) error { if err := os.MkdirAll(s.reposDir, 0o755); err != nil { return err } - args := []string{"clone"} + args := append(s.gitAuthArgs(), "clone") if branch != "" { args = append(args, "--branch", branch) } @@ -278,18 +318,34 @@ func (s *Spawner) cloneOrUpdate(repo, branch, slug string) error { return nil } -// cloneURL builds an authenticated https clone URL when a PAT is stored. +// cloneURL builds the clean https clone URL (never credential-bearing). func (s *Spawner) cloneURL(repo string) (string, error) { u, err := url.Parse(s.baseURL + "/" + repo + ".git") if err != nil { return "", err } - if token, ok, _ := s.store.GetSetting(settingGitLabToken); ok && token != "" && u.User == nil { - u.User = url.User(token) - } return u.String(), nil } +// gitAuthArgs returns per-invocation git args carrying the stored PAT via +// an HTTP header, or nil when no PAT is stored. +func (s *Spawner) gitAuthArgs() []string { + token, ok, _ := s.store.GetSetting(settingGitLabToken) + if !ok || token == "" { + return nil + } + return []string{"-c", "http.extraHeader=Authorization: token " + token} +} + +// gitBranch returns the checked-out branch of an existing clone, "" on failure. +func gitBranch(dir string) string { + out, err := exec.Command("git", "-C", dir, "rev-parse", "--abbrev-ref", "HEAD").Output() + if err != nil { + return "" + } + return strings.TrimSpace(string(out)) +} + func gitRun(dir string, args ...string) error { cmd := exec.Command("git", args...) if dir != "" { @@ -327,17 +383,29 @@ func (s *Spawner) ensureImage(ctx context.Context) error { if err != nil { return err } - var buf bytes.Buffer - if err := tarDir(&buf, s.buildContext); err != nil { - return fmt.Errorf("build context %s: %w", s.buildContext, err) - } - resp, err := s.cli.ImageBuild(ctx, &buf, build.ImageBuildOptions{ + pr, pw := io.Pipe() + tarDone := make(chan error, 1) + go func() { + err := tarDir(pw, s.buildContext) + _ = pw.CloseWithError(err) + tarDone <- err + }() + resp, buildErr := s.cli.ImageBuild(ctx, pr, build.ImageBuildOptions{ Tags: []string{imageRefWorker}, Dockerfile: relDockerfile, Remove: true, }) - if err != nil { - return fmt.Errorf("docker build: %w", err) + if buildErr != nil { + _ = pr.CloseWithError(buildErr) // unblock the tar goroutine + } + if tarErr := <-tarDone; tarErr != nil { + if resp.Body != nil { + _ = resp.Body.Close() + } + return fmt.Errorf("build context %s: %w", s.buildContext, tarErr) + } + if buildErr != nil { + return fmt.Errorf("docker build: %w", buildErr) } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, buildContextReadLimit)) @@ -378,6 +446,11 @@ func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID stri } if fresh { if err := s.seedVolume(ctx, slug, repoVolume); err != nil { + // drop the half-seeded volume so the next spawn retries fresh + // instead of silently booting into an empty workspace. + if rmErr := s.cli.VolumeRemove(context.Background(), repoVolume, true); rmErr != nil { + log.Printf("spawner: remove failed seed volume %s: %v", repoVolume, rmErr) + } return "", fmt.Errorf("seed %s: %w", repoVolume, err) } } @@ -409,6 +482,7 @@ func (s *Spawner) createAndStart(ctx context.Context, repo, slug, sessionID stri Binds: binds, NetworkMode: container.NetworkMode(s.network), AutoRemove: false, + Init: &[]bool{true}[0], // tini reaps bridge setup-hook zombies } name := "lvmh-agent-" + strings.ReplaceAll(sessionID, "-", "")[:12] created, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, name) @@ -445,18 +519,32 @@ func (s *Spawner) seedVolume(ctx context.Context, slug, repoVolume string) error } hostCfg := &container.HostConfig{ Binds: []string{repoVolume + ":" + workspaceMount}, + Init: &[]bool{true}[0], } created, err := s.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, "lvmh-seed-"+slug) if err != nil { return err } defer s.removeContainer(context.Background(), created.ID) - var buf bytes.Buffer - if err := tarDir(&buf, repoDir); err != nil { - return fmt.Errorf("tar clone: %w", err) + pr, pw := io.Pipe() + tarDone := make(chan error, 1) + go func() { + err := tarDir(pw, repoDir) + _ = pw.CloseWithError(err) + tarDone <- err + }() + copyErr := s.cli.CopyToContainer(ctx, created.ID, workspaceMount, pr, container.CopyToContainerOptions{}) + if copyErr != nil { + _ = pr.CloseWithError(copyErr) // unblock the tar goroutine } - if err := s.cli.CopyToContainer(ctx, created.ID, workspaceMount, &buf, container.CopyToContainerOptions{}); err != nil { - return fmt.Errorf("copy into volume: %w", err) + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } + if tarErr := <-tarDone; tarErr != nil { + return fmt.Errorf("tar clone: %w", tarErr) + } + if copyErr != nil { + return fmt.Errorf("copy into volume: %w", copyErr) } return nil } @@ -484,13 +572,7 @@ func (s *Spawner) RemoveSession(ctx context.Context, sessionID string) error { if err := s.store.DeleteContainer(sessionID); err != nil { return err } - s.mu.Lock() - if j, ok := s.jobs[sessionID]; ok { - j.State = stateError - j.Message = "container removed" - j.UpdatedAt = time.Now().UnixMilli() - } - s.mu.Unlock() + s.deleteJob(sessionID) s.hub.BroadcastSpawnStatus() return nil } diff --git a/daemon/docker_test.go b/daemon/docker_test.go index 87607c2..ec5af04 100644 --- a/daemon/docker_test.go +++ b/daemon/docker_test.go @@ -4,6 +4,7 @@ package main // fake git binary (PATH shim recording invocations). import ( + "context" "encoding/json" "fmt" "io" @@ -38,6 +39,7 @@ type recordedCreate struct { HostConfig struct { Binds []string `json:"Binds"` NetworkMode string `json:"NetworkMode"` + Init *bool `json:"Init"` } `json:"HostConfig"` } @@ -58,6 +60,7 @@ type fakeDocker struct { failStop bool failWait bool failVolumeCreate bool + failVolumeDelete bool failArchive bool archiveHang bool waitHang bool @@ -205,6 +208,16 @@ func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } writeJSONNow(w, http.StatusNotFound, `{"message":"no such volume"}`) + case call.Method == http.MethodDelete && strings.HasPrefix(call.Path, "/volumes/"): + if f.failVolumeDelete { + writeJSONNow(w, http.StatusInternalServerError, `{"message":"volume delete failed"}`) + return + } + name := strings.TrimPrefix(call.Path, "/volumes/") + f.mu.Lock() + delete(f.volume, name) + f.mu.Unlock() + w.WriteHeader(http.StatusNoContent) case call.Method == http.MethodPost && call.Path == "/containers/create": var rc recordedCreate if err := json.Unmarshal(body, &rc); err != nil { @@ -290,8 +303,11 @@ func useFakeGit(t *testing.T, mode string) string { " noisy) printf '%s\\n' \"" + strings.Repeat("x", 600) + "\"; exit 1;;\n" + " silent) exit 1;;\n" + "esac\n" + + "# branch probe: emit the configured HEAD name\n" + + "if [ \"$1\" = '-C' ] && [ \"$3\" = rev-parse ]; then printf '%s' \"$FAKE_GIT_HEAD\"; exit 0; fi\n" + "# clone: create a real worktree so seeding (tarDir) has files to copy\n" + - "if [ \"$1\" = clone ]; then d=$(eval \"echo \\${$#}\"); mkdir -p \"$d\" && printf 'fake-repo\n' > \"$d/README.md\"; fi\n" + + "# (clone may be $1 or $3, depending on leading -c auth args)\n" + + "case \" $* \" in *\" clone \"*) d=$(eval \"echo \\${$#}\"); mkdir -p \"$d\" && printf 'fake-repo\\n' > \"$d/README.md\";; esac\n" + "exit 0\n" if err := os.WriteFile(filepath.Join(dir, "git"), []byte(script), 0o755); err != nil { t.Fatalf("write fake git: %v", err) @@ -345,7 +361,7 @@ func newTestSpawner(t *testing.T, f *fakeDocker) (*Spawner, *Store) { daemonToken = testToken store := openTestStore(t) hub := NewHub(store) - sp, err := NewSpawner(store, hub, "https://gitlab.example/") + sp, err := NewSpawner(context.Background(), store, hub, "https://gitlab.example/") if err != nil { t.Fatalf("NewSpawner: %v", err) } diff --git a/daemon/gitlab.go b/daemon/gitlab.go index 0f22f23..548b84d 100644 --- a/daemon/gitlab.go +++ b/daemon/gitlab.go @@ -18,6 +18,7 @@ import ( const ( settingGitLabToken string = "gitlab_pat" projectsPerPage int = 50 + maxRepoPages int = 5 // pagination cap: 5 pages / 250 repos gitlabTimeout time.Duration = 15 * time.Second ) @@ -133,30 +134,36 @@ func (g *GitLab) token() (string, error) { return token, nil } -// Repos lists member projects sorted by most recent activity. +// Repos lists member projects sorted by most recent activity, paging +// upstream until a short page (capped at maxRepoPages). func (g *GitLab) Repos(ctx context.Context) ([]GitLabRepo, error) { token, err := g.token() if err != nil { return nil, err } - var projects []giteaRepo - path := fmt.Sprintf("/api/v1/user/repos?limit=%d", projectsPerPage) - if err := g.do(ctx, path, token, &projects); err != nil { - return nil, err - } - repos := make([]GitLabRepo, 0, len(projects)) - for _, p := range projects { - if p.FullName == "" { - continue + var repos []GitLabRepo + for page := 1; page <= maxRepoPages; page++ { + path := fmt.Sprintf("/api/v1/user/repos?limit=%d&page=%d", projectsPerPage, page) + var projects []giteaRepo + if err := g.do(ctx, path, token, &projects); err != nil { + return nil, err + } + for _, p := range projects { + if p.FullName == "" { + continue + } + repos = append(repos, GitLabRepo{ + Path: p.FullName, + Name: p.Name, + Namespace: p.Owner.Login, + LastActivityAt: p.UpdatedAt, + WebURL: p.HTMLURL, + DefaultBranch: p.DefaultBranch, + }) + } + if len(projects) < projectsPerPage { + break } - repos = append(repos, GitLabRepo{ - Path: p.FullName, - Name: p.Name, - Namespace: p.Owner.Login, - LastActivityAt: p.UpdatedAt, - WebURL: p.HTMLURL, - DefaultBranch: p.DefaultBranch, - }) } sort.SliceStable(repos, func(i, j int) bool { return repos[i].LastActivityAt > repos[j].LastActivityAt diff --git a/daemon/gitlab_extra_test.go b/daemon/gitlab_extra_test.go index 6779718..b583756 100644 --- a/daemon/gitlab_extra_test.go +++ b/daemon/gitlab_extra_test.go @@ -4,8 +4,10 @@ package main import ( "context" + "fmt" "net/http" "net/http/httptest" + "strconv" "strings" "testing" ) @@ -131,3 +133,63 @@ func TestGitLabReposSkipsEmptyPaths(t *testing.T) { t.Fatalf("repos = %+v, want empty (blank path skipped)", repos) } } + +// newPagedGitLab serves full pages until pagesToFull+1, then a short page. +func newPagedGitLab(t *testing.T, alwaysFull bool) *GitLab { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("/api/v1/user/repos", func(w http.ResponseWriter, r *http.Request) { + page, _ := strconv.Atoi(r.URL.Query().Get("page")) + if page < 1 { + page = 1 + } + count := projectsPerPage + if !alwaysFull && page > 2 { + count = 1 + } + var b strings.Builder + b.WriteString("[") + for i := 0; i < count; i++ { + if i > 0 { + b.WriteString(",") + } + fmt.Fprintf(&b, `{"full_name":"team/p%d-%d","name":"P","owner":{"login":"team"},`+ + `"updated_at":"2024-01-01T00:00:00Z","html_url":"https://gl","default_branch":"main"}`, page, i) + } + b.WriteString("]") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(b.String())) + }) + up := httptest.NewServer(mux) + t.Cleanup(up.Close) + store := openTestStore(t) + if err := store.SetSetting(settingGitLabToken, "pat"); err != nil { + t.Fatalf("set token: %v", err) + } + return NewGitLab(store, up.URL) +} + +func TestGitLabReposPaginatesUntilShortPage(t *testing.T) { + gl := newPagedGitLab(t, false) // 50 + 50 + 1 (short) → stops after page 3 + repos, err := gl.Repos(context.Background()) + if err != nil { + t.Fatalf("repos: %v", err) + } + if len(repos) != 2*projectsPerPage+1 { + t.Fatalf("repos = %d, want %d (pages 1-2 full + short page 3)", len(repos), 2*projectsPerPage+1) + } + if repos[0].Path != "team/p1-0" || repos[len(repos)-1].Path != "team/p3-0" { + t.Fatalf("first/last = %q..%q, want page1..page3", repos[0].Path, repos[len(repos)-1].Path) + } +} + +func TestGitLabReposPaginationCapped(t *testing.T) { + gl := newPagedGitLab(t, true) // upstream always returns full pages + repos, err := gl.Repos(context.Background()) + if err != nil { + t.Fatalf("repos: %v", err) + } + if len(repos) != maxRepoPages*projectsPerPage { + t.Fatalf("repos = %d, want capped at %d", len(repos), maxRepoPages*projectsPerPage) + } +} diff --git a/daemon/hub.go b/daemon/hub.go index c926317..99211d7 100644 --- a/daemon/hub.go +++ b/daemon/hub.go @@ -44,8 +44,11 @@ const ( agentSendQueue int = 32 webSendQueue int = 128 webFlushInterval time.Duration = 40 * time.Millisecond - webMaxPending int = 4096 // drop slow clients beyond this backlog + webMaxPending int = 4096 // drop slow clients beyond this backlog + webMaxPendingByte int = 64 << 20 // ...or beyond this many queued bytes promptSendTimeout time.Duration = 3 * time.Second + pongWait time.Duration = 60 * time.Second + pingPeriod time.Duration = 30 * time.Second writeWait time.Duration = 5 * time.Second daemonVersion int = 1 // envelope "v" ) @@ -109,11 +112,12 @@ type webClient struct { hub *Hub conn *websocket.Conn - mu sync.Mutex - sub string // subscribed sessionId, "" when none - control [][]byte - events []pendingEvent - dropped bool + mu sync.Mutex + sub string // subscribed sessionId, "" when none + control [][]byte + events []pendingEvent + pendingBytes int + dropped bool closeOnce sync.Once done chan struct{} @@ -126,18 +130,25 @@ func (c *webClient) drop() { }) } +// overflows reports whether one more item of size n would exceed the caps. +func (c *webClient) overflows(n int) bool { + return len(c.control)+len(c.events) >= webMaxPending || + c.pendingBytes+n > webMaxPendingByte +} + func (c *webClient) deliverControl(b []byte) { c.mu.Lock() defer c.mu.Unlock() if c.dropped { return } - if len(c.control)+len(c.events) >= webMaxPending { + if c.overflows(len(b)) { c.dropped = true go c.drop() return } c.control = append(c.control, b) + c.pendingBytes += len(b) } func (c *webClient) deliverEvent(e pendingEvent) { @@ -146,30 +157,68 @@ func (c *webClient) deliverEvent(e pendingEvent) { if c.dropped { return } - if len(c.control)+len(c.events) >= webMaxPending { + if c.overflows(len(e.raw)) { c.dropped = true go c.drop() return } c.events = append(c.events, e) + c.pendingBytes += len(e.raw) +} + +// writeEventsFrame flushes one batched events frame for a single session. +func (c *webClient) writeEventsFrame(batch []pendingEvent) error { + items := make([]json.RawMessage, 0, len(batch)) + for _, e := range batch { + items = append(items, e.raw) + } + payload, err := json.Marshal(map[string]any{ + "type": frameEvents, + "sessionId": batch[0].sessionID, + "after": batch[0].seq - 1, + "events": items, + }) + if err != nil { + return nil // unmarshalable raw JSON cannot happen; drop the batch + } + _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + return c.conn.WriteMessage(websocket.TextMessage, payload) } // writePump flushes queued frames every webFlushInterval, batching events of -// the subscribed session into single frames. Never blocks the hub. +// the subscribed session into single frames (one frame per contiguous +// sessionID run — a mid-queue resubscribe must never mix sessions). +// Never blocks the hub. func (c *webClient) writePump() { ticker := time.NewTicker(webFlushInterval) + pings := time.NewTicker(pingPeriod) defer ticker.Stop() + defer pings.Stop() for { select { case <-c.done: return case <-ticker.C: + case <-pings.C: + if !c.sendPing() { + c.hub.dropWeb(c) + return + } + continue } c.mu.Lock() control := c.control c.control = nil events := c.events c.events = nil + drained := 0 + for _, b := range control { + drained += len(b) + } + for _, e := range events { + drained += len(e.raw) + } + c.pendingBytes -= drained c.mu.Unlock() for _, b := range control { _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait)) @@ -178,35 +227,34 @@ func (c *webClient) writePump() { return } } - if len(events) == 0 { - continue - } - items := make([]json.RawMessage, 0, len(events)) - after := events[0].seq - 1 - for _, e := range events { - items = append(items, e.raw) - } - payload, err := json.Marshal(map[string]any{ - "type": frameEvents, - "sessionId": events[0].sessionID, - "after": after, - "events": items, - }) - if err != nil { - continue - } - _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait)) - if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil { - c.hub.dropWeb(c) - return + for i := 0; i < len(events); { + j := i + for j < len(events) && events[j].sessionID == events[i].sessionID { + j++ + } + if err := c.writeEventsFrame(events[i:j]); err != nil { + c.hub.dropWeb(c) + return + } + i = j } } } +// sendPing writes one ping frame honoring the write deadline. +func (c *webClient) sendPing() bool { + _ = c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + return c.conn.WriteMessage(websocket.PingMessage, nil) == nil +} + // readPump consumes subscribe/unsubscribe frames; malformed input never kills the server. func (c *webClient) readPump() { defer c.hub.dropWeb(c) c.conn.SetReadLimit(maxFrameSize) + _ = c.conn.SetReadDeadline(time.Now().Add(pongWait)) + c.conn.SetPongHandler(func(string) error { + return c.conn.SetReadDeadline(time.Now().Add(pongWait)) + }) for { _, data, err := c.conn.ReadMessage() if err != nil { @@ -252,8 +300,11 @@ func (a *agentConn) drop() { }) } -// writePump serializes daemon→plugin frames (prompt/abort/welcome). +// writePump serializes daemon→plugin frames (prompt/abort/welcome) and +// keeps the conn alive with periodic pings. func (a *agentConn) writePump() { + pings := time.NewTicker(pingPeriod) + defer pings.Stop() for { select { case <-a.done: @@ -264,10 +315,21 @@ func (a *agentConn) writePump() { a.drop() return } + case <-pings.C: + if !a.sendPing() { + a.drop() + return + } } } } +// sendPing writes one ping frame honoring the write deadline. +func (a *agentConn) sendPing() bool { + _ = a.conn.SetWriteDeadline(time.Now().Add(writeWait)) + return a.conn.WriteMessage(websocket.PingMessage, nil) == nil +} + // Hub tracks live agent conns and web subscribers; it is the only component // that mutates online state. type Hub struct { @@ -418,6 +480,10 @@ func (h *Hub) ServeAgentWS(w http.ResponseWriter, r *http.Request) { func (h *Hub) readAgentLoop(ac *agentConn) { defer ac.drop() ac.conn.SetReadLimit(maxFrameSize) + _ = ac.conn.SetReadDeadline(time.Now().Add(pongWait)) + ac.conn.SetPongHandler(func(string) error { + return ac.conn.SetReadDeadline(time.Now().Add(pongWait)) + }) registered := false defer func() { if registered { @@ -543,11 +609,15 @@ func (h *Hub) handleEvent(f frame) { func (h *Hub) unregister(ac *agentConn) { h.mu.Lock() + removed := false if cur, ok := h.agents[ac.sessionID]; ok && cur == ac { delete(h.agents, ac.sessionID) + removed = true } h.mu.Unlock() - if ac.sessionID != "" { + // Only flip the persisted flag when no newer conn replaced this one; + // a reconnect races this unregister path. + if removed && ac.sessionID != "" { if err := h.store.SetOnline(ac.sessionID, false); err != nil { log.Printf("hub: mark offline %s: %v", ac.sessionID, err) } diff --git a/daemon/hub_extra_test.go b/daemon/hub_extra_test.go index 1d6efd5..ef6f9b5 100644 --- a/daemon/hub_extra_test.go +++ b/daemon/hub_extra_test.go @@ -491,3 +491,226 @@ func TestWebClientDeliverToDropped(t *testing.T) { t.Fatalf("dropped client queued %d events, want %d", len(c.events), webMaxPending) } } + +func TestWebClientByteOverflowDrops(t *testing.T) { + hub := NewHub(openTestStore(t)) + c := &webClient{hub: hub, conn: throwawayConn(t), done: make(chan struct{})} + // stay under the frame-count cap; cross the byte cap on the last event. + meg := make([]byte, 1<<20) + var i int + for ; (i+1)*(1<<20) <= webMaxPendingByte; i++ { + c.deliverEvent(pendingEvent{sessionID: "s", seq: int64(i), raw: meg}) + } + if c.dropped { + t.Fatalf("client dropped at %dMiB, want only beyond %dMiB", i, webMaxPendingByte>>20) + } + c.deliverEvent(pendingEvent{sessionID: "s", seq: int64(i), raw: meg}) // one over the cap + select { + case <-c.done: + case <-time.After(2 * time.Second): + t.Fatal("byte-overflowed web client was not dropped") + } + if !c.dropped { + t.Fatal("client must be marked dropped on byte overflow") + } +} + +func TestUnregisterReplacedConnKeepsOnline(t *testing.T) { + store := openTestStore(t) + hub := NewHub(store) + if err := store.UpsertSession(SessionInfo{ID: "s1", Cwd: "/w", Model: "m", Provider: "p"}); err != nil { + t.Fatalf("upsert: %v", err) + } + if err := store.SetOnline("s1", true); err != nil { + t.Fatalf("online: %v", err) + } + old := &agentConn{hub: hub, conn: throwawayConn(t), send: make(chan []byte, 1), done: make(chan struct{})} + old.sessionID = "s1" + fresh := &agentConn{hub: hub, conn: throwawayConn(t), send: make(chan []byte, 1), done: make(chan struct{})} + fresh.sessionID = "s1" + hub.agents["s1"] = old + hub.agents["s1"] = fresh // reconnect replaced old + + hub.unregister(old) + rows, err := store.Sessions() + if err != nil || len(rows) != 1 { + t.Fatalf("sessions: %v %d", err, len(rows)) + } + if !rows[0].OnlineDB { + t.Fatal("replaced conn unregistering must not flip the live session offline") + } + + hub.unregister(fresh) + rows, err = store.Sessions() + if err != nil { + t.Fatalf("sessions: %v", err) + } + if rows[0].OnlineDB { + t.Fatal("last live conn unregistering must mark the session offline") + } +} + +func TestPingPongConstants(t *testing.T) { + if pingPeriod >= pongWait { + t.Fatalf("pingPeriod %v must be < pongWait %v", pingPeriod, pongWait) + } +} + +func TestWebClientSendPing(t *testing.T) { + serverConn := make(chan *websocket.Conn, 1) + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + return + } + serverConn <- c + for { + if _, _, err := c.ReadMessage(); err != nil { + return + } + } + })) + t.Cleanup(up.Close) + clientConn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(up.URL, "http"), nil) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = clientConn.Close() }) + var server *websocket.Conn + select { + case server = <-serverConn: + case <-time.After(2 * time.Second): + t.Fatal("server conn never upgraded") + } + hub := NewHub(openTestStore(t)) + c := &webClient{hub: hub, conn: server, done: make(chan struct{})} + if !c.sendPing() { + t.Fatal("sendPing over a live conn must succeed") + } + ac := &agentConn{hub: hub, conn: server, send: make(chan []byte, 1), done: make(chan struct{})} + if !ac.sendPing() { + t.Fatal("agent sendPing over a live conn must succeed") + } +} + +func TestPongFramesKeepReadLoopsAlive(t *testing.T) { + // unsolicited pongs must pass through both read pumps (refreshing the + // read deadline) without killing the conn. + ts, _ := newTestServer(t) + ws := dialAgent(t, ts) + if err := ws.WriteMessage(websocket.PongMessage, nil); err != nil { + t.Fatalf("agent pong: %v", err) + } + _ = ws.WriteJSON(helloFrame("s1")) + if welcome := readFrame(t, ws); welcome["type"] != evWelcome { + t.Fatalf("agent read loop died after pong: %v", welcome) + } + + web := dialWeb(t, ts) + if first := readFrame(t, web); first["type"] != frameSessionList { + t.Fatalf("first web frame = %v", first) + } + if err := web.WriteMessage(websocket.PongMessage, nil); err != nil { + t.Fatalf("web pong: %v", err) + } + // subscribe processed by the (still live) read pump → live events flow. + if err := web.WriteJSON(map[string]any{"type": frameSubscribe, "sessionId": "s1"}); err != nil { + t.Fatalf("subscribe: %v", err) + } + // A reader goroutine avoids deadline-based reads (gorilla conns fail + // permanently after a read timeout); resend until a batch arrives. + frames := make(chan map[string]any, 16) + go func() { + defer close(frames) + for { + var m map[string]any + if err := web.ReadJSON(&m); err != nil { + return + } + frames <- m + } + }() + start := time.Now() + for i := 0; ; i++ { + _ = ws.WriteJSON(map[string]any{"v": 1, "type": evMessageUpdate, "sessionId": "s1", "seq": int64(i + 1), "ts": 10, "delta": "x"}) + select { + case m, ok := <-frames: + if !ok { + t.Fatal("web conn closed; read loop died after pong") + } + if m["type"] == frameEvents { + return // read pump survived the pong and routed the subscription + } + case <-time.After(200 * time.Millisecond): + } + if time.Since(start) > 3*time.Second { + t.Fatal("no events frame delivered after pong") + } + } +} + +func TestWebResubscribeSplitsEventBatches(t *testing.T) { + ts, _, hub := newTestServerHub(t) + agent1 := dialAgent(t, ts) + _ = agent1.WriteJSON(helloFrame("s1")) + _ = readFrame(t, agent1) + agent2 := dialAgent(t, ts) + _ = agent2.WriteJSON(helloFrame("s2")) + _ = readFrame(t, agent2) + + web := dialWeb(t, ts) + if first := readFrame(t, web); first["type"] != frameSessionList { + t.Fatalf("first web frame = %v", first) + } + + webSubscribed := func(want string) func() bool { + return func() bool { + hub.mu.Lock() + defer hub.mu.Unlock() + for c := range hub.webs { + c.mu.Lock() + sub := c.sub + c.mu.Unlock() + if sub == want { + return true + } + } + return false + } + } + _ = web.WriteJSON(map[string]any{"type": frameSubscribe, "sessionId": "s1"}) + waitFor(t, 2*time.Second, webSubscribed("s1")) + _ = agent1.WriteJSON(map[string]any{"v": 1, "type": evMessageUpdate, "sessionId": "s1", "seq": 1, "ts": 10, "delta": "one"}) + + readBatch := func(want string) map[string]any { + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + _ = web.SetReadDeadline(time.Now().Add(2 * time.Second)) + var m map[string]any + if err := web.ReadJSON(&m); err != nil { + t.Fatalf("read frame (want batch %s): %v", want, err) + } + if m["type"] != frameEvents { + continue + } + if m["sessionId"] != want { + t.Fatalf("events frame sessionId = %v, want %q (sessions must not mix)", m["sessionId"], want) + } + for _, e := range m["events"].([]any) { + if e.(map[string]any)["sessionId"] != want { + t.Fatalf("batch for %s contains event of %v", want, e.(map[string]any)["sessionId"]) + } + } + return m + } + t.Fatalf("no events frame for %s before timeout", want) + return nil + } + + readBatch("s1") // event A flushed while subscribed to A + + _ = web.WriteJSON(map[string]any{"type": frameSubscribe, "sessionId": "s2"}) + waitFor(t, 2*time.Second, webSubscribed("s2")) + _ = agent2.WriteJSON(map[string]any{"v": 1, "type": evMessageUpdate, "sessionId": "s2", "seq": 1, "ts": 11, "delta": "two"}) + readBatch("s2") // event B must arrive as its own frame, never mixed with A's +} diff --git a/daemon/main.go b/daemon/main.go index d4e1c23..cf912cc 100644 --- a/daemon/main.go +++ b/daemon/main.go @@ -67,7 +67,9 @@ func run(args []string) error { hub := NewHub(store) gitlab := NewGitLab(store, envOr(envGitLabBaseURL, defaultGitLabBase)) - spawner, err := NewSpawner(store, hub, envOr(envGitLabBaseURL, defaultGitLabBase)) + spawnerCtx, spawnerCtxCancel := context.WithCancel(context.Background()) + defer spawnerCtxCancel() + spawner, err := NewSpawner(spawnerCtx, store, hub, envOr(envGitLabBaseURL, defaultGitLabBase)) if err != nil { return fmt.Errorf("docker client: %w", err) } @@ -94,10 +96,12 @@ func run(args []string) error { defer signal.Stop(stop) select { case err := <-serveErr: + spawnerCtxCancel() return fmt.Errorf("listen: %w", err) case <-stop: } log.Printf("shutting down") + spawnerCtxCancel() ctx, cancel := context.WithTimeout(context.Background(), shutdownGrace) defer cancel() if err := httpServer.Shutdown(ctx); err != nil { diff --git a/daemon/spawner_test.go b/daemon/spawner_test.go index 78221b7..c7d9ae0 100644 --- a/daemon/spawner_test.go +++ b/daemon/spawner_test.go @@ -75,7 +75,7 @@ func TestSpawnerStartHappyPath(t *testing.T) { t.Fatalf("missing env %v", wantEnv) } wantBinds := []string{ - "lvmh-repo-group-project:" + workspaceMount, + "lvmh-repo-group--project:" + workspaceMount, volumeSessions + ":" + sessionsMount, volumePiCache + ":" + cacheMount, } @@ -90,6 +90,9 @@ func TestSpawnerStartHappyPath(t *testing.T) { if c.HostConfig.NetworkMode != defaultNetwork { t.Fatalf("network = %q", c.HostConfig.NetworkMode) } + if c.HostConfig.Init == nil || !*c.HostConfig.Init { + t.Fatalf("agent container Init = %v, want true (tini zombie reaping)", c.HostConfig.Init) + } // fresh repo volume seeded from the clone via CopyToContainer (tar) seed := f.createsByName("lvmh-seed-") @@ -99,10 +102,13 @@ func TestSpawnerStartHappyPath(t *testing.T) { if seed[0].Image != imageRefWorker { t.Fatalf("seed create = %+v", seed[0]) } - wantSeedBinds := []string{"lvmh-repo-group-project:" + workspaceMount} + wantSeedBinds := []string{"lvmh-repo-group--project:" + workspaceMount} if !reflect.DeepEqual(seed[0].HostConfig.Binds, wantSeedBinds) { t.Fatalf("seed binds = %v, want %v (no host-path binds)", seed[0].HostConfig.Binds, wantSeedBinds) } + if seed[0].HostConfig.Init == nil || !*seed[0].HostConfig.Init { + t.Fatalf("seed container Init = %v, want true", seed[0].HostConfig.Init) + } if len(f.archives) != 1 || f.archives[0] == 0 { t.Fatalf("CopyToContainer archives = %v, want one non-empty tar", f.archives) } @@ -150,7 +156,7 @@ func TestSpawnerStartValidatesDockerAndDockerfile(t *testing.T) { // docker reachable but image absent and no Dockerfile anywhere → clear error t.Setenv(envWorkerDockerfile, filepath.Join(t.TempDir(), "missing.Dockerfile")) - sp2, err := NewSpawner(sp.store, sp.hub, "https://gitlab.example/") + sp2, err := NewSpawner(context.Background(), sp.store, sp.hub, "https://gitlab.example/") if err != nil { t.Fatalf("NewSpawner: %v", err) } @@ -250,7 +256,7 @@ func TestSpawnerCloneOrUpdatePullsExisting(t *testing.T) { sp, _ := newTestSpawner(t, f) // existing clone → pull --ff-only in that dir, no clone - dir := filepath.Join(sp.reposDir, "group-project") + dir := filepath.Join(sp.reposDir, repoSlug("group/project")) if err := os.MkdirAll(filepath.Join(dir, ".git"), 0o755); err != nil { t.Fatalf("mkdir .git: %v", err) } @@ -263,7 +269,7 @@ func TestSpawnerCloneOrUpdatePullsExisting(t *testing.T) { } } -func TestSpawnerCloneURLInjectsPAT(t *testing.T) { +func TestSpawnerCloneURLNeverCarriesPAT(t *testing.T) { f := newFakeDocker() sp, store := newTestSpawner(t, f) @@ -273,12 +279,13 @@ func TestSpawnerCloneURLInjectsPAT(t *testing.T) { if err := store.SetSetting(settingGitLabToken, "pat-1"); err != nil { t.Fatalf("set token: %v", err) } + // A stored PAT must never leak into the persisted clone URL. u, err := sp.cloneURL("group/project") if err != nil { t.Fatalf("cloneURL: %v", err) } - if u != "https://pat-1@gitlab.example/group/project.git" { - t.Fatalf("cloneURL with PAT = %q", u) + if u != "https://gitlab.example/group/project.git" || strings.Contains(u, "pat-1") { + t.Fatalf("cloneURL with stored PAT = %q, want clean URL", u) } } @@ -305,8 +312,8 @@ func TestSpawnerRemoveSession(t *testing.T) { t.Fatal("container row must be deleted") } for _, j := range sp.JobsSnapshot() { - if j.SessionID == "s1" && j.State != stateError { - t.Fatalf("job after removal = %+v, want error state", j) + if j.SessionID == "s1" { + t.Fatalf("job after removal = %+v, want entry deleted", j) } } @@ -326,17 +333,17 @@ func TestSpawnerSeedVolumeCancel(t *testing.T) { f := newFakeDocker() f.archiveHang = true // server never answers the archive PUT sp, _ := newTestSpawner(t, f) - if err := os.MkdirAll(filepath.Join(sp.reposDir, "group-project"), 0o755); err != nil { + if err := os.MkdirAll(filepath.Join(sp.reposDir, repoSlug("group/project")), 0o755); err != nil { t.Fatal(err) } - if err := os.WriteFile(filepath.Join(sp.reposDir, "group-project", "README.md"), []byte("x"), 0o644); err != nil { + if err := os.WriteFile(filepath.Join(sp.reposDir, repoSlug("group/project"), "README.md"), []byte("x"), 0o644); err != nil { t.Fatal(err) } ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan error, 1) - go func() { done <- sp.seedVolume(ctx, "group-project", "lvmh-repo-group-project") }() + go func() { done <- sp.seedVolume(ctx, repoSlug("group/project"), "lvmh-repo-group--project") }() waitFor(t, 5*time.Second, func() bool { return f.hasCallSuffix(http.MethodPut, "/archive") }) cancel() select { @@ -379,9 +386,13 @@ func TestSpawnerSameRepoSpawnsSerialize(t *testing.T) { } func TestRepoSlugAndUUID(t *testing.T) { - if got := repoSlug("a/b/c"); got != "a-b-c" { + if got := repoSlug("a/b/c"); got != "a--b--c" { t.Fatalf("repoSlug = %q", got) } + // "a/b/c" and "a/b-c" must map to distinct slugs (no collision). + if repoSlug("a/b/c") == repoSlug("a/b-c") { + t.Fatalf("slug collision: %q", repoSlug("a/b/c")) + } id := newUUID() if len(id) != 36 || id[8] != '-' || id[13] != '-' || id[18] != '-' || id[23] != '-' { t.Fatalf("newUUID shape = %q", id) @@ -391,6 +402,195 @@ func TestRepoSlugAndUUID(t *testing.T) { } } +func TestSpawnerCloneUsesHeaderAuthNotURLCredentials(t *testing.T) { + gitLog := useFakeGit(t, fakeGitModeOK) + f := newFakeDocker() + sp, store := newTestSpawner(t, f) + if err := store.SetSetting(settingGitLabToken, "pat-1"); err != nil { + t.Fatalf("set token: %v", err) + } + + res, err := sp.Start(context.Background(), "group/project", "") + if err != nil { + t.Fatalf("Start: %v", err) + } + waitJobState(t, sp, res.SessionID, stateRunning) + + calls := readGitLog(t, gitLog) + if len(calls) != 1 { + t.Fatalf("git calls = %v", calls) + } + call := calls[0] + // auth travels as a per-invocation -c http.extraHeader arg… + if !strings.HasPrefix(call, "-c http.extraHeader=Authorization: token pat-1 clone ") { + t.Fatalf("git call = %q, want -c http.extraHeader auth before clone", call) + } + // …and the URL recorded into .git/config stays credential-free. + if !strings.Contains(call, " -- https://gitlab.example/group/project.git ") { + t.Fatalf("git call = %q, want clean clone URL", call) + } + if strings.Contains(call, "pat-1@") { + t.Fatalf("git call = %q leaks the PAT into the URL", call) + } +} + +func TestSpawnerJobsPrunedToCap(t *testing.T) { + f := newFakeDocker() + sp, _ := newTestSpawner(t, f) + + for i := 0; i < maxSpawnJobs+10; i++ { + sp.setJob(fmt.Sprintf("s%d", i), "group/project", stateCloning, "", "") + } + jobs := sp.JobsSnapshot() + if len(jobs) != maxSpawnJobs { + t.Fatalf("jobs = %d, want capped at %d", len(jobs), maxSpawnJobs) + } + seen := map[string]bool{} + for _, j := range jobs { + seen[j.SessionID] = true + } + if seen["s0"] { + t.Fatal("oldest job must be pruned first") + } + for _, id := range []string{fmt.Sprintf("s%d", maxSpawnJobs), fmt.Sprintf("s%d", maxSpawnJobs+9)} { + if !seen[id] { + t.Fatalf("newest job %s pruned; kept = %v", id, seen) + } + } +} + +func TestSpawnerFailedSeedRemovesRepoVolume(t *testing.T) { + useFakeGit(t, fakeGitModeOK) + f := newFakeDocker() + f.failArchive = true // CopyToContainer fails → seed fails + sp, _ := newTestSpawner(t, f) + + res, err := sp.Start(context.Background(), "group/project", "") + if err != nil { + t.Fatalf("Start: %v", err) + } + job := waitJobState(t, sp, res.SessionID, stateError) + if !strings.Contains(job.Message, "seed") { + t.Fatalf("job message = %q, want seed failure", job.Message) + } + if !f.hasCall(http.MethodDelete, "/volumes/lvmh-repo-group--project") { + t.Fatal("failed seed must force-remove the repo volume for a fresh retry") + } + if f.volumeExists("lvmh-repo-group--project") { + t.Fatal("repo volume must not linger half-seeded") + } +} + +func TestSpawnerFailedSeedSurvivesVolumeRemoveFailure(t *testing.T) { + useFakeGit(t, fakeGitModeOK) + f := newFakeDocker() + f.failArchive = true + f.failVolumeDelete = true // cleanup itself fails; seed error still surfaces + sp, _ := newTestSpawner(t, f) + + res, err := sp.Start(context.Background(), "group/project", "") + if err != nil { + t.Fatalf("Start: %v", err) + } + job := waitJobState(t, sp, res.SessionID, stateError) + if !strings.Contains(job.Message, "seed") { + t.Fatalf("job message = %q, want seed failure despite cleanup failure", job.Message) + } +} + +func TestSpawnerCloneOrUpdateBranchFetchFails(t *testing.T) { + useFakeGit(t, fakeGitModeFail) // branch probe and fetch both fail + f := newFakeDocker() + sp, _ := newTestSpawner(t, f) + slug := repoSlug("group/project") + if err := os.MkdirAll(filepath.Join(sp.reposDir, slug, ".git"), 0o755); err != nil { + t.Fatalf("mkdir .git: %v", err) + } + err := sp.cloneOrUpdate("group/project", "dev", slug) + if err == nil || !strings.Contains(err.Error(), "git fetch") { + t.Fatalf("cloneOrUpdate branch fetch failure = %v, want git fetch error", err) + } +} + +func TestSpawnerDeleteJobMissing(t *testing.T) { + sp, _ := newTestSpawner(t, newFakeDocker()) + sp.deleteJob("never-existed") // no-op, must not panic + sp.setJob("s1", "group/project", stateRunning, "cid", "") + sp.deleteJob("s1") + for _, j := range sp.JobsSnapshot() { + if j.SessionID == "s1" { + t.Fatalf("job %s still present after deleteJob", j.SessionID) + } + } +} + +func TestSpawnerCloneOrUpdateSwitchesBranch(t *testing.T) { + cases := []struct { + name string + branch string + headOut string + wantTail []string + }{ + { + name: "same-branch-skips-fetch-checkout", + branch: "main", + headOut: "main\n", + wantTail: []string{"pull --ff-only"}, + }, + { + name: "different-branch-fetches-and-checks-out", + branch: "dev", + headOut: "main\n", + wantTail: []string{"fetch origin dev", "checkout dev", "pull --ff-only"}, + }, + { + name: "unknown-head-falls-through-to-fetch", + branch: "dev", + headOut: "", + wantTail: []string{"fetch origin dev", "checkout dev", "pull --ff-only"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + gitLog := useFakeGit(t, fakeGitModeOK) + t.Setenv("FAKE_GIT_HEAD", tc.headOut) + f := newFakeDocker() + sp, store := newTestSpawner(t, f) + if err := store.SetSetting(settingGitLabToken, "pat-1"); err != nil { + t.Fatalf("set token: %v", err) + } + slug := repoSlug("group/project") + dir := filepath.Join(sp.reposDir, slug) + if err := os.MkdirAll(filepath.Join(dir, ".git"), 0o755); err != nil { + t.Fatalf("mkdir .git: %v", err) + } + if err := sp.cloneOrUpdate("group/project", tc.branch, slug); err != nil { + t.Fatalf("cloneOrUpdate: %v", err) + } + calls := readGitLog(t, gitLog) + if len(calls) != len(tc.wantTail)+1 { // +1: rev-parse probe + t.Fatalf("git calls = %v, want %v (+rev-parse)", calls, tc.wantTail) + } + if !strings.HasPrefix(calls[0], "-C ") || !strings.HasSuffix(calls[0], "rev-parse --abbrev-ref HEAD") { + t.Fatalf("first call = %q, want branch probe", calls[0]) + } + authPrefix := "-c http.extraHeader=Authorization: token pat-1 " + for i, want := range tc.wantTail { + got := calls[i+1] + if want == "checkout dev" { + if got != want { // checkout needs no auth + t.Fatalf("call %d = %q, want %q", i+1, got, want) + } + continue + } + if got != authPrefix+want { + t.Fatalf("call %d = %q, want %q%q", i+1, got, authPrefix, want) + } + } + }) + } +} + func TestEnvOr(t *testing.T) { t.Setenv("LVMH_TEST_ENV_OR", " value ") if got := envOr("LVMH_TEST_ENV_OR", "def"); got != "value" { @@ -484,7 +684,7 @@ func TestGitRunSilentFailure(t *testing.T) { func TestSpawnerNewBadDockerHost(t *testing.T) { t.Setenv("DOCKER_HOST", "http://") store := openTestStore(t) - if _, err := NewSpawner(store, NewHub(store), "https://gitlab.example"); err == nil { + if _, err := NewSpawner(context.Background(), store, NewHub(store), "https://gitlab.example"); err == nil { t.Fatal("NewSpawner must fail on an unparseable DOCKER_HOST") } } @@ -533,7 +733,7 @@ func TestSpawnerCloneOrUpdateErrors(t *testing.T) { t.Fatalf("write file: %v", err) } sp.reposDir = file - if err := sp.cloneOrUpdate("group/project", "", "group-project"); err == nil { + if err := sp.cloneOrUpdate("group/project", "", repoSlug("group/project")); err == nil { t.Fatal("cloneOrUpdate with file reposDir must fail") } } @@ -570,7 +770,7 @@ func TestSpawnerEnsureImageErrors(t *testing.T) { func TestSpawnerSessionsVolumeCreateFails(t *testing.T) { useFakeGit(t, fakeGitModeOK) f := newFakeDocker() - f.volume["lvmh-repo-group-project"] = true // repo volume exists → skip seed + f.volume["lvmh-repo-group--project"] = true // repo volume exists → skip seed f.failVolumeCreate = true sp, _ := newTestSpawner(t, f) res, err := sp.Start(context.Background(), "group/project", "") @@ -590,12 +790,12 @@ func TestSpawnerSeedVolumeCreateStartFail(t *testing.T) { ctx := context.Background() f.failCreate = true - if err := sp.seedVolume(ctx, "group-project", "lvmh-repo-group-project"); err == nil { + if err := sp.seedVolume(ctx, repoSlug("group/project"), "lvmh-repo-group--project"); err == nil { t.Fatal("seedVolume with failing create must fail") } f.failCreate = false f.failStart = true - if err := sp.seedVolume(ctx, "group-project", "lvmh-repo-group-project"); err == nil { + if err := sp.seedVolume(ctx, repoSlug("group/project"), "lvmh-repo-group--project"); err == nil { t.Fatal("seedVolume with failing start must fail") } } @@ -632,7 +832,7 @@ func TestSpawnerCloneOrUpdatePullFails(t *testing.T) { useFakeGit(t, fakeGitModeFail) f := newFakeDocker() sp, _ := newTestSpawner(t, f) - dir := filepath.Join(sp.reposDir, "group-project") + dir := filepath.Join(sp.reposDir, repoSlug("group/project")) if err := os.MkdirAll(filepath.Join(dir, ".git"), 0o755); err != nil { t.Fatalf("mkdir .git: %v", err) } @@ -647,7 +847,7 @@ func TestSpawnerWorkerCreateStartFailures(t *testing.T) { // worker container create/start instead of at the seed container. useFakeGit(t, fakeGitModeOK) f := newFakeDocker() - f.volume["lvmh-repo-group-project"] = true + f.volume["lvmh-repo-group--project"] = true sp, _ := newTestSpawner(t, f) f.failCreate = true diff --git a/daemon/store.go b/daemon/store.go index 486d6f5..e327be7 100644 --- a/daemon/store.go +++ b/daemon/store.go @@ -131,6 +131,47 @@ func (s *Store) EventsAfter(sessionID string, after int64, limit int) ([]Event, return out, rows.Err() } +// EventsLatest returns the newest limit events for the session, ascending. +func (s *Store) EventsLatest(sessionID string, limit int) ([]Event, error) { + return s.eventsDesc(sessionID, ` WHERE sessionId=? ORDER BY seq DESC LIMIT ?`, limit, sessionID) +} + +// EventsBefore returns the newest window of at most limit events with +// seq < before, ascending. +func (s *Store) EventsBefore(sessionID string, before int64, limit int) ([]Event, error) { + return s.eventsDesc(sessionID, ` WHERE sessionId=? AND seq f.seq <= lastSeq).length + - sendQueue.filter( - (f) => !TRANSIENT_TYPES.has(f.type) && f.seq <= lastSeq, - ).length; + sendQueue.filter((f) => !TRANSIENT_TYPES.has(f.type) && f.seq <= lastSeq) + .length; if (covered > 0) { droppedEvents += covered; log( diff --git a/web/src/App.tsx b/web/src/App.tsx index a0deb1e..0bccb99 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -18,7 +18,10 @@ interface BoundaryState { error: Error | null; } -export class ErrorBoundary extends Component<{ children: ReactNode }, BoundaryState> { +export class ErrorBoundary extends Component< + { children: ReactNode }, + BoundaryState +> { state: BoundaryState = { error: null }; static getDerivedStateFromError(error: Error): BoundaryState { @@ -102,85 +105,85 @@ export default function App() { const archived = store.sessions.filter((s) => !s.online).sort(byActivity); return ( -
- - {sidebarOpen && ( -
setSidebarOpen(false)} - /> - )} -
- {store.state !== "open" && ( -
- - connection {store.state} — reconnecting… -
- )} - - } /> - -
-
- {toasts.map((t) => ( -
- {t.text} -
- ))} + +
+ {toasts.map((t) => ( +
+ {t.text} +
+ ))} +
-
+ ); } diff --git a/web/src/ChatStream.test.tsx b/web/src/ChatStream.test.tsx index d239135..f64baa4 100644 --- a/web/src/ChatStream.test.tsx +++ b/web/src/ChatStream.test.tsx @@ -1,336 +1,483 @@ -import { fireEvent, render, screen, waitFor } from "@testing-library/react"; -import userEvent from "@testing-library/user-event"; -import { describe, expect, it } from "vitest"; -import type { ChatMessage, ToolState } from "./derive"; -import ChatStream, { Bubble, TypingIndicator } from "./ChatStream"; +import { act, fireEvent, render, screen } from "@testing-library/react"; +import { MemoryRouter, Route, Routes } from "react-router-dom"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { Repo, SessionListItem } from "./protocol"; +import { Route as ApiRoute } from "./protocol"; +import { fetchJson } from "./api"; +import SpawnView from "./SpawnView"; +import type { SessionsStore } from "./store"; +import { jsonResponse, mockFetchJson, seedSettings } from "./test/setup"; -function msg(partial: Partial): ChatMessage { +const repo = (path: string, branch = "main"): Repo => ({ + path, + name: path.split("/")[1] ?? path, + namespace: path.split("/")[0] ?? "g", + lastActivityAt: "2024-05-01T00:00:00Z", + webUrl: `https://gl/${path}`, + defaultBranch: branch, +}); + +const PUSH_TOAST = vi.fn(); + +function makeStore(over: Partial = {}): SessionsStore { return { - key: `k-${Math.random()}`, - role: "assistant", - text: "", - thinking: null, - toolCalls: [], - toolCallId: null, - streaming: false, - ...partial, + sessions: [], + state: "open", + spawnJobs: [], + refresh: async (): Promise => + fetchJson(ApiRoute.Sessions), + subscribe: (): (() => void) => () => undefined, + ...over, }; } -const tool = (p: Partial): ToolState => ({ - id: "c1", - name: "bash", - args: "", - running: false, - isError: false, - preview: "", - ...p, +function tree(store: SessionsStore): React.ReactElement { + return ( + + + } + /> + } /> + + + ); +} + +/** Flush pending microtasks + React effects (works under fake timers). */ +async function flush(ticks = 4): Promise { + for (let i = 0; i < ticks; i += 1) { + // eslint-disable-next-line no-await-in-loop + await act(async () => { + await Promise.resolve(); + }); + } +} + +beforeEach(() => { + PUSH_TOAST.mockClear(); + seedSettings(); }); -describe("Bubble", () => { - it("renders plain text per role class", () => { - const { container } = render( - , - ); - expect(container.querySelector(".bubble-row.user")).not.toBeNull(); - expect(container.textContent).toContain("hi there"); +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("SpawnView status", () => { + it("shows checking state, then error when gitlab status fails", async () => { + mockFetchJson((url) => { + if (url.includes("/api/gitlab/status")) + return jsonResponse({ error: "down" }, 500); + return []; + }); + render(tree(makeStore())); + expect(screen.getByText("checking gitlab…")).toBeInTheDocument(); + await flush(); + expect(screen.getByText(/gitlab status failed: down/)).toBeInTheDocument(); + }); +}); + +describe("SpawnView connect flow", () => { + it("requires a token", async () => { + mockFetchJson((url) => { + if (url.endsWith("/api/gitlab/status")) + return { connected: false, baseUrl: "https://gl" }; + return []; + }); + render(tree(makeStore())); + await flush(); + fireEvent.click(screen.getByRole("button", { name: "Connect" })); + await flush(); + expect(screen.getByText("token required")).toBeInTheDocument(); }); - it("renders empty assistant text as nothing but shows nothing when empty", () => { - const { container } = render( - , - ); - expect(container.querySelector(".bubble")?.children).toHaveLength(0); + it("connects, clears the PAT, loads repos and shows the picker", async () => { + let connected = false; + mockFetchJson((url, init) => { + if (url.endsWith("/api/gitlab/status")) + return { + connected, + baseUrl: "https://gl", + username: connected ? "alice" : undefined, + }; + if (url.endsWith("/api/gitlab/connect") && init?.method === "POST") { + connected = true; + return { username: "alice" }; + } + if (url.endsWith("/api/gitlab/repos")) + return [repo("g/one"), repo("g/two")]; + return []; + }); + render(tree(makeStore())); + await flush(); + expect(screen.getByText("Connect GitLab")).toBeInTheDocument(); + + const pat = screen.getByLabelText( + "GitLab personal access token", + ) as HTMLInputElement; + fireEvent.input(pat, { target: { value: "glpat-x" } }); + fireEvent.click(screen.getByRole("button", { name: "Connect" })); + await flush(); + + expect(screen.getByText("g/one")).toBeInTheDocument(); + expect(screen.getByText("g/two")).toBeInTheDocument(); + expect(screen.queryByLabelText("GitLab personal access token")).toBeNull(); + expect(screen.getByText("Repository")).toBeInTheDocument(); }); - it("toolResult collapses to a one-line preview, expands to full text", async () => { - const long = `${"x".repeat(200)}`; - render( - , - ); - const details = screen - .getByText("result") - .closest("details") as HTMLDetailsElement; - expect(details.open).toBe(false); - expect(details.textContent).toContain("…"); - await userEvent.click(screen.getByText("result")); - expect(details.open).toBe(true); - expect(details.textContent).toContain(long); + it("connect failure shows the error and keeps the gate", async () => { + mockFetchJson((url, init) => { + if (url.endsWith("/api/gitlab/status")) + return { connected: false, baseUrl: "https://gl" }; + if (url.endsWith("/api/gitlab/connect") && init?.method === "POST") + return jsonResponse({ error: "bad pat" }, 401); + return []; + }); + render(tree(makeStore())); + await flush(); + fireEvent.input(screen.getByLabelText("GitLab personal access token"), { + target: { value: "glpat-bad" }, + }); + fireEvent.click(screen.getByRole("button", { name: "Connect" })); + await flush(); + expect(screen.getByText("bad pat")).toBeInTheDocument(); + expect(screen.getByText("Connect GitLab")).toBeInTheDocument(); }); - it("toolResult with short text keeps full one-line preview", () => { - render( - , - ); - const details = screen - .getByText("result") - .closest("details") as HTMLDetailsElement; - expect(details.textContent).toContain("short out"); - expect(details.textContent).not.toContain("…"); + it("connected on load fetches repos immediately", async () => { + mockFetchJson((url) => { + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/quick")]; + return []; + }); + render(tree(makeStore())); + await flush(); + expect(screen.getByText("g/quick")).toBeInTheDocument(); + }); +}); + +describe("SpawnView repo picker", () => { + function connectedMock(): void { + mockFetchJson((url) => { + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) + return [repo("g/alpha"), repo("g/beta", "dev")]; + return []; + }); + } + + it("filters repos by query, keyboard selects, empty filter message", async () => { + connectedMock(); + render(tree(makeStore())); + await flush(); + expect(screen.getByText("g/alpha")).toBeInTheDocument(); + + const filter = screen.getByLabelText("Filter repositories"); + fireEvent.input(filter, { target: { value: "beta" } }); + expect(screen.queryByText("g/alpha")).toBeNull(); + expect(screen.getByText("g/beta")).toBeInTheDocument(); + + const item = screen + .getByText("g/beta") + .closest(".repo-item") as HTMLElement; + fireEvent.keyDown(item, { key: "Enter" }); + expect(screen.getByLabelText("Branch")).toHaveValue("dev"); + + fireEvent.input(filter, { target: { value: "zzz" } }); + expect(screen.getByText("no matching repos")).toBeInTheDocument(); }); - it("flattens whitespace in previews", () => { - render( - , - ); - expect(screen.getAllByText("a b c").length).toBeGreaterThan(0); + it("shows loading repos while the list is pending", async () => { + const gate = { resolve: null as ((v: unknown) => void) | null }; + mockFetchJson((url) => { + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) + return new Promise((res) => { + gate.resolve = res; + }); + return []; + }); + render(tree(makeStore())); + await flush(); + expect(screen.getByText("loading repos…")).toBeInTheDocument(); + gate.resolve?.([repo("g/late")]); + await flush(); + expect(await screen.findByText("g/late")).toBeInTheDocument(); }); - it("assistant tool calls attach tool cards", async () => { - const tools = new Map([ - [ - "c1", - tool({ - id: "c1", - name: "bash", - args: "ls -la", - running: false, - isError: false, - preview: "file", - }), + it("spawn without selection shows pick-a-repo error", async () => { + connectedMock(); + render(tree(makeStore())); + await flush(); + expect(screen.getByLabelText("Spawn container")).toBeDisabled(); + expect(screen.getByText("select a repo above")).toBeInTheDocument(); + }); +}); + +describe("SpawnView spawn+poll", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + it("spawns, polls until online, then navigates to the chat", async () => { + const posts: Array<[string, RequestInit | undefined]> = []; + let sessionsOnline = false; // flipped after the first poll tick + const fetchMock = mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) { + posts.push([url, init]); + return { sessionId: "new-1", containerId: "abc123def456" }; + } + if (url.endsWith("/api/sessions")) + return sessionsOnline + ? [ + { + id: "new-1", + name: "spawned", + cwd: "/w", + model: "m", + provider: "p", + agent: true, + repo: "g/proj", + startedAt: 1, + online: true, + lastEventAt: 1, + }, + ] + : []; + if (url.endsWith("/api/spawn/status")) return []; + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/proj")]; + return []; + }); + const store = makeStore(); + const { unmount } = render(tree(store)); + await flush(); + fireEvent.click(screen.getByText("g/proj")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + + expect(posts).toHaveLength(1); + expect((posts[0]?.[1] as RequestInit).body).toBe( + JSON.stringify({ repo: "g/proj", branch: "main" }), + ); + expect(screen.getByText("Spawning…")).toBeInTheDocument(); + expect(screen.getByText(/container abc123def456/)).toBeInTheDocument(); + expect( + screen.getByText("waiting for session to come online…"), + ).toBeInTheDocument(); + + // first tick: session not online yet + await act(async () => { + vi.advanceTimersByTime(1500); + }); + await flush(); + expect(screen.queryByTestId("chat-route")).toBeNull(); + + // each poll tick issues exactly one sessions fetch (refresh reuse, S7) + const sessionsFetches = fetchMock.mock.calls.filter(([u]) => + String(u).endsWith("/api/sessions"), + ).length; + expect(sessionsFetches).toBe(1); + + // session comes online -> next tick navigates + sessionsOnline = true; + await act(async () => { + vi.advanceTimersByTime(1500); + }); + await flush(); + expect(screen.getByTestId("chat-route")).toBeInTheDocument(); + unmount(); + }); + + it("spawning view shows job line from status and spawn error text", async () => { + mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) + return { sessionId: "sx", containerId: "" }; + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/p")]; + return []; + }); + const store = makeStore({ + spawnJobs: [ + { sessionId: "sx", repo: "g/p", state: "cloning", containerId: "" }, ], - ]); - render( - , - ); - expect(screen.getByText("🛠 bash")).toBeInTheDocument(); - expect(screen.getByText("finished")).toBeInTheDocument(); - - const summary = screen - .getByText("🛠 bash") - .closest("summary") as HTMLElement; - const card = summary.closest("details") as HTMLDetailsElement; - expect(card.open).toBe(false); - await userEvent.click(summary); - expect(card.open).toBe(true); - expect(card.textContent).toContain("ls -la"); - expect(card.textContent).toContain("file"); + }); + render(tree(store)); + await flush(); + fireEvent.click(screen.getByText("g/p")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + expect(screen.getByText("Spawning…")).toBeInTheDocument(); + expect(screen.getByText("g/p: cloning")).toBeInTheDocument(); }); - it("tool card status variants: running, error, done", () => { - const tools = new Map([ - ["c1", tool({ id: "c1", running: true })], - ["c2", tool({ id: "c2", running: false, isError: true })], - ["c3", tool({ id: "c3", running: false, isError: false })], - ]); - render( - ({ - id, - name: `t-${id}`, - argsJson: "{}", - })), - })} - tools={tools} - />, - ); - expect(screen.getByText("working…")).toBeInTheDocument(); - expect(screen.getByText("error")).toBeInTheDocument(); - expect(screen.getByText("done")).toBeInTheDocument(); + it("branch left blank sends repo only; poll refresh failure toasts", async () => { + const bodies: string[] = []; + mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) { + bodies.push(String(init.body)); + return { sessionId: "new-2", containerId: "cccccccccccc" }; + } + if (url.endsWith("/api/spawn/status")) return []; + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/blank")]; + return []; + }); + const failingRefresh = makeStore({ + refresh: (): Promise => + Promise.reject(new Error("boom")), + }); + render(tree(failingRefresh)); + await flush(); + fireEvent.click(screen.getByText("g/blank")); + fireEvent.change(screen.getByLabelText("Branch"), { + target: { value: "" }, + }); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + expect(bodies).toEqual([JSON.stringify({ repo: "g/blank" })]); + + await act(async () => { + vi.advanceTimersByTime(1500); + }); + await flush(); + expect(PUSH_TOAST).toHaveBeenCalledWith("boom"); }); - it("tool call with no matching state renders no card", () => { - const { container } = render( - , - ); - expect(container.querySelectorAll(".tool-card")).toHaveLength(0); + it("spawn POST failure shows the error", async () => { + mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) + return jsonResponse({ error: "no docker" }, 500); + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/x")]; + return []; + }); + render(tree(makeStore())); + await flush(); + fireEvent.click(screen.getByText("g/x")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + expect(screen.getByText("no docker")).toBeInTheDocument(); }); - it("thinking block only for non-empty thinking", async () => { - render(); - const details = screen - .getByText("thinking") - .closest("details") as HTMLDetailsElement; - await userEvent.click(screen.getByText("thinking")); - expect(details.open).toBe(true); - expect(details.textContent).toContain("because"); + it("polling gives up after the tick cap and stays on the page", async () => { + let statusCalls = 0; + mockFetchJson((url) => { + if (url.endsWith("/api/spawn/status")) { + statusCalls += 1; + return []; + } + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/slow")]; + if (url.endsWith("/api/spawn")) + return { sessionId: "slow-1", containerId: "d" }; + return []; + }); + const { unmount } = render(tree(makeStore())); + await flush(); + fireEvent.click(screen.getByText("g/slow")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + expect(screen.getByText("Spawning…")).toBeInTheDocument(); - const { container } = render( - , - ); - expect(container.querySelector(".thinking")).toBeNull(); + await act(async () => { + vi.advanceTimersByTime(1500 * 402); + }); + await flush(); + const afterCap = statusCalls; + await act(async () => { + vi.advanceTimersByTime(1500 * 10); + }); + await flush(); + expect(statusCalls).toBe(afterCap); + expect(screen.getByText("Spawning…")).toBeInTheDocument(); + unmount(); }); - it("streaming bubble shows the caret", () => { - const { container } = render( - , - ); - expect(container.querySelector(".stream-caret")).not.toBeNull(); + it("spawn job line renders from store.spawnJobs", async () => { + mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) + return { sessionId: "sj-1", containerId: "cid" }; + if (url.endsWith("/api/spawn/status")) return []; + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "alice" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/p")]; + return []; + }); + const store = makeStore(); + const { rerender } = render(tree(store)); + await flush(); + fireEvent.click(screen.getByText("g/p")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + expect(screen.getByText("Spawning…")).toBeInTheDocument(); + + store.spawnJobs = [{ repo: "g/p", state: "cloning", sessionId: "sj-1" }]; + act(() => { + rerender(tree(store)); + }); + expect(screen.getByText("g/p: cloning")).toBeInTheDocument(); }); }); -describe("TypingIndicator", () => { - it("renders three dots with aria-live", () => { - const { container } = render(); - expect(container.querySelector('[aria-live="polite"]')).not.toBeNull(); - expect(container.querySelectorAll(".dot")).toHaveLength(3); - }); -}); - -describe("ChatStream", () => { - it("renders messages and typing indicator while busy with no open stream", () => { - const { container, rerender } = render( - , - ); - expect(container.querySelector(".typing")).not.toBeNull(); - - rerender( - , - ); - expect(container.querySelector(".typing")).toBeNull(); - rerender( - , - ); - expect(container.querySelector(".typing")).toBeNull(); - }); - - it("sticks to bottom on new messages, un-pins on user scroll up, re-pins near bottom", () => { - const { container, rerender } = render( - , - ); - const scroller = container.querySelector(".chat-scroll") as HTMLElement; - Object.defineProperty(scroller, "scrollHeight", { - configurable: true, - value: 1000, - }); - Object.defineProperty(scroller, "clientHeight", { - configurable: true, - value: 300, +describe("SpawnSteps", () => { + it("renders progress steps matching job state", async () => { + mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) + return { sessionId: "sp1", containerId: "" }; + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "a" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/p")]; + return []; }); - - scroller.dispatchEvent(new Event("scroll")); - // pinned: scrollTop at bottom - scroller.scrollTop = 700; - Object.defineProperty(scroller, "scrollTop", { - configurable: true, - writable: true, - value: 700, + const store = makeStore({ + spawnJobs: [ + { sessionId: "sp1", repo: "g/p", state: "building", containerId: "" }, + ], }); - scroller.dispatchEvent(new Event("scroll")); - rerender( - , - ); - expect(scroller.scrollTop).toBe(1000); + render(tree(store)); + await flush(); + fireEvent.click(screen.getByText("g/p")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + const steps = document.querySelectorAll(".spawn-progress .step"); + expect(steps).toHaveLength(4); + expect(steps[0]?.className).toContain("done"); + expect(steps[1]?.className).toContain("current"); + expect(steps[2]?.className).toBe("step"); + }); - // scroll far up -> unpin - Object.defineProperty(scroller, "scrollTop", { - configurable: true, - writable: true, - value: 0, + it("progress element carries progressbar semantics for the current step", async () => { + mockFetchJson((url, init) => { + if (init?.method === "POST" && url.endsWith("/api/spawn")) + return { sessionId: "sp2", containerId: "" }; + if (url.endsWith("/api/gitlab/status")) + return { connected: true, baseUrl: "https://gl", username: "a" }; + if (url.endsWith("/api/gitlab/repos")) return [repo("g/p")]; + return []; }); - scroller.dispatchEvent(new Event("scroll")); - rerender( - , - ); - expect(scroller.scrollTop).toBe(0); - - // scroll near bottom (within 80px) -> pinned again - Object.defineProperty(scroller, "scrollTop", { - configurable: true, - writable: true, - value: 940, + const store = makeStore({ + spawnJobs: [ + { sessionId: "sp2", repo: "g/p", state: "building", containerId: "" }, + ], }); - scroller.dispatchEvent(new Event("scroll")); - rerender( - , - ); - expect(scroller.scrollTop).toBe(1000); - }); -}); - -describe("copy button", () => { - it("copy button writes message text and flashes copied", async () => { - const writeText = vi.fn(() => Promise.resolve()); - Object.assign(navigator, { clipboard: { writeText } }); - render( - , - ); - const btn = screen.getByRole("button", { name: "Copy message" }); - fireEvent.click(btn); - expect(writeText).toHaveBeenCalledWith("copy me"); - await waitFor(() => expect(screen.getByText("copied")).toBeInTheDocument()); - }); - - it("user bubbles get a copy button, toolResults do not", () => { - const { rerender } = render( - , - ); - expect(screen.getByRole("button", { name: "Copy message" })).toBeInTheDocument(); - rerender( - , - ); - expect(screen.queryByRole("button", { name: "Copy message" })).toBeNull(); + render(tree(store)); + await flush(); + fireEvent.click(screen.getByText("g/p")); + fireEvent.click(screen.getByLabelText("Spawn container")); + await flush(); + const bar = screen.getByRole("progressbar"); + expect(bar).toHaveAttribute("aria-valuemin", "1"); + expect(bar).toHaveAttribute("aria-valuemax", "4"); + expect(bar).toHaveAttribute("aria-valuenow", "2"); // building = step 2 }); }); diff --git a/web/src/ChatStream.tsx b/web/src/ChatStream.tsx index e5c9e22..2804af4 100644 --- a/web/src/ChatStream.tsx +++ b/web/src/ChatStream.tsx @@ -136,9 +136,20 @@ interface Props { messages: ChatMessage[]; tools: Map; busy: boolean; + /** an older page exists beyond the loaded window (B1) */ + hasOlder: boolean; + loadingOlder: boolean; + onLoadOlder: () => void; } -export default function ChatStream({ messages, tools, busy }: Props) { +export default function ChatStream({ + messages, + tools, + busy, + hasOlder, + loadingOlder, + onLoadOlder, +}: Props) { const scrollRef = useRef(null); const pinnedRef = useRef(true); @@ -166,6 +177,19 @@ export default function ChatStream({ messages, tools, busy }: Props) { return (
+ {hasOlder && ( +
+ +
+ )} {messages.map((m) => ( ))} diff --git a/web/src/ChatView.test.tsx b/web/src/ChatView.test.tsx index b1d74e7..ce21565 100644 --- a/web/src/ChatView.test.tsx +++ b/web/src/ChatView.test.tsx @@ -7,7 +7,7 @@ import { } from "@testing-library/react"; import userEvent from "@testing-library/user-event"; import type { ReactElement } from "react"; -import { MemoryRouter, Route, Routes } from "react-router-dom"; +import { MemoryRouter, Route, Routes, useNavigate } from "react-router-dom"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { EventFrame } from "./protocol"; import { ApiError } from "./api"; @@ -42,7 +42,7 @@ function makeStore(over: Partial = {}): SessionsStore { sessions, state: "open", spawnJobs: [], - refresh: async () => undefined, + refresh: async () => [], subscribe: ( sessionId: string, onEvents: (events: EventFrame[]) => void, @@ -131,6 +131,33 @@ function historyEvents(): EventFrame[] { ]; } +/** seq from..to inclusive, one user message per seq. */ +function pageEvents( + from: number, + to: number, + text: (n: number) => string, +): EventFrame[] { + const out: EventFrame[] = []; + for (let n = from; n <= to; n += 1) { + out.push({ + v: 1, + sessionId: "s1", + seq: n, + ts: 0, + type: "message_end", + message: { + role: "user", + id: `u${n}`, + text: text(n), + thinking: null, + toolCalls: [], + toolCallId: null, + }, + }); + } + return out; +} + beforeEach(() => { seq = 0; currentSub = null; @@ -362,7 +389,8 @@ describe("ChatView", () => { , ); await screen.findByText("caught up"); - expect(after).toBe("0"); + // initial load is latest=1: no after= cursor has been issued yet + expect(after).toBe(""); // reconnect: state closed -> open triggers the after=N refetch act(() => { @@ -418,10 +446,174 @@ describe("ChatView", () => { }); }); +describe("ChatView history pagination (B1)", () => { + it("initial history fetch requests the newest page via latest=1", async () => { + const eventsUrls: string[] = []; + mockFetchJson((url) => { + if (url.includes("/events")) { + eventsUrls.push(url); + return historyEvents(); + } + return []; + }); + renderChat(makeStore()); + await screen.findByText("hello there"); + await waitFor(() => expect(eventsUrls.length).toBeGreaterThan(0)); + expect(eventsUrls[0]).toContain("latest=1"); + expect(eventsUrls[0]).toContain("limit=1000"); + expect(eventsUrls[0]).not.toContain("after="); + }); + + it("short first page renders no Load older button", async () => { + mockFetchJson((url) => (url.includes("/events") ? historyEvents() : [])); + renderChat(makeStore()); + await screen.findByText("hello there"); + expect( + screen.queryByRole("button", { name: "Load older messages" }), + ).toBeNull(); + }); + + it("Load older prepends the previous page and hides at a short page", async () => { + const olderUrls: string[] = []; + mockFetchJson((url) => { + if (url.includes("/events")) { + if (url.includes("before=")) { + olderUrls.push(url); + return pageEvents(1, 2, (n) => `oldest-${n}`); + } + return pageEvents(3, 1002, (n) => `m${n}`); + } + return []; + }); + renderChat(makeStore()); + await screen.findByText("m1002"); + expect( + screen.getByRole("button", { name: "Load older messages" }), + ).toBeInTheDocument(); + + await userEvent.click( + screen.getByRole("button", { name: "Load older messages" }), + ); + await screen.findByText("oldest-1"); + expect(screen.getByText("m1002")).toBeInTheDocument(); + expect(olderUrls).toHaveLength(1); + expect(olderUrls[0]).toContain("before=3"); + expect(olderUrls[0]).toContain("limit=1000"); + // short page (2 < 1000): the button disappears + await waitFor(() => + expect( + screen.queryByRole("button", { name: "Load older messages" }), + ).toBeNull(), + ); + }); + + it("cursor refetch after loading older pages still uses the max seq", async () => { + let after = ""; + mockFetchJson((url) => { + const m = /[?&]after=(\d+)/.exec(url); + if (m !== null) after = m[1] ?? ""; + if (url.includes("/events")) { + if (url.includes("before=")) + return pageEvents(1, 2, (n) => `oldest-${n}`); + return pageEvents(3, 1002, (n) => `m${n}`); + } + return []; + }); + const { rerender } = renderChat(makeStore()); + await screen.findByText("m1002"); + await userEvent.click( + screen.getByRole("button", { name: "Load older messages" }), + ); + await screen.findByText("oldest-1"); + + // reconnect: cursor must still point at the newest seq (1002), not at the + // older batch's max (2) — older pages must not poison lastSeqRef + rerenderChatAgain(rerender, makeStore({ state: "connecting" })); + rerenderChatAgain(rerender, makeStore({ state: "open" })); + await vi.waitFor(() => expect(after).toBe("1002")); + }); + + it("Load older failure toasts and keeps the button", async () => { + mockFetchJson((url) => { + if (url.includes("before=")) + return jsonResponse({ error: "older fail" }, 500); + if (url.includes("/events")) return pageEvents(3, 1002, (n) => `m${n}`); + return []; + }); + renderChat(makeStore()); + await screen.findByText("m1002"); + await userEvent.click( + screen.getByRole("button", { name: "Load older messages" }), + ); + await vi.waitFor(() => + expect(pushToast).toHaveBeenCalledWith("older fail"), + ); + expect( + screen.getByRole("button", { name: "Load older messages" }), + ).toBeEnabled(); + }); + + it("a second click while a page load is in flight is ignored", async () => { + let releaseOlder: ((v: EventFrame[]) => void) | null = null; + let beforeCalls = 0; + mockFetchJson((url) => { + if (url.includes("/events")) { + if (url.includes("before=")) { + beforeCalls += 1; + return new Promise((res) => { + releaseOlder = res; + }); + } + return pageEvents(3, 1002, (n) => `m${n}`); + } + return []; + }); + renderChat(makeStore()); + await screen.findByText("m1002"); + const btn = screen.getByRole("button", { name: "Load older messages" }); + await userEvent.click(btn); + // still pending: the button is disabled and re-entry is a no-op + expect(btn).toBeDisabled(); + fireEvent.click(btn); + expect(beforeCalls).toBe(1); + await act(async () => { + releaseOlder?.(pageEvents(1, 2, (n) => `oldest-${n}`)); + }); + await screen.findByText("oldest-1"); + expect(beforeCalls).toBe(1); + }); +}); + +describe("ChatView busy vs offline (B2)", () => { + it("offline session with history ending at agent_start shows no typing and the send button", async () => { + mockFetchJson((url) => + url.includes("/events") ? [ev("agent_start")] : [], + ); + const offline = [{ ...sessions[0]!, online: false }]; + const { container } = renderChat(makeStore({ sessions: offline })); + await waitFor(() => expect(currentSub).not.toBeNull()); + await vi.waitFor(() => + expect(container.querySelector(".typing")).toBeNull(), + ); + expect(screen.queryByLabelText("Abort current run")).toBeNull(); + expect(screen.getByLabelText("Send message")).toBeInTheDocument(); + }); + + it("online session with history ending at agent_start still shows the stop button", async () => { + mockFetchJson((url) => + url.includes("/events") ? [ev("agent_start")] : [], + ); + renderChat(makeStore()); + await waitFor(() => expect(currentSub).not.toBeNull()); + expect(screen.getByLabelText("Abort current run")).toBeInTheDocument(); + expect(screen.queryByLabelText("Send message")).toBeNull(); + }); +}); + describe("ChatView close pi", () => { it("close button deletes the container, toasts and refreshes (agent session)", async () => { const agent = [{ ...sessions[0]!, id: "s1", agent: true }]; - const refresh = vi.fn(async () => undefined); + const refresh = vi.fn(async () => []); const deletes: string[] = []; mockFetchJson((url, init) => { if (init?.method === "DELETE" && url.includes("/container")) { @@ -485,6 +677,141 @@ describe("ChatView refetch failure", () => { }); }); +describe("ChatView stale async guards (S1/S4) and send draft (S6)", () => { + // same-tree navigation: ChatView stays mounted while the session id changes + function Switcher({ + state, + }: { + state: "connecting" | "open"; + }): React.ReactElement { + const nav = useNavigate(); + return ( + <> + + + + } + /> + OTHER
} /> + + + ); + } + + it("refetch resolving after a session switch does not merge stale events (S1)", async () => { + let releaseRefetch: ((v: EventFrame[]) => void) | null = null; + const staleFrame: EventFrame = { + v: 1, + sessionId: "s1", + seq: 77, + ts: 0, + type: "message_end", + message: { + role: "user", + id: "u77", + text: "stale s1 frame", + thinking: null, + toolCalls: [], + toolCallId: null, + }, + }; + let ghostAfter = "unfetched"; + mockFetchJson((url) => { + if (url.startsWith("http://srv/api/sessions/ghost/events")) { + const m = /[?&]after=(\d+)/.exec(url); + if (m !== null) ghostAfter = m[1] ?? ""; + return []; + } + if (url.startsWith("http://srv/api/sessions/s1/events")) { + if (url.includes("after=")) + return new Promise((res) => { + releaseRefetch = res; + }); + return []; + } + return []; + }); + const { rerender } = render( + + + , + ); + // let the initial history load settle (loadedRef) before the ws opens + await act(async () => { + await Promise.resolve(); + }); + // ws (re)opens so the after=N refetch fires for s1 + rerender( + + + , + ); + await waitFor(() => expect(releaseRefetch).not.toBeNull()); + + // navigate away to ghost while the s1 refetch is in flight + await userEvent.click( + screen.getByRole("button", { name: "switch session" }), + ); + await screen.findByText("ghost"); + // ghost reconnects: its own cursor refetch must still start at 0 — the + // stale s1 batch resolving concurrently must not poison it + rerender( + + + , + ); + rerender( + + + , + ); + await waitFor(() => expect(ghostAfter).toBe("0")); + + await act(async () => { + releaseRefetch?.([staleFrame]); + }); + expect(screen.queryByText("stale s1 frame")).toBeNull(); + }); + + it("ChatStream remounts on session switch (scroll pin reset, S4)", async () => { + mockFetchJson(() => []); + render( + + + , + ); + await screen.findByRole("button", { name: "switch session" }); + const firstScroller = document.querySelector(".chat-scroll"); + expect(firstScroller).not.toBeNull(); + await userEvent.click( + screen.getByRole("button", { name: "switch session" }), + ); + await screen.findByText("ghost"); + expect(document.querySelector(".chat-scroll")).not.toBe(firstScroller); + }); + + it("failed send restores the draft (S6)", async () => { + mockFetchJson((_url, init) => + init?.method === "POST" ? jsonResponse({ error: "nope" }, 500) : [], + ); + renderChat(makeStore()); + const ta = screen.getByLabelText("Message") as HTMLTextAreaElement; + await userEvent.type(ta, "precious draft"); + await userEvent.click(screen.getByLabelText("Send message")); + await vi.waitFor(() => expect(pushToast).toHaveBeenCalledWith("nope")); + expect(ta.value).toBe("precious draft"); + }); +}); + describe("ChatView unknown session", () => { it("unknown id falls back to id title and hides agent-only controls", async () => { mockFetchJson((url) => { @@ -505,7 +832,9 @@ describe("ChatView usage chip", () => { seq = 0; return [ ...historyEvents(), - ev("agent_end", { usage: { inputTokens: 1200, outputTokens: 340, totalCost: 0.02 } }), + ev("agent_end", { + usage: { inputTokens: 1200, outputTokens: 340, totalCost: 0.02 }, + }), ]; } return []; @@ -518,7 +847,8 @@ describe("ChatView usage chip", () => { it("usage chip hidden when no usage seen", async () => { mockFetchJson((url) => { - if (url.startsWith("http://srv/api/sessions/s1/events")) return historyEvents(); + if (url.startsWith("http://srv/api/sessions/s1/events")) + return historyEvents(); return []; }); renderChat(makeStore()); diff --git a/web/src/ChatView.tsx b/web/src/ChatView.tsx index dfe1719..09554e7 100644 --- a/web/src/ChatView.tsx +++ b/web/src/ChatView.tsx @@ -15,6 +15,7 @@ import ChatStream from "./ChatStream"; import TaskPanel from "./TaskPanel"; const HISTORY_LIMIT: number = 1000; +const LATEST_QUERY: string = "latest=1"; const TEXTAREA_MAX_H: number = 200; const SEND_KEY: string = "Enter"; @@ -32,10 +33,15 @@ export default function ChatView({ store, pushToast }: Props) { const [draft, setDraft] = useState(""); const [sending, setSending] = useState(false); const [tasksOpen, setTasksOpen] = useState(false); + const [hasOlder, setHasOlder] = useState(false); + const [loadingOlder, setLoadingOlder] = useState(false); const lastSeqRef = useRef(0); const loadedRef = useRef(false); const taRef = useRef(null); + // guards async fetches against session switches (S1) + const sessionIdRef = useRef(sessionId); + sessionIdRef.current = sessionId; const session = store.sessions.find((s) => s.id === sessionId); @@ -62,21 +68,23 @@ export default function ChatView({ store, pushToast }: Props) { }); }, []); - // history load on mount / session switch + // history load on mount / session switch: newest page, ascending (B1) useEffect(() => { if (sessionId.length === 0) return; loadedRef.current = false; setLoadError(""); setEvents([]); + setHasOlder(false); lastSeqRef.current = 0; let alive = true; void (async () => { try { const evts = await fetchJson( - `${Route.SessionEvents(sessionId)}?after=0&limit=${HISTORY_LIMIT}`, + `${Route.SessionEvents(sessionId)}?${LATEST_QUERY}&limit=${HISTORY_LIMIT}`, ); if (!alive) return; applyEvents(evts); + setHasOlder(evts.length >= HISTORY_LIMIT); } catch (err) { if (alive) setLoadError(errMessage(err)); } finally { @@ -94,20 +102,52 @@ export default function ChatView({ store, pushToast }: Props) { return store.subscribe(sessionId, applyEvents); }, [sessionId, store.state, store, applyEvents]); - // missed events after reconnect (persisted only) + // missed events after reconnect (persisted only); a response for a + // previous session must not merge here nor touch the cursor (S1) useEffect(() => { if (store.state !== "open" || !loadedRef.current) return; + const id: string = sessionId; const after: number = lastSeqRef.current; void fetchJson( - `${Route.SessionEvents(sessionId)}?after=${after}&limit=${HISTORY_LIMIT}`, + `${Route.SessionEvents(id)}?after=${after}&limit=${HISTORY_LIMIT}`, ) - .then(applyEvents) + .then((evts) => { + if (sessionIdRef.current === id) applyEvents(evts); + }) .catch(() => undefined); }, [store.state, sessionId, applyEvents]); const chat = useMemo(() => deriveChat(events), [events]); const tasks = useMemo(() => deriveTasks(events), [events]); + // a closed/crashed container can never emit agent_settled: an offline + // session must never look busy (B2) + const busy: boolean = chat.busy && session?.online !== false; + + const minSeq: number = useMemo( + () => (events.length === 0 ? 0 : Math.min(...events.map((e) => e.seq))), + [events], + ); + + // previous page (seq < oldest loaded), ascending (B1) + const loadOlder = useCallback(async (): Promise => { + // the button only renders while a full page is loaded, so minSeq > 0 + if (sessionIdRef.current !== sessionId || loadingOlder) return; + setLoadingOlder(true); + try { + const evts = await fetchJson( + `${Route.SessionEvents(sessionId)}?before=${minSeq}&limit=${HISTORY_LIMIT}`, + ); + if (sessionIdRef.current !== sessionId) return; + applyEvents(evts); + setHasOlder(evts.length >= HISTORY_LIMIT); + } catch (err) { + pushToast(errMessage(err)); + } finally { + setLoadingOlder(false); + } + }, [sessionId, minSeq, loadingOlder, applyEvents, pushToast]); + const autosize = useCallback((): void => { const el = taRef.current; if (el === null) return; @@ -128,6 +168,8 @@ export default function ChatView({ store, pushToast }: Props) { body: JSON.stringify({ message: text }), }); } catch (err) { + // the message never reached the session: put it back (S6) + setDraft(text); if (err instanceof ApiError && err.status === 409) pushToast("session offline"); else pushToast(errMessage(err)); @@ -215,9 +257,13 @@ export default function ChatView({ store, pushToast }: Props) {
) : ( void loadOlder()} /> )}
@@ -231,7 +277,7 @@ export default function ChatView({ store, pushToast }: Props) { onChange={(e) => setDraft(e.target.value)} onKeyDown={onKeyDown} /> - {chat.busy ? ( + {busy ? (