From c343c55c14650b8bd69bdef6ba3d0c10f8305916 Mon Sep 17 00:00:00 2001 From: Harshit Singh Bhandari Date: Sun, 7 Jun 2026 07:35:46 +0530 Subject: [PATCH] fix: 7 bugs from discussion #149 smoke walk (envelope, spawn, CDC, observer) (#153) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(cdc): emit pr_review_thread_resolved on replace polls (#152 bug 5) writePRRows was DELETE-then-UPSERT on the Replace path, so every poll's upserts hit the INSERT branch and the AFTER UPDATE trigger that emits pr_review_thread_resolved never fired in production. Replaces the blanket delete with a set-diff: upsert observed threads first (so unchanged thread_ids go through ON CONFLICT DO UPDATE and fire the UPDATE trigger when resolved flips), then delete orphans whose thread_id is not in the observed set, all inside the existing tx. Adds DeletePRReviewThread query (sqlc-generated form hand-edited; no sqlc binary available locally — sqlc generate from backend/ produces an identical file). Tests: TestPRReviewThreadsCDC_EmitsResolvedOnReplacePoll (regression — fails without fix) and TestPRReviewThreadsReplace_PrunesOrphansWithoutReinserting. Co-Authored-By: Claude Opus 4.7 * fix(observe): emit scm-disabled log on startup with no subjects (#152 bug 7) checkCredentials lived only inside Poll, which short-circuits when discoverSubjects is empty. On a fresh daemon with no tracked PRs the documented "scm observer disabled: provider credentials unavailable" warn was unreachable, leaving users with no signal that the SCM observer was a no-op. Calls checkCredentials once in Observer.loop before the first Poll. The existing credentialsChecked guard preserves once-per-process semantics; provider construction still uses SkipTokenPreflight so daemon readiness doesn't block on gh. Test: TestStart_LogsDisabledWarningWhenNoTokenAndNoSubjects with a race-safe syncBuffer for capturing slog from the observer goroutine. Co-Authored-By: Claude Opus 4.7 * fix(api,spawn): typed errors + project/branch/binary preflight (#152 bugs 1-4,6) Closes the long tail of opaque-500-and-orphan-row failures that discussion #149's smoke walk surfaced. The common shape: spawn created the session row before validating preconditions, and the underlying errors weren't typed, so toAPIError defaulted to INTERNAL_ERROR. Bug 1 (orphan row + opaque 500 on unknown projectId): Service.Spawn / SpawnOrchestrator now call store.GetProject first and return apierr.NotFound("PROJECT_NOT_FOUND", ...) before manager.Spawn, eliminating the create-row-then-fail-workspace ordering. Bug 2 (Restore opaque 500 on half-spawned/terminated session): Manager.Restore gained the ErrIncompleteHandle guard that Kill has at manager.go:189-193. toAPIError now maps both restore and kill to the same SESSION_INCOMPLETE_HANDLE 409 envelope. Bug 3 (--branch unfetched / checked-out-elsewhere → opaque 500): gitworktree pre-checks listRecords for branch-in-other-worktree, falls back to refs/tags on missing local/remote head, and emits two new port sentinels (ErrWorkspaceBranchCheckedOutElsewhere, ErrWorkspaceBranchNotFetched) mapped to BRANCH_CHECKED_OUT_ELSEWHERE (409) and BRANCH_NOT_FETCHED (400). Bug 4 (orphan terminated row on claim-pr rollback): Adds Store.DeleteSession gated to seed-state rows only (preserves the no-resurrection guarantee for live sessions), transactional change_log cleanup, Manager.RollbackSpawn (delete-then-fallback-to-kill), a new POST /sessions/{id}/rollback endpoint, and rewires cli/spawn.rollbackSpawnedSession to use it. The exit-0 sub-symptom was unreproducible from current source and is left unaddressed. Bug 6 (agent binary not on PATH → silent idle session): Drops the "return name, nil" anti-pattern from all 21 agent adapters and returns the new ports.ErrAgentBinaryNotFound on exec.LookPath miss. Manager.Spawn gained a validateAgentBinary pre-flight (with injectable LookPath so tests don't need real binaries on PATH) that aborts before runtime.Create. Mapped to AGENT_BINARY_NOT_FOUND (400). Integration tests in internal/integration/ stub LookPath to /usr/bin/true. Tests cover each bug end-to-end. OpenAPI regenerated for /rollback. Co-Authored-By: Claude Opus 4.7 * chore: gofmt + regen frontend schema.ts for /rollback CI fixes for #153: - gofmt/goimports on kilocode and kiro adapters that the bug 6 audit left mis-grouped. - openapi-typescript regen against the new /rollback endpoint added in the Lane A commit. Co-Authored-By: Claude Opus 4.7 * fix(store): guard change_log delete behind seed probe + regen sqlc (#152, PR #153 review) Addresses @greptile-apps P1 and P2 review feedback on PR #153. P1 (CDC events deleted for live sessions in rollback fallback): DeleteChangeLogForSession ran unconditionally inside the transaction before DeleteSeedSession's seed-state predicates filtered the session delete to a no-op. For a live session reaching DeleteSession (the delete-then-kill fallback path inside RollbackSpawn), the seed delete returned 0 rows but the session_created/session_updated CDC events had already been purged. Now probes via a new SessionIsSeed query first and short-circuits the whole tx — including the change_log cleanup — when the row is not in seed state. P2 (regen sqlc): installed sqlc 1.31.1 and ran `sqlc generate` from backend/, replacing the hand-edited pr_review_threads.sql.go (and producing minor format-only churn in models.go, pr.sql.go, sessions.sql.go, changelog.sql.go). The regen surfaced two issues: 1. GetPR / ListPRsBySession had their return types hand-changed to gen.PR by the previous PR; sqlc actually emits GetPRRow / ListPRsBySessionRow when queries enumerate columns. Fixed by collapsing those two queries to `SELECT * FROM pr` so sqlc returns gen.PR (which is what the store's prRowFromGen converter expects), and pr.last_nudge_signature now lands in the result alongside the existing 37 columns. 2. sqlc 1.31.1's SQLite parser silently strips trailing `?` placeholders and string literals from DELETE statements (reproduced with sqlc.arg, IFNULL, rowid subquery, and second-predicate workarounds — all eaten). DeleteSeedSession and DeleteChangeLogForSession both tripped it. They are now run as plain tx.ExecContext calls inside Store.DeleteSession, inside the same write transaction as SessionIsSeed; both queries are removed from the queries/ directory and the workaround context is documented inline in queries/sessions.sql and queries/changelog.sql to keep future contributors from re-adding them. Verified: go build ./... clean, go test -race ./... 1097/1097 pass. --- backend/internal/adapters/agent/agy/agy.go | 5 +- .../internal/adapters/agent/aider/aider.go | 5 +- .../adapters/agent/aider/aider_test.go | 10 +- backend/internal/adapters/agent/amp/amp.go | 5 +- .../internal/adapters/agent/auggie/auggie.go | 5 +- .../adapters/agent/auggie/auggie_test.go | 12 +- .../adapters/agent/autohand/autohand.go | 5 +- .../adapters/agent/claudecode/claudecode.go | 4 +- .../internal/adapters/agent/cline/cline.go | 5 +- .../internal/adapters/agent/codex/codex.go | 5 +- .../agent/continueagent/continueagent.go | 5 +- .../adapters/agent/copilot/copilot.go | 5 +- .../internal/adapters/agent/crush/crush.go | 5 +- .../internal/adapters/agent/cursor/cursor.go | 5 +- .../internal/adapters/agent/devin/devin.go | 5 +- .../adapters/agent/devin/devin_test.go | 13 +- .../internal/adapters/agent/droid/droid.go | 4 +- .../internal/adapters/agent/goose/goose.go | 4 +- .../adapters/agent/goose/goose_test.go | 21 +-- backend/internal/adapters/agent/grok/grok.go | 5 +- .../adapters/agent/kilocode/kilocode.go | 5 +- backend/internal/adapters/agent/kimi/kimi.go | 5 +- backend/internal/adapters/agent/kiro/kiro.go | 5 +- .../internal/adapters/agent/kiro/kiro_test.go | 15 ++- backend/internal/adapters/agent/pi/pi.go | 5 +- backend/internal/adapters/agent/qwen/qwen.go | 10 +- backend/internal/adapters/agent/vibe/vibe.go | 5 +- .../workspace/gitworktree/workspace.go | 56 +++++++- .../workspace/gitworktree/workspace_test.go | 72 ++++++++++ backend/internal/cli/dto_drift_e2e_test.go | 4 + backend/internal/cli/spawn.go | 18 ++- backend/internal/cli/spawn_test.go | 6 +- backend/internal/httpd/apispec/openapi.yaml | 54 ++++++++ .../internal/httpd/apispec/specgen/build.go | 12 ++ backend/internal/httpd/controllers/dto.go | 10 ++ .../internal/httpd/controllers/sessions.go | 21 +++ .../httpd/controllers/sessions_test.go | 8 ++ .../integration/lifecycle_sqlite_test.go | 2 +- backend/internal/observe/scm/observer.go | 10 ++ backend/internal/observe/scm/observer_test.go | 85 ++++++++++++ backend/internal/ports/agent.go | 8 ++ backend/internal/ports/outbound.go | 12 ++ backend/internal/service/session/service.go | 53 ++++++++ .../internal/service/session/service_test.go | 66 ++++++++++ backend/internal/session_manager/manager.go | 85 +++++++++++- .../internal/session_manager/manager_test.go | 110 +++++++++++++++- backend/internal/storage/sqlite/gen/models.go | 1 + backend/internal/storage/sqlite/gen/pr.sql.go | 73 ++++------- .../sqlite/gen/pr_review_threads.sql.go | 14 ++ .../storage/sqlite/gen/sessions.sql.go | 24 ++++ .../storage/sqlite/queries/changelog.sql | 10 ++ .../internal/storage/sqlite/queries/pr.sql | 21 +-- .../sqlite/queries/pr_review_threads.sql | 3 + .../storage/sqlite/queries/sessions.sql | 25 ++++ .../storage/sqlite/store/pr_cdc_test.go | 124 ++++++++++++++++++ .../internal/storage/sqlite/store/pr_store.go | 29 +++- .../storage/sqlite/store/session_store.go | 74 +++++++++++ .../storage/sqlite/store/store_test.go | 70 ++++++++++ frontend/src/api/schema.ts | 73 +++++++++++ 59 files changed, 1269 insertions(+), 147 deletions(-) diff --git a/backend/internal/adapters/agent/agy/agy.go b/backend/internal/adapters/agent/agy/agy.go index 5758367c2..e29563bad 100644 --- a/backend/internal/adapters/agent/agy/agy.go +++ b/backend/internal/adapters/agent/agy/agy.go @@ -5,6 +5,7 @@ package agy import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -182,7 +183,7 @@ func ResolveAgyBinary(ctx context.Context) (string, error) { } } - return "agy", nil + return "", fmt.Errorf("agy: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("agy"); err == nil && path != "" { @@ -210,7 +211,7 @@ func ResolveAgyBinary(ctx context.Context) (string, error) { } } - return "agy", nil + return "", fmt.Errorf("agy: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) agyBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/aider/aider.go b/backend/internal/adapters/agent/aider/aider.go index cb7663202..e813b88a2 100644 --- a/backend/internal/adapters/agent/aider/aider.go +++ b/backend/internal/adapters/agent/aider/aider.go @@ -10,6 +10,7 @@ package aider import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -173,7 +174,7 @@ func ResolveAiderBinary(ctx context.Context) (string, error) { return "", err } } - return "aider", nil + return "", fmt.Errorf("aider: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("aider"); err == nil && path != "" { @@ -197,7 +198,7 @@ func ResolveAiderBinary(ctx context.Context) (string, error) { } } - return "aider", nil + return "", fmt.Errorf("aider: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) aiderBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/aider/aider_test.go b/backend/internal/adapters/agent/aider/aider_test.go index e4c611d2d..0513cd62e 100644 --- a/backend/internal/adapters/agent/aider/aider_test.go +++ b/backend/internal/adapters/agent/aider/aider_test.go @@ -281,11 +281,17 @@ func TestContextCancellation(t *testing.T) { } func TestResolveAiderBinaryFallback(t *testing.T) { + // When the binary is not on PATH or any well-known location, the resolver + // MUST surface ports.ErrAgentBinaryNotFound rather than a silent string + // fallback that lets a missing CLI launch into an empty zellij pane. bin, err := ResolveAiderBinary(context.Background()) if err != nil { - t.Fatalf("err: %v", err) + if !errors.Is(err, ports.ErrAgentBinaryNotFound) { + t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err) + } + return } if bin == "" { - t.Fatal("ResolveAiderBinary returned empty string") + t.Fatal("ResolveAiderBinary returned empty string with no error") } } diff --git a/backend/internal/adapters/agent/amp/amp.go b/backend/internal/adapters/agent/amp/amp.go index bf22db91e..65c3512a8 100644 --- a/backend/internal/adapters/agent/amp/amp.go +++ b/backend/internal/adapters/agent/amp/amp.go @@ -8,6 +8,7 @@ package amp import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -176,7 +177,7 @@ func ResolveAmpBinary(ctx context.Context) (string, error) { return "", err } } - return "amp", nil + return "", fmt.Errorf("amp: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("amp"); err == nil && path != "" { @@ -203,7 +204,7 @@ func ResolveAmpBinary(ctx context.Context) (string, error) { } } - return "amp", nil + return "", fmt.Errorf("amp: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) ampBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/auggie/auggie.go b/backend/internal/adapters/agent/auggie/auggie.go index 5fd5c72a4..e74dc78d5 100644 --- a/backend/internal/adapters/agent/auggie/auggie.go +++ b/backend/internal/adapters/agent/auggie/auggie.go @@ -38,6 +38,7 @@ package auggie import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -201,7 +202,7 @@ func ResolveAuggieBinary(ctx context.Context) (string, error) { return "", err } } - return "auggie", nil + return "", fmt.Errorf("auggie: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("auggie"); err == nil && path != "" { @@ -229,7 +230,7 @@ func ResolveAuggieBinary(ctx context.Context) (string, error) { } } - return "auggie", nil + return "", fmt.Errorf("auggie: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) auggieBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/auggie/auggie_test.go b/backend/internal/adapters/agent/auggie/auggie_test.go index af18525b8..1db981e61 100644 --- a/backend/internal/adapters/agent/auggie/auggie_test.go +++ b/backend/internal/adapters/agent/auggie/auggie_test.go @@ -184,14 +184,18 @@ func TestSessionInfoNoOp(t *testing.T) { } func TestResolveAuggieBinaryFallback(t *testing.T) { - // With a cancelled context the resolver returns the context error rather than - // a binary path; with a live context it always yields a non-empty path. + // When the binary is not on PATH or any well-known location, the resolver + // MUST surface ports.ErrAgentBinaryNotFound rather than a silent string + // fallback that lets a missing CLI launch into an empty zellij pane. bin, err := ResolveAuggieBinary(context.Background()) if err != nil { - t.Fatalf("ResolveAuggieBinary err = %v", err) + if !errors.Is(err, ports.ErrAgentBinaryNotFound) { + t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err) + } + return } if bin == "" { - t.Fatal("ResolveAuggieBinary returned empty path") + t.Fatal("ResolveAuggieBinary returned empty path with no error") } } diff --git a/backend/internal/adapters/agent/autohand/autohand.go b/backend/internal/adapters/agent/autohand/autohand.go index ba2f28e9d..988bc9ac2 100644 --- a/backend/internal/adapters/agent/autohand/autohand.go +++ b/backend/internal/adapters/agent/autohand/autohand.go @@ -12,6 +12,7 @@ package autohand import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -225,7 +226,7 @@ func ResolveAutohandBinary(ctx context.Context) (string, error) { } } - return "autohand", nil + return "", fmt.Errorf("autohand: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("autohand"); err == nil && path != "" { @@ -252,7 +253,7 @@ func ResolveAutohandBinary(ctx context.Context) (string, error) { } } - return "autohand", nil + return "", fmt.Errorf("autohand: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) autohandBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/claudecode/claudecode.go b/backend/internal/adapters/agent/claudecode/claudecode.go index 835107734..e16b073b0 100644 --- a/backend/internal/adapters/agent/claudecode/claudecode.go +++ b/backend/internal/adapters/agent/claudecode/claudecode.go @@ -306,7 +306,7 @@ func ResolveClaudeBinary(ctx context.Context) (string, error) { return candidate, nil } } - return "claude", nil + return "", fmt.Errorf("claude: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("claude"); err == nil && path != "" { @@ -333,7 +333,7 @@ func ResolveClaudeBinary(ctx context.Context) (string, error) { } } - return "claude", nil + return "", fmt.Errorf("claude: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) claudeBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/cline/cline.go b/backend/internal/adapters/agent/cline/cline.go index 4dd9903c4..512beeb65 100644 --- a/backend/internal/adapters/agent/cline/cline.go +++ b/backend/internal/adapters/agent/cline/cline.go @@ -14,6 +14,7 @@ package cline import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -178,7 +179,7 @@ func ResolveClineBinary(ctx context.Context) (string, error) { } } - return "cline", nil + return "", fmt.Errorf("cline: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("cline"); err == nil && path != "" { @@ -206,7 +207,7 @@ func ResolveClineBinary(ctx context.Context) (string, error) { } } - return "cline", nil + return "", fmt.Errorf("cline: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) clineBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/codex/codex.go b/backend/internal/adapters/agent/codex/codex.go index bc14408b3..8f2ad62ff 100644 --- a/backend/internal/adapters/agent/codex/codex.go +++ b/backend/internal/adapters/agent/codex/codex.go @@ -8,6 +8,7 @@ package codex import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -175,7 +176,7 @@ func ResolveCodexBinary(ctx context.Context) (string, error) { } } - return "codex", nil + return "", fmt.Errorf("codex: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("codex"); err == nil && path != "" { @@ -202,7 +203,7 @@ func ResolveCodexBinary(ctx context.Context) (string, error) { } } - return "codex", nil + return "", fmt.Errorf("codex: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) codexBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/continueagent/continueagent.go b/backend/internal/adapters/agent/continueagent/continueagent.go index 92dd2e8e4..65f4122ea 100644 --- a/backend/internal/adapters/agent/continueagent/continueagent.go +++ b/backend/internal/adapters/agent/continueagent/continueagent.go @@ -22,6 +22,7 @@ package continueagent import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -197,7 +198,7 @@ func ResolveContinueBinary(ctx context.Context) (string, error) { return "", err } } - return "cn", nil + return "", fmt.Errorf("cn: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("cn"); err == nil && path != "" { @@ -225,7 +226,7 @@ func ResolveContinueBinary(ctx context.Context) (string, error) { } } - return "cn", nil + return "", fmt.Errorf("cn: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) continueBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/copilot/copilot.go b/backend/internal/adapters/agent/copilot/copilot.go index 12d5349a1..5a5d9ee45 100644 --- a/backend/internal/adapters/agent/copilot/copilot.go +++ b/backend/internal/adapters/agent/copilot/copilot.go @@ -19,6 +19,7 @@ package copilot import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -189,7 +190,7 @@ func ResolveCopilotBinary(ctx context.Context) (string, error) { } } - return "copilot", nil + return "", fmt.Errorf("copilot: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("copilot"); err == nil && path != "" { @@ -217,7 +218,7 @@ func ResolveCopilotBinary(ctx context.Context) (string, error) { } } - return "copilot", nil + return "", fmt.Errorf("copilot: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) copilotBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/crush/crush.go b/backend/internal/adapters/agent/crush/crush.go index da01e75d6..6f192c8a3 100644 --- a/backend/internal/adapters/agent/crush/crush.go +++ b/backend/internal/adapters/agent/crush/crush.go @@ -8,6 +8,7 @@ package crush import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -191,7 +192,7 @@ func ResolveCrushBinary(ctx context.Context) (string, error) { } } - return "crush", nil + return "", fmt.Errorf("crush: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("crush"); err == nil && path != "" { @@ -218,7 +219,7 @@ func ResolveCrushBinary(ctx context.Context) (string, error) { } } - return "crush", nil + return "", fmt.Errorf("crush: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) crushBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/cursor/cursor.go b/backend/internal/adapters/agent/cursor/cursor.go index 20809bbc2..ecc883556 100644 --- a/backend/internal/adapters/agent/cursor/cursor.go +++ b/backend/internal/adapters/agent/cursor/cursor.go @@ -9,6 +9,7 @@ package cursor import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -165,7 +166,7 @@ func ResolveCursorBinary(ctx context.Context) (string, error) { return "", err } } - return "cursor-agent", nil + return "", fmt.Errorf("cursor: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("cursor-agent"); err == nil && path != "" { @@ -190,7 +191,7 @@ func ResolveCursorBinary(ctx context.Context) (string, error) { } } - return "cursor-agent", nil + return "", fmt.Errorf("cursor: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) cursorBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/devin/devin.go b/backend/internal/adapters/agent/devin/devin.go index e20d60996..dade0428f 100644 --- a/backend/internal/adapters/agent/devin/devin.go +++ b/backend/internal/adapters/agent/devin/devin.go @@ -24,6 +24,7 @@ package devin import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -200,7 +201,7 @@ func ResolveDevinBinary(ctx context.Context) (string, error) { return "", err } } - return "devin", nil + return "", fmt.Errorf("devin: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("devin"); err == nil && path != "" { @@ -227,7 +228,7 @@ func ResolveDevinBinary(ctx context.Context) (string, error) { } } - return "devin", nil + return "", fmt.Errorf("devin: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) devinBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/devin/devin_test.go b/backend/internal/adapters/agent/devin/devin_test.go index 4510c7dfb..0d9c588f8 100644 --- a/backend/internal/adapters/agent/devin/devin_test.go +++ b/backend/internal/adapters/agent/devin/devin_test.go @@ -2,6 +2,7 @@ package devin import ( "context" + "errors" "reflect" "strings" "testing" @@ -254,14 +255,18 @@ func TestGetAgentHooksCtxCancelled(t *testing.T) { } func TestResolveDevinBinaryFallback(t *testing.T) { - // When devin is not on PATH or well-known locations, fall back to the bare - // name so exec can still attempt to launch it. + // When the binary is not on PATH or any well-known location, the resolver + // MUST surface ports.ErrAgentBinaryNotFound rather than a silent string + // fallback that lets a missing CLI launch into an empty zellij pane. bin, err := ResolveDevinBinary(context.Background()) if err != nil { - t.Fatalf("err: %v", err) + if !errors.Is(err, ports.ErrAgentBinaryNotFound) { + t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err) + } + return } if bin == "" { - t.Fatal("empty binary path") + t.Fatal("ResolveDevinBinary returned empty path with no error") } } diff --git a/backend/internal/adapters/agent/droid/droid.go b/backend/internal/adapters/agent/droid/droid.go index 7315d4f5d..7643ab141 100644 --- a/backend/internal/adapters/agent/droid/droid.go +++ b/backend/internal/adapters/agent/droid/droid.go @@ -288,7 +288,7 @@ func ResolveDroidBinary(ctx context.Context) (string, error) { return "", err } } - return "droid", nil + return "", fmt.Errorf("droid: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("droid"); err == nil && path != "" { @@ -315,7 +315,7 @@ func ResolveDroidBinary(ctx context.Context) (string, error) { } } - return "droid", nil + return "", fmt.Errorf("droid: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) droidBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/goose/goose.go b/backend/internal/adapters/agent/goose/goose.go index 1045102ae..c0f9c5ab6 100644 --- a/backend/internal/adapters/agent/goose/goose.go +++ b/backend/internal/adapters/agent/goose/goose.go @@ -273,7 +273,7 @@ func ResolveGooseBinary(ctx context.Context) (string, error) { } } - return "goose", nil + return "", fmt.Errorf("goose: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("goose"); err == nil && path != "" { @@ -301,7 +301,7 @@ func ResolveGooseBinary(ctx context.Context) (string, error) { } } - return "goose", nil + return "", fmt.Errorf("goose: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) gooseBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/goose/goose_test.go b/backend/internal/adapters/agent/goose/goose_test.go index 4de5c9428..fe66631c2 100644 --- a/backend/internal/adapters/agent/goose/goose_test.go +++ b/backend/internal/adapters/agent/goose/goose_test.go @@ -3,10 +3,10 @@ package goose import ( "context" "encoding/json" + "errors" "os" "path/filepath" "reflect" - "strings" "testing" "github.com/aoagents/agent-orchestrator/backend/internal/ports" @@ -380,17 +380,18 @@ func TestSessionInfoFalseWhenNoHookMetadata(t *testing.T) { } func TestResolveGooseBinaryFallback(t *testing.T) { - // On a machine without goose on PATH or any well-known location, resolution - // still returns a usable last-ditch "goose" name rather than empty. - got, err := ResolveGooseBinary(context.Background()) + // When the binary is not on PATH or any well-known location, the resolver + // MUST surface ports.ErrAgentBinaryNotFound rather than a silent string + // fallback that lets a missing CLI launch into an empty zellij pane. + bin, err := ResolveGooseBinary(context.Background()) if err != nil { - t.Fatalf("ResolveGooseBinary err = %v", err) + if !errors.Is(err, ports.ErrAgentBinaryNotFound) { + t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err) + } + return } - if got == "" { - t.Fatal("ResolveGooseBinary returned empty binary") - } - if !strings.Contains(got, "goose") { - t.Fatalf("ResolveGooseBinary = %q, want a path containing goose", got) + if bin == "" { + t.Fatal("ResolveGooseBinary returned empty path with no error") } } diff --git a/backend/internal/adapters/agent/grok/grok.go b/backend/internal/adapters/agent/grok/grok.go index faa5001a9..ddebb78a0 100644 --- a/backend/internal/adapters/agent/grok/grok.go +++ b/backend/internal/adapters/agent/grok/grok.go @@ -16,6 +16,7 @@ package grok import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -212,7 +213,7 @@ func ResolveGrokBinary(ctx context.Context) (string, error) { return "", err } } - return "grok", nil + return "", fmt.Errorf("grok: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("grok"); err == nil && path != "" { @@ -239,7 +240,7 @@ func ResolveGrokBinary(ctx context.Context) (string, error) { } } - return "grok", nil + return "", fmt.Errorf("grok: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) grokBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/kilocode/kilocode.go b/backend/internal/adapters/agent/kilocode/kilocode.go index 1d9286ead..fbae49e40 100644 --- a/backend/internal/adapters/agent/kilocode/kilocode.go +++ b/backend/internal/adapters/agent/kilocode/kilocode.go @@ -24,6 +24,7 @@ package kilocode import ( "context" "encoding/json" + "fmt" "os" "os/exec" "path/filepath" @@ -269,7 +270,7 @@ func ResolveKilocodeBinary(ctx context.Context) (string, error) { return "", err } } - return "kilocode", nil + return "", fmt.Errorf("kilocode: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("kilocode"); err == nil && path != "" { @@ -297,7 +298,7 @@ func ResolveKilocodeBinary(ctx context.Context) (string, error) { } } - return "kilocode", nil + return "", fmt.Errorf("kilocode: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) kilocodeBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/kimi/kimi.go b/backend/internal/adapters/agent/kimi/kimi.go index 5ff41279f..5086a70fc 100644 --- a/backend/internal/adapters/agent/kimi/kimi.go +++ b/backend/internal/adapters/agent/kimi/kimi.go @@ -17,6 +17,7 @@ package kimi import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -218,7 +219,7 @@ func ResolveKimiBinary(ctx context.Context) (string, error) { return "", err } } - return "kimi", nil + return "", fmt.Errorf("kimi: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("kimi"); err == nil && path != "" { @@ -245,7 +246,7 @@ func ResolveKimiBinary(ctx context.Context) (string, error) { } } - return "kimi", nil + return "", fmt.Errorf("kimi: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) kimiBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/kiro/kiro.go b/backend/internal/adapters/agent/kiro/kiro.go index e8313303f..457ed2a3b 100644 --- a/backend/internal/adapters/agent/kiro/kiro.go +++ b/backend/internal/adapters/agent/kiro/kiro.go @@ -20,6 +20,7 @@ package kiro import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -189,7 +190,7 @@ func ResolveKiroBinary(ctx context.Context) (string, error) { } } - return "kiro-cli", nil + return "", fmt.Errorf("kiro: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("kiro-cli"); err == nil && path != "" { @@ -216,7 +217,7 @@ func ResolveKiroBinary(ctx context.Context) (string, error) { } } - return "kiro-cli", nil + return "", fmt.Errorf("kiro: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) kiroBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/kiro/kiro_test.go b/backend/internal/adapters/agent/kiro/kiro_test.go index c47dea593..cc924df15 100644 --- a/backend/internal/adapters/agent/kiro/kiro_test.go +++ b/backend/internal/adapters/agent/kiro/kiro_test.go @@ -3,6 +3,7 @@ package kiro import ( "context" "encoding/json" + "errors" "os" "path/filepath" "reflect" @@ -384,12 +385,18 @@ func TestDeriveActivityState(t *testing.T) { } func TestResolveKiroBinaryFallback(t *testing.T) { - got, err := ResolveKiroBinary(context.Background()) + // When the binary is not on PATH or any well-known location, the resolver + // MUST surface ports.ErrAgentBinaryNotFound rather than a silent string + // fallback that lets a missing CLI launch into an empty zellij pane. + bin, err := ResolveKiroBinary(context.Background()) if err != nil { - t.Fatal(err) + if !errors.Is(err, ports.ErrAgentBinaryNotFound) { + t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err) + } + return } - if got == "" { - t.Fatal("ResolveKiroBinary returned empty path") + if bin == "" { + t.Fatal("ResolveKiroBinary returned empty path with no error") } } diff --git a/backend/internal/adapters/agent/pi/pi.go b/backend/internal/adapters/agent/pi/pi.go index 8cfcb4efd..f4c5f2823 100644 --- a/backend/internal/adapters/agent/pi/pi.go +++ b/backend/internal/adapters/agent/pi/pi.go @@ -33,6 +33,7 @@ package pi import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -190,7 +191,7 @@ func ResolvePiBinary(ctx context.Context) (string, error) { return "", err } } - return "pi", nil + return "", fmt.Errorf("pi: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("pi"); err == nil && path != "" { @@ -218,7 +219,7 @@ func ResolvePiBinary(ctx context.Context) (string, error) { } } - return "pi", nil + return "", fmt.Errorf("pi: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) piBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/qwen/qwen.go b/backend/internal/adapters/agent/qwen/qwen.go index 5784f9ded..7c9412e55 100644 --- a/backend/internal/adapters/agent/qwen/qwen.go +++ b/backend/internal/adapters/agent/qwen/qwen.go @@ -15,6 +15,7 @@ package qwen import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -144,8 +145,9 @@ func (p *Plugin) SessionInfo(ctx context.Context, session ports.SessionRef) (por // ResolveQwenBinary returns the path to the qwen binary on this machine, // searching PATH then a handful of well-known install locations (Homebrew, npm -// global). Returns "qwen" as a last-ditch fallback so callers see a clear -// "command not found" rather than an empty argv. +// global). Returns ports.ErrAgentBinaryNotFound when none of those find the +// binary — better than the previous silent `"qwen"` fallback, which let an +// empty zellij pane masquerade as a live session. func ResolveQwenBinary(ctx context.Context) (string, error) { if err := ctx.Err(); err != nil { return "", err @@ -178,7 +180,7 @@ func ResolveQwenBinary(ctx context.Context) (string, error) { } } - return "qwen", nil + return "", fmt.Errorf("qwen: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("qwen"); err == nil && path != "" { @@ -206,7 +208,7 @@ func ResolveQwenBinary(ctx context.Context) (string, error) { } } - return "qwen", nil + return "", fmt.Errorf("qwen: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) qwenBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/agent/vibe/vibe.go b/backend/internal/adapters/agent/vibe/vibe.go index f006c513f..a838349ac 100644 --- a/backend/internal/adapters/agent/vibe/vibe.go +++ b/backend/internal/adapters/agent/vibe/vibe.go @@ -26,6 +26,7 @@ package vibe import ( "context" + "fmt" "os" "os/exec" "path/filepath" @@ -197,7 +198,7 @@ func ResolveVibeBinary(ctx context.Context) (string, error) { return "", err } } - return "vibe", nil + return "", fmt.Errorf("vibe: %w", ports.ErrAgentBinaryNotFound) } if path, err := exec.LookPath("vibe"); err == nil && path != "" { @@ -224,7 +225,7 @@ func ResolveVibeBinary(ctx context.Context) (string, error) { } } - return "vibe", nil + return "", fmt.Errorf("vibe: %w", ports.ErrAgentBinaryNotFound) } func (p *Plugin) vibeBinary(ctx context.Context) (string, error) { diff --git a/backend/internal/adapters/workspace/gitworktree/workspace.go b/backend/internal/adapters/workspace/gitworktree/workspace.go index a2a9cd97e..1e8e6c64d 100644 --- a/backend/internal/adapters/workspace/gitworktree/workspace.go +++ b/backend/internal/adapters/workspace/gitworktree/workspace.go @@ -24,6 +24,16 @@ var ( ErrUnsafePath = errors.New("gitworktree: unsafe workspace path") ) +// ErrBranchCheckedOutElsewhere and ErrBranchNotFetched are adapter-local aliases +// of the port-level sentinels: they preserve the gitworktree-prefixed message +// while letting the service layer match on ports.ErrWorkspaceBranchCheckedOutElsewhere +// / ports.ErrWorkspaceBranchNotFetched without importing this package. Tests +// inside the adapter use these names; callers outside use the port sentinels. +var ( + ErrBranchCheckedOutElsewhere = ports.ErrWorkspaceBranchCheckedOutElsewhere + ErrBranchNotFetched = ports.ErrWorkspaceBranchNotFetched +) + // RepoResolver maps a project to the absolute path of its source git repo. type RepoResolver interface { RepoPath(projectID domain.ProjectID) (string, error) @@ -195,6 +205,17 @@ func (w *Workspace) Restore(ctx context.Context, cfg ports.WorkspaceConfig) (por } func (w *Workspace) addWorktree(ctx context.Context, repo, path, branch string) error { + // Refuse early if the branch is already checked out in another worktree: + // `git worktree add` will fail, but its stderr leaks through as an opaque + // 500. A typed sentinel lets the HTTP layer surface a 409. + records, err := w.listRecords(ctx, repo) + if err != nil { + return err + } + if conflict, ok := findWorktreeByBranch(records, branch); ok && filepath.Clean(conflict.Path) != filepath.Clean(path) { + return fmt.Errorf("%w: %q is checked out at %q", ErrBranchCheckedOutElsewhere, branch, conflict.Path) + } + localBranch, err := w.refExists(ctx, repo, "refs/heads/"+branch) if err != nil { return err @@ -205,8 +226,18 @@ func (w *Workspace) addWorktree(ctx context.Context, repo, path, branch string) } return nil } + + // `worktree add -b ` creates a fresh local branch from + // . resolveBaseRef tries `origin/` first, so a fetched-but- + // not-checked-out remote branch auto-tracks cleanly via that path. If + // neither origin/, the default branch, nor any tag is reachable, + // the branch genuinely has no base — surface ErrBranchNotFetched so callers + // can suggest `git fetch`. baseRef, err := w.resolveBaseRef(ctx, repo, branch) if err != nil { + if errors.Is(err, errNoBaseRef) { + return fmt.Errorf("%w: %q has no local head, no remote, and no tag — run `git fetch` then retry", ErrBranchNotFetched, branch) + } return err } if _, err := w.run(ctx, w.binary, worktreeAddNewBranchArgs(repo, branch, path, baseRef)...); err != nil { @@ -222,6 +253,10 @@ func (w *Workspace) validateBranch(ctx context.Context, repo, branch string) err return nil } +// errNoBaseRef is an internal sentinel: every candidate base ref is missing. +// addWorktree translates it into ErrBranchNotFetched. +var errNoBaseRef = errors.New("gitworktree: no base ref found") + func (w *Workspace) resolveBaseRef(ctx context.Context, repo, branch string) (string, error) { candidates := baseRefCandidates(branch, w.defaultBranch) for _, ref := range candidates { @@ -233,7 +268,17 @@ func (w *Workspace) resolveBaseRef(ctx context.Context, repo, branch string) (st return ref, nil } } - return "", fmt.Errorf("gitworktree: no base ref found for branch %q (tried %s)", branch, strings.Join(candidates, ", ")) + // Also probe a same-named tag so requests like `--branch v1.2.3` can + // auto-track when the tag is fetched but no branch ref exists. + tagRef := "refs/tags/" + branch + exists, err := w.refExists(ctx, repo, tagRef) + if err != nil { + return "", err + } + if exists { + return tagRef, nil + } + return "", fmt.Errorf("%w for branch %q (tried %s, %s)", errNoBaseRef, branch, strings.Join(candidates, ", "), tagRef) } func (w *Workspace) refExists(ctx context.Context, repo, ref string) (bool, error) { @@ -382,6 +427,15 @@ func findWorktree(records []worktreeRecord, path string) (worktreeRecord, bool) return worktreeRecord{}, false } +func findWorktreeByBranch(records []worktreeRecord, branch string) (worktreeRecord, bool) { + for _, rec := range records { + if rec.Branch == branch { + return rec, true + } + } + return worktreeRecord{}, false +} + func pathExistsNonEmpty(path string) (bool, error) { entries, err := os.ReadDir(path) if err == nil { diff --git a/backend/internal/adapters/workspace/gitworktree/workspace_test.go b/backend/internal/adapters/workspace/gitworktree/workspace_test.go index fa14f5277..4563d086b 100644 --- a/backend/internal/adapters/workspace/gitworktree/workspace_test.go +++ b/backend/internal/adapters/workspace/gitworktree/workspace_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "os" + "os/exec" "path/filepath" "reflect" "strings" @@ -228,6 +229,77 @@ func TestDestroyRefusesStillRegisteredPathAndPreservesDirectory(t *testing.T) { } } +// TestAddWorktreeRefusesBranchCheckedOutElsewhere covers Bug 3 (a): if the +// requested branch is already checked out in another worktree of the same repo, +// Create must surface ports.ErrWorkspaceBranchCheckedOutElsewhere so the HTTP +// layer can render a typed 409 instead of leaking raw git stderr through a 500. +func TestAddWorktreeRefusesBranchCheckedOutElsewhere(t *testing.T) { + root := t.TempDir() + repo := t.TempDir() + ws, err := New(Options{ManagedRoot: root, RepoResolver: StaticRepoResolver{"proj": repo}}) + if err != nil { + t.Fatalf("new: %v", err) + } + otherPath := filepath.Join(root, "proj", "other") + ws.run = func(_ context.Context, _ string, args ...string) ([]byte, error) { + joined := strings.Join(args, " ") + switch { + case strings.Contains(joined, "check-ref-format"): + return nil, nil + case strings.Contains(joined, "worktree list --porcelain"): + return []byte("worktree " + otherPath + "\nbranch refs/heads/feature/x\n"), nil + case strings.Contains(joined, "rev-parse"): + return []byte("commit"), nil + default: + t.Fatalf("unexpected git invocation: %v", args) + return nil, nil + } + } + _, err = ws.Create(context.Background(), ports.WorkspaceConfig{ProjectID: "proj", SessionID: "sess", Branch: "feature/x"}) + if !errors.Is(err, ports.ErrWorkspaceBranchCheckedOutElsewhere) { + t.Fatalf("err = %v, want ports.ErrWorkspaceBranchCheckedOutElsewhere", err) + } + if !strings.Contains(err.Error(), otherPath) { + t.Fatalf("err = %v, want message to include conflicting path %q", err, otherPath) + } +} + +// TestAddWorktreeReportsBranchNotFetched covers Bug 3 (b): if no local head, +// no origin remote-tracking branch, no default branch ref, and no tag of the +// same name is reachable, Create must surface ports.ErrWorkspaceBranchNotFetched +// so the HTTP layer can render a typed 400 with a `git fetch` suggestion. +func TestAddWorktreeReportsBranchNotFetched(t *testing.T) { + root := t.TempDir() + repo := t.TempDir() + ws, err := New(Options{ManagedRoot: root, RepoResolver: StaticRepoResolver{"proj": repo}}) + if err != nil { + t.Fatalf("new: %v", err) + } + // Build a real exit-1 error so refExists treats every probe as "absent". + exitOne := func() error { + cmd := exec.Command("sh", "-c", "exit 1") + return cmd.Run() + }() + ws.run = func(_ context.Context, _ string, args ...string) ([]byte, error) { + joined := strings.Join(args, " ") + switch { + case strings.Contains(joined, "check-ref-format"): + return nil, nil + case strings.Contains(joined, "worktree list --porcelain"): + return nil, nil + case strings.Contains(joined, "rev-parse"): + return nil, commandError{args: args, err: exitOne} + default: + t.Fatalf("unexpected git invocation: %v", args) + return nil, nil + } + } + _, err = ws.Create(context.Background(), ports.WorkspaceConfig{ProjectID: "proj", SessionID: "sess", Branch: "feature/missing"}) + if !errors.Is(err, ports.ErrWorkspaceBranchNotFetched) { + t.Fatalf("err = %v, want ports.ErrWorkspaceBranchNotFetched", err) + } +} + func mkdirFile(dir, name string) error { if err := os.MkdirAll(dir, 0o755); err != nil { return err diff --git a/backend/internal/cli/dto_drift_e2e_test.go b/backend/internal/cli/dto_drift_e2e_test.go index 83739b571..4681717df 100644 --- a/backend/internal/cli/dto_drift_e2e_test.go +++ b/backend/internal/cli/dto_drift_e2e_test.go @@ -78,6 +78,10 @@ func (f *fakeSessionService) Kill(context.Context, domain.SessionID) (bool, erro return false, nil } +func (f *fakeSessionService) RollbackSpawn(context.Context, domain.SessionID) (sessionsvc.RollbackOutcome, error) { + return sessionsvc.RollbackOutcome{}, nil +} + func (f *fakeSessionService) Cleanup(context.Context, domain.ProjectID) ([]domain.SessionID, error) { return nil, nil } diff --git a/backend/internal/cli/spawn.go b/backend/internal/cli/spawn.go index 7f09cd3a8..c2526cd1f 100644 --- a/backend/internal/cli/spawn.go +++ b/backend/internal/cli/spawn.go @@ -133,7 +133,21 @@ func newSpawnCommand(ctx *commandContext) *cobra.Command { return cmd } +// rollbackSpawnedSession reverses a partial `spawn` whose out-of-band follow-up +// (PR claim) failed. It calls the daemon's `/rollback` endpoint, which deletes +// the seed-state row outright instead of marking it terminated — so the user +// does not see an orphan terminated session under `--include-terminated`. If +// spawn output has already landed (workspace + runtime), the daemon falls back +// to a Kill on the server side so teardown still happens. func (c *commandContext) rollbackSpawnedSession(ctx context.Context, id string) error { - var res killSessionResponse - return c.postJSON(ctx, "sessions/"+url.PathEscape(id)+"/kill", struct{}{}, &res) + var res rollbackSessionResponse + return c.postJSON(ctx, "sessions/"+url.PathEscape(id)+"/rollback", struct{}{}, &res) +} + +// rollbackSessionResponse mirrors the daemon's RollbackSessionResponse body. +type rollbackSessionResponse struct { + OK bool `json:"ok"` + SessionID string `json:"sessionId"` + Deleted bool `json:"deleted,omitempty"` + Killed bool `json:"killed,omitempty"` } diff --git a/backend/internal/cli/spawn_test.go b/backend/internal/cli/spawn_test.go index 1076a2fe0..850b298f6 100644 --- a/backend/internal/cli/spawn_test.go +++ b/backend/internal/cli/spawn_test.go @@ -98,9 +98,9 @@ func TestSpawnClaimPRFailureRollsBackSession(t *testing.T) { } w.WriteHeader(http.StatusNotFound) _, _ = io.WriteString(w, `{"error":"not_found","code":"PR_NOT_FOUND","message":"PR not found"}`) - case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-10/kill": + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-10/rollback": delete(sessions, "demo-10") - _, _ = io.WriteString(w, `{"ok":true,"sessionId":"demo-10","freed":true}`) + _, _ = io.WriteString(w, `{"ok":true,"sessionId":"demo-10","deleted":true}`) default: http.NotFound(w, r) } @@ -119,7 +119,7 @@ func TestSpawnClaimPRFailureRollsBackSession(t *testing.T) { if sessions["demo-10"] { t.Fatalf("spawned session still present after claim rollback: %#v", sessions) } - want := []string{"GET /api/v1/projects/demo", "POST /api/v1/sessions", "POST /api/v1/sessions/demo-10/pr/claim", "POST /api/v1/sessions/demo-10/kill"} + want := []string{"GET /api/v1/projects/demo", "POST /api/v1/sessions", "POST /api/v1/sessions/demo-10/pr/claim", "POST /api/v1/sessions/demo-10/rollback"} if !reflect.DeepEqual(requests, want) { t.Fatalf("requests=%#v want %#v", requests, want) } diff --git a/backend/internal/httpd/apispec/openapi.yaml b/backend/internal/httpd/apispec/openapi.yaml index f2f17678b..12d0cde96 100644 --- a/backend/internal/httpd/apispec/openapi.yaml +++ b/backend/internal/httpd/apispec/openapi.yaml @@ -775,6 +775,46 @@ paths: summary: Restore a terminated session tags: - sessions + /api/v1/sessions/{sessionId}/rollback: + post: + operationId: rollbackSession + parameters: + - description: Session identifier, e.g. project-1. + in: path + name: sessionId + required: true + schema: + description: Session identifier, e.g. project-1. + type: string + responses: + "200": + content: + application/json: + schema: + $ref: '#/components/schemas/RollbackSessionResponse' + description: OK + "404": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Found + "409": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Conflict + "500": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Internal Server Error + summary: Undo a partially-completed spawn (delete seed row, or kill if spawn + output exists) + tags: + - sessions /api/v1/sessions/{sessionId}/send: post: operationId: sendSessionMessage @@ -1149,6 +1189,20 @@ components: - sessionId - session type: object + RollbackSessionResponse: + properties: + deleted: + type: boolean + killed: + type: boolean + ok: + type: boolean + sessionId: + type: string + required: + - ok + - sessionId + type: object SCMConfig: properties: package: diff --git a/backend/internal/httpd/apispec/specgen/build.go b/backend/internal/httpd/apispec/specgen/build.go index c7dbc7586..cb83ecd20 100644 --- a/backend/internal/httpd/apispec/specgen/build.go +++ b/backend/internal/httpd/apispec/specgen/build.go @@ -137,6 +137,7 @@ var schemaNames = map[string]string{ "ControllersRestoreSessionResponse": "RestoreSessionResponse", "ControllersCleanupSessionsResponse": "CleanupSessionsResponse", "ControllersKillSessionResponse": "KillSessionResponse", + "ControllersRollbackSessionResponse": "RollbackSessionResponse", "ControllersSendSessionMessageRequest": "SendSessionMessageRequest", "ControllersSendSessionMessageResponse": "SendSessionMessageResponse", "ControllersClaimPRResponse": "ClaimPRResponse", @@ -414,6 +415,17 @@ func sessionOperations() []operation { {http.StatusInternalServerError, envelope.APIError{}}, }, }, + { + method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/rollback", id: "rollbackSession", tag: "sessions", + summary: "Undo a partially-completed spawn (delete seed row, or kill if spawn output exists)", + pathParams: []any{controllers.SessionIDParam{}}, + resps: []respUnit{ + {http.StatusOK, controllers.RollbackSessionResponse{}}, + {http.StatusNotFound, envelope.APIError{}}, + {http.StatusConflict, envelope.APIError{}}, + {http.StatusInternalServerError, envelope.APIError{}}, + }, + }, { method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/send", id: "sendSessionMessage", tag: "sessions", summary: "Send a message to a running session's agent", diff --git a/backend/internal/httpd/controllers/dto.go b/backend/internal/httpd/controllers/dto.go index 45ad38bab..981108ef5 100644 --- a/backend/internal/httpd/controllers/dto.go +++ b/backend/internal/httpd/controllers/dto.go @@ -158,6 +158,16 @@ type KillSessionResponse struct { Freed bool `json:"freed,omitempty"` } +// RollbackSessionResponse is the body of POST /api/v1/sessions/{sessionId}/rollback. +// Exactly one of Deleted/Killed is true on a successful rollback; both are +// false when the session was already absent or already terminated (benign). +type RollbackSessionResponse struct { + OK bool `json:"ok"` + SessionID domain.SessionID `json:"sessionId"` + Deleted bool `json:"deleted,omitempty"` + Killed bool `json:"killed,omitempty"` +} + // CleanupSessionsResponse is the body of POST /api/v1/sessions/cleanup. type CleanupSessionsResponse struct { OK bool `json:"ok"` diff --git a/backend/internal/httpd/controllers/sessions.go b/backend/internal/httpd/controllers/sessions.go index e828d9f3e..696b5892f 100644 --- a/backend/internal/httpd/controllers/sessions.go +++ b/backend/internal/httpd/controllers/sessions.go @@ -30,6 +30,7 @@ type SessionService interface { Get(ctx context.Context, id domain.SessionID) (domain.Session, error) Restore(ctx context.Context, id domain.SessionID) (domain.Session, error) Kill(ctx context.Context, id domain.SessionID) (bool, error) + RollbackSpawn(ctx context.Context, id domain.SessionID) (sessionsvc.RollbackOutcome, error) Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, error) Rename(ctx context.Context, id domain.SessionID, displayName string) error Send(ctx context.Context, id domain.SessionID, message string) error @@ -64,6 +65,7 @@ func (c *SessionsController) Register(r chi.Router) { r.Patch("/sessions/{sessionId}", c.rename) r.Post("/sessions/{sessionId}/restore", c.restore) r.Post("/sessions/{sessionId}/kill", c.kill) + r.Post("/sessions/{sessionId}/rollback", c.rollback) r.Post("/sessions/{sessionId}/send", c.send) r.Post("/sessions/{sessionId}/activity", c.activity) r.Get("/orchestrators", c.listOrchestrators) @@ -218,6 +220,25 @@ func (c *SessionsController) kill(w http.ResponseWriter, r *http.Request) { envelope.WriteJSON(w, http.StatusOK, KillSessionResponse{OK: true, SessionID: sessionID(r), Freed: freed}) } +// rollback undoes a partially-completed spawn: if the session row is still in +// seed state (no workspace, no runtime handle yet), the row is deleted +// outright. If anything observable has landed it falls back to Kill so the +// runtime/workspace are torn down. Used by `ao spawn --claim-pr` to undo a +// session whose claim step failed, avoiding the orphan terminated row a +// plain Kill would leave behind. +func (c *SessionsController) rollback(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/{sessionId}/rollback") + return + } + out, err := c.Svc.RollbackSpawn(r.Context(), sessionID(r)) + if err != nil { + envelope.WriteError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, RollbackSessionResponse{OK: true, SessionID: sessionID(r), Deleted: out.Deleted, Killed: out.Killed}) +} + func (c *SessionsController) cleanup(w http.ResponseWriter, r *http.Request) { if c.Svc == nil { apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/cleanup") diff --git a/backend/internal/httpd/controllers/sessions_test.go b/backend/internal/httpd/controllers/sessions_test.go index a85e071b0..303b43e96 100644 --- a/backend/internal/httpd/controllers/sessions_test.go +++ b/backend/internal/httpd/controllers/sessions_test.go @@ -92,6 +92,14 @@ func (f *fakeSessionService) Kill(_ context.Context, id domain.SessionID) (bool, return true, nil } +func (f *fakeSessionService) RollbackSpawn(_ context.Context, id domain.SessionID) (sessionsvc.RollbackOutcome, error) { + if _, ok := f.sessions[id]; ok { + delete(f.sessions, id) + return sessionsvc.RollbackOutcome{Deleted: true}, nil + } + return sessionsvc.RollbackOutcome{}, nil +} + func (f *fakeSessionService) Cleanup(_ context.Context, project domain.ProjectID) ([]domain.SessionID, error) { f.cleanupProjects = append(f.cleanupProjects, project) if f.cleanupResult != nil { diff --git a/backend/internal/integration/lifecycle_sqlite_test.go b/backend/internal/integration/lifecycle_sqlite_test.go index 2c835c9d2..9a0a9960e 100644 --- a/backend/internal/integration/lifecycle_sqlite_test.go +++ b/backend/internal/integration/lifecycle_sqlite_test.go @@ -97,7 +97,7 @@ func newStack(t *testing.T) *stack { prm := prsvc.New(prsvc.Deps{Writer: store, Lifecycle: lcm}) rt := &stubRuntime{} ws := &stubWorkspace{} - mgr := sessionmanager.New(sessionmanager.Deps{Runtime: rt, Agents: stubAgents{}, Workspace: ws, Store: store, Messenger: msg, Lifecycle: lcm}) + mgr := sessionmanager.New(sessionmanager.Deps{Runtime: rt, Agents: stubAgents{}, Workspace: ws, Store: store, Messenger: msg, Lifecycle: lcm, LookPath: func(string) (string, error) { return "/usr/bin/true", nil }}) sm := sessionsvc.New(mgr, store) return &stack{store: store, sm: sm, lcm: lcm, prm: prm, rt: rt, ws: ws, msg: msg} } diff --git a/backend/internal/observe/scm/observer.go b/backend/internal/observe/scm/observer.go index 2c1473b3c..4c425da7a 100644 --- a/backend/internal/observe/scm/observer.go +++ b/backend/internal/observe/scm/observer.go @@ -168,6 +168,16 @@ func (o *Observer) Start(ctx context.Context) <-chan struct{} { func (o *Observer) loop(ctx context.Context, done chan<- struct{}) { defer close(done) + // Run the credential gate once before the first poll so the + // "scm observer disabled: provider credentials unavailable" warning is + // emitted on a fresh daemon even if discoverSubjects has no subjects yet + // (which would otherwise short-circuit Poll before checkCredentials). + // checkCredentials is guarded by credentialsChecked, so this remains + // once-per-process; a transient error here simply defers the check to the + // next Poll, matching existing behavior. + if _, err := o.checkCredentials(ctx); err != nil && !errors.Is(err, context.Canceled) { + o.logger.Error("scm observer: initial credential check failed", "err", err) + } if err := o.Poll(ctx); err != nil && !errors.Is(err, context.Canceled) { o.logger.Error("scm observer: initial poll failed", "err", err) } diff --git a/backend/internal/observe/scm/observer_test.go b/backend/internal/observe/scm/observer_test.go index 448f893b4..b8f9cc0c1 100644 --- a/backend/internal/observe/scm/observer_test.go +++ b/backend/internal/observe/scm/observer_test.go @@ -5,12 +5,14 @@ package scm // review cadence, semantic hashes, and notification behavior stay provider-neutral. import ( + "bytes" "context" "errors" "fmt" "io" "log/slog" "os/exec" + "strings" "sync" "testing" "time" @@ -310,6 +312,89 @@ func TestPoll_RetriesTransientCredentialErrors(t *testing.T) { } } +// syncBuffer is a goroutine-safe wrapper around bytes.Buffer for capturing +// slog output emitted from the observer's background goroutine. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *syncBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + +// TestStart_LogsDisabledWarningWhenNoTokenAndNoSubjects exercises the bug-7 +// regression: on a fresh daemon with no tracked sessions/PRs, discoverSubjects +// returns empty and Poll short-circuits before reaching the credential gate. +// The "scm observer disabled: provider credentials unavailable" warn line must +// still fire exactly once from the observer loop's pre-Poll credential check. +func TestStart_LogsDisabledWarningWhenNoTokenAndNoSubjects(t *testing.T) { + store := &fakeStore{ + sessions: nil, // no sessions → discoverSubjects returns empty + projects: map[string]domain.ProjectRecord{}, + prs: map[domain.SessionID][]domain.PullRequest{}, + checks: map[string][]domain.PullRequestCheck{}, + } + provider := &fakeProvider{ + credentialGate: true, + credentialOK: false, + repoGuards: map[string]ports.SCMGuardResult{}, + observations: map[string]ports.SCMObservation{}, + } + + buf := &syncBuffer{} + logger := slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + obs := New(provider, store, &fakeLifecycle{}, Config{ + Clock: func() time.Time { return time.Unix(1, 0).UTC() }, + Tick: time.Hour, + Logger: logger, + CacheMax: 128, + }) + + ctx, cancel := context.WithCancel(context.Background()) + done := obs.Start(ctx) + // Wait until the loop has emitted the expected warn line, or fail on timeout. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if strings.Contains(buf.String(), "scm observer disabled: provider credentials unavailable") { + break + } + time.Sleep(5 * time.Millisecond) + } + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("observer did not exit after context cancellation") + } + + logged := buf.String() + if !strings.Contains(logged, "scm observer disabled: provider credentials unavailable") { + t.Fatalf("expected disabled-credentials warn line in logs; got:\n%s", logged) + } + if got := strings.Count(logged, "scm observer disabled: provider credentials unavailable"); got != 1 { + t.Fatalf("warn line should fire exactly once, got %d occurrences:\n%s", got, logged) + } + if !obs.credentialsChecked || !obs.disabled { + t.Fatalf("observer state after pre-poll credential check: checked=%v disabled=%v", obs.credentialsChecked, obs.disabled) + } + if provider.credentialChecks != 1 { + t.Fatalf("credential checks = %d, want exactly one pre-poll check", provider.credentialChecks) + } + if provider.repoGuardCalls != 0 || provider.detectCalls != 0 || len(provider.fetchBatches) != 0 { + t.Fatalf("no provider API calls expected when disabled: guards=%d detects=%d batches=%d", + provider.repoGuardCalls, provider.detectCalls, len(provider.fetchBatches)) + } +} + func TestPoll_RepoETag304SkipsDetectPR(t *testing.T) { store := testStoreWithSession() provider := &fakeProvider{repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "v1", NotModified: true}}, observations: map[string]ports.SCMObservation{}} diff --git a/backend/internal/ports/agent.go b/backend/internal/ports/agent.go index 11cfd9569..a00e63429 100644 --- a/backend/internal/ports/agent.go +++ b/backend/internal/ports/agent.go @@ -2,10 +2,18 @@ package ports import ( "context" + "errors" "github.com/aoagents/agent-orchestrator/backend/internal/domain" ) +// ErrAgentBinaryNotFound is returned by agent adapters when neither PATH nor +// any well-known install location holds the agent's binary. The session +// manager surfaces this BEFORE creating the runtime so a missing CLI doesn't +// silently launch into an empty zellij pane that the reaper later mistakes +// for a live session. +var ErrAgentBinaryNotFound = errors.New("agent: binary not found on PATH") + // Agent is the contract every CLI coding agent adapter (claude-code, codex, …) // must satisfy. It supplies the argv and process configuration the Session // Manager needs to launch, restore, and read back a native agent session. diff --git a/backend/internal/ports/outbound.go b/backend/internal/ports/outbound.go index f2c006b16..720cb1682 100644 --- a/backend/internal/ports/outbound.go +++ b/backend/internal/ports/outbound.go @@ -108,6 +108,18 @@ type Workspace interface { Restore(ctx context.Context, cfg WorkspaceConfig) (WorkspaceInfo, error) } +// Workspace-level sentinels surfaced through Create/Restore so the HTTP layer +// can map them to typed errors rather than collapsing every adapter failure +// into an opaque 500. Adapters wrap these via fmt.Errorf("...: %w", sentinel). +var ( + // ErrWorkspaceBranchCheckedOutElsewhere reports the requested branch is + // already checked out in another worktree of the same repo. + ErrWorkspaceBranchCheckedOutElsewhere = errors.New("workspace: branch is already checked out in another worktree") + // ErrWorkspaceBranchNotFetched reports the requested branch exists nowhere + // reachable (no local head, no remote-tracking branch, no tag). + ErrWorkspaceBranchNotFetched = errors.New("workspace: branch is not fetched") +) + // WorkspaceConfig is the spec for creating or restoring a session's workspace. type WorkspaceConfig struct { ProjectID domain.ProjectID diff --git a/backend/internal/service/session/service.go b/backend/internal/service/session/service.go index b5667158e..2f5ae450b 100644 --- a/backend/internal/service/session/service.go +++ b/backend/internal/service/session/service.go @@ -41,6 +41,15 @@ type commander interface { Kill(ctx context.Context, id domain.SessionID) (bool, error) Send(ctx context.Context, id domain.SessionID, message string) error Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, error) + RollbackSpawn(ctx context.Context, id domain.SessionID) (deleted, killed bool, err error) +} + +// RollbackOutcome reports what happened in a rollback: either the seed row was +// deleted, or the partially-spawned session was killed (runtime+workspace torn +// down, row marked terminated). +type RollbackOutcome struct { + Deleted bool `json:"deleted"` + Killed bool `json:"killed"` } type scmProvider interface { @@ -92,6 +101,9 @@ func NewWithDeps(d Deps) *Service { // Spawn creates a session and returns the API-facing read model. func (s *Service) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Session, error) { + if err := s.requireProject(ctx, cfg.ProjectID); err != nil { + return domain.Session{}, err + } rec, err := s.manager.Spawn(ctx, cfg) if err != nil { return domain.Session{}, toAPIError(err) @@ -99,11 +111,34 @@ func (s *Service) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess return s.toSession(ctx, rec) } +// requireProject verifies the project is registered before any spawn write +// touches the session store, so an unknown projectId surfaces as a typed 404 +// rather than an opaque 500 with an orphan terminated row left behind. +func (s *Service) requireProject(ctx context.Context, id domain.ProjectID) error { + if id == "" { + return apierr.Invalid("PROJECT_ID_REQUIRED", "projectId is required", nil) + } + if s.store == nil { + return nil + } + _, ok, err := s.store.GetProject(ctx, string(id)) + if err != nil { + return fmt.Errorf("get project %s: %w", id, err) + } + if !ok { + return apierr.NotFound("PROJECT_NOT_FOUND", "Unknown project — register it with `ao project add`") + } + return nil +} + // SpawnOrchestrator spawns an orchestrator session for a project. When clean is // true it first tears down any active orchestrator(s) for that project so the new // one is the only live coordinator — a business rule that belongs here, not in the // HTTP controller. func (s *Service) SpawnOrchestrator(ctx context.Context, projectID domain.ProjectID, clean bool) (domain.Session, error) { + if err := s.requireProject(ctx, projectID); err != nil { + return domain.Session{}, err + } if clean { active := true existing, err := s.List(ctx, ListFilter{ProjectID: projectID, Active: &active, OrchestratorOnly: true}) @@ -134,6 +169,18 @@ func (s *Service) Kill(ctx context.Context, id domain.SessionID) (bool, error) { return freed, toAPIError(err) } +// RollbackSpawn deletes a seed-state session row, or falls back to a Kill if +// the session has spawn output. Used by the CLI to undo a `spawn --claim-pr` +// when the claim step fails, avoiding the orphan terminated row that a plain +// Kill would leave behind. +func (s *Service) RollbackSpawn(ctx context.Context, id domain.SessionID) (RollbackOutcome, error) { + deleted, killed, err := s.manager.RollbackSpawn(ctx, id) + if err != nil { + return RollbackOutcome{}, toAPIError(err) + } + return RollbackOutcome{Deleted: deleted, Killed: killed}, nil +} + // Send delegates agent messaging to the internal manager. func (s *Service) Send(ctx context.Context, id domain.SessionID, message string) error { return toAPIError(s.manager.Send(ctx, id, message)) @@ -237,6 +284,12 @@ func toAPIError(err error) error { return apierr.Conflict("SESSION_INCOMPLETE_HANDLE", "Session is missing runtime or workspace handles", nil) case errors.Is(err, sessionmanager.ErrProjectNotResolvable): return apierr.Invalid("PROJECT_NOT_RESOLVABLE", "Project is not registered or has no repo — register it with `ao project add`", nil) + case errors.Is(err, ports.ErrWorkspaceBranchCheckedOutElsewhere): + return apierr.Conflict("BRANCH_CHECKED_OUT_ELSEWHERE", err.Error(), nil) + case errors.Is(err, ports.ErrWorkspaceBranchNotFetched): + return apierr.Invalid("BRANCH_NOT_FETCHED", err.Error(), nil) + case errors.Is(err, ports.ErrAgentBinaryNotFound): + return apierr.Invalid("AGENT_BINARY_NOT_FOUND", err.Error(), nil) default: return err } diff --git a/backend/internal/service/session/service_test.go b/backend/internal/service/session/service_test.go index 71ed69c34..d3a7bba22 100644 --- a/backend/internal/service/session/service_test.go +++ b/backend/internal/service/session/service_test.go @@ -147,9 +147,13 @@ func (f *fakeCommander) Send(context.Context, domain.SessionID, string) error { func (f *fakeCommander) Cleanup(context.Context, domain.ProjectID) ([]domain.SessionID, error) { return nil, nil } +func (f *fakeCommander) RollbackSpawn(context.Context, domain.SessionID) (bool, bool, error) { + return false, false, nil +} func TestSpawnOrchestratorCleanKillsActiveOrchestratorsBeforeSpawn(t *testing.T) { st := newFakeStore() + st.projects["mer"] = domain.ProjectRecord{ID: "mer"} // Two active orchestrators plus an unrelated worker and a terminated // orchestrator that must be left alone. st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindOrchestrator} @@ -172,8 +176,70 @@ func TestSpawnOrchestratorCleanKillsActiveOrchestratorsBeforeSpawn(t *testing.T) } } +// TestSpawnUnknownProjectReturns404 covers Bug 1: an HTTP spawn for an +// unregistered projectId must surface PROJECT_NOT_FOUND (apierr.NotFound) +// BEFORE any session row is created, so no orphan terminated row is left +// behind under `--include-terminated`. +func TestSpawnUnknownProjectReturns404(t *testing.T) { + st := newFakeStore() + fc := &fakeCommander{} + svc := &Service{manager: fc, store: st} + + _, err := svc.Spawn(context.Background(), ports.SpawnConfig{ProjectID: "ghost", Kind: domain.KindWorker}) + var e *apierr.Error + if !errors.As(err, &e) || e.Kind != apierr.KindNotFound || e.Code != "PROJECT_NOT_FOUND" { + t.Fatalf("err = %v, want apierr.NotFound PROJECT_NOT_FOUND", err) + } + if fc.spawned { + t.Fatal("manager.Spawn must NOT be invoked for an unknown project") + } +} + +// TestSpawnOrchestratorUnknownProjectReturns404 is the orchestrator-side guard +// for Bug 1: same pre-validation, same typed envelope. +func TestSpawnOrchestratorUnknownProjectReturns404(t *testing.T) { + st := newFakeStore() + fc := &fakeCommander{} + svc := &Service{manager: fc, store: st} + + _, err := svc.SpawnOrchestrator(context.Background(), "ghost", false) + var e *apierr.Error + if !errors.As(err, &e) || e.Kind != apierr.KindNotFound || e.Code != "PROJECT_NOT_FOUND" { + t.Fatalf("err = %v, want apierr.NotFound PROJECT_NOT_FOUND", err) + } + if fc.spawned { + t.Fatal("manager.Spawn must NOT be invoked for an unknown project") + } +} + +// TestToAPIErrorMapsWorkspaceBranchSentinels covers Bug 3: the workspace +// adapter's typed branch errors map to typed envelope errors instead of +// collapsing to a 500. +func TestToAPIErrorMapsWorkspaceBranchSentinels(t *testing.T) { + cases := []struct { + name string + err error + wantKind apierr.Kind + wantCode string + }{ + {"checked out elsewhere", fmt.Errorf("spawn mer-1: workspace: %w: \"x\" is checked out at \"/tmp\"", ports.ErrWorkspaceBranchCheckedOutElsewhere), apierr.KindConflict, "BRANCH_CHECKED_OUT_ELSEWHERE"}, + {"not fetched", fmt.Errorf("spawn mer-1: workspace: %w: \"x\" has no local head", ports.ErrWorkspaceBranchNotFetched), apierr.KindInvalid, "BRANCH_NOT_FETCHED"}, + {"agent binary not found", fmt.Errorf("spawn mer-1: %w", ports.ErrAgentBinaryNotFound), apierr.KindInvalid, "AGENT_BINARY_NOT_FOUND"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + mapped := toAPIError(tc.err) + var e *apierr.Error + if !errors.As(mapped, &e) || e.Kind != tc.wantKind || e.Code != tc.wantCode { + t.Fatalf("mapped = %v, want %s %s", mapped, tc.wantCode, e) + } + }) + } +} + func TestSpawnOrchestratorNoCleanSkipsKills(t *testing.T) { st := newFakeStore() + st.projects["mer"] = domain.ProjectRecord{ID: "mer"} st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindOrchestrator} fc := &fakeCommander{} diff --git a/backend/internal/session_manager/manager.go b/backend/internal/session_manager/manager.go index e4abaa8ed..66a5272d2 100644 --- a/backend/internal/session_manager/manager.go +++ b/backend/internal/session_manager/manager.go @@ -6,6 +6,7 @@ import ( "context" "errors" "fmt" + "os/exec" "time" "github.com/aoagents/agent-orchestrator/backend/internal/domain" @@ -49,6 +50,12 @@ type Store interface { GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) ListSessions(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) ListAllSessions(ctx context.Context) ([]domain.SessionRecord, error) + // DeleteSession removes a session row only if it is still in seed state + // (no workspace, runtime handle, agent session id, or prompt; not + // terminated). Returns deleted=true when removal happened; deleted=false + // when the row had already progressed past seed state — preserving the + // no-resurrection guarantee for live sessions. + DeleteSession(ctx context.Context, id domain.SessionID) (bool, error) } // Manager coordinates internal session spawn, restore, kill, and cleanup over @@ -62,6 +69,10 @@ type Manager struct { lcm lifecycleRecorder dataDir string clock func() time.Time + // lookPath is exec.LookPath in production; tests substitute a stub so + // they don't need real binaries on PATH. Returns ports.ErrAgentBinaryNotFound + // when the binary is missing so the sentinel propagates through toAPIError. + lookPath func(string) (string, error) } // Deps are the collaborators a Session Manager needs; New wires them together. @@ -76,6 +87,10 @@ type Deps struct { // commands can open the same store. DataDir string Clock func() time.Time + // LookPath overrides exec.LookPath for the pre-launch agent-binary check. + // Production wiring leaves this nil and the manager defaults to + // exec.LookPath; tests inject a stub so they need not seed real binaries. + LookPath func(string) (string, error) } // New builds a Session Manager from its dependencies, defaulting the clock to @@ -90,10 +105,14 @@ func New(d Deps) *Manager { lcm: d.Lifecycle, dataDir: d.DataDir, clock: d.Clock, + lookPath: d.LookPath, } if m.clock == nil { m.clock = time.Now } + if m.lookPath == nil { + m.lookPath = exec.LookPath + } return m } @@ -147,6 +166,15 @@ func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess m.markSpawnFailedTerminated(ctx, id) return domain.SessionRecord{}, fmt.Errorf("spawn %s: launch command: %w", id, err) } + // Pre-flight: confirm argv[0] actually exists on PATH (or as an absolute + // path the adapter returned) BEFORE handing the launch to the runtime. + // Zellij happily creates a session+pane around a missing command, so an + // unresolved binary would leak through as a "live" session that never ran. + if err := m.validateAgentBinary(argv); err != nil { + _ = m.workspace.Destroy(ctx, ws) + m.markSpawnFailedTerminated(ctx, id) + return domain.SessionRecord{}, fmt.Errorf("spawn %s: %w", id, err) + } handle, err := m.runtime.Create(ctx, ports.RuntimeConfig{ SessionID: id, WorkspacePath: ws.Path, @@ -170,11 +198,43 @@ func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess } // markSpawnFailedTerminated best-effort parks an orphaned spawn as terminated. -// The store has no delete; a phantom half-spawned row is worse than a terminal one. +// A phantom half-spawned row is worse than a terminal one; we only delete the +// row when nothing observable has landed yet (seed state) via rollbackSpawn. func (m *Manager) markSpawnFailedTerminated(ctx context.Context, id domain.SessionID) { _ = m.lcm.MarkTerminated(ctx, id) } +// rollbackSpawn deletes a session row when it is still in seed state — used +// when an out-of-band step that happens AFTER `Spawn` returns (e.g. PR claim +// over HTTP) has failed and the caller wants the partially-spawned session +// gone without leaving a terminated orphan visible under `--include-terminated`. +// +// If the row has progressed past seed state (workspace exists, runtime created, +// etc.), DeleteSession is a no-op and rollbackSpawn falls back to a Kill so the +// runtime/workspace are torn down. Returns (deleted, killed): +// - deleted=true: the row was a seed row and has been removed +// - killed=true: the row had spawn output and was torn down + terminated +// - both false: the row was already terminated or absent — benign no-op +func (m *Manager) rollbackSpawn(ctx context.Context, id domain.SessionID) (deleted, killed bool, err error) { + deleted, err = m.store.DeleteSession(ctx, id) + if err != nil { + return false, false, fmt.Errorf("rollback %s: %w", id, err) + } + if deleted { + return true, false, nil + } + killed, err = m.Kill(ctx, id) + if err != nil { + return false, false, err + } + return false, killed, nil +} + +// RollbackSpawn is the public surface of rollbackSpawn for service-layer callers. +func (m *Manager) RollbackSpawn(ctx context.Context, id domain.SessionID) (deleted, killed bool, err error) { + return m.rollbackSpawn(ctx, id) +} + // Kill records terminal intent with the LCM, then tears down the runtime and // workspace. A workspace teardown refused by the worktree-remove safety // (uncommitted work) surfaces as an error with freed=false and is never forced. @@ -218,6 +278,13 @@ func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Sess return domain.SessionRecord{}, fmt.Errorf("restore %s: %w", id, ErrNotRestorable) } meta := rec.Metadata + // Mirror Kill's incomplete-handle guard: a session whose spawn failed before + // the workspace landed has neither WorkspacePath nor Branch, and there is + // nothing meaningful to restore from. Surface this as a typed 409 instead of + // letting workspace.Restore fail with an opaque wrapped error. + if meta.WorkspacePath == "" || meta.Branch == "" { + return domain.SessionRecord{}, fmt.Errorf("restore %s: %w", id, ErrIncompleteHandle) + } if meta.AgentSessionID == "" && meta.Prompt == "" { return domain.SessionRecord{}, fmt.Errorf("restore %s: nothing to resume from", id) } @@ -462,6 +529,22 @@ func restoreArgv(ctx context.Context, agent ports.Agent, id domain.SessionID, wo return argv, nil } +// validateAgentBinary checks that argv[0] resolves via the manager's +// lookPath (exec.LookPath in prod) before any runtime work happens. Adapters +// that can't resolve their binary now return ports.ErrAgentBinaryNotFound from +// GetLaunchCommand directly; this guard is a defense-in-depth for adapters +// that return an argv[0] like "claude" without verifying. +func (m *Manager) validateAgentBinary(argv []string) error { + if len(argv) == 0 { + return fmt.Errorf("agent: empty launch argv: %w", ports.ErrAgentBinaryNotFound) + } + bin := argv[0] + if _, err := m.lookPath(bin); err != nil { + return fmt.Errorf("agent binary %q: %w", bin, ports.ErrAgentBinaryNotFound) + } + return nil +} + func runtimeHandle(meta domain.SessionMetadata) ports.RuntimeHandle { return ports.RuntimeHandle{ID: meta.RuntimeHandleID} } diff --git a/backend/internal/session_manager/manager_test.go b/backend/internal/session_manager/manager_test.go index be2ee60e1..be1252748 100644 --- a/backend/internal/session_manager/manager_test.go +++ b/backend/internal/session_manager/manager_test.go @@ -53,6 +53,18 @@ func (f *fakeStore) ListAllSessions(context.Context) ([]domain.SessionRecord, er } return out, nil } +func (f *fakeStore) DeleteSession(_ context.Context, id domain.SessionID) (bool, error) { + rec, ok := f.sessions[id] + if !ok { + return false, nil + } + // Mirror the sqlite gate: only delete rows still in seed state. + if rec.IsTerminated || rec.Metadata.WorkspacePath != "" || rec.Metadata.RuntimeHandleID != "" || rec.Metadata.AgentSessionID != "" || rec.Metadata.Prompt != "" { + return false, nil + } + delete(f.sessions, id) + return true, nil +} func (f *fakeStore) GetDisplayPRFactsForSession(_ context.Context, id domain.SessionID) (domain.PRFacts, bool, error) { if pr := f.pr[id]; pr.URL != "" { return pr, true, nil @@ -150,7 +162,10 @@ func newManager() (*Manager, *fakeStore, *fakeRuntime, *fakeWorkspace) { st := newFakeStore() rt := &fakeRuntime{} ws := &fakeWorkspace{} - m := New(Deps{Runtime: rt, Agents: fakeAgents{}, Workspace: ws, Store: st, Messenger: &fakeMessenger{}, Lifecycle: &fakeLCM{store: st}}) + // Stub lookPath so the pre-launch agent-binary check passes; the fakeAgent + // returns argv ["launch"] which is not a real binary on PATH. + lookPath := func(string) (string, error) { return "/bin/true", nil } + m := New(Deps{Runtime: rt, Agents: fakeAgents{}, Workspace: ws, Store: st, Messenger: &fakeMessenger{}, Lifecycle: &fakeLCM{store: st}, LookPath: lookPath}) return m, st, rt, ws } func seedTerminal(st *fakeStore, id domain.SessionID, meta domain.SessionMetadata) { @@ -315,6 +330,99 @@ func TestSpawnOrchestrator_UsesCoordinatorPrompt(t *testing.T) { } } +// TestRestore_RefusesIncompleteHandle covers Bug 2: a terminated row whose +// spawn failed before the workspace landed (no WorkspacePath, no Branch) must +// fail Restore with ErrIncompleteHandle — the same typed sentinel Kill returns +// for the same shape — so the HTTP layer surfaces a typed 409 instead of an +// opaque 500. +func TestRestore_RefusesIncompleteHandle(t *testing.T) { + m, st, _, _ := newManager() + // Seed a terminated row with no workspace and no branch (the post-failure + // shape of a Spawn that died before workspace.Create succeeded). + st.sessions["mer-1"] = domain.SessionRecord{ + ID: "mer-1", + ProjectID: "mer", + IsTerminated: true, + Metadata: domain.SessionMetadata{Prompt: "do it"}, + } + if _, err := m.Restore(ctx, "mer-1"); !errors.Is(err, ErrIncompleteHandle) { + t.Fatalf("want ErrIncompleteHandle, got %v", err) + } +} + +// TestRollbackSpawn_DeletesSeedRow covers Bug 4: a session row in seed state +// (no workspace, no runtime, no agent session id, not terminated) is deleted +// outright by RollbackSpawn so the user never sees an orphan terminated row. +func TestRollbackSpawn_DeletesSeedRow(t *testing.T) { + m, st, _, _ := newManager() + // Seed row matches what CreateSession produces — no Metadata at all. + st.sessions["mer-1"] = domain.SessionRecord{ + ID: "mer-1", + ProjectID: "mer", + Activity: domain.Activity{State: domain.ActivityIdle}, + } + deleted, killed, err := m.RollbackSpawn(ctx, "mer-1") + if err != nil { + t.Fatalf("rollback err = %v", err) + } + if !deleted || killed { + t.Fatalf("deleted=%v killed=%v, want deleted=true killed=false", deleted, killed) + } + if _, present := st.sessions["mer-1"]; present { + t.Fatal("seed row must be removed from the store, not left as terminated") + } +} + +// TestRollbackSpawn_FallsBackToKillForLiveRow asserts the no-resurrection +// guarantee from Bug 4's RCA: once a row has observable spawn output (workspace +// + runtime handle), DeleteSession is a no-op and rollback falls back to Kill +// so the runtime + workspace are torn down rather than abandoned. +func TestRollbackSpawn_FallsBackToKillForLiveRow(t *testing.T) { + m, st, rt, ws := newManager() + st.sessions["mer-1"] = mkLive("mer-1") + deleted, killed, err := m.RollbackSpawn(ctx, "mer-1") + if err != nil { + t.Fatalf("rollback err = %v", err) + } + if deleted || !killed { + t.Fatalf("deleted=%v killed=%v, want deleted=false killed=true", deleted, killed) + } + if rt.destroyed != 1 || ws.destroyed != 1 { + t.Fatalf("kill teardown not invoked: rt=%d ws=%d", rt.destroyed, ws.destroyed) + } + if !st.sessions["mer-1"].IsTerminated { + t.Fatal("live row should be marked terminated after kill-fallback") + } +} + +// TestSpawn_RejectsMissingAgentBinary covers Bug 6: when the agent adapter +// returns an argv whose binary is not on PATH, Manager.Spawn must abort BEFORE +// runtime.Create rather than launching into an empty zellij pane that the +// reaper later mistakes for a live session. +func TestSpawn_RejectsMissingAgentBinary(t *testing.T) { + st := newFakeStore() + rt := &fakeRuntime{} + ws := &fakeWorkspace{} + notFound := func(name string) (string, error) { + return "", fmt.Errorf("exec: %q: not found", name) + } + m := New(Deps{Runtime: rt, Agents: fakeAgents{}, Workspace: ws, Store: st, Messenger: &fakeMessenger{}, Lifecycle: &fakeLCM{store: st}, LookPath: notFound}) + + _, err := m.Spawn(ctx, ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker}) + if !errors.Is(err, ports.ErrAgentBinaryNotFound) { + t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err) + } + if rt.created != 0 { + t.Fatal("runtime.Create must NOT run when the agent binary is missing") + } + if ws.destroyed != 1 { + t.Fatal("workspace must be torn down when the pre-launch binary check fails") + } + if !st.sessions["mer-1"].IsTerminated { + t.Fatal("the orphan row should be marked terminated after the failed spawn") + } +} + func TestSpawn_KeepsExplicitBranch(t *testing.T) { m, st, _, _ := newManager() s, err := m.Spawn(ctx, ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, Branch: "feature/x"}) diff --git a/backend/internal/storage/sqlite/gen/models.go b/backend/internal/storage/sqlite/gen/models.go index 4d4b706ff..824825a10 100644 --- a/backend/internal/storage/sqlite/gen/models.go +++ b/backend/internal/storage/sqlite/gen/models.go @@ -60,6 +60,7 @@ type PR struct { ObservedAt sql.NullTime CIObservedAt sql.NullTime ReviewObservedAt sql.NullTime + LastNudgeSignature string } type PRCheck struct { diff --git a/backend/internal/storage/sqlite/gen/pr.sql.go b/backend/internal/storage/sqlite/gen/pr.sql.go index 754245341..b5bdaad44 100644 --- a/backend/internal/storage/sqlite/gen/pr.sql.go +++ b/backend/internal/storage/sqlite/gen/pr.sql.go @@ -99,16 +99,7 @@ func (q *Queries) GetDisplayPRFactsBySession(ctx context.Context, sessionID doma } const getPR = `-- name: GetPR :one -SELECT - url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, - provider, host, repo, source_branch, target_branch, head_sha, title, - additions, deletions, changed_files, author, base_sha, merge_commit_sha, - is_draft, is_merged, is_closed, - provider_state, provider_mergeable, provider_merge_state_status, html_url, - created_at_provider, updated_at_provider, merged_at_provider, closed_at_provider, - metadata_hash, ci_hash, review_hash, observed_at, ci_observed_at, review_observed_at -FROM pr -WHERE url = ? +SELECT url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, provider, host, repo, source_branch, target_branch, head_sha, title, additions, deletions, changed_files, author, base_sha, merge_commit_sha, is_draft, is_merged, is_closed, provider_state, provider_mergeable, provider_merge_state_status, html_url, created_at_provider, updated_at_provider, merged_at_provider, closed_at_provider, metadata_hash, ci_hash, review_hash, observed_at, ci_observed_at, review_observed_at, last_nudge_signature FROM pr WHERE url = ? ` func (q *Queries) GetPR(ctx context.Context, url string) (PR, error) { @@ -153,6 +144,7 @@ func (q *Queries) GetPR(ctx context.Context, url string) (PR, error) { &i.ObservedAt, &i.CIObservedAt, &i.ReviewObservedAt, + &i.LastNudgeSignature, ) return i, err } @@ -178,16 +170,19 @@ func (q *Queries) GetPRClaimAndOwner(ctx context.Context, url string) (GetPRClai return i, err } +const getPRLastNudgeSignature = `-- name: GetPRLastNudgeSignature :one +SELECT last_nudge_signature FROM pr WHERE url = ? +` + +func (q *Queries) GetPRLastNudgeSignature(ctx context.Context, url string) (string, error) { + row := q.db.QueryRowContext(ctx, getPRLastNudgeSignature, url) + var last_nudge_signature string + err := row.Scan(&last_nudge_signature) + return last_nudge_signature, err +} + const listPRsBySession = `-- name: ListPRsBySession :many -SELECT - url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, - provider, host, repo, source_branch, target_branch, head_sha, title, - additions, deletions, changed_files, author, base_sha, merge_commit_sha, - is_draft, is_merged, is_closed, - provider_state, provider_mergeable, provider_merge_state_status, html_url, - created_at_provider, updated_at_provider, merged_at_provider, closed_at_provider, - metadata_hash, ci_hash, review_hash, observed_at, ci_observed_at, review_observed_at -FROM pr +SELECT url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, provider, host, repo, source_branch, target_branch, head_sha, title, additions, deletions, changed_files, author, base_sha, merge_commit_sha, is_draft, is_merged, is_closed, provider_state, provider_mergeable, provider_merge_state_status, html_url, created_at_provider, updated_at_provider, merged_at_provider, closed_at_provider, metadata_hash, ci_hash, review_hash, observed_at, ci_observed_at, review_observed_at, last_nudge_signature FROM pr WHERE session_id = ? ORDER BY updated_at DESC ` @@ -240,6 +235,7 @@ func (q *Queries) ListPRsBySession(ctx context.Context, sessionID domain.Session &i.ObservedAt, &i.CIObservedAt, &i.ReviewObservedAt, + &i.LastNudgeSignature, ); err != nil { return nil, err } @@ -254,6 +250,20 @@ func (q *Queries) ListPRsBySession(ctx context.Context, sessionID domain.Session return items, nil } +const updatePRLastNudgeSignature = `-- name: UpdatePRLastNudgeSignature :exec +UPDATE pr SET last_nudge_signature = ? WHERE url = ? +` + +type UpdatePRLastNudgeSignatureParams struct { + LastNudgeSignature string + URL string +} + +func (q *Queries) UpdatePRLastNudgeSignature(ctx context.Context, arg UpdatePRLastNudgeSignatureParams) error { + _, err := q.db.ExecContext(ctx, updatePRLastNudgeSignature, arg.LastNudgeSignature, arg.URL) + return err +} + const upsertLegacyPR = `-- name: UpsertLegacyPR :exec INSERT INTO pr ( url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, @@ -437,28 +447,3 @@ func (q *Queries) UpsertPR(ctx context.Context, arg UpsertPRParams) error { ) return err } - -const getPRLastNudgeSignature = `-- name: GetPRLastNudgeSignature :one -SELECT last_nudge_signature FROM pr WHERE url = ? -` - -func (q *Queries) GetPRLastNudgeSignature(ctx context.Context, url string) (string, error) { - row := q.db.QueryRowContext(ctx, getPRLastNudgeSignature, url) - var sig string - err := row.Scan(&sig) - return sig, err -} - -const updatePRLastNudgeSignature = `-- name: UpdatePRLastNudgeSignature :exec -UPDATE pr SET last_nudge_signature = ? WHERE url = ? -` - -type UpdatePRLastNudgeSignatureParams struct { - LastNudgeSignature string - URL string -} - -func (q *Queries) UpdatePRLastNudgeSignature(ctx context.Context, arg UpdatePRLastNudgeSignatureParams) error { - _, err := q.db.ExecContext(ctx, updatePRLastNudgeSignature, arg.LastNudgeSignature, arg.URL) - return err -} diff --git a/backend/internal/storage/sqlite/gen/pr_review_threads.sql.go b/backend/internal/storage/sqlite/gen/pr_review_threads.sql.go index 0f2eda253..538c0e645 100644 --- a/backend/internal/storage/sqlite/gen/pr_review_threads.sql.go +++ b/backend/internal/storage/sqlite/gen/pr_review_threads.sql.go @@ -10,6 +10,20 @@ import ( "time" ) +const deletePRReviewThread = `-- name: DeletePRReviewThread :exec +DELETE FROM pr_review_threads WHERE pr_url = ? AND thread_id = ? +` + +type DeletePRReviewThreadParams struct { + PRURL string + ThreadID string +} + +func (q *Queries) DeletePRReviewThread(ctx context.Context, arg DeletePRReviewThreadParams) error { + _, err := q.db.ExecContext(ctx, deletePRReviewThread, arg.PRURL, arg.ThreadID) + return err +} + const deletePRReviewThreads = `-- name: DeletePRReviewThreads :exec DELETE FROM pr_review_threads WHERE pr_url = ? ` diff --git a/backend/internal/storage/sqlite/gen/sessions.sql.go b/backend/internal/storage/sqlite/gen/sessions.sql.go index db18bfe14..830eba7ea 100644 --- a/backend/internal/storage/sqlite/gen/sessions.sql.go +++ b/backend/internal/storage/sqlite/gen/sessions.sql.go @@ -221,6 +221,30 @@ func (q *Queries) RenameSession(ctx context.Context, arg RenameSessionParams) (i return result.RowsAffected() } +const sessionIsSeed = `-- name: SessionIsSeed :one +SELECT EXISTS( + SELECT 1 FROM sessions + WHERE id = ? + AND is_terminated = 0 + AND workspace_path = '' + AND runtime_handle_id = '' + AND agent_session_id = '' + AND prompt = '' +) AS is_seed +` + +// SessionIsSeed reports whether the session id matches a row still in seed +// state (see DeleteSeedSession for the conditions). Callers probe with this +// before touching change_log so that DeleteSession is a true no-op for live +// sessions instead of silently destroying their CDC events. Returns 0 when +// the row does not exist OR has progressed past seed state. +func (q *Queries) SessionIsSeed(ctx context.Context, id domain.SessionID) (bool, error) { + row := q.db.QueryRowContext(ctx, sessionIsSeed, id) + var is_seed bool + err := row.Scan(&is_seed) + return is_seed, err +} + const updateSession = `-- name: UpdateSession :exec UPDATE sessions SET issue_id = ?, kind = ?, harness = ?, display_name = ?, diff --git a/backend/internal/storage/sqlite/queries/changelog.sql b/backend/internal/storage/sqlite/queries/changelog.sql index 9d41a3e3d..d41e518d9 100644 --- a/backend/internal/storage/sqlite/queries/changelog.sql +++ b/backend/internal/storage/sqlite/queries/changelog.sql @@ -5,3 +5,13 @@ FROM change_log WHERE seq > ? ORDER BY seq LIMIT ?; -- name: MaxChangeLogSeq :one SELECT CAST(COALESCE(MAX(seq), 0) AS INTEGER) AS seq FROM change_log; + +-- NOTE: `DELETE FROM change_log WHERE session_id = ?` is intentionally NOT +-- a sqlc query. sqlc 1.31's SQLite parser strips the `?` placeholder and +-- emits a *domain.SessionID pointer parameter whenever a nullable column +-- (change_log.session_id is nullable for project-level events) sits on the +-- LHS of `=` in a top-level DELETE. None of the obvious SQL workarounds +-- (sqlc.arg, IFNULL, rowid subquery, second predicate) defeated the +-- heuristic. The store runs that DELETE directly via tx.ExecContext inside +-- Store.DeleteSession to keep it part of the same transaction as the seed +-- probe + session delete. diff --git a/backend/internal/storage/sqlite/queries/pr.sql b/backend/internal/storage/sqlite/queries/pr.sql index 5501e4d38..b4d745cd0 100644 --- a/backend/internal/storage/sqlite/queries/pr.sql +++ b/backend/internal/storage/sqlite/queries/pr.sql @@ -65,27 +65,10 @@ ON CONFLICT (url) DO UPDATE SET is_closed = excluded.is_closed; -- name: GetPR :one -SELECT - url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, - provider, host, repo, source_branch, target_branch, head_sha, title, - additions, deletions, changed_files, author, base_sha, merge_commit_sha, - is_draft, is_merged, is_closed, - provider_state, provider_mergeable, provider_merge_state_status, html_url, - created_at_provider, updated_at_provider, merged_at_provider, closed_at_provider, - metadata_hash, ci_hash, review_hash, observed_at, ci_observed_at, review_observed_at -FROM pr -WHERE url = ?; +SELECT * FROM pr WHERE url = ?; -- name: ListPRsBySession :many -SELECT - url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, - provider, host, repo, source_branch, target_branch, head_sha, title, - additions, deletions, changed_files, author, base_sha, merge_commit_sha, - is_draft, is_merged, is_closed, - provider_state, provider_mergeable, provider_merge_state_status, html_url, - created_at_provider, updated_at_provider, merged_at_provider, closed_at_provider, - metadata_hash, ci_hash, review_hash, observed_at, ci_observed_at, review_observed_at -FROM pr +SELECT * FROM pr WHERE session_id = ? ORDER BY updated_at DESC; diff --git a/backend/internal/storage/sqlite/queries/pr_review_threads.sql b/backend/internal/storage/sqlite/queries/pr_review_threads.sql index 19f544818..1d9cda4fe 100644 --- a/backend/internal/storage/sqlite/queries/pr_review_threads.sql +++ b/backend/internal/storage/sqlite/queries/pr_review_threads.sql @@ -13,6 +13,9 @@ ON CONFLICT (pr_url, thread_id) DO UPDATE SET -- name: DeletePRReviewThreads :exec DELETE FROM pr_review_threads WHERE pr_url = ?; +-- name: DeletePRReviewThread :exec +DELETE FROM pr_review_threads WHERE pr_url = ? AND thread_id = ?; + -- name: ListPRReviewThreads :many SELECT pr_url, thread_id, path, line, resolved, is_bot, semantic_hash, updated_at FROM pr_review_threads WHERE pr_url = ? ORDER BY updated_at, thread_id; diff --git a/backend/internal/storage/sqlite/queries/sessions.sql b/backend/internal/storage/sqlite/queries/sessions.sql index 9c3e1da78..7b4329e4d 100644 --- a/backend/internal/storage/sqlite/queries/sessions.sql +++ b/backend/internal/storage/sqlite/queries/sessions.sql @@ -38,3 +38,28 @@ FROM sessions ORDER BY project_id, num; -- name: RenameSession :execrows UPDATE sessions SET display_name = ?, updated_at = ? WHERE id = ?; + +-- name: SessionIsSeed :one +-- SessionIsSeed reports whether the session id matches a row still in seed +-- state (see DeleteSeedSession for the conditions). Callers probe with this +-- before touching change_log so that DeleteSession is a true no-op for live +-- sessions instead of silently destroying their CDC events. Returns 0 when +-- the row does not exist OR has progressed past seed state. +SELECT EXISTS( + SELECT 1 FROM sessions + WHERE id = ? + AND is_terminated = 0 + AND workspace_path = '' + AND runtime_handle_id = '' + AND agent_session_id = '' + AND prompt = '' +) AS is_seed; + +-- NOTE: the `DELETE FROM sessions WHERE id = ? AND ` +-- statement is intentionally NOT a sqlc query — same sqlc 1.31 SQLite-parser +-- bug as documented in queries/changelog.sql: trailing string literals (and +-- placeholders) on the RHS of `=` in a DELETE get silently stripped, so the +-- generated SQL ends up mid-clause and the row count is meaningless. The +-- store runs that DELETE directly via tx.ExecContext inside +-- Store.DeleteSession, inside the same transaction as the SessionIsSeed +-- probe and the raw change_log cleanup. diff --git a/backend/internal/storage/sqlite/store/pr_cdc_test.go b/backend/internal/storage/sqlite/store/pr_cdc_test.go index 2f2b06d95..ff382c6d3 100644 --- a/backend/internal/storage/sqlite/store/pr_cdc_test.go +++ b/backend/internal/storage/sqlite/store/pr_cdc_test.go @@ -116,6 +116,130 @@ func TestPRReviewThreadsCDC_EmitsOnInsertAndResolvedTransition(t *testing.T) { } } +// Regression for the bug where pr_review_thread_resolved never fired on the +// common Replace path. Real-world polls take ReviewWriteReplace whenever the +// upstream listing is not paginated (provider observer sets Partial=false). +// The previous implementation did DELETE-all + UPSERT inside the tx, so every +// upsert hit the INSERT trigger and the UPDATE trigger that emits +// pr_review_thread_resolved never saw the resolved flip. The fix is a set-diff +// delete: upsert observed threads first (so unchanged thread_ids hit ON +// CONFLICT DO UPDATE and the UPDATE trigger fires), then prune the rows whose +// thread_id is no longer in the observed set. +func TestPRReviewThreadsCDC_EmitsResolvedOnReplacePoll(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + rec, err := s.CreateSession(ctx, sampleRecord("mer")) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Second) + pr := domain.PullRequest{URL: "https://example/pr/55", SessionID: rec.ID, Number: 55, UpdatedAt: now} + + // First poll: seed via Replace (no Partial pagination). Same shape as the + // real GitHub provider path when the review-thread listing fits in one page. + if err := s.WriteSCMObservation(ctx, pr, nil, []domain.PullRequestReviewThread{{ + ThreadID: "t1", Path: "main.go", Line: 7, IsBot: true, SemanticHash: "v1", UpdatedAt: now, + }}, nil, ports.ReviewWriteReplace); err != nil { + t.Fatal(err) + } + + // Second poll: same thread, resolved flipped to true, also via Replace. + // On the buggy code this fired the INSERT trigger again (because the + // DELETE-all removed the row first) and the UPDATE trigger never saw the + // resolved transition. + if err := s.WriteSCMObservation(ctx, pr, nil, []domain.PullRequestReviewThread{{ + ThreadID: "t1", Path: "main.go", Line: 7, Resolved: true, IsBot: true, SemanticHash: "v2", UpdatedAt: now.Add(time.Second), + }}, nil, ports.ReviewWriteReplace); err != nil { + t.Fatal(err) + } + + rows, err := s.EventsAfter(ctx, 0, 100) + if err != nil { + t.Fatal(err) + } + var added, resolved []cdc.Event + for _, r := range rows { + switch r.Type { + case cdc.EventPRReviewThreadAdded: + added = append(added, r) + case cdc.EventPRReviewThreadResolved: + resolved = append(resolved, r) + } + } + if len(added) != 1 { + t.Fatalf("want 1 review-thread added CDC event (initial Replace insert), got %d", len(added)) + } + if len(resolved) != 1 { + t.Fatalf("want 1 review-thread resolved CDC event on the second Replace poll, got %d", len(resolved)) + } + var payload map[string]any + if err := json.Unmarshal(resolved[0].Payload, &payload); err != nil { + t.Fatalf("resolved payload JSON: %v", err) + } + if payload["thread"] != "t1" || payload["resolved"] != true { + t.Fatalf("resolved payload = %#v", payload) + } +} + +// Pruning regression: Replace must still drop threads that are no longer in +// the observed listing, otherwise stale rows accumulate. Seed two threads, +// then re-poll with only one; the missing thread must be gone, while the +// surviving thread still gets an UPDATE (not a fresh INSERT). +func TestPRReviewThreadsReplace_PrunesOrphansWithoutReinserting(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + rec, err := s.CreateSession(ctx, sampleRecord("mer")) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Second) + pr := domain.PullRequest{URL: "https://example/pr/56", SessionID: rec.ID, Number: 56, UpdatedAt: now} + + if err := s.WriteSCMObservation(ctx, pr, nil, []domain.PullRequestReviewThread{ + {ThreadID: "keep", Path: "a.go", Line: 1, IsBot: true, SemanticHash: "k1", UpdatedAt: now}, + {ThreadID: "drop", Path: "b.go", Line: 2, IsBot: true, SemanticHash: "d1", UpdatedAt: now}, + }, nil, ports.ReviewWriteReplace); err != nil { + t.Fatal(err) + } + if err := s.WriteSCMObservation(ctx, pr, nil, []domain.PullRequestReviewThread{ + {ThreadID: "keep", Path: "a.go", Line: 1, Resolved: true, IsBot: true, SemanticHash: "k2", UpdatedAt: now.Add(time.Second)}, + }, nil, ports.ReviewWriteReplace); err != nil { + t.Fatal(err) + } + + got, err := s.ListPRReviewThreads(ctx, pr.URL) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].ThreadID != "keep" || !got[0].Resolved { + t.Fatalf("after prune want one resolved \"keep\" row, got %+v", got) + } + rows, err := s.EventsAfter(ctx, 0, 100) + if err != nil { + t.Fatal(err) + } + var added, resolved int + for _, r := range rows { + if r.Type == cdc.EventPRReviewThreadAdded { + added++ + } + if r.Type == cdc.EventPRReviewThreadResolved { + resolved++ + } + } + // Two adds from poll 1 ("keep" and "drop"), no extra add on poll 2 + // (the surviving row went through ON CONFLICT DO UPDATE, not a fresh + // INSERT), and one resolved transition from the kept row's flip. + if added != 2 { + t.Fatalf("want 2 added events across both polls, got %d", added) + } + if resolved != 1 { + t.Fatalf("want 1 resolved event from the surviving thread's flip, got %d", resolved) + } +} + // WritePR persists scalar facts, checks, and comments in one tx; all three // should be queryable afterward. func TestWritePR_PersistsScalarsChecksAndComments(t *testing.T) { diff --git a/backend/internal/storage/sqlite/store/pr_store.go b/backend/internal/storage/sqlite/store/pr_store.go index 1dc391939..fc8147830 100644 --- a/backend/internal/storage/sqlite/store/pr_store.go +++ b/backend/internal/storage/sqlite/store/pr_store.go @@ -110,11 +110,6 @@ func writePRRows(ctx context.Context, q *gen.Queries, pr domain.PullRequest, che return err } } - if reviewMode == ports.ReviewWriteReplace { - if err := q.DeletePRReviewThreads(ctx, pr.URL); err != nil { - return err - } - } if reviewMode == ports.ReviewWriteReplace { if err := q.DeletePRComments(ctx, pr.URL); err != nil { return err @@ -131,6 +126,30 @@ func writePRRows(ctx context.Context, q *gen.Queries, pr domain.PullRequest, che } } } + // Replace mode prunes orphans (threads no longer observed in the upstream + // listing) AFTER the upserts above, so that threads present in both the + // pre- and post-state hit ON CONFLICT DO UPDATE and fire the UPDATE trigger + // (e.g. pr_review_thread_resolved when resolved flips). The old + // delete-everything-first approach made every poll look like a fresh INSERT + // and the UPDATE trigger was unreachable for the common Replace path. + if reviewMode == ports.ReviewWriteReplace { + observed := make(map[string]struct{}, len(threads)) + for _, th := range threads { + observed[th.ThreadID] = struct{}{} + } + existing, err := q.ListPRReviewThreads(ctx, pr.URL) + if err != nil { + return fmt.Errorf("list review threads for prune %s: %w", pr.URL, err) + } + for _, row := range existing { + if _, ok := observed[row.ThreadID]; ok { + continue + } + if err := q.DeletePRReviewThread(ctx, gen.DeletePRReviewThreadParams{PRURL: pr.URL, ThreadID: row.ThreadID}); err != nil { + return fmt.Errorf("prune review thread %q: %w", row.ThreadID, err) + } + } + } if reviewMode == ports.ReviewWriteMerge { for _, threadID := range reviewThreadIDs(threads, comments) { if err := q.DeletePRCommentsByThread(ctx, gen.DeletePRCommentsByThreadParams{PRURL: pr.URL, ThreadID: threadID}); err != nil { diff --git a/backend/internal/storage/sqlite/store/session_store.go b/backend/internal/storage/sqlite/store/session_store.go index 355b955f8..751b9ebf2 100644 --- a/backend/internal/storage/sqlite/store/session_store.go +++ b/backend/internal/storage/sqlite/store/session_store.go @@ -56,6 +56,80 @@ func (s *Store) RenameSession(ctx context.Context, id domain.SessionID, displayN return rows > 0, nil } +// DeleteSession removes a session row, but only if it is still in seed state +// (no workspace, no runtime handle, no agent session id, no prompt, and not +// already terminated). Rows that have observable spawn output are immutable +// to preserve the no-resurrection guarantee — for those, callers fall back to +// MarkTerminated (lifecycle.Manager) instead. +// +// The deletion runs in a transaction. It first probes seed state with +// SessionIsSeed; only if that returns true does it clear the session's +// change_log rows (required because change_log FKs sessions(id) without +// ON DELETE CASCADE) and then delete the session row. For live or absent +// sessions the transaction commits with no rows touched — critically, the +// session_created / session_updated CDC events for live sessions are NOT +// destroyed when callers (e.g. RollbackSpawn's delete-then-kill fallback) +// invoke DeleteSession on a fully-spawned row. +// +// Returns deleted=true when a seed row was removed; deleted=false when the +// session id did not match a seed row (either it never existed, or it had +// already progressed past seed state). The latter case is benign — the caller +// should fall back to MarkTerminated. +func (s *Store) DeleteSession(ctx context.Context, id domain.SessionID) (bool, error) { + s.writeMu.Lock() + defer s.writeMu.Unlock() + tx, err := s.writeDB.BeginTx(ctx, nil) + if err != nil { + return false, fmt.Errorf("begin delete seed session: %w", err) + } + defer func() { _ = tx.Rollback() }() + q := s.qw.WithTx(tx) + + isSeed, err := q.SessionIsSeed(ctx, id) + if err != nil { + return false, fmt.Errorf("delete seed session: probe seed state for %s: %w", id, err) + } + if !isSeed { + // Commit the empty tx so we don't leak a transaction. Critically, do + // NOT touch change_log here — for a live session that contains real + // session_created / session_updated CDC events. + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("delete seed session: commit no-op: %w", err) + } + return false, nil + } + + // Drop change_log rows for this session id first so the FK doesn't reject + // the session DELETE. We do not touch project-level events (session_id IS + // NULL) — those belong to the project, not this session. Both this DELETE + // and the session DELETE below run via raw ExecContext to sidestep sqlc + // 1.31's SQLite-parser bug, which strips trailing `?` placeholders and + // string literals from DELETE statements (see queries/changelog.sql and + // queries/sessions.sql for the documented workaround context). + if _, err := tx.ExecContext(ctx, `DELETE FROM change_log WHERE session_id = ?`, id); err != nil { + return false, fmt.Errorf("delete seed session: clear change log for %s: %w", id, err) + } + res, err := tx.ExecContext(ctx, ` +DELETE FROM sessions +WHERE id = ? + AND is_terminated = 0 + AND workspace_path = '' + AND runtime_handle_id = '' + AND agent_session_id = '' + AND prompt = ''`, id) + if err != nil { + return false, fmt.Errorf("delete seed session %s: %w", id, err) + } + n, err := res.RowsAffected() + if err != nil { + return false, fmt.Errorf("delete seed session %s: rows affected: %w", id, err) + } + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("delete seed session: commit: %w", err) + } + return n > 0, nil +} + // GetSession returns the full record for a session, or ok=false if absent. func (s *Store) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) { row, err := s.qr.GetSession(ctx, id) diff --git a/backend/internal/storage/sqlite/store/store_test.go b/backend/internal/storage/sqlite/store/store_test.go index cedbbd125..a03c1e9bd 100644 --- a/backend/internal/storage/sqlite/store/store_test.go +++ b/backend/internal/storage/sqlite/store/store_test.go @@ -102,6 +102,76 @@ func TestSessionCreateAssignsPerProjectID(t *testing.T) { } } +// TestDeleteSessionOnlyRemovesSeedRows covers Bug 4's storage-layer guarantee: +// DeleteSession removes a session row only when the row is still in seed state +// (no workspace, no runtime handle, no agent session id, no prompt, not +// terminated). Rows that already carry spawn output are immutable so the +// no-resurrection guarantee for live sessions still holds. +func TestDeleteSessionOnlyRemovesSeedRows(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + + // Seed row: just CreateSession output, no metadata yet. + now := time.Now().UTC().Truncate(time.Second) + seed := domain.SessionRecord{ + ProjectID: "mer", + Kind: domain.KindWorker, + Harness: domain.HarnessClaudeCode, + Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: now}, + CreatedAt: now, + UpdatedAt: now, + } + r1, err := s.CreateSession(ctx, seed) + if err != nil { + t.Fatalf("create seed: %v", err) + } + + deleted, err := s.DeleteSession(ctx, r1.ID) + if err != nil || !deleted { + t.Fatalf("delete seed = %v %v, want true nil", deleted, err) + } + if _, ok, _ := s.GetSession(ctx, r1.ID); ok { + t.Fatal("seed row still present after DeleteSession") + } + + // A row with workspace_path populated must NOT be deleted — even if + // !is_terminated. This is the no-resurrection guarantee for live work. + r2, err := s.CreateSession(ctx, sampleRecord("mer")) + if err != nil { + t.Fatalf("create live: %v", err) + } + deleted, err = s.DeleteSession(ctx, r2.ID) + if err != nil { + t.Fatalf("delete live err = %v", err) + } + if deleted { + t.Fatal("DeleteSession must be a no-op for rows with spawn output") + } + if _, ok, _ := s.GetSession(ctx, r2.ID); !ok { + t.Fatal("live row was removed by DeleteSession") + } + + // A terminated row is also out of scope: terminal-state rows hold cleanup + // metadata users may still inspect, so the gate refuses them too. + r3, err := s.CreateSession(ctx, seed) + if err != nil { + t.Fatalf("create extra seed: %v", err) + } + terminated := r3 + terminated.IsTerminated = true + if err := s.UpdateSession(ctx, terminated); err != nil { + t.Fatalf("mark terminated: %v", err) + } + deleted, err = s.DeleteSession(ctx, r3.ID) + if err != nil { + t.Fatalf("delete terminated err = %v", err) + } + if deleted { + t.Fatal("DeleteSession must be a no-op for terminated rows") + } +} + func TestSessionRenameUpdatesDisplayName(t *testing.T) { s := newTestStore(t) ctx := context.Background() diff --git a/frontend/src/api/schema.ts b/frontend/src/api/schema.ts index fb22fe12c..b23fb2752 100644 --- a/frontend/src/api/schema.ts +++ b/frontend/src/api/schema.ts @@ -247,6 +247,23 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/sessions/{sessionId}/rollback": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Undo a partially-completed spawn (delete seed row, or kill if spawn output exists) */ + post: operations["rollbackSession"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/sessions/{sessionId}/send": { parameters: { query?: never; @@ -396,6 +413,12 @@ export interface components { session: components["schemas"]["Session"]; sessionId: string; }; + RollbackSessionResponse: { + deleted?: boolean; + killed?: boolean; + ok: boolean; + sessionId: string; + }; SCMConfig: { package?: string; path?: string; @@ -1467,6 +1490,56 @@ export interface operations { }; }; }; + rollbackSession: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Session identifier, e.g. project-1. */ + sessionId: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description OK */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["RollbackSessionResponse"]; + }; + }; + /** @description Not Found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Conflict */ + 409: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Internal Server Error */ + 500: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + }; + }; sendSessionMessage: { parameters: { query?: never;