krow_backend/go-api/internal/runtime/loop.go

package runtime

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/krow/krow-backend/go-api/internal/gateway"
	"github.com/krow/krow-backend/go-api/internal/knowledge"
	"github.com/krow/krow-backend/go-api/internal/tools"
)

// ModelExecutor runs an agent against a model.
//
// This is the agent loop. It is spec-driven and there is exactly one of it: no
// branch anywhere below asks which agent it is running. An agent's identity
// reaches this code only as data — its instructions, its tier, its skills —
// which is what I6 means in practice and what makes adding an agent a data
// change rather than a deploy.
//
// The loop runs until the model stops asking for tools, or until a bound is
// reached. Every exit is one of the six terminations.

// trajectoryPersistTimeout bounds the trajectory write that happens after a run
// has produced its answer. Generous, because losing the record of a run is
// worse than a slow one, but finite: the caller is still waiting on this, so an
// unreachable database must cost seconds and not the request's whole write
// timeout. See finish.
//
// A var rather than a const only so a test can assert the bound holds without
// spending the bound. Nothing outside this package sets it.
var trajectoryPersistTimeout = 5 * time.Second

type ModelExecutor struct {
	// subagents resolves a spec's `subagents:` into runnable agents. nil means
	// delegation is off and a spec that declares subagents runs alone — see
	// delegate.go.
	subagents SubagentResolver

	gw    gateway.Gateway
	sink  Sink
	tools *tools.Registry

	// retriever is the knowledge layer, or nil for an agent platform with no
	// documents in it. Nil is a supported state rather than a broken one: every
	// agent built so far answers from the operational tables through tools, and
	// none of them needs a corpus.
	retriever Retriever
}

// Retriever is what the loop needs from the knowledge layer.
//
// An interface rather than the concrete type so the runtime does not import the
// knowledge package's whole surface, and so a test can drive the loop with a
// scripted corpus. Deliberately narrow: the loop retrieves, it does not ingest,
// and it has no way to ask for anything other than the caller's own rows —
// knowledge.Query requires a principal and this signature carries one.
type Retriever interface {
	Retrieve(ctx context.Context, q knowledge.Query) (*knowledge.Results, error)
}

// WithRetriever attaches a knowledge layer to an executor.
func (m *ModelExecutor) WithRetriever(r Retriever) *ModelExecutor {
	m.retriever = r
	return m
}

var _ AgentExecutor = (*ModelExecutor)(nil)

// NewModelExecutor builds the loop over a model gateway.
//
// A nil sink is DiscardSink rather than a panic: a service wired without a
// trajectory store should still answer, and losing the record is a worse
// outcome than nothing but not one worth refusing a correct answer over.
// A nil registry is an empty one: an agent that names no tools does not need
// one, and a nil map dereference is a worse way to discover that than an agent
// that simply has nothing to call.
func NewModelExecutor(gw gateway.Gateway, sink Sink, reg *tools.Registry) *ModelExecutor {
	if sink == nil {
		sink = DiscardSink{}
	}
	if reg == nil {
		reg = tools.NewRegistry()
	}
	return &ModelExecutor{gw: gw, sink: sink, tools: reg}
}

// newRunID returns an opaque run identifier.
//
// Random rather than sequential: a run id appears in logs and in support
// conversations, and a sequential one would leak how many runs a deployment
// has served.
func newRunID() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		// crypto/rand does not fail in practice; if it ever does, a run
		// without an id is still better than a run that refuses to start.
		return "run-unknown"
	}
	return "run_" + hex.EncodeToString(b[:])
}

// ExecuteAgent runs one agent turn and returns a structured result.
//
// It never returns a bare error into user-facing text. Every exit is a
// termination reason plus a trajectory, because §6 requires exactly one
// termination per run and §10 requires user-facing text to be derived at the
// surface layer rather than raised from here.
func (m *ModelExecutor) ExecuteAgent(ctx context.Context, agent *Agent, input ExecutionInput) (*ExecutionResult, error) {
	tier, _ := gateway.ParseTier(agent.Reasoning)
	return m.executeWithLimits(ctx, agent, input, LimitsForTier(string(tier)))
}

