Merge pull request 'feat(skillpack): SKILL.md-subscription battery' (#22) from feat/skillpack-battery into main

Replace Activate's stagedDir string with a BundleStager callback invoked
lazily inside skill_use: when the model loads a pack with bundled files, the
host stages them (mort: into run-scoped file storage) and the returned note is
appended to the body so the model knows how to reach them. A nil stager (or a
stager error) degrades gracefully to just listing the file names.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitea/workflows/adversarial-review.yml

@@ -42,7 +42,7 @@ jobs:
     # 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@7bc3c982fa7b72367034c673f7812bf05e9c503e
+    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:

CLAUDE.md

@@ -85,6 +85,14 @@ BATTERIES (opt-in siblings, each nil-safe + a default):
              (throttled Save/Complete/Fail) + Memory
   budget/    DBBudget rolling-7d + NoOp (run.Budget);    [P4 ✓]
              BudgetStorage iface + Memory default
+  skillpack/ SKILL.md-subscription battery: Manifest +   [P5 ✓]
+             Source (Dir/Git) + Subscription/Store +
+             content-addressed PackCache + Syncer
+             (pending-only; Apply re-pins) + Activate →
+             majordomo agent.Skill (catalog + skill_use,
+             progressive disclosure) + Memory defaults.
+             NOT executus/skill (saved-agent noun) nor
+             majordomo/skill (eager capability bundle).
 
 contrib/store/   SECOND module (+ modernc.org/sqlite):   [P4 ✓]
   pure-Go SQLite impls of ALL store seams: budget +

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
+// zero-dependency default; contrib/store can add a SQLite one. The executor calls
-// executor's call into run.Ports.Checkpointer is a P2 follow-up — this battery
+// run.Ports.Checkpointer (a CheckpointerFactory) during the run loop; NewFactory
-// provides the seam + impls ahead of that wiring.
+// 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
+// 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 {
 	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/critic.go

@@ -22,6 +22,14 @@ type criticBinding struct {
 	h CriticHandle
 }
 
+// criticOwnsDeadline reports whether a critic is configured AND this run enables
+// it — the single predicate that decides the two-tier-timeout path. Used by BOTH
+// Run (to choose the generous runaway ceiling over the literal MaxRuntime cap) and
+// startCritic (the arm/no-op gate), so the two can never drift.
+func (e *Executor) criticOwnsDeadline(ra RunnableAgent) bool {
+	return e.cfg.Ports.Critic != nil && ra.Critic.Enabled
+}
+
 // startCritic begins critic monitoring for this run when one is configured and
 // the agent enables it. It launches a goroutine that cancels runCtx (via
 // cancelCause) the moment the critic's hard deadline passes — the critic may
@@ -31,16 +39,20 @@ type criticBinding struct {
 // "killed"); when the backstop simply expired, it is context.DeadlineExceeded (→
 // "timeout"). Returns (nil, no-op stop) when there is no critic. The caller MUST
 // defer the returned stop.
-func (e *Executor) startCritic(runCtx context.Context, cancelCause context.CancelCauseFunc, ra RunnableAgent, info RunInfo) (*criticBinding, func()) {
+//
+// softTrigger is the run's resolved MaxRuntime: for a critic-owned run MaxRuntime
+// is the soft wake (mort's two-tier semantics — the critic first reviews once the
+// run exceeds its nominal budget, and its backstop = softTrigger × multiplier).
+// The caller (Run) always passes the resolved MaxRuntime, which withFallbacks
+// guarantees is > 0, so no fallback is needed here. (A non-positive soft would make
+// the host Monitor return no handle, and Run's unsupervised-run failsafe then bounds
+// the run at MaxRuntime — so even that impossible case stays bounded.)
+func (e *Executor) startCritic(runCtx context.Context, cancelCause context.CancelCauseFunc, ra RunnableAgent, info RunInfo, softTrigger time.Duration) (*criticBinding, func()) {
 	noop := func() {}
-	if e.cfg.Ports.Critic == nil || !ra.Critic.Enabled {
+	if !e.criticOwnsDeadline(ra) {
 		return nil, noop
 	}
-	soft := e.cfg.Defaults.CriticSoftTimeout
+	h := e.cfg.Ports.Critic.Monitor(runCtx, info, softTrigger)
-	if soft <= 0 {
-		soft = 90 * time.Second // defensive: withFallbacks normally guarantees >0
-	}
-	h := e.cfg.Ports.Critic.Monitor(runCtx, info, soft)
 	if h == nil {
 		return nil, noop
 	}

run/critic_deadline_test.go

@@ -0,0 +1,153 @@
+package run_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
+	"gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake"
+
+	"gitea.stevedudenhoeffer.com/steve/executus/run"
+	"gitea.stevedudenhoeffer.com/steve/executus/tool"
+)
+
+// slowToolInvocation builds an Invocation whose session factory adds a "slow"
+// tool that sleeps for d (respecting ctx). The model script calls it once, then
+// answers — so the run's wall-clock is dominated by d, letting a test set a tiny
+// MaxRuntime and observe whether MaxRuntime hard-cancels the run.
+func slowToolInvocation(runID string, d time.Duration) tool.Invocation {
+	slow := llm.DefineTool("slow", "sleeps for a while",
+		func(ctx context.Context, _ struct{}) (any, error) {
+			select {
+			case <-time.After(d):
+				return "ok", nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+		})
+	return tool.Invocation{
+		RunID: runID,
+		SessionToolFactory: func(_ tool.AgentSession) tool.SessionTools {
+			return tool.SessionTools{Tools: []llm.Tool{slow}}
+		},
+	}
+}
+
+func slowModel() llm.Model {
+	fp := fake.New("fake")
+	fp.Enqueue("m",
+		fake.ReplyWith(llm.Response{ToolCalls: []llm.ToolCall{{ID: "c1", Name: "slow", Arguments: []byte(`{}`)}}}),
+		fake.Reply("done"),
+	)
+	m, _ := fp.Model("m")
+	return m
+}
+
+// TestNoCritic_MaxRuntimeIsHardCap: the legacy contract is preserved — without a
+// critic, MaxRuntime is a literal WithTimeout that kills a run whose work outlasts
+// it. The slow tool (200ms) outlasts MaxRuntime (20ms), so runCtx cancels mid-tool
+// and the run ends in error (timeout).
+func TestNoCritic_MaxRuntimeIsHardCap(t *testing.T) {
+	m := slowModel()
+	ex := run.New(run.Config{
+		Registry: tool.NewRegistry(),
+		Models:   func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
+	})
+	res := ex.Run(context.Background(),
+		run.RunnableAgent{Name: "x", ModelTier: "m", MaxIterations: 5, MaxRuntime: 20 * time.Millisecond},
+		slowToolInvocation("r", 200*time.Millisecond), "go")
+	if res.Err == nil {
+		t.Fatalf("non-critic run should hard-timeout at MaxRuntime; got output=%q err=nil", res.Output)
+	}
+}
+
+// TestCriticOwnsDeadline_SurvivesPastMaxRuntime: the fix — when the critic owns the
+// deadline (Ports.Critic set + Critic.Enabled), MaxRuntime becomes the SOFT trigger
+// and is NOT a hard cap. The fake critic exposes no hard deadline (Deadline()==zero,
+// no kill), so the only hard ceiling is CriticAbsoluteMax (10s here). The slow tool
+// (200ms) outlasts the tiny MaxRuntime (20ms) but the run completes — proving the
+// old agentexec two-tier semantics are restored.
+func TestCriticOwnsDeadline_SurvivesPastMaxRuntime(t *testing.T) {
+	m := slowModel()
+	h := &fakeCriticHandle{} // Deadline()==zero → no hard deadline, no kill
+	ex := run.New(run.Config{
+		Registry: tool.NewRegistry(),
+		Models:   func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
+		Ports:    run.Ports{Critic: &fakeCritic{h: h}},
+		Defaults: run.Defaults{CriticAbsoluteMax: 10 * time.Second},
+	})
+	res := ex.Run(context.Background(),
+		run.RunnableAgent{Name: "watched", ModelTier: "m", MaxIterations: 5, MaxRuntime: 20 * time.Millisecond,
+			Critic: run.CriticConfig{Enabled: true}},
+		slowToolInvocation("r", 200*time.Millisecond), "go")
+	if res.Err != nil {
+		t.Fatalf("critic-owned run must survive past MaxRuntime (soft trigger); got err=%v", res.Err)
+	}
+	if res.Output != "done" {
+		t.Errorf("output = %q, want %q", res.Output, "done")
+	}
+}
+
+// capturingCritic records the soft trigger the executor passes to Monitor.
+type capturingCritic struct {
+	mu   sync.Mutex
+	soft time.Duration
+	h    run.CriticHandle
+}
+
+func (c *capturingCritic) Monitor(_ context.Context, _ run.RunInfo, soft time.Duration) run.CriticHandle {
+	c.mu.Lock()
+	c.soft = soft
+	c.mu.Unlock()
+	return c.h
+}
+
+// TestCriticSoftTriggerIsMaxRuntime: the soft trigger handed to the host critic is
+// the run's resolved MaxRuntime (mort's two-tier model — the critic first wakes once
+// the run exceeds its nominal budget), not some global/default value.
+func TestCriticSoftTriggerIsMaxRuntime(t *testing.T) {
+	fp := fake.New("fake")
+	fp.Enqueue("m", fake.Reply("done"))
+	m, _ := fp.Model("m")
+	cc := &capturingCritic{h: &fakeCriticHandle{}}
+	ex := run.New(run.Config{
+		Registry: tool.NewRegistry(),
+		Models:   func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
+		Ports:    run.Ports{Critic: cc},
+	})
+	const wantSoft = 7 * time.Minute
+	ex.Run(context.Background(),
+		run.RunnableAgent{Name: "x", ModelTier: "m", MaxRuntime: wantSoft, Critic: run.CriticConfig{Enabled: true}},
+		tool.Invocation{RunID: "r"}, "go")
+	cc.mu.Lock()
+	got := cc.soft
+	cc.mu.Unlock()
+	if got != wantSoft {
+		t.Errorf("soft trigger = %v, want the agent's MaxRuntime %v", got, wantSoft)
+	}
+}
+
+// TestCriticOwnsDeadline_NilHandleFallsBackToMaxRuntime: the agent enables the
+// critic but the host Monitor returns NO handle (nil) — there is no deadline-watch,
+// so the run is unsupervised. It must fall back to the nominal MaxRuntime hard cap
+// (the slow 200ms tool outlasts the 20ms MaxRuntime → the run errors), NOT run free
+// up to the generous CriticAbsoluteMax runaway ceiling.
+func TestCriticOwnsDeadline_NilHandleFallsBackToMaxRuntime(t *testing.T) {
+	m := slowModel()
+	cc := &capturingCritic{} // h is the nil interface → Monitor returns a nil handle
+	ex := run.New(run.Config{
+		Registry: tool.NewRegistry(),
+		Models:   func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
+		Ports:    run.Ports{Critic: cc},
+		Defaults: run.Defaults{CriticAbsoluteMax: time.Hour}, // generous ceiling; must NOT be what bounds the run
+	})
+	res := ex.Run(context.Background(),
+		run.RunnableAgent{Name: "x", ModelTier: "m", MaxIterations: 5, MaxRuntime: 20 * time.Millisecond,
+			Critic: run.CriticConfig{Enabled: true}},
+		slowToolInvocation("r", 200*time.Millisecond), "go")
+	if res.Err == nil {
+		t.Fatalf("critic-enabled run with a nil Monitor handle must fall back to the MaxRuntime hard cap; got output=%q err=nil", res.Output)
+	}
+}

run/critic_test.go

@@ -61,8 +61,8 @@ func TestCriticRaisesStepCeiling(t *testing.T) {
 		Registry: tool.NewRegistry(),
 		Models:   func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
 		Ports:    run.Ports{Critic: &fakeCritic{h: h}},
-		// large soft timeout so the deadline-watch never interferes in the test
+		// The fake handle's Deadline() is zero (no hard deadline), so the
-		Defaults: run.Defaults{CriticSoftTimeout: time.Hour},
+		// deadline-watch never interferes regardless of the soft trigger.
 	})
 	res := ex.Run(context.Background(),
 		run.RunnableAgent{Name: "x", ModelTier: "m", MaxIterations: 1, Critic: run.CriticConfig{Enabled: true}},

run/executor.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
+	"log/slog"
 	"time"
 
 	"gitea.stevedudenhoeffer.com/steve/majordomo/agent"
@@ -29,7 +29,17 @@ type Defaults struct {
 	MaxConsecutiveToolErrors int           // loop guard; default 3
 	MaxSameToolCallRepeats   int           // retry-storm guard; default 3
 	CompactionThresholdRatio float64       // fraction of model context to compact at; default 0.7
-	CriticSoftTimeout        time.Duration // idle window before the critic wakes; default 90s
+	// CriticAbsoluteMax is the RUNAWAY ceiling for a critic-OWNED run (Ports.Critic
+	// set AND the agent enables it). For such a run MaxRuntime is the SOFT trigger,
+	// not a hard cap, and the critic's own extendable backstop is the normal
+	// deadline. This ceiling exists ONLY to stop a critic that never advances its
+	// deadline (a broken host handle) from running forever, so it is deliberately
+	// set FAR beyond any realistic backstop (default 24h): the host clamps its own
+	// backstop to a much smaller absolute max (e.g. a 6h host convar), so the ceiling
+	// never pre-empts a healthy supervised run. Keep it well above the host's
+	// absolute max. Never shorter than the run's MaxRuntime. Non-critic runs ignore
+	// it (they keep the literal MaxRuntime kill).
+	CriticAbsoluteMax time.Duration
 }
 
 func (d Defaults) withFallbacks() Defaults {
@@ -51,8 +61,8 @@ func (d Defaults) withFallbacks() Defaults {
 	if d.CompactionThresholdRatio <= 0 {
 		d.CompactionThresholdRatio = 0.7
 	}
-	if d.CriticSoftTimeout <= 0 {
+	if d.CriticAbsoluteMax <= 0 {
-		d.CriticSoftTimeout = 90 * time.Second
+		d.CriticAbsoluteMax = 24 * time.Hour
 	}
 	return d
 }
@@ -114,13 +124,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 +189,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 +206,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.
@@ -231,18 +275,40 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 		postRun = st.PostRun
 	}
 
-	// Run context: bound by MaxRuntime, detached from the caller's deadline so a
+	// Run context: detached from the caller's deadline so a lane/queue wait doesn't
-	// lane/queue wait doesn't eat the run budget (mort's V10 lesson). Caller
+	// eat the run budget (mort's V10 lesson). Caller cancellation still propagates
-	// cancellation still propagates via MergeCancellation. Created BEFORE the
+	// via MergeCancellation. Created BEFORE the step observer so the observer
-	// step observer so the observer forwards the merged run context (not a
+	// forwards the merged run context (not a possibly-cancelled caller ctx) to
-	// possibly-cancelled caller ctx) to OnStep consumers.
+	// OnStep consumers.
-	// MaxRuntime stays a WithTimeout so its DeadlineExceeded propagates through the
+	//
-	// child chain (→ "timeout"), preserving the run's-own-timeout vs caller-cancel
+	// Two-tier timeout: who owns the hard deadline depends on the critic.
-	// distinction. A NESTED cause-carrying layer lets a critic kill surface as a
+	//   - NO critic (the default): MaxRuntime is a literal WithTimeout. Its
-	// distinct "killed" without disturbing that: only an ErrCriticKill cause is
+	//     DeadlineExceeded propagates through the child chain (→ "timeout"),
-	// consulted in statusFor; a generic run error or a caller cancel is classified
+	//     preserving the run's-own-timeout vs caller-cancel distinction.
-	// by the run error itself.
+	//   - critic OWNS the deadline (Ports.Critic set + ra.Critic.Enabled):
-	timeoutCtx, cancelTimeout := context.WithTimeout(context.WithoutCancel(ctx), maxRuntime)
+	//     MaxRuntime becomes the SOFT trigger (passed to startCritic), and the
+	//     critic's extendable backstop — watched in startCritic, which cancels via
+	//     cancelCause — is the real deadline. A slow-but-progressing run is given
+	//     room up to that backstop; only a stalled one is killed. The base context
+	//     gets a WithTimeout at CriticAbsoluteMax (default 24h) purely as a RUNAWAY
+	//     guard for a critic that never advances its deadline: it is set FAR beyond
+	//     any realistic backstop (the host clamps its own backstop to a much smaller
+	//     absolute max, e.g. a 6h host convar), so it does NOT pre-empt a healthy
+	//     supervised run. If the host critic fails to ARM (nil handle), the run is
+	//     unsupervised and we tighten the cap back down to MaxRuntime below.
+	// A NESTED cause-carrying layer (cancelCause) lets a critic kill surface as a
+	// distinct "killed": only an ErrCriticKill cause is consulted in statusFor; a
+	// generic run error, a backstop expiry, or a caller cancel is classified by the
+	// run error itself.
+	criticOwns := e.criticOwnsDeadline(ra)
+	hardCap := maxRuntime
+	if criticOwns {
+		// Runaway guard only — the critic's own (extendable) deadline-watch is the
+		// normal cap. max() keeps it from being shorter than the nominal budget if an
+		// operator sets MaxRuntime above the runaway ceiling (a degenerate config).
+		hardCap = max(e.cfg.Defaults.CriticAbsoluteMax, maxRuntime)
+	}
+	timeoutCtx, cancelTimeout := context.WithTimeout(context.WithoutCancel(ctx), hardCap)
 	defer cancelTimeout()
 	runCtx, cancelCause := context.WithCancelCause(timeoutCtx)
 	defer cancelCause(nil)
@@ -250,11 +316,29 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 	defer mergeCancel()
 
 	// 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).
+	// its host Escalator. When it owns the deadline, MaxRuntime is its soft trigger
-	// nil-safe: no-op when no critic is configured or the agent doesn't enable it.
+	// (so a slow-but-progressing run survives past it); its extendable backstop is
-	critic, stopCritic := e.startCritic(runCtx, cancelCause, ra, info)
+	// bound to runCtx (cancel on pass). nil-safe: no-op when no critic is configured
+	// or the agent doesn't enable it.
+	critic, stopCritic := e.startCritic(runCtx, cancelCause, ra, info, maxRuntime)
 	defer stopCritic()
 
+	// Unsupervised-run failsafe: the agent enabled the critic (so the base context
+	// got the generous runaway ceiling instead of MaxRuntime), but the host Monitor
+	// returned no handle — there is no deadline-watch. Without this the run would be
+	// bounded only by the 24h ceiling. Tighten it back to the nominal MaxRuntime so
+	// an unsupervised run can't hold its slot far past budget. mort's adapter always
+	// arms when the flag is set, so this is pure defence in depth.
+	if criticOwns && critic == nil {
+		var cancelUnsupervised context.CancelFunc
+		runCtx, cancelUnsupervised = context.WithTimeout(runCtx, maxRuntime)
+		defer cancelUnsupervised()
+	}
+	// 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). Set
+	// AFTER the unsupervised-failsafe re-wrap so it reads the context the loop runs on.
+	checkpointCause = func() error { return context.Cause(runCtx) }
+
 	// Step instrumentation: accumulate Result.Steps + fire inv.OnStep, feed the
 	// audit recorder, and keep the live iteration counter fresh. majordomo's
 	// step observer hands us each completed iteration; we zip the model's tool
@@ -289,14 +373,12 @@ 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 and optional compaction. The toolbox, step ceiling, AND
-		// Step ceiling: a fixed WithMaxSteps(maxIter) normally, but when a critic is
+	// step observer are added per path (the observer is wrapped for checkpointing,
-		// active it owns a DYNAMIC ceiling (WithMaxStepsFunc) so it can raise a
+	// which differs single-loop vs per-phase).
-		// healthy-but-long run's budget mid-flight. Falls back to maxIter.
+	sharedOpts := []agent.Option{
-		critic.maxStepsOption(maxIter),
 		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,11 +395,10 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 					})
 				}
 			}
-			opts = append(opts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire)))
+			sharedOpts = append(sharedOpts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire)))
 		}
 	}
 
