Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7a074d8fa2 | ||
|
|
007b7539c2 |
+76
-2
@@ -5,6 +5,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.stevedudenhoeffer.com/steve/majordomo/agent"
|
"gitea.stevedudenhoeffer.com/steve/majordomo/agent"
|
||||||
@@ -429,8 +430,10 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
|
|||||||
// store and fold an [ATTACHED FILES] descriptor into the prompt so the agent
|
// store and fold an [ATTACHED FILES] descriptor into the prompt so the agent
|
||||||
// can reach them by file_id. No-op when Ports.InputFiles is nil or there are
|
// can reach them by file_id. No-op when Ports.InputFiles is nil or there are
|
||||||
// no files. Done after the model/toolbox build but before the loop, so the
|
// no files. Done after the model/toolbox build but before the loop, so the
|
||||||
// descriptor rides the very first user turn.
|
// descriptor rides the very first user turn. The staged file ids are kept for
|
||||||
input = e.stageInputFiles(runCtx, inv.RunID, ra.ID, inv.InputFiles, input)
|
// the FinalGuard: they live under run scope but are the user's INPUTS, not
|
||||||
|
// run-produced artifacts, so an undelivered-artifact check must exclude them.
|
||||||
|
input, stagedFileIDs := e.stageInputFiles(runCtx, inv.RunID, ra.ID, inv.InputFiles, input)
|
||||||
// One WithSteer drains BOTH the session mailbox (a tool's AttachImages) and
|
// One WithSteer drains BOTH the session mailbox (a tool's AttachImages) and
|
||||||
// the critic's nudges before each step.
|
// the critic's nudges before each step.
|
||||||
steer := func() []llm.Message { return append(mailbox.drain(), critic.drainSteer()...) }
|
steer := func() []llm.Message { return append(mailbox.drain(), critic.drainSteer()...) }
|
||||||
@@ -452,6 +455,10 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
|
|||||||
// step responses, not the loop's compacted history) — a host that caps size
|
// step responses, not the loop's compacted history) — a host that caps size
|
||||||
// bounds it. A recovered run seeds the saved transcript and continues.
|
// bounds it. A recovered run seeds the saved transcript and continues.
|
||||||
obs := stepObserver
|
obs := stepObserver
|
||||||
|
// noteCheckpoint appends an out-of-loop message (the FinalGuard nudge) to
|
||||||
|
// the checkpoint transcript so a shutdown-resume of a nudged run replays
|
||||||
|
// the nudge turn too. nil when the run is non-durable.
|
||||||
|
var noteCheckpoint func(llm.Message)
|
||||||
if ckpt != nil {
|
if ckpt != nil {
|
||||||
var acc []llm.Message
|
var acc []llm.Message
|
||||||
if resuming {
|
if resuming {
|
||||||
@@ -469,6 +476,7 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
|
|||||||
}
|
}
|
||||||
_ = ckpt.Save(runCtx, RunCheckpointState{Messages: acc, Iteration: s.Index + 1})
|
_ = ckpt.Save(runCtx, RunCheckpointState{Messages: acc, Iteration: s.Index + 1})
|
||||||
}
|
}
|
||||||
|
noteCheckpoint = func(m llm.Message) { acc = append(acc, m) }
|
||||||
}
|
}
|
||||||
opts := append([]agent.Option{
|
opts := append([]agent.Option{
|
||||||
agent.WithToolbox(toolbox),
|
agent.WithToolbox(toolbox),
|
||||||
@@ -483,6 +491,72 @@ func (e *Executor) Run(ctx context.Context, ra RunnableAgent, inv tool.Invocatio
|
|||||||
} else {
|
} else {
|
||||||
runRes, runErr = runAgent(runCtx, ag, input, inv.Images, agent.WithSteer(steer))
|
runRes, runErr = runAgent(runCtx, ag, input, inv.Images, agent.WithSteer(steer))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// FinalGuard (optional): the run finished successfully — before the result
|
||||||
|
// is finalized, let the host inspect it for undelivered work (the
|
||||||
|
// announce-then-stop failure: an artifact rendered into run storage, a
|
||||||
|
// final text claiming "Done", and no delivery tool call). A non-empty
|
||||||
|
// nudge runs ONE bounded extra round on the same conversation: same model,
|
||||||
|
// same toolbox, same observers (audit/steps/critic all see it), but a
|
||||||
|
// small fixed step cap so a misread nudge can't start a new work spree.
|
||||||
|
// The nudge round's failure is non-fatal — the original successful result
|
||||||
|
// stands (any attachments it queued before failing still deliver).
|
||||||
|
//
|
||||||
|
// The nudge round's job is to QUEUE deliveries (tool calls); its TEXT is
|
||||||
|
// meta-commentary ("answer above stands", "sent!") that must never replace
|
||||||
|
// the answer the user asked for — hosts deliver Output text + separately
|
||||||
|
// drained attachments, so replacing Output with the nudge round's closer
|
||||||
|
// LOSES the real answer (live regression: run d5ec39f4, a research run's
|
||||||
|
// page-cache false positive whose nudge reply "my answer above stands
|
||||||
|
// as-is" was delivered INSTEAD of the answer). Keep the original Output
|
||||||
|
// unless it was blank; log what was discarded so hosts can audit.
|
||||||
|
if runErr == nil && runRes != nil && e.cfg.Ports.FinalGuard != nil {
|
||||||
|
state := FinalState{
|
||||||
|
Output: runRes.Output,
|
||||||
|
ToolNames: toolboxNames(toolbox),
|
||||||
|
StagedFileIDs: stagedFileIDs,
|
||||||
|
}
|
||||||
|
if nudge := safeFinalNudge(runCtx, e.cfg.Ports.FinalGuard, info, state); nudge != "" {
|
||||||
|
if rec != nil {
|
||||||
|
rec.LogEvent("final_nudge", map[string]any{"nudge": nudge})
|
||||||
|
}
|
||||||
|
if noteCheckpoint != nil {
|
||||||
|
noteCheckpoint(llm.UserText(nudge))
|
||||||
|
}
|
||||||
|
nudgeOpts := append([]agent.Option{
|
||||||
|
agent.WithToolbox(toolbox),
|
||||||
|
agent.WithMaxSteps(finalNudgeMaxSteps),
|
||||||
|
agent.WithStepObserver(obs),
|
||||||
|
}, sharedOpts...)
|
||||||
|
nudgeAg := agent.New(model, e.systemPrompt(ra), nudgeOpts...)
|
||||||
|
nudgeRes, nudgeErr := nudgeAg.Run(runCtx, nudge,
|
||||||
|
agent.WithSteer(steer), agent.WithHistory(runRes.Messages))
|
||||||
|
if nudgeRes != nil {
|
||||||
|
// The extra round's tokens are real spend either way.
|
||||||
|
runRes.Usage.Add(nudgeRes.Usage)
|
||||||
|
}
|
||||||
|
if nudgeErr != nil {
|
||||||
|
slog.Warn("run: final-nudge round failed; keeping original result",
|
||||||
|
"run_id", inv.RunID, "error", nudgeErr)
|
||||||
|
} else if nudgeRes != nil {
|
||||||
|
if strings.TrimSpace(runRes.Output) == "" {
|
||||||
|
// Degenerate original (blank stop turn): the nudge round's
|
||||||
|
// text is the only answer there is.
|
||||||
|
runRes.Output = nudgeRes.Output
|
||||||
|
} else if rec != nil {
|
||||||
|
// Discarded meta-text stays observable: a nudge round that
|
||||||
|
// DECLINED to queue while the original claims a send shows
|
||||||
|
// up here, not in the user's reply.
|
||||||
|
rec.LogEvent("final_nudge_result", map[string]any{
|
||||||
|
"discarded_output": nudgeRes.Output,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
// The transcript keeps the nudge round either way (audit,
|
||||||
|
// PostRun, checkpoint continuity).
|
||||||
|
runRes.Messages = nudgeRes.Messages
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
// Multi-phase pipeline: each phase runs its own prompt/tier/tools/step-cap
|
// Multi-phase pipeline: each phase runs its own prompt/tier/tools/step-cap
|
||||||
// sequentially, threading outputs through {{.<PhaseName>}} templates. The
|
// sequentially, threading outputs through {{.<PhaseName>}} templates. The
|
||||||
|
|||||||
@@ -0,0 +1,87 @@
|
|||||||
|
package run
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"log/slog"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
|
||||||
|
)
|
||||||
|
|
||||||
|
// FinalGuard is the optional host seam consulted when a single-loop run is
|
||||||
|
// about to finalize SUCCESSFULLY (the model produced a final answer and no run
|
||||||
|
// error occurred). It exists for the "announce-then-stop" failure: an agent
|
||||||
|
// renders an artifact into host storage, narrates "Done — sent you the file",
|
||||||
|
// and ends the run without ever calling the host's delivery tool — so nothing
|
||||||
|
// reaches the user while the final text claims otherwise (observed live:
|
||||||
|
// gifsmith run 29170c93 2026-07-12, general run e9e7bf40 2026-07-14).
|
||||||
|
//
|
||||||
|
// FinalNudge returns "" to let the run finish as-is (the overwhelmingly common
|
||||||
|
// case — the host should make its check cheap), or a non-empty nudge message.
|
||||||
|
// A non-empty nudge sends the loop back for ONE bounded extra round: the nudge
|
||||||
|
// is appended to the SAME conversation as the next user turn and the agent
|
||||||
|
// runs again with the same toolbox, capped at finalNudgeMaxSteps steps — enough
|
||||||
|
// to deliver a stranded artifact, not enough to start a new project. The guard
|
||||||
|
// is consulted at most ONCE per run; the nudge round's own stop turn is final.
|
||||||
|
// Multi-phase runs are not guarded (their output contract is the phase
|
||||||
|
// pipeline's, not a delivery surface's).
|
||||||
|
//
|
||||||
|
// OUTPUT CONTRACT: the nudge round exists to make TOOL CALLS (queue the
|
||||||
|
// delivery); its final TEXT is discarded — the run's Output stays the original
|
||||||
|
// answer unless that answer was blank. A host's nudge message should say so
|
||||||
|
// explicitly ("any text you write here is not shown to the user"), and must
|
||||||
|
// not instruct the model to compose a corrected reply. Discarded nudge text is
|
||||||
|
// recorded as a final_nudge_result audit event.
|
||||||
|
//
|
||||||
|
// The call runs on the run goroutine, inside the run's deadline, and is
|
||||||
|
// panic-isolated: a guard panic is logged and treated as "" (the run's
|
||||||
|
// successful result must never be lost to a guard bug).
|
||||||
|
type FinalGuard interface {
|
||||||
|
FinalNudge(ctx context.Context, info RunInfo, state FinalState) string
|
||||||
|
}
|
||||||
|
|
||||||
|
// FinalState is the snapshot a FinalGuard receives about the finishing run.
|
||||||
|
type FinalState struct {
|
||||||
|
// Output is the run's final answer text (what the host would deliver).
|
||||||
|
Output string
|
||||||
|
// ToolNames lists the tools that were available to the run, so a guard can
|
||||||
|
// skip nudging a run that has no way to act (e.g. no send_attachments).
|
||||||
|
ToolNames []string
|
||||||
|
// StagedFileIDs are the file ids the executor staged from the invocation's
|
||||||
|
// input attachments (Ports.InputFiles). They live under the run's file scope
|
||||||
|
// but are the USER's inputs, not run-produced artifacts — a guard checking
|
||||||
|
// for undelivered artifacts must exclude them.
|
||||||
|
StagedFileIDs []string
|
||||||
|
}
|
||||||
|
|
||||||
|
// finalNudgeMaxSteps caps the nudge round's tool-dispatch steps. Delivering a
|
||||||
|
// stranded artifact needs at most a file_list + a send_attachments + a final
|
||||||
|
// text turn; the cap keeps a model that misreads the nudge from launching a
|
||||||
|
// whole new work spree on the host's budget.
|
||||||
|
const finalNudgeMaxSteps = 6
|
||||||
|
|
||||||
|
// safeFinalNudge consults the guard behind a recover so a guard panic degrades
|
||||||
|
// to "no nudge" instead of converting a successful run into a panic-error (the
|
||||||
|
// executor's top-level recover would otherwise stamp res.Err).
|
||||||
|
func safeFinalNudge(ctx context.Context, g FinalGuard, info RunInfo, state FinalState) (nudge string) {
|
||||||
|
defer func() {
|
||||||
|
if r := recover(); r != nil {
|
||||||
|
slog.Error("run: FinalGuard panicked; skipping nudge",
|
||||||
|
"run_id", info.RunID, "panic", r)
|
||||||
|
nudge = ""
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return g.FinalNudge(ctx, info, state)
|
||||||
|
}
|
||||||
|
|
||||||
|
// toolboxNames flattens the run toolbox's tool names for FinalState (nil-safe).
|
||||||
|
func toolboxNames(b *llm.Toolbox) []string {
|
||||||
|
if b == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
tools := b.Tools()
|
||||||
|
names := make([]string, 0, len(tools))
|
||||||
|
for _, t := range tools {
|
||||||
|
names = append(names, t.Name)
|
||||||
|
}
|
||||||
|
return names
|
||||||
|
}
|
||||||
@@ -0,0 +1,328 @@
|
|||||||
|
package run
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/majordomo/llm"
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/executus/tool"
|
||||||
|
)
|
||||||
|
|
||||||
|
// guardFunc adapts a func to FinalGuard, counting calls and capturing the
|
||||||
|
// snapshot it received.
|
||||||
|
type guardFunc struct {
|
||||||
|
nudge func(info RunInfo, state FinalState) string
|
||||||
|
calls int
|
||||||
|
info RunInfo
|
||||||
|
state FinalState
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *guardFunc) FinalNudge(_ context.Context, info RunInfo, state FinalState) string {
|
||||||
|
g.calls++
|
||||||
|
g.info = info
|
||||||
|
g.state = state
|
||||||
|
return g.nudge(info, state)
|
||||||
|
}
|
||||||
|
|
||||||
|
// eventRecorder is captureRecorder plus LogEvent capture (for the final_nudge
|
||||||
|
// audit event).
|
||||||
|
type eventRecorder struct {
|
||||||
|
captureRecorder
|
||||||
|
events []string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *eventRecorder) LogEvent(eventType string, _ map[string]any) {
|
||||||
|
r.events = append(r.events, eventType)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardNudgeRunsOneExtraRound is the core announce-then-stop fix: the
|
||||||
|
// model stops with "Done — sent!" (nothing delivered), the guard returns a
|
||||||
|
// nudge, and the SAME conversation runs one more round — here a tool call
|
||||||
|
// (queueing the delivery) then a closing text turn. The guard is consulted
|
||||||
|
// exactly once (the nudge round's own stop is final), the audit log records
|
||||||
|
// the nudge, and the ORIGINAL answer text survives: the nudge round's closer
|
||||||
|
// is discarded (recorded as final_nudge_result), because hosts deliver Output
|
||||||
|
// text + separately-drained attachments — replacing Output with the closer is
|
||||||
|
// the run-d5ec39f4 lost-answer regression.
|
||||||
|
func TestFinalGuardNudgeRunsOneExtraRound(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m",
|
||||||
|
fake.Reply("Done — sent you the file!"),
|
||||||
|
// Nudge round: an (unregistered → tolerated error result) delivery call,
|
||||||
|
// then a meta closer that must NOT become the run output.
|
||||||
|
fake.ReplyWith(llm.Response{ToolCalls: []llm.ToolCall{{ID: "c1", Name: "send_attachments", Arguments: []byte(`{}`)}}}),
|
||||||
|
fake.Reply("queued the file — answer above stands"),
|
||||||
|
)
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
rec := &eventRecorder{}
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "you did not deliver the file" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{
|
||||||
|
Audit: auditFunc{start: func(RunInfo) RunRecorder { return rec }},
|
||||||
|
FinalGuard: guard,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{Name: "g", ModelTier: "m"},
|
||||||
|
tool.Invocation{RunID: "r-nudge", CallerID: "c"}, "make me a file")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("run error: %v", res.Err)
|
||||||
|
}
|
||||||
|
if res.Output != "Done — sent you the file!" {
|
||||||
|
t.Fatalf("output = %q, want the ORIGINAL answer preserved (nudge closer discarded)", res.Output)
|
||||||
|
}
|
||||||
|
if n := fp.CallCount("m"); n != 3 {
|
||||||
|
t.Fatalf("model called %d times, want 3 (the nudge round must still run)", n)
|
||||||
|
}
|
||||||
|
if guard.calls != 1 {
|
||||||
|
t.Fatalf("guard consulted %d times, want exactly 1 (no re-guard of the nudge round)", guard.calls)
|
||||||
|
}
|
||||||
|
if guard.info.RunID != "r-nudge" {
|
||||||
|
t.Errorf("guard info.RunID = %q", guard.info.RunID)
|
||||||
|
}
|
||||||
|
if guard.state.Output != "Done — sent you the file!" {
|
||||||
|
t.Errorf("guard state.Output = %q, want the pre-nudge final text", guard.state.Output)
|
||||||
|
}
|
||||||
|
var sawNudge, sawResult bool
|
||||||
|
for _, ev := range rec.events {
|
||||||
|
switch ev {
|
||||||
|
case "final_nudge":
|
||||||
|
sawNudge = true
|
||||||
|
case "final_nudge_result":
|
||||||
|
sawResult = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !sawNudge {
|
||||||
|
t.Errorf("audit events %v missing final_nudge", rec.events)
|
||||||
|
}
|
||||||
|
if !sawResult {
|
||||||
|
t.Errorf("audit events %v missing final_nudge_result (the discarded closer must stay observable)", rec.events)
|
||||||
|
}
|
||||||
|
if rec.stats.Output != "Done — sent you the file!" {
|
||||||
|
t.Errorf("audit close output = %q, want the original answer", rec.stats.Output)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardNudgeFillsEmptyOriginal: when the ORIGINAL stop turn was blank
|
||||||
|
// (a degenerate run), the nudge round's text is the only answer there is — it
|
||||||
|
// becomes the output instead of being discarded.
|
||||||
|
func TestFinalGuardNudgeFillsEmptyOriginal(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m",
|
||||||
|
fake.Reply(" \n"),
|
||||||
|
fake.Reply("actual answer from the nudge round"),
|
||||||
|
)
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "deliver or correct" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{FinalGuard: guard},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m"}, tool.Invocation{RunID: "r"}, "hi")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("run error: %v", res.Err)
|
||||||
|
}
|
||||||
|
if res.Output != "actual answer from the nudge round" {
|
||||||
|
t.Fatalf("output = %q, want the nudge round's text to fill a blank original", res.Output)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardEmptyNudgeNoExtraRound: a guard returning "" adds zero model
|
||||||
|
// calls — the common path stays one round.
|
||||||
|
func TestFinalGuardEmptyNudgeNoExtraRound(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m", fake.Reply("plain answer"))
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{FinalGuard: guard},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m"}, tool.Invocation{RunID: "r"}, "hi")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("run error: %v", res.Err)
|
||||||
|
}
|
||||||
|
if res.Output != "plain answer" {
|
||||||
|
t.Fatalf("output = %q", res.Output)
|
||||||
|
}
|
||||||
|
if guard.calls != 1 {
|
||||||
|
t.Errorf("guard consulted %d times, want 1", guard.calls)
|
||||||
|
}
|
||||||
|
if n := fp.CallCount("m"); n != 1 {
|
||||||
|
t.Errorf("model called %d times, want 1 (empty nudge must not re-run)", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardNudgeRoundFailureKeepsResult: the nudge round erroring (model
|
||||||
|
// failure, deadline, step cap) must NOT damage the original successful result.
|
||||||
|
func TestFinalGuardNudgeRoundFailureKeepsResult(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m",
|
||||||
|
fake.Reply("original answer"),
|
||||||
|
fake.Fail(errors.New("provider fell over")),
|
||||||
|
)
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "deliver it" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{FinalGuard: guard},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m"}, tool.Invocation{RunID: "r"}, "hi")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("a failed nudge round must not fail the run: %v", res.Err)
|
||||||
|
}
|
||||||
|
if res.Output != "original answer" {
|
||||||
|
t.Fatalf("output = %q, want the original preserved", res.Output)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardNotConsultedOnRunError: a failed run is never nudged — the
|
||||||
|
// guard exists for successful-but-undelivered finishes only.
|
||||||
|
func TestFinalGuardNotConsultedOnRunError(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m", fake.Fail(errors.New("boom")))
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "nudge" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{FinalGuard: guard},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m"}, tool.Invocation{RunID: "r"}, "hi")
|
||||||
|
|
||||||
|
if res.Err == nil {
|
||||||
|
t.Fatal("expected a run error")
|
||||||
|
}
|
||||||
|
if guard.calls != 0 {
|
||||||
|
t.Errorf("guard consulted %d times on a failed run, want 0", guard.calls)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardPanicIsolated: a panicking guard degrades to "no nudge" — the
|
||||||
|
// successful result survives untouched.
|
||||||
|
func TestFinalGuardPanicIsolated(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m", fake.Reply("safe answer"))
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{FinalGuard: panicGuard{}},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m"}, tool.Invocation{RunID: "r"}, "hi")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("guard panic must not fail the run: %v", res.Err)
|
||||||
|
}
|
||||||
|
if res.Output != "safe answer" {
|
||||||
|
t.Fatalf("output = %q", res.Output)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type panicGuard struct{}
|
||||||
|
|
||||||
|
func (panicGuard) FinalNudge(context.Context, RunInfo, FinalState) string {
|
||||||
|
panic("guard bug")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardReceivesToolNamesAndStagedIDs: the guard's snapshot carries the
|
||||||
|
// run's tool names (via ExtraTools here) and the input-staged file ids, so a
|
||||||
|
// host can skip runs that can't deliver and exclude the user's own attachments.
|
||||||
|
func TestFinalGuardReceivesToolNamesAndStagedIDs(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m", fake.Reply("done"))
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{
|
||||||
|
FinalGuard: guard,
|
||||||
|
InputFiles: guardStager(func(name string) string { return "staged-" + name }),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m"},
|
||||||
|
tool.Invocation{
|
||||||
|
RunID: "r",
|
||||||
|
ExtraTools: []llm.Tool{{
|
||||||
|
Name: "send_attachments",
|
||||||
|
Description: "deliver",
|
||||||
|
Handler: func(context.Context, json.RawMessage) (any, error) { return "ok", nil },
|
||||||
|
}},
|
||||||
|
InputFiles: []tool.InputFile{{Name: "input.wav", MimeType: "audio/wave", Data: []byte("x")}},
|
||||||
|
}, "hi")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("run error: %v", res.Err)
|
||||||
|
}
|
||||||
|
var sawTool bool
|
||||||
|
for _, n := range guard.state.ToolNames {
|
||||||
|
if n == "send_attachments" {
|
||||||
|
sawTool = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !sawTool {
|
||||||
|
t.Errorf("guard state.ToolNames = %v, want send_attachments present", guard.state.ToolNames)
|
||||||
|
}
|
||||||
|
if len(guard.state.StagedFileIDs) != 1 || guard.state.StagedFileIDs[0] != "staged-input.wav" {
|
||||||
|
t.Errorf("guard state.StagedFileIDs = %v, want [staged-input.wav]", guard.state.StagedFileIDs)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// guardStager maps an input file's name to a deterministic file id.
|
||||||
|
type guardStager func(name string) string
|
||||||
|
|
||||||
|
func (f guardStager) StageInputFile(_ context.Context, _, _, name, _ string, _ []byte) (string, error) {
|
||||||
|
return f(name), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFinalGuardSkipsMultiPhaseRuns: a phase pipeline's output contract is the
|
||||||
|
// pipeline's, not a delivery surface's — the guard must not be consulted.
|
||||||
|
func TestFinalGuardSkipsMultiPhaseRuns(t *testing.T) {
|
||||||
|
fp := fake.New("fake")
|
||||||
|
fp.Enqueue("m", fake.Reply("phase-out"))
|
||||||
|
m, _ := fp.Model("m")
|
||||||
|
guard := &guardFunc{nudge: func(RunInfo, FinalState) string { return "nudge" }}
|
||||||
|
ex := New(Config{
|
||||||
|
Registry: tool.NewRegistry(),
|
||||||
|
Models: func(ctx context.Context, _ string) (context.Context, llm.Model, error) { return ctx, m, nil },
|
||||||
|
Ports: Ports{FinalGuard: guard},
|
||||||
|
})
|
||||||
|
|
||||||
|
res := ex.Run(context.Background(),
|
||||||
|
RunnableAgent{ModelTier: "m", Phases: []Phase{{Name: "a", SystemPrompt: "do a"}}},
|
||||||
|
tool.Invocation{RunID: "r"}, "hi")
|
||||||
|
|
||||||
|
if res.Err != nil {
|
||||||
|
t.Fatalf("run error: %v", res.Err)
|
||||||
|
}
|
||||||
|
if guard.calls != 0 {
|
||||||
|
t.Errorf("guard consulted %d times on a phase run, want 0", guard.calls)
|
||||||
|
}
|
||||||
|
}
|
||||||
+11
-6
@@ -30,10 +30,11 @@ const maxInputFiles = 32
|
|||||||
//
|
//
|
||||||
// Best-effort: a nil stager, no files, or a per-file save error degrades to
|
// 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)
|
// "skip that file" — the run still proceeds. Returns the (possibly augmented)
|
||||||
// prompt.
|
// prompt plus the staged file ids (for the FinalGuard: staged inputs live under
|
||||||
func (e *Executor) stageInputFiles(ctx context.Context, runID, agentID string, files []tool.InputFile, prompt string) string {
|
// run scope but are not run-produced artifacts).
|
||||||
|
func (e *Executor) stageInputFiles(ctx context.Context, runID, agentID string, files []tool.InputFile, prompt string) (string, []string) {
|
||||||
if e.cfg.Ports.InputFiles == nil || len(files) == 0 {
|
if e.cfg.Ports.InputFiles == nil || len(files) == 0 {
|
||||||
return prompt
|
return prompt, nil
|
||||||
}
|
}
|
||||||
// Count cap: bound how many attachments one run can stage, independent of the
|
// 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).
|
// per-file byte cap (defense-in-depth against a flood of tiny files).
|
||||||
@@ -87,7 +88,11 @@ func (e *Executor) stageInputFiles(ctx context.Context, runID, agentID string, f
|
|||||||
staged = append(staged, stagedFile{name: name, mime: mime, fileID: sanitizeField(fileID), size: len(f.Data)})
|
staged = append(staged, stagedFile{name: name, mime: mime, fileID: sanitizeField(fileID), size: len(f.Data)})
|
||||||
}
|
}
|
||||||
if len(staged) == 0 {
|
if len(staged) == 0 {
|
||||||
return prompt
|
return prompt, nil
|
||||||
|
}
|
||||||
|
stagedIDs := make([]string, 0, len(staged))
|
||||||
|
for _, s := range staged {
|
||||||
|
stagedIDs = append(stagedIDs, s.fileID)
|
||||||
}
|
}
|
||||||
|
|
||||||
var b strings.Builder
|
var b strings.Builder
|
||||||
@@ -101,9 +106,9 @@ func (e *Executor) stageInputFiles(ctx context.Context, runID, agentID string, f
|
|||||||
}
|
}
|
||||||
|
|
||||||
if strings.TrimSpace(prompt) == "" {
|
if strings.TrimSpace(prompt) == "" {
|
||||||
return b.String()
|
return b.String(), stagedIDs
|
||||||
}
|
}
|
||||||
return prompt + "\n\n" + b.String()
|
return prompt + "\n\n" + b.String(), stagedIDs
|
||||||
}
|
}
|
||||||
|
|
||||||
// sanitizeName reduces an untrusted attachment filename to a safe base name. It
|
// sanitizeName reduces an untrusted attachment filename to a safe base name. It
|
||||||
|
|||||||
+10
-10
@@ -44,7 +44,7 @@ func newStagerExecutor(s InputFileStager) *Executor {
|
|||||||
func TestStageInputFiles(t *testing.T) {
|
func TestStageInputFiles(t *testing.T) {
|
||||||
st := &stagerFunc{}
|
st := &stagerFunc{}
|
||||||
ex := newStagerExecutor(st)
|
ex := newStagerExecutor(st)
|
||||||
out := ex.stageInputFiles(context.Background(), "run-1", "agent-1",
|
out, _ := ex.stageInputFiles(context.Background(), "run-1", "agent-1",
|
||||||
[]tool.InputFile{{Name: "clip.mp3", MimeType: "audio/mpeg", Data: []byte("abcd")}},
|
[]tool.InputFile{{Name: "clip.mp3", MimeType: "audio/mpeg", Data: []byte("abcd")}},
|
||||||
"transcribe this")
|
"transcribe this")
|
||||||
|
|
||||||
@@ -65,7 +65,7 @@ func TestStageInputFiles(t *testing.T) {
|
|||||||
// drops the run.
|
// drops the run.
|
||||||
func TestStageInputFilesNoStager(t *testing.T) {
|
func TestStageInputFilesNoStager(t *testing.T) {
|
||||||
ex := newStagerExecutor(nil) // Ports.InputFiles == nil
|
ex := newStagerExecutor(nil) // Ports.InputFiles == nil
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a",
|
out, _ := ex.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "prompt")
|
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "prompt")
|
||||||
if out != "prompt" {
|
if out != "prompt" {
|
||||||
t.Errorf("nil stager changed the prompt: %q", out)
|
t.Errorf("nil stager changed the prompt: %q", out)
|
||||||
@@ -75,7 +75,7 @@ func TestStageInputFilesNoStager(t *testing.T) {
|
|||||||
// TestStageInputFilesNoFiles: no attachments leaves the prompt untouched.
|
// TestStageInputFilesNoFiles: no attachments leaves the prompt untouched.
|
||||||
func TestStageInputFilesNoFiles(t *testing.T) {
|
func TestStageInputFilesNoFiles(t *testing.T) {
|
||||||
ex := newStagerExecutor(&stagerFunc{})
|
ex := newStagerExecutor(&stagerFunc{})
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a", nil, "prompt")
|
out, _ := ex.stageInputFiles(context.Background(), "r", "a", nil, "prompt")
|
||||||
if out != "prompt" {
|
if out != "prompt" {
|
||||||
t.Errorf("no files changed the prompt: %q", out)
|
t.Errorf("no files changed the prompt: %q", out)
|
||||||
}
|
}
|
||||||
@@ -86,7 +86,7 @@ func TestStageInputFilesNoFiles(t *testing.T) {
|
|||||||
func TestStageInputFilesDedup(t *testing.T) {
|
func TestStageInputFilesDedup(t *testing.T) {
|
||||||
st := &stagerFunc{}
|
st := &stagerFunc{}
|
||||||
ex := newStagerExecutor(st)
|
ex := newStagerExecutor(st)
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a", []tool.InputFile{
|
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("1")},
|
||||||
{Name: "a.wav", MimeType: "audio/wav", Data: []byte("2")},
|
{Name: "a.wav", MimeType: "audio/wav", Data: []byte("2")},
|
||||||
}, "go")
|
}, "go")
|
||||||
@@ -106,13 +106,13 @@ func TestStageInputFilesDedup(t *testing.T) {
|
|||||||
func TestStageInputFilesSkipsBad(t *testing.T) {
|
func TestStageInputFilesSkipsBad(t *testing.T) {
|
||||||
// Empty data → skipped; with no good files the prompt is returned as-is.
|
// Empty data → skipped; with no good files the prompt is returned as-is.
|
||||||
ex := newStagerExecutor(&stagerFunc{})
|
ex := newStagerExecutor(&stagerFunc{})
|
||||||
if out := ex.stageInputFiles(context.Background(), "r", "a",
|
if out, _ := ex.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "empty.bin", Data: nil}}, "p"); out != "p" {
|
[]tool.InputFile{{Name: "empty.bin", Data: nil}}, "p"); out != "p" {
|
||||||
t.Errorf("empty file should be skipped, got %q", out)
|
t.Errorf("empty file should be skipped, got %q", out)
|
||||||
}
|
}
|
||||||
// A stager error → that file is dropped; nothing staged → prompt unchanged.
|
// A stager error → that file is dropped; nothing staged → prompt unchanged.
|
||||||
exErr := newStagerExecutor(&stagerFunc{err: errors.New("disk full")})
|
exErr := newStagerExecutor(&stagerFunc{err: errors.New("disk full")})
|
||||||
if out := exErr.stageInputFiles(context.Background(), "r", "a",
|
if out, _ := exErr.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p"); out != "p" {
|
[]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)
|
t.Errorf("save error should drop the file and leave the prompt, got %q", out)
|
||||||
}
|
}
|
||||||
@@ -124,7 +124,7 @@ func TestStageInputFilesOversize(t *testing.T) {
|
|||||||
st := &stagerFunc{}
|
st := &stagerFunc{}
|
||||||
ex := newStagerExecutor(st)
|
ex := newStagerExecutor(st)
|
||||||
big := make([]byte, maxInputFileBytes+1)
|
big := make([]byte, maxInputFileBytes+1)
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a",
|
out, _ := ex.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "huge.bin", Data: big}}, "p")
|
[]tool.InputFile{{Name: "huge.bin", Data: big}}, "p")
|
||||||
if out != "p" || len(st.staged) != 0 {
|
if out != "p" || len(st.staged) != 0 {
|
||||||
t.Errorf("oversized file should be skipped: out=%q staged=%d", out, len(st.staged))
|
t.Errorf("oversized file should be skipped: out=%q staged=%d", out, len(st.staged))
|
||||||
@@ -177,7 +177,7 @@ func TestSanitizeName(t *testing.T) {
|
|||||||
func TestStageInputFilesSanitizesTraversal(t *testing.T) {
|
func TestStageInputFilesSanitizesTraversal(t *testing.T) {
|
||||||
st := &stagerFunc{}
|
st := &stagerFunc{}
|
||||||
ex := newStagerExecutor(st)
|
ex := newStagerExecutor(st)
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a",
|
out, _ := ex.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "../../../etc/passwd", MimeType: "text/plain", Data: []byte("x")}}, "go")
|
[]tool.InputFile{{Name: "../../../etc/passwd", MimeType: "text/plain", Data: []byte("x")}}, "go")
|
||||||
if len(st.staged) != 1 || st.staged[0].name != "passwd" {
|
if len(st.staged) != 1 || st.staged[0].name != "passwd" {
|
||||||
t.Fatalf("staged name = %+v, want passwd", st.staged)
|
t.Fatalf("staged name = %+v, want passwd", st.staged)
|
||||||
@@ -202,7 +202,7 @@ func TestSanitizeFieldStripsBidiAndControl(t *testing.T) {
|
|||||||
func TestStageInputFilesSanitizesMime(t *testing.T) {
|
func TestStageInputFilesSanitizesMime(t *testing.T) {
|
||||||
st := &stagerFunc{}
|
st := &stagerFunc{}
|
||||||
ex := newStagerExecutor(st)
|
ex := newStagerExecutor(st)
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a",
|
out, _ := ex.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "c.wav", MimeType: "audio/wav\ninjected", Data: []byte("x")}}, "go")
|
[]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") {
|
if len(st.staged) != 1 || strings.ContainsAny(st.staged[0].mime, "\n\r") {
|
||||||
t.Errorf("mime not sanitized before staging: %+v", st.staged)
|
t.Errorf("mime not sanitized before staging: %+v", st.staged)
|
||||||
@@ -216,7 +216,7 @@ func TestStageInputFilesSanitizesMime(t *testing.T) {
|
|||||||
// file (no blank file_id in the descriptor).
|
// file (no blank file_id in the descriptor).
|
||||||
func TestStageInputFilesEmptyFileID(t *testing.T) {
|
func TestStageInputFilesEmptyFileID(t *testing.T) {
|
||||||
ex := newStagerExecutor(emptyIDStager{})
|
ex := newStagerExecutor(emptyIDStager{})
|
||||||
out := ex.stageInputFiles(context.Background(), "r", "a",
|
out, _ := ex.stageInputFiles(context.Background(), "r", "a",
|
||||||
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p")
|
[]tool.InputFile{{Name: "x.bin", Data: []byte("z")}}, "p")
|
||||||
if out != "p" {
|
if out != "p" {
|
||||||
t.Errorf("empty file_id should drop the file, got %q", out)
|
t.Errorf("empty file_id should drop the file, got %q", out)
|
||||||
|
|||||||
@@ -54,6 +54,11 @@ type Ports struct {
|
|||||||
// loader tool. nil = SkillPacks are inert. The executus/skillpack battery
|
// loader tool. nil = SkillPacks are inert. The executus/skillpack battery
|
||||||
// ships a default impl (skillpack.Activator).
|
// ships a default impl (skillpack.Activator).
|
||||||
SkillPacks SkillPackActivator
|
SkillPacks SkillPackActivator
|
||||||
|
// FinalGuard is consulted when a single-loop run is about to finalize
|
||||||
|
// successfully; a non-empty nudge sends the loop back for one bounded extra
|
||||||
|
// round (the announce-then-stop undelivered-artifact check). nil = runs
|
||||||
|
// finalize unguarded. See FinalGuard in final_guard.go.
|
||||||
|
FinalGuard FinalGuard
|
||||||
}
|
}
|
||||||
|
|
||||||
// SkillPackActivator resolves an agent's subscribed skill-pack names for a run
|
// SkillPackActivator resolves an agent's subscribed skill-pack names for a run
|
||||||
|
|||||||
Reference in New Issue
Block a user