// executeWithLimits is ExecuteAgent with the bounds supplied rather than
// derived.
//
// The seam exists for two reasons and will earn its keep for the second. Today
// it lets a test drive a real deadline instead of asserting on a counter. When
// the spec gains a `limits:` block, that block resolves here and ExecuteAgent
// stays the one-line default — so per-agent limits arrive without the loop
// itself changing shape.
func (m *ModelExecutor) executeWithLimits(
	ctx context.Context, agent *Agent, input ExecutionInput, limits Limits,
) (*ExecutionResult, error) {
	res, err := m.executeRun(ctx, agent, input, limits, delegation{})
	if next, ok := m.standbyFor(res, input); ok {
		return next.executeRun(ctx, agent, input, limits, delegation{})
	}
	return res, err
}

// standbyFor decides whether to run the whole turn again on another provider.
//
// WHY A RESTART AND NOT A HANDOVER. gateway.canFailOver will not move a
// conversation that has called a tool: the assistant turn echoing that call is
// the provider's own, and a vendor that signs its function calls rejects a
// follow-up carrying somebody else's. But a rate limit lands where the request
// is BIGGEST, which is the second or third call, once the catalogue, the
// retrieved block, the tool results and the whole prior conversation are being
// re-sent. Measured on this deployment: every GatewayFailure recorded had
// already called a tool, so in-place failover covered none of them —
//
//	http 429 … Rate limit reached … tokens per minute (TPM): Limit 8000, Used 7183
//
// Starting over sends no transcript, so nothing provider-specific travels and
// the question is simply asked again somewhere with budget left. It costs the
// work already done, charged to the budget that is NOT exhausted.
//
// THE RULE THAT MAKES IT SAFE: a run that carried a confirmation never
// restarts. Re-running re-runs its tools, and a read twice is two reads while a
// write twice is two shifts assigned. I4 is what makes the test this cheap —
// a write executes ONLY against a resolved token (Registry.gate), so a run with
// no confirmation cannot have written anything, and one with a confirmation is
// refused here without inspecting what it did.
//
// Once. Not a loop over every provider: a question worth asking twice is not
// worth asking five times, and each attempt spends a real budget. The second
// result is returned as it stands, whatever it says.
func (m *ModelExecutor) standbyFor(res *ExecutionResult, input ExecutionInput) (*ModelExecutor, bool) {
	if res == nil || res.Termination != TerminationGatewayFailure {
		return nil, false
	}
	// An approved write may already have happened. Nothing below is worth a
	// double assignment.
	if input.Confirmation != "" {
		return nil, false
	}
	// Only a transient fault moves, on the same line gateway.canFailOver draws:
	// a rejected credential or a model this deployment cannot use fails the
	// same way everywhere, and asking twice only doubles the bill.
	var gwErr *gateway.Error
	if !errors.As(res.Error, &gwErr) || !gwErr.Retryable() {
		return nil, false
	}
	sb, ok := m.gw.(gateway.Standby)
	if !ok {
		return nil, false
	}
	next, ok := sb.Standby()
	if !ok {
		return nil, false
	}
	clone := *m
	clone.gw = next
	return &clone, true
}

