1 Commits

Author SHA1 Message Date
Steve Dudenhoeffer d5a24c6f6e ci: inherit gadfly's default swarm (slim caller, re-pin @b02b11d)
executus CI / test (pull_request) Successful in 1m50s
steve/gadfly#10 moved the curated swarm (3 cloud + Claude Code sonnet/opus/
opus:max, 5-lens suite) into the reusable workflow's input defaults. Drop this
repo's explicit `with:` swarm block and inherit it — only the consumer-specific
allow-list remains. Re-pin to the post-#10 gadfly commit (@b02b11d).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 22:44:48 -04:00
33 changed files with 68 additions and 3785 deletions
+3 -5
View File
@@ -38,11 +38,9 @@ jobs:
&& (github.actor == 'steve'
|| github.actor == 'fizi'
|| github.actor == 'dazed'))
# Pinned to an immutable gadfly commit (not @v1): our act_runners are long-lived
# and cache the reusable-workflow ref, so a moved v1 tag keeps resolving to the
# stale cached copy. A unique sha forces a cache miss → fresh fetch. Bump this
# sha to adopt central swarm changes.
uses: steve/gadfly/.gitea/workflows/review-reusable.yml@5007597cf921dc3f0a83c708878facfe65fd8e8b
# Pinned to an immutable gadfly commit (not @main): a push to gadfly can't
# silently change the code that runs with our forwarded secrets.
uses: steve/gadfly/.gitea/workflows/review-reusable.yml@b02b11d69139843665da4cdbf776bc0b3583490d
# Least privilege: forward only the review secrets (not `secrets: inherit`,
# which would expose every repo secret). GITEA_TOKEN is the automatic token.
secrets:
-8
View File
@@ -85,14 +85,6 @@ 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 +
+1 -1
View File
@@ -37,7 +37,7 @@ bot) — mort and gadfly are the first two consumers (heavy and light). See
tool registry, majordomo's agent loop, context compaction, run-bounding, and
step/audit instrumentation into one `Run(ctx, RunnableAgent, inv) Result`, with
every host concern behind a nil-safe `run.Ports` (Audit/Budget/Critic/
Checkpointer/PaletteSource/Delivery/InputFiles). See `examples/minimal`.
Checkpointer/PaletteSource/Delivery). See `examples/minimal`.
- `model/` — config-driven tier resolution + failover over majordomo, with
pluggable `UsageSink`/`TraceSink` and `GenerateWith[T]` structured output.
- `tool/` — the tool registry + 3-stage permission model + SSRF guard.
+8 -10
View File
@@ -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. The executor calls
// run.Ports.Checkpointer (a CheckpointerFactory) during the run loop; NewFactory
// wires this battery into that seam.
// zero-dependency default; contrib/store can add a SQLite one. NOTE: the
// executor's call into run.Ports.Checkpointer is a P2 follow-up — this battery
// provides the seam + impls ahead of that wiring.
package checkpoint
import (
@@ -14,8 +14,6 @@ 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
@@ -34,11 +32,11 @@ 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 (single-loop runs)
Iteration int // completed agent-loop iterations
CompletedPhases []run.PhaseOutput // finished phases, in order (multi-phase agents)
UpdatedAt time.Time
Meta RunCheckpointMeta
Messages []llm.Message // conversation so far
Iteration int // completed agent-loop iterations
ActivePhase string // current phase name (multi-phase agents); "" otherwise
UpdatedAt time.Time
}
// CheckpointStore persists run checkpoints keyed by run id. A live checkpoint
+4 -42
View File
@@ -54,11 +54,10 @@ func (h *handle) Save(ctx context.Context, st run.RunCheckpointState) error {
// caller believes was saved. (A run drives one Save goroutine, so the brief
// unguarded window here can't double-write.)
if err := h.store.Save(ctx, RunCheckpoint{
Meta: h.meta,
Messages: st.Messages,
Iteration: st.Iteration,
CompletedPhases: st.CompletedPhases,
UpdatedAt: now,
Meta: h.meta,
Messages: st.Messages,
Iteration: st.Iteration,
UpdatedAt: now,
}); err != nil {
return err
}
@@ -82,40 +81,3 @@ 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
}
+3 -15
View File
@@ -55,27 +55,15 @@ type RunnableAgent struct {
}
// Phase is one step of a multi-step run: its own system prompt, model tier,
// iteration cap, and tool subset. Phase prompts are Go text/template strings
// expanded against {{.Query}} (the original input) and {{.<PhaseName>}} (a
// prior phase's output) before the phase runs, so a phase can consume earlier
// work. The final phase's output is the run's output.
// iteration cap, and tool subset. Optional phases may be skipped by the
// pipeline when their precondition isn't met.
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
Optional bool
}
// CriticConfig configures the optional run-critic. Enabled gates whether a
-103
View File
@@ -1,103 +0,0 @@
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.
}
}
-200
View File
@@ -1,200 +0,0 @@
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)
}
}
+7 -19
View File
@@ -22,14 +22,6 @@ 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
@@ -39,20 +31,16 @@ func (e *Executor) criticOwnsDeadline(ra RunnableAgent) bool {
// "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.
//
// 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()) {
func (e *Executor) startCritic(runCtx context.Context, cancelCause context.CancelCauseFunc, ra RunnableAgent, info RunInfo) (*criticBinding, func()) {
noop := func() {}
if !e.criticOwnsDeadline(ra) {
if e.cfg.Ports.Critic == nil || !ra.Critic.Enabled {
return nil, noop
}
h := e.cfg.Ports.Critic.Monitor(runCtx, info, softTrigger)
soft := e.cfg.Defaults.CriticSoftTimeout
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
}
-153
View File
@@ -1,153 +0,0 @@
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)
}
}
+2 -2
View File
@@ -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}},
// The fake handle's Deadline() is zero (no hard deadline), so the
// deadline-watch never interferes regardless of the soft trigger.
// large soft timeout so the deadline-watch never interferes in the test
Defaults: run.Defaults{CriticSoftTimeout: time.Hour},
})
res := ex.Run(context.Background(),
run.RunnableAgent{Name: "x", ModelTier: "m", MaxIterations: 1, Critic: run.CriticConfig{Enabled: true}},
+32 -215
View File
@@ -4,7 +4,6 @@ import (
"context"
"errors"
"fmt"
"log/slog"
"time"
"gitea.stevedudenhoeffer.com/steve/majordomo/agent"
@@ -29,17 +28,7 @@ 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
// 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
CriticSoftTimeout time.Duration // idle window before the critic wakes; default 90s
}
func (d Defaults) withFallbacks() Defaults {
@@ -61,8 +50,8 @@ func (d Defaults) withFallbacks() Defaults {
if d.CompactionThresholdRatio <= 0 {
d.CompactionThresholdRatio = 0.7
}
if d.CriticAbsoluteMax <= 0 {
d.CriticAbsoluteMax = 24 * time.Hour
if d.CriticSoftTimeout <= 0 {
d.CriticSoftTimeout = 90 * time.Second
}
return d
}
@@ -124,26 +113,13 @@ 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. 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).
// rather than unwinding into the caller.
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
@@ -189,9 +165,7 @@ 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,
@@ -206,25 +180,6 @@ 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.
@@ -275,40 +230,18 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
postRun = st.PostRun
}
// Run context: detached from the caller's deadline so a lane/queue wait doesn't
// eat the run budget (mort's V10 lesson). Caller cancellation still propagates
// via MergeCancellation. Created BEFORE the step observer so the observer
// forwards the merged run context (not a possibly-cancelled caller ctx) to
// OnStep consumers.
//
// Two-tier timeout: who owns the hard deadline depends on the critic.
// - NO critic (the default): MaxRuntime is a literal WithTimeout. Its
// DeadlineExceeded propagates through the child chain (→ "timeout"),
// preserving the run's-own-timeout vs caller-cancel distinction.
// - critic OWNS the deadline (Ports.Critic set + ra.Critic.Enabled):
// 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)
// Run context: bound by MaxRuntime, detached from the caller's deadline so a
// lane/queue wait doesn't eat the run budget (mort's V10 lesson). Caller
// cancellation still propagates via MergeCancellation. Created BEFORE the
// step observer so the observer forwards the merged run context (not a
// possibly-cancelled caller ctx) to OnStep consumers.
// MaxRuntime stays a WithTimeout so its DeadlineExceeded propagates through the
// child chain (→ "timeout"), preserving the run's-own-timeout vs caller-cancel
// distinction. A NESTED cause-carrying layer lets a critic kill surface as a
// distinct "killed" without disturbing that: only an ErrCriticKill cause is
// consulted in statusFor; a generic run error or a caller cancel is classified
// by the run error itself.
timeoutCtx, cancelTimeout := context.WithTimeout(context.WithoutCancel(ctx), maxRuntime)
defer cancelTimeout()
runCtx, cancelCause := context.WithCancelCause(timeoutCtx)
defer cancelCause(nil)
@@ -316,29 +249,11 @@ 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. When it owns the deadline, MaxRuntime is its soft trigger
// (so a slow-but-progressing run survives past it); its extendable backstop is
// 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)
// its host Escalator. Its hard deadline is 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)
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
@@ -373,12 +288,14 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
}
}
// Shared agent options used by BOTH the single-loop path and every phase: the
// tool-error guards and optional compaction. The toolbox, step ceiling, AND
// step observer are added per path (the observer is wrapped for checkpointing,
// which differs single-loop vs per-phase).
sharedOpts := []agent.Option{
opts := []agent.Option{
agent.WithToolbox(toolbox),
// Step ceiling: a fixed WithMaxSteps(maxIter) normally, but when a critic is
// active it owns a DYNAMIC ceiling (WithMaxStepsFunc) so it can raise a
// healthy-but-long run's budget mid-flight. Falls back to maxIter.
critic.maxStepsOption(maxIter),
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 {
@@ -395,89 +312,15 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
})
}
}
sharedOpts = append(sharedOpts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire)))
opts = append(opts, agent.WithCompactor(e.cfg.Compactor(threshold, onFire)))
}
}
// Stage non-image input attachments (audio/PDF/binary) into the host file
// store and fold an [ATTACHED FILES] descriptor into the prompt so the agent
// can reach them by file_id. No-op when Ports.InputFiles is nil or there are
// no files. Done after the model/toolbox build but before the loop, so the
// descriptor rides the very first user turn.
input = e.stageInputFiles(runCtx, inv.RunID, ra.ID, inv.InputFiles, input)
ag := agent.New(model, e.systemPrompt(ra), opts...)
// 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()...) }
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.
runRes, runErr := ag.Run(runCtx, input, agent.WithSteer(steer))
status := statusFor(runCtx, runErr)
if runRes != nil {
@@ -553,20 +396,13 @@ func (e *Executor) finishAudit(ctx context.Context, rec RunRecorder, status stri
}
func (e *Executor) systemPrompt(ra RunnableAgent) string {
return e.systemPromptWithBody(ra.SystemPrompt)
}
// systemPromptWithBody composes the optional platform header with an arbitrary
// body. The single-loop path passes ra.SystemPrompt; the phase runner passes a
// phase's expanded instructions, so each phase keeps the platform header.
func (e *Executor) systemPromptWithBody(body string) string {
if e.cfg.SystemHeader == "" {
return body
return ra.SystemPrompt
}
if body == "" {
if ra.SystemPrompt == "" {
return e.cfg.SystemHeader
}
return e.cfg.SystemHeader + "\n\n" + body
return e.cfg.SystemHeader + "\n\n" + ra.SystemPrompt
}
// compactionThreshold returns the token threshold for the tier's model context
@@ -604,22 +440,3 @@ func detach(ctx context.Context) context.Context {
_ = cancel // bounded by the timeout; nothing to cancel early
return c
}
// runAgent dispatches the majordomo agent loop. majordomo's Run takes a text-only
// input arg, so when the invocation carries images they're folded into the first
// user message (text + image parts) via WithHistory and Run is called with an
// empty input — the model then sees a multimodal opening turn. The image-less path
// passes the prompt straight through.
//
// The text part is omitted when input is blank (image-only run), matching
// runSession.AttachImages so no empty TextPart is sent.
func runAgent(ctx context.Context, ag *agent.Agent, input string, images []llm.ImagePart, opts ...agent.RunOption) (*agent.Result, error) {
if len(images) == 0 {
return ag.Run(ctx, input, opts...)
}
// Copy opts before appending so a caller-supplied backing array is never
// mutated/aliased (the variadic slice can have spare capacity). The multimodal
// opening turn (text + image parts) is built by the shared helper.
opts = append(opts[:len(opts):len(opts)], agent.WithHistory([]llm.Message{multimodalUserMessage(input, images)}))
return ag.Run(ctx, "", opts...)
}
-121
View File
@@ -1,121 +0,0 @@
package run_test
import (
"context"
"strings"
"testing"
"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"
)
// TestExecutorFoldsInitialImages: when the invocation carries Images, they're
// folded into the first user message (alongside the prompt text) instead of being
// dropped — majordomo's Run input arg is text-only, so the executor seeds the
// multimodal opening turn via history.
func TestExecutorFoldsInitialImages(t *testing.T) {
fp := fake.New("fake")
fp.Enqueue("m", fake.Reply("saw the image"))
m, _ := fp.Model("m")
img := llm.ImagePart{MIME: "image/png", Data: []byte("PNGDATA")}
inv := tool.Invocation{RunID: "r1", Images: []llm.ImagePart{img}}
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{ModelTier: "m"}, inv, "describe this")
if res.Err != nil {
t.Fatalf("run error: %v", res.Err)
}
calls := fp.Calls()
if len(calls) == 0 {
t.Fatal("no model calls recorded")
}
// The text + image must be CO-LOCATED in a single user message (not split
// across two), so the model reads them as one multimodal turn.
coLocated := false
for _, msg := range calls[0].Request.Messages {
sawImage, sawText := false, false
for _, p := range msg.Parts {
switch pp := p.(type) {
case llm.ImagePart:
if string(pp.Data) == "PNGDATA" {
sawImage = true
}
case llm.TextPart:
if strings.Contains(pp.Text, "describe this") {
sawText = true
}
}
}
if sawImage && sawText {
coLocated = true
}
}
if !coLocated {
t.Error("image + prompt text were not folded into the SAME user message")
}
}
// TestExecutorImageOnlyNoBlankText: an image-only run (blank prompt) must NOT emit
// an empty TextPart — the message carries just the image, matching
// runSession.AttachImages's guard.
func TestExecutorImageOnlyNoBlankText(t *testing.T) {
fp := fake.New("fake")
fp.Enqueue("m", fake.Reply("saw it"))
m, _ := fp.Model("m")
inv := tool.Invocation{RunID: "r3", Images: []llm.ImagePart{{MIME: "image/png", Data: []byte("IMG")}}}
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{ModelTier: "m"}, inv, " ")
if res.Err != nil {
t.Fatalf("run error: %v", res.Err)
}
for _, msg := range fp.Calls()[0].Request.Messages {
for _, p := range msg.Parts {
if tp, ok := p.(llm.TextPart); ok && strings.TrimSpace(tp.Text) == "" {
t.Error("image-only run emitted a blank TextPart")
}
}
}
}
// TestExecutorTextOnlyUnchanged: with no Images, the prompt flows through as the
// text input (regression guard that the fold path didn't break the common case).
func TestExecutorTextOnlyUnchanged(t *testing.T) {
fp := fake.New("fake")
fp.Enqueue("m", fake.Reply("ok"))
m, _ := fp.Model("m")
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{ModelTier: "m"}, tool.Invocation{RunID: "r2"}, "plain prompt")
if res.Err != nil {
t.Fatalf("run error: %v", res.Err)
}
calls := fp.Calls()
if len(calls) == 0 {
t.Fatal("no model calls recorded")
}
sawText := false
for _, msg := range calls[0].Request.Messages {
for _, p := range msg.Parts {
if tp, ok := p.(llm.TextPart); ok && strings.Contains(tp.Text, "plain prompt") {
sawText = true
}
}
}
if !sawText {
t.Error("text-only prompt did not reach the model")
}
}
-179
View File
@@ -1,179 +0,0 @@
package run
import (
"context"
"fmt"
"log/slog"
"path"
"strings"
"unicode"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// maxInputFileBytes is a defense-in-depth cap at the staging boundary. A host's
// extraction path may already cap downloads, but stageInputFiles is the trust
// boundary for the InputFiles seam: a call site or bug that populates InputFiles
// directly must not write an unbounded blob to the host file store.
const maxInputFileBytes = 50_000_000
// maxInputFiles bounds how many attachments a single run stages, independent of
// the per-file byte cap — defense-in-depth against a flood of tiny files.
const maxInputFiles = 32
// stageInputFiles persists each non-image input attachment into the host file
// store (Ports.InputFiles) under run scope and appends a descriptor block to the
// prompt so the agent knows the file_ids it can pass to a worker tool. The bytes
// are NOT inlined into the model context — the LLM can't read raw audio/binary —
// so the agent reaches them via a file_id-aware tool (e.g. code_exec files_in,
// which writes the file to /workspace/<name>).
//
// Best-effort: a nil stager, no files, or a per-file save error degrades to
// "skip that file" — the run still proceeds. Returns the (possibly augmented)
// prompt.
func (e *Executor) stageInputFiles(ctx context.Context, runID, agentID string, files []tool.InputFile, prompt string) string {
if e.cfg.Ports.InputFiles == nil || len(files) == 0 {
return prompt
}
// Count cap: bound how many attachments one run can stage, independent of the
// per-file byte cap (defense-in-depth against a flood of tiny files).
if len(files) > maxInputFiles {
slog.Warn("run: too many input files, truncating",
"agent", agentID, "run_id", runID, "count", len(files), "cap", maxInputFiles)
files = files[:maxInputFiles]
}
type stagedFile struct {
name, mime, fileID string
size int
}
var staged []stagedFile
seenNames := make(map[string]int, len(files))
for _, f := range files {
if len(f.Data) == 0 {
slog.Warn("run: skipping empty input file",
"agent", agentID, "run_id", runID, "name", f.Name)
continue
}
if len(f.Data) > maxInputFileBytes {
slog.Warn("run: skipping oversized input file",
"agent", agentID, "run_id", runID, "name", f.Name,
"size", len(f.Data), "cap", maxInputFileBytes)
continue
}
// Reduce the untrusted filename to a safe base name BEFORE staging or
// inlining: strips ../ and absolute-path components (so it can't escape
// the host store or /workspace/<name>) and drops control chars/newlines
// (so a crafted name can't inject text into the descriptor block below).
// Then disambiguate colliding base names so two attachments don't both map
// to /workspace/<name> (the second would clobber the first).
name := uniqueName(sanitizeName(f.Name), seenNames)
// Sanitize the mime ONCE and pass the clean value to both the host store
// and the descriptor (don't hand the raw value to StageInputFile).
mime := sanitizeField(f.MimeType)
fileID, err := e.cfg.Ports.InputFiles.StageInputFile(ctx, runID, agentID, name, mime, f.Data)
if err != nil {
slog.Warn("run: failed to stage input file",
"agent", agentID, "run_id", runID, "name", name, "error", err)
continue
}
if fileID == "" {
slog.Warn("run: stager returned empty file_id, skipping",
"agent", agentID, "run_id", runID, "name", name)
continue
}
// fileID is host-generated, but sanitize it too before inlining — the
// descriptor must never carry control chars no matter the stager impl.
staged = append(staged, stagedFile{name: name, mime: mime, fileID: sanitizeField(fileID), size: len(f.Data)})
}
if len(staged) == 0 {
return prompt
}
var b strings.Builder
b.WriteString("[ATTACHED FILES]\n")
b.WriteString("The user attached the following file(s). Their contents are NOT included in this prompt and you cannot read them directly. ")
b.WriteString("To work with one, call the code_exec tool with a files_in entry — e.g. ")
b.WriteString(`files_in: [{"name": "<name>", "file_id": "<file_id>"}]`)
b.WriteString(" — which writes it to /workspace/<name> inside the Python sandbox. You may also pass a file_id to any other tool that accepts one.\n")
for _, s := range staged {
fmt.Fprintf(&b, "- %s (%s, %s) → file_id: %s\n", s.name, s.mime, humanizeBytes(s.size), s.fileID)
}
if strings.TrimSpace(prompt) == "" {
return b.String()
}
return prompt + "\n\n" + b.String()
}
// sanitizeName reduces an untrusted attachment filename to a safe base name. It
// drops control characters / newlines (which would otherwise let a crafted name
// inject text into the [ATTACHED FILES] descriptor) and strips every directory
// component — defeating ../ traversal, nested dirs, and absolute / drive paths
// both in the host file store and at /workspace/<name>. Returns "attachment"
// when nothing usable remains (empty, ".", "..").
func sanitizeName(name string) string {
name = sanitizeField(name)
// Normalize backslashes so a Windows-style path also reduces to its base.
base := path.Base(strings.ReplaceAll(name, `\`, "/"))
base = strings.TrimSpace(base)
if base == "" || base == "." || base == ".." {
return "attachment"
}
return base
}
// sanitizeField strips characters that could let a value inlined verbatim into
// the prompt descriptor break out of its line or visually mislead: control
// characters (IsControl covers newlines/tabs) AND Unicode format characters
// (category Cf — e.g. the bidi overrides U+202AU+202E, which can reorder how
// the descriptor renders).
func sanitizeField(s string) string {
return strings.Map(func(r rune) rune {
if unicode.IsControl(r) || unicode.Is(unicode.Cf, r) {
return -1
}
return r
}, s)
}
// uniqueName returns name unchanged the first time it's seen, then name-2,
// name-3, … (suffix inserted before the extension) on repeats, recording each
// result in seen so later collisions keep counting up.
func uniqueName(name string, seen map[string]int) string {
if seen[name] == 0 {
seen[name]++
return name
}
ext := path.Ext(name)
base := strings.TrimSuffix(name, ext)
for {
seen[name]++
candidate := fmt.Sprintf("%s-%d%s", base, seen[name], ext)
if seen[candidate] == 0 {
seen[candidate]++
return candidate
}
}
}
// humanizeBytes renders a byte count as a short human-readable string (e.g.
// "2.1 MB") for the attached-files descriptor block.
func humanizeBytes(n int) string {
if n < 0 {
n = 0
}
const unit = 1024
if n < unit {
return fmt.Sprintf("%d B", n)
}
const prefixes = "KMGTPE"
div, exp := int64(unit), 0
// Clamp exp to the last prefix so an absurd size (≥1024^7) can't index past
// "KMGTPE" and panic — a no-panic guarantee independent of the per-file cap.
for v := int64(n) / unit; v >= unit && exp < len(prefixes)-1; v /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.1f %cB", float64(n)/float64(div), prefixes[exp])
}
-243
View File
@@ -1,243 +0,0 @@
package run
import (
"context"
"errors"
"strings"
"testing"
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
"gitea.stevedudenhoeffer.com/steve/executus/tool"
)
// stagerFunc is a test InputFileStager: it records each staged file and returns
// a deterministic file_id ("file_<name>"), or an error if err is set.
type stagerFunc struct {
staged []stagedRec
err error
}
type stagedRec struct {
runID, agentID, name, mime string
size int
}
func (s *stagerFunc) StageInputFile(_ context.Context, runID, agentID, name, mime string, content []byte) (string, error) {
if s.err != nil {
return "", s.err
}
s.staged = append(s.staged, stagedRec{runID, agentID, name, mime, len(content)})
return "file_" + name, nil
}
func newStagerExecutor(s InputFileStager) *Executor {
return New(Config{
Registry: tool.NewRegistry(),
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, nil, nil },
Ports: Ports{InputFiles: s},
})
}
// TestStageInputFiles: files are staged via the port and an [ATTACHED FILES]
// descriptor (with each file_id) is appended to the prompt.
func TestStageInputFiles(t *testing.T) {
st := &stagerFunc{}
ex := newStagerExecutor(st)
out := ex.stageInputFiles(context.Background(), "run-1", "agent-1",
[]tool.InputFile{{Name: "clip.mp3", MimeType: "audio/mpeg", Data: []byte("abcd")}},
"transcribe this")
if len(st.staged) != 1 || st.staged[0].name != "clip.mp3" {
t.Fatalf("staged = %+v, want one clip.mp3", st.staged)
}
if st.staged[0].runID != "run-1" || st.staged[0].agentID != "agent-1" {
t.Errorf("stager got runID/agentID = %q/%q, want run-1/agent-1", st.staged[0].runID, st.staged[0].agentID)
}
for _, want := range []string{"transcribe this", "[ATTACHED FILES]", "clip.mp3", "file_clip.mp3", "audio/mpeg"} {
if !strings.Contains(out, want) {
t.Errorf("output missing %q:\n%s", want, out)
}
}
}
// TestStageInputFilesNoStager: a nil port leaves the prompt untouched and never
// drops the run.
func TestStageInputFilesNoStager(t *testing.T) {
ex := newStagerExecutor(nil) // Ports.InputFiles == nil
out := ex.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "prompt")
if out != "prompt" {
t.Errorf("nil stager changed the prompt: %q", out)
}
}
// TestStageInputFilesNoFiles: no attachments leaves the prompt untouched.
func TestStageInputFilesNoFiles(t *testing.T) {
ex := newStagerExecutor(&stagerFunc{})
out := ex.stageInputFiles(context.Background(), "r", "a", nil, "prompt")
if out != "prompt" {
t.Errorf("no files changed the prompt: %q", out)
}
}
// TestStageInputFilesDedup: colliding base names are disambiguated so they don't
// clobber each other at /workspace/<name>.
func TestStageInputFilesDedup(t *testing.T) {
st := &stagerFunc{}
ex := newStagerExecutor(st)
out := ex.stageInputFiles(context.Background(), "r", "a", []tool.InputFile{
{Name: "a.wav", MimeType: "audio/wav", Data: []byte("1")},
{Name: "a.wav", MimeType: "audio/wav", Data: []byte("2")},
}, "go")
if len(st.staged) != 2 {
t.Fatalf("staged %d files, want 2", len(st.staged))
}
if st.staged[0].name != "a.wav" || st.staged[1].name != "a-2.wav" {
t.Errorf("dedup names = %q, %q; want a.wav, a-2.wav", st.staged[0].name, st.staged[1].name)
}
if !strings.Contains(out, "a-2.wav") {
t.Errorf("output missing disambiguated name:\n%s", out)
}
}
// TestStageInputFilesSkipsBad: empty + oversized files are skipped; a save error
// drops only that file. With nothing staged, the prompt is unchanged.
func TestStageInputFilesSkipsBad(t *testing.T) {
// Empty data → skipped; with no good files the prompt is returned as-is.
ex := newStagerExecutor(&stagerFunc{})
if out := ex.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "empty.bin", Data: nil}}, "p"); out != "p" {
t.Errorf("empty file should be skipped, got %q", out)
}
// A stager error → that file is dropped; nothing staged → prompt unchanged.
exErr := newStagerExecutor(&stagerFunc{err: errors.New("disk full")})
if out := exErr.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p"); out != "p" {
t.Errorf("save error should drop the file and leave the prompt, got %q", out)
}
}
// TestStageInputFilesOversize: a file past the byte cap is skipped (prompt
// unchanged), exercising the size guard directly.
func TestStageInputFilesOversize(t *testing.T) {
st := &stagerFunc{}
ex := newStagerExecutor(st)
big := make([]byte, maxInputFileBytes+1)
out := ex.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "huge.bin", Data: big}}, "p")
if out != "p" || len(st.staged) != 0 {
t.Errorf("oversized file should be skipped: out=%q staged=%d", out, len(st.staged))
}
}
// TestStageInputFilesCountCap: more than maxInputFiles attachments are truncated
// to the cap.
func TestStageInputFilesCountCap(t *testing.T) {
st := &stagerFunc{}
ex := newStagerExecutor(st)
files := make([]tool.InputFile, maxInputFiles+5)
for i := range files {
files[i] = tool.InputFile{Name: "f.bin", Data: []byte("x")}
}
ex.stageInputFiles(context.Background(), "r", "a", files, "p")
if len(st.staged) != maxInputFiles {
t.Errorf("count cap: staged %d, want %d", len(st.staged), maxInputFiles)
}
}
// TestSanitizeName: traversal + absolute + control-char filenames are reduced to
// a safe base name (no path separators, no newlines), with a fallback.
func TestSanitizeName(t *testing.T) {
cases := map[string]string{
"../../etc/passwd": "passwd",
"/etc/cron.d/x": "x",
`..\..\windows\sys`: "sys",
"clip.mp3": "clip.mp3",
"": "attachment",
"..": "attachment",
".": "attachment",
"evil\n- injected": "evil- injected",
"a/b/c.wav": "c.wav",
}
for in, want := range cases {
if got := sanitizeName(in); got != want {
t.Errorf("sanitizeName(%q) = %q, want %q", in, got, want)
}
// A sanitized name must never carry a path separator or newline.
got := sanitizeName(in)
if strings.ContainsAny(got, "/\\\n\r") {
t.Errorf("sanitizeName(%q) = %q still contains a separator/newline", in, got)
}
}
}
// TestStageInputFilesSanitizesTraversal: a traversal filename is staged AND
// described under its safe base name only.
func TestStageInputFilesSanitizesTraversal(t *testing.T) {
st := &stagerFunc{}
ex := newStagerExecutor(st)
out := ex.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "../../../etc/passwd", MimeType: "text/plain", Data: []byte("x")}}, "go")
if len(st.staged) != 1 || st.staged[0].name != "passwd" {
t.Fatalf("staged name = %+v, want passwd", st.staged)
}
if strings.Contains(out, "..") || strings.Contains(out, "/etc/") {
t.Errorf("descriptor leaked the traversal path:\n%s", out)
}
}
// TestSanitizeFieldStripsBidiAndControl: control chars AND Unicode format/bidi
// overrides are removed from inlined values.
func TestSanitizeFieldStripsBidiAndControl(t *testing.T) {
in := "audio/mpg\n; rm -rf" // bidi override + newline
got := sanitizeField(in)
if strings.ContainsAny(got, "\n\r\t") || strings.ContainsRune(got, '') {
t.Errorf("sanitizeField left control/bidi chars: %q", got)
}
}
// TestStageInputFilesSanitizesMime: a mime with a control char is cleaned in BOTH
// the staged value and the descriptor.
func TestStageInputFilesSanitizesMime(t *testing.T) {
st := &stagerFunc{}
ex := newStagerExecutor(st)
out := ex.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "c.wav", MimeType: "audio/wav\ninjected", Data: []byte("x")}}, "go")
if len(st.staged) != 1 || strings.ContainsAny(st.staged[0].mime, "\n\r") {
t.Errorf("mime not sanitized before staging: %+v", st.staged)
}
if strings.Contains(out, "\ninjected") {
t.Errorf("descriptor carried an unsanitized mime newline:\n%s", out)
}
}
// TestStageInputFilesEmptyFileID: a stager returning an empty file_id drops the
// file (no blank file_id in the descriptor).
func TestStageInputFilesEmptyFileID(t *testing.T) {
ex := newStagerExecutor(emptyIDStager{})
out := ex.stageInputFiles(context.Background(), "r", "a",
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p")
if out != "p" {
t.Errorf("empty file_id should drop the file, got %q", out)
}
}
type emptyIDStager struct{}
func (emptyIDStager) StageInputFile(context.Context, string, string, string, string, []byte) (string, error) {
return "", nil
}
// TestHumanizeBytesNoPanic: an absurd size clamps to the last prefix instead of
// indexing past "KMGTPE".
func TestHumanizeBytesNoPanic(t *testing.T) {
defer func() {
if r := recover(); r != nil {
t.Fatalf("humanizeBytes panicked: %v", r)
}
}()
for _, n := range []int{0, 512, 2048, 5_000_000, 1 << 62} {
_ = humanizeBytes(n)
}
}
-398
View File
@@ -1,398 +0,0 @@
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
}
-278
View File
@@ -1,278 +0,0 @@
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")
}
}
+5 -45
View File
@@ -33,30 +33,15 @@ type Ports struct {
Budget Budget
// Critic optionally monitors a long run for hangs/runaways. nil = none.
Critic Critic
// Checkpointer mints a per-run Checkpointer for durable recovery (it decides
// per run whether the run is durable). nil = no checkpointing (a run
// interrupted by shutdown is simply lost).
Checkpointer CheckpointerFactory
// Checkpointer persists resumable progress for durable recovery. nil = no
// checkpointing (a run interrupted by shutdown is simply lost).
Checkpointer Checkpointer
// Palette resolves SkillPalette / SubAgentPalette entries into delegation
// tools (skill__<name> / agent__<name>). nil = those entries are inert.
Palette PaletteSource
// Delivery is where the run's output + artifacts go. nil = the caller
// reads the Result in-process (the light-host default).
Delivery deliver.Delivery
// InputFiles persists non-image input attachments (audio, PDF, binary)
// carried on Invocation.InputFiles into a host file store under run scope,
// returning file_ids the agent can hand to a worker tool. nil = input files
// are silently ignored (the run still proceeds, text-only). The bytes are
// never inlined into the model context — the LLM can't read raw audio/binary.
InputFiles InputFileStager
}
// InputFileStager persists a single non-image input attachment into a host file
// store under run scope and returns a file_id the run can reference. It is the
// seam mort's skill FileStorage (and any host blob store) implements so the
// kernel can stage Invocation.InputFiles without importing a storage layer.
type InputFileStager interface {
StageInputFile(ctx context.Context, runID, agentID, name, mime string, content []byte) (fileID string, err error)
}
// RunInfo describes a run at start time — the attribution a recorder/critic
@@ -67,9 +52,7 @@ 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
@@ -175,16 +158,6 @@ 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 {
@@ -197,24 +170,11 @@ type Checkpointer interface {
Fail(ctx context.Context, err error) error
}
// RunCheckpointState is the resumable snapshot a Checkpointer persists.
// RunCheckpointState is the resumable snapshot a Checkpointer persists. Kept
// 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 ---
-156
View File
@@ -1,156 +0,0 @@
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
}
-154
View File
@@ -1,154 +0,0 @@
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)
}
}
-208
View File
@@ -1,208 +0,0 @@
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
}
-96
View File
@@ -1,96 +0,0 @@
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)
}
}
-114
View File
@@ -1,114 +0,0 @@
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
}
-139
View File
@@ -1,139 +0,0 @@
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))
}
-105
View File
@@ -1,105 +0,0 @@
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")
}
}
-54
View File
@@ -1,54 +0,0 @@
// 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")
-149
View File
@@ -1,149 +0,0 @@
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
}
-97
View File
@@ -1,97 +0,0 @@
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)
}
}
-27
View File
@@ -1,27 +0,0 @@
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)
}
-65
View File
@@ -1,65 +0,0 @@
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{}
}
-203
View File
@@ -1,203 +0,0 @@
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
}
-177
View File
@@ -1,177 +0,0 @@
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)
}
}
+3 -4
View File
@@ -154,10 +154,9 @@ type ContinuationContext struct {
// InputFile is a non-image file the user supplied with a run (audio,
// etc.). The executor stages it into the file store under run scope and
// surfaces its file_id to the agent. Name may be an untrusted attachment
// filename — the executor reduces it to a safe base name (stripping path
// separators + control chars) before staging or exposing it as
// /workspace/<name>; MimeType is the resolved content type; Data is the raw bytes.
// surfaces its file_id to the agent. Name is a safe base name (no path
// separators) suitable for /workspace/<name>; MimeType is the resolved
// content type; Data is the raw bytes.
type InputFile struct {
Name string
MimeType string