//! Sub-agent runner.
//!
//! A sub-agent is a fresh agent loop spawned from inside the
//! parent's tool dispatch. It has its own short-lived `messages`
//! history, its own [`ToolContext`] (folder + mode), and its own
//! cancellation token derived from the parent's so an abort
//! cascades. Sub-agents return a single text result that the parent
//! sees as the `task` tool's return value; the rest of the
//! sub-agent's transcript is kept addressable via its
//! `sub_session_id` for the UI's pop-out view (Phase E).
//!
//! The wire-level tool name on the parent is `task` (matches the
//! convention from every other agent product the team has used and
//! the model picked up on it without prompting). Internally we keep
//! the `subagent` naming for types and the module — that's what
//! actually gets spawned, and it disambiguates from the `Session`
//! type that backs both parents and sub-agents.
//!
//! Today's contract:
//!
//! - Mode is `Research` or `Coder`. `Research` blocks `write_file`
//!   and `edit_file` at the dispatch boundary (see
//!   [`ToolRegistry::dispatch`]); the "no mutation via `bash`" half
//!   is behavioural and lives in the system prompt.
//! - Sub-agents inherit the parent's everyday driver model
//!   ([`crate::models::CoderModels::standard`]). There used to be
//!   a `fast`/`large` selector; we dropped it because (1) it
//!   implied sub-agents were second-class workers, which made the
//!   parent reluctant to delegate non-trivial tasks; and (2) it
//!   was unused complexity — the team uses one model for actual
//!   work and a separate cheap model for the auto-rename title
//!   generator + compaction summaries. The cheap model
//!   ([`crate::models::CoderModels::cheap`]) is still used
//!   internally by sub-agents for compaction; it's not selectable
//!   per-call.
//! - Iteration cap: same [`MAX_TURN_ITERATIONS`] as the parent.
//!   Sub-agents used to run a tighter cap (50) on the assumption
//!   they were scoped tasks, but in practice that just made them
//!   bail mid-refactor when delegated a meaty chunk of work; auto-
//!   compaction handles the "context too big" failure mode the
//!   older byte cap was approximating, so there's no reason to
//!   throttle a sub-agent harder than its parent.
//! - Token-aware compaction kicks in at the same threshold as the
//!   parent loop ([`crate::compaction::COMPACT_THRESHOLD`]); the
//!   summary becomes a synthetic `system` message that replaces
//!   the older prefix.
//! - Depth cap: hardcoded to 1. The sub-agent's own tool list does
//!   not include `task`, so a sub-agent literally cannot spawn a
//!   sub-sub-agent.
//! - User continuation: the pop-out composer steers a running
//!   sub-agent mid-flight (queued, drained at the next iteration
//!   top) and resumes a finished one with a follow-up (history
//!   rebuilt from its JSONL, fresh detached loop, parent agent not
//!   involved). See [`queue_subagent_steer`] / [`resume_subagent`].
//!   The parent agent rides the same channel for its *detached*
//!   runs via the `task_steer` tool (ADR 0071).

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use camino::{Utf8Path, Utf8PathBuf};
use moon_core::WorkspaceFolderEntry;
use serde_json::{json, Value};
use tokio_util::sync::CancellationToken;

use crate::compaction;
use crate::defaults::{MAX_TURN_ITERATIONS, OUTPUT_CAP_CONTINUATIONS, OUTPUT_CAP_CONTINUATION_PROMPT};
use crate::error::CoderError;
use crate::event::CoderEvent;
use crate::inference::{AssistantResponse, ChatMessage, InferenceClient, StreamEvent, TokenUsage, ToolDefinition};
use crate::models::SharedCoderModels;
use crate::runner::FolderEventSink;
use crate::sessions::{
	self, current_time_ms, new_session_id, sessions_dir, subagent_session_dir, SessionHeader, SessionRecord,
	SESSION_SCHEMA_VERSION,
};
use crate::tools::{CoderMode, ToolContext, ToolRegistry};

/// Generated as `sub-<19-char-id>`. Different prefix from session
/// ids (`sess-...`) so a tail of the events can tell them apart at
/// a glance. Reuses [`new_session_id`]'s entropy generator so we
/// don't ship a second timestamp scheme.
fn new_subagent_id() -> String {
	format!("sub-{}", new_session_id().trim_start_matches("sess-"))
}

/// Minimum gap between auto-fetches against the same folder. A
/// parent that fans out four `task` calls against one folder in a
/// single assistant message should only trigger one `git fetch`,
/// not four; and a chatty session that keeps delegating to the
/// same sibling shouldn't re-fetch on every spawn. 5 minutes is
/// well below any realistic "refs went stale" window while still
/// collapsing every burst.
const SUBAGENT_FETCH_THROTTLE: Duration = Duration::from_secs(5 * 60);

/// Last successful auto-fetch dispatch time, keyed by folder root.
/// Process-global because sub-agents across every session and
/// folder share the throttle — the thing we're rate-limiting is
/// "hitting a given remote", which is per-folder, not per-session.
fn subagent_fetch_throttle() -> &'static Mutex<HashMap<String, Instant>> {
	static MAP: std::sync::OnceLock<Mutex<HashMap<String, Instant>>> = std::sync::OnceLock::new();
	MAP.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Kick off a best-effort `git fetch` against the sub-agent's
/// target folder so a sub-agent that inspects `origin/main`
/// (`git log origin/main`, `git diff origin/main...HEAD`, …) sees
/// fresh remote-tracking refs rather than whatever was last
/// fetched. Demand-driven: the human-facing active folder already
/// has its own periodic auto-fetch loop, but a sibling folder a
/// sub-agent is dispatched against may not have been fetched in a
/// long time — this closes that staleness window exactly when it
/// would otherwise bite.
///
/// Fire-and-forget on purpose: a fetch is up to a 30s network call
/// (capped inside `git_fetch`) and the sub-agent's opening tool
/// calls are almost never an `origin/main` read, so blocking the
/// dispatch on it would add latency for no benefit. The fetch races
/// ahead of the sub-agent's reasoning and lands before it matters.
/// A `git fetch` only moves remote-tracking refs — never the
/// working tree, index, or `HEAD` — so there's no conflict risk and
/// nothing to surprise an in-flight edit. Failures (offline, no
/// upstream, auth refused, timeout) are swallowed by `git_fetch`'s
/// own quiet path; the sub-agent just runs against whatever refs it
/// has, same as before this existed.
fn kick_off_subagent_fetch(folder: &Arc<WorkspaceFolderEntry>) {
	let root = folder.folder.path.clone();
	{
		let mut throttle = subagent_fetch_throttle().lock().expect("throttle map poisoned");
		let now = Instant::now();
		if let Some(last) = throttle.get(&root) {
			if now.duration_since(*last) < SUBAGENT_FETCH_THROTTLE {
				return;
			}
		}
		throttle.insert(root, now);
	}
	let host = folder.host.clone();
	tokio::spawn(async move {
		if let Err(err) = host.git_fetch().await {
			tracing::debug!(%err, "sub-agent pre-fetch failed (ignored)");
		}
	});
}

/// One steer queued against a running sub-agent from the pop-out
/// composer. Mirrors the parent's `PendingSteer` contract: the
/// panel renders a muted "queued" row the moment the steer is
/// accepted (via a wrapped `UserMessage { queued: true }`), and
/// the sub-agent drains it into its message history at the top of
/// its next iteration, flipping the row with a `SteerDrained`.
#[derive(Debug, Clone)]
struct QueuedSubagentSteer {
	id: String,
	text: String,
	/// `true` when the steer came from the supervising agent (the
	/// parent's `task_steer` tool) rather than the user's pop-out
	/// composer. Rides the existing `from_coordinator` wire flag so
	/// the transcript tags the row as agent-sent, not user-sent.
	from_coordinator: bool,
}

/// Live steer channel for one running sub-agent. An entry exists
/// exactly while the sub-agent's loop runs — registered before
/// `SubagentSpawned` is announced, removed once the loop settles —
/// so a steer against an absent id means "not running" and the
/// caller gets `false` back. The sink is the sub-agent's own
/// (parent-tagged) sink, captured so queue-time emission lands in
/// the same UI bucket as the rest of the sub-agent's events. The
/// cancel token is the run's own (a child of the parent turn's for
/// tool-dispatch runs, a fresh root for user-resumed runs), so the
/// pop-out's stop button can abort one sub-agent without touching
/// its siblings or the parent.
struct SubagentSteerChannel {
	queue: Vec<QueuedSubagentSteer>,
	sink: FolderEventSink,
	cancel: CancellationToken,
}

/// Process-global registry, same pattern as the fetch throttle
/// above: sub-agents run deep inside the parent's tool dispatch,
/// so the Tauri command layer can't reach them through `Coder`'s
/// session state — the id-keyed map is the rendezvous point.
fn subagent_steer_registry() -> &'static Mutex<HashMap<String, SubagentSteerChannel>> {
	static MAP: std::sync::OnceLock<Mutex<HashMap<String, SubagentSteerChannel>>> = std::sync::OnceLock::new();
	MAP.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Queue a mid-flight steer against a running sub-agent. Emits the
