diff --git a/AGENTS.md b/AGENTS.md index 4bfaa1a662..b78eebec0b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -174,6 +174,7 @@ history and remains immutable. - In dev mode (`KANDEV_MOCK_AGENT=true` or `debug.pprofEnabled`), `/debug/vars` exposes the stdlib expvar handler. Office provider-routing metrics live under `routing_*` (route attempts, fallbacks, parked runs, provider degraded/recovered counters). The metrics are also still emitted as structured `routing.metric.*` zap logs for human debugging. - ADR 0015's step-completion-signal telemetry lives under `workflow_*`: `workflow_step_completion_signal_received_total` (`internal/workflow/signalmetrics/`), labelled by `source` and `agent_type`, counts accepted `step_complete_kandev` signals. Its separate `workflow_step_completion_signal_fallback_used_total` counter is labelled by `agent_type` and counts manual fallback uses. The fallback button does not have a production increment site yet, so the counter remains zero until that UI ships. +- Office stall detection (REQ-OFFICE-STALL-VISIBILITY) lives under `office_stall_*`: `office_stall_stranded_signal_total` (labelled by `gate`, which names which of the two watchdog gate sites saw it), `office_stall_decision_waiting_total`, and `office_stall_detector_skipped_total` (labelled by `reason`, so a detector that fails closed is visible rather than silent). Detection only: these counters never accompany a transition, a synthesized decision, or a queued run, because Office tasks are surfaced and never reclaimed. Also emitted as structured zap logs. ### GitHub Operations diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index 90ed0403bb..0cc0261799 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -1623,6 +1623,10 @@ func wireWorkflowEngineForOffice( // REQ-OFFICE-REVIEW-SEATS-004.3: drop a required seat whose agent profile // was deleted after casting, rather than waiting forever on it. orchestratorSvc.SetEngineAgentProfileResolver(officeengineadapters.NewAgentProfileResolverAdapter(repos.Office)) + // REQ-OFFICE-STALL-VISIBILITY-002: the decision-waiting detector's + // false-positive guard. Without this the detector reports nothing rather + // than reporting tasks it cannot prove are idle. + orchestratorSvc.SetOfficeRunInFlightReader(repos.Office) eng := orchestratorSvc.WorkflowEngine() if eng == nil { log.Warn("workflow engine not initialised; office engine dispatcher disabled") diff --git a/apps/backend/internal/office/repository/sqlite/runs_inflight.go b/apps/backend/internal/office/repository/sqlite/runs_inflight.go new file mode 100644 index 0000000000..2b910c9a0c --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/runs_inflight.go @@ -0,0 +1,52 @@ +package sqlite + +import ( + "context" + "fmt" + + "github.com/kandev/kandev/internal/db/dialect" + "github.com/kandev/kandev/internal/office/models" +) + +// inFlightRunStatuses are the run statuses that mean work is still owed on a +// task. The runs table has no "running" status: a run is `queued` until the +// scheduler claims it and `claimed` while it executes, reaching `finished`, +// `failed` or `cancelled` only at the end. Anything outside this pair is +// terminal. +var inFlightRunStatuses = []string{ + string(models.RunStatusQueued), + string(models.RunStatusClaimed), +} + +// HasInFlightRunForTask reports whether the task has a queued or claimed run. +// +// The runs table carries no task_id column — a run's task lives in its payload +// JSON — so the predicate reads it out with dialect.JSONExtract rather than a +// literal json_extract. Postgres is a supported driver +// (internal/persistence/provider.go), and a SQLite-only expression here would +// pass every SQLite test and fail at runtime on Postgres. Precedent: +// HasPriorTasklessFailedRun in failure.go, covered by failure_postgres_test.go. +// +// Used by the Office decision-waiting detector as its false-positive guard: a +// task with work in flight is being worked on, not stalled, however long it has +// carried an undecided review seat. +func (r *Repository) HasInFlightRunForTask(ctx context.Context, taskID string) (bool, error) { + if taskID == "" { + return false, nil + } + taskIDExpr := dialect.JSONExtract(r.ro.DriverName(), "payload", "task_id") + query := fmt.Sprintf(` + SELECT EXISTS ( + SELECT 1 FROM runs + WHERE %s = ? + AND status IN (?, ?) + ) + `, taskIDExpr) + var exists bool + if err := r.ro.QueryRowxContext( + ctx, r.ro.Rebind(query), taskID, inFlightRunStatuses[0], inFlightRunStatuses[1], + ).Scan(&exists); err != nil { + return false, fmt.Errorf("check in-flight run for task: %w", err) + } + return exists, nil +} diff --git a/apps/backend/internal/office/repository/sqlite/runs_inflight_postgres_test.go b/apps/backend/internal/office/repository/sqlite/runs_inflight_postgres_test.go new file mode 100644 index 0000000000..801887bcae --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/runs_inflight_postgres_test.go @@ -0,0 +1,69 @@ +package sqlite_test + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/office/repository/sqlite" + taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/testutil" +) + +// TestPostgresHasInFlightRunForTask is the PostgreSQL twin of +// TestHasInFlightRunForTask. HasInFlightRunForTask reads task_id out of the +// runs payload, and a literal json_extract(...) there is a syntax error on +// Postgres (payload->>'task_id' is the Postgres form) — the same trap +// TestPostgresHasPriorTasklessFailedRun was written for. Running the predicate +// against a real Postgres backend makes a dialect regression fail loudly +// instead of only in production, where its failure mode is a detector that +// errors on every tick and surfaces nothing. +// Skips unless KANDEV_TEST_POSTGRES_DSN is set. +func TestPostgresHasInFlightRunForTask(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + ctx := context.Background() + + // runs is created by the task repository's schema init, mirroring + // production boot order (see failure_postgres_test.go). + if _, err := taskrepo.NewWithDB(db, db, nil); err != nil { + t.Fatalf("init task repo: %v", err) + } + repo, err := sqlite.NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init office repo: %v", err) + } + + now := time.Now().UTC() + cases := []struct { + status string + want bool + }{ + {"queued", true}, + {"claimed", true}, + {"finished", false}, + {"failed", false}, + {"cancelled", false}, + } + for _, tc := range cases { + taskID := "pg-inflight-" + tc.status + seedPostgresRun(t, ctx, repo, "pg-run-"+tc.status, "agent-a", tc.status, now, + `{"task_id":"`+taskID+`"}`) + + got, err := repo.HasInFlightRunForTask(ctx, taskID) + if err != nil { + t.Fatalf("HasInFlightRunForTask(%s): %v", tc.status, err) + } + if got != tc.want { + t.Errorf("HasInFlightRunForTask(status=%s) = %v, want %v", tc.status, got, tc.want) + } + } + + // Scoping: the in-flight rows above must not answer for a different task. + got, err := repo.HasInFlightRunForTask(ctx, "pg-inflight-absent") + if err != nil { + t.Fatalf("HasInFlightRunForTask (absent task): %v", err) + } + if got { + t.Error("got true, want false — no run carries this task_id") + } +} diff --git a/apps/backend/internal/office/repository/sqlite/runs_inflight_test.go b/apps/backend/internal/office/repository/sqlite/runs_inflight_test.go new file mode 100644 index 0000000000..2506f01bb3 --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/runs_inflight_test.go @@ -0,0 +1,102 @@ +package sqlite_test + +import ( + "context" + "testing" + "time" +) + +// HasInFlightRunForTask is the false-positive guard of the Office +// decision-waiting detector (REQ-OFFICE-STALL-VISIBILITY-002): a task with a +// queued or claimed run is being worked on, so it must never be surfaced as +// waiting on a decision. These tests pin both halves of that predicate — the +// two in-flight statuses answer true, every terminal status answers false — +// because a guard that silently answers false for `claimed` turns the whole +// detector into a false-positive generator. + +func TestHasInFlightRunForTask(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + cases := []struct { + status string + want bool + }{ + {"queued", true}, + {"claimed", true}, + {"finished", false}, + {"failed", false}, + {"cancelled", false}, + } + for _, tc := range cases { + t.Run(tc.status, func(t *testing.T) { + taskID := "task-" + tc.status + finishedAt := time.Now().UTC() + var stamp *time.Time + if !tc.want { + stamp = &finishedAt + } + seedCancelRun(t, repo, taskID, tc.status, nil, stamp) + + got, err := repo.HasInFlightRunForTask(ctx, taskID) + if err != nil { + t.Fatalf("HasInFlightRunForTask(%s): %v", tc.status, err) + } + if got != tc.want { + t.Errorf("HasInFlightRunForTask(status=%s) = %v, want %v", tc.status, got, tc.want) + } + }) + } +} + +func TestHasInFlightRunForTask_ScopedToTheNamedTask(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + seedCancelRun(t, repo, "other-task", "claimed", nil, nil) + + got, err := repo.HasInFlightRunForTask(ctx, "quiet-task") + if err != nil { + t.Fatalf("HasInFlightRunForTask: %v", err) + } + if got { + t.Error("got true, want false — the only in-flight run belongs to another task") + } +} + +func TestHasInFlightRunForTask_EmptyTaskIDIsNotInFlight(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + // An empty task ID names no task, so it must never be considered in flight. + // HasInFlightRunForTask returns false before the query runs, making this + // independent of the dialect; this test pins that early return. + seedCancelRun(t, repo, "", "queued", nil, nil) + + got, err := repo.HasInFlightRunForTask(ctx, "") + if err != nil { + t.Fatalf("HasInFlightRunForTask(\"\"): %v", err) + } + if got { + t.Error("got true, want false — an empty task ID names no task") + } +} + +func TestHasInFlightRunForTask_AnyInFlightRunCounts(t *testing.T) { + repo := newTestRepo(t) + ctx := context.Background() + + // A task that has already run once and is now running again: the terminal + // run must not mask the live one. + finishedAt := time.Now().UTC().Add(-time.Hour) + seedCancelRun(t, repo, "busy-task", "finished", nil, &finishedAt) + seedCancelRun(t, repo, "busy-task", "queued", nil, nil) + + got, err := repo.HasInFlightRunForTask(ctx, "busy-task") + if err != nil { + t.Fatalf("HasInFlightRunForTask: %v", err) + } + if !got { + t.Error("got false, want true — the task has a queued run alongside a finished one") + } +} diff --git a/apps/backend/internal/orchestrator/idle_session_reaper.go b/apps/backend/internal/orchestrator/idle_session_reaper.go index 9c228ab774..46efcf893d 100644 --- a/apps/backend/internal/orchestrator/idle_session_reaper.go +++ b/apps/backend/internal/orchestrator/idle_session_reaper.go @@ -157,7 +157,8 @@ func (r *idleSessionReaper) runLoop(ctx context.Context, tick func(ctx context.C // startIdleSessionReaper wires the reaper into Service.Start. The // reaper captures the canonical fail-closed reclaim primitive so // the per-tick work is a thin scan + loop, not a new policy. The tick also -// drives reclaimStuckSignalSessionsOnce (stuck_signal_watchdog.go) and +// drives reclaimStuckSignalSessionsOnce (stuck_signal_watchdog.go), +// detectOfficeDecisionWaitingOnce (office_decision_stall_watchdog.go) and // reapStalePendingMovesOnce (pending_move_reaper.go) — further scans // sharing this ticker rather than background goroutines of their own, // preserving the reaper's single-goroutine-owner invariant. The 30s @@ -170,6 +171,7 @@ func (s *Service) startIdleSessionReaper(ctx context.Context) { if !s.idleReaper.start(ctx, func(tickCtx context.Context) { s.reclaimIdleSessionsOnce(tickCtx) s.reclaimStuckSignalSessionsOnce(tickCtx) + s.detectOfficeDecisionWaitingOnce(tickCtx) s.reapStalePendingMovesOnce(tickCtx) }) { return diff --git a/apps/backend/internal/orchestrator/office_decision_stall_test.go b/apps/backend/internal/orchestrator/office_decision_stall_test.go new file mode 100644 index 0000000000..dc99b03092 --- /dev/null +++ b/apps/backend/internal/orchestrator/office_decision_stall_test.go @@ -0,0 +1,613 @@ +package orchestrator + +import ( + "context" + "errors" + "expvar" + "fmt" + "sync" + "testing" + "time" + + taskmodels "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/workflow/engine" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// Doubles for the decision-waiting detector. Each can be told to fail, because +// "the input could not be read" is a behaviour the detector must get right +// (fail closed, count the reason) and not merely a plumbing accident. + +type stallParticipantStore struct { + participants []engine.ParticipantInfo + err error +} + +func (f *stallParticipantStore) ListStepParticipants(_ context.Context, _, _ string) ([]engine.ParticipantInfo, error) { + return f.participants, f.err +} + +func (f *stallParticipantStore) ListTaskParticipants(_ context.Context, _ string) ([]engine.ParticipantInfo, error) { + return f.participants, f.err +} + +type stallDecisionStore struct { + decisions []engine.DecisionInfo + err error + // recorded and cleared count writes. The detector must never perform + // either: recording a decision would forge the quorum the Office gates + // exist to require, and clearing one would discard a real reviewer's. + recorded int + cleared int +} + +func (f *stallDecisionStore) ListStepDecisions(_ context.Context, _, _ string) ([]engine.DecisionInfo, error) { + return f.decisions, f.err +} + +func (f *stallDecisionStore) RecordStepDecision(_ context.Context, _ engine.DecisionInfo) error { + f.recorded++ + return nil +} + +func (f *stallDecisionStore) ClearStepDecisions(_ context.Context, _, _ string) (int64, error) { + f.cleared++ + return 0, nil +} + +// stallRunQueue counts QueueRun calls so "the detector queues no run" +// (AC-003.2) is an assertion rather than an appeal to reading the source. +type stallRunQueue struct { + calls int +} + +func (q *stallRunQueue) QueueRun(_ context.Context, _ engine.QueueRunRequest) (engine.QueueOutcome, error) { + q.calls++ + return engine.QueueOutcomeQueued, nil +} + +type stallRunReader struct { + inFlight bool + err error + calls int +} + +func (f *stallRunReader) HasInFlightRunForTask(_ context.Context, _ string) (bool, error) { + f.calls++ + return f.inFlight, f.err +} + +func decidingSeat() []engine.ParticipantInfo { + return []engine.ParticipantInfo{{ + ID: "seat-1", + StepID: "step1", + Role: string(wfmodels.ParticipantRoleReviewer), + AgentProfileID: "agent-1", + DecisionRequired: true, + }} +} + +func officeDecisionWaitingCount() int64 { + v := officeStallDecisionWaitingTotal.Get(officeStallLabel("detector", "decision_waiting")) + iv, ok := v.(*expvar.Int) + if !ok { + return 0 + } + return iv.Value() +} + +func officeStallSkipCount(reason string) int64 { + v := officeStallDetectorSkippedTotal.Get(officeStallLabel("reason", reason)) + iv, ok := v.(*expvar.Int) + if !ok { + return 0 + } + return iv.Value() +} + +// seedDecisionWaitingTask makes an existing task look Office-owned, seats a +// decision-required reviewer at its step, and backdates the task past the +// threshold — the three facts the candidate query selects on. +func seedDecisionWaitingTask(t *testing.T, repo *sqliterepo.Repository, taskID, stepID string) { + t.Helper() + ctx := context.Background() + + // IsFromOffice is a derived projection, not a stored column, so take the + // project_id branch rather than setting the struct field (which the write + // path discards). Same reason as seedOfficeStrandedSignal. + if _, err := repo.DB().ExecContext(ctx, + `UPDATE tasks SET project_id = 'proj1' WHERE id = ?`, taskID); err != nil { + t.Fatalf("mark task as office: %v", err) + } + if _, err := repo.DB().ExecContext(ctx, ` + INSERT INTO workflow_step_participants + (id, step_id, task_id, role, agent_profile_id, decision_required, position) + VALUES ('seat-'||?, ?, '', 'reviewer', 'agent-1', 1, 0) + `, taskID, stepID); err != nil { + t.Fatalf("seed decision seat: %v", err) + } + quiet := time.Now().UTC().Add(-2 * officeDecisionWaitingThreshold) + if _, err := repo.DB().ExecContext(ctx, + `UPDATE tasks SET updated_at = ? WHERE id = ?`, quiet, taskID); err != nil { + t.Fatalf("backdate task: %v", err) + } + + task, err := repo.GetTask(ctx, taskID) + if err != nil { + t.Fatalf("get task: %v", err) + } + if !task.IsFromOffice { + t.Fatalf("precondition: task %s should read as an Office task", taskID) + } +} + +// decisionWaitingService builds a Service with the detector's three +// dependencies wired to the supplied doubles. +func decisionWaitingService( + t *testing.T, repo *sqliterepo.Repository, + participants *stallParticipantStore, decisions *stallDecisionStore, runs *stallRunReader, +) *Service { + t.Helper() + svc := createTestService(repo, officeStallStepGetter(), newMockTaskRepo()) + if participants != nil { + svc.SetEngineParticipantStore(participants) + } + if decisions != nil { + svc.SetEngineDecisionStore(decisions) + } + if runs != nil { + svc.SetOfficeRunInFlightReader(runs) + } + return svc +} + +// AC-002.1: an Office task past the threshold with an undecided seat and no +// run in flight is surfaced. +func TestDetectOfficeDecisionWaitingOnce_SurfacesWaitingTask(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + runs := &stallRunReader{} + decisions := &stallDecisionStore{} + queue := &stallRunQueue{} + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, decisions, runs) + svc.SetEngineRunQueue(queue) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - before; got != 1 { + t.Fatalf("decision-waiting counter delta = %d, want 1", got) + } + if runs.calls != 1 { + t.Fatalf("run reader called %d times, want 1", runs.calls) + } + assertOfficeDecisionNotActedOn(t, repo, "t1", "step1", decisions, queue) +} + +// A bounded page must not make the oldest candidates a permanent exclusion +// list. Every candidate past the first page still needs one evaluation during +// a scan pass. +func TestDetectOfficeDecisionWaitingOnce_ProcessesCandidatesBeyondFirstPage(t *testing.T) { + const candidateCount = 201 + repo := setupTestRepo(t) + seedSession(t, repo, "task-000", "s-seed", "step1") + seedDecisionWaitingTask(t, repo, "task-000", "step1") + for i := 1; i < candidateCount; i++ { + taskID := fmt.Sprintf("task-%03d", i) + now := time.Now().UTC() + if err := repo.CreateTask(context.Background(), &taskmodels.Task{ + ID: taskID, + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Title: taskID, + State: v1.TaskStateInProgress, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("create task %s: %v", taskID, err) + } + seedDecisionWaitingTask(t, repo, taskID, "step1") + } + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, &stallDecisionStore{}, + &stallRunReader{}) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - before; got != candidateCount { + t.Fatalf("decision-waiting counter delta = %d, want %d", got, candidateCount) + } + if _, seen := svc.officeDecisionWaiting.Load(officeDecisionWaitingKey("task-200", "step1")); !seen { + t.Fatal("candidate beyond the first page was not evaluated") + } +} + +// The Office dependency readers are wired after Service.Start. The detector +// can run during that interval, so late wiring and detector reads must be race +// free. +func TestOfficeDecisionWaitingLateRunReaderWiringIsRaceFree(t *testing.T) { + repo := setupTestRepo(t) + seedTaskWithoutSession(t, repo, "t1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, &stallDecisionStore{}, + &stallRunReader{}) + ctx := context.Background() + + const iterations = 1000 + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + <-start + for i := 0; i < iterations; i++ { + svc.detectOfficeDecisionWaitingOnce(ctx) + } + }() + go func() { + defer wg.Done() + close(start) + for i := 0; i < iterations; i++ { + svc.SetOfficeRunInFlightReader(&stallRunReader{inFlight: i%2 == 0}) + } + }() + wg.Wait() +} + +// AC-002.2: the false-positive guard. A task with a claimed run is being +// worked on, so it must not be reported however long its seat has gone +// undecided. This is the case named in the definition of done. +func TestDetectOfficeDecisionWaitingOnce_ClaimedRunIsNotSurfaced(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, &stallDecisionStore{}, + &stallRunReader{inFlight: true}) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - before; got != 0 { + t.Fatalf("counter delta = %d, want 0 — a task with a run in flight is healthy", got) + } +} + +// AC-002.3: a decision already recorded for the current step suppresses +// surfacing. +func TestDetectOfficeDecisionWaitingOnce_RecordedDecisionSuppresses(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + runs := &stallRunReader{} + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, + &stallDecisionStore{decisions: []engine.DecisionInfo{{ + ID: "d1", TaskID: "t1", StepID: "step1", Decision: "approve", + }}}, + runs) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - before; got != 0 { + t.Fatalf("counter delta = %d, want 0 — the step already has a decision", got) + } + if runs.calls != 0 { + t.Fatalf("run reader called %d times; the decision check must reject first", runs.calls) + } +} + +// AC-002.3 (superseded case): ListStepDecisions returns superseded rows +// alongside active ones (its own documented contract), so a step whose only +// decision is superseded must still read as undecided. maybeSupersedeOnRework +// leaves the row in place with superseded_at set rather than deleting it, so +// treating any non-empty result as "decided" would permanently suppress the +// repeat stall this detector exists to find. +func TestDetectOfficeDecisionWaitingOnce_SupersededDecisionDoesNotSuppress(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + runs := &stallRunReader{} + supersededAt := time.Now().UTC().Add(-time.Hour) + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, + &stallDecisionStore{decisions: []engine.DecisionInfo{{ + ID: "d1", TaskID: "t1", StepID: "step1", Decision: "approve", + SupersededAt: &supersededAt, + }}}, + runs) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - before; got != 1 { + t.Fatalf("counter delta = %d, want 1 — a superseded-only decision set is undecided", got) + } + if runs.calls != 1 { + t.Fatalf("run reader called %d times, want 1 — the superseded row must not short-circuit the check", runs.calls) + } +} + +// AC-002.4: seats that owe no decision cannot strand a task. +func TestDetectOfficeDecisionWaitingOnce_NonDecidingSeatsAreNotSurfaced(t *testing.T) { + cases := []struct { + name string + seat engine.ParticipantInfo + }{ + {"watcher", engine.ParticipantInfo{Role: string(wfmodels.ParticipantRoleWatcher), DecisionRequired: true}}, + {"collaborator", engine.ParticipantInfo{Role: string(wfmodels.ParticipantRoleCollaborator), DecisionRequired: true}}, + {"runner", engine.ParticipantInfo{Role: string(wfmodels.ParticipantRoleRunner), DecisionRequired: true}}, + {"reviewer not required", engine.ParticipantInfo{Role: string(wfmodels.ParticipantRoleReviewer), DecisionRequired: false}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + runs := &stallRunReader{} + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: []engine.ParticipantInfo{tc.seat}}, + &stallDecisionStore{}, runs) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - before; got != 0 { + t.Fatalf("counter delta = %d, want 0 — a %s seat owes no decision", got, tc.name) + } + if runs.calls != 0 { + t.Fatalf("run reader called %d times; the seat check must reject first", runs.calls) + } + }) + } +} + +// AC-002.5: every unreadable input fails closed and is counted, so a silently +// degraded detector is distinguishable from a quiet system. +func TestDetectOfficeDecisionWaitingOnce_UnreadableInputsFailClosed(t *testing.T) { + readErr := errors.New("boom") + cases := []struct { + name string + participants *stallParticipantStore + decisions *stallDecisionStore + runs *stallRunReader + reason string + }{ + { + name: "participant store unwired", + runs: &stallRunReader{}, + reason: officeStallSkipParticipantStore, + }, + { + name: "decision store unwired", + participants: &stallParticipantStore{participants: decidingSeat()}, + runs: &stallRunReader{}, + reason: officeStallSkipDecisionStore, + }, + { + name: "participant read failed", + participants: &stallParticipantStore{err: readErr}, + decisions: &stallDecisionStore{}, + runs: &stallRunReader{}, + reason: officeStallSkipParticipantReadFailed, + }, + { + name: "decision read failed", + participants: &stallParticipantStore{participants: decidingSeat()}, + decisions: &stallDecisionStore{err: readErr}, + runs: &stallRunReader{}, + reason: officeStallSkipDecisionReadFailed, + }, + { + name: "run reader unwired", + participants: &stallParticipantStore{participants: decidingSeat()}, + decisions: &stallDecisionStore{}, + reason: officeStallSkipRunReaderUnwired, + }, + { + name: "run reader errored", + participants: &stallParticipantStore{participants: decidingSeat()}, + decisions: &stallDecisionStore{}, + runs: &stallRunReader{err: readErr}, + reason: officeStallSkipRunReaderError, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + svc := decisionWaitingService(t, repo, tc.participants, tc.decisions, tc.runs) + + beforeSurfaced := officeDecisionWaitingCount() + beforeSkipped := officeStallSkipCount(tc.reason) + svc.detectOfficeDecisionWaitingOnce(context.Background()) + + if got := officeDecisionWaitingCount() - beforeSurfaced; got != 0 { + t.Errorf("counter delta = %d, want 0 — an unreadable input must not surface", got) + } + if got := officeStallSkipCount(tc.reason) - beforeSkipped; got != 1 { + t.Errorf("skip counter %q delta = %d, want 1", tc.reason, got) + } + }) + } +} + +// AC-002.6: a non-Office task is never evaluated. The exclusion here is the +// mirror image of the stranded-signal path's: there, Office must not be +// excluded from detection; here, non-Office must not be included. +func TestDetectOfficeDecisionWaitingOnce_NonOfficeTaskIsNeverEvaluated(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + ctx := context.Background() + // Everything a candidate needs except Office ownership. + if _, err := repo.DB().ExecContext(ctx, ` + INSERT INTO workflow_step_participants + (id, step_id, task_id, role, agent_profile_id, decision_required, position) + VALUES ('seat-kanban', 'step1', '', 'reviewer', 'agent-1', 1, 0) + `); err != nil { + t.Fatalf("seed decision seat: %v", err) + } + quiet := time.Now().UTC().Add(-2 * officeDecisionWaitingThreshold) + if _, err := repo.DB().ExecContext(ctx, + `UPDATE tasks SET updated_at = ? WHERE id = 't1'`, quiet); err != nil { + t.Fatalf("backdate task: %v", err) + } + task, err := repo.GetTask(ctx, "t1") + if err != nil { + t.Fatalf("get task: %v", err) + } + if task.IsFromOffice { + t.Fatalf("precondition: t1 must read as a kanban task") + } + + runs := &stallRunReader{} + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, &stallDecisionStore{}, runs) + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(ctx) + + if got := officeDecisionWaitingCount() - before; got != 0 { + t.Fatalf("counter delta = %d, want 0 — a kanban task is out of scope", got) + } + if runs.calls != 0 { + t.Fatalf("run reader called %d times for a kanban task", runs.calls) + } +} + +// A still-waiting task is reported once, not on every 30-second tick. +func TestDetectOfficeDecisionWaitingOnce_ReportsOncePerTaskAndStep(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, &stallDecisionStore{}, + &stallRunReader{}) + + before := officeDecisionWaitingCount() + for range 3 { + svc.detectOfficeDecisionWaitingOnce(context.Background()) + } + if got := officeDecisionWaitingCount() - before; got != 1 { + t.Fatalf("counter delta across three scans = %d, want 1", got) + } +} + +// A task that stops being a candidate is pruned from the dedupe map, so if it +// stalls again later it is reported again rather than silently swallowed. +func TestDetectOfficeDecisionWaitingOnce_PrunesResolvedTasks(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, &stallDecisionStore{}, + &stallRunReader{}) + ctx := context.Background() + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(ctx) + + // The reviewer finally acts: the task row is touched, so it is no longer + // quiet and drops out of the candidate set. + if _, err := repo.DB().ExecContext(ctx, + `UPDATE tasks SET updated_at = ? WHERE id = 't1'`, time.Now().UTC()); err != nil { + t.Fatalf("touch task: %v", err) + } + svc.detectOfficeDecisionWaitingOnce(ctx) + if _, seen := svc.officeDecisionWaiting.Load(officeDecisionWaitingKey("t1", "step1")); seen { + t.Fatal("dedupe entry survived a scan in which the task was no longer a candidate") + } + + // It stalls again: a new report, not a swallowed repeat. + quiet := time.Now().UTC().Add(-2 * officeDecisionWaitingThreshold) + if _, err := repo.DB().ExecContext(ctx, + `UPDATE tasks SET updated_at = ? WHERE id = 't1'`, quiet); err != nil { + t.Fatalf("re-backdate task: %v", err) + } + svc.detectOfficeDecisionWaitingOnce(ctx) + if got := officeDecisionWaitingCount() - before; got != 2 { + t.Fatalf("counter delta = %d, want 2 — a fresh stall is a fresh report", got) + } +} + +// A task whose decision is recorded, then superseded by a fresh requirement, +// must be reported again even though tasks.updated_at never moves and the +// task never leaves the cheap SQL candidate window: recording a decision +// must clear the dedupe key, not merely suppress the tick it was recorded on +// (regression for the "live" set tracking the SQL prefilter instead of the +// full predicate). +func TestDetectOfficeDecisionWaitingOnce_RestallAfterDecisionIsReportedAgain(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedDecisionWaitingTask(t, repo, "t1", "step1") + decisions := &stallDecisionStore{} + svc := decisionWaitingService(t, repo, + &stallParticipantStore{participants: decidingSeat()}, decisions, + &stallRunReader{}) + ctx := context.Background() + + before := officeDecisionWaitingCount() + svc.detectOfficeDecisionWaitingOnce(ctx) + if got := officeDecisionWaitingCount() - before; got != 1 { + t.Fatalf("first scan counter delta = %d, want 1", got) + } + + // The reviewer decides. tasks.updated_at is untouched, so the task stays + // in the cheap SQL candidate window even though the full predicate no + // longer holds. + decisions.decisions = []engine.DecisionInfo{{ + ID: "d1", TaskID: "t1", StepID: "step1", Decision: "reject", + }} + svc.detectOfficeDecisionWaitingOnce(ctx) + if _, seen := svc.officeDecisionWaiting.Load(officeDecisionWaitingKey("t1", "step1")); seen { + t.Fatal("dedupe entry survived a scan in which the decision was recorded") + } + + // Rework: a fresh requirement supersedes the old decision, still at the + // same step, still with the same stale tasks.updated_at. + supersededAt := time.Now().UTC() + decisions.decisions = []engine.DecisionInfo{{ + ID: "d1", TaskID: "t1", StepID: "step1", Decision: "reject", + SupersededAt: &supersededAt, + }} + svc.detectOfficeDecisionWaitingOnce(ctx) + if got := officeDecisionWaitingCount() - before; got != 2 { + t.Fatalf("counter delta after rework = %d, want 2 — the rework stall must be reported again", got) + } +} + +// AC-003.1 through AC-003.3: surfacing is the entire action. No decision row, +// no queued run, no step transition — forging any of those would manufacture +// the quorum the Office gates exist to require. +func assertOfficeDecisionNotActedOn( + t *testing.T, repo *sqliterepo.Repository, taskID, stepID string, + decisions *stallDecisionStore, queue *stallRunQueue, +) { + t.Helper() + + task, err := repo.GetTask(context.Background(), taskID) + if err != nil { + t.Fatalf("get task: %v", err) + } + if task.WorkflowStepID != stepID { + t.Fatalf("workflow step moved to %q; the detector applied a transition", task.WorkflowStepID) + } + if decisions.recorded != 0 { + t.Fatalf("detector recorded %d decisions; it must never manufacture a quorum", decisions.recorded) + } + if decisions.cleared != 0 { + t.Fatalf("detector cleared decisions %d times; it must never discard a real one", decisions.cleared) + } + if queue.calls != 0 { + t.Fatalf("detector queued %d runs; surfacing is the whole action", queue.calls) + } +} diff --git a/apps/backend/internal/orchestrator/office_decision_stall_watchdog.go b/apps/backend/internal/orchestrator/office_decision_stall_watchdog.go new file mode 100644 index 0000000000..b8a5191044 --- /dev/null +++ b/apps/backend/internal/orchestrator/office_decision_stall_watchdog.go @@ -0,0 +1,288 @@ +package orchestrator + +import ( + "context" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/workflow/engine" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// Office decision-waiting detector. +// +// The stuck-signal watchdog next door only sees a task whose session accepted +// a completion signal. It cannot see the other Office stall: a task parked at +// a review or approval step where the seat that owes a decision never records +// one. Nothing is running, nothing failed, no signal is pending — the card +// simply stops moving, and the only present symptom is the absence of an +// event. This pass detects that state and reports it. +// +// It never acts on what it finds (REQ-OFFICE-STALL-VISIBILITY-003). Advancing +// the task would mean manufacturing the decision the quorum gate is there to +// require, so this file contains no write to workflow_step_decisions, no +// queued run, and no step transition — only a log line and a counter. That +// restraint is the feature, not a limitation of it. +const ( + // officeDecisionWaitingThreshold is how long an Office task may sit at a + // step with an undecided decision-required seat before the detector + // reports it. + // + // PROVISIONAL. 60 minutes is a first guess, not a measurement: a human + // reviewer who steps away for ten minutes is behaving normally, and an + // agent reviewer that has genuinely died is equally stuck at ten minutes + // or at sixty, so the cost of waiting is only reporting latency while the + // cost of being early is a log full of healthy tasks. Re-tune once the + // counter has produced real distribution data. + // + // Deliberately NOT stuckSignalWatchdogThreshold. The two constants + // measure unrelated phenomena — one an agent process that stopped + // emitting events mid-turn, the other a decision nobody has made yet — + // and sharing the value would make a future tuning of session silence + // silently retune this, in the wrong direction and with no test to catch + // it. + officeDecisionWaitingThreshold = 60 * time.Minute +) + +// officeDecisionCandidateLister is implemented by the durable task repository. +// A narrow optional interface, mirroring stuckSignalSessionLister, so test +// doubles that do not exercise this scan stay compatible. +type officeDecisionCandidateLister interface { + ListOfficeDecisionWaitCandidates( + ctx context.Context, + quietSince time.Time, + after *models.OfficeDecisionWaitCursor, + ) ([]models.OfficeDecisionWaitCandidate, *models.OfficeDecisionWaitCursor, error) +} + +// officeRunInFlightReader answers whether a task still has a queued or claimed +// run. It is the detector's false-positive guard: a task with work in flight +// is being worked on, not stalled, however long its seat has gone undecided. +// Implemented by the Office SQLite repository (HasInFlightRunForTask) and +// wired from internal/backendapp, where that repository is constructed; the +// orchestrator does not otherwise reach the runs table, so this stays an +// injected dependency rather than a repo cast. +type officeRunInFlightReader interface { + HasInFlightRunForTask(ctx context.Context, taskID string) (bool, error) +} + +// SetOfficeRunInFlightReader wires the run-queue read the Office +// decision-waiting detector needs. Until it is wired the detector reports +// nothing and counts every evaluation as skipped, rather than reporting +// tasks it cannot prove are idle. +func (s *Service) SetOfficeRunInFlightReader(reader officeRunInFlightReader) { + s.officeStallDependenciesMu.Lock() + s.officeRunInFlight = reader + s.officeStallDependenciesMu.Unlock() +} + +// officeDecisionWaitingKey identifies one observation for dedupe purposes. +// Keyed by task and step so a task that moves to another decision step — or +// returns to this one after a rejection round — is reported again, while a +// task sitting still is reported once. +func officeDecisionWaitingKey(taskID, stepID string) string { + return taskID + "\x00" + stepID +} + +// detectOfficeDecisionWaitingOnce is the per-tick scan, called from the +// idle-session reaper's existing ticker alongside reclaimStuckSignalSessionsOnce +// rather than owning a second background goroutine. It walks bounded keyset +// pages until the scan budget expires or the candidate set is complete. +func (s *Service) detectOfficeDecisionWaitingOnce(ctx context.Context) { + lister, ok := s.repo.(officeDecisionCandidateLister) + if !ok { + officeStallSkipped(officeStallSkipCandidateListUnwired) + return + } + now := time.Now().UTC() + scanCtx, cancel := context.WithTimeout(ctx, stuckSignalScanBudget) + defer cancel() + live := make(map[string]struct{}) + var cursor *models.OfficeDecisionWaitCursor + scanned := 0 + for { + if scanCtx.Err() != nil { + s.logOfficeDecisionScanBudgetExceeded(scanned) + return + } + candidates, next, err := lister.ListOfficeDecisionWaitCandidates( + scanCtx, now.Add(-officeDecisionWaitingThreshold), cursor, + ) + if err != nil { + if scanCtx.Err() != nil { + s.logOfficeDecisionScanBudgetExceeded(scanned) + return + } + officeStallSkipped(officeStallSkipCandidateListFailed) + s.logger.Warn("office stall watchdog: listing decision-wait candidates failed; tick skipped", + zap.Error(err)) + return + } + for _, candidate := range candidates { + if scanCtx.Err() != nil { + s.logOfficeDecisionScanBudgetExceeded(scanned) + return + } + scanned++ + if s.evaluateOfficeDecisionWaiting(scanCtx, candidate, now) { + live[officeDecisionWaitingKey(candidate.TaskID, candidate.StepID)] = struct{}{} + } + } + if next == nil || len(candidates) == 0 { + break + } + cursor = next + } + s.pruneOfficeDecisionWaiting(live) +} + +func (s *Service) logOfficeDecisionScanBudgetExceeded(scanned int) { + s.logger.Warn("office stall watchdog: decision-wait scan budget exceeded; deferring remaining candidates to the next tick", + zap.Int("candidates_scanned", scanned)) +} + +// evaluateOfficeDecisionWaiting applies the remaining predicate to one +// candidate, reports it if every clause holds, and reports whether the +// candidate should stay live for pruning purposes. The order is deliberate: +// the cheapest, most-selective reads run first and the runs-table read runs +// last, because most candidates are rejected before reaching it. +// +// Live must track the full predicate, not just the cheap SQL prefilter that +// produced the candidate: the query has no way to see a recorded decision or +// an in-flight run, so a resolved task keeps matching it for as long as +// tasks.updated_at stays quiet. Marking it live from the SQL match alone +// would keep its dedupe key past resolution, and a rework round that +// re-stalls the same task at the same step — without ever touching +// updated_at — would then find the old key already present and be silently +// swallowed by surfaceOfficeDecisionWaiting's LoadOrStore, contradicting +// officeDecisionWaitingKey's own "returns to this one after a rejection +// round — is reported again" contract. +// +// Every unreadable input fails closed to "do not report" and is counted. A +// detector that guessed on missing input would report healthy tasks, and the +// only thing this feature produces is reports. Failing closed here also +// means an unreadable input is not live, so a transient read failure costs at +// most one duplicate report on a later successful tick rather than risking a +// permanently swallowed one. +func (s *Service) evaluateOfficeDecisionWaiting( + ctx context.Context, candidate models.OfficeDecisionWaitCandidate, now time.Time, +) bool { + if !s.officeStepAwaitsDecision(ctx, candidate) { + return false + } + _, _, reader := s.officeStallDependencies() + if reader == nil { + officeStallSkipped(officeStallSkipRunReaderUnwired) + return false + } + inFlight, err := reader.HasInFlightRunForTask(ctx, candidate.TaskID) + if err != nil { + officeStallSkipped(officeStallSkipRunReaderError) + return false + } + if inFlight { + return false + } + s.surfaceOfficeDecisionWaiting(candidate, now) + return true +} + +// officeStepAwaitsDecision reports whether the candidate's current step holds +// a decision-required reviewer or approver seat with no decision recorded +// against it. Both stores are required: without them the detector cannot tell +// a genuinely undecided step from one it simply could not read. +func (s *Service) officeStepAwaitsDecision( + ctx context.Context, candidate models.OfficeDecisionWaitCandidate, +) bool { + participantsStore, decisionsStore, _ := s.officeStallDependencies() + if participantsStore == nil { + officeStallSkipped(officeStallSkipParticipantStore) + return false + } + if decisionsStore == nil { + officeStallSkipped(officeStallSkipDecisionStore) + return false + } + participants, err := participantsStore.ListStepParticipants(ctx, candidate.StepID, candidate.TaskID) + if err != nil { + officeStallSkipped(officeStallSkipParticipantReadFailed) + return false + } + if !hasDecidingSeat(participants) { + return false + } + decisions, err := decisionsStore.ListStepDecisions(ctx, candidate.TaskID, candidate.StepID) + if err != nil { + officeStallSkipped(officeStallSkipDecisionReadFailed) + return false + } + return !hasActiveDecision(decisions) +} + +// hasActiveDecision reports whether any row is not superseded. +// ListStepDecisions returns superseded rows alongside active ones (see +// engine.DecisionInfo), and a rework round leaves the superseded row in +// place rather than deleting it, so a re-entered step must be judged on +// active rows only or it reads as permanently decided. +func hasActiveDecision(decisions []engine.DecisionInfo) bool { + for _, d := range decisions { + if d.SupersededAt == nil { + return true + } + } + return false +} + +// hasDecidingSeat reports whether any seat at the step owes a decision. +// Watcher, collaborator and runner seats never do, and a reviewer seat with +// DecisionRequired false is an observer — none of them can strand a task, so +// none of them makes it a candidate. +func hasDecidingSeat(participants []engine.ParticipantInfo) bool { + for _, p := range participants { + if !p.DecisionRequired { + continue + } + switch wfmodels.ParticipantRole(p.Role) { + case wfmodels.ParticipantRoleReviewer, wfmodels.ParticipantRoleApprover: + return true + } + } + return false +} + +// surfaceOfficeDecisionWaiting reports one decision-waiting task exactly once +// per (task, step). Reporting only — see this file's header for why there is +// no recovery here. +func (s *Service) surfaceOfficeDecisionWaiting( + candidate models.OfficeDecisionWaitCandidate, now time.Time, +) { + key := officeDecisionWaitingKey(candidate.TaskID, candidate.StepID) + if _, seen := s.officeDecisionWaiting.LoadOrStore(key, struct{}{}); seen { + return + } + officeStallDecisionWaitingTotal.Add(officeStallLabel("detector", "decision_waiting"), 1) + s.logger.Warn("office stall watchdog: Office task is waiting on a decision that has not been recorded", + zap.String("task_id", candidate.TaskID), + zap.String("step_id", candidate.StepID), + zap.Time("task_updated_at", candidate.UpdatedAt), + zap.Duration("quiet_for", now.Sub(candidate.UpdatedAt)), + zap.Duration("threshold", officeDecisionWaitingThreshold)) +} + +// pruneOfficeDecisionWaiting drops dedupe entries for tasks that are no longer +// candidates, so the map stays bounded by the number of currently-waiting +// tasks rather than growing for the process's lifetime. A task that becomes a +// candidate again after being pruned is correctly reported again: it is a new +// stall, not a repeat of the old one. +func (s *Service) pruneOfficeDecisionWaiting(live map[string]struct{}) { + s.officeDecisionWaiting.Range(func(key, _ any) bool { + if k, ok := key.(string); ok { + if _, still := live[k]; !still { + s.officeDecisionWaiting.Delete(k) + } + } + return true + }) +} diff --git a/apps/backend/internal/orchestrator/office_stall_metrics.go b/apps/backend/internal/orchestrator/office_stall_metrics.go new file mode 100644 index 0000000000..b8d3776ebb --- /dev/null +++ b/apps/backend/internal/orchestrator/office_stall_metrics.go @@ -0,0 +1,61 @@ +package orchestrator + +import ( + "expvar" + "strings" +) + +// expvar maps for the Office stall detectors, exposed via stdlib's +// /debug/vars handler. The label model is "key=value;key=value..." so a +// Prometheus translation layer can split on `;` and `=` later, mirroring +// internal/office/scheduler/metrics_vars.go. +// +// Counters only. A consumer that wants a snapshot of what is currently stuck +// re-runs the detectors' own predicates against the database; these record +// events. +var ( + // officeStallStrandedSignalTotal counts Office tasks observed holding a + // stranded step-completion signal. Labelled by `gate` so it is visible + // which of the two watchdog sites observed it. + officeStallStrandedSignalTotal = expvar.NewMap("office_stall_stranded_signal_total") + + // officeStallDecisionWaitingTotal counts Office tasks observed parked at + // a decision-required step with no run in flight. + officeStallDecisionWaitingTotal = expvar.NewMap("office_stall_decision_waiting_total") + + // officeStallDetectorSkippedTotal counts evaluations abandoned because an + // input could not be read. Both detectors fail closed, so without this a + // silently degraded detector is indistinguishable from a quiet system — + // which is the failure mode that matters for a surface-only capability. + officeStallDetectorSkippedTotal = expvar.NewMap("office_stall_detector_skipped_total") +) + +// Skip reasons for officeStallDetectorSkippedTotal. +const ( + officeStallSkipRunReaderUnwired = "run_reader_unwired" + officeStallSkipRunReaderError = "run_reader_error" + officeStallSkipParticipantStore = "participant_store_unwired" + officeStallSkipDecisionStore = "decision_store_unwired" + officeStallSkipParticipantReadFailed = "participant_read_failed" + officeStallSkipDecisionReadFailed = "decision_read_failed" + officeStallSkipCandidateListUnwired = "candidate_lister_unwired" + officeStallSkipCandidateListFailed = "candidate_list_failed" +) + +// officeStallLabel builds a "k1=v1;k2=v2;..." expvar map key. Returns an +// empty label for an odd number of arguments rather than guessing. +func officeStallLabel(pairs ...string) string { + if len(pairs)%2 != 0 { + return "" + } + parts := make([]string, 0, len(pairs)/2) + for i := 0; i < len(pairs); i += 2 { + parts = append(parts, pairs[i]+"="+pairs[i+1]) + } + return strings.Join(parts, ";") +} + +// officeStallSkipped records a fail-closed skip from either detector. +func officeStallSkipped(reason string) { + officeStallDetectorSkippedTotal.Add(officeStallLabel("reason", reason), 1) +} diff --git a/apps/backend/internal/orchestrator/office_stall_visibility_test.go b/apps/backend/internal/orchestrator/office_stall_visibility_test.go new file mode 100644 index 0000000000..24703d15cc --- /dev/null +++ b/apps/backend/internal/orchestrator/office_stall_visibility_test.go @@ -0,0 +1,328 @@ +package orchestrator + +import ( + "context" + "expvar" + "testing" + "time" + + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// officeStallStepGetter builds a two-step workflow whose first step would +// auto-advance on turn completion. If the Office exclusion ever regressed, the +// task would move to step2 and the not-acted-on assertions would fail. +func officeStallStepGetter() *mockStepGetter { + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + AutoAdvanceRequiresSignal: true, + Events: wfmodels.StepEvents{ + OnTurnComplete: []wfmodels.OnTurnCompleteAction{ + {Type: wfmodels.OnTurnCompleteMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + return stepGetter +} + +// The Office exclusion at both watchdog gates is load-bearing: passing the +// reconcileWaitingStuckSignalSessionIfDue gate falls through to +// reconcileStepCompletionSignalLocked, and passing the stuckSignalCandidate +// gate leads to reclaimStuckSignalSessionOwned. Either one applies the step +// transition the agent asked for, which for an Office task would record it +// without the decisions its quorum gate requires. These tests pin the +// disposition that keeps that from happening. +func TestStuckSignalDispositionFor(t *testing.T) { + const stepID = "step1" + signal := models.PendingStepCompletionSignal{StepID: stepID} + + tests := []struct { + name string + task *models.Task + session *models.TaskSession + want stuckSignalDisposition + }{ + { + name: "ordinary task at the signalled step is reclaimable", + task: &models.Task{ID: "t1", WorkflowStepID: stepID}, + session: &models.TaskSession{ID: "s1"}, + want: stuckSignalReclaimable, + }, + { + name: "office task is surface-only, never reclaimable", + task: &models.Task{ID: "t1", WorkflowStepID: stepID, IsFromOffice: true}, + session: &models.TaskSession{ID: "s1"}, + want: stuckSignalSurfaceOnly, + }, + { + name: "passthrough session is excluded even when the task is office", + task: &models.Task{ID: "t1", WorkflowStepID: stepID, IsFromOffice: true}, + session: &models.TaskSession{ID: "s1", IsPassthrough: true}, + want: stuckSignalNotCandidate, + }, + { + name: "passthrough session is excluded on an ordinary task", + task: &models.Task{ID: "t1", WorkflowStepID: stepID}, + session: &models.TaskSession{ID: "s1", IsPassthrough: true}, + want: stuckSignalNotCandidate, + }, + { + name: "stale signal on an office task is not reported as a stall", + task: &models.Task{ID: "t1", WorkflowStepID: "step2", IsFromOffice: true}, + session: &models.TaskSession{ID: "s1"}, + want: stuckSignalNotCandidate, + }, + { + name: "stale signal on an ordinary task is left to the signal-consuming paths", + task: &models.Task{ID: "t1", WorkflowStepID: "step2"}, + session: &models.TaskSession{ID: "s1"}, + want: stuckSignalNotCandidate, + }, + { + name: "nil task", + task: nil, + session: &models.TaskSession{ID: "s1"}, + want: stuckSignalNotCandidate, + }, + { + name: "nil session", + task: &models.Task{ID: "t1", WorkflowStepID: stepID}, + session: nil, + want: stuckSignalNotCandidate, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := stuckSignalDispositionFor(tc.task, tc.session, signal); got != tc.want { + t.Fatalf("disposition = %v, want %v", got, tc.want) + } + }) + } +} + +// An Office task must never be classified reclaimable, whatever else is true +// of it. This is the invariant the whole unit rests on; if it ever fails, the +// watchdog can forge a quorum decision. +func TestStuckSignalDispositionFor_OfficeIsNeverReclaimable(t *testing.T) { + const stepID = "step1" + signal := models.PendingStepCompletionSignal{ + StepID: stepID, + SignaledAt: time.Now().UTC().Add(-30 * time.Minute), + } + for _, passthrough := range []bool{false, true} { + task := &models.Task{ID: "t1", WorkflowStepID: stepID, IsFromOffice: true} + session := &models.TaskSession{ID: "s1", IsPassthrough: passthrough} + if got := stuckSignalDispositionFor(task, session, signal); got == stuckSignalReclaimable { + t.Fatalf("office task classified reclaimable (passthrough=%v)", passthrough) + } + } +} + +// seedOfficeStrandedSignal creates an Office task holding a step-completion +// signal older than the watchdog threshold, in the given session state. +func seedOfficeStrandedSignal( + t *testing.T, svc *Service, repo *sqliterepo.Repository, + taskID, sessionID, stepID string, state models.TaskSessionState, +) models.PendingStepCompletionSignal { + t.Helper() + ctx := context.Background() + + // models.Task.IsFromOffice is a derived projection, not a stored column + // (isFromOfficeProjection in the task repository): a task is Office when it + // has a non-empty project_id, or its workflow is the workspace's canonical + // office workflow. Setting the struct field would be silently discarded, so + // take the project_id branch. + if _, err := repo.DB().ExecContext(ctx, + `UPDATE tasks SET project_id = 'proj1' WHERE id = ?`, taskID); err != nil { + t.Fatalf("mark task as office: %v", err) + } + task, err := repo.GetTask(ctx, taskID) + if err != nil { + t.Fatalf("get task: %v", err) + } + if !task.IsFromOffice { + t.Fatalf("precondition: task %s should read as an Office task", taskID) + } + + signal := models.PendingStepCompletionSignal{ + StepID: stepID, + Source: models.StepCompletionSourceAgent, + Summary: "all done", + SignaledAt: time.Now().UTC().Add(-30 * time.Minute), + } + if err := repo.SetSessionMetadataKey(ctx, sessionID, models.SessionMetaKeyPendingStepCompletion, signal); err != nil { + t.Fatalf("seed pending signal: %v", err) + } + if state != models.TaskSessionStateRunning { + if err := repo.UpdateTaskSessionState(ctx, sessionID, state, ""); err != nil { + t.Fatalf("set session state %q: %v", state, err) + } + } + return signal +} + +func officeStrandedCount(gate stuckSignalGate) int64 { + v := officeStallStrandedSignalTotal.Get(officeStallLabel("gate", string(gate))) + iv, ok := v.(*expvar.Int) + if !ok { + return 0 + } + return iv.Value() +} + +// AC-001.2: an Office task entering through the stuckSignalCandidate gate is +// surfaced, and is not reclaimed. +func TestReclaimStuckSignalSessionsOnce_SurfacesOfficeTaskAtCandidateGate(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + svc := createTestService(repo, officeStallStepGetter(), newMockTaskRepo()) + seedOfficeStrandedSignal(t, svc, repo, "t1", "s1", "step1", models.TaskSessionStateRunning) + + before := officeStrandedCount(stuckSignalGateCandidate) + svc.reclaimStuckSignalSessionsOnce(ctx) + + if got := officeStrandedCount(stuckSignalGateCandidate) - before; got != 1 { + t.Fatalf("stranded-signal counter delta = %d, want 1", got) + } + assertOfficeStallNotActedOn(t, repo, "t1", "s1", "step1") +} + +// Signal age alone is not inactivity (stuck_signal_watchdog.go's +// stuckSignalWatchdogThreshold comment). An Office agent still producing +// events within the inactivity window — a long tool call, a provider +// retry/backoff — must not be surfaced just because its signal cleared the +// age gate. Only the candidate gate needs this: the waiting gate's session is +// already WAITING_FOR_INPUT, not running. +func TestReclaimStuckSignalSessionsOnce_ActiveOfficeAgentNotSurfacedAtCandidateGate(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + agentMgr := &mockAgentManager{ + currentPromptExecutionID: "exec-1", + } + agentMgr.currentPromptLastActivityAt = time.Now().UTC().Add(-time.Minute) + svc := createTestServiceWithAgent(repo, officeStallStepGetter(), newMockTaskRepo(), agentMgr) + seedOfficeStrandedSignal(t, svc, repo, "t1", "s1", "step1", models.TaskSessionStateRunning) + + before := officeStrandedCount(stuckSignalGateCandidate) + svc.reclaimStuckSignalSessionsOnce(ctx) + + if got := officeStrandedCount(stuckSignalGateCandidate) - before; got != 0 { + t.Fatalf("stranded-signal counter delta = %d, want 0 — the agent is still active", got) + } + assertOfficeStallNotActedOn(t, repo, "t1", "s1", "step1") +} + +// AC-001.3: an Office task entering through the +// reconcileWaitingStuckSignalSessionIfDue gate is surfaced from that gate. +// The two gates are near-identical predicates, so each needs its own proof: +// covering only one leaves the other as a silent second gate. +func TestReclaimStuckSignalSessionsOnce_SurfacesOfficeTaskAtWaitingGate(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + svc := createTestService(repo, officeStallStepGetter(), newMockTaskRepo()) + seedOfficeStrandedSignal(t, svc, repo, "t1", "s1", "step1", models.TaskSessionStateWaitingForInput) + + before := officeStrandedCount(stuckSignalGateWaiting) + svc.reclaimStuckSignalSessionsOnce(ctx) + + if got := officeStrandedCount(stuckSignalGateWaiting) - before; got != 1 { + t.Fatalf("stranded-signal counter delta = %d, want 1", got) + } + assertOfficeStallNotActedOn(t, repo, "t1", "s1", "step1") +} + +// The waiting gate receives a session snapshot from the active-session scan. +// A replacement signal on the same step must not inherit the old signal's age +// while the detector is between that scan and its report. +func TestReconcileWaitingStuckSignalSessionIfDue_SkipsReplacedSignal(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + svc := createTestService(repo, officeStallStepGetter(), newMockTaskRepo()) + oldSignal := seedOfficeStrandedSignal(t, svc, repo, "t1", "s1", "step1", models.TaskSessionStateWaitingForInput) + newSignal := models.PendingStepCompletionSignal{ + StepID: oldSignal.StepID, + Source: models.StepCompletionSourceAgent, + SignaledAt: time.Now().UTC(), + } + if err := repo.SetSessionMetadataKey(ctx, "s1", models.SessionMetaKeyPendingStepCompletion, newSignal); err != nil { + t.Fatalf("replace pending signal: %v", err) + } + + snapshot := &models.TaskSession{ + ID: "s1", + TaskID: "t1", + State: models.TaskSessionStateWaitingForInput, + Metadata: map[string]interface{}{ + models.SessionMetaKeyPendingStepCompletion: oldSignal, + }, + } + before := officeStrandedCount(stuckSignalGateWaiting) + svc.reconcileWaitingStuckSignalSessionIfDue(ctx, snapshot, time.Now().UTC()) + if got := officeStrandedCount(stuckSignalGateWaiting) - before; got != 0 { + t.Fatalf("stranded-signal counter delta = %d, want 0", got) + } +} + +// AC-001.5: a still-stranded signal is reported once, not on every tick. +func TestReclaimStuckSignalSessionsOnce_SurfacesOfficeSignalOncePerSignal(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + svc := createTestService(repo, officeStallStepGetter(), newMockTaskRepo()) + seedOfficeStrandedSignal(t, svc, repo, "t1", "s1", "step1", models.TaskSessionStateRunning) + + before := officeStrandedCount(stuckSignalGateCandidate) + for range 3 { + svc.reclaimStuckSignalSessionsOnce(ctx) + } + if got := officeStrandedCount(stuckSignalGateCandidate) - before; got != 1 { + t.Fatalf("counter delta across three scans = %d, want 1", got) + } + + // A genuinely new signal on the same session and step reports again. + newer := models.PendingStepCompletionSignal{ + StepID: "step1", + Source: models.StepCompletionSourceAgent, + SignaledAt: time.Now().UTC().Add(-20 * time.Minute), + } + if err := repo.SetSessionMetadataKey(ctx, "s1", models.SessionMetaKeyPendingStepCompletion, newer); err != nil { + t.Fatalf("reseed signal: %v", err) + } + svc.reclaimStuckSignalSessionsOnce(ctx) + if got := officeStrandedCount(stuckSignalGateCandidate) - before; got != 2 { + t.Fatalf("counter delta after a new signal = %d, want 2", got) + } +} + +// AC-003.1 through AC-003.4: surfacing is the entire action. +func assertOfficeStallNotActedOn(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID, stepID string) { + t.Helper() + ctx := context.Background() + + task, err := repo.GetTask(ctx, taskID) + if err != nil { + t.Fatalf("get task: %v", err) + } + if task.WorkflowStepID != stepID { + t.Fatalf("workflow step moved to %q; the watchdog applied a transition an Office quorum never approved", task.WorkflowStepID) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if _, stillPending := models.LoadPendingStepSignal(session.Metadata); !stillPending { + t.Error("pending signal was consumed; surfacing must leave it for a human") + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index fab56becb1..dc3afa2073 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -662,6 +662,30 @@ type Service struct { // queuedMoveLifecycleLocks serializes source-exit work per task. The // completion marker remains durable so a restart can safely resume work. queuedMoveLifecycleLocks sync.Map + // officeStalledSignals dedupes the Office stranded-signal surfacing so a + // still-stranded signal is reported once, not on every 30-second reaper + // tick. Keyed by session ID, step ID and the signal's SignaledAt, so a + // genuinely new signal on the same session reports again. Deliberately + // in-memory: the surfaced state is derived and cheap to recompute, and a + // backend restart re-reporting each stranded signal once is preferable to + // a schema migration. Pruned against the live session set at the end of + // every complete scan, so it cannot outgrow the active-session list. + officeStalledSignals sync.Map + // officeDecisionWaiting dedupes the Office decision-waiting surfacing on + // the same terms as officeStalledSignals above. Keyed by task ID and step + // ID, so a task that moves on and later returns to a decision step (a + // rejection round) is reported again while a task sitting still is + // reported once. Pruned against the live candidate set at the end of + // every complete scan. + officeDecisionWaiting sync.Map + // officeStallDependenciesMu protects detector dependencies that are wired + // after Service.Start while the idle reaper can already read them. + officeStallDependenciesMu sync.RWMutex + // officeRunInFlight answers whether a task still has a queued or claimed + // run, the decision-waiting detector's false-positive guard. Nil until + // wired (SetOfficeRunInFlightReader); the detector reports nothing and + // counts a skip while it is. + officeRunInFlight officeRunInFlightReader // engineOptions are applied each time initWorkflowEngine runs. Wired // from cmd/kandev (Phase 3.2) to plug Phase 2 ADR-0004 dependencies // — RunQueueAdapter, ParticipantStore, DecisionStore, and the CEO / @@ -1104,6 +1128,16 @@ type Service struct { ciAutomationWorkers sync.WaitGroup } +func (s *Service) officeStallDependencies() ( + engine.ParticipantStore, + engine.DecisionStore, + officeRunInFlightReader, +) { + s.officeStallDependenciesMu.RLock() + defer s.officeStallDependenciesMu.RUnlock() + return s.engineParticipants, s.engineDecisions, s.officeRunInFlight +} + type RouteAction string const ( @@ -1816,14 +1850,18 @@ func (s *Service) SetEngineRunQueue(adapter engine.RunQueueAdapter) { // SetEngineParticipantStore wires the engine's ParticipantStore. func (s *Service) SetEngineParticipantStore(store engine.ParticipantStore) { + s.officeStallDependenciesMu.Lock() s.engineParticipants = store + s.officeStallDependenciesMu.Unlock() s.engineOptions = append(s.engineOptions, engine.WithParticipantStore(store)) s.reinitWorkflowEngine() } // SetEngineDecisionStore wires the engine's DecisionStore. func (s *Service) SetEngineDecisionStore(store engine.DecisionStore) { + s.officeStallDependenciesMu.Lock() s.engineDecisions = store + s.officeStallDependenciesMu.Unlock() s.engineOptions = append(s.engineOptions, engine.WithDecisionStore(store)) s.reinitWorkflowEngine() } diff --git a/apps/backend/internal/orchestrator/stuck_signal_watchdog.go b/apps/backend/internal/orchestrator/stuck_signal_watchdog.go index ff8f1ab74c..487dc2df7d 100644 --- a/apps/backend/internal/orchestrator/stuck_signal_watchdog.go +++ b/apps/backend/internal/orchestrator/stuck_signal_watchdog.go @@ -44,7 +44,9 @@ import ( // (handleRecoverableFailureLocked, PR #2963): only a session holding a // signal for the task's current step is touched. A long turn that never // signalled at all is out of scope and keeps running. Office sessions are -// excluded — this path must not advance an Office task's step. Passthrough +// excluded from the reclaim — this path must not advance an Office task's +// step — but are still reported; see stuckSignalDispositionFor and +// surfaceOfficeStalledSignal. Passthrough // (PTY) sessions are excluded too, mirroring the guard // flipStaleRunningToWaiting and markIdleAfterReset already apply // (event_handlers_workflow.go): a passthrough session manages its own @@ -108,18 +110,28 @@ func (s *Service) reclaimStuckSignalSessionsOnce(ctx context.Context) { scanCtx, cancel := context.WithTimeout(ctx, stuckSignalScanBudget) defer cancel() now := time.Now().UTC() + live := make(map[string]struct{}, len(sessions)) for i, session := range sessions { if scanCtx.Err() != nil { s.logger.Warn("stuck-signal watchdog: per-tick scan budget exceeded; deferring remaining candidates to the next tick", zap.Int("candidates_scanned", i), zap.Int("candidates_deferred", len(sessions)-i)) + // Deliberately no prune: `live` is incomplete for a scan cut + // short here, and pruning against it would re-report signals on + // the sessions this tick never reached. return } + if session != nil { + if signal, ok := models.LoadPendingStepSignal(session.Metadata); ok { + live[officeStalledSignalKey(session.ID, signal)] = struct{}{} + } + } if s.reconcileWaitingStuckSignalSessionIfDue(scanCtx, session, now) { continue } s.reclaimStuckSignalSessionIfDue(scanCtx, session, now) } + s.pruneOfficeStalledSignals(live) } // reconcileWaitingStuckSignalSessionIfDue retries an accepted signal after a @@ -135,7 +147,29 @@ func (s *Service) reconcileWaitingStuckSignalSessionIfDue(ctx context.Context, s return false } task, err := s.repo.GetTask(ctx, session.TaskID) - if err != nil || task == nil || task.WorkflowStepID != signal.StepID || task.IsFromOffice || session.IsPassthrough { + if err != nil || task == nil { + return false + } + switch stuckSignalDispositionFor(task, session, signal) { + case stuckSignalSurfaceOnly: + guard := s.lockCancelInFlightGuard(session.ID) + defer guard.release() + if s.isCancelInFlight(session.ID) { + return true + } + latest, latestSignal, stillPending := s.stuckSignalWaitingStillPending(ctx, session.ID, signal) + if !stillPending { + return true + } + latestTask, err := s.repo.GetTask(ctx, latest.TaskID) + if err != nil || latestTask == nil || + stuckSignalDispositionFor(latestTask, latest, latestSignal) != stuckSignalSurfaceOnly { + return true + } + s.surfaceOfficeStalledSignal(latestTask, latest, latestSignal, stuckSignalGateWaiting, now) + return true + case stuckSignalReclaimable: + default: return false } guard := s.lockCancelInFlightGuard(session.ID) @@ -143,7 +177,7 @@ func (s *Service) reconcileWaitingStuckSignalSessionIfDue(ctx context.Context, s if s.isCancelInFlight(session.ID) { return true } - _, stillPending := s.stuckSignalWaitingStillPending(ctx, session.ID, signal.StepID) + _, _, stillPending := s.stuckSignalWaitingStillPending(ctx, session.ID, signal) if !stillPending { return true } @@ -151,16 +185,20 @@ func (s *Service) reconcileWaitingStuckSignalSessionIfDue(ctx context.Context, s return true } -func (s *Service) stuckSignalWaitingStillPending(ctx context.Context, sessionID, stepID string) (*models.TaskSession, bool) { +func (s *Service) stuckSignalWaitingStillPending( + ctx context.Context, + sessionID string, + expected models.PendingStepCompletionSignal, +) (*models.TaskSession, models.PendingStepCompletionSignal, bool) { latest, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil || latest == nil || latest.State != models.TaskSessionStateWaitingForInput { - return nil, false + return nil, models.PendingStepCompletionSignal{}, false } signal, ok := models.LoadPendingStepSignal(latest.Metadata) - if !ok || signal.StepID != stepID { - return nil, false + if !ok || officeStalledSignalKey(sessionID, signal) != officeStalledSignalKey(sessionID, expected) { + return nil, models.PendingStepCompletionSignal{}, false } - return latest, true + return latest, signal, true } // reclaimStuckSignalSessionIfDue filters a single session against the @@ -357,13 +395,10 @@ func (s *Service) prepareStuckSignalCancellation( } // stuckSignalCandidate applies the unguarded eligibility filter: RUNNING or -// STARTING state, a pending signal older than stuckSignalWatchdogThreshold, -// matching the task's CURRENT step, on a non-Office, non-passthrough task. -// Office sessions go FAILED, not WAITING_FOR_INPUT, and must not advance -// their step from this path — same exclusion PR #2963 applies to -// handleRecoverableFailureLocked's reconciliation hook. Passthrough sessions -// are excluded because this watchdog's inactivity gate cannot see them (see -// the package doc and stuckSignalInactiveLongEnough). +// STARTING state, plus a pending signal older than +// stuckSignalWatchdogThreshold that stuckSignalDispositionFor classifies as +// reclaimable. See that function for why Office and passthrough sessions never +// reach the reclaim path. func (s *Service) stuckSignalCandidate( ctx context.Context, session *models.TaskSession, now time.Time, ) (*models.Task, models.PendingStepCompletionSignal, bool) { @@ -381,14 +416,145 @@ func (s *Service) stuckSignalCandidate( if err != nil || task == nil { return nil, models.PendingStepCompletionSignal{}, false } - // Only a signal that matches the task's CURRENT step is this watchdog's - // job. A stale signal (step already moved on) is left for the ordinary - // signal-consuming paths, which clear a bag entry that no longer - // matches the current step. - if task.WorkflowStepID != signal.StepID || task.IsFromOffice || session.IsPassthrough { + switch stuckSignalDispositionFor(task, session, signal) { + case stuckSignalSurfaceOnly: + // Signal age alone is not inactivity (see stuckSignalWatchdogThreshold's + // comment above): an Office agent can still be mid-tool-call when its + // signal clears this gate. The waiting gate needs no equivalent check — + // that session is WAITING_FOR_INPUT, not running — but this gate's + // session is RUNNING/STARTING, so it is read here exactly as the + // reclaim path below would read it, just without registering a + // reclaim. + if _, inactive := s.stuckSignalInactiveLongEnough(ctx, session.ID, now); inactive { + s.surfaceOfficeStalledSignal(task, session, signal, stuckSignalGateCandidate, now) + } + return nil, models.PendingStepCompletionSignal{}, false + case stuckSignalReclaimable: + return task, signal, true + default: return nil, models.PendingStepCompletionSignal{}, false } - return task, signal, true +} + +// stuckSignalDisposition is the verdict both watchdog gate sites consult. It +// exists so the Office boundary is expressed exactly once: the two gates +// (reconcileWaitingStuckSignalSessionIfDue and stuckSignalCandidate) carried +// near-identical inline predicates, and a change applied to one of them +// silently left the other as a second, divergent gate. +type stuckSignalDisposition int + +const ( + // stuckSignalNotCandidate: this watchdog has no business with the + // session at all. + stuckSignalNotCandidate stuckSignalDisposition = iota + // stuckSignalReclaimable: the session may be reclaimed - cancel the + // execution, force-close the turn, and apply the signalled transition. + stuckSignalReclaimable + // stuckSignalSurfaceOnly: the stuck state is real and worth reporting, + // but this watchdog must not act on it. Office tasks only. + stuckSignalSurfaceOnly +) + +// stuckSignalDispositionFor classifies a session holding a pending signal. +// +// Office tasks are stuckSignalSurfaceOnly, never stuckSignalReclaimable. Both +// gate sites lead to recovery: passing the +// reconcileWaitingStuckSignalSessionIfDue gate falls through to +// reconcileStepCompletionSignalLocked, which applies the step transition the +// agent asked for, and passing the stuckSignalCandidate gate leads to +// reclaimStuckSignalSessionOwned, which cancels the execution and force-closes +// the turn before doing the same. Advancing an Office step from here would +// record the transition without the decisions the step's quorum gate requires - +// the forged approval those gates exist to prevent. Office therefore stays +// excluded from recovery, and is lifted only for detection. +// +// Passthrough (PTY) sessions are stuckSignalNotCandidate regardless of Office +// status: none of the ACP-only lastActivityAt writers ever run for one, so +// there is no activity signal to distinguish a stuck passthrough session from a +// busy one. Excluding them from detection as well as recovery keeps this +// watchdog from reporting a state it cannot actually observe. +func stuckSignalDispositionFor( + task *models.Task, session *models.TaskSession, signal models.PendingStepCompletionSignal, +) stuckSignalDisposition { + if task == nil || session == nil { + return stuckSignalNotCandidate + } + // A stale signal (step already moved on) is left for the ordinary + // signal-consuming paths, which clear a bag entry that no longer matches + // the current step. Checked before the Office branch so a stale Office + // signal is not reported as a stall. + if task.WorkflowStepID != signal.StepID { + return stuckSignalNotCandidate + } + if session.IsPassthrough { + return stuckSignalNotCandidate + } + if task.IsFromOffice { + return stuckSignalSurfaceOnly + } + return stuckSignalReclaimable +} + +// stuckSignalGate names which of the two watchdog gate sites observed a +// stranded signal, so the report says where it was seen. +type stuckSignalGate string + +const ( + stuckSignalGateWaiting stuckSignalGate = "waiting_reconcile" + stuckSignalGateCandidate stuckSignalGate = "candidate" +) + +// officeStalledSignalKey identifies one stranded signal for dedupe purposes. +// SignaledAt is part of the key so a genuinely new signal on the same session +// and step is reported again rather than suppressed by its predecessor. +func officeStalledSignalKey(sessionID string, signal models.PendingStepCompletionSignal) string { + return sessionID + "\x00" + signal.StepID + "\x00" + signal.SignaledAt.UTC().Format(time.RFC3339Nano) +} + +// surfaceOfficeStalledSignal reports an Office task holding a stranded +// completion signal, and does nothing else. No session write, no turn close, +// no step transition, no decision row, no queued run — surfacing this state is +// the whole action, because repairing it would apply a step transition the +// task's quorum gate never approved. +func (s *Service) surfaceOfficeStalledSignal( + task *models.Task, + session *models.TaskSession, + signal models.PendingStepCompletionSignal, + gate stuckSignalGate, + now time.Time, +) { + if task == nil || session == nil { + return + } + if _, seen := s.officeStalledSignals.LoadOrStore(officeStalledSignalKey(session.ID, signal), struct{}{}); seen { + return + } + officeStallStrandedSignalTotal.Add(officeStallLabel("gate", string(gate)), 1) + s.logger.Warn("office stall watchdog: Office task holds a stranded completion signal", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID), + zap.String("step_id", signal.StepID), + zap.Duration("signal_age", now.Sub(signal.SignaledAt)), + zap.String("gate", string(gate))) +} + +// pruneOfficeStalledSignals drops dedupe entries for signals that are no +// longer present on any active session, bounding the map by the active-session +// list. Only ever called after a scan that visited every session: a scan cut +// short by the budget has an incomplete live set, and pruning against it would +// re-report signals that are still stranded. +func (s *Service) pruneOfficeStalledSignals(live map[string]struct{}) { + s.officeStalledSignals.Range(func(key, _ any) bool { + k, ok := key.(string) + if !ok { + s.officeStalledSignals.Delete(key) + return true + } + if _, stillLive := live[k]; !stillLive { + s.officeStalledSignals.Delete(k) + } + return true + }) } // stuckSignalStillPending re-reads the session under the cancelInFlight diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 35986f9c28..c2d14d8b05 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -920,6 +920,27 @@ func (t *Task) IsOfficeOwnedAndAssigned() bool { return t != nil && t.IsFromOffice && t.AssigneeAgentProfileID != "" } +// OfficeDecisionWaitCandidate is the compact projection the Office +// decision-waiting detector scans (REQ-OFFICE-STALL-VISIBILITY-002). It is a +// candidate, not a finding: the repository query only establishes that the +// task is Office-owned, sits at a step carrying a decision-required seat, and +// has been quiet since UpdatedAt. Whether a decision was already recorded and +// whether a run is still in flight are judged by the detector, so each +// rejection has its own countable reason. +type OfficeDecisionWaitCandidate struct { + TaskID string `db:"id"` + StepID string `db:"workflow_step_id"` + UpdatedAt time.Time `db:"updated_at"` +} + +// OfficeDecisionWaitCursor identifies the last candidate in one ordered page. +// The task repository uses it to continue a bounded scan without repeatedly +// returning the same oldest rows. +type OfficeDecisionWaitCursor struct { + UpdatedAt time.Time + TaskID string +} + // ChildCompletionRow is the compact active-child projection used to decide // whether a parent task's on_children_completed trigger is ready to fire. type ChildCompletionRow struct { diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index d7e1ddc0c7..10605605f3 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -107,6 +107,7 @@ func (r *Repository) runMigrations() error { return err } r.migrate.Apply("idx_tasks_queued_for_step", `CREATE INDEX IF NOT EXISTS idx_tasks_queued_for_step ON tasks(queued_for_step_id, queued_at)`) + r.migrate.Apply("idx_tasks_updated_at_id", `CREATE INDEX IF NOT EXISTS idx_tasks_updated_at_id ON tasks(updated_at, id)`) // Remove deprecated workflow_step_id column from task_sessions if err := r.migrateSessionsRemoveWorkflowStepID(); err != nil { return err diff --git a/apps/backend/internal/task/repository/sqlite/base_schema.go b/apps/backend/internal/task/repository/sqlite/base_schema.go index ae3a4ed68f..505828dfd2 100644 --- a/apps/backend/internal/task/repository/sqlite/base_schema.go +++ b/apps/backend/internal/task/repository/sqlite/base_schema.go @@ -553,6 +553,7 @@ func (r *Repository) initCoreIndexes() error { CREATE INDEX IF NOT EXISTS idx_tasks_workflow_id ON tasks(workflow_id); CREATE INDEX IF NOT EXISTS idx_tasks_workflow_step_id ON tasks(workflow_step_id); CREATE INDEX IF NOT EXISTS idx_tasks_archived_at ON tasks(archived_at); + CREATE INDEX IF NOT EXISTS idx_tasks_updated_at_id ON tasks(updated_at, id); CREATE INDEX IF NOT EXISTS idx_task_repositories_task_id ON task_repositories(task_id); CREATE INDEX IF NOT EXISTS idx_task_repositories_repository_id ON task_repositories(repository_id); CREATE INDEX IF NOT EXISTS idx_task_workspace_folders_task_position ON task_workspace_folders(task_id, position); diff --git a/apps/backend/internal/task/repository/sqlite/office_decision_candidates.go b/apps/backend/internal/task/repository/sqlite/office_decision_candidates.go new file mode 100644 index 0000000000..7fe0217de1 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/office_decision_candidates.go @@ -0,0 +1,97 @@ +package sqlite + +import ( + "context" + "fmt" + "time" + + "github.com/kandev/kandev/internal/agentctl/tracing" + "github.com/kandev/kandev/internal/task/models" +) + +// officeDecisionWaitScanLimit bounds one page. The detector runs on the idle +// reaper's tick and does per-candidate follow-up reads, so an unbounded result +// set would let a large parked backlog stretch a single tick past the scan +// budget. Keyset pagination lets one tick continue with later rows until that +// budget expires. +const officeDecisionWaitScanLimit = 200 + +// ListOfficeDecisionWaitCandidates returns Office tasks that sit at a workflow +// step carrying a decision-required seat and have not been touched since +// quietSince. Pass after to continue after a previous ordered page. The return +// cursor is nil when the result set is complete. +// +// This is deliberately only the cheap, indexable half of the +// decision-waiting predicate (REQ-OFFICE-STALL-VISIBILITY-002). It does not +// look at workflow_step_decisions or at the runs queue: those are judged by +// the orchestrator's detector so a candidate rejected for "a decision already +// exists" is distinguishable from one rejected for "a run is still in flight", +// and so an unreadable input can fail closed with its own counted reason +// instead of silently shrinking a SQL result set. +// +// Office identity comes from IsFromOfficePredicate — the same expression that +// backs models.Task.IsFromOffice — rather than a hand-rolled project_id test, +// because IsFromOffice is a read-time projection and a second, drifting +// definition of "Office task" is exactly the bug that would reintroduce the +// exclusion this feature exists to remove. +// +// tasks.updated_at is the quiet-time anchor. It is an approximation of "when +// the task entered this step": an unrelated edit to the task row (a title +// change, a reorder) restarts the clock and delays surfacing. That error is +// one-directional — it under-surfaces, never over-surfaces — which is the +// right direction for a detector whose stated non-goal is acting on what it +// finds. The exact anchor is task_step_transitions.occurred_at, which has no +// read API yet; tightening to it is a follow-up, not a correction. +func (r *Repository) ListOfficeDecisionWaitCandidates( + ctx context.Context, + quietSince time.Time, + after *models.OfficeDecisionWaitCursor, +) ([]models.OfficeDecisionWaitCandidate, *models.OfficeDecisionWaitCursor, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListOfficeDecisionWaitCandidates") + defer span.End() + + cursorPredicate := "" + args := []interface{}{quietSince} + if after != nil { + cursorPredicate = ` + AND (t.updated_at > ? OR (t.updated_at = ? AND t.id > ?))` + args = append(args, after.UpdatedAt, after.UpdatedAt, after.TaskID) + } + query := fmt.Sprintf(` + SELECT t.id, t.workflow_step_id, t.updated_at + FROM tasks t + WHERE t.archived_at IS NULL + AND COALESCE(t.workflow_step_id, '') != '' + AND t.state NOT IN ('COMPLETED', 'FAILED', 'CANCELLED') + AND t.updated_at < ? + %s + AND %s + AND %s + AND EXISTS ( + SELECT 1 FROM workflow_step_participants p + WHERE p.step_id = t.workflow_step_id + AND p.decision_required = 1 + AND (COALESCE(p.task_id, '') = '' OR p.task_id = t.id) + ) + ORDER BY t.updated_at ASC, t.id ASC + LIMIT %d + `, + cursorPredicate, + isFromOfficeProjection("t"), + excludeConfigModePredicate(r.ro.DriverName(), "t.metadata"), + officeDecisionWaitScanLimit, + ) + + var out []models.OfficeDecisionWaitCandidate + if err := r.ro.SelectContext(ctx, &out, r.ro.Rebind(query), args...); err != nil { + return nil, nil, fmt.Errorf("list office decision-wait candidates: %w", err) + } + if len(out) == officeDecisionWaitScanLimit { + last := out[len(out)-1] + return out, &models.OfficeDecisionWaitCursor{ + UpdatedAt: last.UpdatedAt, + TaskID: last.TaskID, + }, nil + } + return out, nil, nil +} diff --git a/apps/backend/internal/task/repository/sqlite/office_decision_candidates_postgres_test.go b/apps/backend/internal/task/repository/sqlite/office_decision_candidates_postgres_test.go new file mode 100644 index 0000000000..bcd999c354 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/office_decision_candidates_postgres_test.go @@ -0,0 +1,60 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/testutil" +) + +// TestPostgresListOfficeDecisionWaitCandidates is the PostgreSQL twin of the +// SQLite candidate-enumeration tests. The query is dialect-sensitive in two +// places the SQLite run cannot exercise: excludeConfigModePredicate expands to +// a JSON expression that differs per driver, and the office-identity EXISTS +// subquery joins workspaces on a nullable column. A regression in either is a +// query that errors on every tick in production while every SQLite test stays +// green — the detector would then surface nothing and look healthy. +// Skips unless KANDEV_TEST_POSTGRES_DSN is set. +func TestPostgresListOfficeDecisionWaitCandidates(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + + const workspaceID = "ws-pg-decision-wait" + if err := repo.CreateWorkspace(ctx, &models.Workspace{ + ID: workspaceID, Name: "pg-decision-wait", OwnerID: "u-1", + }); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + officeWorkflowID, err := repo.EnsureOfficeWorkflow(ctx, workspaceID) + if err != nil { + t.Fatalf("EnsureOfficeWorkflow: %v", err) + } + + seatRow(t, repo, "pg-seat-review", "pg-step-review", "", "reviewer", true) + seatRow(t, repo, "pg-seat-plain", "pg-step-plain", "", "reviewer", false) + + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "pg-t-waiting", "pg-step-review", "", "IN_PROGRESS") + seedCandidateTask(t, repo, workspaceID, "wf-pg-kanban", "pg-t-kanban", "pg-step-review", "", "IN_PROGRESS") + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "pg-t-no-seat", "pg-step-plain", "", "IN_PROGRESS") + + quiet := time.Now().UTC().Add(-2 * time.Hour) + for _, id := range []string{"pg-t-waiting", "pg-t-kanban", "pg-t-no-seat"} { + stampTaskQuiet(t, repo, id, quiet) + } + + got := candidateIDs(t, repo, time.Now().UTC().Add(-time.Hour)) + if !hasID(got, "pg-t-waiting") { + t.Errorf("pg-t-waiting missing from candidates %v", got) + } + for _, unwanted := range []string{"pg-t-kanban", "pg-t-no-seat"} { + if hasID(got, unwanted) { + t.Errorf("candidate %q must not be selected; got %v", unwanted, got) + } + } +} diff --git a/apps/backend/internal/task/repository/sqlite/office_decision_candidates_test.go b/apps/backend/internal/task/repository/sqlite/office_decision_candidates_test.go new file mode 100644 index 0000000000..d448586d36 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/office_decision_candidates_test.go @@ -0,0 +1,209 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// seatRow inserts a workflow_step_participants row. taskID "" is a +// template-level seat (applies to every task at the step); a non-empty taskID +// is a per-task override. +func seatRow(t *testing.T, repo *Repository, id, stepID, taskID, role string, decisionRequired bool) { + t.Helper() + required := 0 + if decisionRequired { + required = 1 + } + if _, err := repo.db.ExecContext(context.Background(), repo.db.Rebind(` + INSERT INTO workflow_step_participants + (id, step_id, task_id, role, agent_profile_id, decision_required, position) + VALUES (?, ?, ?, ?, 'agent-1', ?, 0) + `), id, stepID, taskID, role, required); err != nil { + t.Fatalf("seed seat %s: %v", id, err) + } +} + +// stampTaskQuiet backdates a task's updated_at. CreateTask stamps "now", and +// the candidate query's whole purpose is to select on that column, so every +// fixture has to be aged explicitly. +func stampTaskQuiet(t *testing.T, repo *Repository, taskID string, at time.Time) { + t.Helper() + if _, err := repo.db.ExecContext(context.Background(), + repo.db.Rebind(`UPDATE tasks SET updated_at = ? WHERE id = ?`), at, taskID); err != nil { + t.Fatalf("backdate %s: %v", taskID, err) + } +} + +func candidateIDs(t *testing.T, repo *Repository, quietSince time.Time) []string { + t.Helper() + var ( + rows []models.OfficeDecisionWaitCandidate + cursor *models.OfficeDecisionWaitCursor + ) + for { + page, next, err := repo.ListOfficeDecisionWaitCandidates(context.Background(), quietSince, cursor) + if err != nil { + t.Fatalf("ListOfficeDecisionWaitCandidates: %v", err) + } + rows = append(rows, page...) + if next == nil { + break + } + cursor = next + } + ids := make([]string, 0, len(rows)) + for _, row := range rows { + ids = append(ids, row.TaskID) + } + return ids +} + +func hasID(ids []string, want string) bool { + for _, id := range ids { + if id == want { + return true + } + } + return false +} + +// officeCandidateFixture builds one workspace with a materialised office +// workflow and returns (repo, workspaceID, officeWorkflowID). +func officeCandidateFixture(t *testing.T) (*Repository, string, string) { + t.Helper() + repo := newRepoForBuiltinWorkflowTests(t) + ctx := context.Background() + const workspaceID = "ws-decision-wait" + if err := repo.CreateWorkspace(ctx, &models.Workspace{ + ID: workspaceID, Name: "decision-wait", OwnerID: "u-1", + }); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + officeWorkflowID, err := repo.EnsureOfficeWorkflow(ctx, workspaceID) + if err != nil { + t.Fatalf("EnsureOfficeWorkflow: %v", err) + } + return repo, workspaceID, officeWorkflowID +} + +func seedCandidateTask( + t *testing.T, repo *Repository, workspaceID, workflowID, taskID, stepID, projectID string, + state string, +) { + t.Helper() + task := &models.Task{ + ID: taskID, + WorkspaceID: workspaceID, + WorkflowID: workflowID, + WorkflowStepID: stepID, + Title: taskID, + State: v1.TaskState(state), + ProjectID: projectID, + } + if err := repo.CreateTask(context.Background(), task); err != nil { + t.Fatalf("CreateTask %s: %v", taskID, err) + } +} + +// TestListOfficeDecisionWaitCandidates_SelectsOnlyOfficeTasksAtDecisionSeats +// pins the cheap half of REQ-OFFICE-STALL-VISIBILITY-002's predicate. Each +// negative row differs from the positive one in exactly one respect, so a +// broken clause fails a named case rather than merely shrinking a list. +func TestListOfficeDecisionWaitCandidates_SelectsOnlyOfficeTasksAtDecisionSeats(t *testing.T) { + repo, workspaceID, officeWorkflowID := officeCandidateFixture(t) + quiet := time.Now().UTC().Add(-2 * time.Hour) + + const ( + reviewStep = "step-review" + plainStep = "step-plain" + ) + seatRow(t, repo, "seat-review", reviewStep, "", "reviewer", true) + seatRow(t, repo, "seat-plain", plainStep, "", "reviewer", false) + + // Positive: office-by-workflow, decision-required seat, quiet. + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-waiting", reviewStep, "", "IN_PROGRESS") + // Office by project link rather than workflow — must also qualify. + seedCandidateTask(t, repo, workspaceID, "wf-kanban", "t-waiting-project", reviewStep, "proj-1", "IN_PROGRESS") + // Kanban task at the same step: excluded by the Office predicate. + seedCandidateTask(t, repo, workspaceID, "wf-kanban", "t-kanban", reviewStep, "", "IN_PROGRESS") + // Office task at a step whose only seat is not decision-required. + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-no-decision-seat", plainStep, "", "IN_PROGRESS") + // Office task at a step with no seats at all. + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-seatless", "step-none", "", "IN_PROGRESS") + // Office task in a terminal state. + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-done", reviewStep, "", "COMPLETED") + + for _, id := range []string{ + "t-waiting", "t-waiting-project", "t-kanban", + "t-no-decision-seat", "t-seatless", "t-done", + } { + stampTaskQuiet(t, repo, id, quiet) + } + + got := candidateIDs(t, repo, time.Now().UTC().Add(-time.Hour)) + + for _, want := range []string{"t-waiting", "t-waiting-project"} { + if !hasID(got, want) { + t.Errorf("candidate %q missing from %v", want, got) + } + } + for _, unwanted := range []string{"t-kanban", "t-no-decision-seat", "t-seatless", "t-done"} { + if hasID(got, unwanted) { + t.Errorf("candidate %q must not be selected; got %v", unwanted, got) + } + } +} + +func TestListOfficeDecisionWaitCandidates_RecentlyTouchedTaskIsNotQuiet(t *testing.T) { + repo, workspaceID, officeWorkflowID := officeCandidateFixture(t) + seatRow(t, repo, "seat-review", "step-review", "", "approver", true) + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-fresh", "step-review", "", "IN_PROGRESS") + stampTaskQuiet(t, repo, "t-fresh", time.Now().UTC().Add(-5*time.Minute)) + + got := candidateIDs(t, repo, time.Now().UTC().Add(-time.Hour)) + if hasID(got, "t-fresh") { + t.Errorf("a task touched 5 minutes ago must not be a candidate at a 1h threshold; got %v", got) + } +} + +func TestListOfficeDecisionWaitCandidates_PerTaskSeatQualifiesOnlyItsOwnTask(t *testing.T) { + repo, workspaceID, officeWorkflowID := officeCandidateFixture(t) + quiet := time.Now().UTC().Add(-2 * time.Hour) + + // The only decision-required seat at this step is a per-task override + // belonging to t-mine, so t-theirs sits at a step with no seat of its own. + seatRow(t, repo, "seat-mine", "step-review", "t-mine", "reviewer", true) + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-mine", "step-review", "", "IN_PROGRESS") + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-theirs", "step-review", "", "IN_PROGRESS") + stampTaskQuiet(t, repo, "t-mine", quiet) + stampTaskQuiet(t, repo, "t-theirs", quiet) + + got := candidateIDs(t, repo, time.Now().UTC().Add(-time.Hour)) + if !hasID(got, "t-mine") { + t.Errorf("t-mine holds the per-task decision seat and must be a candidate; got %v", got) + } + if hasID(got, "t-theirs") { + t.Errorf("t-theirs has no seat of its own and must not borrow t-mine's; got %v", got) + } +} + +func TestListOfficeDecisionWaitCandidates_ArchivedTaskIsExcluded(t *testing.T) { + repo, workspaceID, officeWorkflowID := officeCandidateFixture(t) + seatRow(t, repo, "seat-review", "step-review", "", "reviewer", true) + seedCandidateTask(t, repo, workspaceID, officeWorkflowID, "t-archived", "step-review", "", "IN_PROGRESS") + archivedAt := time.Now().UTC().Add(-3 * time.Hour) + if _, err := repo.db.ExecContext(context.Background(), + repo.db.Rebind(`UPDATE tasks SET archived_at = ?, updated_at = ? WHERE id = ?`), + archivedAt, archivedAt, "t-archived"); err != nil { + t.Fatalf("archive: %v", err) + } + + got := candidateIDs(t, repo, time.Now().UTC().Add(-time.Hour)) + if hasID(got, "t-archived") { + t.Errorf("an archived task must not be surfaced as decision-waiting; got %v", got) + } +} diff --git a/apps/backend/internal/workflow/adapters/adapters.go b/apps/backend/internal/workflow/adapters/adapters.go index 444021c543..7e137c3fb4 100644 --- a/apps/backend/internal/workflow/adapters/adapters.go +++ b/apps/backend/internal/workflow/adapters/adapters.go @@ -164,6 +164,7 @@ func (a *DecisionAdapter) ListStepDecisions( DeciderID: d.DeciderID, Role: d.Role, Comment: d.Comment, + SupersededAt: d.SupersededAt, }) } return out, nil diff --git a/apps/backend/internal/workflow/engine/adapters.go b/apps/backend/internal/workflow/engine/adapters.go index b5de4a286d..f72c2ca22e 100644 --- a/apps/backend/internal/workflow/engine/adapters.go +++ b/apps/backend/internal/workflow/engine/adapters.go @@ -92,6 +92,13 @@ type DecisionInfo struct { DeciderID string Role string Comment string + + // SupersededAt is non-nil once a rework round has superseded this row + // (workflow_step_decisions.superseded_at). ListStepDecisions returns + // superseded rows alongside active ones so timelines render full history; + // a caller that needs only the currently-decided state must filter on + // this field itself rather than treating a non-empty result as "decided". + SupersededAt *time.Time } // ParticipantStore reads the workflow_step_participants table for an engine diff --git a/docs/plans/office-stall-visibility/plan.md b/docs/plans/office-stall-visibility/plan.md new file mode 100644 index 0000000000..482a4bfa88 --- /dev/null +++ b/docs/plans/office-stall-visibility/plan.md @@ -0,0 +1,135 @@ +--- +spec: docs/specs/office/requirements/stall-visibility.md +design: docs/specs/office/system-design/stall-visibility.md +created: 2026-09-01 +status: done +--- + +# Implementation Plan: Office Stall Visibility + +## Overview + +Make two invisible Office stall states visible from the backend, without giving +any sweep permission to repair them. Work lands in three waves: restructure the +watchdog so its Office boundary is expressed once (Wave 1), add stranded-signal +surfacing for Office tasks (Wave 2), and add the decision-waiting detector +(Wave 3). + +## Confirmed root cause + +Verified against the working tree on 2026-09-01. + +`internal/orchestrator/stuck_signal_watchdog.go` excludes Office at two sites: + +- **:138** — `reconcileWaitingStuckSignalSessionIfDue` +- **:388** — `stuckSignalCandidate` + +Both are **recovery** gates, not detection gates. Passing either leads to a +reclaim that cancels the agent execution, force-closes the turn, and calls +`reconcileStepCompletionSignalLocked` to apply the requested step transition. +The originating plan unit describes these exclusions as the defect and asks for +them to be lifted, while separately forbidding any automatic transition for +Office. Both cannot hold: lifting the term at either site is what causes the +automatic transition. This plan resolves the conflict in favour of the stated +non-goal — detection is lifted for Office, recovery is not. + +The second gap is real and unqualified. No detector covers a task parked at a +review or approval step with no run in flight and no decision recorded. The +Office inbox's `inboxTaskReviewRequests` surfaces "a decision is outstanding", +but has no age threshold and no in-flight-run condition, so a healthy task being +actively worked is indistinguishable from a stalled one. + +## Schema findings that shape the work + +Confirmed before implementation, as the plan unit asked: + +- The `runs` table has **no `task_id` column**. Task linkage is + `payload -> '$.task_id'`. Use `dialect.JSONExtract(driver, "payload", + "task_id")`, not a literal `json_extract` — Postgres is a supported driver. + Precedent: `internal/office/repository/sqlite/failure.go:351` with + `failure_postgres_test.go`. +- Run statuses are `queued`, `claimed`, `finished`, `failed`, `cancelled`. There + is no `running`; **`claimed` is the executing state**. "No queued or running + run" maps to `status IN ('queued','claimed')`. Precedent for the whole + predicate: `ReapStaleCheckouts` in `internal/office/repository/sqlite/tasks.go`. +- `orchestrator.Service` already holds step-scoped `engineParticipants` + (`ListStepParticipants`) and `engineDecisions` (`ListStepDecisions`). No new + wiring is needed for seats or decisions. +- `engineRunQueue` is **write-only** (`QueueRun`). A new narrow reader interface + is required for the in-flight-run term. +- The Office repository's `ListActiveTaskDecisions` is deliberately task-scoped + across all steps. Use the engine's step-scoped `ListStepDecisions` instead. + +--- + +## Backend + +### Wave 1: express the Office boundary once + +Replace the two inline `task.IsFromOffice` terms with a single +`stuckSignalDisposition` helper returning `notCandidate` / `reclaimable` / +`surfaceOnly`. Pure refactor: Office still returns a non-recovery outcome at both +sites, so behavior is unchanged and the existing suite must stay green. + +### Wave 2: surface a stranded signal on an Office task + +Add `surfaceOfficeStalledSignal`, called on the `surfaceOnly` outcome from both +sites, with an in-memory dedupe keyed by session, step, and `SignaledAt`. Emits +a `Warn` log carrying the observing gate, and increments +`office_stall_stranded_signal_total`. + +### Wave 3: decision-waiting detector + +Add `office_decision_stall_watchdog.go` with a second pass on the existing reaper +tick, plus `officeRunInFlightReader` and its Office repository implementation, +wired from `internal/backendapp`. Predicate order puts the run check last, since it is the +most expensive term. Every unreadable input fails closed to "do not surface". + +## Waves + +Wave 1: + +- [x] [Task 01: Express the watchdog's Office boundary once](task-01-single-office-disposition.md) + +Wave 2: + +- [x] [Task 02: Surface an Office task's stranded signal](task-02-surface-stranded-signal.md) + +Wave 3: + +- [x] [Task 03: Detect an Office task waiting on a decision](task-03-decision-waiting-detector.md) + +## Required workflow verification + +After all targeted task checks pass: + +1. Run `make -C apps/backend fmt`. +2. Run `make -C apps/backend test lint`. +3. Commit the explicit changed paths with a Conventional Commit message. + +## Risks and non-goals + +- **The reclaim path must stay closed to Office.** Wave 1 is the load-bearing + change: if a later edit routes `surfaceOnly` into the reclaim branch, an Office + step advances without its quorum. The Wave 1 test asserting that no Office task + reaches the reclaim path from either site is the guard against that, and must + not be weakened. +- **A wrong alert is the failure mode that matters here.** Nothing is repaired, + so the only way this capability can do harm is by training operators to ignore + it. Every unreadable input fails closed, and + `office_stall_detector_skipped_total` exists so a silently degraded detector is + distinguishable from a quiet system. +- **The decision-waiting threshold is provisional and deliberately separate.** + Wave 3 uses its own `officeDecisionWaitingThreshold`, set to 60 minutes, and + must not reuse `stuckSignalWatchdogThreshold`. Ten minutes of session silence + is anomalous; ten minutes of a human not yet deciding is a reviewer getting + coffee. Sharing the constant would also couple two unrelated phenomena, so + tuning session silence would silently retune this. Sixty minutes is expected to + move once there is field data. +- **No user-visible surface in this plan.** Detection lands as logs and expvar + counters only. Promoting it to an Office inbox row is deliberately deferred: a + new inbox type carries a frontend change plus copy in five languages, and it + should be designed against real detector output rather than in advance. +- No schema migration, no new table, no new HTTP endpoint, no change to + non-Office reclaim behavior, and no retirement of the workspace-scoped + `Stall Session Watchdog` automation. diff --git a/docs/plans/office-stall-visibility/task-01-single-office-disposition.md b/docs/plans/office-stall-visibility/task-01-single-office-disposition.md new file mode 100644 index 0000000000..41e40d77f8 --- /dev/null +++ b/docs/plans/office-stall-visibility/task-01-single-office-disposition.md @@ -0,0 +1,54 @@ +--- +id: "01-single-office-disposition" +title: "Express the watchdog's Office boundary once" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/office/requirements/stall-visibility.md" +--- + +# Task 01: Express the watchdog's Office boundary once + +Pure refactor. No behavior change; the existing suite must stay green. + +## Acceptance + +- A single `stuckSignalDisposition` helper is the only place `task.IsFromOffice` + is evaluated in `stuck_signal_watchdog.go`, returning `stuckSignalNotCandidate`, + `stuckSignalReclaimable`, or `stuckSignalSurfaceOnly`. +- `reconcileWaitingStuckSignalSessionIfDue` (currently :138) and + `stuckSignalCandidate` (currently :388) both consult it; neither retains an + inline `IsFromOffice` term. +- An Office task returns `stuckSignalSurfaceOnly` from both sites and reaches + neither reclaim path. +- A non-Office task's reclaim behavior is byte-for-byte unchanged. +- Passthrough sessions still return `stuckSignalNotCandidate` from both sites. + +## Verification + +- `make -C apps/backend test` — the whole existing + `internal/orchestrator` suite, including + `stuck_signal_watchdog_test.go` and + `stuck_signal_watchdog_cancellation_test.go`, passes unchanged. +- New table-driven test asserting the disposition for each of: Office task, + non-Office task, passthrough session, step mismatch, signal too young. +- New test asserting an Office task reaches neither reclaim path, entered + independently through each of the two sites. This is the guard named in the + plan's risk section and must not be weakened. + +Write the disposition tests first against the current inline predicates so they +pass before and after the refactor; that is what proves it is behavior-preserving. + +## Files likely touched + +- `apps/backend/internal/orchestrator/stuck_signal_watchdog.go` +- `apps/backend/internal/orchestrator/stuck_signal_watchdog_test.go` + +## Dependencies + +None. + +## Parallelism + +Blocks Tasks 02 and 03. Both build on the disposition helper. diff --git a/docs/plans/office-stall-visibility/task-02-surface-stranded-signal.md b/docs/plans/office-stall-visibility/task-02-surface-stranded-signal.md new file mode 100644 index 0000000000..9089cc54fb --- /dev/null +++ b/docs/plans/office-stall-visibility/task-02-surface-stranded-signal.md @@ -0,0 +1,58 @@ +--- +id: "02-surface-stranded-signal" +title: "Surface an Office task's stranded signal" +status: done +wave: 2 +depends_on: ["01-single-office-disposition"] +plan: "plan.md" +spec: "../../specs/office/requirements/stall-visibility.md" +--- + +# Task 02: Surface an Office task's stranded signal + +Satisfies REQ-OFFICE-STALL-VISIBILITY-001. + +## Acceptance + +- `surfaceOfficeStalledSignal` fires on the `stuckSignalSurfaceOnly` outcome from + both gate sites, emitting one `Warn` log with `task_id`, `session_id`, + `step_id`, `signal_age`, and `gate`. +- `office_stall_stranded_signal_total` increments, labelled by `gate`, using the + `expvar.NewMap` + `metricLabel` convention in + `internal/office/scheduler/metrics_vars.go`. +- An in-memory dedupe keyed by session ID, step ID, and the signal's `SignaledAt` + suppresses re-surfacing across the 30-second reaper ticks; the entry is dropped + when the signal clears. +- Nothing else changes: no session state write, no turn close, no step + transition, no decision row, no queued run. + +## Verification + +- `make -C apps/backend test` +- Test: an Office task with a stranded signal surfaces exactly once across three + consecutive scans. +- Test: entered through `stuckSignalCandidate`, it surfaces (AC-001.2). +- Test: entered through `reconcileWaitingStuckSignalSessionIfDue`, it surfaces + (AC-001.3). +- Test: after surfacing, task step, session state, decision rows and run queue + are all unchanged (AC-003.1 through AC-003.4). +- Test: a new signal on the same session with a later `SignaledAt` surfaces + again. + +Each test must first fail on the Task 01 tree, where the surface-only outcome is +silent. + +## Files likely touched + +- `apps/backend/internal/orchestrator/stuck_signal_watchdog.go` +- `apps/backend/internal/orchestrator/stuck_signal_watchdog_test.go` +- `apps/backend/internal/orchestrator/service.go` (dedupe map field) +- new: `apps/backend/internal/orchestrator/office_stall_metrics.go` + +## Dependencies + +Task 01. + +## Parallelism + +Independent of Task 03 once Task 01 lands. diff --git a/docs/plans/office-stall-visibility/task-03-decision-waiting-detector.md b/docs/plans/office-stall-visibility/task-03-decision-waiting-detector.md new file mode 100644 index 0000000000..8fe1d923f3 --- /dev/null +++ b/docs/plans/office-stall-visibility/task-03-decision-waiting-detector.md @@ -0,0 +1,71 @@ +--- +id: "03-decision-waiting-detector" +title: "Detect an Office task waiting on a decision" +status: done +wave: 3 +depends_on: ["01-single-office-disposition"] +plan: "plan.md" +spec: "../../specs/office/requirements/stall-visibility.md" +--- + +# Task 03: Detect an Office task waiting on a decision + +Satisfies REQ-OFFICE-STALL-VISIBILITY-002. + +## Acceptance + +- A second detection pass runs on the existing idle-reaper tick, bounded by the + same `stuckSignalScanBudget` and sharing its defer-to-next-tick behavior. +- Predicate, evaluated in this order: task is Office → current step has a + `decision_required` reviewer or approver seat + (`engineParticipants.ListStepParticipants`) → no active decision for + (task, current step) (`engineDecisions.ListStepDecisions`) → age over threshold + → no in-flight run. The run check is last because it is the most expensive. +- `officeRunInFlightReader.HasInFlightRunForTask` is implemented in the Office + SQLite repository using `dialect.JSONExtract(driver, "payload", "task_id")` and + `status IN ('queued','claimed')`, and is wired from `internal/backendapp`, where the Office repository is constructed. +- Surfacing is a `Warn` log plus `office_stall_decision_waiting_total`, with the + same in-memory dedupe (keyed by task ID and step ID). +- Every unreadable input fails closed to "do not surface" and increments + `office_stall_detector_skipped_total` labelled by `reason`. +- Threshold is its own named constant, `officeDecisionWaitingThreshold` = 60 + minutes, commented as provisional. Reusing `stuckSignalWatchdogThreshold` is + explicitly rejected: it couples two unrelated phenomena. + +## Verification + +- `make -C apps/backend test` +- Test: a decision-waiting Office task past the threshold is surfaced + (AC-002.1). +- Test: **the same task with a `claimed` run is not surfaced** (AC-002.2). This + is the false-positive guard named in the definition of done. +- Test: the same task with a `queued` run is not surfaced. +- Test: an active decision row for the current step suppresses surfacing + (AC-002.3). +- Test: a step with only non-decision-required or watcher/collaborator seats is + not surfaced (AC-002.4). +- Test: run reader unwired, and run reader returning an error, each skip and + count (AC-002.5). +- Test: a non-Office task is never evaluated (AC-002.6). +- Test: surfacing writes no decision row, queues no run, and does not change the + task's step (AC-003.1 through AC-003.3). +- Postgres-dialect test for the new repository query, mirroring + `internal/office/repository/sqlite/failure_postgres_test.go`. + +## Files likely touched + +- new: `apps/backend/internal/orchestrator/office_decision_stall_watchdog.go` +- new: `apps/backend/internal/orchestrator/office_decision_stall_watchdog_test.go` +- `apps/backend/internal/orchestrator/service.go` (reader setter) +- `apps/backend/internal/orchestrator/idle_session_reaper.go` (tick call site) +- `apps/backend/internal/office/repository/sqlite/runs.go` +- `apps/backend/internal/office/repository/sqlite/runs_test.go` +- `apps/backend/internal/backendapp/main.go` (wiring) + +## Dependencies + +Task 01. Independent of Task 02. + +## Parallelism + +Can run alongside Task 02. diff --git a/docs/specs/office/README.md b/docs/specs/office/README.md index 4230eb818b..fbe03e8f75 100644 --- a/docs/specs/office/README.md +++ b/docs/specs/office/README.md @@ -52,6 +52,7 @@ dashboard projections, and Office testing contracts. - [Office Review Participant Seats](requirements/review-participant-seats.md) - [Office Agent Runtime — Error Handling Contract](requirements/runtime.md) - [Office Scheduler](requirements/scheduler.md) +- [Office Stall Visibility](requirements/stall-visibility.md) - [Step Entry Sequence Execution](requirements/step-entry-sequence-execution.md) - [Office Tasks](requirements/tasks.md) - [Office: E2E Mock Harness for Task Sessions and Messages](requirements/testing.md) @@ -99,6 +100,7 @@ dashboard projections, and Office testing contracts. - [Office Agent Runtime — Error Handling Contract System Design Part 1](system-design/runtime-01.md) - [Office Agent Runtime — Error Handling Contract System Design Part 2](system-design/runtime-02.md) - [Office Scheduler System Design Part 1](system-design/scheduler-01.md) +- [Office Stall Visibility System Design](system-design/stall-visibility.md) - [Step Entry Sequence Execution System Design](system-design/step-entry-sequence-execution.md) - [Office Scheduler System Design Part 2](system-design/scheduler-02.md) - [Office Tasks System Design Part 1](system-design/tasks-01.md) diff --git a/docs/specs/office/requirements/stall-visibility.md b/docs/specs/office/requirements/stall-visibility.md new file mode 100644 index 0000000000..70f4636557 --- /dev/null +++ b/docs/specs/office/requirements/stall-visibility.md @@ -0,0 +1,140 @@ +--- +status: active +system: office +created: 2026-09-01 +owners: + - kandev +--- + +# Office Stall Visibility Requirements + +## Overview + +An Office task can stop making progress in two ways that no current backend +detector reports. First, an Office session can accept a step-completion signal +and never reach turn-end; the stuck-signal watchdog +(`internal/orchestrator/stuck_signal_watchdog.go`) excludes every Office task by +construction, so the state is invisible. Second, an Office task can sit at a +review or approval step with a decision-required seat, no decision recorded, and +no run in flight; nothing is stalled in the session sense, because there is no +session, so no session-oriented detector can see it. + +The Office system owns this contract because the stuck state is defined by +Office primitives: `tasks.is_from_office`, the participant slate, the decision +ledger, and the Office run queue. The orchestrator's stuck-signal watchdog is an +adjacent contract this capability reads and constrains, but does not own. + +This capability makes both states **visible**. It does not recover either one. + +## Terminology + +- **Stranded signal:** a pending `step_complete_kandev` signal recorded in a + session's metadata bag that matches the task's current step and has not been + applied. +- **Reclaim:** the stuck-signal watchdog's recovery action, which cancels the + agent execution, force-closes the stale turn, moves the session to + `WAITING_FOR_INPUT`, and applies the requested step transition through + `reconcileStepCompletionSignalLocked`. +- **Surface:** report a stuck state to an operator or user without changing task, + session, decision, or run state. +- **Decision-waiting task:** an Office task whose current step carries at least + one `decision_required` reviewer or approver seat with no active decision row + for that task and step. +- **In-flight run:** a row in the Office `runs` table for the task with status + `queued` or `claimed`. The table has no terminal `running` status; `claimed` + is the executing state. + +## Requirements + +### REQ-OFFICE-STALL-VISIBILITY-001: Surface an Office task holding a stranded signal + +**Intent:** An Office task whose session accepted a completion signal and then +went quiet is currently invisible to every backend detector. Make it visible +without granting the watchdog permission to advance an Office step. + +**User story:** As an operator, I want a stranded completion signal on an Office +task to be reported, so that I can intervene before the card sits indefinitely. + +The stuck-signal watchdog applies its Office exclusion at two independent sites +(`reconcileWaitingStuckSignalSessionIfDue` and `stuckSignalCandidate`). Both are +**recovery** gates: passing either one leads to a reclaim. Removing the +`IsFromOffice` term from either predicate would make the watchdog cancel an +Office agent and apply the step transition it requested, which is the forged +quorum decision that REQ-OFFICE-STALL-VISIBILITY-003 forbids. The exclusion is +therefore preserved for recovery and lifted only for detection. + +#### Acceptance criteria + +- **AC-OFFICE-STALL-VISIBILITY-001.1:** When an Office task's session holds a + stranded signal older than the watchdog threshold, the system shall surface it + exactly once per distinct signal, and shall not reclaim the session. +- **AC-OFFICE-STALL-VISIBILITY-001.2:** When an Office task reaches the + `stuckSignalCandidate` evaluation path, the system shall surface the stranded + signal from that path. +- **AC-OFFICE-STALL-VISIBILITY-001.3:** When an Office task reaches the + `reconcileWaitingStuckSignalSessionIfDue` evaluation path, the system shall + surface the stranded signal from that path. +- **AC-OFFICE-STALL-VISIBILITY-001.4:** When a task is not an Office task, the + system shall preserve the existing reclaim behavior with no change in outcome. +- **AC-OFFICE-STALL-VISIBILITY-001.5:** While a surfaced signal remains pending + and unchanged, the system shall not re-surface it on subsequent scans. + +### REQ-OFFICE-STALL-VISIBILITY-002: Surface a decision-waiting Office task + +**Intent:** The most common Office stall is a task parked at a review or approval +step with nobody coming. No session is stuck, so no session-oriented detector +observes it. + +**User story:** As an operator, I want an Office task that has been waiting on a +decision with no work in flight to be reported, so that a human can decide it. + +#### Acceptance criteria + +- **AC-OFFICE-STALL-VISIBILITY-002.1:** When an Office task's current step + carries at least one `decision_required` reviewer or approver seat, has no + active decision row for that task and step, has no in-flight run, and has been + in that state for longer than the configured threshold, the system shall + surface the task as decision-waiting. +- **AC-OFFICE-STALL-VISIBILITY-002.2:** When the task has an in-flight run, the + system shall not surface it, regardless of decision state or age. +- **AC-OFFICE-STALL-VISIBILITY-002.3:** When an active decision row exists for + the task and its current step, the system shall not surface it. +- **AC-OFFICE-STALL-VISIBILITY-002.4:** When the current step carries no + `decision_required` reviewer or approver seat, the system shall not surface it. +- **AC-OFFICE-STALL-VISIBILITY-002.5:** When the in-flight run state cannot be + read, the system shall not surface the task and shall record the skip. +- **AC-OFFICE-STALL-VISIBILITY-002.6:** When a task is not an Office task, the + system shall not evaluate it for this detector. + +### REQ-OFFICE-STALL-VISIBILITY-003: Never act on a surfaced Office stall + +**Intent:** Office quorum gates exist to stop a step advancing without the +decisions it requires. A detector that repaired what it found would forge those +decisions. + +**User story:** As a workspace owner, I want stall detection to report and never +remediate, so that an automated sweep cannot manufacture an approval. + +#### Acceptance criteria + +- **AC-OFFICE-STALL-VISIBILITY-003.1:** When either detector surfaces an Office + task, the system shall not change the task's workflow step. +- **AC-OFFICE-STALL-VISIBILITY-003.2:** When either detector surfaces an Office + task, the system shall not write a decision row. +- **AC-OFFICE-STALL-VISIBILITY-003.3:** When either detector surfaces an Office + task, the system shall not queue a run, cancel an agent, or close a turn. +- **AC-OFFICE-STALL-VISIBILITY-003.4:** When either detector surfaces an Office + task, the system shall not change any session state. + +## Out of scope + +- Automatic recovery, step advancement, decision synthesis, or run queueing for + any Office stall. This is the explicit non-goal above. +- Changing reclaim behavior for non-Office tasks. The existing stuck-signal + watchdog contract is unchanged for them. +- Making `lastActivityAt` meaningful for passthrough sessions. Passthrough + remains excluded from both detection and recovery. +- Detecting a long-running Office turn that never signalled at all. That is the + in-process stall ticker's domain. +- Replacing the workspace-scoped `Stall Session Watchdog` automation. Retiring it + is a follow-up once backend coverage is confirmed in the field. diff --git a/docs/specs/office/system-design/stall-visibility.md b/docs/specs/office/system-design/stall-visibility.md new file mode 100644 index 0000000000..0ae1e2c455 --- /dev/null +++ b/docs/specs/office/system-design/stall-visibility.md @@ -0,0 +1,257 @@ +--- +status: current +system: office +requirements: + - REQ-OFFICE-STALL-VISIBILITY-001 + - REQ-OFFICE-STALL-VISIBILITY-002 + - REQ-OFFICE-STALL-VISIBILITY-003 +--- + +# Office Stall Visibility System Design + +## Purpose and boundaries + +This design adds two detection-only passes to the orchestrator's existing +stuck-signal watchdog scan. The Office system owns the outcome because the +states are defined by Office primitives; the orchestrator owns the scan loop and +the session lifecycle, which this design reads and constrains but does not +change for non-Office tasks. + +Adjacent contracts used and not owned: + +- `internal/orchestrator/stuck_signal_watchdog.go` — the scan loop, the + threshold, the cancellation guards, and the reclaim path. +- `internal/workflow/engine` `ParticipantStore` / `DecisionStore` — already wired + onto `orchestrator.Service` as `engineParticipants` / `engineDecisions`, both + step-scoped and nil-safe. +- `internal/office/repository/sqlite` — the `runs` queue. +- `internal/task/repository/sqlite` — the `tasks` table, the + `workflow_step_participants` table, and `IsFromOfficePredicate`, the exported + SQL expression that defines Office ownership. + +The detector lives beside the existing watchdog rather than in the Office +scheduler for two reasons: the step-scoped participant and decision readers this +detector needs are already wired onto `orchestrator.Service` (the Office +repository's `ListActiveTaskDecisions` is deliberately task-scoped across all +steps), and co-locating keeps one place to look for stall detection. + +## Requirement mapping + +| Requirement | Design section | +| --- | --- | +| `REQ-OFFICE-STALL-VISIBILITY-001` | [Splitting detection from recovery](#splitting-detection-from-recovery) | +| `REQ-OFFICE-STALL-VISIBILITY-002` | [Decision-waiting detector](#decision-waiting-detector) | +| `REQ-OFFICE-STALL-VISIBILITY-003` | [Failure and recovery](#failure-and-recovery) | + +## Components and responsibilities + +- **`stuckSignalDisposition`** (new, `stuck_signal_watchdog.go`) — the single + predicate both existing gate sites consult. Returns one of three outcomes and + is the only place the `IsFromOffice` term is evaluated. +- **`surfaceOfficeStalledSignal`** (new) — emits the stranded-signal surfacing + for an Office task and records the dedupe key. +- **`detectOfficeDecisionWaitingOnce`** (new file, + `office_decision_stall_watchdog.go`) — the second detection pass, driven from + the same reaper tick. +- **`officeDecisionCandidateLister`** (new narrow optional interface) — the + candidate enumeration, implemented by the task repository as + `ListOfficeDecisionWaitCandidates`. +- **`officeRunInFlightReader`** (new narrow optional interface) — reports whether + a task has a `queued` or `claimed` run. +- **Office SQLite repository** — implements the reader with a dialect-portable + JSON predicate. + +## Data and contracts + +### Shared disposition + +```go +type stuckSignalDisposition int + +const ( + stuckSignalNotCandidate stuckSignalDisposition = iota + stuckSignalReclaimable + stuckSignalSurfaceOnly // Office: visible, never reclaimed +) +``` + +Both `reconcileWaitingStuckSignalSessionIfDue` and +`stuckSignalCandidate` drop their inline +`task.IsFromOffice` term and call this helper instead. Routing both sites through +one predicate is what makes AC-001.2 and AC-001.3 structurally true rather than +separately maintained: the two predicates are near-identical today, and fixing +one in isolation would leave the other as a silent second gate. + +`stuckSignalSurfaceOnly` is returned for an Office task that satisfies every +other candidacy term. Callers treat it as "not a recovery candidate" — the same +control flow the current `false` return produces — after calling +`surfaceOfficeStalledSignal`. + +Passthrough sessions continue to return `stuckSignalNotCandidate` from both +paths. The inactivity gate cannot read them, so they are excluded from detection +as well as recovery. + +### Run-queue reader + +```go +type officeRunInFlightReader interface { + HasInFlightRunForTask(ctx context.Context, taskID string) (bool, error) +} +``` + +Follows the optional-interface idiom already used by `stuckSignalSessionLister` +and `promptActivitySessionReader`: asserted at use, nil-safe, so lightweight test +doubles stay compatible. Wired from `internal/backendapp`, where the Office +repository is constructed, keeping the orchestrator decoupled from the Office +packages. + +The repository implementation must use +`dialect.JSONExtract(r.ro.DriverName(), "payload", "task_id")` rather than a +literal `json_extract`. The `runs` table has no `task_id` column — task linkage +lives in the payload JSON — and Postgres is a supported driver. Precedent: +`internal/office/repository/sqlite/failure.go` and its +`failure_postgres_test.go`. The status predicate is `status IN ('queued', +'claimed')`; the table has no `running` status. + +### Decision and participant reads + +- Seats: `engineParticipants.ListStepParticipants(ctx, stepID, taskID)`, filtered + to `DecisionRequired` seats with role `reviewer` or `approver`. +- Decisions: `engineDecisions.ListStepDecisions(ctx, taskID, stepID)`. + +Both are step-scoped, which is what AC-002.1 requires. `ListStepDecisions` +returns superseded rows alongside active ones by its own documented contract +(quorum guards filter at the call site), and `SupersedeTaskDecisions` leaves a +reworked task's prior decision rows in place with `superseded_at` set rather +than deleting or re-scoping them. The detector must therefore filter on +`DecisionInfo.SupersededAt` itself — treating any non-empty result as +"decided" reads a re-entered step as permanently decided after the first +rework round, which is exactly the repeat stall this detector exists to find. + +## Control flow + +Both passes run inside `reclaimStuckSignalSessionsOnce`, on the idle reaper's +existing 30-second ticker. No new goroutine. + +1. Existing per-session loop. For each session, the disposition helper decides + `notCandidate` / `reclaimable` / `surfaceOnly`. `surfaceOnly` surfaces and + continues; `reclaimable` follows the unchanged reclaim path. +2. A second pass enumerates candidates, then applies the remaining + decision-waiting predicate to each: current step has a decision-required + reviewer/approver seat → no decision recorded for (task, step) → no in-flight + run. Order matters: the run check is the most expensive term and is evaluated + last. + +The two passes are siblings on the reaper tick, not nested: the decision-waiting +pass is a separate call in `startIdleSessionReaper`'s tick closure, because a +task waiting on a decision need not have any session at all and so cannot be +found by iterating sessions. + +The decision-waiting pass is bounded by the same `stuckSignalScanBudget` already +guarding the tick, and shares its "defer the remainder to the next tick" +behavior so it cannot starve `reclaimIdleSessionsOnce`. + +### Candidate enumeration + +The driving query lives in the task repository, because it reads `tasks` and +must define Office ownership with `IsFromOfficePredicate` — the same expression +that backs `models.Task.IsFromOffice`. A second, hand-rolled definition of +"Office task" would drift, which is exactly the class of bug this feature +exists to remove. + +```go +ListOfficeDecisionWaitCandidates(ctx, quietSince) ([]models.OfficeDecisionWaitCandidate, error) +``` + +It selects only the cheap, indexable half of the predicate: not archived, not in +a terminal state, at a step carrying a `decision_required` seat (template-level +or that task's own override), Office-owned, not a config-mode task, and +`updated_at < quietSince`. Everything else is judged in the orchestrator so each +rejection has its own countable reason and each unreadable input can fail closed +independently. The result is ordered oldest-first and capped, so a truncated +scan drops the freshest candidates and the next tick still sees them. + +**Age anchor.** `tasks.updated_at` is an approximation of "when the task entered +this step". An unrelated edit to the task row restarts the clock and delays +surfacing. That error is one-directional — it under-surfaces, never +over-surfaces — which is the correct direction for a detector whose stated +non-goal is acting on what it finds. The exact anchor is +`task_step_transitions.occurred_at`, which has no read API yet; tightening to it +is a follow-up, not a correction. + +### Threshold + +The decision-waiting pass uses its own constant, +`officeDecisionWaitingThreshold`, set provisionally to 60 minutes. It must not +reuse `stuckSignalWatchdogThreshold`. The two measure unrelated phenomena: a +session that has emitted no events for ten minutes is anomalous, whereas a +decision waiting on a human for ten minutes is ordinary. Sharing the constant +would also mean a future tuning of session silence silently retunes +decision-waiting. Sixty minutes is still plausibly normal but worth counting; a +first alert that cries wolf gets muted, and a muted detector is worth nothing. + +## Failure and recovery + +This capability has no recovery path by design (REQ-OFFICE-STALL-VISIBILITY-003). +Every failure mode resolves to "do not surface, record the skip": + +- Reader unwired or type assertion fails → skip. Absence of the run reader must + never be read as absence of runs; that would surface every decision-waiting + task including healthy ones. +- Reader returns an error → skip and count. Unknown is not "no run in flight". +- Participant or decision store unwired, or either read failing → skip and + count. Mirrors the engine's existing `ReasonParticipantStoreUnwired` / + `ReasonDecisionStoreUnwired` discipline. +- Candidate lister unwired, or the enumeration query failing → skip the whole + tick and count. The next tick retries. + +The asymmetry with the existing watchdog is deliberate. The reclaim path fails +closed because a wrong reclaim corrupts a live session; these passes fail closed +because a wrong alert trains operators to ignore the signal. + +## Persistence + +No schema change. No migration. No new table. + +The stranded-signal dedupe key is in-memory on `orchestrator.Service`, keyed by +session ID, step ID, and the signal's `SignaledAt`, and dropped when the signal +clears or the session is reclaimed. A backend restart re-surfaces each still- +stranded signal once; that is acceptable and preferable to a schema change, +because the surfaced state is itself derived and cheap to recompute. + +The decision-waiting pass is stateless per tick. Re-surfacing is suppressed by +the same in-memory dedupe keyed by task ID and step ID. + +## Security + +No new trust boundary. Both passes read state the orchestrator already reads +within the same process, and neither exposes a new endpoint. Surfaced payloads +carry task, session, and step identifiers plus a duration; they must not carry +prompt text, agent output, or decision notes. + +## Observability + +Structured logs, at `Warn`, on the existing watchdog logger: + +- `office stall watchdog: Office task holds a stranded completion signal` with + `task_id`, `session_id`, `step_id`, `signal_age`, and `gate` (which of the two + sites observed it). +- `office stall watchdog: Office task is waiting on a decision that has not + been recorded` with `task_id`, `step_id`, `task_updated_at`, `quiet_for`, and + `threshold`. + +Counters, following the `expvar.NewMap` + `metricLabel` convention already used +by `internal/office/scheduler/metrics_vars.go` and exposed through `/debug/vars`: + +- `office_stall_stranded_signal_total`, labelled by `gate`. +- `office_stall_decision_waiting_total`. +- `office_stall_detector_skipped_total`, labelled by `reason` + (`run_reader_unwired`, `run_reader_error`, `participant_store_unwired`, + `decision_store_unwired`, `participant_read_failed`, `decision_read_failed`, + `candidate_lister_unwired`, `candidate_list_failed`), so a silently degraded + detector is distinguishable from a genuinely quiet system. + +## Related decisions + +None. This design adds no architectural decision beyond the existing watchdog +contract; the Office recovery exclusion it preserves was established by PR #2963.