use std::collections::BTreeSet;
use std::fmt;

use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

use crate::capability::CapabilityRegistry;
use crate::completion::{CompletionBlocker, CompletionPolicy};
use crate::contract::{validate_description, validate_identifier, validate_metadata};

pub const RUNTIME_SNAPSHOT_SCHEMA_VERSION: &str = "agent-runtime-core-snapshot.v1";
const MAX_DELEGATION_CHILDREN: usize = 64;

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum RunStatus {
    Pending,
    Running,
    WaitingForAction,
    WaitingForChildren,
    Paused,
    Completed,
    Failed,
    Cancelled,
    NeedsReconciliation,
}

impl RunStatus {
    pub fn is_terminal(self) -> bool {
        matches!(self, Self::Completed | Self::Failed | Self::Cancelled)
    }

    fn occupies_agent_lane(self) -> bool {
        !matches!(self, Self::Pending) && !self.is_terminal()
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum ActionStatus {
    Queued,
    Executing,
    Observed,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum ObservationStatus {
    Completed,
    Failed,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum JoinMode {
    All,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RunSpec {
    run_id: String,
    agent_id: String,
    task: String,
    metadata: Value,
}

impl RunSpec {
    pub fn try_new(
        run_id: impl Into<String>,
        agent_id: impl Into<String>,
        task: impl Into<String>,
    ) -> Result<Self, RuntimeError> {
        let run_id = run_id.into();
        let agent_id = agent_id.into();
        let task = task.into();
        validate_identifier(&run_id, "run id")?;
        validate_identifier(&agent_id, "agent id")?;
        validate_description(&task, "run task")?;
        Ok(Self {
            run_id,
            agent_id,
            task,
            metadata: Value::Object(Map::new()),
        })
    }

    pub fn with_metadata(mut self, metadata: Value) -> Result<Self, RuntimeError> {
        validate_metadata(&metadata, "run metadata")?;
        self.metadata = metadata;
        Ok(self)
    }

    pub fn run_id(&self) -> &str {
        &self.run_id
    }

    pub fn agent_id(&self) -> &str {
        &self.agent_id
    }

    pub fn task(&self) -> &str {
        &self.task
    }

    pub fn metadata(&self) -> &Value {
        &self.metadata
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RuntimeAction {
    id: String,
    capability_id: String,
    input: Value,
}

impl RuntimeAction {
    pub fn try_new(
        id: impl Into<String>,
        capability_id: impl Into<String>,
        input: Value,
    ) -> Result<Self, RuntimeError> {
        let id = id.into();
        let capability_id = capability_id.into();
        validate_identifier(&id, "action id")?;
        validate_identifier(&capability_id, "capability id")?;
        if !input.is_object() {
            return Err(RuntimeError::invalid("action input 必须是 JSON object"));
        }
        Ok(Self {
            id,
            capability_id,
            input,
        })
    }

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

    pub fn capability_id(&self) -> &str {
        &self.capability_id
    }

    pub fn input(&self) -> &Value {
        &self.input
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ActionRecord {
    action: RuntimeAction,
    status: ActionStatus,
    updated_at_ms: u64,
}

impl ActionRecord {
    pub fn action(&self) -> &RuntimeAction {
        &self.action
    }

    pub fn status(&self) -> ActionStatus {
        self.status
    }

    pub fn updated_at_ms(&self) -> u64 {
        self.updated_at_ms
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ToolOutput {
    status: ObservationStatus,
    summary: String,
    output: Value,
}

impl ToolOutput {
    pub fn completed(summary: impl Into<String>, output: Value) -> Result<Self, RuntimeError> {
        Self::try_new(ObservationStatus::Completed, summary, output)
    }

    pub fn failed(summary: impl Into<String>, output: Value) -> Result<Self, RuntimeError> {
        Self::try_new(ObservationStatus::Failed, summary, output)
    }

    fn try_new(
        status: ObservationStatus,
        summary: impl Into<String>,
        output: Value,
    ) -> Result<Self, RuntimeError> {
        let summary = summary.into();
        validate_description(&summary, "tool output summary")?;
        Ok(Self {
            status,
            summary,
            output,
        })
    }

    pub fn status(&self) -> ObservationStatus {
        self.status
    }

    pub fn summary(&self) -> &str {
        &self.summary
    }

    pub fn output(&self) -> &Value {
        &self.output
    }
}

#[derive(Clone, Debug, PartialEq)]
pub enum ToolExecution {
    Observed(ToolOutput),
    Unknown,
}

pub trait ToolHost {
    fn execute(&mut self, run: &RunRecord, action: &RuntimeAction) -> ToolExecution;
}

pub trait RuntimeClock {
    fn now_millis(&self) -> u64;
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RuntimeObservation {
    action_id: String,
    capability_id: String,
    status: ObservationStatus,
    summary: String,
    output: Value,
    observed_at_ms: u64,
}

impl RuntimeObservation {
    pub fn action_id(&self) -> &str {
        &self.action_id
    }

    pub fn capability_id(&self) -> &str {
        &self.capability_id
    }

    pub fn status(&self) -> ObservationStatus {
        self.status
    }

    pub fn summary(&self) -> &str {
        &self.summary
    }

    pub fn output(&self) -> &Value {
        &self.output
    }

    pub fn observed_at_ms(&self) -> u64 {
        self.observed_at_ms
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RunRecord {
    run_id: String,
    agent_id: String,
    task: String,
    metadata: Value,
    parent_run_id: Option<String>,
    delegation_group_id: Option<String>,
    status: RunStatus,
    actions: Vec<ActionRecord>,
    observations: Vec<RuntimeObservation>,
    terminal_summary: Option<String>,
    terminal_error: Option<String>,
    created_at_ms: u64,
    updated_at_ms: u64,
}

impl RunRecord {
    pub fn run_id(&self) -> &str {
        &self.run_id
    }

    pub fn agent_id(&self) -> &str {
        &self.agent_id
    }

    pub fn task(&self) -> &str {
        &self.task
    }

    pub fn metadata(&self) -> &Value {
        &self.metadata
    }

    pub fn parent_run_id(&self) -> Option<&str> {
        self.parent_run_id.as_deref()
    }

    pub fn delegation_group_id(&self) -> Option<&str> {
        self.delegation_group_id.as_deref()
    }

    pub fn status(&self) -> RunStatus {
        self.status
    }

    pub fn actions(&self) -> &[ActionRecord] {
        &self.actions
    }

    pub fn observations(&self) -> &[RuntimeObservation] {
        &self.observations
    }

    pub fn terminal_summary(&self) -> Option<&str> {
        self.terminal_summary.as_deref()
    }

    pub fn terminal_error(&self) -> Option<&str> {
        self.terminal_error.as_deref()
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RunTerminalResult {
    run_id: String,
    status: RunStatus,
    summary: Option<String>,
    error: Option<String>,
}

impl RunTerminalResult {
    pub fn run_id(&self) -> &str {
        &self.run_id
    }

    pub fn status(&self) -> RunStatus {
        self.status
    }

    pub fn summary(&self) -> Option<&str> {
        self.summary.as_deref()
    }

    pub fn error(&self) -> Option<&str> {
        self.error.as_deref()
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct DelegationGroup {
    id: String,
    parent_run_id: String,
    child_run_ids: Vec<String>,
    join_mode: JoinMode,
    resolved: bool,
    results: Vec<RunTerminalResult>,
}

impl DelegationGroup {
    pub fn id(&self) -> &str {
        &self.id
    }

    pub fn parent_run_id(&self) -> &str {
        &self.parent_run_id
    }

    pub fn child_run_ids(&self) -> &[String] {
        &self.child_run_ids
    }

    pub fn join_mode(&self) -> JoinMode {
        self.join_mode
    }

    pub fn is_resolved(&self) -> bool {
        self.resolved
    }

    pub fn results(&self) -> &[RunTerminalResult] {
        &self.results
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RuntimeSnapshot {
    schema_version: String,
    runtime_id: String,
    revision: u64,
    runs: Vec<RunRecord>,
    delegations: Vec<DelegationGroup>,
}

impl RuntimeSnapshot {
    pub fn schema_version(&self) -> &str {
        &self.schema_version
    }

    pub fn runtime_id(&self) -> &str {
        &self.runtime_id
    }

    pub fn revision(&self) -> u64 {
        self.revision
    }

    pub fn runs(&self) -> &[RunRecord] {
        &self.runs
    }

    pub fn run(&self, run_id: &str) -> Option<&RunRecord> {
        self.runs.iter().find(|run| run.run_id == run_id)
    }

    pub fn delegations(&self) -> &[DelegationGroup] {
        &self.delegations
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum RuntimeEventKind {
    RuntimeCreated,
    RunAdded,
    RunStarted,
    ActionQueued,
    ActionExecuting,
    ActionObserved,
    ActionNeedsReconciliation,
    DelegationSpawned,
    JoinResolved,
    RunPaused,
    RunResumed,
    RunCompleted,
    RunFailed,
    RunCancelled,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct RuntimeEvent {
    runtime_id: String,
    revision: u64,
    occurred_at_ms: u64,
    kind: RuntimeEventKind,
    run_id: Option<String>,
    detail: Value,
}

impl RuntimeEvent {
    pub fn runtime_id(&self) -> &str {
        &self.runtime_id
    }

    pub fn revision(&self) -> u64 {
        self.revision
    }

    pub fn occurred_at_ms(&self) -> u64 {
        self.occurred_at_ms
    }

    pub fn kind(&self) -> RuntimeEventKind {
        self.kind
    }

    pub fn run_id(&self) -> Option<&str> {
        self.run_id.as_deref()
    }

    pub fn detail(&self) -> &Value {
        &self.detail
    }
}

pub trait RuntimeStore {
    fn load(&self, runtime_id: &str) -> Result<Option<RuntimeSnapshot>, String>;

    fn commit(
        &mut self,
        runtime_id: &str,
        expected_revision: Option<u64>,
        snapshot: &RuntimeSnapshot,
        events: &[RuntimeEvent],
    ) -> Result<(), String>;
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RuntimeErrorKind {
    InvalidInput,
    NotFound,
    Conflict,
    InvalidTransition,
    Store,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RuntimeError {
    kind: RuntimeErrorKind,
    detail: String,
}

impl RuntimeError {
    fn invalid(detail: impl Into<String>) -> Self {
        Self {
            kind: RuntimeErrorKind::InvalidInput,
            detail: detail.into(),
        }
    }

    fn not_found(detail: impl Into<String>) -> Self {
        Self {
            kind: RuntimeErrorKind::NotFound,
            detail: detail.into(),
        }
    }

    fn conflict(detail: impl Into<String>) -> Self {
        Self {
            kind: RuntimeErrorKind::Conflict,
            detail: detail.into(),
        }
    }

    fn transition(detail: impl Into<String>) -> Self {
        Self {
            kind: RuntimeErrorKind::InvalidTransition,
            detail: detail.into(),
        }
    }

    fn store(detail: impl Into<String>) -> Self {
        Self {
            kind: RuntimeErrorKind::Store,
            detail: detail.into(),
        }
    }

    pub fn kind(&self) -> RuntimeErrorKind {
        self.kind
    }

    pub fn detail(&self) -> &str {
        &self.detail
    }
}

impl fmt::Display for RuntimeError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.detail)
    }
}

impl std::error::Error for RuntimeError {}

impl From<crate::ContractError> for RuntimeError {
    fn from(error: crate::ContractError) -> Self {
        Self::invalid(error.to_string())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ActionExecutionOutcome {
    Observed,
    NeedsReconciliation,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum CompletionAttempt {
    Completed,
    Blocked(Vec<CompletionBlocker>),
}

struct PendingEvent {
    kind: RuntimeEventKind,
    run_id: Option<String>,
    detail: Value,
}

pub struct RuntimeEngine<S, C> {
    store: S,
    clock: C,
}

impl<S, C> RuntimeEngine<S, C>
where
    S: RuntimeStore,
    C: RuntimeClock,
{
    pub fn new(store: S, clock: C) -> Self {
        Self { store, clock }
    }

    pub fn store(&self) -> &S {
        &self.store
    }

    pub fn store_mut(&mut self) -> &mut S {
        &mut self.store
    }

    pub fn into_parts(self) -> (S, C) {
        (self.store, self.clock)
    }

    pub fn load(&self, runtime_id: &str) -> Result<Option<RuntimeSnapshot>, RuntimeError> {
        validate_identifier(runtime_id, "runtime id")?;
        let snapshot = self.store.load(runtime_id).map_err(RuntimeError::store)?;
        if let Some(snapshot) = &snapshot {
            validate_runtime_snapshot(snapshot, runtime_id)?;
        }
        Ok(snapshot)
    }

    pub fn create_runtime(
        &mut self,
        runtime_id: impl Into<String>,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        let runtime_id = runtime_id.into();
        validate_identifier(&runtime_id, "runtime id")?;
        if self
            .store
            .load(&runtime_id)
            .map_err(RuntimeError::store)?
            .is_some()
        {
            return Err(RuntimeError::conflict(format!(
                "runtime 已存在：{runtime_id}"
            )));
        }
        let snapshot = RuntimeSnapshot {
            schema_version: RUNTIME_SNAPSHOT_SCHEMA_VERSION.to_string(),
            runtime_id: runtime_id.clone(),
            revision: 1,
            runs: Vec::new(),
            delegations: Vec::new(),
        };
        let event = RuntimeEvent {
            runtime_id: runtime_id.clone(),
            revision: 1,
            occurred_at_ms: self.clock.now_millis(),
            kind: RuntimeEventKind::RuntimeCreated,
            run_id: None,
            detail: Value::Object(Map::new()),
        };
        self.store
            .commit(&runtime_id, None, &snapshot, &[event])
            .map_err(RuntimeError::store)?;
        Ok(snapshot)
    }

    pub fn add_run(
        &mut self,
        runtime_id: &str,
        spec: RunSpec,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        let mut snapshot = self.load_required(runtime_id)?;
        if snapshot.run(&spec.run_id).is_some() {
            return Err(RuntimeError::conflict(format!(
                "run 已存在：{}",
                spec.run_id
            )));
        }
        let now = self.clock.now_millis();
        snapshot.runs.push(RunRecord {
            run_id: spec.run_id.clone(),
            agent_id: spec.agent_id,
            task: spec.task,
            metadata: spec.metadata,
            parent_run_id: None,
            delegation_group_id: None,
            status: RunStatus::Pending,
            actions: Vec::new(),
            observations: Vec::new(),
            terminal_summary: None,
            terminal_error: None,
            created_at_ms: now,
            updated_at_ms: now,
        });
        self.commit_transition(
            snapshot,
            vec![PendingEvent {
                kind: RuntimeEventKind::RunAdded,
                run_id: Some(spec.run_id),
                detail: Value::Object(Map::new()),
            }],
        )
    }

    pub fn schedule_ready(&mut self, runtime_id: &str) -> Result<Vec<String>, RuntimeError> {
        let mut snapshot = self.load_required(runtime_id)?;
        let mut occupied = snapshot
            .runs
            .iter()
            .filter(|run| run.status.occupies_agent_lane())
            .map(|run| run.agent_id.clone())
            .collect::<BTreeSet<_>>();
        let now = self.clock.now_millis();
        let mut started = Vec::new();
        for run in &mut snapshot.runs {
            if run.status == RunStatus::Pending && occupied.insert(run.agent_id.clone()) {
                run.status = RunStatus::Running;
                run.updated_at_ms = now;
                started.push(run.run_id.clone());
            }
        }
        if started.is_empty() {
            return Ok(started);
        }
        let events = started
            .iter()
            .map(|run_id| PendingEvent {
                kind: RuntimeEventKind::RunStarted,
                run_id: Some(run_id.clone()),
                detail: Value::Object(Map::new()),
            })
            .collect();
        self.commit_transition(snapshot, events)?;
        Ok(started)
    }

    pub fn enqueue_action<D>(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        capabilities: &CapabilityRegistry<D>,
        action: RuntimeAction,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        if capabilities.get(action.capability_id()).is_none() {
            return Err(RuntimeError::invalid(format!(
                "action 引用了未注册 capability：{}",
                action.capability_id()
            )));
        }
        let mut snapshot = self.load_required(runtime_id)?;
        if snapshot
            .runs
            .iter()
            .flat_map(|run| run.actions.iter())
            .any(|record| record.action.id == action.id)
        {
            return Err(RuntimeError::conflict(format!(
                "action id 已存在：{}",
                action.id
            )));
        }
        let now = self.clock.now_millis();
        let run = run_mut(&mut snapshot, run_id)?;
        require_status(run, &[RunStatus::Running], "enqueue action")?;
        run.actions.push(ActionRecord {
            action: action.clone(),
            status: ActionStatus::Queued,
            updated_at_ms: now,
        });
        run.status = RunStatus::WaitingForAction;
        run.updated_at_ms = now;
        self.commit_transition(
            snapshot,
            vec![PendingEvent {
                kind: RuntimeEventKind::ActionQueued,
                run_id: Some(run_id.to_string()),
                detail: serde_json::json!({"actionId": action.id, "capabilityId": action.capability_id}),
            }],
        )
    }

    pub fn execute_next_action<H: ToolHost>(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        host: &mut H,
    ) -> Result<ActionExecutionOutcome, RuntimeError> {
        let mut snapshot = self.load_required(runtime_id)?;
        let now = self.clock.now_millis();
        let (action, was_executing, already_reconciling) = {
            let run = run_mut(&mut snapshot, run_id)?;
            let current_status = run.status;
            require_status(
                run,
                &[RunStatus::WaitingForAction, RunStatus::NeedsReconciliation],
                "execute action",
            )?;
            let record = run
                .actions
                .last_mut()
                .ok_or_else(|| RuntimeError::transition("run 缺少待执行 action"))?;
            match record.status {
                ActionStatus::Observed => {
                    return Err(RuntimeError::transition("最新 action 已 observed"));
                }
                ActionStatus::Executing => (
                    record.action.clone(),
                    true,
                    current_status == RunStatus::NeedsReconciliation,
                ),
                ActionStatus::Queued => {
                    record.status = ActionStatus::Executing;
                    record.updated_at_ms = now;
                    (record.action.clone(), false, false)
                }
            }
        };
        if was_executing {
            if already_reconciling {
                return Ok(ActionExecutionOutcome::NeedsReconciliation);
            }
            let run = run_mut(&mut snapshot, run_id)?;
            run.status = RunStatus::NeedsReconciliation;
            run.updated_at_ms = now;
            self.commit_transition(
                snapshot,
                vec![PendingEvent {
                    kind: RuntimeEventKind::ActionNeedsReconciliation,
                    run_id: Some(run_id.to_string()),
                    detail: serde_json::json!({"actionId": action.id}),
                }],
            )?;
            return Ok(ActionExecutionOutcome::NeedsReconciliation);
        }
        let executing = self.commit_transition(
            snapshot,
            vec![PendingEvent {
                kind: RuntimeEventKind::ActionExecuting,
                run_id: Some(run_id.to_string()),
                detail: serde_json::json!({"actionId": action.id}),
            }],
        )?;
        let execution = host.execute(
            executing
                .run(run_id)
                .expect("committed executing run must exist"),
            &action,
        );
        match execution {
            ToolExecution::Observed(output) => {
                self.persist_observation(runtime_id, run_id, &action, output)?;
                Ok(ActionExecutionOutcome::Observed)
            }
            ToolExecution::Unknown => {
                let mut snapshot = self.load_required(runtime_id)?;
                let run = run_mut(&mut snapshot, run_id)?;
                run.status = RunStatus::NeedsReconciliation;
                run.updated_at_ms = self.clock.now_millis();
                self.commit_transition(
                    snapshot,
                    vec![PendingEvent {
                        kind: RuntimeEventKind::ActionNeedsReconciliation,
                        run_id: Some(run_id.to_string()),
                        detail: serde_json::json!({"actionId": action.id}),
                    }],
                )?;
                Ok(ActionExecutionOutcome::NeedsReconciliation)
            }
        }
    }

    pub fn reconcile_action(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        output: ToolOutput,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        let snapshot = self.load_required(runtime_id)?;
        let run = snapshot
            .run(run_id)
            .ok_or_else(|| RuntimeError::not_found(format!("run 不存在：{run_id}")))?;
        require_status(
            run,
            &[RunStatus::WaitingForAction, RunStatus::NeedsReconciliation],
            "reconcile action",
        )?;
        let action = run
            .actions
            .last()
            .filter(|record| record.status == ActionStatus::Executing)
            .map(|record| record.action.clone())
            .ok_or_else(|| RuntimeError::transition("run 缺少 executing action"))?;
        self.persist_observation(runtime_id, run_id, &action, output)
    }

    pub fn spawn(
        &mut self,
        runtime_id: &str,
        parent_run_id: &str,
        group_id: impl Into<String>,
        children: impl IntoIterator<Item = RunSpec>,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        let group_id = group_id.into();
        validate_identifier(&group_id, "delegation group id")?;
        let children = children.into_iter().collect::<Vec<_>>();
        if children.is_empty() || children.len() > MAX_DELEGATION_CHILDREN {
            return Err(RuntimeError::invalid(format!(
                "delegation children 必须为 1..={MAX_DELEGATION_CHILDREN}"
            )));
        }
        let mut child_ids = BTreeSet::new();
        for child in &children {
            if !child_ids.insert(child.run_id.clone()) {
                return Err(RuntimeError::conflict(format!(
                    "child run id 重复：{}",
                    child.run_id
                )));
            }
        }
        let mut snapshot = self.load_required(runtime_id)?;
        if snapshot
            .delegations
            .iter()
            .any(|group| group.id == group_id)
        {
            return Err(RuntimeError::conflict(format!(
                "delegation group 已存在：{group_id}"
            )));
        }
        if children
            .iter()
            .any(|child| snapshot.run(&child.run_id).is_some())
        {
            return Err(RuntimeError::conflict("child run id 已存在"));
        }
        let now = self.clock.now_millis();
        let parent = run_mut(&mut snapshot, parent_run_id)?;
        require_status(parent, &[RunStatus::Running], "spawn children")?;
        parent.status = RunStatus::WaitingForChildren;
        parent.updated_at_ms = now;
        let child_run_ids = children
            .iter()
            .map(|child| child.run_id.clone())
            .collect::<Vec<_>>();
        for child in children {
            snapshot.runs.push(RunRecord {
                run_id: child.run_id,
                agent_id: child.agent_id,
                task: child.task,
                metadata: child.metadata,
                parent_run_id: Some(parent_run_id.to_string()),
                delegation_group_id: Some(group_id.clone()),
                status: RunStatus::Pending,
                actions: Vec::new(),
                observations: Vec::new(),
                terminal_summary: None,
                terminal_error: None,
                created_at_ms: now,
                updated_at_ms: now,
            });
        }
        snapshot.delegations.push(DelegationGroup {
            id: group_id.clone(),
            parent_run_id: parent_run_id.to_string(),
            child_run_ids: child_run_ids.clone(),
            join_mode: JoinMode::All,
            resolved: false,
            results: Vec::new(),
        });
        self.commit_transition(
            snapshot,
            vec![PendingEvent {
                kind: RuntimeEventKind::DelegationSpawned,
                run_id: Some(parent_run_id.to_string()),
                detail: serde_json::json!({"groupId": group_id, "childRunIds": child_run_ids}),
            }],
        )
    }

    pub fn complete_if_ready<Context, Policy>(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        policy: &Policy,
        context: &Context,
        summary: impl Into<String>,
    ) -> Result<CompletionAttempt, RuntimeError>
    where
        Policy: CompletionPolicy<Context>,
    {
        validate_identifier(policy.id(), "completion policy id")?;
        let snapshot = self.load_required(runtime_id)?;
        let run = snapshot
            .run(run_id)
            .ok_or_else(|| RuntimeError::not_found(format!("run 不存在：{run_id}")))?;
        require_status(run, &[RunStatus::Running], "evaluate completion")?;
        let decision = policy.evaluate(context);
        if decision.is_ready() {
            self.finish_run(
                runtime_id,
                run_id,
                RunStatus::Completed,
                summary.into(),
                None,
            )?;
            Ok(CompletionAttempt::Completed)
        } else {
            Ok(CompletionAttempt::Blocked(decision.blockers().to_vec()))
        }
    }

    pub fn fail_run(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        summary: impl Into<String>,
        error: impl Into<String>,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        self.finish_run(
            runtime_id,
            run_id,
            RunStatus::Failed,
            summary.into(),
            Some(error.into()),
        )
    }

    pub fn cancel_run(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        summary: impl Into<String>,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        self.finish_run(
            runtime_id,
            run_id,
            RunStatus::Cancelled,
            summary.into(),
            None,
        )
    }

    pub fn pause_run(
        &mut self,
        runtime_id: &str,
        run_id: &str,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        self.change_status(
            runtime_id,
            run_id,
            &[RunStatus::Running],
            RunStatus::Paused,
            RuntimeEventKind::RunPaused,
        )
    }

    pub fn resume_run(
        &mut self,
        runtime_id: &str,
        run_id: &str,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        self.change_status(
            runtime_id,
            run_id,
            &[RunStatus::Paused],
            RunStatus::Running,
            RuntimeEventKind::RunResumed,
        )
    }

    fn load_required(&self, runtime_id: &str) -> Result<RuntimeSnapshot, RuntimeError> {
        self.load(runtime_id)?
            .ok_or_else(|| RuntimeError::not_found(format!("runtime 不存在：{runtime_id}")))
    }

    fn persist_observation(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        action: &RuntimeAction,
        output: ToolOutput,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        let mut snapshot = self.load_required(runtime_id)?;
        let now = self.clock.now_millis();
        let run = run_mut(&mut snapshot, run_id)?;
        require_status(
            run,
            &[RunStatus::WaitingForAction, RunStatus::NeedsReconciliation],
            "persist observation",
        )?;
        let record = run
            .actions
            .last_mut()
            .filter(|record| {
                record.action.id == action.id && record.status == ActionStatus::Executing
            })
            .ok_or_else(|| RuntimeError::transition("executing action 与 observation 不匹配"))?;
        record.status = ActionStatus::Observed;
        record.updated_at_ms = now;
        run.observations.push(RuntimeObservation {
            action_id: action.id.clone(),
            capability_id: action.capability_id.clone(),
            status: output.status,
            summary: output.summary.clone(),
            output: output.output,
            observed_at_ms: now,
        });
        run.status = RunStatus::Running;
        run.updated_at_ms = now;
        self.commit_transition(
            snapshot,
            vec![PendingEvent {
                kind: RuntimeEventKind::ActionObserved,
                run_id: Some(run_id.to_string()),
                detail: serde_json::json!({
                    "actionId": action.id,
                    "status": output.status,
                    "summary": output.summary,
                }),
            }],
        )
    }

    fn finish_run(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        terminal_status: RunStatus,
        summary: String,
        error: Option<String>,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        if !terminal_status.is_terminal() {
            return Err(RuntimeError::invalid("finish_run 需要终态 status"));
        }
        validate_description(&summary, "terminal summary")?;
        if let Some(error) = &error {
            validate_description(error, "terminal error")?;
        }
        if terminal_status == RunStatus::Failed && error.is_none() {
            return Err(RuntimeError::invalid("failed run 必须携带 error"));
        }
        let mut snapshot = self.load_required(runtime_id)?;
        let now = self.clock.now_millis();
        let run = run_mut(&mut snapshot, run_id)?;
        if run.status.is_terminal() {
            return Err(RuntimeError::transition(format!("run 已是终态：{run_id}")));
        }
        if terminal_status == RunStatus::Completed && run.status != RunStatus::Running {
            return Err(RuntimeError::transition(
                "只有 running run 可以进入 completed",
            ));
        }
        if matches!(
            run.status,
            RunStatus::WaitingForAction
                | RunStatus::WaitingForChildren
                | RunStatus::NeedsReconciliation
        ) {
            return Err(RuntimeError::transition(
                "未收束 action/delegation 时不能结束 run",
            ));
        }
        run.status = terminal_status;
        run.terminal_summary = Some(summary);
        run.terminal_error = error;
        run.updated_at_ms = now;
        let mut events = vec![PendingEvent {
            kind: match terminal_status {
                RunStatus::Completed => RuntimeEventKind::RunCompleted,
                RunStatus::Failed => RuntimeEventKind::RunFailed,
                RunStatus::Cancelled => RuntimeEventKind::RunCancelled,
                _ => unreachable!(),
            },
            run_id: Some(run_id.to_string()),
            detail: Value::Object(Map::new()),
        }];
        resolve_ready_joins(&mut snapshot, now, &mut events)?;
        self.commit_transition(snapshot, events)
    }

    fn change_status(
        &mut self,
        runtime_id: &str,
        run_id: &str,
        allowed: &[RunStatus],
        next: RunStatus,
        kind: RuntimeEventKind,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        let mut snapshot = self.load_required(runtime_id)?;
        let run = run_mut(&mut snapshot, run_id)?;
        require_status(run, allowed, "change status")?;
        run.status = next;
        run.updated_at_ms = self.clock.now_millis();
        self.commit_transition(
            snapshot,
            vec![PendingEvent {
                kind,
                run_id: Some(run_id.to_string()),
                detail: Value::Object(Map::new()),
            }],
        )
    }

    fn commit_transition(
        &mut self,
        mut snapshot: RuntimeSnapshot,
        pending_events: Vec<PendingEvent>,
    ) -> Result<RuntimeSnapshot, RuntimeError> {
        if pending_events.is_empty() {
            return Err(RuntimeError::invalid(
                "runtime transition 至少需要一个 event",
            ));
        }
        let expected_revision = snapshot.revision;
        snapshot.revision = expected_revision
            .checked_add(1)
            .ok_or_else(|| RuntimeError::conflict("runtime revision 溢出"))?;
        let now = self.clock.now_millis();
        let events = pending_events
            .into_iter()
            .map(|event| RuntimeEvent {
                runtime_id: snapshot.runtime_id.clone(),
                revision: snapshot.revision,
                occurred_at_ms: now,
                kind: event.kind,
                run_id: event.run_id,
                detail: event.detail,
            })
            .collect::<Vec<_>>();
        self.store
            .commit(
                &snapshot.runtime_id,
                Some(expected_revision),
                &snapshot,
                &events,
            )
            .map_err(RuntimeError::store)?;
        Ok(snapshot)
    }
}

fn run_mut<'a>(
    snapshot: &'a mut RuntimeSnapshot,
    run_id: &str,
) -> Result<&'a mut RunRecord, RuntimeError> {
    snapshot
        .runs
        .iter_mut()
        .find(|run| run.run_id == run_id)
        .ok_or_else(|| RuntimeError::not_found(format!("run 不存在：{run_id}")))
}

fn require_status(
    run: &RunRecord,
    allowed: &[RunStatus],
    operation: &str,
) -> Result<(), RuntimeError> {
    if allowed.contains(&run.status) {
        Ok(())
    } else {
        Err(RuntimeError::transition(format!(
            "{operation} 不允许 run {} 处于 {:?}",
            run.run_id, run.status
        )))
    }
}

fn resolve_ready_joins(
    snapshot: &mut RuntimeSnapshot,
    now: u64,
    events: &mut Vec<PendingEvent>,
) -> Result<(), RuntimeError> {
    let ready = snapshot
        .delegations
        .iter()
        .enumerate()
        .filter(|(_, group)| {
            !group.resolved
                && group.child_run_ids.iter().all(|child_id| {
                    snapshot
                        .run(child_id)
                        .is_some_and(|run| run.status.is_terminal())
                })
        })
        .map(|(index, _)| index)
        .collect::<Vec<_>>();
    for index in ready {
        let child_ids = snapshot.delegations[index].child_run_ids.clone();
        let results = child_ids
            .iter()
            .map(|child_id| {
                let child = snapshot.run(child_id).expect("ready child must exist");
                RunTerminalResult {
                    run_id: child.run_id.clone(),
                    status: child.status,
                    summary: child.terminal_summary.clone(),
                    error: child.terminal_error.clone(),
                }
            })
            .collect::<Vec<_>>();
        let parent_run_id = snapshot.delegations[index].parent_run_id.clone();
        let group_id = snapshot.delegations[index].id.clone();
        snapshot.delegations[index].resolved = true;
        snapshot.delegations[index].results = results;
        let parent = run_mut(snapshot, &parent_run_id)?;
        require_status(parent, &[RunStatus::WaitingForChildren], "resolve join")?;
        parent.status = RunStatus::Running;
        parent.updated_at_ms = now;
        events.push(PendingEvent {
            kind: RuntimeEventKind::JoinResolved,
            run_id: Some(parent_run_id),
            detail: serde_json::json!({"groupId": group_id, "childRunIds": child_ids}),
        });
    }
    Ok(())
}

fn validate_runtime_snapshot(
    snapshot: &RuntimeSnapshot,
    expected_runtime_id: &str,
) -> Result<(), RuntimeError> {
    if snapshot.schema_version != RUNTIME_SNAPSHOT_SCHEMA_VERSION {
        return Err(RuntimeError::conflict(format!(
            "runtime snapshot schemaVersion 不支持：{}",
            snapshot.schema_version
        )));
    }
    validate_identifier(&snapshot.runtime_id, "runtime id")?;
    if snapshot.runtime_id != expected_runtime_id || snapshot.revision == 0 {
        return Err(RuntimeError::conflict(
            "runtime snapshot identity/revision 与读取请求不一致",
        ));
    }
    let mut run_ids = BTreeSet::new();
    let mut action_ids = BTreeSet::new();
    for run in &snapshot.runs {
        validate_identifier(&run.run_id, "run id")?;
        validate_identifier(&run.agent_id, "agent id")?;
        validate_description(&run.task, "run task")?;
        validate_metadata(&run.metadata, "run metadata")?;
        if !run_ids.insert(run.run_id.as_str()) {
            return Err(RuntimeError::conflict(format!(
                "runtime snapshot run id 重复：{}",
                run.run_id
            )));
        }
        for action in &run.actions {
            validate_identifier(&action.action.id, "action id")?;
            validate_identifier(&action.action.capability_id, "capability id")?;
            if !action.action.input.is_object() || !action_ids.insert(action.action.id.as_str()) {
                return Err(RuntimeError::conflict(
                    "runtime snapshot action input 非 object 或 action id 重复",
                ));
            }
        }
        let active_action = run
            .actions
            .last()
            .filter(|action| action.status != ActionStatus::Observed);
        match run.status {
            RunStatus::WaitingForAction
                if !active_action.is_some_and(|action| {
                    matches!(
                        action.status,
                        ActionStatus::Queued | ActionStatus::Executing
                    )
                }) =>
            {
                return Err(RuntimeError::conflict(
                    "waiting-for-action run 缺少 queued/executing action",
                ));
            }
            RunStatus::NeedsReconciliation
                if !active_action
                    .is_some_and(|action| action.status == ActionStatus::Executing) =>
            {
                return Err(RuntimeError::conflict(
                    "needs-reconciliation run 缺少 executing action",
                ));
            }
            status
                if !matches!(
                    status,
                    RunStatus::WaitingForAction | RunStatus::NeedsReconciliation
                ) && active_action.is_some() =>
            {
                return Err(RuntimeError::conflict(
                    "非 action 等待态不能保留未观察 action",
                ));
            }
            _ => {}
        }
    }
    let mut group_ids = BTreeSet::new();
    for group in &snapshot.delegations {
        validate_identifier(&group.id, "delegation group id")?;
        let unique_child_ids =
            group.child_run_ids.iter().collect::<BTreeSet<_>>().len() == group.child_run_ids.len();
        if !group_ids.insert(group.id.as_str())
            || !run_ids.contains(group.parent_run_id.as_str())
            || group.child_run_ids.is_empty()
            || !unique_child_ids
            || group
                .child_run_ids
                .iter()
                .any(|child_id| !run_ids.contains(child_id.as_str()))
        {
            return Err(RuntimeError::conflict(
                "runtime snapshot delegation identity/reference 无效",
            ));
        }
        if group.resolved && group.results.len() != group.child_run_ids.len() {
            return Err(RuntimeError::conflict(
                "resolved delegation results 数量不匹配",
            ));
        }
        if group.resolved
            && group
                .results
                .iter()
                .zip(&group.child_run_ids)
                .any(|(result, child_id)| {
                    result.run_id != *child_id
                        || !result.status.is_terminal()
                        || snapshot.run(child_id).is_none_or(|child| {
                            child.status != result.status
                                || child.terminal_summary != result.summary
                                || child.terminal_error != result.error
                        })
                })
        {
            return Err(RuntimeError::conflict(
                "resolved delegation result identity/order 无效",
            ));
        }
        let parent = snapshot
            .run(&group.parent_run_id)
            .expect("validated parent reference");
        if !group.resolved && parent.status != RunStatus::WaitingForChildren {
            return Err(RuntimeError::conflict(
                "unresolved delegation parent 必须等待 children",
            ));
        }
        for child_id in &group.child_run_ids {
            let child = snapshot.run(child_id).expect("validated child reference");
            if child.parent_run_id.as_deref() != Some(group.parent_run_id.as_str())
                || child.delegation_group_id.as_deref() != Some(group.id.as_str())
            {
                return Err(RuntimeError::conflict(
                    "delegation child parent/group binding 不匹配",
                ));
            }
        }
    }
    for run in &snapshot.runs {
        if run.status == RunStatus::WaitingForChildren
            && snapshot
                .delegations
                .iter()
                .filter(|group| group.parent_run_id == run.run_id && !group.resolved)
                .count()
                != 1
        {
            return Err(RuntimeError::conflict(
                "waiting-for-children run 必须绑定唯一 unresolved delegation",
            ));
        }
    }
    Ok(())
}