deepstrike_core/scheduler/state_machine/

mod.rs

use std::collections::{HashMap, HashSet, VecDeque};

use compact_str::CompactString;

use super::entropy::{EntropyTracker, EntropyWatchConfig};
use super::milestone::MilestoneTracker;
use super::policy::SchedulerBudget;
use super::tcb::{DurableWaitSet, TaskLifecycle, TaskTable, Tcb, WaitSet};
use crate::AgentRunSpec;
use crate::context::manager::ContextManager;
use crate::context::measurement::ToolMeasurement;
use crate::context::renderer::InternalRenderedContext;
use crate::governance::pipeline::GovernancePipeline;
use crate::governance::repeat_fuse::RepeatFuseConfig;
use crate::signals::router::SignalRouter;
use crate::types::result::SubAgentResult;
// `pub use` so external integration tests that glob `state_machine::*` resolve the observation
// type here — exactly as they did for the former `pub enum LoopObservation` this replaced.
pub use crate::runtime::kernel::KernelObservation;
use crate::runtime::session::RollbackReason;
use crate::types::message::{
    Content, ContentPart, CoreMessage, ToolCall, ToolErrorKind, ToolResult, ToolSchema,
};
use crate::types::milestone::MilestoneCheckResult;
use crate::types::result::{LoopResult, TerminationReason};
use crate::types::task::RuntimeTask;

/// Compact digest of a tool call's arguments for the recency log (2b). Kept short and CJK-safe — it
/// only needs to make `same-tool / different-args` calls distinguishable (so a legit loop isn't
/// flagged as a no-progress repeat) and to read sensibly in the "just did: …" footer. Empty for
/// no-arg / `{}` calls. Lives in the volatile State turn, so length here never churns the cache.
///
/// The 2c STOP and the O6 fuse compare these digests for EQUALITY, so identity must cover the
/// FULL arguments even though the display truncates: serde_json orders keys alphabetically, and
/// a long leading value (an `edit` call's `file_path`) otherwise swallows the whole window —
/// collapsing distinct same-file edits into one signature. A truncated digest therefore carries
/// a hash of the complete canonical JSON as its suffix.
fn compact_tool_args(args: &serde_json::Value) -> String {
    if args.is_null() {
        return String::new();
    }
    let s = args.to_string();
    if s == "{}" {
        return String::new();
    }
    const MAX: usize = 48;
    if s.chars().count() <= MAX {
        s
    } else {
        // FNV-1a 64 folded to 32 bits: deterministic across processes/replays (no SipHash
        // random keys), 8 hex chars of noise in a footer line that already truncates.
        let mut h: u64 = 0xcbf2_9ce4_8422_2325;
        for b in s.as_bytes() {
            h ^= u64::from(*b);
            h = h.wrapping_mul(0x0000_0100_0000_01b3);
        }
        let fold = (h ^ (h >> 32)) as u32;
        format!("{}…#{fold:08x}", s.chars().take(MAX).collect::<String>())
    }
}

/// The *turn step* of the L* execution loop (M1d).
///
/// Schedulability (`Ready/Running/Blocked/Suspended/Done`) is no longer carried here — it lives
/// on the root task's [`TaskLifecycle`] in the kernel's `TaskTable`, queried via
/// [`LoopStateMachine::lifecycle`]. `LoopPhase` is now orthogonal: it only records *which step of a
/// running turn* the loop is in. When the task is `Ready/Suspended/Done`, the phase value is
/// inert (left at its last step) and ignored.
#[derive(Debug, Clone)]
pub enum LoopPhase {
    Reason,
    Act { tool_calls: Vec<ToolCall> },
}

/// Events fed into the state machine from the SDK layer.
#[derive(Debug)]
pub enum LoopEvent {
    LLMResponse {
        message: CoreMessage,
    },
    ToolResults {
        results: Vec<ToolResult>,
    },
    /// Result of evaluating the current milestone phase's criteria.
    /// Feed this back after handling `LoopAction::EvaluateMilestone`.
    MilestoneResult {
        result: MilestoneCheckResult,
    },
    /// Sub-agent run completed — result is injected into the loop as context.
    SubAgentCompleted {
        result: SubAgentResult,
    },
    Complete,
    Timeout,
}

/// Actions the state machine outputs — SDK layer executes the I/O.
#[derive(Debug, Clone)]
pub enum LoopAction {
    /// Structured context ready for a provider call.
    /// `context.system_text` → provider system param.
    /// `context.turns`       → provider messages array (strictly alternating).
    /// `tools`               → tool schemas (skill / memory / knowledge / user tools).
    CallLLM {
        context: InternalRenderedContext,
        tools: Vec<ToolSchema>,
    },
    ExecuteTools {
        calls: Vec<ToolCall>,
    },
    /// Host-owned approval effect. The kernel remains suspended until the host
    /// returns the correlated result through the ABI.
    RequestApproval {
        requests: Vec<ApprovalRequest>,
    },
    /// Host-owned workflow orchestration effect. The kernel has reserved the
    /// batch but records no spawn fact until the correlated result arrives.
    SpawnWorkflow {
        nodes: Vec<crate::orchestration::workflow::WorkflowSpawnInfo>,
        budget: Option<crate::orchestration::workflow::WorkflowBudget>,
    },
    /// Host-owned cancellation of in-flight child agents.
    PreemptSubAgents {
        agent_ids: Vec<String>,
        reason: String,
    },
    PersistMemory {
        memory: crate::mm::memory::MemoryRecord,
    },
    QueryMemory {
        query: crate::mm::memory::MemoryQuery,
        requested_k: usize,
    },
    ArchivePageOut {
        turn: u32,
        action: crate::runtime::kernel::KernelPressureAction,
        summary: Option<String>,
        archived: Vec<CoreMessage>,
        tier: String,
    },
    Done {
        result: LoopResult,
    },
    /// Kernel requests the SDK to evaluate the current milestone phase.
    ///
    /// The SDK should assess `criteria` against the agent's output using the
    /// specified `verifier`, then feed back `LoopEvent::MilestoneResult { result }`.
    EvaluateMilestone {
        phase_id: String,
        criteria: Vec<String>,
        verifier: Option<crate::types::milestone::MilestoneVerifier>,
        required_evidence: Vec<String>,
    },
    /// Kernel is suspended awaiting a non-approval internal continuation.
    AwaitingResume,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ApprovalRequest {
    pub call_id: String,
    pub tool: String,
    pub arguments: serde_json::Value,
    pub reason: String,
}

#[derive(Debug, Clone)]
pub(super) struct PendingWorkflowSpawn {
    pub nodes: Vec<crate::orchestration::workflow::WorkflowSpawnInfo>,
    pub budget: Option<crate::orchestration::workflow::WorkflowBudget>,
}

#[derive(Debug, Clone)]
pub(super) struct PendingPreempt {
    pub agent_ids: Vec<String>,
    pub reason: String,
}

#[derive(Debug, Clone)]
pub(super) enum PendingHostEffect {
    ArchivePageOut {
        turn: u32,
        action: crate::runtime::kernel::KernelPressureAction,
        summary: Option<String>,
        archived: Vec<CoreMessage>,
        tier: String,
    },
}

impl PendingHostEffect {
    fn action(&self) -> LoopAction {
        match self {
            Self::ArchivePageOut {
                turn,
                action,
                summary,
                archived,
                tier,
            } => LoopAction::ArchivePageOut {
                turn: *turn,
                action: *action,
                summary: summary.clone(),
                archived: archived.clone(),
                tier: tier.clone(),
            },
        }
    }
}

/// Payload held while the loop is in `Suspended`.
#[derive(Debug, Clone)]
pub(super) enum SuspendState {
    /// Governance AskUser — awaiting a correlated approval result.
    AskUser {
        calls: Vec<ToolCall>,
        gated_reasons: HashMap<String, String>,
    },
    /// Sub-agent spawn — awaiting `SubAgentCompleted` for each listed agent id.
    SubAgentAwait { agent_ids: Vec<String> },
}

pub(super) enum GateToolOutcome {
    Proceed,
    Blocked(LoopAction),
    ApprovalRequired(Vec<ApprovalRequest>),
}

/// One P1 syscall the kernel adjudicated itself, and the answer the model reads for it.
///
/// A syscall call is never dispatched to a host, but it *is* a tool call the model made, so it
/// still gets a tool result — the v0.2.42 rule that the model-facing surface stays a training-set
/// convention. `is_error` is what distinguishes "the kernel did it" from "the kernel refused".
#[derive(Debug, Clone)]
pub struct AnsweredCall {
    pub call_id: CompactString,
    pub output: String,
    pub is_error: bool,
}

/// What a provider turn does when the kernel's own adjudication left nothing for a host to run.
///
/// Spec adjudication §5k: a batch of pure control-plane calls (`skill`, `update_plan`) publishes no
/// effect, so without this the operation would have nothing outstanding and stall.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum IdleContinuation {
    /// §5k · nothing is outstanding, so the kernel continues the turn the only way a turn
    /// continues: it calls the provider again.
    #[default]
    CallProvider,
    /// The adjudication already published kernel-owned work (a memory effect, a spawn round). The
    /// turn resumes when that work resolves; issuing a provider call now would race it.
    Await,
}

/// What the canonical driver adjudicated *before* the provider turn it is about to feed (§7.6).
///
/// Staged rather than passed as an argument so `feed(LLMResponse)` stays the single place that
/// decides what a provider turn means. An absent value means there were no pre-turn answers and
/// the provider is the idle continuation.
#[derive(Debug, Clone, Default)]
pub struct AdjudicatedTurn {
    pub answered_calls: Vec<AnsweredCall>,
    pub idle_continuation: IdleContinuation,
}

/// Outcome of the P1 fail-closed exposure gate: the calls that survived, or the committed action
/// when every call in the batch was denied.
pub(super) enum ExposureGateOutcome {
    Proceed(Vec<ToolCall>),
    Blocked(LoopAction),
}

/// Snapshot of context lengths captured just before each LLM call.
/// Used internally to restore state on rollback.
#[derive(Debug, Clone, Default)]
pub struct TurnCheckpoint {
    pub history_len: usize,
    pub signals_len: usize,
    pub task_state: Option<crate::context::task_state::TaskState>,
}

