agent_engine/session/

actor.rs

//! `SessionActor` — owns THE `Runtime` + `ConversationState` for one
//! conversation and runs its turn machine (PLAN-phase2 §2.5).
//!
//! Every method body is moved from an existing reactor site (TUI
//! `dispatch.rs` Abort/Submit/StreamingInput, `stream_handler.rs`
//! event-queue + stream arms, `tui/mod.rs` teardown, `cmd/chat.rs`
//! post-turn); the presentation halves stay client-side and are fed by
//! the envelopes this actor emits. Each `StreamEvent` is forwarded to
//! clients BEFORE the actor acts on it, so a client sees the same order it
//! sees today.
//!
//! Invariants:
//! - `Runtime::clone()` resets TTL latches. The actor clones it only for
//!   driver stream-start (Prepared→Started) and compaction. The stream-start
//!   clone shares the original's TTL atomics via `share_ttl_latches` so a
//!   redundant 1h→5m downgrade notice per driver turn is avoided. The
//!   compaction clone is read-only+discarded and needs no sharing.
//! - `emit` is the ONLY `seq` increment site.
//! - `Detach` never touches `stream`/`cancel`; only `End` (and `Cancel`)
//!   stop a turn.

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

use futures::StreamExt;
use tokio::sync::{broadcast, mpsc, oneshot};

use crate::engine::reactor::{
    auto_turn_cap_reached, claim_auto_turn_with_cap, drain_event_queue, wake_action_with_cap,
    EventDisposition, WakeAction,
};
use crate::engine::session::ConversationState;
use crate::engine::setup::BackgroundTasks;
use crate::extensions::invoke_output::{invoke_event_channel, InvokeOutputBudget};
use crate::extensions::session_driver::{self as protocol, Grant, Reply};
use crate::runtime::compaction::{
    apply_compaction, compact_conversation, preview_compaction_disclosure, CompactionPolicy,
    CompactionTransition,
};
use crate::tools::{SecretPromptHandle, SecretPromptRequest};
use crate::{
    AgentEvent, CancellationToken, EngineHost, Result, Runtime, SessionEvent,
    StreamEvent,
};

use super::budgets;
use super::handle::{SessionEndpoints, SessionHandle};
use super::types::*;
use super::view::RuntimeView;

/// The in-flight response stream (tui/stream_handler.rs `ActiveStream`).
pub type ActiveStream = std::pin::Pin<Box<dyn futures::Stream<Item = StreamEvent> + Send>>;

/// `turn_replay` cap (envelopes). §6 #9: the 2 MiB text bound is day 3.
const TURN_REPLAY_CAP: usize = 4096;

/// The in-flight turn draft (`sessions/<id>.turn`, see
/// `agent_core::core::session_draft`): what the actor last knew about the
/// response being streamed, flushed at most once per 1 Hz turn tick.
#[derive(Default)]
pub(crate) struct TurnDraftState {
    /// Session id the open draft is persisted under; `None` = not persisted
    /// (no turn open, or this actor doesn't write drafts). `base_len` and
    /// `text` are tracked for every turn: a cancel whose drain times out
    /// keeps the streamed text (`SessionActor::keep_streamed_partial`).
    open: Option<String>,
    /// History length the in-flight response continues from.
    base_len: usize,
    /// Text streamed in the in-flight response so far.
    text: String,
    /// Tools the in-flight response has called (complete `ToolUse`s), which
    /// may be running. Not persisted.
    tools: Vec<String>,
    /// Changed since the last flush.
    dirty: bool,
}

/// What a started turn needs to open its stream (`start_turn`), held while
/// its request waits for `on_session_start`.
pub(crate) struct TurnStart {
    cancel: CancellationToken,
    steer_rx: mpsc::UnboundedReceiver<String>,
    auto_approve: bool,
    completion: crate::runtime::TurnCompletion,
    /// Events started this turn (nobody typed it): its stream runs under the
    /// event turn budget (#251). Decided from the turn's trigger at
    /// `start_turn`, applied when the stream opens (`open_turn_stream`),
    /// which may be after `on_session_start` has run.
    event_turn: bool,
}

/// Why autonomous work may not start now
/// ([`SessionActor::autonomous_work_blocked`]).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum WorkBlock {
    /// A reload checkpointed this session: nothing may start work the reload
    /// record does not carry until the exec, or until the daemon reports
    /// that it failed (#255's reload fence).
    Fenced,
    /// This session's `on_session_start` has not run yet: a request sent
    /// now would lack its injection.
    SessionStartPending,
}

/// A source of autonomous work in the actor's loop: work nobody asked for
/// with a command ([`SessionActor::holds`]).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum WorkSource {
    /// Queued events (inbox, `synaps send`, worker completions).
    QueueWake,
    /// The park grace or a worker re-check.
    ParkTimer,
    /// A due prompt-cache keep-warm ping.
    KeepWarmTimer,
    /// The armed session driver's 200 ms tick.
    DriverTick,
}

/// What `SessionActor::drain_cancelled_stream` recovered from a cancelled
/// turn's stream.
#[derive(Default)]
pub(crate) struct CancelDrain {
    /// The engine's final history for the turn (last `MessageHistory` seen).
    pub(crate) history: Option<Vec<crate::SharedMessage>>,
    /// The stream ended (`Done` / EOF) inside the drain budget.
    pub(crate) closed: bool,
}

/// A field that is `None` exactly while the session is `Parked` (B3).
/// Derefs to the value so the turn machine reads `self.runtime` / `self.conv`
/// unchanged; every command that needs them goes through `ensure_live()`
/// first, so the panic path is a programming error, not a runtime state.
pub(crate) struct Live<T>(Option<T>);

impl<T> Live<T> {
    pub(crate) fn new(v: T) -> Self {
        Self(Some(v))
    }
    pub(crate) fn is_live(&self) -> bool {
        self.0.is_some()
    }
    pub(crate) fn park_take(&mut self) -> Option<T> {
        self.0.take()
    }
    pub(crate) fn unpark_set(&mut self, v: T) {
        self.0 = Some(v);
    }
}

impl<T> std::ops::Deref for Live<T> {
    type Target = T;
    fn deref(&self) -> &T {
        self.0.as_ref().expect("session is parked: runtime/conv unavailable")
    }
}

impl<T> std::ops::DerefMut for Live<T> {
    fn deref_mut(&mut self) -> &mut T {
        self.0.as_mut().expect("session is parked: runtime/conv unavailable")
    }
}

/// `SYNAPS_DAEMON_PARKED_EVICT_SECS`: how long a PARKED session stays in the
/// daemon's map before it is evicted (`EndReason::Evicted`). Its state is on
/// disk; `--continue <id>` rebuilds it exactly as after a daemon restart. The
/// row only exists so `--attach <id>` / the adopt banner can find it quickly.
/// Default 1 h; `never` → keep forever.
pub fn parked_evict_after() -> Option<std::time::Duration> {
    // Precedence: env > config (`daemon.parked_evict_secs`) > builtin default.
    let cfg_secs = crate::config::load_config().daemon.parked_evict_secs;
    parked_evict_after_from(
        std::env::var("SYNAPS_DAEMON_PARKED_EVICT_SECS").ok().as_deref(),
        cfg_secs,
    )
}

fn parked_evict_after_from(v: Option<&str>, cfg_secs: u64) -> Option<std::time::Duration> {
    let from_cfg = || (cfg_secs != 0).then(|| std::time::Duration::from_secs(cfg_secs));
    match v.map(str::trim) {
        Some("never" | "0" | "off") => None,
        Some(n) => n
            .parse::<u64>()
            .ok()
            .map(std::time::Duration::from_secs)
            .or_else(from_cfg),
        None => from_cfg(),
    }
}

/// `SYNAPS_DAEMON_IDLE_END_GRACE_SECS`: how long a ZERO-turn session with no
/// clients lingers before it ends (F18). Default 5 s — long enough for a
/// client that disconnected mid-reconnect, short enough that open-and-close
/// does not pin a Runtime for a minute.
pub fn idle_end_grace() -> std::time::Duration {
    std::env::var("SYNAPS_DAEMON_IDLE_END_GRACE_SECS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .map(std::time::Duration::from_secs)
        .unwrap_or(std::time::Duration::from_secs(5))
}

/// `SYNAPS_DAEMON_PARK_GRACE_SECS`: `never` → `None` (Parked disabled);
/// `n` → n seconds after the last detach once idle; default 60.
pub fn park_grace() -> Option<std::time::Duration> {
    match std::env::var("SYNAPS_DAEMON_PARK_GRACE_SECS") {
        Ok(v) if v.trim().eq_ignore_ascii_case("never") => None,
        Ok(v) => v
            .trim()
            .parse::<u64>()
            .ok()
            .map(std::time::Duration::from_secs)
            .or(Some(DEFAULT_PARK_GRACE)),
        Err(_) => Some(DEFAULT_PARK_GRACE),
    }
}

pub const DEFAULT_PARK_GRACE: std::time::Duration = std::time::Duration::from_secs(60);

/// How often a detached idle session re-checks a park its workers hold
/// (`can_park_ignoring_keepwarm`).
pub const WORKER_PARK_RECHECK: std::time::Duration = std::time::Duration::from_secs(30);

#[cfg(any(test, feature = "testing"))]
static WORKER_PARK_RECHECK_OVERRIDE_MS: std::sync::atomic::AtomicU64 =
    std::sync::atomic::AtomicU64::new(0);

/// [`WORKER_PARK_RECHECK`], or the test override. Code-only, like
/// `cache_keepwarm::interval_override`: no binary input can reach it.
fn worker_park_recheck() -> std::time::Duration {
    #[cfg(any(test, feature = "testing"))]
    {
        let ms = WORKER_PARK_RECHECK_OVERRIDE_MS.load(std::sync::atomic::Ordering::Relaxed);
        if ms > 0 {
            return std::time::Duration::from_millis(ms);
        }
    }
    WORKER_PARK_RECHECK
}

/// Test-only controls (`cfg(test)` / `testing` feature).
#[cfg(any(test, feature = "testing"))]
pub mod testing {
    use std::sync::atomic::Ordering;
    use std::time::Duration;

    /// Replaces the worker park re-check interval for every session in
    /// this process while alive (tests that use it must be `#[serial]`).
    #[must_use = "the override is cleared when the guard drops"]
    pub struct WorkerParkRecheck(());

    impl WorkerParkRecheck {
        pub fn set(interval: Duration) -> Self {
            let ms = u64::try_from(interval.as_millis()).unwrap_or(u64::MAX).max(1);
            super::WORKER_PARK_RECHECK_OVERRIDE_MS.store(ms, Ordering::Relaxed);
            Self(())
        }
    }

    impl Drop for WorkerParkRecheck {
        fn drop(&mut self) {
            super::WORKER_PARK_RECHECK_OVERRIDE_MS.store(0, Ordering::Relaxed);
        }
    }
}

/// `SYNAPS_DAEMON_PROMPT_ABANDON_SECS`: how long a pending host confirmation on
/// a NON-driver session survives after the LAST client detaches before it is
/// fail-closed answered `None` (deny) so `can_park()` can proceed. This bounds
/// the resource pin from an ABANDONED prompt WITHOUT punishing a transient
/// disconnect (SSH drop, sleep, wifi): a re-attach before expiry cancels it and
/// the user answers the prompt normally. Default 1 h — generous enough that a
/// human coming back from lunch still answers; `never`/`0`/`off` → `None` =
/// disabled = pure pre-#112 behaviour (a prompt survives detach forever).
///
/// (The driver-armed case fail-closes immediately on last detach; this deadline
/// is only for a plain interactive session — see `detach` / `rearm_prompt_abandon`.)
pub fn prompt_abandon_timeout() -> Option<std::time::Duration> {
    // Precedence: env > config (`daemon.prompt_abandon_secs`) > builtin default.
    let cfg_secs = crate::config::load_config().daemon.prompt_abandon_secs;
    prompt_abandon_timeout_from(
        std::env::var("SYNAPS_DAEMON_PROMPT_ABANDON_SECS").ok().as_deref(),
        cfg_secs,
    )
}

fn prompt_abandon_timeout_from(v: Option<&str>, cfg_secs: u64) -> Option<std::time::Duration> {
    let from_cfg = || (cfg_secs != 0).then(|| std::time::Duration::from_secs(cfg_secs));
    match v.map(str::trim) {
        Some("never" | "0" | "off") => None,
        Some(n) => n
            .parse::<u64>()
            .ok()
            .map(std::time::Duration::from_secs)
            .or_else(from_cfg),
        None => from_cfg(),
    }
}

/// Non-persisted runtime knobs replayed after unpark (last-wins per
/// variant). Model/reasoning live in the journal already; `/system` is
/// runtime-only (never journaled) so it replays too (M7).
fn replayable(setting: &SessionSetting) -> bool {
    matches!(
        setting,
        SessionSetting::SystemPrompt { .. }
            | SessionSetting::ContextWindow { .. }
            | SessionSetting::CompactionModel { .. }
            | SessionSetting::ApiRetries { .. }
            | SessionSetting::SubagentTimeout { .. }
            | SessionSetting::MaxToolOutput { .. }
            | SessionSetting::BashTimeout { .. }
            | SessionSetting::BashMaxTimeout { .. }
            | SessionSetting::GrantWorkerModel { .. }
    )
}

fn replay_setting(rt: &mut Runtime, setting: &SessionSetting) {
    match setting {
        SessionSetting::SystemPrompt { text } => rt.set_system_prompt(text.clone()),
        SessionSetting::ContextWindow { tokens } => rt.set_context_window(*tokens),
        SessionSetting::CompactionModel { model } => rt.set_compaction_model(model.clone()),
        SessionSetting::ApiRetries { n } => rt.set_api_retries(*n),
        SessionSetting::SubagentTimeout { secs } => rt.set_subagent_timeout(*secs),
        SessionSetting::MaxToolOutput { bytes } => rt.set_max_tool_output(*bytes),
        SessionSetting::BashTimeout { secs } => rt.set_bash_timeout(*secs),
        SessionSetting::BashMaxTimeout { secs } => rt.set_bash_max_timeout(*secs),
        SessionSetting::GrantWorkerModel { model } => {
            let _ = rt.grant_worker_model(model);
        }
        _ => {}
    }
}

/// One attached client: its meta + the mode it attached with (B1 ownership
/// needs the mode to pick the next owner when the owner detaches).
pub(crate) struct AttachedClient {
    pub(crate) meta: ClientMeta,
    pub(crate) mode: AttachMode,
    /// Resolved at attach ([`resolve_client_name`]); unique in the session.
    pub(crate) name: String,
    /// Unix ms.
    pub(crate) attached_at_ms: u64,
    /// `SessionActor::input_seq` at this client's last input; 0 = none yet.
    pub(crate) last_input: u64,
}

/// B2: a spawned `compact_conversation` (§2.5).
pub(crate) struct CompactionJob {
    pub(crate) task: tokio::task::JoinHandle<Result<crate::runtime::compaction::CompactionOutcome>>,
    pub(crate) source: String,
    #[allow(dead_code)]
    pub(crate) started: tokio::time::Instant,
}

/// Commands that only the input owner may send (B1). Everything else
/// (`Answer`, `Query`, `Save`, `Detach`, `Attach`, `Resync`, `HostEvent`)
/// is open to every attached client — except that `answer` refuses an
/// observer's answer to a Confirm prompt and accepts "Allow all" only from
/// the owner (#148 C2). `Checkpoint` cancels the owner's turn
/// and kills its PTYs, so it is owner-only from a client (M1); the daemon
/// sends it `from: None`.
fn is_input_command(cmd: &SessionCommand) -> bool {
    matches!(
        cmd,
        SessionCommand::Checkpoint { .. }
            | SessionCommand::Submit { .. }
            | SessionCommand::SubmitPrepared { .. }
            | SessionCommand::Steer { .. }
            | SessionCommand::SubagentSteer { .. }
            | SessionCommand::Cancel
            | SessionCommand::Set { .. }
            | SessionCommand::Compact { .. }
            | SessionCommand::NewSession
            | SessionCommand::EngineCommand { .. }
            | SessionCommand::PluginCommand { .. }
            | SessionCommand::Resume { .. }
            | SessionCommand::KeepWarm { .. }
            | SessionCommand::DriverStart { .. }
            | SessionCommand::End {
                reason: EndReason::ClientQuit
            }
    )
}

/// The MAIN-conversation note for a user steer of a worker (`SubagentSteer`):
/// a low-severity event, formatted for the model like any drained event.
fn subagent_steer_event(handle_id: &str, agent_name: &str, text: &str) -> crate::events::types::Event {
    use crate::events::types::{Event, EventContent, EventSource, Severity};
    Event {
        id: uuid::Uuid::new_v4().to_string(),
        timestamp: chrono::Utc::now(),
        source: EventSource {
            source_type: "user".to_string(),
            name: "user".to_string(),
            callback: None,
        },
        channel: None,
        sender: None,
        content: EventContent {
            text: format!("User steered subagent '{agent_name}' ({handle_id}) directly: {text}"),
            content_type: "subagent_steer".to_string(),
            severity: Some(Severity::Low),
            data: Some(serde_json::json!({ "handle_id": handle_id, "agent_name": agent_name })),
            disclosure: None,
        },
        expects_response: false,
        reply_to: None,
    }
}

/// The MAIN-conversation note for a worker the user cancelled with
/// `/workers cancel`: its completion wake is withheld, so without this the
/// main agent would wait for it (or dispatch the work again).
pub(crate) fn subagent_user_cancel_event(
    handle_id: &str,
    agent_name: &str,
) -> crate::events::types::Event {
    use crate::events::types::{Event, EventContent, EventSource, Severity};
    Event {
        id: uuid::Uuid::new_v4().to_string(),
        timestamp: chrono::Utc::now(),
        source: EventSource {
            source_type: "user".to_string(),
            name: "user".to_string(),
            callback: None,
        },
        channel: None,
        sender: None,
        content: EventContent {
            text: format!(
                "User cancelled subagent '{agent_name}' ({handle_id}) with /workers cancel. It stops at \
                 its next step and sends no completion wake; do not wait for it. Its partial output is \
                 readable with subagent_collect (handle_id \"{handle_id}\") only while this session \
                 stays loaded: the handle does not survive the session being parked."
            ),
            content_type: "subagent_cancelled".to_string(),
            severity: Some(Severity::Low),
            data: Some(serde_json::json!({ "handle_id": handle_id, "agent_name": agent_name })),
            disclosure: None,
        },
        expects_response: false,
        reply_to: None,
    }
}

/// The MAIN-conversation note for a worker `/close force` cancelled. The
/// session ends right after, so this is for whoever continues it later
/// (`--continue`): the worker never finished and no wake is coming.
pub(crate) fn subagent_close_cancel_event(
    handle_id: &str,
    agent_name: &str,
) -> crate::events::types::Event {
    let mut event = subagent_user_cancel_event(handle_id, agent_name);
    event.content.text = format!(
        "User closed this session with /close force while subagent '{agent_name}' ({handle_id}) was \
         running, so it was cancelled before it finished. It sends no completion wake; do not wait for \
         it. If its work is still needed, check what it left behind and dispatch it again."
    );
    event
}

/// `KeepWarm{on: true}`'s notice (`/pin on`, `daemon pin`, `--keep-warm`).
pub const PIN_ON_NOTICE: &str = "pinned: this session stays loaded (never parked or idle-ended, \
     and the daemon will not idle-exit while it is pinned) and comes back after daemon restarts. \
     /pin off to undo.";
/// `KeepWarm{on: false}`'s notice.
pub const PIN_OFF_NOTICE: &str = "unpinned: this session parks again when idle and won't come back after a restart";

/// The engine command behind `/close` (`SessionActor::close_command`).
pub(crate) const CLOSE_COMMAND: &str = "close";

fn is_close_command(cmd: &SessionCommand) -> bool {
    matches!(cmd, SessionCommand::EngineCommand { name, .. } if name == CLOSE_COMMAND)
}

fn command_name(cmd: &SessionCommand) -> &'static str {
    match cmd {
        SessionCommand::Submit { .. } => "submit",
        SessionCommand::SubmitPrepared { .. } => "submit_prepared",
        SessionCommand::Steer { .. } => "steer",
        SessionCommand::SubagentSteer { .. } => "subagent_steer",
        SessionCommand::Cancel => "cancel",
        SessionCommand::Answer { .. } => "answer",
        SessionCommand::Set { .. } => "set",
        SessionCommand::Compact { .. } => "compact",
        SessionCommand::NewSession => "new_session",
        SessionCommand::Save => "save",
        SessionCommand::Query { .. } => "query",
        SessionCommand::Attach { .. } => "attach",
        SessionCommand::Detach { .. } => "detach",
        SessionCommand::End { .. } => "end",
        SessionCommand::Resync { .. } => "resync",
        SessionCommand::EngineCommand { name, .. } if name == CLOSE_COMMAND => "close",
        SessionCommand::EngineCommand { .. } => "engine_command",
        SessionCommand::PluginCommand { .. } => "plugin_command",
        SessionCommand::Resume { .. } => "resume",
        SessionCommand::Checkpoint { .. } => "checkpoint",
        SessionCommand::KeepWarm { .. } => "keep_warm",
        SessionCommand::TakeInput => "take_input",
        SessionCommand::Park => "park",
        SessionCommand::AdoptWorkers { .. } => "adopt_workers",
        SessionCommand::RestoreToolActivations { .. } => "restore_tool_activations",
        SessionCommand::ReloadAborted => "reload_aborted",
        SessionCommand::ArmPark => "arm_park",
        SessionCommand::HostEvent(_) => "host_event",
        SessionCommand::DriverStart { .. } => "driver_start",
    }
}

pub struct SessionActor {
    pub(crate) id: SessionId,
    pub(crate) meta: SessionMeta,
    pub(crate) config: SessionConfig,
    /// THE runtime; `session_id` + `cwd` set by `create`. `None` ⇔ Parked.
    pub(crate) runtime: Live<Runtime>,
    /// `None` ⇔ Parked.
    pub(crate) conv: Live<ConversationState>,
    /// B3: unpark needs a fresh `foreground_runtime()`.
    pub(crate) host: Arc<EngineHost>,
    /// Kept across Park (`finish()` needs it without a Runtime).
    pub(crate) hook_bus: Arc<crate::extensions::hooks::HookBus>,
    /// Session-lifetime queue; `background` pushes into it, every Runtime
    /// this actor builds is pointed at it (`Runtime::set_event_queue`).
    pub(crate) event_queue: Arc<crate::events::EventQueue>,
    /// Last-wins per variant; replayed after unpark (non-persisted knobs).
    pub(crate) settings_replay: Vec<SessionSetting>,
    pub(crate) keep_warm: bool,
    /// Opt-in prompt-cache keep-warm pings while idle (task #444). Unrelated
    /// to `keep_warm` above, which only stops the session from parking.
    pub(crate) cache_keepwarm: super::cache_keepwarm::KeepWarmState,
    /// Armed on last-detach-while-idle / idle-while-detached.
    pub(crate) park_deadline: Option<tokio::time::Instant>,
    /// `park_deadline` when it was armed as a worker re-check (workers held
    /// the park), not as the reconnect grace: a re-check that finds the
    /// session parkable arms the grace instead of parking.
    pub(crate) park_recheck_at: Option<tokio::time::Instant>,
    /// (P11) Armed when the last client detaches while a pending host prompt is
    /// held on a NON-driver session; a re-attach cancels it, expiry denies the
    /// prompt(s) so `can_park()` can proceed. `None` = no abandoned prompt (or
    /// the deadline is disabled by config).
    pub(crate) prompt_abandon_deadline: Option<tokio::time::Instant>,
    // ── turn machine: the run() loop locals + App fields ──
    pub(crate) stream: Option<ActiveStream>,
    pub(crate) cancel: Option<CancellationToken>,
    pub(crate) steer_tx: Option<mpsc::UnboundedSender<String>>,
    /// An event-started turn's way onto the session budget: `steer` sends
    /// the owner's message through it, and the turn's stream switches to the
    /// session budget when it delivers that message. `None` for every other
    /// turn (they already run on the session budget).
    pub(crate) owner_steer_budget: Option<Arc<crate::runtime::budget::OwnerSteerBudget>>,
    pub(crate) streaming: bool,
    pub(crate) turn_baseline: usize,
    pub(crate) consecutive_auto_turns: u32,
    /// Formatted events `on_queue_wake` steered into this turn's stream
    /// that the engine has not acknowledged (`SteeringDelivered`) yet. On
    /// cancel they would die with the steering channel; `cancel_turn` moves
    /// them into history after the interruption marker instead.
    pub(crate) turn_steered_events: Vec<String>,
    /// Set by the engine when the running turn reaches its normal end
    /// (`TurnCompletion`); read when a cancel races that end.
    pub(crate) turn_completion: Option<crate::runtime::TurnCompletion>,
    /// A turn cut from outside `cancel_turn` (driver revocation cancels the
    /// driver turn's token): the stream is given until this instant to
    /// deliver its terminal event, then dropped and the turn finished as
    /// interrupted (`finish_revoked_turn`).
    pub(crate) revoked_turn_deadline: Option<tokio::time::Instant>,
    /// In-flight turn draft for crash recovery (persisting sessions only).
    pub(crate) turn_draft: TurnDraftState,
    /// Session saves and turn-draft writes, off the turn machine
    /// (`session::persister`).
    pub(crate) persister: super::persister::Persister,
    /// The snapshot last handed to the persister: an identical one is not
    /// queued again (a turn's end asks several times for the same state).
    pub(crate) last_queued_save: Option<super::persister::SnapshotKey>,
    // ── prompts ──
    pub(crate) secret_prompt_handle: SecretPromptHandle,
    pub(crate) secret_prompt_rx: mpsc::UnboundedReceiver<SecretPromptRequest>,
    /// Held across detach; replayed in `AttachSnapshot.pending_prompts`.
    pub(crate) pending_prompts: VecDeque<(PromptRequest, oneshot::Sender<Option<String>>)>,
    pub(crate) next_prompt_id: u64,
    // ── clients ──
    pub(crate) cmd_rx: mpsc::Receiver<Addressed>,
    pub(crate) events: broadcast::Sender<Envelope>,
    pub(crate) view: Arc<arc_swap::ArcSwap<RuntimeView>>,
    pub(crate) attached: HashMap<ClientId, AttachedClient>,
    /// B1: the one client whose input commands are honoured (`None` = any
    /// host-originated sender only). With `client_input = shared` it is the
    /// *active* client: the drive client that sent input last.
    pub(crate) input_owner: Option<ClientId>,
    /// Who may drive (config `client_input`, or the create override).
    pub(crate) client_input: ClientInput,
    /// Bumped on each input from a client ([`AttachedClient::last_input`]).
    pub(crate) input_seq: u64,
    pub(crate) next_client_id: u64,
    pub(crate) seq: u64,
    pub(crate) turn_replay: VecDeque<Envelope>,
    /// Last Codex hosted-web_search state line shown (once per session per
    /// model/state; see `codex_wire::session_web_search_notice`).
    pub(crate) web_search_notice: Option<String>,
    pub(crate) state: AttachState,
    pub(crate) background: BackgroundTasks,
    /// B2: spawned compaction in flight (`busy` while `Some`).
    pub(crate) compact: Option<CompactionJob>,
    /// `compact_inline` is running (it blocks the actor; listings still
    /// see the session busy).
    pub(crate) compacting_inline: bool,
    /// `on_session_start` for this session. `await_extensions=false`: a
    /// task waits for process-level discovery and its turn at the host's
    /// gate (one hook at a time), then calls it, so boot never blocks on
    /// it. A turn that comes first moves it ahead of background hooks and
    /// waits for it without blocking the loop (`turn_awaiting_start`); a
    /// keep-warm ping waits with its timer off (`holds`).
    pub(crate) session_start: super::session_start::SessionStart,
    /// A started turn whose request waits for `on_session_start`: its
    /// stream opens when the hook has run. The turn is live meanwhile
    /// (`streaming`), so Cancel, End and a driver revocation end it as any
    /// turn, with nothing sent.
    pub(crate) turn_awaiting_start: Option<TurnStart>,
    /// 1 Hz `SubagentRows` cadence while a turn runs (`None` when idle).
    pub(crate) subagent_tick: Option<tokio::time::Interval>,
    /// Worker-registry generation of the current/last turn: a user
    /// interrupt cancels only the workers this turn launched.
    pub(crate) turn_worker_generation: u64,
    /// Mirrors `handle.lifecycle()` (B3 writes Parking/Parked).
    pub(crate) lifecycle: Arc<std::sync::atomic::AtomicU8>,
    /// Mirrors `handle.journal_id()` (B2 stores the successor id).
    pub(crate) journal_id: Arc<arc_swap::ArcSwap<String>>,
    /// Mirrors `handle.presence()` (clients / owner / pending prompts).
    pub(crate) presence: Arc<arc_swap::ArcSwap<super::handle::Presence>>,
    /// Mirrors `handle.name()`; `sync_name` after any rename.
    pub(crate) name: Arc<arc_swap::ArcSwap<Option<String>>>,
    /// Mirrors `handle.pinned()`; written with `keep_warm` (create seeds it
    /// from `meta.pinned`, `KeepWarm{on}` stores it).
    pub(crate) keep_warm_cell: Arc<std::sync::atomic::AtomicBool>,
    /// F10: per-session journal ownership lock. Held while Live, released on Park.
    pub(crate) session_lock: Option<agent_core::session_lock::SessionLock>,
    // ── E: session driver (actor-side, behind `session.drive` permission) ──
    pub(crate) driver: Option<super::driver::DriverState>,
    pub(crate) driver_pending: Option<super::driver::DriverPending>,
    /// Generation counter for invalidating stale pending results.
    pub(crate) driver_generation: u64,
    /// Interrupted owner (for generic stop commands across revocation).
    pub(crate) driver_interrupted_owner: Option<String>,
    /// 200 ms cadence for `driver_tick()`, always present.
    pub(crate) driver_tick_interval: tokio::time::Interval,
    /// Set by a reload checkpoint until the `execv` (or until the daemon
    /// reports it failed): see [`ReloadFence`].
    pub(crate) reload_fence: Option<ReloadFence>,
    /// Shared with the handle: `config.memory_mode`, readable by a reload
    /// that gets no checkpoint reply.
    pub(crate) memory_mode_slot: super::handle::MemoryModeSlot,
    /// This runtime's memory has been seen active (a grant took effect).
    /// Only then does "memory is off" mean consent was revoked; see
    /// `effective_memory_mode`. Reset whenever the runtime is rebuilt.
    pub(crate) memory_seen_active: bool,
    /// While parked: the exact tool activations the dropped runtime held,
    /// handed to the runtime the unpark builds (and to a reload record
    /// written while parked). Empty while live.
    pub(crate) parked_tool_activations: Vec<crate::tools::activation::CarriedActivation>,
}

