leviath_runtime/

world.rs

//! The pipeline driver: a single [`PipelineWorld`] that hosts every agent as
//! ECS data and ticks the [`crate::pipeline`] systems over all of them - the
//! traditional-game-loop core of the shared world.
//!
//! The world owns the bevy [`World`], the tick [`Schedule`], the per-model
//! inference pools, and the async bridges (inference jobs + the tool worker).
//! Systems never block: they dispatch async work to the bridges and collect the
//! results on a later tick. Between ticks the driver **parks** on a wake
//! [`Notify`] until an async result lands or an external message arrives, so an
//! idle world costs ~0 CPU regardless of how many (paused/blocked) agents it
//! holds.
//!
//! ## Idle detection (no busy-spin)
//!
//! Each outer iteration drives the schedule to a **fixed point**: it ticks until
//! a tick produces no change in the per-phase marker counts (the "fingerprint").
//! At quiescence every remaining agent is either waiting on an in-flight async
//! job (which will `notify` on completion) or blocked on a resource that only an
//! async completion can free (a full pool) or on nothing at all (a missing
//! provider / no input) - so the driver parks on the wake instead of spinning.
//! A fresh async result or an external `send_message` fires the wake and the
//! fixed-point loop re-runs.

use std::sync::Arc;

use bevy_ecs::prelude::*;
use bevy_ecs::query::QueryFilter;
use leviath_providers::ProviderError;
use tokio::runtime::Handle;
use tokio::sync::Notify;
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
use tokio::task::JoinHandle;

use crate::components::{AgentMessage, AgentState, AgentStatus};
use crate::inference_pool::{InferencePoolConfig, InferencePools};
use crate::persistence_bridge::persistence_worker;
use crate::pipeline::{
    AwaitingCompaction, AwaitingInference, AwaitingTools, AwaitingTransitionChoice,
    AwaitingTransitionResponse, CompactionResults, InferenceResults, InferenceStage, MessageIntake,
    PersistenceStage, ProcessResponse, Providers, ReadyForTools, ReadyForTransition, ReadyToInfer,
    ResolveTransition, ToolResults, ToolService, ToolServiceRes, ToolStage, TransitionResults,
    abort_terminal_work, check_workspace_health, collect_compaction, collect_inference,
    collect_tools, collect_transition_choice, deliver_messages, detect_stuck_stage,
    dispatch_compaction, dispatch_edge_compact, dispatch_inference, dispatch_persistence,
    dispatch_tools, dispatch_transition_choice, enforce_max_iterations, fail_stalled_dispatch,
    fail_wedged_runs, gate_requires_children, handle_empty_response, poll_dynamic_tool_refresh,
    process_response, reflect_interaction_status, refresh_advertised_tools,
    require_context_regions, require_final_output, resolve_transition, run_after_inference_hooks,
    run_before_inference_hooks, run_stage_enter_hooks, run_stage_exit_hooks, run_terminal_hooks,
    run_tool_call_hooks, sync_tool_stages,
};
use crate::providers::ProviderRegistry;
use crate::tool_bridge::ToolLane;

/// What a tick can change, as one comparable value. Two consecutive equal
/// fingerprints mean a tick changed nothing (quiescence).
///
/// Marker counts alone are not enough, because a tick can move an agent out of a
/// marker and back into it. A stage that ends on `max_iterations` does exactly
/// that: `enforce_max_iterations` swaps `ReadyToInfer` for `ResolveTransition` in
/// the first chained group, and `resolve_transition` enters the next stage and
/// re-arms `ReadyToInfer` in the second - one tick, a whole stage transition, and
/// every count identical either side of it. The driver read that as quiescence
/// and parked on an agent that no dispatch system had yet seen in its new stage,
/// leaving the 30s re-drive to start the next stage (issue #197).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Fingerprint {
    /// How many agents hold each phase marker.
    markers: [usize; 12],
    /// Per-agent run progress that no marker reflects (see
    /// [`PipelineWorld::agent_digest`]).
    agents: u64,
}

/// How many attributed system panics one [`PipelineWorld::run_to_fixed_point`]
/// round will absorb before it stops driving. Each one fails a different agent,
/// so this only bites if the world is thoroughly broken - it exists so a
/// pathological agent can't spin the loop.
const MAX_TICK_FAILURES_PER_ROUND: usize = 8;

/// A schedule configured the way the pipeline needs it.
///
/// Every pipeline system is `.chain()`ed, so the multi-threaded executor can
/// never overlap two of them - it only adds a hop through the compute task
/// pool. Running single-threaded keeps systems on the thread that catches their
/// panics, which is what lets [`run_isolated`] read the offending agent out of
/// the (thread-local) [`crate::tick_scope`].
fn tick_schedule() -> Schedule {
    let mut schedule = Schedule::default();
    // bevy_ecs 0.19 replaced `set_executor_kind(ExecutorKind::…)` with
    // `set_executor(<executor instance>)`.
    schedule.set_executor(bevy_ecs::schedule::SingleThreadedExecutor::new());
    schedule
}

/// What one [`PipelineWorld::tick`] did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TickOutcome {
    /// Every system ran to completion.
    Clean,
    /// A system panicked and the agent responsible was failed; the rest of the
    /// world is unaffected and can keep being driven.
    AgentFailed,
    /// A system panicked with no agent in scope, so nothing could be failed.
    /// Re-ticking would just re-panic.
    Unattributed,
}

/// How many agents are in each status. See [`PipelineWorld::lane_snapshot`].
#[derive(Debug, Default, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AgentCounts {
    /// Doing work, or ready to.
    pub active: usize,
    /// Blocked on input, a child, or a prompt.
    pub waiting: usize,
    /// Parked by the user.
    pub paused: usize,
    /// Spawned but not yet started.
    pub idle: usize,
    /// Finished, still loaded pending reaping.
    pub terminal: usize,
}

impl std::fmt::Display for AgentCounts {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "active={} waiting={} paused={} idle={} terminal={}",
            self.active, self.waiting, self.paused, self.idle, self.terminal
        )
    }
}

/// What the world is holding and what it is waiting on, at one instant.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LaneSnapshot {
    /// Loaded agents by status.
    pub agents: AgentCounts,
    /// Inference-pool occupancy, one entry per model actually used.
    pub inference: Vec<crate::inference_pool::PoolOccupancy>,
    /// Tool batches holding lane capacity and running.
    pub tools_busy: usize,
    /// Tool batches waiting for lane capacity.
    pub tools_queued: usize,
    /// Tool batches parked on an unbounded wait, holding no capacity.
    pub tools_parked: usize,
    /// The tool lane's concurrency cap.
    pub tools_workers: usize,
    /// The lane full with batches still queued behind it.
    pub tools_saturated: bool,
}

impl LaneSnapshot {
    /// Whether some lane is at capacity with work queued behind it - the shape
    /// worth raising the log level for.
    #[must_use]
    pub fn is_under_pressure(&self) -> bool {
        self.tools_saturated
            || (self.agents.active > 0 && self.inference.iter().any(|p| p.is_full()))
    }

    /// The per-model inference occupancy, rendered for a log line.
    #[must_use]
    pub fn inference_summary(&self) -> String {
        if self.inference.is_empty() {
            return "none".to_string();
        }
        self.inference
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join(" ")
    }
}

/// Identifies one [`PipelineWorld`] within this process.
///
/// Only ever compared, never interpreted. A counter rather than a random value
/// because a mismatch is easier to read in a test failure as `1 != 2`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct WorldId(u64);

/// The identity of the world this resource lives in.
///
/// Stored *inside* the world so that code holding only a
/// [`bevy_ecs::world::World`] - a system, or a free function called from one -
/// can still tell whether an [`AgentId`] belongs to it. Without this the check
/// would only be possible on [`PipelineWorld`], which is not what a system has.
#[derive(Resource, Debug, Clone, Copy, PartialEq, Eq)]
pub struct OwnWorldId(pub WorldId);

/// An agent, together with the world that spawned it.
///
/// `Entity` is an index plus a generation minted *per world*, so two worlds
/// hand out the same id for their first agent. Nothing in `Entity` records
/// which one it came from, so passing one world's entity to another was not
/// refused - it named a real, different agent there, and the call acted on that
/// one instead. `b.pause(a_entity)` paused B's own agent while the caller
/// believed it had paused A's, silently.
///
/// The provenance has to travel *with* the id, which is what this is. It cannot
/// be built outside this module: the only sources are [`PipelineWorld::spawn_agent`]
/// and [`PipelineWorld::spawn_from_blueprint`], so an id always names an agent
/// in the world that minted it.
///
/// A tag component on the agent was tried first and does not work: looking the
/// tag up on the foreign id resolves to the *local* agent, whose tag naturally
/// matches, so the check passes and guards nothing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct AgentId {
    world: WorldId,
    entity: Entity,
}

impl AgentId {
    /// Scope a raw entity to the world it came out of.
    ///
    /// The reverse of [`Self::resolve_in`], for a system holding a query result
    /// that needs to call something taking an [`AgentId`]. Wrapping and then
    /// resolving inside the same world always round-trips; an id built this way
    /// in one world and resolved in another does not, which is the point.
    pub fn in_world(world: &World, entity: Entity) -> Self {
        Self {
            // A world assembled by hand in a test has no identity to borrow;
            // `resolve_in` accepts any id against such a world, so the pair
            // still round-trips.
            world: world
                .get_resource::<OwnWorldId>()
                .map_or(WorldId(0), |own| own.0),
            entity,
        }
    }

    /// The entity, if this id belongs to `world`.
    ///
    /// The check any code holding a raw [`World`] should make before touching an
    /// agent it was handed. `None` means the id came from a different world, in
    /// which case its raw entity would name some *other* agent here - which is
    /// the whole failure this type exists to prevent.
    ///
    /// A world with no [`OwnWorldId`] resource (a bare test world assembled by
    /// hand) accepts any id: it never minted one, so there is nothing to
    /// disagree with.
    pub fn resolve_in(self, world: &World) -> Option<Entity> {
        match world.get_resource::<OwnWorldId>() {
            Some(own) if own.0 != self.world => None,
            _ => Some(self.entity),
        }
    }

    /// The raw ECS entity.
    ///
    /// For code already inside the owning world - systems, queries, direct
    /// `World` access - where same-world is true by construction. Crossing a
    /// world boundary with the result is the bug this type exists to prevent.
    pub fn entity(self) -> Entity {
        self.entity
    }

    /// Which world minted this id.
    pub fn world(self) -> WorldId {
        self.world
    }
}

/// The shared ECS world that hosts and drives every agent.
pub struct PipelineWorld {
    /// This world's identity, carried by every [`AgentId`] it mints.
    id: WorldId,
    world: World,
    schedule: Schedule,
    wake: Arc<Notify>,
    shutdown: Arc<Notify>,
    msg_tx: UnboundedSender<AgentMessage>,
    /// The tool lane, kept so the world can widen it under relief.
    tool_lane: Arc<ToolLane>,
    /// The task serving the tool lane; kept so it lives as long as the world. It
    /// exits on its own once the world (and thus the [`ToolStage`] sender) is
    /// dropped and the batches it started have finished.
    _tool_task: JoinHandle<()>,
    /// The persistence worker task. Retained (rather than detached) so
    /// [`Self::flush_and_stop`] can close its channel and `await` it, guaranteeing
    /// every queued snapshot reaches disk before shutdown. `None` once flushed.
    persist_task: Option<JoinHandle<()>>,
}

impl PipelineWorld {
    /// Build a world: wire the pool/bridge resources, register the providers and
    /// tool service, spawn the tool worker onto `runtime`, and assemble the tick
    /// schedule. Agents are added later via [`Self::spawn_agent`].
    ///
    /// `runs_dir` is where agent snapshots persist (`<runs_dir>/<run_id>/`, the
    /// daemon's on-disk layout). `None` keeps the world entirely in memory:
    /// snapshots are still produced and drained (so log events and watermarks
    /// behave identically) but nothing is ever written to disk.
    pub fn new(
        providers: ProviderRegistry,
        tool_service: Arc<dyn ToolService>,
        pool_config: InferencePoolConfig,
        tool_concurrency: usize,
        runs_dir: Option<std::path::PathBuf>,
        runtime: Handle,
    ) -> Self {
        // `Query::par_iter` fans out over the compute task pool; initialize it
        // once (idempotent) so per-agent request assembly in `dispatch_inference`
        // runs in parallel. (The schedule executor itself is single-threaded -
        // see `tick_schedule`.)
        bevy_tasks::ComputeTaskPool::get_or_init(bevy_tasks::TaskPool::default);
        static NEXT_WORLD_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
        let id = WorldId(NEXT_WORLD_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed));