/// Pure state machine for the L* execution loop. No I/O — only state transitions.
///
/// Internal engine backing the canonical operation driver. Exposed for in-crate use and tests;
/// external callers drive it through [`crate::runtime::kernel::wire::CanonicalKernel`].
#[doc(hidden)]
pub struct LoopStateMachine {
    pub phase: LoopPhase,
    pub turn: u32,
    pub ctx: ContextManager,
    pub tools: Vec<ToolSchema>,
    pub observations: Vec<KernelObservation>,
    pub(super) policy: SchedulerBudget,
    pub(super) scheduler_policy: crate::scheduler::policy::SchedulerPolicyConfig,
    pub(super) total_tokens: u64,
    /// Reservation-backed hard limits for this operation. Shared accounting stays in the host;
    /// the kernel tracks only this run's local usage.
    pub(super) budget_grant: Option<crate::runtime::kernel::wire::BudgetGrant>,
    pub(super) local_rounds_completed: u32,
    /// spc_005-04: the `BudgetGrant` `evaluate_spawn_quota_inner` just reserved from the parent's
    /// `child_budget_remaining`, awaiting attachment to the child `Tcb` the caller is about to
    /// construct. Set only on a successful hierarchical-budget check; taken (and cleared) by the
    /// spawn call site immediately after, so it never survives past the syscall that produced it.
    pub(super) pending_budget_grant: Option<crate::scheduler::budget_grant::BudgetGrant>,
    /// ③ the adjudicated `pace` decision awaiting attachment to this round's LoopResult.
    pub(super) pending_pace: Option<crate::types::result::PaceDecision>,
    /// When set, the next LLM call strips tools to force a text response,
    /// then terminates with this reason once the response arrives.
    pub(super) pending_termination: Option<TerminationReason>,
    /// Reactive context-overflow recovery: consecutive compact-and-retry attempts since the last
    /// successful provider turn. Bounds the recovery ladder (anti-spiral) and resets to 0 on any
    /// `LLMResponse`, mirroring the per-turn `hasAttemptedReactiveCompact` reset the SDK runners
    /// used to own. See `recover_from_provider_error`.
    pub(super) recovery_attempts: u8,
    pub(crate) provider_recovery_attempt_limit: u8,
    /// Max-output-tokens recovery: consecutive continue-and-retry turns since the model last
    /// finished a response WITHOUT hitting the output cap. When a turn is cut off at the cap
    /// (provider `stop_reason` = max_tokens/length) the kernel keeps the partial, nudges the model
    /// to resume mid-thought, and re-calls — bounded by `MAX_OUTPUT_RECOVERY` (mirrors query.ts's
    /// MAX_OUTPUT_TOKENS_RECOVERY_LIMIT). Resets to 0 on any non-truncated response.
    pub(super) output_recovery_attempts: u8,
    pub(crate) output_recovery_attempt_limit: u8,
    /// Whether the in-flight response was cut off at the provider's output cap. Set just before
    /// `feed(LLMResponse)` and taken (cleared) inside it.
    ///
    /// A **classified** fact, never vendor text (§22.8): the canonical provider outcome supplies a
    /// typed stop reason and [`Self::set_output_truncated`] stages the derived boolean.
    pub(super) pending_output_truncated: bool,
    /// §7.6 · what the canonical driver adjudicated for the provider turn about to be fed. Consumed
    /// (and cleared) inside `feed(LLMResponse)`; `None` until the canonical driver stages a turn.
    pub(super) adjudicated_turn: Option<AdjudicatedTurn>,
    /// Number of history messages present at session start (after preload_history).
    /// drain_new_messages() returns the slice from this offset onward.
    pub(super) session_history_baseline: usize,
    pub(super) checkpoint: TurnCheckpoint,
    /// Milestone contract tracker (extracted to reduce state machine bloat).
    pub(super) milestone: MilestoneTracker,
    pub run_spec: Option<AgentRunSpec>,
    /// M1 収口: the single source of truth for schedulability *and* sub-agent lineage. Root is
    /// task `"root"`; each sub-agent is a child task carrying its `ProcInfo`. The former
    /// `ProcessTable` is now a derived view over this (`agent_process(es)` rebuild `AgentProcess`
    /// rows on demand via `AgentProcess::from_tcb`).
    pub(super) tasks: TaskTable,
    /// Optional governance pipeline. When set, every tool call proposed by the
    /// model is evaluated before `ExecuteTools` is emitted. `None` (default)
    /// skips the gate entirely, preserving the pre-governance behavior.
    pub(super) governance: Option<GovernancePipeline>,
    /// P1 fail-closed dispatch: the toolset advertised to the model on the most recent `CallLLM`.
    /// Refreshed at every emission (`call_llm_action`), consumed by `gate_exposed_tool_calls`.
    ///
    /// `None` means this process has not advertised a toolset yet. It remains fail-closed for task
    /// tools, and canonical checkpoints preserve the distinction from an advertised empty set.
    pub(super) exposed_tool_names: Option<HashSet<CompactString>>,
    /// Optional resource quota evaluated at the syscall trap (M2). `None` (default) leaves spawn /
    /// memory syscalls unconditionally allowed, preserving pre-M2 behavior.
    pub(super) resource_quota: Option<crate::governance::quota::ResourceQuota>,
    /// Timestamps of recent allowed `WriteMemory` syscalls, for the rolling-window rate limit.
    /// Only populated when `resource_quota.memory_writes_per_window` is set.
    pub(super) memory_write_times: Vec<u64>,
    /// Kernel-owned signal routing: dedup set + attention policy + bounded queue.
    /// Always initialized; `set_attention` rebuilds it with a new queue size.
    pub(super) signal_router: SignalRouter,
    /// Prefix of `ctx.partitions.signals` included in the currently pending provider request.
    /// A correlated `ProviderResult` consumes exactly this prefix; signals arriving while the
    /// provider is in flight remain for the next request.
    pub(super) delivered_signals_len: usize,
    /// Accepted envelope time of the operation's first timed transition.
    /// Used by the wall-time budget axis in `SchedulerBudget::should_terminate`.
    pub(super) started_at_ms: Option<u64>,
    /// Most-recent accepted envelope time, forwarded to the budget check.
    pub(super) last_now_ms: Option<u64>,
    /// Tool batch awaiting `Resume` after an AskUser suspend.
    pub(super) suspend_state: Option<SuspendState>,
    /// Denied tool results to merge into the next `ToolResults` feed after resume.
    pub(super) pending_denied_results: Vec<ToolResult>,
    /// Host-provided tool accounting keyed by call id. Measurements stay outside canonical
    /// `ToolResult` state and are consumed exactly once at the tool-result boundary.
    pub(super) pending_tool_measurements: HashMap<CompactString, u32>,
    /// W0: an in-flight workflow DAG, when one is loaded. The kernel spawns its ready nodes as
    /// gated batches (each through `evaluate_syscall(Syscall::Spawn)`) and advances on
    /// completions. `None` (default) preserves the single-spawn `spawn_sub_agent` behavior.
    pub(super) workflow: Option<crate::orchestration::workflow::WorkflowRun>,
    /// Whether the in-flight workflow **is** this operation's root (spec §6.1 invariant 7).
    ///
    /// Immutable for the lifetime of a run: it mirrors the canonical `RootKind`, which no committed
    /// transition may change. `false` means a workflow nested inside an agent loop, whose completion resumes the parent
    /// agent with another provider call. `true` makes that completion the operation's terminal
    /// instead, which is what deletes the host-side `CompleteRun` race (§10.1 现状注记).
    pub(super) root_workflow: bool,
    /// Spec §10.4 / §15.3 · whether a spawned child waits for the host's launch acknowledgement
    /// before it counts as `Running`.
    ///
    /// Child launch follows the single canonical arc: `PendingLaunch` when the kernel mints
    /// identity, `Starting` when it publishes the effect, and `Running` only after `TasksSpawned`.
    /// Workflow batch reserved by the kernel and awaiting the host's correlated
    /// spawn result. This is intent, not an observed external fact.
    pub(super) pending_workflow_spawn: Option<PendingWorkflowSpawn>,
    pub(super) pending_preempt: Option<PendingPreempt>,
    /// Ordered host-owned durability effects produced during a pure state-machine
    /// transition. The normal continuation is held until every effect commits.
    pub(super) pending_host_effects: VecDeque<PendingHostEffect>,
    pub(super) active_host_effect: Option<PendingHostEffect>,
    pub(super) deferred_action: Option<Box<LoopAction>>,
    /// O6: repeat-fuse thresholds (the hard rungs above the 2c soft STOP). Default enabled with
    /// generous thresholds; tune/disable via `SetRepeatFuse` / `ConfigureRun.repeat_fuse`.
    pub(super) repeat_fuse: RepeatFuseConfig,
    /// O6: the previous turn's action signature (non-meta `name(args)` joined — the same key the
    /// 2c STOP uses). NOT part of the turn checkpoint: a fuse deny's rollback must not launder
    /// the streak it just tripped on.
    pub(super) repeat_sig: Option<String>,
    /// O6: consecutive turns whose signature equalled `repeat_sig` (1 = first occurrence).
    pub(super) repeat_count: u32,
    /// O4: turn-end criteria gate (the Stop-hook analog). When the model finishes (no tool calls)
    /// while explicit acceptance criteria stand, inject ONE bounded self-check turn before
    /// accepting `Completed`. 2c guards "won't stop"; this guards "stops too early".
    pub(super) criteria_gate_enabled: bool,
    /// O4: whether the gate already fired this run (it fires at most once — no nag loops).
    pub(super) criteria_gate_fired: bool,
    /// Session-entropy sliding window + watch state (see `scheduler::entropy`). Like the
    /// RepeatFuse streak, NOT part of the turn checkpoint — a rollback must not launder
    /// the disorder it just evidenced.
    pub(super) entropy: EntropyTracker,
    /// Opt-in threshold watch over the per-turn entropy score. Default disabled; the
    /// unconditional per-turn `EntropySample` observation does not depend on it.
    pub(super) entropy_watch: EntropyWatchConfig,
}

mod cancellation;
mod capability;
mod eviction;
mod gate;
mod milestone_exec;
mod process;
mod signal;
mod workflow;

impl LoopStateMachine {
    fn message_tokens(&self, message: &CoreMessage) -> u32 {
        self.ctx.engine.count_message(message)
    }

    pub fn new(policy: SchedulerBudget) -> Self {
        let mut tasks = TaskTable::new();
        // M1d: the root task carries the authoritative schedulability lifecycle. It starts
        // `Ready`; `start()`/`resume_*` flip it to `Running`, suspends set `Suspended`, and
        // `terminate()` sets `Done`. `phase` is now only the intra-turn step.
        tasks.insert(Tcb::root("root", policy.clone()));
        Self {
            // Inert placeholder step; meaningful only while the root task is `Running`.
            phase: LoopPhase::Reason,
            turn: 0,
            ctx: ContextManager::new(policy.max_tokens),
            tools: Vec::new(),
            observations: Vec::new(),
            policy,
            scheduler_policy: crate::scheduler::policy::SchedulerPolicyConfig::default(),
            total_tokens: 0,
            budget_grant: None,
            pending_budget_grant: None,
            local_rounds_completed: 0,
            pending_pace: None,
            pending_termination: None,
            recovery_attempts: 0,
            provider_recovery_attempt_limit: 2,
            output_recovery_attempts: 0,
            output_recovery_attempt_limit: 3,
            pending_output_truncated: false,
            adjudicated_turn: None,
            session_history_baseline: 0,
            checkpoint: TurnCheckpoint::default(),
            milestone: MilestoneTracker::new(),
            run_spec: None,
            tasks,
            governance: None,
            exposed_tool_names: None,
            resource_quota: None,
            memory_write_times: Vec::new(),
            signal_router: SignalRouter::new(64),
            delivered_signals_len: 0,
            started_at_ms: None,
            last_now_ms: None,
            suspend_state: None,
            pending_denied_results: Vec::new(),
            pending_tool_measurements: HashMap::new(),
            workflow: None,
            root_workflow: false,
            pending_workflow_spawn: None,
            pending_preempt: None,
            pending_host_effects: VecDeque::new(),
            active_host_effect: None,
            deferred_action: None,
            repeat_fuse: RepeatFuseConfig::default(),
            repeat_sig: None,
            repeat_count: 0,
            criteria_gate_enabled: true,
            criteria_gate_fired: false,
            entropy: EntropyTracker::default(),
            entropy_watch: EntropyWatchConfig::default(),
        }
    }

