feat(orchestrator): surface stalled Office tasks without reclaiming them (#3279)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-02 19:44:20 +00:00
committed by GitHub
parent e66bf620cd
commit 750b23a745
27 changed files with 2860 additions and 22 deletions
+1
View File
@@ -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. - 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. - 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 ### GitHub Operations
+4
View File
@@ -1623,6 +1623,10 @@ func wireWorkflowEngineForOffice(
// REQ-OFFICE-REVIEW-SEATS-004.3: drop a required seat whose agent profile // REQ-OFFICE-REVIEW-SEATS-004.3: drop a required seat whose agent profile
// was deleted after casting, rather than waiting forever on it. // was deleted after casting, rather than waiting forever on it.
orchestratorSvc.SetEngineAgentProfileResolver(officeengineadapters.NewAgentProfileResolverAdapter(repos.Office)) 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() eng := orchestratorSvc.WorkflowEngine()
if eng == nil { if eng == nil {
log.Warn("workflow engine not initialised; office engine dispatcher disabled") log.Warn("workflow engine not initialised; office engine dispatcher disabled")
@@ -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
}
@@ -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")
}
}
@@ -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")
}
}
@@ -157,7 +157,8 @@ func (r *idleSessionReaper) runLoop(ctx context.Context, tick func(ctx context.C
// startIdleSessionReaper wires the reaper into Service.Start. The // startIdleSessionReaper wires the reaper into Service.Start. The
// reaper captures the canonical fail-closed reclaim primitive so // 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 // 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 // reapStalePendingMovesOnce (pending_move_reaper.go) — further scans
// sharing this ticker rather than background goroutines of their own, // sharing this ticker rather than background goroutines of their own,
// preserving the reaper's single-goroutine-owner invariant. The 30s // 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) { if !s.idleReaper.start(ctx, func(tickCtx context.Context) {
s.reclaimIdleSessionsOnce(tickCtx) s.reclaimIdleSessionsOnce(tickCtx)
s.reclaimStuckSignalSessionsOnce(tickCtx) s.reclaimStuckSignalSessionsOnce(tickCtx)
s.detectOfficeDecisionWaitingOnce(tickCtx)
s.reapStalePendingMovesOnce(tickCtx) s.reapStalePendingMovesOnce(tickCtx)
}) { }) {
return return
@@ -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)
}
}
@@ -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
})
}
@@ -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)
}
@@ -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")
}
}
@@ -662,6 +662,30 @@ type Service struct {
// queuedMoveLifecycleLocks serializes source-exit work per task. The // queuedMoveLifecycleLocks serializes source-exit work per task. The
// completion marker remains durable so a restart can safely resume work. // completion marker remains durable so a restart can safely resume work.
queuedMoveLifecycleLocks sync.Map 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 // engineOptions are applied each time initWorkflowEngine runs. Wired
// from cmd/kandev (Phase 3.2) to plug Phase 2 ADR-0004 dependencies // from cmd/kandev (Phase 3.2) to plug Phase 2 ADR-0004 dependencies
// — RunQueueAdapter, ParticipantStore, DecisionStore, and the CEO / // — RunQueueAdapter, ParticipantStore, DecisionStore, and the CEO /
@@ -1104,6 +1128,16 @@ type Service struct {
ciAutomationWorkers sync.WaitGroup 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 type RouteAction string
const ( const (
@@ -1816,14 +1850,18 @@ func (s *Service) SetEngineRunQueue(adapter engine.RunQueueAdapter) {
// SetEngineParticipantStore wires the engine's ParticipantStore. // SetEngineParticipantStore wires the engine's ParticipantStore.
func (s *Service) SetEngineParticipantStore(store engine.ParticipantStore) { func (s *Service) SetEngineParticipantStore(store engine.ParticipantStore) {
s.officeStallDependenciesMu.Lock()
s.engineParticipants = store s.engineParticipants = store
s.officeStallDependenciesMu.Unlock()
s.engineOptions = append(s.engineOptions, engine.WithParticipantStore(store)) s.engineOptions = append(s.engineOptions, engine.WithParticipantStore(store))
s.reinitWorkflowEngine() s.reinitWorkflowEngine()
} }
// SetEngineDecisionStore wires the engine's DecisionStore. // SetEngineDecisionStore wires the engine's DecisionStore.
func (s *Service) SetEngineDecisionStore(store engine.DecisionStore) { func (s *Service) SetEngineDecisionStore(store engine.DecisionStore) {
s.officeStallDependenciesMu.Lock()
s.engineDecisions = store s.engineDecisions = store
s.officeStallDependenciesMu.Unlock()
s.engineOptions = append(s.engineOptions, engine.WithDecisionStore(store)) s.engineOptions = append(s.engineOptions, engine.WithDecisionStore(store))
s.reinitWorkflowEngine() s.reinitWorkflowEngine()
} }
@@ -44,7 +44,9 @@ import (
// (handleRecoverableFailureLocked, PR #2963): only a session holding a // (handleRecoverableFailureLocked, PR #2963): only a session holding a
// signal for the task's current step is touched. A long turn that never // 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 // 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 // (PTY) sessions are excluded too, mirroring the guard
// flipStaleRunningToWaiting and markIdleAfterReset already apply // flipStaleRunningToWaiting and markIdleAfterReset already apply
// (event_handlers_workflow.go): a passthrough session manages its own // (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) scanCtx, cancel := context.WithTimeout(ctx, stuckSignalScanBudget)
defer cancel() defer cancel()
now := time.Now().UTC() now := time.Now().UTC()
live := make(map[string]struct{}, len(sessions))
for i, session := range sessions { for i, session := range sessions {
if scanCtx.Err() != nil { if scanCtx.Err() != nil {
s.logger.Warn("stuck-signal watchdog: per-tick scan budget exceeded; deferring remaining candidates to the next tick", 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_scanned", i),
zap.Int("candidates_deferred", len(sessions)-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 return
} }
if session != nil {
if signal, ok := models.LoadPendingStepSignal(session.Metadata); ok {
live[officeStalledSignalKey(session.ID, signal)] = struct{}{}
}
}
if s.reconcileWaitingStuckSignalSessionIfDue(scanCtx, session, now) { if s.reconcileWaitingStuckSignalSessionIfDue(scanCtx, session, now) {
continue continue
} }
s.reclaimStuckSignalSessionIfDue(scanCtx, session, now) s.reclaimStuckSignalSessionIfDue(scanCtx, session, now)
} }
s.pruneOfficeStalledSignals(live)
} }
// reconcileWaitingStuckSignalSessionIfDue retries an accepted signal after a // reconcileWaitingStuckSignalSessionIfDue retries an accepted signal after a
@@ -135,7 +147,29 @@ func (s *Service) reconcileWaitingStuckSignalSessionIfDue(ctx context.Context, s
return false return false
} }
task, err := s.repo.GetTask(ctx, session.TaskID) 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 return false
} }
guard := s.lockCancelInFlightGuard(session.ID) guard := s.lockCancelInFlightGuard(session.ID)
@@ -143,7 +177,7 @@ func (s *Service) reconcileWaitingStuckSignalSessionIfDue(ctx context.Context, s
if s.isCancelInFlight(session.ID) { if s.isCancelInFlight(session.ID) {
return true return true
} }
_, stillPending := s.stuckSignalWaitingStillPending(ctx, session.ID, signal.StepID) _, _, stillPending := s.stuckSignalWaitingStillPending(ctx, session.ID, signal)
if !stillPending { if !stillPending {
return true return true
} }
@@ -151,16 +185,20 @@ func (s *Service) reconcileWaitingStuckSignalSessionIfDue(ctx context.Context, s
return true 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) latest, err := s.repo.GetTaskSession(ctx, sessionID)
if err != nil || latest == nil || latest.State != models.TaskSessionStateWaitingForInput { if err != nil || latest == nil || latest.State != models.TaskSessionStateWaitingForInput {
return nil, false return nil, models.PendingStepCompletionSignal{}, false
} }
signal, ok := models.LoadPendingStepSignal(latest.Metadata) signal, ok := models.LoadPendingStepSignal(latest.Metadata)
if !ok || signal.StepID != stepID { if !ok || officeStalledSignalKey(sessionID, signal) != officeStalledSignalKey(sessionID, expected) {
return nil, false return nil, models.PendingStepCompletionSignal{}, false
} }
return latest, true return latest, signal, true
} }
// reclaimStuckSignalSessionIfDue filters a single session against the // reclaimStuckSignalSessionIfDue filters a single session against the
@@ -357,13 +395,10 @@ func (s *Service) prepareStuckSignalCancellation(
} }
// stuckSignalCandidate applies the unguarded eligibility filter: RUNNING or // stuckSignalCandidate applies the unguarded eligibility filter: RUNNING or
// STARTING state, a pending signal older than stuckSignalWatchdogThreshold, // STARTING state, plus a pending signal older than
// matching the task's CURRENT step, on a non-Office, non-passthrough task. // stuckSignalWatchdogThreshold that stuckSignalDispositionFor classifies as
// Office sessions go FAILED, not WAITING_FOR_INPUT, and must not advance // reclaimable. See that function for why Office and passthrough sessions never
// their step from this path — same exclusion PR #2963 applies to // reach the reclaim path.
// handleRecoverableFailureLocked's reconciliation hook. Passthrough sessions
// are excluded because this watchdog's inactivity gate cannot see them (see
// the package doc and stuckSignalInactiveLongEnough).
func (s *Service) stuckSignalCandidate( func (s *Service) stuckSignalCandidate(
ctx context.Context, session *models.TaskSession, now time.Time, ctx context.Context, session *models.TaskSession, now time.Time,
) (*models.Task, models.PendingStepCompletionSignal, bool) { ) (*models.Task, models.PendingStepCompletionSignal, bool) {
@@ -381,14 +416,145 @@ func (s *Service) stuckSignalCandidate(
if err != nil || task == nil { if err != nil || task == nil {
return nil, models.PendingStepCompletionSignal{}, false return nil, models.PendingStepCompletionSignal{}, false
} }
// Only a signal that matches the task's CURRENT step is this watchdog's switch stuckSignalDispositionFor(task, session, signal) {
// job. A stale signal (step already moved on) is left for the ordinary case stuckSignalSurfaceOnly:
// signal-consuming paths, which clear a bag entry that no longer // Signal age alone is not inactivity (see stuckSignalWatchdogThreshold's
// matches the current step. // comment above): an Office agent can still be mid-tool-call when its
if task.WorkflowStepID != signal.StepID || task.IsFromOffice || session.IsPassthrough { // 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 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 // stuckSignalStillPending re-reads the session under the cancelInFlight
@@ -920,6 +920,27 @@ func (t *Task) IsOfficeOwnedAndAssigned() bool {
return t != nil && t.IsFromOffice && t.AssigneeAgentProfileID != "" 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 // ChildCompletionRow is the compact active-child projection used to decide
// whether a parent task's on_children_completed trigger is ready to fire. // whether a parent task's on_children_completed trigger is ready to fire.
type ChildCompletionRow struct { type ChildCompletionRow struct {
@@ -107,6 +107,7 @@ func (r *Repository) runMigrations() error {
return err 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_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 // Remove deprecated workflow_step_id column from task_sessions
if err := r.migrateSessionsRemoveWorkflowStepID(); err != nil { if err := r.migrateSessionsRemoveWorkflowStepID(); err != nil {
return err return err
@@ -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_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_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_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_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_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); CREATE INDEX IF NOT EXISTS idx_task_workspace_folders_task_position ON task_workspace_folders(task_id, position);
@@ -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
}
@@ -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)
}
}
}
@@ -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)
}
}
@@ -164,6 +164,7 @@ func (a *DecisionAdapter) ListStepDecisions(
DeciderID: d.DeciderID, DeciderID: d.DeciderID,
Role: d.Role, Role: d.Role,
Comment: d.Comment, Comment: d.Comment,
SupersededAt: d.SupersededAt,
}) })
} }
return out, nil return out, nil
@@ -92,6 +92,13 @@ type DecisionInfo struct {
DeciderID string DeciderID string
Role string Role string
Comment 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 // ParticipantStore reads the workflow_step_participants table for an engine
+135
View File
@@ -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.
@@ -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.
@@ -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.
@@ -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.
+2
View File
@@ -52,6 +52,7 @@ dashboard projections, and Office testing contracts.
- [Office Review Participant Seats](requirements/review-participant-seats.md) - [Office Review Participant Seats](requirements/review-participant-seats.md)
- [Office Agent Runtime — Error Handling Contract](requirements/runtime.md) - [Office Agent Runtime — Error Handling Contract](requirements/runtime.md)
- [Office Scheduler](requirements/scheduler.md) - [Office Scheduler](requirements/scheduler.md)
- [Office Stall Visibility](requirements/stall-visibility.md)
- [Step Entry Sequence Execution](requirements/step-entry-sequence-execution.md) - [Step Entry Sequence Execution](requirements/step-entry-sequence-execution.md)
- [Office Tasks](requirements/tasks.md) - [Office Tasks](requirements/tasks.md)
- [Office: E2E Mock Harness for Task Sessions and Messages](requirements/testing.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 1](system-design/runtime-01.md)
- [Office Agent Runtime — Error Handling Contract System Design Part 2](system-design/runtime-02.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 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) - [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 Scheduler System Design Part 2](system-design/scheduler-02.md)
- [Office Tasks System Design Part 1](system-design/tasks-01.md) - [Office Tasks System Design Part 1](system-design/tasks-01.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.
@@ -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.