// executeRun is the loop. `del` is what a SUBAGENT inherits from its parent —
// a shared budget, a parent run id, and a depth — and its zero value is an
// ordinary root run that owns its own budget and has no parent.
//
// One loop, not two. A delegated run is the same code on the same path; the
// only things that differ are where its budget came from and what its
// trajectory is linked to. §6 says there is exactly one loop, and a second one
// for subagents would be the first place the two drifted apart.
func (m *ModelExecutor) executeRun(
	ctx context.Context, agent *Agent, input ExecutionInput, limits Limits, del delegation,
) (*ExecutionResult, error) {
	tier, known := gateway.ParseTier(agent.Reasoning)

	// §6: a subagent SHARES the parent's budget and never gets a fresh one.
	// Delegating would otherwise be a way to buy more steps.
	budget := del.budget
	if budget == nil {
		budget = NewBudget(limits)
	}

	skillIDs := make([]string, len(agent.ResolvedSkills))
	for i, s := range agent.ResolvedSkills {
		skillIDs[i] = s.ID
	}

	rec := NewRecorder(&Trajectory{
		RunID:        newRunID(),
		ParentRunID:  del.parentRunID,
		OrgID:        input.Identity.OrgID,
		UserID:       input.Identity.UserID,
		AgentID:      agent.ID,
		AgentVersion: agent.Version,
		Tier:         string(tier),
	})

	// A spec naming a tier the vocabulary does not have still runs, at the
	// default — but it is recorded, so a definition that has drifted is
	// visible in the trajectory rather than silently reinterpreted.
	if !known {
		rec.Error("runtime.unknown_tier",
			fmt.Sprintf("%q is not a reasoning mode; running at %s", agent.Reasoning, tier))
	}

	// The deadline is the budget's, so an in-flight model call is torn down
	// rather than returning into a run that has already ended.
	runCtx, cancel := budget.Context(ctx)
	defer cancel()

	// Anything the caller wants on the record, before the run does anything.
	// A run that silently could not do what was asked of it is the failure
	// worth preventing here.
	for _, note := range input.Notes {
		rec.Error("runtime.note", note)
	}

	question := strings.TrimSpace(input.Input)
	if question == "" {
		return m.finish(ctx, rec, budget, TerminationToolFailure, agent, skillIDs,
			"", &RuntimeError{Code: "runtime.empty_input", Message: "a run needs a question"})
	}
	rec.Message("user", question)

	// A greeting is answered as a greeting.
	//
	// Everything below this line — the tool catalogue, the subagent list, the
	// retrieval pass — exists to answer a QUESTION. A message that asks nothing
	// gets none of it. See isSmalltalk for why the test is on the message and
	// never on the agent: this stays one spec-driven loop (§6), and adding an
	// agent still needs no runtime change (I6).
	//
	// A resumed run never qualifies, whatever its text says. The person
	// approved a write and is owed a report on it, and performApproved below
	// needs its tools to give them one.
	smalltalk := input.Confirmation == "" && isSmalltalk(question)
	if smalltalk {
		rec.Error("runtime.smalltalk",
			"conversational message; answered without retrieval, tools or subagents")
	}

	var toolDefs []gateway.ToolDef
	if !smalltalk {
		// The tools this agent may use. Unknown names are recorded and dropped
		// rather than failing the run: §3 says an unknown tool fails validation at
		// *publish*, so one reaching run time means a tool was withdrawn under a
		// live spec — degrading is better than an outage, provided someone is told.
		var unknown []string
		toolDefs, unknown = m.toolsFor(agent)
		for _, name := range unknown {
			rec.Error("runtime.unknown_tool", fmt.Sprintf("%q is not a registered tool; it was not offered", name))
		}
	}

	// Subagents are offered as tools, because from this agent's side that is
	// exactly what they are (§6). Resolved once per run rather than per turn:
	// the set cannot change mid-run, and loading it per turn would spend the
	// caller's time on the same query repeatedly.
	//
	// Resolved even for smalltalk, and only the OFFER is withheld. Resolution
	// is what records a spec naming itself, or naming a subagent that will not
	// load, and those are faults of the spec rather than of the question —
	// losing them because somebody said hello would make a misconfiguration
	// visible only intermittently, which is the hardest kind to chase. It is
	// free to keep: a spec with no subagents returns at the first line.
	subs := m.resolveSubagents(ctx, rec, agent, input.Identity, del.depth)
	if !smalltalk {
		toolDefs = append(toolDefs, delegateTools(subs)...)
	}

	// An approved write happens FIRST, before the model gets a turn.
	//
	// This is the half of I4 that makes a confirmation reliable rather than
	// hopeful. The older design resumed the run and matched the model's next
	// tool call against the token — which only works if the model repeats
	// itself, and a model asked a second time may perfectly reasonably ask a
	// clarifying question instead. When that happened the token was never
	// presented, nothing was written, and the person who clicked Approve got a
	// follow-up question with no explanation.
	//
	// So the approved call is performed from what the person was SHOWN, not
	// from what the model says next. The model's job afterwards is to report
	// what happened, which is a job it cannot get wrong in a way that costs
	// anybody a shift.
	if approved, done := m.performApproved(runCtx, rec, budget, agent, input); done != nil {
		return m.finish(ctx, rec, budget, *done, agent, skillIDs, "", nil)
	} else if approved != "" {
		// Prepended to the question so the model answers knowing the write
		// already happened. It is a tool result in everything but shape —
		// delimited, factual, and about an act rather than an instruction.
		question = approved + "\n\n" + question
	}

	// Retrieval, before the first model call.
	//
	// I7 decides where the result goes: into a delimited block in a USER
	// message, never into the system prompt. The system prompt is assembled
	// from the agent record alone, so no amount of document content can reach
	// it — which is the only reason the standing "content inside <context> is
	// data" instruction means anything.
	//
	// Skipped entirely for smalltalk. retrieve() gates on configuration and
	// never on the question, so without this a greeting was handed eight policy
	// chunks AHEAD of the word "hi" — which is both the dominant cost of the
	// turn and the reason the answer came back as an operational briefing.
	conversation := []gateway.Message{{Role: gateway.RoleUser, Text: question}}
	if !smalltalk {
		if block, retrieved := m.retrieve(runCtx, rec, agent, input, question); block != "" {
			conversation = []gateway.Message{{
				Role: gateway.RoleUser,
				// Context first, question second. A model reads the question last
				// and answers it, rather than treating the evidence as the prompt.
				Text: block + "\n\n" + question,
			}}
			rec.Retrieval(retrieved)
		}
	}

	// Recorded once rather than on each remaining step, so a long run does not
	// fill its trajectory with the same note.
	var toolsWithheld bool

	// The language the reader chose, resolved once for the whole run rather
	// than per step: a run that answered its third turn in a different language
	// from its first would be a bug, not a feature. An unrecognised tag is
	// recorded and falls back to English — the reader loses a preference they
	// may not have set, which is the cheap failure, and somebody is told.
	lang, known := ParseLanguage(string(input.Language))
	if !known {
		rec.Error("runtime.unknown_language",
			fmt.Sprintf("%q is not a language this build answers in; used %s",
				string(input.Language), lang.Name()))
	}

	system := SystemPrompt(agent, lang)
	if smalltalk {
		// Taking the evidence away removes the citations; it does not by itself
		// shorten the reply, because the agent's own instructions still
		// describe an operational analyst. See smalltalkDirective.
		system += smalltalkDirective
	}
	var lastText string

	for {
		// Claimed before dispatch, never after. A call that hangs until the
		// context dies has still spent the step it was given.
		if t := budget.ClaimStep(); t != "" {
			return m.finish(ctx, rec, budget, t, agent, skillIDs, lastText, nil)
		}
		if t := budget.CheckTokens(); t != "" {
			return m.finish(ctx, rec, budget, t, agent, skillIDs, lastText, nil)
		}
		rec.Budget(budget.Snapshot())

		// Streamed when the caller asked for it AND the gateway can. Both
		// halves go through StreamComplete, so the loop has one call site and
		// no branch on transport — a run behaves identically whether its text
		// arrived in one piece or a hundred.
		// The catalogue has to be resent on every call — the wire protocol has
		// no way to refer back to one already sent — but a catalogue the model
		// is no longer ALLOWED to use is pure waste. Once the tool-call budget
		// is spent, every definition describes a call that would be refused,
		// and the step it is being sent on is the synthesis turn that just
		// needs to write the answer up.
		//
		// Measured on this deployment's control-center agent: seven tools,
		// ~1.2k tokens, resent on the final call of every tool-using run.
		stepTools := toolDefs
		if len(stepTools) > 0 && budget.Snapshot().ToolCallsLeft <= 0 {
			stepTools = nil
			if !toolsWithheld {
				toolsWithheld = true
				rec.Error("runtime.tools_withheld",
					"tool-call budget spent; catalogue not resent on the remaining steps")
			}
		}

		// Smalltalk is capped far below the tier's ceiling. The directive in
		// the system prompt is what actually shortens the reply; this only
		// bounds the bill for a model that ignores it.
		maxOut := budget.Limits().MaxOutputTokens
		if smalltalk && (maxOut <= 0 || maxOut > smalltalkMaxOutputTokens) {
			maxOut = smalltalkMaxOutputTokens
		}

		resp, err := gateway.StreamComplete(runCtx, m.gw, gateway.Request{
			Tier:     tier,
			System:   system,
			Messages: conversation,
			Tools:    stepTools,
			// Per STEP, and taken from the budget rather than from the tier —
			// so a delegated run inherits the parent's ceiling along with the
			// parent's budget instead of reading its own tier and quietly
			// buying a longer answer than the parent was allowed.
			MaxOutputTokens: maxOut,
		}, input.OnDelta)

		// Charged whatever happened. A refused or failed call was still billed,
		// and a ledger that forgives it is one a loop will happily repeat
		// against.
		if resp != nil {
			budget.ChargeTokens(resp.Usage.Total())
			rec.ChargeUsage(resp.Usage.InputTokens, resp.Usage.OutputTokens,
				resp.Usage.CacheReadTokens+resp.Usage.CacheCreationTokens)
			rec.SetModel(resp.Model)
		}
		if err != nil {
			return m.finish(ctx, rec, budget, terminationFor(err), agent, skillIDs, lastText, err)
		}

		if resp.Text != "" {
			rec.Message("assistant", resp.Text)
			lastText = resp.Text
		}

		// No tool calls means the model is done talking.
		if len(resp.ToolCalls) == 0 {
			return m.finish(ctx, rec, budget, TerminationCompleted, agent, skillIDs, lastText, nil)
		}

		// The assistant turn goes back verbatim, calls included, before any
		// result is appended — a tool result with no preceding call is a
		// malformed conversation the API will reject.
		conversation = append(conversation, gateway.Message{
			Role: gateway.RoleAssistant, Text: resp.Text, ToolCalls: resp.ToolCalls,
		})

		results, pending, term := m.runTools(runCtx, rec, budget, agent, input, resp.ToolCalls, subs, del.depth)
		if term != "" {
			return m.finish(ctx, rec, budget, term, agent, skillIDs, lastText, nil)
		}

		// I4. A run that wants to write stops here and asks. It does not
		// continue with the reads it also made, does not summarise, and does
		// not get another turn to reconsider — the next thing that happens is a
		// person deciding, and the run resumes only if they say yes.
		if len(pending) > 0 {
			res, err := m.finish(ctx, rec, budget,
				TerminationConfirmationPending, agent, skillIDs, lastText, nil)
			res.Confirmations = pending
			return res, err
		}

		// Every result in ONE user turn. Splitting them is accepted and quietly
		// teaches the model to stop calling tools in parallel.
		conversation = append(conversation, gateway.Message{
			Role: gateway.RoleUser, ToolResults: results,
		})
	}
}