    /// Feed external tool results with host-owned accounting evidence. The evidence is consumed
    /// exactly once by the ordinary `ToolResults` transition and never becomes canonical message
    /// state.
    pub fn feed_tool_results_with_measurements(
        &mut self,
        results: Vec<ToolResult>,
        measurements: impl IntoIterator<Item = ToolMeasurement>,
    ) -> LoopAction {
        self.pending_tool_measurements.extend(
            measurements
                .into_iter()
                .map(|measurement| (CompactString::from(measurement.call_id), measurement.tokens)),
        );
        let action = self.feed(LoopEvent::ToolResults { results });
        self.pending_tool_measurements.clear();
        action
    }

    /// O4: enable/disable the turn-end criteria gate (default enabled; no-op without criteria).
    pub fn set_criteria_gate(&mut self, enabled: bool) {
        self.criteria_gate_enabled = enabled;
    }

    /// Declare that the workflow this machine runs **is** the operation's root (spec §6.1.7).
    ///
    /// Set once, before the DAG is installed, and never unset — it mirrors the immutable
    /// `RootKind`. Its only consumer is `finish_workflow`, which terminates instead of calling the
    /// provider again. A nested (agent-authored) workflow leaves it `false`.
    pub fn set_root_workflow(&mut self, is_root: bool) {
        self.root_workflow = is_root;
    }

    /// Whether the in-flight workflow is this operation's root.
    pub fn is_root_workflow(&self) -> bool {
        self.root_workflow
    }

    /// The schedulability state of one task, for host projections and tests.
    pub fn task_lifecycle(&self, task_id: &str) -> Option<TaskLifecycle> {
        self.tasks.get(task_id).map(|task| task.state)
    }

    /// Fine-grained capabilities held by a task. Skill activation uses the derived syscall caller
    /// as the parent set for the same attenuation rule that governs child-task delegation.
    pub fn task_capabilities(&self, task_id: &str) -> &[crate::types::capability::Capability] {
        self.tasks
            .get(task_id)
            .map(|task| task.capabilities.as_slice())
            .unwrap_or(&[])
    }

    /// The root operation's capability set, used by host control-plane mutations.
    pub fn root_capabilities(&self) -> &[crate::types::capability::Capability] {
        self.task_capabilities("root")
    }

    /// §10.4 · the launch effect for these tasks has been published. Moves each of them from
    /// `PendingLaunch` to `Starting`, and never downgrades a task that already advanced.
    pub fn mark_tasks_starting(&mut self, task_ids: &[String]) {
        for id in task_ids {
            if let Some(task) = self.tasks.get_mut(id)
                && task.state == TaskLifecycle::PendingLaunch
            {
                task.state = TaskLifecycle::Starting;
            }
        }
    }

    /// §7.6 · whether the named workflow node ran under quarantine (it read untrusted content).
    /// A quarantined caller may not use a P1 syscall to widen its own authority, so the canonical
    /// driver consults this before it admits a workflow / memory / capability request.
    /// Errs closed only in the sense that an unknown id is *not* quarantined — an id the kernel
    /// never issued is refused earlier, by causation derivation.
    pub fn task_quarantined(&self, task_id: &str) -> bool {
        self.workflow
            .as_ref()
            .is_some_and(|run| run.is_agent_quarantined(task_id))
    }

    /// Test instrument for the §7.6 quarantine refusal. The canonical wire `WorkflowNode` carries
    /// no trust level yet (SPEC-ISSUE in the canonical driver), so a test cannot declare a
    /// quarantined node through the contract the driver reads.
    #[cfg(test)]
    pub(crate) fn quarantine_task_for_test(&mut self, task_id: &str) -> bool {
        self.workflow
            .as_mut()
            .is_some_and(|run| run.quarantine_agent(task_id))
    }

    pub(crate) fn set_scheduler_policy(
        &mut self,
        policy: crate::scheduler::policy::SchedulerPolicyConfig,
    ) {
        self.scheduler_policy = policy;
        if let Some(workflow) = self.workflow.as_mut() {
            workflow.set_scheduler_policy(policy);
        }
    }

    /// Install the two semantic recovery ladders the kernel owns (§13.2 `ReplaceRecoveryPolicy`).
    /// Both ladders are always stated here — a resolved policy has no "unset" — so a patch can
    /// lower *and* raise within the validated ceiling.
    pub fn set_recovery_limits(&mut self, provider_attempts: u8, output_attempts: u8) {
        self.provider_recovery_attempt_limit = provider_attempts;
        self.output_recovery_attempt_limit = output_attempts;
    }

    pub(crate) fn externalize_pending_host_effect(
        &mut self,
        continuation: LoopAction,
    ) -> LoopAction {
        if self.active_host_effect.is_some() {
            return continuation;
        }
        let Some(pending) = self.pending_host_effects.pop_front() else {
            return continuation;
        };
        assert!(
            self.deferred_action.is_none(),
            "host effect continuation must be unique"
        );
        self.deferred_action = Some(Box::new(continuation));
        self.active_host_effect = Some(pending);
        self.active_host_effect
            .as_ref()
            .expect("host effect was just activated")
            .action()
    }

    fn next_after_host_effect(&mut self) -> LoopAction {
        if let Some(pending) = self.pending_host_effects.pop_front() {
            self.active_host_effect = Some(pending);
            self.active_host_effect
                .as_ref()
                .expect("host effect was just activated")
                .action()
        } else {
            match self.deferred_action.take().map(|action| *action) {
                // Durability effects can change rendered context and conditional meta-tools
                // (notably `read_result`). Never return the pre-commit frozen provider action.
                Some(LoopAction::CallLLM { .. }) => self.emit_call_llm(),
                Some(action) => action,
                None => LoopAction::AwaitingResume,
            }
        }
    }

    /// Commit the archive the host performed and release the continuation it was holding.
    pub(crate) fn commit_page_out_archive(&mut self, archive_ref: Option<String>) -> LoopAction {
        let pending = self
            .active_host_effect
            .as_ref()
            .expect("page-out result requires an active host effect");
        let PendingHostEffect::ArchivePageOut {
            turn,
            action,
            summary,
            archived,
            tier,
        } = pending;
        self.observations.push(KernelObservation::PageOutArchived {
            turn: *turn,
            action: *action,
            summary: summary.clone(),
            tier: tier.clone(),
            message_count: archived.len() as u32,
            archive_ref,
        });
        self.active_host_effect = None;
        self.next_after_host_effect()
    }

    /// DEC-5 · the canonical decision for an archive the host could not perform: **abandon it**.
    ///
    /// The compaction it belongs to already happened in this kernel — the summary is in context and
    /// the evicted bodies are gone either way — so the run stays live and degraded rather than
    /// dying on a best-effort durability effect. The failure is a typed, replayable audit fact and
    /// the kernel never re-emits the same archive; a host that wants another attempt asks again
    /// with a new causation.
    pub(crate) fn abandon_page_out_archive(&mut self, error: String) -> LoopAction {
        self.push_page_out_archive_failure(error);
        self.active_host_effect = None;
        self.next_after_host_effect()
    }

    fn push_page_out_archive_failure(&mut self, error: String) {
        let pending = self
            .active_host_effect
            .as_ref()
            .expect("page-out failure requires an active host effect");
        let PendingHostEffect::ArchivePageOut {
            turn,
            action,
            archived,
            tier,
            ..
        } = pending;
        self.observations
            .push(KernelObservation::PageOutArchiveFailed {
                turn: *turn,
                action: *action,
                tier: tier.clone(),
                message_count: archived.len() as u32,
                error,
            });
    }

    /// O6: tune or disable the repeat fuse (see [`RepeatFuseConfig`]).
    pub fn set_repeat_fuse(&mut self, config: RepeatFuseConfig) {
        self.repeat_fuse = config;
    }

    /// Configure the opt-in entropy threshold watch (see [`EntropyWatchConfig`]).
    /// The per-turn `EntropySample` observation is unconditional and unaffected.
    pub fn set_entropy_watch(&mut self, config: EntropyWatchConfig) {
        self.entropy_watch = config;
    }

    pub fn entropy_watch_config(&self) -> EntropyWatchConfig {
        self.entropy_watch
    }

    pub(crate) fn entropy_checkpoint_state(
        &self,
    ) -> crate::scheduler::entropy::EntropyTrackerRuntimeState {
        self.entropy.checkpoint_state()
    }

    pub(crate) fn restore_entropy_checkpoint_state(
        &mut self,
        state: crate::scheduler::entropy::EntropyTrackerRuntimeState,
    ) -> Result<(), String> {
        self.entropy.restore_state(state, self.turn)
    }

    /// O6: the active repeat-fuse config (for read-modify-write from the ABI event).
    pub fn repeat_fuse_config(&self) -> RepeatFuseConfig {
        self.repeat_fuse
    }

    /// The authoritative schedulability lifecycle of the loop (root task state). Replaces the
    /// removed `LoopPhase::{Idle,Suspended,Blocked,Terminal}` reads.
    pub fn lifecycle(&self) -> TaskLifecycle {
        self.tasks
            .get("root")
            .map(|t| t.state)
            .unwrap_or(TaskLifecycle::Ready)
    }

    /// The canonical durable wait set while suspended, if any.
    pub fn wait_set(&self) -> Option<DurableWaitSet> {
        self.tasks
            .get("root")
            .and_then(|task| task.wait_set.clone())
    }

    /// Whether the loop has terminated.
    pub fn is_terminal(&self) -> bool {
        matches!(self.lifecycle(), TaskLifecycle::Done(_))
    }

    /// Whether the loop is suspended awaiting external resolution.
    pub fn is_suspended(&self) -> bool {
        matches!(self.lifecycle(), TaskLifecycle::Suspended)
    }

    /// §7.9 · the operation ended because a host executor failed on an effect the loop cannot do
    /// without. Closes the loop so a later input finds a terminated kernel rather than one that
    /// still believes it is running.
    ///
    /// Deliberately produces no `LoopResult`: the terminal belongs to the canonical driver, and a
    /// second one minted here would give the same event two representations (§7.12).
    pub fn close_for_host_effect_failure(&mut self) {
        self.set_lifecycle(TaskLifecycle::Done(TerminationReason::Error), None);
    }

