package builtin

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"

	"github.com/denn-gubsky/loomcycle/internal/config"
	"github.com/denn-gubsky/loomcycle/internal/providers"
	"github.com/denn-gubsky/loomcycle/internal/tools"
)

// SubAgentRunner runs a named sub-agent with a given user prompt and
// returns the sub-agent's final assistant text. Implementations are
// expected to:
//
//   - Reject unknown agent names (operator-controlled allowlist:
//     cfg.Agents).
//   - Apply the sub-agent's own Tools as the security floor; the
//     parent's tool set does NOT widen the child's. Parent and child are
//     both operator-vetted definitions, so each agent's declared
//     allow-list is authoritative for itself.
//   - When defID is non-empty, resolve the sub-agent against the
//     agent_defs row at that id (v0.8.5 substrate). The row's Name MUST
//     match the requested name — cross-name pinning is refused. The
//     def's mutable fields (system_prompt, tools, model, tier,
//     etc.) override the static cfg.Agents entry for this sub-run. The
//     pinned defID is also persisted on the sub-run row so evaluations
//     denormalise correctly.
//   - Persist the sub-agent's run as a separate session in the Store so
//     transcripts are replayable independently of the parent.
//   - Carry the parent's context.Context (so cancellation, deadlines,
//     and the loomcycle agent-depth value all propagate).
type SubAgentRunner func(ctx context.Context, name string, prompt string, defID string) (output string, err error)

// SubAgentRunnerDetailed is SubAgentRunner that ALSO surfaces the child's run
// row id (RFC X Phase 3). parallel_spawn uses it (via AgentTool.RunDetailed)
// to record each child in the spawn ledger so a snapshotted+restored fan-out
// parent can re-find and re-collect its children. Additive: when RunDetailed
// is nil, parallel_spawn falls back to Run (no run_id, no ledger) — existing
// callers + tests are unaffected.
type SubAgentRunnerDetailed func(ctx context.Context, name string, prompt string, defID string) (output string, runID string, err error)

// OpenChildRunner starts a PERSISTENT interactive sub-run (RFC BK), runs its
// first turn, parks it at awaiting_input, and returns the child's run id, the
// first turn's assistant output, and the child's state ("awaiting_input" when
// parked, "completed"/"failed" if the first turn already ended the run). The
// child stays RESIDENT — reachable via SendChildRunner / CloseChildRunner —
// until it is closed, idle-reaped, or its parent run's tree tears down. Unlike
// SubAgentRunner (fire-and-return), the child keeps its conversation + any
// resources it holds (e.g. a warm sandbox container) between sends.
//
// idleTTLSeconds overrides the operator-default idle reap window for THIS child
// (0 = use the default). The parent owns the child's lifecycle.
type OpenChildRunner func(ctx context.Context, name, prompt, defID string, idleTTLSeconds int) (childRunID, output, state string, err error)

// SendChildRunner injects the next instruction into a resident child (RFC BK)
// and waits for that turn's result. timeoutMs bounds the wait: 0 blocks until the
// child re-parks at awaiting_input (or terminates), matching P1; >0 returns
// early with state "running" + the partial output-so-far if the turn is still in
// flight after timeoutMs. The caller must own the child (same tenant); an unknown
// or unowned child_run_id is an error.
type SendChildRunner func(ctx context.Context, childRunID, prompt string, timeoutMs int) (output, state string, err error)

// PollChildRunner checks on a resident child WITHOUT sending new input (RFC BK
// P2): it returns the current turn's output-so-far + state. timeoutMs=0 is a
// non-blocking snapshot; >0 waits up to that long for the child to park. Used to
// await a child after a send returned state "running".
type PollChildRunner func(ctx context.Context, childRunID string, timeoutMs int) (output, state string, err error)

// CancelChildRunner turn-cancels a resident child's CURRENT turn (RFC BK P2):
// it stops the in-flight model call + tool dispatch and re-parks the child at
// awaiting_input (the child stays alive — unlike close, which terminates it).
// A child that is already parked is a no-op.
type CancelChildRunner func(ctx context.Context, childRunID string) (output, state string, err error)

// CloseChildRunner finalizes a resident child (RFC BK), firing its teardown
// (e.g. sandbox_close_run). Idempotent — closing an already-gone child is not an
// error. The caller must own the child.
type CloseChildRunner func(ctx context.Context, childRunID string) (err error)

// MaxConcurrentChildrenLookup resolves the per-agent
// `max_concurrent_children` cap for the named CALLING agent (the
// parent that is invoking Agent.parallel_spawn). Returns 0 when the
// agent has no explicit override; the tool then falls back to
// DefaultMaxConcurrentChildren.
//
// The lookup runs against the same resolver chain that drives sub-run
// dispatch (yaml > dynamic_agents > AgentDef substrate), so an agent
// edited via the substrate UI sees the updated cap on its next
// parallel_spawn without a runtime restart.
//
// nil = "no lookup wired" — every call falls through to
// DefaultMaxConcurrentChildren. Acceptable for tests and zero-config
// dev runs where every agent shares the default.
type MaxConcurrentChildrenLookup func(ctx context.Context, callingAgentName string) int