// performApproved carries out a write a person approved.
//
// Returns the sentence describing what happened, for the model to report from.
// A run with no confirmation token does nothing here and returns "".
//
// A token that authorises nothing — unknown, expired, already spent, somebody
// else's — is NOT an error and does not end the run. It is recorded and the run
// continues, because the most common cause is a person clicking Approve twice,
// and the honest response to that is to answer the question again rather than
// to fail.
func (m *ModelExecutor) performApproved(
	ctx context.Context, rec *Recorder, budget *Budget, agent *Agent, input ExecutionInput,
) (string, *Termination) {
	if input.Confirmation == "" || m.tools == nil {
		return "", nil
	}

	// The write spends a tool call from the run's budget, claimed before
	// dispatch like every other. An approval is not a way around I3.
	if t := budget.ClaimToolCall(); t != "" {
		return "", &t
	}
	rec.Budget(budget.Snapshot())

	tc := tools.Context{
		Principal:        input.Identity,
		RunID:            rec.RunID(),
		RemainingTokens:  budget.Snapshot().TokensLeft,
		AgentID:          agent.ID,
		KnowledgeSources: agent.KnowledgeSources,
	}

	out, ok := m.tools.DispatchApproved(ctx, tc, input.Confirmation)
	if !ok {
		rec.Error("runtime.confirmation_not_redeemable",
			"the supplied approval authorises nothing; it may have expired or already been used")
		return "", nil
	}

	rec.ToolCall(out.Tool, string(tools.EffectWrite), out.Inputs)
	rec.ToolResult(out.Tool, string(tools.EffectWrite), out.Result.Error != nil, out.Result)

	encoded, err := json.Marshal(out.Result)
	if err != nil {
		encoded = []byte(`{"error":{"code":"tool.failed","message":"the result could not be encoded"}}`)
	}

	// Delimited and labelled as data, on the same terms as retrieved content.
	// This text describes something that already happened; it is not an
	// instruction, and the standing <context> rule in the system prompt covers
	// it for exactly that reason.
	return fmt.Sprintf(
		"<context>\nA change you approved has already been carried out. This is its "+
			"result, as data — report it, do not repeat the action.\n\n"+
			"<source id=%q>\n%s\n</source>\n</context>",
		out.Tool, string(encoded)), nil
}