/// A reload checkpoint wrote this session's reload record; until the
/// `execv` replaces the process nothing may start new work the record does
/// not know about. Queue wakes stay queued (the record carries the wakes it
/// owes), and keep-warm pings, driver ticks and parking wait. Lifted by
/// `ReloadAborted` (the exec failed) or, as a backstop, at `deadline`.
pub(crate) struct ReloadFence {
    pub(crate) deadline: tokio::time::Instant,
    /// What the checkpoint recorded: re-announced to this image when the
    /// exec does not happen.
    pub(crate) workers: Vec<crate::runtime::subagent::CarriedWorker>,
}

/// A create that fails after taking the journal lock: a FRESH session never
/// journaled, so its `.lock` would be litter next to no `.json` (F18). A
/// continued session's lock file sits next to its journal and stays.
fn release_fresh_lock(lock: Option<agent_core::session_lock::SessionLock>, continued: bool) {
    if let (Some(lock), false) = (lock, continued) {
        lock.release_and_remove();
    }
}

impl SessionActor {
    /// = today's `setup::boot()` per session (foreground_runtime →
    /// resolve_session_and_prompt → journal lock → set_cwd → model override →
    /// spawn_session_background → finish_session_setup) then
    /// `on_session_start` (keyed injection).
    pub(crate) async fn create(
        host: &Arc<EngineHost>,
        mut cfg: SessionConfig,
    ) -> Result<(SessionHandle, SessionTask)> {
        // cwd goes in BEFORE apply_config (inside foreground_runtime_for) so the
        // one-shot memory binding scopes to the session's project (§4).
        // A session that made its own `/memory` choice (carried by a reload)
        // gets that, not the operator default.
        let mut runtime = host
            .foreground_runtime_with(cfg.cwd.clone(), cfg.memory_mode.is_none())
            .await?;
        if let Some(mode) = cfg.memory_mode {
            runtime.restore_session_memory_mode(mode);
        }
        let config: crate::SynapsConfig = (**host.config()).clone();
        let client_input = cfg.client_input.unwrap_or(config.client_input);

        let mut sb = crate::engine::setup::resolve_session_and_prompt(
            &mut runtime,
            &cfg.continue_session,
            cfg.system.as_deref(),
            cfg.prompt_manifest.as_deref(),
        )?;
        // F10: acquire the per-session journal ownership lock BEFORE anything
        // publishes or writes for this id: the `--name` save, the registry
        // card + per-session socket (`spawn_session_background`), the START
        // index record and `on_session_start`. A refused `--continue` of a
        // session live in another process used to publish its own card over
        // the owner's first, then delete it on drop: the owner stayed live
        // (socket up, journal locked) but `synaps send` and peers lost it.
        let lock_kind = if cfg.await_extensions { "daemon" } else { "tui" };
        let session_lock = if sb.continued {
            // Continuing an existing session — the lock MUST succeed.
            // If another process holds it, refuse with an actionable error.
            let dir = agent_core::session_lock::sessions_dir();
            let holder = agent_core::session_lock::LockHolder {
                pid: std::process::id(),
                kind: lock_kind.to_string(),
            };
            match agent_core::session_lock::SessionLock::try_acquire(&dir, &sb.session.id, holder) {
                Ok(lock) => Some(lock),
                Err(agent_core::session_lock::SessionLockError::Held { session_id, holder }) => {
                    let msg = agent_core::session_lock::SessionLockError::Held { session_id, holder };
                    return Err(crate::RuntimeError::Session(msg.to_string()));
                }
                Err(e) => {
                    tracing::warn!(session = %sb.session.id, "session lock: {e}");
                    None
                }
            }
        } else {
            // Fresh session — best-effort lock.
            let dir = agent_core::session_lock::sessions_dir();
            let holder = agent_core::session_lock::LockHolder {
                pid: std::process::id(),
                kind: lock_kind.to_string(),
            };
            match agent_core::session_lock::SessionLock::try_acquire(&dir, &sb.session.id, holder) {
                Ok(lock) => Some(lock),
                Err(e) => {
                    tracing::warn!(session = %sb.session.id, "session lock: {e}");
                    None
                }
            }
        };

        // `--name` at create: apply BEFORE the registry entry is written so
        // `synaps send --session <name>` and `--continue <name>` resolve
        // from the first second — not only after a later `--continue`.
        if let Some(name) = cfg.name.as_deref().map(str::trim).filter(|n| !n.is_empty()) {
            if let Err(e) = sb.session.set_name(name) {
                release_fresh_lock(session_lock, sb.continued);
                return Err(crate::RuntimeError::Session(format!("--name {name:?}: {e}")));
            }
            if cfg.persist {
                // Persist the name now so disk resolution (chain/name → id)
                // agrees with the live map even before the first turn.
                if let Err(e) = sb.session.save().await {
                    tracing::warn!(session = %sb.session.id, "failed to save named session at create: {e}");
                }
            }
        }
        runtime.set_cwd(cfg.cwd.clone());
        // T5: rehydrate env from journal when the client sent none.
        // Precedence: explicit Hello.env from the continuing client > journal > None.
        // Never fall back to the daemon's own process env.
        if cfg.env.is_none() {
            if let Some(ref journal_env) = sb.session.env {
                cfg.env = Some(journal_env.clone());
            }
            if cfg.env_stripped.is_empty() && !sb.session.env_stripped.is_empty() {
                cfg.env_stripped = sb.session.env_stripped.clone();
            }
        }
        // Always persist the latest env on the session so the next
        // restart/continue sees it — a fresh client env supersedes
        // whatever was journaled previously.
        if cfg.env.is_some() {
            sb.session.env = cfg.env.clone();
            sb.session.env_stripped = cfg.env_stripped.clone();
        }
        runtime.set_env(cfg.env.clone());
        runtime.set_env_stripped(cfg.env_stripped.clone());
        // CLI `--model` overrides whatever was persisted (rpc.rs precedent).
        if let Some(ref m) = cfg.model_override {
            runtime.set_model(m.clone());
        }

        // Shared-inbox files are for whoever is at a keyboard: the gate stays
        // shut until a client owns input (`publish_presence`), so a detached,
        // parked or freshly rehydrated session is never woken by them.
        let background = match crate::engine::setup::spawn_session_background_gated(
            &runtime,
            &sb.session,
            Arc::new(std::sync::atomic::AtomicBool::new(false)),
        ) {
            Ok(background) => background,
            Err(e) => {
                release_fresh_lock(session_lock, sb.continued);
                return Err(e);
            }
        };
        crate::engine::setup::finish_session_setup(
            &mut runtime,
            &config,
            &sb.session,
            cfg.cwd.clone(),
            crate::engine::setup::IndexRecord::Start,
        );

        // C2: per-session on_session_start (keyed injection) once the
        // process-level discovery is known-finished. Never re-runs discovery.
        // Every call goes through the host's session-start gate.
        // `await_extensions=false` (TUI, a reload's restore): a spawned task
        // runs it (`session_start` below), so boot never blocks on it.
        // `startup.extensions_ready_timeout_secs` (config, default 30) bounds
        // the wait on extension discovery only; the const remains the
        // fallback. `run_at_create` then waits its turn at the gate, behind a
        // running hook and older foreground ones (see `session_start`).
        let ext_ready_timeout = extensions_ready_timeout(&config);
        if cfg.await_extensions {
            if tokio::time::timeout(ext_ready_timeout, host.extensions_ready())
                .await
                .is_err()
            {
                tracing::warn!(
                    budget_secs = ext_ready_timeout.as_secs(),
                    "extensions_ready timed out — on_session_start may miss late extensions"
                );
            }
            super::session_start::run_at_create(host, runtime.hook_bus(), &sb.session.id).await;
        }

        let id = SessionId::from(sb.session.id.clone());

        let meta = SessionMeta {
            id: id.clone(),
            name: sb.session.name.clone(),
            model: runtime.model().to_string(),
            cwd: cfg.cwd.clone(),
            created_at: sb.session.created_at,
            continued: sb.continued,
            continue_info: sb.continue_info.as_ref().map(ContinueInfoWire::from),
            host_pid: std::process::id(),
            lifecycle: SessionLifecycle::Live,
            clients: 0,
            input_owner: None,
            awaiting_input: 0,
            roster: Vec::new(),
            client_input: Some(client_input),
            journal_id: sb.session.id.clone(),
            locked_by: None,
            pinned: cfg.keep_warm,
            busy: false,
            state_since: None,
            park_due: None,
            evict_due: None,
            title: None,
            cost_usd: 0.0,
            messages: 0,
        };
        let mut conv = if sb.continued {
            ConversationState::from_resumed(sb.session)
        } else {
            ConversationState::new(sb.session)
        };
        conv.api_messages = sb.api_messages;
        conv.total_input_tokens = sb.total_input_tokens;
        conv.total_output_tokens = sb.total_output_tokens;
        conv.session_cost = sb.session_cost;
        // Crash recovery (the last process died mid-turn): only the lock
        // holder may fold the turn draft in, persist it and remove it.
        if sb.continued && cfg.persist && session_lock.is_some() {
            crate::engine::setup::recover_turn_draft(&mut conv).await;
        }

        let view = RuntimeView::from_runtime(&runtime).await;
        let hook_bus = Arc::clone(runtime.hook_bus());
        let event_queue = Arc::clone(runtime.event_queue());
        let keep_warm = cfg.keep_warm;
        let (
            handle,
            SessionEndpoints {
                cmd_rx,
                events,
                view,
                lifecycle,
                journal_id,
                presence,
                name,
                keep_warm: keep_warm_cell,
                cache_keepwarm,
                memory_mode,
            },
        ) = SessionHandle::new(meta.clone(), view);
        // The mode is fixed for the session: seed the listing with it now.
        // Otherwise only a presence publish (attach, detach, input) carries
        // it, so a session restored at reload, with no client back yet,
        // listed as `owner=-`. Not `publish_presence()`: that also sets the
        // inbox gate, which stays as `SessionHandle::new` left it.
        presence.store(Arc::new(super::handle::Presence {
            client_input: Some(client_input),
            ..Default::default()
        }));
        *memory_mode.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = cfg.memory_mode;
        let (sp_tx, secret_prompt_rx) = mpsc::unbounded_channel();
        let session_start = if cfg.await_extensions {
            super::session_start::SessionStart::Done
        } else if cfg.restore_parked {
            // Parked right after create: the hook waits for the unpark.
            super::session_start::SessionStart::Pending
        } else {
            super::session_start::SessionStart::queue(
                host,
                Arc::clone(&hook_bus),
                Arc::clone(&journal_id),
                ext_ready_timeout,
            )
        };

        let mut actor = SessionActor {
            id,
            meta,
            config: cfg,
            runtime: Live::new(runtime),
            conv: Live::new(conv),
            host: Arc::clone(host),
            hook_bus,
            event_queue,
            settings_replay: Vec::new(),
            keep_warm,
            cache_keepwarm: super::cache_keepwarm::KeepWarmState::new(
                config.cache_keepwarm,
                config.cache_keepwarm_auto_max_hours,
                cache_keepwarm,
            ),
            park_deadline: None,
            park_recheck_at: None,
            prompt_abandon_deadline: None,
            stream: None,
            cancel: None,
            steer_tx: None,
            owner_steer_budget: None,
            streaming: false,
            turn_baseline: 0,
            consecutive_auto_turns: 0,
            turn_steered_events: Vec::new(),
            turn_completion: None,
            revoked_turn_deadline: None,
            turn_draft: TurnDraftState::default(),
            persister: super::persister::Persister::new(agent_core::session_lock::sessions_dir()),
            last_queued_save: None,
            secret_prompt_handle: SecretPromptHandle::new(sp_tx),
            secret_prompt_rx,
            pending_prompts: VecDeque::new(),
            next_prompt_id: 1,
            cmd_rx,
            events,
            view,
            attached: HashMap::new(),
            input_owner: None,
            client_input,
            input_seq: 0,
            next_client_id: 1,
            seq: 0,
            turn_replay: VecDeque::new(),
            web_search_notice: None,
            state: AttachState::Detached { running: false },
            background,
            compact: None,
            compacting_inline: false,
            session_start,
            turn_awaiting_start: None,
            subagent_tick: None,
            turn_worker_generation: 0,
            lifecycle,
            journal_id,
            presence,
            name,
            keep_warm_cell,
            session_lock,
            driver: None,
            driver_pending: None,
            driver_generation: 0,
            driver_interrupted_owner: None,
            driver_tick_interval: {
                let mut interval = tokio::time::interval(Duration::from_millis(200));
                interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
                interval
            },
            reload_fence: None,
            memory_mode_slot: memory_mode,
            memory_seen_active: false,
            parked_tool_activations: Vec::new(),
        };
        // Whether the build's grant (default or restored choice) took.
        actor.observe_memory();
        if actor.keep_warm {
            actor.note_pin();
        }
        Ok((handle, SessionTask(actor)))
    }

    /// Publish `conv.session.name` to the handle (listings, `--continue
    /// <name>`) and rewrite this session's registry entry so `synaps send
    /// --session <name>` resolves the new name. Call after any rename.
    /// Write the pin as it stands now to the daemon's pin book (no-op
    /// in-process): the entry is added while pinned, dropped otherwise.
    pub(crate) fn note_pin(&self) {
        let Some(sink) = self.host.pin_sink() else { return };
        let journal = (**self.journal_id.load()).clone();
        if self.keep_warm {
            let kw = self.cache_keepwarm.override_mode.map(|m| m.to_string());
            sink.pinned(self.id.as_str(), &journal, self.conv.session.name.as_deref(), self.config.cwd.as_deref(), kw.as_deref());
        } else {
            sink.unpinned(self.id.as_str(), &journal);
        }
    }

    /// Publish a new journal id; a pinned session's book entry follows it.
    pub(crate) fn set_journal_id(&mut self, new_id: String) {
        let old = self.journal_id.swap(Arc::new(new_id.clone()));
        if self.keep_warm && *old != new_id {
            if let Some(sink) = self.host.pin_sink() {
                sink.moved(&old, &new_id);
            }
        }
    }

    pub(crate) fn sync_name(&mut self) {
        let name = self.conv.session.name.clone();
        self.meta.name = name.clone();
        self.name.store(Arc::new(name.clone()));
        if self.keep_warm {
            self.note_pin();
        }
        if let Err(e) = self.background.update_session_name(name.as_deref()) {
            tracing::warn!(session = %self.id, "registry name update failed: {e}");
        }
    }

    // ── emit ─────────────────────────────────────────────────────────────

    /// The ONLY seq++ site. Pushes to `turn_replay` while streaming, except
    /// prompt traffic (never replayed), per-client replies, and full-history
    /// envelopes (`MessageHistory`, `Conversation`): the engine publishes one
    /// per round, and an attaching client already gets the LATEST history in
    /// its snapshot's `conversation` — replaying older ones would ship the
    /// whole history once per round and roll its mirror back to a stale
    /// state. The ring's `TurnStarted` drops `user_text`: that prompt is
    /// already in the snapshot's history (a replayed copy would show it
    /// twice). Completed rounds leave the ring when their checkpoint is
    /// adopted (`trim_replay_to_checkpoint`).
    pub(crate) fn emit(&mut self, event: SessionEventWire) {
        let replay = self.streaming
            && !matches!(
                event,
                SessionEventWire::Prompt(_)
                    | SessionEventWire::PromptResolved { .. }
                    | SessionEventWire::Attached { .. }
                    | SessionEventWire::QueryResult { .. }
                    | SessionEventWire::Conversation(_)
                    | SessionEventWire::Stream(StreamEvent::Session(
                        SessionEvent::MessageHistory(_)
                    ))
            );
        let env = Envelope {
            session_id: self.id.clone(),
            seq: self.seq,
            ts: chrono::Utc::now(),
            event,
        };
        self.seq += 1;
        if replay {
            if self.turn_replay.len() >= TURN_REPLAY_CAP {
                self.turn_replay.pop_front();
            }
            let mut ring = env.clone();
            if let SessionEventWire::TurnStarted { user_text, .. } = &mut ring.event {
                *user_text = None;
            }
            self.turn_replay.push_back(ring);
        }
        // No receivers is not an error: streams are not tied to clients.
        let _ = self.events.send(env);
    }

    /// A round checkpoint was adopted: every round in it is now in the
    /// attach snapshot's `conversation`, so the replay ring drops those
    /// rounds' display events — a mid-turn attach would otherwise render
    /// them twice (once from history, once from the ring). The engine
    /// publishes the checkpoint at a round boundary, so a tool call and its
    /// result are always on the same side of the cut. Kept: the turn's
    /// `TurnStarted`, notices, subagent progress, `Usage`.
    fn trim_replay_to_checkpoint(&mut self) {
        self.turn_replay.retain(|env| {
            !matches!(
                env.event,
                SessionEventWire::Stream(StreamEvent::Llm(_))
                    | SessionEventWire::Stream(StreamEvent::Agent(
                        AgentEvent::SteeringDelivered { .. }
                    ))
            )
        });
    }

    pub(crate) fn emit_conversation(&mut self) {
        let snap = self.conv.snapshot(self.consecutive_auto_turns);
        self.emit(SessionEventWire::Conversation(snap));
    }

    pub(crate) async fn publish_view(&mut self) {
        let v = RuntimeView::from_runtime(&self.runtime).await;
        self.view.store(Arc::new(v));
    }

    /// Queue a save of the current conversation. Never waits: the write
    /// runs on the actor's persister (latest wins), so a slow disk cannot
    /// stall the turn machine.
    pub(crate) fn request_save(&mut self) {
        if !self.config.persist || !self.conv.is_live() {
            return;
        }
        let Some(session) = self.conv.prepare_save() else {
            return;
        };
        let key = super::persister::SnapshotKey::of(&session);
        // This exact state is already queued or on disk — unless the last
        // write failed, in which case it is queued again.
        if self.last_queued_save.as_ref() == Some(&key) && self.persister.saves_ok() {
            return;
        }
        self.last_queued_save = Some(key);
        self.persister.save(session);
    }

    /// Wait, at most `SAVE_TIMEOUT`, until every queued save and draft
    /// operation has been applied. `false` on a failed save or the timeout
    /// (the queued writes still land, in order, when the disk catches up).
    pub(crate) async fn flush_saves(&mut self) -> bool {
        let ok = match tokio::time::timeout(budgets::SAVE_TIMEOUT, self.persister.flush()).await {
            Ok(ok) => ok,
            Err(_) => {
                tracing::warn!(session = %self.id, "session save still pending after the save budget");
                false
            }
        };
        if !ok {
            // Whatever failed is not known to be on disk: queue it again next
            // time instead of skipping it as a duplicate.
            self.last_queued_save = None;
        }
        ok
    }

    /// Save now: queue the current conversation and wait for it (bounded,
    /// see `flush_saves`).
    pub(crate) async fn save(&mut self) -> bool {
        self.request_save();
        self.flush_saves().await
    }

    /// A turn is over: announce `Idle` only once everything it queued is on
    /// disk (bounded, `flush_saves`). `Idle` has always meant "the session
    /// is saved": a client or script that reads the session, quits or
    /// hands off on `Idle` must never see the previous state. Per-round
    /// saves stay in the background; this waits once per turn.
    pub(crate) async fn announce_idle(&mut self) {
        // A turn may have revoked memory consent (the `memory_context`
        // tool): keep the handle's slot current for a reload that gets no
        // checkpoint reply.
        self.sync_memory_choice();
        self.flush_saves().await;
        // Before `Idle`, so a test that saw `Idle` knows the hook is in line.
        #[cfg(any(test, feature = "testing"))]
        if super::session_start::testing::take_requeue(&self.journal_id.load()) {
            self.queue_session_start();
        }
        self.emit(SessionEventWire::Idle);
        self.cache_keepwarm_on_idle().await;
    }