    /// Set the root task's lifecycle and canonical wait state.
    fn set_lifecycle(&mut self, state: TaskLifecycle, wait: Option<WaitSet>) {
        if self.tasks.get("root").is_none() {
            self.tasks.insert(Tcb::root("root", self.policy.clone()));
        }
        if let Some(root) = self.tasks.get_mut("root") {
            root.state = state;
        }
        if let Some(wait) = wait {
            self.tasks.register_wait_set("root", wait);
        } else {
            self.tasks.clear_wait("root");
        }
    }

    /// spc_009-05: convert the RunGroup admission grant (`tokens`/`subagents`/`rounds`, a coarse
    /// whole-operation ceiling) into the [`crate::scheduler::budget_grant::ResourceBudget`] shape
    /// spc_005's hierarchical checks understand. Axes with no source field (`cost_microunits`/
    /// `wall_ms`/`concurrent_children`/`tool_calls`/`memory_writes`/`object_bytes`) stay `None` —
    /// `ResourceBudget`'s own "unset means unbounded" convention, not an omission.
    fn root_child_budget_seed(&self) -> Option<super::budget_grant::ResourceBudget> {
        self.budget_grant
            .as_ref()
            .map(|grant| super::budget_grant::ResourceBudget {
                tokens: grant.tokens.map(|t| t.get()),
                child_tasks: grant.subagents,
                turns: grant.rounds,
                ..super::budget_grant::ResourceBudget::default()
            })
    }

    /// Build a transient root [`Tcb`] mirroring the current scheduling facts (budget counters,
    /// wall-clock anchors, lifecycle) so the pure scheduler applies the same budget verdict.
    fn root_tcb(&self) -> Tcb {
        let mut tcb = Tcb::root("root", self.policy.clone());
        tcb.budget.turns = self.turn;
        tcb.budget.total_tokens = self.total_tokens;
        if let Some(tokens) = self
            .budget_grant
            .as_ref()
            .and_then(|grant| grant.tokens)
            .map(crate::runtime::kernel::wire::WireU64::get)
        {
            tcb.budget.limits.max_total_tokens = tcb.budget.limits.max_total_tokens.min(tokens);
        }
        tcb.budget.started_at_ms = self.started_at_ms;
        tcb.state = self.lifecycle();
        tcb
    }

    /// Adjust the wall-clock budget axis at runtime.
    pub fn set_wall_budget(&mut self, max_wall_ms: Option<u64>) {
        self.policy.max_wall_ms = max_wall_ms;
    }

    /// The wall-clock budget axis as it currently stands — the value an `UpdateDeadline` command
    /// last projected onto it. Read by the §12.1 checkpoint projection, because a deadline that a
    /// restore forgot would silently un-bound the run.
    pub fn wall_budget(&self) -> Option<u64> {
        self.policy.max_wall_ms
    }

    // ----- §12.2 · restore -----
    //
    // Narrow, one-fact-each setters, not a "load this snapshot" door. Each one writes back exactly
    // one value the §12.1 projection reads, so "what a checkpoint restores" is decided by the DTO
    // and enforced by the restore's own digest re-check — never by whatever this struct happens to
    // hold. Anything not reachable through these is, by construction, state the checkpoint declares
    // rebuildable (the loop phase, the exposed toolset, the frozen-prefix marker).

    /// Reinstall the counters the budget axes are evaluated against.
    pub fn restore_budget_usage(&mut self, total_tokens: u64, rounds_completed: u32) {
        self.total_tokens = total_tokens;
        self.local_rounds_completed = rounds_completed;
    }

    /// Reinstall the anchor the wall-clock axis measures from.
    pub fn restore_started_at_ms(&mut self, started_at_ms: Option<u64>) {
        self.started_at_ms = started_at_ms;
    }

    /// The task table, for a restore to repopulate. `insert` is idempotent per task id, so a
    /// restore that runs twice produces one table, not two.
    pub fn task_table_mut(&mut self) -> &mut TaskTable {
        &mut self.tasks
    }

    /// Reinstall the rolling memory-write window the syscall gate rate-limits against.
    pub fn restore_memory_write_window(&mut self, window: Vec<u64>) {
        self.memory_write_times = window;
    }

    /// The accepted time this operation started measuring wall-clock budget from, if any input has
    /// carried a clock yet. The wall axis is a *duration* from here, so an absolute deadline is
    /// projected onto it as `deadline − start`.
    pub fn started_at_ms(&self) -> Option<u64> {
        self.started_at_ms
    }

    /// Install a governance pipeline. Once set, all model-proposed tool calls
    /// are evaluated before execution. Denied/rate-limited calls commit visible
    /// error tool results; `AskUser` calls surface a `ToolGated` observation for
    /// the SDK to enforce.
    pub fn set_governance(&mut self, pipeline: GovernancePipeline) {
        self.governance = Some(pipeline);
    }

    /// Install resource quotas (M2). Once set, `Spawn` and `WriteMemory` syscalls are bounded by
    /// the quota at the trap. Not setting it (the default) leaves them unconditionally allowed.
    pub fn set_resource_quota(&mut self, quota: crate::governance::quota::ResourceQuota) {
        self.resource_quota = Some(quota);
    }

    pub fn set_budget_grant(&mut self, grant: crate::runtime::kernel::wire::BudgetGrant) {
        self.budget_grant = Some(grant);
        // spc_009-05: seed root's own grantable pool from the RunGroup admission grant the Host
        // just declared. `new()` already inserted root (eagerly, before any grant is known), so
        // this is the first point a grant actually exists to seed from — `set_lifecycle`'s
        // lazy-insert branch never fires for root and cannot be the hook. `None` stays `None`:
        // this only activates the spc_005 hierarchical check when the Host declared a real
        // admission grant, never unconditionally.
        let seed = self.root_child_budget_seed();
        if let Some(root) = self.tasks.get_mut("root") {
            root.child_budget_remaining = seed;
        }
    }

    /// spc_009-04: seed the operation root's own delegatable `Capability` set from
    /// `InitialContext.requested_capabilities` — the Host input a `StartOperation` carries.
    /// Mirrors `set_budget_grant`'s hook: root already exists (`new()` inserts it eagerly), so
    /// this writes directly rather than waiting on `set_lifecycle`'s lazy-insert branch, which
    /// never fires for root. An empty request leaves `capabilities` at its `Vec::new()` default —
    /// no unconditional grant for operations that declared none.
    pub fn set_requested_capabilities(
        &mut self,
        capabilities: Vec<crate::types::capability::Capability>,
    ) {
        if let Some(root) = self.tasks.get_mut("root") {
            root.capabilities = capabilities;
        }
    }

    /// L1: this vehicle's cumulative sub-agent spawns this run — every child task ever registered in
    /// the `TaskTable` (running + completed), distinct from the *instantaneous* running count. Used
    /// for the cumulative spawn quota and read back by the SDK to charge the group ledger at run end.
    pub fn local_subagents_spawned(&self) -> u32 {
        self.tasks.all().iter().filter(|t| t.proc.is_some()).count() as u32
    }

    pub fn local_budget_usage(&self) -> (u64, u32, u32) {
        (
            self.total_tokens,
            self.local_subagents_spawned(),
            self.local_rounds_completed,
        )
    }

    pub fn budget_grant(&self) -> Option<&crate::runtime::kernel::wire::BudgetGrant> {
        self.budget_grant.as_ref()
    }

    /// Timestamps of the recent allowed memory writes — the rolling window the syscall-gate rate
    /// limit is evaluated against.
    ///
    /// Read by the §12.1 checkpoint projection: the window is a gate *input*, so a checkpoint that
    /// dropped it would hand the restored run a fresh quota.
    pub fn memory_write_window(&self) -> &[u64] {
        &self.memory_write_times
    }

    /// §11.2 · ingest the **accepted envelope time** of the input being planned.
    ///
    /// The canonical driver calls this once per transition, before any semantic call, so that
    /// every clock-dependent decision this step makes — signal TTL and deadline escalation, the
    /// governance rate-limit window, the wall-time budget axis, idle time-decay — reads the one
    /// host clock fact the journal already holds. The kernel itself never reads a system clock.
    ///
    /// Beyond [`Self::set_observed_time`] it anchors context *activity* at the first accepted
    /// time. Without the anchor, `last_activity_ms` starts at 0 while the accepted clock is an
    /// epoch value, so the very first turn would look idle for ~55 years and trip time-decay
    /// compaction on an empty context.
    pub fn observe_accepted_time(&mut self, now_ms: u64) {
        let first = self.started_at_ms.is_none();
        self.set_observed_time(now_ms);
        if first {
            self.ctx.record_activity(now_ms);
        }
    }

    /// Feed the current wall-clock time (ms) to scheduler/governance budget axes.
    pub fn set_observed_time(&mut self, now_ms: u64) {
        if self.started_at_ms.is_none() {
            self.started_at_ms = Some(now_ms);
        }
        self.last_now_ms = Some(now_ms);
        if let Some(pipeline) = self.governance.as_mut() {
            pipeline.set_time(now_ms);
        }
    }

    /// The provider's typed stop reason says whether the response was cut off at the output cap;
    /// core never parses vendor-specific text.
    pub fn set_output_truncated(&mut self, truncated: bool) {
        self.pending_output_truncated = truncated;
    }

    /// §7.6 · stage the canonical driver's adjudication of the provider turn about to be fed.
    pub fn stage_adjudicated_turn(&mut self, adjudicated: AdjudicatedTurn) {
        self.adjudicated_turn = Some(adjudicated);
    }

    /// The agent ids of the spawn batch the kernel published and is still waiting on. Empty when no
    /// launch is outstanding. A batch-level launch failure is charged against exactly this set.
    pub fn pending_spawn_agent_ids(&self) -> Vec<String> {
        self.pending_workflow_spawn
            .as_ref()
            .map(|pending| {
                pending
                    .nodes
                    .iter()
                    .map(|node| node.agent_id.clone())
                    .collect()
            })
            .unwrap_or_default()
    }

    /// The tool calls this turn dispatched and is still waiting on. Empty outside an `Act` phase.
    pub fn dispatched_tool_calls(&self) -> Vec<ToolCall> {
        match &self.phase {
            LoopPhase::Act { tool_calls } => tool_calls.clone(),
            LoopPhase::Reason => Vec::new(),
        }
    }

    /// Pre-populate the history partition with messages from a prior session.
    ///
    /// Call **before** `start()` when resuming a conversation. Sets the baseline
    /// so `drain_new_messages()` returns only the messages from the current run.
    pub fn preload_history(&mut self, messages: Vec<CoreMessage>) {
        for msg in messages {
            let tokens = self.message_tokens(&msg);
            self.ctx.push_history(msg, tokens);
        }
        self.session_history_baseline = self.ctx.partitions.history.messages.len();
    }

