From 30b79a330f811941aa25172502b5750baed79434 Mon Sep 17 00:00:00 2001 From: Steve Dudenhoeffer Date: Mon, 29 Jun 2026 15:14:45 -0400 Subject: [PATCH 1/2] feat(run): execute multi-phase pipelines (RunnableAgent.Phases) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The kernel carried RunnableAgent.Phases as a DTO but never executed it — Run always ran a single agent loop with ra.SystemPrompt, so a phased agent (mort's deepresearch/research) silently ran one loop with the base prompt instead of its pipeline. This implements the phase loop, ported from mort's agentexec pipeline but reusing the kernel's own machinery. - run/phases.go: runPhases / runOnePhase. Phases run sequentially; each is a fresh agent loop (or a bare LLM call for IsRunFunc phases) with its own template-expanded system prompt ({{.Query}} + {{.}}), model tier, step cap, and tool subset. Outputs thread into later phases; the final phase's output is the run output. Optional phases swallow errors and substitute FallbackMessage; a non-optional phase that merely exhausts its step/tool budget salvages its partial transcript and continues (a hard error still aborts); per-phase tier-resolve failures fall back with a WARN. - run/agent.go: Phase gains IsRunFunc + FallbackMessage (the kernel Phase struct previously omitted them). - run/executor.go: Run factors the shared agent options (tool-error limits, step observer, compactor) and branches — single loop (critic's dynamic step ceiling) vs the phase runner (fixed per-phase caps; the run-level critic's steer + hard deadline still apply across phases). systemPrompt now delegates to systemPromptWithBody so each phase keeps the platform header. The same step observer feeds audit/steps/critic across all phases. Tests (run/phases_test.go): sequential output threading + template expansion, Optional-failure → FallbackMessage continues, hard-error abort, IsRunFunc bare call, per-phase SystemHeader, filterToolbox subset, template expansion. Full ./... suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- run/agent.go | 18 ++- run/executor.go | 56 ++++++-- run/phases.go | 311 +++++++++++++++++++++++++++++++++++++++++++++ run/phases_test.go | 233 +++++++++++++++++++++++++++++++++ 4 files changed, 603 insertions(+), 15 deletions(-) create mode 100644 run/phases.go create mode 100644 run/phases_test.go diff --git a/run/agent.go b/run/agent.go index 18a0a98..9b46fca 100644 --- a/run/agent.go +++ b/run/agent.go @@ -55,15 +55,27 @@ type RunnableAgent struct { } // Phase is one step of a multi-step run: its own system prompt, model tier, -// iteration cap, and tool subset. Optional phases may be skipped by the -// pipeline when their precondition isn't met. +// iteration cap, and tool subset. Phase prompts are Go text/template strings +// expanded against {{.Query}} (the original input) and {{.}} (a +// prior phase's output) before the phase runs, so a phase can consume earlier +// work. The final phase's output is the run's output. type Phase struct { Name string SystemPrompt string ModelTier string MaxIterations int Tools []string - Optional bool + // Optional swallows a phase's error and substitutes FallbackMessage (or a + // generated note) as its output, so a non-critical phase failing does not + // abort the pipeline. + Optional bool + // FallbackMessage is the substitute output when an Optional phase fails. + // Empty → a generated "(phase %q encountered an error…)" note. + FallbackMessage string + // IsRunFunc marks a phase as a single bare LLM call (no tool loop, no tools + // array) — a deterministic transform step (plan/synthesize) rather than an + // agentic loop. Its Tools/MaxIterations are ignored. + IsRunFunc bool } // CriticConfig configures the optional run-critic. Enabled gates whether a diff --git a/run/executor.go b/run/executor.go index ef29daf..76955f3 100644 --- a/run/executor.go +++ b/run/executor.go @@ -289,12 +289,10 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio } } - opts := []agent.Option{ - agent.WithToolbox(toolbox), - // Step ceiling: a fixed WithMaxSteps(maxIter) normally, but when a critic is - // active it owns a DYNAMIC ceiling (WithMaxStepsFunc) so it can raise a - // healthy-but-long run's budget mid-flight. Falls back to maxIter. - critic.maxStepsOption(maxIter), + // Shared agent options used by BOTH the single-loop path and every phase: the + // tool-error guards, the step observer, and optional compaction. The toolbox + + // step ceiling are NOT shared (they vary per phase), so they're added per path. + sharedOpts := []agent.Option{ agent.WithToolErrorLimits(e.cfg.Defaults.MaxConsecutiveToolErrors, e.cfg.Defaults.MaxSameToolCallRepeats), agent.WithStepObserver(stepObserver), } @@ -313,11 +311,10 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio }) } } - opts = append(opts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire))) + sharedOpts = append(sharedOpts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire))) } } - ag := agent.New(model, e.systemPrompt(ra), opts...) // Stage non-image input attachments (audio/PDF/binary) into the host file // store and fold an [ATTACHED FILES] descriptor into the prompt so the agent // can reach them by file_id. No-op when Ports.InputFiles is nil or there are @@ -327,7 +324,35 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio // One WithSteer drains BOTH the session mailbox (a tool's AttachImages) and // the critic's nudges before each step. steer := func() []llm.Message { return append(mailbox.drain(), critic.drainSteer()...) } - runRes, runErr := runAgent(runCtx, ag, input, inv.Images, agent.WithSteer(steer)) + + var runRes *agent.Result + var runErr error + if len(ra.Phases) == 0 { + // Single-loop run: the agent's base prompt + full toolbox, with the + // critic's DYNAMIC step ceiling (WithMaxStepsFunc, so it can raise a + // healthy-but-long run's budget mid-flight; falls back to maxIter). + opts := append([]agent.Option{ + agent.WithToolbox(toolbox), + critic.maxStepsOption(maxIter), + }, sharedOpts...) + ag := agent.New(model, e.systemPrompt(ra), opts...) + runRes, runErr = runAgent(runCtx, ag, input, inv.Images, agent.WithSteer(steer)) + } else { + // Multi-phase pipeline: each phase runs its own prompt/tier/tools/step-cap + // sequentially, threading outputs through {{.}} templates. Reuses + // the shared opts so audit/steps/critic-steer accumulate across every phase. + // (Per-phase step caps are fixed — the critic's dynamic ceiling is not + // propagated to phases — but its steer + hard deadline still apply.) + runRes, runErr = e.runPhases(runCtx, ra, phaseDeps{ + baseModel: model, + baseTier: tier, + baseToolbox: toolbox, + baseMaxIter: maxIter, + sharedOpts: sharedOpts, + steer: steer, + rec: rec, + }, input, inv.Images) + } status := statusFor(runCtx, runErr) if runRes != nil { @@ -403,13 +428,20 @@ func (e *Executor) finishAudit(ctx context.Context, rec RunRecorder, status stri } func (e *Executor) systemPrompt(ra RunnableAgent) string { + return e.systemPromptWithBody(ra.SystemPrompt) +} + +// systemPromptWithBody composes the optional platform header with an arbitrary +// body. The single-loop path passes ra.SystemPrompt; the phase runner passes a +// phase's expanded instructions, so each phase keeps the platform header. +func (e *Executor) systemPromptWithBody(body string) string { if e.cfg.SystemHeader == "" { - return ra.SystemPrompt + return body } - if ra.SystemPrompt == "" { + if body == "" { return e.cfg.SystemHeader } - return e.cfg.SystemHeader + "\n\n" + ra.SystemPrompt + return e.cfg.SystemHeader + "\n\n" + body } // compactionThreshold returns the token threshold for the tier's model context diff --git a/run/phases.go b/run/phases.go new file mode 100644 index 0000000..add1fa4 --- /dev/null +++ b/run/phases.go @@ -0,0 +1,311 @@ +package run + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "strings" + "text/template" + + "gitea.stevedudenhoeffer.com/steve/majordomo/agent" + "gitea.stevedudenhoeffer.com/steve/majordomo/llm" +) + +// The multi-step phase runner. A phased RunnableAgent (ra.Phases non-empty) runs +// its phases in order; each phase is a fresh majordomo agent loop (or a single +// bare LLM call for IsRunFunc phases) with its own template-expanded system +// prompt, model tier, step cap, and tool subset. Phase outputs feed later phases +// through {{.}} template variables; {{.Query}} is the original input. +// The final phase's output is the run's output. +// +// Ported from mort's agentexec pipeline so the executus kernel — which already +// carries RunnableAgent.Phases as a DTO — actually EXECUTES them (it previously +// ignored the slice and ran a single loop with the base prompt). It reuses the +// shared run machinery built once in Run: the same stepObserver (so audit/steps/ +// critic accumulate across every phase), the same critic steer, and the same +// compaction option. +// +// Semantics preserved from mort's pipeline: +// - phases run sequentially; ctx cancellation between phases aborts the run. +// - IsRunFunc = one bare LLM call, no tools, no loop. +// - Optional phases swallow errors and substitute FallbackMessage. +// - a non-optional phase that merely exhausts its step/tool budget is NOT fatal: +// its partial transcript is salvaged and the pipeline continues (so a long +// verify phase never discards every earlier phase's work). A hard error +// (cancellation, model failure) still aborts. +// - per-phase ModelTier resolve failures fall back to the base model with a WARN. +// +// Deliberately NOT carried over (kernel is leaner than mort's legacy pipeline): +// the legacy `submit` capture tool (the kernel relies on majordomo's +// no-tool-call-is-final-answer termination, like its single-loop path), and the +// critic's dynamic iteration ceiling (per-phase caps are fixed at phase start — +// the run-level critic's steer + hard deadline still apply across phases). + +// phaseDeps carries the per-run state the phase runner shares with Run: the base +// model + tier, the full decorated toolbox (filtered per phase), the base step +// cap, the shared agent options (tool-error limits + step observer + compactor), +// the critic/session steer, and the audit recorder (phase events). +type phaseDeps struct { + baseModel llm.Model + baseTier string + baseToolbox *llm.Toolbox + baseMaxIter int + sharedOpts []agent.Option + steer func() []llm.Message + rec RunRecorder +} + +// runPhases executes ra.Phases sequentially and returns a synthetic agent.Result +// whose Output is the final phase's output, with Usage aggregated across phases +// and Messages set to the last phase's transcript (for the PostRun hook). A hard +// (non-optional, non-budget) phase failure returns the error. +func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phaseDeps, query string, images []llm.ImagePart) (*agent.Result, error) { + outputs := make(map[string]string, len(ra.Phases)) + var lastResult *agent.Result + var lastOutput string + var totalUsage llm.Usage + + for i, phase := range ra.Phases { + // A killed/timed-out/cancelled run must not start its next phase. + if err := runCtx.Err(); err != nil { + return lastResult, err + } + + instructions := expandPhaseTemplate(phase.SystemPrompt, query, outputs) + if deps.rec != nil { + deps.rec.LogEvent("phase_start", map[string]any{"phase": phase.Name}) + } + + output, res, err := e.runOnePhase(runCtx, ra, deps, phase, instructions, query, images) + if res != nil { + lastResult = res + totalUsage = addUsage(totalUsage, res.Usage) + } + if err != nil { + isLast := i == len(ra.Phases)-1 + switch { + case phase.Optional: + output = phase.FallbackMessage + if output == "" { + output = fmt.Sprintf("(Phase %q encountered an error -- proceeding without its results)", phase.Name) + } + slog.Warn("run: optional pipeline phase failed", + "agent", ra.Name, "phase", phase.Name, "error", err) + if deps.rec != nil { + deps.rec.LogEvent("phase_failed_optional", map[string]any{"phase": phase.Name, "error": err.Error()}) + } + + case isPhaseBudgetExhaustion(err) && (!isLast || strings.TrimSpace(output) != ""): + // Soft stop: the phase ran out of its step/tool budget before + // composing a final answer. Not fatal — it did real work + // (runOnePhase salvaged its partial transcript into output), and + // aborting would discard every completed phase before it. Degrade + // gracefully and continue. + if strings.TrimSpace(output) == "" { + output = fmt.Sprintf("(Phase %q reached its step budget before producing a consolidated result; continuing with its partial findings.)", phase.Name) + } else { + output += fmt.Sprintf("\n\n(Note: phase %q reached its step budget before fully completing; the above is its partial output.)", phase.Name) + } + slog.Warn("run: pipeline phase exhausted its budget; salvaging partial output and continuing", + "agent", ra.Name, "phase", phase.Name, "last_phase", isLast, "error", err) + if deps.rec != nil { + deps.rec.LogEvent("phase_budget_exhausted", map[string]any{"phase": phase.Name, "error": err.Error(), "last_phase": isLast}) + } + + default: + return lastResult, fmt.Errorf("pipeline phase %q: %w", phase.Name, err) + } + } + + outputs[phase.Name] = output + lastOutput = output + } + + // Synthesize the run result: the final phase's output, usage aggregated over + // all phases, and the last phase's transcript for the PostRun hook. + if lastResult == nil { + lastResult = &agent.Result{} + } + lastResult.Output = lastOutput + lastResult.Usage = totalUsage + return lastResult, nil +} + +// runOnePhase runs a single phase: a bare LLM call for IsRunFunc phases, a fresh +// agent loop otherwise. Returns the phase output, the loop result (nil for a +// failed bare call), and any error. On a budget-exhaustion error the loop's +// partial transcript is salvaged into the returned output. +func (e *Executor) runOnePhase(runCtx context.Context, ra RunnableAgent, deps phaseDeps, phase Phase, instructions, query string, images []llm.ImagePart) (string, *agent.Result, error) { + model := e.phaseModel(runCtx, deps, ra, phase) + // The phase's expanded instructions are the system prompt (with the platform + // header so tools keep their run ids); the original query is the user message. + system := e.systemPromptWithBody(instructions) + + if phase.IsRunFunc { + // Bare LLM call: no tool loop, no tools array (some models 400 on an empty + // tools list). The response still lands in the audit token tally. + msgs := []llm.Message{phaseUserMessage(query, images)} + resp, err := model.Generate(runCtx, llm.Request{System: system, Messages: msgs}) + if err != nil { + return "", nil, fmt.Errorf("phase %q model call: %w", phase.Name, err) + } + if deps.rec != nil { + deps.rec.OnStep(1, resp) + } + return resp.Text(), &agent.Result{ + Output: resp.Text(), + Usage: resp.Usage, + Messages: append(msgs, resp.Message()), + }, nil + } + + toolbox := filterToolbox(deps.baseToolbox, phase.Tools) + maxIter := phase.MaxIterations + if maxIter <= 0 { + maxIter = deps.baseMaxIter + } + // Per-phase opts: a fixed step ceiling for this phase (the critic's dynamic + // ceiling is intentionally not propagated to phases) + the phase toolbox, on + // top of the shared opts (tool-error limits, step observer, compactor). + opts := append([]agent.Option{ + agent.WithToolbox(toolbox), + agent.WithMaxSteps(maxIter), + }, deps.sharedOpts...) + ag := agent.New(model, system, opts...) + + res, runErr := runAgent(runCtx, ag, query, images, agent.WithSteer(deps.steer)) + output := "" + if res != nil { + output = res.Output + } + // Budget/guard exhaustion leaves a usable partial transcript but an empty + // final answer; salvage the narrated work so the pipeline can carry it forward. + if runErr != nil && isPhaseBudgetExhaustion(runErr) { + if salvaged := salvagePhaseTranscript(res); salvaged != "" { + output = salvaged + } + } + return output, res, runErr +} + +// phaseModel resolves the phase's model tier, falling back to the base model on +// an empty tier or a resolution failure (WARN — visible, non-fatal). +func (e *Executor) phaseModel(ctx context.Context, deps phaseDeps, ra RunnableAgent, phase Phase) llm.Model { + if phase.ModelTier == "" { + return deps.baseModel + } + _, m, err := e.cfg.Models(ctx, phase.ModelTier) + if err != nil || m == nil { + slog.Warn("run: pipeline phase model resolve failed; using base model", + "agent", ra.Name, "phase", phase.Name, "tier", phase.ModelTier, "error", err) + return deps.baseModel + } + return m +} + +// isPhaseBudgetExhaustion reports whether err is a soft budget/guard stop (the +// loop hit its step cap or tripped a tool-error guard) — which leaves a usable +// partial transcript — as opposed to a hard error (cancellation, model failure). +func isPhaseBudgetExhaustion(err error) bool { + return errors.Is(err, agent.ErrMaxSteps) || errors.Is(err, agent.ErrToolLoop) +} + +// salvagePhaseTranscript reconstructs a best-effort phase output from a loop that +// ended without a final answer: the assistant's narrated text across every step, +// tail-trimmed to a bound. Returns "" when the model wrote no prose. +func salvagePhaseTranscript(res *agent.Result) string { + if res == nil { + return "" + } + var b strings.Builder + for _, step := range res.Steps { + if step.Response == nil { + continue + } + if t := strings.TrimSpace(step.Response.Text()); t != "" { + if b.Len() > 0 { + b.WriteString("\n\n") + } + b.WriteString(t) + } + } + out := strings.TrimSpace(b.String()) + const maxSalvage = 8000 + if len(out) > maxSalvage { + out = "...(earlier reasoning trimmed)...\n" + out[len(out)-maxSalvage:] + } + return out +} + +// phaseUserMessage builds a phase's user message: the original query text plus +// any inline images. Mirrors the single-loop multimodal seeding in runAgent. +func phaseUserMessage(query string, images []llm.ImagePart) llm.Message { + if len(images) == 0 { + return llm.UserText(query) + } + parts := make([]llm.Part, 0, len(images)+1) + if strings.TrimSpace(query) != "" { + parts = append(parts, llm.Text(query)) + } + for _, img := range images { + parts = append(parts, img) + } + return llm.UserParts(parts...) +} + +// expandPhaseTemplate applies Go text/template substitution to a phase prompt, +// replacing {{.Query}} with the original query and {{.}} with a prior +// phase's output. Returns the original string unchanged if parsing or execution +// fails (best-effort, not fatal). +func expandPhaseTemplate(tmpl, query string, priorOutputs map[string]string) string { + t, err := template.New("phase").Option("missingkey=zero").Parse(tmpl) + if err != nil { + return tmpl + } + data := map[string]string{"Query": query} + for k, v := range priorOutputs { + data[k] = v + } + var buf bytes.Buffer + if err := t.Execute(&buf, data); err != nil { + return tmpl + } + return buf.String() +} + +// filterToolbox returns a new toolbox restricted to the named tools (preserving +// palette order). Empty names = the full palette. Unknown names are skipped with +// a WARN — a typo'd phase tool list should not abort a run mid-pipeline. +func filterToolbox(box *llm.Toolbox, names []string) *llm.Toolbox { + out := llm.NewToolbox(box.Name()) + if len(names) == 0 { + for _, t := range box.Tools() { + _ = out.Add(t) + } + return out + } + for _, name := range names { + t, ok := box.Get(name) + if !ok { + slog.Warn("run: pipeline phase references unknown tool; skipping", "tool", name) + continue + } + if err := out.Add(t); err != nil { + slog.Warn("run: pipeline phase tool duplicated; skipping", "tool", name, "error", err) + } + } + return out +} + +// addUsage sums two llm.Usage tallies field-by-field so a phased run reports the +// total tokens across all phases. +func addUsage(a, b llm.Usage) llm.Usage { + a.InputTokens += b.InputTokens + a.OutputTokens += b.OutputTokens + a.CacheReadTokens += b.CacheReadTokens + a.CacheWriteTokens += b.CacheWriteTokens + a.ReasoningTokens += b.ReasoningTokens + return a +} diff --git a/run/phases_test.go b/run/phases_test.go new file mode 100644 index 0000000..54db56c --- /dev/null +++ b/run/phases_test.go @@ -0,0 +1,233 @@ +package run + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + "gitea.stevedudenhoeffer.com/steve/majordomo/llm" + "gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake" + + "gitea.stevedudenhoeffer.com/steve/executus/tool" +) + +// phaseProvider builds a fake provider scripted with the given per-call steps +// (consumed in order across every phase's model call) and a resolver over it, +// returning both so a test can read back each call's request. +func phaseProvider(t *testing.T, steps ...fake.Step) (ModelResolver, *fake.Provider) { + t.Helper() + fp := fake.New("fake") + fp.Enqueue("test-model", steps...) + m, err := fp.Model("test-model") + if err != nil { + t.Fatalf("fake model: %v", err) + } + return func(ctx context.Context, _ string) (context.Context, llm.Model, error) { + return ctx, m, nil + }, fp +} + +// TestPhases_SequentialThreadsOutputs: phases run in order, each phase's output +// is threaded into the next via {{.}}, {{.Query}} reaches a phase, and +// the final phase's output is the run output. +func TestPhases_SequentialThreadsOutputs(t *testing.T) { + models, fp := phaseProvider(t, + fake.Reply("out-a"), + fake.Reply("out-b"), + fake.Reply("out-c"), + ) + ex := New(Config{Registry: tool.NewRegistry(), Models: models}) + + ra := RunnableAgent{ + Name: "pipeline", + ModelTier: "test-model", + Phases: []Phase{ + {Name: "a", SystemPrompt: "Phase A instructions"}, + {Name: "b", SystemPrompt: "B saw: {{.a}}"}, + {Name: "c", SystemPrompt: "C saw: {{.b}} and query {{.Query}}"}, + }, + } + res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r", CallerID: "c"}, "QUERY-TEXT") + if res.Err != nil { + t.Fatalf("run error: %v", res.Err) + } + if res.Output != "out-c" { + t.Fatalf("final output = %q, want the LAST phase's output out-c", res.Output) + } + calls := fp.Calls() + if len(calls) != 3 { + t.Fatalf("want 3 model calls (one per phase), got %d", len(calls)) + } + if got := calls[0].Request.System; got != "Phase A instructions" { + t.Errorf("phase a system = %q", got) + } + if got := calls[1].Request.System; got != "B saw: out-a" { + t.Errorf("phase b should see phase a's output threaded; system = %q", got) + } + if got := calls[2].Request.System; got != "C saw: out-b and query QUERY-TEXT" { + t.Errorf("phase c should see phase b's output + {{.Query}}; system = %q", got) + } +} + +// TestPhases_OptionalFailureSubstitutesFallback: an Optional phase that errors +// does not abort the pipeline — its FallbackMessage becomes its output and is +// threaded into later phases, which still run. +func TestPhases_OptionalFailureSubstitutesFallback(t *testing.T) { + models, fp := phaseProvider(t, + fake.Fail(errors.New("provider exploded")), // phase a fails + fake.Reply("out-b"), // phase b runs + ) + ex := New(Config{Registry: tool.NewRegistry(), Models: models}) + + ra := RunnableAgent{ + Name: "pipeline", + ModelTier: "test-model", + Phases: []Phase{ + {Name: "a", SystemPrompt: "Phase A", Optional: true, FallbackMessage: "FALLBACK-A"}, + {Name: "b", SystemPrompt: "B saw: {{.a}}"}, + }, + } + res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r", CallerID: "c"}, "Q") + if res.Err != nil { + t.Fatalf("optional-phase failure must not fail the run: %v", res.Err) + } + if res.Output != "out-b" { + t.Fatalf("final output = %q, want out-b", res.Output) + } + calls := fp.Calls() + if len(calls) != 2 { + t.Fatalf("want 2 calls (failed phase a + phase b), got %d", len(calls)) + } + if got := calls[1].Request.System; got != "B saw: FALLBACK-A" { + t.Errorf("phase b should see the fallback threaded; system = %q", got) + } +} + +// TestPhases_HardErrorAborts: a NON-optional phase that hits a hard error (not a +// budget/step exhaustion) aborts the pipeline; later phases do not run. +func TestPhases_HardErrorAborts(t *testing.T) { + boom := errors.New("model down") + models, fp := phaseProvider(t, + fake.Fail(boom), // phase a (non-optional) fails hard + fake.Reply("out-b"), // must NOT be consumed + ) + ex := New(Config{Registry: tool.NewRegistry(), Models: models}) + + ra := RunnableAgent{ + Name: "pipeline", + ModelTier: "test-model", + Phases: []Phase{ + {Name: "a", SystemPrompt: "Phase A"}, + {Name: "b", SystemPrompt: "Phase B"}, + }, + } + res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r", CallerID: "c"}, "Q") + if res.Err == nil { + t.Fatal("a hard non-optional phase error must fail the run") + } + if !errors.Is(res.Err, boom) { + t.Errorf("run error %v should wrap the phase's model error", res.Err) + } + if n := len(fp.Calls()); n != 1 { + t.Errorf("pipeline must abort after phase a; got %d calls (phase b should not run)", n) + } +} + +// TestPhases_IsRunFuncBareCall: an IsRunFunc phase produces output via a bare LLM +// call and that output threads into a following loop phase. +func TestPhases_IsRunFuncBareCall(t *testing.T) { + models, fp := phaseProvider(t, + fake.Reply("plan-output"), // IsRunFunc phase a + fake.Reply("final"), // loop phase b + ) + ex := New(Config{Registry: tool.NewRegistry(), Models: models}) + + ra := RunnableAgent{ + Name: "pipeline", + ModelTier: "test-model", + Phases: []Phase{ + {Name: "plan", SystemPrompt: "Make a plan for {{.Query}}", IsRunFunc: true}, + {Name: "exec", SystemPrompt: "Execute: {{.plan}}"}, + }, + } + res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r", CallerID: "c"}, "do-thing") + if res.Err != nil { + t.Fatalf("run error: %v", res.Err) + } + if res.Output != "final" { + t.Fatalf("output = %q, want final", res.Output) + } + calls := fp.Calls() + if len(calls) != 2 { + t.Fatalf("want 2 calls, got %d", len(calls)) + } + if got := calls[0].Request.System; got != "Make a plan for do-thing" { + t.Errorf("IsRunFunc phase system = %q", got) + } + if got := calls[1].Request.System; got != "Execute: plan-output" { + t.Errorf("exec phase should see the plan output threaded; system = %q", got) + } +} + +// TestPhases_SystemHeaderAppliedPerPhase: the platform SystemHeader is prepended +// to every phase's prompt (each phase keeps it). +func TestPhases_SystemHeaderAppliedPerPhase(t *testing.T) { + models, fp := phaseProvider(t, fake.Reply("a"), fake.Reply("b")) + ex := New(Config{Registry: tool.NewRegistry(), Models: models, SystemHeader: "PLATFORM"}) + + ra := RunnableAgent{ + Name: "p", + ModelTier: "test-model", + Phases: []Phase{{Name: "one", SystemPrompt: "P1"}, {Name: "two", SystemPrompt: "P2"}}, + } + if res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r"}, "Q"); res.Err != nil { + t.Fatalf("run error: %v", res.Err) + } + for i, want := range []string{"PLATFORM\n\nP1", "PLATFORM\n\nP2"} { + if got := fp.Calls()[i].Request.System; got != want { + t.Errorf("phase %d system = %q, want %q", i, got, want) + } + } +} + +// TestFilterToolbox: a named subset restricts the toolbox (preserving order); +// empty names = the full palette; unknown names are skipped. +func TestFilterToolbox(t *testing.T) { + box := llm.NewToolbox("base") + noop := func(context.Context, json.RawMessage) (any, error) { return "", nil } + for _, name := range []string{"alpha", "beta", "gamma"} { + if err := box.Add(llm.Tool{Name: name, Description: "d", Handler: noop}); err != nil { + t.Fatalf("add %s: %v", name, err) + } + } + + full := filterToolbox(box, nil) + if len(full.Tools()) != 3 { + t.Errorf("nil names = full palette; got %d tools", len(full.Tools())) + } + + sub := filterToolbox(box, []string{"gamma", "alpha", "nonexistent"}) + names := make([]string, 0) + for _, tl := range sub.Tools() { + names = append(names, tl.Name) + } + if strings.Join(names, ",") != "gamma,alpha" { + t.Errorf("subset (order-preserving, unknown skipped) = %v, want [gamma alpha]", names) + } +} + +// TestExpandPhaseTemplate: {{.Query}} + prior outputs substitute; a parse error +// returns the template unchanged (best-effort). +func TestExpandPhaseTemplate(t *testing.T) { + got := expandPhaseTemplate("q={{.Query}} a={{.a}}", "QQ", map[string]string{"a": "AA"}) + if got != "q=QQ a=AA" { + t.Errorf("expand = %q", got) + } + // Malformed template → returned unchanged. + bad := "{{.Unclosed" + if expandPhaseTemplate(bad, "QQ", nil) != bad { + t.Errorf("malformed template should pass through unchanged") + } +} -- 2.52.0 From 0dd2ced717fdc54d3bce9b604b7293a91acf5593 Mon Sep 17 00:00:00 2001 From: Steve Dudenhoeffer Date: Mon, 29 Jun 2026 15:44:04 -0400 Subject: [PATCH 2/2] fix(run): address gadfly review of the phases PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real findings from the consensus review (37 raw; many devstral dups/noise): - Optional/budget-salvage branches no longer swallow a context cancellation / deadline / critic-kill: such errors return immediately so the run is classified cancelled/timeout/killed, not "ok" with a fallback. (the most serious finding — an Optional final phase could mask a killed run) - IsRunFunc bare phase now feeds the SHARED step observer (not just the audit recorder), so the critic's activity clock + Result.Steps see it — a long synthesize phase no longer looks idle to the critic. - phaseModel returns the resolver's enriched (usage-attribution) context and the phase's calls use it, mirroring the single-loop path (non-base-tier phases were mis-attributed). - salvagePhaseTranscript trims the tail on a rune boundary (was a raw byte slice that could split a UTF-8 rune); maxSalvage is now a named const with rationale. - expandPhaseTemplate logs a WARN on parse/execute failure instead of silently returning the unexpanded template; documented the phase-name identifier requirement + the "Query" shadow. - removed the dead phaseDeps.baseTier field. - extracted multimodalUserMessage, shared by runAgent + the phase runner (was duplicated image-folding). - aggregated phase usage is stamped onto the result even on a hard-error return; TrimSpace computed once; filterToolbox returns the base toolbox as-is for the empty-names (full-palette) case instead of copying; phaseModel WARN no longer prints error=. New test: Optional phase does not swallow a cancellation. Full ./... green. Co-Authored-By: Claude Opus 4.8 (1M context) --- run/executor.go | 27 +++---- run/phases.go | 186 ++++++++++++++++++++++++++++----------------- run/phases_test.go | 25 ++++++ 3 files changed, 151 insertions(+), 87 deletions(-) diff --git a/run/executor.go b/run/executor.go index 76955f3..41546dc 100644 --- a/run/executor.go +++ b/run/executor.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "strings" "time" "gitea.stevedudenhoeffer.com/steve/majordomo/agent" @@ -344,13 +343,13 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio // (Per-phase step caps are fixed — the critic's dynamic ceiling is not // propagated to phases — but its steer + hard deadline still apply.) runRes, runErr = e.runPhases(runCtx, ra, phaseDeps{ - baseModel: model, - baseTier: tier, - baseToolbox: toolbox, - baseMaxIter: maxIter, - sharedOpts: sharedOpts, - steer: steer, - rec: rec, + baseModel: model, + baseToolbox: toolbox, + baseMaxIter: maxIter, + sharedOpts: sharedOpts, + stepObserver: stepObserver, + steer: steer, + rec: rec, }, input, inv.Images) } @@ -492,15 +491,9 @@ func runAgent(ctx context.Context, ag *agent.Agent, input string, images []llm.I if len(images) == 0 { return ag.Run(ctx, input, opts...) } - parts := make([]llm.Part, 0, len(images)+1) - if strings.TrimSpace(input) != "" { - parts = append(parts, llm.Text(input)) - } - for _, img := range images { - parts = append(parts, img) - } // Copy opts before appending so a caller-supplied backing array is never - // mutated/aliased (the variadic slice can have spare capacity). - opts = append(opts[:len(opts):len(opts)], agent.WithHistory([]llm.Message{llm.UserParts(parts...)})) + // mutated/aliased (the variadic slice can have spare capacity). The multimodal + // opening turn (text + image parts) is built by the shared helper. + opts = append(opts[:len(opts):len(opts)], agent.WithHistory([]llm.Message{multimodalUserMessage(input, images)})) return ag.Run(ctx, "", opts...) } diff --git a/run/phases.go b/run/phases.go index add1fa4..7db85f4 100644 --- a/run/phases.go +++ b/run/phases.go @@ -8,6 +8,7 @@ import ( "log/slog" "strings" "text/template" + "unicode/utf8" "gitea.stevedudenhoeffer.com/steve/majordomo/agent" "gitea.stevedudenhoeffer.com/steve/majordomo/llm" @@ -24,17 +25,17 @@ import ( // carries RunnableAgent.Phases as a DTO — actually EXECUTES them (it previously // ignored the slice and ran a single loop with the base prompt). It reuses the // shared run machinery built once in Run: the same stepObserver (so audit/steps/ -// critic accumulate across every phase), the same critic steer, and the same -// compaction option. +// critic-activity accumulate across every phase, including IsRunFunc bare calls), +// the same critic steer, and the same compaction option. // // Semantics preserved from mort's pipeline: -// - phases run sequentially; ctx cancellation between phases aborts the run. +// - phases run sequentially; ctx cancellation/deadline/critic-kill aborts the +// run (even mid-phase and even for an Optional phase). // - IsRunFunc = one bare LLM call, no tools, no loop. -// - Optional phases swallow errors and substitute FallbackMessage. +// - Optional phases swallow NON-context errors and substitute FallbackMessage. // - a non-optional phase that merely exhausts its step/tool budget is NOT fatal: -// its partial transcript is salvaged and the pipeline continues (so a long -// verify phase never discards every earlier phase's work). A hard error -// (cancellation, model failure) still aborts. +// its partial transcript is salvaged and the pipeline continues — EXCEPT a +// final phase that salvaged nothing, which is a genuine empty-result failure. // - per-phase ModelTier resolve failures fall back to the base model with a WARN. // // Deliberately NOT carried over (kernel is leaner than mort's legacy pipeline): @@ -42,35 +43,56 @@ import ( // no-tool-call-is-final-answer termination, like its single-loop path), and the // critic's dynamic iteration ceiling (per-phase caps are fixed at phase start — // the run-level critic's steer + hard deadline still apply across phases). +// +// NOTE on phase names: {{.}} resolves a map key, so a phase whose name +// is not a Go-template identifier (hyphens, spaces, leading digit) cannot be +// referenced as {{.my-phase}} — authors must use {{index . "my-phase"}}. A +// template that fails to parse/execute is logged (WARN) and passed through +// unchanged rather than silently dropped (see expandPhaseTemplate). Avoid naming +// a phase "Query" — it shadows the original-input variable. // phaseDeps carries the per-run state the phase runner shares with Run: the base -// model + tier, the full decorated toolbox (filtered per phase), the base step -// cap, the shared agent options (tool-error limits + step observer + compactor), -// the critic/session steer, and the audit recorder (phase events). +// model, the full decorated toolbox (filtered per phase), the base step cap, the +// shared agent options (tool-error limits + step observer + compactor), the +// shared step observer (also fed by IsRunFunc bare calls), the critic/session +// steer, and the audit recorder (phase events). type phaseDeps struct { - baseModel llm.Model - baseTier string - baseToolbox *llm.Toolbox - baseMaxIter int - sharedOpts []agent.Option - steer func() []llm.Message - rec RunRecorder + baseModel llm.Model + baseToolbox *llm.Toolbox + baseMaxIter int + sharedOpts []agent.Option + stepObserver func(agent.Step) + steer func() []llm.Message + rec RunRecorder } // runPhases executes ra.Phases sequentially and returns a synthetic agent.Result // whose Output is the final phase's output, with Usage aggregated across phases // and Messages set to the last phase's transcript (for the PostRun hook). A hard -// (non-optional, non-budget) phase failure returns the error. +// (non-optional, non-budget) phase failure — and any context cancellation/ +// deadline/critic-kill — returns the error. func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phaseDeps, query string, images []llm.ImagePart) (*agent.Result, error) { outputs := make(map[string]string, len(ra.Phases)) var lastResult *agent.Result var lastOutput string var totalUsage llm.Usage + // finish stamps the aggregated usage + final output onto the synthetic result. + finish := func(err error) (*agent.Result, error) { + if lastResult == nil { + lastResult = &agent.Result{} + } + lastResult.Usage = totalUsage + if err == nil { + lastResult.Output = lastOutput + } + return lastResult, err + } + for i, phase := range ra.Phases { // A killed/timed-out/cancelled run must not start its next phase. if err := runCtx.Err(); err != nil { - return lastResult, err + return finish(err) } instructions := expandPhaseTemplate(phase.SystemPrompt, query, outputs) @@ -84,7 +106,14 @@ func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phas totalUsage = addUsage(totalUsage, res.Usage) } if err != nil { + // A context cancellation / deadline / critic-kill is NEVER swallowed by + // the Optional or budget-salvage branches — the run genuinely ended and + // must surface as cancelled/timeout/killed (statusFor classifies it). + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return finish(err) + } isLast := i == len(ra.Phases)-1 + trimmed := strings.TrimSpace(output) switch { case phase.Optional: output = phase.FallbackMessage @@ -97,13 +126,14 @@ func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phas deps.rec.LogEvent("phase_failed_optional", map[string]any{"phase": phase.Name, "error": err.Error()}) } - case isPhaseBudgetExhaustion(err) && (!isLast || strings.TrimSpace(output) != ""): + case isPhaseBudgetExhaustion(err) && (!isLast || trimmed != ""): // Soft stop: the phase ran out of its step/tool budget before - // composing a final answer. Not fatal — it did real work - // (runOnePhase salvaged its partial transcript into output), and - // aborting would discard every completed phase before it. Degrade - // gracefully and continue. - if strings.TrimSpace(output) == "" { + // composing a final answer. Not fatal — it did real work (runOnePhase + // salvaged its partial transcript into output), and aborting would + // discard every completed phase before it. Degrade and continue. + // (A FINAL phase that salvaged nothing falls through to the hard error + // below: there is no result to return.) + if trimmed == "" { output = fmt.Sprintf("(Phase %q reached its step budget before producing a consolidated result; continuing with its partial findings.)", phase.Name) } else { output += fmt.Sprintf("\n\n(Note: phase %q reached its step budget before fully completing; the above is its partial output.)", phase.Name) @@ -115,7 +145,7 @@ func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phas } default: - return lastResult, fmt.Errorf("pipeline phase %q: %w", phase.Name, err) + return finish(fmt.Errorf("pipeline phase %q: %w", phase.Name, err)) } } @@ -123,14 +153,7 @@ func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phas lastOutput = output } - // Synthesize the run result: the final phase's output, usage aggregated over - // all phases, and the last phase's transcript for the PostRun hook. - if lastResult == nil { - lastResult = &agent.Result{} - } - lastResult.Output = lastOutput - lastResult.Usage = totalUsage - return lastResult, nil + return finish(nil) } // runOnePhase runs a single phase: a bare LLM call for IsRunFunc phases, a fresh @@ -138,21 +161,23 @@ func (e *Executor) runPhases(runCtx context.Context, ra RunnableAgent, deps phas // failed bare call), and any error. On a budget-exhaustion error the loop's // partial transcript is salvaged into the returned output. func (e *Executor) runOnePhase(runCtx context.Context, ra RunnableAgent, deps phaseDeps, phase Phase, instructions, query string, images []llm.ImagePart) (string, *agent.Result, error) { - model := e.phaseModel(runCtx, deps, ra, phase) + phaseCtx, model := e.phaseModel(runCtx, deps, ra, phase) // The phase's expanded instructions are the system prompt (with the platform // header so tools keep their run ids); the original query is the user message. system := e.systemPromptWithBody(instructions) if phase.IsRunFunc { // Bare LLM call: no tool loop, no tools array (some models 400 on an empty - // tools list). The response still lands in the audit token tally. - msgs := []llm.Message{phaseUserMessage(query, images)} - resp, err := model.Generate(runCtx, llm.Request{System: system, Messages: msgs}) + // tools list). The response is fed through the SAME step observer as a loop + // step so the audit token tally, Result.Steps, AND the critic's activity + // clock all see it (a long synthesize phase must not look idle to the critic). + msgs := []llm.Message{multimodalUserMessage(query, images)} + resp, err := model.Generate(phaseCtx, llm.Request{System: system, Messages: msgs}) if err != nil { return "", nil, fmt.Errorf("phase %q model call: %w", phase.Name, err) } - if deps.rec != nil { - deps.rec.OnStep(1, resp) + if deps.stepObserver != nil { + deps.stepObserver(agent.Step{Index: 0, Response: resp}) } return resp.Text(), &agent.Result{ Output: resp.Text(), @@ -175,7 +200,7 @@ func (e *Executor) runOnePhase(runCtx context.Context, ra RunnableAgent, deps ph }, deps.sharedOpts...) ag := agent.New(model, system, opts...) - res, runErr := runAgent(runCtx, ag, query, images, agent.WithSteer(deps.steer)) + res, runErr := runAgent(phaseCtx, ag, query, images, agent.WithSteer(deps.steer)) output := "" if res != nil { output = res.Output @@ -190,19 +215,26 @@ func (e *Executor) runOnePhase(runCtx context.Context, ra RunnableAgent, deps ph return output, res, runErr } -// phaseModel resolves the phase's model tier, falling back to the base model on -// an empty tier or a resolution failure (WARN — visible, non-fatal). -func (e *Executor) phaseModel(ctx context.Context, deps phaseDeps, ra RunnableAgent, phase Phase) llm.Model { +// phaseModel resolves the phase's model tier, returning the resolver's enriched +// context (usage attribution) alongside the model. An empty tier or a resolution +// failure falls back to the base model + the run context (WARN — visible, not +// fatal). Returning the enriched ctx mirrors the single-loop path, which adopts +// ctx = modelCtx, so a non-base-tier phase's calls are attributed correctly. +func (e *Executor) phaseModel(ctx context.Context, deps phaseDeps, ra RunnableAgent, phase Phase) (context.Context, llm.Model) { if phase.ModelTier == "" { - return deps.baseModel + return ctx, deps.baseModel } - _, m, err := e.cfg.Models(ctx, phase.ModelTier) + modelCtx, m, err := e.cfg.Models(ctx, phase.ModelTier) if err != nil || m == nil { + reason := "resolver returned a nil model" + if err != nil { + reason = err.Error() + } slog.Warn("run: pipeline phase model resolve failed; using base model", - "agent", ra.Name, "phase", phase.Name, "tier", phase.ModelTier, "error", err) - return deps.baseModel + "agent", ra.Name, "phase", phase.Name, "tier", phase.ModelTier, "reason", reason) + return ctx, deps.baseModel } - return m + return modelCtx, m } // isPhaseBudgetExhaustion reports whether err is a soft budget/guard stop (the @@ -212,9 +244,15 @@ func isPhaseBudgetExhaustion(err error) bool { return errors.Is(err, agent.ErrMaxSteps) || errors.Is(err, agent.ErrToolLoop) } +// maxSalvageBytes bounds a salvaged partial transcript so a long phase's narrated +// reasoning doesn't blow up the next phase's prompt (the tail is the most recent, +// most relevant reasoning). Matches mort's pipeline cap. +const maxSalvageBytes = 8000 + // salvagePhaseTranscript reconstructs a best-effort phase output from a loop that // ended without a final answer: the assistant's narrated text across every step, -// tail-trimmed to a bound. Returns "" when the model wrote no prose. +// tail-trimmed to maxSalvageBytes on a rune boundary. Returns "" when the model +// wrote no prose. func salvagePhaseTranscript(res *agent.Result) string { if res == nil { return "" @@ -232,22 +270,27 @@ func salvagePhaseTranscript(res *agent.Result) string { } } out := strings.TrimSpace(b.String()) - const maxSalvage = 8000 - if len(out) > maxSalvage { - out = "...(earlier reasoning trimmed)...\n" + out[len(out)-maxSalvage:] + if len(out) > maxSalvageBytes { + tail := out[len(out)-maxSalvageBytes:] + // Advance to the next rune boundary so the cut never splits a UTF-8 rune. + for len(tail) > 0 && !utf8.RuneStart(tail[0]) { + tail = tail[1:] + } + out = "...(earlier reasoning trimmed)...\n" + tail } return out } -// phaseUserMessage builds a phase's user message: the original query text plus -// any inline images. Mirrors the single-loop multimodal seeding in runAgent. -func phaseUserMessage(query string, images []llm.ImagePart) llm.Message { +// multimodalUserMessage builds a user message from text + inline images. Shared +// by the phase runner and runAgent so the image-folding lives in one place. +// Empty text with images yields an image-only message (no empty text part). +func multimodalUserMessage(text string, images []llm.ImagePart) llm.Message { if len(images) == 0 { - return llm.UserText(query) + return llm.UserText(text) } parts := make([]llm.Part, 0, len(images)+1) - if strings.TrimSpace(query) != "" { - parts = append(parts, llm.Text(query)) + if strings.TrimSpace(text) != "" { + parts = append(parts, llm.Text(text)) } for _, img := range images { parts = append(parts, img) @@ -257,11 +300,13 @@ func phaseUserMessage(query string, images []llm.ImagePart) llm.Message { // expandPhaseTemplate applies Go text/template substitution to a phase prompt, // replacing {{.Query}} with the original query and {{.}} with a prior -// phase's output. Returns the original string unchanged if parsing or execution -// fails (best-effort, not fatal). +// phase's output. On a parse/execute error it logs a WARN and returns the +// template unchanged (best-effort, non-fatal) so a misconfigured prompt is +// visible rather than silently masked. func expandPhaseTemplate(tmpl, query string, priorOutputs map[string]string) string { t, err := template.New("phase").Option("missingkey=zero").Parse(tmpl) if err != nil { + slog.Warn("run: pipeline phase template parse failed; using it unexpanded", "error", err) return tmpl } data := map[string]string{"Query": query} @@ -270,22 +315,21 @@ func expandPhaseTemplate(tmpl, query string, priorOutputs map[string]string) str } var buf bytes.Buffer if err := t.Execute(&buf, data); err != nil { + slog.Warn("run: pipeline phase template execute failed; using it unexpanded", "error", err) return tmpl } return buf.String() } -// filterToolbox returns a new toolbox restricted to the named tools (preserving -// palette order). Empty names = the full palette. Unknown names are skipped with -// a WARN — a typo'd phase tool list should not abort a run mid-pipeline. +// filterToolbox returns a toolbox restricted to the named tools (preserving +// palette order). Empty names = the full palette (the base toolbox is returned +// as-is — it is read-only during a run, like the single-loop path). Unknown names +// are skipped with a WARN — a typo'd phase tool list should not abort a run. func filterToolbox(box *llm.Toolbox, names []string) *llm.Toolbox { - out := llm.NewToolbox(box.Name()) if len(names) == 0 { - for _, t := range box.Tools() { - _ = out.Add(t) - } - return out + return box } + out := llm.NewToolbox(box.Name()) for _, name := range names { t, ok := box.Get(name) if !ok { @@ -300,7 +344,9 @@ func filterToolbox(box *llm.Toolbox, names []string) *llm.Toolbox { } // addUsage sums two llm.Usage tallies field-by-field so a phased run reports the -// total tokens across all phases. +// total tokens across all phases. NOTE: if llm.Usage gains a field, add it here +// too — the audit recorder (rec) is the authoritative per-run token source, this +// is the secondary Result.Usage roll-up. func addUsage(a, b llm.Usage) llm.Usage { a.InputTokens += b.InputTokens a.OutputTokens += b.OutputTokens diff --git a/run/phases_test.go b/run/phases_test.go index 54db56c..a8fe249 100644 --- a/run/phases_test.go +++ b/run/phases_test.go @@ -105,6 +105,31 @@ func TestPhases_OptionalFailureSubstitutesFallback(t *testing.T) { } } +// TestPhases_OptionalDoesNotSwallowCancellation: an Optional phase that fails +// with a context cancellation must NOT be swallowed into its FallbackMessage — +// the run genuinely ended (cancel/deadline/critic-kill) and must surface the +// error so the run is classified cancelled/timeout/killed, not "ok". +func TestPhases_OptionalDoesNotSwallowCancellation(t *testing.T) { + models, _ := phaseProvider(t, fake.Fail(context.Canceled)) + ex := New(Config{Registry: tool.NewRegistry(), Models: models}) + + ra := RunnableAgent{ + Name: "pipeline", + ModelTier: "test-model", + Phases: []Phase{ + // IsRunFunc so the cancellation surfaces directly wrapped (%w). + {Name: "a", SystemPrompt: "Phase A", IsRunFunc: true, Optional: true, FallbackMessage: "FB"}, + }, + } + res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r", CallerID: "c"}, "Q") + if !errors.Is(res.Err, context.Canceled) { + t.Fatalf("Optional phase must NOT swallow a cancellation; res.Err = %v", res.Err) + } + if res.Output == "FB" { + t.Error("a cancelled run must not report the fallback message as output") + } +} + // TestPhases_HardErrorAborts: a NON-optional phase that hits a hard error (not a // budget/step exhaustion) aborts the pipeline; later phases do not run. func TestPhases_HardErrorAborts(t *testing.T) { -- 2.52.0