af_agent_session/

lib.rs

//! Event-sourced Agent sessions. The append-only log is the conversation SSOT;
//! messages, runs, interactions and UI state are deterministic projections.

#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]

mod corrections;
pub use corrections::{
    AppliedMeteringCorrection, MeteringCorrection, OperationUsage, RunOperationUsage,
};

mod metering;
pub use metering::{MeteringDetails, MeteringOutcome, MeteringSource};

use af_context::{InputId, InteractionId, ProfileRevisionId, RunId, SessionId, ToolCallId};
use std::collections::{BTreeMap, BTreeSet};

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;

/// Wire/storage format of [`Event`]; only structural changes bump it.
pub const SESSION_EVENT_FORMAT_VERSION: u32 = 1;

/// Versioned organizational metadata projected from the Session log.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SessionMetadata {
    /// User title; empty means the consumer derives an automatic title.
    pub title: String,
    /// Archived Sessions reject new input; already accepted work may converge.
    pub archived: bool,
    /// Monotonic metadata version independent of running event traffic.
    pub version: u64,
}

impl SessionMetadata {
    /// Validate display metadata at every public write and replay boundary.
    pub fn validate(&self) -> Result<(), EventError> {
        if self.title.chars().count() > 200 || self.title.chars().any(char::is_control) {
            return Err(EventError::InvalidSessionMetadata);
        }
        Ok(())
    }
}

/// One committed Session event with its position in the log.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SessionEvent {
    /// Session this record belongs to.
    pub session_id: SessionId,
    /// Strictly increasing sequence inside the Session.
    pub seq: u64,
    /// When the event happened.
    pub occurred_at: DateTime<Utc>,
    /// The event payload.
    pub event: Event,
}

impl SessionEvent {
    /// An event awaiting append: sequence 0 until the store assigns it.
    pub fn pending(session_id: impl Into<SessionId>, event: Event) -> Self {
        Self {
            session_id: session_id.into(),
            seq: 0,
            occurred_at: Utc::now(),
            event,
        }
    }

    /// Format version of the payload.
    pub fn format_version(&self) -> u32 {
        self.event.format_version()
    }

    /// Stable snake_case event type.
    pub fn event_type(&self) -> &str {
        self.event.event_type()
    }

    /// Whether a reader that does not know this event may skip it.
    pub fn ignorable(&self) -> bool {
        self.event.ignorable()
    }
}

/// Every model-visible fact and lifecycle transition of a Session. The log is the source of truth; everything else is a projection.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum Event {
    /// First event of a Session, pinning its Profile revision.
    SessionCreated {
        /// Profile revision the Session pins for its lifetime.
        profile_revision_id: ProfileRevisionId,
    },
    /// First event of a forked Session, citing the parent position it copied through.
    SessionForked {
        /// Session the fork was cut from.
        parent_session_id: SessionId,
        /// Last parent sequence included in the fork.
        parent_seq: u64,
    },
    /// Atomically replace organizational metadata at the next metadata version.
    SessionMetadataUpdated {
        /// Full immutable metadata fact.
        metadata: SessionMetadata,
    },
    /// Tombstone: active and queued Runs are closed, listing hides the Session, history stays readable.
    SessionDeleted {
        /// Why the Session was deleted.
        reason: String,
    },
    /// Input entered the inbox for `run_id`.
    InputQueued {
        /// Queued input.
        input_id: InputId,
        /// Run that will consume the input (a new Run for `followup`).
        run_id: RunId,
        /// Delivery semantics.
        mode: DeliveryMode,
        /// Submitted content.
        content: Vec<ContentBlock>,
        /// Skill slug invoked explicitly with this input.
        explicit_skill: Option<String>,
    },
    /// The Run took the input out of the inbox.
    InputClaimed {
        /// Claimed input.
        input_id: InputId,
        /// Claiming Run.
        run_id: RunId,
    },
    /// The input was dropped before a Run consumed it.
    InputCancelled {
        /// Cancelled input.
        input_id: InputId,
        /// Run that would have consumed it.
        run_id: RunId,
        /// Machine-readable cancellation code.
        error_code: String,
    },
    /// A Run began executing.
    RunStarted {
        /// The Run.
        run_id: RunId,
        /// Input that started it.
        input_id: InputId,
    },
    /// The Run parked on an interaction.
    RunWaiting {
        /// The Run.
        run_id: RunId,
        /// Interaction it waits for.
        interaction_id: InteractionId,
    },
    /// The Run resumed after its interaction resolved.
    RunResumed {
        /// The Run.
        run_id: RunId,
        /// Resolved interaction.
        interaction_id: InteractionId,
    },
    /// The Run reached a terminal status.
    RunFinished {
        /// The Run.
        run_id: RunId,
        /// Terminal status.
        status: RunStatus,
        /// Machine-readable failure code.
        error_code: Option<String>,
    },
    /// A Turn opened.
    TurnStarted {
        /// The Run.
        run_id: RunId,
        /// Turn number.
        turn: u32,
    },
    /// A Turn closed.
    TurnFinished {
        /// The Run.
        run_id: RunId,
        /// Turn number.
        turn: u32,
    },
    /// A Step opened.
    StepStarted {
        /// The Run.
        run_id: RunId,
        /// Step number.
        step: u32,
    },
    /// A Step closed.
    StepFinished {
        /// The Run.
        run_id: RunId,
        /// Step number.
        step: u32,
    },
    /// User content entered the transcript.
    UserMessage {
        /// The Run.
        run_id: RunId,
        /// Content blocks.
        content: Vec<ContentBlock>,
    },
    /// Streamed assistant text fragment.
    AssistantDelta {
        /// The Run.
        run_id: RunId,
        /// Step producing it.
        step: u32,
        /// Provider attempt producing it.
        attempt: u32,
        /// Text fragment.
        content: String,
    },
    /// Final assistant content for a step.
    AssistantMessage {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Provider attempt that succeeded.
        attempt: u32,
        /// Content blocks.
        content: Vec<ContentBlock>,
    },
    /// The model requested tool calls.
    AssistantToolCalls {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Text accompanying the calls.
        content: Option<String>,
        /// Requested calls in model order.
        calls: Vec<RecordedToolCall>,
    },
    /// One tool call was admitted with immutable arguments.
    ToolCall {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Call identity.
        call_id: ToolCallId,
        /// Tool name.
        tool: String,
        /// Immutable arguments.
        arguments: Value,
    },
    /// Authorization outcome for a tool call.
    ToolAuthorization {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Call identity.
        call_id: ToolCallId,
        /// Outcome.
        status: ToolAuthorizationStatus,
        /// Why it was denied or parked.
        reason: Option<String>,
    },
    /// Tool execution began.
    ToolExecutionStarted {
        /// Immutable tool attribution; absent only in legacy logs.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        metering: Option<MeteringDetails>,
        /// Declared resource weight frozen before dispatch.
        #[serde(default)]
        reserved_cost_units: u64,
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Call identity.
        call_id: ToolCallId,
    },
    /// Tool execution finished.
    ToolResult {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Call identity.
        call_id: ToolCallId,
        /// Result value.
        result: Value,
        /// Whether the result is an error.
        is_error: bool,
    },
    /// Privileged correction; does not reopen execution or its reservation settlement.
    UsageCorrected {
        /// Immutable reported measurement and idempotency key.
        correction: MeteringCorrection,
        /// Authenticated reporting subject.
        actor_id: af_context::SubjectId,
    },
    /// Usage settled for one idempotent operation.
    UsageRecorded {
        /// Immutable attribution; absent only on legacy events.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        metering: Option<MeteringDetails>,
        /// The Run.
        run_id: RunId,
        /// Idempotent operation identity.
        operation_id: String,
        /// Prompt tokens.
        prompt_tokens: u64,
        /// Completion tokens.
        completion_tokens: u64,
        /// Provider-neutral cost units.
        #[serde(default)]
        cost_units: u64,
    },
    /// A model retry was scheduled.
    RetryScheduled {
        /// The Run.
        run_id: RunId,
        /// Failed attempt number.
        attempt: u32,
        /// Backoff before the next attempt.
        delay_ms: u64,
        /// Failure reason.
        reason: String,
    },
    /// The frozen model request before it is sent; the durable attempt identity makes crashes reconcilable.
    ModelRequestPrepared {
        /// Attribution frozen before dispatch; absent on legacy events.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        metering: Option<MeteringDetails>,
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Attempt number.
        attempt: u32,
        /// Durable provider attempt identity.
        #[serde(default)]
        provider_attempt_id: String,
        /// Usage operation identity.
        #[serde(default)]
        operation_id: String,
        /// Prompt tokens reserved.
        #[serde(default)]
        reserved_prompt_tokens: u64,
        /// Completion tokens reserved.
        #[serde(default)]
        reserved_completion_tokens: u64,
        /// Redacted request as sent.
        request: Value,
        /// Prompt sections the request was composed from.
        prompt_sections: Value,
    },
    /// Context a contributor injected into a step.
    ContextInjected {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Contribution identity.
        contribution_id: String,
        /// Contributor source.
        source: String,
        /// Contributor version.
        version: String,
        /// `trusted` or `untrusted`.
        authority: String,
        /// Rendering form.
        form: String,
        /// Content blocks.
        content: Vec<ContentBlock>,
    },
    /// A model attempt failed.
    ModelAttemptFailed {
        /// The Run.
        run_id: RunId,
        /// Step.
        step: u32,
        /// Attempt number.
        attempt: u32,
        /// Failure.
        error: String,
        /// Whether a retry may help.
        retryable: bool,
    },
    /// Compaction began.
    CompactionStarted {
        /// The Run.
        run_id: RunId,
        /// Compaction identity.
        compaction_id: String,
        /// Last sequence the summary covers.
        source_through_seq: u64,
    },
    /// Large tool results were truncated in the transcript.
    ToolResultsPruned {
        /// The Run.
        run_id: RunId,
        /// Truncated calls.
        call_ids: Vec<ToolCallId>,
    },
    /// A summary replaced transcript through `through_seq`.
    SummaryReplaced {
        /// The Run.
        run_id: RunId,
        /// Last sequence replaced.
        through_seq: u64,
        /// Summary text.
        summary: String,
        /// Compactor name.
        compactor: String,
        /// Model used.
        model: String,
    },
    /// Compaction finished.
    CompactionFinished {
        /// The Run.
        run_id: RunId,
        /// Compaction identity.
        compaction_id: String,
        /// `completed` or `failed`.
        status: String,
        /// Failure, when failed.
        error: Option<String>,
    },
    /// The Run asked the user for an approval or answer.
    InteractionRequested {
        /// The Run.
        run_id: RunId,
        /// Interaction identity.
        interaction_id: InteractionId,
        /// Approval or question.
        kind: InteractionKind,
        /// What is being asked.
        payload: Value,
    },
    /// The user resolved an interaction.
    InteractionResolved {
        /// The Run.
        run_id: RunId,
        /// Interaction identity.
        interaction_id: InteractionId,
        /// Outcome.
        resolution: InteractionResolution,
        /// Answer payload.
        payload: Value,
    },
    /// A child Session was created for this Run.
    ChildSessionLinked {
        /// The Run.
        run_id: RunId,
        /// Child Session.
        child_session_id: SessionId,
        /// Subagent provider.
        provider: String,
    },
    /// Plugin-defined event; readers that do not know `event_type` treat it as data.
    Extension {
        /// The Run.
        run_id: RunId,
        /// Plugin that emitted it.
        plugin_id: String,
        /// Plugin event type.
        event_type: String,
        /// Plugin payload.
        payload: Value,
    },
    /// An event this build cannot decode; `ignorable` decides whether replay may skip it.
    #[serde(skip)]
    Opaque {
        /// Producer format version.
        format_version: u32,
        /// Producer event type.
        event_type: String,
        /// Whether readers may skip it.
        ignorable: bool,
        /// Raw payload.
        payload: Value,
    },
}

