diff --git a/apps/backend/internal/office/engine_dispatcher/dispatcher.go b/apps/backend/internal/office/engine_dispatcher/dispatcher.go index 426bf8494f..ad6cfa0a81 100644 --- a/apps/backend/internal/office/engine_dispatcher/dispatcher.go +++ b/apps/backend/internal/office/engine_dispatcher/dispatcher.go @@ -443,6 +443,35 @@ func isActiveSessionState(state taskmodels.TaskSessionState) bool { // filters by task, so unlike resolveDeciderSessionID (which validates a // caller-supplied session id and must check task ownership itself), there // is no cross-task session id to smuggle in here. +// +// This resolver is deliberately task-scoped, and stays that way after +// features.officeSessionIdentity gives each participant agent its own +// session per task. Picking "the newest session" only stays well-defined +// because EvaluateStepQuorum — its sole caller — evaluates guards off +// TaskID, CurrentStepID and WorkflowID, all derived from the task row +// rather than the session (see orchestrator.assembleMachineState). +// +// MachineState.SessionID, SessionState, Data and IsPassthrough are the +// session-derived fields, so with several live sessions per task "newest" no +// longer names a specific agent's session. That is latent, not live: none of +// the four is read by the guard-evaluation path today (Data is written by +// set_workflow_data through SetSessionMetadataKey and read back into +// MachineState.Data on every state assembly, but never consumed by guard +// evaluation), so which session is chosen is unobservable today. +// +// The first caller that genuinely needs per-session state must session-scope +// its own call — pass the deciding session explicitly, as RecordDecision now +// does via resolveDeciderSessionID — rather than adding a defensive check +// here. Guarding this resolver would make a wrong-scoped read look safe and +// remove the pressure to fix it properly. +// +// The invariant is pinned by +// TestEvaluateStepQuorum_InsensitiveToWhichLiveSessionIsNewest +// (internal/workflow/engine): it evaluates one step through two live +// sessions carrying different SessionID, SessionState and Data, and requires +// an identical snapshot — so a guard whose outcome starts depending on any +// of the three changes that test's result instead of silently resolving +// another agent's session through this function. func (d *Dispatcher) resolveLatestSessionID(ctx context.Context, taskID string) (string, error) { session, err := d.sessions.GetTaskSessionByTaskID(ctx, taskID) if err != nil { diff --git a/apps/backend/internal/workflow/engine/quorum_reevaluation_test.go b/apps/backend/internal/workflow/engine/quorum_reevaluation_test.go index dd8c2f6284..17cdc28b2d 100644 --- a/apps/backend/internal/workflow/engine/quorum_reevaluation_test.go +++ b/apps/backend/internal/workflow/engine/quorum_reevaluation_test.go @@ -3,6 +3,7 @@ package engine import ( "context" "expvar" + "reflect" "testing" "github.com/kandev/kandev/internal/common/logger" @@ -489,3 +490,197 @@ func TestEvaluateStepQuorum_DoesNotRecordSlateEmptySideEffects(t *testing.T) { t.Fatalf("expected no slate-empty warning log from a read-only snapshot, got %d: %+v", len(entries), entries) } } + +// --- B1 companion: EvaluateStepQuorum must not depend on WHICH session --- + +// sessionScopedStateStore mirrors production assembleMachineState +// (orchestrator/event_handlers_workflow.go): CurrentStepID and WorkflowID are +// derived from the TASK row and are therefore identical for every session, +// while SessionID, SessionState, Data and IsPassthrough are genuinely +// per-session — Data being loaded from that session's own +// metadata["workflow_data"] bag. +type sessionScopedStateStore struct { + step StepSpec + dataBySess map[string]map[string]any + stateBySess map[string]string + passthroughBySess map[string]bool + loadedData []map[string]any + loadedStates []string + loadedPassthrough []bool + applied map[string]bool +} + +func (s *sessionScopedStateStore) LoadState(_ context.Context, taskID, sessionID string) (MachineState, error) { + data := s.dataBySess[sessionID] + sessionState := s.stateBySess[sessionID] + isPassthrough := s.passthroughBySess[sessionID] + s.loadedData = append(s.loadedData, data) + s.loadedStates = append(s.loadedStates, sessionState) + s.loadedPassthrough = append(s.loadedPassthrough, isPassthrough) + return MachineState{ + TaskID: taskID, + SessionID: sessionID, + WorkflowID: "wf", + CurrentStepID: "review", + SessionState: sessionState, + IsPassthrough: isPassthrough, + Data: data, + }, nil +} +func (s *sessionScopedStateStore) LoadStep(_ context.Context, _, _ string) (StepSpec, error) { + return s.step, nil +} +func (s *sessionScopedStateStore) LoadNextStep(_ context.Context, _ string, _ int) (StepSpec, error) { + return StepSpec{ID: "approval", Position: 2}, nil +} +func (s *sessionScopedStateStore) LoadPreviousStep(_ context.Context, _ string, _ int) (StepSpec, error) { + return StepSpec{}, nil +} +func (s *sessionScopedStateStore) ApplyTransition(_ context.Context, _, _, _, _ string, _ Trigger) error { + return nil +} +func (s *sessionScopedStateStore) ApplyTransitionIfAtStep( + _ context.Context, _, _, _, _ string, _ Trigger, +) (bool, error) { + return false, nil +} +func (s *sessionScopedStateStore) PersistData(_ context.Context, _ string, _ map[string]any) error { + return nil +} +func (s *sessionScopedStateStore) IsOperationApplied(_ context.Context, op string) (bool, error) { + return s.applied[op], nil +} +func (s *sessionScopedStateStore) MarkOperationApplied(_ context.Context, op string) error { + s.applied[op] = true + return nil +} + +// TestEvaluateStepQuorum_InsensitiveToWhichLiveSessionIsNewest pins the +// invariant that lets engine_dispatcher's resolveLatestSessionID stay +// task-scoped ("the task's most recent session, any state") once +// features.officeSessionIdentity gives each participant agent its own +// session per task and a task therefore has SEVERAL live sessions. +// +// Today EvaluateStepQuorum's guard evaluation reads TaskID, CurrentStepID +// and WorkflowID off the loaded MachineState, all task-derived, so which +// session the resolver happens to pick cannot change the answer. SessionID, +// SessionState, Data and IsPassthrough are the session-derived fields, and +// none of them is read by guard evaluation. +// +// If a guard ever starts reading SessionID, SessionState, Data or +// IsPassthrough, that +// stops being true silently: the resolver would hand the engine whichever +// agent's state happened to be newest, and a decision could be evaluated +// against another agent's state with nothing failing. This test makes that +// regression loud. The fix at that point is to session-scope the CALL SITE +// — pass the deciding session, as RecordDecision does via +// resolveDeciderSessionID — not to add a defensive check inside the +// task-scoped resolver, which would only make a wrong-scoped read look +// safe. +func TestEvaluateStepQuorum_InsensitiveToWhichLiveSessionIsNewest(t *testing.T) { + newStore := func() *sessionScopedStateStore { + return &sessionScopedStateStore{ + step: StepSpec{ + ID: "review", WorkflowID: "wf", Position: 1, + Events: map[Trigger][]Action{ + TriggerOnTurnComplete: { + {Kind: ActionMoveToNext, Guard: approvedGuard("reviewer")}, + }, + }, + }, + // The two live sessions carry DIFFERENT per-session bags and + // states, so a guard that started reading either would diverge + // between them. + dataBySess: map[string]map[string]any{ + "sess-runner": {"owner": "runner", "attempts": 3}, + "sess-reviewer": {"owner": "reviewer", "attempts": 1}, + }, + stateBySess: map[string]string{ + "sess-runner": "RUNNING", + "sess-reviewer": "WAITING_FOR_INPUT", + }, + passthroughBySess: map[string]bool{ + "sess-runner": true, + "sess-reviewer": false, + }, + applied: map[string]bool{}, + } + } + parts := fakeParticipants{list: []ParticipantInfo{ + {ID: "p1", Role: "reviewer", DecisionRequired: true, AgentProfileID: "rev-A"}, + }} + newDecisions := func() *fakeDecisionStore { + d := newFakeDecisionStore() + d.byKey[dkey("task-1", "review")] = []DecisionInfo{{ParticipantID: "p1", Decision: DecisionApproved}} + return d + } + + runnerStore := newStore() + fromRunner, err := New(runnerStore, MapRegistry{}, + WithDecisionStore(newDecisions()), WithParticipantStore(parts)). + EvaluateStepQuorum(context.Background(), "task-1", "sess-runner") + if err != nil { + t.Fatalf("EvaluateStepQuorum(sess-runner): %v", err) + } + + reviewerStore := newStore() + fromReviewer, err := New(reviewerStore, MapRegistry{}, + WithDecisionStore(newDecisions()), WithParticipantStore(parts)). + EvaluateStepQuorum(context.Background(), "task-1", "sess-reviewer") + if err != nil { + t.Fatalf("EvaluateStepQuorum(sess-reviewer): %v", err) + } + + // Guard against a vacuous pass: if the fake ever stops handing the two + // runs genuinely different per-session bags and states, the comparison + // below proves nothing. Assert the divergence actually reached the engine. + if len(runnerStore.loadedData) != 1 || len(reviewerStore.loadedData) != 1 { + t.Fatalf("expected exactly one LoadState per run, got %d and %d", + len(runnerStore.loadedData), len(reviewerStore.loadedData)) + } + if reflect.DeepEqual(runnerStore.loadedData[0], reviewerStore.loadedData[0]) { + t.Fatalf("test is vacuous: both sessions loaded the same Data %v", runnerStore.loadedData[0]) + } + if len(runnerStore.loadedStates) != 1 || len(reviewerStore.loadedStates) != 1 { + t.Fatalf("expected exactly one LoadState per run, got %d and %d states", + len(runnerStore.loadedStates), len(reviewerStore.loadedStates)) + } + if runnerStore.loadedStates[0] == reviewerStore.loadedStates[0] { + t.Fatalf("test is vacuous: both sessions loaded the same SessionState %q", runnerStore.loadedStates[0]) + } + if len(runnerStore.loadedPassthrough) != 1 || len(reviewerStore.loadedPassthrough) != 1 { + t.Fatalf("expected exactly one LoadState per run, got %d and %d passthrough values", + len(runnerStore.loadedPassthrough), len(reviewerStore.loadedPassthrough)) + } + if runnerStore.loadedPassthrough[0] == reviewerStore.loadedPassthrough[0] { + t.Fatalf("test is vacuous: both sessions loaded the same IsPassthrough value %v", + runnerStore.loadedPassthrough[0]) + } + + // The snapshot must be identical despite that divergence. + if fromRunner.StepID != fromReviewer.StepID { + t.Errorf("StepID differs by session: %q vs %q", fromRunner.StepID, fromReviewer.StepID) + } + if fromRunner.ReevaluationBlocked != fromReviewer.ReevaluationBlocked { + t.Errorf("ReevaluationBlocked differs by session: %v vs %v", + fromRunner.ReevaluationBlocked, fromReviewer.ReevaluationBlocked) + } + if len(fromRunner.Guards) != len(fromReviewer.Guards) { + t.Fatalf("guard count differs by session: %d vs %d", + len(fromRunner.Guards), len(fromReviewer.Guards)) + } + for i := range fromRunner.Guards { + if !reflect.DeepEqual(fromRunner.Guards[i], fromReviewer.Guards[i]) { + t.Errorf("guard %d differs by session:\n runner: %#v\n reviewer: %#v", + i, fromRunner.Guards[i], fromReviewer.Guards[i]) + } + } + + // Sanity: the snapshot is non-trivial, so the equality above has content. + if fromRunner.StepID != "review" || len(fromRunner.Guards) != 1 { + t.Fatalf("expected a one-guard snapshot at step review, got %#v", fromRunner) + } + if !fromRunner.Guards[0].Satisfied { + t.Fatalf("expected the quorum guard to be satisfied, got %#v", fromRunner.Guards[0]) + } +}