    /// Wall 1 — actor-owned context-head checkpoint persistence.
    ///
    /// The receipt stays in-process (actor task → runtime stream task).
    /// It never crosses the wire — `SessionEventWire` maps this to `Done`.
    ///
    /// Park ordering: `can_park()` requires `!self.streaming`, and streaming
    /// is only cleared on stream completion/error. The checkpoint event
    /// arrives mid-stream, so park cannot race with a pending checkpoint.
    /// No explicit drain is needed.
    async fn handle_context_head_checkpoint(
        &mut self,
        session_id: String,
        messages: Vec<crate::SharedMessage>,
        receipt: agent_core::core::context_head::ContextHeadReceipt,
    ) {
        self.turn_draft_rolled_over();
        // Stale session id guard (compaction may have changed it).
        if session_id != self.conv.session.id {
            tracing::warn!(
                event_id = %session_id,
                actor_id = %self.conv.session.id,
                "context head checkpoint: stale session id — ignoring"
            );
            receipt.complete(Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "stale session id after compaction",
            )));
            return;
        }

        // Non-persistent sessions (tests / --no-persist): complete without I/O.
        if !self.config.persist {
            receipt.complete(Ok(()));
            return;
        }

        // A queued ordinary save must not land after (and over) the durable
        // head: drain the persister first.
        if !self.persister.flush().await {
            tracing::warn!(session = %self.id, "context head checkpoint: an earlier save failed");
        }
        let result = self.conv.persist_context_head(&session_id, messages).await;
        if let Err(ref e) = result {
            tracing::error!(session = %session_id, "context head checkpoint save failed: {e}");
        }
        let succeeded = result.is_ok();
        receipt.complete(result);

        // P5: observe_checkpoint — revoke driver on failure or mismatch.
        if let Some(driver) = &self.driver {
            if !succeeded
                || driver.grant.session_id != session_id
                || session_id != self.conv.session.id
            {
                self.driver_revoke("context checkpoint failed or session replaced");
            }
        }
    }


    /// C3: the checkpoint reply payload (`SessionReloadRecord`).
    pub(crate) fn reload_record(&self) -> SessionReloadRecord {
        let view = self.view.load();
        SessionReloadRecord {
            // A consent revocation since the last record counts.
            config: SessionConfig {
                memory_mode: self.effective_memory_mode(),
                ..self.config.clone()
            },
            keep_warm: self.keep_warm,
            lifecycle: SessionLifecycle::from_u8(
                self.lifecycle.load(std::sync::atomic::Ordering::Acquire),
            ),
            settings_replay: self.settings_replay.clone(),
            cache_keepwarm: self.cache_keepwarm.override_mode.map(|m| m.to_string()),
            model: view.model.clone(),
            thinking_level: view.thinking_level.clone(),
            workers: Vec::new(),
            tool_activations: if self.runtime.is_live() {
                self.runtime.carried_tool_activations()
            } else {
                self.parked_tool_activations.clone()
            },
        }
    }

    /// Also the one place the inbox gate follows `input_owner`: every owner
    /// change (attach, detach, takeover) publishes presence. `Observe`
    /// clients never own input, so a keep-warm observer keeps it shut.
    ///
    /// Also the `/sessions` state: busy, `state_since` (restamped whenever
    /// idle/busy/parked changes), the park / eviction deadline as wall
    /// clock, and the conversation's title / cost / message count (kept as
    /// last published while Parked). Every transition that moves them —
    /// turn and compaction start/end, attach/detach, `rearm_park`,
    /// park/unpark — ends in a publish; stream deltas do not.
    pub(crate) fn publish_presence(&self) {
        use super::handle::{Presence, PresenceState};
        self.background
            .set_inbox_accepting(self.input_owner.is_some());
        let prev = self.presence.load();
        let busy = self.busy() || self.compacting_inline;
        let parked = self.is_parked();
        let state = if parked {
            PresenceState::Parked
        } else if busy {
            PresenceState::Busy
        } else {
            PresenceState::Idle
        };
        let now = chrono::Utc::now();
        let state_since = match prev.state_since {
            Some(t) if prev.state == state => Some(t),
            _ => Some(now),
        };
        let wall = |at: tokio::time::Instant| {
            let left = at.saturating_duration_since(tokio::time::Instant::now());
            chrono::Duration::from_std(left)
                .ok()
                .and_then(|d| now.checked_add_signed(d))
        };
        // A worker re-check is not a park (it only looks again), nor is a
        // grace that workers now hold: its firing would only re-check. The
        // F18 idle end of a zero-turn session counts — it leaves Live too.
        let park_due = self
            .park_deadline
            .filter(|d| !parked && Some(*d) != self.park_recheck_at)
            .filter(|_| self.can_end_idle() || self.can_park())
            .and_then(wall);
        let evict_due = self.park_deadline.filter(|_| parked).and_then(wall);
        let (title, cost_usd, messages) = if self.conv.is_live() {
            (
                conversation_title(&self.conv),
                self.conv.session_cost,
                self.conv.api_messages.len(),
            )
        } else {
            (prev.title.clone(), prev.cost_usd, prev.messages)
        };
        self.presence.store(Arc::new(Presence {
            clients: self.attached.len(),
            input_owner: self.input_owner,
            awaiting_input: self.pending_prompts.len(),
            roster: self.roster(),
            client_input: Some(self.client_input),
            busy,
            state,
            state_since,
            park_due,
            evict_due,
            title,
            cost_usd,
            messages,
        }));
    }

    /// Every attached client, by id.
    pub(crate) fn roster(&self) -> Vec<ClientInfo> {
        let mut roster: Vec<ClientInfo> = self
            .attached
            .iter()
            .map(|(id, a)| ClientInfo {
                id: *id,
                name: a.name.clone(),
                kind: a.meta.kind,
                mode: a.mode,
                attached_at_ms: a.attached_at_ms,
            })
            .collect();
        roster.sort_by_key(|c| c.id.0);
        roster
    }

    /// `tui (#1)`, or `client #1` for an id that isn't attached.
    fn client_label(&self, id: ClientId) -> String {
        match self.attached.get(&id) {
            Some(a) => format!("{} (#{})", a.name, id.0),
            None => format!("client #{}", id.0),
        }
    }

    /// Why client `c` may not send an input command, if it may not.
    /// `exclusive`: anyone but the owner. `shared`: observers (and clients
    /// that aren't attached); every drive client may.
    fn input_refusal(&self, c: ClientId) -> Option<String> {
        match self.client_input {
            ClientInput::Exclusive => (self.input_owner != Some(c)).then(|| match self.input_owner {
                Some(o) => format!("input owned by {}", self.client_label(o)),
                None => "input has no owner; attach with --takeover".to_string(),
            }),
            ClientInput::Shared => match self.attached.get(&c).map(|a| a.mode) {
                Some(AttachMode::Mirror | AttachMode::Takeover) => None,
                Some(AttachMode::Observe) => {
                    Some("this client is read-only (attached with --observe)".to_string())
                }
                None => Some("this client is not attached".to_string()),
            },
        }
    }

    /// Client `c` sent input (or answered a prompt): remember when, and with
    /// `client_input = shared` make it the active client. Observers never
    /// become active.
    fn note_input(&mut self, c: ClientId) {
        let Some(a) = self.attached.get_mut(&c) else { return };
        if a.mode == AttachMode::Observe {
            return;
        }
        self.input_seq += 1;
        a.last_input = self.input_seq;
        if self.client_input == ClientInput::Shared && self.input_owner != Some(c) {
            let from = self.input_owner;
            self.input_owner = Some(c);
            self.emit(SessionEventWire::InputOwnerChanged { from, to: Some(c), reason: OwnerChangeReason::Input });
            self.publish_presence();
        }
    }

    /// Recomputes `state` from clients/streaming and (re)arms or disarms the
    /// park grace timer. A no-op while Parking/Parked.
    pub(crate) fn update_attach_state(&mut self) {
        self.publish_presence();
        if matches!(self.state, AttachState::Parking | AttachState::Parked) {
            return;
        }
        self.state = if self.attached.is_empty() {
            AttachState::Detached {
                running: self.streaming,
            }
        } else {
            AttachState::Attached(self.attached.len())
        };
        self.rearm_park();
        self.rearm_prompt_abandon();
    }

    // ── Parked (B3) ──────────────────────────────────────────────────────

    pub(crate) fn is_parked(&self) -> bool {
        matches!(self.state, AttachState::Parked)
    }

    /// A session with nothing to save (`ConversationState::save` skips an
    /// empty `api_messages`, so no journal ever exists) never parks: it
    /// costs nothing warm and could not be unparked from disk (H2).
    pub(crate) fn can_park(&self) -> bool {
        // An active cache keep-warm period defers the idle auto-park (its
        // cost is bounded by the ski-rental ledger); when it ends,
        // `rearm_park`. An explicit `Park` ends the period first.
        self.can_park_ignoring_keepwarm() && !self.cache_keepwarm.active()
    }

    /// `can_park` minus the keep-warm deferral.
    pub(crate) fn can_park_ignoring_keepwarm(&self) -> bool {
        self.can_park_but_for_workers() && !self.workers_hold_park()
    }

    /// Whether parking would cost a worker (`SubagentRegistry::holds_park`):
    /// the park waits and re-checks every [`WORKER_PARK_RECHECK`].
    fn workers_hold_park(&self) -> bool {
        self.runtime.is_live()
            && self
                .runtime
                .subagent_registry()
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .holds_park()
    }

    /// Every park condition except the session's workers.
    fn can_park_but_for_workers(&self) -> bool {
        self.attached.is_empty()
            && !self.streaming
            && self.compact.is_none()
            && self.pending_prompts.is_empty()
            && self.conv.is_live()
            && !self.conv.api_messages.is_empty()
            && self.conv.queued_message.is_none()
            && !self.keep_warm
            && self.config.persist
            && !self.is_parked()
            && self.driver.is_none()
    }

    /// (E-P7, §S3) Return the first breached spend ceiling, if any, as
    /// `(scope, cost_observed, cap)`. Checks the host-owned per-session cap
    /// (`config.max_session_cost`) and, when a driver is armed, the effective
    /// per-run cap (`min(plugin grant, host)`) measured from `cost_at_arm`.
    /// `None` means every configured ceiling still has headroom.
    fn cost_cap_breach(&self) -> Option<(&'static str, f64, f64)> {
        let total = self.conv.session_cost;
        if let Some(cap) = self.config.max_session_cost {
            if cap.is_finite() && total >= cap {
                return Some(("session", total, cap));
            }
        }
        if let Some(driver) = &self.driver {
            // Host cap for the run is not yet a config field (S3 should-have
            // #4, daemon-level); pass `None` so the plugin's proposal stands
            // alone until the daemon cap lands.
            if let Some(cap) = driver.grant.effective_cost_cap(None) {
                let run_cost = total - driver.cost_at_arm;
                if cap.is_finite() && run_cost >= cap {
                    return Some(("run", run_cost, cap));
                }
            }
        }
        None
    }

    /// `<sessions>/<id>.json` — written by both persistence modes.
    fn journal_exists(&self) -> bool {        let id = (**self.journal_id.load()).clone();
        crate::config::resolve_write_path("sessions")
            .join(format!("{id}.json"))
            .is_file()
    }

    // ── E: driver arm/revoke ─────────────────────────────────────────────

    /// Invalidate the driver: drop `DriverState`, cancel pending work, emit
    /// `DriverRevoked` with any undelivered steering, and release the host grant.
    pub(crate) fn driver_revoke(&mut self, reason: &str) {
        let undelivered = self
            .driver
            .as_mut()
            .map(|d| d.steering.drain(..).collect::<Vec<_>>())
            .unwrap_or_default();
        let was_active = self.driver.is_some() || self.driver_pending.is_some();
        // Dropping the driver below cancels the driver turn's token.
        if self.driver.as_ref().is_some_and(|d| d.awaiting_terminal) {
            self.note_cancel_cause(crate::engine::interrupt::InterruptReason::Driver);
        }
        if let Some(driver) = &self.driver {
            self.driver_interrupted_owner = Some(driver.grant.plugin_id.clone());
            self.host
                .release_driver(&driver.grant.plugin_id, &self.id);
        }
        self.driver_generation = self.driver_generation.wrapping_add(1);
        self.driver_pending = None;
        self.driver = None;
        if was_active {
            self.emit(SessionEventWire::DriverRevoked {
                reason: reason.to_string(),
                undelivered_steering: undelivered,
            });
        }
        // Dropping `DriverState` cancelled the driver turn's token (a child
        // of the driver's): that turn is over, but not through `cancel_turn`.
        // Unwind prompts now (as `cancel_turn` does), then give its stream
        // the drain budget to deliver its cancel-path history and terminal
        // event (`finish_revoked_turn`); a stream that never started (its
        // start task was just aborted) ends at the next loop turn. A caller
        // that cancels the turn itself right after (`finish`, `checkpoint`,
        // the cost cap) supersedes this: `cancel_turn` clears the deadline.
        if self.turn_cancelled() && self.revoked_turn_deadline.is_none() {
            self.resolve_pending_prompts();
            let wait = if self.stream.is_some() {
                budgets::CANCEL_DRAIN_TIMEOUT
            } else {
                std::time::Duration::ZERO
            };
            self.revoked_turn_deadline = Some(tokio::time::Instant::now() + wait);
        }
        self.rearm_park();
    }

    /// Spawn the plugin's start command as an async task; the result is
    /// processed in `driver_tick()` (P4). Invokes the command via the
    /// extension manager, just like the TUI's `start_command`.
    pub(crate) fn driver_start(&mut self, plugin: String, command: String, arg: String) {
        // Revoke any existing driver first (TUI :369).
        if self.driver.is_some() || self.driver_pending.is_some() {
            self.driver_revoke("explicit command");
        }

        let manager = self.host.ext_manager().clone();
        let session_id = self.id.0.clone();

        // Resolve handler + check permissions synchronously.
        let (handler, timeout) = match manager.try_read() {
            Ok(mgr) => {
                let handler = match mgr.user_action_handler(&plugin) {
                    Ok(h) => h,
                    Err(error) => {
                        self.emit(SessionEventWire::SystemNotice(error));
                        return;
                    }
                };
                let timeout = if mgr.session_driver_handler(&plugin).is_ok() {
                    5
                } else {
                    self.emit(SessionEventWire::SystemNotice(
                        "session driver extension lacks validated session.drive permission".into(),
                    ));
                    return;
                };
                (handler, timeout)
            }
            Err(_) => {
                self.emit(SessionEventWire::SystemNotice(
                    "extensions are loading or busy — try again shortly".into(),
                ));
                return;
            }
        };

        let handler_generation = super::driver::live_generation(&handler).ok();

        let owner = plugin.clone();
        let cmd = command.clone();
        let args: Vec<String> = arg.split_whitespace().map(str::to_owned).collect();

        let generation = self.driver_generation;
        let task_handler = handler.clone();
        self.driver_pending = Some(super::driver::DriverPending {
            generation,
            session_id: session_id.clone(),
            task: super::driver::Task(tokio::spawn(async move {
                let (sink, collector) =
                    invoke_event_channel(InvokeOutputBudget::default());
                let request_id = uuid::Uuid::new_v4().to_string();
                let (result, report) =
                    tokio::time::timeout(Duration::from_secs(timeout), async {
                        tokio::join!(
                            task_handler.invoke_command(&cmd, args, &request_id, sink),
                            collector.collect()
                        )
                    })
                    .await
                    .unwrap_or_else(|_| {
                        (
                            Err(format!("interactive command timed out ({timeout}s)")),
                            crate::extensions::invoke_output::InvokeOutputReport {
                                events: Vec::new(),
                                counters: Default::default(),
                            },
                        )
                    });
                super::driver::TaskResult::Command {
                    owner,
                    command,
                    handler: task_handler,
                    handler_generation,
                    result,
                    report,
                }
            })),
        });
    }

    /// Arm the driver after a successful start command result. Mirrors
    /// the TUI's `arm()` (:538-602). Called from P4's `driver_tick()`.
    pub(crate) fn driver_arm(
        &mut self,
        owner: String,
        handler: Arc<dyn crate::extensions::runtime::ExtensionHandler>,
        handler_generation: u64,
        reply: Reply,
    ) -> std::result::Result<(), String> {
        let manager = self.host.ext_manager().clone();
        super::driver::same_handler(&manager, &owner, &handler, handler_generation)?;

        if self.driver.is_some() {
            return Err("cannot replace an armed grant".into());
        }

        let session_id = self.id.0.clone();
        let (grant, proposal) = Grant::from_start(&owner, &session_id, reply)?;

        // Claim the host-level single-tenancy grant.
        self.host
            .claim_driver(&owner, &self.id)
            .map_err(|e| e.to_string())?;

        let cancel = CancellationToken::new();
        let workers = self.runtime.subagent_registry().clone();
        let worker_epoch = workers
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .set_spawn_cancellation(Some(cancel.clone()));
        let deadline_task = grant.deadline().map(|deadline| {
            let c = cancel.clone();
            let w = workers.clone();
            super::driver::Task(tokio::spawn(async move {
                tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
                c.cancel();
                super::driver::cancel_workers(&w, worker_epoch);
            }))
        });

        let models = grant.models().to_vec();
        let proposal_notice = proposal.notice.clone();
        let selection = proposal.selection.clone();
        let deadline_ms = super::driver::deadline_ms(&grant);
        let run_id = grant.run_id.clone();
        let cost_at_arm = self.conv.session_cost;

        let mut state = super::driver::DriverState {
            selection: proposal.selection.clone(),
            grant,
            handler,
            handler_generation,
            cancel,
            workers,
            worker_epoch,
            deadline_task,
            proposal: None,
            awaiting_terminal: false,
            outcome: None,
            feedback: crate::extensions::feedback::Tracker::default(),
            completed_feedback: "unknown",
            steering: VecDeque::new(),
            auto_wakes_blocked: false,
            cost_at_arm,
        };

        super::driver::schedule(&mut state, proposal)?;

        // Apply context mode (session-only, like /context auto/off).
        let context_notice = state.grant.apply_context_mode(&self.runtime)?;
        if let Some(text) = context_notice {
            self.emit(SessionEventWire::SystemNotice(text));
        }

        self.driver_interrupted_owner = None;
        self.driver = Some(state);

        self.emit(SessionEventWire::DriverArmed {
            plugin_id: owner,
            run_id,
            models,
            selection,
            deadline_ms,
            notice: proposal_notice,
        });

        Ok(())
    }

    // ── autonomous work: the reload fence and on_session_start ───────────

    /// Whether autonomous work may start now, and if not, why. Two states
    /// hold it back: a reload's fence (#255) until the exec or its abort, and
    /// this session's `on_session_start` until it has run. `Fenced` wins when
    /// both hold. Each source of autonomous work asks [`Self::holds`]; none
    /// reads the fence or the hook's state itself.
    ///
    /// | source          | `Fenced`  | `SessionStartPending`                                         |
    /// |-----------------|-----------|---------------------------------------------------------------|
    /// | queue wake      | waits     | runs: the turn starts, and its request waits for the hook      |
    /// | park timer      | waits     | runs: a park that happens takes a hook not called out of line  |
    /// | keep-warm timer | waits     | waits (timer off): a due ping goes out once the hook has run   |
    /// | driver tick     | waits     | runs: an armed driver is revoked (`driver_idle_conflict`)      |
    /// | driver start    | refused * | rejected in the tick (`driver_idle_conflict`)                  |
    ///
    /// \* `DriverStart` is a command: under the fence the command fence
    /// refuses it first ("daemon reloading"); the tick never sees it.
    ///
    /// The queue wake must not wait for the hook: a parked session's hook is
    /// out of line (`SessionStart::Pending`) and joins it only at the unpark,
    /// and the unpark is what the queue wake does. Waiting there would leave
    /// an event for a parked session queued for good. The turn's own wait
    /// (`open_turn_stream` after `session_start_ready`) holds its request
    /// instead, without blocking the loop.
    pub(crate) fn autonomous_work_blocked(&self) -> Option<WorkBlock> {
        if self.reload_fence.is_some() {
            Some(WorkBlock::Fenced)
        } else if !self.session_start.is_done() {
            Some(WorkBlock::SessionStartPending)
        } else {
            None
        }
    }

    /// Whether `source` waits now (the table on
    /// [`Self::autonomous_work_blocked`]).
    pub(crate) fn holds(&self, source: WorkSource) -> bool {
        match self.autonomous_work_blocked() {
            None => false,
            Some(WorkBlock::Fenced) => true,
            Some(WorkBlock::SessionStartPending) => matches!(source, WorkSource::KeepWarmTimer),
        }
    }

    // ── E-P4: driver_tick ────────────────────────────────────────────────

    /// Actor-equivalent of the TUI's `idle_conflict()` (TUI :437-463).
    /// Returns `Some(reason)` when the driver must defer or revoke.
    fn driver_idle_conflict(&self) -> Option<&'static str> {
        if self.conv.queued_message.is_some() || !self.conv.pending_events.is_empty() {
            return Some("other queued work");
        }
        if self.compact.is_some() {
            return Some("compaction");
        }
        match self.autonomous_work_blocked() {
            Some(WorkBlock::Fenced) => return Some("daemon reloading"),
            Some(WorkBlock::SessionStartPending) => return Some("extensions loading or reloading"),
            None => {}
        }
        if self.conv.context_head.is_blocked(&self.conv.session) {
            return Some("unverified context head");
        }
        if self.driver_completion_blocked() {
            return Some("outstanding workers require collection/reconciliation");
        }
        None
    }

    /// Actor-equivalent of the TUI's `completion_blocked()` (TUI :465-480).
    fn driver_completion_blocked(&self) -> bool {
        use agent_core::orchestration::CompletionGate;
        self.runtime.orchestration().is_some_and(|o| {
            !matches!(o.completion_gate(), CompletionGate::Allowed)
        }) || self
            .runtime
            .subagent_registry()
            .lock()
            .map(|r| {
                r.list_active().iter().any(|(_, _, status)| {
                    *status == crate::runtime::subagent::SubagentStatus::Running
                })
            })
            .unwrap_or(true)
    }

    /// 200 ms timer arm — the core driver loop. Every invocation performs only
    /// bounded state transitions; all plugin and provider IO is in abort-on-drop
    /// spawned tasks. Mirrors the TUI's `tick()` (TUI :695-1027).
    pub(crate) async fn driver_tick(&mut self) {
        use crate::extensions::session_driver::Outcome;
        if self.holds(WorkSource::DriverTick) {
            return;
        }

        // ── 1. Cleanup cancelled stream setup ────────────────────────────
        // (e.g. the grant's deadline cancelled the driver token before the
        // turn's stream started): an interrupted turn, ended like any other.
        if self.stream.is_none() && self.turn_cancelled() {
            self.finish_revoked_turn(false).await;
        }

        // ── 2. Validate lifecycle ────────────────────────────────────────
        if let Some(driver) = &self.driver {
            let invalid = if driver.grant.session_id != self.id.0 {
                Some("session replaced".to_string())
            } else if driver.grant.expired() {
                Some("grant deadline reached".into())
            } else if driver.cancel.is_cancelled() {
                Some("canceled".into())
            } else {
                let manager = self.host.ext_manager().clone();
                super::driver::same_handler(
                    &manager,
                    &driver.grant.plugin_id,
                    &driver.handler,
                    driver.handler_generation,
                )
                .err()
            };
            if let Some(reason) = invalid {
                self.driver_revoke(&reason);
                return;
            }
        }

        // ── 3. Idle conflict (only when driver is armed) ─────────────────
        if self.driver.is_some() {
            // (E-P7, §S1) No clients → no headless spend. `detach()` revokes on
            // the last-client transition; this is the belt-and-suspenders check
            // for any path that leaves a grant armed with zero clients.
            if self.attached.is_empty() {
                self.driver_revoke("no clients attached");
                return;
            }
            // F10/F14: use the full driver_idle_conflict() instead of an
            // inline subset that was missing `completion_blocked`.
            if let Some(reason) = self.driver_idle_conflict() {
                self.driver_revoke(reason);
                return;
            }

            // Let the reactor classify queued events before deciding whether they
            // are competing work or steering within this owned turn.
            if !self.runtime.event_queue().is_empty() {
                return;
            }
        }

        // ── 4. Process finished pending task ─────────────────────────────
        if self
            .driver_pending
            .as_ref()
            .is_some_and(|p| p.task.0.is_finished())
        {
            let mut pending = self.driver_pending.take().expect("checked pending");
            if pending.generation != self.driver_generation
                || pending.session_id != self.id.0
            {
                return;
            }
            let result = match (&mut pending.task.0).await {
                Ok(result) => result,
                Err(error) => {
                    self.driver_revoke("driver task failed");
                    self.emit(SessionEventWire::SystemNotice(format!(
                        "Session driver task failed: {error}"
                    )));
                    return;
                }
            };
            match result {
                super::driver::TaskResult::Command {
                    owner,
                    command: _,
                    handler,
                    handler_generation,
                    result,
                    report,
                } => {
                    let current = self
                        .host
                        .ext_manager()
                        .try_read()
                        .ok()
                        .and_then(|m| m.user_action_handler(&owner).ok());
                    if !current
                        .as_ref()
                        .is_some_and(|current| Arc::ptr_eq(current, &handler))
                    {
                        self.driver_revoke(
                            "command owner unloaded, replaced, or unavailable",
                        );
                        return;
                    }
                    let output_failed = report.is_limited()
                        || report.events.iter().any(|event| {
                            matches!(
                                event,
                                crate::extensions::runtime::InvokeCommandEvent::Output(
                                    crate::extensions::commands::CommandOutputEvent::Error { .. }
                                )
                            )
                        });
                    // F7 (shady should-fix): plugin command output is visible
                    // to the user, but BATCHED into one notice (cap 20 lines) —
                    // a chatty `/auto start` must not flood the client with one
                    // wire event per line.
                    let mut lines: Vec<String> = Vec::new();
                    for event in &report.events {
                        if let crate::extensions::runtime::InvokeCommandEvent::Output(out) = event {
                            let text = match out {
                                crate::extensions::commands::CommandOutputEvent::Text { content }
                                | crate::extensions::commands::CommandOutputEvent::System { content } => {
                                    Some(content.clone())
                                }
                                crate::extensions::commands::CommandOutputEvent::Error { content } => {
                                    Some(format!("Error: {content}"))
                                }
                                _ => None,
                            };
                            if let Some(text) = text {
                                lines.push(text);
                            }
                        }
                    }
                    if !lines.is_empty() {
                        const MAX_LINES: usize = 20;
                        let truncated = lines.len() > MAX_LINES;
                        if truncated {
                            lines.truncate(MAX_LINES);
                            lines.push("…output truncated".into());
                        }
                        self.emit(SessionEventWire::SystemNotice(lines.join("\n")));
                    }
                    if let Some(notice) = report.limit_notice() {
                        self.emit(SessionEventWire::SystemNotice(format!(
                            "command output truncated: {}",
                            notice.message,
                        )));
                    }
                    if let Ok(value) = result {
                        let result = match protocol::parse_reply(&value) {
                            Ok(Some(reply @ Reply::Start { .. })) => {
                                if output_failed {
                                    Err("command reported an error or exceeded its output budget"
                                        .into())
                                } else if self.streaming || self.stream.is_some() {
                                    Err("foreground turn still active".into())
                                } else if let Some(reason) = self.driver_idle_conflict() {
                                    Err(reason.into())
                                } else {
                                    match handler_generation {
                                        Some(generation) => {
                                            self.driver_arm(
                                                owner, handler, generation, reply,
                                            )
                                        }
                                        None => Err(
                                            "command owner did not expose a live lifecycle at dispatch"
                                                .into(),
                                        ),
                                    }
                                }
                            }
                            Ok(Some(
                                Reply::Stop { notice: text }
                                | Reply::Status { notice: text },
                            )) => {
                                self.emit(SessionEventWire::SystemNotice(text));
                                Ok(())
                            }
                            Ok(Some(Reply::Next { .. })) => Err(
                                "next requires an armed poll, not an interactive command"
                                    .into(),
                            ),
                            Ok(None) => Ok(()),
                            Err(error) => Err(error),
                        };
                        if let Err(error) = result {
                            self.emit(SessionEventWire::SystemNotice(format!(
                                "Session driver rejected: {error}"
                            )));
                        }
                    }
                }
                super::driver::TaskResult::Poll(result) => {
                    let Some(driver) = self.driver.as_mut() else {
                        return;
                    };
                    let stop_notice = match &result {
                        Ok(Reply::Stop { notice }) => Some(notice.clone()),
                        _ => None,
                    };
                    match result.and_then(|reply| driver.grant.accept(reply)) {
                        Ok(Some(proposal)) => {
                            let text = proposal.notice.clone();
                            if let Err(error) =
                                super::driver::schedule(driver, proposal)
                            {
                                self.driver_revoke(&error);
                            } else {
                                self.emit(SessionEventWire::SystemNotice(text));
                            }
                        }
                        Ok(None) => {
                            self.driver_revoke("owner requested stop");
                            if let Some(text) = stop_notice {
                                self.emit(SessionEventWire::SystemNotice(text));
                            }
                        }
                        Err(error) => {
                            self.driver_revoke(&format!("invalid poll response: {error}"))
                        }
                    }
                }
                super::driver::TaskResult::Prepared { proposal, result } => {
                    if self.streaming || self.stream.is_some() {
                        self.driver_revoke("foreground work took priority");
                        return;
                    }
                    // This includes a pending `on_session_start`: a driver
                    // turn opens its stream below, not through `start_turn`,
                    // so it never waits for the hook and must not run before it.
                    if let Some(reason) = self.driver_idle_conflict() {
                        self.driver_revoke(reason);
                        return;
                    }
                    match *result {
                        Err(protocol::PrepareError::Selection(error)) => {
                            self.emit(SessionEventWire::SystemNotice(format!(
                                "Session driver selection rejected (nothing sent): {error}"
                            )));
                            if let Some(driver) = self.driver.as_mut() {
                                driver.outcome = Some((
                                    Outcome::SelectionRejected,
                                    "unknown".into(),
                                ));
                            }
                        }
                        Err(protocol::PrepareError::Blocked(error)) => {
                            self.driver_revoke(&format!("preflight blocked: {error}"));
                        }
                        Ok(candidate) => {
                            let validated = proposal.clone();
                            let history = super::driver::history_with_steering(
                                &self.conv.api_messages,
                                &self
                                    .driver
                                    .as_ref()
                                    .map(|d| &d.steering)
                                    .cloned()
                                    .unwrap_or_default(),
                            );
                            if let Err(error) = protocol::validate_prepared(
                                &candidate, &validated, &history,
                            ) {
                                self.driver_revoke(&format!(
                                    "prepared selection changed before commit: {error}"
                                ));
                                return;
                            }
                            // F3: commit steering + prompt via the
                            // driver.rs helper (single call site).
                            // Destructure to split borrows across conv fields.
                            {
                                let driver = self.driver.as_mut().expect("driver present");
                                let conv = &mut *self.conv;
                                super::driver::commit_submission(
                                    &mut conv.api_messages,
                                    &mut driver.steering,
                                    &proposal.prompt,
                                );
                            }
                            // Apply the prepared runtime.
                            *self.runtime = candidate;
                            self.conv.session.model = self.runtime.model().to_owned();
                            self.conv.session.thinking_level =
                                self.runtime.thinking_level().to_owned();
                            self.consecutive_auto_turns = 0;
                            self.turn_baseline = self.conv.api_messages.len();
                            self.turn_steered_events.clear();
                            self.turn_replay.clear();
                            self.streaming = true;
                            self.open_turn_draft();
                            self.update_attach_state();
                            self.emit(SessionEventWire::TurnStarted {
                                turn_baseline: self.turn_baseline,
                                trigger: TurnTrigger::DriverAuto,
                                user_text: Some(proposal.prompt.clone()),
                            });
                            self.announce_web_search_state();

                            // Set up the driver's awaiting_terminal.
                            if let Some(driver) = self.driver.as_mut() {
                                driver.awaiting_terminal = true;
                                driver.completed_feedback = "unknown";
                                if driver.grant.feedback_enabled() {
                                    driver.feedback.begin_turn(
                                        &driver.selection.model,
                                        &driver.selection.effort,
                                    );
                                }
                            }

                            // Spawn stream start as a task (never block tick).
                            let handler = self
                                .driver
                                .as_ref()
                                .expect("driver")
                                .handler
                                .clone();
                            let handler_generation = self
                                .driver
                                .as_ref()
                                .expect("driver")
                                .handler_generation;
                            let ct = self
                                .driver
                                .as_ref()
                                .expect("driver")
                                .cancel
                                .child_token();
                            self.cancel = Some(ct.clone());
                            // Cancelled by anything that notes no cause (the
                            // grant's deadline task): the driver ending.
                            let completion = crate::runtime::TurnCompletion::with_default_cause(
                                crate::runtime::CancelCause::Driver,
                            );
                            self.turn_completion = Some(completion.clone());
                            let (tx, rx) = mpsc::unbounded_channel();
                            self.steer_tx = Some(tx);
                            self.owner_steer_budget = None;
                            let mut runtime = self.runtime.clone();
                            // F2: share the parent's TTL latches so the
                            // driver turn doesn't fire a redundant 1h→5m
                            // downgrade notice.
                            runtime.share_ttl_latches(&self.runtime);
                            let history = self.conv.api_messages.clone();
                            let secret = self.secret_prompt_handle.clone();
                            let session_id = self.id.0.clone();
                            let generation = self.driver_generation;
                            self.driver_pending = Some(super::driver::DriverPending {
                                generation,
                                session_id,
                                task: super::driver::Task(tokio::spawn(async move {
                                    if ct.is_cancelled() {
                                        return super::driver::TaskResult::Started(
                                            Err("canceled before stream setup".into()),
                                        );
                                    }
                                    if let Err(error) =
                                        super::driver::same_lifecycle(
                                            &handler,
                                            handler_generation,
                                        )
                                    {
                                        return super::driver::TaskResult::Started(
                                            Err(error),
                                        );
                                    }
                                    tokio::select! {
                                        biased;
                                        _ = ct.cancelled() => super::driver::TaskResult::Started(Err("canceled during stream setup".into())),
                                        started = runtime.run_stream_tracked(history, ct.clone(), Some(rx), Some(secret), false, completion) => super::driver::TaskResult::Started(Ok(started)),
                                    }
                                })),
                            });

                            // Start subagent tick (same as normal start_turn).
                            let mut tick = tokio::time::interval(
                                std::time::Duration::from_secs(1),
                            );
                            tick.set_missed_tick_behavior(
                                tokio::time::MissedTickBehavior::Delay,
                            );
                            tick.reset();
                            self.subagent_tick = Some(tick);
                            self.request_save();
                        }
                    }
                }
                super::driver::TaskResult::Started(started) => match started {
                    Ok(started) if self.driver.is_some() => {
                        self.stream = Some(started);
                    }
                    Err(error) => {
                        self.driver_revoke(&error);
                    }
                    _ => {}
                },
            }
        }

        // ── 5. Idle scheduling ───────────────────────────────────────────
        if self.streaming || self.stream.is_some() || self.driver_pending.is_some() {
            return;
        }
        if let Some(reason) = self.driver_idle_conflict() {
            self.driver_revoke(reason);
            return;
        }
        let session_cost = self.conv.session_cost;
        let Some(driver) = self.driver.as_mut() else {
            return;
        };
        if let Some((outcome, error_kind)) = driver.outcome.take() {
            let request = super::driver::poll_request(driver, outcome, error_kind, session_cost);
            let handler = driver.handler.clone();
            let session_id = self.id.0.clone();
            let generation = self.driver_generation;
            self.driver_pending = Some(super::driver::DriverPending {
                generation,
                session_id,
                task: super::driver::Task(tokio::spawn(async move {
                    // F4: poll must not hang forever — 30 s timeout like
                    // the prepare path's 5 s.
                    let result = tokio::time::timeout(
                        Duration::from_secs(30),
                        protocol::poll(handler, request),
                    )
                    .await
                    .unwrap_or_else(|_| Err("poll timed out (30s)".into()));
                    super::driver::TaskResult::Poll(result)
                })),
            });
        } else if driver
            .proposal
            .as_ref()
            .is_some_and(|p| Instant::now() >= p.due)
        {
            let proposal =
                driver.proposal.take().expect("checked proposal").proposal;
            // Clone is read-only (prepare never sends a request) and
            // discarded — no TTL latch sharing needed.
            let runtime = self.runtime.clone();
            let history = super::driver::history_with_steering(
                &self.conv.api_messages,
                &driver.steering,
            );
            let session_id = self.id.0.clone();
            let generation = self.driver_generation;
            self.driver_pending = Some(super::driver::DriverPending {
                generation,
                session_id,
                task: super::driver::Task(tokio::spawn(async move {
                    let result = tokio::time::timeout(
                        Duration::from_secs(5),
                        protocol::prepare(&runtime, &proposal, &history),
                    )
                    .await
                    .unwrap_or_else(|_| {
                        Err(protocol::PrepareError::Blocked(
                            "preparation timed out (5s)".into(),
                        ))
                    });
                    super::driver::TaskResult::Prepared {
                        proposal,
                        result: Box::new(result),
                    }
                })),
            });
        }
    }

    /// F10: the session lock follows the conversation to `new_id` (a fresh
    /// id from NewSession or compaction). Best-effort: on failure the old
    /// lock is still released (the conversation left it) and the failure
    /// logged. The new lock is taken BEFORE the old one is dropped.
    pub(crate) fn reacquire_session_lock(&mut self, new_id: &str) {
        match Self::try_lock_session(new_id) {
            Ok(lock) => self.session_lock = Some(lock),
            Err(e) => {
                tracing::warn!(session = %new_id, "reacquire session lock: {e}");
                self.session_lock = None;
            }
        }
    }

    /// Try to take the journal lock on `id` for this daemon process.
    pub(crate) fn try_lock_session(
        id: &str,
    ) -> std::result::Result<agent_core::session_lock::SessionLock, agent_core::session_lock::SessionLockError>
    {
        let dir = agent_core::session_lock::sessions_dir();
        let holder = agent_core::session_lock::LockHolder {
            pid: std::process::id(),
            kind: "daemon".to_string(),
        };
        agent_core::session_lock::SessionLock::try_acquire(&dir, id, holder)
    }

    /// F18: a session with NO history can never park (nothing to journal),
    /// so with no clients it would stay Live — Runtime resident — forever.
    /// Same gates as `can_park` minus the history requirement: at the park
    /// deadline such a session ends instead.
    pub(crate) fn can_end_idle(&self) -> bool {
        self.attached.is_empty()
            && !self.streaming
            && self.compact.is_none()
            && self.pending_prompts.is_empty()
            && self.conv.is_live()
            && self.conv.api_messages.is_empty()
            && self.conv.queued_message.is_none()
            && !self.keep_warm
            && !self.is_parked()
            // An armed driver keeps the session alive between turns exactly
            // like an attached client would (shady P3/P4 F6: without this, a
            // zero-turn armed session is idle-ended 5 s after the last client
            // detaches, killing the run mid-arm). `can_park` already guards.
            && self.driver.is_none()
            && self.driver_pending.is_none()
    }

    pub(crate) fn rearm_park(&mut self) {
        self.rearm_park_deadline();
        self.publish_presence();
    }

    fn rearm_park_deadline(&mut self) {
        if self.is_parked() {
            // Eviction deadline is owned by `park()`; the attach path unparks
            // (which rebuilds the runtime and clears the deadline).
            return;
        }
        if self.can_end_idle() {
            // F18: nothing to keep warm — a zero-turn session with no
            // clients ends after a short grace, not the full park grace
            // (which exists to keep a runtime WITH history warm for a
            // quick reconnect).
            if self.park_deadline.is_none() {
                self.park_deadline =
                    Some(tokio::time::Instant::now() + idle_end_grace());
            }
            return;
        }
        if !self.can_park() {
            // Only workers in the way: look again later (a worker finishing
            // or being collected does not always pass through here).
            let workers_only = self.can_park_but_for_workers()
                && !self.cache_keepwarm.active()
                && park_grace().is_some();
            self.park_deadline = if workers_only {
                self.park_deadline.or_else(|| {
                    let at = tokio::time::Instant::now() + worker_park_recheck();
                    self.park_recheck_at = Some(at);
                    Some(at)
                })
            } else {
                None
            };
            return;
        }
        match park_grace() {
            Some(grace) => {
                // The earlier of a pending deadline and a fresh grace: a
                // grace armed at detach keeps counting, while a far-off
                // worker re-check gives way once the workers let go.
                let due = tokio::time::Instant::now() + grace;
                // A re-check due before the grace stays a re-check: when it
                // fires it arms the grace rather than parking early.
                self.park_deadline = Some(self.park_deadline.map_or(due, |d| d.min(due)));
            }
            None => self.park_deadline = None,
        }
    }

    /// (P11) Bound the resource pin from an ABANDONED prompt. A pending host
    /// confirmation on a plain interactive session survives detach so the user
    /// reattaches and answers it — but if NObody ever comes back it pins the
    /// session forever (`can_park()` requires `pending_prompts` empty). Arm a
    /// deadline only in that abandoned state (zero clients, prompt pending, no
    /// driver); a re-attach clears it (prompt survives, user answers); expiry
    /// denies the prompt(s). The driver-armed case fail-closes immediately in
    /// `detach` and never reaches here. Idempotent: safe to call on every
    /// attach/detach/prompt transition.
    fn rearm_prompt_abandon(&mut self) {
        let armed_driver = self.driver.is_some() || self.driver_pending.is_some();
        let abandoned = self.attached.is_empty()
            && !self.pending_prompts.is_empty()
            && !armed_driver
            && !self.is_parked();
        if !abandoned {
            // Someone is attached, the prompt was answered, or a driver took
            // over the fail-closed path — never auto-deny while any of these.
            self.prompt_abandon_deadline = None;
            return;
        }
        if self.prompt_abandon_deadline.is_none() {
            // `None` config ⇒ deadline stays `None` ⇒ disabled (pre-#112).
            self.prompt_abandon_deadline =
                prompt_abandon_timeout().map(|d| tokio::time::Instant::now() + d);
        }
    }

    /// (P11) The prompt-abandonment deadline expired with still zero clients:
    /// deny every pending prompt (send `None` + emit `PromptResolved`) so the
    /// session stops zombie-blocking and `can_park()` becomes true. A re-attach
    /// would have cleared the deadline first; re-check defensively.
    fn on_prompt_abandon_deadline(&mut self) {
        self.prompt_abandon_deadline = None;
        if !self.attached.is_empty() {
            return;
        }
        let had_prompts = !self.pending_prompts.is_empty();
        while let Some((pr, tx)) = self.pending_prompts.pop_front() {
            let _ = tx.send(None);
            self.emit(SessionEventWire::PromptResolved { prompt_id: pr.id });
        }
        if had_prompts {
            tracing::info!(session = %self.id, "pending prompt abandoned (no clients before deadline) — denied");
            self.publish_presence();
            // `can_park()` may now be true — re-evaluate the park deadline.
            self.rearm_park();
        }
    }

    fn set_lifecycle(&mut self, l: SessionLifecycle) {
        self.lifecycle
            .store(l as u8, std::sync::atomic::Ordering::Release);
        self.emit(SessionEventWire::Lifecycle(l));
    }

    /// Save, close PTYs, drop `conv` THEN `runtime`. `background` (inbox
    /// watcher + per-session UDS + registry) stays: `synaps send` keeps
    /// resolving and its push into `event_queue` is the wake-up.
    pub(crate) async fn park(&mut self) -> std::ops::ControlFlow<EndReason> {
        let flow = self.park_now().await;
        self.publish_presence();
        // Restored to be parked but still live (it could not park): its
        // `on_session_start` runs now, as for any live session.
        if flow.is_continue()
            && !self.is_parked()
            && matches!(self.session_start, super::session_start::SessionStart::Pending)
        {
            self.queue_session_start();
        }
        flow
    }

    async fn park_now(&mut self) -> std::ops::ControlFlow<EndReason> {
        // Fired by a worker re-check (not the grace, not an explicit Park)?
        let recheck = self.park_deadline.is_some() && self.park_deadline == self.park_recheck_at;
        self.park_recheck_at = None;
        self.park_deadline = None;
        if self.is_parked() {
            if self.attached.is_empty() {
                tracing::info!(session = %self.id, "parked past the eviction age — leaving the map (journal kept)");
                return std::ops::ControlFlow::Break(EndReason::Evicted);
            }
            return std::ops::ControlFlow::Continue(());
        }
        if self.can_end_idle() {
            tracing::info!(session = %self.id, "idle with no history — ending instead of parking (F18)");
            // No journal will ever exist for this id: take the lock file
            // with us instead of leaving an orphan `.lock` (RC soak F-NEW-1).
            if let Some(lock) = self.session_lock.take() {
                lock.release_and_remove();
            }
            return std::ops::ControlFlow::Break(EndReason::Idle);
        }
        if !self.can_park() {
            // Re-arms a re-check while only workers hold the park.
            self.rearm_park();
            return std::ops::ControlFlow::Continue(());
        }
        if recheck {
            // The workers let go: now the reconnect grace runs, from here.
            // A re-check is not the user's grace having elapsed.
            if let Some(grace) = park_grace() {
                self.park_deadline = Some(tokio::time::Instant::now() + grace);
                return std::ops::ControlFlow::Continue(());
            }
        }
        self.state = AttachState::Parking;
        self.set_lifecycle(SessionLifecycle::Parking);
        // Parking drops the conversation from memory: everything queued must
        // be on disk first, and a failed save keeps the session live.
        if !self.save().await {
            tracing::warn!(session = %self.id, "park: save failed or timed out — staying live");
            self.state = AttachState::Detached { running: false };
            self.set_lifecycle(SessionLifecycle::Live);
            return std::ops::ControlFlow::Continue(());
        }
        if !self.journal_exists() {
            // Never park what cannot be restored (H2).
            tracing::warn!(session = %self.id, "park: no journal on disk — staying live");
            self.state = AttachState::Detached { running: false };
            self.set_lifecycle(SessionLifecycle::Live);
            return std::ops::ControlFlow::Continue(());
        }
        if self.runtime.session_manager().active_count() > 0 {
            self.emit(SessionEventWire::SystemNotice(
                "parked: background shells closed".into(),
            ));
        }
        self.runtime.session_manager().shutdown_all();
        self.turn_replay.clear();
        // The unparked runtime is built from `config.memory_mode`: a
        // revocation made without `/memory` must be recorded first.
        self.sync_memory_choice();
        // The runtime's exact tool activations outlive it: the unparked
        // runtime restores them, so the tools array (the start of the cached
        // prompt prefix) and the model's tools are what they were.
        self.parked_tool_activations = self.runtime.carried_tool_activations();
        // Drop order: conv first, then runtime — a panic between them can
        // never leave a Runtime without state to save.
        // Last look at the conversation for listings (title/cost/messages).
        self.publish_presence();
        let conv = self.conv.park_take();
        drop(conv);
        let runtime = self.runtime.park_take();
        drop(runtime);
        // Hand the freed pages back: jemalloc otherwise keeps them as
        // dirty/muzzy for its decay window, and a parked session that
        // still shows in RssAnon buys nothing.
        crate::core::memstat::purge_arenas();
        // F10: release the journal lock so an in-process --continue can
        // pick up the parked session (the daemon will fail to re-acquire
        // on unpark and surface a clear error instead of a zombie).
        self.session_lock = None;
        // A hook not called yet waits for the unpark, out of line.
        self.session_start.withdraw_for_park();
        self.state = AttachState::Parked;
        self.set_lifecycle(SessionLifecycle::Parked);
        tracing::info!(session = %self.id, "session parked");
        // The park timer is reused as the eviction timer: a session nobody
        // re-attaches to within `parked_evict_after` leaves the map (its
        // journal stays; `--continue` brings it back).
        self.park_deadline = parked_evict_after().map(|d| tokio::time::Instant::now() + d);
        std::ops::ControlFlow::Continue(())
    }

    /// Rebuild `runtime` + `conv` from the journal (`load_session_in_dir`
    /// via `resolve_session_and_prompt`), replay non-persisted settings,
    /// `finish_session_setup(IndexRecord::Skip)`. On error the session
    /// stays Parked (nothing is half-built).
    pub(crate) async fn unpark(&mut self) -> Result<()> {
        let started = std::time::Instant::now();
        // Somebody came back: the eviction deadline armed by `park()` is void.
        self.park_deadline = None;
        let journal_id = (**self.journal_id.load()).clone();

        // F10: re-acquire the journal lock before rebuilding the runtime.
        // If an in-process --continue took over while we were parked, fail
        // loudly so the attach gets a clear error instead of a zombie.
        {
            let dir = agent_core::session_lock::sessions_dir();
            let holder = agent_core::session_lock::LockHolder {
                pid: std::process::id(),
                kind: "daemon".to_string(),
            };
            match agent_core::session_lock::SessionLock::try_acquire(&dir, &journal_id, holder) {
                Ok(lock) => self.session_lock = Some(lock),
                Err(agent_core::session_lock::SessionLockError::Held { .. }) => {
                    return Err(crate::RuntimeError::Session(format!(
                        "cannot unpark session {}: journal locked by another process",
                        journal_id
                    )));
                }
                Err(e) => {
                    tracing::warn!(session = %journal_id, "unpark: session lock: {e}");
                    // Non-fatal: proceed without the lock (e.g. read-only fs).
                }
            }
        }

        let host = Arc::clone(&self.host);
        let cfg = self.config.clone();
        let queue = Arc::clone(&self.event_queue);
        let replay = self.settings_replay.clone();
        let journal_present = self.journal_exists();
        let view = self.view.load_full();
        let build = async move {
            let config: crate::SynapsConfig = (**host.config()).clone();
            // §4: cwd before apply_config — see `create`. The session's own
            // `/memory` choice, if it made one, instead of the default.
            let mut runtime = host
                .foreground_runtime_with(cfg.cwd.clone(), cfg.memory_mode.is_none())
                .await?;
            if let Some(mode) = cfg.memory_mode {
                runtime.restore_session_memory_mode(mode);
            }
            runtime.set_event_queue(queue);
            let mut sb = if journal_present {
                crate::engine::setup::resolve_session_and_prompt(
                    &mut runtime,
                    &Some(Some(journal_id)),
                    cfg.system.as_deref(),
                    cfg.prompt_manifest.as_deref(),
                )?
            } else {
                // The journal vanished under us (H2): rebuild an empty
                // conversation under the SAME id with the last-published
                // model/thinking rather than leaving a zombie Parked session.
                tracing::warn!(
                    session = %journal_id,
                    "unpark: journal missing — restoring as a fresh conversation"
                );
                let mut sb = crate::engine::setup::resolve_session_and_prompt(
                    &mut runtime,
                    &None,
                    cfg.system.as_deref(),
                    cfg.prompt_manifest.as_deref(),
                )?;
                sb.session.id = journal_id.clone();
                runtime.set_session_id(Some(journal_id));
                sb
            };
            runtime.set_cwd(cfg.cwd.clone());
            runtime.set_env(cfg.env.clone());
            runtime.set_env_stripped(cfg.env_stripped.clone());
            // The CURRENT model/thinking (the last published view), not
            // `cfg.model_override` frozen at create: `/model` survives park.
            runtime.set_model(view.model.clone());
            let _ = runtime.restore_session_reasoning(&view.thinking_level);
            sb.session.model = view.model.clone();
            sb.session.thinking_level = runtime.thinking_level().to_string();
            crate::engine::setup::finish_session_setup(
                &mut runtime,
                &config,
                &sb.session,
                cfg.cwd.clone(),
                crate::engine::setup::IndexRecord::Skip,
            );
            for s in &replay {
                replay_setting(&mut runtime, s);
            }
            Ok::<_, crate::RuntimeError>((runtime, sb))
        };
        let (runtime, sb) = match tokio::time::timeout(budgets::UNPARK_TIMEOUT, build).await {
            Ok(Ok(v)) => v,
            Ok(Err(e)) => return Err(e),
            Err(_) => {
                return Err(crate::RuntimeError::Session(format!(
                    "unpark timed out after {}s",
                    budgets::UNPARK_TIMEOUT_SECS
                )))
            }
        };
        let mut conv = ConversationState::from_resumed(sb.session);
        conv.api_messages = sb.api_messages;
        conv.total_input_tokens = sb.total_input_tokens;
        conv.total_output_tokens = sb.total_output_tokens;
        conv.session_cost = sb.session_cost;
        if journal_present && self.config.persist && self.session_lock.is_some() {
            crate::engine::setup::recover_turn_draft(&mut conv).await;
        }
        runtime.restore_tool_activations(std::mem::take(&mut self.parked_tool_activations));
        self.runtime.unpark_set(runtime);
        self.conv.unpark_set(conv);
        // A fresh runtime: whether its grant took effect is its own.
        self.memory_seen_active = false;
        self.observe_memory();
        self.publish_view().await;
        self.state = AttachState::Detached { running: false };
        self.set_lifecycle(SessionLifecycle::Live);
        self.update_attach_state();
        // A hook that waited for the unpark runs now (once per session).
        if matches!(self.session_start, super::session_start::SessionStart::Pending) {
            self.queue_session_start();
        }
        tracing::info!(session = %self.id, ms = started.elapsed().as_millis() as u64, "session unparked");
        Ok(())
    }

    /// Unpark if Parked. `Err` = still Parked; the caller reports it
    /// (`Refused` to a client, `AttachRefused` to an attacher, a warning
    /// for host wakes).
    pub(crate) async fn ensure_live(&mut self) -> std::result::Result<(), String> {
        if !self.is_parked() {
            return Ok(());
        }
        let res = self
            .unpark()
            .await
            .map_err(|e| format!("session parked and could not be restored: {e}"));
        self.publish_presence();
        res
    }

    /// `stream_handler.rs:264`: a turn OR a compaction blocks auto-turns.
    pub(crate) fn busy(&self) -> bool {
        self.streaming || self.compact.is_some()
    }

    /// `runtime.subagent_registry().display_rows()` → `SubagentRows`, only
    /// when there is something to show (the TUI's reconcile is a no-op on
    /// an empty registry).
    ///
    /// The 1 Hz tick that drives this starts with a turn and outlives it
    /// while a background worker (`subagent_start`) is still running: those
    /// workers' progress events rode the ended turn's stream, so these rows
    /// (which carry each worker's step and tool count) are how clients keep
    /// up. The tick stops once no worker is running after the turn, having
    /// published the terminal rows.
    pub(crate) fn publish_subagent_rows(&mut self) {
        let rows = self
            .runtime
            .subagent_registry()
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .display_rows();
        let running = rows
            .iter()
            .any(|r| matches!(r.status, crate::runtime::subagent::SubagentStatus::Running));
        if !rows.is_empty() {
            self.emit(SessionEventWire::SubagentRows(rows));
        }
        if !keep_subagent_tick(self.streaming, running) {
            self.subagent_tick = None;
            // The last worker finished: a park it held may be due now.
            self.rearm_park();
        }
    }

    /// Whether any worker in the registry is still running.
    fn subagents_running(&self) -> bool {
        self.runtime
            .subagent_registry()
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .display_rows()
            .iter()
            .any(|r| matches!(r.status, crate::runtime::subagent::SubagentStatus::Running))
    }

    /// Codex hosted web_search is read from this (engine/daemon) process's
    /// env, so say so in the session — once per session per model/state,
    /// inside the turn so its replay ring carries it to clients attaching
    /// mid-turn.
    fn announce_web_search_state(&mut self) {
        let notice = crate::runtime::openai::codex_wire::session_web_search_notice(
            self.runtime.model(),
            crate::runtime::openai::codex_wire::engine_env,
        );
        if notice.is_some() && notice != self.web_search_notice {
            self.web_search_notice = notice.clone();
            if let Some(text) = notice {
                self.emit(SessionEventWire::Stream(StreamEvent::Session(
                    SessionEvent::Notice(text),
                )));
            }
        }
    }

    // ── turn start (dispatch.rs Submit tail / stream_handler.rs RunTurn) ──

    /// This session's `on_session_start` task ended: the hook ran (or the
    /// task failed, which counts as done: fail-open like a hook timeout).
    /// Refresh the view, which shows the injection, then send what waited
    /// for it: a started turn's request. A due keep-warm ping waited with its
    /// timer off (`holds`); the loop's next pass arms it again.
    async fn session_start_finished(
        &mut self,
        ran: std::result::Result<bool, tokio::task::JoinError>,
    ) {
        use super::session_start::SessionStart;
        match ran {
            Ok(true) => {}
            // Withdrawn before it was called (a park): not run yet. Anything
            // waiting for it puts it back in line.
            Ok(false) => {
                self.session_start = SessionStart::Pending;
                if self.turn_awaiting_start.is_some() || self.cache_keepwarm.deadline().is_some() {
                    self.session_start_ready();
                }
                return;
            }
            Err(e) => tracing::warn!(session = %self.id, "on_session_start task failed: {e}"),
        }
        self.session_start = SessionStart::Done;
        if self.runtime.is_live() {
            self.publish_view().await;
        }
        if let Some(start) = self.turn_awaiting_start.take() {
            // A turn canceled from outside (a driver revocation) ends on its
            // own path (`finish_revoked_turn`) with nothing sent.
            if !start.cancel.is_cancelled() {
                self.open_turn_stream(start).await;
            }
        }
    }

    /// No turn and no keep-warm ping goes out before `on_session_start` has
    /// run. Its injection is part of their system prompt, so a request sent
    /// before it would lack the extension context and the next request would
    /// change the cached system prompt. (A manual `/compact` does not wait:
    /// it builds its own prompt; a rolling summary reuses the request of the
    /// turn before it or builds its own prompt.) A session created with
    /// `await_extensions=false` (quick start, every session a reload
    /// restores) has a task in line at the host's gate for it, or, while
    /// parked, nothing in line until the unpark.
    ///
    /// `true` once the hook has run. Otherwise it puts the hook in line if
    /// it is not (after a park) and moves it ahead of background hooks; the
    /// caller waits for `session_start_finished` without blocking the loop.
    /// The hook itself always runs in that task, through the gate, so it is
    /// never called twice or beside another session's.
    pub(crate) fn session_start_ready(&mut self) -> bool {
        use super::session_start::SessionStart;
        if matches!(self.session_start, SessionStart::Pending) {
            self.queue_session_start();
        }
        match &self.session_start {
            SessionStart::Done => true,
            SessionStart::Queued { ticket, .. } => {
                ticket.prefer();
                false
            }
            SessionStart::Pending => false,
        }
    }

    /// Put this session's `on_session_start` in line (in the background).
    fn queue_session_start(&mut self) {
        self.session_start = super::session_start::SessionStart::queue(
            &self.host,
            Arc::clone(&self.hook_bus),
            Arc::clone(&self.journal_id),
            extensions_ready_timeout(&self.host.config()),
        );
    }

    /// Open the turn's stream: the request goes out.
    async fn open_turn_stream(&mut self, start: TurnStart) {
        // A turn started by events (nobody typed it) runs under the event
        // turn budget for this turn only; the stream copies the budget when
        // it starts, so the session budget is restored right after. Here and
        // not in `start_turn`: a turn whose request waited for
        // `on_session_start` opens its stream later, under its own budget.
        let session_budget = start.event_turn.then(|| {
            let event_budget = self
                .runtime
                .turn_budget()
                .for_event_turn(&self.host.config().turn_budgets);
            let session_budget = self.runtime.turn_budget().clone();
            self.runtime.set_turn_budget(event_budget);
            self.runtime
                .set_owner_steer_budget(self.owner_steer_budget.clone());
            session_budget
        });
        let stream = self
            .runtime
            .run_stream_tracked(
                self.conv.api_messages.clone(),
                start.cancel,
                Some(start.steer_rx),
                Some(self.secret_prompt_handle.clone()),
                start.auto_approve,
                start.completion,
            )
            .await;
        if let Some(session_budget) = session_budget {
            self.runtime.set_turn_budget(session_budget);
            self.runtime.set_owner_steer_budget(None);
        }
        self.stream = Some(stream);
    }

    /// Start a turn. Its request goes out at once, or, while this session's
    /// `on_session_start` has not run, once it has (`session_start_ready`);
    /// the turn is live from here either way.
    pub(crate) async fn start_turn(&mut self, trigger: TurnTrigger, user_text: Option<String>) {
        self.cache_keepwarm_turn_started();
        let ct = CancellationToken::new();
        let (s_tx, s_rx) = mpsc::unbounded_channel::<String>();
        self.streaming = true;
        self.turn_baseline = self.conv.api_messages.len();
        self.open_turn_draft();
        self.turn_steered_events.clear();
        self.turn_replay.clear();
        self.update_attach_state();
        self.emit(SessionEventWire::TurnStarted {
            turn_baseline: self.turn_baseline,
            trigger,
            user_text,
        });
        self.announce_web_search_state();
        // Blocks command processing during setup exactly like the TUI loop.
        // (E-P7, §S9) An armed driver MUST NOT auto-approve tool activation —
        // the session-driver contract keeps ordinary tool-approval gates in
        // force. Override the session config to false whenever a grant is
        // armed, regardless of what the client requested. (Driver-initiated
        // turns already pass `false` from the tick's Prepared arm; this guards
        // any foreground turn that starts while armed.)
        let auto_approve = self.config.auto_approve_confirms && self.driver.is_none();
        // Stamp this turn's worker launches before the stream can start any.
        // A grant made since the last turn (e.g. while the session was
        // being rebuilt) is seen before this turn can revoke it.
        self.observe_memory();
        self.turn_worker_generation = self
            .runtime
            .subagent_registry()
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .begin_turn();
        let completion = crate::runtime::TurnCompletion::new();
        self.turn_completion = Some(completion.clone());
        self.cancel = Some(ct.clone());
        self.steer_tx = Some(s_tx);
        let event_turn = trigger == TurnTrigger::EventAuto;
        self.owner_steer_budget = event_turn.then(Default::default);
        let mut tick = tokio::time::interval(std::time::Duration::from_secs(1));
        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        tick.reset();
        self.subagent_tick = Some(tick);
        let start = TurnStart {
            cancel: ct,
            steer_rx: s_rx,
            auto_approve,
            completion,
            event_turn,
        };
        if self.session_start_ready() {
            self.open_turn_stream(start).await;
        } else {
            tracing::debug!(session = %self.id, "turn waits for on_session_start");
            self.turn_awaiting_start = Some(start);
            // Quick Start's first turn can wait here for extension discovery
            // (`startup.extensions_ready_timeout_secs`) and then the hook:
            // say so, or the silence looks like a hang.
            self.emit(SessionEventWire::SystemNotice(
                "waiting for extensions to start (on_session_start) before sending...".into(),
            ));
        }
    }

    /// Every turn-end path (Done/Error/Cancel/stream EOF). `streaming=false`
    /// makes a `Cancel` racing a `Done` the idle no-op. Events steered into
    /// the stream that the engine never took (`turn_steered_events`) go back
    /// to `pending_events`, first: they are flushed into history like any
    /// event that arrived during the turn (the interrupted path takes them
    /// earlier, to place them after the marker). Queues the save of the
    /// turn's final history, then the draft removal (`close_turn_draft`).
    pub(crate) fn clear_stream(&mut self) {
        self.stream = None;
        self.turn_awaiting_start = None;
        self.cancel = None;
        self.steer_tx = None;
        self.owner_steer_budget = None;
        self.streaming = false;
        self.turn_completion = None;
        self.revoked_turn_deadline = None;
        // Background workers outlive the turn: keep publishing their rows
        // until the last one finishes (see `publish_subagent_rows`).
        if !keep_subagent_tick(false, self.subagents_running()) {
            self.subagent_tick = None;
        }
        if !self.turn_steered_events.is_empty() && self.conv.is_live() {
            let undelivered = std::mem::take(&mut self.turn_steered_events);
            self.conv.pending_events.splice(0..0, undelivered);
        }
        // The history that ends the turn — whether or not a draft is open
        // (no draft without the session lock) — then the draft removal.
        self.request_save();
        self.close_turn_draft();
        self.update_attach_state();
    }

    // ── in-flight turn draft (crash recovery) ──────────────────────────

    /// Turn start: the draft's existence is the "turn open" signal a
    /// loader uses to detect a process that died mid-turn. Only the session
    /// lock holder writes one: without the lock another process may own
    /// this session, and a draft is a claim that a turn is running HERE.
    fn open_turn_draft(&mut self) {
        let persist =
            self.config.persist && self.conv.is_live() && self.session_lock.is_some();
        self.turn_draft = TurnDraftState {
            open: persist.then(|| self.conv.session.id.clone()),
            base_len: self.conv.api_messages.len(),
            text: String::new(),
            tools: Vec::new(),
            dirty: true,
        };
        self.flush_turn_draft();
    }

    fn turn_draft_text(&mut self, delta: &str) {
        let d = &mut self.turn_draft;
        if d.text.len() < agent_core::core::session_draft::TURN_DRAFT_MAX_TEXT_BYTES {
            d.text.push_str(delta);
            d.dirty = true;
        }
    }

    fn turn_draft_reset_text(&mut self) {
        let d = &mut self.turn_draft;
        d.tools.clear();
        if !d.text.is_empty() {
            d.text.clear();
            d.dirty = true;
        }
    }

    /// A round checkpoint was adopted and saved: later text continues from it.
    fn turn_draft_committed(&mut self) {
        self.turn_draft.base_len = self.conv.api_messages.len();
        self.turn_draft.text.clear();
        self.turn_draft.tools.clear();
        self.turn_draft.dirty = true;
        self.flush_turn_draft();
    }

    /// A context rollover replaced the history head, and `base_len` no
    /// longer refers to it. Until the engine's next round checkpoint
    /// re-bases the draft, nothing streamed can be placed: an impossible
    /// length makes the cancel fallback (`keep_streamed_partial`) and crash
    /// recovery (`recover_crashed_turn`) both skip the text.
    fn turn_draft_rolled_over(&mut self) {
        self.turn_draft.base_len = usize::MAX;
        self.turn_draft.text.clear();
        self.turn_draft.tools.clear();
        self.turn_draft.dirty = true;
        self.flush_turn_draft();
    }

    /// Follow the in-flight response for the draft: its text, a new
    /// response or retry starting over, the tools it calls. Shared by the
    /// stream handler and the cancel drain, so text the actor only receives
    /// while draining (already shown to clients) is tracked too.
    fn observe_turn_draft(&mut self, event: &crate::LlmEvent) {
        match event {
            crate::LlmEvent::Text(text) => self.turn_draft_text(text),
            crate::LlmEvent::ResponseStart | crate::LlmEvent::ResponseReset => {
                self.turn_draft_reset_text()
            }
            crate::LlmEvent::ToolUse { tool_name, .. } => {
                self.turn_draft.tools.push(tool_name.clone())
            }
            _ => {}
        }
    }

    /// Write the draft if it changed (1 Hz turn tick; round checkpoints).
    /// Non-blocking and ordered (`persister`).
    pub(crate) fn flush_turn_draft(&mut self) {
        let Some(open) = self.turn_draft.open.clone() else {
            return;
        };
        if !self.turn_draft.dirty {
            return;
        }
        // Defensive: follow the conversation if its id changed mid-turn.
        let id = self.conv.session.id.clone();
        if open != id {
            self.persister.remove_draft(&open);
            self.turn_draft.open = Some(id.clone());
        }
        self.turn_draft.dirty = false;
        self.persister.write_draft(
            &id,
            agent_core::core::session_draft::TurnDraft {
                base_len: self.turn_draft.base_len,
                partial_text: self.turn_draft.text.clone(),
            },
        );
    }

    /// Turn end (every path goes through `clear_stream`, which has just
    /// queued the save of the history that ends the turn): remove the draft.
    /// Queued after that save, and the persister never removes a draft while
    /// its session's latest save has failed, so the draft cannot disappear
    /// before the history that concludes its turn is on disk.
    fn close_turn_draft(&mut self) {
        if let Some(id) = self.turn_draft.open.take() {
            self.persister.remove_draft(&id);
        }
        self.turn_draft = TurnDraftState::default();
    }

    /// dispatch.rs Submit (:1231-1288) minus presentation.
    pub(crate) async fn submit(
        &mut self,
        text: String,
        attachments: Vec<serde_json::Value>,
        from: Option<ClientId>,
    ) {
        // Wall 1 defense-in-depth: a latched (unverified) context head must
        // not accept new inference. The stream would refuse via
        // `durability_blocked` anyway, but that leaves the user message
        // orphaned in history; refuse here, before it is pushed.
        if self.conv.context_head.is_blocked(&self.conv.session) {
            let reason = "context head is unverified after a failed checkpoint; reload the \
                          session (`--continue`) or start a new one before continuing"
                .to_string();
            // Refused to the submitter (its editor text comes back and it
            // stops expecting this turn), a notice for everyone else.
            match from {
                Some(client) => self.emit(SessionEventWire::Refused {
                    client,
                    command: "submit".into(),
                    reason,
                }),
                None => self.emit(SessionEventWire::SystemNotice(reason)),
            }
            return;
        }
        if self.streaming {
            // A Submit while streaming is what the TUI calls StreamingInput.
            // Attachments during streaming are rejected by the client; if they
            // arrive anyway, ignore them (text-only steer).
            // P6: if the driver is armed and streaming, steer AND update
            // the driver's steering FIFO.
            if let Some(driver) = self.driver.as_mut() {
                let byte_total: usize = driver.steering.iter().map(|s| s.len()).sum();
                if driver.steering.len() >= super::driver::STEERING_MAX_MESSAGES {
                    self.emit(SessionEventWire::SystemNotice(
                        "steering queue full (16 messages) — message dropped".into(),
                    ));
                    return;
                }
                if byte_total + text.len() > super::driver::STEERING_MAX_BYTES {
                    self.emit(SessionEventWire::SystemNotice(
                        "steering queue full (256 KiB) — message dropped".into(),
                    ));
                    return;
                }
                driver.steering.push_back(text.clone());
            }
            self.steer(text);
            return;
        }
        // P6: while the driver is armed and idle, route to steering FIFO
        // instead of starting a normal turn.
        if self.driver.is_some() {
            let driver = self.driver.as_mut().unwrap();
            let byte_total: usize = driver.steering.iter().map(|s| s.len()).sum();
            if driver.steering.len() >= super::driver::STEERING_MAX_MESSAGES {
                self.emit(SessionEventWire::SystemNotice(
                    "steering queue full (16 messages) — message dropped".into(),
                ));
                return;
            }
            if byte_total + text.len() > super::driver::STEERING_MAX_BYTES {
                self.emit(SessionEventWire::SystemNotice(
                    "steering queue full (256 KiB) — message dropped".into(),
                ));
                return;
            }
            driver.steering.push_back(text.clone());
            // Clear auto_wakes_blocked on explicit user submit (user takeover).
            driver.auto_wakes_blocked = false;
            self.emit(SessionEventWire::Steered {
                text,
                delivered: false,
            });
            return;
        }
        if self.compact.is_some() {
            // dispatch.rs:1233-1237: queued until the compaction lands.
            self.emit(SessionEventWire::Steered {
                text: text.clone(),
                delivered: false,
            });
            self.conv.queued_message = Some(text);
            return;
        }
        // Real user send — reset auto-turn counter.
        self.consecutive_auto_turns = 0;
        // For every OTHER attached client: the submitter drew its own card.
        let user_text = text.clone();
        let api_content = text;

        if attachments.is_empty() {
            // Text-only: existing path.
            self.conv.api_messages.push(std::sync::Arc::new(
                serde_json::json!({"role": "user", "content": api_content}),
            ));
        } else {
            // Build multipart content: text block (if non-empty) + attachment blocks.
            let mut blocks = Vec::with_capacity(attachments.len() + 1);
            if !api_content.is_empty() {
                blocks.push(serde_json::json!({"type": "text", "text": api_content}));
            }
            blocks.extend(attachments);
            let candidate = std::sync::Arc::new(
                serde_json::json!({"role": "user", "content": blocks}),
            );
            // Validate the complete history (including this message) for the
            // session's current model before accepting.
            let mut proposed = self.conv.api_messages.clone();
            proposed.push(candidate.clone());
            let model = self.runtime.model().to_string();
            if let Err(e) = crate::runtime::attachments::validate_history_for_request(&model, &proposed) {
                // Typed refusal: the submitting client gets its editor text
                // back (stream_handler restores `last_submitted` on `Refused`)
                // and keeps its attachment drafts; mirrors see a notice.
                match from {
                    Some(client) => self.emit(SessionEventWire::Refused {
                        client,
                        command: "submit".into(),
                        reason: format!("attachments rejected: {e}"),
                    }),
                    None => self.emit(SessionEventWire::SystemNotice(format!(
                        "attachments rejected: {e}"
                    ))),
                }
                return;
            }
            self.conv.api_messages.push(candidate);
        }
        self.start_turn(TurnTrigger::User, Some(user_text)).await;
    }

    /// dispatch.rs StreamingInput plain-text branch (:1369-1378).
    ///
    /// Only the input owner steers (`is_input_command`). In an event-started
    /// turn, the owner's message also moves the turn onto the session budget
    /// (`self.runtime.turn_budget()`, which a `/budget` changes) once the
    /// stream delivers it. A steer that misses the turn (its stream ended)
    /// stays queued and starts a typed turn, on that budget anyway.
    pub(crate) fn steer(&mut self, text: String) {
        let send = |tx: &mpsc::UnboundedSender<String>| tx.send(text.clone()).is_ok();
        let delivered = match (self.steer_tx.as_ref(), self.owner_steer_budget.as_deref()) {
            (Some(tx), Some(lift)) => {
                lift.send_owner_steer(self.runtime.turn_budget().clone(), || send(tx))
            }
            (Some(tx), None) => send(tx),
            (None, _) => false,
        };
        self.emit(SessionEventWire::Steered {
            text: text.clone(),
            delivered,
        });
        self.conv.queued_message = Some(text);
    }

    /// `SubagentSteer`: hand the user's text to one of this session's running
    /// workers, then note it in the MAIN conversation so the main agent knows
    /// the user redirected its worker. The note travels like a drained event
    /// (steered into a live turn, buffered while busy, injected while idle)
    /// but never starts a turn on its own; clients render it from `External`.
    pub(crate) fn steer_subagent(&mut self, from: Option<ClientId>, handle_id: &str, text: String) {
        let steered = {
            let registry = self
                .runtime
                .subagent_registry()
                .lock()
                .unwrap_or_else(|p| p.into_inner());
            match registry.get(handle_id) {
                None => Err(format!("no subagent {handle_id} in this session")),
                Some(h) => {
                    let running = h
                        .state
                        .read()
                        .unwrap_or_else(|p| p.into_inner())
                        .public_status()
                        == crate::runtime::subagent::SubagentStatus::Running;
                    if running {
                        h.steer(&text).map(|()| h.agent_name.clone())
                    } else {
                        Err(format!("{} ({handle_id}) is not running", h.agent_name))
                    }
                }
            }
        };
        let agent_name = match steered {
            Ok(name) => name,
            Err(reason) => {
                match from {
                    Some(c) => self.emit(SessionEventWire::Refused {
                        client: c,
                        command: "subagent_steer".to_string(),
                        reason,
                    }),
                    None => self.emit(SessionEventWire::SystemNotice(reason)),
                }
                return;
            }
        };
        self.note_in_main_conversation(subagent_steer_event(handle_id, &agent_name, &text));
    }

    /// Tell the MAIN agent something the user did to its workers: the note
    /// travels like a drained event (steered into a live turn, buffered while
    /// busy, injected while idle) but never starts a turn on its own; clients
    /// render it from `External`.
    pub(crate) fn note_in_main_conversation(&mut self, event: crate::events::types::Event) {
        let formatted = crate::events::format_event_for_agent(&event);
        if self.busy() {
            let into_turn = self
                .steer_tx
                .as_ref()
                .is_some_and(|tx| tx.send(formatted.clone()).is_ok());
            if into_turn {
                self.turn_steered_events.push(formatted);
            } else {
                self.conv.pending_events.push(formatted);
            }
        } else {
            self.conv
                .api_messages
                .push(std::sync::Arc::new(serde_json::json!({
                    "role": "user",
                    "content": formatted
                })));
            self.request_save();
        }
        self.emit(SessionEventWire::External(event));
        self.emit_conversation();
    }

    /// Every cancel path — `Cancel` (Esc), quit mid-turn (`finish`), the cost
    /// cap, a reload/host checkpoint — behind the TUI's `if streaming` guard
    /// (input.rs:350): a `Cancel` while idle is a no-op that only re-announces
    /// `Idle` — it must never touch history, save, or emit `Aborted`.
    ///
    /// The interrupted turn is recorded as REAL history, never as a recap:
    /// 1. the token is cancelled, then pending host prompts are answered
    ///    `None` (a tool or `before_tool_call` hook blocked on one must unwind);
    /// 2. the still-live stream is drained for a
    ///    bounded time (`drain_cancelled_stream`) so the engine's cancel-path
    ///    history — partial assistant message, completed tool rounds,
    ///    delivered steering, canceled `tool_result`s — plus its final Usage
    ///    and any in-flight context-head checkpoint reach the actor;
    /// 3. that history is adopted verbatim (else the last adopted history is
    ///    kept — every history the engine publishes is valid), then
    ///    `finish_interrupted_turn` APPENDS one interruption marker
    ///    (`engine::interrupt`). Nothing already sent is edited, so the
    ///    provider's cached prefix survives.
    ///
    /// A turn the engine had already finished when the cancel reached it
    /// (Esc a moment after the answer ended) is not interrupted: no marker.
    pub(crate) async fn cancel_turn(&mut self, reason: crate::engine::interrupt::InterruptReason) {
        if !self.streaming {
            if self.compact.is_some() {
                self.abort_compaction();
                if let Some(q) = self.conv.queued_message.take() {
                    self.emit(SessionEventWire::Dequeued { text: q });
                }
                self.update_attach_state();
            }
            self.emit(SessionEventWire::Idle);
            return;
        }
        self.note_cancel_cause(reason);
        if let Some(ref ct) = self.cancel {
            ct.cancel();
        }
        // After the cancel, so a tool awaiting a prompt observes the cancel
        // (not a `None` it could read as "declined"); a `before_tool_call`
        // hook awaiting a confirm prompt is not cancel-aware and needs this
        // answer to unwind at all.
        self.resolve_pending_prompts();
        let drain = self.drain_cancelled_stream().await;
        match drain.history {
            Some(history) => self.conv.api_messages = history,
            None => {
                if !drain.closed {
                    tracing::warn!(
                        session = %self.id,
                        budget_ms = budgets::CANCEL_DRAIN_TIMEOUT_MS,
                        "cancelled turn did not publish its history within the drain budget; \
                         keeping the last adopted history plus the streamed partial reply"
                    );
                }
                self.keep_streamed_partial(reason);
            }
        }
        // Complete only if the drain consumed the whole stream: the final
        // history is published right after the completion is recorded.
        let completed = drain.closed && self.turn_completed();
        self.finish_interrupted_turn(reason, completed).await;
    }

    /// An interrupted turn whose engine never published its post-cancel
    /// history (the drain budget ran out, or the stream ended without it):
    /// keep the reply text clients were already shown, as a real assistant
    /// message, so history matches the screen and the marker that follows
    /// doesn't merge into the next prompt. Same rule as crash recovery
    /// (`engine::interrupt::recover_crashed_turn`): only when the history is
    /// still exactly where that text began (`base_len`), never a stale one,
    /// and right after a user message. Text only, so it can't leave a
    /// `tool_use` without its result: an in-flight tool call and unsigned
    /// thinking are dropped. (Rebuilding the call isn't safe: with extended
    /// thinking a `tool_use` turn needs its signed thinking block.) A tool
    /// the response called instead gets a host-authored note after it
    /// (`engine::interrupt::tool_in_flight_note_message`), worded like the
    /// engine's canceled `tool_result`, so the next turn doesn't blindly
    /// repeat a non-idempotent action; clients get the same note. The
    /// engine's late publication is not applied: its stream is gone with the
    /// drain.
    fn keep_streamed_partial(&mut self, reason: crate::engine::interrupt::InterruptReason) {
        let d = &self.turn_draft;
        let follows_user = self.conv.api_messages.last().is_some_and(|m| m["role"] == "user");
        if d.base_len != self.conv.api_messages.len() || !follows_user {
            return;
        }
        let (text, tools) = (d.text.clone(), d.tools.clone());
        if !text.trim().is_empty() {
            self.conv.api_messages.push(std::sync::Arc::new(serde_json::json!({
                "role": "assistant",
                "content": [{"type": "text", "text": text}]
            })));
        }
        if !tools.is_empty() {
            let note = crate::engine::interrupt::tool_in_flight_note_message(reason, &tools);
            let text = note["content"].as_str().unwrap_or_default().to_owned();
            self.conv.api_messages.push(note);
            self.emit(SessionEventWire::SystemNotice(text));
        }
    }

    /// The engine recorded the running turn's normal end (`TurnCompletion`).
    fn turn_completed(&self) -> bool {
        self.turn_completion.as_ref().is_some_and(|c| c.completed())
    }

    /// Tell the engine why the running turn is being cancelled, BEFORE its
    /// token is: its canceled tool results say so ("Canceled by user" only
    /// for the user). First cause wins, so a teardown that revokes a driver
    /// and then cancels notes its own cause first.
    fn note_cancel_cause(&self, reason: crate::engine::interrupt::InterruptReason) {
        if !self.streaming {
            return;
        }
        if let (Some(turn), Some(cause)) = (&self.turn_completion, reason.cancel_cause()) {
            turn.note_cancel_cause(cause);
        }
    }

    /// The ONE tail of every interrupted turn, once its stream has been
    /// drained or dropped (`cancel_turn`, a revoked driver turn): record the
    /// interruption in history, end the turn, tell clients. Never runs the
    /// post-turn machinery of a normal end (queued auto-send, event
    /// auto-turns, auto-compaction).
    ///
    /// `completed`: the engine had finished the turn normally before the
    /// cancel reached it — the history is a complete answer, so no marker is
    /// appended and clients get the `Done` they would have had.
    async fn finish_interrupted_turn(
        &mut self,
        reason: crate::engine::interrupt::InterruptReason,
        completed: bool,
    ) {
        // Defensive: only trailing invalid messages this turn appended.
        crate::engine::stream::repair_history_after_failure(
            &mut self.conv.api_messages,
            self.turn_baseline,
        );
        let kept_partial = self.conv.api_messages.len() > self.turn_baseline;
        if !completed {
            crate::engine::interrupt::append_marker(&mut self.conv.api_messages, reason);
        }
        // A user steer the engine never picked up was not delivered.
        if let Some(q) = self.conv.queued_message.take() {
            self.emit(SessionEventWire::Dequeued { text: q });
        }
        // Events that arrived during the turn go in after the marker: first
        // those steered into the stream but never drained by the engine
        // (otherwise lost with the channel), then those buffered.
        {
            let undelivered = std::mem::take(&mut self.turn_steered_events);
            let conv: &mut ConversationState = &mut self.conv;
            for formatted in undelivered.into_iter().chain(conv.pending_events.drain(..)) {
                conv.api_messages.push(std::sync::Arc::new(serde_json::json!({
                    "role": "user",
                    "content": formatted
                })));
            }
        }
        // Queues the save of this history, then the draft removal.
        self.clear_stream();
        // Cancel running reactive subagents; recover a poisoned guard rather
        // than skip cancellation. A user interrupt (Esc) stops the turn it
        // interrupted and the workers THAT turn launched; reactive workers
        // started by earlier turns are background work and keep running
        // (`/workers cancel` stops them). Every other reason — session
        // teardown, cost cap, host checkpoint, driver revocation — stops all.
        {
            use crate::runtime::subagent::WorkerCancelScope;
            let ending = self.lifecycle.load(std::sync::atomic::Ordering::Acquire)
                == SessionLifecycle::Ending as u8;
            let scope = worker_cancel_scope(reason, ending, self.turn_worker_generation);
            let mut registry = match self.runtime.subagent_registry().lock() {
                Ok(g) => g,
                Err(poisoned) => {
                    tracing::warn!(
                        "subagent registry mutex poisoned during abort; recovering to cancel running handles"
                    );
                    poisoned.into_inner()
                }
            };
            let cancelled = registry.cancel_running(scope);
            if !cancelled.is_empty() {
                tracing::info!(
                    ?scope,
                    workers = ?cancelled,
                    "turn interrupted: cancelled running workers"
                );
            }
            if let WorkerCancelScope::Turn(_) = scope {
                // Still running and staying so: not the ones just cancelled
                // (they read as running until their next step).
                let spared = registry
                    .display_rows()
                    .iter()
                    .filter(|r| {
                        r.status == crate::runtime::subagent::SubagentStatus::Running
                            && !r.cancel_requested
                    })
                    .count();
                drop(registry);
                if spared > 0 {
                    self.emit(SessionEventWire::Stream(StreamEvent::Session(
                        SessionEvent::Notice(format!(
                            "{spared} background worker{} from earlier turns still running — /workers to list, /workers cancel all to stop",
                            if spared == 1 { "" } else { "s" }
                        )),
                    )));
                }
            }
        }
        if completed {
            self.emit(SessionEventWire::Stream(StreamEvent::Session(SessionEvent::Done)));
        } else {
            // Typed event. `context_saved` (wire name kept for compatibility)
            // now means "the turn's partial work is kept in history".
            self.emit(SessionEventWire::Aborted {
                context_saved: kept_partial,
            });
        }
        self.emit_conversation();
        self.announce_idle().await;
    }

    /// A turn whose token was cancelled from OUTSIDE `cancel_turn` — a
    /// driver revocation drops `DriverState`, cancelling the driver turn's
    /// child token (the grant's deadline task cancels that token too; the
    /// next `driver_tick` then revokes and arms the deadline) — ends here: on its `Done`, its stream's end, or when
    /// `revoked_turn_deadline` passes (a tool that ignores the cancel must
    /// not keep the session busy forever; the same budget as
    /// `drain_cancelled_stream`). What is left of the stream is dropped.
    async fn finish_revoked_turn(&mut self, stream_closed: bool) {
        if !self.streaming {
            return;
        }
        let completed = stream_closed && self.turn_completed();
        self.stream = None;
        if !completed {
            self.keep_streamed_partial(crate::engine::interrupt::InterruptReason::Driver);
        }
        self.finish_interrupted_turn(crate::engine::interrupt::InterruptReason::Driver, completed)
            .await;
    }

    /// Whether the running turn's token has been cancelled (by `cancel_turn`
    /// or a driver revocation).
    fn turn_cancelled(&self) -> bool {
        self.streaming && self.cancel.as_ref().is_some_and(|ct| ct.is_cancelled())
    }

    /// Answer every pending host prompt `None` (same as `checkpoint`/`finish`).
    fn resolve_pending_prompts(&mut self) {
        if self.pending_prompts.is_empty() {
            return;
        }
        while let Some((pr, tx)) = self.pending_prompts.pop_front() {
            let _ = tx.send(None);
            self.emit(SessionEventWire::PromptResolved { prompt_id: pr.id });
        }
        self.publish_presence();
        self.rearm_prompt_abandon();
    }

    /// Consume what a CANCELLED turn's stream still carries, for at most
    /// `budgets::CANCEL_DRAIN_TIMEOUT`, then drop it.
    ///
    /// Deliberately NOT `on_stream_event`: that handler would re-check the
    /// cost cap on the final `Usage` (recursive `cancel_turn`), run the
    /// failure path on the engine's typed `Canceled` error, and treat `Done`
    /// as a normal completion (queued auto-send, auto-compaction).
    ///
    /// Forwarded to clients: display events (partial text, tool results —
    /// the history being adopted contains them), `Usage`, agent events.
    /// Handled: `MessageHistory` (kept, last wins), `ContextHeadCheckpoint`
    /// (serviced — dropping its receipt would latch `durability_blocked`;
    /// the head it installs replaces any history kept before it),
    /// `SteeringDelivered`. Swallowed: `Done`, the `Canceled` error. A typed
    /// `InterruptedAfterSideEffect` becomes a notice (it is not a failure of
    /// the turn; the canceled `tool_result` already tells the model). A tool
    /// raising a host prompt before it observed the cancel is answered
    /// `None` at once, not at the deadline.
    async fn drain_cancelled_stream(&mut self) -> CancelDrain {
        let mut out = CancelDrain::default();
        let Some(mut stream) = self.stream.take() else {
            out.closed = true;
            return out;
        };
        let deadline = tokio::time::Instant::now() + budgets::CANCEL_DRAIN_TIMEOUT;
        loop {
            let event = tokio::select! {
                biased;
                Some(req) = self.secret_prompt_rx.recv() => {
                    let _ = req.response_tx.send(None);
                    continue;
                }
                next = tokio::time::timeout_at(deadline, stream.next()) => match next {
                    Err(_) => break,
                    Ok(None) => {
                        out.closed = true;
                        break;
                    }
                    Ok(Some(event)) => event,
                },
            };
            match event {
                StreamEvent::Session(SessionEvent::Done) => {
                    out.closed = true;
                    break;
                }
                StreamEvent::Session(SessionEvent::MessageHistory(history)) => {
                    out.history = Some(history);
                }
                StreamEvent::Session(SessionEvent::ContextHeadCheckpoint {
                    session_id,
                    messages,
                    receipt,
                }) => {
                    self.handle_context_head_checkpoint(session_id, messages, receipt)
                        .await;
                    // The adopted head is the actor's history now: a history
                    // kept from BEFORE it is stale and must not override it
                    // (a later `MessageHistory`, if any, still wins).
                    out.history = None;
                }
                StreamEvent::Session(SessionEvent::Error(err)) => match err.outcome {
                    crate::TurnOutcome::Canceled => {}
                    crate::TurnOutcome::InterruptedAfterSideEffect { .. } => {
                        self.emit(SessionEventWire::SystemNotice(err.message));
                    }
                    _ => {
                        tracing::warn!(
                            session = %self.id,
                            category = err.category_label(),
                            "cancelled turn reported an error while draining"
                        );
                    }
                },
                StreamEvent::Agent(AgentEvent::SteeringDelivered { ref message }) => {
                    self.note_steering_delivered(message);
                    self.emit(SessionEventWire::Stream(event));
                }
                StreamEvent::Session(SessionEvent::Usage { .. }) => {
                    self.record_usage(&event);
                    self.emit(SessionEventWire::Stream(event));
                }
                StreamEvent::Llm(ref llm) => {
                    self.observe_turn_draft(llm);
                    self.emit(SessionEventWire::Stream(event));
                }
                other => self.emit(SessionEventWire::Stream(other)),
            }
        }
        drop(stream);
        out
    }

    /// `SteeringDelivered`: the engine injected `message` into history.
    fn note_steering_delivered(&mut self, message: &str) {
        if self.conv.queued_message.as_deref() == Some(message) {
            self.conv.queued_message = None;
            self.emit_conversation();
        }
        // P5/P6: pop the driver's steering FIFO on delivery ack.
        if let Some(driver) = self.driver.as_mut() {
            if driver.steering.front().map(String::as_str) == Some(message) {
                driver.steering.pop_front();
            }
        }
        if let Some(pos) = self.turn_steered_events.iter().position(|e| e == message) {
            self.turn_steered_events.remove(pos);
        }
    }

    /// Accumulate one `SessionEvent::Usage` into the conversation totals.
    fn record_usage(&mut self, event: &StreamEvent) {
        if let StreamEvent::Session(SessionEvent::Usage {
            input_tokens,
            output_tokens,
            cache_read_input_tokens,
            cache_creation_input_tokens,
            cache_creation_5m,
            cache_creation_1h,
            model: usage_model,
        }) = event
        {
            let model_for_pricing = usage_model
                .as_deref()
                .unwrap_or(self.runtime.model())
                .to_string();
            self.cache_keepwarm_note_usage(crate::runtime::keepwarm::KeepWarmUsage {
                input_tokens: *input_tokens,
                output_tokens: *output_tokens,
                cache_read_tokens: *cache_read_input_tokens,
                cache_write_tokens: *cache_creation_input_tokens,
                cache_write_5m: *cache_creation_5m,
                cache_write_1h: *cache_creation_1h,
                model: usage_model.clone(),
            });
            self.conv.add_usage(
                *input_tokens,
                *output_tokens,
                *cache_read_input_tokens,
                *cache_creation_input_tokens,
                *cache_creation_5m,
                *cache_creation_1h,
                &model_for_pricing,
            );
        }
    }

    // ── event-queue wake (stream_handler.rs handle_event_queue_arm) ──────

    async fn on_queue_wake(&mut self) {
        if self.holds(WorkSource::QueueWake) {
            // Fenced for a reload: the record carries what is queued.
            return;
        }
        if self.is_parked() {
            if self.event_queue.is_empty() {
                return;
            }
            if let Err(e) = self.ensure_live().await {
                tracing::warn!(session = %self.id, error = %e, "event wake could not unpark; events stay queued");
                return;
            }
        }
        let busy = self.busy();
        let conv: &mut ConversationState = &mut self.conv;
        let drained = drain_event_queue(
            &self.event_queue,
            &mut conv.api_messages,
            &mut conv.pending_events,
            busy,
            self.steer_tx.as_ref(),
        );
        if drained.is_empty() {
            return;
        }
        for de in &drained {
            if de.disposition == EventDisposition::Steered {
                self.turn_steered_events.push(de.formatted.clone());
            }
            self.emit(SessionEventWire::External(de.event.clone()));
        }
        let injected = drained
            .iter()
            .any(|d| d.disposition == EventDisposition::Injected);
        if injected || busy {
            self.emit_conversation();
        }

        // P5: observe_events — if non-steering events arrive during a
        // driver-OWNED turn, revoke (competing work). Coordinates with F8.
        if self.streaming
            && self
                .driver
                .as_ref()
                .is_some_and(|d| d.awaiting_terminal)
        {
            let competing = drained
                .iter()
                .any(|d| !matches!(d.disposition, EventDisposition::Steered | EventDisposition::DisplayOnly));
            if competing {
                self.driver_revoke("event-bus work took priority");
            }
        }

        // `events.auto_turn = false` opts the session out of event-driven
        // turns (events are still injected/forwarded; the spend governor
        // for an ambient session). Read live from host config, like the
        // RPC/server hosts do — it used to be hardcoded `true` here.
        let auto_turn_enabled = self.host.config().events.auto_turn;
        let auto_turn_cap = self.host.config().events.auto_turn_cap;
        let action = wake_action_with_cap(
            &drained,
            &self.conv.api_messages,
            busy,
            auto_turn_enabled,
            self.consecutive_auto_turns,
            auto_turn_cap,
        );
        match action {
            WakeAction::RunTurn => {
                // F8: while a driver is armed, its tick owns turn scheduling.
                // Starting a competing turn here would race the driver.
                if self.driver.is_some() {
                    tracing::debug!("on_queue_wake: RunTurn inhibited — driver armed");
                } else if self.streaming || self.stream.is_some() {
                    tracing::warn!("handle_event_arm: RunTurn with active turn — skipping");
                } else {
                    self.consecutive_auto_turns += 1;
                    self.start_turn(TurnTrigger::EventAuto, None).await;
                }
            }
            WakeAction::Forward => {
                let hit_cap = injected
                    && !busy
                    && auto_turn_enabled
                    && auto_turn_cap_reached(self.consecutive_auto_turns, auto_turn_cap);
                if hit_cap {
                    self.emit(SessionEventWire::AutoTurnCapReached { cap: auto_turn_cap });
                }
            }
            WakeAction::Nothing => {}
        }
    }

    // ── stream events (stream_handler.rs handle_stream_event + arm tail) ──

    async fn on_stream_event(&mut self, event: StreamEvent) {
        // A turn cancelled from outside `cancel_turn` (a driver revocation):
        // its terminal events are not a normal end — the turn is finished as
        // interrupted, never through the post-turn machinery below.
        if self.turn_cancelled() {
            match event {
                StreamEvent::Session(SessionEvent::Done) => {
                    self.finish_revoked_turn(true).await;
                    return;
                }
                // Swallowed as in `drain_cancelled_stream`; `Done` follows.
                StreamEvent::Session(SessionEvent::Error(err)) => {
                    match err.outcome {
                        crate::TurnOutcome::Canceled => {}
                        crate::TurnOutcome::InterruptedAfterSideEffect { .. } => {
                            self.emit(SessionEventWire::SystemNotice(err.message));
                        }
                        _ => tracing::warn!(
                            session = %self.id,
                            category = err.category_label(),
                            "revoked turn reported an error"
                        ),
                    }
                    return;
                }
                _ => {}
            }
        }
        // Forward first: clients see the same order they see today. A
        // context-head checkpoint is actor-internal (its receipt is ours to
        // complete) and must not reach clients: the wire has no such event
        // and maps it to `Done`, which ended the turn on socket clients.
        if !matches!(
            event,
            StreamEvent::Session(SessionEvent::ContextHeadCheckpoint { .. })
        ) {
            self.emit(SessionEventWire::Stream(event.clone()));
        }

        // P5: observe_feedback — feed opted-in driver turns.
        if let Some(driver) = self.driver.as_mut() {
            if driver.awaiting_terminal && driver.grant.feedback_enabled() {
                driver.feedback.observe(&event);
            }
        }

        // P5: capture_terminal BEFORE Done/Error handlers modify state.
        let is_canceled = self
            .cancel
            .as_ref()
            .is_some_and(|ct| ct.is_cancelled());
        let terminal = if self
            .driver
            .as_ref()
            .is_some_and(|d| d.awaiting_terminal)
        {
            super::driver::capture_terminal(Some(&event), is_canceled)
        } else {
            None
        };

        enum After {
            Continue,
            AutoSendQueued(String),
            AutoTriggerEvents,
            Failed,
        }
        let mut after = After::Continue;

        match event {
            // Text, a new provider response (or a retry's reset: the
            // in-flight text starts over), tool calls.
            StreamEvent::Llm(ref llm) => self.observe_turn_draft(llm),
            StreamEvent::Session(SessionEvent::MessageHistory(history)) => {
                // Published at every round boundary as well as at the end of
                // the turn (`runtime/stream.rs` ROUND CHECKPOINT): the session
                // on disk follows the turn as it progresses. Queued, never
                // awaited — the persister writes it (latest wins), so a slow
                // disk can never stall the turn machine.
                self.conv.api_messages = history;
                self.request_save();
                // The round is committed: the draft now continues from here.
                self.turn_draft_committed();
                self.trim_replay_to_checkpoint();
                self.emit_conversation();
            }
            StreamEvent::Agent(AgentEvent::SteeringDelivered { ref message }) => {
                self.note_steering_delivered(message);
            }
            StreamEvent::Agent(_) => {}
            StreamEvent::Session(SessionEvent::Usage { .. }) => {
                self.record_usage(&event);
                // (E-P7, §S3) Host-owned spend circuit breaker. Checked after
                // every Usage event — a breach cancels the in-flight turn,
                // revokes any armed driver (so it cannot re-poll and keep
                // spending), and tells clients why.
                if let Some((scope, cost, cap)) = self.cost_cap_breach() {
                    self.emit(SessionEventWire::CostCapReached {
                        scope: scope.to_string(),
                        cost,
                        cap,
                    });
                    self.note_cancel_cause(crate::engine::interrupt::InterruptReason::CostCap);
                    if self.driver.is_some() || self.driver_pending.is_some() {
                        self.driver_revoke(&format!("{scope} cost cap reached (${cost:.4} ≥ ${cap:.4})"));
                    }
                    self.cancel_turn(crate::engine::interrupt::InterruptReason::CostCap)
                        .await;
                    return;
                }
            }
            StreamEvent::Session(SessionEvent::Notice(_) | SessionEvent::Retry(_)) => {}
            StreamEvent::Session(SessionEvent::Done) => {
                self.clear_stream();
                self.publish_subagent_rows();
                // Flush events that arrived during streaming into api_messages
                let had_pending = !self.conv.pending_events.is_empty();
                {
                    let conv: &mut ConversationState = &mut self.conv;
                    for formatted in conv.pending_events.drain(..) {
                        conv.api_messages
                            .push(std::sync::Arc::new(serde_json::json!({
                                "role": "user",
                                "content": formatted
                            })));
                    }
                }
                if let Some(queued) = self.conv.queued_message.take() {
                    after = After::AutoSendQueued(queued);
                } else if had_pending {
                    self.request_save();
                    after = After::AutoTriggerEvents;
                }
                self.emit_conversation();
                if matches!(after, After::Continue) {
                    if self.config.auto_compact {
                        self.post_turn_chat().await;
                    }
                    // A spawned compaction emits Idle when it lands.
                    if self.compact.is_none() {
                        self.announce_idle().await;
                    }
                }
            }
            StreamEvent::Session(SessionEvent::Error(_)) => {
                // Remove only invalid messages appended by the ACTIVE turn —
                // before `clear_stream` queues the turn's final save.
                crate::engine::stream::repair_history_after_failure(
                    &mut self.conv.api_messages,
                    self.turn_baseline,
                );
                self.clear_stream();
                self.publish_subagent_rows();
                self.emit_conversation();
                after = After::Failed;
            }
            StreamEvent::Session(SessionEvent::ContextHeadCheckpoint {
                session_id,
                messages,
                receipt,
            }) => {
                self.handle_context_head_checkpoint(session_id, messages, receipt)
                    .await;
            }
        }

        match after {
            After::Continue => {}
            After::Failed => {
                if self.config.auto_compact {
                    self.post_turn_chat().await;
                }
                if self.compact.is_none() {
                    self.announce_idle().await;
                }
            }
            After::AutoSendQueued(queued) => {
                if self.config.auto_compact {
                    self.post_turn_chat().await;
                }
                if self.compact.is_some() {
                    // Rides the compaction transition (queued_message
                    // restored last); the TUI reports it, never re-sends.
                    self.conv.queued_message = Some(queued);
                    return;
                }
                // Auto-send the queued message (user-authored — reset counter)
                self.consecutive_auto_turns = 0;
                let user_text = queued.clone();
                self.conv.api_messages.push(std::sync::Arc::new(
                    serde_json::json!({"role": "user", "content": queued}),
                ));
                self.start_turn(TurnTrigger::QueuedAuto, Some(user_text)).await;
            }
            After::AutoTriggerEvents => {
                if self.config.auto_compact {
                    self.post_turn_chat().await;
                }
                if self.compact.is_some() {
                    return;
                }
                // Central claim gate: allows turns while under the configured
                // cap (events.auto_turn_cap; 0 = unlimited), denies once reached.
                let auto_turn_cap = self.host.config().events.auto_turn_cap;
                if claim_auto_turn_with_cap(&mut self.consecutive_auto_turns, auto_turn_cap) {
                    self.start_turn(TurnTrigger::EventAuto, None).await;
                } else {
                    self.emit(SessionEventWire::AutoTurnCapReached { cap: auto_turn_cap });
                    self.announce_idle().await;
                }
            }
        }

        // P5: observe_terminal — set driver outcome or revoke.
        if let Some(terminal) = terminal {
            let revoke_reason = if let Some(driver) = self.driver.as_mut() {
                super::driver::observe_terminal(driver, &self.runtime, terminal)
            } else {
                None
            };
            // Emit DriverTurnOutcome if an outcome was just set.
            if let Some(driver) = self.driver.as_ref() {
                if let Some((outcome, _)) = &driver.outcome {
                    self.emit(SessionEventWire::DriverTurnOutcome {
                        outcome: *outcome,
                        selection: driver.selection.clone(),
                        feedback: driver.grant.feedback_enabled().then(|| {
                            driver.completed_feedback.to_string()
                        }),
                    });
                }
            }
            if let Some(reason) = revoke_reason {
                self.driver_revoke(&reason);
            }
        }
    }

    /// chat.rs post-turn block: save + engine-budget auto-compaction.
    async fn post_turn_chat(&mut self) {
        self.request_save();
        let assessment = self.runtime.assess_context(&self.conv.api_messages).await;
        if assessment.should_compact() {
            self.emit(SessionEventWire::SystemNotice(format!(
                "[auto-compacting ~{} tokens...]",
                assessment.used_tokens()
            )));
            self.compact(None, "auto").await;
        }
    }

    /// `SYNAPS_SESSION_COMPACT_INLINE=1`: one-release kill-switch back to the
    /// #107 inline body (deleted in phase 4).
    fn compact_inline_env() -> bool {
        matches!(
            std::env::var("SYNAPS_SESSION_COMPACT_INLINE").as_deref(),
            Ok("1") | Ok("true")
        )
    }

    /// `/compact` + auto-compaction entry (B2): spawns `compact_conversation`
    /// on a `Runtime::clone()` (the same clone the TUI makes at
    /// dispatch.rs:450 — the clone never runs turns) and returns; the
    /// `select!` arm `on_compaction_done` applies the outcome. While a job
    /// is in flight Attach/Detach/Cancel/Query are serviced, `Submit` is
    /// queued (`Steered{delivered:false}`), auto-turns see `busy`.
    pub(crate) async fn compact(&mut self, instructions: Option<String>, source: &str) {
        // Compaction replaces the history: the old cached prefix is dead.
        self.cache_keepwarm_compaction();
        if Self::compact_inline_env() {
            return self.compact_inline(instructions, source).await;
        }
        if self.compact.is_some() {
            self.emit(SessionEventWire::SystemNotice(
                "compaction already in progress".into(),
            ));
            return;
        }
        if self.streaming {
            self.emit(SessionEventWire::SystemNotice(
                "cannot compact while a turn is running".into(),
            ));
            return;
        }
        let disclosure =
            preview_compaction_disclosure(&self.runtime, &self.conv.api_messages).render_line();
        self.emit(SessionEventWire::CompactionStarted {
            source: source.to_string(),
            disclosure,
        });
        let msgs = self.conv.api_messages.clone();
        let rt: Runtime = (*self.runtime).clone();
        let task = tokio::spawn(async move {
            compact_conversation(&msgs, &rt, instructions.as_deref()).await
        });
        self.compact = Some(CompactionJob {
            task,
            source: source.to_string(),
            started: tokio::time::Instant::now(),
        });
        self.update_attach_state();
    }

    /// `select!` arm: the spawned job finished (or was aborted). Applies via
    /// the ONE engine transition with the configured policy; pending events
    /// and the queued message ride the transition (loop_arms.rs:761-836).
    async fn on_compaction_done(
        &mut self,
        res: std::result::Result<
            Result<crate::runtime::compaction::CompactionOutcome>,
            tokio::task::JoinError,
        >,
    ) {
        let Some(job) = self.compact.take() else {
            return;
        };
        let msg_count = self.conv.api_messages.len();
        match res {
            Err(join) => {
                if !join.is_cancelled() {
                    self.emit(SessionEventWire::CompactionFailed {
                        message: join.to_string(),
                        panicked: true,
                    });
                }
            }
            Ok(Err(e)) => self.emit(SessionEventWire::CompactionFailed {
                message: e.to_string(),
                panicked: false,
            }),
            Ok(Ok(outcome)) => {
                // The transition rewrites the predecessor on disk (its
                // `compacted_into` link): a queued save of it must land
                // first, never after.
                if !self.persister.flush().await {
                    tracing::warn!(session = %self.id, "compaction: an earlier save failed");
                }
                let policy: CompactionPolicy = self.config.compaction_policy.into();
                let queued = self.conv.queued_message.clone();
                let applied = apply_compaction(
                    &self.runtime,
                    &self.conv.session,
                    &self.conv.api_messages,
                    &outcome,
                    CompactionTransition {
                        policy,
                        pending_events: self.conv.pending_events.clone(),
                        queued_message: queued.clone(),
                        hook_source: job.source.clone(),
                    },
                )
                .await;
                match applied {
                    Ok(applied) => {
                        let previous = applied.previous_session_id.clone();
                        self.conv.session = applied.session;
                        self.conv.api_messages = applied.api_messages;
                        self.conv.pending_events.clear();
                        self.conv.queued_message = None;
                        if policy == CompactionPolicy::LinkedSuccessor {
                            self.conv.total_input_tokens = 0;
                            self.conv.total_output_tokens = 0;
                            self.conv.session_cost = 0.0;
                            // The keep-warm subtotal follows the totals it
                            // is part of: input/output/cost restart, the
                            // cache totals (not reset here) keep their
                            // keep-warm share.
                            self.conv.keepwarm_cost = 0.0;
                            self.conv.keepwarm_tokens.input = 0;
                            self.conv.keepwarm_tokens.output = 0;
                        }
                        let new_id = self.conv.session.id.clone();
                        self.runtime.set_session_id(Some(new_id.clone()));
                        self.set_journal_id(new_id.clone());
                        // F10: lock follows the new journal id.
                        self.reacquire_session_lock(&new_id);
                        // Wall 1: reset continuation state for the new session id
                        // so stale epoch checks in persist_head() don't reject
                        // future checkpoints.
                        self.conv.context_head = crate::engine::session::ContextHeadPersistence::default();
                        self.runtime
                            .reset_context_continuation(&new_id, &self.conv.api_messages);
                        self.emit(SessionEventWire::CompactionApplied {
                            previous_session_id: previous,
                            session_id: new_id,
                            chains_advanced: applied.chains_advanced,
                            queued_restored: queued,
                            msg_count,
                        });
                    }
                    Err(e) => self.emit(SessionEventWire::CompactionFailed {
                        message: e.to_string(),
                        panicked: false,
                    }),
                }
            }
        }
        self.emit_conversation();
        self.update_attach_state();
        self.announce_idle().await;
    }

    /// #107 inline body (kill-switch only).
    async fn compact_inline(&mut self, instructions: Option<String>, source: &str) {
        self.compacting_inline = true;
        self.publish_presence();
        self.compact_inline_inner(instructions, source).await;
        self.compacting_inline = false;
        self.publish_presence();
    }

    async fn compact_inline_inner(&mut self, instructions: Option<String>, source: &str) {
        self.emit(SessionEventWire::SystemNotice(format!(
            "[{}]",
            preview_compaction_disclosure(&self.runtime, &self.conv.api_messages).render_line()
        )));
        let outcome =
            compact_conversation(&self.conv.api_messages, &self.runtime, instructions.as_deref())
                .await;
        let applied = match outcome {
            Ok(outcome) => {
                // Queued saves land before the transition rewrites the file.
                if !self.persister.flush().await {
                    tracing::warn!(session = %self.id, "compaction: an earlier save failed");
                }
                apply_compaction(
                    &self.runtime,
                    &self.conv.session,
                    &self.conv.api_messages,
                    &outcome,
                    CompactionTransition {
                        policy: CompactionPolicy::InPlace,
                        pending_events: Vec::new(),
                        queued_message: None,
                        hook_source: source.to_string(),
                    },
                )
                .await
            }
            Err(e) => Err(e),
        };
        match applied {
            Ok(applied) => {
                self.conv.session = applied.session;
                self.conv.api_messages = applied.api_messages;
                let after = self.runtime.assess_context(&self.conv.api_messages).await;
                self.emit(SessionEventWire::SystemNotice(format!(
                    "[compacted → ~{} tokens]",
                    after.used_tokens()
                )));
            }
            Err(e) => self.emit(SessionEventWire::SystemNotice(format!(
                "[compaction failed: {}]",
                e
            ))),
        }
        self.emit_conversation();
    }

    // ── prompts ──────────────────────────────────────────────────────────

    fn on_prompt_request(&mut self, req: SecretPromptRequest) {
        let id = self.next_prompt_id;
        self.next_prompt_id += 1;
        let pr = PromptRequest {
            id,
            kind: PromptKind::from_title(&req.title),
            title: req.title,
            prompt: req.prompt,
            raised_at: chrono::Utc::now(),
            confirm_scope: req.confirm_scope,
        };
        self.pending_prompts.push_back((pr.clone(), req.response_tx));
        self.publish_presence();
        self.emit(SessionEventWire::Prompt(pr));
        // A prompt can be raised while already detached (a turn running with no
        // client). Arm the abandonment deadline if this left us pinned.
        self.rearm_prompt_abandon();
    }

    /// Why `from` may not give this answer to `prompt`, if it may not.
    /// Confirm prompts gate tool execution, so an `Observe` client cannot
    /// answer them at all, and only the input owner (or the host, `from:
    /// None`) can choose "Allow all", which arms a session-wide latch.
    fn confirm_answer_refusal(
        &self,
        from: Option<ClientId>,
        prompt: &PromptRequest,
        allow_all: bool,
    ) -> Option<&'static str> {
        let c = from?;
        if prompt.kind != PromptKind::Confirm {
            return None;
        }
        if self.attached.get(&c).map(|a| a.mode) == Some(AttachMode::Observe) {
            return Some("observe-mode clients cannot answer confirm prompts");
        }
        if allow_all && self.input_owner != Some(c) {
            return Some("only the input owner can choose Allow all");
        }
        None
    }

    fn answer(&mut self, from: Option<ClientId>, prompt_id: u64, value: Option<String>) {
        let Some(pos) = self.pending_prompts.iter().position(|(p, _)| p.id == prompt_id) else {
            return; // unknown or already answered — dedup on id
        };
        let allow_all = value.as_deref().is_some_and(|v| {
            v.trim()
                .eq_ignore_ascii_case(crate::tools::CONFIRM_ANSWER_ALLOW_ALL)
        });
        // An answer is input too: with `client_input = shared` the answering
        // drive client becomes active first, so "Allow all" (owner only)
        // is its to choose.
        if let Some(c) = from {
            self.note_input(c);
        }
        if let Some(reason) = self.confirm_answer_refusal(from, &self.pending_prompts[pos].0, allow_all) {
            if let Some(c) = from {
                self.emit(SessionEventWire::Refused {
                    client: c,
                    command: "answer".to_string(),
                    reason: reason.to_string(),
                });
            }
            return; // the prompt stays pending for an eligible client
        }
        let approved = allow_all
            || value
                .as_deref()
                .map(str::trim)
                .is_some_and(|v| v.eq_ignore_ascii_case("y") || v.eq_ignore_ascii_case("yes"));
        let (prompt, tx) = self.pending_prompts.remove(pos).expect("position");
        let _ = tx.send(value);
        self.publish_presence();
        self.emit(SessionEventWire::PromptResolved { prompt_id });
        let by = match from {
            Some(c) => self.client_label(c),
            None => "the host".to_string(),
        };
        // prA N3: the tool card shows the model's original input; when the
        // approved call runs input rewritten by an extension, every client's
        // transcript records that (the prompt showed the final input).
        if approved && prompt.title == CONFIRM_PROMPT_TITLE {
            if let Some(scope) = prompt.confirm_scope.as_ref().filter(|s| !s.rewritten_by.is_empty()) {
                self.emit(SessionEventWire::SystemNotice(format!(
                    "Approved by {by}: {} runs with input rewritten by {} (not the input shown \
                     on its tool card; the approval prompt showed what runs).",
                    scope.tool.as_deref().unwrap_or("the tool call"),
                    scope.rewritten_by.join(", ")
                )));
            }
        }
        // Arming the latch is announced to every attached client, not just
        // the one that pressed the key.
        if allow_all && prompt.title == CONFIRM_PROMPT_TITLE {
            if let Some(scope) = prompt.confirm_scope.as_ref().filter(|s| !s.extensions.is_empty()) {
                self.emit(SessionEventWire::SystemNotice(format!(
                    "Allow all armed by {by}: tool-call confirms from {} are auto-approved \
                     until /new. Other extensions still ask. /confirms reset revokes.",
                    scope.extensions_label()
                )));
            }
        }
        // The last pending prompt may have just cleared — drop the deadline.
        self.rearm_prompt_abandon();
    }

    // ── settings / queries / engine commands ─────────────────────────────

    pub(crate) async fn apply_setting(&mut self, id: u64, setting: SessionSetting) {
        if replayable(&setting) {
            let disc = std::mem::discriminant(&setting);
            self.settings_replay
                .retain(|s| std::mem::discriminant(s) != disc);
            self.settings_replay.push(setting.clone());
        }
        let name = match &setting {
            SessionSetting::Model { .. } => "model",
            SessionSetting::ReasoningLevel { .. } => "reasoning_level",
            SessionSetting::ContextWindow { .. } => "context_window",
            SessionSetting::CompactionModel { .. } => "compaction_model",
            SessionSetting::ApiRetries { .. } => "api_retries",
            SessionSetting::SubagentTimeout { .. } => "subagent_timeout",
            SessionSetting::MaxToolOutput { .. } => "max_tool_output",
            SessionSetting::BashTimeout { .. } => "bash_timeout",
            SessionSetting::BashMaxTimeout { .. } => "bash_max_timeout",
            SessionSetting::SystemPrompt { .. } => "system_prompt",
            SessionSetting::ReloadPrompt => "reload_prompt",
            SessionSetting::GrantWorkerModel { .. } => "grant_worker_model",
        };
        // Capability evidence must exist in THIS (possibly daemon) process
        // before the mutation is validated and a reasoning clamp decided.
        match &setting {
            SessionSetting::Model { model } => self.runtime.prepare_model_capabilities(model).await,
            SessionSetting::ReasoningLevel { .. } => {
                let model = self.runtime.model().to_string();
                self.runtime.prepare_model_capabilities(&model).await;
            }
            _ => {}
        }
        let rt = &mut self.runtime;
        let mut clamp_wire = None;
        let result: std::result::Result<Option<String>, String> = match setting {
            SessionSetting::Model { model } => rt.try_set_model(model).map(|clamp| {
                self.conv.session.model = rt.model().to_string();
                clamp.map(|c| {
                    self.conv.session.thinking_level = rt.thinking_level().to_string();
                    clamp_wire = Some(ReasoningClampWire {
                        from: c.from.as_str().to_string(),
                        to: c.to.as_str().to_string(),
                    });
                    format!(
                        "thinking → {} (clamped from {}: not supported by {})",
                        c.to.as_str(),
                        c.from.as_str(),
                        rt.model()
                    )
                })
            }),
            SessionSetting::ReasoningLevel { level } => {
                rt.set_reasoning_level_checked(level).map(|_| {
                    self.conv.session.thinking_level = rt.thinking_level().to_string();
                    None
                })
            }
            SessionSetting::ContextWindow { tokens } => {
                rt.set_context_window(tokens);
                Ok(None)
            }
            SessionSetting::CompactionModel { model } => {
                rt.set_compaction_model(model);
                Ok(None)
            }
            SessionSetting::ApiRetries { n } => {
                rt.set_api_retries(n);
                Ok(None)
            }
            SessionSetting::SubagentTimeout { secs } => {
                rt.set_subagent_timeout(secs);
                Ok(None)
            }
            SessionSetting::MaxToolOutput { bytes } => {
                rt.set_max_tool_output(bytes);
                Ok(None)
            }
            SessionSetting::BashTimeout { secs } => {
                rt.set_bash_timeout(secs);
                Ok(None)
            }
            SessionSetting::BashMaxTimeout { secs } => {
                rt.set_bash_max_timeout(secs);
                Ok(None)
            }
            SessionSetting::SystemPrompt { text } => {
                rt.set_system_prompt(text);
                Ok(None)
            }
            SessionSetting::ReloadPrompt => rt
                .reload_prompt()
                .map(|generation| Some(format!("prompt reloaded (generation {generation})")))
                .map_err(|e| e.to_string()),
            SessionSetting::GrantWorkerModel { model } => {
                rt.grant_worker_model(&model).map(|_| None)
            }
        };
        self.publish_view().await;
        let view = (**self.view.load()).clone();
        let (ok, message) = match result {
            Ok(m) => (true, m),
            Err(e) => (false, Some(e)),
        };
        self.emit(SessionEventWire::SettingChanged(SettingApplied {
            id,
            setting: name.to_string(),
            ok,
            message,
            view,
            clamp: clamp_wire,
        }));
    }

    pub(crate) async fn query(&mut self, id: u64, query: SessionQuery) {
        // Status/View never wake a parked session (the idle probe must not).
        if self.is_parked() {
            let value = match query {
                SessionQuery::Status => serde_json::json!({
                    "session": (**self.journal_id.load()).clone(),
                    "model": self.view.load().model,
                    "lifecycle": SessionLifecycle::Parked,
                    "streaming": false,
                    "auto_turns": self.consecutive_auto_turns,
                    "attached": 0,
                    "pending_prompts": 0,
                    "pinned": self.keep_warm,
                }),
                SessionQuery::View => serde_json::to_value(&**self.view.load()).unwrap_or_default(),
                // A parked session has no live workers; a watch poll must not wake it.
                SessionQuery::SubagentWatch { handle_id, .. } => {
                    serde_json::json!({ "error": format!("no subagent {handle_id} in this session") })
                }
                _ => match self.ensure_live().await {
                    Ok(()) => return Box::pin(self.query(id, query)).await,
                    Err(e) => serde_json::json!({ "error": e }),
                },
            };
            self.emit(SessionEventWire::QueryResult { id, value });
            return;
        }
        let value = match query {
            SessionQuery::Status => serde_json::json!({
                "session": self.conv.session.id,
                "model": self.runtime.model(),
                "lifecycle": SessionLifecycle::from_u8(self.lifecycle.load(std::sync::atomic::Ordering::Acquire)),
                "tokens": { "input": self.conv.total_input_tokens, "output": self.conv.total_output_tokens },
                "cost": self.conv.session_cost,
                // Included in `cost`; labelled separately (task #444).
                "keepwarm_cost": self.conv.keepwarm_cost,
                "keepwarm": self.cache_keepwarm_status_json(None),
                "messages": self.conv.api_messages.len(),
                "streaming": self.streaming,
                "auto_turns": self.consecutive_auto_turns,
                "attached": self.attached.len(),
                "pending_prompts": self.pending_prompts.len(),
                // `/pin` status in the TUI.
                "pinned": self.keep_warm,
            }),
            SessionQuery::Messages => {
                serde_json::to_value(&self.conv.api_messages).unwrap_or_default()
            }
            SessionQuery::DisplayTail { items } => {
                serde_json::to_value(super::display::display_tail(&self.conv.api_messages, items))
                    .unwrap_or_default()
            }
            SessionQuery::SubagentRows => {
                let rows = self
                    .runtime
                    .subagent_registry()
                    .lock()
                    .unwrap_or_else(|p| p.into_inner())
                    .display_rows();
                serde_json::Value::Array(
                    rows.iter()
                        .map(|r| {
                            serde_json::json!({
                                "subagent_id": r.subagent_id,
                                "agent_name": r.agent_name,
                                "status": format!("{:?}", r.status),
                                "cancel_requested": r.cancel_requested,
                                "elapsed_secs": r.elapsed_secs,
                            })
                        })
                        .collect(),
                )
            }
            SessionQuery::SubagentWatch { handle_id, since, tool_queue } => {
                let mut registry = self
                    .runtime
                    .subagent_registry()
                    .lock()
                    .unwrap_or_else(|p| p.into_inner());
                match registry.watch_page(&handle_id, since) {
                    // An older client did not ask for queue entries and
                    // could not decode a page that had them.
                    Some(page) if !tool_queue => {
                        serde_json::to_value(page.without_tool_queue()).unwrap_or_default()
                    }
                    Some(page) => serde_json::to_value(page).unwrap_or_default(),
                    None => serde_json::json!({
                        "error": format!("no subagent {handle_id} in this session")
                    }),
                }
            }
            SessionQuery::PromptInspection => {
                serde_json::to_value(self.runtime.prompt_inspection_json()).unwrap_or_default()
            }
            SessionQuery::ToolsSchema => serde_json::json!({ "unsupported": "tools_schema" }),
            SessionQuery::View => serde_json::to_value(&**self.view.load()).unwrap_or_default(),
            SessionQuery::ContextAssessment => {
                let a = self.runtime.assess_context(&self.conv.api_messages).await;
                serde_json::json!({
                    "used_tokens": a.used_tokens(),
                    "budget_tokens": a.budget_tokens(),
                    "provider_window": a.provider_window,
                    "should_compact": a.should_compact(),
                })
            }
            SessionQuery::ContextReport => {
                use crate::engine::commands::{context_command, CommandResult};
                match context_command(&self.runtime, Some(&self.conv.api_messages)) {
                    CommandResult::Output(text) => serde_json::json!({ "text": text }),
                    other => serde_json::json!({ "unsupported": format!("{other:?}") }),
                }
            }
        };
        self.emit(SessionEventWire::QueryResult { id, value });
    }

    // ── attach / detach ──────────────────────────────────────────────────

    pub(crate) fn snapshot(&self) -> AttachSnapshot {
        AttachSnapshot {
            meta: self.meta.clone(),
            view: (**self.view.load()).clone(),
            conversation: self.conv.snapshot(self.consecutive_auto_turns),
            streaming: self.streaming,
            replay: self.turn_replay.iter().cloned().collect(),
            pending_prompts: self.pending_prompts.iter().map(|(p, _)| p.clone()).collect(),
            clients: self.attached.iter().map(|(c, a)| (*c, a.meta.kind)).collect(),
            roster: self.roster(),
            client_input: self.client_input,
            input_owner: self.input_owner,
            display_tail: None,
        }
    }

    /// `snapshot()` shaped for one client (phase 4 §2.3): `Digest` clients
    /// get an empty `api_messages` (`messages_len` kept), a daemon-projected
    /// `display_tail`, and a replay without the per-round `MessageHistory`
    /// envelopes (the trailing `Conversation` digest carries len/hash).
    pub(crate) fn snapshot_for(&self, client: &ClientMeta) -> AttachSnapshot {
        let mut snap = self.snapshot();
        if client.history == HistoryMode::Digest {
            snap.display_tail = Some(super::display::display_tail(
                &snap.conversation.api_messages,
                client.tail_items,
            ));
            snap.conversation.api_messages = Vec::new();
            snap.replay.retain(|e| {
                !matches!(
                    e.event,
                    SessionEventWire::Stream(StreamEvent::Session(SessionEvent::MessageHistory(_)))
                )
            });
        }
        snap
    }

    /// B1 ownership: `Observe` never owns; `Mirror` owns iff nobody does
    /// (else read-only + notice); `Takeover` steals with
    /// `InputOwnerChanged{reason: Takeover}` to everyone (the old owner's
    /// client renders the toast).
    async fn attach(&mut self, client: ClientMeta, mode: AttachMode) {
        if let Err(e) = self.ensure_live().await {
            self.emit(SessionEventWire::AttachRefused { message: e });
            return;
        }
        let cid = ClientId(self.next_client_id);
        self.next_client_id += 1;
        let kind = client.kind;
        self.cache_keepwarm_note_attach(kind, mode);
        let snapshot_meta = client.clone();
        let name = resolve_client_name(
            client.name.as_deref(),
            kind,
            self.attached.values().map(|a| a.name.as_str()),
        );
        let attached_at_ms = chrono::Utc::now().timestamp_millis().max(0) as u64;
        self.attached.insert(
            cid,
            AttachedClient { meta: client, mode, name: name.clone(), attached_at_ms, last_input: 0 },
        );
        self.update_attach_state();
        let mut owner_change: Option<(Option<ClientId>, OwnerChangeReason)> = None;
        match mode {
            AttachMode::Observe => {}
            // Input owned elsewhere: the joiner learns it from its snapshot
            // (`AttachSnapshot::input_owned_elsewhere`), never from a notice
            // broadcast to every client.
            AttachMode::Mirror => {
                if self.input_owner.is_none() {
                    owner_change = Some((None, OwnerChangeReason::Attach));
                }
            }
            AttachMode::Takeover => {
                let reason = if self.input_owner.is_some() {
                    OwnerChangeReason::Takeover
                } else {
                    OwnerChangeReason::Attach
                };
                owner_change = Some((self.input_owner, reason));
            }
        }
        if let Some((from, reason)) = owner_change {
            self.input_owner = Some(cid);
            self.emit(SessionEventWire::InputOwnerChanged {
                from,
                to: Some(cid),
                reason,
            });
        }
        self.publish_presence();
        let snapshot = self.snapshot_for(&snapshot_meta);
        self.emit(SessionEventWire::Attached {
            client: cid,
            snapshot,
        });
        self.emit(SessionEventWire::ClientJoined { client: cid, kind, name, mode });
    }

    /// B1 `/takeover`: the sending client takes input in place, as an
    /// `AttachMode::Takeover` attach would, without re-attaching. An
    /// `Observe` client is refused (it attached read-only on purpose); the
    /// owner, the host and unknown clients change nothing. Any turn keeps
    /// running: only who may drive the session changes.
    fn take_input(&mut self, from: Option<ClientId>) {
        let Some(c) = from else { return };
        let Some(mode) = self.attached.get(&c).map(|a| a.mode) else { return };
        if mode == AttachMode::Observe {
            self.emit(SessionEventWire::Refused {
                client: c,
                command: command_name(&SessionCommand::TakeInput).to_string(),
                reason: "this client attached with --observe and never owns input; \
                         attach without --observe to take it"
                    .to_string(),
            });
            return;
        }
        if self.input_owner == Some(c) {
            return;
        }
        let from = self.input_owner;
        let reason = if from.is_some() {
            OwnerChangeReason::Takeover
        } else {
            OwnerChangeReason::Attach
        };
        self.input_owner = Some(c);
        self.emit(SessionEventWire::InputOwnerChanged { from, to: Some(c), reason });
        self.publish_presence();
        self.note_input(c); // most recent, for promotion on detach
    }

    /// Never touches `stream`/`cancel`: the turn keeps running and its
    /// events buffer in the broadcast (§8 detach-without-abort). The owner
    /// leaving passes input to the oldest attached non-`Observe` client.
    fn detach(&mut self, client: ClientId) {
        if self.attached.remove(&client).is_none() {
            return;
        }
        self.update_attach_state();
        self.emit(SessionEventWire::ClientLeft { client });
        if self.input_owner == Some(client) {
            // `shared`: back to whoever typed last (a script that sent one
            // prompt and left hands input back to the tab in use);
            // `exclusive`: the lowest id, as before. Ties: the lowest id.
            let shared = self.client_input == ClientInput::Shared;
            let next = self
                .attached
                .iter()
                .filter(|(_, a)| a.mode != AttachMode::Observe)
                .max_by_key(|(c, a)| (if shared { a.last_input } else { 0 }, std::cmp::Reverse(c.0)))
                .map(|(c, _)| *c);
            self.input_owner = next;
            self.emit(SessionEventWire::InputOwnerChanged {
                from: Some(client),
                to: next,
                reason: OwnerChangeReason::OwnerDetached,
            });
            self.publish_presence();
        }
        // (E-P7 §S1 / P11) The last client just left. Fail-closed gates apply
        // ONLY when a driver is (or was about to be) armed:
        //
        //  1. An armed driver is revoked: it must never run turns headless with
        //     nobody watching and no way to answer a confirmation.
        //  2. Its pending host confirmations are answered `None` (deny) — a
        //     headless autonomous run must not sit on an unanswerable prompt.
        //     `tools/discovery.rs` treats `None` as Unauthorized (deny).
        //
        // For a PLAIN interactive session (no driver), a pending prompt SURVIVES
        // detach: the user reattaches and answers it — the daemon's core
        // detach/reattach contract, and detach is often involuntary (SSH drop,
        // sleep, wifi). Denying on detach would let a transient disconnect
        // silently reject the user's action. The resource pin from an
        // *abandoned* prompt is bounded by the pending-prompt deadline instead
        // (see the actor select loop), not by punishing every detach.
        if self.attached.is_empty() && (self.driver.is_some() || self.driver_pending.is_some()) {
            self.driver_revoke("no clients attached");
            let had_prompts = !self.pending_prompts.is_empty();
            while let Some((pr, tx)) = self.pending_prompts.pop_front() {
                let _ = tx.send(None);
                self.emit(SessionEventWire::PromptResolved { prompt_id: pr.id });
            }
            if had_prompts {
                self.publish_presence();
            }
        }
    }

    /// B1 (used by C3 reload): checkpoint the session without ending it —
    /// cancel any turn (partial history kept + interruption marker), abort compaction, answer
    /// pending prompts `None`, save, close PTYs. Replies on
    /// `CHECKPOINT_QUERY_ID` so `reload.rs` can await it per session.
    pub(crate) async fn checkpoint(&mut self, reason: CheckpointReason) {
        let interrupt = match reason {
            CheckpointReason::Reload => crate::engine::interrupt::InterruptReason::Restart,
            CheckpointReason::HostRequest => crate::engine::interrupt::InterruptReason::Host,
        };
        // Before the revoke below cancels a driver turn.
        self.note_cancel_cause(interrupt);
        // E-P3: driver does NOT survive reload (§3 S2/S5).
        if self.driver.is_some() {
            self.driver_revoke("daemon reloaded");
        }
        // Workers cannot survive the exec either (they are threads of this
        // process). Flag + cancel them BEFORE `cancel_turn` (whose own
        // cancel would otherwise label them user-cancelled). This also
        // fences new launches: a worker the turn registers before its
        // cancel lands is stopped at birth as the reload's.
        let reload = matches!(reason, CheckpointReason::Reload);
        if reload {
            self.stop_workers_for_reload();
        }
        if self.streaming {
            self.cancel_turn(interrupt).await;
        }
        // Again after the turn cancel: every worker to await, including
        // any the cancelled turn registered meanwhile.
        let stopping = if reload { self.stop_workers_for_reload() } else { Vec::new() };
        let stopped_workers = await_stopped_workers(self.id.as_str(), stopping).await;
        self.abort_compaction();
        while let Some((pr, tx)) = self.pending_prompts.pop_front() {
            let _ = tx.send(None);
            self.emit(SessionEventWire::PromptResolved { prompt_id: pr.id });
        }
        // One bounded wait covers the cancelled turn's save too (queued, not
        // awaited, by `cancel_turn`).
        if !self.save().await {
            tracing::warn!(session = %self.id, "checkpoint: save failed or timed out");
        }
        let notice = match reason {
            CheckpointReason::Reload => {
                "daemon reloading — background shells/PTYs will be closed; the turn was checkpointed"
            }
            CheckpointReason::HostRequest => {
                "checkpoint — background shells/PTYs closed; the turn was checkpointed"
            }
        };
        self.emit(SessionEventWire::SystemNotice(notice.to_string()));
        if !stopped_workers.is_empty() {
            self.emit(SessionEventWire::SystemNotice(format!(
                "daemon reloading — stopped worker{} {} (archived; resumable after the reload)",
                if stopped_workers.len() == 1 { "" } else { "s" },
                stopped_workers.join(", ")
            )));
        }
        self.runtime.session_manager().shutdown_all();
        let mut record = self.reload_record();
        if reload {
            record.workers = self.reload_carried_workers();
            // Recorded: from here to the exec, nothing may start work the
            // record does not carry (a queued wake must stay queued).
            self.reload_fence = Some(ReloadFence {
                deadline: tokio::time::Instant::now() + budgets::RELOAD_FENCE_TIMEOUT,
                workers: record.workers.clone(),
            });
        }
        let record = serde_json::to_value(record).unwrap_or_default();
        self.emit(SessionEventWire::QueryResult {
            id: super::wire::CHECKPOINT_QUERY_ID,
            value: serde_json::json!({ "ok": true, "record": record }),
        });
    }

    /// The workers a reload carries for this session: every registry entry,
    /// plus whether a finished worker's completion wake is still queued
    /// (the exec discards the queue, so the next image re-posts it).
    pub(crate) fn reload_carried_workers(&self) -> Vec<crate::runtime::subagent::CarriedWorker> {
        let undelivered: std::collections::HashSet<String> = self
            .event_queue
            .snapshot()
            .iter()
            .filter(|e| e.content.content_type == "subagent_completion")
            .filter_map(|e| {
                e.content
                    .data
                    .as_ref()?
                    .get("handle_id")?
                    .as_str()
                    .map(str::to_string)
            })
            .collect();
        self.runtime
            .subagent_registry()
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .reload_carry(&undelivered)
    }

    /// Host-only (`SessionCommand::RestoreToolActivations`, sent by
    /// reload's rehydrate): the activations this conversation held in the
    /// previous image. A live runtime takes them (restored at its first turn
    /// start); a parked session keeps them for the runtime its unpark
    /// builds, without unparking.
    pub(crate) fn restore_tool_activations(
        &mut self,
        activations: Vec<crate::tools::activation::CarriedActivation>,
    ) {
        if activations.is_empty() {
            return;
        }
        tracing::info!(
            session = %self.id,
            count = activations.len(),
            "reload: tool activations handed over from the previous image"
        );
        if self.runtime.is_live() {
            self.runtime.restore_tool_activations(activations);
        } else {
            for a in activations {
                if !self.parked_tool_activations.iter().any(|p| p.tool_id == a.tool_id) {
                    self.parked_tool_activations.push(a);
                }
            }
        }
    }

    /// Host-only (`SessionCommand::AdoptWorkers`, sent by reload's
    /// rehydrate): the workers the previous daemon image stopped or still
    /// owed a wake. Their ids resolve to their archives again, and the
    /// parent is woken once per worker — it was waiting for completions the
    /// exec destroyed.
    pub(crate) fn adopt_workers(&mut self, workers: Vec<crate::runtime::subagent::CarriedWorker>) {
        self.adopt_carried_workers(workers, false);
    }

    /// [`Self::adopt_workers`]; `failed_reload`: this image's own exec
    /// failed (`ReloadAborted`), so the notice must not say it reloaded.
    fn adopt_carried_workers(
        &mut self,
        workers: Vec<crate::runtime::subagent::CarriedWorker>,
        failed_reload: bool,
    ) {
        if workers.is_empty() {
            return;
        }
        {
            let mut registry = self
                .runtime
                .subagent_registry()
                .lock()
                .unwrap_or_else(|p| p.into_inner());
            for w in &workers {
                registry.adopt_carried(w);
                if w.needs_wake() {
                    // The next reload repeats this wake only while it is
                    // still queued.
                    registry.note_reload_wake(w);
                }
            }
        }
        let woken: Vec<&crate::runtime::subagent::CarriedWorker> =
            workers.iter().filter(|w| w.needs_wake()).collect();
        for w in &woken {
            let ev = crate::tools::build_reload_carry_event(w);
            if self.event_queue.push(ev.clone()).is_err() {
                self.event_queue.push_priority(ev);
            }
        }
        let stopped: Vec<&str> = woken
            .iter()
            .filter(|w| w.stopped_by_reload)
            .map(|w| w.id.as_str())
            .collect();
        tracing::info!(
            session = %self.id,
            carried = workers.len(),
            stopped = ?stopped,
            woken = woken.len(),
            "reload: adopted workers from the previous image"
        );
        if !stopped.is_empty() {
            self.emit(SessionEventWire::SystemNotice(format!(
                "{} — worker{} {} stopped by the reload; subagent_resume continues {} from {} archive",
                if failed_reload { "daemon reload failed" } else { "daemon reloaded" },
                if stopped.len() == 1 { "" } else { "s" },
                stopped.join(", "),
                if stopped.len() == 1 { "it" } else { "them" },
                if stopped.len() == 1 { "its" } else { "their" },
            )));
        }
    }

    /// `stop_for_reload` on this session's registry (fences new launches).
    fn stop_workers_for_reload(
        &self,
    ) -> Vec<(String, Arc<std::sync::RwLock<crate::runtime::subagent::SubagentState>>)> {
        self.runtime
            .subagent_registry()
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .stop_for_reload()
    }

    /// Host-only (`SessionCommand::ReloadAborted`): the reload this session
    /// checkpointed for did not exec, so the old image keeps serving it.
    /// Lift the reload fence, hand the workers the checkpoint stopped back
    /// to the parent, and deliver what queued up meanwhile.
    pub(crate) async fn reload_aborted(&mut self) {
        self.lift_reload_fence(
            "daemon reload failed — this session continues on the running daemon",
        )
        .await;
    }

    /// Lift the reload fence (see [`Self::reload_aborted`]), telling the
    /// user `notice` (the reload did not happen).
    async fn lift_reload_fence(&mut self, notice: &str) {
        let Some(fence) = self.reload_fence.take() else {
            return;
        };
        self.emit(SessionEventWire::SystemNotice(notice.to_string()));
        let workers = if self.runtime.is_live() {
            // From the LIVE registry, not only the checkpoint's snapshot: a
            // worker registered after it was stopped at birth by the
            // registry's reload fence and is owed its wake too. Snapshot
            // entries the registry no longer holds ride along.
            let mut carried = self.reload_carried_workers();
            carried.extend(
                fence
                    .workers
                    .into_iter()
                    .filter(|w| !carried.iter().any(|c| c.id == w.id))
                    .collect::<Vec<_>>(),
            );
            let mut registry = self
                .runtime
                .subagent_registry()
                .lock()
                .unwrap_or_else(|p| p.into_inner());
            registry.lift_reload_fence();
            // The queue the record assumed lost is still here: only the
            // reload's own stops owe a wake.
            registry.rewake_after_failed_exec(carried)
        } else {
            Vec::new()
        };
        tracing::info!(session = %self.id, "reload did not exec: fence lifted");
        self.adopt_carried_workers(workers, true);
        self.rearm_park();
        if !self.event_queue.is_empty() {
            self.on_queue_wake().await;
        }
    }

    /// Backstop: no exec and no `ReloadAborted` within
    /// [`budgets::RELOAD_FENCE_TIMEOUT`] of the checkpoint.
    async fn reload_fence_expired(&mut self) {
        tracing::warn!(
            session = %self.id,
            "reload fence expired with no exec and no abort from the daemon; lifting it"
        );
        self.lift_reload_fence(
            "daemon reload never completed or reported back — this session continues on the running daemon",
        )
        .await;
    }

    /// Abort the spawned job (`CompactionCancelled`); prior state intact.
    pub(crate) fn abort_compaction(&mut self) {
        if let Some(job) = self.compact.take() {
            job.task.abort();
            self.emit(SessionEventWire::CompactionCancelled);
            self.publish_presence();
        }
    }

    // ── command dispatch ─────────────────────────────────────────────────

    /// B1: a client-stamped input command from anyone but the owner is
    /// `Refused` with no side effect; `from: None` (host) bypasses.
    async fn handle(&mut self, addressed: Addressed) -> std::ops::ControlFlow<EndReason> {
        use std::ops::ControlFlow;
        let Addressed { from, cmd } = addressed;
        if let Some(c) = from {
            // Only the daemon's reload sends a reload checkpoint (it stops
            // every worker and fences the session for the exec after it).
            if matches!(cmd, SessionCommand::Checkpoint { reason: CheckpointReason::Reload }) {
                self.emit(SessionEventWire::Refused {
                    client: c,
                    command: command_name(&cmd).to_string(),
                    reason: "checkpoint(reload) is host-only; use `synaps daemon reload`".into(),
                });
                return ControlFlow::Continue(());
            }
            if let Some(reason) = is_input_command(&cmd).then(|| self.input_refusal(c)).flatten() {
                self.emit(SessionEventWire::Refused {
                    client: c,
                    command: command_name(&cmd).to_string(),
                    reason,
                });
                return ControlFlow::Continue(());
            }
        }
        // A reload checkpointed this session: nothing may start work or
        // change what the reload record captured until the exec (or until
        // the daemon reports it failed). The wake sources are fenced in the
        // select loop; commands are fenced here.
        if self.reload_fence.is_some() && cmd.refused_while_reloading() {
            let reason = super::RELOADING_REFUSAL.to_string();
            match from {
                Some(c) => self.emit(SessionEventWire::Refused {
                    client: c,
                    command: command_name(&cmd).to_string(),
                    reason,
                }),
                None => self.emit(SessionEventWire::SystemNotice(reason)),
            }
            return ControlFlow::Continue(());
        }
        // Accepted input: with `client_input = shared` its sender becomes the
        // active client. `KeepWarm` doesn't count: `--keep-warm` sends it
        // on attach, not a person typing.
        if let Some(c) = from {
            if is_input_command(&cmd) && !matches!(cmd, SessionCommand::KeepWarm { .. }) {
                self.note_input(c);
            }
        }
        // B3: anything that needs the runtime wakes a parked session first.
        if self.is_parked() {
            let needs_runtime = !matches!(
                cmd,
                SessionCommand::Attach { .. }
                    | SessionCommand::Detach { .. }
                    | SessionCommand::End { .. }
                    | SessionCommand::Query { .. }
                    | SessionCommand::Answer { .. }
                    | SessionCommand::Save
                    | SessionCommand::Cancel
                    | SessionCommand::Checkpoint { .. }
                    | SessionCommand::Park
                    | SessionCommand::ReloadAborted
                    | SessionCommand::ArmPark
                    | SessionCommand::RestoreToolActivations { .. }
                    | SessionCommand::Resync { .. }
                    | SessionCommand::HostEvent(_)
                    | SessionCommand::TakeInput
            ) && !is_close_command(&cmd) // closing a parked session must not rebuild it
                // Unpinning a parked session only clears the flag; pinning
                // one wakes it (a pinned session is live by definition).
                && !matches!(cmd, SessionCommand::KeepWarm { on: false });
            if matches!(cmd, SessionCommand::HostEvent(_)) {
                return ControlFlow::Continue(()); // nobody attached, nothing to render
            }
            if matches!(
                cmd,
                SessionCommand::Save
                    | SessionCommand::Cancel
                    | SessionCommand::Park
                    | SessionCommand::ArmPark
                    | SessionCommand::Checkpoint { .. }
            ) {
                if let SessionCommand::Checkpoint { .. } = cmd {
                    let record = serde_json::to_value(self.reload_record()).unwrap_or_default();
                    self.emit(SessionEventWire::QueryResult {
                        id: super::wire::CHECKPOINT_QUERY_ID,
                        value: serde_json::json!({ "ok": true, "parked": true, "record": record }),
                    });
                }
                return ControlFlow::Continue(()); // already on disk / idle
            }
            if needs_runtime {
                if let Err(e) = self.ensure_live().await {
                    match from {
                        Some(c) => self.emit(SessionEventWire::Refused {
                            client: c,
                            command: command_name(&cmd).to_string(),
                            reason: e,
                        }),
                        None => self.emit(SessionEventWire::SystemNotice(e)),
                    }
                    return ControlFlow::Continue(());
                }
            }
        }
        match cmd {
            SessionCommand::Submit { text, attachments } => self.submit(text, attachments, from).await,
            SessionCommand::Steer { text } => {
                if self.streaming {
                    self.steer(text)
                } else {
                    self.submit(text, vec![], from).await
                }
            }
            SessionCommand::SubagentSteer { handle_id, text } => {
                self.steer_subagent(from, &handle_id, text)
            }
            SessionCommand::Cancel => {
                // The user's cancel, even of a driver's turn: noted before
                // the revoke below cancels that turn.
                self.note_cancel_cause(crate::engine::interrupt::InterruptReason::User);
                if self.driver.is_some() {
                    self.driver_revoke("canceled");
                }
                self.cancel_turn(crate::engine::interrupt::InterruptReason::User)
                    .await;
            }
            SessionCommand::Answer { prompt_id, value } => self.answer(from, prompt_id, value),
            SessionCommand::Set { id, setting } => self.apply_setting(id, setting).await,
            // `CompactionStarted` is the contract; no notice before it.
            SessionCommand::Compact { instructions } => {
                self.compact(instructions, "manual").await;
            }
            SessionCommand::NewSession => {
                if self.driver.is_some() {
                    self.driver_revoke("session replaced");
                }
                // A turn still waiting for `on_session_start` would open its
                // stream later on the cleared history, under the new journal
                // id: cancel it first, so it ends (and is saved) in the old
                // session with nothing sent.
                if self.turn_awaiting_start.is_some() {
                    self.cancel_turn(crate::engine::interrupt::InterruptReason::User).await;
                }
                // Before `clear` saves the old session: a ping in flight is
                // aborted and booked to it, not to the new one.
                self.cache_keepwarm_conversation_replaced();
                // `clear` saves the old session directly: queued saves of it
                // must land first, never after.
                self.persister.flush().await;
                self.conv.clear(&self.runtime).await;
                self.runtime
                    .begin_conversation(Some(self.conv.session.id.clone()));
                self.set_journal_id(self.conv.session.id.clone());
                // F10: lock follows the new session id.
                self.reacquire_session_lock(&self.conv.session.id.clone());
                self.emit(SessionEventWire::Cleared {
                    session_id: self.conv.session.id.clone(),
                });
                self.emit_conversation();
            }
            SessionCommand::Save => {
                self.save().await;
            }
            SessionCommand::Query { id, query } => self.query(id, query).await,
            // `/close` ends the session, which no other engine command can.
            SessionCommand::EngineCommand { id, name, arg } if name == CLOSE_COMMAND => {
                if let Some(reason) = self.close_command(id, &arg) {
                    return ControlFlow::Break(reason);
                }
            }
            SessionCommand::EngineCommand { id, name, arg } => {
                self.engine_command(id, name, arg).await
            }
            SessionCommand::Attach { client, mode } => self.attach(client, mode).await,
            SessionCommand::Detach { client } => {
                self.detach(client);
                // Nobody left to notice a stale period: confirm the prefix
                // now rather than at the next ping, so a session whose next
                // request already changed is free to park.
                if self.attached.is_empty() && self.cache_keepwarm.active() {
                    self.cache_keepwarm_recheck().await;
                }
            }
            SessionCommand::End { reason } => {
                // `finish` cancels a running turn for this reason; note it
                // before the revoke below cancels a driver turn first.
                self.note_cancel_cause(match reason {
                    EndReason::ClientQuit => crate::engine::interrupt::InterruptReason::User,
                    _ => crate::engine::interrupt::InterruptReason::Host,
                });
                if self.driver.is_some() {
                    self.driver_revoke("session ending");
                }
                return ControlFlow::Break(reason);
            }
            SessionCommand::Resync { .. } => self.emit(SessionEventWire::SystemNotice(
                "resync not supported yet".into(),
            )),
            // A3 bodies live in actor_cmds.rs; B1 (Checkpoint), B3 (KeepWarm)
            // fill in the rest.
            SessionCommand::SubmitPrepared {
                messages,
                user_text,
            } => self.submit_prepared(messages, user_text).await,
            SessionCommand::PluginCommand {
                id,
                plugin,
                name,
                arg,
            } => self.plugin_command(id, plugin, name, arg).await,
            SessionCommand::Resume { id, query } => self.resume(id, query).await,
            SessionCommand::Checkpoint { reason } => self.checkpoint(reason).await,
            SessionCommand::AdoptWorkers { workers } => self.adopt_workers(workers),
            SessionCommand::RestoreToolActivations { activations } => {
                self.restore_tool_activations(activations)
            }
            SessionCommand::ReloadAborted => self.reload_aborted().await,
            // Parked sessions returned above (their evict deadline stands).
            SessionCommand::ArmPark => self.rearm_park(),
            SessionCommand::Park => {
                // Explicit request: end an active keep-warm period, then park.
                self.cache_keepwarm_park_requested();
                if let std::ops::ControlFlow::Break(reason) = self.park().await {
                    return ControlFlow::Break(reason);
                }
            }
            SessionCommand::TakeInput => self.take_input(from),
            SessionCommand::KeepWarm { on } => {
                self.keep_warm = on;
                self.meta.pinned = on; // the attach snapshot's copy
                self.keep_warm_cell.store(on, std::sync::atomic::Ordering::Release);
                self.note_pin();
                // Unpinned with nobody who ever typed: back to the old rule.
                if !on && !self.cache_keepwarm.interactive_seen && self.cache_keepwarm.active() {
                    self.cache_keepwarm_end(super::cache_keepwarm::PeriodEnd::Disabled);
                }
                self.rearm_park();
                self.emit(SessionEventWire::SystemNotice(if on {
                    PIN_ON_NOTICE.to_string()
                } else {
                    PIN_OFF_NOTICE.to_string()
                }));
            }
            SessionCommand::HostEvent(ev) => match ev {
                HostEvent::ExtensionNotification {
                    extension_id,
                    method,
                    params,
                } => self.emit(SessionEventWire::ExtensionNotification {
                    extension_id,
                    method,
                    params,
                }),
                HostEvent::LoaderProgress(ev) => self.emit(SessionEventWire::LoaderProgress(ev)),
            },
            // E-P3: driver start handler.
            SessionCommand::DriverStart { plugin, command, arg } => {
                self.driver_start(plugin, command, arg);
            }
        }
        ControlFlow::Continue(())
    }

    // ── teardown (tui/mod.rs:352-432 + chat.rs shutdown) ─────────────────

    async fn finish(&mut self, reason: EndReason) {
        // Ended on purpose (`/close`, `daemon close`, a client's quit): the
        // pin goes with it. A daemon stop, reload or crash keeps it.
        if self.keep_warm && reason == EndReason::ClientQuit {
            self.keep_warm = false;
            self.note_pin();
        }
        self.cache_keepwarm_end(super::cache_keepwarm::PeriodEnd::SessionEnding);
        // Before on_session_end clears the keyed injection (STEP 2): a hook
        // not yet called leaves the line; one in flight finishes holding the
        // gate and drops what it injects (`SessionStart::end`).
        self.session_start.end();
        self.lifecycle.store(
            SessionLifecycle::Ending as u8,
            std::sync::atomic::Ordering::Release,
        );
        // Quitting the client mid-turn is the user's interruption; every
        // other end reason is the host's.
        let interrupt = match reason {
            EndReason::ClientQuit => crate::engine::interrupt::InterruptReason::User,
            _ => crate::engine::interrupt::InterruptReason::Host,
        };
        // Before the revoke below cancels a driver turn.
        self.note_cancel_cause(interrupt);
        // E-P3: clean up driver before teardown.
        if self.driver.is_some() {
            self.driver_revoke("session ending");
        }
        if self.streaming {
            self.cancel_turn(interrupt).await;
        }
        self.abort_compaction();
        // Outstanding prompts are cancelled (tool sees `None`).
        while let Some((pr, tx)) = self.pending_prompts.pop_front() {
            let _ = tx.send(None);
            self.emit(SessionEventWire::PromptResolved { prompt_id: pr.id });
        }

        let parked = !self.conv.is_live();
        let session_id = (**self.journal_id.load()).clone();
        let api_messages = if parked {
            None
        } else {
            Some(self.conv.api_messages.clone())
        };

        // STEP 1: save — own bounded budget, highest priority. A parked
        // session is already on disk: end record only. The wait covers
        // everything queued, including the cancelled turn's save.
        let persist = self.config.persist;
        if !parked {
            self.request_save();
        }
        let persister = &self.persister;
        let save_fut = async {
            if persist {
                if !persister.flush().await {
                    tracing::warn!(session = %session_id, "session end: save failed");
                }
                let mut index_record =
                    crate::core::session_index::SessionIndexRecord::end(&session_id);
                index_record.turns = api_messages.as_ref().map(|m| m.len());
                if let Err(err) = crate::core::session_index::append_record(&index_record) {
                    tracing::warn!("failed to append session end index record: {}", err);
                }
            }
        };
        if tokio::time::timeout(budgets::SAVE_TIMEOUT, save_fut)
            .await
            .is_err()
        {
            tracing::warn!(
                budget_secs = budgets::SAVE_TIMEOUT_SECS,
                "session save timed out — data may be incomplete"
            );
        }

        // STEP 2 (C2): on_session_end — per session, concurrent, fail-open,
        // own budget; clears this session's keyed injection.
        let hook_bus = Arc::clone(&self.hook_bus);
        crate::extensions::loader::emit_session_end(
            &hook_bus,
            &session_id,
            api_messages,
            budgets::HOOKS_TIMEOUT,
        )
        .await;

        // STEP 3: bounded observability flush (live only).
        if self.runtime.is_live() {
            if let Some(outcome) = self
                .runtime
                .shutdown_observability_async(
                    crate::runtime::telemetry::DEFAULT_SHUTDOWN_FLUSH_TIMEOUT,
                )
                .await
            {
                if !outcome.is_flushed() {
                    tracing::warn!(
                        stats = ?outcome.stats(),
                        "observability flush timed out — detached worker keeps draining"
                    );
                }
            }
        }

        // Inbox watcher, per-session UDS, registry entry, keyed injection.
        self.background.shutdown();
        self.emit(SessionEventWire::Ended { reason });
    }

    pub fn id(&self) -> &SessionId {
        &self.id
    }
}