        let wake = Arc::new(Notify::new());
        let shutdown = Arc::new(Notify::new());

        let (inf_tx, inf_rx) = unbounded_channel();
        let (trans_tx, trans_rx) = unbounded_channel();
        let (compact_tx, compact_rx) = unbounded_channel();
        let (tool_job_tx, tool_job_rx) = unbounded_channel();
        let (tool_res_tx, tool_res_rx) = unbounded_channel();
        let (persist_tx, persist_rx) = unbounded_channel();
        let (msg_tx, msg_rx) = unbounded_channel();
        let (ip_tx, ip_rx) = unbounded_channel();
        let (gp_tx, gp_rx) = unbounded_channel();
        let (cs_tx, cs_rx) = unbounded_channel();
        let (title_tx, title_rx) = unbounded_channel();

        let tool_stats = Arc::new(crate::tool_bridge::ToolLaneStats::new(tool_concurrency));
        let tool_lane = ToolLane::new(
            runtime.clone(),
            tool_res_tx,
            wake.clone(),
            tool_concurrency,
            tool_stats.clone(),
        );
        let tool_task = tool_lane.serve(tool_job_rx);
        // Retained so `flush_and_stop` can drain it on shutdown. Left to its own
        // devices otherwise: it exits when the world (and thus its PersistenceStage
        // sender) is dropped.
        let persist_task = runtime.spawn(persistence_worker(runs_dir, persist_rx));
        let ip_runtime = runtime.clone();
        let gp_runtime = runtime.clone();

        let mut world = World::new();
        world.insert_resource(OwnWorldId(id));
        world.insert_resource(Providers(providers));
        world.insert_resource(InferenceStage {
            // The wake goes into the pools, not just the bridges: freeing a slot
            // has to re-drive dispatch, or the agents parked on a full pool never
            // learn that capacity came back (issue #189).
            pools: Arc::new(InferencePools::new(pool_config).with_wake(wake.clone())),
            outcomes: inf_tx,
            transition_outcomes: trans_tx,
            compaction_outcomes: compact_tx,
            content_summary_outcomes: cs_tx,
            wake: wake.clone(),
            runtime,
            exact_token_counting: false,
        });
        world.insert_resource(crate::context_transform::ContentSummaryResults(cs_rx));
        world.insert_resource(crate::title::TitleSink(title_tx));
        world.insert_resource(crate::title::TitleResults(title_rx));
        world.insert_resource(crate::interaction_points::InteractionPointStage {
            outcomes: ip_tx,
            wake: wake.clone(),
            runtime: ip_runtime,
        });
        world.insert_resource(crate::interaction_points::InteractionPointResults(ip_rx));
        world.insert_resource(crate::gate_prompt::GatePromptStage {
            outcomes: gp_tx,
            wake: wake.clone(),
            runtime: gp_runtime,
        });
        world.insert_resource(crate::gate_prompt::GatePromptResults(gp_rx));
        world.insert_resource(InferenceResults(inf_rx));
        world.insert_resource(TransitionResults(trans_rx));
        world.insert_resource(CompactionResults(compact_rx));
        world.insert_resource(ToolServiceRes(tool_service));
        world.insert_resource(ToolStage::new(tool_job_tx, tool_stats));
        world.insert_resource(ToolResults(tool_res_rx));
        world.insert_resource(PersistenceStage(persist_tx));
        world.insert_resource(MessageIntake(msg_rx));
        // Telemetry defaults to the no-op sink; a host that wants export
        // replaces the resource after construction (as `build_host` does).
        world.insert_resource(crate::telemetry::Telemetry(std::sync::Arc::new(
            leviath_core::telemetry::NoopSink,
        )));

        // The tick chain is split into two `.chain()`ed groups (bevy caps a
        // system tuple at 20); the second group runs strictly after the first.
        let mut schedule = tick_schedule();
        schedule.add_systems(
            (
                // First: stop whatever a now-terminal agent still has running in
                // the async lanes. Ahead of everything else so a cancel frees its
                // inference permit and tool-lane capacity on the very next tick,
                // rather than whenever the provider or tool happens to answer.
                abort_terminal_work,
                deliver_messages,
                collect_compaction,
                // Apply any completed Summarize context-transform summaries into
                // the child's regions, then dispatch newly-queued ones.
                crate::context_transform::collect_content_summary,
                crate::context_transform::dispatch_content_summary,
                // Route edge-transform compaction through the compaction lane
                // before the threshold-based pass.
                dispatch_edge_compact,
                dispatch_compaction,
                // Cap a stage at its max_iterations before running more inference.
                enforce_max_iterations,
                // …then the softer guard: bail out of a stage that is burning
                // turns/edits without progress, when the blueprint declares a
                // `stuck` escape edge. Runs after the hard cap so that always wins.
                detect_stuck_stage,
                // Stop a run whose working directory vanished, rather than let
                // every tool fail with ENOENT for the rest of the run.
                check_workspace_health,
                // Tag dynamic_tools agents that have pending tool changes, then
                // apply the re-advertisement before the next request is assembled
                // so a newly-discovered tool is visible.
                poll_dynamic_tool_refresh,
                refresh_advertised_tools,
                // Move ready agents off any provider whose circuit is open, so
                // dispatch only ever considers one still in service. Serial,
                // because it needs `&mut StageInference` and dispatch fans out.
                // Nested rather than inline: the outer tuple is at bevy's
                // 20-system limit for `.chain()`.
                // Nested tuples here and below for the same reason the circuit
                // pair already was: the outer tuple is at bevy's 20-system
                // `.chain()` limit, and grouping preserves the ordering.
                //
                // `before_inference` runs with the window assembled and before
                // the request is built from it.
                (
                    run_before_inference_hooks,
                    crate::pipeline::rotate_open_circuits,
                    dispatch_inference,
                )
                    .chain(),
                collect_inference,
                // Intercept a fan-out stage's split response before normal routing.
                crate::fanout::fan_out_split,
                // `after_inference` sees the response before anything is
                // written to context or dispatched from it.
                (run_after_inference_hooks, process_response).chain(),
                // Apply resolved taint gate prompts (re-arming ReadyForTools)
                // before the tool dispatch re-runs the held batch.
                crate::gate_prompt::collect_gate_prompt,
                // `on_tool_call` before the policy and taint layers see the
                // calls, so a hook can narrow what runs and never widen it.
                (run_tool_call_hooks, dispatch_tools).chain(),
                collect_tools,
                // Apply any resolved stage-boundary interaction-point answers
                // before the stage decides its transition.
                crate::interaction_points::collect_interaction_point,
            )
                .chain(),
        );
        schedule.add_systems(
            (
                handle_empty_response,
                // Hold a `requires_children` stage until its sub-agents finish.
                gate_requires_children,
                // Re-run a stage that left a `required` context region empty
                // before it may transition or ask for approval.
                require_context_regions,
                // Same, for a stage that owes a final output and has not
                // submitted one. Beside its sibling and before any transition
                // resolves, so an unfinished stage is sent back rather than
                // silently ending the run with nothing to hand back.
                require_final_output,
                // Intercept a would-be transition for an interactive-points stage
                // (e.g. plan_approval) and drive the interaction-point lane.
                crate::interaction_points::gate_interaction_points,
                crate::interaction_points::dispatch_interaction_point,
                // `on_stage_exit` while the finishing stage is still current
                // and before the edge that leaves it is picked.
                (run_stage_exit_hooks, resolve_transition).chain(),
                dispatch_transition_choice,
                collect_transition_choice,
                // Drive fan-out workers and merge once they finish.
                crate::fanout::fan_out_collect,
                // Narrate lifecycle/activity into the telemetry sink. Must run
                // before `sync_tool_stages` (which consumes the transient
                // `StageJustEntered` marker) and before `dispatch_persistence`
                // (which drains the log buffer this system only reads).
                crate::telemetry::observe_lifecycle,
                // A stage's `on_stage_enter` script, before `sync_tool_stages`
                // consumes the `StageJustEntered` marker it fires on - so the
                // hook sees the stage's layout and prompt already in place, and
                // whatever it writes is in the stage's first request.
                run_stage_enter_hooks,
                sync_tool_stages,
                // Store any finished run title, then start newly-marked ones.
                // Collect precedes persistence so a landed title is written on
                // this same tick.
                // `on_completion` / `on_error`, once, as a run finishes.
                // Grouped for the 20-system `.chain()` limit, as above.
                (run_terminal_hooks, crate::title::collect_title).chain(),
                crate::title::dispatch_title,
                // Fail a run whose dispatch has been declining for something
                // that will never arrive. Last of the guards, and after *both*
                // dispatch systems, so it reads stall records both lanes have
                // refreshed on this same tick - and before persistence, so the
                // failure reaches disk immediately.
                fail_stalled_dispatch,
                // Mirror open interaction-hub requests into agent status
                // (Active ↔ Waiting) so the dashboard surfaces blocked prompts;
                // must run before persistence so the status change is written.
                reflect_interaction_status,
                // Fail a run nothing can drive at all. After every dispatch and
                // collect system, so a marker set anywhere on this tick counts;
                // after the interaction reflection, so an agent that just parked
                // on a prompt is already wearing its marker and is exempt; and
                // before persistence, so the failure reaches meta.json on the
                // same tick rather than waiting for the next one.
                fail_wedged_runs,
                dispatch_persistence,
                // After persistence: a merged fan-out worker is only slimmed
                // once its terminal snapshot has been dispatched to the lane,
                // and running behind `dispatch_persistence` means the check
                // reads this tick's watermark, not last tick's.
                crate::fanout::slim_merged_workers,
            )
                .chain()
                .after(crate::interaction_points::collect_interaction_point),
        );