-	ag := agent.New(model, e.systemPrompt(ra), opts...)
 	// Stage non-image input attachments (audio/PDF/binary) into the host file
 	// store and fold an [ATTACHED FILES] descriptor into the prompt so the agent
 	// can reach them by file_id. No-op when Ports.InputFiles is nil or there are
@@ -327,7 +408,76 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
 	// One WithSteer drains BOTH the session mailbox (a tool's AttachImages) and
 	// the critic's nudges before each step.
 	steer := func() []llm.Message { return append(mailbox.drain(), critic.drainSteer()...) }
-	runRes, runErr := runAgent(runCtx, ag, input, inv.Images, agent.WithSteer(steer))
+
+	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 {
@@ -403,13 +553,20 @@ func (e *Executor) finishAudit(ctx context.Context, rec RunRecorder, status stri
 }
 
 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 + "\n\n" + ra.SystemPrompt
+	return e.cfg.SystemHeader + "\n\n" + body
 }
 
 // compactionThreshold returns the token threshold for the tier's model context
@@ -460,15 +617,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).
+	// mutated/aliased (the variadic slice can have spare capacity). The multimodal
-	opts = append(opts[:len(opts):len(opts)], agent.WithHistory([]llm.Message{llm.UserParts(parts...)}))
+	// 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/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,9 +33,10 @@ 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
+	// Checkpointer mints a per-run Checkpointer for durable recovery (it decides
-	// checkpointing (a run interrupted by shutdown is simply lost).
+	// per run whether the run is durable). nil = no checkpointing (a run
-	Checkpointer Checkpointer
+	// 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
@@ -66,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
@@ -172,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 {
@@ -184,11 +197,24 @@ type Checkpointer interface {
 	Fail(ctx context.Context, err error) error
 }
 
-// RunCheckpointState is the resumable snapshot a Checkpointer persists. Kept
+// RunCheckpointState is the resumable snapshot a Checkpointer persists.
-// minimal here; the executor extends what it records during the merge.
 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 ---

skillpack/activation.go

@@ -0,0 +1,156 @@
+package skillpack
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+
+	mdagent "gitea.stevedudenhoeffer.com/steve/majordomo/agent"
+	"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
+	mdskill "gitea.stevedudenhoeffer.com/steve/majordomo/skill"
+)
+
+// Resolve loads the pinned Pack for each enabled subscription from the cache. It
+// is how a host turns "this agent subscribes to these packs" into activatable
+// packs at run time without touching the network. A pinned digest missing from
+// the cache is an error (the host should have cached it at pin/apply time).
+// Disabled subscriptions are skipped.
+func Resolve(ctx context.Context, cache PackCache, subs []Subscription) ([]*Pack, error) {
+	out := make([]*Pack, 0, len(subs))
+	for i := range subs {
+		s := &subs[i]
+		if !s.Enabled {
+			continue
+		}
+		tree, err := cache.Get(ctx, s.PinnedDigest)
+		if err != nil {
+			return nil, fmt.Errorf("skillpack: resolving %q: %w", s.Name, err)
+		}
+		pack, err := LoadPack(tree)
+		if err != nil {
+			return nil, fmt.Errorf("skillpack: loading %q: %w", s.Name, err)
+		}
+		out = append(out, pack)
+	}
+	return out, nil
+}
+
+// Catalog renders the always-in-prompt block for a set of packs: one line per
+// pack (name + description) plus how to load one. This is the whole prompt cost
+// of a subscription — the bodies stay out until skill_use is called.
+func Catalog(packs []*Pack) string {
+	sorted := make([]*Pack, 0, len(packs))
+	for _, p := range packs {
+		if p != nil && p.Manifest != nil {
+			sorted = append(sorted, p)
+		}
+	}
+	if len(sorted) == 0 {
+		return ""
+	}
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Manifest.Name < sorted[j].Manifest.Name })
+
+	var b strings.Builder
+	b.WriteString("You have access to skills — packaged instructions for specific tasks. ")
+	b.WriteString("When a task matches one, call skill_use with its name to load its full instructions before proceeding.\n\n")
+	b.WriteString("Available skills:\n")
+	for _, p := range sorted {
+		fmt.Fprintf(&b, "- %s: %s\n", p.Manifest.Name, p.Manifest.Description)
+	}
+	return strings.TrimRight(b.String(), "\n")
+}
+
+type skillUseArgs struct {
+	Name string `json:"name" description:"the exact name of the skill to load, from the Available skills list"`
+}
+
+// BundleStager makes a pack's bundled files available to the current run and
+// returns a short note the model can act on (e.g. where the files are and how to
+// reference them). It is called LAZILY, inside the skill_use tool, so a pack's
+// files are staged only when the model actually loads that pack — not for every
+// subscribed pack on every run. A host implements it over its own file plumbing
+// (mort saves the files to run-scoped storage and returns their file_ids). nil =
+// no staging: skill_use just lists the bundled file names.
+type BundleStager func(ctx context.Context, p *Pack) (string, error)
+
+// Activate turns a set of resolved packs into a majordomo agent.Skill: its
+// Instructions are the Catalog, and it contributes a single skill_use tool that
+// returns a named pack's full body (progressive disclosure). Attach the result
+// to an agent with agent.WithSkill. Returns nil when there are no packs, which
+// agent.WithSkill tolerates (a nil skill contributes nothing).
+//
+// stager, if non-nil, is invoked when skill_use loads a pack with bundled files;
+// its returned note is appended to the body so the model knows how to reach the
+// staged scripts/references. A stager error degrades gracefully (the
+// instructions still return, with a note that the files are unavailable).
+func Activate(packs []*Pack, stager BundleStager) mdagent.Skill {
+	byName := make(map[string]*Pack, len(packs))
+	for _, p := range packs {
+		if p != nil && p.Manifest != nil {
+			byName[p.Manifest.Name] = p
+		}
+	}
+	if len(byName) == 0 {
+		return nil
+	}
+
+	tool := llm.DefineTool("skill_use",
+		"Load the full instructions for a skill by name before doing a task it covers. Returns the skill's instructions and, if it has bundled files, how to access them.",
+		func(ctx context.Context, args skillUseArgs) (any, error) {
+			p, ok := byName[strings.TrimSpace(args.Name)]
+			if !ok {
+				return fmt.Sprintf("No skill named %q. Use one of the names from the Available skills list.", args.Name), nil
+			}
+			body := renderPackBody(p)
+			if stager != nil && len(p.Bundled) > 0 {
+				note, err := stager(ctx, p)
+				switch {
+				case err != nil:
+					body += "\n\n(bundled files could not be staged: " + err.Error() + ")"
+				case note != "":
+					body += "\n\n" + note
+				}
+			}
+			return body, nil
+		})
+
+	tb := llm.NewToolbox("skillpack", tool)
+	return mdskill.New("skillpacks",
+		mdskill.WithInstructions(Catalog(packs)),
+		mdskill.WithToolbox(tb),
+	)
+}
+
+// renderPackBody is the base skill_use payload: the pack's instructions plus, if
+// it has any, a list of its bundled file names. A stager (see Activate) appends
+// the concrete access note.
+func renderPackBody(p *Pack) string {
+	if p == nil || p.Manifest == nil {
+		return "Error: invalid skill pack."
+	}
+	var b strings.Builder
+	fmt.Fprintf(&b, "# Skill: %s\n\n%s\n", p.Manifest.Name, p.Manifest.Body)
+	if len(p.Bundled) > 0 {
+		b.WriteString("\nBundled files:\n")
+		for _, f := range p.Bundled {
+			fmt.Fprintf(&b, "- %s\n", f)
+		}
+	}
+	return strings.TrimRight(b.String(), "\n")
+}
+
+// Stage materializes a pack's files under baseDir/<pack name>/ so a host can
+// mount them (read-only is the host's concern) into a sandbox the agent's file
+// tools can read. Returns the pack's staged directory.
+func Stage(p *Pack, baseDir string) (string, error) {
+	if p == nil || p.Manifest == nil {
+		return "", errors.New("skillpack: Stage requires a non-nil pack")
+	}
+	dir := baseDir + "/" + p.Manifest.Name
+	if err := p.Tree.WriteTo(dir); err != nil {
+		return "", err
+	}
+	return dir, nil
+}