/// queued `UserMessage` row into the sub-agent's transcript
/// immediately; the text reaches the model at the top of the
/// sub-agent's next iteration. `from_coordinator` marks steers sent
/// by the supervising agent (`task_steer`) rather than the user's
/// pop-out composer. Returns `false` when the sub-agent isn't
/// running (finished, aborted, or unknown id) — the message is not
/// delivered and the caller should keep it.
pub(crate) fn queue_subagent_steer(subagent_id: &str, text: String, from_coordinator: bool) -> bool {
	let mut registry = subagent_steer_registry().lock().expect("steer registry poisoned");
	let Some(channel) = registry.get_mut(subagent_id) else {
		return false;
	};
	let steer_id = crate::runner::new_message_id();
	channel.queue.push(QueuedSubagentSteer {
		id: steer_id.clone(),
		text: text.clone(),
		from_coordinator,
	});
	channel.sink.send(wrap_inner(
		subagent_id,
		CoderEvent::UserMessage {
			id: steer_id,
			text,
			images: Vec::new(),
			queued: true,
			created_at_ms: Some(current_time_ms()),
			from_coordinator,
		},
	));
	true
}

fn register_subagent_steer_channel(subagent_id: &str, sink: FolderEventSink, cancel: CancellationToken) {
	subagent_steer_registry()
		.lock()
		.expect("steer registry poisoned")
		.insert(
			subagent_id.to_string(),
			SubagentSteerChannel {
				queue: Vec::new(),
				sink,
				cancel,
			},
		);
}

/// Atomically claim `subagent_id` for a user-driven resume: if no
/// channel exists, register one (with a fresh root token) and
/// return it for the resumed run; if one appeared in the meantime
/// (the sub-agent is running after all), queue `text` as an
/// ordinary steer instead and return `None`. Single registry lock
/// so a double-click / concurrent continue can't spawn two loops
/// over the same JSONL.
pub(crate) fn claim_subagent_for_resume(
	subagent_id: &str,
	text: &str,
	sink: &FolderEventSink,
) -> Option<CancellationToken> {
	let mut registry = subagent_steer_registry().lock().expect("steer registry poisoned");
	if registry.contains_key(subagent_id) {
		drop(registry);
		queue_subagent_steer(subagent_id, text.to_string(), false);
		return None;
	}
	let cancel = CancellationToken::new();
	registry.insert(
		subagent_id.to_string(),
		SubagentSteerChannel {
			queue: Vec::new(),
			sink: sink.clone(),
			cancel: cancel.clone(),
		},
	);
	Some(cancel)
}

/// Cancel a running sub-agent's own token — the pop-out's stop
/// button. Scoped to the one sub-agent: a tool-dispatch run's
/// token is a *child* of the parent turn's, so cancelling it never
/// propagates upward (the parent just sees the `task` call return
/// `Aborted` and reacts). Returns `false` when nothing is running
/// under that id.
pub(crate) fn abort_subagent(subagent_id: &str) -> bool {
	let registry = subagent_steer_registry().lock().expect("steer registry poisoned");
	match registry.get(subagent_id) {
		Some(channel) => {
			channel.cancel.cancel();
			true
		}
		None => false,
	}
}

/// Drop the channel. Any still-queued steers are discarded — same
/// contract as the parent's abort dropping its queue; the frontend
/// removes the corresponding queued rows on `SubagentFinished`.
fn unregister_subagent_steer_channel(subagent_id: &str) {
	subagent_steer_registry()
		.lock()
		.expect("steer registry poisoned")
		.remove(subagent_id);
}

fn take_queued_subagent_steers(subagent_id: &str) -> Vec<QueuedSubagentSteer> {
	let mut registry = subagent_steer_registry().lock().expect("steer registry poisoned");
	match registry.get_mut(subagent_id) {
		Some(channel) => std::mem::take(&mut channel.queue),
		None => Vec::new(),
	}
}

fn has_queued_subagent_steers(subagent_id: &str) -> bool {
	let registry = subagent_steer_registry().lock().expect("steer registry poisoned");
	registry.get(subagent_id).map(|c| !c.queue.is_empty()).unwrap_or(false)
}

/// Caller-provided plan for one sub-agent run. Built by the
/// parent's `task` tool dispatch from validated args.
#[derive(Debug, Clone)]
pub struct Subagent {
	pub id: String,
	pub parent_session_id: String,
	pub parent_tool_call_id: String,
	/// Bound folder of the **parent** session that originated this
	/// sub-agent. Persistence routes the sub-agent's JSONL under
	/// this folder's slug so it shows up in the parent project's
	/// session list (per the multi-session decision: sub-agents
	/// belong to whichever project originated them, not whichever
	/// folder they happened to operate against).
	pub parent_folder: Utf8PathBuf,
	pub task: String,
	pub system_prompt_override: Option<String>,
	pub mode: CoderMode,
	/// Whether this run was spawned with `detach: true` ([ADR
	/// 0053]): the parent's tool call returns a handle immediately
	/// and the run settles in the background, surfaced via the
	/// detached registry + `task_collect` instead of a blocking
	/// tool result. `false` for every synchronous `task` call.
	pub detach: bool,
	/// Folder the sub-agent's tools operate against. May differ
	/// from `parent_folder` when the model passed an explicit
	/// `folder` argument to `task`. Surfaced as `target_folder`
	/// in events / persistence metadata.
	pub folder: Arc<WorkspaceFolderEntry>,
}

/// What the sub-agent runner returns to the parent's tool
/// dispatch. `result` is the only field the parent's model sees
/// (as the tool's stringified return value); the others are
/// metadata the UI uses to render the collapsed card / pop-out.
///
/// `Serialize` so a detached run ([ADR 0053]) can stash the
/// finished report in the registry and have `task_collect`
/// return it verbatim.
#[derive(Debug, Clone, serde::Serialize)]
pub struct SubagentReport {
	pub result: String,
	pub tokens_used_estimate: u32,
	pub sub_session_id: String,
	#[serde(serialize_with = "serialize_mode_wire")]
	pub mode: CoderMode,
	pub iterations_used: u32,
}

/// `CoderMode` isn't `Serialize` (it's a dispatch-level enum, not
/// a wire type); the report serialises it as its wire string.
fn serialize_mode_wire<S: serde::Serializer>(mode: &CoderMode, s: S) -> Result<S::Ok, S::Error> {
	s.serialize_str(mode.as_wire())
}

/// JSON tool definition for the `task` tool (delegation primitive
/// — what kicks off a sub-agent). Lives outside the `ToolRegistry`
/// so sub-agents (which use the registry's `definitions()`) don't
/// see it — that's how depth is enforced. The parent's `run_turn`
/// appends this to the tool list it advertises.
///
/// Wire name is `task` (matching the convention every other agent
/// product the team has used). The Rust types and module are still
/// named `Subagent` / `subagent.rs` because internally that's what
/// the call spawns; only the model-facing tool name reflects the
/// user-facing concept.
pub fn task_tool_definition() -> ToolDefinition {
	ToolDefinition::function(
		"task",
		"Delegate a self-contained task to a sub-agent and get back a single summarised string. Sub-agents run in their own context — you spend tokens on the task description and the final summary, not on every intermediate read or edit. See \"When to use sub-agents\" in your instructions for when to reach for this (context preservation, parallelism, scoped delegation); it is for *delegation*, not for *access* — your own tools already reach every bound folder. \
\
The sub-agent has no access to your conversation history — describe the task self-containedly. Sub-agents cannot spawn further sub-agents.",
		json!({
			"type": "object",
			"properties": {
				"task": {
					"type": "string",
					"description": "Self-contained description of what the sub-agent should do. Include any context the sub-agent needs — it does not see the parent's transcript."
				},
				"folder": {
					"type": "string",
					"description": "Basename of a currently-bound workspace folder to scope the sub-agent against (matches one of the names listed in the Bound folders section). Omit (or set to the active folder's basename) to target the parent's active folder — useful for context-isolation even within the same folder. Targeting an unbound folder errors."
				},
				"mode": {
					"type": "string",
					"enum": ["research", "agent"],
					"description": "`research` is read-only intent (read_file, list_dir, grep, bash for inspection); `agent` (default) is the full toolkit and may edit files. The sub-agent in `agent` mode has the same capabilities you do."
				},
				"system_prompt": {
					"type": "string",
					"description": "Optional override for the sub-agent's system prompt. Most callers should leave this empty and rely on the mode-default prompt."
				},
				"detach": {
					"type": "boolean",
					"description": "When true, the call returns immediately with a `subagent_id` handle instead of blocking on the report: the sub-agent runs in the background, its finish wakes you with a notice, and you fetch the report with `task_collect(subagent_id)` (steer it mid-run with `task_steer`, or stop it with `task_abort`). Default false (synchronous). Prefer detach for slow, independent work you don't need the answer to before continuing — long test suites, background audits."
				}
			},
			"required": ["task"]
		}),
	)
}

/// `task_collect` — fetch a detached sub-agent's report ([ADR
/// 0053]). Lives outside `ToolRegistry::definitions()` alongside
/// `task_tool_definition` so sub-agents never see it.
pub fn task_collect_tool_definition() -> ToolDefinition {
	ToolDefinition::function(
			"task_collect",
			"Fetch the report of a detached sub-agent (one spawned with `task({ ..., detach: true })`). Returns `{ status: \"done\", result, tokens_used_estimate, iterations_used }` once it settles, `{ status: \"error\", error }` on failure, or `{ status: \"running\" }` while it's still going. Pass `wait_ms` to block until it settles (capped at 60 s) instead of busy-polling. Only ids your own session spawned with `detach: true` are collectable.",
			json!({
				"type": "object",
				"properties": {
					"subagent_id": {
						"type": "string",
						"description": "The `subagent_id` a detached `task` call returned."
					},
					"wait_ms": {
						"type": "integer",
						"description": "Optional. Block up to this many milliseconds for the run to settle (capped at 60000). Omit to return immediately with the current status."
					}
				},
				"required": ["subagent_id"]
			}),
		)
}