        Self {
            id,
            world,
            schedule,
            wake,
            shutdown,
            msg_tx,
            tool_lane,
            _tool_task: tool_task,
            persist_task: Some(persist_task),
        }
    }

    /// Mutable access to the underlying ECS world, for spawning agents (the CLI /
    /// daemon builds each agent's component bundle) and inspection.
    ///
    /// This is the unstable layer: it exposes raw `bevy_ecs` (re-exported as
    /// [`crate::ecs`] so versions stay aligned) and carries no compatibility
    /// promise across releases. Prefer [`crate::AgentWorld`] or
    /// [`crate::host::WorldHost`] unless you are building your own assembly.
    pub fn world_mut(&mut self) -> &mut World {
        &mut self.world
    }

    /// Read-only access to the underlying ECS world.
    pub fn world(&self) -> &World {
        &self.world
    }

    /// Enable (or disable) the opt-in exact pre-inference budget guard for this
    /// world - see `inference_bridge::InferenceJob::exact_token_counting`.
    /// Call once at startup when the run config requests it, before serving.
    pub fn set_exact_token_counting(&mut self, enabled: bool) {
        // `InferenceStage` is inserted by every `PipelineWorld::new` path, so it
        // is a hard invariant here - `resource_mut` (which panics if absent) is
        // correct and keeps this branch-free.
        self.world
            .resource_mut::<crate::pipeline::InferenceStage>()
            .exact_token_counting = enabled;
    }

    /// Install the shared interaction hub as a world resource and attach this
    /// world's wake handle to it, so opening/answering a prompt wakes the driver
    /// and [`reflect_interaction_status`]
    /// mirrors the change into agent status. Call once at startup, before
    /// serving. Without this, that system is a no-op (test worlds).
    pub fn insert_interaction_hub(&mut self, hub: crate::interaction_hub::InteractionHub) {
        hub.attach_wake(self.wake.clone());
        self.world.insert_resource(hub);
    }

    /// Spawn an agent from its pre-built component bundle and wake the driver so
    /// the next fixed-point picks it up. Returns the new entity.
    pub fn spawn_agent(&mut self, bundle: impl Bundle) -> AgentId {
        let entity = self.world.spawn(bundle).id();
        self.wake.notify_one();
        AgentId {
            world: self.id,
            entity,
        }
    }

    /// Spawn an agent from a blueprint + task + per-stage resolution (see
    /// [`crate::pipeline::spawn_agent`]) and wake the driver. Returns the new
    /// entity, or an error if the first stage's system prompt doesn't fit.
    pub fn spawn_from_blueprint(
        &mut self,
        agent_id: String,
        blueprint: leviath_core::Blueprint,
        task: &str,
        stages: Vec<crate::pipeline::ResolvedStage>,
        global_hints: leviath_core::config::PromptHints,
    ) -> Result<AgentId, String> {
        let entity = crate::pipeline::spawn_agent(
            &mut self.world,
            agent_id,
            blueprint,
            task,
            stages,
            global_hints,
        )?;
        self.wake.notify_one();
        Ok(AgentId {
            world: self.id,
            entity,
        })
    }

    /// Deliver a message to a running agent (routed to its inbox on the next
    /// tick) and wake the driver.
    pub fn send_message(&self, msg: AgentMessage) -> Result<(), ProviderError> {
        self.msg_tx
            .send(msg)
            .map_err(|e| ProviderError::Other(format!("world message channel closed: {e}")))?;
        self.wake.notify_one();
        Ok(())
    }

    /// A clone of the wake handle, so external producers (e.g. a control socket)
    /// can nudge the driver after mutating the world directly.
    pub fn wake_handle(&self) -> Arc<Notify> {
        self.wake.clone()
    }

    /// Request the [`Self::run`] loop to stop after its current fixed point.
    pub fn shutdown(&self) {
        self.shutdown.notify_one();
    }

    /// A clone of the shutdown handle, so a supervisor can stop a [`Self::run`]
    /// loop that has taken ownership of the world on another task.
    pub fn shutdown_handle(&self) -> Arc<Notify> {
        self.shutdown.clone()
    }

    /// Cleanly stop the world, guaranteeing every queued snapshot reaches disk.
    ///
    /// The persistence lane is async and fire-and-forget, so a plain shutdown (the
    /// [`Self::run`]/`serve` loop returning, then the world dropping) can lose
    /// snapshots still queued in the channel. This method closes that gap: it
    /// signals shutdown, drives one last fixed point so any state that settled
    /// after the loop parked is dispatched to the lane, then **closes the lane and
    /// awaits the worker** so all queued writes (`meta.json` / `context.json` /
    /// `run.lvr`) land before it returns.
    ///
    /// Call it after the serve loop has returned (the tokio runtime must still be
    /// alive for the worker to be scheduled). Idempotent: a second call is a no-op
    /// because the persistence resource is already removed and the task taken.
    pub async fn flush_and_stop(&mut self) {
        // Idempotent - the serve loop has usually already returned on this signal.
        self.shutdown.notify_one();
        // Dispatch anything that settled between the last park and now (e.g. an
        // inference result that woke the loop the same instant shutdown fired).
        self.run_to_fixed_point();
        // Drop the *only* `PersistJob` sender so the worker's `recv()` loop drains
        // its queue and then ends.
        self.world.remove_resource::<PersistenceStage>();
        // Wait for every queued write to hit disk.
        if let Some(task) = self.persist_task.take() {
            let _ = task.await;
        }
        // Push any buffered telemetry export out before the process goes away;
        // the final fixed point above already emitted the last events. The
        // resource always exists - `new()` installs the no-op default.
        self.world
            .resource::<crate::telemetry::Telemetry>()
            .0
            .force_flush();
    }

    /// A point-in-time read of what the world is holding and what it is waiting
    /// on: agents by status, per-model inference-pool occupancy, and tool-lane
    /// occupancy.
    ///
    /// Providers currently taken out of service by their circuit breaker.
    ///
    /// Empty when the breaker is not installed, so an embedded world that never
    /// inserted the resource simply reports nothing wrong (issue #201).
    pub fn open_circuits(&self) -> Vec<crate::pipeline::ProviderCircuitState> {
        let Some(circuits) = self
            .world
            .get_resource::<crate::pipeline::ProviderCircuits>()
        else {
            return Vec::new();
        };
        let policy = self
            .world
            .get_resource::<crate::pipeline::CircuitPolicy>()
            .copied()
            .unwrap_or_default();
        circuits.open_circuits(chrono::Utc::now().timestamp(), &policy)
    }

    /// This is the answer to "the daemon has been quiet for hours - is anything
    /// actually running?", which issue #189 had no way to ask.
    pub fn lane_snapshot(&self) -> LaneSnapshot {
        let mut agents = AgentCounts::default();
        for state in self
            .world
            .iter_entities()
            .filter_map(|e| e.get::<AgentState>())
        {
            match state.status {
                AgentStatus::Active => agents.active += 1,
                AgentStatus::Waiting => agents.waiting += 1,
                AgentStatus::Paused => agents.paused += 1,
                AgentStatus::Idle => agents.idle += 1,
                // Terminal agents linger until the reaper unloads them; counting
                // them apart keeps "nothing is running" honest.
                AgentStatus::Complete | AgentStatus::Error { .. } | AgentStatus::Cancelled => {
                    agents.terminal += 1
                }
            }
        }
        let tools = self.world.resource::<ToolStage>().stats.clone();
        LaneSnapshot {
            agents,
            inference: self.world.resource::<InferenceStage>().pools.occupancy(),
            tools_busy: tools.busy(),
            tools_queued: tools.queued(),
            tools_parked: tools.parked(),
            tools_workers: tools.workers(),
            tools_saturated: tools.is_saturated(),
        }
    }

    /// Widen the tool lane by `extra` batches.
    ///
    /// The relief valve: when the lane has stopped draining, handing out more
    /// capacity lets the queued batches through without cancelling anything.
    /// Returns how many were added.
    pub fn relieve_tool_lane(&self, extra: usize) -> usize {
        self.tool_lane.relieve(extra)
    }

    /// Reclaim up to `upto` idle permits from the tool lane (the relief valve's
    /// give-back half). Returns how many were reclaimed; never touches a permit
    /// a running batch holds.
    pub fn narrow_tool_lane(&self, upto: usize) -> usize {
        self.tool_lane.narrow(upto)
    }

    /// Wrap an entity that came out of this world, for callers that hold one.
    ///
    /// Exposed for the host and for recovery: both query this world directly,
    /// so the entities they get back are ours by construction. Not a general
    /// escape - there is no way to build an [`AgentId`] for a world you do not
    /// already have in hand.
    pub fn own_agent(&self, entity: Entity) -> AgentId {
        self.own(entity)
    }

    /// Wrap an entity this world already owns.
    ///
    /// For entities that came out of this world's own queries, where same-world
    /// is true by construction. Private: outside code must get its ids from a
    /// spawn, which is what makes [`AgentId`] mean anything.
    fn own(&self, entity: Entity) -> AgentId {
        AgentId {
            world: self.id,
            entity,
        }
    }

    /// The status of an agent, if it still exists.
    ///
    /// An id another world minted reports `None` rather than this world's agent
    /// of the same raw entity - see [`AgentId`]. `pause`, `resume` and `cancel`
    /// all read status through here, so guarding it guards them.
    pub fn agent_status(&self, agent: AgentId) -> Option<AgentStatus> {
        if agent.world != self.id {
            return None;
        }
        self.world
            .get::<AgentState>(agent.entity)
            .map(|s| s.status.clone())
    }

    /// Set an agent's status and wake the driver. Returns `false` if the agent no
    /// longer exists. The async-starting dispatchers only act on `Active` agents,
    /// so this is how the world pauses/resumes/cancels an agent - a non-`Active`
    /// agent is simply data the systems skip until it is `Active` again.
    pub fn set_status(&mut self, agent: AgentId, status: AgentStatus) -> bool {
        // Every status mutation funnels through here, so this is the one place a
        // foreign id has to be refused.
        if agent.world != self.id {
            return false;
        }
        let Some(mut state) = self.world.get_mut::<AgentState>(agent.entity) else {
            return false;
        };
        state.status = status;
        self.wake.notify_one();
        true
    }

    /// Pause an agent (it finishes any in-flight step, then stops before starting
    /// new work). Only `Active` and `Idle` agents can be paused: a `Waiting`
    /// agent's status is the marker the fan-out merge poll and interaction
    /// resolution depend on, so overwriting it would wedge the run, and pausing
    /// a terminal agent is meaningless. Returns `false` if the agent no longer
    /// exists or is not in a pausable state.
    pub fn pause(&mut self, agent: AgentId) -> bool {
        match self.agent_status(agent) {
            Some(AgentStatus::Active | AgentStatus::Idle) => {
                self.set_status(agent, AgentStatus::Paused)
            }
            _ => false,
        }
    }

    /// Resume a paused agent. `Idle` is also accepted (resume-as-nudge for an
    /// agent that has not ticked yet); anything else returns `false`.
    pub fn resume(&mut self, agent: AgentId) -> bool {
        // Resolved once, up front. Doing it again inside the arm below would
        // be a second check that cannot fail - the status lookup already
        // proved the entity is here - and an arm nothing can reach is an arm
        // nothing can test.
        let Some(entity) = agent.resolve_in(&self.world) else {
            return false;
        };
        match self.agent_status(agent) {
            Some(AgentStatus::Paused | AgentStatus::Idle) => {
                // An explicit resume says conditions have changed - most often
                // a top-up after a run paused on exhausted credits (issue
                // #413). A tripped breaker would otherwise hold the retry
                // until its cooldown lapses, making the resume look ignored.
                if let Some(mut circuits) = self
                    .world
                    .get_resource_mut::<crate::pipeline::ProviderCircuits>()
                {
                    circuits.reset();
                }
                // The run no longer claims to need setup. If the machine was
                // not actually fixed the watchdog re-parks it a minute later
                // with a fresh message, which is better than carrying a stale
                // one that describes a problem somebody may have just solved.
                self.world
                    .entity_mut(entity)
                    .remove::<crate::pipeline::PausedForSetup>();
                self.set_status(agent, AgentStatus::Active)
            }
            _ => false,
        }
    }

    /// Cancel an agent (it stops starting new work; in-flight results still land).
    pub fn cancel(&mut self, agent: AgentId) -> bool {
        self.set_status(agent, AgentStatus::Cancelled)
    }

    /// Run one schedule tick over every agent, catching a panic from any system
    /// so one bad agent can't crash the daemon and take every other hosted agent
    /// with it.
    ///
    /// When the panic can be traced to a specific agent (the usual case - see
    /// `tick_scope`), that agent is failed with the panic message so it
    /// stops being driven, its run is persisted as errored, and the host reaps
    /// it. Without that, the world would re-tick the same unchanged state on
    /// every wake and panic again indefinitely.
    pub fn tick(&mut self) -> TickOutcome {
        let Err(panicked) = run_isolated(&mut self.schedule, &mut self.world) else {
            // A clean unwind doesn't mean a clean tick: work that ran on the
            // compute pool catches its own panics, since they can't unwind back
            // here, and leaves a marker instead.
            return self.fail_agents_panicked_in_parallel();
        };
        let message = panic_status_message(&panicked.message);
        match panicked.entity {
            Some(entity) if self.set_status(self.own(entity), AgentStatus::Error { message }) => {
                tracing::error!(
                    ?entity,
                    panic = %panicked.message,
                    "a pipeline system panicked; failing that agent - the daemon and every \
                     other run keep going"
                );
                TickOutcome::AgentFailed
            }
            _ => {
                tracing::error!(
                    panic = %panicked.message,
                    "a pipeline system panicked outside any agent's scope; the daemon survived \
                     (an agent may be wedged - cancel it via `lev cancel <run-id>`)"
                );
                TickOutcome::Unattributed
            }
        }
    }

    /// Fail every agent that a compute-pool body marked
    /// [`PanickedInParallel`](crate::tick_scope::PanickedInParallel), and report
    /// whether there were any.
    ///
    /// These panics were caught on a task-pool thread rather than unwinding into
    /// `tick`, so the marker component is how they reach the driver - but from
    /// here on they are handled exactly like an attributed unwind: the agent is
    /// failed, stops being driven, and its run persists as errored.
    fn fail_agents_panicked_in_parallel(&mut self) -> TickOutcome {
        let mut query = self
            .world
            .query::<(Entity, &crate::tick_scope::PanickedInParallel)>();
        let failed: Vec<(Entity, String)> = query
            .iter(&self.world)
            .map(|(entity, p)| (entity, p.message.clone()))
            .collect();
        if failed.is_empty() {
            return TickOutcome::Clean;
        }
        for (entity, message) in failed {
            self.world
                .entity_mut(entity)
                .remove::<crate::tick_scope::PanickedInParallel>();
            let status = AgentStatus::Error {
                message: panic_status_message(&message),
            };
            // The entity came straight out of the query above, so it exists.
            let _ = self.set_status(self.own(entity), status);
        }
        TickOutcome::AgentFailed
    }

    /// Append a system to the schedule (test-only, for panic-isolation tests).
    #[cfg(test)]
    pub(crate) fn add_test_system<M>(
        &mut self,
        // `IntoSystemConfigs` became `IntoScheduleConfigs<ScheduleSystem, _>` in
        // bevy_ecs 0.19 (it now also describes observer and other schedulables,
        // so the schedulable kind is an explicit parameter).
        system: impl bevy_ecs::schedule::IntoScheduleConfigs<bevy_ecs::system::ScheduleSystem, M>,
    ) {
        self.schedule.add_systems(system);
    }

    fn count<F: QueryFilter>(&mut self) -> usize {
        let mut q = self.world.query_filtered::<(), F>();
        q.iter(&self.world).count()
    }

    /// Digest the run progress a phase marker cannot show: each agent's status,
    /// which stage it is in, and its per-stage counters.
    ///
    /// Only values that step on a real event go in. Anything that moves on its
    /// own (a clock, a stall timestamp) would keep the fixed-point loop from ever
    /// converging, which is a spinning daemon rather than a parked one.
    ///
    /// The per-agent digests are XOR-folded, so archetype iteration order doesn't
    /// matter; each one includes the entity id so two agents swapping states
    /// can't cancel out.
    fn agent_digest(&mut self) -> u64 {
        use std::hash::{Hash, Hasher};
        let mut query = self.world.query::<(
            Entity,
            &AgentState,
            Option<&crate::pipeline::StageCursor>,
            Option<&crate::pipeline::StageProgress>,
        )>();
        query
            .iter(&self.world)
            .map(|(entity, state, cursor, progress)| {
                let mut hasher = std::collections::hash_map::DefaultHasher::new();
                entity.to_bits().hash(&mut hasher);
                state.status.hash(&mut hasher);
                state.current_stage.hash(&mut hasher);
                state.iteration.hash(&mut hasher);
                cursor.map(|c| c.index).hash(&mut hasher);
                progress
                    .map(|p| {
                        (
                            p.iterations,
                            p.total_tool_calls,
                            p.modifying_tool_calls,
                            p.gate_reentries,
                            p.stuck_fired,
                        )
                    })
                    .hash(&mut hasher);
                hasher.finish()
            })
            .fold(0, |acc, digest| acc ^ digest)
    }

    /// Snapshot the per-phase marker counts and the per-agent progress digest.
    fn fingerprint(&mut self) -> Fingerprint {
        let markers = [
            self.count::<With<ReadyToInfer>>(),
            self.count::<With<AwaitingInference>>(),
            self.count::<With<ProcessResponse>>(),
            self.count::<With<ReadyForTools>>(),
            self.count::<With<ReadyForTransition>>(),
            self.count::<With<ResolveTransition>>(),
            self.count::<With<AwaitingTools>>(),
            self.count::<With<AwaitingTransitionChoice>>(),
            self.count::<With<AwaitingTransitionResponse>>(),
            self.count::<With<AwaitingCompaction>>(),
            self.count::<With<crate::title::PendingTitle>>(),
            self.count::<With<crate::title::AwaitingTitle>>(),
        ];
        Fingerprint {
            markers,
            agents: self.agent_digest(),
        }
    }

    /// Any agent waiting on an in-flight async job (inference, tools, a
    /// transition choice, or compaction) whose completion will wake the driver.
    fn has_async_inflight(&mut self) -> bool {
        self.count::<With<AwaitingInference>>() > 0
            || self.count::<With<AwaitingTools>>() > 0
            || self.count::<With<AwaitingTransitionResponse>>() > 0
            || self.count::<With<AwaitingCompaction>>() > 0
            || self.count::<With<crate::title::AwaitingTitle>>() > 0
    }

    /// Drive the schedule until a tick changes nothing (quiescence). Public so a
    /// host loop can interleave control operations between quiescent points.
    pub fn run_to_fixed_point(&mut self) {
        let mut prev = self.fingerprint();
        let mut failures = 0;
        loop {
            let outcome = self.tick();
            match outcome {
                TickOutcome::Clean => {}
                // The offending agent has been failed, so it won't be driven
                // again. Keep ticking: the rest of the world still has work to
                // do, and only a later tick reaches `dispatch_persistence` (the
                // last system in the chain) to record the failure on disk. The
                // budget stops a pathological agent that somehow panics again
                // from spinning this loop.
                TickOutcome::AgentFailed if failures < MAX_TICK_FAILURES_PER_ROUND => {
                    failures += 1;
                }
                // Nothing to fail, so re-ticking would just re-panic: stop
                // driving this round. The daemon stays alive, other agents keep
                // running, and a wedged agent can be cancelled via the control
                // socket (dispatch systems skip non-Active agents once
                // cancelled).
                TickOutcome::AgentFailed | TickOutcome::Unattributed => break,
            }
            let now = self.fingerprint();
            // Quiescence, but only trust it after a clean tick: a panicking tick
            // abandons the rest of the chain (and its buffered commands), so the
            // markers can look unchanged while the world very much has changed.
            // Force at least one more tick so the failed agent gets persisted.
            if now == prev && outcome == TickOutcome::Clean {
                break;
            }
            prev = now;
        }
    }

    /// Drive every agent as far as it can go **right now**, then, while async
    /// work is in flight, wait for each completion and drive again - returning
    /// once the world is fully quiescent with nothing in flight. Bounded by
    /// `max_waits` wake-waits as a safety valve so a lost/never-arriving wake
    /// can't hang a caller (e.g. a test) forever.
    pub async fn run_until_idle(&mut self, max_waits: usize) {
        self.run_to_fixed_point();
        let mut waits = 0;
        while self.has_async_inflight() && waits < max_waits {
            self.wake.notified().await;
            waits += 1;
            self.run_to_fixed_point();
        }
    }

    /// Run forever: drive to quiescence, then park until an async completion or
    /// an external `send_message`/`spawn_agent` wakes the driver. Returns when
    /// [`Self::shutdown`] is signalled.
    pub async fn run(&mut self) {
        loop {
            self.run_to_fixed_point();
            tokio::select! {
                _ = self.wake.notified() => {}
                _ = self.shutdown.notified() => return,
            }
        }
    }
}