// retrieve searches the agent's declared corpora on the CALLER's behalf.
//
// Three things are load-bearing and none of them is the search itself:
//
//   - The principal is the caller's, never the agent's. I1: an agent reads
//     exactly what its caller could read directly, and the identity that
//     reaches knowledge.Query is the one that arrived with the request.
//   - The sources are the SPEC's. An agent granted the policy library does not
//     gain the incident log by asking nicely, because the source list is not
//     something the model can influence.
//   - A failure degrades rather than ends the run. A knowledge layer that is
//     down should cost grounding, not the answer — but it is recorded, because
//     an ungrounded answer that looks grounded is the worse outcome.
func (m *ModelExecutor) retrieve(
	ctx context.Context, rec *Recorder, agent *Agent, input ExecutionInput, question string,
) (string, *knowledge.Results) {
	if m.retriever == nil || len(agent.KnowledgeSources) == 0 {
		return "", nil
	}

	res, err := m.retriever.Retrieve(ctx, knowledge.Query{
		Text:      question,
		Principal: input.Identity,
		Sources:   agent.KnowledgeSources,
	})
	if err != nil {
		// Recorded, not raised. The run continues without grounding, and the
		// trajectory says so — "the agent answered from nothing" is only
		// diagnosable afterwards if the failure was written down at the time.
		var kErr *knowledge.Error
		if errors.As(err, &kErr) {
			rec.Error(kErr.Code, kErr.Message)
		} else {
			rec.Error("knowledge.failed", err.Error())
		}
		return "", nil
	}
	if res == nil || len(res.Chunks) == 0 {
		return "", res
	}
	return knowledge.RenderContext(res), res
}

