Merge pull request 'feat(run): durable checkpoint + resume (wire Ports.Checkpointer)' (#20) from feat/kernel-checkpoint into main

Real findings from the consensus review (44 raw; heavy devstral noise):

- finalizeCheckpoint is now fired from the top-of-Run defer, so it runs on
  EVERY exit: a panic, an early build-error return (before the run loop), AND
  normal completion. Previously an early return on a recovered run left its
  durable record unfinalized → boot recovery would retry it forever on a
  persistent build error. (opus + glm)
- Removed the dead ActivePhase field from run.RunCheckpointState +
  run.ResumeState (and the battery RunCheckpoint) — phase recovery is
  boundary-granular (skip completed phases; the interrupted phase re-runs from
  its start), so ActivePhase was never written nor read. Docs across
  ports/checkpoint/phases now state this plainly (5-model consensus that the
  field + docs over-promised mid-phase resume).
- CheckpointerFactory.Begin error is now logged (WARN) before degrading to
  non-durable, per the documented contract (was silently swallowed). (4 models)
- finalizeCheckpoint logs Complete/Fail errors (was silent).
- Resume phase-skip now keys off a SEPARATE resumeSkip set, not the live
  outputs map — a fresh run with two same-named phases no longer skips the
  second (the outputs map fills as phases run). (opus:max) + regression test.
- Removed the dead checkpoint.factory.now field (never set). (opus + glm)
- Fixed the stale phaseDeps doc (the step observer moved out of sharedOpts to
  per-path). Hoisted the resume guard to a local; dropped the wasted acc
  allocation on the resume path; documented that Save throttling is the
  Checkpointer's responsibility and the accumulated transcript is pre-compaction
  (host size-caps it).

Note (carried from the PR): classifyCheckpointOutcome keys shutdown on
run.ErrShutdown; mort stamps its own runengine.ErrShutdown — the mort wiring PR
aliases them so errors.Is matches.

New test: duplicate phase names both run on a fresh run. Full ./... green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.gitea/workflows/adversarial-review.yml

@@ -38,10 +38,11 @@ jobs:
           && (github.actor == 'steve'
               || github.actor == 'fizi'
               || github.actor == 'dazed'))
-    # Tracks gadfly's v1 release tag — a curated pointer re-moved on each release
-    # (unlike @main, which moves on every push). Central swarm tuning propagates
-    # here automatically; the tradeoff vs a full sha pin is that v1 is mutable.
-    uses: steve/gadfly/.gitea/workflows/review-reusable.yml@v1
+    # Pinned to an immutable gadfly commit (not @v1): our act_runners are long-lived
+    # and cache the reusable-workflow ref, so a moved v1 tag keeps resolving to the
+    # stale cached copy. A unique sha forces a cache miss → fresh fetch. Bump this
+    # sha to adopt central swarm changes.
+    uses: steve/gadfly/.gitea/workflows/review-reusable.yml@5007597cf921dc3f0a83c708878facfe65fd8e8b
     # Least privilege: forward only the review secrets (not `secrets: inherit`,
     # which would expose every repo secret). GITEA_TOKEN is the automatic token.
     secrets:

README.md

@@ -37,7 +37,7 @@ bot) — mort and gadfly are the first two consumers (heavy and light). See
   tool registry, majordomo's agent loop, context compaction, run-bounding, and
   step/audit instrumentation into one `Run(ctx, RunnableAgent, inv) Result`, with
   every host concern behind a nil-safe `run.Ports` (Audit/Budget/Critic/
-  Checkpointer/PaletteSource/Delivery). See `examples/minimal`.
+  Checkpointer/PaletteSource/Delivery/InputFiles). See `examples/minimal`.
 - `model/` — config-driven tier resolution + failover over majordomo, with
   pluggable `UsageSink`/`TraceSink` and `GenerateWith[T]` structured output.
 - `tool/` — the tool registry + 3-stage permission model + SSRF guard.

checkpoint/checkpoint.go

@@ -4,9 +4,9 @@
 // run.Ports.Checkpointer.
 //
 // Mort backs CheckpointStore with its durable-job table; Memory() is the
