feat(run): execute multi-phase pipelines (RunnableAgent.Phases)
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}} + {{.<PhaseName>}}), 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) <noreply@anthropic.com>
This commit is contained in:
+14
-2
@@ -55,15 +55,27 @@ type RunnableAgent struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Phase is one step of a multi-step run: its own system prompt, model tier,
|
// 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
|
// iteration cap, and tool subset. Phase prompts are Go text/template strings
|
||||||
// pipeline when their precondition isn't met.
|
// expanded against {{.Query}} (the original input) and {{.<PhaseName>}} (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 {
|
type Phase struct {
|
||||||
Name string
|
Name string
|
||||||
SystemPrompt string
|
SystemPrompt string
|
||||||
ModelTier string
|
ModelTier string
|
||||||
MaxIterations int
|
MaxIterations int
|
||||||
Tools []string
|
Tools []string
|
||||||
|
// 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
|
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
|
// CriticConfig configures the optional run-critic. Enabled gates whether a
|
||||||
|
|||||||
+45
-13
@@ -289,12 +289,10 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
opts := []agent.Option{
|
// Shared agent options used by BOTH the single-loop path and every phase: the
|
||||||
agent.WithToolbox(toolbox),
|
// tool-error guards, the step observer, and optional compaction. The toolbox +
|
||||||
// Step ceiling: a fixed WithMaxSteps(maxIter) normally, but when a critic is
|
// step ceiling are NOT shared (they vary per phase), so they're added per path.
|
||||||
// active it owns a DYNAMIC ceiling (WithMaxStepsFunc) so it can raise a
|
sharedOpts := []agent.Option{
|
||||||
// healthy-but-long run's budget mid-flight. Falls back to maxIter.
|
|
||||||
critic.maxStepsOption(maxIter),
|
|
||||||
agent.WithToolErrorLimits(e.cfg.Defaults.MaxConsecutiveToolErrors, e.cfg.Defaults.MaxSameToolCallRepeats),
|
agent.WithToolErrorLimits(e.cfg.Defaults.MaxConsecutiveToolErrors, e.cfg.Defaults.MaxSameToolCallRepeats),
|
||||||
agent.WithStepObserver(stepObserver),
|
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
|
// 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
|
// 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
|
// 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
|
// One WithSteer drains BOTH the session mailbox (a tool's AttachImages) and
|
||||||
// the critic's nudges before each step.
|
// the critic's nudges before each step.
|
||||||
steer := func() []llm.Message { return append(mailbox.drain(), critic.drainSteer()...) }
|
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 {{.<PhaseName>}} 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)
|
status := statusFor(runCtx, runErr)
|
||||||
if runRes != nil {
|
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 {
|
func (e *Executor) systemPrompt(ra RunnableAgent) string {
|
||||||
if e.cfg.SystemHeader == "" {
|
return e.systemPromptWithBody(ra.SystemPrompt)
|
||||||
return ra.SystemPrompt
|
|
||||||
}
|
}
|
||||||
if 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 body
|
||||||
|
}
|
||||||
|
if body == "" {
|
||||||
return e.cfg.SystemHeader
|
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
|
// compactionThreshold returns the token threshold for the tier's model context
|
||||||
|
|||||||
+311
@@ -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 {{.<PhaseName>}} 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 {{.<PhaseName>}} 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
|
||||||
|
}
|
||||||
@@ -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 {{.<PhaseName>}}, {{.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")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user