impl Event {
    /// Format version of this event.
    pub fn format_version(&self) -> u32 {
        match self {
            Self::Opaque { format_version, .. } => *format_version,
            _ => SESSION_EVENT_FORMAT_VERSION,
        }
    }

    /// Stable snake_case event type.
    pub fn event_type(&self) -> &str {
        match self {
            Self::SessionCreated { .. } => "session_created",
            Self::SessionMetadataUpdated { .. } => "session_metadata_updated",
            Self::SessionForked { .. } => "session_forked",
            Self::SessionDeleted { .. } => "session_deleted",
            Self::InputQueued { .. } => "input_queued",
            Self::InputClaimed { .. } => "input_claimed",
            Self::InputCancelled { .. } => "input_cancelled",
            Self::RunStarted { .. } => "run_started",
            Self::RunWaiting { .. } => "run_waiting",
            Self::RunResumed { .. } => "run_resumed",
            Self::RunFinished { .. } => "run_finished",
            Self::TurnStarted { .. } => "turn_started",
            Self::TurnFinished { .. } => "turn_finished",
            Self::StepStarted { .. } => "step_started",
            Self::StepFinished { .. } => "step_finished",
            Self::UserMessage { .. } => "user_message",
            Self::AssistantDelta { .. } => "assistant_delta",
            Self::AssistantMessage { .. } => "assistant_message",
            Self::AssistantToolCalls { .. } => "assistant_tool_calls",
            Self::ToolCall { .. } => "tool_call",
            Self::ToolAuthorization { .. } => "tool_authorization",
            Self::ToolExecutionStarted { .. } => "tool_execution_started",
            Self::ToolResult { .. } => "tool_result",
            Self::UsageRecorded { .. } => "usage_recorded",
            Self::UsageCorrected { .. } => "usage_corrected",
            Self::RetryScheduled { .. } => "retry_scheduled",
            Self::ModelRequestPrepared { .. } => "model_request_prepared",
            Self::ContextInjected { .. } => "context_injected",
            Self::ModelAttemptFailed { .. } => "model_attempt_failed",
            Self::CompactionStarted { .. } => "compaction_started",
            Self::ToolResultsPruned { .. } => "tool_results_pruned",
            Self::SummaryReplaced { .. } => "summary_replaced",
            Self::CompactionFinished { .. } => "compaction_finished",
            Self::InteractionRequested { .. } => "interaction_requested",
            Self::InteractionResolved { .. } => "interaction_resolved",
            Self::ChildSessionLinked { .. } => "child_session_linked",
            Self::Extension { .. } => "extension",
            Self::Opaque { event_type, .. } => event_type,
        }
    }

    /// Whether an unknown reader may skip this event.
    pub fn ignorable(&self) -> bool {
        match self {
            Self::Opaque { ignorable, .. } => *ignorable,
            _ => false,
        }
    }
}

/// How queued input reaches a Run.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeliveryMode {
    /// Runs as its own Turn after the current Run.
    Followup,
    /// Interrupts the current Run at the next step boundary.
    Steer,
    /// Added as context to the next step without waking the loop.
    Inject,
}

/// Terminal Run status.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunStatus {
    /// Finished with an answer.
    Completed,
    /// Failed with an error code.
    Failed,
    /// Cancelled by the user or platform.
    Cancelled,
    /// Stopped at a step, tool-call or token limit.
    MaxStepsReached,
}

impl RunStatus {
    /// Stable snake_case name.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Completed => "completed",
            Self::Failed => "failed",
            Self::Cancelled => "cancelled",
            Self::MaxStepsReached => "max_steps_reached",
        }
    }
}

/// What the Run asks the user for.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InteractionKind {
    /// Approve or reject a side effect.
    Action,
    /// Answer a question.
    UserQuestion,
}