skillpack/activation_test.go

@@ -0,0 +1,154 @@
+package skillpack
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func mustPack(t *testing.T, name, body string, extra map[string]string) *Pack {
+	t.Helper()
+	tr := packTree(name, body)
+	for k, v := range extra {
+		tr[k] = []byte(v)
+	}
+	p, err := LoadPack(tr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return p
+}
+
+func TestCatalog(t *testing.T) {
+	packs := []*Pack{
+		mustPack(t, "zebra", "z", nil),
+		mustPack(t, "alpha", "a", nil),
+	}
+	cat := Catalog(packs)
+	if !strings.Contains(cat, "skill_use") {
+		t.Error("catalog should tell the model how to load a skill")
+	}
+	ai := strings.Index(cat, "alpha")
+	zi := strings.Index(cat, "zebra")
+	if ai < 0 || zi < 0 || ai > zi {
+		t.Errorf("catalog should list packs sorted by name:\n%s", cat)
+	}
+	if Catalog(nil) != "" {
+		t.Error("empty catalog should be empty string")
+	}
+}
+
+func TestActivate_SkillUseTool(t *testing.T) {
+	ctx := context.Background()
+	packs := []*Pack{
+		mustPack(t, "pdf", "Use pdfplumber.", map[string]string{"scripts/x.py": "print()"}),
+	}
+	staged := 0
+	stager := func(_ context.Context, p *Pack) (string, error) {
+		staged++
+		return "staged " + p.Manifest.Name + " (file_id=abc)", nil
+	}
+	sk := Activate(packs, stager)
+	if sk == nil {
+		t.Fatal("expected a non-nil skill")
+	}
+	if sk.Instructions() != Catalog(packs) {
+		t.Error("skill instructions should be the catalog")
+	}
+	tb := sk.Tools()
+	tool, ok := tb.Get("skill_use")
+	if !ok {
+		t.Fatal("skill_use tool missing from toolbox")
+	}
+	if staged != 0 {
+		t.Error("stager must be lazy — not called until skill_use runs")
+	}
+
+	// load an existing pack
+	out, err := tool.Handler(ctx, json.RawMessage(`{"name":"pdf"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, _ := out.(string)
+	if !strings.Contains(body, "Use pdfplumber.") {
+		t.Errorf("skill_use body missing instructions: %q", body)
+	}
+	if !strings.Contains(body, "scripts/x.py") {
+		t.Errorf("skill_use should list bundled files: %q", body)
+	}
+	if staged != 1 || !strings.Contains(body, "file_id=abc") {
+		t.Errorf("stager should run on load and its note append to the body: staged=%d body=%q", staged, body)
+	}
+
+	// unknown pack returns guidance, not an error
+	out, err = tool.Handler(ctx, json.RawMessage(`{"name":"nope"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if s, _ := out.(string); !strings.Contains(s, "No skill named") {
+		t.Errorf("unknown skill should return guidance: %q", s)
+	}
+}
+
+func TestActivate_Empty(t *testing.T) {
+	if Activate(nil, nil) != nil {
+		t.Error("no packs should activate to a nil skill")
+	}
+}
+
+func TestNilPackElementsAreSafe(t *testing.T) {
+	packs := []*Pack{nil, mustPack(t, "real", "b", nil), {Manifest: nil}}
+	// Neither Catalog nor Activate may panic on nil / malformed elements.
+	if got := Catalog(packs); !strings.Contains(got, "real") {
+		t.Errorf("catalog should include the valid pack and skip nils: %q", got)
+	}
+	sk := Activate(packs, nil)
+	if sk == nil {
+		t.Fatal("a valid pack among nils should still activate")
+	}
+	if _, ok := sk.Tools().Get("skill_use"); !ok {
+		t.Error("skill_use missing")
+	}
+	// All-nil activates to nothing rather than panicking.
+	if Activate([]*Pack{nil, {Manifest: nil}}, nil) != nil {
+		t.Error("only-nil packs should activate to nil")
+	}
+}
+
+func TestResolveFromCache(t *testing.T) {
+	ctx := context.Background()
+	cache := NewMemoryPackCache()
+	p := mustPack(t, "alpha", "a", nil)
+	cache.Put(ctx, p.Digest, p.Tree)
+
+	subs := []Subscription{
+		{Name: "alpha", PinnedDigest: p.Digest, Enabled: true},
+		{Name: "disabled", PinnedDigest: p.Digest, Enabled: false},
+	}
+	packs, err := Resolve(ctx, cache, subs)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(packs) != 1 || packs[0].Manifest.Name != "alpha" {
+		t.Fatalf("resolve should skip disabled subs; got %d packs", len(packs))
+	}
+
+	// missing from cache is an error
+	subs = []Subscription{{Name: "ghost", PinnedDigest: "deadbeef", Enabled: true}}
+	if _, err := Resolve(ctx, cache, subs); err == nil {
+		t.Fatal("expected error resolving an uncached pin")
+	}
+}
+
+func TestStage(t *testing.T) {
+	dir := t.TempDir()
+	p := mustPack(t, "pdf", "b", map[string]string{"scripts/x.py": "print()"})
+	staged, err := Stage(p, dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.HasSuffix(staged, "/pdf") {
+		t.Errorf("staged dir = %q", staged)
+	}
+}

skillpack/manifest.go

@@ -0,0 +1,208 @@
+package skillpack
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+)
+
+// ManifestName is the required filename at a pack's root.
+const ManifestName = "SKILL.md"
+
+// Limits on manifest fields, matching the Anthropic agent-skills constraints so
+// packs authored against that ecosystem validate here unchanged.
+const (
+	maxNameLen        = 64
+	maxDescriptionLen = 1024
+	maxBodyBytes      = 1 << 20 // 1 MiB of instruction text is already excessive
+)
+
+// Manifest is a parsed SKILL.md: YAML frontmatter plus the markdown body. Only
+// Name and Description are required; everything else is optional and passes
+// through so a host can honor it (or ignore it) without this package growing a
+// policy opinion.
+type Manifest struct {
+	// Name is the pack's stable identifier (kebab-case, unique within a host's
+	// subscriptions). It is what the model passes to skill_use.
+	Name string
+	// Description is the one-liner shown in the catalog — the ONLY text loaded
+	// into the prompt up front, so it must convey when to reach for the skill.
+	Description string
+	// License is an optional SPDX-ish tag, informational only.
+	License string
+	// AllowedTools is the pack author's declared tool allow-list. It is advisory
+	// here: a host MAY intersect it with the agent's real toolset, but it can
+	// only ever NARROW, never grant (see the host wiring, not this package).
+	AllowedTools []string
+	// Metadata is arbitrary passthrough frontmatter (e.g. version) the host may
+	// use; this package does not interpret it.
+	Metadata map[string]string
+	// Body is the markdown instruction text after the frontmatter — the payload
+	// skill_use returns on demand.
+	Body string
+}
+
+// ParseManifest parses a SKILL.md byte slice into a validated Manifest. The
+// input must begin with a `---` YAML frontmatter block; the remainder is the
+// body. It returns a descriptive error on malformed frontmatter or a field that
+// violates the limits, so a bad pack fails loudly at subscribe/sync time rather
+// than silently activating.
+func ParseManifest(raw []byte) (*Manifest, error) {
+	front, body, err := splitFrontmatter(raw)
+	if err != nil {
+		return nil, err
+	}
+
+	// Decode into a permissive intermediate: SKILL.md uses hyphenated keys
+	// (allowed-tools) and lets metadata values be scalars of any type.
+	var fm struct {
+		Name         string         `yaml:"name"`
+		Description  string         `yaml:"description"`
+		License      string         `yaml:"license"`
+		AllowedTools stringList     `yaml:"allowed-tools"`
+		Metadata     map[string]any `yaml:"metadata"`
+	}
+	if err := yaml.Unmarshal(front, &fm); err != nil {
+		return nil, fmt.Errorf("skillpack: invalid SKILL.md frontmatter: %w", err)
+	}
+
+	m := &Manifest{
+		Name:         strings.TrimSpace(fm.Name),
+		Description:  strings.TrimSpace(fm.Description),
+		License:      strings.TrimSpace(fm.License),
+		AllowedTools: []string(fm.AllowedTools),
+		Body:         strings.TrimSpace(string(body)),
+	}
+	if len(fm.Metadata) > 0 {
+		m.Metadata = make(map[string]string, len(fm.Metadata))
+		for k, v := range fm.Metadata {
+			m.Metadata[k] = fmt.Sprintf("%v", v)
+		}
+	}
+
+	if err := m.Validate(); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+// Validate reports the first field that violates the manifest contract.
+func (m *Manifest) Validate() error {
+	switch {
+	case m.Name == "":
+		return fmt.Errorf("skillpack: SKILL.md missing required 'name'")
+	case len(m.Name) > maxNameLen:
+		return fmt.Errorf("skillpack: name %q exceeds %d chars", m.Name, maxNameLen)
+	case !isKebab(m.Name):
+		return fmt.Errorf("skillpack: name %q must be lowercase kebab-case (a-z, 0-9, -)", m.Name)
+	case m.Description == "":
+		return fmt.Errorf("skillpack: SKILL.md missing required 'description'")
+	case len(m.Description) > maxDescriptionLen:
+		return fmt.Errorf("skillpack: description exceeds %d chars", maxDescriptionLen)
+	case len(m.Body) > maxBodyBytes:
+		return fmt.Errorf("skillpack: body exceeds %d bytes", maxBodyBytes)
+	}
+	return nil
+}
+
+// splitFrontmatter separates a leading `---`-delimited YAML block from the body.
+// Leading blank lines/BOM are tolerated. A missing or unterminated block is an
+// error — a SKILL.md without frontmatter has no name/description to catalog.
+func splitFrontmatter(raw []byte) (front, body []byte, err error) {
+	// Strip a leading UTF-8 BOM: editors on some platforms prepend one, and
+	// bytes.TrimSpace (used below) does not remove it, so a BOM would otherwise
+	// make the first "---" fence unrecognizable.
+	raw = bytes.TrimPrefix(raw, []byte{0xEF, 0xBB, 0xBF})
+	s := bufio.NewScanner(bytes.NewReader(raw))
+	s.Buffer(make([]byte, 0, 64*1024), maxBodyBytes+64*1024)
+
+	var frontLines [][]byte
+	var bodyLines [][]byte
+	state := 0 // 0=before open fence, 1=in frontmatter, 2=in body
+	sawOpen := false
+	for s.Scan() {
+		line := s.Bytes()
+		trimmed := bytes.TrimRight(line, "\r")
+		switch state {
+		case 0:
+			if len(bytes.TrimSpace(trimmed)) == 0 {
+				continue // skip leading blanks
+			}
+			if string(bytes.TrimSpace(trimmed)) != "---" {
+				return nil, nil, fmt.Errorf("skillpack: SKILL.md must start with a '---' frontmatter block")
+			}
+			sawOpen = true
+			state = 1
+		case 1:
+			if string(bytes.TrimSpace(trimmed)) == "---" {
+				state = 2
+				continue
+			}
+			frontLines = append(frontLines, append([]byte(nil), trimmed...))
+		case 2:
+			bodyLines = append(bodyLines, append([]byte(nil), trimmed...))
+		}
+	}
+	if err := s.Err(); err != nil {
+		return nil, nil, fmt.Errorf("skillpack: reading SKILL.md: %w", err)
+	}
+	if !sawOpen || state != 2 {
+		return nil, nil, fmt.Errorf("skillpack: SKILL.md frontmatter block is not terminated by a closing '---'")
+	}
+	return bytes.Join(frontLines, []byte("\n")), bytes.Join(bodyLines, []byte("\n")), nil
+}
+
+// stringList decodes either a YAML sequence or a comma-separated scalar into a
+// []string, so `allowed-tools: [Read, Bash]` and `allowed-tools: "Read, Bash"`
+// both work.
+type stringList []string
+
+func (l *stringList) UnmarshalYAML(node *yaml.Node) error {
+	var seq []string
+	if err := node.Decode(&seq); err == nil {
+		*l = trimAll(seq)
+		return nil
+	}
+	var scalar string
+	if err := node.Decode(&scalar); err != nil {
+		return err
+	}
+	*l = trimAll(strings.Split(scalar, ","))
+	return nil
+}
+
+func trimAll(in []string) []string {
+	out := in[:0]
+	for _, s := range in {
+		if t := strings.TrimSpace(s); t != "" {
+			out = append(out, t)
+		}
+	}
+	return out
+}
+
+// isKebab reports whether s is strict lowercase kebab-case: [a-z0-9] segments
+// joined by single hyphens, with no leading, trailing, or consecutive hyphens.
+func isKebab(s string) bool {
+	if s == "" || s[0] == '-' || s[len(s)-1] == '-' {
+		return false
+	}
+	prevHyphen := false
+	for _, r := range s {
+		switch {
+		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
+			prevHyphen = false
+		case r == '-':
+			if prevHyphen {
+				return false
+			}
+			prevHyphen = true
+		default:
+			return false
+		}
+	}
+	return true
+}

skillpack/manifest_test.go

@@ -0,0 +1,96 @@
+package skillpack
+
+import (
+	"strings"
+	"testing"
+)
+
+const goodManifest = `---
+name: pdf-processing
+description: Extract text and tables from PDF files and fill forms.
+license: MIT
+allowed-tools: [Read, Bash]
+metadata:
+  version: 1.2.0
+---
+# PDF Processing
+
+Use pdfplumber for extraction.
+`
+
+func TestParseManifest_Good(t *testing.T) {
+	m, err := ParseManifest([]byte(goodManifest))
+	if err != nil {
+		t.Fatalf("ParseManifest: %v", err)
+	}
+	if m.Name != "pdf-processing" {
+		t.Errorf("name = %q", m.Name)
+	}
+	if !strings.HasPrefix(m.Description, "Extract text") {
+		t.Errorf("description = %q", m.Description)
+	}
+	if m.License != "MIT" {
+		t.Errorf("license = %q", m.License)
+	}
+	if len(m.AllowedTools) != 2 || m.AllowedTools[0] != "Read" || m.AllowedTools[1] != "Bash" {
+		t.Errorf("allowed-tools = %v", m.AllowedTools)
+	}
+	if m.Metadata["version"] != "1.2.0" {
+		t.Errorf("metadata version = %q", m.Metadata["version"])
+	}
+	if !strings.Contains(m.Body, "pdfplumber") || strings.Contains(m.Body, "---") {
+		t.Errorf("body not cleanly extracted: %q", m.Body)
+	}
+}
+
+func TestParseManifest_AllowedToolsScalar(t *testing.T) {
+	m, err := ParseManifest([]byte("---\nname: n\ndescription: d\nallowed-tools: \"Read, Bash , Grep\"\n---\nbody\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(m.AllowedTools) != 3 || m.AllowedTools[2] != "Grep" {
+		t.Errorf("scalar allowed-tools = %v", m.AllowedTools)
+	}
+}
+
+func TestParseManifest_Errors(t *testing.T) {
+	cases := map[string]string{
+		"no frontmatter":     "# just a heading\n",
+		"unterminated":       "---\nname: x\ndescription: y\n",
+		"missing name":       "---\ndescription: y\n---\nb\n",
+		"missing desc":       "---\nname: x\n---\nb\n",
+		"bad name uppercase": "---\nname: PdfProcessing\ndescription: d\n---\nb\n",
+		"bad name space":     "---\nname: pdf processing\ndescription: d\n---\nb\n",
+		"bad name leading -": "---\nname: -pdf\ndescription: d\n---\nb\n",
+		"bad name trailing-": "---\nname: pdf-\ndescription: d\n---\nb\n",
+		"bad name double -":  "---\nname: pdf--tools\ndescription: d\n---\nb\n",
+		"bad yaml":           "---\nname: [unclosed\n---\nb\n",
+	}
+	for label, in := range cases {
+		if _, err := ParseManifest([]byte(in)); err == nil {
+			t.Errorf("%s: expected error, got nil", label)
+		}
+	}
+}
+
+func TestParseManifest_LeadingBlanksAndCRLF(t *testing.T) {
+	in := "\r\n\n---\r\nname: ok-name\r\ndescription: fine\r\n---\r\nbody line\r\n"
+	m, err := ParseManifest([]byte(in))
+	if err != nil {
+		t.Fatalf("tolerant parse: %v", err)
+	}
+	if m.Name != "ok-name" || m.Body != "body line" {
+		t.Errorf("got name=%q body=%q", m.Name, m.Body)
+	}
+}
+
+func TestParseManifest_BOM(t *testing.T) {
+	in := append([]byte{0xEF, 0xBB, 0xBF}, []byte("---\nname: bom-ok\ndescription: d\n---\nbody\n")...)
+	m, err := ParseManifest(in)
+	if err != nil {
+		t.Fatalf("BOM-prefixed SKILL.md should parse: %v", err)
+	}
+	if m.Name != "bom-ok" {
+		t.Errorf("name = %q", m.Name)
+	}
+}

skillpack/memory.go

@@ -0,0 +1,114 @@
+package skillpack
+
+import (
+	"context"
+	"sort"
+	"sync"
+)
+
+// Memory is a zero-dependency in-process Store — a light host or a test gets
+// subscription persistence with no DB. Returned values are copies, so callers
+// can mutate them without corrupting the store.
+type Memory struct {
+	mu   sync.RWMutex
+	subs map[string]*Subscription // by ID
+}
+
+// NewMemory returns an empty in-memory Store.
+func NewMemory() *Memory {
+	return &Memory{subs: map[string]*Subscription{}}
+}
+
+var _ Store = (*Memory)(nil)
+
+func (m *Memory) Initialize(context.Context) error { return nil }
+
+func (m *Memory) Save(_ context.Context, s *Subscription) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	cp := *s
+	m.subs[s.ID] = &cp
+	return nil
+}
+
+func (m *Memory) Get(_ context.Context, id string) (*Subscription, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	s, ok := m.subs[id]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	cp := *s
+	return &cp, nil
+}
+
+func (m *Memory) GetByName(_ context.Context, name string) (*Subscription, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	for _, s := range m.subs {
+		if s.Name == name {
+			cp := *s
+			return &cp, nil
+		}
+	}
+	return nil, ErrNotFound
+}
+
+func (m *Memory) List(context.Context) ([]Subscription, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	out := make([]Subscription, 0, len(m.subs))
+	for _, s := range m.subs {
+		out = append(out, *s)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
+	return out, nil
+}
+
+func (m *Memory) Delete(_ context.Context, id string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.subs, id)
+	return nil
+}
+
+// MemoryPackCache is a zero-dependency in-process PackCache. Trees are copied on
+// the way in and out so a cached pin is immutable in practice.
+type MemoryPackCache struct {
+	mu    sync.RWMutex
+	trees map[string]Tree
+}
+
+// NewMemoryPackCache returns an empty in-memory PackCache.
+func NewMemoryPackCache() *MemoryPackCache {
+	return &MemoryPackCache{trees: map[string]Tree{}}
+}
+
+var _ PackCache = (*MemoryPackCache)(nil)
+
+func (c *MemoryPackCache) Put(_ context.Context, digest string, t Tree) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.trees[digest] = cloneTree(t)
+	return nil
+}
+
+func (c *MemoryPackCache) Get(_ context.Context, digest string) (Tree, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	t, ok := c.trees[digest]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	return cloneTree(t), nil
+}
+
+func cloneTree(t Tree) Tree {
+	cp := make(Tree, len(t))
+	for k, v := range t {
+		b := make([]byte, len(v))
+		copy(b, v)
+		cp[k] = b
+	}
+	return cp
+}

skillpack/pack.go

@@ -0,0 +1,139 @@
+package skillpack
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io/fs"
+	"os"
+	"path"
+	"path/filepath"
+	"sort"
+	"strings"
+)
+
+// Tree is a pack's file set: relative slash-separated path -> file bytes,
+// including the SKILL.md itself. It is self-contained (no live filesystem
+// handle) so it can be cached, digested, and staged without worrying about the
+// lifetime of a clone or temp dir.
+type Tree map[string][]byte
+
+// Digest is the content address of the tree: a SHA-256 over every file's path
+// and bytes, order-independent. Two trees with identical contents produce the
+// same digest regardless of how they were fetched — this is the pin identity
+// and the change-detection signal (a git SHA is provenance, but the digest is
+// what says "the bytes an agent runs changed").
+func (t Tree) Digest() string {
+	paths := t.Paths()
+	h := sha256.New()
+	for _, p := range paths {
+		fh := sha256.Sum256(t[p])
+		// path \x00 filehash \n — the NUL prevents path/content boundary games.
+		fmt.Fprintf(h, "%s\x00%s\n", p, hex.EncodeToString(fh[:]))
+	}
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+// Paths returns the tree's file paths, sorted.
+func (t Tree) Paths() []string {
+	out := make([]string, 0, len(t))
+	for p := range t {
+		out = append(out, p)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// WriteTo materializes the tree under dir (creating it and any parents). It is
+// how a host stages a pack's files for a sandbox; the host owns mount/read-only
+// policy. Paths are cleaned and constrained to dir — a tree entry that escapes
+// (via .. or an absolute path) is rejected rather than written outside dir.
+func (t Tree) WriteTo(dir string) error {
+	for _, p := range t.Paths() {
+		dest := filepath.Join(dir, filepath.FromSlash(p))
+		if !within(dir, dest) {
+			return fmt.Errorf("skillpack: refusing to stage %q outside %q", p, dir)
+		}
+		if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
+			return err
+		}
+		if err := os.WriteFile(dest, t[p], 0o644); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Pack is a fetched, parsed pack: its manifest, its file tree, the tree's
+// content digest, and the non-manifest ("bundled") file paths a host can stage.
+type Pack struct {
+	Manifest *Manifest
+	Tree     Tree
+	Digest   string
+	// Bundled is every tree path except the SKILL.md, sorted — the scripts and
+	// reference files skill_use points the model at.
+	Bundled []string
+}
+
+// LoadPack parses a fetched Tree into a Pack: it requires a root SKILL.md,
+// parses+validates it, computes the digest, and lists the bundled files.
+func LoadPack(t Tree) (*Pack, error) {
+	raw, ok := t[ManifestName]
+	if !ok {
+		return nil, ErrNoManifest
+	}
+	m, err := ParseManifest(raw)
+	if err != nil {
+		return nil, err
+	}
+	bundled := make([]string, 0, len(t))
+	for _, p := range t.Paths() {
+		if p != ManifestName {
+			bundled = append(bundled, p)
+		}
+	}
+	return &Pack{Manifest: m, Tree: t, Digest: t.Digest(), Bundled: bundled}, nil
+}
+
+// readTree reads an entire fs.FS (rooted at ".") into a Tree, skipping
+// directories. It is the shared reader for DirSource and GitSource, so both
+// produce identical self-contained trees.
+func readTree(fsys fs.FS) (Tree, error) {
+	t := Tree{}
+	err := fs.WalkDir(fsys, ".", func(p string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			return nil
+		}
+		// Skip symlinks. A pack must be self-contained; os.DirFS + ReadFile
+		// follows symlinks, so a malicious pack with `SKILL.md -> /etc/passwd`
+		// or `scripts/x -> ../../.ssh/id_rsa` would otherwise read host files
+		// into the tree. WalkDir yields a symlink-to-dir as a non-dir entry
+		// carrying ModeSymlink, so this one check covers file and dir symlinks.
+		if d.Type()&fs.ModeSymlink != 0 {
+			return nil
+		}
+		b, err := fs.ReadFile(fsys, p)
+		if err != nil {
+			return err
+		}
+		t[path.Clean(p)] = b
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return t, nil
+}
+
+// within reports whether dest is inside dir (defense against path traversal in
+// a staged tree).
+func within(dir, dest string) bool {
+	rel, err := filepath.Rel(dir, dest)
+	if err != nil {
+		return false
+	}
+	return rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))
+}

skillpack/pack_test.go

@@ -0,0 +1,105 @@
+package skillpack
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func sampleTree() Tree {
+	return Tree{
+		ManifestName:         []byte(goodManifest),
+		"scripts/fill.py":    []byte("print('hi')\n"),
+		"references/spec.md": []byte("# spec\n"),
+	}
+}
+
+func TestTreeDigest_StableAndContentSensitive(t *testing.T) {
+	a := sampleTree()
+	b := sampleTree()
+	if a.Digest() != b.Digest() {
+		t.Fatal("identical trees must share a digest")
+	}
+	b["scripts/fill.py"] = []byte("print('bye')\n")
+	if a.Digest() == b.Digest() {
+		t.Fatal("content change must change the digest")
+	}
+	// Adding a file changes the digest.
+	c := sampleTree()
+	c["extra.txt"] = []byte("x")
+	if a.Digest() == c.Digest() {
+		t.Fatal("added file must change the digest")
+	}
+}
+
+func TestLoadPack(t *testing.T) {
+	p, err := LoadPack(sampleTree())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p.Manifest.Name != "pdf-processing" {
+		t.Errorf("name = %q", p.Manifest.Name)
+	}
+	if len(p.Bundled) != 2 || p.Bundled[0] != "references/spec.md" || p.Bundled[1] != "scripts/fill.py" {
+		t.Errorf("bundled = %v (want sorted, sans SKILL.md)", p.Bundled)
+	}
+	if p.Digest == "" {
+		t.Error("digest empty")
+	}
+}
+
+func TestLoadPack_NoManifest(t *testing.T) {
+	if _, err := LoadPack(Tree{"readme.md": []byte("x")}); err != ErrNoManifest {
+		t.Fatalf("want ErrNoManifest, got %v", err)
+	}
+}
+
+func TestTreeWriteTo(t *testing.T) {
+	dir := t.TempDir()
+	if err := sampleTree().WriteTo(dir); err != nil {
+		t.Fatal(err)
+	}
+	got, err := os.ReadFile(filepath.Join(dir, "scripts", "fill.py"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(got) != "print('hi')\n" {
+		t.Errorf("staged content = %q", got)
+	}
+}
+
+func TestReadTree_SkipsSymlinks(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, ManifestName), []byte(goodManifest), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	// A malicious pack pointing at a host file must NOT be read into the tree.
+	secret := filepath.Join(t.TempDir(), "secret")
+	if err := os.WriteFile(secret, []byte("TOPSECRET"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Symlink(secret, filepath.Join(dir, "leak")); err != nil {
+		t.Skipf("symlink unsupported: %v", err)
+	}
+	tree, err := readTree(os.DirFS(dir))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := tree["leak"]; ok {
+		t.Fatal("symlink was followed into the tree — arbitrary host file read")
+	}
+	if _, ok := tree[ManifestName]; !ok {
+		t.Fatal("real file should still be read")
+	}
+}
+
+func TestTreeWriteTo_RejectsTraversal(t *testing.T) {
+	dir := t.TempDir()
+	evil := Tree{"../escape.txt": []byte("nope")}
+	if err := evil.WriteTo(dir); err == nil {
+		t.Fatal("expected traversal rejection")
+	}
+	if _, err := os.Stat(filepath.Join(filepath.Dir(dir), "escape.txt")); err == nil {
+		t.Fatal("traversal file was written outside dir")
+	}
+}

skillpack/skillpack.go

@@ -0,0 +1,54 @@
+// Package skillpack is the SKILL.md-subscription battery: it lets an agent host
+// subscribe to skill packages published as directories/git repos in the
+// Anthropic "agent skills" format (a SKILL.md manifest plus optional bundled
+// scripts and reference files) and activate them for a run with progressive
+// disclosure.
+//
+// It is a THIRD, distinct concept from the two "skill" nouns already in the
+// stack — do not conflate them:
+//
+//   - majordomo/skill  — a lightweight capability bundle (instructions + tools)
+//     appended to an agent eagerly at construction.
+//   - executus/skill   — a heavyweight persisted "saved agent" noun.
+//   - executus/skillpack (this package) — an externally-authored, versioned,
+//     on-demand-loaded instruction pack fetched from a Source and pinned by
+//     content digest.
+//
+// Progressive disclosure is the reason this is not just a majordomo/skill:
+// majordomo skills inject their whole instruction text into the system prompt
+// up front, which does not scale to a catalog of large third-party packs. Here
+// only each pack's name+description sits in the prompt permanently (the
+// Catalog); the full body is loaded lazily when the model calls the single
+// skill_use tool (see Activate).
+//
+// Design shape (each piece is nil-safe / host-agnostic, mirroring the other
+// executus batteries):
+//
+//   - Manifest / ParseManifest — parse+validate a SKILL.md.
+//   - Tree / Pack / LoadPack    — a fetched pack's files, content digest, and
+//     parsed manifest.
+//   - Source (Dir, Git)         — where packs come from; Fetch returns the file
+//     tree and the source's resolved ref.
+//   - Subscription + Store      — the persisted "this host tracks this pack at
+//     this pinned digest" record; Memory is the zero-dep default.
+//   - PackCache                 — content-addressed store of pinned pack trees
+//     so activation never re-fetches; Memory default.
+//   - Syncer                    — checks the tracked ref and records a PENDING
+//     update; applying it is an explicit, separate re-pin (supply-chain guard —
+//     upstream can never silently change what an agent runs).
+//   - Catalog / Activate / Stage — turn a set of resolved packs into a
+//     majordomo agent.Skill (catalog instructions + skill_use tool) and
+//     materialize a pack's files for a sandbox.
+//
+// The host (e.g. mort) supplies policy: which sources are allowed, who may
+// subscribe, and where staged files are mounted. This package supplies only the
+// mechanism.
+package skillpack
+
+import "errors"
+
+// ErrNotFound is returned when a subscription or cached pack lookup misses.
+var ErrNotFound = errors.New("skillpack: not found")
+
+// ErrNoManifest is returned when a fetched tree has no SKILL.md at its root.
+var ErrNoManifest = errors.New("skillpack: tree has no SKILL.md")

skillpack/source.go

@@ -0,0 +1,149 @@
+package skillpack
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path"
+	"path/filepath"
+	"strings"
+)
+
+// Source is where a pack's files come from. Fetch retrieves the tree at ref and
+// returns it together with the source's own resolved ref (a git commit SHA, or
+// the content digest for a plain directory) — provenance a host can show and
+// pin against. ref semantics are source-specific and may be empty ("the
+// default": a dir's current contents, a repo's default branch).
+type Source interface {
+	Fetch(ctx context.Context, ref string) (Tree, string, error)
+	// Kind is a short stable tag ("dir", "git") for persistence + display.
+	Kind() string
+	// String is a human-readable identifier (path or URL[/subpath]).
+	String() string
+}
+
+// DirSource reads a pack from a local directory. ref is ignored (a directory
+// has no versions); the resolved ref is the content digest. Useful for
+// first-party/builtin packs shipped on disk and for tests.
+type DirSource struct {
+	Path string
+}
+
+func (d DirSource) Kind() string   { return "dir" }
+func (d DirSource) String() string { return d.Path }
+
+func (d DirSource) Fetch(_ context.Context, _ string) (Tree, string, error) {
+	info, err := os.Stat(d.Path)
+	if err != nil {
+		return nil, "", fmt.Errorf("skillpack: dir source %q: %w", d.Path, err)
+	}
+	if !info.IsDir() {
+		return nil, "", fmt.Errorf("skillpack: dir source %q is not a directory", d.Path)
+	}
+	t, err := readTree(os.DirFS(d.Path))
+	if err != nil {
+		return nil, "", err
+	}
+	return t, t.Digest(), nil
+}
+
+// GitSource fetches a pack from a git repository, optionally from a Subpath
+// within it (for repos that publish several packs). ref is any git commit-ish
+// (branch, tag, or SHA); empty means the default branch. The resolved ref is
+// the checked-out commit SHA.
+//
+// Fetch clones into a temp dir, reads the subpath tree into memory, and removes
+// the clone before returning — the returned Tree is self-contained, so there is
+// no clone lifetime to manage and nothing left on disk. Git runs via the system
+// `git`; GitRunner is overridable for tests.
+type GitSource struct {
+	URL     string
+	Subpath string
+	// GitRunner runs a git command in dir and returns combined output. Nil uses
+	// the system git.
+	GitRunner func(ctx context.Context, dir string, args ...string) ([]byte, error)
+}
+
+func (g GitSource) Kind() string { return "git" }
+
+func (g GitSource) String() string {
+	if g.Subpath != "" {
+		return g.URL + "//" + g.Subpath
+	}
+	return g.URL
+}
+
+func (g GitSource) run(ctx context.Context, dir string, args ...string) ([]byte, error) {
+	if g.GitRunner != nil {
+		return g.GitRunner(ctx, dir, args...)
+	}
+	cmd := exec.CommandContext(ctx, "git", args...)
+	cmd.Dir = dir
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return out, fmt.Errorf("skillpack: git %s: %w: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
+	}
+	return out, nil
+}
+
+func (g GitSource) Fetch(ctx context.Context, ref string) (Tree, string, error) {
+	// Argument-injection guard: a URL or ref beginning with "-" would be parsed
+	// by git as an option (e.g. --upload-pack=…), not a value. Reject it rather
+	// than rely solely on the "--" separator, which checkout does not honor for
+	// a rev. Hosts should also allow-list sources, but this is defense-in-depth
+	// for a library.
+	if strings.HasPrefix(g.URL, "-") {
+		return nil, "", fmt.Errorf("skillpack: git url must not start with '-': %q", g.URL)
+	}
+	if strings.HasPrefix(ref, "-") {
+		return nil, "", fmt.Errorf("skillpack: git ref must not start with '-': %q", ref)
+	}
+
+	tmp, err := os.MkdirTemp("", "skillpack-git-*")
+	if err != nil {
+		return nil, "", err
+	}
+	defer os.RemoveAll(tmp)
+
+	// --filter=blob:none: a blobless partial clone gets the ref graph cheaply
+	// and fetches only the blobs the checkout needs — much less than the full
+	// history, while still supporting an arbitrary commit-ish ref. "--" ends
+	// option parsing before the URL.
+	if _, err := g.run(ctx, "", "clone", "--quiet", "--filter=blob:none", "--", g.URL, tmp); err != nil {
+		return nil, "", err
+	}
+	if ref != "" {
+		if _, err := g.run(ctx, tmp, "checkout", "--quiet", "--detach", ref); err != nil {
+			return nil, "", err
+		}
+	}
+	shaOut, err := g.run(ctx, tmp, "rev-parse", "HEAD")
+	if err != nil {
+		return nil, "", err
+	}
+	sha := strings.TrimSpace(string(shaOut))
+
+	root := tmp
+	if g.Subpath != "" {
+		clean := path.Clean("/" + g.Subpath) // normalize, strip leading ../
+		root = filepath.Join(tmp, filepath.FromSlash(strings.TrimPrefix(clean, "/")))
+		if !within(tmp, root) {
+			return nil, "", fmt.Errorf("skillpack: subpath %q escapes the repo", g.Subpath)
+		}
+		if info, err := os.Stat(root); err != nil || !info.IsDir() {
+			return nil, "", fmt.Errorf("skillpack: subpath %q not found in %s", g.Subpath, g.URL)
+		}
+	}
+	t, err := readTree(os.DirFS(root))
+	if err != nil {
+		return nil, "", err
+	}
+	// Drop a nested .git if the subpath was the repo root.
+	for p := range t {
+		if p == ".git" || strings.HasPrefix(p, ".git/") {
+			delete(t, p)
+		}
+	}
+	return t, sha, nil
+}

skillpack/source_test.go

@@ -0,0 +1,97 @@
+package skillpack
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+)
+
+func writePack(t *testing.T, dir string) {
+	t.Helper()
+	if err := os.MkdirAll(filepath.Join(dir, "scripts"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, ManifestName), []byte(goodManifest), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, "scripts", "fill.py"), []byte("print('hi')\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestDirSource(t *testing.T) {
+	dir := t.TempDir()
+	writePack(t, dir)
+
+	tree, ref, err := DirSource{Path: dir}.Fetch(context.Background(), "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ref != tree.Digest() {
+		t.Errorf("dir resolved ref should be the content digest")
+	}
+	p, err := LoadPack(tree)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p.Manifest.Name != "pdf-processing" || len(p.Bundled) != 1 {
+		t.Errorf("loaded pack wrong: name=%q bundled=%v", p.Manifest.Name, p.Bundled)
+	}
+}
+
+func TestDirSource_NotADir(t *testing.T) {
+	f := filepath.Join(t.TempDir(), "file")
+	os.WriteFile(f, []byte("x"), 0o644)
+	if _, _, err := (DirSource{Path: f}).Fetch(context.Background(), ""); err == nil {
+		t.Fatal("expected error for non-directory source")
+	}
+}
+
+// TestGitSource drives a real local git repo (no network) to exercise clone +
+// checkout + subpath + SHA resolution. Skipped when git is unavailable.
+func TestGitSource(t *testing.T) {
+	if _, err := exec.LookPath("git"); err != nil {
+		t.Skip("git not installed")
+	}
+	repo := t.TempDir()
+	git := func(args ...string) {
+		t.Helper()
+		cmd := exec.Command("git", args...)
+		cmd.Dir = repo
+		cmd.Env = append(os.Environ(),
+			"GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@t")
+		if out, err := cmd.CombinedOutput(); err != nil {
+			t.Fatalf("git %v: %v: %s", args, err, out)
+		}
+	}
+	git("init", "-q", "-b", "main")
+	// pack lives under packs/pdf/
+	sub := filepath.Join(repo, "packs", "pdf")
+	writePack(t, sub)
+	git("add", "-A")
+	git("commit", "-q", "-m", "add pack")
+
+	src := GitSource{URL: repo, Subpath: "packs/pdf"}
+	tree, sha, err := src.Fetch(context.Background(), "main")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(sha) != 40 {
+		t.Errorf("resolved ref should be a full SHA, got %q", sha)
+	}
+	if _, ok := tree[ManifestName]; !ok {
+		t.Errorf("subpath tree missing SKILL.md; got %v", tree.Paths())
+	}
+	if _, ok := tree[".git"]; ok {
+		t.Error(".git leaked into the tree")
+	}
+	p, err := LoadPack(tree)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p.Manifest.Name != "pdf-processing" {
+		t.Errorf("name = %q", p.Manifest.Name)
+	}
+}

skillpack/store.go

@@ -0,0 +1,27 @@
+package skillpack
+
+import "context"
+
+// Store is the persistence seam for subscriptions (metadata + the current pin).
+// It is deliberately small; a host backs it with its DB, Memory is the zero-dep
+// default, and contrib/store can add durable SQLite alongside the other
+// executus store impls.
+type Store interface {
+	Initialize(ctx context.Context) error
+	Save(ctx context.Context, s *Subscription) error
+	Get(ctx context.Context, id string) (*Subscription, error)
+	GetByName(ctx context.Context, name string) (*Subscription, error)
+	List(ctx context.Context) ([]Subscription, error)
+	Delete(ctx context.Context, id string) error
+}
+
+// PackCache is the content-addressed store of pinned pack trees, keyed by
+// content digest. It exists so activating an agent never re-fetches from the
+// Source (no clone per run) and so a pinned digest's exact bytes survive even if
+// upstream later force-pushes or disappears. A host may back it with disk;
+// Memory is the default. Because the key IS the content digest, entries are
+// immutable and safe to share across subscriptions that pin the same bytes.
+type PackCache interface {
+	Put(ctx context.Context, digest string, t Tree) error
+	Get(ctx context.Context, digest string) (Tree, error)
+}

skillpack/subscription.go

@@ -0,0 +1,65 @@
+package skillpack
+
+import "time"
+
+// Subscription is a host's persisted "I track this pack, pinned here" record. It
+// is metadata only — the pinned pack's bytes live in a PackCache keyed by
+// PinnedDigest. A subscription is only ever advanced to new content by an
+// explicit Apply (see Syncer): a sync records a PendingDigest, it never moves
+// the pin. That is the supply-chain guard — a compromised or careless upstream
+// cannot change what an agent runs without a human re-pin.
+type Subscription struct {
+	// ID is a stable host-assigned identifier.
+	ID string
+	// Name is the pack's manifest name (unique per host); what an agent lists in
+	// its SkillPacks and what skill_use receives.
+	Name string
+	// Description is the pinned manifest's description, cached so the catalog
+	// renders without opening the PackCache.
+	Description string
+
+	// Source coordinates.
+	SourceKind string // "dir" | "git"
+	SourceURL  string // dir path or git URL
+	Subpath    string // git subpath, if any
+	// TrackRef is the git commit-ish the host follows (branch/tag/SHA); empty =
+	// default branch. Sync fetches THIS; the pin only moves on Apply.
+	TrackRef string
+
+	// Pinned* describe the currently-active content.
+	PinnedDigest    string // content digest = PackCache key + change signal
+	PinnedSourceRef string // source's resolved ref (git SHA) — provenance
+	PinnedAt        time.Time
+	PinnedBy        string
+
+	// Pending* describe an update a sync found but has NOT applied. Empty
+	// PendingDigest = no pending update. A pending digest equal to the pinned
+	// one is impossible by construction (Syncer clears it).
+	PendingDigest    string
+	PendingSourceRef string
+	PendingAt        time.Time
+
+	// Enabled lets a host keep a subscription but deactivate it without
+	// deleting the pin/history.
+	Enabled bool
+}
+
+// HasPending reports whether a sync found an unapplied update.
+func (s *Subscription) HasPending() bool {
+	return s.PendingDigest != "" && s.PendingDigest != s.PinnedDigest
+}
+
+// pinTo advances the active pin to a fetched pack and clears any pending state.
+// Used by initial pin and by Apply. It does NOT set Name: a subscription's name
+// is its stable host handle, fixed at Subscribe time — letting an upstream pack
+// rename move it would silently collide with another subscription on Apply.
+func (s *Subscription) pinTo(p *Pack, sourceRef, by string, now time.Time) {
+	s.Description = p.Manifest.Description
+	s.PinnedDigest = p.Digest
+	s.PinnedSourceRef = sourceRef
+	s.PinnedAt = now
+	s.PinnedBy = by
+	s.PendingDigest = ""
+	s.PendingSourceRef = ""
+	s.PendingAt = time.Time{}
+}

skillpack/sync.go

@@ -0,0 +1,203 @@
+package skillpack
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+// Syncer ties a Store, a PackCache, and Sources together into the subscription
+// lifecycle: subscribe (initial pin), check (record a PENDING update, never move
+// the pin), and apply (the explicit re-pin). It owns the supply-chain invariant
+// — the only call that changes the bytes an agent runs is Apply, always with an
+// actor recorded.
+type Syncer struct {
+	Cache PackCache // content store for pinned trees
+	Subs  Store     // subscription metadata store
+
+	// SourceFor builds the Source for a stored subscription. A host overrides
+	// this to enforce its allow-list (reject a disallowed URL/kind before any
+	// fetch). Nil uses DefaultSourceFor (dir + git, no allow-list).
+	SourceFor func(*Subscription) (Source, error)
+
+	// Now/NewID are injectable for deterministic tests.
+	Now   func() time.Time
+	NewID func() string
+}
+
+func (y *Syncer) now() time.Time {
+	if y.Now != nil {
+		return y.Now()
+	}
+	return time.Now()
+}
+
+func (y *Syncer) newID() string {
+	if y.NewID != nil {
+		return y.NewID()
+	}
+	return uuid.NewString()
+}
+
+func (y *Syncer) sourceFor(s *Subscription) (Source, error) {
+	if y.SourceFor != nil {
+		return y.SourceFor(s)
+	}
+	return DefaultSourceFor(s)
+}
+
+// DefaultSourceFor reconstructs a Source from a subscription's stored
+// coordinates, with no allow-list. A host that cares about which sources are
+// permitted should set Syncer.SourceFor instead of using this.
+func DefaultSourceFor(s *Subscription) (Source, error) {
+	switch s.SourceKind {
+	case "dir":
+		return DirSource{Path: s.SourceURL}, nil
+	case "git":
+		return GitSource{URL: s.SourceURL, Subpath: s.Subpath}, nil
+	default:
+		return nil, fmt.Errorf("skillpack: unknown source kind %q", s.SourceKind)
+	}
+}
+
+// fetchPack fetches src at ref, caches the resulting tree, and returns the
+// parsed pack plus the source's resolved ref.
+func (y *Syncer) fetchPack(ctx context.Context, src Source, ref string) (*Pack, string, error) {
+	tree, sourceRef, err := src.Fetch(ctx, ref)
+	if err != nil {
+		return nil, "", err
+	}
+	pack, err := LoadPack(tree)
+	if err != nil {
+		return nil, "", err
+	}
+	if err := y.Cache.Put(ctx, pack.Digest, pack.Tree); err != nil {
+		return nil, "", err
+	}
+	return pack, sourceRef, nil
+}
+
+// Subscribe fetches a pack from src at trackRef, caches it, and persists a new
+// Subscription pinned to that exact content, attributed to by. It rejects a
+// second subscription to the same pack name.
+func (y *Syncer) Subscribe(ctx context.Context, src Source, trackRef, by string) (*Subscription, error) {
+	pack, sourceRef, err := y.fetchPack(ctx, src, trackRef)
+	if err != nil {
+		return nil, err
+	}
+	existing, err := y.Subs.GetByName(ctx, pack.Manifest.Name)
+	if err == nil {
+		return nil, fmt.Errorf("skillpack: already subscribed to %q (id %s)", pack.Manifest.Name, existing.ID)
+	}
+	if !errors.Is(err, ErrNotFound) {
+		// A transient store error must NOT fall through to creating a row — that
+		// would produce a duplicate subscription the uniqueness check missed.
+		return nil, fmt.Errorf("skillpack: checking for existing subscription %q: %w", pack.Manifest.Name, err)
+	}
+
+	sub := &Subscription{
+		ID:         y.newID(),
+		Name:       pack.Manifest.Name,
+		SourceKind: src.Kind(),
+		SourceURL:  src.String(),
+		TrackRef:   trackRef,
+		Enabled:    true,
+	}
+	// Store the raw URL + subpath separately (String() may combine them for
+	// display). GitSource methods have value receivers, so a caller may pass
+	// either GitSource or *GitSource — handle both.
+	switch gs := src.(type) {
+	case GitSource:
+		sub.SourceURL, sub.Subpath = gs.URL, gs.Subpath
+	case *GitSource:
+		sub.SourceURL, sub.Subpath = gs.URL, gs.Subpath
+	}
+	sub.pinTo(pack, sourceRef, by, y.now())
+	if err := y.Subs.Save(ctx, sub); err != nil {
+		return nil, err
+	}
+	return sub, nil
+}
+
+// Check fetches the subscription's tracked ref and, if the content digest
+// differs from the current pin, caches the new tree and records it as PENDING —
+// it never moves the pin. If the tracked ref matches the pin, any stale pending
+// state is cleared. The updated subscription is saved and returned.
+func (y *Syncer) Check(ctx context.Context, id string) (*Subscription, error) {
+	sub, err := y.Subs.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	src, err := y.sourceFor(sub)
+	if err != nil {
+		return nil, err
+	}
+	pack, sourceRef, err := y.fetchPack(ctx, src, sub.TrackRef)
+	if err != nil {
+		return nil, err
+	}
+	if pack.Digest == sub.PinnedDigest {
+		// No change upstream; drop any previously-recorded pending update.
+		sub.PendingDigest, sub.PendingSourceRef, sub.PendingAt = "", "", time.Time{}
+	} else {
+		sub.PendingDigest = pack.Digest
+		sub.PendingSourceRef = sourceRef
+		sub.PendingAt = y.now()
+	}
+	if err := y.Subs.Save(ctx, sub); err != nil {
+		return nil, err
+	}
+	return sub, nil
+}
+
+// CheckAll runs Check on every subscription and returns the ones that now have a
+// pending update. Errors on individual subscriptions are collected, not fatal —
+// one unreachable source shouldn't stop the sweep. A host calls this on its own
+// ticker (skillpack has no cron opinion; the update is never auto-applied so the
+// cadence only affects how fresh the "pending" signal is).
+func (y *Syncer) CheckAll(ctx context.Context) (pending []Subscription, errs []error) {
+	subs, err := y.Subs.List(ctx)
+	if err != nil {
+		return nil, []error{err}
+	}
+	for i := range subs {
+		updated, err := y.Check(ctx, subs[i].ID)
+		if err != nil {
+			errs = append(errs, fmt.Errorf("skillpack: check %q: %w", subs[i].Name, err))
+			continue
+		}
+		if updated.HasPending() {
+			pending = append(pending, *updated)
+		}
+	}
+	return pending, errs
+}
+
+// Apply promotes a subscription's pending update to the active pin, attributed
+// to by. This is the ONLY call that changes what an agent runs. It errors if
+// there is no pending update or the pending tree is missing from the cache.
+func (y *Syncer) Apply(ctx context.Context, id, by string) (*Subscription, error) {
+	sub, err := y.Subs.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	if !sub.HasPending() {
+		return nil, fmt.Errorf("skillpack: %q has no pending update to apply", sub.Name)
+	}
+	tree, err := y.Cache.Get(ctx, sub.PendingDigest)
+	if err != nil {
+		return nil, fmt.Errorf("skillpack: pending tree for %q missing from cache: %w", sub.Name, err)
+	}
+	pack, err := LoadPack(tree)
+	if err != nil {
+		return nil, err
+	}
+	sub.pinTo(pack, sub.PendingSourceRef, by, y.now())
+	if err := y.Subs.Save(ctx, sub); err != nil {
+		return nil, err
+	}
+	return sub, nil
+}

skillpack/sync_test.go

@@ -0,0 +1,177 @@
+package skillpack
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+)
+
+// fakeSource returns a caller-controlled tree, so sync behavior is tested with
+// no filesystem or git.
+type fakeSource struct {
+	tree Tree
+	ref  string
+	err  error
+}
+
+func (f *fakeSource) Fetch(context.Context, string) (Tree, string, error) {
+	return f.tree, f.ref, f.err
+}
+func (f *fakeSource) Kind() string   { return "fake" }
+func (f *fakeSource) String() string { return "fake://pack" }
+
+func packTree(name, body string) Tree {
+	return Tree{ManifestName: []byte("---\nname: " + name + "\ndescription: does " + name + "\n---\n" + body + "\n")}
+}
+
+func newTestSyncer(src *fakeSource) *Syncer {
+	n := 0
+	return &Syncer{
+		Cache:     NewMemoryPackCache(),
+		Subs:      NewMemory(),
+		Now:       func() time.Time { return time.Unix(1000, 0) },
+		NewID:     func() string { n++; return fmt.Sprintf("id-%d", n) },
+		SourceFor: func(*Subscription) (Source, error) { return src, nil },
+	}
+}
+
+func TestSubscribeAndPin(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "sha-v1"}
+	y := newTestSyncer(src)
+
+	sub, err := y.Subscribe(ctx, src, "main", "steve")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if sub.Name != "alpha" || sub.PinnedSourceRef != "sha-v1" || sub.PinnedBy != "steve" {
+		t.Fatalf("bad pin: %+v", sub)
+	}
+	if sub.HasPending() {
+		t.Fatal("fresh subscription should have no pending update")
+	}
+	// pinned tree is cached under its digest
+	if _, err := y.Cache.Get(ctx, sub.PinnedDigest); err != nil {
+		t.Fatalf("pinned tree not cached: %v", err)
+	}
+}
+
+func TestSubscribe_DuplicateName(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "r"}
+	y := newTestSyncer(src)
+	if _, err := y.Subscribe(ctx, src, "", "s"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := y.Subscribe(ctx, src, "", "s"); err == nil {
+		t.Fatal("expected duplicate-name error")
+	}
+}
+
+func TestCheck_RecordsPendingButDoesNotMovePin(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "sha-v1"}
+	y := newTestSyncer(src)
+	sub, _ := y.Subscribe(ctx, src, "main", "s")
+	pinnedBefore := sub.PinnedDigest
+
+	// upstream changes
+	src.tree = packTree("alpha", "v2-new-instructions")
+	src.ref = "sha-v2"
+
+	updated, err := y.Check(ctx, sub.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !updated.HasPending() {
+		t.Fatal("expected a pending update after upstream change")
+	}
+	if updated.PinnedDigest != pinnedBefore {
+		t.Fatal("Check must NOT move the pin — that is the supply-chain guard")
+	}
+	if updated.PendingSourceRef != "sha-v2" {
+		t.Errorf("pending ref = %q", updated.PendingSourceRef)
+	}
+	// the pending tree is cached, ready for Apply
+	if _, err := y.Cache.Get(ctx, updated.PendingDigest); err != nil {
+		t.Fatalf("pending tree not cached: %v", err)
+	}
+}
+
+func TestCheck_ClearsStalePendingWhenUpstreamMatches(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "r1"}
+	y := newTestSyncer(src)
+	sub, _ := y.Subscribe(ctx, src, "main", "s")
+
+	src.tree = packTree("alpha", "v2")
+	src.ref = "r2"
+	sub, _ = y.Check(ctx, sub.ID) // records pending
+	if !sub.HasPending() {
+		t.Fatal("precondition: pending expected")
+	}
+	// upstream reverts to the pinned content
+	src.tree = packTree("alpha", "v1")
+	src.ref = "r1"
+	sub, _ = y.Check(ctx, sub.ID)
+	if sub.HasPending() {
+		t.Fatal("pending should be cleared once upstream matches the pin again")
+	}
+}
+
+func TestApply_MovesPinAndClearsPending(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "sha-v1"}
+	y := newTestSyncer(src)
+	sub, _ := y.Subscribe(ctx, src, "main", "s")
+
+	src.tree = packTree("alpha", "v2")
+	src.ref = "sha-v2"
+	sub, _ = y.Check(ctx, sub.ID)
+	pendingDigest := sub.PendingDigest
+
+	applied, err := y.Apply(ctx, sub.ID, "admin")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if applied.PinnedDigest != pendingDigest {
+		t.Fatal("Apply must move the pin to the pending digest")
+	}
+	if applied.PinnedSourceRef != "sha-v2" || applied.PinnedBy != "admin" {
+		t.Errorf("bad post-apply pin: %+v", applied)
+	}
+	if applied.HasPending() {
+		t.Fatal("Apply must clear the pending update")
+	}
+}
+
+func TestApply_NoPending(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "r"}
+	y := newTestSyncer(src)
+	sub, _ := y.Subscribe(ctx, src, "", "s")
+	if _, err := y.Apply(ctx, sub.ID, "admin"); err == nil {
+		t.Fatal("expected error applying with no pending update")
+	}
+}
+
+func TestCheckAll(t *testing.T) {
+	ctx := context.Background()
+	src := &fakeSource{tree: packTree("alpha", "v1"), ref: "r1"}
+	y := newTestSyncer(src)
+	sub, _ := y.Subscribe(ctx, src, "main", "s")
+
+	if pend, errs := y.CheckAll(ctx); len(pend) != 0 || len(errs) != 0 {
+		t.Fatalf("no change: pend=%v errs=%v", pend, errs)
+	}
+	src.tree = packTree("alpha", "v2")
+	src.ref = "r2"
+	pend, errs := y.CheckAll(ctx)
+	if len(errs) != 0 {
+		t.Fatalf("errs: %v", errs)
+	}
+	if len(pend) != 1 || pend[0].ID != sub.ID {
+		t.Fatalf("expected 1 pending, got %v", pend)
+	}
+}