/// How a caught panic is recorded on the agent it is blamed on. Shared by the
/// unwind path and the compute-pool path so a run's `error` reads the same
/// either way.
fn panic_status_message(panic: &str) -> String {
    format!("internal error: a pipeline system panicked: {panic}")
}

/// A panic caught while ticking the schedule, and the agent it belongs to.
struct TickPanic {
    /// The agent being processed when the panic fired, if the pipeline had
    /// recorded one (see [`crate::tick_scope`]).
    entity: Option<Entity>,
    /// The panic payload rendered as text.
    message: String,
}

/// Run a schedule over a world, catching a panic from any system so it can't
/// unwind the daemon's drive loop and take down every hosted agent.
///
/// The world may be partially updated after a panic: the panicking system's
/// buffered `Commands` are lost, but resources and components already written
/// are intact, so the caller can still fail the offending agent.
fn run_isolated(schedule: &mut Schedule, world: &mut World) -> Result<(), TickPanic> {
    // Clear first: the slot is thread-local and survives across ticks, so a
    // stale entity from an earlier tick must not be blamed for this one.
    crate::tick_scope::clear();
    match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| schedule.run(world))) {
        Ok(()) => Ok(()),
        Err(payload) => {
            reset_executor(schedule);
            Err(TickPanic {
                entity: crate::tick_scope::current(),
                message: leviath_core::panic_message(payload.as_ref()),
            })
        }
    }
}