// MaxAgentDepth caps recursion: a top-level run is depth 0, the agents
// it spawns are depth 1, etc. The cap is a safety rail against runaway
// self-spawning prompt loops; cv-batch-adapter spawning cv-adapter is
// depth 1 today. Hardcoded for v0.4.0; lifted to config in a later
// release if a real workflow demands deeper.
const MaxAgentDepth = 3

// DefaultMaxConcurrentChildren caps how many sub-agents Agent.parallel_spawn
// will spin up concurrently for one call when the calling agent has
// no explicit `max_concurrent_children` yaml override. Four is a
// pragmatic default for fan-out workflows on a single VM: high
// enough to amortize the latency of slow-model children (claude-opus
// at ~10-15s per response) but low enough to avoid blowing past the
// per-tenant fairness cap (v0.10.1 default = 4) on the global
// semaphore.
//
// Sequential Agent.spawn calls are unaffected — the cap only applies
// inside a single parallel_spawn op's `spawns` array. Operators
// override per-agent via `max_concurrent_children: N` in
// loomcycle.yaml or via the AgentDef substrate overlay.
const DefaultMaxConcurrentChildren = 4

// MaxParallelSpawns caps how many entries one parallel_spawn op's
// `spawns` array may carry, regardless of the calling agent's
// max_concurrent_children. The cap is the absolute ceiling on
// per-call fan-out: a single tool call can't enqueue more than
// MaxParallelSpawns children, even if the per-agent cap is higher
// (a poorly-written prompt asking for 100 specialists would
// otherwise saturate the global semaphore from a single tool call).
// The per-agent cap still applies on top — it controls how many of
// those MaxParallelSpawns actually run concurrently vs serialized.
//
// 32 is a deliberately wide ceiling: large enough that no real fan-
// out workflow hits it, low enough that a runaway prompt can't kite
// the substrate. Hardcoded; not yaml-configurable in v1.
const MaxParallelSpawns = 32

// AgentTool is the built-in `Agent` tool that spawns one or more
// named sub-agents in fresh sessions. The model invokes it with
// either `{op:"spawn", name, prompt, def_id?}` (the v0.4.0 single-
// spawn shape; `op` is omittable and defaults to "spawn") or
// `{op:"parallel_spawn", spawns:[{name, prompt, def_id?}, ...]}`
// for concurrent fan-out (v0.11.8+).
//
// `op:"spawn"` returns the sub-agent's final assistant text as a
// tool_result. `op:"parallel_spawn"` returns a JSON-encoded
// `{results:[{agent,ok,output|error},...]}` envelope in input order
// — sub-agent errors are captured per-child and surfaced inside the
// envelope, NOT escalated to a parent tool error. The parent's
// model decides whether to retry, fall back, or give up.
//
// Wire shape:
//
//	// Single spawn — v0.4.0 default, op optional.
//	{
//	  "op":     "spawn",                // optional; default
//	  "name":   "cv-adapter",          // a key in cfg.Agents
//	  "prompt": "Generate CV for ...", // user-message body the sub sees
//	  "def_id": "def_abc123"            // optional; pin to a specific row
//	}
//
//	// Parallel spawn — v0.11.8+, op required.
//	{
//	  "op": "parallel_spawn",
//	  "spawns": [
//	    {"name": "researcher", "prompt": "Topic A"},
//	    {"name": "researcher", "prompt": "Topic B"},
//	    {"name": "summarizer", "prompt": "..."}
//	  ]
//	}
//
// The sub-agent's `tools` (from its YAML AgentDef) is the sole
// authority on what the sub can use — the parent's allow-list does not
// widen or narrow it. This matches the trust model: each agent
// definition is operator-curated and self-describing.
//
// The parent observes only the tool_call (with input) and the
// tool_result (with the sub's final text). The sub's intermediate
// events go to the sub's own persisted transcript, retrievable via
// GET /v1/sessions/{sub-session-id}/transcript.
type AgentTool struct {
	// Run is the closure provided by the runtime that knows how to
	// look up sub-agent definitions, build the tool dispatcher, and
	// drive a fresh loop. Constructed by the HTTP server at boot.
	Run SubAgentRunner

	// RunDetailed is the run_id-surfacing variant used by parallel_spawn
	// for the RFC X Phase 3 spawn ledger (optional). When nil, parallel_spawn
	// falls back to Run with no run_id (no ledger) — existing behavior.
	// Wired by the HTTP server at boot to the same runner as Run.
	RunDetailed SubAgentRunnerDetailed

	// SpawnLedger gates the RFC X Phase 3 spawn-ledger emits (and the
	// per-child index threading) in parallel_spawn. Set from
	// LOOMCYCLE_RESUME_FANOUT at boot; default false → no ledger events, no
	// behavior change. Requires RunDetailed + an EventEmitter in ctx.
	SpawnLedger bool

	// CapLookup resolves the calling agent's per-agent
	// `max_concurrent_children` override. nil = always use
	// DefaultMaxConcurrentChildren. Wired by the HTTP server at boot.
	CapLookup MaxConcurrentChildrenLookup

	// OpenChild / SendChild / PollChild / CancelChild / CloseChild wire the RFC
	// BK resident-child ops (open/send/poll/cancel/close). nil = the op is
	// unavailable (the tool refuses with a clear "not wired" message), so a
	// runtime without interactive sub-agents stays byte-identical for
	// spawn/parallel_spawn. Set by the HTTP server.
	OpenChild   OpenChildRunner
	SendChild   SendChildRunner
	PollChild   PollChildRunner
	CancelChild CancelChildRunner
	CloseChild  CloseChildRunner
}