/// `task_steer` — queue a steering message into a running detached
/// sub-agent (ADR 0071). Rides the same steer channel as the
/// pop-out composer, so delivery semantics are identical: queued
/// now, drained at the top of the sub-agent's next iteration.
/// Advertised alongside `task_collect` / `task_abort` to `Agent`
/// mode only.
pub fn task_steer_tool_definition() -> ToolDefinition {
	ToolDefinition::function(
			"task_steer",
			"Send a steering message to a running detached sub-agent (one spawned with `task({ ..., detach: true })`). The message is queued and delivered at the top of the sub-agent's next iteration — the same way a user steers a session. Use it to redirect a run that's drifting, narrow or extend its scope, or feed it something you just learned; don't use it to poll (use `task_collect` with `wait_ms` for that). Returns `{ status: \"steered\" }`, or `{ status: \"not_running\" }` when the run already settled (collect its report instead). Only ids your own session spawned with `detach: true` can be steered.",
			json!({
				"type": "object",
				"properties": {
					"subagent_id": {
						"type": "string",
						"description": "The `subagent_id` a detached `task` call returned."
					},
					"text": {
						"type": "string",
						"description": "The steering message. Self-contained — the sub-agent doesn't see your conversation history."
					}
				},
				"required": ["subagent_id", "text"]
			}),
		)
}

/// `task_abort` — cancel a running detached sub-agent ([ADR
/// 0053]). Scoped to the one sub-agent; never touches the parent
/// turn or sibling sub-agents.
pub fn task_abort_tool_definition() -> ToolDefinition {
	ToolDefinition::function(
			"task_abort",
			"Cancel a running detached sub-agent. Its partial work is discarded (the turn's format-on-save flush still runs for whatever it already wrote). Returns `{ status: \"aborted\" }` when it was running, `{ status: \"not_running\" }` when it had already settled (its report is still available via `task_collect`). Never affects your own turn or other sub-agents.",
			json!({
				"type": "object",
				"properties": {
					"subagent_id": {
						"type": "string",
						"description": "The `subagent_id` a detached `task` call returned."
					}
				},
				"required": ["subagent_id"]
			}),
		)
}

/// Run a sub-agent to completion (or a budget cap, or
/// cancellation). Mirrors the structure of `run_turn` but emits
/// every event wrapped in [`CoderEvent::SubagentEvent`] so the UI
/// can route updates to the right collapsed card / pop-out pane.
/// Persists a JSONL transcript at
/// `<coder_sessions_dir>/<parent-folder-slug>/<parent-session-id>/<sub-id>.jsonl`,
/// with a header carrying the parent's session id + tool_call_id
/// so a "pop out" lookup survives IDE restarts.
pub(crate) async fn run_subagent(
	tools: &ToolRegistry,
	inference: &InferenceClient,
	sink: &FolderEventSink,
	coder_sessions_dir: &Utf8Path,
	models: &SharedCoderModels,
	spec: Subagent,
	cancel: CancellationToken,
) -> Result<SubagentReport, CoderError> {
	let id = spec.id.clone();
	let mode = spec.mode;
	// Open the steer channel before announcing the spawn, so the
	// pop-out can never see a running sub-agent whose composer
	// bounces. Closed (and any undelivered steers dropped) after
	// the loop settles, before `SubagentFinished` goes out.
	register_subagent_steer_channel(&id, sink.clone(), cancel.clone());
	sink.send(CoderEvent::SubagentSpawned {
		tool_call_id: spec.parent_tool_call_id.clone(),
		subagent_id: id.clone(),
		target_folder: spec.folder.folder.path.clone(),
		mode: mode.as_wire().to_string(),
		worktree_root: None,
		worker: false,
		detached: spec.detach,
	});

	// Mirror the task into the sub-agent's own transcript as a
	// user message, so the pop-out shows what the sub-agent was
	// asked to do. Matches the persisted `SessionRecord::User`
	// that `subagent_replay_inners` re-emits on reload — without
	// this the live view starts at the first assistant delta.
	sink.send(wrap_inner(
		&id,
		CoderEvent::UserMessage {
			id: format!("{id}::task"),
			text: spec.task.clone(),
			images: Vec::new(),
			queued: false,
			created_at_ms: Some(current_time_ms()),
			from_coordinator: false,
		},
	));

	// Header seed only — the per-round-trip model comes from the
	// fresh snapshot at the top of every loop iteration.
	let standard_model = models.read().await.standard().to_owned();
	let system_prompt = spec
		.system_prompt_override
		.clone()
		.unwrap_or_else(|| build_subagent_system_prompt(spec.mode, &spec.folder, &spec.task));
	let messages: Vec<ChatMessage> = vec![
		ChatMessage::System { content: system_prompt },
		ChatMessage::user(spec.task.clone()),
	];

	// JSONL transcript lives under the **parent** folder's slug,
	// nested inside a per-parent-session subdirectory:
	// `<sessions_dir>/<parent_session_id>/<sub-id>.jsonl`. Sub-agents
	// belong to whichever project originated them, and grouping by
	// parent session means listing the sessions dir flat returns
	// only top-level sessions (the picker stays clean) while
	// `<parent_id>/` keeps every sub-agent that ran during that
	// conversation in one obvious spot. The header carries
	// `target_folder` as metadata so the UI can still show which
	// folder the sub-agent was scoped to.
	let parent_dir = sessions_dir(coder_sessions_dir, spec.parent_folder.as_path());
	let session_dir = subagent_session_dir(&parent_dir, &spec.parent_session_id);
	let now = current_time_ms();
	// `subagent_target_folder` is `Some(...)` only when the
	// sub-agent operated against a folder different from the
	// parent's (i.e. an explicit `folder` argument was passed to
	// `task`). When the sub-agent targets the same folder as its
	// parent, `None` keeps the on-disk header tidy.
	let target_folder_path = spec.folder.folder.path.clone();
	let target_differs = Utf8Path::new(target_folder_path.as_str()) != spec.parent_folder.as_path();
	let header = SessionHeader {
		schema: SESSION_SCHEMA_VERSION,
		id: id.clone(),
		// `task` sub-agents have a parent, not an orchestrator.
		orchestrator_session_id: None,
		// Sub-agents operate against their target folder; that's
		// the `cwd` the pi viewer should show for this trace.
		// Sub-agents that fall back to the parent's folder still
		// get a real path here (we use `target_folder_path`
		// regardless of `target_differs`).
		cwd: target_folder_path.clone(),
		title: subagent_session_title(&spec.task),
		created_at_ms: now,
		updated_at_ms: now,
		// Informational seed; the actual model used per round-trip
		// comes from the parent's `CoderModels` snapshot at
		// `run_subagent` time. See note in
		// [`crate::runner::Session::new_blank`].
		model: standard_model.clone(),
		parent_session_id: Some(spec.parent_session_id.clone()),
		parent_tool_call_id: Some(spec.parent_tool_call_id.clone()),
		subagent_mode: Some(spec.mode.as_wire().to_string()),
		// Sub-agents are never top-level, so they carry no top-level
		// `mode` — their mode lives in `subagent_mode` above.
		mode: None,
		subagent_target_folder: if target_differs { Some(target_folder_path) } else { None },
		// Sub-agents always run with auto bash routing; a forced-host
		// parent session doesn't leak its override into delegated work.
		bash_target_override: None,
		// Sub-agents don't re-route via the header: their tools already
		// run against `spec.folder` (which is the worktree when the
		// parent is worktree-backed), so `cwd` above already points
		// there and no extra routing field is needed.
		worktree_root: None,
		worktree_branch: None,
		// Branch association is a top-level-session concern (the user
		// commits from the panel); sub-agents never carry one.
		committed_branch: None,
	};

	// Best-effort: persistence failures log at warn but never
	// fail the sub-agent. The model still gets its answer; the
	// only loss is an absent transcript on disk for the pop-out.
	persist_subagent(
		&session_dir,
		&header,
		&SessionRecord::User {
			text: spec.task.clone(),
			images: Vec::new(),
			from_coordinator: false,
		},
	)
	.await;

	let state = SubagentLoopState {
		messages,
		last_usage: None,
		todos: Vec::new(),
	};
	run_subagent_wrapped(
		tools,
		inference,
		sink,
		models,
		&spec,
		session_dir,
		header,
		state,
		cancel,
	)
	.await
}

/// Resume a *finished* sub-agent with a user follow-up (the
/// pop-out composer after the run settled). The caller —
/// [`crate::runner::Coder::continue_subagent`] — already rebuilt
/// the message history from the sub-agent's JSONL and claimed the
/// steer channel via [`claim_subagent_for_resume`], so this only
/// re-announces the spawn (the frontend keeps the existing
/// transcript rows and flips the card back to `running`), appends
/// and persists the follow-up, and re-enters the shared loop. The
/// parent agent is deliberately not involved: a follow-up is
/// user ↔ sub-agent; the parent's history still ends at the
/// original `task` tool result.
pub(crate) async fn resume_subagent(
	tools: &ToolRegistry,
	inference: &InferenceClient,
	sink: &FolderEventSink,
	models: &SharedCoderModels,
	resume: SubagentResume,
	cancel: CancellationToken,
) -> Result<SubagentReport, CoderError> {
	let SubagentResume {
		spec,
		session_dir,
		header,
		mut messages,
		last_usage,
		todos,
		follow_up,
	} = resume;
	let id = spec.id.clone();
	sink.send(CoderEvent::SubagentSpawned {
		tool_call_id: spec.parent_tool_call_id.clone(),
		subagent_id: id.clone(),
		target_folder: spec.folder.folder.path.clone(),
		mode: spec.mode.as_wire().to_string(),
		worktree_root: None,
		worker: false,
		// A user-driven resume keeps the original spawn's detach
		// semantics; the resume itself doesn't restamp the flag
		// (`spec.detach` is false here by construction).
		detached: spec.detach,
	});
	sink.send(wrap_inner(
		&id,
		CoderEvent::UserMessage {
			id: crate::runner::new_message_id(),
			text: follow_up.clone(),
			images: Vec::new(),
			queued: false,
			created_at_ms: Some(current_time_ms()),
			from_coordinator: false,
		},
	));
	messages.push(ChatMessage::user(follow_up.clone()));
	persist_subagent(
		&session_dir,
		&header,
		&SessionRecord::User {
			text: follow_up,
			images: Vec::new(),
			from_coordinator: false,
		},
	)
	.await;

	let state = SubagentLoopState {
		messages,
		last_usage,
		todos,
	};
	run_subagent_wrapped(
		tools,
		inference,
		sink,
		models,
		&spec,
		session_dir,
		header,
		state,
		cancel,
	)
	.await
}

