diff --git a/apps/backend/internal/backendapp/cron.go b/apps/backend/internal/backendapp/cron.go index 5b65e05add..a1c825601c 100644 --- a/apps/backend/internal/backendapp/cron.go +++ b/apps/backend/internal/backendapp/cron.go @@ -32,6 +32,7 @@ func startCronScheduler( dispatcher *officeenginedispatcher.Dispatcher, routineSvc *officeroutines.RoutineService, officeRecovery schedulercron.Handler, + parentWakeReconciler schedulercron.Handler, log *logger.Logger, ) *schedulercron.Loop { heartbeat := buildHeartbeatHandler(repos, dispatcher, log) @@ -46,7 +47,7 @@ func startCronScheduler( } routines := schedulercron.NewRoutinesHandler(routineTicker, nil, log) loop := schedulercron.NewLoop(schedulercron.DefaultTickInterval, log, - heartbeat, budget, routines, officeRecovery) + heartbeat, budget, routines, officeRecovery, parentWakeReconciler) loop.Start(ctx) log.Info("phase 5 cron loop started", zap.Duration("interval", schedulercron.DefaultTickInterval)) diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index 62c2355a03..2a93f72d32 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -1529,11 +1529,13 @@ func startSchedulingRuntime( officeRoutines = services.OfficeSvcs.Routines } var officeRecovery schedulercron.Handler + var parentWakeReconciler schedulercron.Handler if services.Office != nil { officeRecovery = officeservice.NewOfficeRecoveryHandler(orchScheduler) + parentWakeReconciler = officeservice.NewParentWakeReconciler(orchScheduler) } cronLoop := startCronScheduler( - ctx, repos, engineDispatcher, officeRoutines, officeRecovery, log, + ctx, repos, engineDispatcher, officeRoutines, officeRecovery, parentWakeReconciler, log, ) return &schedulingRuntime{runs: runScheduler, cron: cronLoop} } diff --git a/apps/backend/internal/backendapp/scheduler_wiring_test.go b/apps/backend/internal/backendapp/scheduler_wiring_test.go index 0b65960f60..6b929cb6be 100644 --- a/apps/backend/internal/backendapp/scheduler_wiring_test.go +++ b/apps/backend/internal/backendapp/scheduler_wiring_test.go @@ -88,6 +88,160 @@ func TestRoutinesHandlerNotWiredWithTypedNilService(t *testing.T) { } } +// TestParentWakeReconcilerWiredWhenOfficeEnabled guards ParentWakeReconciler's +// registration end to end: the service tests (scheduler_wake_reconciler_test.go) +// only exercise the handler directly, so nothing proves Office-enabled startup +// actually constructs it and hands it to the cron loop rather than leaving the +// typed-nil schedulercron.Handler the "Office disabled" path relies on. +func TestParentWakeReconcilerWiredWhenOfficeEnabled(t *testing.T) { + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, "main.go", nil, 0) + if err != nil { + t.Fatalf("parse main.go: %v", err) + } + + var startFn *ast.FuncDecl + for _, decl := range file.Decls { + if fn, ok := decl.(*ast.FuncDecl); ok && fn.Name.Name == "startSchedulingRuntime" { + startFn = fn + break + } + } + if startFn == nil { + t.Fatal("startSchedulingRuntime not found in main.go") + } + + var constructedUnderOfficeGate bool + ast.Inspect(startFn, func(n ast.Node) bool { + ifStmt, ok := n.(*ast.IfStmt) + if !ok || !isServicesOfficeNotNil(ifStmt.Cond) { + return true + } + ast.Inspect(ifStmt.Body, func(n ast.Node) bool { + assign, ok := n.(*ast.AssignStmt) + if !ok { + return true + } + for i, lhs := range assign.Lhs { + ident, ok := lhs.(*ast.Ident) + if !ok || ident.Name != "parentWakeReconciler" || i >= len(assign.Rhs) { + continue + } + if isSelectorCall(assign.Rhs[i], "NewParentWakeReconciler") { + constructedUnderOfficeGate = true + } + } + return true + }) + return false + }) + if !constructedUnderOfficeGate { + t.Error("parentWakeReconciler is not constructed via officeservice.NewParentWakeReconciler " + + "inside the `services.Office != nil` gate; ParentWakeReconciler would never run") + } + + var passedToCronLoop bool + ast.Inspect(startFn, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + ident, ok := call.Fun.(*ast.Ident) + if !ok || ident.Name != "startCronScheduler" { + return true + } + for _, arg := range call.Args { + if a, ok := arg.(*ast.Ident); ok && a.Name == "parentWakeReconciler" { + passedToCronLoop = true + } + } + return true + }) + if !passedToCronLoop { + t.Error("parentWakeReconciler is not passed as an argument to startCronScheduler") + } +} + +// TestParentWakeReconcilerRegisteredInCronLoop guards the second half of the +// same registration chain: startCronScheduler must actually forward its +// parentWakeReconciler parameter into the loop it builds, not just accept and +// drop it. +func TestParentWakeReconcilerRegisteredInCronLoop(t *testing.T) { + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, "cron.go", nil, 0) + if err != nil { + t.Fatalf("parse cron.go: %v", err) + } + + var startFn *ast.FuncDecl + for _, decl := range file.Decls { + if fn, ok := decl.(*ast.FuncDecl); ok && fn.Name.Name == "startCronScheduler" { + startFn = fn + break + } + } + if startFn == nil { + t.Fatal("startCronScheduler not found in cron.go") + } + + var hasParam bool + for _, field := range startFn.Type.Params.List { + for _, name := range field.Names { + if name.Name == "parentWakeReconciler" { + hasParam = true + } + } + } + if !hasParam { + t.Fatal("startCronScheduler no longer takes a parentWakeReconciler parameter") + } + + var passedToNewLoop bool + ast.Inspect(startFn, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok || !isSelectorCall(call, "NewLoop") { + return true + } + for _, arg := range call.Args { + if a, ok := arg.(*ast.Ident); ok && a.Name == "parentWakeReconciler" { + passedToNewLoop = true + } + } + return true + }) + if !passedToNewLoop { + t.Error("startCronScheduler does not pass parentWakeReconciler into schedulercron.NewLoop; " + + "it would be accepted but never ticked") + } +} + +// isServicesOfficeNotNil matches the `services.Office != nil` guard condition +// as an AST shape, without depending on operand identifier names beyond that. +func isServicesOfficeNotNil(cond ast.Expr) bool { + bin, ok := cond.(*ast.BinaryExpr) + if !ok || bin.Op != token.NEQ { + return false + } + sel, ok := bin.X.(*ast.SelectorExpr) + if !ok || sel.Sel.Name != "Office" { + return false + } + ident, ok := bin.Y.(*ast.Ident) + return ok && ident.Name == "nil" +} + +// isSelectorCall reports whether expr is a call whose function is a +// package-qualified selector ending in name, e.g. officeservice.NewX(...) or +// schedulercron.NewLoop(...). +func isSelectorCall(expr ast.Expr, name string) bool { + call, ok := expr.(*ast.CallExpr) + if !ok { + return false + } + sel, ok := call.Fun.(*ast.SelectorExpr) + return ok && sel.Sel.Name == name +} + func countNamedCalls(fn *ast.FuncDecl, name string) int { count := 0 ast.Inspect(fn, func(n ast.Node) bool { diff --git a/apps/backend/internal/office/repository/sqlite/base.go b/apps/backend/internal/office/repository/sqlite/base.go index 8501617ffc..9b13a514ce 100644 --- a/apps/backend/internal/office/repository/sqlite/base.go +++ b/apps/backend/internal/office/repository/sqlite/base.go @@ -185,6 +185,9 @@ func (r *Repository) createExtensionTables() error { if err := r.createAgentWakeupRequestTable(); err != nil { return err } + if err := r.createParentChildWakeReceiptsTable(); err != nil { + return err + } return nil } @@ -692,3 +695,30 @@ func (r *Repository) createAgentWakeupRequestTable() error { `) return err } + +// createParentChildWakeReceiptsTable creates the table backing +// ParentWakeReconciler's level-triggered sweep (see +// scheduler_wake_reconciler.go). One row per parent task records the +// child set (sorted child IDs + terminal states, compared directly — no +// hashing) a task_children_completed run was last delivered for, so a +// healthy steady state costs one indexed lookup per tick and emits +// nothing. +// +// The companion tasks(parent_id) index ListStuckParents needs lives in +// runMigrations (migrateParentWakeIndexes) via r.migrate.Apply, not here: +// tasks is a table this package doesn't own (see runMigrations' doc +// comment), so a plain r.db.Exec against it is fatal in the minimal +// single-domain test repos under internal/office/repository/sqlite, +// which build a tasks table only after NewWithDB returns. +func (r *Repository) createParentChildWakeReceiptsTable() error { + _, err := r.db.Exec(` + CREATE TABLE IF NOT EXISTS parent_child_wake_receipts ( + parent_task_id TEXT PRIMARY KEY, + child_set_key TEXT NOT NULL, + delivered_run_id TEXT NOT NULL DEFAULT '', + delivery_operation_id TEXT NOT NULL DEFAULT '', + delivered_at TIMESTAMP NOT NULL + ); + `) + return err +} diff --git a/apps/backend/internal/office/repository/sqlite/base_migrations.go b/apps/backend/internal/office/repository/sqlite/base_migrations.go index 5fb8513ae1..a8cc0bb899 100644 --- a/apps/backend/internal/office/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/office/repository/sqlite/base_migrations.go @@ -39,6 +39,8 @@ func (r *Repository) runMigrations() { // schemas include the columns inline; ALTERs converge existing databases. r.migrateProviderRouting() r.migrateContinuationScope() + r.migrateParentWakeIndexes() + r.migrateParentWakeReceiptColumns() } // migrateContinuationScope adds runs.continuation_scope for databases @@ -127,6 +129,34 @@ func (r *Repository) migrateRunPayloadIndexes() { ) } +// migrateParentWakeIndexes adds the tasks(parent_id) index +// ListStuckParents' three parent_id-correlated subqueries (has-a-live- +// child, no-non-terminal-child, the child-set-key GROUP_CONCAT) need to +// avoid a full tasks scan on every 30-second reconciler tick. Also +// benefits AreAllChildrenTerminal and GetChildSummaries (blockers.go), +// which query the same shape. Applied via r.migrate (non-fatal) rather +// than createParentChildWakeReceiptsTable's r.db.Exec: tasks belongs to +// a different package's schema, so a plain Exec against it is fatal in +// the minimal single-domain test repos that don't create tasks until +// after NewWithDB returns. +func (r *Repository) migrateParentWakeIndexes() { + r.migrate.Apply( + "idx_tasks_parent_id", + `CREATE INDEX IF NOT EXISTS idx_tasks_parent_id ON tasks(parent_id)`, + ) +} + +// migrateParentWakeReceiptColumns adds operation identity for receipts +// created by workflow-engine dispatch. Existing direct-run receipts keep +// their delivered_run_id and receive the empty operation id default. +func (r *Repository) migrateParentWakeReceiptColumns() { + r.migrate.Apply( + "parent_child_wake_receipts.delivery_operation_id", + `ALTER TABLE parent_child_wake_receipts + ADD COLUMN delivery_operation_id TEXT NOT NULL DEFAULT ''`, + ) +} + // migrateProviderRouting creates the office_workspace_routing, // office_run_route_attempts, and office_provider_health tables. The // routing columns are also added with replayable ALTER statements so diff --git a/apps/backend/internal/office/repository/sqlite/blockers.go b/apps/backend/internal/office/repository/sqlite/blockers.go index 5d0f04f645..93e8410682 100644 --- a/apps/backend/internal/office/repository/sqlite/blockers.go +++ b/apps/backend/internal/office/repository/sqlite/blockers.go @@ -162,7 +162,29 @@ func (r *Repository) GetTaskAssignee(ctx context.Context, taskID string) (string return assignee, nil } -// AreAllChildrenTerminal checks if all child tasks of a parent are in terminal state. +// GetTaskAssigneeTx is GetTaskAssignee scoped to a caller-owned transaction, +// for a caller that must read the effective runner immediately before an +// insert whose validity depends on it not having changed since an earlier, +// separate read (ParentWakeReconciler.recordReceipt, scheduler_wake_reconciler.go: +// closing the race between ListStuckParents' SELECT and this transaction's +// run insert, where a reassignment in between would otherwise queue a run +// for a runner that is no longer this parent's assignee). +func (r *Repository) GetTaskAssigneeTx(ctx context.Context, tx *sqlx.Tx, taskID string) (string, error) { + var assignee string + err := tx.QueryRowxContext(ctx, tx.Rebind( + `SELECT `+RunnerProjection("tasks")+` FROM tasks WHERE id = ?`), taskID).Scan(&assignee) + if err != nil { + return "", err + } + return assignee, nil +} + +// AreAllChildrenTerminal checks if all child tasks of a parent are in +// terminal state. Unlike ListStuckParents (wake_receipts.go), this counts +// every child regardless of archived_at, so an archived child stuck +// mid-flight can block it forever — that divergence is intentional on the +// reconciler's side (see ListStuckParents), not a bug here; do not add an +// archived_at filter to this query to "match" it. func (r *Repository) AreAllChildrenTerminal(ctx context.Context, parentID string) (bool, error) { var nonTerminal int err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(` diff --git a/apps/backend/internal/office/repository/sqlite/blockers_test.go b/apps/backend/internal/office/repository/sqlite/blockers_test.go index 6d0ccf1475..eb632336b7 100644 --- a/apps/backend/internal/office/repository/sqlite/blockers_test.go +++ b/apps/backend/internal/office/repository/sqlite/blockers_test.go @@ -261,6 +261,60 @@ func TestListBlockersForTasks(t *testing.T) { } } +// TestGetTaskAssigneeTx_ReflectsCurrentRunner covers the primitive +// ParentWakeReconciler.recordReceipt (scheduler_wake_reconciler.go) uses to close the +// TOCTOU race between ListStuckParents' SELECT and its own transactional run +// insert: a runner reassignment committed on a separate connection before +// the transaction begins must be visible inside it, matching GetTaskAssignee +// (the non-transactional analog this method mirrors) rather than some stale +// snapshot taken before the reassignment. +func TestGetTaskAssigneeTx_ReflectsCurrentRunner(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + insertTask(t, repo, ctx, "parent-1", "ws-1", "Parent", "", "") + seedWakeAgentProfile(t, repo, ctx, "agent-a", "idle") + seedWakeAgentProfile(t, repo, ctx, "agent-b", "idle") + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workflow_step_participants (id, step_id, task_id, role, agent_profile_id) + VALUES ('p-runner-parent-1', '', 'parent-1', 'runner', 'agent-a') + `); err != nil { + t.Fatalf("seed runner: %v", err) + } + + got, err := repo.GetTaskAssignee(ctx, "parent-1") + if err != nil { + t.Fatalf("GetTaskAssignee: %v", err) + } + if got != "agent-a" { + t.Fatalf("GetTaskAssignee = %q, want agent-a", got) + } + + // Reassign on a separate, already-committed statement, simulating a + // reassignment that lands between a candidate's capture (the earlier + // ListStuckParents SELECT) and the transaction recordReceipt() opens to record it. + if _, err := repo.ExecRaw(ctx, ` + UPDATE workflow_step_participants SET agent_profile_id = 'agent-b' + WHERE task_id = 'parent-1' AND role = 'runner' + `); err != nil { + t.Fatalf("reassign runner: %v", err) + } + + tx, err := repo.Writer().BeginTxx(ctx, nil) + if err != nil { + t.Fatalf("begin tx: %v", err) + } + defer func() { _ = tx.Rollback() }() + + gotTx, err := repo.GetTaskAssigneeTx(ctx, tx, "parent-1") + if err != nil { + t.Fatalf("GetTaskAssigneeTx: %v", err) + } + if gotTx != "agent-b" { + t.Fatalf("GetTaskAssigneeTx = %q, want agent-b (the reassigned runner, not the stale agent-a)", gotTx) + } +} + func TestTaskBlocker_SelfReferenceBlocked(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() diff --git a/apps/backend/internal/office/repository/sqlite/wake_receipts.go b/apps/backend/internal/office/repository/sqlite/wake_receipts.go new file mode 100644 index 0000000000..0a70643b72 --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/wake_receipts.go @@ -0,0 +1,280 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "strings" + "time" + + "github.com/jmoiron/sqlx" + + taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// StuckParentCandidate is a parent task whose non-archived children are +// all terminal but which may not yet have received its +// task_children_completed wake (or received it for a since-changed child +// set). ChildSetKey is the deterministic "id:state" concatenation compared +// directly against the last-delivered receipt — no hashing, SQLite has no +// built-in hash function — computed here, in SQL, so the sweep stays a +// single query per tick. +type StuckParentCandidate struct { + ParentTaskID string `db:"parent_task_id"` + AssigneeAgentProfileID string `db:"assignee_agent_profile_id"` + WorkflowStepID string `db:"workflow_step_id"` + ChildSetKey string `db:"child_set_key"` +} + +// ListStuckParents finds parent tasks that look done from their children's +// perspective — not archived, not ephemeral, not already terminal, with at +// least one non-archived child, and with no non-archived child left in a +// non-terminal state — and that have not yet had a task_children_completed +// wake delivered for their current child set, by anyone. +// +// The p.archived_at/is_ephemeral/state filters run over every task row +// regardless of Office adoption; taskrepo.IsFromOfficePredicate narrows +// that down to parents this reconciler may actually act on. HasOfficeAdoption +// (the Tick-level gate in ParentWakeReconciler) only proves some workspace +// somewhere has adopted Office — it says nothing about this parent's own +// task or workspace, so without this predicate a Kanban-only parent in an +// unrelated workspace, or in the same workspace as an Office project it has +// no connection to, could still match every other filter and receive an +// unrequested autonomous run. ListUnstartedTasks +// (tasks.go, the sibling query this reconciler mirrors) applies the same +// predicate for the same reason. +// +// Archived children are deliberately excluded from both the "has a child" +// and "any non-terminal" checks below — this is a divergence from +// AreAllChildrenTerminal (blockers.go), which counts every child +// regardless of archived_at and so can be blocked forever by an archived +// child stuck mid-flight. Do not "fix" this back to match +// AreAllChildrenTerminal; the divergence is intentional so an archived +// child can never wedge the reconciler, and a parent whose only children +// are archived is never swept with a spurious empty child-set key. +// +// Every remaining filter runs in SQL, ahead of LIMIT, not in the caller +// afterward: +// - the LEFT JOIN against parent_child_wake_receipts drops a candidate +// whose stored receipt already matches its current child set and whose +// delivery evidence still exists. A receipt created by the workflow +// engine stores a child-set operation id. A legacy receipt stores a run +// id, and any existing run is evidence that the wake entered the queue. +// Terminal execution failures stay terminal under the Office runtime +// contract; they require an explicit user retry and must not become a +// cron retry loop. A missing referenced run remains eligible because a +// cleanup or migration can remove the delivery evidence. +// - the NOT EXISTS against runs drops a candidate with a queued or +// claimed task_children_completed run (still in flight, regardless of +// which child set it was requested for — wait for it to resolve rather +// than race a duplicate) or a terminal one requested at or after +// newest_child_updated_at (it already reflects the current child set, +// including a failed or cancelled wake that must remain terminal under +// the Office runtime contract). A terminal run requested *before* the +// newest child update does NOT block: that run saw a stale child set, so +// it must not be treated as having delivered this one — this is R3-A's +// fix, replacing a plain "any terminal run ever" check that let one +// finished run permanently immunize a parent against every later +// child-set change. +// queueChildrenCompletedRun and cascadeChildrenCompleted (the +// edge-triggered delivery paths) never write a receipt, so the receipt +// alone cannot tell a healthy edge-delivered wake from a lost one; +// evidence of delivery has to come from runs itself. +// - requiring a non-empty assignee_agent_profile_id drops a candidate +// with no resolvable runner, and the INNER JOIN against agent_profiles +// drops one whose runner is paused, stopped, pending approval, or +// altogether missing (a dangling assignee_agent_profile_id with no +// matching row) — R3-B's fix: a LEFT JOIN treated "no matching row" as +// COALESCE'd-to-idle, which passed a dangling reference straight +// through as a sticky, unresolvable, LIMIT-slot-consuming candidate. +// +// This is an invariant, not four independent filters: any predicate that +// can stay true for the same parent across consecutive ticks MUST be +// applied here, before LIMIT — never in Go after ListStuckParents returns. +// A parent with no runner, or a paused runner, does not resolve itself on +// its own; left as a Go-side rejection it would occupy a LIMIT slot on +// every tick forever, permanently starving any genuinely actionable +// candidate behind it. Go-side rejection is only admissible for a +// condition that can change between this SELECT and the write a moment +// later (guardAgentStatus in the caller is exactly that — a cheap, +// redundant closing of that race window, not the primary filter). +func (r *Repository) ListStuckParents(ctx context.Context, reason string, limit int) ([]StuckParentCandidate, error) { + var rows []StuckParentCandidate + err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(` + WITH stuck AS ( + SELECT + p.id AS parent_task_id, + `+RunnerProjection("p")+` AS assignee_agent_profile_id, + p.workflow_step_id AS workflow_step_id, + COALESCE(( + SELECT GROUP_CONCAT(c.id || ':' || c.state, ',') + FROM ( + SELECT id, state FROM tasks + WHERE parent_id = p.id AND archived_at IS NULL + ORDER BY id + ) c + ), '') AS child_set_key, + ( + SELECT MAX(c.updated_at) FROM tasks c + WHERE c.parent_id = p.id AND c.archived_at IS NULL + ) AS newest_child_updated_at + FROM tasks p + WHERE p.archived_at IS NULL + AND p.is_ephemeral = 0 + AND p.state NOT IN ('COMPLETED', 'CANCELLED') + AND `+taskrepo.IsFromOfficePredicate("p")+` + AND EXISTS ( + SELECT 1 FROM tasks c + WHERE c.parent_id = p.id AND c.archived_at IS NULL + ) + AND NOT EXISTS ( + SELECT 1 FROM tasks c + WHERE c.parent_id = p.id + AND c.archived_at IS NULL + AND c.state NOT IN ('COMPLETED', 'CANCELLED') + ) + ) + SELECT s.parent_task_id, s.assignee_agent_profile_id, s.workflow_step_id, s.child_set_key + FROM stuck s + LEFT JOIN parent_child_wake_receipts r ON r.parent_task_id = s.parent_task_id + INNER JOIN agent_profiles ap ON ap.id = s.assignee_agent_profile_id + WHERE s.assignee_agent_profile_id != '' + AND ap.status NOT IN ('paused', 'stopped', 'pending_approval') + AND ( + r.child_set_key IS NOT s.child_set_key + OR ( + NOT EXISTS ( + SELECT 1 FROM runs delivered + WHERE delivered.id = r.delivered_run_id + ) + AND COALESCE(r.delivery_operation_id, '') = '' + ) + ) + AND NOT EXISTS ( + SELECT 1 FROM runs w + WHERE json_extract(w.payload, '$.task_id') = s.parent_task_id + AND w.reason = ? + AND ( + w.status IN ('queued', 'claimed') + OR ( + w.status IN ('finished', 'failed', 'cancelled') + AND w.requested_at >= s.newest_child_updated_at + ) + ) + ) + ORDER BY s.parent_task_id + LIMIT ? + `), reason, limit) + if err != nil { + return nil, err + } + if rows == nil { + rows = []StuckParentCandidate{} + } + return rows, nil +} + +// WakeReceipt is the last-delivered task_children_completed wake for a +// parent task, keyed by the child set it was delivered for. +type WakeReceipt struct { + ParentTaskID string `db:"parent_task_id"` + ChildSetKey string `db:"child_set_key"` + DeliveredRunID string `db:"delivered_run_id"` + DeliveryOperationID string `db:"delivery_operation_id"` + DeliveredAt time.Time `db:"delivered_at"` +} + +// GetWakeReceipt returns the receipt for a parent task, or nil if none +// has been recorded yet. +func (r *Repository) GetWakeReceipt(ctx context.Context, parentTaskID string) (*WakeReceipt, error) { + var rec WakeReceipt + err := r.ro.GetContext(ctx, &rec, r.ro.Rebind(` + SELECT parent_task_id, child_set_key, delivered_run_id, + delivery_operation_id, delivered_at + FROM parent_child_wake_receipts + WHERE parent_task_id = ? + `), parentTaskID) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return &rec, nil +} + +// UpsertWakeReceiptTx records (or updates) the delivery receipt for a +// parent task's current child set, using a transaction the caller owns. +// deliveredRunID is populated by legacy direct-run callers. The workflow +// engine path uses deliveryOperationID because one trigger can fan out to +// several runs and the engine owns their admission. +func (r *Repository) UpsertWakeReceiptTx( + ctx context.Context, tx *sqlx.Tx, + parentTaskID, childSetKey, deliveredRunID, deliveryOperationID string, + deliveredAt time.Time, +) error { + _, err := tx.ExecContext(ctx, tx.Rebind(` + INSERT INTO parent_child_wake_receipts ( + parent_task_id, child_set_key, delivered_run_id, + delivery_operation_id, delivered_at + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT (parent_task_id) DO UPDATE SET + child_set_key = excluded.child_set_key, + delivered_run_id = excluded.delivered_run_id, + delivery_operation_id = excluded.delivery_operation_id, + delivered_at = excluded.delivered_at + `), parentTaskID, childSetKey, deliveredRunID, deliveryOperationID, deliveredAt) + return err +} + +type childSetKeyRow struct { + ID string `db:"id"` + State string `db:"state"` +} + +// GetChildSetKey returns the deterministic key for the parent's current +// active child set. It reads child ids and states separately from the +// aggregate query so the same logic works on SQLite and PostgreSQL. +func (r *Repository) GetChildSetKey(ctx context.Context, parentTaskID string) (string, error) { + var rows []childSetKeyRow + if err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(` + SELECT id, state + FROM tasks + WHERE parent_id = ? AND archived_at IS NULL + ORDER BY id + `), parentTaskID); err != nil { + return "", err + } + return formatChildSetKey(rows), nil +} + +// GetChildSetKeyTx is the transaction-scoped counterpart to GetChildSetKey. +// Reconciler admission uses it immediately before recording a receipt, so a +// child-set change does not get hidden by a receipt for the old generation. +func (r *Repository) GetChildSetKeyTx( + ctx context.Context, tx *sqlx.Tx, parentTaskID string, +) (string, error) { + var rows []childSetKeyRow + if err := tx.SelectContext(ctx, &rows, tx.Rebind(` + SELECT id, state + FROM tasks + WHERE parent_id = ? AND archived_at IS NULL + ORDER BY id + `), parentTaskID); err != nil { + return "", err + } + return formatChildSetKey(rows), nil +} + +func formatChildSetKey(rows []childSetKeyRow) string { + var b strings.Builder + for i, row := range rows { + if i > 0 { + b.WriteByte(',') + } + b.WriteString(row.ID) + b.WriteByte(':') + b.WriteString(row.State) + } + return b.String() +} diff --git a/apps/backend/internal/office/repository/sqlite/wake_receipts_test.go b/apps/backend/internal/office/repository/sqlite/wake_receipts_test.go new file mode 100644 index 0000000000..87899999fd --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/wake_receipts_test.go @@ -0,0 +1,535 @@ +package sqlite_test + +import ( + "context" + "fmt" + "testing" + + "github.com/kandev/kandev/internal/office/repository/sqlite" +) + +// seedWakeCandidate creates a parent task with one non-archived, terminal +// (COMPLETED) child and a resolvable runner backed by a real agent_profiles +// row, so it satisfies every one of ListStuckParents' predicates on its +// own — including the assignee filter (R2-A), the INNER JOIN against +// agent_profiles (R3-B), and the authoritative-Office-task predicate +// (project_id set, matching scheduler_recovery_test.go's own +// 'office-project' marker) — and would actually be woken end-to-end by the +// full reconciler, not just accepted by this SQL layer in isolation. +// Returns the child_set_key ListStuckParents will compute for it. +func seedWakeCandidate(t *testing.T, repo *sqlite.Repository, ctx context.Context, wsID, parentID string) string { + t.Helper() + insertTask(t, repo, ctx, parentID, wsID, "Parent", "", "") + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, parentID, + ); err != nil { + t.Fatalf("mark parent as Office task: %v", err) + } + childID := parentID + "-child-0" + insertTask(t, repo, ctx, childID, wsID, "Child", "", "") + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET parent_id = ?, state = 'COMPLETED' WHERE id = ?`, parentID, childID, + ); err != nil { + t.Fatalf("set child state: %v", err) + } + agentID := parentID + "-agent" + seedWakeAgentProfile(t, repo, ctx, agentID, "idle") + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workflow_step_participants (id, step_id, task_id, role, agent_profile_id) + VALUES (?, '', ?, 'runner', ?) + `, "p-runner-"+parentID, parentID, agentID); err != nil { + t.Fatalf("seed runner: %v", err) + } + return childID + ":COMPLETED" +} + +// seedWakeAgentProfile inserts (or, for a second call against the same +// agentID, replaces) an agent_profiles row with the given status, using +// the same columns TestListStuckParents_ExcludesUnresolvedOrPausedAssignee +// always has: agent_profiles is created by settingsstore.Provide (see +// newSearchTestRepo) with its full production schema and NOT NULL columns. +func seedWakeAgentProfile(t *testing.T, repo *sqlite.Repository, ctx context.Context, agentID, status string) { + t.Helper() + if _, err := repo.ExecRaw(ctx, ` + INSERT OR REPLACE INTO agent_profiles (id, agent_id, name, agent_display_name, status, created_at, updated_at) + VALUES (?, '', ?, ?, ?, datetime('now'), datetime('now')) + `, agentID, agentID, agentID, status); err != nil { + t.Fatalf("seed agent profile %s: %v", agentID, err) + } +} + +// seedWakeReceipt records a legacy run-backed receipt for parentID as already +// delivered for childSetKey. It also seeds a backing 'finished' run for the +// receipt's delivered_run_id. Workflow-engine receipts use the operation-id +// column instead; this helper keeps the legacy shape covered because existing +// databases can still contain those rows. +func seedWakeReceipt(t *testing.T, repo *sqlite.Repository, ctx context.Context, parentID, childSetKey string) { + t.Helper() + runID := "prior-run-" + parentID + seedWakeRun(t, repo, ctx, runID, parentID, "task_children_completed", "finished") + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO parent_child_wake_receipts (parent_task_id, child_set_key, delivered_run_id, delivered_at) + VALUES (?, ?, ?, datetime('now')) + `, parentID, childSetKey, runID); err != nil { + t.Fatalf("seed receipt: %v", err) + } +} + +// seedWakeRun inserts a runs row for parentID the way CreateRunTx would, +// so tests can simulate a wake already queued/claimed/finished/failed for +// this parent without going through the full reconciler. +func seedWakeRun(t *testing.T, repo *sqlite.Repository, ctx context.Context, runID, parentID, reason, status string) { + t.Helper() + seedWakeRunAt(t, repo, ctx, runID, parentID, reason, status, "datetime('now')") +} + +// seedWakeRunAt is seedWakeRun with an explicit requested_at expression +// (a SQL literal or expression, not a bound parameter — callers pass +// either "datetime('now')" or a quoted absolute timestamp literal), so +// tests can control a run's ordering relative to child updates without +// relying on wall-clock delay between statements. +func seedWakeRunAt(t *testing.T, repo *sqlite.Repository, ctx context.Context, runID, parentID, reason, status, requestedAtExpr string) { + t.Helper() + if _, err := repo.ExecRaw(ctx, fmt.Sprintf(` + INSERT INTO runs (id, agent_profile_id, reason, payload, status, requested_at) + VALUES (?, 'agent-x', ?, ?, ?, %s) + `, requestedAtExpr), runID, reason, fmt.Sprintf(`{"task_id":%q}`, parentID), status); err != nil { + t.Fatalf("seed run: %v", err) + } +} + +// insertTaskAt is insertTask with an explicit created_at/updated_at +// timestamp, so tests can control a child's ordering relative to a run's +// requested_at without relying on wall-clock delay between statements. +func insertTaskAt(t *testing.T, repo *sqlite.Repository, ctx context.Context, id, wsID, timestamp string) { + t.Helper() + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO tasks (id, workspace_id, title, description, identifier, created_at, updated_at) + VALUES (?, ?, 'Task', '', '', ?, ?) + `, id, wsID, timestamp, timestamp); err != nil { + t.Fatalf("insert task %s: %v", id, err) + } +} + +// setChildStateAt sets a child's parent, state, and updated_at in one +// statement, mirroring production's state-transition writes (which always +// bump updated_at alongside state) rather than the plain +// `UPDATE tasks SET parent_id = ?, state = ?` other helpers in this package +// use — those are fine for tests that don't assert on timestamps, but a +// timestamp-comparison predicate needs a child's updated_at to actually +// move when its state does. +func setChildStateAt(t *testing.T, repo *sqlite.Repository, ctx context.Context, parentID, childID, state, timestamp string) { + t.Helper() + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET parent_id = ?, state = ?, updated_at = ? WHERE id = ?`, + parentID, state, timestamp, childID, + ); err != nil { + t.Fatalf("set child state: %v", err) + } +} + +// seedRunner inserts a workflow_step_participants runner row for parentID, +// the same shape seedWakeCandidate writes, factored out so callers that +// build up a candidate's children by hand can still get a resolvable +// runner without duplicating this insert. +func seedRunner(t *testing.T, repo *sqlite.Repository, ctx context.Context, parentID string) { + t.Helper() + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workflow_step_participants (id, step_id, task_id, role, agent_profile_id) + VALUES (?, '', ?, 'runner', ?) + `, "p-runner-"+parentID, parentID, parentID+"-agent"); err != nil { + t.Fatalf("seed runner: %v", err) + } +} + +func TestGetChildSetKey_UsesActiveChildren(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + insertTask(t, repo, ctx, "parent-1", "ws-1", "Parent", "", "") + insertTask(t, repo, ctx, "child-b", "ws-1", "Child B", "", "") + insertTask(t, repo, ctx, "child-a", "ws-1", "Child A", "", "") + if _, err := repo.ExecRaw(ctx, ` + UPDATE tasks SET parent_id = ?, state = ? WHERE id = ? + `, "parent-1", "COMPLETED", "child-b"); err != nil { + t.Fatalf("set child-b: %v", err) + } + if _, err := repo.ExecRaw(ctx, ` + UPDATE tasks SET parent_id = ?, state = ?, archived_at = CURRENT_TIMESTAMP WHERE id = ? + `, "parent-1", "FAILED", "child-a"); err != nil { + t.Fatalf("archive child-a: %v", err) + } + insertTask(t, repo, ctx, "child-c", "ws-1", "Child C", "", "") + if _, err := repo.ExecRaw(ctx, ` + UPDATE tasks SET parent_id = ?, state = ? WHERE id = ? + `, "parent-1", "CANCELLED", "child-c"); err != nil { + t.Fatalf("set child-c: %v", err) + } + + got, err := repo.GetChildSetKey(ctx, "parent-1") + if err != nil { + t.Fatalf("GetChildSetKey: %v", err) + } + if got != "child-b:COMPLETED,child-c:CANCELLED" { + t.Fatalf("child set key = %q, want child-b:COMPLETED,child-c:CANCELLED", got) + } +} + +// TestListStuckParents_LimitDoesNotStarveLaterCandidates is R1-A's +// regression test: candidates with an already-current receipt (nothing to +// do) must not consume LIMIT slots ahead of candidates that genuinely need +// a wake. Before the fix, the receipt comparison ran in Go after the SQL +// LIMIT, so the first 5 "resting" parents in id order permanently starved +// every parent past them. +func TestListStuckParents_LimitDoesNotStarveLaterCandidates(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + // p1..p5: already delivered for their current (unchanged) child set — + // not actionable. + for i := 1; i <= 5; i++ { + id := fmt.Sprintf("p%d", i) + key := seedWakeCandidate(t, repo, ctx, "ws-1", id) + seedWakeReceipt(t, repo, ctx, id, key) + } + // p6..p8: no receipt at all — genuinely stuck, need a wake. + want := map[string]bool{} + for i := 6; i <= 8; i++ { + id := fmt.Sprintf("p%d", i) + seedWakeCandidate(t, repo, ctx, "ws-1", id) + want[id] = true + } + + candidates, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 3 { + t.Fatalf("candidates = %d, want 3 (p6,p7,p8): %#v", len(candidates), candidates) + } + got := map[string]bool{} + for _, c := range candidates { + got[c.ParentTaskID] = true + } + for id := range want { + if !got[id] { + t.Errorf("expected %s among candidates, got %v", id, got) + } + } +} + +// TestListStuckParents_ExcludesParentWithCoveredRun is R1-B's +// regression test: a parent whose wake was already delivered via the +// edge-triggered path (queueChildrenCompletedRun / cascadeChildrenCompleted) +// never gets a receipt written for it — those paths don't write one — so +// only a runs-backed check, not the receipt, can tell the sweep the wake was +// already delivered. Terminal failures are covered too: the runtime contract +// requires an explicit user retry, so the reconciler must not retry them. +func TestListStuckParents_ExcludesParentWithCoveredRun(t *testing.T) { + const reason = "task_children_completed" + + blocking := []string{"queued", "claimed", "finished", "failed", "cancelled"} + for _, status := range blocking { + t.Run(status, func(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + seedWakeCandidate(t, repo, ctx, "ws-1", "parent-1") + seedWakeRunAt(t, repo, ctx, "run-1", "parent-1", reason, status, "datetime('now', '+1 second')") + + candidates, err := repo.ListStuckParents(ctx, reason, 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 0 { + t.Fatalf("status %q: expected the already-delivered parent to be excluded, got %#v", status, candidates) + } + }) + } +} + +// TestListStuckParents_TerminalRunDoesNotRetry verifies that a receipt whose +// delivered run failed or was cancelled still suppresses automatic retries. +// The Office runtime contract makes adapter failures terminal until an +// explicit user action starts a new run. +func TestListStuckParents_TerminalRunDoesNotRetry(t *testing.T) { + const reason = "task_children_completed" + + terminal := []string{"failed", "cancelled"} + for _, status := range terminal { + t.Run(status, func(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + key := seedWakeCandidate(t, repo, ctx, "ws-1", "parent-1") + seedWakeRun(t, repo, ctx, "run-1", "parent-1", reason, status) + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO parent_child_wake_receipts (parent_task_id, child_set_key, delivered_run_id, delivered_at) + VALUES ('parent-1', ?, 'run-1', datetime('now')) + `, key); err != nil { + t.Fatalf("seed receipt: %v", err) + } + + candidates, err := repo.ListStuckParents(ctx, reason, 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 0 { + t.Fatalf("status %q: terminal run must not be retried automatically: candidates = %#v", status, candidates) + } + }) + } + + // Control: a receipt whose referenced run actually finished must still + // permanently exclude the parent for that unchanged child set. + t.Run("finished_stays_excluded", func(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + key := seedWakeCandidate(t, repo, ctx, "ws-1", "parent-1") + seedWakeRun(t, repo, ctx, "run-1", "parent-1", reason, "finished") + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO parent_child_wake_receipts (parent_task_id, child_set_key, delivered_run_id, delivered_at) + VALUES ('parent-1', ?, 'run-1', datetime('now')) + `, key); err != nil { + t.Fatalf("seed receipt: %v", err) + } + + candidates, err := repo.ListStuckParents(ctx, reason, 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 0 { + t.Fatalf("finished delivery: candidates = %#v, want none", candidates) + } + }) + + // An edge-triggered terminal run has no reconciler receipt, but it still + // represents an attempted wake for the current child set. It must not be + // retried on every tick. + for _, status := range terminal { + t.Run("edge_"+status, func(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + seedWakeCandidate(t, repo, ctx, "ws-1", "parent-1") + seedWakeRunAt(t, repo, ctx, "run-1", "parent-1", reason, status, "datetime('now', '+1 second')") + + candidates, err := repo.ListStuckParents(ctx, reason, 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 0 { + t.Fatalf("status %q: edge-triggered terminal run must not be retried: candidates = %#v", status, candidates) + } + }) + } +} + +// TestListStuckParents_ExcludesNonOfficeParent is the fixup regression test +// for the missing authoritative-Office-task predicate: HasOfficeAdoption +// (the Tick-level gate in ParentWakeReconciler) only proves some workspace +// somewhere has adopted Office, not that this parent's own task is an +// Office task, so ListStuckParents must apply its own row-level check — +// mirroring ListUnstartedTasks (tasks.go), the sibling query this +// reconciler mirrors. Without it, an ordinary Kanban parent with terminal +// children and a resolvable runner would receive an unrequested autonomous +// run merely because Office happens to be adopted somewhere in the install. +func TestListStuckParents_ExcludesNonOfficeParent(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + // A Kanban-only parent: seedWakeCandidate's shape, minus the project_id + // marker, so it satisfies every other ListStuckParents predicate. + insertTask(t, repo, ctx, "parent-kanban", "ws-1", "Parent", "", "") + childID := "parent-kanban-child-0" + insertTask(t, repo, ctx, childID, "ws-1", "Child", "", "") + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET parent_id = ?, state = 'COMPLETED' WHERE id = ?`, "parent-kanban", childID, + ); err != nil { + t.Fatalf("set child state: %v", err) + } + seedWakeAgentProfile(t, repo, ctx, "parent-kanban-agent", "idle") + seedRunner(t, repo, ctx, "parent-kanban") + + // Control: a genuine Office candidate, so an empty result can't be + // mistaken for a broken query. + seedWakeCandidate(t, repo, ctx, "ws-1", "parent-office") + + candidates, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 1 || candidates[0].ParentTaskID != "parent-office" { + t.Fatalf("candidates = %#v, want exactly [parent-office] (parent-kanban is not an Office task)", candidates) + } +} + +// TestListStuckParents_ExcludesArchivedOnlyChildren is R1-C's regression +// test: a parent whose only children are archived (including one that +// never reached a terminal state) must not be swept with a spurious empty +// child-set key. +func TestListStuckParents_ExcludesArchivedOnlyChildren(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + insertTask(t, repo, ctx, "parent-archived-only", "ws-1", "Parent", "", "") + insertTask(t, repo, ctx, "parent-archived-only-child-0", "ws-1", "Child", "", "") + if _, err := repo.ExecRaw(ctx, ` + UPDATE tasks SET parent_id = 'parent-archived-only', state = 'IN_PROGRESS', archived_at = datetime('now') + WHERE id = 'parent-archived-only-child-0' + `); err != nil { + t.Fatalf("archive child: %v", err) + } + + // Control: a normal candidate, so an empty result can't be mistaken + // for a broken query. + seedWakeCandidate(t, repo, ctx, "ws-1", "parent-normal") + + candidates, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 1 || candidates[0].ParentTaskID != "parent-normal" { + t.Fatalf("candidates = %#v, want exactly [parent-normal]", candidates) + } +} + +// TestListStuckParents_ExcludesUnresolvedOrPausedAssignee is R2-A's SQL-layer +// regression test: a candidate with no resolvable runner, or whose runner is +// paused/stopped/pending_approval, must never be returned at all — those +// states are sticky (nothing about the row changes on the next tick), so +// returning them lets them permanently occupy a LIMIT slot. See +// TestParentWakeReconciler_UnresolvedAndPausedDoNotStarveHealthyParent +// (scheduler_wake_reconciler_test.go) for the end-to-end version of this +// same defect. +func TestListStuckParents_ExcludesUnresolvedOrPausedAssignee(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + // No runner at all. + insertTask(t, repo, ctx, "parent-no-runner", "ws-1", "Parent", "", "") + insertTask(t, repo, ctx, "parent-no-runner-child-0", "ws-1", "Child", "", "") + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET parent_id = 'parent-no-runner', state = 'COMPLETED' WHERE id = 'parent-no-runner-child-0'`, + ); err != nil { + t.Fatalf("set child state: %v", err) + } + + // Runner resolves, but the agent_profiles row says paused. seedWakeCandidate + // already inserted an 'idle' row for parent-paused-agent; seedWakeAgentProfile's + // INSERT OR REPLACE overwrites it with 'paused'. + seedWakeCandidate(t, repo, ctx, "ws-1", "parent-paused") + seedWakeAgentProfile(t, repo, ctx, "parent-paused-agent", "paused") + + // Runner resolves to an agent_profile_id with no matching agent_profiles + // row at all (R3-B): a dangling reference, not merely a paused one. + insertTask(t, repo, ctx, "parent-dangling", "ws-1", "Parent", "", "") + insertTask(t, repo, ctx, "parent-dangling-child-0", "ws-1", "Child", "", "") + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET parent_id = 'parent-dangling', state = 'COMPLETED' WHERE id = 'parent-dangling-child-0'`, + ); err != nil { + t.Fatalf("set child state: %v", err) + } + if _, err := repo.ExecRaw(ctx, ` + INSERT INTO workflow_step_participants (id, step_id, task_id, role, agent_profile_id) + VALUES ('p-runner-parent-dangling', '', 'parent-dangling', 'runner', 'parent-dangling-nonexistent-agent') + `); err != nil { + t.Fatalf("seed dangling runner: %v", err) + } + + // Control: a normal, healthy candidate. + seedWakeCandidate(t, repo, ctx, "ws-1", "parent-normal") + + candidates, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 1 || candidates[0].ParentTaskID != "parent-normal" { + t.Fatalf("candidates = %#v, want exactly [parent-normal]", candidates) + } +} + +// TestListStuckParents_RecoversAfterChildSetChangesPastFinishedRun is R3-A's +// regression test: a parent whose task_children_completed wake was already +// delivered by a finished run must become a candidate again once its child +// set changes afterward. Before this fix, the NOT EXISTS against runs +// treated any 'finished' run for the parent+reason as permanent evidence of +// delivery, regardless of which child set it was requested for, so a +// second child completing after that run finished could never re-trigger +// the sweep — the exact "lost wake, no recovery path" failure mode this +// reconciler exists to fix, but for a wake it delivered itself. +func TestListStuckParents_RecoversAfterChildSetChangesPastFinishedRun(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + const ( + parentID = "parent-1" + wsID = "ws-1" + oldTime = "2026-01-01 00:00:00" + runTime = "2026-01-01 00:05:00" + newTime = "2026-01-01 00:10:00" + ) + + insertTaskAt(t, repo, ctx, parentID, wsID, oldTime) + if _, err := repo.ExecRaw(ctx, + `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, parentID, + ); err != nil { + t.Fatalf("mark parent as Office task: %v", err) + } + child0 := parentID + "-child-0" + insertTaskAt(t, repo, ctx, child0, wsID, oldTime) + setChildStateAt(t, repo, ctx, parentID, child0, "COMPLETED", oldTime) + seedWakeAgentProfile(t, repo, ctx, parentID+"-agent", "idle") + seedRunner(t, repo, ctx, parentID) + + seedWakeRunAt(t, repo, ctx, "run-1", parentID, "task_children_completed", "finished", fmt.Sprintf("'%s'", runTime)) + + before, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents (before child set changed): %v", err) + } + if len(before) != 0 { + t.Fatalf("before child set changed: candidates = %#v, want none (the finished run already reflects this child set)", before) + } + + // A second child completes after the run finished: the child set has + // changed, and no run reflects it yet. + child1 := parentID + "-child-1" + insertTaskAt(t, repo, ctx, child1, wsID, newTime) + setChildStateAt(t, repo, ctx, parentID, child1, "COMPLETED", newTime) + + after, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents (after child set changed): %v", err) + } + if len(after) != 1 || after[0].ParentTaskID != parentID { + t.Fatalf("LOST WAKE NOT RECOVERABLE: after the child set changed, ListStuckParents returned %#v; want exactly [%s]", after, parentID) + } +} + +// TestListStuckParents_OrdersDeterministically is R2-C's regression test: +// the capped query must not depend on incidental scan order. +func TestListStuckParents_OrdersDeterministically(t *testing.T) { + repo := newSearchTestRepo(t) + ctx := context.Background() + + ids := []string{"p-z", "p-a", "p-m"} + for _, id := range ids { + seedWakeCandidate(t, repo, ctx, "ws-1", id) + } + + candidates, err := repo.ListStuckParents(ctx, "task_children_completed", 5) + if err != nil { + t.Fatalf("ListStuckParents: %v", err) + } + if len(candidates) != 3 { + t.Fatalf("candidates = %d, want 3: %#v", len(candidates), candidates) + } + want := []string{"p-a", "p-m", "p-z"} + for i, c := range candidates { + if c.ParentTaskID != want[i] { + t.Fatalf("candidates[%d] = %q, want %q (candidates not ordered): %#v", i, c.ParentTaskID, want[i], candidates) + } + } +} diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 25993ce0ba..5deaf7cccf 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -52,6 +52,47 @@ func (s *Service) dispatchEngineTrigger( return nil } +// dispatchEngineTriggerForRecovery dispatches a level-triggered wake through +// the workflow engine and reports whether the engine accepted the trigger. +// A missing session is a normal no-op: recording a receipt in that case would +// suppress a later delivery after the session is created. +// +// The production dispatcher exposes HandleTriggerHandled so this path can +// distinguish a missing session from a valid no-action step. Small test +// dispatchers only implement WorkflowEngineDispatcher, so they use the +// existing error-only contract and count a successful call as accepted. +func (s *Service) dispatchEngineTriggerForRecovery( + ctx context.Context, taskID string, trigger engine.Trigger, payload any, opID string, +) (bool, error) { + if s.engineDispatcher == nil { + return false, nil + } + + type handledDispatcher interface { + HandleTriggerHandled( + context.Context, string, engine.Trigger, any, string, + ) (bool, error) + } + if d, ok := s.engineDispatcher.(handledDispatcher); ok { + _, err := d.HandleTriggerHandled(ctx, taskID, trigger, payload, opID) + if errors.Is(err, shared.ErrEngineNoSession) { + s.logger.Debug("engine recovery trigger skipped: no active session", + zap.String("task_id", taskID), + zap.String("trigger", string(trigger))) + return false, nil + } + if err != nil { + return false, err + } + return true, nil + } + + if err := s.dispatchEngineTrigger(ctx, taskID, trigger, payload, opID); err != nil { + return false, err + } + return true, nil +} + // TaskMovedData represents the payload of a task.moved event. type TaskMovedData struct { TaskID string `json:"task_id"` diff --git a/apps/backend/internal/office/service/scheduler_wake_reconciler.go b/apps/backend/internal/office/service/scheduler_wake_reconciler.go new file mode 100644 index 0000000000..62b004a6a7 --- /dev/null +++ b/apps/backend/internal/office/service/scheduler_wake_reconciler.go @@ -0,0 +1,235 @@ +package service + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/office/repository/sqlite" + "github.com/kandev/kandev/internal/workflow/engine" +) + +// maxWakeReconcilePerTick caps the number of stuck parents processed in one +// tick, mirroring maxRecoveryPerTick (scheduler_recovery.go). ListStuckParents +// filters out every sticky non-actionable candidate — stale-or-missing +// receipt, active/terminal run, unresolved or paused/stopped/pending-approval +// assignee — in SQL before this cap is applied, so it bounds real work, not +// resting or permanently-blocked parents. +const maxWakeReconcilePerTick = 5 + +// ParentWakeReconciler is a level-triggered backstop for the +// task_children_completed wake: queueChildrenCompletedRun +// (event_subscribers.go) fires it edge-triggered off the child-completion +// event, and that dispatch can be lost without re-delivery. This handler +// re-derives "is this parent stuck" from current task state every tick, then +// sends the trigger through the same workflow engine used by the edge path. +type ParentWakeReconciler struct { + scheduler *SchedulerIntegration + logger *logger.Logger +} + +// NewParentWakeReconciler constructs the parent-wake reconciliation handler +// for a scheduler integration. +func NewParentWakeReconciler(si *SchedulerIntegration) *ParentWakeReconciler { + return &ParentWakeReconciler{ + scheduler: si, + logger: si.svc.logger.WithFields(zap.String("component", "parent-wake-reconciler")), + } +} + +// Name implements scheduler/cron.Handler. +func (h *ParentWakeReconciler) Name() string { return "parent_wake_reconciler" } + +// Tick implements scheduler/cron.Handler. The adoption check prevents an +// Office-enabled but Kanban-only installation from scanning task rows. +func (h *ParentWakeReconciler) Tick(ctx context.Context) error { + adopted, err := h.scheduler.svc.repo.HasOfficeAdoption(ctx) + if err != nil { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("check Office adoption: %w", err) + } + if !adopted { + return nil + } + h.reconcile(ctx) + return nil +} + +// reconcile sweeps for stuck parents and re-delivers a wake for any whose +// last-delivered receipt no longer matches their current child set. +func (h *ParentWakeReconciler) reconcile(ctx context.Context) { + svc := h.scheduler.svc + + candidates, err := svc.repo.ListStuckParents(ctx, RunReasonTaskChildrenCompleted, maxWakeReconcilePerTick) + if err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: list stuck parents failed", zap.Error(err)) + return + } + + for _, c := range candidates { + if ctx.Err() != nil { + return + } + svc.recordWakeCandidate(c.ParentTaskID) + h.reconcileOne(ctx, svc, c) + } +} + +// reconcileOne re-delivers the wake for a single stuck parent, unless its +// assignee cannot accept a run right now. It re-reads the child-set key before +// dispatch and again before recording the receipt. This keeps a child update +// from making an old candidate look delivered for the new generation. +func (h *ParentWakeReconciler) reconcileOne( + ctx context.Context, svc *Service, c sqlite.StuckParentCandidate, +) { + if err := svc.guardAgentStatus(ctx, c.AssigneeAgentProfileID); err != nil { + svc.recordWakeAssigneeUnresolved(c.ParentTaskID, err.Error()) + return + } + + payload, err := h.buildPayload(ctx, svc, c.ParentTaskID) + if err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: build payload failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + + currentKey, err := svc.repo.GetChildSetKey(ctx, c.ParentTaskID) + if err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: revalidate child set failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + if currentKey != c.ChildSetKey { + return + } + + operationID := wakeOperationID(c.ParentTaskID, c.ChildSetKey) + accepted, err := svc.dispatchEngineTriggerForRecovery( + ctx, + c.ParentTaskID, + engine.TriggerOnChildrenCompleted, + payload, + operationID, + ) + if err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: engine dispatch failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + if !accepted { + return + } + + h.recordReceipt(ctx, svc, c, operationID) +} + +// buildPayload assembles the typed payload expected by the workflow engine's +// on_children_completed trigger. GetChildSummaries counts all children, and +// the candidate query separately excludes archived children from readiness. +func (h *ParentWakeReconciler) buildPayload( + ctx context.Context, svc *Service, parentTaskID string, +) (engine.OnChildrenCompletedPayload, error) { + children, truncated, err := svc.repo.GetChildSummaries(ctx, parentTaskID) + if err != nil { + return engine.OnChildrenCompletedPayload{}, fmt.Errorf("get child summaries: %w", err) + } + + prsByTask := svc.lookupChildPRLinks(ctx, children) + summaries := make([]engine.ChildSummary, 0, len(children)) + for _, c := range children { + summaries = append(summaries, engine.ChildSummary{ + TaskID: c.TaskID, + Status: c.State, + Summary: c.LastComment, + PRLinks: prsByTask[c.TaskID], + }) + } + + if truncated { + // The engine payload has no truncation field. The list is already + // capped by the repository, so the available summaries remain the + // same as the normal edge-triggered path. + h.logger.Debug("wake sweep: child summaries truncated", + zap.String("parent_task_id", parentTaskID)) + } + return engine.OnChildrenCompletedPayload{ChildSummaries: summaries}, nil +} + +// recordReceipt stores the operation-backed receipt after the workflow engine +// accepts the trigger. The engine owns run admission and can fan out to more +// than one target, so a single delivered run id cannot represent this wake. +func (h *ParentWakeReconciler) recordReceipt( + ctx context.Context, svc *Service, c sqlite.StuckParentCandidate, operationID string, +) { + tx, err := svc.repo.Writer().BeginTxx(ctx, nil) + if err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: begin tx failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + defer func() { _ = tx.Rollback() }() + + currentKey, err := svc.repo.GetChildSetKeyTx(ctx, tx, c.ParentTaskID) + if err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: revalidate child set in receipt tx failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + if currentKey != c.ChildSetKey { + return + } + + deliveredAt := time.Now().UTC() + if err := svc.repo.UpsertWakeReceiptTx( + ctx, tx, c.ParentTaskID, c.ChildSetKey, "", operationID, deliveredAt, + ); err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: upsert wake receipt failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + + if err := tx.Commit(); err != nil { + if ctx.Err() != nil { + return + } + h.logger.Error("wake sweep: commit failed", + zap.String("parent_task_id", c.ParentTaskID), zap.Error(err)) + return + } + + svc.recordWakeEmitted(c.ParentTaskID, operationID) +} + +func wakeOperationID(parentTaskID, childSetKey string) string { + sum := sha256.Sum256([]byte(parentTaskID + "\x00" + childSetKey)) + return fmt.Sprintf("task_children_completed:%s:%s", parentTaskID, hex.EncodeToString(sum[:])) +} diff --git a/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go b/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go new file mode 100644 index 0000000000..bc8a412a4c --- /dev/null +++ b/apps/backend/internal/office/service/scheduler_wake_reconciler_test.go @@ -0,0 +1,324 @@ +package service_test + +import ( + "context" + "fmt" + "testing" + + "github.com/kandev/kandev/internal/office/models" + "github.com/kandev/kandev/internal/office/service" +) + +// adoptOffice satisfies HasOfficeAdoption via an office_projects row, the +// same shape scheduler_recovery_test.go uses. +func adoptOffice(t *testing.T, svc *service.Service, wsID string) { + t.Helper() + svc.ExecSQL(t, ` + INSERT INTO office_projects (id, workspace_id, name, created_at, updated_at) + VALUES (?, ?, 'Office Project', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + `, "office-project-"+wsID, wsID) +} + +// seedStuckParent creates a parent task with three terminal (non-archived) +// children and an assignee, satisfying ListStuckParents' predicate. The +// project_id marker (matching scheduler_recovery_test.go's own +// 'office-project' marker) satisfies the authoritative-Office-task +// predicate ListStuckParents applies alongside adoptOffice's workspace-level +// HasOfficeAdoption signal — the two are independent checks. +func seedStuckParent(t *testing.T, svc *service.Service, wsID, parentID, agentID string) { + t.Helper() + createTestAgent(t, svc, wsID, agentID) + insertTestTask(t, svc, parentID, wsID) + svc.ExecSQL(t, `UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, parentID) + setTestTaskAssignee(t, svc, parentID, agentID) + + states := []string{"COMPLETED", "COMPLETED", "CANCELLED"} + for i, state := range states { + childID := fmt.Sprintf("%s-child-%d", parentID, i) + insertTestTask(t, svc, childID, wsID) + svc.ExecSQL(t, `UPDATE tasks SET parent_id = ?, state = ? WHERE id = ?`, + parentID, state, childID) + } +} + +func TestParentWakeReconciler_SkipsWithoutOfficeAdoption(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("reconciler ran without Office adoption: %#v", runs) + } +} + +func TestParentWakeReconciler_EmitsRunForStuckParent(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + dispatcher := &fakeDispatcher{} + svc.SetWorkflowEngineDispatcher(dispatcher) + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick 1: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("reconciler inserted a run instead of dispatching through the workflow engine: %#v", runs) + } + calls := dispatcher.Calls() + if len(calls) != 1 { + t.Fatalf("want exactly one engine dispatch after tick 1, got %d: %#v", len(calls), calls) + } + if calls[0].taskID != "parent-1" { + t.Fatalf("engine dispatch task = %q, want parent-1", calls[0].taskID) + } + if calls[0].opID == "" { + t.Fatal("engine dispatch operation id is empty") + } + receipt, err := svc.GetWakeReceiptForTest(ctx, "parent-1") + if err != nil { + t.Fatalf("get wake receipt: %v", err) + } + if receipt == nil || receipt.DeliveryOperationID == "" { + t.Fatalf("engine dispatch did not persist an operation-backed receipt: %#v", receipt) + } + + // Tick 2: the operation-backed receipt excludes parent-1 for the same + // child set, so the sweep must be a no-op. + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick 2: %v", err) + } + runsAfter, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs after tick 2: %v", err) + } + if len(runsAfter) != 0 { + t.Fatalf("tick 2 created an unexpected run: %#v", runsAfter) + } + if got := len(dispatcher.Calls()); got != 1 { + t.Fatalf("tick 2 dispatched a duplicate engine operation: got %d calls", got) + } +} + +func TestParentWakeReconciler_SkipsWithoutEngineDispatcher(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("reconciler inserted a run without an engine dispatcher: %#v", runs) + } + receipt, err := svc.GetWakeReceiptForTest(ctx, "parent-1") + if err != nil { + t.Fatalf("get wake receipt: %v", err) + } + if receipt != nil { + t.Fatalf("missing dispatcher left a delivery receipt: %#v", receipt) + } +} + +func TestParentWakeReconciler_SkipsCompletedParent(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + svc.ExecSQL(t, `UPDATE tasks SET state = 'COMPLETED' WHERE id = ?`, "parent-1") + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("reconciler swept an already-terminal parent: %#v", runs) + } +} + +func TestParentWakeReconciler_SkipsEphemeralParent(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + svc.ExecSQL(t, `UPDATE tasks SET is_ephemeral = 1 WHERE id = ?`, "parent-1") + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("reconciler swept an ephemeral parent: %#v", runs) + } +} + +func TestParentWakeReconciler_SkipsPausedAssigneeWithoutPartialReceipt(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + + adoptOffice(t, svc, "ws-1") + seedStuckParent(t, svc, "ws-1", "parent-1", "worker-1") + if err := svc.UpdateAgentStatusFields(ctx, "worker-1", string(models.AgentStatusPaused), "test"); err != nil { + t.Fatalf("pause agent: %v", err) + } + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick: %v", err) + } + + runs, err := svc.ListRuns(ctx, "ws-1") + if err != nil { + t.Fatalf("list runs: %v", err) + } + if len(runs) != 0 { + t.Fatalf("reconciler emitted a run for a paused assignee: %#v", runs) + } + + receipt, err := svc.GetWakeReceiptForTest(ctx, "parent-1") + if err != nil { + t.Fatalf("get wake receipt: %v", err) + } + if receipt != nil { + t.Fatalf("paused-assignee skip left a partial receipt: %#v", receipt) + } +} + +// seedStuckParentNoAssignee creates a parent task with terminal children +// but no runner participant row, so RunnerProjection returns an empty +// string and ListStuckParents' assignee_agent_profile_id != ” guard +// filters it out. Used to verify that unresolvable candidates cannot +// starve the LIMIT ahead of a healthy parent. +func seedStuckParentNoAssignee(t *testing.T, svc *service.Service, wsID, parentID string) { + t.Helper() + insertTestTask(t, svc, parentID, wsID) + states := []string{"COMPLETED", "COMPLETED", "CANCELLED"} + for i, state := range states { + childID := fmt.Sprintf("%s-child-%d", parentID, i) + insertTestTask(t, svc, childID, wsID) + svc.ExecSQL(t, `UPDATE tasks SET parent_id = ?, state = ? WHERE id = ?`, + parentID, state, childID) + } +} + +// seedStuckParentDanglingAssignee is seedStuckParent but points the runner +// participant row at an agent_profile_id with no matching agent_profiles +// row — R3-B's sticky case: a dangling reference that a LEFT JOIN against +// agent_profiles let through as COALESCE'd-to-idle, occupying a LIMIT slot +// forever since nothing about a dangling reference ever resolves itself. +func seedStuckParentDanglingAssignee(t *testing.T, svc *service.Service, wsID, parentID string) { + t.Helper() + insertTestTask(t, svc, parentID, wsID) + setTestTaskAssignee(t, svc, parentID, parentID+"-dangling-agent") + states := []string{"COMPLETED", "COMPLETED", "CANCELLED"} + for i, state := range states { + childID := fmt.Sprintf("%s-child-%d", parentID, i) + insertTestTask(t, svc, childID, wsID) + svc.ExecSQL(t, `UPDATE tasks SET parent_id = ?, state = ? WHERE id = ?`, + parentID, state, childID) + } +} + +// TestParentWakeReconciler_UnresolvedAndPausedDoNotStarveHealthyParent is +// R2-A's regression test, tightened by R3-C after review proved it vacuous: +// deleting both SQL assignee predicates and re-running left it unchanged +// green, because ORDER BY parent_task_id already put "healthy-parent" ahead +// of every "stuck-*"/"zz-*" candidate (h < s, h < z) regardless of whether +// the fix was present. The healthy parent is now named to sort strictly +// after every sticky candidate ("zz-healthy-parent" > "stuck-*"), so this +// test goes red against a reintroduced ordering-only guard instead of +// passing by accident. +// +// The unresolved-runner and paused-agent candidates are excluded by +// ListStuckParents' WHERE clause before its LIMIT is applied, so — now +// that both live in SQL — they cost no LIMIT slot at all and cannot by +// themselves starve anything; they stay here purely as regression coverage +// that this fix does not regress. The 5 dangling-agent-profile candidates +// are what actually drives this test red against R3-B's unfixed LEFT JOIN: +// a LEFT JOIN against agent_profiles COALESCEs a dangling reference's NULL +// status to 'idle', which passes the assignee filter and — unlike the +// unresolved/paused cases — genuinely consumes a LIMIT slot, is exactly as +// sticky (nothing about a dangling reference ever starts resolving), and +// five of them alone are enough to fill maxWakeReconcilePerTick (5) ahead +// of "zz-healthy-parent" on every tick, forever. +func TestParentWakeReconciler_UnresolvedAndPausedDoNotStarveHealthyParent(t *testing.T) { + svc := newTestService(t) + ctx := context.Background() + dispatcher := &fakeDispatcher{} + svc.SetWorkflowEngineDispatcher(dispatcher) + + adoptOffice(t, svc, "ws-1") + + // 3 candidates with no resolvable runner at all. + for i := 0; i < 3; i++ { + seedStuckParentNoAssignee(t, svc, "ws-1", fmt.Sprintf("stuck-unresolved-%d", i)) + } + // 2 candidates with a paused assignee. + for i := 0; i < 2; i++ { + id := fmt.Sprintf("stuck-paused-%d", i) + agentID := fmt.Sprintf("paused-agent-%d", i) + seedStuckParent(t, svc, "ws-1", id, agentID) + if err := svc.UpdateAgentStatusFields(ctx, agentID, string(models.AgentStatusPaused), "test"); err != nil { + t.Fatalf("pause agent %s: %v", agentID, err) + } + } + // 5 candidates with a dangling agent_profile_id (R3-B) — enough on + // their own to fill maxWakeReconcilePerTick ahead of the healthy parent. + for i := 0; i < 5; i++ { + seedStuckParentDanglingAssignee(t, svc, "ws-1", fmt.Sprintf("stuck-dangling-%d", i)) + } + // The one genuinely healthy stuck parent, named to sort strictly after + // every sticky candidate above so an ordered scan reaches it last. + seedStuckParent(t, svc, "ws-1", "zz-healthy-parent", "zz-healthy-worker") + + handler := service.NewParentWakeReconciler(service.NewSchedulerIntegration(svc, 0)) + for tick := 0; tick < 3; tick++ { + if err := handler.Tick(ctx); err != nil { + t.Fatalf("tick %d: %v", tick, err) + } + } + + for _, call := range dispatcher.Calls() { + if call.taskID == "zz-healthy-parent" { + return + } + } + t.Fatalf("zz-healthy-parent was never dispatched across 3 ticks behind 10 sticky candidates: %#v", dispatcher.Calls()) +} diff --git a/apps/backend/internal/office/service/test_helpers.go b/apps/backend/internal/office/service/test_helpers.go index f5b2237c76..79735fa819 100644 --- a/apps/backend/internal/office/service/test_helpers.go +++ b/apps/backend/internal/office/service/test_helpers.go @@ -182,3 +182,10 @@ func BuildSkillManifestForTest( // Skill delivery test helpers were removed in ADR 0005 Wave E along // with the office-tier delivery code. Coverage moved into // internal/agent/runtime/lifecycle/skill. + +// GetWakeReceiptForTest exposes the repo's parent_child_wake_receipts read +// so ParentWakeReconciler tests can assert a paused/unresolved-assignee +// skip leaves no partial receipt behind. +func (s *Service) GetWakeReceiptForTest(ctx context.Context, parentTaskID string) (*sqlite.WakeReceipt, error) { + return s.repo.GetWakeReceipt(ctx, parentTaskID) +} diff --git a/apps/backend/internal/office/service/wake_metrics.go b/apps/backend/internal/office/service/wake_metrics.go new file mode 100644 index 0000000000..4f53c6c0bf --- /dev/null +++ b/apps/backend/internal/office/service/wake_metrics.go @@ -0,0 +1,83 @@ +package service + +import ( + "expvar" + + "go.uber.org/zap" +) + +// expvar counters published at package init, exposed via stdlib's +// /debug/vars handler (dev mode). Mirrors cost_metrics.go's shape, but as +// plain counters rather than label maps — a parent task ID is the only +// candidate dimension here and it's unbounded, so there is nothing worth +// labeling on. +// +// These are ParentWakeReconciler's only steady-state signal +// (scheduler_wake_reconciler.go): production cannot otherwise tell a +// working sweep (candidates found, nothing to do) from a dead one +// (handler not ticking at all). The steady-state no-op case — a parent +// whose receipt already matches its current child set, whose wake already +// has an active/finished run, or whose assignee is unresolved/paused/ +// stopped/pending-approval — is filtered out in ListStuckParents' SQL +// before a tick ever sees it, so parent_wake_candidates_total itself +// staying flat is that steady-state signal; there is no separate +// "unchanged, skipped" counter, because nothing ever reaches Go to skip. +// +// There were briefly four counters here. parent_wake_receipt_stale_total +// was removed: reconcileOne bumped it unconditionally as its first +// statement, with no branch between it and recordWakeCandidate, so the two +// could never diverge for a processed candidate — pure duplication, not a +// second signal. parent_wake_assignee_unresolved_total stays, but its +// trigger narrowed: SQL now excludes the common unresolved/paused/ +// stopped/pending-approval case before Go ever sees the row (see +// ListStuckParents), so this only fires for the residual TOCTOU race +// window between that SELECT and guardAgentStatus's own read a moment +// later — a real, if rare, case (unlike the receipt, an agent's status +// isn't reconciler-exclusive; anything can change it mid-tick). A nonzero +// rate here is itself a signal worth alerting on, since it should be +// close to zero in steady state. +var ( + parentWakeCandidatesTotal = expvar.NewInt("parent_wake_candidates_total") + parentWakeEmittedTotal = expvar.NewInt("parent_wake_emitted_total") + parentWakeAssigneeUnresolvedTotal = expvar.NewInt("parent_wake_assignee_unresolved_total") +) + +const ( + metricWakeCandidate = "wake.metric.candidate" + metricWakeEmitted = "wake.metric.emitted" + metricWakeAssigneeUnresolved = "wake.metric.assignee_unresolved" +) + +// recordWakeCandidate bumps the count of stuck-parent rows the sweep +// observed this tick, before any receipt comparison. +func (s *Service) recordWakeCandidate(parentTaskID string) { + parentWakeCandidatesTotal.Add(1) + if s.logger == nil { + return + } + s.logger.Debug(metricWakeCandidate, zap.String("parent_task_id", parentTaskID)) +} + +// recordWakeEmitted bumps the count of task_children_completed operations +// the reconciler dispatched through the workflow engine. +func (s *Service) recordWakeEmitted(parentTaskID, operationID string) { + parentWakeEmittedTotal.Add(1) + if s.logger == nil { + return + } + s.logger.Info(metricWakeEmitted, + zap.String("parent_task_id", parentTaskID), zap.String("operation_id", operationID)) +} + +// recordWakeAssigneeUnresolved bumps the count of candidates skipped +// because the assignee failed guardAgentStatus's race-window recheck +// after already passing ListStuckParents' SQL-side status filter (see the +// package doc comment above). +func (s *Service) recordWakeAssigneeUnresolved(parentTaskID, reason string) { + parentWakeAssigneeUnresolvedTotal.Add(1) + if s.logger == nil { + return + } + s.logger.Debug(metricWakeAssigneeUnresolved, + zap.String("parent_task_id", parentTaskID), zap.String("reason", reason)) +} diff --git a/apps/backend/internal/runs/repository/sqlite/runs.go b/apps/backend/internal/runs/repository/sqlite/runs.go index 5f99f20bf4..f45c792cb0 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs.go +++ b/apps/backend/internal/runs/repository/sqlite/runs.go @@ -9,13 +9,18 @@ import ( "time" "github.com/google/uuid" + "github.com/jmoiron/sqlx" "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/office/models" "github.com/kandev/kandev/internal/runs/commentkeys" ) -// CreateRun creates a new run queue entry. +// CreateRunTx creates a new run queue entry using a transaction the caller +// owns. Callers that need to combine the insert with other transactional +// writes (for example the parent-wake reconciler's receipt upsert) use +// this directly; CreateRun wraps it with a private transaction for the +// common single-statement case. // // ContinuationScope is decided here, once, before the row exists to be // coalesced into — never at read or write time downstream. A routine @@ -23,7 +28,7 @@ import ( // only ever patches context_snapshot, so every later reader/writer of // this run's continuation summary reads the value persisted here instead // of re-deriving it against a snapshot that may have drifted. -func (r *Repository) CreateRun(ctx context.Context, req *models.Run) error { +func (r *Repository) CreateRunTx(ctx context.Context, tx *sqlx.Tx, req *models.Run) error { if req.ID == "" { req.ID = uuid.New().String() } @@ -31,7 +36,7 @@ func (r *Repository) CreateRun(ctx context.Context, req *models.Run) error { req.RequestedAt = time.Now().UTC() req.ContinuationScope = models.ContinuationScopeForRun(req, req.AgentProfileID) - _, err := r.db.ExecContext(ctx, r.db.Rebind(` + _, err := tx.ExecContext(ctx, tx.Rebind(` INSERT INTO runs ( id, agent_profile_id, reason, payload, status, coalesced_count, idempotency_key, context_snapshot, capabilities, input_snapshot, @@ -46,6 +51,21 @@ func (r *Repository) CreateRun(ctx context.Context, req *models.Run) error { return err } +// CreateRun creates a new run queue entry in a transaction owned by this +// method. See CreateRunTx for the field-defaulting / continuation-scope +// derivation this delegates to. +func (r *Repository) CreateRun(ctx context.Context, req *models.Run) error { + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if err := r.CreateRunTx(ctx, tx, req); err != nil { + return err + } + return tx.Commit() +} + func ensureRunDefaults(req *models.Run) { if req.Payload == "" { req.Payload = "{}"