// runChild drives one sub-agent, preferring RunDetailed (surfaces the child
// run_id for the spawn ledger) and falling back to Run. runID is "" on the
// fallback path or when RunDetailed couldn't create a run.
func (a *AgentTool) runChild(ctx context.Context, name, prompt, defID string) (output string, runID string, err error) {
	if a.RunDetailed != nil {
		return a.RunDetailed(ctx, name, prompt, defID)
	}
	output, err = a.Run(ctx, name, prompt, defID)
	return output, "", err
}

// agentInput is the JSON shape the model sends. The discriminator is
// the optional `op` field — empty / "spawn" routes to the single-
// child path; "parallel_spawn" routes to the fan-out path.
type agentInput struct {
	Op string `json:"op,omitempty"`

	// Single-spawn fields (op="" or "spawn").
	Name   string `json:"name,omitempty"`
	Prompt string `json:"prompt,omitempty"`
	DefID  string `json:"def_id,omitempty"`
	// Compaction optionally overrides the child's resolved compaction settings
	// PER FIELD (the parent steering its child's context management). Merged on
	// top of what the child inherits from this parent. nil = inherit unchanged.
	Compaction *config.Compaction `json:"compaction,omitempty"`

	// Parallel-spawn fields (op="parallel_spawn").
	Spawns []parallelSpawnEntry `json:"spawns,omitempty"`

	// Resident-child fields (RFC BK). op="open" reuses Name/Prompt/DefID and
	// optionally IdleTTLSeconds; op="send" uses ChildRunID + Prompt (+ optional
	// TimeoutMs); op="poll" uses ChildRunID (+ optional TimeoutMs); op="close"
	// and op="cancel" use ChildRunID.
	ChildRunID     string `json:"child_run_id,omitempty"`
	IdleTTLSeconds int    `json:"idle_ttl_seconds,omitempty"`
	// TimeoutMs bounds a send/poll (RFC BK P2). send: 0 = block until the child
	// parks (P1 behavior); >0 = return {state:"running", partial output} if the
	// turn is still going after this long. poll: 0 = a non-blocking snapshot; >0 =
	// wait up to this long for the child to park.
	TimeoutMs int `json:"timeout_ms,omitempty"`
}

// parallelSpawnEntry is one row in a parallel_spawn `spawns` array.
type parallelSpawnEntry struct {
	Name   string `json:"name"`
	Prompt string `json:"prompt"`
	DefID  string `json:"def_id,omitempty"`
	// Compaction: per-spawn compaction override for THIS child (see agentInput).
	Compaction *config.Compaction `json:"compaction,omitempty"`
}

// ParallelSpawnResult is one entry in the JSON envelope the
// parallel_spawn op returns to the calling model. `Ok` discriminates
// success vs failure: when true, `Output` carries the child's final
// text; when false, `Error` carries the human-readable error string.
// Index preserves input ordering when the model needs to correlate.
type ParallelSpawnResult struct {
	Index  int    `json:"index"`
	Agent  string `json:"agent"`
	Ok     bool   `json:"ok"`
	Output string `json:"output,omitempty"`
	Error  string `json:"error,omitempty"`
	// RunID is the child's run row id (RFC X Phase 3), captured for the
	// spawn ledger so a restored fan-out parent can re-find this child.
	// Internal-only — omitted from the tool_result envelope the model sees.
	RunID string `json:"-"`
}