/// The tokio task. `SessionTask::run` is the reactor loop.
pub struct SessionTask(SessionActor);

async fn next_stream_event(stream: &mut Option<ActiveStream>) -> Option<StreamEvent> {
    match stream {
        Some(s) => s.next().await,
        None => std::future::pending().await,
    }
}

/// The conversation's title for listings: the saved title, else what
/// `Session::auto_title` will make of the first user message.
fn conversation_title(conv: &ConversationState) -> Option<String> {
    if !conv.session.title.is_empty() {
        return Some(conv.session.title.clone());
    }
    conv.api_messages
        .iter()
        .filter(|m| m["role"].as_str() == Some("user"))
        .find_map(|m| agent_core::core::session::user_content_for_display(&m["content"]))
        .map(|t| t.chars().take(80).collect())
}

/// Whether the 1 Hz `SubagentRows` tick keeps running: always during a turn,
/// and after it while any worker is still running (a background worker's
/// progress events rode the ended turn's stream; the rows carry it now).
fn keep_subagent_tick(streaming: bool, any_running: bool) -> bool {
    streaming || any_running
}

/// Which running workers an interrupted turn cancels. Only the user's own
/// interrupt of a live session is turn-scoped: Esc stops what that turn
/// started, not reactive workers inherited from earlier turns (which the
/// user may be waiting on). Quitting/teardown (`ending`), the cost cap, host
/// checkpoints/restarts and driver revocation stop every running worker.
fn worker_cancel_scope(
    reason: crate::engine::interrupt::InterruptReason,
    ending: bool,
    turn_generation: u64,
) -> crate::runtime::subagent::WorkerCancelScope {
    use crate::engine::interrupt::InterruptReason;
    use crate::runtime::subagent::WorkerCancelScope;
    match reason {
        InterruptReason::User if !ending => WorkerCancelScope::Turn(turn_generation),
        _ => WorkerCancelScope::All,
    }
}

