fix(office): recover lost children_completed wakes for stalled parents (#3103)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-08-28 15:54:39 +00:00
committed by GitHub
parent 1744c8b830
commit 061c7d1bca
15 changed files with 1824 additions and 6 deletions
+2 -1
View File
@@ -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))
+3 -1
View File
@@ -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}
}
@@ -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 {
@@ -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
}
@@ -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
@@ -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(`
@@ -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()
@@ -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()
}
@@ -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)
}
}
}
@@ -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"`
@@ -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[:]))
}
@@ -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())
}
@@ -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)
}
@@ -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))
}
@@ -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 = "{}"