impl InteractionKind {
    /// Stable snake_case name.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Action => "action",
            Self::UserQuestion => "user_question",
        }
    }
}

/// How the user resolved an interaction.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InteractionResolution {
    /// Approved.
    Confirmed,
    /// Rejected.
    Rejected,
    /// Answered a question.
    Answered,
}

impl InteractionResolution {
    /// Stable snake_case name.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Confirmed => "confirmed",
            Self::Rejected => "rejected",
            Self::Answered => "answered",
        }
    }
}

/// Authorization outcome of a tool call.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ToolAuthorizationStatus {
    /// May execute.
    Allowed,
    /// Parked until the interaction resolves.
    Waiting,
    /// Refused.
    Denied,
}

/// A tool call as recorded in `AssistantToolCalls`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RecordedToolCall {
    /// Tool call this record refers to.
    pub call_id: ToolCallId,
    /// Tool name.
    pub tool: String,
    /// JSON arguments passed to the tool.
    pub arguments: Value,
}

impl ToolAuthorizationStatus {
    /// Stable snake_case name.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Allowed => "allowed",
            Self::Waiting => "waiting",
            Self::Denied => "denied",
        }
    }
}

/// Typed content unit; UI and models parse types, never prose conventions.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ContentBlock {
    /// Plain text.
    Text {
        /// The text.
        text: String,
    },
    /// Attached resource resolved by the host.
    Resource {
        /// Resource identity.
        resource_id: String,
        /// MIME type.
        media_type: String,
    },
    /// Product-typed data for a registered UI slot.
    Data {
        /// Slot name.
        slot: String,
        /// Slot payload.
        value: Value,
    },
    /// Citation of a retrieved resource.
    Citation {
        /// Resource identity.
        resource_id: String,
        /// Display label.
        label: String,
        /// Where to open it.
        uri: String,
        /// Supporting excerpt.
        excerpt: Option<String>,
    },
}

/// Deterministic projection of a Session log used by the runtime, recovery and UI.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct SessionProjection {
    /// Latest versioned Session title and archival state.
    pub metadata: SessionMetadata,
    /// Session this record belongs to; `None` until the first event is applied.
    pub session_id: Option<SessionId>,
    /// Immutable Agent Profile revision pinned by the Session; `None` before
    /// `SessionCreated`.
    pub profile_revision_id: Option<ProfileRevisionId>,
    /// Whether the Session was deleted.
    pub deleted: bool,
    /// Sequence of the last applied event.
    pub last_seq: u64,
    /// Run currently executing, if any.
    pub active_run_id: Option<RunId>,
    /// Interaction the active Run waits for, if any.
    pub waiting_interaction_id: Option<InteractionId>,
    /// Conversation messages in request order.
    pub messages: Vec<ProjectedMessage>,
    /// Context injected so far, in order.
    pub injected_context: Vec<ProjectedContext>,
    /// State of every Run seen.
    pub run_status: BTreeMap<RunId, RunState>,
    /// Tool calls admitted but not yet resulted.
    pub open_tool_calls: BTreeMap<ToolCallId, OpenToolCall>,
    /// Open tool calls whose execution started.
    pub started_tool_calls: BTreeSet<ToolCallId>,
    /// Inputs queued but not claimed.
    pub queued_inputs: BTreeMap<InputId, (RunId, DeliveryMode)>,
    /// Inputs claimed by a Run.
    pub claimed_inputs: BTreeMap<InputId, RunId>,
    /// Open Turn, if any.
    pub open_turn: Option<u32>,
    /// Open steps (at most one).
    pub open_steps: BTreeSet<u32>,
    /// Next step number.
    pub next_step: u32,
    /// Open compaction `(id, through_seq)`, if any.
    pub open_compaction: Option<(String, u64)>,
    /// Short human-readable summary.
    pub summary: Option<String>,
    corrected_usage: BTreeMap<(RunId, String), OperationUsage>,
    metering_corrections: BTreeMap<af_context::MeteringCorrectionId, AppliedMeteringCorrection>,
    usage_operations: BTreeMap<(RunId, String), RecordedUsage>,
    pending_usage_operations: BTreeMap<(RunId, String), PreparedUsage>,
    seen_tool_calls: BTreeSet<ToolCallId>,
}

type PreparedUsage = RecordedUsage;

type RecordedUsage = (u64, u64, u64, Option<MeteringDetails>);

/// Projected lifecycle state of one Run: live, parked on an interaction, or terminal.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunState {
    /// Executing.
    Running,
    /// Parked on an interaction.
    WaitingForInput,
    /// Finished.
    Terminal(RunStatus),
}

impl RunState {
    /// Stable snake_case name.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Running => "running",
            Self::WaitingForInput => "waiting_for_input",
            Self::Terminal(status) => status.as_str(),
        }
    }

    /// Whether the Run has finished.
    pub const fn is_terminal(self) -> bool {
        matches!(self, Self::Terminal(_))
    }
}

/// Injected context as projected for UI and replay.
#[derive(Debug, Clone, PartialEq)]
pub struct ProjectedContext {
    /// Run this record belongs to.
    pub run_id: RunId,
    /// 1-based step number inside the Turn.
    pub step: u32,
    /// Contribution identity.
    pub contribution_id: String,
    /// Contributor source.
    pub source: String,
    /// Semantic version string.
    pub version: String,
    /// `trusted` or `untrusted`.
    pub authority: String,
    /// Rendering form.
    pub form: String,
    /// Content blocks carried by this record.
    pub content: Vec<ContentBlock>,
}

/// Message as projected for UI and replay.
#[derive(Debug, Clone, PartialEq)]
pub struct ProjectedMessage {
    /// `user` or `assistant`.
    pub role: &'static str,
    /// Run this record belongs to.
    pub run_id: RunId,
    /// Content blocks carried by this record.
    pub content: Vec<ContentBlock>,
}

/// A tool call awaiting its result.
#[derive(Debug, Clone, PartialEq)]
pub struct OpenToolCall {
    /// Run this record belongs to.
    pub run_id: RunId,
    /// 1-based step number inside the Turn.
    pub step: u32,
    /// Tool name.
    pub tool: String,
    /// JSON arguments passed to the tool.
    pub arguments: Value,
    /// Sequence of the Session event this was derived from.
    pub source_event_seq: u64,
}

impl SessionProjection {
    /// Rebuild the projection from a committed log; fails on any invariant violation.
    pub fn replay(events: &[SessionEvent]) -> Result<Self, EventError> {
        let mut projection = Self::default();
        for event in events {
            projection.apply(event)?;
        }
        Ok(projection)
    }

    /// Apply one committed event, enforcing sequence, lifecycle and pairing invariants.
    pub fn apply(&mut self, envelope: &SessionEvent) -> Result<(), EventError> {
        self.apply_internal(envelope, true)
    }

    /// Reduce execution/authorization/usage facts without retaining UI messages
    /// or injected prompt content. Suitable for incremental replay of bounded pages.
    pub fn apply_facts(&mut self, envelope: &SessionEvent) -> Result<(), EventError> {
        self.messages.clear();
        self.injected_context.clear();
        self.apply_internal(envelope, false)
    }

    /// Fold the same lifecycle invariants without building a transcript copy.
    pub fn replay_facts(events: &[SessionEvent]) -> Result<Self, EventError> {
        let mut projection = Self::default();
        for event in events {
            projection.apply_facts(event)?;
        }
        Ok(projection)
    }