async fn next_tick(tick: &mut Option<tokio::time::Interval>) {
    match tick {
        Some(t) => {
            t.tick().await;
        }
        None => std::future::pending().await,
    }
}

async fn poll_compaction(
    job: &mut Option<CompactionJob>,
) -> std::result::Result<Result<crate::runtime::compaction::CompactionOutcome>, tokio::task::JoinError> {
    match job {
        Some(j) => (&mut j.task).await,
        None => std::future::pending().await,
    }
}

async fn park_timer(deadline: Option<tokio::time::Instant>) {
    match deadline {
        Some(t) => tokio::time::sleep_until(t).await,
        None => std::future::pending().await,
    }
}

/// `revoked_turn_deadline`: same shape as `park_timer`.
async fn revoked_turn_timer(deadline: Option<tokio::time::Instant>) {
    match deadline {
        Some(t) => tokio::time::sleep_until(t).await,
        None => std::future::pending().await,
    }
}

/// (P11) Prompt-abandonment deadline: same shape as `park_timer`. `None` =
/// disabled (no abandoned prompt, or the feature is off) → pends forever.
async fn prompt_abandon_timer(deadline: Option<tokio::time::Instant>) {
    match deadline {
        Some(t) => tokio::time::sleep_until(t).await,
        None => std::future::pending().await,
    }
}

