fix: 7 bugs from discussion #149 smoke walk (envelope, spawn, CDC, observer) (#153)

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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.
This commit is contained in:
Harshit Singh Bhandari 2026-06-07 07:35:46 +05:30 committed by GitHub
parent c7e3a0336e
commit c343c55c14
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
59 changed files with 1269 additions and 147 deletions

View File

@ -5,6 +5,7 @@ package agy
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) agyBinary(ctx context.Context) (string, error) {

View File

@ -10,6 +10,7 @@ package aider
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -173,7 +174,7 @@ func ResolveAiderBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "aider", nil return "", fmt.Errorf("aider: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("aider"); err == nil && path != "" { 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) { func (p *Plugin) aiderBinary(ctx context.Context) (string, error) {

View File

@ -281,11 +281,17 @@ func TestContextCancellation(t *testing.T) {
} }
func TestResolveAiderBinaryFallback(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()) bin, err := ResolveAiderBinary(context.Background())
if err != nil { 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 == "" { if bin == "" {
t.Fatal("ResolveAiderBinary returned empty string") t.Fatal("ResolveAiderBinary returned empty string with no error")
} }
} }

View File

@ -8,6 +8,7 @@ package amp
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -176,7 +177,7 @@ func ResolveAmpBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "amp", nil return "", fmt.Errorf("amp: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("amp"); err == nil && path != "" { 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) { func (p *Plugin) ampBinary(ctx context.Context) (string, error) {

View File

@ -38,6 +38,7 @@ package auggie
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -201,7 +202,7 @@ func ResolveAuggieBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "auggie", nil return "", fmt.Errorf("auggie: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("auggie"); err == nil && path != "" { 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) { func (p *Plugin) auggieBinary(ctx context.Context) (string, error) {

View File

@ -184,14 +184,18 @@ func TestSessionInfoNoOp(t *testing.T) {
} }
func TestResolveAuggieBinaryFallback(t *testing.T) { func TestResolveAuggieBinaryFallback(t *testing.T) {
// With a cancelled context the resolver returns the context error rather than // When the binary is not on PATH or any well-known location, the resolver
// a binary path; with a live context it always yields a non-empty path. // 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()) bin, err := ResolveAuggieBinary(context.Background())
if err != nil { 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 == "" { if bin == "" {
t.Fatal("ResolveAuggieBinary returned empty path") t.Fatal("ResolveAuggieBinary returned empty path with no error")
} }
} }

View File

@ -12,6 +12,7 @@ package autohand
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) autohandBinary(ctx context.Context) (string, error) {

View File

@ -306,7 +306,7 @@ func ResolveClaudeBinary(ctx context.Context) (string, error) {
return candidate, nil return candidate, nil
} }
} }
return "claude", nil return "", fmt.Errorf("claude: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("claude"); err == nil && path != "" { 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) { func (p *Plugin) claudeBinary(ctx context.Context) (string, error) {

View File

@ -14,6 +14,7 @@ package cline
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) clineBinary(ctx context.Context) (string, error) {

View File

@ -8,6 +8,7 @@ package codex
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) codexBinary(ctx context.Context) (string, error) {

View File

@ -22,6 +22,7 @@ package continueagent
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -197,7 +198,7 @@ func ResolveContinueBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "cn", nil return "", fmt.Errorf("cn: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("cn"); err == nil && path != "" { 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) { func (p *Plugin) continueBinary(ctx context.Context) (string, error) {

View File

@ -19,6 +19,7 @@ package copilot
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) copilotBinary(ctx context.Context) (string, error) {

View File

@ -8,6 +8,7 @@ package crush
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) crushBinary(ctx context.Context) (string, error) {

View File

@ -9,6 +9,7 @@ package cursor
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -165,7 +166,7 @@ func ResolveCursorBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "cursor-agent", nil return "", fmt.Errorf("cursor: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("cursor-agent"); err == nil && path != "" { 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) { func (p *Plugin) cursorBinary(ctx context.Context) (string, error) {

View File

@ -24,6 +24,7 @@ package devin
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -200,7 +201,7 @@ func ResolveDevinBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "devin", nil return "", fmt.Errorf("devin: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("devin"); err == nil && path != "" { 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) { func (p *Plugin) devinBinary(ctx context.Context) (string, error) {

View File

@ -2,6 +2,7 @@ package devin
import ( import (
"context" "context"
"errors"
"reflect" "reflect"
"strings" "strings"
"testing" "testing"
@ -254,14 +255,18 @@ func TestGetAgentHooksCtxCancelled(t *testing.T) {
} }
func TestResolveDevinBinaryFallback(t *testing.T) { func TestResolveDevinBinaryFallback(t *testing.T) {
// When devin is not on PATH or well-known locations, fall back to the bare // When the binary is not on PATH or any well-known location, the resolver
// name so exec can still attempt to launch it. // 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()) bin, err := ResolveDevinBinary(context.Background())
if err != nil { 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 == "" { if bin == "" {
t.Fatal("empty binary path") t.Fatal("ResolveDevinBinary returned empty path with no error")
} }
} }

View File

@ -288,7 +288,7 @@ func ResolveDroidBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "droid", nil return "", fmt.Errorf("droid: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("droid"); err == nil && path != "" { 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) { func (p *Plugin) droidBinary(ctx context.Context) (string, error) {

View File

@ -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 != "" { 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) { func (p *Plugin) gooseBinary(ctx context.Context) (string, error) {

View File

@ -3,10 +3,10 @@ package goose
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"os" "os"
"path/filepath" "path/filepath"
"reflect" "reflect"
"strings"
"testing" "testing"
"github.com/aoagents/agent-orchestrator/backend/internal/ports" "github.com/aoagents/agent-orchestrator/backend/internal/ports"
@ -380,17 +380,18 @@ func TestSessionInfoFalseWhenNoHookMetadata(t *testing.T) {
} }
func TestResolveGooseBinaryFallback(t *testing.T) { func TestResolveGooseBinaryFallback(t *testing.T) {
// On a machine without goose on PATH or any well-known location, resolution // When the binary is not on PATH or any well-known location, the resolver
// still returns a usable last-ditch "goose" name rather than empty. // MUST surface ports.ErrAgentBinaryNotFound rather than a silent string
got, err := ResolveGooseBinary(context.Background()) // fallback that lets a missing CLI launch into an empty zellij pane.
bin, err := ResolveGooseBinary(context.Background())
if err != nil { 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 == "" { if bin == "" {
t.Fatal("ResolveGooseBinary returned empty binary") t.Fatal("ResolveGooseBinary returned empty path with no error")
}
if !strings.Contains(got, "goose") {
t.Fatalf("ResolveGooseBinary = %q, want a path containing goose", got)
} }
} }

View File

@ -16,6 +16,7 @@ package grok
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -212,7 +213,7 @@ func ResolveGrokBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "grok", nil return "", fmt.Errorf("grok: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("grok"); err == nil && path != "" { 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) { func (p *Plugin) grokBinary(ctx context.Context) (string, error) {

View File

@ -24,6 +24,7 @@ package kilocode
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -269,7 +270,7 @@ func ResolveKilocodeBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "kilocode", nil return "", fmt.Errorf("kilocode: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("kilocode"); err == nil && path != "" { 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) { func (p *Plugin) kilocodeBinary(ctx context.Context) (string, error) {

View File

@ -17,6 +17,7 @@ package kimi
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -218,7 +219,7 @@ func ResolveKimiBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "kimi", nil return "", fmt.Errorf("kimi: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("kimi"); err == nil && path != "" { 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) { func (p *Plugin) kimiBinary(ctx context.Context) (string, error) {

View File

@ -20,6 +20,7 @@ package kiro
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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 != "" { 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) { func (p *Plugin) kiroBinary(ctx context.Context) (string, error) {

View File

@ -3,6 +3,7 @@ package kiro
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"os" "os"
"path/filepath" "path/filepath"
"reflect" "reflect"
@ -384,12 +385,18 @@ func TestDeriveActivityState(t *testing.T) {
} }
func TestResolveKiroBinaryFallback(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 { if err != nil {
t.Fatal(err) if !errors.Is(err, ports.ErrAgentBinaryNotFound) {
t.Fatalf("err = %v, want ports.ErrAgentBinaryNotFound", err)
}
return
} }
if got == "" { if bin == "" {
t.Fatal("ResolveKiroBinary returned empty path") t.Fatal("ResolveKiroBinary returned empty path with no error")
} }
} }

View File

@ -33,6 +33,7 @@ package pi
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -190,7 +191,7 @@ func ResolvePiBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "pi", nil return "", fmt.Errorf("pi: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("pi"); err == nil && path != "" { 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) { func (p *Plugin) piBinary(ctx context.Context) (string, error) {

View File

@ -15,6 +15,7 @@ package qwen
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "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, // ResolveQwenBinary returns the path to the qwen binary on this machine,
// searching PATH then a handful of well-known install locations (Homebrew, npm // 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 // global). Returns ports.ErrAgentBinaryNotFound when none of those find the
// "command not found" rather than an empty argv. // 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) { func ResolveQwenBinary(ctx context.Context) (string, error) {
if err := ctx.Err(); err != nil { if err := ctx.Err(); err != nil {
return "", err 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 != "" { 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) { func (p *Plugin) qwenBinary(ctx context.Context) (string, error) {

View File

@ -26,6 +26,7 @@ package vibe
import ( import (
"context" "context"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@ -197,7 +198,7 @@ func ResolveVibeBinary(ctx context.Context) (string, error) {
return "", err return "", err
} }
} }
return "vibe", nil return "", fmt.Errorf("vibe: %w", ports.ErrAgentBinaryNotFound)
} }
if path, err := exec.LookPath("vibe"); err == nil && path != "" { 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) { func (p *Plugin) vibeBinary(ctx context.Context) (string, error) {

View File

@ -24,6 +24,16 @@ var (
ErrUnsafePath = errors.New("gitworktree: unsafe workspace path") 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. // RepoResolver maps a project to the absolute path of its source git repo.
type RepoResolver interface { type RepoResolver interface {
RepoPath(projectID domain.ProjectID) (string, error) 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 { 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) localBranch, err := w.refExists(ctx, repo, "refs/heads/"+branch)
if err != nil { if err != nil {
return err return err
@ -205,8 +226,18 @@ func (w *Workspace) addWorktree(ctx context.Context, repo, path, branch string)
} }
return nil return nil
} }
// `worktree add -b <branch> <path> <base>` creates a fresh local branch from
// <base>. resolveBaseRef tries `origin/<branch>` first, so a fetched-but-
// not-checked-out remote branch auto-tracks cleanly via that path. If
// neither origin/<branch>, 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) baseRef, err := w.resolveBaseRef(ctx, repo, branch)
if err != nil { 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 return err
} }
if _, err := w.run(ctx, w.binary, worktreeAddNewBranchArgs(repo, branch, path, baseRef)...); err != nil { 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 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) { func (w *Workspace) resolveBaseRef(ctx context.Context, repo, branch string) (string, error) {
candidates := baseRefCandidates(branch, w.defaultBranch) candidates := baseRefCandidates(branch, w.defaultBranch)
for _, ref := range candidates { for _, ref := range candidates {
@ -233,7 +268,17 @@ func (w *Workspace) resolveBaseRef(ctx context.Context, repo, branch string) (st
return ref, nil 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) { 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 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) { func pathExistsNonEmpty(path string) (bool, error) {
entries, err := os.ReadDir(path) entries, err := os.ReadDir(path)
if err == nil { if err == nil {

View File

@ -4,6 +4,7 @@ import (
"context" "context"
"errors" "errors"
"os" "os"
"os/exec"
"path/filepath" "path/filepath"
"reflect" "reflect"
"strings" "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 { func mkdirFile(dir, name string) error {
if err := os.MkdirAll(dir, 0o755); err != nil { if err := os.MkdirAll(dir, 0o755); err != nil {
return err return err

View File

@ -78,6 +78,10 @@ func (f *fakeSessionService) Kill(context.Context, domain.SessionID) (bool, erro
return false, nil 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) { func (f *fakeSessionService) Cleanup(context.Context, domain.ProjectID) ([]domain.SessionID, error) {
return nil, nil return nil, nil
} }

View File

@ -133,7 +133,21 @@ func newSpawnCommand(ctx *commandContext) *cobra.Command {
return cmd 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 { func (c *commandContext) rollbackSpawnedSession(ctx context.Context, id string) error {
var res killSessionResponse var res rollbackSessionResponse
return c.postJSON(ctx, "sessions/"+url.PathEscape(id)+"/kill", struct{}{}, &res) 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"`
} }

View File

@ -98,9 +98,9 @@ func TestSpawnClaimPRFailureRollsBackSession(t *testing.T) {
} }
w.WriteHeader(http.StatusNotFound) w.WriteHeader(http.StatusNotFound)
_, _ = io.WriteString(w, `{"error":"not_found","code":"PR_NOT_FOUND","message":"PR not found"}`) _, _ = 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") 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: default:
http.NotFound(w, r) http.NotFound(w, r)
} }
@ -119,7 +119,7 @@ func TestSpawnClaimPRFailureRollsBackSession(t *testing.T) {
if sessions["demo-10"] { if sessions["demo-10"] {
t.Fatalf("spawned session still present after claim rollback: %#v", sessions) 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) { if !reflect.DeepEqual(requests, want) {
t.Fatalf("requests=%#v want %#v", requests, want) t.Fatalf("requests=%#v want %#v", requests, want)
} }

View File

@ -775,6 +775,46 @@ paths:
summary: Restore a terminated session summary: Restore a terminated session
tags: tags:
- sessions - 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: /api/v1/sessions/{sessionId}/send:
post: post:
operationId: sendSessionMessage operationId: sendSessionMessage
@ -1149,6 +1189,20 @@ components:
- sessionId - sessionId
- session - session
type: object type: object
RollbackSessionResponse:
properties:
deleted:
type: boolean
killed:
type: boolean
ok:
type: boolean
sessionId:
type: string
required:
- ok
- sessionId
type: object
SCMConfig: SCMConfig:
properties: properties:
package: package:

View File

@ -137,6 +137,7 @@ var schemaNames = map[string]string{
"ControllersRestoreSessionResponse": "RestoreSessionResponse", "ControllersRestoreSessionResponse": "RestoreSessionResponse",
"ControllersCleanupSessionsResponse": "CleanupSessionsResponse", "ControllersCleanupSessionsResponse": "CleanupSessionsResponse",
"ControllersKillSessionResponse": "KillSessionResponse", "ControllersKillSessionResponse": "KillSessionResponse",
"ControllersRollbackSessionResponse": "RollbackSessionResponse",
"ControllersSendSessionMessageRequest": "SendSessionMessageRequest", "ControllersSendSessionMessageRequest": "SendSessionMessageRequest",
"ControllersSendSessionMessageResponse": "SendSessionMessageResponse", "ControllersSendSessionMessageResponse": "SendSessionMessageResponse",
"ControllersClaimPRResponse": "ClaimPRResponse", "ControllersClaimPRResponse": "ClaimPRResponse",
@ -414,6 +415,17 @@ func sessionOperations() []operation {
{http.StatusInternalServerError, envelope.APIError{}}, {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", method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/send", id: "sendSessionMessage", tag: "sessions",
summary: "Send a message to a running session's agent", summary: "Send a message to a running session's agent",

View File

@ -158,6 +158,16 @@ type KillSessionResponse struct {
Freed bool `json:"freed,omitempty"` 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. // CleanupSessionsResponse is the body of POST /api/v1/sessions/cleanup.
type CleanupSessionsResponse struct { type CleanupSessionsResponse struct {
OK bool `json:"ok"` OK bool `json:"ok"`

View File

@ -30,6 +30,7 @@ type SessionService interface {
Get(ctx context.Context, id domain.SessionID) (domain.Session, error) Get(ctx context.Context, id domain.SessionID) (domain.Session, error)
Restore(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) 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) Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, error)
Rename(ctx context.Context, id domain.SessionID, displayName string) error Rename(ctx context.Context, id domain.SessionID, displayName string) error
Send(ctx context.Context, id domain.SessionID, message 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.Patch("/sessions/{sessionId}", c.rename)
r.Post("/sessions/{sessionId}/restore", c.restore) r.Post("/sessions/{sessionId}/restore", c.restore)
r.Post("/sessions/{sessionId}/kill", c.kill) 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}/send", c.send)
r.Post("/sessions/{sessionId}/activity", c.activity) r.Post("/sessions/{sessionId}/activity", c.activity)
r.Get("/orchestrators", c.listOrchestrators) 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}) 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) { func (c *SessionsController) cleanup(w http.ResponseWriter, r *http.Request) {
if c.Svc == nil { if c.Svc == nil {
apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/cleanup") apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/cleanup")

View File

@ -92,6 +92,14 @@ func (f *fakeSessionService) Kill(_ context.Context, id domain.SessionID) (bool,
return true, nil 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) { func (f *fakeSessionService) Cleanup(_ context.Context, project domain.ProjectID) ([]domain.SessionID, error) {
f.cleanupProjects = append(f.cleanupProjects, project) f.cleanupProjects = append(f.cleanupProjects, project)
if f.cleanupResult != nil { if f.cleanupResult != nil {

View File

@ -97,7 +97,7 @@ func newStack(t *testing.T) *stack {
prm := prsvc.New(prsvc.Deps{Writer: store, Lifecycle: lcm}) prm := prsvc.New(prsvc.Deps{Writer: store, Lifecycle: lcm})
rt := &stubRuntime{} rt := &stubRuntime{}
ws := &stubWorkspace{} 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) sm := sessionsvc.New(mgr, store)
return &stack{store: store, sm: sm, lcm: lcm, prm: prm, rt: rt, ws: ws, msg: msg} return &stack{store: store, sm: sm, lcm: lcm, prm: prm, rt: rt, ws: ws, msg: msg}
} }

View File

@ -168,6 +168,16 @@ func (o *Observer) Start(ctx context.Context) <-chan struct{} {
func (o *Observer) loop(ctx context.Context, done chan<- struct{}) { func (o *Observer) loop(ctx context.Context, done chan<- struct{}) {
defer close(done) 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) { if err := o.Poll(ctx); err != nil && !errors.Is(err, context.Canceled) {
o.logger.Error("scm observer: initial poll failed", "err", err) o.logger.Error("scm observer: initial poll failed", "err", err)
} }

View File

@ -5,12 +5,14 @@ package scm
// review cadence, semantic hashes, and notification behavior stay provider-neutral. // review cadence, semantic hashes, and notification behavior stay provider-neutral.
import ( import (
"bytes"
"context" "context"
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log/slog" "log/slog"
"os/exec" "os/exec"
"strings"
"sync" "sync"
"testing" "testing"
"time" "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) { func TestPoll_RepoETag304SkipsDetectPR(t *testing.T) {
store := testStoreWithSession() store := testStoreWithSession()
provider := &fakeProvider{repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "v1", NotModified: true}}, observations: map[string]ports.SCMObservation{}} provider := &fakeProvider{repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "v1", NotModified: true}}, observations: map[string]ports.SCMObservation{}}

View File

@ -2,10 +2,18 @@ package ports
import ( import (
"context" "context"
"errors"
"github.com/aoagents/agent-orchestrator/backend/internal/domain" "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, …) // Agent is the contract every CLI coding agent adapter (claude-code, codex, …)
// must satisfy. It supplies the argv and process configuration the Session // must satisfy. It supplies the argv and process configuration the Session
// Manager needs to launch, restore, and read back a native agent session. // Manager needs to launch, restore, and read back a native agent session.

View File

@ -108,6 +108,18 @@ type Workspace interface {
Restore(ctx context.Context, cfg WorkspaceConfig) (WorkspaceInfo, error) 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. // WorkspaceConfig is the spec for creating or restoring a session's workspace.
type WorkspaceConfig struct { type WorkspaceConfig struct {
ProjectID domain.ProjectID ProjectID domain.ProjectID

View File

@ -41,6 +41,15 @@ type commander interface {
Kill(ctx context.Context, id domain.SessionID) (bool, error) Kill(ctx context.Context, id domain.SessionID) (bool, error)
Send(ctx context.Context, id domain.SessionID, message string) error Send(ctx context.Context, id domain.SessionID, message string) error
Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, 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 { type scmProvider interface {
@ -92,6 +101,9 @@ func NewWithDeps(d Deps) *Service {
// Spawn creates a session and returns the API-facing read model. // Spawn creates a session and returns the API-facing read model.
func (s *Service) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Session, error) { 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) rec, err := s.manager.Spawn(ctx, cfg)
if err != nil { if err != nil {
return domain.Session{}, toAPIError(err) 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) 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 // 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 // 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 // one is the only live coordinator — a business rule that belongs here, not in the
// HTTP controller. // HTTP controller.
func (s *Service) SpawnOrchestrator(ctx context.Context, projectID domain.ProjectID, clean bool) (domain.Session, error) { 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 { if clean {
active := true active := true
existing, err := s.List(ctx, ListFilter{ProjectID: projectID, Active: &active, OrchestratorOnly: 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) 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. // Send delegates agent messaging to the internal manager.
func (s *Service) Send(ctx context.Context, id domain.SessionID, message string) error { func (s *Service) Send(ctx context.Context, id domain.SessionID, message string) error {
return toAPIError(s.manager.Send(ctx, id, message)) 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) return apierr.Conflict("SESSION_INCOMPLETE_HANDLE", "Session is missing runtime or workspace handles", nil)
case errors.Is(err, sessionmanager.ErrProjectNotResolvable): 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) 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: default:
return err return err
} }

View File

@ -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) { func (f *fakeCommander) Cleanup(context.Context, domain.ProjectID) ([]domain.SessionID, error) {
return nil, nil return nil, nil
} }
func (f *fakeCommander) RollbackSpawn(context.Context, domain.SessionID) (bool, bool, error) {
return false, false, nil
}
func TestSpawnOrchestratorCleanKillsActiveOrchestratorsBeforeSpawn(t *testing.T) { func TestSpawnOrchestratorCleanKillsActiveOrchestratorsBeforeSpawn(t *testing.T) {
st := newFakeStore() st := newFakeStore()
st.projects["mer"] = domain.ProjectRecord{ID: "mer"}
// Two active orchestrators plus an unrelated worker and a terminated // Two active orchestrators plus an unrelated worker and a terminated
// orchestrator that must be left alone. // orchestrator that must be left alone.
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindOrchestrator} 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) { func TestSpawnOrchestratorNoCleanSkipsKills(t *testing.T) {
st := newFakeStore() st := newFakeStore()
st.projects["mer"] = domain.ProjectRecord{ID: "mer"}
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindOrchestrator} st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindOrchestrator}
fc := &fakeCommander{} fc := &fakeCommander{}

View File

@ -6,6 +6,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"os/exec"
"time" "time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain" "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) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error)
ListSessions(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) ListSessions(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error)
ListAllSessions(ctx context.Context) ([]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 // Manager coordinates internal session spawn, restore, kill, and cleanup over
@ -62,6 +69,10 @@ type Manager struct {
lcm lifecycleRecorder lcm lifecycleRecorder
dataDir string dataDir string
clock func() time.Time 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. // 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. // commands can open the same store.
DataDir string DataDir string
Clock func() time.Time 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 // New builds a Session Manager from its dependencies, defaulting the clock to
@ -90,10 +105,14 @@ func New(d Deps) *Manager {
lcm: d.Lifecycle, lcm: d.Lifecycle,
dataDir: d.DataDir, dataDir: d.DataDir,
clock: d.Clock, clock: d.Clock,
lookPath: d.LookPath,
} }
if m.clock == nil { if m.clock == nil {
m.clock = time.Now m.clock = time.Now
} }
if m.lookPath == nil {
m.lookPath = exec.LookPath
}
return m return m
} }
@ -147,6 +166,15 @@ func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess
m.markSpawnFailedTerminated(ctx, id) m.markSpawnFailedTerminated(ctx, id)
return domain.SessionRecord{}, fmt.Errorf("spawn %s: launch command: %w", id, err) 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{ handle, err := m.runtime.Create(ctx, ports.RuntimeConfig{
SessionID: id, SessionID: id,
WorkspacePath: ws.Path, 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. // 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) { func (m *Manager) markSpawnFailedTerminated(ctx context.Context, id domain.SessionID) {
_ = m.lcm.MarkTerminated(ctx, id) _ = 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 // Kill records terminal intent with the LCM, then tears down the runtime and
// workspace. A workspace teardown refused by the worktree-remove safety // workspace. A workspace teardown refused by the worktree-remove safety
// (uncommitted work) surfaces as an error with freed=false and is never forced. // (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) return domain.SessionRecord{}, fmt.Errorf("restore %s: %w", id, ErrNotRestorable)
} }
meta := rec.Metadata 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 == "" { if meta.AgentSessionID == "" && meta.Prompt == "" {
return domain.SessionRecord{}, fmt.Errorf("restore %s: nothing to resume from", id) 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 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 { func runtimeHandle(meta domain.SessionMetadata) ports.RuntimeHandle {
return ports.RuntimeHandle{ID: meta.RuntimeHandleID} return ports.RuntimeHandle{ID: meta.RuntimeHandleID}
} }

View File

@ -53,6 +53,18 @@ func (f *fakeStore) ListAllSessions(context.Context) ([]domain.SessionRecord, er
} }
return out, nil 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) { func (f *fakeStore) GetDisplayPRFactsForSession(_ context.Context, id domain.SessionID) (domain.PRFacts, bool, error) {
if pr := f.pr[id]; pr.URL != "" { if pr := f.pr[id]; pr.URL != "" {
return pr, true, nil return pr, true, nil
@ -150,7 +162,10 @@ func newManager() (*Manager, *fakeStore, *fakeRuntime, *fakeWorkspace) {
st := newFakeStore() st := newFakeStore()
rt := &fakeRuntime{} rt := &fakeRuntime{}
ws := &fakeWorkspace{} 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 return m, st, rt, ws
} }
func seedTerminal(st *fakeStore, id domain.SessionID, meta domain.SessionMetadata) { 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) { func TestSpawn_KeepsExplicitBranch(t *testing.T) {
m, st, _, _ := newManager() m, st, _, _ := newManager()
s, err := m.Spawn(ctx, ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, Branch: "feature/x"}) s, err := m.Spawn(ctx, ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, Branch: "feature/x"})

View File

@ -60,6 +60,7 @@ type PR struct {
ObservedAt sql.NullTime ObservedAt sql.NullTime
CIObservedAt sql.NullTime CIObservedAt sql.NullTime
ReviewObservedAt sql.NullTime ReviewObservedAt sql.NullTime
LastNudgeSignature string
} }
type PRCheck struct { type PRCheck struct {

View File

@ -99,16 +99,7 @@ func (q *Queries) GetDisplayPRFactsBySession(ctx context.Context, sessionID doma
} }
const getPR = `-- name: GetPR :one const getPR = `-- name: GetPR :one
SELECT 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 = ?
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 = ?
` `
func (q *Queries) GetPR(ctx context.Context, url string) (PR, error) { 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.ObservedAt,
&i.CIObservedAt, &i.CIObservedAt,
&i.ReviewObservedAt, &i.ReviewObservedAt,
&i.LastNudgeSignature,
) )
return i, err return i, err
} }
@ -178,16 +170,19 @@ func (q *Queries) GetPRClaimAndOwner(ctx context.Context, url string) (GetPRClai
return i, err 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 const listPRsBySession = `-- name: ListPRsBySession :many
SELECT 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
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 session_id = ? WHERE session_id = ?
ORDER BY updated_at DESC ORDER BY updated_at DESC
` `
@ -240,6 +235,7 @@ func (q *Queries) ListPRsBySession(ctx context.Context, sessionID domain.Session
&i.ObservedAt, &i.ObservedAt,
&i.CIObservedAt, &i.CIObservedAt,
&i.ReviewObservedAt, &i.ReviewObservedAt,
&i.LastNudgeSignature,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@ -254,6 +250,20 @@ func (q *Queries) ListPRsBySession(ctx context.Context, sessionID domain.Session
return items, nil 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 const upsertLegacyPR = `-- name: UpsertLegacyPR :exec
INSERT INTO pr ( INSERT INTO pr (
url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, 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 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
}

View File

@ -10,6 +10,20 @@ import (
"time" "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 const deletePRReviewThreads = `-- name: DeletePRReviewThreads :exec
DELETE FROM pr_review_threads WHERE pr_url = ? DELETE FROM pr_review_threads WHERE pr_url = ?
` `

View File

@ -221,6 +221,30 @@ func (q *Queries) RenameSession(ctx context.Context, arg RenameSessionParams) (i
return result.RowsAffected() 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 const updateSession = `-- name: UpdateSession :exec
UPDATE sessions SET UPDATE sessions SET
issue_id = ?, kind = ?, harness = ?, display_name = ?, issue_id = ?, kind = ?, harness = ?, display_name = ?,

View File

@ -5,3 +5,13 @@ FROM change_log WHERE seq > ? ORDER BY seq LIMIT ?;
-- name: MaxChangeLogSeq :one -- name: MaxChangeLogSeq :one
SELECT CAST(COALESCE(MAX(seq), 0) AS INTEGER) AS seq FROM change_log; 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.

View File

@ -65,27 +65,10 @@ ON CONFLICT (url) DO UPDATE SET
is_closed = excluded.is_closed; is_closed = excluded.is_closed;
-- name: GetPR :one -- name: GetPR :one
SELECT SELECT * FROM pr WHERE url = ?;
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 = ?;
-- name: ListPRsBySession :many -- name: ListPRsBySession :many
SELECT SELECT * FROM pr
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 session_id = ? WHERE session_id = ?
ORDER BY updated_at DESC; ORDER BY updated_at DESC;

View File

@ -13,6 +13,9 @@ ON CONFLICT (pr_url, thread_id) DO UPDATE SET
-- name: DeletePRReviewThreads :exec -- name: DeletePRReviewThreads :exec
DELETE FROM pr_review_threads WHERE pr_url = ?; 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 -- name: ListPRReviewThreads :many
SELECT pr_url, thread_id, path, line, resolved, is_bot, semantic_hash, updated_at 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; FROM pr_review_threads WHERE pr_url = ? ORDER BY updated_at, thread_id;

View File

@ -38,3 +38,28 @@ FROM sessions ORDER BY project_id, num;
-- name: RenameSession :execrows -- name: RenameSession :execrows
UPDATE sessions SET display_name = ?, updated_at = ? WHERE id = ?; 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 <seed-state predicates>`
-- 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.

View File

@ -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 // WritePR persists scalar facts, checks, and comments in one tx; all three
// should be queryable afterward. // should be queryable afterward.
func TestWritePR_PersistsScalarsChecksAndComments(t *testing.T) { func TestWritePR_PersistsScalarsChecksAndComments(t *testing.T) {

View File

@ -110,11 +110,6 @@ func writePRRows(ctx context.Context, q *gen.Queries, pr domain.PullRequest, che
return err return err
} }
} }
if reviewMode == ports.ReviewWriteReplace {
if err := q.DeletePRReviewThreads(ctx, pr.URL); err != nil {
return err
}
}
if reviewMode == ports.ReviewWriteReplace { if reviewMode == ports.ReviewWriteReplace {
if err := q.DeletePRComments(ctx, pr.URL); err != nil { if err := q.DeletePRComments(ctx, pr.URL); err != nil {
return err 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 { if reviewMode == ports.ReviewWriteMerge {
for _, threadID := range reviewThreadIDs(threads, comments) { for _, threadID := range reviewThreadIDs(threads, comments) {
if err := q.DeletePRCommentsByThread(ctx, gen.DeletePRCommentsByThreadParams{PRURL: pr.URL, ThreadID: threadID}); err != nil { if err := q.DeletePRCommentsByThread(ctx, gen.DeletePRCommentsByThreadParams{PRURL: pr.URL, ThreadID: threadID}); err != nil {

View File

@ -56,6 +56,80 @@ func (s *Store) RenameSession(ctx context.Context, id domain.SessionID, displayN
return rows > 0, nil 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. // 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) { func (s *Store) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) {
row, err := s.qr.GetSession(ctx, id) row, err := s.qr.GetSession(ctx, id)

View File

@ -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) { func TestSessionRenameUpdatesDisplayName(t *testing.T) {
s := newTestStore(t) s := newTestStore(t)
ctx := context.Background() ctx := context.Background()

View File

@ -247,6 +247,23 @@ export interface paths {
patch?: never; patch?: never;
trace?: 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": { "/api/v1/sessions/{sessionId}/send": {
parameters: { parameters: {
query?: never; query?: never;
@ -396,6 +413,12 @@ export interface components {
session: components["schemas"]["Session"]; session: components["schemas"]["Session"];
sessionId: string; sessionId: string;
}; };
RollbackSessionResponse: {
deleted?: boolean;
killed?: boolean;
ok: boolean;
sessionId: string;
};
SCMConfig: { SCMConfig: {
package?: string; package?: string;
path?: 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: { sendSessionMessage: {
parameters: { parameters: {
query?: never; query?: never;