    fn apply_internal(
        &mut self,
        envelope: &SessionEvent,
        include_content: bool,
    ) -> Result<(), EventError> {
        if envelope.seq != self.last_seq + 1 {
            return Err(EventError::Sequence {
                expected: self.last_seq + 1,
                actual: envelope.seq,
            });
        }
        let session_id = self
            .session_id
            .get_or_insert_with(|| envelope.session_id.clone());
        if *session_id != envelope.session_id {
            return Err(EventError::SessionMismatch);
        }
        if envelope.format_version() != SESSION_EVENT_FORMAT_VERSION {
            if envelope.ignorable() {
                self.last_seq = envelope.seq;
                return Ok(());
            }
            return Err(EventError::UnsupportedFormat(envelope.format_version()));
        }
        if self.deleted && !matches!(envelope.event, Event::UsageCorrected { .. }) {
            return Err(EventError::SessionClosed);
        }
        match &envelope.event {
            Event::UsageCorrected {
                correction,
                actor_id,
            } => self.apply_metering_correction(correction, actor_id, envelope.seq)?,
            Event::SessionCreated {
                profile_revision_id,
            } => {
                if envelope.seq != 1 || self.profile_revision_id.is_some() {
                    return Err(EventError::DuplicateSession);
                }
                self.profile_revision_id = Some(profile_revision_id.clone());
            }
            Event::SessionMetadataUpdated { metadata } => {
                metadata.validate()?;
                if self.profile_revision_id.is_none()
                    || self.metadata.version.checked_add(1) != Some(metadata.version)
                {
                    return Err(EventError::InvalidSessionMetadata);
                }
                self.metadata = metadata.clone();
            }
            Event::SessionDeleted { .. } => {
                if let Some(run_id) = self.active_run_id.take() {
                    self.run_status
                        .insert(run_id, RunState::Terminal(RunStatus::Cancelled));
                }
                for (_, (run_id, _)) in std::mem::take(&mut self.queued_inputs) {
                    self.run_status
                        .insert(run_id, RunState::Terminal(RunStatus::Cancelled));
                }
                self.waiting_interaction_id = None;
                self.open_turn = None;
                self.open_steps.clear();
                self.open_tool_calls.clear();
                self.started_tool_calls.clear();
                self.open_compaction = None;
                self.deleted = true;
            }
            Event::RunStarted { run_id, input_id } => {
                if self.active_run_id.is_some() {
                    return Err(EventError::ConcurrentRun);
                }
                if self.claimed_inputs.get(input_id) != Some(run_id) {
                    return Err(EventError::UnclaimedInput(input_id.clone()));
                }
                self.active_run_id = Some(run_id.clone());
                self.next_step = 1;
                self.run_status.insert(run_id.clone(), RunState::Running);
            }
            Event::RunWaiting {
                run_id,
                interaction_id,
            } => {
                self.require_active(run_id)?;
                self.waiting_interaction_id = Some(interaction_id.clone());
                self.run_status
                    .insert(run_id.clone(), RunState::WaitingForInput);
            }
            Event::RunResumed {
                run_id,
                interaction_id,
            } => {
                self.require_active(run_id)?;
                if self.waiting_interaction_id.as_ref() != Some(interaction_id) {
                    return Err(EventError::InteractionMismatch);
                }
                self.waiting_interaction_id = None;
                self.run_status.insert(run_id.clone(), RunState::Running);
            }
            Event::RunFinished { run_id, status, .. } => {
                self.require_active(run_id)?;
                if !self.open_tool_calls.is_empty()
                    || !self.open_steps.is_empty()
                    || self.open_turn.is_some()
                    || self.open_compaction.is_some()
                    || self
                        .queued_inputs
                        .values()
                        .any(|(target_run_id, _)| target_run_id == run_id)
                {
                    return Err(EventError::OpenLifecycle);
                }
                self.run_status
                    .insert(run_id.clone(), RunState::Terminal(*status));
                self.active_run_id = None;
                self.waiting_interaction_id = None;
            }
            Event::UserMessage { run_id, content } if include_content => {
                self.messages.push(ProjectedMessage {
                    role: "user",
                    run_id: run_id.clone(),
                    content: content.clone(),
                })
            }
            Event::AssistantMessage {
                run_id, content, ..
            } if include_content => self.messages.push(ProjectedMessage {
                role: "assistant",
                run_id: run_id.clone(),
                content: content.clone(),
            }),
            Event::ContextInjected {
                run_id,
                step,
                contribution_id,
                source,
                version,
                authority,
                form,
                content,
            } => {
                self.require_active(run_id)?;
                if !self.open_steps.contains(step) {
                    return Err(EventError::LifecycleMismatch);
                }
                if include_content {
                    self.injected_context.push(ProjectedContext {
                        run_id: run_id.clone(),
                        step: *step,
                        contribution_id: contribution_id.clone(),
                        source: source.clone(),
                        version: version.clone(),
                        authority: authority.clone(),
                        form: form.clone(),
                        content: content.clone(),
                    });
                }
            }
            Event::InputQueued {
                input_id,
                run_id,
                mode,
                ..
            } => {
                if self.metadata.archived {
                    return Err(EventError::SessionArchived);
                }
                if *mode != DeliveryMode::Followup {
                    self.require_active(run_id)?;
                }
                if self
                    .queued_inputs
                    .insert(input_id.clone(), (run_id.clone(), *mode))
                    .is_some()
                {
                    return Err(EventError::DuplicateInput(input_id.clone()));
                }
            }
            Event::InputClaimed { input_id, run_id } => {
                if self
                    .queued_inputs
                    .remove(input_id)
                    .map(|value| value.0)
                    .as_ref()
                    != Some(run_id)
                    || self
                        .claimed_inputs
                        .insert(input_id.clone(), run_id.clone())
                        .is_some()
                {
                    return Err(EventError::UnqueuedInput(input_id.clone()));
                }
            }
            Event::InputCancelled {
                input_id,
                run_id,
                error_code,
            } => {
                if self
                    .queued_inputs
                    .remove(input_id)
                    .map(|value| value.0)
                    .as_ref()
                    != Some(run_id)
                {
                    return Err(EventError::UnqueuedInput(input_id.clone()));
                }
                self.run_status.insert(
                    run_id.clone(),
                    RunState::Terminal(if error_code == "cancelled" {
                        RunStatus::Cancelled
                    } else {
                        RunStatus::Failed
                    }),
                );
            }
            Event::TurnStarted { run_id, turn } => {
                self.require_active(run_id)?;
                if self.open_turn.replace(*turn).is_some() {
                    return Err(EventError::ConcurrentTurn);
                }
            }
            Event::TurnFinished { run_id, turn } => {
                self.require_active(run_id)?;
                if self.open_turn != Some(*turn)
                    || !self.open_steps.is_empty()
                    || !self.open_tool_calls.is_empty()
                {
                    return Err(EventError::LifecycleMismatch);
                }
                self.open_turn = None;
            }
            Event::StepStarted { run_id, step } => {
                self.require_active(run_id)?;
                if self.open_turn.is_none()
                    || !self.open_steps.is_empty()
                    || *step != self.next_step
                    || !self.open_steps.insert(*step)
                {
                    return Err(EventError::ConcurrentStep);
                }
            }
            Event::StepFinished { run_id, step } => {
                self.require_active(run_id)?;
                if self
                    .open_tool_calls
                    .values()
                    .any(|call| call.run_id == *run_id && call.step == *step)
                    || !self.open_steps.remove(step)
                {
                    return Err(EventError::LifecycleMismatch);
                }
                self.next_step = step.saturating_add(1);
            }
            Event::ToolCall {
                run_id,
                step,
                call_id,
                tool,
                arguments,
            } => {
                self.require_active(run_id)?;
                if !self.open_steps.contains(step) {
                    return Err(EventError::LifecycleMismatch);
                }
                if !self.seen_tool_calls.insert(call_id.clone())
                    || self
                        .open_tool_calls
                        .insert(
                            call_id.clone(),
                            OpenToolCall {
                                run_id: run_id.clone(),
                                step: *step,
                                tool: tool.clone(),
                                arguments: arguments.clone(),
                                source_event_seq: envelope.seq,
                            },
                        )
                        .is_some()
                {
                    return Err(EventError::DuplicateToolCall(call_id.clone()));
                }
            }
            Event::ToolResult {
                run_id,
                step,
                call_id,
                ..
            } => {
                self.require_active(run_id)?;
                if !self.open_steps.contains(step) {
                    return Err(EventError::LifecycleMismatch);
                }
                let Some(call) = self.open_tool_calls.get(call_id) else {
                    return Err(EventError::OrphanToolResult(call_id.clone()));
                };
                if call.run_id != *run_id || call.step != *step {
                    return Err(EventError::ToolResultMismatch(call_id.clone()));
                }
                self.open_tool_calls.remove(call_id);
                self.started_tool_calls.remove(call_id);
            }
            Event::ToolExecutionStarted {
                run_id,
                step,
                call_id,
                metering,
                reserved_cost_units,
            } => {
                self.require_active(run_id)?;
                let Some(call) = self.open_tool_calls.get(call_id) else {
                    return Err(EventError::OrphanToolResult(call_id.clone()));
                };
                if call.run_id != *run_id
                    || call.step != *step
                    || self.started_tool_calls.contains(call_id)
                {
                    return Err(EventError::ToolResultMismatch(call_id.clone()));
                }
                if let Some(details) = metering {
                    details.validate()?;
                    if !matches!(details, MeteringDetails::Tool { call_id: id, name, source: MeteringSource::Estimated, outcome: MeteringOutcome::Unknown } if id == call_id && name == &call.tool)
                    {
                        return Err(EventError::InvalidMetering);
                    }
                }
                self.started_tool_calls.insert(call_id.clone());
                self.pending_usage_operations.insert(
                    (run_id.clone(), format!("tool:{call_id}")),
                    (0, 0, *reserved_cost_units, metering.clone()),
                );
            }
            Event::ModelRequestPrepared {
                run_id,
                operation_id,
                reserved_prompt_tokens,
                reserved_completion_tokens,
                metering,
                ..
            } if !operation_id.is_empty() => {
                self.require_active(run_id)?;
                let key = (run_id.clone(), operation_id.clone());
                if !self.usage_operations.contains_key(&key) {
                    if let Some(details) = metering {
                        details.validate()?;
                        if !matches!(
                            details,
                            MeteringDetails::Model {
                                source: MeteringSource::Estimated,
                                outcome: MeteringOutcome::Unknown,
                                ..
                            }
                        ) {
                            return Err(EventError::InvalidMetering);
                        }
                    }
                    let reservation = (
                        *reserved_prompt_tokens,
                        *reserved_completion_tokens,
                        0,
                        metering.clone(),
                    );
                    match self.pending_usage_operations.get(&key) {
                        Some(existing) if *existing != reservation => {
                            return Err(EventError::UsageConflict(operation_id.clone()));
                        }
                        Some(_) => {}
                        None => {
                            self.pending_usage_operations.insert(key, reservation);
                        }
                    }
                }
            }
            Event::UsageRecorded {
                run_id,
                operation_id,
                prompt_tokens,
                completion_tokens,
                cost_units,
                metering,
            } => {
                self.require_active(run_id)?;
                if let Some(details) = metering {
                    details.validate()?;
                }
                let key = (run_id.clone(), operation_id.clone());
                if let Some((_, _, _, Some(prepared))) = self.pending_usage_operations.get(&key) {
                    if !metering
                        .as_ref()
                        .is_some_and(|details| details.completes(prepared))
                    {
                        return Err(EventError::UsageConflict(operation_id.clone()));
                    }
                }
                let usage = (
                    *prompt_tokens,
                    *completion_tokens,
                    *cost_units,
                    metering.clone(),
                );
                match self.usage_operations.get(&key) {
                    Some(existing) if *existing != usage => {
                        return Err(EventError::UsageConflict(operation_id.clone()));
                    }
                    Some(_) => {}
                    None => {
                        self.pending_usage_operations.remove(&key);
                        self.usage_operations.insert(key, usage);
                    }
                }
            }
            Event::CompactionStarted {
                run_id,
                compaction_id,
                source_through_seq,
            } => {
                self.require_active(run_id)?;
                if self.open_compaction.is_some() {
                    return Err(EventError::ConcurrentCompaction);
                }
                self.open_compaction = Some((compaction_id.clone(), *source_through_seq));
            }
            Event::CompactionFinished {
                run_id,
                compaction_id,
                ..
            } => {
                self.require_active(run_id)?;
                if self.open_compaction.as_ref().map(|value| value.0.as_str())
                    != Some(compaction_id.as_str())
                {
                    return Err(EventError::CompactionMismatch);
                }
                self.open_compaction = None;
            }
            Event::SummaryReplaced { summary, .. } => self.summary = Some(summary.clone()),
            Event::Opaque {
                event_type,
                ignorable: false,
                ..
            } => return Err(EventError::UnknownRequired(event_type.clone())),
            _ => {}
        }
        self.last_seq = envelope.seq;
        Ok(())
    }