// agentInputSchema is the JSON Schema the model sees. Discriminated
// by `op`; both shapes documented in `oneOf` so providers that surface
// schema-driven help (Claude, OpenAI) show the right field set per op.
// v0.8.10's Gemini schema sanitizer merges `oneOf` branches so the
// schema lands cleanly there too.
const agentInputSchema = `{
  "type": "object",
  "oneOf": [
    {
      "title": "spawn — single sub-agent (default)",
      "properties": {
        "op":     {"type": "string", "enum": ["spawn"], "description": "Optional; defaults to spawn when omitted."},
        "name":   {"type": "string", "description": "Sub-agent name. Must match a key in the loomcycle.yaml agents map."},
        "prompt": {"type": "string", "description": "User-message body the sub-agent sees. Treat as the task description; do not include auth tokens (the sub-agent gets its own auth context)."},
        "def_id": {"type": "string", "description": "Optional. Pin this sub-run to a specific agent_defs row id (returned by AgentDef.create or AgentDef.fork). The row's name must match the 'name' field."},
        "compaction": {"type": "object", "description": "Optional. Override this child's context-compaction settings (it inherits yours by default). Per field: enabled (auto-compact on/off), target_percentage (10-50), keep_last_n, keep_first, autocompact_at_pct (50-95), model.", "properties": {"enabled": {"type": "boolean"}, "target_percentage": {"type": "integer"}, "keep_last_n": {"type": "integer"}, "keep_first": {"type": "boolean"}, "autocompact_at_pct": {"type": "integer"}, "model": {"type": "string"}}}
      },
      "required": ["name", "prompt"]
    },
    {
      "title": "parallel_spawn — fan out to N sub-agents concurrently",
      "properties": {
        "op": {"type": "string", "enum": ["parallel_spawn"], "description": "Required for the fan-out shape."},
        "spawns": {
          "type": "array",
          "minItems": 1,
          "description": "Sub-agents to spawn concurrently. Returns when ALL children complete (success or error). Per-child errors are captured inside the result envelope, not escalated.",
          "items": {
            "type": "object",
            "properties": {
              "name":   {"type": "string", "description": "Sub-agent name. Must match a key in the loomcycle.yaml agents map."},
              "prompt": {"type": "string", "description": "User-message body the sub-agent sees."},
              "def_id": {"type": "string", "description": "Optional. Pin this sub-run to a specific agent_defs row id."},
              "compaction": {"type": "object", "description": "Optional per-child context-compaction override (inherits yours by default). Fields: enabled, target_percentage (10-50), keep_last_n, keep_first, autocompact_at_pct (50-95), model.", "properties": {"enabled": {"type": "boolean"}, "target_percentage": {"type": "integer"}, "keep_last_n": {"type": "integer"}, "keep_first": {"type": "boolean"}, "autocompact_at_pct": {"type": "integer"}, "model": {"type": "string"}}}
            },
            "required": ["name", "prompt"]
          }
        }
      },
      "required": ["op", "spawns"]
    },
    {
      "title": "open — start a resident, steerable sub-agent",
      "properties": {
        "op":     {"type": "string", "enum": ["open"], "description": "Start a PERSISTENT sub-agent you can steer over multiple turns. Returns {child_run_id, state, output}. The child keeps its conversation AND any resources it holds (e.g. a warm sandbox container) between sends — use this instead of re-spawning when the child must stay stateful."},
        "name":   {"type": "string", "description": "Sub-agent name. Must match a key in the loomcycle.yaml agents map."},
        "prompt": {"type": "string", "description": "The first instruction for the child. It runs one turn, then parks awaiting your next send."},
        "def_id": {"type": "string", "description": "Optional. Pin this child to a specific agent_defs row id."},
        "idle_ttl_seconds": {"type": "integer", "description": "Optional. Reap the child after this many seconds with no send (0 = operator default). You own the child's lifecycle — close it when done."}
      },
      "required": ["op", "name", "prompt"]
    },
    {
      "title": "send — give a resident sub-agent its next instruction",
      "properties": {
        "op":           {"type": "string", "enum": ["send"], "description": "Steer a resident child's next turn. Waits for the turn and returns its output."},
        "child_run_id": {"type": "string", "description": "The child_run_id returned by op=open."},
        "prompt":       {"type": "string", "description": "The next instruction for the child. It sees its full prior conversation."},
        "timeout_ms":   {"type": "integer", "description": "Optional. 0 (default) blocks until the child parks. >0 returns early with state \"running\" + the partial output if the turn is still going after this long — then use op=poll to await it, or op=cancel to interrupt."}
      },
      "required": ["op", "child_run_id", "prompt"]
    },
    {
      "title": "poll — check on a resident sub-agent without sending input",
      "properties": {
        "op":           {"type": "string", "enum": ["poll"], "description": "Return a resident child's current output-so-far + state, without giving it new input. Use it to await a child after send returned state \"running\"."},
        "child_run_id": {"type": "string", "description": "The child_run_id to check."},
        "timeout_ms":   {"type": "integer", "description": "Optional. 0 (default) is a non-blocking snapshot; >0 waits up to this long for the child to park."}
      },
      "required": ["op", "child_run_id"]
    },
    {
      "title": "cancel — stop a resident sub-agent's current turn",
      "properties": {
        "op":           {"type": "string", "enum": ["cancel"], "description": "Turn-cancel a resident child: stop its in-flight turn and re-park it (the child stays alive — use op=close to shut it down). A no-op if it's already parked."},
        "child_run_id": {"type": "string", "description": "The child_run_id whose current turn to stop."}
      },
      "required": ["op", "child_run_id"]
    },
    {
      "title": "close — shut down a resident sub-agent",
      "properties": {
        "op":           {"type": "string", "enum": ["close"], "description": "Finalize a resident child and free its resources (idempotent). Always close a child you opened once you're done with it."},
        "child_run_id": {"type": "string", "description": "The child_run_id to close."}
      },
      "required": ["op", "child_run_id"]
    }
  ]
}`

const agentDescription = `Spawn or drive named sub-agents, each with its own tool allowlist (your tool set does not transfer). ` +
	`Stateless ops: 'spawn' (default; one child, return its final text) and 'parallel_spawn' (N children concurrently, JSON envelope with per-child ok/output/error) — best when you describe the whole task up front. ` +
	`Resident ops (stateful): 'open' starts a persistent sub-agent and returns a child_run_id; 'send' gives it the next instruction and returns that turn's output (optional timeout_ms bounds the wait — a long turn returns state "running" + partial output); 'poll' checks a running child without new input; 'cancel' stops a child's current turn (it stays alive); 'close' shuts it down. Use these when the child must keep state between steps — a warm sandbox container, a REPL, a multi-turn analysis — instead of re-spawning and re-threading state by hand. Close what you open. ` +
	`See Context.help(topic="fan-out-patterns") for spawn vs parallel_spawn vs Channel.publish, and Context.help(topic="resident-sub-agents") for the open/send/close lifecycle.`