/// Everything [`crate::runner::Coder::continue_subagent`] rebuilds
/// from a finished sub-agent's JSONL before handing off to
/// [`resume_subagent`].
pub(crate) struct SubagentResume {
	pub spec: Subagent,
	pub session_dir: Utf8PathBuf,
	pub header: SessionHeader,
	pub messages: Vec<ChatMessage>,
	pub last_usage: Option<TokenUsage>,
	pub todos: Vec<crate::TodoItem>,
	pub follow_up: String,
}

/// Mutable loop state threaded into [`run_subagent_loop`]. Fresh
/// runs start with just the system + task messages; resumed runs
/// carry the history rebuilt from the sub-agent's JSONL.
struct SubagentLoopState {
	messages: Vec<ChatMessage>,
	/// Most-recent provider-supplied usage. Populated whenever an
	/// LLM call returns a `usage` block; `None` when every round
	/// fell back to the bytes/4 estimate. Drives the compaction
	/// trigger and the eventual `tokens_used_estimate` the parent
	/// sees back.
	last_usage: Option<TokenUsage>,
	/// Sub-agent-local todo list. Sub-agents see `todo_write`
	/// advertised on the same footing the parent does and maintain
	/// their own scratchpad; it never bubbles up to the parent's
	/// pill but the per-call result renders inside the sub-agent's
	/// transcript card.
	todos: Vec<crate::TodoItem>,
}

/// Shared body of [`run_subagent`] and [`resume_subagent`]: format
/// queue + background-process lifecycle around the LLM loop, then
/// channel teardown and the `SubagentFinished` announcement.
#[allow(clippy::too_many_arguments)]
async fn run_subagent_wrapped(
	tools: &ToolRegistry,
	inference: &InferenceClient,
	sink: &FolderEventSink,
	models: &SharedCoderModels,
	spec: &Subagent,
	session_dir: Utf8PathBuf,
	header: SessionHeader,
	state: SubagentLoopState,
	cancel: CancellationToken,
) -> Result<SubagentReport, CoderError> {
	let id = spec.id.clone();
	// Refresh the target folder's remote-tracking refs so a
	// sub-agent that inspects `origin/main` sees fresh data.
	// Fire-and-forget + per-folder throttled; see the helper.
	kick_off_subagent_fetch(&spec.folder);

	// Fresh per-turn format queue. The sub-agent's write/edit
	// tools record into it instead of running the lint-staged
	// chain inline; we drain it below regardless of how the
	// sub-agent ended. See the ADR superseding 0013's per-tool-
	// call invocation.
	let format_queue = Arc::new(crate::tools::FormatQueue::default());
	// Per-turn background-process registry (ADR 0034). Same
	// lifetime as the format queue — cleaned up regardless of how
	// the sub-agent ended.
	let background = Arc::new(crate::tools::BackgroundProcessRegistry::default());

	// Scoped per sub-agent id, not inherited from the parent turn:
	// each sub-agent has its own stable prompt prefix, so pinning it
	// to its own fireworks replica (via `X-HF-Session-Id`) gets its
	// cache hits without competing with the parent's. Also covers
	// detached sub-agents, which run outside the parent's task scope.
	let outcome = crate::inference::SESSION_HINT
		.scope(
			id.clone(),
			run_subagent_loop(
				tools,
				inference,
				sink,
				models,
				spec,
				&session_dir,
				&header,
				state,
				cancel,
				format_queue.clone(),
				background.clone(),
			),
		)
		.await;

	// Flush before announcing the finish so by the time the
	// parent's transcript shows `SubagentFinished` the files are
	// already formatted on disk — important because the parent's
	// next iteration may `read_file` against one of those paths
	// to verify the sub-agent's work. Sub-agent tools today only
	// write inside `spec.folder` (path routing happens in the
	// parent's `ToolContext`, not the sub-agent's), so we can
	// pull the host straight off the spec instead of looking it
	// up through the workspace registry.
	for (_folder_path, rel) in format_queue.drain() {
		if let Err(err) = spec.folder.host.format_file(&rel).await {
			tracing::warn!(path = %rel, %err, "format-on-save (sub-agent turn end): format_file failed");
		}
	}
	// Kill + reap any detached processes the sub-agent left running.
	background.cleanup().await;
	unregister_subagent_steer_channel(&id);

	let was_error = outcome.is_err();
	sink.send(CoderEvent::SubagentFinished {
		subagent_id: id,
		tokens_used_estimate: outcome.as_ref().ok().map(|r| r.tokens_used_estimate).unwrap_or(0),
		was_error,
	});
	outcome
}