    fn require_active(&self, run_id: &RunId) -> Result<(), EventError> {
        if self.active_run_id.as_ref() == Some(run_id) {
            Ok(())
        } else {
            Err(EventError::RunMismatch)
        }
    }

    /// `(prompt, completion)` tokens settled for `run_id`.
    pub fn usage_for(&self, run_id: &str) -> (u64, u64) {
        self.usage_operations
            .iter()
            .filter(|((recorded_run_id, _), _)| recorded_run_id.as_str() == run_id)
            .fold((0, 0), |total, (_, usage)| {
                (total.0 + usage.0, total.1 + usage.1)
            })
    }

    /// Settled plus reserved-but-unsettled units for `run_id`, for conservative billing.
    pub fn billable_units_for(&self, run_id: &str) -> u64 {
        let recorded = self
            .usage_operations
            .iter()
            .filter(|((recorded_run_id, _), _)| recorded_run_id.as_str() == run_id)
            .map(|(_, usage)| usage.0 + usage.1 + usage.2)
            .sum::<u64>();
        recorded
            + self
                .pending_usage_operations
                .iter()
                .filter(|((recorded_run_id, _), _)| recorded_run_id.as_str() == run_id)
                .map(|(_, usage)| usage.0 + usage.1 + usage.2)
                .sum::<u64>()
    }
}

/// Invariant violated while applying an event.
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum EventError {
    /// Metadata syntax or version does not match the prior state.
    #[error("invalid Session metadata or metadata version conflict")]
    InvalidSessionMetadata,
    /// Metering attribution is malformed.
    #[error("invalid metering attribution")]
    InvalidMetering,
    /// New input requires an unarchived Session.
    #[error("Session is archived")]
    SessionArchived,
    /// Unsupported session event format version.
    #[error("unsupported session event format version {0}")]
    UnsupportedFormat(u32),
    /// Unknown required session event type.
    #[error("unknown required session event type {0}")]
    UnknownRequired(String),
    /// Event sequence mismatch: expected `expected`, got `actual`.
    #[error("event sequence mismatch: expected {expected}, got {actual}")]
    Sequence {
        /// Sequence the projection expected next.
        expected: u64,
        /// Sequence the event carried.
        actual: u64,
    },
    /// Event belongs to another session.
    #[error("event belongs to another session")]
    SessionMismatch,
    /// Session creation must be the first and only creation event.
    #[error("session creation must be the first and only creation event")]
    DuplicateSession,
    /// Session already has an active run.
    #[error("session already has an active run")]
    ConcurrentRun,
    /// Session is closed.
    #[error("session is closed")]
    SessionClosed,
    /// Event does not match the active run.
    #[error("event does not match the active run")]
    RunMismatch,
    /// Interaction does not match the waiting run.
    #[error("interaction does not match the waiting run")]
    InteractionMismatch,
    /// Run cannot finish with an open turn, step or tool call.
    #[error("run cannot finish with an open turn, step or tool call")]
    OpenLifecycle,
    /// Input was queued twice.
    #[error("input was queued twice: {0}")]
    DuplicateInput(InputId),
    /// Input was claimed before it was queued.
    #[error("input was claimed before it was queued: {0}")]
    UnqueuedInput(InputId),
    /// Run started from an unclaimed input.
    #[error("run started from an unclaimed input: {0}")]
    UnclaimedInput(InputId),
    /// Session already has an active turn.
    #[error("session already has an active turn")]
    ConcurrentTurn,
    /// Turn already has this active step.
    #[error("turn already has this active step")]
    ConcurrentStep,
    /// Turn or step lifecycle does not pair.
    #[error("turn or step lifecycle does not pair")]
    LifecycleMismatch,
    /// Duplicate tool call.
    #[error("duplicate tool call {0}")]
    DuplicateToolCall(ToolCallId),
    /// Tool result has no matching call.
    #[error("tool result has no matching call {0}")]
    OrphanToolResult(ToolCallId),
    /// Tool result does not match the call run and step.
    #[error("tool result does not match the call run and step: {0}")]
    ToolResultMismatch(ToolCallId),
    /// Usage operation was recorded with different totals.
    #[error("usage operation was recorded with different totals: {0}")]
    UsageConflict(String),
    /// Session already has an active compaction.
    #[error("session already has an active compaction")]
    ConcurrentCompaction,
    /// Compaction lifecycle does not pair.
    #[error("compaction lifecycle does not pair")]
    CompactionMismatch,
    /// Event store conflict.
    #[error("event store conflict: {0}")]
    Conflict(String),
    /// Event store unavailable.
    #[error("event store unavailable: {0}")]
    Unavailable(String),
}