    /// Continue from preloaded history without appending a new user turn.
    /// Use after `preload_history` when recovering a session that ended mid-run.
    ///
    /// If the last assistant turn has tool calls without matching tool results,
    /// resumes with `ExecuteTools` instead of calling the LLM again.
    ///
    /// "Unanswered" is read from history PLUS the results already synthesized this turn but not yet
    /// committed (`pending_denied_results`). Both matter because this is also the mid-turn
    /// continuation point: the kernel answers a `memory`/`knowledge` call by pushing hits into
    /// history and resuming here, while a denial from the same batch (fail-closed dispatch or a
    /// governance verdict) is still in flight and therefore invisible to a history-only scan.
    /// Re-dispatching such a call would execute a tool the kernel just refused AND give the model
    /// two results for one call_id. The filter keys on answered call_ids only, so a call that was
    /// never denied is still resumed — the wake-path behavior is untouched.
    pub fn resume_after_preload(&mut self) -> LoopAction {
        self.observations.clear();
        let mut calls = crate::runtime::repair::pending_tool_calls_from_messages(
            &self.ctx.partitions.history.messages,
        );
        if !self.pending_denied_results.is_empty() {
            let answered: HashSet<CompactString> = self
                .pending_denied_results
                .iter()
                .map(|result| result.call_id.clone())
                .collect();
            calls.retain(|call| !answered.contains(&call.id));
        }
        if !calls.is_empty() {
            self.phase = LoopPhase::Act {
                tool_calls: calls.clone(),
            };
            self.set_lifecycle(TaskLifecycle::Running, None);
            return LoopAction::ExecuteTools { calls };
        }
        self.phase = LoopPhase::Reason;
        self.emit_call_llm()
    }

    /// Return all messages added to history during the current run
    /// (since the last `preload_history` call or since construction).
    ///
    /// Call after `LoopAction::Done` to get the complete turn transcript
    /// for persistence to a SessionStore.
    pub fn drain_new_messages(&self) -> Vec<CoreMessage> {
        let history = &self.ctx.partitions.history.messages;
        let start = self.session_history_baseline.min(history.len());
        history[start..].to_vec()
    }

    pub fn start(&mut self, task: RuntimeTask) -> LoopAction {
        self.observations.clear();
        self.ctx.init_task(task.goal.clone(), task.criteria.clone());

        // A loop vehicle with no admitted round capacity must not make even one provider call.
        // The host may have raced another member between reading its durable loop log and reserve;
        // the reservation is the authoritative admission decision.
        let zero_round_grant = self
            .run_spec
            .as_ref()
            .and_then(|spec| spec.loop_round.as_ref())
            .is_some()
            && self.budget_grant.as_ref().and_then(|grant| grant.rounds) == Some(0);
        // A zero-token grant is the same exhausted admission on the token axis, but it binds
        // every vehicle, loop or not. Both axes can race to zero on one reservation; report
        // each before terminating so no provider call is ever dispatched.
        let zero_token_grant = self
            .budget_grant
            .as_ref()
            .and_then(|grant| grant.tokens)
            .is_some_and(|tokens| tokens.get() == 0);
        if zero_round_grant || zero_token_grant {
            if zero_round_grant {
                self.observations.push(KernelObservation::BudgetExceeded {
                    turn: self.turn,
                    budget: "rounds".into(),
                    operation_id: String::new(),
                    reservation_id: self
                        .budget_grant
                        .as_ref()
                        .map(|grant| grant.reservation_id.clone()),
                });
                self.pending_pace = Some(crate::types::result::PaceDecision {
                    action: crate::types::result::PaceAction::Stop,
                    delay_ms: None,
                    reason: "round budget grant exhausted before start".into(),
                    coerced_from: None,
                });
            }
            if zero_token_grant {
                self.observations.push(KernelObservation::BudgetExceeded {
                    turn: self.turn,
                    budget: "tokens".into(),
                    operation_id: String::new(),
                    reservation_id: self
                        .budget_grant
                        .as_ref()
                        .map(|grant| grant.reservation_id.clone()),
                });
            }
            // Token exhaustion is the harder stop: prefer it when both axes are zero.
            return self.terminate(
                if zero_token_grant {
                    TerminationReason::TokenBudget
                } else {
                    TerminationReason::Completed
                },
                None,
            );
        }

        let user_msg = "Proceed with the task described in [TASK STATE].".to_string();

        // User message goes into history so it appears at the correct chronological
        // position: [prior turns...] → [current user message] — LLM reads left-to-right
        // and responds to the last message. working is reserved for runtime signals only.
        // Estimate tokens (1 token ≈ 4 chars) with a minimum of 1 so the renderer
        // does not skip this message (it skips zero-token entries).
        let user_tokens = self.ctx.engine.count(&user_msg).max(1);
        self.ctx
            .push_history(CoreMessage::user(user_msg), user_tokens);
        self.phase = LoopPhase::Reason;
        // Root task (seeded `Ready` in `new()`) becomes `Running`; `emit_call_llm` sets it.
        self.emit_call_llm()
    }