#[allow(clippy::too_many_arguments)]
async fn run_subagent_loop(
	tools: &ToolRegistry,
	inference: &InferenceClient,
	sink: &FolderEventSink,
	models: &SharedCoderModels,
	spec: &Subagent,
	session_dir: &Utf8Path,
	header: &SessionHeader,
	state: SubagentLoopState,
	cancel: CancellationToken,
	format_queue: Arc<crate::tools::FormatQueue>,
	background: Arc<crate::tools::BackgroundProcessRegistry>,
) -> Result<SubagentReport, CoderError> {
	let id = spec.id.clone();
	let SubagentLoopState {
		mut messages,
		mut last_usage,
		mut todos,
	} = state;

	// The sub-agent's tool list deliberately omits `task`. That's
	// how the depth=1 cap is enforced: a sub-agent literally cannot
	// describe a sub-sub-agent because the model never sees the
	// tool.
	let mut tool_defs = tools.definitions();
	// MCP meta-tools ride along when the workspace has enabled
	// servers — a sub-agent driving playwright against its target
	// folder's dev server is a natural delegation.
	tool_defs.extend(tools.mcp_definitions().await);
	let cx = ToolContext::with_format_queue(spec.folder.clone(), spec.mode, format_queue).with_background(background);
	// Terminals of the sub-agent's *own* target folder (ADR 0048) —
	// the same folder its `bash` and `grep` run against.
	tool_defs.extend(tools.terminal_definitions(&cx).await);

	// Consecutive empty-shell responses — same retry-then-fail
	// posture as the parent loop in `run_turn`. Without it a
	// provider that bails mid-stream makes the sub-agent return an
	// empty `result` as a *success*, which the parent model (and
	// the user) reads as "the sub-agent found nothing".
	let mut empty_shell_attempts: usize = 0;
	// Fragments of a report the provider cut off at the
	// output-token ceiling, in order. The parent gets them joined
	// back together — a sub-agent report that stops mid-sentence is
	// worse than useless, because the parent has no way to tell a
	// truncated audit from a complete one.
	let mut report_fragments: Vec<String> = Vec::new();
	let mut output_cap_continuations: usize = 0;
	for iter in 0..MAX_TURN_ITERATIONS {
		if cancel.is_cancelled() {
			return Err(CoderError::Aborted);
		}

		// Drain any steers the user queued from the pop-out
		// composer: append to the history + persist, then swap the
		// provisional queued bubble for a real message at the
		// bottom — remove the placeholder via `SteerDrained` and
		// re-append a fresh `UserMessage { queued: false }`. Same
		// contract as the parent loop's `drain_pending_steers`
		// (steers persist at drain time, in queue order, and land in
		// history after the answer that was already streaming).
		for steer in take_queued_subagent_steers(&id) {
			messages.push(ChatMessage::user(steer.text.clone()));
			persist_subagent(
				session_dir,
				header,
				&SessionRecord::User {
					text: steer.text.clone(),
					images: Vec::new(),
					from_coordinator: steer.from_coordinator,
				},
			)
			.await;
			sink.send(wrap_inner(&id, CoderEvent::SteerDrained { id: steer.id }));
			sink.send(wrap_inner(
				&id,
				CoderEvent::UserMessage {
					id: crate::runner::new_message_id(),
					text: steer.text,
					images: Vec::new(),
					queued: false,
					created_at_ms: Some(crate::sessions::current_time_ms()),
					from_coordinator: steer.from_coordinator,
				},
			));
		}

		// Fresh model snapshot per round-trip — same discipline as
		// the parent loop in `run_turn`: the inference client
		// resolves the *route* from the live shared handle per
		// request, so the model slug must come from the same
		// generation or a provider switch mid-run pairs the old
		// provider's slug with the new provider's endpoint.
		let models = models.read().await.clone();
		let standard_model = models.standard().to_owned();
		let pi_model = models.resolve_route().pi_provider_model(&standard_model);

		// Compaction-before-send: if the last round's prompt size
		// is bumping into the model's context window, fold the
		// older messages into a synthetic system summary before
		// the next request goes out. No-op early in the run.
		// Persist a `Compaction` record into the sub-agent's
		// JSONL alongside the parent path, so a future "reopen
		// sub-agent transcript" feature replays into the same
		// compacted shape instead of re-inflating the prefix.
		// Sub-agents don't run the image-elision budget, so an empty
		// set reproduces their previous round-trips' wire shape.
		let no_elisions = std::collections::HashSet::new();
		let compaction_outcome = compaction::compact_if_needed(
			inference,
			sink,
			Some(&id),
			&models,
			&tool_defs,
			last_usage.as_ref(),
			&mut messages,
			&no_elisions,
			&cancel,
		)
		.await;
		if let Some(applied) = compaction_outcome {
			// Re-anchor on the compacted prompt's estimate (not
			// `None`), so the next iteration's compaction check
			// stays armed and re-fires if a single pass didn't get
			// us under the threshold. Mirrors the parent loop.
			let estimate = crate::runner::estimate_prompt_tokens(&messages);
			last_usage = Some(TokenUsage {
				prompt_tokens: estimate,
				completion_tokens: 0,
				total_tokens: estimate,
				cache_read_input_tokens: 0,
				cache_creation_input_tokens: 0,
			});
			persist_subagent(
				session_dir,
				header,
				&SessionRecord::Compaction {
					summary: applied.summary,
					messages_compacted: applied.messages_compacted,
					messages_kept: applied.messages_kept,
				},
			)
			.await;
		}

		let assistant_id = format!("{id}::msg-{iter}");
		let id_for_cb = assistant_id.clone();
		let id_for_subagent = id.clone();
		let sink_for_cb = sink.clone();
		let started = std::sync::atomic::AtomicBool::new(false);
		let mut response = inference
			.chat_completion_stream(
				&standard_model,
				&messages,
				&tool_defs,
				Some(crate::defaults::TURN_MAX_OUTPUT_TOKENS),
				&cancel,
				|event| match event {
					StreamEvent::ContentDelta { delta } => {
						if !started.swap(true, std::sync::atomic::Ordering::Relaxed) {
							sink_for_cb.send(wrap_inner(
								&id_for_subagent,
								CoderEvent::AssistantMessageStart { id: id_for_cb.clone() },
							));
						}
						sink_for_cb.send(wrap_inner(
							&id_for_subagent,
							CoderEvent::AssistantMessageDelta {
								id: id_for_cb.clone(),
								delta: delta.to_string(),
							},
						));
					}
					StreamEvent::ThinkingDelta { delta } => {
						if !started.swap(true, std::sync::atomic::Ordering::Relaxed) {
							sink_for_cb.send(wrap_inner(
								&id_for_subagent,
								CoderEvent::AssistantMessageStart { id: id_for_cb.clone() },
							));
						}
						sink_for_cb.send(wrap_inner(
							&id_for_subagent,
							CoderEvent::AssistantThinkingDelta {
								id: id_for_cb.clone(),
								delta: delta.to_string(),
							},
						));
					}
					StreamEvent::ToolCallDelta { .. } => {}
				},
			)
			.await?;
		// Same recycled-id guard as the parent loop — must run
		// before the response is pushed / persisted / dispatched.
		crate::runner::dedupe_response_tool_call_ids(&messages, &mut response.tool_calls);
		// Same empty-shell guard the parent runner uses — skip
		// pushing / persisting / emitting `End` for an Anthropic
		// turn that bailed mid-stream. See
		// [`crate::runner::assistant_response_is_empty`].
		let response_is_empty = crate::runner::assistant_response_is_empty(&response);
		if started.into_inner() && !response_is_empty {
			sink.send(wrap_inner(
				&id,
				CoderEvent::AssistantMessageEnd {
					id: assistant_id.clone(),
					text: response.content.clone().unwrap_or_default(),
					thinking: response.thinking.clone(),
					created_at_ms: Some(current_time_ms()),
				},
			));
		}

		if let Some(u) = response.usage {
			last_usage = Some(u);
		}
		// Wrap the parent runner's helper so the sub-agent's ring
		// updates land on the same wire as parent updates. The
		// envelope is `SubagentEvent { inner: TokenUsage … }`.
		emit_subagent_token_usage(sink, &id, &models, &standard_model, &messages, &response);
		if !response_is_empty {
			empty_shell_attempts = 0;
			messages.push(response_to_message(&response));
			persist_subagent(
				session_dir,
				header,
				&SessionRecord::Assistant {
					content: response.content.clone(),
					thinking: response.thinking.clone(),
					thinking_blocks: response.thinking_blocks.clone(),
					tool_calls: response.tool_calls.clone(),
					model: Some(pi_model.clone()),
					stop_reason: response.stop_reason.clone(),
				},
			)
			.await;
		}

		if response.tool_calls.is_empty() {
			// Empty shell → retry the round-trip (nothing was
			// appended to `messages`), then fail loudly. The error
			// lands in the parent as the `task` tool's
			// `is_error: true` result, so the parent model can
			// react instead of trusting an empty summary.
			if response_is_empty {
				empty_shell_attempts += 1;
				if empty_shell_attempts > crate::defaults::EMPTY_RESPONSE_RETRIES {
					return Err(CoderError::EmptyResponse {
						attempts: empty_shell_attempts as u32,
					});
				}
				tracing::warn!(
					attempt = empty_shell_attempts,
					subagent = %id,
					"provider returned an empty response; retrying the round-trip"
				);
				continue;
			}
			// A steer that landed during the final assistant
			// message would otherwise be dropped on the floor —
			// give it another round-trip instead (the loop-top
			// drain picks it up). Mirrors the parent runner's
			// extra iteration for late steers.
			if has_queued_subagent_steers(&id) {
				continue;
			}
			// Cut off at the output-token ceiling: keep the
			// fragment and ask for the rest, so the parent gets
			// the whole report instead of everything up to the
			// sentence the ceiling landed in.
			if response.hit_output_cap() && output_cap_continuations < OUTPUT_CAP_CONTINUATIONS {
				output_cap_continuations += 1;
				tracing::warn!(
					subagent = %id,
					continuation = output_cap_continuations,
					"sub-agent report hit the output-token ceiling; asking the model to continue"
				);
				report_fragments.push(response.content.clone().unwrap_or_default());
				messages.push(ChatMessage::user(OUTPUT_CAP_CONTINUATION_PROMPT.to_owned()));
				persist_subagent(
					session_dir,
					header,
					&SessionRecord::User {
						text: OUTPUT_CAP_CONTINUATION_PROMPT.to_owned(),
						images: Vec::new(),
						from_coordinator: false,
					},
				)
				.await;
				continue;
			}
			report_fragments.push(response.content.clone().unwrap_or_default());
			// Still truncated after the continuation budget: say
			// so in the report itself. The parent model reads the
			// result as prose, so an inline marker is the only
			// channel that reliably reaches it.
			if response.hit_output_cap() {
				report_fragments.push(format!(
					"\n\n[Report truncated: the model hit the output-token ceiling after \
{OUTPUT_CAP_CONTINUATIONS} continuations. Treat everything above as partial.]"
				));
			}
			return Ok(SubagentReport {
				result: report_fragments.join(""),
				tokens_used_estimate: tokens_used_for_report(&messages, last_usage),
				sub_session_id: id.clone(),
				mode: spec.mode,
				iterations_used: (iter + 1) as u32,
			});
		}

		// The model went back to tool work after a continuation, so
		// whatever it had started writing is superseded by the
		// report it'll write at the end. Drop the fragments — they
		// stay in `messages` as history either way.
		report_fragments.clear();

		// Sub-agents dispatch their tools sequentially today —
		// recursive parallelism (a sub-agent's own tools running
		// concurrently) is out of scope for the current slice.
		// Parallelism happens *one level up*: multiple sub-agents
		// in the parent's batch run concurrently via the parent's
		// `dispatch_subagent_batch`.
		for call in &response.tool_calls {
			if cancel.is_cancelled() {
				return Err(CoderError::Aborted);
			}
			// Same refusal as the parent loop: arguments that don't
			// parse mean the ceiling landed inside them, so the
			// call never runs. See `tool_args_or_refusal`.
			let refusal = crate::runner::tool_args_or_refusal(&call.function, response.hit_output_cap());
			let args = refusal
				.as_ref()
				.ok()
				.cloned()
				.unwrap_or_else(|| Value::Object(Default::default()));
			sink.send(wrap_inner(
				&id,
				CoderEvent::ToolCall {
					id: call.id.clone(),
					name: call.function.name.clone(),
					args: args.clone(),
					started_at_ms: Some(current_time_ms()),
				},
			));
			let dispatched_at = std::time::Instant::now();
			let outcome = if let Err(err) = refusal {
				Err(err)
			} else if call.function.name == "todo_write" {
				// Same short-circuit shape as the parent runner —
				// `todo_write` mutates per-session state
				// (`todos`), so it can't go through the stateless
				// registry dispatch.
				handle_subagent_todo_write(&mut todos, &args, session_dir, header).await
			} else {
				tools.dispatch(&call.function.name, &args, &cx, &cancel).await
			};
			let duration_ms = u64::try_from(dispatched_at.elapsed().as_millis()).ok();
			// Same image convention as the parent runner: a
			// tool's `images` key leaves the text projection and
			// rides typed (the sub-agent's `read_file` on a
			// screenshot is the case that needs it).
			let (content, is_error, images) = match outcome {
				Ok(value) => {
					let (images, text_value) = crate::runner::split_tool_images(value);
					(text_value.to_string(), false, images)
				}
				Err(CoderError::Aborted) => return Err(CoderError::Aborted),
				Err(err) => (json!({ "error": err.to_string() }).to_string(), true, Vec::new()),
			};
			let payload: Value = serde_json::from_str(&content).unwrap_or_else(|_| Value::String(content.clone()));
			sink.send(wrap_inner(
				&id,
				CoderEvent::ToolResult {
					id: call.id.clone(),
					result: payload,
					is_error,
					duration_ms,
				},
			));
			messages.push(ChatMessage::Tool {
				tool_call_id: call.id.clone(),
				content: content.clone(),
				images: images.clone(),
			});
			persist_subagent(
				session_dir,
				header,
				&SessionRecord::Tool {
					tool_call_id: call.id.clone(),
					tool_name: call.function.name.clone(),
					content,
					duration_ms,
					images,
				},
			)
			.await;
		}
	}

	// Iteration cap reached. Mirror the parent's behaviour: ask
	// the model for one final tools-disabled wrap-up turn so the
	// parent gets a real answer back instead of a canned "stopped
	// after N iterations" stub. The wrap-up reply becomes the
	// sub-agent's `result` string (with a leading note so the
	// parent's model knows the budget was exhausted).
	let wrap_up_models = models.read().await.clone();
	let standard_model = wrap_up_models.standard().to_owned();
	let pi_model = wrap_up_models.resolve_route().pi_provider_model(&standard_model);
	let wrap_up = subagent_wrap_up(
		inference,
		sink,
		spec,
		session_dir,
		header,
		&standard_model,
		&pi_model,
		&mut messages,
		&cancel,
	)
	.await;
	let result = match wrap_up {
		Ok(text) if !text.trim().is_empty() => {
			format!("[Sub-agent reached the {MAX_TURN_ITERATIONS}-iteration cap; final wrap-up follows.]\n\n{text}")
		}
		_ => format!("Sub-agent stopped after {MAX_TURN_ITERATIONS} iterations without producing a final answer.",),
	};
	Ok(SubagentReport {
		result,
		tokens_used_estimate: tokens_used_for_report(&messages, last_usage),
		sub_session_id: id.clone(),
		mode: spec.mode,
		iterations_used: MAX_TURN_ITERATIONS as u32,
	})
}