// Name implements tools.Tool.
func (a *AgentTool) Name() string { return "Agent" }

// Description implements tools.Tool.
func (a *AgentTool) Description() string { return agentDescription }

// InputSchema implements tools.Tool.
func (a *AgentTool) InputSchema() json.RawMessage { return json.RawMessage(agentInputSchema) }

// Execute implements tools.Tool. Validates the input, enforces the
// recursion depth cap, and delegates to the SubAgentRunner. Errors are
// surfaced as IsError tool_result so the model can self-correct rather
// than the run being torn down.
func (a *AgentTool) Execute(ctx context.Context, input json.RawMessage) (tools.Result, error) {
	if a.Run == nil {
		return tools.Result{IsError: true, Text: "Agent tool not wired to a sub-agent runner (operator misconfiguration)"}, nil
	}
	var in agentInput
	if err := json.Unmarshal(input, &in); err != nil {
		return tools.Result{IsError: true, Text: fmt.Sprintf("invalid input JSON: %s", err)}, nil
	}
	op := strings.TrimSpace(in.Op)
	if op == "" {
		op = "spawn"
	}
	switch op {
	case "spawn":
		return a.executeSpawn(ctx, in)
	case "parallel_spawn":
		return a.executeParallelSpawn(ctx, in)
	case "open":
		return a.executeOpen(ctx, in)
	case "send":
		return a.executeSend(ctx, in)
	case "poll":
		return a.executePoll(ctx, in)
	case "cancel":
		return a.executeCancel(ctx, in)
	case "close":
		return a.executeClose(ctx, in)
	default:
		return tools.Result{IsError: true, Text: fmt.Sprintf("unknown op %q (expected 'spawn', 'parallel_spawn', 'open', 'send', 'poll', 'cancel', or 'close')", in.Op)}, nil
	}
}

// executeSpawn handles the v0.4.0 single-child path. Behavior is
// unchanged from the pre-v0.11.8 AgentTool — same depth guard, same
// error surface, same "no final text" hint.
func (a *AgentTool) executeSpawn(ctx context.Context, in agentInput) (tools.Result, error) {
	if len(in.Spawns) > 0 {
		return tools.Result{IsError: true, Text: "op=spawn must not carry a 'spawns' array; use op=parallel_spawn for fan-out"}, nil
	}
	in.Name = strings.TrimSpace(in.Name)
	if in.Name == "" {
		return tools.Result{IsError: true, Text: "missing required field: name"}, nil
	}
	if in.Prompt == "" {
		return tools.Result{IsError: true, Text: "missing required field: prompt"}, nil
	}
	if AgentDepth(ctx) >= MaxAgentDepth {
		return tools.Result{
			IsError: true,
			Text: fmt.Sprintf(
				"max sub-agent recursion depth (%d) reached at agent %q; refusing to spawn deeper",
				MaxAgentDepth, in.Name,
			),
		}, nil
	}
	subCtx := IncrementAgentDepth(ctx)
	// Per-spawn compaction override (the parent steering this child's context
	// management); runSubAgent blends it on top of inheritance.
	if !in.Compaction.IsZero() {
		subCtx = tools.WithCompactionOverride(subCtx, in.Compaction)
	}
	output, err := a.Run(subCtx, in.Name, in.Prompt, in.DefID)
	if err != nil {
		return tools.Result{IsError: true, Text: err.Error()}, nil
	}
	if output == "" {
		return tools.Result{Text: fmt.Sprintf("(sub-agent %q completed with no final text)", in.Name)}, nil
	}
	return tools.Result{Text: output}, nil
}