    pub fn feed(&mut self, event: LoopEvent) -> LoopAction {
        self.observations.clear();
        self.sweep_expired_leases();
        // K3: skill leases expire on the same head-of-event cadence as capability leases.
        self.ctx.sweep_expired_skill_leases(self.turn);

        match event {
            LoopEvent::LLMResponse { message } => {
                // §7.6 · taken unconditionally, so a staged adjudication can never survive into a
                // later turn — a turn with no tool calls at all simply has nothing to apply it to.
                let adjudicated = self.adjudicated_turn.take().unwrap_or_default();
                let delivered = self
                    .delivered_signals_len
                    .min(self.ctx.partitions.signals.len());
                self.ctx.partitions.signals.drain(..delivered);
                self.delivered_signals_len = 0;
                // Signals admitted while the provider was in flight were not in the completed
                // request. Promote queued items at this boundary and keep a no-tool response from
                // terminating before the model receives them in a follow-up request.
                self.drain_queued_signals();
                let signals_waiting_for_followup = !self.ctx.partitions.signals.is_empty();
                // A response arrived ⇒ the prompt fit ⇒ the overflow recovery ladder is reset.
                self.recovery_attempts = 0;
                let tokens = self.message_tokens(&message);
                self.total_tokens += tokens as u64;

                // Max-output-tokens recovery (mirrors query.ts): a response cut off at the output
                // cap reports stop_reason = max_tokens (Anthropic) / length (OpenAI). A clean finish
                // resets the ladder.
                const OUTPUT_TRUNCATION_NUDGE: &str = "Output token limit hit. Resume directly — no apology, no recap of what you were doing. Pick up mid-thought if that is where the cut happened. Break remaining work into smaller pieces.";
                let truncated = std::mem::take(&mut self.pending_output_truncated);
                if !truncated {
                    self.output_recovery_attempts = 0;
                }

                if let Some(reason) = self.pending_termination.take() {
                    return self.terminate(reason, Some(message));
                }

                if message.tool_calls.is_empty() {
                    // The model was cut off at the output cap with no tool call. Keep the partial,
                    // nudge it to resume mid-thought, and re-call — instead of mistaking the
                    // truncation for a finished turn. Bounded by MAX_OUTPUT_RECOVERY; once exhausted
                    // the partial stands and the turn terminates normally below. (A truncated
                    // *tool-call* turn isn't handled here — it falls through to tool execution.)
                    if truncated
                        && self.output_recovery_attempts < self.output_recovery_attempt_limit
                    {
                        self.output_recovery_attempts += 1;
                        self.ctx.push_history(message, tokens);
                        self.ctx.push_signal(OUTPUT_TRUNCATION_NUDGE.to_string());
                        self.phase = LoopPhase::Reason;
                        return self.emit_call_llm();
                    }
                    // When a milestone contract is active and not yet complete,
                    // request evaluation instead of terminating.
                    if !self.milestone.is_complete() {
                        let phase_id = self.milestone.current_phase_id().unwrap_or("").to_string();
                        let criteria = self.milestone.current_criteria().to_vec();
                        let (verifier, required_evidence) = self
                            .milestone
                            .current_phase()
                            .map(|p| (p.verifier.clone(), p.required_evidence.clone()))
                            .unwrap_or_default();
                        // `tokens` was already computed for this message above.
                        self.ctx.push_history(message, tokens);
                        return LoopAction::EvaluateMilestone {
                            phase_id,
                            criteria,
                            verifier,
                            required_evidence,
                        };
                    }
                    // O4 criteria gate (the Stop-hook analog): the model is finishing while explicit
                    // acceptance criteria stand. Before accepting `Completed`, inject ONE bounded
                    // self-check at the peak-attention slot — verify each criterion, continue if any
                    // is unmet, else confirm. Fires at most once per run (no nag loop); runs with no
                    // criteria are untouched. 2c guards "won't stop"; this guards "stops too early".
                    if self.criteria_gate_enabled
                        && !self.criteria_gate_fired
                        && !self.ctx.partitions.task_state.criteria.is_empty()
                    {
                        self.criteria_gate_fired = true;
                        let criteria = self.ctx.partitions.task_state.criteria.clone();
                        self.ctx.push_history(message, tokens);
                        self.ctx.push_signal(format!(
                            "[CRITERIA CHECK] You are about to finish. Verify each acceptance \
                             criterion first: {}. If any is NOT met, continue working on it now. \
                             If all are met, give the final answer.",
                            criteria.join(" | ")
                        ));
                        self.observations
                            .push(KernelObservation::CriteriaGateFired {
                                turn: self.turn,
                                criteria,
                            });
                        self.phase = LoopPhase::Reason;
                        return self.emit_call_llm();
                    }
                    if signals_waiting_for_followup {
                        self.ctx.push_history(message, tokens);
                        self.phase = LoopPhase::Reason;
                        return self.emit_call_llm();
                    }
                    return self.terminate(TerminationReason::Completed, Some(message));
                }

                let calls = message.tool_calls.clone();
                self.ctx.push_history(message, tokens);

                // ━━ 记录活动时间（Layer 3时间衰减使用）
                if let Some(now_ms) = self.last_now_ms {
                    self.ctx.record_activity(now_ms);
                }

                // §7.6 · the P1 syscalls in this batch were already adjudicated by the canonical
                // driver. The assistant message keeps every call the model made — it must see the
                // turn it emitted — but a syscall is never dispatched: it is closed here with the
                // kernel's own answer, before the fuse/gate, which meter host work only.
                let calls = if adjudicated.answered_calls.is_empty() {
                    calls
                } else {
                    let answered: HashSet<CompactString> = adjudicated
                        .answered_calls
                        .iter()
                        .map(|answer| answer.call_id.clone())
                        .collect();
                    for answer in &adjudicated.answered_calls {
                        self.push_synthetic_tool_result(
                            &answer.call_id,
                            &answer.output,
                            answer.is_error,
                        );
                    }
                    calls
                        .into_iter()
                        .filter(|call| !answered.contains(&call.id))
                        .collect()
                };
                if calls.is_empty() {
                    // §5k · a syscall-only batch leaves nothing for a host to execute. Either the
                    // adjudication already published kernel-owned work — and the turn resumes when
                    // that resolves — or the turn continues the only way a turn can continue: the
                    // kernel calls the provider again, itself, in this same committed step.
                    self.phase = LoopPhase::Reason;
                    return match adjudicated.idle_continuation {
                        IdleContinuation::CallProvider => self.emit_call_llm(),
                        IdleContinuation::Await => LoopAction::AwaitingResume,
                    };
                }

                // ③ pacing trap: a `pace` call is a kernel-adjudicated round-end proposal,
                // never an SDK tool. Handled before the fuse/gate — it is a control verb,
                // not task work.
                if self
                    .run_spec
                    .as_ref()
                    .and_then(|r| r.loop_round.as_ref())
                    .is_some()
                {
                    if let Some(pace_call) = calls.iter().find(|c| c.name.as_str() == "pace") {
                        let call = pace_call.clone();
                        // The assistant message carrying `pace` is already committed to history
                        // and the kernel adjudicates pace itself — sibling calls batched with it
                        // are never executed, so close their transcript pairs here or they remain
                        // orphaned tool_use blocks (wire-invalid on several vendors).
                        for sibling in calls.iter().filter(|c| c.id != call.id) {
                            self.push_synthetic_tool_result(
                                &sibling.id,
                                "not executed: superseded by pace — the round is ending",
                                false,
                            );
                        }
                        return self.handle_pace_call(call);
                    }
                }

                // 2b: record this turn's tool activity into the task-state recency log (meta-tools
                // filtered inside). The State-turn footer renders it as "just did: …" + a forward
                // nudge / STOP, so progress is kernel-derived and never depends on the model
                // remembering to call `update_plan`. Tool *names* live only on the request (results
                // carry call_id only), so this is the turn to capture them.
                //
                // Capture name AND a compact arg digest: the no-progress STOP keys on whether the
                // SAME call repeats, and a legit loop (same tool, DIFFERENT args — e.g. processing 20
                // items) is real progress, not a stall. Keying on the name alone false-positives those
                // loops; including args distinguishes "step(n=1), step(n=2)…" from a true repeat.
                let action_sigs: Vec<(String, String)> = calls
                    .iter()
                    .map(|c| (c.name.to_string(), compact_tool_args(&c.arguments)))
                    .collect();
                self.ctx.note_tool_actions(&action_sigs);

                // O6 RepeatFuse: the hard rungs above the 2c soft STOP. Runs BEFORE the governance
                // gate and independent of whether a policy is loaded — a batteries-included kernel
                // protection, not a policy feature. Deny commits a visible synthetic error result;
                // the terminate rung ends the run `NoProgress` after one final no-tools report turn.
                if let Some(action) = self.check_repeat_fuse(&calls) {
                    return action;
                }

                // P1 fail-closed dispatch: drop calls to tools this run never advertised, BEFORE the
                // governance gate (a tool that was never exposed is not a policy question) and AFTER
                // the fuse (a model retrying the same phantom tool must still burn the fuse, exactly
                // like a repeated host denial). Denials commit as visible error results; surviving
                // siblings execute this turn.
                let calls = match self.gate_exposed_tool_calls(calls) {
                    ExposureGateOutcome::Blocked(action) => return action,
                    ExposureGateOutcome::Proceed(calls) => calls,
                };

                match self.gate_tool_calls(&calls) {
                    GateToolOutcome::Blocked(action) => return action,
                    GateToolOutcome::ApprovalRequired(requests) => {
                        return LoopAction::RequestApproval { requests };
                    }
                    GateToolOutcome::Proceed => {}
                }
                self.phase = LoopPhase::Act {
                    tool_calls: calls.clone(),
                };
                self.set_lifecycle(TaskLifecycle::Running, None);
                LoopAction::ExecuteTools { calls }
            }

            LoopEvent::ToolResults { mut results } => {
                if !self.pending_denied_results.is_empty() {
                    results.append(&mut self.pending_denied_results);
                }
                if let Some(reason) = results
                    .iter()
                    .find_map(|result| self.rollback_reason_for_tool_result(result))
                {
                    let note = CoreMessage::user(super::rollback::build_rollback_note(
                        &reason,
                        self.ctx.config.verbose_control_notes,
                    ));
                    self.rollback(reason);
                    self.ctx
                        .push_signal(note.content.as_text().unwrap_or_default().to_string());
                    self.phase = LoopPhase::Reason;
                    return self.emit_call_llm();
                }
                // Tool errors are committed to history so the LLM can see them and self-correct
                // without losing turn state. UserInterrupt was handled by the rollback arm above.

                // Entropy: this completed turn's failure tally. All model-visible tool failures,
                // including fatal and timeout results, accrue at this committed boundary.
                let errored_results = results.iter().filter(|r| r.is_error).count() as u32;
                let total_results = results.len() as u32;
                for r in &results {
                    let measured_tokens = self.pending_tool_measurements.remove(&r.call_id);
                    self.total_tokens += u64::from(measured_tokens.unwrap_or(0));
                    // Preserve Content::Parts (structured / multimodal tool output).
                    // Parts are serialised to JSON so the text can be restored faithfully.
                    let output = match &r.output {
                        Content::Text(s) => s.clone(),
                        Content::Parts(parts) => serde_json::to_string(parts).unwrap_or_default(),
                    };
                    let parts = vec![ContentPart::ToolResult {
                        call_id: r.call_id.clone(),
                        output,
                        is_error: r.is_error,
                        durable_content: r.durable_content.clone(),
                    }];
                    let tool_msg = CoreMessage::tool(parts);
                    let tokens =
                        measured_tokens.unwrap_or_else(|| self.ctx.engine.count_message(&tool_msg));
                    self.ctx.push_history(tool_msg, tokens);
                }
                self.turn += 1;
                // The budget verdict (turn/token/wall) fires inside `emit_call_llm` at the end of
                // this arm — the single provider-call funnel — so the eviction checkpoint and
                // entropy sample below still run on the exhaustion turn before the final report.

                // ━━ Eviction checkpoint (M3): one decision model (`plan_eviction`), one
                // execution funnel (`execute_eviction_op`). Layer 3 (idle/time-decay) must run
                // before the rho recommendation is read, since it mutates token usage — so the
                // plan is built in that interleaved order and the ops are executed in plan order.
                let idle_decay = self
                    .last_now_ms
                    .is_some_and(|now_ms| self.ctx.should_time_decay_compact(now_ms));
                if idle_decay {
                    self.execute_eviction_op(&crate::mm::EvictionOp::TimeDecayMicro);
                }

                // Layer 4 read-time projection: recompute handle residency on the post-time-decay rho.
                self.ctx.recompute_handle_residency();
                // K2: knowledge budget check — marks over-budget unpinned entries for the next
                // boundary sweep (marks are idempotent; drops only apply there) and stashes a
                // warn-once-per-generation notice, drained into an observation here.
                if let Some((used, budget)) = self.ctx.enforce_knowledge_budget() {
                    self.observations
                        .push(KernelObservation::KnowledgeBudgetExceeded {
                            turn: self.turn,
                            used,
                            budget,
                        });
                }
                // Layers 2/4/5: execute the pressure-driven ops from the plan (skip TimeDecayMicro
                // if already executed). The plan carries specific ops stamped with real config-derived
                // params (W1-1 収口 — no magic-number placeholders), not the umbrella `Pressure` wrapper.
                let (target_tokens, preserve_turns) = self.ctx.plan_compaction_params();
                let plan = crate::mm::plan_eviction(
                    self.ctx.should_compress(),
                    idle_decay,
                    target_tokens,
                    preserve_turns,
                );
                // `idle_decay` ⇒ the plan carries a `TimeDecayMicro` (so the skip-on-already-executed
                // below is meaningful). The converse does NOT hold: a pressure-driven `MicroCompact`
                // also emits `TimeDecayMicro` independent of `idle_decay` (W1 unified planner), so we
                // assert the implication, not equality.
                debug_assert!(!idle_decay || plan.has_time_decay());
                for op in &plan.ops {
                    // Skip TimeDecayMicro if we already executed it (prevents double-execution).
                    if matches!(op, crate::mm::EvictionOp::TimeDecayMicro) && idle_decay {
                        continue;
                    }
                    self.execute_eviction_op(op);
                }

                // Renewal: when compression alone cannot recover enough headroom,
                // start a new sprint — carry forward system + memory + last N history turns.
                if self.ctx.should_renew() {
                    self.ctx.renew();
                    // A new sprint is a session boundary for signal identity: clear the dedup set so
                    // it cannot grow unbounded across a long run, and so a signal seen in a prior
                    // sprint may legitimately re-fire in the new one.
                    self.signal_router.clear_dedup();
                    self.observations.push(KernelObservation::Renewed {
                        sprint: self.ctx.sprint,
                    });
                    // K1: renewal is a boundary — surface the knowledge sweep it just ran.
                    self.emit_knowledge_sweep_observations();
                }

                // Session-entropy sample (the heartbeat watch source): fold this completed
                // turn's outcomes into the sliding window and surface the measurement.
                // Unconditional, like `CheckpointTaken`; only the watch alert below is opt-in.
                let repeat_streak = if self.repeat_fuse.enabled {
                    self.repeat_count
                } else {
                    0
                };
                let sample = self.entropy.sample(
                    self.turn,
                    self.ctx.rho(),
                    repeat_streak,
                    self.repeat_fuse.deny_after,
                    errored_results,
                    total_results,
                );
                self.observations.push(KernelObservation::EntropySample {
                    turn: sample.turn,
                    score: sample.score,
                    rho: sample.rho,
                    repeat_pressure: sample.repeat_pressure,
                    failure_rate: sample.failure_rate,
                    rollbacks_in_window: sample.rollbacks_in_window,
                    window_turns: sample.window_turns,
                });
                // Opt-in entropy watch: threshold + hysteresis + cooldown. The alert is an
                // observation (host-facing); with `notify_model` it is ALSO routed through
                // the kernel's own signal dispatch as a Heartbeat/Alert directive — High
                // urgency while running ⇒ a durable [SIGNAL] note on the turn we are about
                // to emit anyway, never an extra provider call.
                if self.entropy.should_alert(&self.entropy_watch, &sample) {
                    self.observations.push(KernelObservation::EntropyAlert {
                        turn: sample.turn,
                        score: sample.score,
                        threshold: self.entropy_watch.threshold,
                    });
                    if self.entropy_watch.notify_model {
                        use crate::types::signal::{
                            RuntimeSignal, SignalSource, SignalType, Urgency,
                        };
                        let signal = RuntimeSignal::new(
                            SignalSource::Heartbeat,
                            SignalType::Alert,
                            Urgency::High,
                            format!(
                                "[entropy] session disorder {:.2} ≥ {:.2} (repeat {:.2} / failures {:.2} / pressure {:.2}). \
                                 Stop and reassess: state what is not working and try a different approach.",
                                sample.score,
                                self.entropy_watch.threshold,
                                sample.repeat_pressure,
                                sample.failure_rate,
                                sample.rho,
                            ),
                        )
                        .with_dedupe(format!("entropy_alert:{}", sample.turn));
                        let _ = self.dispatch_signal(signal);
                    }
                }

                // Turn boundary: drain any kernel-queued signals into context so they
                // are seen on the next reasoning turn (ready queue → running).
                self.drain_queued_signals();

                self.phase = LoopPhase::Reason;
                self.emit_call_llm()
            }

            LoopEvent::MilestoneResult { result } => self.handle_milestone_result(result),

            LoopEvent::SubAgentCompleted { result } => self.handle_sub_agent_completed(result),

            LoopEvent::Complete => self.terminate(TerminationReason::Completed, None),

            LoopEvent::Timeout => {
                // A timed-out tool batch commits per-call timeout error results — the trained
                // convention ("command timed out" as a visible error) — so the model sees which
                // call stalled and can verify or change approach. Only a Reason-phase timeout
                // (nothing model-visible pending) keeps the rollback + note path.
                if let LoopPhase::Act { tool_calls } = &self.phase {
                    if !tool_calls.is_empty() {
                        let results: Vec<ToolResult> = tool_calls
                            .iter()
                            .map(|call| ToolResult {
                                call_id: call.id.clone(),
                                output: Content::Text(format!(
                                    "Tool call `{}` timed out before completing. The operation \
                                     may or may not have taken effect — verify before assuming, \
                                     then retry with a smaller step or a faster approach.",
                                    call.name
                                )),
                                durable_content: None,
                                is_error: true,
                                is_fatal: false,
                                error_kind: Some(ToolErrorKind::Timeout),
                            })
                            .collect();
                        return self.feed(LoopEvent::ToolResults { results });
                    }
                }
                let reason = RollbackReason::Timeout;
                let note = CoreMessage::user(super::rollback::build_rollback_note(
                    &reason,
                    self.ctx.config.verbose_control_notes,
                ));
                self.rollback(reason);
                self.ctx
                    .push_signal(note.content.as_text().unwrap_or_default().to_string());
                self.phase = LoopPhase::Reason;
                self.emit_call_llm()
            }
        }
    }