/// Final tools-disabled round-trip the sub-agent runs after its
/// iteration cap is hit. Same idea as the parent's
/// [`wrap_up_final_answer`](crate::runner::wrap_up_final_answer):
/// inject a sentinel user message asking the model to write its
/// best answer with what it has, then call inference with
/// `tools = []` so it can't loop again. Streams the response
/// inside [`CoderEvent::SubagentEvent`] envelopes so the pop-out
/// UI sees it land like any other sub-agent turn.
///
/// Returns the text of the wrap-up answer for the caller to wire
/// into [`SubagentReport::result`]. Errors / empty responses fall
/// back to the historical canned message in the caller.
// Eight args is a hair over clippy's seven-arg cap, but every one
// of them is genuinely needed by the wrap-up flow — `header` for
// persistence (sub-agents write their wrap-up reply into the same
// JSONL as the rest of the run), `session_dir` likewise,
// `standard_model` for the LLM call. Bundling any of these into
// an aux struct just for the arg-count gate would obscure the
// call site for no real signal.
#[allow(clippy::too_many_arguments)]
async fn subagent_wrap_up(
	inference: &InferenceClient,
	sink: &FolderEventSink,
	spec: &Subagent,
	session_dir: &Utf8Path,
	header: &SessionHeader,
	standard_model: &str,
	pi_model: &str,
	messages: &mut Vec<ChatMessage>,
	cancel: &CancellationToken,
) -> Result<String, CoderError> {
	let id = spec.id.as_str();
	tracing::info!(
		subagent_id = %id,
		iterations = MAX_TURN_ITERATIONS,
		"sub-agent iteration cap reached; running final tools-disabled wrap-up",
	);
	let sentinel = format!(
		"[Tool-call budget exhausted: you've used all {MAX_TURN_ITERATIONS} tool-call iterations available for this sub-agent. \
Do not call any more tools. Write a final response now using only what you've already gathered: summarise findings, what's still unfinished, and any uncertainty.]"
	);
	messages.push(ChatMessage::user(sentinel.clone()));
	persist_subagent(
		session_dir,
		header,
		&SessionRecord::User {
			text: sentinel.clone(),
			images: Vec::new(),
			from_coordinator: false,
		},
	)
	.await;
	// Live-mirror the sentinel into the pop-out, same as the
	// task message at spawn — the persisted record above covers
	// the reload/replay path.
	sink.send(wrap_inner(
		id,
		CoderEvent::UserMessage {
			id: format!("{id}::wrap-up-sentinel"),
			text: sentinel,
			images: Vec::new(),
			queued: false,
			created_at_ms: Some(current_time_ms()),
			from_coordinator: false,
		},
	));

	let assistant_id = format!("{id}::wrap-up");
	let id_for_cb = assistant_id.clone();
	let id_for_subagent = id.to_string();
	let sink_for_cb = sink.clone();
	let started = std::sync::atomic::AtomicBool::new(false);
	let response = inference
		.chat_completion_stream(standard_model, messages, &[], None, cancel, |event| match event {
			StreamEvent::ContentDelta { delta } => {
				if !started.swap(true, std::sync::atomic::Ordering::Relaxed) {
					sink_for_cb.send(wrap_inner(
						&id_for_subagent,
						CoderEvent::AssistantMessageStart { id: id_for_cb.clone() },
					));
				}
				sink_for_cb.send(wrap_inner(
					&id_for_subagent,
					CoderEvent::AssistantMessageDelta {
						id: id_for_cb.clone(),
						delta: delta.to_string(),
					},
				));
			}
			StreamEvent::ThinkingDelta { delta } => {
				if !started.swap(true, std::sync::atomic::Ordering::Relaxed) {
					sink_for_cb.send(wrap_inner(
						&id_for_subagent,
						CoderEvent::AssistantMessageStart { id: id_for_cb.clone() },
					));
				}
				sink_for_cb.send(wrap_inner(
					&id_for_subagent,
					CoderEvent::AssistantThinkingDelta {
						id: id_for_cb.clone(),
						delta: delta.to_string(),
					},
				));
			}
			// Tools were disabled in the request; drop any stragglers.
			StreamEvent::ToolCallDelta { .. } => {}
		})
		.await?;

	// Same empty-shell guard as the main sub-agent loop.
	let response_is_empty = crate::runner::assistant_response_is_empty(&response);
	if started.into_inner() && !response_is_empty {
		sink.send(wrap_inner(
			id,
			CoderEvent::AssistantMessageEnd {
				id: assistant_id,
				text: response.content.clone().unwrap_or_default(),
				thinking: response.thinking.clone(),
				created_at_ms: Some(current_time_ms()),
			},
		));
	}

	let final_text = response.content.clone().unwrap_or_default();
	if !response_is_empty {
		messages.push(response_to_message(&response));
		persist_subagent(
			session_dir,
			header,
			&SessionRecord::Assistant {
				content: response.content.clone(),
				thinking: response.thinking.clone(),
				thinking_blocks: response.thinking_blocks.clone(),
				tool_calls: response.tool_calls.clone(),
				model: Some(pi_model.to_string()),
				stop_reason: response.stop_reason.clone(),
			},
		)
		.await;
	}
	Ok(final_text)
}

/// Sub-agent-specific wrapper around the parent runner's
/// [`emit_token_usage`](crate::runner::emit_token_usage). The
/// inner event is a [`CoderEvent::TokenUsage`] just like the
/// parent's, but wrapped in [`CoderEvent::SubagentEvent`] so the
/// frontend can target the right nested ring.
fn emit_subagent_token_usage(
	sink: &FolderEventSink,
	subagent_id: &str,
	models: &crate::models::CoderModels,
	model_slug: &str,
	messages: &[ChatMessage],
	response: &AssistantResponse,
) {
	let context_window = models.context_window(model_slug);
	let (prompt_tokens, completion_tokens, total_tokens, cache_read_tokens, cache_creation_tokens, source) =
		match response.usage {
			Some(u) => (
				u.prompt_tokens,
				u.completion_tokens,
				u.total_tokens,
				u.cache_read_input_tokens,
				u.cache_creation_input_tokens,
				crate::event::TokenUsageSource::Provider,
			),
			None => {
				let prompt = crate::runner::estimate_prompt_tokens(messages);
				// Mirror the parent's bytes/4 for completion side; the
				// inference module's helper isn't reachable here and
				// duplicating four lines is cleaner than threading it.
				let completion = (response.content.as_deref().map(str::len).unwrap_or(0)
					+ response.thinking.as_deref().map(str::len).unwrap_or(0)
					+ response
						.tool_calls
						.iter()
						.map(|c| c.function.name.len() + c.function.arguments.len())
						.sum::<usize>()) as u32
					/ 4;
				(
					prompt,
					completion,
					prompt + completion,
					0,
					0,
					crate::event::TokenUsageSource::Estimate,
				)
			}
		};
	sink.send(wrap_inner(
		subagent_id,
		CoderEvent::TokenUsage {
			prompt_tokens,
			completion_tokens,
			total_tokens,
			context_window,
			source,
			cache_read_tokens,
			cache_creation_tokens,
			// Sub-agent pop-outs don't track a session scoreboard;
			// 0/0 renders as "no data" on both frontends.
			session_cache_hits: 0,
			session_requests: 0,
		},
	));
}

/// Choose the best available "tokens used" number for the report
/// the parent's tool dispatch sees. Prefer the final
/// provider-supplied usage; fall back to the parent runner's
/// bytes/4 estimator computed from the message history.
fn tokens_used_for_report(messages: &[ChatMessage], last_usage: Option<TokenUsage>) -> u32 {
	if let Some(u) = last_usage {
		return u.total_tokens.max(u.prompt_tokens + u.completion_tokens);
	}
	crate::runner::estimate_prompt_tokens(messages)
}

