fix(run): address gadfly review of the phases PR
executus CI / test (pull_request) Successful in 48s
executus CI / test (pull_request) Successful in 48s
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=<nil>. New test: Optional phase does not swallow a cancellation. Full ./... green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+116
-70
@@ -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: {{.<PhaseName>}} 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 {{.<PhaseName>}} 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
|
||||
|
||||
Reference in New Issue
Block a user