    /// Drain observations emitted during the last `start`/`feed` call.
    pub fn take_observations(&mut self) -> Vec<KernelObservation> {
        std::mem::take(&mut self.observations)
    }

    /// ③ the pacing trap. The model PROPOSES `pace(next, delay_ms?, reason)`; the kernel
    /// ADJUDICATES: malformed → governance-style rollback note; sleep delay clamped into
    /// the spec's [min,max]; continue/sleep at the round cap coerced to stop("max_rounds");
    /// stop with standing acceptance criteria routes through the O4 criteria gate ONCE
    /// (one bounded self-check turn) before being honored. An allowed pace ends the round:
    /// the decision is stashed for LoopResult, a synthetic tool result closes the
    /// transcript pair, and the strip-tools final-report turn finishes the round.
    fn handle_pace_call(&mut self, call: ToolCall) -> LoopAction {
        use crate::types::result::{PaceAction, PaceDecision};

        let spec = self
            .run_spec
            .as_ref()
            .and_then(|r| r.loop_round.as_ref())
            .cloned()
            .unwrap_or_default();

        let next = call
            .arguments
            .get("next")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let reason = call
            .arguments
            .get("reason")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();
        let proposed_delay = call.arguments.get("delay_ms").and_then(|v| v.as_u64());

        let mut action = match next {
            "continue" => PaceAction::Continue,
            "sleep" => PaceAction::Sleep,
            "stop" => PaceAction::Stop,
            other => {
                // Malformed proposal: governance-style directive note + fresh reason turn.
                let rejection_reason =
                    format!("invalid pace next={other:?} (expected continue|sleep|stop)");
                let note = super::rollback::build_control_rejection_note(
                    "pace",
                    &rejection_reason,
                    self.ctx.config.verbose_control_notes,
                );
                self.push_synthetic_tool_result(
                    &call.id,
                    "pace rejected: next must be continue|sleep|stop",
                    false,
                );
                self.ctx.push_signal(note);
                self.phase = LoopPhase::Reason;
                return self.emit_call_llm();
            }
        };
        let mut coerced_from: Option<String> = None;

        // Round-cap coercion: both the run spec and reservation grant bound local rounds.
        if action != PaceAction::Stop {
            let granted_rounds = self.budget_grant.as_ref().and_then(|grant| grant.rounds);
            let max_rounds = if granted_rounds == Some(0) {
                Some(0)
            } else {
                spec.max_rounds
            };
            if let Some(max) = max_rounds {
                if self.local_rounds_completed.saturating_add(1) >= max {
                    coerced_from = Some(format!("{} (max_rounds={max})", action.label()));
                    action = PaceAction::Stop;
                }
            }
        }

        // O4 routing: a stop with standing criteria takes the existing criteria-gate
        // self-check turn first; the model re-decides with the checklist in view.
        if action == PaceAction::Stop
            && self.criteria_gate_enabled
            && !self.criteria_gate_fired
            && !self.ctx.partitions.task_state.criteria.is_empty()
        {
            self.criteria_gate_fired = true;
            let criteria = self.ctx.partitions.task_state.criteria.clone();
            self.push_synthetic_tool_result(
                &call.id,
                "pace(stop) noted — verify the acceptance criteria first, then pace again.",
                false,
            );
            self.ctx.push_signal(format!(
                "[CRITERIA CHECK] You proposed stopping the loop. Verify each acceptance \
                 criterion first: {}. If any is NOT met, continue working (or pace(continue)). \
                 If all are met, call pace(stop) again.",
                criteria.join(" | ")
            ));
            self.observations
                .push(KernelObservation::CriteriaGateFired {
                    turn: self.turn,
                    criteria,
                });
            self.phase = LoopPhase::Reason;
            return self.emit_call_llm();
        }

        // Sleep clamp into [min, max].
        let delay_ms = if action == PaceAction::Sleep {
            let raw = proposed_delay.unwrap_or(spec.min_sleep_ms.unwrap_or(60_000));
            let mut clamped = raw;
            if let Some(min) = spec.min_sleep_ms {
                clamped = clamped.max(min);
            }
            if let Some(max) = spec.max_sleep_ms {
                clamped = clamped.min(max);
            }
            if clamped != raw && coerced_from.is_none() {
                coerced_from = Some(format!("sleep {raw}ms (clamped)"));
            }
            Some(clamped)
        } else {
            None
        };

        self.local_rounds_completed = self.local_rounds_completed.saturating_add(1);
        let decision = PaceDecision {
            action,
            delay_ms,
            reason,
            coerced_from,
        };
        self.observations.push(KernelObservation::RoundPaced {
            turn: self.turn,
            round: self.local_rounds_completed,
            decision: decision.clone(),
        });
        self.push_synthetic_tool_result(
            &call.id,
            &format!(
                "pace acknowledged: {}{} — wrap up with a brief round report.",
                decision.action.label(),
                decision
                    .delay_ms
                    .map(|d| format!(" {d}ms"))
                    .unwrap_or_default()
            ),
            false,
        );
        self.pending_pace = Some(decision);
        self.pending_termination = Some(TerminationReason::Completed);
        self.phase = LoopPhase::Reason;
        self.emit_call_llm()
    }

    /// Close a kernel-handled tool call's transcript pair with a synthetic result so
    /// providers always see call → result.
    fn push_synthetic_tool_result(&mut self, call_id: &str, output: &str, is_error: bool) {
        let msg = CoreMessage::tool(vec![crate::types::message::ContentPart::ToolResult {
            call_id: call_id.into(),
            output: output.to_string(),
            is_error,
            durable_content: None,
        }]);
        let tokens = self.message_tokens(&msg);
        self.ctx.push_history(msg, tokens);
    }

    fn terminate(
        &mut self,
        termination: TerminationReason,
        final_message: Option<CoreMessage>,
    ) -> LoopAction {
        // Commit the final response into history so subsequent session restores
        // include the complete transcript: user → [tool turns] → final assistant.
        if let Some(ref msg) = final_message {
            let tokens = self.message_tokens(msg);
            self.ctx.push_history(msg.clone(), tokens);
        }
        // ③ attach the round's pacing decision. Stashed by the trap when the model
        // called `pace`; otherwise the spec's default_action ("stop" for goal loops,
        // "sleep" for cron loops) — but ONLY on a clean Completed. NoProgress /
        // ContextOverflow / Error rounds stop and surface (nothing nags the model).
        let pace_decision = self.pending_pace.take().or_else(|| {
            let spec = self.run_spec.as_ref()?.loop_round.as_ref()?;
            if termination != TerminationReason::Completed {
                return Some(crate::types::result::PaceDecision {
                    action: crate::types::result::PaceAction::Stop,
                    delay_ms: None,
                    reason: format!("round terminated: {}", termination.label()),
                    coerced_from: None,
                });
            }
            match spec.default_action.as_deref() {
                Some("sleep") => Some(crate::types::result::PaceDecision {
                    action: crate::types::result::PaceAction::Sleep,
                    delay_ms: spec.min_sleep_ms.or(Some(60_000)),
                    reason: "default_action: sleep (cron loop)".to_string(),
                    coerced_from: None,
                }),
                _ => Some(crate::types::result::PaceDecision {
                    action: crate::types::result::PaceAction::Stop,
                    delay_ms: None,
                    reason: "default_action: stop (no pace call this round)".to_string(),
                    coerced_from: None,
                }),
            }
        });
        let result = LoopResult {
            termination,
            final_message,
            turns_used: self.turn,
            total_tokens_used: self.total_tokens,
            loop_continue: None,
            classify_branch: None,
            tournament_winner: None,
            pace_decision,
        };
        self.set_lifecycle(TaskLifecycle::Done(termination), None);
        // spc_008-04: `SupervisionPolicy.child_failure` was structural-only (spc_002-07) until now
        // — the root's own terminal transition is the one place in production a task with children
        // (workflow nodes; see spc_002-03's finding that only root ever has children today) can
        // reliably be observed terminating. `Restart`/`Retry`/`Ignore` are deliberately treated as
        // `Isolate` for now (no policy distinguishes them yet) — a documented placeholder per
        // spc_008-04's own scope, not silently dropped.
        match self
            .tasks
            .get("root")
            .map(|root| root.supervision.child_failure)
            .unwrap_or_default()
        {
            crate::scheduler::tcb::ChildFailurePolicy::Propagate => {
                self.tasks.cancel_children("root");
            }
            crate::scheduler::tcb::ChildFailurePolicy::Isolate
            | crate::scheduler::tcb::ChildFailurePolicy::Restart
            | crate::scheduler::tcb::ChildFailurePolicy::Retry
            | crate::scheduler::tcb::ChildFailurePolicy::Ignore => {}
        }
        LoopAction::Done { result }
    }