/// The reload fence's backstop deadline; pends forever when unfenced.
async fn reload_fence_timer(deadline: Option<tokio::time::Instant>) {
    match deadline {
        Some(t) => tokio::time::sleep_until(t).await,
        None => std::future::pending().await,
    }
}

/// 200 ms cadence, gated on driver state — `None` driver+pending → pending forever.
async fn driver_tick_timer(
    has_driver: bool,
    has_pending: bool,
    interval: &mut tokio::time::Interval,
) {
    if has_driver || has_pending {
        interval.tick().await;
    } else {
        std::future::pending().await
    }
}

/// How long a session waits for process-level extension discovery before
/// `on_session_start`: `startup.extensions_ready_timeout_secs`, or the
/// budget constant when that is 0.
fn extensions_ready_timeout(config: &crate::SynapsConfig) -> Duration {
    match config.startup.extensions_ready_timeout_secs {
        0 => budgets::EXTENSIONS_READY_TIMEOUT,
        secs => Duration::from_secs(secs),
    }
}

/// This session's `on_session_start` task, while it is in line or running.
async fn session_start_task(
    start: &mut super::session_start::SessionStart,
) -> std::result::Result<bool, tokio::task::JoinError> {
    match start {
        super::session_start::SessionStart::Queued { task, .. } => task.await,
        _ => std::future::pending().await,
    }
}