/// Append-only Session event storage.
#[async_trait]
pub trait SessionEventStore: Send + Sync {
    /// Append events atomically after `expected_seq`; returns them with sequences.
    async fn append(
        &self,
        tenant_id: &str,
        session_id: &str,
        expected_seq: u64,
        events: Vec<Event>,
    ) -> Result<Vec<SessionEvent>, EventError>;
    /// Events after `after_seq`.
    async fn load(
        &self,
        tenant_id: &str,
        session_id: &str,
        after_seq: u64,
    ) -> Result<Vec<SessionEvent>, EventError>;
}

/// One text content block.
pub fn text(value: impl Into<String>) -> Vec<ContentBlock> {
    vec![ContentBlock::Text { text: value.into() }]
}

/// Events that close an interrupted Run as failed with `worker_restarted`, so recovery starts from a clean lifecycle.
pub fn recovery_events(projection: &SessionProjection) -> Vec<Event> {
    failure_events(projection, "worker_restarted")
}

/// Events that close an interrupted Run as failed with `error_code`.
pub fn failure_events(projection: &SessionProjection, error_code: &str) -> Vec<Event> {
    termination_events(projection, RunStatus::Failed, error_code)
}

/// Events that close an interrupted Run as cancelled.
pub fn cancel_events(projection: &SessionProjection) -> Vec<Event> {
    termination_events(projection, RunStatus::Cancelled, "cancelled")
}

/// Events that tombstone a Session and close its Runs.
pub fn session_deletion_events(projection: &SessionProjection, reason: &str) -> Vec<Event> {
    let active = projection.active_run_id.as_deref();
    let mut events = cancel_events(projection);
    events.extend(
        projection
            .queued_inputs
            .iter()
            .filter(|(_, (run_id, _))| Some(run_id.as_str()) != active)
            .map(|(input_id, (run_id, _))| Event::InputCancelled {
                input_id: input_id.clone(),
                run_id: run_id.clone(),
                error_code: "cancelled".into(),
            }),
    );
    events.push(Event::SessionDeleted {
        reason: reason.into(),
    });
    events
}

fn termination_events(
    projection: &SessionProjection,
    status: RunStatus,
    error_code: &str,
) -> Vec<Event> {
    let Some(run_id) = &projection.active_run_id else {
        return Vec::new();
    };
    let mut events = projection
        .queued_inputs
        .iter()
        .filter(|(_, (target_run_id, _))| target_run_id == run_id)
        .map(|(input_id, _)| Event::InputCancelled {
            input_id: input_id.clone(),
            run_id: run_id.clone(),
            error_code: error_code.into(),
        })
        .collect::<Vec<_>>();
    events.extend(
        projection
            .open_tool_calls
            .iter()
            .map(|(call_id, call)| Event::ToolResult {
                run_id: run_id.clone(),
                step: call.step,
                call_id: call_id.clone(),
                result: termination_result(error_code),
                is_error: true,
            }),
    );
    if let Some((compaction_id, _)) = &projection.open_compaction {
        events.push(Event::CompactionFinished {
            run_id: run_id.clone(),
            compaction_id: compaction_id.clone(),
            status: "failed".into(),
            error: Some(error_code.into()),
        });
    }
    events.extend(
        projection
            .open_steps
            .iter()
            .map(|step| Event::StepFinished {
                run_id: run_id.clone(),
                step: *step,
            }),
    );
    if let Some(turn) = projection.open_turn {
        events.push(Event::TurnFinished {
            run_id: run_id.clone(),
            turn,
        });
    }
    events.push(Event::RunFinished {
        run_id: run_id.clone(),
        status,
        error_code: Some(error_code.into()),
    });
    events
}