// executeParallelSpawn fans out to N children concurrently. Returns
// when ALL children complete (success or error); per-child errors are
// captured inside the JSON envelope rather than escalated.
//
// Concurrency is bounded by min(per-agent max_concurrent_children,
// len(spawns)) — a semaphore-shaped channel rate-limits goroutine
// admission. Depth is guarded once for the entire call: the parent's
// current depth must be < MaxAgentDepth (so children dispatch at
// depth+1, identical to the single-spawn path).
//
// Result text is a deterministic-ordering JSON envelope:
//
//	{"results": [{"index":0,"agent":"researcher","ok":true,"output":"..."},
//	             {"index":1,"agent":"researcher","ok":false,"error":"..."}]}
//
// The envelope is a tool_result Text payload (not IsError) regardless
// of per-child success — the call as a whole succeeded; the per-child
// disposition is the model's to read.
func (a *AgentTool) executeParallelSpawn(ctx context.Context, in agentInput) (tools.Result, error) {
	if in.Name != "" || in.Prompt != "" || in.DefID != "" {
		return tools.Result{IsError: true, Text: "op=parallel_spawn must not carry top-level name/prompt/def_id fields; put each child in the 'spawns' array"}, nil
	}
	if len(in.Spawns) == 0 {
		return tools.Result{IsError: true, Text: "op=parallel_spawn requires a non-empty 'spawns' array"}, nil
	}
	if len(in.Spawns) > MaxParallelSpawns {
		return tools.Result{
			IsError: true,
			Text: fmt.Sprintf(
				"parallel_spawn 'spawns' array has %d entries; the per-call ceiling is %d (split the work across multiple calls if you genuinely need more)",
				len(in.Spawns), MaxParallelSpawns,
			),
		}, nil
	}
	// Per-entry input validation BEFORE we kick anything off — a
	// malformed entry should fail the whole call up-front, not
	// arrive as a per-child error inside an otherwise-successful
	// envelope.
	for i, sp := range in.Spawns {
		name := strings.TrimSpace(sp.Name)
		if name == "" {
			return tools.Result{IsError: true, Text: fmt.Sprintf("spawns[%d]: missing required field: name", i)}, nil
		}
		if sp.Prompt == "" {
			return tools.Result{IsError: true, Text: fmt.Sprintf("spawns[%d] (%s): missing required field: prompt", i, name)}, nil
		}
		in.Spawns[i].Name = name
	}
	// Depth guard fires once for the whole call. Each child
	// dispatches at depth+1 (same as single-spawn).
	if AgentDepth(ctx) >= MaxAgentDepth {
		return tools.Result{
			IsError: true,
			Text: fmt.Sprintf(
				"max sub-agent recursion depth (%d) reached; refusing to parallel_spawn at depth %d",
				MaxAgentDepth, AgentDepth(ctx),
			),
		}, nil
	}
	subCtx := IncrementAgentDepth(ctx)

	// Resolve the per-call concurrency cap: per-agent override (if
	// the calling agent set one + the lookup is wired) else the
	// runtime default. Cap is then min'd with the spawn count so a
	// single-entry call doesn't allocate a buffered chan of size 4
	// for no reason.
	callingAgent := tools.AgentName(ctx)
	// Avoid shadowing the built-in `cap` identifier — `concurrencyCap`
	// reads better at the call site and future-proofs against an edit
	// that wants `cap(slice)` here.
	concurrencyCap := DefaultMaxConcurrentChildren
	if a.CapLookup != nil && callingAgent != "" {
		if override := a.CapLookup(ctx, callingAgent); override > 0 {
			concurrencyCap = override
		}
	}
	if concurrencyCap > len(in.Spawns) {
		concurrencyCap = len(in.Spawns)
	}

	// RFC X Phase 3 spawn ledger (flag-gated). emit + toolUseID are captured
	// from the parent ctx once; the per-child goroutines record spawn_child_*
	// events through the (mutex-guarded, concurrency-safe) parent emitter so a
	// snapshotted+restored fan-out parent can reconstruct this tool_result.
	// Gate on a REAL emitter (tools.HasEventEmitter) + a tool_use id — not on
	// EventEmitter(ctx) != nil, which is always true (the accessor returns a
	// no-op when unset, so the old check could not detect a missing sink and the
	// ledger would silently drain to the no-op).
	toolUseID := tools.ToolUseID(ctx)
	ledger := a.SpawnLedger && a.RunDetailed != nil && toolUseID != "" && tools.HasEventEmitter(ctx)
	var ledgerEmit tools.EventEmitterFunc
	if ledger {
		ledgerEmit = tools.EventEmitter(ctx)
	}

	results := make([]ParallelSpawnResult, len(in.Spawns))
	sem := make(chan struct{}, concurrencyCap)
	var wg sync.WaitGroup
	for i, sp := range in.Spawns {
		i, sp := i, sp
		wg.Add(1)
		go func() {
			defer wg.Done()
			// Acquire a slot or bail on ctx cancellation. ctx
			// cancellation propagates from the parent run, so a
			// cancelled parent reliably terminates outstanding
			// children rather than waiting on the semaphore.
			select {
			case sem <- struct{}{}:
				defer func() { <-sem }()
			case <-subCtx.Done():
				results[i] = ParallelSpawnResult{Index: i, Agent: sp.Name, Ok: false, Error: subCtx.Err().Error()}
				return
			}
			// Thread the child's index so runSubAgent can emit the
			// spawn_child_started ledger row with the right index + run_id.
			childCtx := subCtx
			if ledger {
				childCtx = tools.WithSpawnIndex(subCtx, i)
			}
			// Per-spawn compaction override (the parent steering this child's
			// context management); runSubAgent blends it on top of inheritance.
			if !sp.Compaction.IsZero() {
				childCtx = tools.WithCompactionOverride(childCtx, sp.Compaction)
			}
			out, childRunID, err := a.runChild(childCtx, sp.Name, sp.Prompt, sp.DefID)
			var r ParallelSpawnResult
			if err != nil {
				r = ParallelSpawnResult{Index: i, Agent: sp.Name, Ok: false, Error: err.Error(), RunID: childRunID}
			} else {
				if out == "" {
					out = fmt.Sprintf("(sub-agent %q completed with no final text)", sp.Name)
				}
				r = ParallelSpawnResult{Index: i, Agent: sp.Name, Ok: true, Output: out, RunID: childRunID}
			}
			results[i] = r
			// Record this child's result on the parent's transcript so a
			// child that completed BEFORE a snapshot (whose run row isn't
			// captured) still has a recoverable result on resume.
			if ledger {
				ledgerEmit(providers.Event{
					Type: providers.EventSpawnChildResult,
					SpawnChild: &providers.SpawnChildEventInfo{
						ToolUseID: toolUseID,
						Index:     i,
						RunID:     childRunID,
						Agent:     sp.Name,
						Ok:        r.Ok,
						Output:    r.Output,
						Error:     r.Error,
					},
				})
			}
		}()
	}

	// RFC X Phase 3: park the fan-out PARENT while it's blocked in wg.Wait. The
	// loop's own top-of-iteration park is unreachable mid-tool-call, so a
	// watcher goroutine parks the parent run (persist pause_state=paused + take
	// barrier credit so paused_runs_count is accurate + the warning clears) when
	// a pause is declared, and unparks on resume — WITHOUT touching the
	// wg.Wait collection below (the in-memory results survive a same-instance
	// pause/resume; a cross-instance snapshot reconstructs from the ledger). The
	// deferred close+join guarantees the watcher has fully stopped before we
	// return, so it never races the loop's own next Park.
	if ledger {
		if gate := tools.PauseGateFromContext(ctx); gate != nil {
			watchDone := make(chan struct{})
			watcherExited := make(chan struct{})
			go func() {
				defer close(watcherExited)
				for {
					select {
					case <-watchDone:
						return
					case <-ctx.Done():
						return
					case <-gate.PauseCh():
						if err := gate.Park(ctx); err != nil {
							return // ctx cancelled while parked
						}
						// resumed → re-watch the next pause cycle (PauseCh
						// returns a fresh channel post-resume).
					}
				}
			}()
			defer func() {
				close(watchDone)
				<-watcherExited
			}()
		}
	}

	wg.Wait()

	envelope := struct {
		Results []ParallelSpawnResult `json:"results"`
	}{Results: results}
	body, err := json.Marshal(envelope)
	if err != nil {
		// json.Marshal on a slice of plain structs effectively never
		// fails; keep the defensive path so future field additions
		// surface loudly rather than silently.
		return tools.Result{IsError: true, Text: fmt.Sprintf("internal: marshal parallel_spawn envelope: %s", err)}, nil
	}
	return tools.Result{Text: string(body)}, nil
}