fn wrap_inner(subagent_id: &str, inner: CoderEvent) -> CoderEvent {
	CoderEvent::SubagentEvent {
		subagent_id: subagent_id.to_string(),
		inner: Box::new(inner),
	}
}

/// Compose a short, human-readable title for the sub-agent's
/// session list entry. Prefixed with `Sub-agent:` so the
/// persisted-session UI can tell parent and sub-agent sessions
/// apart at a glance even before the dedicated badge ships.
fn subagent_session_title(task: &str) -> String {
	const TITLE_BODY_LIMIT: usize = 60;
	let mut head = task.trim().lines().next().unwrap_or("").trim().to_string();
	if head.len() > TITLE_BODY_LIMIT {
		let mut idx = TITLE_BODY_LIMIT;
		while idx > 0 && !head.is_char_boundary(idx) {
			idx -= 1;
		}
		head.truncate(idx);
		head.push('…');
	}
	if head.is_empty() {
		"Sub-agent".into()
	} else {
		format!("Sub-agent: {head}")
	}
}

async fn persist_subagent(dir: &Utf8Path, header: &SessionHeader, record: &SessionRecord) {
	if let Err(err) = sessions::append_record(dir, header, record).await {
		// Persistence is best-effort: the parent's tool result
		// still carries the sub-agent's answer; the only thing
		// we lose on a write failure is the per-step transcript
		// the pop-out view would render. Logged at warn so it
		// doesn't slip silently.
		tracing::warn!(error = %err, "failed to persist sub-agent record");
	}
}

fn response_to_message(response: &AssistantResponse) -> ChatMessage {
	ChatMessage::Assistant {
		content: response.content.clone(),
		thinking_blocks: response.thinking_blocks.clone(),
		tool_calls: response.tool_calls.clone(),
	}
}

/// Sub-agent counterpart to [`crate::runner::handle_todo_write`].
/// Same wire shape, same validation, same persistence record —
/// the only difference is the storage cell: this one mutates the
/// sub-agent's local `todos` vec (passed by `&mut`) and writes
/// into the sub-agent's per-parent JSONL via [`persist_subagent`].
/// Sub-agents see `todo_write` advertised the same way the parent
/// does (per the spec), so each sub-agent gets its own scratchpad
/// without bleeding into the parent's plan.
async fn handle_subagent_todo_write(
	todos: &mut Vec<crate::TodoItem>,
	args: &Value,
	session_dir: &Utf8Path,
	header: &SessionHeader,
) -> Result<Value, CoderError> {
	#[derive(serde::Deserialize)]
	struct TodoWriteArgs {
		todos: Vec<crate::TodoItem>,
		#[serde(default)]
		merge: bool,
	}
	let parsed: TodoWriteArgs =
		serde_json::from_value(args.clone()).map_err(|err| CoderError::invalid_args("todo_write", err.to_string()))?;
	for item in &parsed.todos {
		if item.id.trim().is_empty() {
			return Err(CoderError::invalid_args(
				"todo_write",
				"todo item `id` must be a non-empty string",
			));
		}
	}
	let merged = crate::merge_todos(todos, parsed.todos, parsed.merge);
	*todos = merged.clone();
	persist_subagent(
		session_dir,
		header,
		&SessionRecord::TodosUpdate { todos: merged.clone() },
	)
	.await;
	Ok(json!({ "todos": merged }))
}

const RESEARCH_SYSTEM_PROMPT: &str = r#"You are a research sub-agent inside moon-ide. You have been spawned by a parent agent to gather information from a workspace folder and report back. Your job is investigation, not editing.

Tools available: `read_file`, `list_dir`, `grep`, `bash`, `read_process`, and `stop_process`. You are scoped to a single folder (shown below); `grep` and `bash` run against it, and relative paths resolve inside it. You cannot spawn further sub-agents.

The shell is for read-only inspection only — `git log`, `git diff --stat`, `cargo check`, `pytest --collect-only`, `ls -la`, `cat`, `wc`, and similar. Do **not** run commands that mutate the filesystem, network state, or remote services. No `git commit`, `git push`, `cargo build` (it writes to `target/`), `mv`, `rm`, `npm install`, `pip install`, redirection-to-file (`> path`, `>> path`), or anything that would change persistent state.

Return your findings as a single coherent text result when you finish. The parent will see only that string; do not address them as "you" or assume shared context.
"#;

const AGENT_SYSTEM_PROMPT: &str = r#"You are an agent sub-agent inside moon-ide. You have been spawned by a parent agent to perform a focused task in a workspace folder. Your capabilities are the same as the parent's — you can read, search, run commands, and edit files freely.

Tools available: `read_file`, `list_dir`, `grep`, `bash`, `read_process`, `stop_process`, `write_file`, `edit_file`. You are scoped to a single folder (shown below); `grep` and `bash` run against it, and relative paths resolve inside it. You cannot spawn further sub-agents.

Read before you edit — don't invent file paths. Use `edit_file` for surgical changes inside large files; reach for `write_file` for new files and whole-file rewrites.

Return a single coherent text result when you finish. The parent will see only that string and a short transcript of your tool calls; do not address them as "you" or assume shared context.
"#;

pub(crate) fn build_subagent_system_prompt(mode: CoderMode, folder: &Arc<WorkspaceFolderEntry>, task: &str) -> String {
	let base = match mode {
		CoderMode::Research => RESEARCH_SYSTEM_PROMPT,
		CoderMode::Agent => AGENT_SYSTEM_PROMPT,
		// `Coordinator` is a top-level mode (ADR 0030); sub-agents
		// never run in it. Fall back to the agent prompt rather than
		// `unreachable!` so a future call path can't panic a turn.
		CoderMode::Coordinator => AGENT_SYSTEM_PROMPT,
	};
	let header = format!(
		"## Task\n\n{task}\n\n## Working folder\n\n- **{name}** at `{path}`\n",
		task = task.trim(),
		name = folder.folder.name,
		path = folder.folder.path,
	);
	let mut out = String::with_capacity(base.len() + header.len() + 16);
	out.push_str(base.trim_end());
	out.push_str("\n\n");
	out.push_str(&header);
	out
}

/// Validate + materialise a `Subagent` from JSON args + parent
/// context. Surfaces actionable errors back to the parent's model
/// (folder not bound, unknown mode/model strings) so it can
/// recover or adjust.
///
/// `parent_folder` is the absolute path of the parent session's
/// bound folder — the sub-agent's JSONL persists under this
/// folder's slug so the sub-agent file shows up in the parent
/// project's session list, regardless of which folder the
/// sub-agent's tools operate against.
pub fn build_subagent_spec(
	parent_session_id: String,
	parent_tool_call_id: String,
	parent_folder: Utf8PathBuf,
	args: &Value,
	parent_active_folder: &Arc<WorkspaceFolderEntry>,
	bound_folders: &[Arc<WorkspaceFolderEntry>],
) -> Result<Subagent, CoderError> {
	let task = args
		.get("task")
		.and_then(Value::as_str)
		.ok_or_else(|| CoderError::invalid_args("task", "missing required string field `task`"))?
		.to_string();
	if task.trim().is_empty() {
		return Err(CoderError::invalid_args("task", "`task` must not be empty"));
	}

	let folder = match args.get("folder").and_then(Value::as_str) {
		None | Some("") => parent_active_folder.clone(),
		Some(name) => {
			let found = find_bound_folder(bound_folders, name)
				.ok_or_else(|| CoderError::tool_failed("task", format!("folder `{name}` is not bound")))?;
			// A worktree-backed session naming its own parent project
			// means "my project" (ADR 0040): route the sub-agent to
			// the worktree so its work stays on the session's branch
			// instead of leaking into the parent's main working tree.
			match &parent_active_folder.folder.origin {
				moon_protocol::workspace::FolderOrigin::Worktree { parent_path, .. } if *parent_path == found.folder.path => {
					parent_active_folder.clone()
				}
				_ => found,
			}
		}
	};

	let mode = match args.get("mode").and_then(Value::as_str) {
		None | Some("agent") => CoderMode::Agent,
		Some("research") => CoderMode::Research,
		Some(other) => {
			return Err(CoderError::invalid_args(
				"task",
				format!("unknown mode `{other}` — expected `research` or `agent`"),
			));
		}
	};

	let system_prompt_override = args
		.get("system_prompt")
		.and_then(Value::as_str)
		.map(str::trim)
		.filter(|s| !s.is_empty())
		.map(str::to_string);

	let detach = args.get("detach").and_then(Value::as_bool).unwrap_or(false);

	Ok(Subagent {
		id: new_subagent_id(),
		parent_session_id,
		parent_tool_call_id,
		parent_folder,
		task,
		system_prompt_override,
		mode,
		detach,
		folder,
	})
}

fn find_bound_folder(folders: &[Arc<WorkspaceFolderEntry>], name: &str) -> Option<Arc<WorkspaceFolderEntry>> {
	// Match by basename first (the model usually says "moon-landing"
	// not the full absolute path) and fall back to absolute-path
	// equality so a model that does send a full path still works.
	if let Some(by_basename) = folders.iter().find(|f| f.folder.name == name) {
		return Some(by_basename.clone());
	}
	let candidate = Utf8Path::new(name);
	folders.iter().find(|f| f.folder.path == candidate.as_str()).cloned()
}

#[cfg(test)]
mod tests {
	use super::*;
	use moon_core::WorkspaceRegistry;
	use tempfile::TempDir;

	fn make_args(json_text: &str) -> Value {
		serde_json::from_str(json_text).unwrap()
	}

