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 <carlosmflorencio@gmail.com>
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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.
|
||||
+187
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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`.
|
||||
Reference in New Issue
Block a user