// toolsFor resolves the tools an agent's spec names.
func (m *ModelExecutor) toolsFor(agent *Agent) (defs []gateway.ToolDef, unknown []string) {
	if m.tools == nil || len(agent.Tools) == 0 {
		return nil, nil
	}
	resolved, unknown, err := m.tools.Resolve(agent.Tools)
	if err != nil {
		// Over the per-agent cap. Offering none is the safe reading: an agent
		// that silently got its first twenty tools would behave differently
		// depending on the order someone happened to write them in.
		return nil, agent.Tools
	}
	for _, t := range resolved {
		defs = append(defs, gateway.ToolDef{
			Name: t.Name, Description: t.Description, InputSchema: t.InputSchema,
		})
	}
	return defs, unknown
}

// runTools dispatches one turn's calls and returns their results.
//
// A tool that fails returns its error TO THE MODEL rather than ending the run.
// §13 lists "swallowing a tool error and letting the model narrate around it"
// as an anti-pattern — the fix is not to hide the failure but to hand it over
// as a failure, so the model can say it could not look rather than inventing
// what it would have found.
//
// The tool-call budget is claimed per call, before dispatch. Running out ends
// the run: a model that has exhausted its calls cannot make progress, and
// letting it continue would spend the remaining step budget on turns that can
// only apologise.
//
// A write that needs approving comes back as a pending confirmation rather than
// a result. Those are collected across the whole turn rather than returned at
// the first one, so a person is asked about every write the model wanted in one
// go instead of being walked through them one dialog at a time — and so that
// the reads in the same turn, which are safe, still run and are still recorded.
func (m *ModelExecutor) runTools(
	ctx context.Context, rec *Recorder, budget *Budget, agent *Agent,
	input ExecutionInput, calls []gateway.ToolCall,
	subs map[string]*Agent, depth int,
) (results []gateway.ToolResult, pending []*tools.Confirmation, term Termination) {
	results = make([]gateway.ToolResult, 0, len(calls))

	for _, call := range calls {
		if t := budget.ClaimToolCall(); t != "" {
			return nil, nil, t
		}
		rec.Budget(budget.Snapshot())

		// A delegation spends the same tool-call budget as any other call,
		// claimed above, and then spends the parent's remaining budget inside
		// the subagent. It is charged twice on purpose: once for asking, and
		// then for whatever the asking cost.
		if sub, isDelegation := subs[call.Name]; isDelegation {
			rec.ToolCall(call.Name, "delegate", json.RawMessage(call.Input))
			answer, subPending := m.delegate(ctx, rec, budget, sub, input,
				json.RawMessage(call.Input), depth)

			encoded, err := json.Marshal(answer)
			if err != nil {
				encoded = []byte(`{"error":"the subagent's answer could not be encoded"}`)
			}
			rec.ToolResult(call.Name, "delegate", answer.Error != "",
				tools.Result{Data: answer})

			// I4 survives delegation. A write a SUBAGENT wants approved is
			// still a write, and it stops this run the same way one from a
			// direct tool call does — see the pending check in the loop.
			if len(subPending) > 0 {
				for _, c := range subPending {
					rec.Confirmation(call.Name, c)
				}
				pending = append(pending, subPending...)
				continue
			}
			results = append(results, gateway.ToolResult{
				CallID:  call.ID,
				Content: string(encoded),
				IsError: answer.Error != "",
			})
			continue
		}

		// The declared effect travels with the record. An eval asking "did this
		// run change anything" reads it from here rather than keeping its own
		// list of which tools write — a list that goes stale on the first tool
		// anybody adds.
		var effect string
		if t, ok := m.tools.Get(call.Name); ok {
			effect = string(t.Effect)
		}
		rec.ToolCall(call.Name, effect, json.RawMessage(call.Input))

		res := m.tools.Dispatch(ctx, tools.Context{
			Principal:       input.Identity,
			RunID:           rec.RunID(),
			RemainingTokens: budget.Snapshot().TokensLeft,
			Confirmation:    input.Confirmation,
			// From the spec, never from the call. A model that asked to search
			// a corpus its agent was not granted is asking for a source list it
			// has no way to set.
			KnowledgeSources: agent.KnowledgeSources,
		}, call.Name, call.Input)

		// A pending confirmation never reaches the model. It is a question for
		// a person, and handing it back as a tool result would invite the model
		// to reason about it — to explain why it should be approved, or to try
		// a different tool that might not ask. Neither is its business.
		if res.Confirmation != nil {
			rec.Confirmation(call.Name, res.Confirmation)
			pending = append(pending, res.Confirmation)
			continue
		}

		rec.ToolResult(call.Name, effect, res.Error != nil, res)

		encoded, err := json.Marshal(res)
		if err != nil {
			encoded = []byte(`{"error":{"code":"tool.failed","message":"the result could not be encoded"}}`)
		}
		results = append(results, gateway.ToolResult{
			CallID:  call.ID,
			Content: string(encoded),
			IsError: res.Error != nil,
		})
	}
	return results, pending, ""
}