/// Give `schedule` a fresh executor after a caught panic.
///
/// bevy's executors mark a system "completed" *before* running it and only
/// clear that set when `run` returns normally. A panic therefore leaves every
/// system up to and including the offending one marked done, so the **next**
/// tick silently skips them and only runs the tail of the chain - a partial
/// tick that would, among other things, keep `dispatch_persistence` from ever
/// seeing an agent we just failed. Swapping the executor kind and back is the
/// public API for forcing a rebuild.
///
/// One call suffices on bevy_ecs 0.19: `set_executor` takes an executor
/// *instance* and unconditionally replaces `schedule.executor` with it (clearing
/// `executor_initialized` too), so the fresh `SingleThreadedExecutor` arrives
/// with an empty `completed_systems`.
///
/// On 0.15 this had to set two different *kinds* and swap back, because
/// `set_executor_kind` was a no-op when the kind was unchanged - and
/// `SimpleExecutor`, the other kind it used, no longer exists.
fn reset_executor(schedule: &mut Schedule) {
    schedule.set_executor(bevy_ecs::schedule::SingleThreadedExecutor::new());
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Serializes every test in this binary that swaps the **process-global**
    /// panic hook - see the definition for why they can't run concurrently.
    use crate::test_support::{PANIC_HOOK_LOCK, hints};

    /// Run `f` with the process panic hook silenced (the panic is expected), and
    /// serialized against the other hook-swapping tests.
    fn with_silent_panics<T>(f: impl FnOnce() -> T) -> T {
        let _hook_guard = PANIC_HOOK_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let prev_hook = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {}));
        let out = f();
        std::panic::set_hook(prev_hook);
        out
    }

    #[test]
    fn run_isolated_catches_a_system_panic_and_reports_the_agent() {
        fn ok_system() {}
        fn boom_system() {
            panic!("simulated system panic");
        }
        // A system that panics *while working on a specific agent* - the shape
        // every real pipeline system has.
        fn boom_on_agent_system() {
            crate::tick_scope::enter(
                Entity::from_raw_u32(41)
                    .expect("a small literal index is always a valid entity id"),
            );
            panic!("agent-scoped panic");
        }
        let mut world = World::new();

        // A clean schedule ticks normally.
        let mut ok = tick_schedule();
        ok.add_systems(ok_system);
        assert!(run_isolated(&mut ok, &mut world).is_ok());

        // A panicking system is caught (the daemon would survive) and, with no
        // agent in scope, reports no entity to blame.
        let mut bad = tick_schedule();
        bad.add_systems(boom_system);
        let err = with_silent_panics(|| run_isolated(&mut bad, &mut world))
            .expect_err("the panic must be caught");
        assert_eq!(err.entity, None);
        assert_eq!(err.message, "simulated system panic");

        // With an agent in scope, the panic is attributed to it.
        let mut blamed = tick_schedule();
        blamed.add_systems(boom_on_agent_system);
        let err = with_silent_panics(|| run_isolated(&mut blamed, &mut world))
            .expect_err("the panic must be caught");
        assert_eq!(
            err.entity,
            Some(
                Entity::from_raw_u32(41)
                    .expect("a small literal index is always a valid entity id")
            )
        );
        assert_eq!(err.message, "agent-scoped panic");

        // A later clean tick must not inherit the previous tick's entity.
        assert!(run_isolated(&mut ok, &mut world).is_ok());
        assert_eq!(crate::tick_scope::current(), None);
    }

    use crate::components::{AgentState, ContextWindow, InferenceConfig};
    use crate::pipeline::{
        AgentBlueprint, MessageIntake, StageCursor, StageInference, StageInferences, StageProgress,
        StageSetup, StageSetups, VisitCounts,
    };
    use crate::tool_bridge::BoxedToolExec;
    use leviath_core::{Region, RegionKind};
    use leviath_providers::{
        FinishReason, InferenceRequest, InferenceResponse, ModelCapabilities, Provider, TokenUsage,
        ToolCall,
    };
    use std::sync::Mutex;

    /// A provider scripted with a queue of responses; each `infer` pops the next.
    struct Script {
        responses: Mutex<std::collections::VecDeque<InferenceResponse>>,
    }

    #[async_trait::async_trait]
    impl Provider for Script {
        async fn infer(
            &self,
            _req: &InferenceRequest,
        ) -> leviath_providers::Result<InferenceResponse> {
            let next = self.responses.lock().unwrap().pop_front();
            next.ok_or_else(|| ProviderError::Other("script exhausted".to_string()))
        }
        async fn count_tokens(&self, _t: &str, _m: &str) -> usize {
            1
        }
        fn max_context_tokens(&self, _m: &str) -> usize {
            100_000
        }
        fn name(&self) -> &str {
            "script"
        }
        fn capabilities(&self, _m: &str) -> ModelCapabilities {
            ModelCapabilities::default()
        }
    }

    fn text(content: &str) -> InferenceResponse {
        InferenceResponse {
            content: content.to_string(),
            tool_calls: vec![],
            tokens_used: TokenUsage {
                prompt_tokens: 1,
                completion_tokens: 1,
                total_tokens: 2,
                cached_tokens: 0,
                cache_write_tokens: 0,
            },
            finish_reason: FinishReason::Complete,
        }
    }

    fn with_tool(id: &str, name: &str) -> InferenceResponse {
        let mut r = text("");
        r.tool_calls.push(ToolCall {
            id: id.to_string(),
            name: name.to_string(),
            arguments: serde_json::json!({}),
            thought_signature: None,
        });
        r
    }

    /// A tool service that returns a fixed result string for every call.
    struct EchoTools;
    impl ToolService for EchoTools {
        fn exec_for(
            &self,
            _entity: Entity,
            calls: Vec<ToolCall>,
            _progress: crate::pipeline::ToolProgress,
        ) -> BoxedToolExec {
            Box::new(move || {
                Box::pin(async move {
                    calls
                        .into_iter()
                        .map(|c| (c.id, "ok".to_string()))
                        .collect()
                })
            })
        }
    }

    fn window() -> ContextWindow {
        let mut w = ContextWindow::new(10_000);
        w.add_region(Region::new("sys".to_string(), RegionKind::Pinned, 2000));
        w.add_region(Region::new(
            "conversation".to_string(),
            RegionKind::Clearable,
            10_000,
        ));
        w.add_region(Region::new(
            "tool_results".to_string(),
            RegionKind::Temporary,
            5000,
        ));
        w
    }

    fn agent_state() -> AgentState {
        AgentState {
            agent_id: "a".to_string(),
            current_stage: "s".to_string(),
            iteration: 0,
            status: AgentStatus::Active,
            spawned_children_ids: vec![],
            pending_wait: None,
            accepts_messages: true,
        }
    }

    /// A stage advertising the tools the scripted responses here actually call.
    ///
    /// Advertising them is load-bearing: dispatch refuses tools a stage never
    /// offered, so with an empty tool list every end-to-end test that drives a
    /// tool call would short-circuit into a refusal and the tool service would
    /// never be reached at all.
    fn stage(model: &str) -> StageInference {
        StageInference {
            provider_name: "script".to_string(),
            model: model.to_string(),
            tools: ["do", "read"]
                .iter()
                .map(|n| leviath_providers::Tool {
                    name: (*n).to_string(),
                    description: String::new(),
                    parameters: serde_json::json!({}),
                })
                .collect(),
            tool_filter: None,
            fallbacks: Vec::new(),
            output: None,
        }
    }

    fn setup() -> StageSetup {
        StageSetup {
            inference_config: InferenceConfig {
                temperature: None,
                max_output_tokens: None,
                extra_params: Default::default(),
                batch_tool_hint: false,
                shell_hint: false,
                request_timeout_secs: None,
            },
            routing: None,
            accepts_messages: true,
            context_layout: None,
            system_prompt: None,
            output: None,
        }
    }

    fn blueprint() -> leviath_core::Blueprint {
        let layout = leviath_core::layout::ContextLayout::new(
            vec![leviath_core::layout::RegionDefinition::new(
                "conversation".to_string(),
                RegionKind::Clearable,
                10_000,
            )],
            12_000,
        );
        let s = leviath_core::Stage::new(
            "s".to_string(),
            leviath_core::blueprint::ModelConfig::new("script".to_string(), "m".to_string()),
        );
        leviath_core::Blueprint::new("t".to_string(), "d".to_string(), vec![s], layout)
    }

    /// Spawn a single-stage agent, initially ready to infer.
    fn spawn(world: &mut PipelineWorld) -> AgentId {
        world.spawn_agent((
            AgentBlueprint(blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            ReadyToInfer,
        ))
    }

    fn build_world(providers: ProviderRegistry) -> PipelineWorld {
        // These agents carry no RunMetadata, so persistence never fires; run the
        // world fully in memory.
        PipelineWorld::new(
            providers,
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            None,
            Handle::current(),
        )
    }

    #[tokio::test]
    async fn open_circuits_reports_nothing_without_the_breaker() {
        // An embedded world that never installed the resource must report a
        // clean bill of health rather than panicking on a missing resource.
        let world = build_world(ProviderRegistry::new());
        assert!(world.open_circuits().is_empty());
    }

    #[tokio::test]
    async fn open_circuits_reports_a_tripped_provider() {
        let mut world = build_world(ProviderRegistry::new());
        let policy = crate::pipeline::CircuitPolicy {
            failures_before_open: 1,
            cooldown_secs: 300,
        };
        let mut circuits = crate::pipeline::ProviderCircuits::default();
        circuits.record_failure(
            "openrouter",
            leviath_providers::UnavailableReason::CreditsExhausted,
            chrono::Utc::now().timestamp(),
            &policy,
        );
        world.world_mut().insert_resource(circuits);
        world.world_mut().insert_resource(policy);

        let open = world.open_circuits();
        assert_eq!(open.len(), 1);
        assert_eq!(open[0].provider, "openrouter");
        assert_eq!(
            open[0].reason,
            leviath_providers::UnavailableReason::CreditsExhausted
        );
    }

    #[tokio::test]
    async fn open_circuits_falls_back_to_the_default_policy() {
        // Circuits installed, policy not: the default must apply rather than
        // the report silently coming back empty.
        let mut world = build_world(ProviderRegistry::new());
        let default_policy = crate::pipeline::CircuitPolicy::default();
        let mut circuits = crate::pipeline::ProviderCircuits::default();
        for _ in 0..default_policy.failures_before_open {
            circuits.record_failure(
                "openrouter",
                leviath_providers::UnavailableReason::AuthFailed,
                chrono::Utc::now().timestamp(),
                &default_policy,
            );
        }
        world.world_mut().insert_resource(circuits);

        assert_eq!(world.open_circuits().len(), 1);
    }

    #[tokio::test]
    async fn set_exact_token_counting_toggles_the_stage_flag() {
        let mut world = build_world(ProviderRegistry::new());
        // Default is off.
        assert!(
            !world
                .world()
                .resource::<crate::pipeline::InferenceStage>()
                .exact_token_counting
        );
        world.set_exact_token_counting(true);
        assert!(
            world
                .world()
                .resource::<crate::pipeline::InferenceStage>()
                .exact_token_counting
        );
    }

    #[tokio::test]
    async fn run_to_fixed_point_survives_a_panicking_system() {
        // A system that panics must not hang or crash the drive loop - it's
        // caught and the loop breaks (the daemon survives).
        fn boom_system() {
            panic!("simulated system panic");
        }
        let mut world = build_world(ProviderRegistry::new());
        world.add_test_system(boom_system);
        // Unattributed: nothing to fail, so the round stops immediately.
        with_silent_panics(|| world.run_to_fixed_point());
    }

    #[tokio::test]
    async fn a_panic_on_the_compute_pool_is_attributed_to_its_agent() {
        // `dispatch_inference` fans its per-agent work out over the compute task
        // pool, where the thread-local scope can't reach the driver thread that
        // catches unwinds. Those bodies run under `run_agent_parallel`, which
        // catches on the pool thread and marks the agent instead - this proves
        // the marker makes it back and fails the right run (issue #109).
        fn boom_in_parallel(
            agents: Query<(Entity, &AgentState)>,
            par_commands: bevy_ecs::system::ParallelCommands,
        ) {
            agents.par_iter().for_each(|(entity, state)| {
                if state.status != AgentStatus::Active {
                    return; // already failed - nothing left to blow up
                }
                // Clear the thread-local first: whatever attributes this panic,
                // it is demonstrably not the `enter`/`current` mechanism.
                crate::tick_scope::clear();
                crate::tick_scope::run_agent_parallel(entity, &par_commands, &mut || {
                    panic!("blew up on the compute pool");
                });
            });
        }

        let mut world = build_world(ProviderRegistry::new());
        let entity = spawn(&mut world);
        world.add_test_system(boom_in_parallel);
        with_silent_panics(|| world.run_to_fixed_point());

        let status = world.agent_status(entity);
        assert!(
            matches!(status, Some(AgentStatus::Error { ref message })
                if message.contains("a pipeline system panicked")
                    && message.contains("blew up on the compute pool")),
            "got: {status:?}"
        );
        // The marker is consumed, so a later tick doesn't re-fail the agent.
        assert!(
            world
                .world()
                .entity(entity.entity())
                .get::<crate::tick_scope::PanickedInParallel>()
                .is_none(),
            "the marker must be drained once acted on"
        );
    }

    #[tokio::test]
    async fn a_panicking_system_fails_its_agent_instead_of_looping_forever() {
        // Before issue #109 was fixed, a panicking system was swallowed
        // anonymously: nothing changed, so the very next wake re-ticked the same
        // state and panicked again, forever, while every other agent stalled.
        // Now the agent in scope is failed, which takes it out of the dispatch
        // systems (they only act on `Active` agents) and lets the world settle.
        static VICTIM: std::sync::Mutex<Option<Entity>> = std::sync::Mutex::new(None);
        static PANICS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

        fn boom_on_active_agent(agents: Query<(Entity, &AgentState)>) {
            // No trailing statements after the `panic!`: an unreachable tail
            // would read as uncovered under the workspace's 100% gate.
            let Some((entity, _)) = agents
                .iter()
                .find(|(_, state)| state.status == AgentStatus::Active)
            else {
                return; // the agent has been failed - nothing left to blow up
            };
            crate::tick_scope::enter(entity);
            *VICTIM
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(entity);
            PANICS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            panic!("blew up on this agent");
        }

        let mut world = build_world(ProviderRegistry::new());
        let entity = spawn(&mut world);
        world.add_test_system(boom_on_active_agent);
        with_silent_panics(|| world.run_to_fixed_point());

        let victim = VICTIM
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
        assert_eq!(
            victim,
            Some(entity.entity()),
            "the system saw the spawned agent"
        );
        let status = world.agent_status(entity);
        assert!(
            matches!(status, Some(AgentStatus::Error { ref message })
                if message.contains("a pipeline system panicked")
                    && message.contains("blew up on this agent")),
            "got: {status:?}"
        );
        // The loop terminated rather than re-panicking without bound.
        assert!(
            PANICS.load(std::sync::atomic::Ordering::SeqCst) <= MAX_TICK_FAILURES_PER_ROUND + 1,
            "the panic budget must stop the round"
        );
    }

    fn registry_with(responses: Vec<InferenceResponse>) -> ProviderRegistry {
        let mut r = ProviderRegistry::new();
        r.register(
            "script".to_string(),
            Arc::new(Script {
                responses: Mutex::new(responses.into_iter().collect()),
            }),
        );
        r
    }

    #[tokio::test]
    async fn an_agent_whose_provider_is_missing_wedges_at_iteration_zero() {
        // Issue #190. The registry has no `script` provider, so
        // `dispatch_inference` declines and leaves the agent `ReadyToInfer`.
        // Nothing about the world changed, so the fixed point is reached
        // immediately and nothing is in flight to wake the driver - the agent
        // used to sit `Active` at iteration 0 for ever, which on disk reads as
        // a `running` run with no tokens and a frozen `updated_at`.
        let mut world = build_world(ProviderRegistry::new());
        let e = spawn(&mut world);

        world.run_until_idle(30).await;

        // Nothing has dispatched, and within the grace period that is still
        // just a wait - but it is now a *recorded* one.
        let state = world
            .world()
            .get::<AgentState>(e.entity())
            .expect("the agent");
        assert_eq!(state.iteration, 0, "not a single inference happened");
        assert_eq!(state.status, AgentStatus::Active);
        let stall = world
            .world()
            .get::<crate::pipeline::DispatchStall>(e.entity())
            .expect("the decline is recorded");
        assert_eq!(stall.reason, crate::pipeline::StallReason::ProviderMissing);

        // Backdate it past the grace period, as the host's redrive timer would
        // find it on a later tick. The run stops claiming to be working and
        // says what it needs instead of hanging - and, since a missing
        // provider is one config edit away from being present, it parks rather
        // than dies, so the edit can be followed by `lev resume`.
        let past =
            chrono::Utc::now().timestamp() - crate::pipeline::DEFAULT_STALL_TIMEOUT_SECS as i64 - 1;
        world
            .world_mut()
            .get_mut::<crate::pipeline::DispatchStall>(e.entity())
            .expect("the stall record")
            .since = past;
        world.run_to_fixed_point();

        assert_eq!(world.agent_status(e), Some(AgentStatus::Paused));
        let parked = world
            .world()
            .get::<crate::pipeline::PausedForSetup>(e.entity())
            .expect("it says what to do");
        assert!(
            parked.remedy.contains("script") && parked.remedy.contains("not configured"),
            "{}",
            parked.remedy
        );
        assert!(
            world.world().get::<ReadyToInfer>(e.entity()).is_some(),
            "the retry stays staged, so a resume re-dispatches it"
        );
    }

    /// Issue #202, end to end through the real schedule: an agent stripped of
    /// every phase marker is unreachable, and the watchdog registered in the
    /// chain above fails it rather than leaving it `running` for ever.
    ///
    /// This also proves the fixed-point loop still converges with the new system
    /// in it. The watchdog writes a `Wedged` record on its first pass, so a tick
    /// does change the world; if that record fed the fingerprint the loop would
    /// spin instead of parking, which is why it deliberately does not.
    #[tokio::test]
    async fn a_run_nothing_can_drive_is_failed_rather_than_left_running() {
        let mut world = build_world(registry_with(vec![]));
        world
            .world_mut()
            .insert_resource(crate::pipeline::WedgeTimeout(60));
        let e = spawn(&mut world);

        // Strip the agent of the marker it spawned with. Nothing in the engine
        // does this; a panicking system that dropped a marker without landing a
        // successor is what it stands in for.
        world
            .world_mut()
            .entity_mut(e.entity())
            .remove::<ReadyToInfer>();
        world.run_to_fixed_point();

        // First pass records it. Inside the grace period it is still just a wait.
        assert_eq!(
            world.agent_status(e),
            Some(AgentStatus::Active),
            "not failed while it is still inside the grace period"
        );
        let since = world
            .world()
            .get::<crate::pipeline::Wedged>(e.entity())
            .expect("the wedge is recorded")
            .since;

        // Backdate past the grace period, as the host's redrive would find it.
        world
            .world_mut()
            .get_mut::<crate::pipeline::Wedged>(e.entity())
            .expect("the wedge record")
            .since = since - 61;
        world.run_to_fixed_point();

        let status = world.agent_status(e);
        assert!(
            matches!(status, Some(AgentStatus::Error { ref message })
                if message.contains("never move again")),
            "got: {status:?}"
        );
    }

    #[tokio::test]
    async fn agent_completes_after_nudges_exhausted() {
        // Text-only responses with no tool calls get nudged up to the max; the
        // response after the last nudge is accepted and the single-stage
        // blueprint terminates the agent. (Exercises the handle_empty_response
        // nudge loop end-to-end through the driver.)
        let mut world = build_world(registry_with(vec![
            text("thinking"),
            text("still"),
            text("more"),
            text("final"),
        ]));
        let e = spawn(&mut world);

        world.run_until_idle(30).await;

        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
    }

    #[tokio::test]
    async fn agent_nudge_max_bounds_the_loop_end_to_end() {
        // `[agent.nudge] max = 1` (issue #127): the second text-only response
        // is final, so a two-response script finishes where the default cap
        // would have demanded four. A third scripted response left unconsumed
        // would keep the driver looping past run_until_idle's budget.
        let mut world = build_world(registry_with(vec![text("thinking"), text("final")]));
        let mut bp = blueprint();
        bp.nudge = Some(leviath_core::NudgeConfig {
            max: Some(1),
            ..Default::default()
        });
        let e = world.spawn_agent((
            AgentBlueprint(bp),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            ReadyToInfer,
        ));

        world.run_until_idle(30).await;

        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
    }

    #[tokio::test]
    async fn agent_runs_tools_then_completes() {
        // First response calls a tool; after the tool result comes back the
        // second response is text-only, finishing the run.
        let mut world = build_world(registry_with(vec![with_tool("c1", "do"), text("done")]));
        let e = spawn(&mut world);

        world.run_until_idle(20).await;

        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
        // With no routing configured, tool results land in the conversation
        // region.
        assert!(
            world
                .world()
                .get::<ContextWindow>(e.entity())
                .unwrap()
                .get_region("conversation")
                .unwrap()
                .current_tokens
                > 0
        );
    }

    #[tokio::test]
    async fn insert_interaction_hub_installs_resource_and_attaches_wake() {
        use crate::dynamic_interaction::InteractionBackend;
        use crate::interaction_hub::InteractionHub;
        let mut world = build_world(registry_with(vec![]));
        let hub = InteractionHub::new();
        world.insert_interaction_hub(hub.clone());

        // The hub is now a world resource the reflect system reads.
        assert!(world.world().get_resource::<InteractionHub>().is_some());

        // The wake handle was attached: opening a request nudges the same wake
        // the driver parks on (a later notified() returns immediately).
        let backend = hub.backend_for("x");
        let asking = tokio::spawn(async move {
            backend
                .ask(leviath_core::interaction::InteractionRequest::free_text(
                    "q", "p", "s", true,
                ))
                .await
        });
        for _ in 0..8 {
            tokio::task::yield_now().await;
        }
        world.wake_handle().notified().await;
        hub.cancel("q");
        let _ = asking.await;
    }

    #[tokio::test]
    async fn provider_error_marks_agent_error() {
        // Empty script ⇒ the very first infer errors.
        let mut world = build_world(registry_with(vec![]));
        let e = spawn(&mut world);

        world.run_until_idle(20).await;

        assert_eq!(
            std::mem::discriminant(&world.agent_status(e).unwrap()),
            std::mem::discriminant(&AgentStatus::Error {
                message: String::new()
            })
        );
    }

    #[tokio::test]
    async fn send_message_reaches_the_agent_inbox() {
        // No responses queued: the agent dispatches inference and parks awaiting
        // it. We deliver a message; the deliver system routes it to context.
        let mut world = build_world(registry_with(vec![]));
        let e = spawn(&mut world);
        // Drive to the point the first (doomed) inference is dispatched/collected.
        world.run_until_idle(20).await;

        world
            .send_message(AgentMessage {
                agent_id: "a".to_string(),
                content: "hello".to_string(),
                target_region: Some("conversation".to_string()),
            })
            .unwrap();
        world.tick(); // deliver_messages runs

        assert!(
            world
                .world()
                .get::<ContextWindow>(e.entity())
                .unwrap()
                .get_region("conversation")
                .unwrap()
                .current_tokens
                > 0
        );
    }

    #[tokio::test]
    async fn run_returns_on_shutdown() {
        let mut world = build_world(registry_with(vec![text("done")]));
        spawn(&mut world);
        world.shutdown(); // pre-signal: run parks then returns
        // Must return rather than loop forever.
        world.run().await;
    }

    #[tokio::test]
    async fn run_wakes_then_shuts_down() {
        // Drives run() on its own task: a wake makes it loop once (wake branch),
        // then a shutdown makes it return (shutdown branch).
        let mut world = build_world(registry_with(vec![
            text("t1"),
            text("t2"),
            text("t3"),
            text("t4"),
        ]));
        spawn(&mut world);
        let wake = world.wake_handle();
        let shutdown = world.shutdown_handle();
        let handle = tokio::spawn(async move { world.run().await });

        wake.notify_one();
        tokio::task::yield_now().await;
        shutdown.notify_one();

        handle.await.unwrap(); // returns cleanly
    }

    #[tokio::test]
    async fn send_message_errors_when_intake_dropped() {
        let mut world = build_world(registry_with(vec![]));
        // Drop the intake receiver via the world accessor, closing the channel.
        let removed = world.world_mut().remove_resource::<MessageIntake>();
        drop(removed);

        let err = world.send_message(AgentMessage {
            agent_id: "a".to_string(),
            content: "x".to_string(),
            target_region: None,
        });
        assert!(err.is_err());
    }

    #[tokio::test]
    async fn script_provider_metadata_is_exercised() {
        // Keep the mock's non-`infer`/`capabilities` methods measured.
        let p = Script {
            responses: Mutex::new(std::collections::VecDeque::new()),
        };
        assert_eq!(p.name(), "script");
        assert_eq!(p.count_tokens("t", "m").await, 1);
        assert_eq!(p.max_context_tokens("m"), 100_000);
        let _ = p.capabilities("m");
    }

    #[tokio::test]
    async fn agent_status_is_none_for_unknown_entity() {
        let world = build_world(registry_with(vec![]));
        assert_eq!(
            // Scoped to this world, but naming an entity it never spawned.
            world.agent_status(
                world.own_agent(
                    Entity::from_raw_u32(999)
                        .expect("a small literal index is always a valid entity id")
                )
            ),
            None
        );
    }

    #[tokio::test]
    async fn paused_agent_does_not_progress_until_resumed() {
        let mut world = build_world(registry_with(vec![
            text("t1"),
            text("t2"),
            text("t3"),
            text("t4"),
        ]));
        let e = spawn(&mut world);
        assert!(world.pause(e));

        world.run_until_idle(30).await;
        // Paused ⇒ parked, never inferred.
        assert_eq!(world.agent_status(e), Some(AgentStatus::Paused));

        assert!(world.resume(e));
        world.run_until_idle(30).await;
        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
    }

    #[tokio::test]
    async fn resume_resets_the_provider_circuits() {
        // Issue #413: a run paused on exhausted credits comes back through an
        // explicit resume. If the breaker kept its state, the retry would sit
        // out the rest of the cooldown and the resume would look ignored.
        let mut world = build_world(registry_with(vec![text("t1")]));
        let e = spawn(&mut world);
        assert!(world.pause(e));

        let policy = crate::pipeline::CircuitPolicy {
            failures_before_open: 1,
            cooldown_secs: 300,
        };
        let mut circuits = crate::pipeline::ProviderCircuits::default();
        circuits.record_failure(
            "openrouter",
            leviath_providers::UnavailableReason::CreditsExhausted,
            chrono::Utc::now().timestamp(),
            &policy,
        );
        world.world_mut().insert_resource(circuits);
        world.world_mut().insert_resource(policy);
        assert_eq!(world.open_circuits().len(), 1);

        assert!(world.resume(e));
        assert!(world.open_circuits().is_empty());
    }

    /// An id belonging to another world names a different agent here, so
    /// resuming with one refuses rather than resuming whatever happens to sit
    /// at that entity index. This is the check `AgentId` exists for.
    #[tokio::test]
    async fn resume_refuses_an_id_from_another_world() {
        let mut theirs = build_world(registry_with(vec![text("t1")]));
        let e = spawn(&mut theirs);
        assert!(theirs.pause(e));

        let mut ours = build_world(registry_with(vec![text("t1")]));
        assert!(
            !ours.resume(e),
            "an id from elsewhere is not ours to resume"
        );
        // And the run it really names is untouched.
        assert_eq!(theirs.agent_status(e), Some(AgentStatus::Paused));
    }

    /// Resuming drops the "needs setup" note.
    ///
    /// If the machine was not actually fixed the watchdog re-parks the run a
    /// minute later with a fresh message, which is better than carrying a
    /// stale one describing a problem somebody may have just solved.
    #[tokio::test]
    async fn resume_clears_the_note_saying_what_the_run_needed() {
        let mut world = build_world(registry_with(vec![text("t1")]));
        let e = spawn(&mut world);
        assert!(world.pause(e));
        world
            .world_mut()
            .entity_mut(e.entity())
            .insert(crate::pipeline::PausedForSetup {
                blocker: leviath_core::run_meta::SetupBlocker::ProviderMissing,
                remedy: "add it to config.toml".to_string(),
            });

        assert!(world.resume(e));

        assert!(
            world
                .world()
                .get::<crate::pipeline::PausedForSetup>(e.entity())
                .is_none(),
            "a resumed run no longer claims to need setup"
        );
    }

    #[tokio::test]
    async fn pause_refuses_waiting_and_terminal_agents() {
        let mut world = build_world(registry_with(vec![text("t1")]));
        let e = spawn(&mut world);

        // A Waiting agent's status is the marker fan-out merges and interaction
        // resolution key off - pause must not clobber it.
        world.set_status(e, AgentStatus::Waiting);
        assert!(!world.pause(e));
        assert_eq!(world.agent_status(e), Some(AgentStatus::Waiting));

        world.set_status(e, AgentStatus::Cancelled);
        assert!(!world.pause(e));
        assert_eq!(world.agent_status(e), Some(AgentStatus::Cancelled));
    }

    #[tokio::test]
    async fn resume_refuses_agents_that_are_not_paused_or_idle() {
        let mut world = build_world(registry_with(vec![text("t1")]));
        let e = spawn(&mut world);

        // Already running: nothing to resume.
        world.set_status(e, AgentStatus::Active);
        assert!(!world.resume(e));

        world.set_status(e, AgentStatus::Waiting);
        assert!(!world.resume(e));
        assert_eq!(world.agent_status(e), Some(AgentStatus::Waiting));

        world.set_status(e, AgentStatus::Complete);
        assert!(!world.resume(e));
        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
    }

    #[tokio::test]
    async fn resume_nudges_an_idle_agent_active() {
        let mut world = build_world(registry_with(vec![text("t1")]));
        let e = spawn(&mut world);
        world.set_status(e, AgentStatus::Idle);
        assert!(world.resume(e));
        assert_eq!(world.agent_status(e), Some(AgentStatus::Active));
    }

    #[tokio::test]
    async fn cancelled_agent_stops_progressing() {
        let mut world = build_world(registry_with(vec![with_tool("c1", "do"), text("done")]));
        let e = spawn(&mut world);
        assert!(world.cancel(e));

        world.run_until_idle(20).await;

        assert_eq!(world.agent_status(e), Some(AgentStatus::Cancelled));
    }

    #[tokio::test]
    async fn status_ops_return_false_for_unknown_entity() {
        let mut world = build_world(registry_with(vec![]));
        // Scoped to this world, but naming an entity it never spawned.
        let unknown = world.own_agent(
            Entity::from_raw_u32(999).expect("a small literal index is always a valid entity id"),
        );
        assert!(!world.pause(unknown));
        assert!(!world.resume(unknown));
        assert!(!world.cancel(unknown));
    }

    #[tokio::test]
    async fn spawn_from_blueprint_builds_a_runnable_agent() {
        // End-to-end via the blueprint resolver: build → drive → complete.
        let mut world = build_world(registry_with(vec![with_tool("c1", "do"), text("done")]));
        let e = world
            .spawn_from_blueprint(
                "agent-1".to_string(),
                blueprint(),
                "do the task",
                vec![crate::pipeline::ResolvedStage {
                    provider_name: "script".to_string(),
                    model: "m".to_string(),
                    tools: vec![],
                    fallbacks: Vec::new(),
                    output: None,
                }],
                hints(true),
            )
            .unwrap();

        world.run_until_idle(20).await;

        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
    }

    #[tokio::test]
    async fn persists_agent_snapshot_to_runs_dir() {
        // An agent carrying RunMetadata + TokenTotals is snapshotted to disk as it
        // runs; after it completes, meta.json exists with the final status.
        let dir = tempfile::tempdir().unwrap();
        let mut world = PipelineWorld::new(
            registry_with(vec![with_tool("c1", "do"), text("done")]),
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            Some(dir.path().to_path_buf()),
            Handle::current(),
        );
        world.spawn_agent((
            AgentBlueprint(blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            crate::persistence::RunMetadata {
                run_id: "run-42".to_string(),
                agent_name: "a".to_string(),
                agent_path: "/p".to_string(),
                task: "t".to_string(),
                model: None,
                // A real directory: the tick chain fails a run whose workspace is gone.
                workdir: std::env::temp_dir().to_string_lossy().to_string(),
                num_stages: 1,
                started_at: 0,
                parent_run_id: None,
                metadata: std::collections::HashMap::new(),
                callback_url: None,
                callback_secret: None,
                title: None,
                unattended: false,
                read_paths: None,
                output_request: None,
            },
            crate::persistence::TokenTotals::default(),
            crate::pipeline::PersistWatermark::default(),
            ReadyToInfer,
        ));

        world.run_until_idle(20).await;

        // The persistence worker is fire-and-forget on its own task; poll until the
        // final (Complete) snapshot has been flushed. A short real sleep between
        // polls (rather than a bare `yield_now`) gives the worker's write actual
        // wall-clock time to land under load - otherwise the loop can spin through
        // every iteration before the write completes and spuriously time out.
        let meta_path = dir.path().join("run-42").join("meta.json");
        let mut meta = None;
        for _ in 0..200 {
            if let Ok(text) = std::fs::read_to_string(&meta_path)
                && let Ok(m) = serde_json::from_str::<leviath_core::run_meta::RunMeta>(&text)
                && m.status == leviath_core::run_meta::RunStatus::Complete
            {
                meta = Some(m);
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }

        let meta = meta.expect("final Complete snapshot flushed to disk");
        assert_eq!(meta.run_id, "run-42");
        assert!(dir.path().join("run-42").join("context.json").exists());
    }

    #[tokio::test]
    async fn a_panicked_agent_is_recorded_as_errored_on_disk() {
        // The reported symptom in issue #109: a crashed run stayed `"running"`
        // in meta.json forever. `dispatch_persistence` is the *last* system in
        // the chain, so the tick that panics never reaches it - which is exactly
        // why `run_to_fixed_point` keeps driving after failing the agent.
        fn boom_on_active_agent(agents: Query<(Entity, &AgentState)>) {
            let Some((entity, _)) = agents
                .iter()
                .find(|(_, state)| state.status == AgentStatus::Active)
            else {
                return; // the agent has been failed - nothing left to blow up
            };
            crate::tick_scope::enter(entity);
            panic!("exploded mid-stage");
        }

        let dir = tempfile::tempdir().unwrap();
        let mut world = PipelineWorld::new(
            registry_with(vec![]),
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            Some(dir.path().to_path_buf()),
            Handle::current(),
        );
        world.spawn_agent((
            AgentBlueprint(blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            crate::persistence::RunMetadata {
                run_id: "run-boom".to_string(),
                agent_name: "a".to_string(),
                agent_path: "/p".to_string(),
                task: "t".to_string(),
                model: None,
                workdir: "/w".to_string(),
                num_stages: 1,
                started_at: 0,
                parent_run_id: None,
                metadata: std::collections::HashMap::new(),
                callback_url: None,
                callback_secret: None,
                title: None,
                unattended: false,
                read_paths: None,
                output_request: None,
            },
            crate::persistence::TokenTotals::default(),
            crate::pipeline::PersistWatermark::default(),
            ReadyToInfer,
        ));
        world.add_test_system(boom_on_active_agent);
        with_silent_panics(|| world.run_to_fixed_point());

        let meta_path = dir.path().join("run-boom").join("meta.json");
        let mut meta = None;
        for _ in 0..200 {
            if let Ok(text) = std::fs::read_to_string(&meta_path)
                && let Ok(m) = serde_json::from_str::<leviath_core::run_meta::RunMeta>(&text)
                && m.status == leviath_core::run_meta::RunStatus::Error
            {
                meta = Some(m);
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        let meta = meta.expect("the panicked run must be persisted as errored");
        let error = meta.error.unwrap_or_default();
        assert!(error.contains("a pipeline system panicked"), "got: {error}");
        assert!(error.contains("exploded mid-stage"), "got: {error}");
    }

    /// A single-stage blueprint whose stage is an `interactive_points` stage with a
    /// `plan_approval` point (the shape that blocks awaiting human approval).
    fn interactive_blueprint() -> leviath_core::Blueprint {
        use leviath_core::blueprint::{InteractionPoint, InteractionStyle, StageMode};
        let layout = leviath_core::layout::ContextLayout::new(
            vec![leviath_core::layout::RegionDefinition::new(
                "conversation".to_string(),
                RegionKind::Clearable,
                10_000,
            )],
            12_000,
        );
        let mut s = leviath_core::Stage::new(
            "plan".to_string(),
            leviath_core::blueprint::ModelConfig::new("script".to_string(), "m".to_string()),
        );
        s.mode = StageMode::InteractivePoints {
            points: vec![InteractionPoint {
                name: "plan_approval".to_string(),
                prompt: "Approve?".to_string(),
                required: true,
                unattended: leviath_core::blueprint::UnattendedPolicy::AutoApprove,
                style: InteractionStyle::MultipleChoice,
                options: vec!["Approve".to_string(), "Abort".to_string()],
                directives: std::collections::HashMap::new(),
                abort_options: vec!["Abort".to_string()],
                edit_options: vec![],
                document_region: None,
            }],
        };
        leviath_core::Blueprint::new("t".to_string(), "d".to_string(), vec![s], layout)
    }

    #[tokio::test]
    async fn persists_interaction_point_when_a_live_agent_blocks() {
        // Drive a real agent through inference → transition → the interaction-point
        // lane until it blocks awaiting approval, and assert the daemon wrote the
        // `interactions.json` sidecar - the issue #38 persist side, end-to-end
        // through the live lane (a tool call first, then a text "plan", so the stage
        // transitions into the interaction point rather than looping on nudges).
        let dir = tempfile::tempdir().unwrap();
        let mut world = PipelineWorld::new(
            registry_with(vec![with_tool("c1", "read"), text("## Plan\n1. do it")]),
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            Some(dir.path().to_path_buf()),
            Handle::current(),
        );
        world.insert_interaction_hub(crate::interaction_hub::InteractionHub::new());
        let e = world.spawn_agent((
            AgentBlueprint(interactive_blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            crate::persistence::RunMetadata {
                run_id: "run-ip".to_string(),
                agent_name: "a".to_string(),
                agent_path: "/p".to_string(),
                task: "t".to_string(),
                model: None,
                // A real directory: the tick chain fails a run whose workspace is gone.
                workdir: std::env::temp_dir().to_string_lossy().to_string(),
                num_stages: 1,
                started_at: 0,
                parent_run_id: None,
                metadata: std::collections::HashMap::new(),
                callback_url: None,
                callback_secret: None,
                title: None,
                unattended: false,
                read_paths: None,
                output_request: None,
            },
            crate::persistence::TokenTotals::default(),
            crate::pipeline::PersistWatermark::default(),
            ReadyToInfer,
        ));

        world.run_until_idle(30).await;
        // `run_until_idle` stops once no inference/tool is in flight, but the
        // interaction-point ask task registers in the hub just after; the real
        // daemon's `run()` loop catches its wake, so pump fixed points here until
        // `reflect_interaction_status` flips the agent to Waiting (and persistence
        // captures the sidecar).
        for _ in 0..50 {
            if world.agent_status(e) == Some(AgentStatus::Waiting) {
                break;
            }
            tokio::task::yield_now().await;
            world.run_to_fixed_point();
        }
        assert_eq!(world.agent_status(e), Some(AgentStatus::Waiting));

        // Poll until the interaction sidecar lands (the persistence worker writes it
        // on its own task once the agent is parked Waiting at the point).
        let path = dir.path().join("run-ip").join("interactions.json");
        let mut sidecar = None;
        for _ in 0..200 {
            if let Ok(t) = std::fs::read_to_string(&path)
                && let Ok(s) =
                    serde_json::from_str::<crate::interaction_points::InteractionPointState>(&t)
            {
                sidecar = Some(s);
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        let s = sidecar.expect("interaction-point sidecar flushed to disk");
        assert_eq!(s.cursor, 0);
        assert_eq!(s.round, 0);
        assert_eq!(s.body, "## Plan\n1. do it");
    }

    #[tokio::test]
    async fn flush_and_stop_drains_queued_snapshots() {
        // Unlike a plain shutdown, `flush_and_stop` awaits the persistence worker,
        // so the final snapshot is guaranteed on disk the instant it returns - no
        // filesystem polling required (contrast the test above).
        let dir = tempfile::tempdir().unwrap();
        let mut world = PipelineWorld::new(
            registry_with(vec![with_tool("c1", "do"), text("done")]),
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            Some(dir.path().to_path_buf()),
            Handle::current(),
        );
        world.spawn_agent((
            AgentBlueprint(blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            crate::persistence::RunMetadata {
                run_id: "run-flush".to_string(),
                agent_name: "a".to_string(),
                agent_path: "/p".to_string(),
                task: "t".to_string(),
                model: None,
                // A real directory: the tick chain fails a run whose workspace is gone.
                workdir: std::env::temp_dir().to_string_lossy().to_string(),
                num_stages: 1,
                started_at: 0,
                parent_run_id: None,
                metadata: std::collections::HashMap::new(),
                callback_url: None,
                callback_secret: None,
                title: None,
                unattended: false,
                read_paths: None,
                output_request: None,
            },
            crate::persistence::TokenTotals::default(),
            crate::pipeline::PersistWatermark::default(),
            ReadyToInfer,
        ));

        world.run_until_idle(20).await;
        world.flush_and_stop().await;

        // Read immediately - the drain guarantees the write landed.
        let meta_path = dir.path().join("run-flush").join("meta.json");
        let text = std::fs::read_to_string(&meta_path).expect("meta.json flushed on stop");
        let meta: leviath_core::run_meta::RunMeta = serde_json::from_str(&text).unwrap();
        assert_eq!(meta.run_id, "run-flush");
        assert_eq!(meta.status, leviath_core::run_meta::RunStatus::Complete);

        // A second call is a no-op (resource already removed, task taken) - no panic.
        world.flush_and_stop().await;
        assert!(meta_path.exists());
    }

    #[tokio::test]
    async fn in_memory_world_runs_and_flushes_without_touching_disk() {
        // `runs_dir: None` is the embedding mode: the agent runs to completion,
        // snapshots are produced and drained exactly as in the persistent world
        // (same watermark/log behavior), but nothing lands on disk. The tempdir
        // doubles as the agent workdir and as the canary a persistent world
        // would have written run dirs and a machine-id into.
        let dir = tempfile::tempdir().unwrap();
        let mut world = PipelineWorld::new(
            registry_with(vec![with_tool("c1", "do"), text("done")]),
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            None,
            Handle::current(),
        );
        let entity = world.spawn_agent((
            AgentBlueprint(blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            crate::persistence::RunMetadata {
                run_id: "run-inmem".to_string(),
                agent_name: "a".to_string(),
                agent_path: "/p".to_string(),
                task: "t".to_string(),
                model: None,
                workdir: dir.path().to_string_lossy().to_string(),
                num_stages: 1,
                started_at: 0,
                parent_run_id: None,
                metadata: std::collections::HashMap::new(),
                callback_url: None,
                callback_secret: None,
                title: None,
                unattended: false,
                read_paths: None,
                output_request: None,
            },
            crate::persistence::TokenTotals::default(),
            crate::pipeline::PersistWatermark::default(),
            ReadyToInfer,
        ));

        world.run_until_idle(20).await;
        world.flush_and_stop().await;

        assert_eq!(world.agent_status(entity), Some(AgentStatus::Complete));
        assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn world_init_and_restore_needs_no_daemon_infra() {
        // `PipelineWorld::new` + `restore::restore_agent` form a self-contained
        // spin-up→restore path: no control socket, HTTP server, PID files, or build
        // markers - only providers, a tool service, a runs dir, and a runtime. This
        // locks that in so the daemon wiring stays optional.
        use leviath_core::region::EntryKind;
        use leviath_core::run_meta::{ContextSnapshot, RegionEntrySnapshot, RegionSnapshot};

        let dir = tempfile::tempdir().unwrap();
        let mut world = PipelineWorld::new(
            registry_with(vec![text("unused")]),
            Arc::new(EchoTools),
            InferencePoolConfig::new(),
            1,
            Some(dir.path().to_path_buf()),
            Handle::current(),
        );
        let entity = world.spawn_agent((
            AgentBlueprint(blueprint()),
            StageCursor { index: 0 },
            agent_state(),
            crate::components::MessageInbox::default(),
            StageProgress::default(),
            StageInferences(vec![stage("m")]),
            StageSetups(vec![setup()]),
            VisitCounts::default(),
            window(),
            stage("m"),
            setup().inference_config,
            crate::persistence::TokenTotals::default(),
        ));

        let snapshot = ContextSnapshot {
            stage_name: "s0".to_string(),
            total_tokens: 4,
            max_tokens: 10_000,
            regions: vec![RegionSnapshot {
                name: "conversation".to_string(),
                kind: "clearable".to_string(),
                current_tokens: 4,
                max_tokens: 10_000,
                entries: vec![RegionEntrySnapshot {
                    content: "restored turn".to_string(),
                    tokens: 4,
                    kind: EntryKind::UserMessage,
                    metadata: None,
                    key: None,
                    taint: Default::default(),
                }],
            }],
        };
        crate::restore::restore_agent(
            world.world_mut(),
            entity.entity(),
            &snapshot,
            0,
            3,
            crate::persistence::TokenTotals::default(),
        );

        let state = world
            .world()
            .get::<crate::components::AgentState>(entity.entity())
            .unwrap();
        assert_eq!(state.status, AgentStatus::Active);
        assert_eq!(state.iteration, 3);
        let win = world
            .world()
            .get::<crate::components::ContextWindow>(entity.entity())
            .unwrap();
        assert_eq!(
            win.get_region("conversation").unwrap().content[0].content,
            "restored turn"
        );
    }

    #[tokio::test]
    async fn spawn_from_blueprint_errors_on_oversized_system_prompt() {
        let mut world = build_world(registry_with(vec![]));
        // A blueprint whose stage carries an enormous system prompt in a tiny
        // pinned region overflows at spawn.
        let layout = leviath_core::layout::ContextLayout::new(
            vec![leviath_core::layout::RegionDefinition::new(
                "task".to_string(),
                RegionKind::Pinned,
                50,
            )],
            1000,
        );
        let mut s = leviath_core::Stage::new(
            "s".to_string(),
            leviath_core::blueprint::ModelConfig::new("script".to_string(), "m".to_string()),
        );
        s.config.insert(
            "system_prompt".to_string(),
            serde_json::Value::String("x".repeat(100_000)),
        );
        let bp = leviath_core::Blueprint::new("t".to_string(), "d".to_string(), vec![s], layout);

        let err = world.spawn_from_blueprint(
            "a".to_string(),
            bp,
            "task",
            vec![crate::pipeline::ResolvedStage {
                provider_name: "script".to_string(),
                model: "m".to_string(),
                tools: vec![],
                fallbacks: Vec::new(),
                output: None,
            }],
            hints(true),
        );
        assert!(err.is_err());
    }

    #[tokio::test]
    async fn wake_handle_and_run_until_idle_bound_are_exposed() {
        // Exercises the wake handle accessor and the max-waits safety bound on a
        // world with an agent parked on an in-flight inference that never
        // resolves within the bound (script returns after we stop waiting).
        let mut world = build_world(registry_with(vec![with_tool("c1", "do"), text("done")]));
        let _ = world.wake_handle();
        let e = spawn(&mut world);
        world.run_until_idle(0).await; // bound 0 ⇒ no extra waits
        // With no waits allowed we may not have observed completion yet; drain.
        world.run_until_idle(20).await;
        assert_eq!(world.agent_status(e), Some(AgentStatus::Complete));
    }

    // ─── Two worlds at once ─────────────────────────────────────────────────
    //
    // Multi-world is planned, so the properties it rests on are asserted now
    // rather than discovered later. Two of these pass today; the third records
    // a real hazard that is *not* closed, so that it is a known quantity rather
    // than a surprise.

    #[tokio::test]
    async fn two_worlds_each_drive_their_own_agents() {
        let mut a = build_world(ProviderRegistry::new());
        let mut b = build_world(ProviderRegistry::new());
        let in_a = spawn(&mut a);
        let in_b = spawn(&mut b);

        assert!(a.agent_status(in_a).is_some());
        assert!(b.agent_status(in_b).is_some());

        // Pausing in one leaves the other alone: no shared resource ties the
        // two worlds' agent state together.
        assert!(a.pause(in_a));
        assert_eq!(a.agent_status(in_a), Some(AgentStatus::Paused));
        assert_ne!(b.agent_status(in_b), Some(AgentStatus::Paused));
    }

    #[tokio::test]
    async fn a_world_with_no_agents_does_not_answer_for_a_foreign_entity() {
        let mut a = build_world(ProviderRegistry::new());
        let b = build_world(ProviderRegistry::new());
        let in_a = spawn(&mut a);
        // `b` has spawned nothing, so the id names nothing there.
        assert!(b.agent_status(in_a).is_none());
    }

    /// `set_status` guards separately, and needs its own case.
    ///
    /// `pause`/`resume`/`cancel` read status first, so a foreign id stops at
    /// `agent_status` and never reaches the mutation. A caller holding a foreign
    /// id can still call `set_status` directly, which is the path this covers.
    #[tokio::test]
    async fn set_status_refuses_a_foreign_agent_id() {
        let mut a = build_world(ProviderRegistry::new());
        let mut b = build_world(ProviderRegistry::new());
        let in_a = spawn(&mut a);
        let in_b = spawn(&mut b);

        assert!(!b.set_status(in_a, AgentStatus::Complete), "B accepted it");
        // B's own agent, which shares the raw id, is untouched.
        assert_ne!(b.agent_status(in_b), Some(AgentStatus::Complete));
        // And B still works on its own.
        assert!(b.set_status(in_b, AgentStatus::Complete));
        assert_eq!(b.agent_status(in_b), Some(AgentStatus::Complete));
    }

    /// The world carries its own identity, so a *raw* `World` can check too.
    ///
    /// This is what lets the free functions called from inside systems -
    /// `force_transition`, `apply_context_transforms`,
    /// `restore_interaction_point` - refuse a foreign id. They are handed a
    /// `&mut World`, never a `PipelineWorld`, so without the resource there is
    /// nothing for them to compare against.
    #[tokio::test]
    async fn a_raw_world_refuses_an_id_another_world_minted() {
        let mut a = build_world(ProviderRegistry::new());
        let mut b = build_world(ProviderRegistry::new());
        let in_a = spawn(&mut a);
        let in_b = spawn(&mut b);

        // Resolving in its own world yields the entity...
        assert_eq!(in_a.resolve_in(a.world()), Some(in_a.entity()));
        // ...and in the other world, nothing - even though the raw id is valid
        // there and names one of B's own agents.
        assert_eq!(in_a.resolve_in(b.world()), None);
        assert_eq!(in_b.resolve_in(a.world()), None);

        // Round-tripping through the same world always works, which is what the
        // systems do with their query results.
        let round = AgentId::in_world(a.world(), in_a.entity());
        assert_eq!(round.resolve_in(a.world()), Some(in_a.entity()));
    }

    /// The free functions a system calls refuse a foreign id, and do nothing.
    ///
    /// Each takes a `&mut World` and would otherwise act on whichever local
    /// agent happened to share the raw entity: move it to another stage, seed it
    /// from a stranger's context, or park it on a prompt it never asked for.
    #[tokio::test]
    async fn the_world_taking_helpers_refuse_a_foreign_agent_id() {
        let mut a = build_world(ProviderRegistry::new());
        let mut b = build_world(ProviderRegistry::new());
        let in_a = spawn(&mut a);
        let in_b = spawn(&mut b);
        let before = b.agent_status(in_b);

        // Stage transition: B's agent must not move because A asked.
        let stage_before = b
            .world()
            .get::<crate::pipeline::StageCursor>(in_b.entity())
            .map(|c| c.index);
        crate::pipeline::force_transition(b.world_mut(), in_a, 1);
        let stage_after = b
            .world()
            .get::<crate::pipeline::StageCursor>(in_b.entity())
            .map(|c| c.index);
        assert_eq!(stage_before, stage_after, "a foreign id moved a stage");

        // Context seeding: nothing copied between worlds.
        crate::context_transform::apply_context_transforms(b.world_mut(), in_a, in_a);

        // A restored interaction point must not land on B's agent.
        crate::interaction_points::restore_interaction_point(
            b.world_mut(),
            in_a,
            crate::interaction_points::InteractionPointState {
                cursor: 0,
                round: 0,
                body: "not for you".to_string(),
            },
        );
        assert!(
            b.world()
                .get::<crate::components::AwaitingInteraction>(in_b.entity())
                .is_none(),
            "a foreign id parked B's agent on a prompt"
        );

        // And B's agent is exactly as it was.
        assert_eq!(b.agent_status(in_b), before);
    }

    /// The hazard [`AgentId`] exists for, now closed.
    ///
    /// The raw entities still collide - that is a property of bevy, not
    /// something this can change - but an [`AgentId`] carries the world that
    /// minted it, so the collision no longer means the two name the same agent.
    /// Before this, `b.pause(a_entity)` paused B's own agent while the caller
    /// believed it had paused A's, silently.
    #[tokio::test]
    async fn a_foreign_agent_id_is_refused_rather_than_naming_the_wrong_agent() {
        let mut a = build_world(ProviderRegistry::new());
        let mut b = build_world(ProviderRegistry::new());
        let in_a = spawn(&mut a);
        let in_b = spawn(&mut b);

        // The underlying ids do collide - the problem is real, not hypothetical.
        assert_eq!(
            in_a.entity(),
            in_b.entity(),
            "the raw ids collide, which is what made this silent"
        );
        // But the handles do not, because they remember where they came from.
        assert_ne!(in_a, in_b);
        assert_ne!(in_a.world(), in_b.world());

        // B refuses A's agent instead of acting on its own.
        assert!(!b.pause(in_a), "B accepted a foreign id");
        assert!(
            b.agent_status(in_a).is_none(),
            "B answered for a foreign id"
        );
        assert_ne!(b.agent_status(in_b), Some(AgentStatus::Paused));

        // Each world still works normally on its own.
        assert!(a.pause(in_a));
        assert_eq!(a.agent_status(in_a), Some(AgentStatus::Paused));
        assert!(b.pause(in_b));
        assert_eq!(b.agent_status(in_b), Some(AgentStatus::Paused));
    }
}