package orchestrator import ( "context" "encoding/json" "errors" "strings" "sync" "testing" "time" "github.com/stretchr/testify/require" "github.com/tidbcloud/AgentTeam/internal/core/domain" "github.com/tidbcloud/AgentTeam/internal/store/workflow/run" "github.com/tidbcloud/AgentTeam/internal/workflow/plan" runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run" ) type fakeRecoveryRepository struct { candidates []run.Effect tick func(run.RecoveryTickParams) (run.RecoveryTickResult, error) } func (f *fakeRecoveryRepository) ListRecoveryCandidates(context.Context, run.ListRecoveryCandidatesParams) ([]run.Effect, error) { return f.candidates, nil } func (f *fakeRecoveryRepository) RecoveryTick(_ context.Context, params run.RecoveryTickParams) (run.RecoveryTickResult, error) { if f.tick == nil { return run.RecoveryTickResult{}, nil } return f.tick(params) } type recordingRecoveryWriter struct { writes []MessageRequest } func (w *recordingRecoveryWriter) WriteRunMessage(_ context.Context, request MessageRequest) (MessageResult, error) { w.writes = append(w.writes, request) scope := domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_rec"} return MessageResult{MessageID: "msg_rec", DeliveryScope: &scope}, nil } func recoverySendTick(effect run.Effect, displayAttemptIndex, deliverySequence int) run.RecoveryTickResult { effect.RecoveryAttemptCount = deliverySequence return run.RecoveryTickResult{ Effect: effect, Send: true, DisplayAttemptIndex: displayAttemptIndex, FirstNonWorkingAt: effect.FirstNonWorkingAt, AttemptCount: displayAttemptIndex, NextRecoveryAt: effect.NextRecoveryAt, } } func recoveryEffect(t *testing.T, runID, effectID string) run.Effect { t.Helper() request, err := EncodeEffectRequest(EffectRequest{ Intent: runkernel.Intent{ Kind: runkernel.IntentSendMessage, ActivationID: "a1", NodeID: "say", Callback: true, Retry: plan.ArtifactRetryPolicy{MaxAttempts: 1, BackoffNanos: int64(time.Second)}, }, Message: &MessageRequest{ WorkspaceID: "wsp_1", PlanID: "wfp_1", PlanVersionID: "wfv_1", RunID: runID, ActivationID: "a1", EffectID: effectID, NodeID: "say", SourceKind: "manual", OwnerUserID: "usr_1", AuthorMemberID: "mem_1", TargetRef: "#game", Content: "original", ClientMessageID: "workflow-run:" + effectID, Callback: true, }, }) require.NoError(t, err) receipt, err := EncodeEffectResult(EffectResult{ MessageID: "msg_" + effectID, WakeAgentIDs: []string{"agt_1"}, CallbackResponderAgentID: "agt_1", DeliveryScope: &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, }) require.NoError(t, err) return run.Effect{ ID: effectID, WorkspaceID: "wsp_1", RunID: runID, ActivationID: "wra_1", Kind: EffectKindMessage, Status: run.EffectStatusAwaiting, CallbackResponderAgentID: "agt_1", RequestJSON: request, ResultJSON: receipt, } } func TestRecoveryRunnerSendsExactPromptWithRealSelectors(t *testing.T) { effect := recoveryEffect(t, "wri_recovery", "wfx_recovery") repository := &fakeRecoveryRepository{ candidates: []run.Effect{effect}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { return recoverySendTick(effect, 2, 2), nil }, } writer := &recordingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorking{working: false, known: true}, }) require.NoError(t, err) outcome, err := runner.RecoverOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, outcome.Candidates) require.Equal(t, 1, outcome.Sent) require.Len(t, writer.writes, 1) sent := writer.writes[0] // The recovery nudge reuses the callback-shaped message request (Callback=true // + frozen responder ref) so the ordinary message transaction wakes the frozen // responder through the trusted responder-ref chain, with the internal fence // pinning the frozen responder ID. require.True(t, sent.Callback, "the recovery nudge must wake the responder via the callback routing path") require.Equal(t, "agt_1", sent.ExpectedCallbackResponderAgentID) require.Equal(t, "wsp_1", sent.WorkspaceID) require.Equal(t, "wri_recovery", sent.RunID) require.Equal(t, "wfx_recovery", sent.EffectID) require.Equal(t, &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, sent.Scope) require.Empty(t, sent.TargetRef, "recovery reuses the frozen scope, never a re-resolved target") require.Equal(t, "workflow-run-recovery:wri_recovery:wfx_recovery:2", sent.ClientMessageID) require.NotNil(t, sent.Presentation) require.NoError(t, sent.Presentation.Validate()) var presentationArgs domain.WorkflowRecoveryPresentationArgs require.NoError(t, json.Unmarshal(sent.Presentation.Args, &presentationArgs)) require.Equal(t, domain.WorkflowRecoveryPresentationArgs{ RunID: "wri_recovery", StepID: "wfx_recovery", EffectID: "wfx_recovery", Attempt: 2, TotalAttempts: domain.WorkflowRecoveryMaxAttempts, }, presentationArgs) // The content is the frozen English Recovery Prompt with the real trusted // Run/Step selectors materialized into the exact copy-pasteable command. require.Equal(t, `loop exec workflow step '{"action":"get","runId":"wri_recovery","stepId":"wfx_recovery"}'`, extractRecoveryCommand(sent.Content), ) require.Contains(t, sent.Content, "This is automatic recovery attempt 2 of 5.") require.Contains(t, sent.Content, "Use the trusted Run and Step selectors included with this recovery delivery.") require.NotContains(t, sent.Content, "") require.NotContains(t, sent.Content, "") require.NotContains(t, sent.Content, "{{recoveryIndex}}") } func TestRecoveryRunnerSelectOneMaterializesFrozenWinnerOnResend(t *testing.T) { // A selectOne callback effect: the durable message request carries the ordered // candidate pool and an empty responder ref, while the durable receipt/effect // froze the concrete winner (agt_winner). Recovery must materialize the frozen // winner into ResponderAgentRef and clear the pool, exactly like a fresh // dispatch, so the nudge routes to one responder and never re-selects. request, err := EncodeEffectRequest(EffectRequest{ Intent: runkernel.Intent{ Kind: runkernel.IntentSendMessage, ActivationID: "a1", NodeID: "say", Callback: true, ResponderCandidates: []string{"agt_winner", "agt_other"}, Retry: plan.ArtifactRetryPolicy{MaxAttempts: 1, BackoffNanos: int64(time.Second)}, }, Message: &MessageRequest{ WorkspaceID: "wsp_1", PlanID: "wfp_1", PlanVersionID: "wfv_1", RunID: "wri_sel", ActivationID: "a1", EffectID: "wfx_sel", NodeID: "say", SourceKind: "manual", OwnerUserID: "usr_1", AuthorMemberID: "mem_1", TargetRef: "#game", Content: "original", ClientMessageID: "workflow-run:wfx_sel", Callback: true, ResponderCandidates: []string{"agt_winner", "agt_other"}, }, }) require.NoError(t, err) receipt, err := EncodeEffectResult(EffectResult{ MessageID: "msg_wfx_sel", WakeAgentIDs: []string{"agt_winner"}, CallbackResponderAgentID: "agt_winner", DeliveryScope: &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, }) require.NoError(t, err) effect := run.Effect{ ID: "wfx_sel", WorkspaceID: "wsp_1", RunID: "wri_sel", ActivationID: "wra_1", Kind: EffectKindMessage, Status: run.EffectStatusAwaiting, CallbackResponderAgentID: "agt_winner", RequestJSON: request, ResultJSON: receipt, } repository := &fakeRecoveryRepository{ candidates: []run.Effect{effect}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { return recoverySendTick(effect, 1, 1), nil }, } writer := &recordingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorking{working: false, known: true}, }) require.NoError(t, err) outcome, err := runner.RecoverOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, outcome.Candidates) require.Equal(t, 1, outcome.Sent) require.Len(t, writer.writes, 1) sent := writer.writes[0] require.Equal(t, "agt_winner", sent.ResponderAgentRef, "recovery must materialize the durable frozen winner into the responder ref") require.Empty(t, sent.ResponderCandidates, "recovery must clear the candidate pool so the nudge is not re-selected") require.Equal(t, "agt_winner", sent.ExpectedCallbackResponderAgentID, "the trusted fence must still pin the frozen winner") require.True(t, sent.Callback) } func TestRecoveryRunnerSelectOneNoWinnerFailsClosed(t *testing.T) { // A selectOne effect whose durable receipt/effect has no frozen winner is // treated as unknown by the runtime working reader (never send, never start // timing): the recovery pass must not re-select and must not deliver with an // unresolved pool. request, err := EncodeEffectRequest(EffectRequest{ Intent: runkernel.Intent{ Kind: runkernel.IntentSendMessage, ActivationID: "a1", NodeID: "say", Callback: true, ResponderCandidates: []string{"agt_a", "agt_b"}, Retry: plan.ArtifactRetryPolicy{MaxAttempts: 1, BackoffNanos: int64(time.Second)}, }, Message: &MessageRequest{ WorkspaceID: "wsp_1", PlanID: "wfp_1", PlanVersionID: "wfv_1", RunID: "wri_non", ActivationID: "a1", EffectID: "wfx_non", NodeID: "say", SourceKind: "manual", OwnerUserID: "usr_1", AuthorMemberID: "mem_1", TargetRef: "#game", Content: "original", ClientMessageID: "workflow-run:wfx_non", Callback: true, ResponderCandidates: []string{"agt_a", "agt_b"}, }, }) require.NoError(t, err) receipt, err := EncodeEffectResult(EffectResult{ MessageID: "msg_wfx_non", WakeAgentIDs: []string{"agt_a", "agt_b"}, DeliveryScope: &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, }) require.NoError(t, err) effect := run.Effect{ ID: "wfx_non", WorkspaceID: "wsp_1", RunID: "wri_non", ActivationID: "wra_1", Kind: EffectKindMessage, Status: run.EffectStatusAwaiting, RequestJSON: request, ResultJSON: receipt, } repository := &fakeRecoveryRepository{ candidates: []run.Effect{effect}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { return recoverySendTick(effect, 1, 1), nil }, } writer := &recordingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorking{working: false, known: true}, }) require.NoError(t, err) outcome, err := runner.RecoverOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, outcome.Unknown, "a selectOne pool without a durable frozen winner must be skipped as unknown") require.Equal(t, 0, outcome.Sent) require.Empty(t, writer.writes, "no nudge may be delivered without a frozen winner") } func TestRenderRecoveryPromptIsByteExactExceptSelectorsAndIndex(t *testing.T) { runID, stepID, index := "wri_x", "wfx_y", 3 rendered := renderRecoveryPrompt(runID, stepID, index) require.Equal(t, "The Workflow step you were executing is still waiting for its callback, but you have stopped working. Continue the step. First run `loop exec workflow step '{\"action\":\"get\",\"runId\":\"wri_x\",\"stepId\":\"wfx_y\"}'` to inspect the active callback and output contract. Use the trusted Run and Step selectors included with this recovery delivery. Check the work and artifacts already produced; do not restart without reason. Call `complete` when finished, `fail` if the step cannot be completed, `pause` if you need to stop automatic recovery permanently while you continue discussing with the user (only completing, failing, or canceling the step ends the pause), or `park` if you want automatic recovery to stop while you are idle and then resume on a fresh schedule once you start working again. An ordinary chat reply does not advance or pause the Workflow.\n\nThis is automatic recovery attempt 3 of 5.", rendered) } func TestRecoveryRunnerSkipsWorkingAndUnknownCandidates(t *testing.T) { working := recoveryEffect(t, "wri_w", "wfx_w") working.CallbackResponderAgentID = "agt_working" unknown := recoveryEffect(t, "wri_u", "wfx_u") unknown.CallbackResponderAgentID = "agt_unknown" idle := recoveryEffect(t, "wri_i", "wfx_i") idle.CallbackResponderAgentID = "agt_idle" repository := &fakeRecoveryRepository{ candidates: []run.Effect{working, unknown, idle}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { return run.RecoveryTickResult{Effect: idle, Send: false}, nil }, } writer := &recordingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorkingByAgent{ "agt_working": {working: true, known: true}, "agt_unknown": {working: false, known: false}, "agt_idle": {working: false, known: true}, }, }) require.NoError(t, err) outcome, err := runner.RecoverOnce(context.Background()) require.NoError(t, err) require.Equal(t, 3, outcome.Candidates) require.Equal(t, 1, outcome.Working, "a working responder is skipped") require.Equal(t, 1, outcome.Unknown, "an unknown responder is skipped without sending") require.Equal(t, 0, outcome.Sent) require.Empty(t, writer.writes) } func TestRecoveryRunnerTicksAllCandidatesAndCountsFailures(t *testing.T) { send := recoveryEffect(t, "wri_s", "wfx_s") bad := recoveryEffect(t, "wri_b", "wfx_b") repository := &fakeRecoveryRepository{ candidates: []run.Effect{send, bad}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { if params.EffectID == "wfx_s" { return recoverySendTick(send, 1, 1), nil } return run.RecoveryTickResult{}, errors.New("tick failed") }, } writer := &recordingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorking{working: false, known: true}, }) require.NoError(t, err) outcome, err := runner.RecoverOnce(context.Background()) require.Error(t, err) require.Equal(t, 2, outcome.Candidates) require.Equal(t, 1, outcome.Sent) require.Equal(t, 1, outcome.Failed) require.Len(t, writer.writes, 1) } func TestRecoveryRunnerSendFailureStillCountsAttempt(t *testing.T) { effect := recoveryEffect(t, "wri_f", "wfx_f") repository := &fakeRecoveryRepository{ candidates: []run.Effect{effect}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { return recoverySendTick(effect, 1, 1), nil }, } writer := &failingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorking{working: false, known: true}, }) require.NoError(t, err) outcome, err := runner.RecoverOnce(context.Background()) require.Error(t, err) require.Equal(t, 1, outcome.Failed) require.Equal(t, 0, outcome.Sent) } // TestRecoveryRunnerBoundedGraceThenRecoverable proves the #t2596 bounded-grace // contract end to end at the RecoveryRunner boundary: while an online responder // is within the snapshot grace the reader reports known=false and the pass skips // with no send and no timing; once the grace elapses the reader reports the // responder recoverable and the durable schedule advances. func TestRecoveryRunnerBoundedGraceThenRecoverable(t *testing.T) { effect := recoveryEffect(t, "wri_g", "wfx_g") effect.CallbackResponderAgentID = "agt_grace" repository := &fakeRecoveryRepository{ candidates: []run.Effect{effect}, tick: func(params run.RecoveryTickParams) (run.RecoveryTickResult, error) { return recoverySendTick(effect, 1, 1), nil }, } // The reader mirrors the Hub: within grace -> known=false (deferred); after // grace -> recoverable (known=true, not working). reader := &flipAfterCallsRuntimeWorking{deferCalls: 2, thenKnown: true} writer := &recordingRecoveryWriter{} runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-1", Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: reader, }) require.NoError(t, err) // Pass 1 and 2: within grace -> deferred (no send, no timing). for pass := 1; pass <= 2; pass++ { outcome, err := runner.RecoverOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, outcome.Candidates) require.Equal(t, 1, outcome.Unknown, "within grace the responder is deferred") require.Equal(t, 0, outcome.Sent) require.Empty(t, writer.writes, "no nudge may fire within grace") } // Pass 3: grace elapsed -> recoverable -> schedule advances and nudge fires. outcome, err := runner.RecoverOnce(context.Background()) require.NoError(t, err) require.Equal(t, 0, outcome.Unknown) require.Equal(t, 1, outcome.Sent, "after grace the responder becomes recoverable and the schedule advances") require.Len(t, writer.writes, 1) } // flipAfterCallsRuntimeWorking defers (known=false) for deferCalls invocations, // then reports the fixed working/known state. It models the HubWorkingReader // bounded-snapshot-grace transition across recovery passes. type flipAfterCallsRuntimeWorking struct { deferCalls int calls int thenKnown bool } func (reader *flipAfterCallsRuntimeWorking) RuntimeWorking(_ context.Context, _ string, _ string) (bool, bool) { reader.calls++ if reader.calls <= reader.deferCalls { return false, false } return false, reader.thenKnown } // TestRecoveryRunnerConcurrentPassesSingleSend proves multi-worker duplicate // scans cannot double-send one due effect: concurrent RecoverOnce passes racing // the same effect resolve to exactly one RecoveryTick winner (mirroring the // store's FOR UPDATE row lock) and exactly one deterministic recovery message. func TestRecoveryRunnerConcurrentPassesSingleSend(t *testing.T) { effect := recoveryEffect(t, "wri_multi", "wfx_multi") repository := &singleWinnerRecoveryRepository{effect: effect} writer := &recordingRecoveryWriter{} var wg sync.WaitGroup const workers = 6 start := make(chan struct{}) errs := make(chan error, workers) for i := 0; i < workers; i++ { wg.Add(1) go func(worker int) { defer wg.Done() runner, err := NewRecoveryRunner(repository, writer, RecoveryRunnerOptions{ WorkerID: "recovery-" + string(rune('a'+worker)), Now: func() time.Time { return time.Date(2026, 8, 12, 12, 0, 0, 0, time.UTC) }, RuntimeWorking: fakeRuntimeWorking{working: false, known: true}, }) if err != nil { errs <- err return } <-start if _, err := runner.RecoverOnce(context.Background()); err != nil { errs <- err } }(i) } close(start) wg.Wait() close(errs) for err := range errs { t.Fatalf("concurrent recovery pass failed: %v", err) } require.Equal(t, 1, len(writer.writes), "exactly one recovery message may be committed for one due effect") require.Equal(t, 1, repository.sendWins(), "exactly one concurrent pass wins the send slot") require.Equal(t, "workflow-run-recovery:wri_multi:wfx_multi:1", writer.writes[0].ClientMessageID, "the recovery clientMessageId must stay deterministic under concurrent duplicate scans") } // singleWinnerRecoveryRepository models the durable row-lock semantics: the first // RecoveryTick wins the send slot; every later tick finds the slot already // advanced (Send=false). It is safe for concurrent use. type singleWinnerRecoveryRepository struct { effect run.Effect mu sync.Mutex winner int ticks int } func (r *singleWinnerRecoveryRepository) ListRecoveryCandidates(context.Context, run.ListRecoveryCandidatesParams) ([]run.Effect, error) { r.mu.Lock() defer r.mu.Unlock() if r.ticks == 0 { return []run.Effect{r.effect}, nil } // After the winner advances the schedule, the effect is no longer due. return nil, nil } func (r *singleWinnerRecoveryRepository) RecoveryTick(_ context.Context, params run.RecoveryTickParams) (run.RecoveryTickResult, error) { r.mu.Lock() defer r.mu.Unlock() r.ticks++ if r.winner == 0 { r.winner = r.ticks return recoverySendTick(r.effect, 1, 1), nil } return run.RecoveryTickResult{Effect: r.effect, Send: false}, nil } func (r *singleWinnerRecoveryRepository) sendWins() int { r.mu.Lock() defer r.mu.Unlock() return r.winner } // fakeRuntimeWorking is a fixed working-state reader for deterministic tests. type fakeRuntimeWorking struct { working bool known bool } func (reader fakeRuntimeWorking) RuntimeWorking(_ context.Context, _ string, _ string) (bool, bool) { return reader.working, reader.known } // fakeRuntimeWorkingByAgent returns per-responder working state keyed by the // frozen responder agent id. type fakeRuntimeWorkingByAgent map[string]struct { working bool known bool } func (reader fakeRuntimeWorkingByAgent) RuntimeWorking(_ context.Context, _ string, agentID string) (bool, bool) { state, ok := reader[agentID] if !ok { return false, false } return state.working, state.known } type failingRecoveryWriter struct{} func (*failingRecoveryWriter) WriteRunMessage(context.Context, MessageRequest) (MessageResult, error) { return MessageResult{}, errors.New("writer unavailable") } // extractRecoveryCommand pulls the materialized step command out of the rendered // prompt for assertion without coupling the test to the surrounding prose. func extractRecoveryCommand(content string) string { const prefix = "First run `" start := strings.Index(content, prefix) if start < 0 { return "" } rest := content[start+len(prefix):] end := strings.Index(rest, "`") if end < 0 { return "" } return rest[:end] } var _ RecoveryRepository = (*fakeRecoveryRepository)(nil)