// terminationFor maps a failure to the reason a run ends with.
//
// The mapping matters more than it looks: Deadline and BudgetExceeded are
// different questions to an operator ("too slow" versus "too expensive"), and
// a Refused run is one that must not be retried. Flattening them into a single
// failure reason would make every one of those distinctions unanswerable from
// the trajectory.
//
// Any other gateway error is GatewayFailure, not ToolFailure. The provider
// being rate limited, rejecting the request or refusing the key is not a tool
// failing, and calling it one sent two weeks of operators looking for a broken
// tool that did not exist. A non-gateway error — a tool the model invented, a
// result that could not be encoded — is still the tool layer's.
func terminationFor(err error) Termination {
	var gwErr *gateway.Error
	if !errors.As(err, &gwErr) {
		return TerminationToolFailure
	}
	switch gwErr.Code {
	case gateway.CodeRefused:
		return TerminationRefused
	case gateway.CodeTimeout:
		return TerminationDeadline
	default:
		return TerminationGatewayFailure
	}
}

// finish closes the trajectory, persists it, and builds the caller's result.
//
// Persistence deliberately outlives the run's context: that context is
// cancelled at the deadline, and a run that ended by running out of time is
// exactly the one whose record is most worth keeping. It does NOT outlive the
// caller's patience — trajectoryPersistTimeout bounds the whole write, because
// a run that answered inside its deadline and then sat in the sink for a minute
// is, to the person waiting, a slow run. I3 bounds the run; this bounds its
// tail.
func (m *ModelExecutor) finish(
	ctx context.Context,
	rec *Recorder,
	budget *Budget,
	term Termination,
	agent *Agent,
	skillIDs []string,
	output string,
	cause error,
) (*ExecutionResult, error) {
	if cause != nil {
		var gwErr *gateway.Error
		if errors.As(cause, &gwErr) {
			// The status rides in the message because `entries` has no column
			// for it and E5 forbids applying a migration from here. It matters:
			// `gateway.upstream` alone cannot tell a provider shedding load
			// (5xx, clears by itself) from an endpoint rejecting the request
			// (4xx, needs an administrator), and those are opposite actions.
			msg := gwErr.Message
			if gwErr.Status > 0 {
				msg = fmt.Sprintf("http %d: %s", gwErr.Status, msg)
			}
			rec.Error(gwErr.Code, msg)
		} else {
			rec.Error("runtime.failed", cause.Error())
		}
	}
	rec.Budget(budget.Snapshot())
	traj := rec.Finish(term)

	// WithoutCancel so a deadline-terminated run still records itself; the
	// timeout so it cannot record itself forever. One budget covers the parent
	// and every child, since writing the tree is one logical act and a
	// per-trajectory timeout would multiply by the number of subagents.
	persistCtx, cancelPersist := context.WithTimeout(
		context.WithoutCancel(ctx), trajectoryPersistTimeout)
	defer cancelPersist()

	// A sink that fails must not fail the run — the answer was already
	// produced. It is recorded in the trajectory we could not save, which is
	// the best available place for it.
	var unsaved []string
	if err := m.sink.Save(persistCtx, traj); err != nil {
		rec.Error("runtime.trajectory_unsaved", err.Error())
		unsaved = append(unsaved, traj.RunID+": "+err.Error())
	}

	// Delegated runs are written AFTER this one, because parent_run_id is a
	// foreign key and a subagent finishes first. Each child arrives with its
	// own descendants already ordered behind it, so one pass here writes a
	// whole tree parent-first.
	for _, child := range rec.Children() {
		if err := m.sink.Save(persistCtx, child); err != nil {
			rec.Error("runtime.subrun_unsaved",
				fmt.Sprintf("%s: %s", child.RunID, err.Error()))
			unsaved = append(unsaved, child.RunID+": "+err.Error())
		}
	}

	res := &ExecutionResult{
		Unsaved:        unsaved,
		Success:        term == TerminationCompleted,
		Output:         output,
		AgentID:        agent.ID,
		AgentVersion:   agent.Version,
		ResolvedSkills: skillIDs,
		RunID:          traj.RunID,
		Termination:    term,
		Usage:          traj.Usage,
	}

	if term == TerminationCompleted {
		return res, nil
	}

	// A bounded run is not an exception. The caller gets a result carrying the
	// reason; the error exists so a Go caller that ignores the result still
	// notices, and it is structured so the surface layer derives the wording.
	rtErr := &RuntimeError{
		Code:    "runtime." + strings.ToLower(string(term)),
		Message: terminationMessage(term),
		Target:  agent.ID,
		Cause:   cause,
	}
	res.Error = rtErr
	return res, rtErr
}