impl SessionTask {
    pub fn id(&self) -> &SessionId {
        self.0.id()
    }

    /// Test seam: this session's subagent registry, to seed fake workers
    /// before `run()` (see `EngineHost::create_session_unspawned`).
    #[cfg(any(test, feature = "testing"))]
    pub fn subagent_registry(
        &self,
    ) -> Arc<std::sync::Mutex<crate::runtime::subagent::SubagentRegistry>> {
        Arc::clone(self.0.runtime.subagent_registry())
    }

    /// Test seam: a clone of this session's runtime. Clones share the
    /// memory-consent state, so a harness can revoke consent the way the
    /// `memory_context` tool's capability does (`memory_context_disable`).
    #[cfg(any(test, feature = "testing"))]
    pub fn runtime_for_harness(&self) -> crate::Runtime {
        (*self.0.runtime).clone()
    }

    /// Test seam: this session's event queue (what a worker's finalizer
    /// pushes its completion wake into).
    #[cfg(any(test, feature = "testing"))]
    pub fn event_queue(&self) -> Arc<crate::events::EventQueue> {
        Arc::clone(&self.0.event_queue)
    }

    /// Unbiased select (the TUI loop is unbiased too).
    pub async fn run(mut self) {
        let queue = Arc::clone(&self.0.event_queue);
        let reason = loop {
            let actor = &mut self.0;
            tokio::select! {
                cmd = actor.cmd_rx.recv() => match cmd {
                    Some(cmd) => {
                        if let std::ops::ControlFlow::Break(reason) = actor.handle(cmd).await {
                            break reason;
                        }
                    }
                    // Every handle dropped: nobody can ever reach us again.
                    None => break EndReason::HostShutdown,
                },
                Some(req) = actor.secret_prompt_rx.recv() => actor.on_prompt_request(req),
                // A held wake (and its permit) stays queued.
                _ = queue.notified(), if !actor.holds(WorkSource::QueueWake) => actor.on_queue_wake().await,
                _ = next_tick(&mut actor.subagent_tick) => {
                    actor.publish_subagent_rows();
                    actor.flush_turn_draft();
                }
                _ = park_timer(actor.park_deadline), if !actor.holds(WorkSource::ParkTimer) => {
                    if let std::ops::ControlFlow::Break(reason) = actor.park().await {
                        break reason;
                    }
                }
                _ = prompt_abandon_timer(actor.prompt_abandon_deadline) => {
                    actor.on_prompt_abandon_deadline();
                }
                _ = super::cache_keepwarm::keepwarm_timer(actor.cache_keepwarm.deadline()), if !actor.holds(WorkSource::KeepWarmTimer) => {
                    actor.on_cache_keepwarm_due().await;
                }
                res = super::cache_keepwarm::poll_keepwarm_ping(&mut actor.cache_keepwarm.task) => {
                    actor.on_cache_keepwarm_done(res);
                }
                res = poll_compaction(&mut actor.compact) => actor.on_compaction_done(res).await,
                ran = session_start_task(&mut actor.session_start) => {
                    actor.session_start_finished(ran).await;
                },
                _ = driver_tick_timer(actor.driver.is_some(), actor.driver_pending.is_some(), &mut actor.driver_tick_interval), if !actor.holds(WorkSource::DriverTick) => {
                    actor.driver_tick().await;
                }
                _ = reload_fence_timer(actor.reload_fence.as_ref().map(|f| f.deadline)) => {
                    actor.reload_fence_expired().await;
                }
                _ = revoked_turn_timer(actor.revoked_turn_deadline) => {
                    actor.finish_revoked_turn(false).await;
                }
                ev = next_stream_event(&mut actor.stream) => match ev {
                    Some(ev) => actor.on_stream_event(ev).await,
                    // A revoked turn's stream ending is its end.
                    None if actor.turn_cancelled() => actor.finish_revoked_turn(true).await,
                    None => {
                        // Stream ended without a terminal event: defensive reset.
                        // P5: capture_terminal(None) = EOF → revoke driver.
                        if actor.driver.as_ref().is_some_and(|d| d.awaiting_terminal) {
                            let terminal = super::driver::capture_terminal(None, false);
                            if let Some(terminal) = terminal {
                                if let Some(driver) = actor.driver.as_mut() {
                                    let reason = super::driver::observe_terminal(driver, &actor.runtime, terminal);
                                    // F-NEW-1 (shady): EOF is a Blocked terminal.
                                    // Emit a DriverTurnOutcome before revoking so a
                                    // client tracking outcomes sees the same
                                    // outcome+revoke pair the Error path produces —
                                    // not a turn that silently vanishes.
                                    let sel = driver.selection.clone();
                                    let fb = driver.grant.feedback_enabled()
                                        .then(|| driver.completed_feedback.to_string());
                                    actor.emit(SessionEventWire::DriverTurnOutcome {
                                        outcome: crate::extensions::session_driver::Outcome::Blocked,
                                        selection: sel,
                                        feedback: fb,
                                    });
                                    if let Some(reason) = reason {
                                        actor.driver_revoke(&reason);
                                    }
                                }
                            }
                        }
                        actor.clear_stream();
                        actor.emit_conversation();
                        actor.announce_idle().await;
                    }
                },
            }
        };
        self.0.finish(reason).await;
    }
}