fn termination_result(error_code: &str) -> Value {
    if error_code == "tool_outcome_unknown" || error_code == "worker_restarted" {
        serde_json::json!({
            "error": error_code,
            "guidance": "The tool outcome is unknown. Verify external state before retrying any operation with side effects; ask the user when verification is unavailable."
        })
    } else {
        serde_json::json!({"error":error_code})
    }
}

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

    fn event(seq: u64, event: Event) -> SessionEvent {
        SessionEvent {
            session_id: "s".parse().unwrap(),
            seq,
            occurred_at: Utc::now(),
            event,
        }
    }

    #[test]
    fn replay_enforces_single_run_and_tool_pairs() {
        let events = vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("hi"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputClaimed {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                },
            ),
            event(
                4,
                Event::RunStarted {
                    run_id: "r1".parse().unwrap(),
                    input_id: "i1".parse().unwrap(),
                },
            ),
            event(
                5,
                Event::TurnStarted {
                    run_id: "r1".parse().unwrap(),
                    turn: 1,
                },
            ),
            event(
                6,
                Event::StepStarted {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                },
            ),
            event(
                7,
                Event::ToolCall {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                    call_id: "c1".parse().unwrap(),
                    tool: "echo".into(),
                    arguments: serde_json::json!({"x":1}),
                },
            ),
            event(
                8,
                Event::ToolResult {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                    call_id: "c1".parse().unwrap(),
                    result: serde_json::json!({"x":1}),
                    is_error: false,
                },
            ),
            event(
                9,
                Event::StepFinished {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                },
            ),
            event(
                10,
                Event::TurnFinished {
                    run_id: "r1".parse().unwrap(),
                    turn: 1,
                },
            ),
            event(
                11,
                Event::RunFinished {
                    run_id: "r1".parse().unwrap(),
                    status: RunStatus::Completed,
                    error_code: None,
                },
            ),
        ];
        let projection = SessionProjection::replay(&events).unwrap();
        assert_eq!(projection.last_seq, 11);
        assert!(projection.active_run_id.is_none());
    }

    #[test]
    fn replay_rejects_orphan_tool_result() {
        let events = vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("hi"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputClaimed {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                },
            ),
            event(
                4,
                Event::RunStarted {
                    run_id: "r1".parse().unwrap(),
                    input_id: "i1".parse().unwrap(),
                },
            ),
            event(
                5,
                Event::TurnStarted {
                    run_id: "r1".parse().unwrap(),
                    turn: 1,
                },
            ),
            event(
                6,
                Event::StepStarted {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                },
            ),
            event(
                7,
                Event::ToolResult {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                    call_id: "missing".parse().unwrap(),
                    result: Value::Null,
                    is_error: true,
                },
            ),
        ];
        assert_eq!(
            SessionProjection::replay(&events).unwrap_err(),
            EventError::OrphanToolResult("missing".parse().unwrap())
        );
    }

    #[test]
    fn queued_input_failure_is_not_projected_as_cancellation() {
        let events = vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("hi"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputCancelled {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    error_code: "profile_not_found".into(),
                },
            ),
        ];
        let projection = SessionProjection::replay(&events).unwrap();
        assert_eq!(
            projection.run_status.get("r1").map(|state| state.as_str()),
            Some("failed")
        );
    }

    #[test]
    fn session_deletion_closes_active_and_queued_runs_before_tombstone() {
        let mut events = vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("start"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputClaimed {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                },
            ),
            event(
                4,
                Event::RunStarted {
                    run_id: "r1".parse().unwrap(),
                    input_id: "i1".parse().unwrap(),
                },
            ),
            event(
                5,
                Event::TurnStarted {
                    run_id: "r1".parse().unwrap(),
                    turn: 1,
                },
            ),
            event(
                6,
                Event::InputQueued {
                    input_id: "i2".parse().unwrap(),
                    run_id: "r2".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("later"),
                    explicit_skill: None,
                },
            ),
        ];
        let projection = SessionProjection::replay(&events).unwrap();
        for event_value in session_deletion_events(&projection, "api_deleted") {
            let seq = events.len() as u64 + 1;
            events.push(event(seq, event_value));
        }
        let deleted = SessionProjection::replay(&events).unwrap();
        assert!(deleted.deleted);
        assert_eq!(
            deleted.run_status.get("r1").map(|state| state.as_str()),
            Some("cancelled")
        );
        assert_eq!(
            deleted.run_status.get("r2").map(|state| state.as_str()),
            Some("cancelled")
        );
        assert!(events.iter().any(|event| matches!(
            &event.event,
            Event::RunFinished { run_id, status: RunStatus::Cancelled, .. } if run_id == "r1"
        )));
        assert!(matches!(
            events.last().map(|event| &event.event),
            Some(Event::SessionDeleted { .. })
        ));
    }

    #[test]
    fn usage_operations_are_idempotent_and_conflicts_fail_replay() {
        let mut events = vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("hi"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputClaimed {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                },
            ),
            event(
                4,
                Event::RunStarted {
                    run_id: "r1".parse().unwrap(),
                    input_id: "i1".parse().unwrap(),
                },
            ),
            event(
                5,
                Event::UsageRecorded {
                    metering: None,
                    run_id: "r1".parse().unwrap(),
                    operation_id: "model:1:attempt:1".into(),
                    prompt_tokens: 7,
                    completion_tokens: 3,
                    cost_units: 5,
                },
            ),
            event(
                6,
                Event::UsageRecorded {
                    metering: None,
                    run_id: "r1".parse().unwrap(),
                    operation_id: "model:1:attempt:1".into(),
                    prompt_tokens: 7,
                    completion_tokens: 3,
                    cost_units: 5,
                },
            ),
        ];
        assert_eq!(
            SessionProjection::replay(&events).unwrap().usage_for("r1"),
            (7, 3)
        );
        assert_eq!(
            SessionProjection::replay(&events)
                .unwrap()
                .billable_units_for("r1"),
            15
        );

        let mut attributed = events.clone();
        let details = MeteringDetails::Model {
            model: "m".into(),
            provider: Some("p".into()),
            provider_attempt_id: "r1:model:1:attempt:1".parse().unwrap(),
            source: MeteringSource::Estimated,
            outcome: MeteringOutcome::Unknown,
        };
        for entry in &mut attributed[4..] {
            if let Event::UsageRecorded { metering, .. } = &mut entry.event {
                *metering = Some(details.clone());
            }
        }
        assert_eq!(
            SessionProjection::replay(&attributed)
                .unwrap()
                .billable_units_for("r1"),
            15
        );
        if let Event::UsageRecorded {
            metering: Some(MeteringDetails::Model { source, .. }),
            ..
        } = &mut attributed[5].event
        {
            *source = MeteringSource::Reported;
        }
        assert!(matches!(
            SessionProjection::replay(&attributed),
            Err(EventError::UsageConflict(_))
        ));
        let mut prepared = attributed[..4].to_vec();
        prepared.push(event(
            5,
            Event::ModelRequestPrepared {
                metering: Some(details.clone()),
                run_id: "r1".parse().unwrap(),
                step: 1,
                attempt: 1,
                provider_attempt_id: "r1:model:1:attempt:1".into(),
                operation_id: "model:1:attempt:1".into(),
                reserved_prompt_tokens: 10,
                reserved_completion_tokens: 20,
                request: Value::Null,
                prompt_sections: Value::Null,
            },
        ));
        prepared.push(attributed[5].clone());
        assert_eq!(
            SessionProjection::replay(&prepared)
                .unwrap()
                .billable_units_for("r1"),
            15
        );
        if let Event::UsageRecorded {
            metering: Some(MeteringDetails::Model { model, .. }),
            ..
        } = &mut prepared[5].event
        {
            *model = "wrong".into();
        }
        assert!(matches!(
            SessionProjection::replay(&prepared),
            Err(EventError::UsageConflict(_))
        ));
        if let Event::UsageRecorded { metering, .. } = &mut prepared[5].event {
            *metering = None;
        }
        assert!(matches!(
            SessionProjection::replay(&prepared),
            Err(EventError::UsageConflict(_))
        ));
        let serialized = serde_json::to_value(&events[4]).unwrap();
        assert!(serialized["event"].get("metering").is_none());
        assert_eq!(
            serde_json::from_value::<SessionEvent>(serialized).unwrap(),
            events[4]
        );

        events.push(event(
            7,
            Event::UsageRecorded {
                metering: None,
                run_id: "r1".parse().unwrap(),
                operation_id: "model:1:attempt:1".into(),
                prompt_tokens: 8,
                completion_tokens: 3,
                cost_units: 0,
            },
        ));
        assert_eq!(
            SessionProjection::replay(&events).unwrap_err(),
            EventError::UsageConflict("model:1:attempt:1".into())
        );
    }

    #[test]
    fn unresolved_provider_attempt_is_conservatively_billable_and_reconcilable() {
        let mut events = vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("hi"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputClaimed {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                },
            ),
            event(
                4,
                Event::RunStarted {
                    run_id: "r1".parse().unwrap(),
                    input_id: "i1".parse().unwrap(),
                },
            ),
            event(
                5,
                Event::ModelRequestPrepared {
                    metering: None,
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                    attempt: 1,
                    provider_attempt_id: "r1:model:1:attempt:1".into(),
                    operation_id: "model:1:attempt:1".into(),
                    reserved_prompt_tokens: 7,
                    reserved_completion_tokens: 11,
                    request: Value::Null,
                    prompt_sections: Value::Null,
                },
            ),
        ];
        assert_eq!(
            SessionProjection::replay(&events)
                .unwrap()
                .billable_units_for("r1"),
            18
        );
        events.push(event(
            6,
            Event::UsageRecorded {
                metering: None,
                run_id: "r1".parse().unwrap(),
                operation_id: "model:1:attempt:1".into(),
                prompt_tokens: 6,
                completion_tokens: 2,
                cost_units: 0,
            },
        ));
        assert_eq!(
            SessionProjection::replay(&events)
                .unwrap()
                .billable_units_for("r1"),
            8
        );
    }

    #[test]
    fn a3_incremental_facts_preserve_every_lifecycle_and_usage_invariant() {
        let mut events = open_step_events();
        for fact in [
            Event::UserMessage {
                run_id: "r1".parse().unwrap(),
                content: text("history".repeat(10_000)),
            },
            Event::UsageRecorded {
                metering: None,
                run_id: "r1".parse().unwrap(),
                operation_id: "attempt".into(),
                prompt_tokens: 5,
                completion_tokens: 3,
                cost_units: 0,
            },
            Event::ToolCall {
                run_id: "r1".parse().unwrap(),
                step: 1,
                call_id: "pending".parse().unwrap(),
                tool: "echo".into(),
                arguments: Value::Null,
            },
        ] {
            events.push(event(events.len() as u64 + 1, fact));
        }
        let mut full = SessionProjection::replay(&events).unwrap();
        assert!(!full.messages.is_empty());
        full.messages.clear();
        full.injected_context.clear();
        let mut folded = SessionProjection::default();
        for page in events.chunks(3) {
            for fact in page {
                folded.apply_facts(fact).unwrap();
            }
            assert!(folded.messages.is_empty());
        }
        assert_eq!(folded, full);
        assert_eq!(SessionProjection::replay_facts(&events).unwrap(), full);
        assert_eq!(cancel_events(&folded), cancel_events(&full));
        assert_eq!(folded.usage_for("r1"), (5, 3));
        let invalid = event(
            events.len() as u64 + 1,
            Event::ToolResult {
                run_id: "r1".parse().unwrap(),
                step: 1,
                call_id: "missing".parse().unwrap(),
                result: Value::Null,
                is_error: true,
            },
        );
        assert_eq!(folded.apply_facts(&invalid), full.apply(&invalid));
    }

    fn open_step_events() -> Vec<SessionEvent> {
        vec![
            event(
                1,
                Event::SessionCreated {
                    profile_revision_id: "p1".parse().unwrap(),
                },
            ),
            event(
                2,
                Event::InputQueued {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                    mode: DeliveryMode::Followup,
                    content: text("hi"),
                    explicit_skill: None,
                },
            ),
            event(
                3,
                Event::InputClaimed {
                    input_id: "i1".parse().unwrap(),
                    run_id: "r1".parse().unwrap(),
                },
            ),
            event(
                4,
                Event::RunStarted {
                    run_id: "r1".parse().unwrap(),
                    input_id: "i1".parse().unwrap(),
                },
            ),
            event(
                5,
                Event::TurnStarted {
                    run_id: "r1".parse().unwrap(),
                    turn: 1,
                },
            ),
            event(
                6,
                Event::StepStarted {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                },
            ),
        ]
    }

    #[test]
    fn replay_rejects_overlapping_or_out_of_order_steps() {
        let mut overlapping = open_step_events();
        overlapping.push(event(
            7,
            Event::StepStarted {
                run_id: "r1".parse().unwrap(),
                step: 2,
            },
        ));
        assert_eq!(
            SessionProjection::replay(&overlapping).unwrap_err(),
            EventError::ConcurrentStep
        );

        let mut skipped = open_step_events();
        skipped[5] = event(
            6,
            Event::StepStarted {
                run_id: "r1".parse().unwrap(),
                step: 2,
            },
        );
        assert_eq!(
            SessionProjection::replay(&skipped).unwrap_err(),
            EventError::ConcurrentStep
        );
    }

    #[test]
    fn replay_rejects_cross_step_results_and_dangling_calls() {
        let mut events = open_step_events();
        events.push(event(
            7,
            Event::ToolCall {
                run_id: "r1".parse().unwrap(),
                step: 1,
                call_id: "c1".parse().unwrap(),
                tool: "echo".into(),
                arguments: Value::Null,
            },
        ));
        events.push(event(
            8,
            Event::ToolResult {
                run_id: "r1".parse().unwrap(),
                step: 2,
                call_id: "c1".parse().unwrap(),
                result: Value::Null,
                is_error: false,
            },
        ));
        assert_eq!(
            SessionProjection::replay(&events).unwrap_err(),
            EventError::LifecycleMismatch
        );

        let mut dangling = open_step_events();
        dangling.push(event(
            7,
            Event::ToolCall {
                run_id: "r1".parse().unwrap(),
                step: 1,
                call_id: "c1".parse().unwrap(),
                tool: "echo".into(),
                arguments: Value::Null,
            },
        ));
        dangling.push(event(
            8,
            Event::StepFinished {
                run_id: "r1".parse().unwrap(),
                step: 1,
            },
        ));
        assert_eq!(
            SessionProjection::replay(&dangling).unwrap_err(),
            EventError::LifecycleMismatch
        );
    }

    #[test]
    fn replay_rejects_reused_tool_call_ids() {
        let mut events = open_step_events();
        events.extend([
            event(
                7,
                Event::ToolCall {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                    call_id: "c1".parse().unwrap(),
                    tool: "echo".into(),
                    arguments: Value::Null,
                },
            ),
            event(
                8,
                Event::ToolResult {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                    call_id: "c1".parse().unwrap(),
                    result: Value::Null,
                    is_error: false,
                },
            ),
            event(
                9,
                Event::StepFinished {
                    run_id: "r1".parse().unwrap(),
                    step: 1,
                },
            ),
            event(
                10,
                Event::StepStarted {
                    run_id: "r1".parse().unwrap(),
                    step: 2,
                },
            ),
            event(
                11,
                Event::ToolCall {
                    run_id: "r1".parse().unwrap(),
                    step: 2,
                    call_id: "c1".parse().unwrap(),
                    tool: "echo".into(),
                    arguments: Value::Null,
                },
            ),
        ]);
        assert_eq!(
            SessionProjection::replay(&events).unwrap_err(),
            EventError::DuplicateToolCall("c1".parse().unwrap())
        );
    }

    #[test]
    fn replay_keeps_injected_context_without_changing_run_state() {
        let mut events = open_step_events();
        events.push(event(
            7,
            Event::ContextInjected {
                run_id: "r1".parse().unwrap(),
                step: 1,
                contribution_id: "memory:1".into(),
                source: "memory".into(),
                version: "v1".into(),
                authority: "tenant".into(),
                form: "message".into(),
                content: text("tenant context"),
            },
        ));

        let projection = SessionProjection::replay(&events).unwrap();
        assert_eq!(projection.active_run_id.as_deref(), Some("r1"));
        assert_eq!(projection.open_steps, BTreeSet::from([1]));
        assert_eq!(
            projection.injected_context,
            vec![ProjectedContext {
                run_id: "r1".parse().unwrap(),
                step: 1,
                contribution_id: "memory:1".into(),
                source: "memory".into(),
                version: "v1".into(),
                authority: "tenant".into(),
                form: "message".into(),
                content: text("tenant context"),
            }]
        );
    }

    #[test]
    fn replay_skips_unknown_ignorable_formats_and_rejects_required_events() {
        let optional = event(
            1,
            Event::Opaque {
                format_version: SESSION_EVENT_FORMAT_VERSION,
                event_type: "future_optional".into(),
                ignorable: true,
                payload: serde_json::json!({"answer": 42}),
            },
        );
        let projection = SessionProjection::replay(&[optional]).unwrap();
        assert_eq!(projection.last_seq, 1);

        let required = event(
            1,
            Event::Opaque {
                format_version: SESSION_EVENT_FORMAT_VERSION,
                event_type: "future_required".into(),
                ignorable: false,
                payload: Value::Null,
            },
        );
        assert_eq!(
            SessionProjection::replay(&[required]).unwrap_err(),
            EventError::UnknownRequired("future_required".into())
        );

        let unsupported = event(
            1,
            Event::Opaque {
                format_version: SESSION_EVENT_FORMAT_VERSION + 1,
                event_type: "future_optional".into(),
                ignorable: true,
                payload: Value::Null,
            },
        );
        assert_eq!(
            SessionProjection::replay(&[unsupported]).unwrap().last_seq,
            1
        );

        let required_new_format = event(
            1,
            Event::Opaque {
                format_version: SESSION_EVENT_FORMAT_VERSION + 1,
                event_type: "future_required".into(),
                ignorable: false,
                payload: Value::Null,
            },
        );
        assert_eq!(
            SessionProjection::replay(&[required_new_format]).unwrap_err(),
            EventError::UnsupportedFormat(SESSION_EVENT_FORMAT_VERSION + 1)
        );
    }
}