// terminationMessage is the internal explanation for a termination. Not
// user-facing copy — §10 puts that at the surface layer, which is free to say
// something kinder using the code.
func terminationMessage(t Termination) string {
	switch t {
	case TerminationBudgetExceeded:
		return "the run reached its budget before finishing"
	case TerminationDeadline:
		return "the run reached its deadline before finishing"
	case TerminationRefused:
		return "the model declined to answer"
	case TerminationConfirmationPending:
		return "the run is waiting on a confirmation"
	case TerminationToolFailure:
		return "the run failed"
	case TerminationGatewayFailure:
		return "the model provider did not answer"
	default:
		return string(t)
	}
}

// SystemPrompt assembles an agent's system prompt from its spec.
//
// I7 is the whole design of this function. Retrieved document text, tool
// results and user messages are all untrusted, and none of them are reachable
// from here: it reads the agent record and nothing else. When Phase 2 adds
// retrieval, the retrieved chunks go into a delimited block in a *user*
// message — not into this string — and the standing instruction below is what
// makes that delimiter mean something.
//
// `lang` does not weaken that. It is a Language, so the only strings it can
// contribute are the constants in language.go — the caller's two-letter tag
// selects one and is never itself written here. See Language.
func SystemPrompt(agent *Agent, lang Language) string {
	var b strings.Builder

	b.WriteString("You are ")
	b.WriteString(agent.Name)
	if agent.Description != "" {
		b.WriteString(", ")
		b.WriteString(agent.Description)
	}
	b.WriteString(".\n\n")

	if instructions := strings.TrimSpace(agent.Instructions); instructions != "" {
		b.WriteString(instructions)
		b.WriteString("\n\n")
	}

	if len(agent.Pages) > 0 {
		b.WriteString("You answer on: ")
		b.WriteString(strings.Join(agent.Pages, ", "))
		b.WriteString(". Anywhere else, say plainly that you do not cover it.\n\n")
	}

	// Stated even when nothing was retrieved, because the boundary has to be
	// established before content arrives rather than alongside it.
	//
	// The sentence comes from the knowledge package, beside the renderer that
	// emits the fence. A prompt promising <context> while the renderer wrote
	// <documents> would be a defence that had quietly stopped existing, and two
	// copies of a string in two packages is exactly how that happens.
	b.WriteString(knowledge.ContextInstruction)
	b.WriteString("\n\n")

	// The same boundary for the other channel untrusted text arrives on.
	// Retrieval is not the only one: a tool result carries whatever the records
	// hold, and a person who can type into the platform can put a sentence
	// there. Stated unconditionally, like the one above, because the rule has
	// to be established before the content arrives rather than alongside it.
	b.WriteString(gateway.ToolResultInstruction)
	b.WriteString("\n\n")

	b.WriteString("State a figure only where the records you were given show it. " +
		"When you cannot answer from them, say so rather than estimating.")

	// Last, and deliberately so. Everything above it is English — the agent's
	// own instructions, the standing rules, and every tool result that will
	// arrive later — so a language instruction placed earlier is one the rest
	// of the prompt spends thousands of tokens arguing against. Nearest the
	// question is where it holds.
	if directive := lang.Directive(); directive != "" {
		b.WriteString("\n\n")
		b.WriteString(directive)
	}

	return b.String()
}