/// Wait (bounded by [`budgets::WORKER_RELOAD_STOP_TIMEOUT`]) until every worker
/// `stop_for_reload` returned has settled its finalizer: a stopped worker's
/// resume archive is on disk and its in-flight tool was dropped, which kills
/// that tool's process group (after the exec nothing could); a worker that
/// finished on its own has pushed its completion wake into the queue the
/// reload carries. Returns the ids the reload stopped.
async fn await_stopped_workers(
    session: &str,
    stopping: Vec<(String, Arc<std::sync::RwLock<crate::runtime::subagent::SubagentState>>)>,
) -> Vec<String> {
    if stopping.is_empty() {
        return Vec::new();
    }
    let ids: Vec<String> = stopping.iter().map(|(id, _)| id.clone()).collect();
    tracing::info!(session, workers = ?ids, "reload: stopping running workers");
    let deadline = tokio::time::Instant::now() + budgets::WORKER_RELOAD_STOP_TIMEOUT;
    loop {
        let pending: Vec<&str> = stopping
            .iter()
            .filter(|(_, s)| !s.read().unwrap_or_else(|p| p.into_inner()).completion_settled)
            .map(|(id, _)| id.as_str())
            .collect();
        if pending.is_empty() {
            break;
        }
        if tokio::time::Instant::now() >= deadline {
            tracing::warn!(
                session,
                workers = ?pending,
                "reload: workers did not stop in time; only their last archive survives the exec"
            );
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
    }
    stopping
        .iter()
        .filter(|(_, s)| s.read().unwrap_or_else(|p| p.into_inner()).stopped_by_reload)
        .map(|(id, _)| id.clone())
        .collect()
}

#[cfg(test)]
mod worker_cancel_scope_tests {
    use super::worker_cancel_scope;
    use crate::engine::interrupt::InterruptReason;
    use crate::runtime::subagent::WorkerCancelScope;

    #[test]
    fn only_a_live_user_interrupt_is_turn_scoped() {
        assert_eq!(
            worker_cancel_scope(InterruptReason::User, false, 7),
            WorkerCancelScope::Turn(7),
            "Esc stops the interrupted turn's workers only"
        );
        assert_eq!(
            worker_cancel_scope(InterruptReason::User, true, 7),
            WorkerCancelScope::All,
            "quitting the client mid-turn tears every worker down"
        );
        for reason in [
            InterruptReason::CostCap,
            InterruptReason::Restart,
            InterruptReason::Host,
            InterruptReason::Driver,
        ] {
            assert_eq!(
                worker_cancel_scope(reason, false, 7),
                WorkerCancelScope::All,
                "{reason:?} stops every running worker"
            );
        }
    }
}

#[cfg(test)]
mod subagent_tick_tests {
    use super::keep_subagent_tick;

    #[test]
    fn rows_keep_flowing_after_the_turn_while_a_worker_runs() {
        assert!(keep_subagent_tick(true, false), "during a turn, always");
        assert!(keep_subagent_tick(true, true));
        assert!(
            keep_subagent_tick(false, true),
            "after the turn, while a background worker runs"
        );
        assert!(
            !keep_subagent_tick(false, false),
            "stops once the turn is over and no worker runs"
        );
    }
}

#[cfg(test)]
mod parked_evict_tests {
    use super::parked_evict_after_from;
    use super::prompt_abandon_timeout_from;
    use std::time::Duration;

    #[test]
    fn parked_evict_defaults_to_one_hour_and_honours_never() {
        // cfg_secs = 3600 mirrors the DaemonConfig default.
        let d = 3600u64;
        assert_eq!(parked_evict_after_from(None, d), Some(Duration::from_secs(3600)));
        assert_eq!(parked_evict_after_from(Some("120"), d), Some(Duration::from_secs(120)));
        for never in ["never", "0", "off"] {
            assert_eq!(parked_evict_after_from(Some(never), d), None, "{never:?}");
        }
        assert_eq!(parked_evict_after_from(Some("junk"), d), Some(Duration::from_secs(3600)));
        // env absent → config wins; config 0 disables; env still beats config.
        assert_eq!(parked_evict_after_from(None, 42), Some(Duration::from_secs(42)));
        assert_eq!(parked_evict_after_from(None, 0), None);
        assert_eq!(parked_evict_after_from(Some("7"), 42), Some(Duration::from_secs(7)));
    }

    #[test]
    fn prompt_abandon_defaults_to_one_hour_and_honours_never() {
        // cfg_secs = 3600 mirrors the DaemonConfig default.
        let d = 3600u64;
        assert_eq!(prompt_abandon_timeout_from(None, d), Some(Duration::from_secs(3600)));
        assert_eq!(prompt_abandon_timeout_from(Some("30"), d), Some(Duration::from_secs(30)));
        // Disabled sentinels ⇒ None ⇒ pure pre-#112 (prompt survives forever).
        for never in ["never", "0", "off"] {
            assert_eq!(prompt_abandon_timeout_from(Some(never), d), None, "{never:?}");
        }
        // Garbage falls back to the safe default rather than disabling the guard.
        assert_eq!(prompt_abandon_timeout_from(Some("junk"), d), Some(Duration::from_secs(3600)));
        // env absent → config wins; config 0 disables; env still beats config.
        assert_eq!(prompt_abandon_timeout_from(None, 42), Some(Duration::from_secs(42)));
        assert_eq!(prompt_abandon_timeout_from(None, 0), None);
        assert_eq!(prompt_abandon_timeout_from(Some("7"), 42), Some(Duration::from_secs(7)));
    }
}