// residentChildResult is the JSON envelope the resident-child ops (open/send/
// close) return. child_run_id is the handle the model reuses for send/close;
// state tells it whether the child is still awaiting_input; output is the turn's
// assistant text.
type residentChildResult struct {
	ChildRunID string `json:"child_run_id,omitempty"`
	State      string `json:"state"`
	Output     string `json:"output"`
}

// residentResult marshals the resident-child envelope.
func residentResult(childRunID, state, output string) (tools.Result, error) {
	body, err := json.Marshal(residentChildResult{ChildRunID: childRunID, State: state, Output: output})
	if err != nil {
		return tools.Result{IsError: true, Text: fmt.Sprintf("internal: marshal resident-child envelope: %s", err)}, nil
	}
	return tools.Result{Text: string(body)}, nil
}

// executeOpen (RFC BK) starts a resident interactive sub-agent, runs its first
// turn, parks it at awaiting_input, and returns {child_run_id, state, output}.
// The child stays alive for follow-up sends until closed / idle-reaped / the
// parent tree tears down. A resident child counts against the recursion depth
// exactly like a spawn — it is a sub-run one level deeper.
func (a *AgentTool) executeOpen(ctx context.Context, in agentInput) (tools.Result, error) {
	if a.OpenChild == nil {
		return tools.Result{IsError: true, Text: "Agent op=open (resident sub-agent) is not available on this runtime"}, nil
	}
	in.Name = strings.TrimSpace(in.Name)
	if in.Name == "" {
		return tools.Result{IsError: true, Text: "missing required field: name"}, nil
	}
	if in.Prompt == "" {
		return tools.Result{IsError: true, Text: "missing required field: prompt"}, nil
	}
	if in.ChildRunID != "" {
		return tools.Result{IsError: true, Text: "op=open mints a NEW child; do not pass child_run_id (use op=send with the id op=open returns)"}, nil
	}
	if in.IdleTTLSeconds < 0 {
		return tools.Result{IsError: true, Text: "idle_ttl_seconds must be >= 0 (0 = operator default)"}, nil
	}
	if AgentDepth(ctx) >= MaxAgentDepth {
		return tools.Result{IsError: true, Text: fmt.Sprintf(
			"max sub-agent recursion depth (%d) reached at agent %q; refusing to open deeper", MaxAgentDepth, in.Name)}, nil
	}
	subCtx := IncrementAgentDepth(ctx)
	if !in.Compaction.IsZero() {
		subCtx = tools.WithCompactionOverride(subCtx, in.Compaction)
	}
	childRunID, output, state, err := a.OpenChild(subCtx, in.Name, in.Prompt, in.DefID, in.IdleTTLSeconds)
	if err != nil {
		return tools.Result{IsError: true, Text: err.Error()}, nil
	}
	return residentResult(childRunID, state, output)
}

