From 82d52844b7985b9a42065675fa61e2add2378ae1 Mon Sep 17 00:00:00 2001 From: Henry Date: Tue, 22 Sep 2026 16:40:21 +0800 Subject: [PATCH] fix: start the agent again when a task is sent back a workflow step (#3852) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- ...ent_handlers_pending_move_backward_test.go | 154 +++++++++++++++ .../event_handlers_pending_move_test.go | 21 +- .../orchestrator/event_handlers_workflow.go | 10 +- .../internal/orchestrator/workflow_store.go | 19 +- .../plan.md | 148 ++++++++++++++ ...k-01-thread-deferred-move-transition-id.md | 187 ++++++++++++++++++ ...deferred-move-onenter-dispatch-identity.md | 57 ++++++ ...deferred-move-onenter-dispatch-identity.md | 61 ++++++ 8 files changed, 640 insertions(+), 17 deletions(-) create mode 100644 apps/backend/internal/orchestrator/event_handlers_pending_move_backward_test.go create mode 100644 docs/plans/deferred-move-onenter-dispatch-identity/plan.md create mode 100644 docs/plans/deferred-move-onenter-dispatch-identity/task-01-thread-deferred-move-transition-id.md create mode 100644 docs/specs/tasks/requirements/deferred-move-onenter-dispatch-identity.md create mode 100644 docs/specs/tasks/system-design/deferred-move-onenter-dispatch-identity.md diff --git a/apps/backend/internal/orchestrator/event_handlers_pending_move_backward_test.go b/apps/backend/internal/orchestrator/event_handlers_pending_move_backward_test.go new file mode 100644 index 0000000000..bea4fcf234 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_pending_move_backward_test.go @@ -0,0 +1,154 @@ +package orchestrator + +import ( + "testing" + "time" + + "github.com/jmoiron/sqlx" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// TestPendingMove_BackwardMoveDispatchesOnEnterWithStaleRoute reproduces the +// production bug: a deferred move_task_kandev applied while the task already +// carries a workflow_session_route naming a step OTHER than the move's +// destination — the state any task is in after at least one prior workflow +// entry — commits the step change, but processStepExitAndEnterForDeferredMove +// passes a hardcoded transitionID of 0 into processOnEnter. The staleness +// guard's "may replace a stale route" escape hatch requires a positive entry +// ID to recognize a legitimate callback, so with 0 it falls through to the +// plain route-destination comparison, which fails, and on_enter +// (auto_start_agent) is silently skipped even though the step change itself +// commits. +// +// A route is deliberately seeded here rather than reached by two prior real +// transitions: without an established route, the guard's legacy no-route +// compatibility path lets any entry ID — including the buggy 0 — through, +// which is why a task's very first deferred move always "accidentally" +// worked and the existing scenario tests never caught this. +func TestPendingMove_BackwardMoveDispatchesOnEnterWithStaleRoute(t *testing.T) { + sc := buildPendingMoveScenario(t) + assertBackwardMoveDispatchesOnEnter(t, sc) +} + +func TestPendingMove_BackwardMoveDispatchesOnEnterWithStaleRoute_AtomicQueue(t *testing.T) { + sc := buildPendingMoveScenarioWithQueue(t, true) + assertBackwardMoveDispatchesOnEnter(t, sc) +} + +func assertBackwardMoveDispatchesOnEnter(t *testing.T, sc *pendingMoveScenario) { + t.Helper() + + task, err := sc.repo.GetTask(sc.ctx, "task-1") + if err != nil { + t.Fatalf("load task: %v", err) + } + task.Metadata = map[string]interface{}{ + models.MetaKeyWorkflowSessionRoute: models.WorkflowSessionRoute{ + OperationID: "seed-in-review-route", + DestinationStepID: stepInReviewID, + TargetKind: workflowSessionRouteTargetProfile, + DestinationID: sc.reviewSessionID, + Phase: workflowSessionRouteCommitted, + }, + } + if err := sc.repo.UpdateTask(sc.ctx, task); err != nil { + t.Fatalf("seed workflow session route: %v", err) + } + + session, err := sc.repo.GetTaskSession(sc.ctx, sc.reviewSessionID) + if err != nil { + t.Fatalf("load review session: %v", err) + } + + // The backward move: Review bounces the task back to In Progress, exactly + // the production reproducer (a step whose position precedes the current + // one). + move, found, err := sc.svc.messageQueue.GetPendingMoveWithError(sc.ctx, sc.reviewSessionID) + if err != nil { + t.Fatalf("load pending move: %v", err) + } + if !found || move == nil { + t.Fatal("expected a pending move for the review session") + } + sc.svc.applyPendingMove(sc.ctx, "task-1", sc.reviewSessionID, session, move) + + sc.waitForTaskStep(t, stepInProgressID, 2*time.Second) + + // THE ACTUAL ASSERTION: on_enter's auto_start_agent action must dispatch, + // not just the step change. Before the fix, processOnEnter's staleness + // guard rejects this callback (the hardcoded transitionID=0 can't + // identify itself against the seeded stale route) and no session is ever + // started. + if !sc.waitForAgentStart(2 * time.Second) { + t.Fatal("StartAgentProcess was never called — on_enter auto_start_agent did not dispatch after the backward deferred move") + } + + sessions, err := sc.repo.ListTaskSessions(sc.ctx, "task-1") + if err != nil { + t.Fatalf("list sessions: %v", err) + } + var freshImpl *models.TaskSession + for _, s := range sessions { + if s.AgentProfileID == profileImpl && s.ID != sc.implSessionID { + freshImpl = s + } + } + if freshImpl == nil { + t.Fatal("expected a fresh impl-profile session after the backward deferred move dispatched on_enter") + } + if !freshImpl.IsPrimary { + t.Error("fresh impl session must be primary after on_enter dispatches auto_start_agent") + } +} + +// waitForTaskStep polls the task's workflow_step_id until it reaches want and +// fails the test if the deadline elapses first. +func (sc *pendingMoveScenario) waitForTaskStep(t *testing.T, want string, timeout time.Duration) *models.Task { + t.Helper() + deadline := time.Now().Add(timeout) + var task *models.Task + for { + var err error + task, err = sc.repo.GetTask(sc.ctx, "task-1") + if err != nil { + t.Fatalf("load task: %v", err) + } + if task.WorkflowStepID == want { + return task + } + if time.Now().After(deadline) { + t.Fatalf("workflow_step_id = %q, want %q after %s", task.WorkflowStepID, want, timeout) + return task + } + time.Sleep(10 * time.Millisecond) + } +} + +// newSQLiteQueueForTaskRepo shares the task repository's database so the +// deferred transition takes the production atomic task-and-queue path. +func newSQLiteQueueForTaskRepo(t *testing.T, repo *sqliterepo.Repository) *messagequeue.Service { + t.Helper() + db := sqlx.NewDb(repo.DB(), "sqlite3") + queueRepo, err := messagequeue.NewSQLiteRepository(db, db) + if err != nil { + t.Fatalf("create SQLite message queue repository: %v", err) + } + return messagequeue.NewService(queueRepo, messagequeue.DefaultMaxPerSession, testLogger()) +} + +// waitForAgentStart polls the mock agent manager until StartAgentProcess has +// been called at least once or the deadline elapses. +func (sc *pendingMoveScenario) waitForAgentStart(timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for { + sc.agentMgr.mu.Lock() + started := len(sc.agentMgr.startAgentProcessCalls) > 0 + sc.agentMgr.mu.Unlock() + if started || time.Now().After(deadline) { + return started + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go b/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go index aa89af7a57..4e16d3f6a6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_pending_move_test.go @@ -351,6 +351,10 @@ type pendingMoveScenario struct { // - Mock LaunchAgent that fires the boot signal asynchronously so the // resume path can complete in tests without a real agent process. func buildPendingMoveScenario(t *testing.T) *pendingMoveScenario { + return buildPendingMoveScenarioWithQueue(t, false) +} + +func buildPendingMoveScenarioWithQueue(t *testing.T, useSQLiteQueue bool) *pendingMoveScenario { t.Helper() ctx := context.Background() now := time.Now().UTC() @@ -390,6 +394,10 @@ func buildPendingMoveScenario(t *testing.T) *pendingMoveScenario { log := testLogger() exec := executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}) sched := scheduler.NewScheduler(queue.NewTaskQueue(100), exec, taskRepo, log, scheduler.SchedulerConfig{}) + messageQueueService := newAuthoritativeMemoryQueue(repo, log) + if useSQLiteQueue { + messageQueueService = newSQLiteQueueForTaskRepo(t, repo) + } svc := &Service{ logger: log, @@ -397,7 +405,7 @@ func buildPendingMoveScenario(t *testing.T) *pendingMoveScenario { workflowStepGetter: stepGetter, taskRepo: taskRepo, agentManager: agentMgr, - messageQueue: newAuthoritativeMemoryQueue(repo, log), + messageQueue: messageQueueService, executor: exec, scheduler: sched, } @@ -413,10 +421,15 @@ func buildPendingMoveScenario(t *testing.T) *pendingMoveScenario { ); err != nil { t.Fatalf("queue hand-off prompt: %v", err) } + reviewSession, err := repo.GetTaskSession(ctx, reviewSessionID) + if err != nil { + t.Fatalf("load review session for pending move: %v", err) + } if err := svc.messageQueue.SetPendingMove(ctx, reviewSessionID, &messagequeue.PendingMove{ - TaskID: "task-1", - WorkflowID: "wf1", - WorkflowStepID: stepInProgressID, + SessionIncarnationID: reviewSession.QueueIncarnationID, + TaskID: "task-1", + WorkflowID: "wf1", + WorkflowStepID: stepInProgressID, }); err != nil { t.Fatalf("set pending move: %v", err) } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index b2ec6b4255..b3f80b0204 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -4935,12 +4935,13 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string } deferredMoveCtx := steptelemetry.WithAttribution(ctx, deferredMoveAttribution) var transitionErr error + var transitionID int64 if s.messageQueue.SupportsAtomicDeferredMoveTransition() { - transitionErr = s.workflowStore.ApplyDeferredMoveTransition( + transitionID, transitionErr = s.workflowStore.ApplyDeferredMoveTransition( deferredMoveCtx, taskID, sessionID, fromStepID, move.WorkflowStepID, move.MoveID, record, ) } else { - transitionErr = s.workflowStore.applyTransition( + transitionID, transitionErr = s.workflowStore.applyTransition( deferredMoveCtx, taskID, sessionID, fromStepID, move.WorkflowStepID, engine.TriggerOnEnter, move.MoveID, nil, ) @@ -5004,7 +5005,7 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string taskDescription := task.Description go s.processStepExitAndEnterForDeferredMove( context.WithoutCancel(ctx), identity, freshSession, - fromStepID, move.WorkflowStepID, taskDescription, move.EntryOptions, + fromStepID, move.WorkflowStepID, taskDescription, move.EntryOptions, transitionID, ) } @@ -5053,6 +5054,7 @@ func (s *Service) processStepExitAndEnterForDeferredMove( session *models.TaskSession, fromStepID, toStepID, taskDescription string, entryOptions *workflowmove.EntryOptions, + transitionID int64, ) { current, err := s.messageQueue.ResolveSessionIdentity(ctx, identity.TaskID, identity.SessionID) if err != nil || current != identity || session.QueueIncarnationID != identity.SessionIncarnationID { @@ -5083,7 +5085,7 @@ func (s *Service) processStepExitAndEnterForDeferredMove( if entryOptions != nil { entryStep = workflowmove.OverlayStep(targetStep, entryOptions) } - s.processOnEnter(ctx, identity.TaskID, fresh, entryStep, taskDescription, 0, fromStep) + s.processOnEnter(ctx, identity.TaskID, fresh, entryStep, taskDescription, transitionID, fromStep) } func (s *Service) removePendingMoveHandoffPromptForSession( diff --git a/apps/backend/internal/orchestrator/workflow_store.go b/apps/backend/internal/orchestrator/workflow_store.go index 5be555924e..34462440a3 100644 --- a/apps/backend/internal/orchestrator/workflow_store.go +++ b/apps/backend/internal/orchestrator/workflow_store.go @@ -289,14 +289,15 @@ func (s *workflowStore) LoadPreviousStep(ctx context.Context, workflowID string, } func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID, fromStepID, toStepID string, trigger engine.Trigger) error { - return s.applyTransition(ctx, taskID, sessionID, fromStepID, toStepID, trigger, "", nil) + _, err := s.applyTransition(ctx, taskID, sessionID, fromStepID, toStepID, trigger, "", nil) + return err } func (s *workflowStore) ApplyDeferredMoveTransition( ctx context.Context, taskID, sessionID, fromStepID, toStepID, moveID string, record messagequeue.PendingMoveRecord, -) error { +) (int64, error) { return s.applyTransition( ctx, taskID, sessionID, fromStepID, toStepID, engine.TriggerOnEnter, moveID, &record, ) @@ -347,14 +348,14 @@ func (s *workflowStore) applyTransition( trigger engine.Trigger, moveID string, deferredMove *messagequeue.PendingMoveRecord, -) error { +) (int64, error) { task, err := s.repo.GetTask(ctx, taskID) if err != nil { - return fmt.Errorf("load task for transition: %w", err) + return 0, fmt.Errorf("load task for transition: %w", err) } targetStep, err := s.workflowStepGetter.GetStep(ctx, toStepID) if err != nil { - return fmt.Errorf("load target step for transition: %w", err) + return 0, fmt.Errorf("load target step for transition: %w", err) } // Keep WorkflowID in sync with the target step's owning workflow. Most // callers transition within the same workflow (targetStep.WorkflowID == @@ -362,7 +363,7 @@ func (s *workflowStore) applyTransition( // cross-workflow move_task_kandev hand-offs too — without this, the task // would end up with a step ID from a workflow its WorkflowID doesn't match. if err := markDeferredMoveApplied(task, moveID); err != nil { - return err + return 0, err } oldWorkflowID := task.WorkflowID @@ -390,10 +391,10 @@ func (s *workflowStore) applyTransition( if err := s.updateDeferredTransitionTask( transitionCtx, task, fromStepID, targetStep, *deferredMove, ); err != nil { - return fmt.Errorf("update task workflow step: %w", err) + return 0, fmt.Errorf("update task workflow step: %w", err) } } else if err := s.updateTransitionTask(transitionCtx, task, fromStepID, targetStep); err != nil { - return fmt.Errorf("update task workflow step: %w", err) + return 0, fmt.Errorf("update task workflow step: %w", err) } // Pass the pre-move workflow ID through so cross-workflow transitions @@ -418,7 +419,7 @@ func (s *workflowStore) applyTransition( s.pullNextTaskOnVacate(ctx, fromStepID, taskID) - return nil + return task.WorkflowStepTransitionID, nil } func (s *workflowStore) carryStepHandoffForTransition( diff --git a/docs/plans/deferred-move-onenter-dispatch-identity/plan.md b/docs/plans/deferred-move-onenter-dispatch-identity/plan.md new file mode 100644 index 0000000000..fdab2a4aae --- /dev/null +++ b/docs/plans/deferred-move-onenter-dispatch-identity/plan.md @@ -0,0 +1,148 @@ +--- +created: 2026-09-21 +status: implemented +requirements: + - REQ-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001 +system_design: + - ../../specs/tasks/system-design/deferred-move-onenter-dispatch-identity.md +legacy_specs: + - ../../specs/workflow-on-enter-action-dispatch/spec.md + - ../../specs/tasks/system-design/workflow-task-step-transition-ledger.md +--- + +# Implementation Plan: Deferred move on_enter dispatch identity + +## Outcome + +A deferred `move_task_kandev` (E2, applied at turn end via `applyPendingMove` -> +`processStepExitAndEnterForDeferredMove`) currently commits the workflow-step +transition and then has its own `on_enter` dispatch rejected as stale, so +`auto_start_agent` (and any other on_enter action) never runs. This fix threads +the transition ledger ID that the deferred-move commit already produces into +`processOnEnter`, so the existing "may replace a stale route" escape hatch +(`workflowEntryDispatchMayReplaceRoute` in `ceiling_entry.go`, added for #3755) +recognises this callback as the current, legitimate entry instead of falling +through to the route-mismatch rejection. + +## Root cause + +`processStepExitAndEnterForDeferredMove` (`apps/backend/internal/orchestrator/event_handlers_workflow.go:5086`) +calls `s.processOnEnter(ctx, identity.TaskID, fresh, entryStep, taskDescription, 0, fromStep)` +with a hardcoded `transitionID` of `0`. `processOnEnter`'s staleness guard +(`workflowEntryDispatchIsCurrentForSession` -> `workflowEntryDispatchRouteIsCurrent` +in `ceiling_entry.go:303`) checks `workflowEntryDispatchMayReplaceRoute` first, +which requires `entryIDs[0] > 0` to match the callback against +`task.WorkflowStepTransitionID` via `workflowEntryDispatchMatchesCurrentTaskEntry` +(`ceiling_entry.go:367`). With `0`, that check short-circuits to `false` +(`ceiling_entry.go:373`: `len(entryIDs) == 0 || entryIDs[0] <= 0`), so the guard +falls through to comparing the task's stored `workflow_session_route` against +the destination step. That route is only updated to name the destination +*inside* `maybySwitchSessionForProfile`, which runs **after** the staleness +guard (PR #3826 moved route preparation there). The route therefore still +names the source step, the comparison fails, and `processOnEnter` returns +before dispatching `on_enter`. + +The transition ledger ID this callback needs already exists: the deferred-move +commit path (`workflowStore.applyTransition`, called via +`ApplyDeferredMoveTransition` or the unfenced `applyTransition` fallback) writes +it via `updateTaskTx` (`apps/backend/internal/task/repository/sqlite/task.go:1258`, +`task.WorkflowStepTransitionID = transitionID`) into the same `*models.Task` +value `applyTransition` holds locally — but neither `ApplyTransition` nor +`ApplyDeferredMoveTransition` return it, so it is discarded before reaching +`applyPendingMove` and the goroutine it spawns. + +The forward-move path (`applyEngineTransitionWithCommitMode`, +`event_handlers_workflow.go:7666`) already solves this correctly for `E1`, +using the existing `stepentry.AllocationResult` / `stepentry.WithResultHolder` +context out-parameter to capture `stepEntry.TransitionID` and thread it into +`launchProcessOnEnter`. That mechanism is deliberately **not** reused here: it +also causes `transitionContext` to attach a `stepentry.PendingAllocation` for +ledger-owned marker-bearing on_enter actions +(`workflow_store.go:460`), which is unrelated behavior this fix must not +introduce as a side effect on the deferred-move path. Instead, `applyTransition` +is changed to also return the `int64` transition ID it already computes, and +that value is threaded explicitly through the two existing call sites down to +`processOnEnter`. + +## Requirement and system-design mapping + +- **AC-A1**, **AC-A2** (`docs/specs/workflow-on-enter-action-dispatch/spec.md`): + every declared `on_enter` action must execute, in order, exactly once per + step entry, on every entry path including `E2` (manual move, which includes + both immediate and deferred `move_task_kandev`). A backward deferred move + into a step declaring `auto_start_agent` currently executes none of its + `on_enter` actions, violating both criteria for that entry path. +- **Ledger identity contract** (`docs/specs/tasks/system-design/workflow-task-step-transition-ledger.md`): + `task.WorkflowStepTransitionID` is the immutable per-write ledger identity + the repository layer populates after every step-changing write, including + the `mcp_deferred_move` trigger. This fix is purely a consumer-side gap: the + identity is already produced correctly and durably; it was never threaded to + the one caller that needs it to identify its own dispatch as current. + +## Constraints + +- `event_handlers_workflow.go` / `workflow_session_target.go` / `ceiling_entry.go` + are under heavy concurrent upstream churn (8 commits by carlosflorencio, + 2026-09-10 to 2026-09-20). Keep the diff to call-site plumbing only. +- Do not weaken, bypass, or delete `workflowEntryDispatchIsCurrentForSession` or + `workflowEntryDispatchMayReplaceRoute`. They correctly reject a genuinely + stale callback (AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.6-style protection); + the defect is that a legitimate entry had no way to identify itself. +- `engine.TransitionStore.ApplyTransition` is a broad interface implemented by + several test fakes across `internal/workflow/engine`. Its public signature + (`error` only) is unchanged. Only `workflowStore.ApplyDeferredMoveTransition` + — which has exactly one production call site and no direct test fakes — gains + an `int64` return. + +## Architecture + +`workflowStore.applyTransition` (private, `workflow_store.go:344`) already +leaves the transition ID on its local `task.WorkflowStepTransitionID` after a +successful `updateDeferredTransitionTask`/`updateTransitionTask` call. Change +its signature to `(int64, error)`, returning `task.WorkflowStepTransitionID` on +success and `0` alongside every existing error return. + +- `workflowStore.ApplyTransition` (public, part of `engine.TransitionStore`): + keep its `error`-only signature; discard the new first return value. +- `workflowStore.ApplyDeferredMoveTransition` (public, one call site): change + to return `(int64, error)`. +- `Service.applyPendingMove` (`event_handlers_workflow.go:4938`): capture the + transition ID from both the atomic (`ApplyDeferredMoveTransition`) and + unfenced-fallback (`applyTransition`) branches. +- `Service.processStepExitAndEnterForDeferredMove` (`event_handlers_workflow.go:5050`): + add a `transitionID int64` parameter, pass it into the `processOnEnter` call + at line 5086 instead of the hardcoded `0`. +- `applyPendingMove`'s call to `processStepExitAndEnterForDeferredMove` + (`event_handlers_workflow.go:5005`): pass the captured transition ID through. + +`processStepExitForDeferredMove` (the WIP-queued branch, which does not call +`processOnEnter`) is unchanged — its later promotion re-enters through the +ledger dispatcher (`DispatcherLedger`), which already has its own dispatch +identity from the promotion write, not from this call. + +## Verification strategy + +A regression test at the `orchestrator` package level drives a deferred +`move_task_kandev` backward transition (source step B -> destination step A, +where A declares `on_enter: auto_start_agent`) end to end through +`applyPendingMove`, and asserts the auto-start dispatch actually ran (not just +that `tasks.workflow_step_id` changed) — matching the existing shape of +`TestApplyTransitionRecordsEngineTransitionFromSessionID` and neighboring tests +in `workflow_step_ledger_test.go`, but exercising the deferred-move goroutine +path (`processStepExitAndEnterForDeferredMove`) specifically, since the direct +`ApplyTransition`/`applyTransition` ledger tests do not reach `processOnEnter` +at all. + +## Work orders + +- [task-01-thread-deferred-move-transition-id.md](task-01-thread-deferred-move-transition-id.md) + +## Risks + +- Low blast radius: `ApplyDeferredMoveTransition` has one production caller and + no test fakes to update. `applyTransition` is private with two callers + besides the public wrappers, both in this same file. +- The unfenced-fallback branch (`SupportsAtomicDeferredMoveTransition() == + false`) is exercised by the in-memory message queue backend used in tests; + the regression test should cover the path actually reachable in the test + environment and note which backend it exercises. diff --git a/docs/plans/deferred-move-onenter-dispatch-identity/task-01-thread-deferred-move-transition-id.md b/docs/plans/deferred-move-onenter-dispatch-identity/task-01-thread-deferred-move-transition-id.md new file mode 100644 index 0000000000..cc53a54808 --- /dev/null +++ b/docs/plans/deferred-move-onenter-dispatch-identity/task-01-thread-deferred-move-transition-id.md @@ -0,0 +1,187 @@ +--- +id: "01-thread-deferred-move-transition-id" +title: "Thread the deferred-move transition ID into processOnEnter" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001 +acceptance_criteria: + - AC-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001.1 + - AC-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001.2 +system_design: + - ../../specs/tasks/system-design/deferred-move-onenter-dispatch-identity.md +--- + +# Work Order 01: Thread the deferred-move transition ID into processOnEnter + +## Outcome + +`processStepExitAndEnterForDeferredMove` stops passing a hardcoded `0` as the +transition/entry ID to `processOnEnter`. It receives and forwards the real +ledger transition ID that `applyPendingMove` already produces when it commits +the deferred move, so the staleness guard's existing "may replace a stale +route" escape hatch recognises the callback as current and `on_enter` actually +dispatches `on_enter` actions (in particular `auto_start_agent`) for a +backward (or any) deferred `move_task_kandev`. + +## In scope + +- `apps/backend/internal/orchestrator/workflow_store.go`: + - `applyTransition` (private): change return type from `error` to + `(int64, error)`. Return `0` alongside every existing error return; on + success return `task.WorkflowStepTransitionID` (populated by + `updateDeferredTransitionTask`/`updateTransitionTask` via `updateTaskTx`). + - `ApplyTransition` (public, part of `engine.TransitionStore`): keep its + `error`-only signature; discard the new first return value from + `applyTransition`. + - `ApplyDeferredMoveTransition` (public): change return type to + `(int64, error)`, returning `applyTransition`'s two values directly. +- `apps/backend/internal/orchestrator/event_handlers_workflow.go`: + - `applyPendingMove`: capture the transition ID from both branches — the + atomic `ApplyDeferredMoveTransition` call and the unfenced-fallback + `applyTransition` call — into a local `transitionID int64`. + - Thread `transitionID` through the existing call to + `processStepExitAndEnterForDeferredMove` (replacing no existing argument; + add it as a new trailing parameter). + - `processStepExitAndEnterForDeferredMove`: add a `transitionID int64` + parameter; pass it (instead of the current hardcoded `0`) as the + `transitionID` argument to the `s.processOnEnter(...)` call at the end of + the function. + +## Out of scope + +- `processStepExitForDeferredMove` (the WIP-queued branch) — it never calls + `processOnEnter`; no change needed. +- Any change to `workflowEntryDispatchIsCurrentForSession`, + `workflowEntryDispatchMayReplaceRoute`, or any other staleness-guard logic + in `ceiling_entry.go`. +- Reusing the `stepentry.AllocationResult` / `WithResultHolder` context + mechanism the forward-move (`E1`) path uses — it has the side effect of also + attaching a `stepentry.PendingAllocation` for ledger-owned marker-bearing + on_enter actions via `transitionContext`, which is unrelated behavior this + fix must not introduce on the deferred-move path. +- Any change to `engine.TransitionStore`'s interface signature or its test + fakes in `internal/workflow/engine/*_test.go`. + +## Applicable requirements + +- **AC-A1**, **AC-A2** — `docs/specs/workflow-on-enter-action-dispatch/spec.md` +- **Ledger identity contract** — `docs/specs/tasks/system-design/workflow-task-step-transition-ledger.md` + +## Acceptance conditions + +1. A deferred `move_task_kandev` into a step whose `on_enter` declares + `auto_start_agent` actually dispatches that action (asserted via a real + launch/start-agent call, not just `tasks.workflow_step_id`), for both a + forward and a backward transition. +2. `workflowEntryDispatchIsCurrentForSession` still rejects a callback for a + transition that is not the task's current one (no regression to the #3755 + staleness protection) — covered by the existing ledger/staleness tests in + `workflow_step_ledger_test.go` and `ceiling_entry_test.go` (or equivalent), + which must continue to pass unmodified. +3. `golangci-lint` and the package build are clean for the changed files. + +## Regression test + +Add a test in `apps/backend/internal/orchestrator/` (new file or alongside +`event_handlers_pending_move_test.go`) that reuses `buildPendingMoveScenario`'s +shape: two steps each declaring `on_enter: auto_start_agent`, drive +`svc.applyPendingMove` for a **backward** move (mirroring the production +reproducer — Review bounced back to an earlier step), let the spawned +`processStepExitAndEnterForDeferredMove` goroutine settle, and assert the +destination step's agent was actually started — e.g. via `startAgentProcessFunc` +/ `launchAgentFunc` invocation tracking on `mockAgentManager`, following the +`wireBootReadySimulator` pattern already used by +`TestPendingMove_ReviewToInProgress_OneTransitionOnly`. Assert the dispatch, +not merely that `task.WorkflowStepID` changed — the step change already works +today and is what makes the missing dispatch invisible without this assertion. + +Confirm the test fails against the current code (hardcoded `0`) before the fix +and passes after. + +## Verification commands + +```bash +cd apps/backend +go test ./internal/orchestrator/... -run 'PendingMove|DeferredMove' -v +go test ./internal/orchestrator/... -run 'ApplyTransition' -v +go build ./... +golangci-lint run ./internal/orchestrator/... --new-from-rev=HEAD~1 --timeout=5m +``` + +## Likely files + +- `apps/backend/internal/orchestrator/workflow_store.go` +- `apps/backend/internal/orchestrator/event_handlers_workflow.go` +- `apps/backend/internal/orchestrator/event_handlers_pending_move_test.go` (or + a new sibling test file) + +## Dependencies + +None — this is a self-contained call-site change with no upstream work-order +dependency. + +## Results + +Implemented as designed: + +- `workflow_store.go`: private `applyTransition` returns `(int64, error)`, + returning `0` on every existing error path and `task.WorkflowStepTransitionID` + on success. Public `ApplyTransition` discards the new first return value, + keeping its `error`-only signature for `engine.TransitionStore`. Public + `ApplyDeferredMoveTransition` now returns `(int64, error)`. +- `event_handlers_workflow.go`: `applyPendingMove` captures `transitionID` from + both the atomic (`ApplyDeferredMoveTransition`) and unfenced-fallback + (`applyTransition`) branches, and threads it through the existing call to + `processStepExitAndEnterForDeferredMove`. That function gained a + `transitionID int64` parameter, passed into `processOnEnter` instead of the + hardcoded `0`. +- New regression test: + `apps/backend/internal/orchestrator/event_handlers_pending_move_backward_test.go` + (`TestPendingMove_BackwardMoveDispatchesOnEnterWithStaleRoute`), in a new + file per the 800-line test-file convention. It seeds a + `workflow_session_route` naming a step other than the move's destination + (the state any task is in after its first workflow entry — which is exactly + why the pre-existing `TestPendingMove_ReviewToInProgress_OneTransitionOnly` + scenario, with no route ever established, never caught this bug), drives a + backward `applyPendingMove`, and asserts `StartAgentProcess` was actually + called — not just that `workflow_step_id` changed. + - Confirmed RED against pre-fix code: failed with "StartAgentProcess was + never called...". + - Confirmed GREEN after the fix. + +No changes to `ceiling_entry.go` or any staleness-guard logic — the existing +`workflowEntryDispatchIsCurrentForSession` / `workflowEntryDispatchMayReplaceRoute` +protections are exercised unmodified and still pass +(`TestReplayCeilingDeferralRejectsStaleWorkflowEntryBeforeDispatch`, +`TestWorkflowEntryDispatchRejectsStaleCommittedRoute`). + +Commands run (from `apps/backend` unless noted): + +``` +go build -tags fts5 ./... +go test ./internal/orchestrator/ -run 'PendingMove|DeferredMove|ApplyTransition|Ceiling|WorkflowEntryDispatch' -v -timeout=180s -p 1 +golangci-lint run ./internal/orchestrator/... --new-from-rev=a7f7e01af8c04625e5378de6d69fd1756ef2937c --timeout=5m +make fmt # (root) — reverted two unrelated gofmt-drift files by explicit path +make typecheck # (root) +make lint-format # (root) +cd apps/web && pnpm run i18n:ratchet +``` + +Note on `make test` (full-suite `go test ./...`): the whole-repo run is +resource-contention-flaky in this sandbox — a bare `-timeout=10m` panic +surfaced in unrelated SQLite-migration setup code +(`migrateSubagentContextBackfillUnsafe`) inside completely unrelated tests, +alongside failures in packages with no relation to this change (launcher, +subproc, agentctl/server/process, k8s/docker/ssh-dependent tests). Verified +non-diff-caused: the same targeted test set passes cleanly at +`-p 1` (`go test ./internal/orchestrator/ -run '...' -p 1` → `ok`, all PASS +including the new regression test), and re-running the identical targeted +command plus a spot-check of an unrelated failing package +(`internal/common/subproc`, `TestRunGitOutputAfterAcquireStartsTimeoutAfterAdmission`) +against `git merge-base HEAD origin/main` (`a7f7e01af8`, a scratch worktree at +`/tmp/kandev-scratch-verify/kandev`, removed after verification) also passed +cleanly with default parallelism. The hang is load-dependent and reproduces +independently of this diff. diff --git a/docs/specs/tasks/requirements/deferred-move-onenter-dispatch-identity.md b/docs/specs/tasks/requirements/deferred-move-onenter-dispatch-identity.md new file mode 100644 index 0000000000..813d09eecd --- /dev/null +++ b/docs/specs/tasks/requirements/deferred-move-onenter-dispatch-identity.md @@ -0,0 +1,57 @@ +--- +status: draft +system: tasks +created: 2026-09-21 +owners: + - nova28 +--- + +# Deferred Move on_enter Dispatch Identity Requirements + +## Overview + +A deferred `move_task_kandev` (a backward or otherwise non-immediate board +move, applied at turn end) commits a workflow-step transition and then +dispatches the destination step's `on_enter` actions itself. That dispatch +must be able to identify itself as the current entry using the transition it +just committed, so the destination step's declared `on_enter` actions +(including `auto_start_agent`) actually run instead of being rejected by the +staleness guard as if they belonged to an earlier, superseded entry. + +## Requirements + +### REQ-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001: Deferred move on_enter dispatch identity + +**Intent:** A deferred move's own `on_enter` dispatch carries the identity of +the transition it just committed, so the staleness guard recognizes it as +current rather than rejecting it as stale. + +#### Acceptance criteria + +- **AC-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001.1:** When a deferred + `move_task_kandev` commits a workflow-step transition, the system shall pass + that transition's ledger identity to the destination step's `on_enter` + dispatch. +- **AC-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001.2:** When the + `on_enter` staleness guard evaluates a deferred-move dispatch that carries + its own just-committed transition identity, the system shall recognize it + as current and execute the destination step's declared `on_enter` actions, + including `auto_start_agent`, in declared order. A dispatch that does not + carry the current transition identity shall remain rejected. + +## Out of scope + +- The staleness guard's rejection logic itself + (`workflowEntryDispatchMayReplaceRoute`, `workflowEntryDispatchIsCurrentForSession`), + which correctly rejects a genuinely stale callback and is unchanged by this + requirement. +- The forward-move (`E1`, agent-turn auto-advance) `on_enter` dispatch path, + which already threads its own transition identity through + `stepentry.AllocationResult`. +- The broader multi-path `on_enter` dispatch unification (covering + `switch_workflow`, WIP-queue promotion, and the full acceptance-criteria set + `AC-A1`-`AC-A13`) tracked informally in + [`workflow-on-enter-action-dispatch/spec.md`](../../workflow-on-enter-action-dispatch/spec.md). + This requirement formalizes only the narrow deferred-move identity gap that + subset (`AC-A1`, `AC-A2` as applied to the `E2` deferred-move entry path) + describes. diff --git a/docs/specs/tasks/system-design/deferred-move-onenter-dispatch-identity.md b/docs/specs/tasks/system-design/deferred-move-onenter-dispatch-identity.md new file mode 100644 index 0000000000..189165adc7 --- /dev/null +++ b/docs/specs/tasks/system-design/deferred-move-onenter-dispatch-identity.md @@ -0,0 +1,61 @@ +--- +status: draft +system: tasks +requirements: + - REQ-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001 +--- + +# Deferred Move on_enter Dispatch Identity System Design + +## Purpose and boundaries + +The task system owns the workflow-step transition ledger and the `on_enter` +dispatch that follows a committed transition. This design covers only the +identity a deferred move's own `on_enter` dispatch presents to the existing +staleness guard. It does not change the guard's decision logic, the ledger's +write path, or the forward-move dispatch path. + +## Requirement mapping + +| Requirement | Design sections | +| --- | --- | +| `REQ-TASKS-DEFERRED-MOVE-ONENTER-DISPATCH-IDENTITY-001` | [Transition identity threading](#transition-identity-threading) | + +## Components and responsibilities + +`workflowStore.applyTransition` (private, `orchestrator/workflow_store.go`) +computes and durably records the ledger transition ID for every +step-changing write, including a deferred move, via `updateTaskTx`. Its +public wrappers `ApplyTransition` (part of `engine.TransitionStore`) and +`ApplyDeferredMoveTransition` previously discarded that value. + +`Service.applyPendingMove` and `Service.processStepExitAndEnterForDeferredMove` +(`orchestrator/event_handlers_workflow.go`) drive the deferred-move commit and +its own `on_enter` dispatch call. + +`processOnEnter`'s staleness guard (`workflowEntryDispatchIsCurrentForSession` +/ `workflowEntryDispatchMayReplaceRoute` in `ceiling_entry.go`) decides +whether an `on_enter` dispatch is the current entry for a session, using the +transition/entry identity the caller supplies. + +## Transition identity threading + +`workflowStore.applyTransition`, `ApplyDeferredMoveTransition`, and the +private `applyTransition` return the `int64` ledger transition ID alongside +their existing error. `ApplyTransition`'s public, broadly-implemented +interface signature is unchanged; only `ApplyDeferredMoveTransition`, which +has exactly one production call site, gains the new return value. + +`applyPendingMove` captures the transition ID from both the atomic +(`ApplyDeferredMoveTransition`) and unfenced-fallback (`applyTransition`) +commit branches, and threads it through `processStepExitAndEnterForDeferredMove` +into `processOnEnter`, replacing a hardcoded `0`. The staleness guard's +existing `workflowEntryDispatchMayReplaceRoute` escape hatch then recognizes +the dispatch as current by matching it against `task.WorkflowStepTransitionID`, +the same identity the transition write already produced. + +This changes only what identity a deferred move's own dispatch presents to +the guard. It does not change the guard's rejection logic for a genuinely +stale callback, and does not affect the forward-move (`E1`) path, which +already threads its own transition identity through +`stepentry.AllocationResult`.