-// zero-dependency default; contrib/store can add a SQLite one. NOTE: the
-// executor's call into run.Ports.Checkpointer is a P2 follow-up — this battery
-// provides the seam + impls ahead of that wiring.
+// zero-dependency default; contrib/store can add a SQLite one. The executor calls
+// run.Ports.Checkpointer (a CheckpointerFactory) during the run loop; NewFactory
+// wires this battery into that seam.
 package checkpoint
 
 import (
@@ -14,6 +14,8 @@ import (
 	"time"
 
 	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
+
+	"gitea.stevedudenhoeffer.com/steve/executus/run"
 )
 
 // RunCheckpointMeta is the run attribution needed to resume a run from scratch
@@ -33,9 +35,9 @@ type RunCheckpointMeta struct {
 // RunCheckpoint is one persisted snapshot of a run's resumable progress.
 type RunCheckpoint struct {
 	Meta            RunCheckpointMeta
-	Messages    []llm.Message // conversation so far
+	Messages        []llm.Message     // conversation so far (single-loop runs)
 	Iteration       int               // completed agent-loop iterations
-	ActivePhase string        // current phase name (multi-phase agents); "" otherwise
+	CompletedPhases []run.PhaseOutput // finished phases, in order (multi-phase agents)
 	UpdatedAt       time.Time
 }
 

checkpoint/handle.go

@@ -57,6 +57,7 @@ func (h *handle) Save(ctx context.Context, st run.RunCheckpointState) error {
 		Meta:            h.meta,
 		Messages:        st.Messages,
 		Iteration:       st.Iteration,
+		CompletedPhases: st.CompletedPhases,
 		UpdatedAt:       now,
 	}); err != nil {
 		return err
@@ -81,3 +82,40 @@ var _ run.Checkpointer = noop{}
 func (noop) Save(context.Context, run.RunCheckpointState) error { return nil }
 func (noop) Complete(context.Context) error                     { return nil }
 func (noop) Fail(context.Context, error) error                  { return nil }
+
+// factory is a run.CheckpointerFactory that mints a per-run handle over store,
+// deriving the per-run meta from the kernel's RunInfo. It is the battery's glue
+// for the Ports.Checkpointer (factory) seam: every run becomes durable (the
+// store persists snapshots; a host wanting lazy/short-run skipping uses its own
+// factory, as mort does over its durable-job table).
+type factory struct {
+	store    CheckpointStore
+	throttle time.Duration
+}
+
+var _ run.CheckpointerFactory = (*factory)(nil)
+
+// NewFactory returns a run.CheckpointerFactory backed by store: each run gets a
+// per-run Checkpointer (throttled to at most once per throttle). A nil store
+// yields factory.Begin returning a no-op Checkpointer.
+func NewFactory(store CheckpointStore, throttle time.Duration) run.CheckpointerFactory {
+	return &factory{store: store, throttle: throttle}
+}
+
+// Begin mints the per-run Checkpointer. The prompt is read from
+// info.Inputs["prompt"] when present so a recovered run can re-dispatch.
+func (f *factory) Begin(_ context.Context, info run.RunInfo) (run.Checkpointer, error) {
+	prompt, _ := info.Inputs["prompt"].(string)
+	meta := RunCheckpointMeta{
+		RunID:       info.RunID,
+		AgentID:     info.SubjectID,
+		AgentName:   info.Name,
+		CallerID:    info.CallerID,
+		ChannelID:   info.ChannelID,
+		GuildID:     info.GuildID,
+		Prompt:      prompt,
+		ModelTier:   info.ModelTier,
+		ParentRunID: info.ParentRunID,
+	}
+	return New(f.store, meta, f.throttle, nil /* now defaults to time.Now */), nil
+}

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 {{.<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 {
 	Name          string
 	SystemPrompt  string
 	ModelTier     string
 	MaxIterations int
 	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
+	// 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

run/checkpoint.go

@@ -0,0 +1,103 @@
+package run
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+
+	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
+)
+
+// Durable-recovery plumbing for the executor. The Checkpointer port (set via
+// Ports.Checkpointer, a CheckpointerFactory) persists a run's resumable progress
+// during the loop; on boot a host re-dispatches an interrupted run through the
+// executor with a ResumeState (the saved transcript / completed phases) so it
+// CONTINUES rather than restarting, reusing the SAME durable record via an
+// existing Checkpointer. Both are carried into Run via the context (mirrors
+// mort's agentexec.WithResumeState / WithExistingCheckpointer).
+
+// ResumeState carries a recovered run's prior progress into Run so the run
+// continues instead of restarting. The host's recovery path sets it via
+// WithResumeState; the executor reads it:
+//   - single-loop: History seeds the saved transcript (the run continues).
+//   - multi-phase: CompletedPhases are skipped; the interrupted phase re-runs
+//     from its start (boundary-granular — there is no mid-phase transcript
+//     resume, so History is unused for multi-phase runs).
+type ResumeState struct {
+	History         []llm.Message // single-loop transcript (unused for multi-phase)
+	CompletedPhases []PhaseOutput // multi-phase: outputs of finished phases, in order
+}
+
+type resumeStateKey struct{}
+
+// WithResumeState carries a recovered run's prior progress into Run.
+func WithResumeState(ctx context.Context, rs *ResumeState) context.Context {
+	return context.WithValue(ctx, resumeStateKey{}, rs)
+}
+
+func resumeStateFromContext(ctx context.Context) *ResumeState {
+	rs, _ := ctx.Value(resumeStateKey{}).(*ResumeState)
+	return rs
+}
+
+type existingCheckpointerKey struct{}
+
+// WithExistingCheckpointer carries a pre-existing Checkpointer into Run so a
+// recovery re-run reuses the SAME durable record (the executor uses it instead of
+// calling Ports.Checkpointer.Begin).
+func WithExistingCheckpointer(ctx context.Context, cp Checkpointer) context.Context {
+	return context.WithValue(ctx, existingCheckpointerKey{}, cp)
+}
+
+func existingCheckpointerFromContext(ctx context.Context) Checkpointer {
+	cp, _ := ctx.Value(existingCheckpointerKey{}).(Checkpointer)
+	return cp
+}
+
+// checkpointOutcome is the finalize decision for a durable run.
+type checkpointOutcome int
+
+const (
+	checkpointComplete checkpointOutcome = iota
+	checkpointLeaveRunning
+	checkpointFail
+)
+
+// classifyCheckpointOutcome maps (run error, cancellation cause) to the durable
+// finalize action: success clears the checkpoint (Complete); a shutdown-caused
+// cancellation leaves the record so boot recovery picks it up (neither
+// Complete nor Fail); anything else (model error, tool loop, the run's own
+// deadline, a critic kill, a caller cancel) is terminal (Fail). Mirrors mort's
+// agentexec.classifyCheckpointOutcome.
+func classifyCheckpointOutcome(runErr, cause error) checkpointOutcome {
+	switch {
+	case runErr == nil:
+		return checkpointComplete
+	case errors.Is(cause, ErrShutdown):
+		return checkpointLeaveRunning
+	default:
+		return checkpointFail
+	}
+}
+
+// finalizeCheckpoint applies the outcome to the per-run checkpointer (nil-safe).
+// Runs on a detached context so a cancelled run still records its terminal state.
+// Complete/Fail errors are best-effort but logged (a stale record would only
+// cause a wasteful boot-recovery retry, not data loss).
+func finalizeCheckpoint(ctx context.Context, cp Checkpointer, runErr error, cause error) {
+	if cp == nil {
+		return
+	}
+	switch classifyCheckpointOutcome(runErr, cause) {
+	case checkpointComplete:
+		if err := cp.Complete(detach(ctx)); err != nil {
+			slog.Warn("run: checkpoint Complete failed", "error", err)
+		}
+	case checkpointFail:
+		if err := cp.Fail(detach(ctx), runErr); err != nil {
+			slog.Warn("run: checkpoint Fail failed", "error", err)
+		}
+	case checkpointLeaveRunning:
+		// Interrupted by shutdown: leave the record for boot recovery.
+	}
+}

run/checkpoint_test.go

@@ -0,0 +1,200 @@
+package run
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
+	"gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake"
+
+	"gitea.stevedudenhoeffer.com/steve/executus/tool"
+)
+
+// fakeCheckpointer records every Save state + whether Complete/Fail fired.
+type fakeCheckpointer struct {
+	saves     []RunCheckpointState
+	completed bool
+	failed    bool
+	failErr   error
+}
+
+func (c *fakeCheckpointer) Save(_ context.Context, st RunCheckpointState) error {
+	c.saves = append(c.saves, st)
+	return nil
+}
+func (c *fakeCheckpointer) Complete(context.Context) error { c.completed = true; return nil }
+func (c *fakeCheckpointer) Fail(_ context.Context, err error) error {
+	c.failed = true
+	c.failErr = err
+	return nil
+}
+
+// fakeCheckpointFactory hands out one fakeCheckpointer and records the RunInfo.
+type fakeCheckpointFactory struct {
+	cp   *fakeCheckpointer
+	info RunInfo
+}
+
+func (f *fakeCheckpointFactory) Begin(_ context.Context, info RunInfo) (Checkpointer, error) {
+	f.info = info
+	return f.cp, nil
+}
+
+// TestClassifyCheckpointOutcome covers the finalize decision matrix.
+func TestClassifyCheckpointOutcome(t *testing.T) {
+	cases := []struct {
+		name  string
+		err   error
+		cause error
+		want  checkpointOutcome
+	}{
+		{"success", nil, nil, checkpointComplete},
+		{"shutdown", context.Canceled, ErrShutdown, checkpointLeaveRunning},
+		{"critic-kill", context.Canceled, ErrCriticKill, checkpointFail},
+		{"deadline", context.DeadlineExceeded, context.DeadlineExceeded, checkpointFail},
+		{"model-error", errors.New("boom"), nil, checkpointFail},
+		{"caller-cancel", context.Canceled, context.Canceled, checkpointFail},
+	}
+	for _, tc := range cases {
+		if got := classifyCheckpointOutcome(tc.err, tc.cause); got != tc.want {
+			t.Errorf("%s: classifyCheckpointOutcome = %v, want %v", tc.name, got, tc.want)
+		}
+	}
+}
+
+// TestCheckpoint_SingleLoopSaveAndComplete: a durable single-loop run gets a
+// per-run checkpointer (Begin), Saves its transcript each step, and Completes on
+// success (clearing the checkpoint). The RunInfo carries the resume meta.
+func TestCheckpoint_SingleLoopSaveAndComplete(t *testing.T) {
+	models, _ := phaseProvider(t, fake.Reply("done"))
+	cp := &fakeCheckpointer{}
+	f := &fakeCheckpointFactory{cp: cp}
+	ex := New(Config{Registry: tool.NewRegistry(), Models: models, Ports: Ports{Checkpointer: f}})
+
+	res := ex.Run(context.Background(),
+		RunnableAgent{ID: "a1", Name: "boss", ModelTier: "test-model"},
+		tool.Invocation{RunID: "run-x", CallerID: "steve", ChannelID: "chan", GuildID: "g", SkillInputs: map[string]any{"prompt": "go"}},
+		"go")
+	if res.Err != nil {
+		t.Fatalf("run error: %v", res.Err)
+	}
+	if f.info.RunID != "run-x" || f.info.SubjectID != "a1" || f.info.ModelTier != "test-model" || f.info.GuildID != "g" {
+		t.Errorf("Begin RunInfo missing resume meta: %+v", f.info)
+	}
+	if len(cp.saves) == 0 {
+		t.Error("expected at least one checkpoint Save during the run")
+	} else if len(cp.saves[len(cp.saves)-1].Messages) == 0 {
+		t.Error("checkpoint Save should carry the running transcript")
+	}
+	if !cp.completed {
+		t.Error("a successful run must Complete (clear) its checkpoint")
+	}
+	if cp.failed {
+		t.Error("a successful run must NOT Fail its checkpoint")
+	}
+}
+
+// TestCheckpoint_TerminalErrorFails: a run that errors (not shutdown) Fails its
+// checkpoint (clears it — not a recovery candidate).
+func TestCheckpoint_TerminalErrorFails(t *testing.T) {
+	models, _ := phaseProvider(t, fake.Fail(errors.New("model down")))
+	cp := &fakeCheckpointer{}
+	ex := New(Config{Registry: tool.NewRegistry(), Models: models, Ports: Ports{Checkpointer: &fakeCheckpointFactory{cp: cp}}})
+
+	res := ex.Run(context.Background(),
+		RunnableAgent{ID: "a1", ModelTier: "test-model"},
+		tool.Invocation{RunID: "r", CallerID: "c", SkillInputs: map[string]any{"prompt": "go"}}, "go")
+	if res.Err == nil {
+		t.Fatal("expected a run error")
+	}
+	if !cp.failed {
+		t.Error("a terminal (non-shutdown) error must Fail the checkpoint")
+	}
+	if cp.completed {
+		t.Error("a failed run must NOT Complete its checkpoint")
+	}
+}
+
+// TestCheckpoint_ResumeSeedsHistory: a run carrying a ResumeState seeds the saved
+// transcript as the model's opening messages (continues) instead of the input.
+func TestCheckpoint_ResumeSeedsHistory(t *testing.T) {
+	models, fp := phaseProvider(t, fake.Reply("continued"))
+	history := []llm.Message{llm.UserText("prior turn 1"), llm.AssistantText("prior answer 1")}
+	ctx := WithResumeState(context.Background(), &ResumeState{History: history})
+
+	ex := New(Config{Registry: tool.NewRegistry(), Models: models})
+	res := ex.Run(ctx,
+		RunnableAgent{ID: "a1", ModelTier: "test-model"},
+		tool.Invocation{RunID: "r", CallerID: "c", SkillInputs: map[string]any{"prompt": "ignored-on-resume"}}, "ignored-on-resume")
+	if res.Err != nil {
+		t.Fatalf("run error: %v", res.Err)
+	}
+	got := fp.Calls()[0].Request.Messages
+	if len(got) != len(history) {
+		t.Fatalf("resume should seed the saved %d-message transcript, got %d messages", len(history), len(got))
+	}
+}
+
+// TestCheckpoint_PhaseBoundarySavesCompleted: a durable multi-phase run records
+// the completed phases at each boundary, growing the list, and Completes on
+// success.
+func TestCheckpoint_PhaseBoundarySavesCompleted(t *testing.T) {
+	models, _ := phaseProvider(t, fake.Reply("out-a"), fake.Reply("out-b"))
+	cp := &fakeCheckpointer{}
+	ex := New(Config{Registry: tool.NewRegistry(), Models: models, Ports: Ports{Checkpointer: &fakeCheckpointFactory{cp: cp}}})
+
+	ra := RunnableAgent{
+		ID: "p", ModelTier: "test-model",
+		Phases: []Phase{{Name: "a", SystemPrompt: "A"}, {Name: "b", SystemPrompt: "B"}},
+	}
+	if res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r", CallerID: "c"}, "Q"); res.Err != nil {
+		t.Fatalf("run error: %v", res.Err)
+	}
+	// The final phase-boundary Save must list both completed phases.
+	var lastPhaseSave *RunCheckpointState
+	for i := range cp.saves {
+		if len(cp.saves[i].CompletedPhases) > 0 {
+			lastPhaseSave = &cp.saves[i]
+		}
+	}
+	if lastPhaseSave == nil || len(lastPhaseSave.CompletedPhases) != 2 {
+		t.Fatalf("expected a phase-boundary Save listing 2 completed phases; saves=%+v", cp.saves)
+	}
+	if !cp.completed {
+		t.Error("a successful phased run must Complete its checkpoint")
+	}
+}
+
+// TestCheckpoint_ResumeSkipsCompletedPhases: a resumed multi-phase run skips
+// phases already in ResumeState.CompletedPhases (only the remaining phase calls
+// the model) and threads their outputs into the remaining phase's template.
+func TestCheckpoint_ResumeSkipsCompletedPhases(t *testing.T) {
+	models, fp := phaseProvider(t, fake.Reply("out-b")) // ONLY phase b should call the model
+	ctx := WithResumeState(context.Background(), &ResumeState{
+		CompletedPhases: []PhaseOutput{{Name: "a", Output: "saved-a"}},
+	})
+	ex := New(Config{Registry: tool.NewRegistry(), Models: models})
+
+	ra := RunnableAgent{
+		ID: "p", ModelTier: "test-model",
+		Phases: []Phase{
+			{Name: "a", SystemPrompt: "A"},
+			{Name: "b", SystemPrompt: "B saw {{.a}}"},
+		},
+	}
+	res := ex.Run(ctx, ra, tool.Invocation{RunID: "r", CallerID: "c"}, "Q")
+	if res.Err != nil {
+		t.Fatalf("run error: %v", res.Err)
+	}
+	if res.Output != "out-b" {
+		t.Fatalf("output = %q, want out-b", res.Output)
+	}
+	calls := fp.Calls()
+	if len(calls) != 1 {
+		t.Fatalf("only the un-completed phase b should call the model; got %d calls", len(calls))
+	}
+	if calls[0].Request.System != "B saw saved-a" {
+		t.Errorf("resumed phase b should see the completed phase a's saved output; system = %q", calls[0].Request.System)
+	}
+}

run/executor.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
+	"log/slog"
 	"time"
 
 	"gitea.stevedudenhoeffer.com/steve/majordomo/agent"
@@ -114,13 +114,26 @@ type Result struct {
 func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocation, input string) (res Result) {
 	started := time.Now()
 	res = Result{RunID: inv.RunID}
+	// ckpt is the per-run durable checkpointer (resolved below; nil = non-durable).
+	// checkpointCause yields the run context's cancellation cause once the run
+	// context exists; nil before then (an early build-error return).
+	var ckpt Checkpointer
+	var checkpointCause func() error
 	// Enforce the no-panic contract: a panic anywhere in the run (incl. a host
 	// Critic/Audit/Palette callback on the main goroutine) becomes Result.Err
-	// rather than unwinding into the caller.
+	// rather than unwinding into the caller. This defer ALSO finalizes the
+	// checkpoint on EVERY exit path — panic, an early build-error return (before
+	// the run loop), or normal completion — so a recovered run's durable record is
+	// never left dangling (which would loop boot-recovery on a persistent error).
 	defer func() {
 		if r := recover(); r != nil {
 			res.Err = fmt.Errorf("run.Executor: recovered panic: %v", r)
 		}
+		var cause error
+		if checkpointCause != nil {
+			cause = checkpointCause()
+		}
+		finalizeCheckpoint(ctx, ckpt, res.Err, cause)
 	}()
 
 	tier := ra.ModelTier
@@ -166,7 +179,9 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 		Name:          ra.Name,
 		CallerID:      inv.CallerID,
 		ChannelID:     inv.ChannelID,
+		GuildID:       inv.GuildID,
 		ParentRunID:   inv.ParentRunID,
+		ModelTier:     tier,
 		Inputs:        inv.SkillInputs,
 		StartedAt:     started,
 		MaxIterations: maxIter,
@@ -181,6 +196,25 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 		inv.RunState = stateAcc
 	}
 
+	// Durable recovery (optional): a recovered run carries a ResumeState (prior
+	// transcript / completed phases) + an existing Checkpointer in ctx so it
+	// continues on the SAME durable record; a fresh run mints a per-run
+	// Checkpointer via the factory (which decides durability — nil = non-durable).
+	// nil-safe throughout.
+	resume := resumeStateFromContext(ctx)
+	ckpt = existingCheckpointerFromContext(ctx)
+	if ckpt == nil && e.cfg.Ports.Checkpointer != nil {
+		c, cerr := e.cfg.Ports.Checkpointer.Begin(ctx, info)
+		if cerr != nil {
+			// Degrade to non-durable (the documented contract) but log it — a
+			// failing checkpoint store must not fail the run, yet shouldn't be silent.
+			slog.Warn("run: checkpointer Begin failed; running non-durable",
+				"run_id", inv.RunID, "error", cerr)
+		} else {
+			ckpt = c
+		}
+	}
+
 	// Steer mailbox: lets session tools (via inv.AttachImages) feed multimodal
 	// messages into the running conversation before its next step. Created BEFORE
 	// the toolbox build so any tool's handler captures the live AttachImages seam.
@@ -248,6 +282,9 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 	defer cancelCause(nil)
 	runCtx, mergeCancel := MergeCancellation(runCtx, ctx)
 	defer mergeCancel()
+	// The finalize defer (top of Run) now has a run context to read the
+	// cancellation cause from (shutdown vs critic-kill vs deadline vs cancel).
+	checkpointCause = func() error { return context.Cause(runCtx) }
 
 	// Critic (optional): monitors the run for a stall, can nudge/extend/kill via
 	// its host Escalator. Its hard deadline is bound to runCtx (cancel on pass).
@@ -289,14 +326,12 @@ 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 and optional compaction. The toolbox, step ceiling, AND
+	// step observer are added per path (the observer is wrapped for checkpointing,
+	// which differs single-loop vs per-phase).
+	sharedOpts := []agent.Option{
 		agent.WithToolErrorLimits(e.cfg.Defaults.MaxConsecutiveToolErrors, e.cfg.Defaults.MaxSameToolCallRepeats),
-		agent.WithStepObserver(stepObserver),
 	}
 	if e.cfg.Compactor != nil && e.cfg.ContextTokens != nil {
 		if threshold := e.compactionThreshold(tier); threshold > 0 {
@@ -313,15 +348,89 @@ 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
+	// no files. Done after the model/toolbox build but before the loop, so the
+	// descriptor rides the very first user turn.
+	input = e.stageInputFiles(runCtx, inv.RunID, ra.ID, inv.InputFiles, input)
 	// 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))
+
+	resuming := resume != nil && len(resume.History) > 0
+
+	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).
+		//
+		// Checkpointing: wrap the step observer to accumulate the running transcript
+		// and Save it each step. Save is called every step; THROTTLING is the
+		// Checkpointer's responsibility (the battery + mort's durable-job adapter
+		// both throttle + size-cap), so the kernel doesn't gate the hot path. The
+		// accumulated transcript is the pre-compaction one (the observer sees raw
+		// step responses, not the loop's compacted history) — a host that caps size
+		// bounds it. A recovered run seeds the saved transcript and continues.
+		obs := stepObserver
+		if ckpt != nil {
+			var acc []llm.Message
+			if resuming {
+				acc = append([]llm.Message(nil), resume.History...)
+			} else {
+				acc = []llm.Message{multimodalUserMessage(input, inv.Images)}
+			}
+			obs = func(s agent.Step) {
+				stepObserver(s)
+				if s.Response != nil {
+					acc = append(acc, s.Response.Message())
+				}
+				if len(s.Results) > 0 {
+					acc = append(acc, llm.ToolResultsMessage(s.Results...))
+				}
+				_ = ckpt.Save(runCtx, RunCheckpointState{Messages: acc, Iteration: s.Index + 1})
+			}
+		}
+		opts := append([]agent.Option{
+			agent.WithToolbox(toolbox),
+			critic.maxStepsOption(maxIter),
+			agent.WithStepObserver(obs),
+		}, sharedOpts...)
+		ag := agent.New(model, e.systemPrompt(ra), opts...)
+		if resuming {
+			// Resume: seed the saved transcript and continue (no new input — the
+			// completed tool calls in the transcript are NOT re-run).
+			runRes, runErr = ag.Run(runCtx, "", agent.WithSteer(steer), agent.WithHistory(resume.History))
+		} else {
+			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. The
+		// shared step observer (audit/steps/critic) is wired per phase by the phase
+		// runner; checkpointing is phase-boundary granular (completed phases are
+		// recorded so a resumed run skips them).
+		runRes, runErr = e.runPhases(runCtx, ra, phaseDeps{
+			baseModel:    model,
+			baseToolbox:  toolbox,
+			baseMaxIter:  maxIter,
+			sharedOpts:   sharedOpts,
+			stepObserver: stepObserver,
+			steer:        steer,
+			rec:          rec,
+			checkpointer: ckpt,
+			resume:       resume,
+		}, input, inv.Images)
+	}
+
+	// Durable-recovery finalize (Complete/Fail/leave-running) happens in the
+	// top-of-Run defer so it covers panics + early build-error returns too.
 
 	status := statusFor(runCtx, runErr)
 	if runRes != nil {
@@ -397,13 +506,20 @@ func (e *Executor) finishAudit(ctx context.Context, rec RunRecorder, status stri
 }
 
 func (e *Executor) systemPrompt(ra RunnableAgent) string {
-	if e.cfg.SystemHeader == "" {
-		return ra.SystemPrompt
+	return e.systemPromptWithBody(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 + "\n\n" + ra.SystemPrompt
+	return e.cfg.SystemHeader + "\n\n" + body
 }
 
 // compactionThreshold returns the token threshold for the tier's model context
@@ -454,15 +570,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...)
 }

run/input_files.go

@@ -0,0 +1,179 @@
+package run
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"path"
+	"strings"
+	"unicode"
+
+	"gitea.stevedudenhoeffer.com/steve/executus/tool"
+)
+
+// maxInputFileBytes is a defense-in-depth cap at the staging boundary. A host's
+// extraction path may already cap downloads, but stageInputFiles is the trust
+// boundary for the InputFiles seam: a call site or bug that populates InputFiles
+// directly must not write an unbounded blob to the host file store.
+const maxInputFileBytes = 50_000_000
+
+// maxInputFiles bounds how many attachments a single run stages, independent of
+// the per-file byte cap — defense-in-depth against a flood of tiny files.
+const maxInputFiles = 32
+
+// stageInputFiles persists each non-image input attachment into the host file
+// store (Ports.InputFiles) under run scope and appends a descriptor block to the
+// prompt so the agent knows the file_ids it can pass to a worker tool. The bytes
+// are NOT inlined into the model context — the LLM can't read raw audio/binary —
+// so the agent reaches them via a file_id-aware tool (e.g. code_exec files_in,
+// which writes the file to /workspace/<name>).
+//
+// Best-effort: a nil stager, no files, or a per-file save error degrades to
+// "skip that file" — the run still proceeds. Returns the (possibly augmented)
+// prompt.
+func (e *Executor) stageInputFiles(ctx context.Context, runID, agentID string, files []tool.InputFile, prompt string) string {
+	if e.cfg.Ports.InputFiles == nil || len(files) == 0 {
+		return prompt
+	}
+	// Count cap: bound how many attachments one run can stage, independent of the
+	// per-file byte cap (defense-in-depth against a flood of tiny files).
+	if len(files) > maxInputFiles {
+		slog.Warn("run: too many input files, truncating",
+			"agent", agentID, "run_id", runID, "count", len(files), "cap", maxInputFiles)
+		files = files[:maxInputFiles]
+	}
+
+	type stagedFile struct {
+		name, mime, fileID string
+		size               int
+	}
+	var staged []stagedFile
+	seenNames := make(map[string]int, len(files))
+	for _, f := range files {
+		if len(f.Data) == 0 {
+			slog.Warn("run: skipping empty input file",
+				"agent", agentID, "run_id", runID, "name", f.Name)
+			continue
+		}
+		if len(f.Data) > maxInputFileBytes {
+			slog.Warn("run: skipping oversized input file",
+				"agent", agentID, "run_id", runID, "name", f.Name,
+				"size", len(f.Data), "cap", maxInputFileBytes)
+			continue
+		}
+		// Reduce the untrusted filename to a safe base name BEFORE staging or
+		// inlining: strips ../ and absolute-path components (so it can't escape
+		// the host store or /workspace/<name>) and drops control chars/newlines
+		// (so a crafted name can't inject text into the descriptor block below).
+		// Then disambiguate colliding base names so two attachments don't both map
+		// to /workspace/<name> (the second would clobber the first).
+		name := uniqueName(sanitizeName(f.Name), seenNames)
+		// Sanitize the mime ONCE and pass the clean value to both the host store
+		// and the descriptor (don't hand the raw value to StageInputFile).
+		mime := sanitizeField(f.MimeType)
+		fileID, err := e.cfg.Ports.InputFiles.StageInputFile(ctx, runID, agentID, name, mime, f.Data)
+		if err != nil {
+			slog.Warn("run: failed to stage input file",
+				"agent", agentID, "run_id", runID, "name", name, "error", err)
+			continue
+		}
+		if fileID == "" {
+			slog.Warn("run: stager returned empty file_id, skipping",
+				"agent", agentID, "run_id", runID, "name", name)
+			continue
+		}
+		// fileID is host-generated, but sanitize it too before inlining — the
+		// descriptor must never carry control chars no matter the stager impl.
+		staged = append(staged, stagedFile{name: name, mime: mime, fileID: sanitizeField(fileID), size: len(f.Data)})
+	}
+	if len(staged) == 0 {
+		return prompt
+	}
+
+	var b strings.Builder
+	b.WriteString("[ATTACHED FILES]\n")
+	b.WriteString("The user attached the following file(s). Their contents are NOT included in this prompt and you cannot read them directly. ")
+	b.WriteString("To work with one, call the code_exec tool with a files_in entry — e.g. ")
+	b.WriteString(`files_in: [{"name": "<name>", "file_id": "<file_id>"}]`)
+	b.WriteString(" — which writes it to /workspace/<name> inside the Python sandbox. You may also pass a file_id to any other tool that accepts one.\n")
+	for _, s := range staged {
+		fmt.Fprintf(&b, "- %s (%s, %s) → file_id: %s\n", s.name, s.mime, humanizeBytes(s.size), s.fileID)
+	}
+
+	if strings.TrimSpace(prompt) == "" {
+		return b.String()
+	}
+	return prompt + "\n\n" + b.String()
+}
+
+// sanitizeName reduces an untrusted attachment filename to a safe base name. It
+// drops control characters / newlines (which would otherwise let a crafted name
+// inject text into the [ATTACHED FILES] descriptor) and strips every directory
+// component — defeating ../ traversal, nested dirs, and absolute / drive paths
+// both in the host file store and at /workspace/<name>. Returns "attachment"
+// when nothing usable remains (empty, ".", "..").
+func sanitizeName(name string) string {
+	name = sanitizeField(name)
+	// Normalize backslashes so a Windows-style path also reduces to its base.
+	base := path.Base(strings.ReplaceAll(name, `\`, "/"))
+	base = strings.TrimSpace(base)
+	if base == "" || base == "." || base == ".." {
+		return "attachment"
+	}
+	return base
+}
+
+// sanitizeField strips characters that could let a value inlined verbatim into
+// the prompt descriptor break out of its line or visually mislead: control
+// characters (IsControl covers newlines/tabs) AND Unicode format characters
+// (category Cf — e.g. the bidi overrides U+202A–U+202E, which can reorder how
+// the descriptor renders).
+func sanitizeField(s string) string {
+	return strings.Map(func(r rune) rune {
+		if unicode.IsControl(r) || unicode.Is(unicode.Cf, r) {
+			return -1
+		}
+		return r
+	}, s)
+}
+
+// uniqueName returns name unchanged the first time it's seen, then name-2,
+// name-3, … (suffix inserted before the extension) on repeats, recording each
+// result in seen so later collisions keep counting up.
+func uniqueName(name string, seen map[string]int) string {
+	if seen[name] == 0 {
+		seen[name]++
+		return name
+	}
+	ext := path.Ext(name)
+	base := strings.TrimSuffix(name, ext)
+	for {
+		seen[name]++
+		candidate := fmt.Sprintf("%s-%d%s", base, seen[name], ext)
+		if seen[candidate] == 0 {
+			seen[candidate]++
+			return candidate
+		}
+	}
+}
+
+// humanizeBytes renders a byte count as a short human-readable string (e.g.
+// "2.1 MB") for the attached-files descriptor block.
+func humanizeBytes(n int) string {
+	if n < 0 {
+		n = 0
+	}
+	const unit = 1024
+	if n < unit {
+		return fmt.Sprintf("%d B", n)
+	}
+	const prefixes = "KMGTPE"
+	div, exp := int64(unit), 0
+	// Clamp exp to the last prefix so an absurd size (≥1024^7) can't index past
+	// "KMGTPE" and panic — a no-panic guarantee independent of the per-file cap.
+	for v := int64(n) / unit; v >= unit && exp < len(prefixes)-1; v /= unit {
+		div *= unit
+		exp++
+	}
+	return fmt.Sprintf("%.1f %cB", float64(n)/float64(div), prefixes[exp])
+}

run/input_files_test.go

@@ -0,0 +1,243 @@
+package run
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
+
+	"gitea.stevedudenhoeffer.com/steve/executus/tool"
+)
+
+// stagerFunc is a test InputFileStager: it records each staged file and returns
+// a deterministic file_id ("file_<name>"), or an error if err is set.
+type stagerFunc struct {
+	staged []stagedRec
+	err    error
+}
+
+type stagedRec struct {
+	runID, agentID, name, mime string
+	size                       int
+}
+
+func (s *stagerFunc) StageInputFile(_ context.Context, runID, agentID, name, mime string, content []byte) (string, error) {
+	if s.err != nil {
+		return "", s.err
+	}
+	s.staged = append(s.staged, stagedRec{runID, agentID, name, mime, len(content)})
+	return "file_" + name, nil
+}
+
+func newStagerExecutor(s InputFileStager) *Executor {
+	return New(Config{
+		Registry: tool.NewRegistry(),
+		Models:   func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, nil, nil },
+		Ports:    Ports{InputFiles: s},
+	})
+}
+
+// TestStageInputFiles: files are staged via the port and an [ATTACHED FILES]
+// descriptor (with each file_id) is appended to the prompt.
+func TestStageInputFiles(t *testing.T) {
+	st := &stagerFunc{}
+	ex := newStagerExecutor(st)
+	out := ex.stageInputFiles(context.Background(), "run-1", "agent-1",
+		[]tool.InputFile{{Name: "clip.mp3", MimeType: "audio/mpeg", Data: []byte("abcd")}},
+		"transcribe this")
+
+	if len(st.staged) != 1 || st.staged[0].name != "clip.mp3" {
+		t.Fatalf("staged = %+v, want one clip.mp3", st.staged)
+	}
+	if st.staged[0].runID != "run-1" || st.staged[0].agentID != "agent-1" {
+		t.Errorf("stager got runID/agentID = %q/%q, want run-1/agent-1", st.staged[0].runID, st.staged[0].agentID)
+	}
+	for _, want := range []string{"transcribe this", "[ATTACHED FILES]", "clip.mp3", "file_clip.mp3", "audio/mpeg"} {
+		if !strings.Contains(out, want) {
+			t.Errorf("output missing %q:\n%s", want, out)
+		}
+	}
+}
+
+// TestStageInputFilesNoStager: a nil port leaves the prompt untouched and never
+// drops the run.
+func TestStageInputFilesNoStager(t *testing.T) {
+	ex := newStagerExecutor(nil) // Ports.InputFiles == nil
+	out := ex.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "prompt")
+	if out != "prompt" {
+		t.Errorf("nil stager changed the prompt: %q", out)
+	}
+}
+
+// TestStageInputFilesNoFiles: no attachments leaves the prompt untouched.
+func TestStageInputFilesNoFiles(t *testing.T) {
+	ex := newStagerExecutor(&stagerFunc{})
+	out := ex.stageInputFiles(context.Background(), "r", "a", nil, "prompt")
+	if out != "prompt" {
+		t.Errorf("no files changed the prompt: %q", out)
+	}
+}
+
+// TestStageInputFilesDedup: colliding base names are disambiguated so they don't
+// clobber each other at /workspace/<name>.
+func TestStageInputFilesDedup(t *testing.T) {
+	st := &stagerFunc{}
+	ex := newStagerExecutor(st)
+	out := ex.stageInputFiles(context.Background(), "r", "a", []tool.InputFile{
+		{Name: "a.wav", MimeType: "audio/wav", Data: []byte("1")},
+		{Name: "a.wav", MimeType: "audio/wav", Data: []byte("2")},
+	}, "go")
+	if len(st.staged) != 2 {
+		t.Fatalf("staged %d files, want 2", len(st.staged))
+	}
+	if st.staged[0].name != "a.wav" || st.staged[1].name != "a-2.wav" {
+		t.Errorf("dedup names = %q, %q; want a.wav, a-2.wav", st.staged[0].name, st.staged[1].name)
+	}
+	if !strings.Contains(out, "a-2.wav") {
+		t.Errorf("output missing disambiguated name:\n%s", out)
+	}
+}
+
+// TestStageInputFilesSkipsBad: empty + oversized files are skipped; a save error
+// drops only that file. With nothing staged, the prompt is unchanged.
+func TestStageInputFilesSkipsBad(t *testing.T) {
+	// Empty data → skipped; with no good files the prompt is returned as-is.
+	ex := newStagerExecutor(&stagerFunc{})
+	if out := ex.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "empty.bin", Data: nil}}, "p"); out != "p" {
+		t.Errorf("empty file should be skipped, got %q", out)
+	}
+	// A stager error → that file is dropped; nothing staged → prompt unchanged.
+	exErr := newStagerExecutor(&stagerFunc{err: errors.New("disk full")})
+	if out := exErr.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p"); out != "p" {
+		t.Errorf("save error should drop the file and leave the prompt, got %q", out)
+	}
+}
+
+// TestStageInputFilesOversize: a file past the byte cap is skipped (prompt
+// unchanged), exercising the size guard directly.
+func TestStageInputFilesOversize(t *testing.T) {
+	st := &stagerFunc{}
+	ex := newStagerExecutor(st)
+	big := make([]byte, maxInputFileBytes+1)
+	out := ex.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "huge.bin", Data: big}}, "p")
+	if out != "p" || len(st.staged) != 0 {
+		t.Errorf("oversized file should be skipped: out=%q staged=%d", out, len(st.staged))
+	}
+}
+
+// TestStageInputFilesCountCap: more than maxInputFiles attachments are truncated
+// to the cap.
+func TestStageInputFilesCountCap(t *testing.T) {
+	st := &stagerFunc{}
+	ex := newStagerExecutor(st)
+	files := make([]tool.InputFile, maxInputFiles+5)
+	for i := range files {
+		files[i] = tool.InputFile{Name: "f.bin", Data: []byte("x")}
+	}
+	ex.stageInputFiles(context.Background(), "r", "a", files, "p")
+	if len(st.staged) != maxInputFiles {
+		t.Errorf("count cap: staged %d, want %d", len(st.staged), maxInputFiles)
+	}
+}
+
+// TestSanitizeName: traversal + absolute + control-char filenames are reduced to
+// a safe base name (no path separators, no newlines), with a fallback.
+func TestSanitizeName(t *testing.T) {
+	cases := map[string]string{
+		"../../etc/passwd":  "passwd",
+		"/etc/cron.d/x":     "x",
+		`..\..\windows\sys`: "sys",
+		"clip.mp3":          "clip.mp3",
+		"":                  "attachment",
+		"..":                "attachment",
+		".":                 "attachment",
+		"evil\n- injected":  "evil- injected",
+		"a/b/c.wav":         "c.wav",
+	}
+	for in, want := range cases {
+		if got := sanitizeName(in); got != want {
+			t.Errorf("sanitizeName(%q) = %q, want %q", in, got, want)
+		}
+		// A sanitized name must never carry a path separator or newline.
+		got := sanitizeName(in)
+		if strings.ContainsAny(got, "/\\\n\r") {
+			t.Errorf("sanitizeName(%q) = %q still contains a separator/newline", in, got)
+		}
+	}
+}
+
+// TestStageInputFilesSanitizesTraversal: a traversal filename is staged AND
+// described under its safe base name only.
+func TestStageInputFilesSanitizesTraversal(t *testing.T) {
+	st := &stagerFunc{}
+	ex := newStagerExecutor(st)
+	out := ex.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "../../../etc/passwd", MimeType: "text/plain", Data: []byte("x")}}, "go")
+	if len(st.staged) != 1 || st.staged[0].name != "passwd" {
+		t.Fatalf("staged name = %+v, want passwd", st.staged)
+	}
+	if strings.Contains(out, "..") || strings.Contains(out, "/etc/") {
+		t.Errorf("descriptor leaked the traversal path:\n%s", out)
+	}
+}
+
+// TestSanitizeFieldStripsBidiAndControl: control chars AND Unicode format/bidi
+// overrides are removed from inlined values.
+func TestSanitizeFieldStripsBidiAndControl(t *testing.T) {
+	in := "audio/‮mpg\n; rm -rf" // bidi override + newline
+	got := sanitizeField(in)
+	if strings.ContainsAny(got, "\n\r\t") || strings.ContainsRune(got, '‮') {
+		t.Errorf("sanitizeField left control/bidi chars: %q", got)
+	}
+}
+
+// TestStageInputFilesSanitizesMime: a mime with a control char is cleaned in BOTH
+// the staged value and the descriptor.
+func TestStageInputFilesSanitizesMime(t *testing.T) {
+	st := &stagerFunc{}
+	ex := newStagerExecutor(st)
+	out := ex.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "c.wav", MimeType: "audio/wav\ninjected", Data: []byte("x")}}, "go")
+	if len(st.staged) != 1 || strings.ContainsAny(st.staged[0].mime, "\n\r") {
+		t.Errorf("mime not sanitized before staging: %+v", st.staged)
+	}
+	if strings.Contains(out, "\ninjected") {
+		t.Errorf("descriptor carried an unsanitized mime newline:\n%s", out)
+	}
+}
+
+// TestStageInputFilesEmptyFileID: a stager returning an empty file_id drops the
+// file (no blank file_id in the descriptor).
+func TestStageInputFilesEmptyFileID(t *testing.T) {
+	ex := newStagerExecutor(emptyIDStager{})
+	out := ex.stageInputFiles(context.Background(), "r", "a",
+		[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p")
+	if out != "p" {
+		t.Errorf("empty file_id should drop the file, got %q", out)
+	}
+}
+
+type emptyIDStager struct{}
+
+func (emptyIDStager) StageInputFile(context.Context, string, string, string, string, []byte) (string, error) {
+	return "", nil
+}
+
+// TestHumanizeBytesNoPanic: an absurd size clamps to the last prefix instead of
+// indexing past "KMGTPE".
+func TestHumanizeBytesNoPanic(t *testing.T) {
+	defer func() {
+		if r := recover(); r != nil {
+			t.Fatalf("humanizeBytes panicked: %v", r)
+		}
+	}()
+	for _, n := range []int{0, 512, 2048, 5_000_000, 1 << 62} {
+		_ = humanizeBytes(n)
+	}
+}

run/phases.go

@@ -0,0 +1,398 @@
+package run
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"text/template"
+	"unicode/utf8"
+
+	"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-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/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 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 — 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):
+// 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).
+//
+// 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, the full decorated toolbox (filtered per phase), the base step cap, the
+// shared agent options (tool-error limits + compactor — the step observer is
+// added per phase, NOT in sharedOpts, so checkpointing can vary per path), the
+// shared step observer (wired into each phase's loop AND invoked for IsRunFunc
+// bare calls), the critic/session steer, and the audit recorder (phase events).
+type phaseDeps struct {
+	baseModel    llm.Model
+	baseToolbox  *llm.Toolbox
+	baseMaxIter  int
+	sharedOpts   []agent.Option
+	stepObserver func(agent.Step)
+	steer        func() []llm.Message
+	rec          RunRecorder
+	// checkpointer records phase-boundary progress (completed phases) for durable
+	// recovery; nil = non-durable. resume carries a recovered run's completed
+	// phases so they are skipped on re-run. Phase recovery is boundary-granular:
+	// the interrupted (active) phase re-runs from its start (its mid-phase
+	// transcript is NOT resumed — only the single-loop path resumes mid-loop).
+	checkpointer Checkpointer
+	resume       *ResumeState
+}
+
+// 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 — 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 completed []PhaseOutput
+	var lastResult *agent.Result
+	var lastOutput string
+	var totalUsage llm.Usage
+
+	// resumeSkip is the set of phases already finished on a RECOVERED run — kept
+	// SEPARATE from the live `outputs` map (which fills as phases run this time) so
+	// the skip guard only skips RESUME-completed phases, never a fresh run's own
+	// phases. (Reusing `outputs` would make a second phase with a duplicate name
+	// skip itself.) Pre-populate outputs + completed so a resumed run threads the
+	// saved outputs into later phases. The interrupted (active) phase is NOT
+	// pre-populated, so it re-runs from its start (boundary-granular recovery).
+	resumeSkip := map[string]bool{}
+	if deps.resume != nil {
+		for _, pc := range deps.resume.CompletedPhases {
+			outputs[pc.Name] = pc.Output
+			resumeSkip[pc.Name] = true
+			completed = append(completed, pc)
+			lastOutput = pc.Output
+		}
+	}
+
+	// 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 {
+		// Skip phases already completed on a resumed run.
+		if resumeSkip[phase.Name] {
+			continue
+		}
+		// A killed/timed-out/cancelled run must not start its next phase.
+		if err := runCtx.Err(); err != nil {
+			return finish(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 {
+			// 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
+				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 || 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 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)
+				}
+				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 finish(fmt.Errorf("pipeline phase %q: %w", phase.Name, err))
+			}
+		}
+
+		outputs[phase.Name] = output
+		lastOutput = output
+		// Checkpoint the phase boundary: this phase is done, so a resumed run skips
+		// it and continues from the next. (Copy the slice — the checkpointer may
+		// hold/serialize it asynchronously.)
+		completed = append(completed, PhaseOutput{Name: phase.Name, Output: output})
+		if deps.checkpointer != nil {
+			_ = deps.checkpointer.Save(runCtx, RunCheckpointState{
+				CompletedPhases: append([]PhaseOutput(nil), completed...),
+			})
+		}
+	}
+
+	return finish(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) {
+	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 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.stepObserver != nil {
+			deps.stepObserver(agent.Step{Index: 0, Response: 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 + the
+	// shared step observer (audit/steps/critic), on top of the shared opts
+	// (tool-error limits, compactor).
+	opts := append([]agent.Option{
+		agent.WithToolbox(toolbox),
+		agent.WithMaxSteps(maxIter),
+		agent.WithStepObserver(deps.stepObserver),
+	}, deps.sharedOpts...)
+	ag := agent.New(model, system, opts...)
+
+	res, runErr := runAgent(phaseCtx, 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, 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 ctx, deps.baseModel
+	}
+	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, "reason", reason)
+		return ctx, deps.baseModel
+	}
+	return modelCtx, 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)
+}
+
+// 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 maxSalvageBytes on a rune boundary. 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())
+	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
+}
+
+// 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(text)
+	}
+	parts := make([]llm.Part, 0, len(images)+1)
+	if strings.TrimSpace(text) != "" {
+		parts = append(parts, llm.Text(text))
+	}
+	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. 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}
+	for k, v := range priorOutputs {
+		data[k] = v
+	}
+	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 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 {
+	if len(names) == 0 {
+		return box
+	}
+	out := llm.NewToolbox(box.Name())
+	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. 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
+	a.CacheReadTokens += b.CacheReadTokens
+	a.CacheWriteTokens += b.CacheWriteTokens
+	a.ReasoningTokens += b.ReasoningTokens
+	return a
+}

run/phases_test.go

@@ -0,0 +1,278 @@
+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_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_DuplicateNamesBothRun: a fresh (non-resume) run with two phases
+// sharing a name must run BOTH — the resume-skip guard keys off a separate
+// resume set, not the live outputs map (which fills as phases run), so a phase
+// never skips a same-named sibling on a fresh run.
+func TestPhases_DuplicateNamesBothRun(t *testing.T) {
+	models, fp := phaseProvider(t, fake.Reply("first"), fake.Reply("second"))
+	ex := New(Config{Registry: tool.NewRegistry(), Models: models})
+	ra := RunnableAgent{
+		Name: "p", ModelTier: "test-model",
+		Phases: []Phase{{Name: "x", SystemPrompt: "P1"}, {Name: "x", SystemPrompt: "P2"}},
+	}
+	res := ex.Run(context.Background(), ra, tool.Invocation{RunID: "r"}, "Q")
+	if res.Err != nil {
+		t.Fatalf("run error: %v", res.Err)
+	}
+	if n := len(fp.Calls()); n != 2 {
+		t.Fatalf("both same-named phases must run on a fresh run; got %d model calls", n)
+	}
+}
+
+// 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")
+	}
+}

run/ports.go

@@ -33,15 +33,30 @@ type Ports struct {
 	Budget Budget
 	// Critic optionally monitors a long run for hangs/runaways. nil = none.
 	Critic Critic
-	// Checkpointer persists resumable progress for durable recovery. nil = no
-	// checkpointing (a run interrupted by shutdown is simply lost).
-	Checkpointer Checkpointer
+	// Checkpointer mints a per-run Checkpointer for durable recovery (it decides
+	// per run whether the run is durable). nil = no checkpointing (a run
+	// interrupted by shutdown is simply lost).
+	Checkpointer CheckpointerFactory
 	// Palette resolves SkillPalette / SubAgentPalette entries into delegation
 	// tools (skill__<name> / agent__<name>). nil = those entries are inert.
 	Palette PaletteSource
 	// Delivery is where the run's output + artifacts go. nil = the caller
 	// reads the Result in-process (the light-host default).
 	Delivery deliver.Delivery
+	// InputFiles persists non-image input attachments (audio, PDF, binary)
+	// carried on Invocation.InputFiles into a host file store under run scope,
+	// returning file_ids the agent can hand to a worker tool. nil = input files
+	// are silently ignored (the run still proceeds, text-only). The bytes are
+	// never inlined into the model context — the LLM can't read raw audio/binary.
+	InputFiles InputFileStager
+}
+
+// InputFileStager persists a single non-image input attachment into a host file
+// store under run scope and returns a file_id the run can reference. It is the
+// seam mort's skill FileStorage (and any host blob store) implements so the
+// kernel can stage Invocation.InputFiles without importing a storage layer.
+type InputFileStager interface {
+	StageInputFile(ctx context.Context, runID, agentID, name, mime string, content []byte) (fileID string, err error)
 }
 
 // RunInfo describes a run at start time — the attribution a recorder/critic
@@ -52,7 +67,9 @@ type RunInfo struct {
 	Name        string
 	CallerID    string
 	ChannelID   string
+	GuildID     string // the originating guild/server id (empty for DMs/triggers)
 	ParentRunID string
+	ModelTier   string // the run's resolved base tier (for checkpoint re-dispatch)
 	Inputs      map[string]any
 	StartedAt   time.Time
 	// MaxIterations is the run's base tool-dispatch step ceiling, so a critic can
@@ -158,6 +175,16 @@ type CriticHandle interface {
 
 // --- Checkpointer ---
 
+// CheckpointerFactory decides, per run, whether the run is durable and (if so)
+// mints the per-run Checkpointer that records its progress. It returns (nil, nil)
+// for a non-durable run (the common short-run case — no checkpointing overhead).
+// A storage error should be logged and degraded to (nil, nil) so a failing
+// checkpoint store never fails the run. Mirrors mort's
+// agentexec.CheckpointerFactory.
+type CheckpointerFactory interface {
+	Begin(ctx context.Context, info RunInfo) (Checkpointer, error)
+}
+
 // Checkpointer persists a run's resumable progress for durable recovery.
 // Mirrors mort's agentexec.RunCheckpointer.
 type Checkpointer interface {
@@ -170,11 +197,24 @@ type Checkpointer interface {
 	Fail(ctx context.Context, err error) error
 }
 
-// RunCheckpointState is the resumable snapshot a Checkpointer persists. Kept
-// minimal here; the executor extends what it records during the merge.
+// RunCheckpointState is the resumable snapshot a Checkpointer persists.
 type RunCheckpointState struct {
+	// Messages is the running transcript of a SINGLE-LOOP run (grows each step;
+	// resumed via WithHistory). nil for multi-phase runs — phase recovery is
+	// boundary-granular (see CompletedPhases), not mid-phase transcript.
 	Messages  []llm.Message
 	Iteration int
+	// CompletedPhases is set only for multi-phase runs: the outputs of phases
+	// already finished, in phase order, so a resumed run skips them and re-runs
+	// the interrupted phase from its start. nil for single-loop runs.
+	CompletedPhases []PhaseOutput
+}
+
+// PhaseOutput is one completed pipeline phase's name and output text, recorded in
+// a checkpoint so a resumed multi-phase run can skip already-finished phases.
+type PhaseOutput struct {
+	Name   string
+	Output string
 }
 
 // --- PaletteSource ---

tool/registry.go

@@ -154,9 +154,10 @@ type ContinuationContext struct {
 
 // InputFile is a non-image file the user supplied with a run (audio,
 // etc.). The executor stages it into the file store under run scope and
-// surfaces its file_id to the agent. Name is a safe base name (no path
-// separators) suitable for /workspace/<name>; MimeType is the resolved
-// content type; Data is the raw bytes.
+// surfaces its file_id to the agent. Name may be an untrusted attachment
+// filename — the executor reduces it to a safe base name (stripping path
+// separators + control chars) before staging or exposing it as
+// /workspace/<name>; MimeType is the resolved content type; Data is the raw bytes.
 type InputFile struct {
 	Name     string
 	MimeType string