// executeSend (RFC BK) steers a resident child with the next instruction and
// blocks until it re-parks (or terminates), returning that turn's output.
func (a *AgentTool) executeSend(ctx context.Context, in agentInput) (tools.Result, error) {
	if a.SendChild == nil {
		return tools.Result{IsError: true, Text: "Agent op=send (resident sub-agent) is not available on this runtime"}, nil
	}
	in.ChildRunID = strings.TrimSpace(in.ChildRunID)
	if in.ChildRunID == "" {
		return tools.Result{IsError: true, Text: "missing required field: child_run_id (the id op=open returned)"}, nil
	}
	if in.Prompt == "" {
		return tools.Result{IsError: true, Text: "missing required field: prompt"}, nil
	}
	if in.TimeoutMs < 0 {
		return tools.Result{IsError: true, Text: "timeout_ms must be >= 0 (0 = block until the child parks)"}, nil
	}
	output, state, err := a.SendChild(ctx, in.ChildRunID, in.Prompt, in.TimeoutMs)
	if err != nil {
		return tools.Result{IsError: true, Text: err.Error()}, nil
	}
	return residentResult(in.ChildRunID, state, output)
}

// executePoll (RFC BK P2) checks on a resident child without sending new input —
// returns its current output-so-far + state. Used to await a child after a send
// returned state "running" (timeout_ms=0 is a non-blocking snapshot).
func (a *AgentTool) executePoll(ctx context.Context, in agentInput) (tools.Result, error) {
	if a.PollChild == nil {
		return tools.Result{IsError: true, Text: "Agent op=poll (resident sub-agent) is not available on this runtime"}, nil
	}
	in.ChildRunID = strings.TrimSpace(in.ChildRunID)
	if in.ChildRunID == "" {
		return tools.Result{IsError: true, Text: "missing required field: child_run_id"}, nil
	}
	if in.TimeoutMs < 0 {
		return tools.Result{IsError: true, Text: "timeout_ms must be >= 0 (0 = non-blocking snapshot)"}, nil
	}
	output, state, err := a.PollChild(ctx, in.ChildRunID, in.TimeoutMs)
	if err != nil {
		return tools.Result{IsError: true, Text: err.Error()}, nil
	}
	return residentResult(in.ChildRunID, state, output)
}

// executeCancel (RFC BK P2) turn-cancels a resident child's current turn (stops
// it and re-parks the child — the child stays alive, unlike close).
func (a *AgentTool) executeCancel(ctx context.Context, in agentInput) (tools.Result, error) {
	if a.CancelChild == nil {
		return tools.Result{IsError: true, Text: "Agent op=cancel (resident sub-agent) is not available on this runtime"}, nil
	}
	in.ChildRunID = strings.TrimSpace(in.ChildRunID)
	if in.ChildRunID == "" {
		return tools.Result{IsError: true, Text: "missing required field: child_run_id"}, nil
	}
	output, state, err := a.CancelChild(ctx, in.ChildRunID)
	if err != nil {
		return tools.Result{IsError: true, Text: err.Error()}, nil
	}
	return residentResult(in.ChildRunID, state, output)
}

// executeClose (RFC BK) finalizes a resident child (idempotent), firing its
// teardown so any resources it held (e.g. a sandbox container) are released.
func (a *AgentTool) executeClose(ctx context.Context, in agentInput) (tools.Result, error) {
	if a.CloseChild == nil {
		return tools.Result{IsError: true, Text: "Agent op=close (resident sub-agent) is not available on this runtime"}, nil
	}
	in.ChildRunID = strings.TrimSpace(in.ChildRunID)
	if in.ChildRunID == "" {
		return tools.Result{IsError: true, Text: "missing required field: child_run_id"}, nil
	}
	if err := a.CloseChild(ctx, in.ChildRunID); err != nil {
		return tools.Result{IsError: true, Text: err.Error()}, nil
	}
	return residentResult(in.ChildRunID, "closed", "")
}

// errAgentBadInput is reserved for future structured errors. Currently
// we surface every input problem as an IsError tool_result, but a
// caller that wants to break out of the loop entirely (rather than let
// the model self-correct) can return this from a SubAgentRunner.
var errAgentBadInput = errors.New("agent: bad input")

type ctxKeyAgentDepth struct{}

// IncrementAgentDepth bumps the depth counter on ctx. The Agent tool
// applies this before invoking the SubAgentRunner so a recursive child
// inherits a higher depth than the parent.
func IncrementAgentDepth(ctx context.Context) context.Context {
	d, _ := ctx.Value(ctxKeyAgentDepth{}).(int)
	return context.WithValue(ctx, ctxKeyAgentDepth{}, d+1)
}

// AgentDepth reports the current recursion depth (0 at the top level).
// Exposed so tests and the loop can assert depth invariants without
// reaching into the unexported context key.
func AgentDepth(ctx context.Context) int {
	d, _ := ctx.Value(ctxKeyAgentDepth{}).(int)
	return d
}

var _ tools.Tool = (*AgentTool)(nil)