    /// Build the `CallLLM` action with a structured `InternalRenderedContext`.
    /// Meta-tools (skill / memory / knowledge) are appended to the tool list
    /// when configured. When `pending_termination` is set, tools are stripped
    /// to force a plain-text response before the loop terminates.
    fn emit_call_llm(&mut self) -> LoopAction {
        // Calling the provider is definitionally "running" — the single funnel for entering the
        // Running lifecycle (covers start, resume, signal-driven turns, budget final-call).
        self.set_lifecycle(TaskLifecycle::Running, None);

        // M1 収口 (completed): the budget verdict lives at the same single funnel. Every edge that
        // requests a provider call — tool-turn completion, milestone retry, signal-forced turns,
        // criteria gate, recovery ladders — passes the three axes here, so a loop that completes
        // no tool turns (and therefore never increments `turn`) is still bounded by the token and
        // wall axes. The final-report turn itself (`pending_termination` set) is exempt: it is the
        // one bounded call the verdict buys, so the check fires exactly once per exhaustion.
        if self.pending_termination.is_none() {
            if let Some(term) = super::tcb::budget_verdict(&self.root_tcb(), self.last_now_ms) {
                let budget = match term {
                    TerminationReason::MaxTurns => "max_turns",
                    TerminationReason::Timeout => "wall_time",
                    _ => "token_budget",
                };
                self.observations.push(KernelObservation::BudgetExceeded {
                    turn: self.turn,
                    budget: budget.to_string(),
                    operation_id: String::new(),
                    reservation_id: self
                        .budget_grant
                        .as_ref()
                        .map(|grant| grant.reservation_id.clone()),
                });
                self.pending_termination = Some(term);
            }
        }
        self.checkpoint.history_len = self.ctx.partitions.history.messages.len();
        self.checkpoint.signals_len = self.ctx.partitions.signals.len();
        self.checkpoint.task_state = Some(self.ctx.partitions.task_state.clone());
        self.delivered_signals_len = self.ctx.partitions.signals.len();
        self.observations.push(KernelObservation::CheckpointTaken {
            turn: self.turn,
            history_len: self.checkpoint.history_len as u32,
        });

        let context = self.ctx.render();
        if let Some(overflow) = context.budget_overflow.clone() {
            self.observations
                .push(KernelObservation::ContextBudgetExceeded {
                    turn: self.turn,
                    overflow_kind: overflow.kind,
                    required_tokens: overflow.required_tokens,
                    max_tokens: overflow.max_tokens,
                });
            // P0-2 §C: only a `FixedContext` overflow (system + state_turn alone exceed the hard
            // window) is unrecoverable — compaction cannot touch that region, so terminate honestly.
            // A `ProtectedTail` overflow (a protected recent unit tips the budget after compaction
            // already ran) is NOT terminal: the observation records the over-budget tail, and the
            // context is still submitted. The provider decides; if it rejects with a 413, the
            // reactive recovery ladder (`recover_from_provider_error`) is the real backstop. Silently
            // terminating here would kill runs the provider could have accepted or recovered from.
            if matches!(
                overflow.kind,
                crate::context::renderer::ContextBudgetOverflowKind::FixedContext
            ) {
                self.delivered_signals_len = 0;
                return self.terminate(TerminationReason::ContextOverflow, None);
            }
        }
        let tools = if self.pending_termination.is_some() {
            Vec::new()
        } else {
            self.provider_tools()
        };

        self.call_llm_action(context, tools)
    }

    /// Rebuild the provider projection after a same-transition context mutation.
    ///
    /// External payload residency is committed after `ToolResults` has already produced its
    /// continuation. That mutation can change both rendered handle state and the conditional
    /// `read_result` meta-tool, but it must not cross a second scheduler boundary (budget verdict,
    /// checkpoint, or observation). Refresh only the projection and advertised-tool authority.
    pub(crate) fn refresh_call_llm_action(&mut self, action: &mut LoopAction) {
        if !matches!(action, LoopAction::CallLLM { .. }) {
            return;
        }
        let context = self.ctx.render();
        let tools = if self.pending_termination.is_some() {
            Vec::new()
        } else {
            self.provider_tools()
        };
        self.exposed_tool_names = Some(tools.iter().map(|tool| tool.name.clone()).collect());
        *action = LoopAction::CallLLM { context, tools };
    }

    fn provider_tools(&self) -> Vec<ToolSchema> {
        let mut tools = self.tools.clone();
        tools.extend(self.ctx.meta_tool_schemas());

        if let Some(ref spec) = self.run_spec {
            use crate::context::manager::is_exposure_exempt_meta_tool;
            use crate::types::capability::CapabilityKind;
            tools.retain(|tool| {
                let kind = match tool.name.as_str() {
                    "skill" => CapabilityKind::Skill,
                    "memory" => CapabilityKind::Memory,
                    "knowledge" => CapabilityKind::Knowledge,
                    _ => CapabilityKind::Tool,
                };
                // Kernel-owned meta surfaces are exempt from the ID axis: `allowedToolIds` lists
                // the run's *task* tools, and silently deleting the model's route back to kernel
                // state (load a skill, re-read an evicted result) is never what that means — the
                // same rationale the pace tool encodes below. The KIND axis still applies, so a
                // sub-agent isolation filter that admits only `Tool` still excludes
                // skill/memory/knowledge outright. See `EXPOSURE_EXEMPT_META_TOOLS`.
                if is_exposure_exempt_meta_tool(&tool.name) {
                    return spec.capability_filter.allows_kind(kind);
                }
                let desc = crate::types::capability::CapabilityDescriptor::marker(
                    kind,
                    tool.name.clone(),
                    &tool.description,
                );
                spec.capability_filter.allows(&desc)
            });
        }

        // ─── Filter B: which of the ceiling-admitted tools are exposed *this* epoch ───
        //
        // The baseline is the pre-activation exposure policy under the ceiling:
        //
        // ```text
        // exposed = META ∪ ((baseline ∪ stableCore ∪ ⋃ activeSkills.allowed_tools) ∩ ceiling)
        // ```
        //
        // - `ceiling` = the run-level capability filter, already applied above as filter A, so the
        //   `∩ ceiling` term needs no code here: a baseline entry outside the ceiling was simply
        //   never in `tools` to begin with (D3 — silent intersection, no start_run error, the same
        //   fold every id-list surface uses).
        // - `activeSkills.allowed_tools` covers only declared lists. An active skill that declares
        //   nothing contributes ∅ and the surface stays at the baseline.
        // - Missing and empty baselines both mean meta-tools (+stable-core) only.
        //
        // Kernel-owned meta surfaces remain exempt so the model can still load a skill —
        // and still re-read an evicted result, which the truncation marker explicitly instructs it
        // to do. Byte-stable within an epoch either way: the set changes only at an
        // activation/deactivation boundary.
        let baseline = self
            .run_spec
            .as_ref()
            .and_then(|s| s.exposure_baseline.as_ref())
            .map(Vec::as_slice)
            .unwrap_or_default();
        let baseline: std::collections::HashSet<&CompactString> = baseline.iter().collect();
        let declared = self.ctx.active_skill_tool_filter().unwrap_or_default();
        let stable = &self.ctx.stable_core_tools;
        tools.retain(|tool| {
            crate::context::manager::is_exposure_exempt_meta_tool(&tool.name)
                || baseline.contains(&tool.name)
                || stable.contains(&tool.name)
                || declared.contains(&tool.name)
        });

        // ③ pace meta-tool: exposed ONLY when this run is a round of a paced loop
        // (run_spec.loop_round present) — the same conditional-exposure pattern as
        // skill/memory/read_result. Pushed after every filter: pacing is kernel-owned
        // and must never be narrowed away by skills or capability filters.
        if self
            .run_spec
            .as_ref()
            .and_then(|r| r.loop_round.as_ref())
            .is_some()
        {
            tools.push(pace_tool_schema());
        }

        tools
    }

    /// The single exit for every provider call. Records the advertised toolset (P1 fail-closed
    /// dispatch arms against exactly what the model was shown this turn) and returns the action.
    fn call_llm_action(
        &mut self,
        context: crate::context::renderer::InternalRenderedContext,
        tools: Vec<ToolSchema>,
    ) -> LoopAction {
        self.exposed_tool_names = Some(tools.iter().map(|tool| tool.name.clone()).collect());
        LoopAction::CallLLM { context, tools }
    }

    /// Canonical checkpoint projection of the most recently advertised tool surface.
    pub(crate) fn advertised_tool_ids(&self) -> Option<Vec<String>> {
        self.exposed_tool_names.as_ref().map(|names| {
            let mut names: Vec<String> = names.iter().map(ToString::to_string).collect();
            names.sort();
            names
        })
    }

    /// Restore the exact provider-advertised surface captured by a canonical checkpoint.
    pub(crate) fn restore_advertised_tool_ids(&mut self, names: Option<Vec<String>>) {
        self.exposed_tool_names = names.map(|names| names.into_iter().map(Into::into).collect());
    }

    pub fn rollback(&mut self, reason: RollbackReason) {
        self.ctx
            .partitions
            .history
            .messages
            .truncate(self.checkpoint.history_len);
        self.ctx
            .partitions
            .signals
            .truncate(self.checkpoint.signals_len);
        if let Some(ref state) = self.checkpoint.task_state {
            self.ctx.partitions.task_state = state.clone();
        }
        // Rolled-back turns never reach the boundary sample point; accrue here so the
        // disorder they evidence lands in the next completed turn's entropy window.
        self.entropy.note_rollback();
        self.observations.push(KernelObservation::Rollbacked {
            turn: self.turn,
            checkpoint_history_len: self.checkpoint.history_len as u32,
            reason: Some(reason),
        });
    }

    /// Which tool results still roll the turn back. Models are trained on "tool failed → an
    /// error tool result stays in history and the model adapts" — the convention every major
    /// harness produces — so fatal / timeout / provider-failure / denied results all COMMIT as
    /// visible errors (same evidence class as the governance-denial A/B: erasing the attempt
    /// makes the model re-try what it cannot see). The one survivor is `UserInterrupt`: the
    /// user's escape is a host-owned control event, not model feedback.
    fn rollback_reason_for_tool_result(&self, result: &ToolResult) -> Option<RollbackReason> {
        match result.error_kind {
            Some(ToolErrorKind::UserInterrupt) => Some(RollbackReason::UserInterrupt),
            _ => None,
        }
    }
}

#[cfg(test)]
#[path = "tests.rs"]
mod tests;

/// ③ the `pace` meta-tool schema — exposed only on loop-round runs.
fn pace_tool_schema() -> crate::types::message::ToolSchema {
    crate::types::message::ToolSchema {
        name: compact_str::CompactString::new("pace"),
        description: "End this round and decide what happens next: continue immediately, \
sleep then run another round, or stop the loop. Call this when the round's work is done."
            .to_string(),
        parameters: serde_json::json!({
            "type": "object",
            "properties": {
                "next": { "type": "string", "enum": ["continue", "sleep", "stop"] },
                "delay_ms": { "type": "integer", "minimum": 0 },
                "reason": { "type": "string" }
            },
            "required": ["next", "reason"]
        }),
    }
}