	async fn registry_with_folders(paths: &[&Utf8Path]) -> Vec<Arc<WorkspaceFolderEntry>> {
		let registry = WorkspaceRegistry::new("test-workspace".into());
		for path in paths {
			registry.add_folder(path.to_path_buf()).await.unwrap();
		}
		registry.folders().await
	}

	fn parent_folder_for(folders: &[Arc<WorkspaceFolderEntry>], idx: usize) -> Utf8PathBuf {
		Utf8PathBuf::from(folders[idx].folder.path.clone())
	}

	#[tokio::test]
	async fn build_spec_defaults_to_active_folder_and_coder_mode() {
		let dir = TempDir::new().unwrap();
		let path = camino::Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).unwrap();
		let folders = registry_with_folders(&[path.as_path()]).await;
		let active = folders[0].clone();
		let parent_folder = parent_folder_for(&folders, 0);
		let args = make_args(r#"{ "task": "do the thing" }"#);
		let spec = build_subagent_spec(
			"sess-x".into(),
			"call-1".into(),
			parent_folder.clone(),
			&args,
			&active,
			&folders,
		)
		.unwrap();
		assert_eq!(spec.mode, CoderMode::Agent);
		assert_eq!(spec.task, "do the thing");
		assert!(Arc::ptr_eq(&spec.folder, &active));
		assert_eq!(spec.parent_folder, parent_folder);
	}

	#[tokio::test]
	async fn build_spec_collapses_parent_target_to_worktree() {
		// A worktree-backed session that names its own parent
		// project targets the *worktree* (ADR 0040) — the sub-agent's
		// work must stay on the session's branch.
		let dir = TempDir::new().unwrap();
		let parent_path = Utf8PathBuf::from_path_buf(dir.path().canonicalize().unwrap()).unwrap();
		let wt_path = parent_path.join(".worktrees/moon-agent-1");
		std::fs::create_dir_all(wt_path.as_std_path()).unwrap();
		let registry = WorkspaceRegistry::new("test-workspace".into());
		registry.add_folder(parent_path.clone()).await.unwrap();
		registry
			.add_worktree_folder(wt_path, parent_path.to_string(), "moon/agent-1".into())
			.await
			.unwrap();
		let folders = registry.folders().await;
		let parent = folders[0].clone();
		let worktree = folders[1].clone();
		let args = make_args(&format!(
			r#"{{ "task": "do the thing", "folder": "{}" }}"#,
			parent.folder.name
		));
		let spec = build_subagent_spec(
			"sess-x".into(),
			"call-1".into(),
			parent_folder_for(&folders, 0),
			&args,
			&worktree,
			&folders,
		)
		.unwrap();
		assert!(
			Arc::ptr_eq(&spec.folder, &worktree),
			"parent-named target from a worktree session must collapse to the worktree"
		);
	}

	#[tokio::test]
	async fn build_spec_routes_research_mode_correctly() {
		let dir = TempDir::new().unwrap();
		let path = camino::Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).unwrap();
		let folders = registry_with_folders(&[path.as_path()]).await;
		let active = folders[0].clone();
		let args = make_args(r#"{ "task": "find auth code", "mode": "research" }"#);
		let spec = build_subagent_spec(
			"sess-x".into(),
			"call-1".into(),
			parent_folder_for(&folders, 0),
			&args,
			&active,
			&folders,
		)
		.unwrap();
		assert_eq!(spec.mode, CoderMode::Research);
	}

	#[tokio::test]
	async fn build_spec_rejects_unknown_mode_string() {
		let dir = TempDir::new().unwrap();
		let path = camino::Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).unwrap();
		let folders = registry_with_folders(&[path.as_path()]).await;
		let args = make_args(r#"{ "task": "x", "mode": "rogue" }"#);
		let err = build_subagent_spec(
			"sess-x".into(),
			"call-1".into(),
			parent_folder_for(&folders, 0),
			&args,
			&folders[0],
			&folders,
		)
		.unwrap_err();
		assert!(matches!(err, CoderError::InvalidToolArgs { .. }), "got {err:?}");
	}

	#[tokio::test]
	async fn build_spec_rejects_unbound_folder_basename() {
		let dir = TempDir::new().unwrap();
		let path = camino::Utf8PathBuf::from_path_buf(dir.path().to_path_buf()).unwrap();
		let folders = registry_with_folders(&[path.as_path()]).await;
		let args = make_args(r#"{ "task": "x", "folder": "no-such-folder" }"#);
		let err = build_subagent_spec(
			"sess-x".into(),
			"call-1".into(),
			parent_folder_for(&folders, 0),
			&args,
			&folders[0],
			&folders,
		)
		.unwrap_err();
		assert!(matches!(err, CoderError::ToolFailed { .. }), "got {err:?}");
	}

	#[tokio::test]
	async fn build_spec_finds_bound_folder_by_basename() {
		let one = TempDir::new().unwrap();
		let two = TempDir::new().unwrap();
		let one_path = camino::Utf8PathBuf::from_path_buf(one.path().to_path_buf()).unwrap();
		let two_path = camino::Utf8PathBuf::from_path_buf(two.path().to_path_buf()).unwrap();
		let folders = registry_with_folders(&[one_path.as_path(), two_path.as_path()]).await;
		let other_basename = folders[1].folder.name.clone();
		let args_text = format!(r#"{{ "task": "x", "folder": "{other_basename}" }}"#);
		let spec = build_subagent_spec(
			"sess-x".into(),
			"call-1".into(),
			parent_folder_for(&folders, 0),
			&make_args(&args_text),
			&folders[0],
			&folders,
		)
		.unwrap();
		assert!(Arc::ptr_eq(&spec.folder, &folders[1]));
		// Sub-agent's persistence still belongs to the **parent's**
		// folder slug, not the target's.
		assert_eq!(spec.parent_folder, parent_folder_for(&folders, 0));
	}

	#[test]
	fn build_subagent_system_prompt_includes_task_and_folder() {
		// Synthetic entry — we just need a folder shape, not an
		// actual host. Construct from `WorkspaceRegistry` would
		// require an existing dir, this is enough for the prompt
		// composer.
		let folder = Arc::new(WorkspaceFolderEntry {
			folder: moon_protocol::workspace::WorkspaceFolder {
				path: "/abs/path/to/proj".into(),
				name: "proj".into(),
				host: moon_protocol::workspace::HostKind::Local,
				origin: moon_protocol::workspace::FolderOrigin::UserPicked,
			},
			host: Arc::new(moon_core::LocalHost::new(camino::Utf8PathBuf::from(
				"/abs/path/to/proj",
			))),
		});
		let prompt = build_subagent_system_prompt(CoderMode::Research, &folder, "investigate auth flow");
		assert!(prompt.contains("research sub-agent"));
		assert!(prompt.contains("investigate auth flow"));
		assert!(prompt.contains("/abs/path/to/proj"));
	}

	#[test]
	fn steer_registry_queues_drains_and_rejects_after_unregister() {
		let (sender, mut receiver) = tokio::sync::broadcast::channel(16);
		let sink = FolderEventSink::new(sender, "/folder", "sess-parent");

		// Unknown id → rejected, nothing emitted.
		assert!(!queue_subagent_steer("sub-nope", "hello".into(), false));

		register_subagent_steer_channel("sub-steer-test", sink, CancellationToken::new());
		assert!(!has_queued_subagent_steers("sub-steer-test"));
		assert!(queue_subagent_steer("sub-steer-test", "first".into(), false));
		assert!(queue_subagent_steer("sub-steer-test", "second".into(), true));
		assert!(has_queued_subagent_steers("sub-steer-test"));

		// Queue-time emission: one wrapped queued UserMessage per
		// steer, carrying the sender flag (user vs parent agent).
		for (expected, expected_from_coordinator) in [("first", false), ("second", true)] {
			let envelope = receiver.try_recv().unwrap();
			let CoderEvent::SubagentEvent { subagent_id, inner } = envelope.event else {
				panic!("expected SubagentEvent");
			};
			assert_eq!(subagent_id, "sub-steer-test");
			let CoderEvent::UserMessage {
				text,
				queued,
				from_coordinator,
				..
			} = *inner
			else {
				panic!("expected wrapped UserMessage");
			};
			assert_eq!(text, expected);
			assert!(queued);
			assert_eq!(from_coordinator, expected_from_coordinator);
		}

		// Drain preserves queue order (and the sender flag) and
		// empties the channel.
		let drained = take_queued_subagent_steers("sub-steer-test");
		assert_eq!(
			drained
				.iter()
				.map(|s| (s.text.as_str(), s.from_coordinator))
				.collect::<Vec<_>>(),
			vec![("first", false), ("second", true)]
		);
		assert!(!has_queued_subagent_steers("sub-steer-test"));

		// After the run settles, the channel is gone and steers bounce.
		unregister_subagent_steer_channel("sub-steer-test");
		assert!(!queue_subagent_steer("sub-steer-test", "too late".into(), false));

		// Resume claim: no channel → registers one and hands back
		// its token; a raced second claim queues a steer instead.
		let sink = FolderEventSink::new(tokio::sync::broadcast::channel(16).0, "/folder", "sess-parent");
		let token = claim_subagent_for_resume("sub-steer-test", "follow up", &sink).expect("claim should register");
		assert!(claim_subagent_for_resume("sub-steer-test", "raced follow up", &sink).is_none());
		assert!(has_queued_subagent_steers("sub-steer-test"));

		// Per-sub-agent abort cancels the claimed token; after the
		// channel is gone it's a no-op.
		assert!(abort_subagent("sub-steer-test"));
		assert!(token.is_cancelled());
		unregister_subagent_steer_channel("sub-steer-test");
		assert!(!abort_subagent("sub-steer-test"));
	}
}
