//! # Module: cycle_state
//!
//! ## Spec
//! - Persists per-document turn intent as typed checkpoints in the project
//!   Lazily state ledger (`state.db`).
//! - Tracks the exact phase of the current or most recent response cycle:
//!   `preflight_started` → `response_captured` → `write_applied` → `committed`.
//!   A stale `preflight_started` cycle with no response artifact may become
//!   `abandoned` so a later preflight can start a fresh cycle for the same
//!   unresolved prompt.
//! - Stores cycle-scoped snapshot/file content hashes so callers can reason
//!   about exact cycle state instead of inferring from file-size drift or only
//!   the last `ops.log` line.
//! - Stores the preflight baseline reference, prompt targets, queue head identity,
//!   pending-operation facts, and response capture hash as one durable turn
//!   checkpoint so restart/recycle paths can resume or fail closed from the ledger.
//! - `start_preflight()` opens a new cycle for a document and overwrites any
//!   prior committed state for that document.
//! - `mark_response_captured()` advances the open cycle to `response_captured`
//!   once the final parsed response has been durably stored.
//! - `mark_write_applied()` advances the open cycle to `write_applied` (or
//!   creates a synthetic cycle if a write lands without a prior preflight).
//! - `mark_committed()` advances the cycle to `committed` (or creates a
//!   synthetic committed cycle if commit happens without a closable open state,
//!   including after a stale preflight was abandoned).
//! - Lower-rank bookkeeping must never mutate an already-committed or abandoned
//!   cycle; duplicate terminal bookkeeping stays idempotent for already-committed
//!   cycles.
//! - Phase transitions are accepted through `cycle_state_machine`; accepted
//!   transitions append both the complete turn-intent checkpoint and typed
//!   closeout facts into the state backbone.
//! - `load()` returns only complete state-backbone turn checkpoints. It never
//!   imports or synthesizes lifecycle state from a secondary store.
//! - `load_closeout_projection()` returns the state-backbone closeout projection
//!   when lazily facts have been recorded for the document.
//!
//! ## Agentic Contracts
//! - State is per-document, never global across the repo.
//! - Writes are deterministic, idempotently keyed state-ledger events.
//! - Missing project root or state-ledger facts returns `Ok(None)`.
//! - `is_open()` is true for any phase except `Committed` or `Abandoned`.
//!
//! ## Evals
//! - `start_preflight_persists_open_cycle`
//! - `mark_response_captured_sets_capture_metadata`
//! - `mark_write_applied_advances_existing_cycle`
//! - `mark_committed_closes_cycle`
//! - `mark_write_applied_creates_synthetic_cycle_when_missing`

use agent_doc_document::transient_markers::replay_content_hash;
use agent_doc_element_backlog::backlog::normalize_pending_id;
use agent_doc_turn::cycle_policy::{
    is_noop_commit_event, is_stable_commit_event, normalize_checkpoint_task_id,
    normalize_checkpoint_text_list,
};
use agent_doc_turn::{CycleEvent, CyclePhase, CyclePhaseMachine};
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::path::Path;

pub mod command_plane;
pub mod pipeline_frontmatter;

/// Mark whether the current thread is handling a project-controller request
/// in-process. Set `true` for the lifetime of the controller serve loop.
pub fn set_in_controller_request(value: bool) {
    agent_doc_state_wire::set_in_controller_request(value);
}

pub(crate) fn in_controller_request() -> bool {
    agent_doc_state_wire::in_controller_request()
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct BacklogTargetRequirement {
    pub path: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub component: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub baseline_hash: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub baseline_item_ids: Vec<String>,
}

/// Node-keyed reactive advisory projected when semantic merge cannot apply the
/// agent's change verbatim and the concurrent operator value wins.
///
/// The advisory belongs to the cycle that observed the conflict. It is state
/// for diagnostics and editor presentation, never a request for a future turn
/// and never carried into the next preflight.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SemanticMergeConflictAdvisory {
    pub component: String,
    pub id: String,
    pub reason: String,
    pub detail: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recorded_cycle_id: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct BlockedCloseout {
    pub kind: String,
    pub reason: String,
    pub source: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub patch_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recovery: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recovery_command: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CycleState {
    pub cycle_id: String,
    pub file: String,
    pub phase: CyclePhase,
    pub last_event: String,
    pub started_at: u64,
    pub updated_at: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub snapshot_hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub file_hash: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub normalized_snapshot_hash: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub normalized_file_hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub capture_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub response_sha256: Option<String>,
    #[serde(default)]
    pub had_pending_mutations: bool,
    #[serde(default)]
    pub requires_backlog_capture: bool,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub required_backlog_targets: Vec<BacklogTargetRequirement>,
    #[serde(default, skip_serializing_if = "is_zero")]
    pub required_explicit_backlog_item_count: usize,
    #[serde(default, skip_serializing_if = "is_zero")]
    pub required_plan_reference_count: usize,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub prompt_targets: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub queue_task_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub turn_id: Option<String>,
    /// `#midturn-recycle-resume` Phase B: latched true once a fresh supervisor boot
    /// has re-dispatched this exact open cycle's interrupted turn from the
    /// `#durablerecycle` checkpoint (the harness child died across the `execve`
    /// recycle). The idempotency guard: a SECOND boot reading the same still-open
    /// checkpoint must not re-dispatch the turn again (it would double-run it). A
    /// committed cycle is never open so it never re-dispatches regardless of this
    /// flag; this flag covers the rarer case where two boots both observe the same
    /// open-but-child-dead checkpoint before the re-dispatched turn commits.
    #[serde(default, skip_serializing_if = "is_false")]
    pub recycle_resume_consumed: bool,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_done_ids: Vec<String>,
    /// A tracked-work-only `write --commit` whose document projection was
    /// retained before the git effect could run. The target hash is the
    /// idempotency fence for resuming that exact commit after editor/disk
    /// convergence; absence means no pending-only commit is authorized.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pending_only_commit_target_hash: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_kept_open_ids: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub reaped_pending_ids: Vec<String>,
    /// `#do-id-closeout-open-backlog`: tracked-work ids named by an explicit
    /// `do [#id]` / `do #id` prompt directive that were still open in the live
    /// `agent:backlog` at preflight time. A successful closeout must end each of
    /// these with an explicit lifecycle outcome (`--done`, `--pending-gate`, an
    /// explicit kept-open edit, or reap); otherwise `session-check` fails closed
    /// so a directive cannot clear the queue while leaving its target `[ ]`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub expect_done_or_gate_ids: Vec<String>,
    /// `#blocked-closeout-followup-capture`: tracked ids moved to the
    /// review/gated component this cycle via `--pending-gate`. A gate removes an
    /// id from active `agent:backlog`, so when the response also signals the
    /// work is blocked / still needs future action, `session-check` requires a
    /// captured follow-up (kept-open edit, new backlog item, or explicit
    /// no-follow-up justification) before clean closeout.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_gated_ids: Vec<String>,
    /// `#blocked-closeout-followup-capture`: true when this cycle added at least
    /// one follow-up item via any `--pending-add*` primitive (active backlog or
    /// gated). Satisfies the blocked-closeout follow-up requirement.
    #[serde(default)]
    pub pending_added_this_cycle: bool,
    /// `#opsproof-samecycle-add`: ids of tracked-work items added THIS cycle via
    /// `--pending-add` / `--pending-add-gated` / `--review-add`. Opportunistic
    /// ops-proof auto-completion must never reap an id that first appeared this
    /// cycle — the on-disk snapshot it compares against is updated by the same
    /// write invocation, so the snapshot baseline alone cannot distinguish a
    /// brand-new same-cycle add from a pre-existing item.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_added_ids: Vec<String>,
    /// `#mutprovenancepreresponse`: tracked-work ids this cycle REQUESTED,
    /// recorded before the response cell is published.
    ///
    /// `pending_done_ids` / `pending_added_ids` are written by the mutation
    /// phase, which runs AFTER the response write. When that write fails once
    /// the response has already landed, the mutation phase never runs, nothing
    /// is recorded, and the closeout-provenance predicate correctly reports "no
    /// unlanded mutations of mine" — so the divergence is misclassified as a
    /// fresh operator edit and swept. Intent recorded up front survives a
    /// failure at any position.
    ///
    /// Deliberately separate from `pending_done_ids` rather than recorded early
    /// into it: closeout-signal logic reads that field as "this cycle recorded a
    /// completion outcome", and a failed cycle claiming its queue head was
    /// answered would strike a head that never got done.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub requested_done_ids: Vec<String>,
    /// Explicit ids this cycle requested to ADD (`#mutprovenancepreresponse`).
    ///
    /// Only ids the caller named (`id=<id>` / `[#id]`) are knowable before the
    /// add runs; generated ids are not, and stay covered by the post-hoc
    /// `pending_added_ids` record.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub requested_added_ids: Vec<String>,
    /// `#backlogqueuepopulation`: tracked-work ids that became actionable this
    /// cycle and therefore may need insert-only mirroring into an explicit
    /// go-mode `agent:queue`. Adds and ungates record here; gates, done
    /// transitions, and unrelated operator queue deletions do not.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_actionable_ids: Vec<String>,
    /// `#queuemirrororder`: the subset of `pending_actionable_ids` filed with an
    /// EXPLICIT backlog anchor (`--backlog-add-after` / `--backlog-add-before`).
    ///
    /// Position cannot be inferred after the fact: a `--backlog-add-back` item and
    /// an anchored item both end up below an already-queued id, and only the
    /// anchored one asked for that placement. Recording the intent at the mutation
    /// site keeps the queue mirror following the anchor while leaving the
    /// `#queueatcreate` head default intact for everything else.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_anchored_ids: Vec<String>,
    /// True when the preflight-current document already had completed tracked
    /// work that closeout pending maintenance should reap. This lets closeout
    /// distinguish status-only cycles from tracked-work cycles without reading
    /// a non-authoritative disk replica when editor authority is unavailable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tracked_work_maintenance_required_at_preflight: Option<bool>,
    #[serde(default)]
    pub ipc_snapshot_adoption_blocked: bool,
    /// `#exchange-prompt-dropped-on-merge`: user-authored exchange prompt lines
    /// that were dropped when `content_ours` was adopted over a divergent IPC
    /// candidate (live prompt drift after preflight). Recorded at adoption time
    /// so `session-check` can fail closed on the data-loss class even if the
    /// editor later overwrites the disk prompt via IPC buffer convergence
    /// (the silent-loss race the post-commit disk diff cannot win).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub dropped_exchange_prompts: Vec<String>,
    /// `#queue-user-edit-overwrite`: user-authored `agent:queue` prompt lines
    /// (e.g. `- do [#gscaccess]`) that were dropped when `content_ours` was
    /// adopted over a divergent IPC candidate. Recorded at adoption time so
    /// `session-check` can fail closed if a user queue edit was silently
    /// deleted by write/reset/commit convergence instead of being consumed by
    /// the current response.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub dropped_queue_prompts: Vec<String>,
    /// `#queue-clear-unrun-items`: `do [#id]` queue head prompt texts present in
    /// the visible `agent:queue` at preflight time. An active queue head is
    /// executable user intent, so a closeout / reset / commit may delete one only
    /// with proof that it was consumed (this cycle's directive target), resolved
    /// (its `#id` left `agent:backlog` via done/gate/reap), or removed by an
    /// explicit user edit. `session-check` fails closed when a recorded head
    /// disappears from the committed queue while its `#id` is still open in
    /// `agent:backlog` and the cycle never targeted it.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub active_queue_heads: Vec<String>,
    /// `#lr-queue-patchback-miss`: free-text (non-`do [#id]`) queue head prompt
    /// texts present in the visible `agent:queue` at preflight time. Unlike
    /// `active_queue_heads` (which carries id-backed directive heads), these
    /// free-text heads have no backlog id to track — the guard instead checks
    /// for a committed response, binary consume, or explicit deferral proof.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub active_free_text_queue_heads: Vec<String>,
    /// Reactive semantic-merge conflicts observed during this cycle.
    #[serde(
        default,
        alias = "pending_semantic_merge_acks",
        skip_serializing_if = "Vec::is_empty"
    )]
    pub semantic_merge_conflict_advisories: Vec<SemanticMergeConflictAdvisory>,
    /// `#closeoutstall`: typed operator-gated closeout state. A response may be
    /// safely captured and queued for editor IPC while the live editor has not
    /// proven application. This keeps the cycle open and gives session-check,
    /// doctor, and hooks one canonical recovery surface.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub blocked_closeout: Option<BlockedCloseout>,
    /// `#queueskip`: id-backed queue heads the binary has SKIPPED — each was
    /// dispatched, came back unconsumed, and the queue advanced past it to a
    /// non-dependent drainable head so the loop does not wedge re-dispatching a
    /// dead ref. Accumulated across cycles (carried forward by
    /// [`start_preflight_with_task`]) and cleared for an id once it is consumed
    /// (resolved/reaped) or no longer present. Preflight stamps the visible `⏭️`
    /// skip marker on these heads and excludes them from selection.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub skipped_queue_head_ids: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AdmitOutput {
    pub admitted: bool,
    pub file: String,
    pub cycle_id: String,
    pub cycle_phase: String,
    pub last_event: String,
    pub source: String,
    pub maintenance_required: bool,
    pub preflight_required: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub snapshot_hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub file_hash: Option<String>,
}

/// `#suprecyclespin` — seconds an open cycle may sit untouched (no IPC advisory
/// connection in flight) at a harness turn boundary before the supervisor
/// recycle/restart defer path force-closes it as abandoned. Bounded so a
/// crashed/superseded older turn cannot wedge the recycle in `DeferCycleOpen`
/// (~2/sec `cycle_open` spin-loop, `idle_watch.rs`) forever.
///
/// `#suprecyclespin-falseabandon`: the earlier 45s value assumed "each phase
/// ticks `updated_at` and finalize holds IPC inflight", so a live cycle could
/// never look stale. That assumption is FALSE for a slow first response: the
/// cycle sits in `preflight_started` with an untouched `updated_at` and
/// `inflight == 0` for the ENTIRE harness generation, which for a large prompt
/// routinely exceeds 45s. Combined with a transiently-misread `turn_boundary`
/// (`prompt_visible && !turn_active`, both pane-scraped) that abandoned a live
/// turn mid-generation. The deadline is now generous enough to clear normal
/// first-response latency; the primary guard is the consecutive-tick debounce
/// below (a genuine generation does not hold `turn_boundary && stalled` across
/// many back-to-back polls) plus the `MAX_CYCLE_OPEN_DEFER_TICKS` recycle
/// escalation backstop, which force-recycles a never-closing cycle WITHOUT
/// abandoning it (the durable checkpoint survives for the fresh boot).
pub const STALLED_CYCLE_RESOLVE_SECS: u64 = 120;

/// `#suprecyclespin-falseabandon` — consecutive idle-watch polls
/// (`AUTO_TRIGGER_POLL_INTERVAL`, 500ms) for which `stalled_before_response_capture_cycle`
/// must hold at a `turn_boundary` before the supervisor force-abandons the
/// cycle. A live harness generation never sustains `turn_boundary && stalled`
/// for this long (the pane is not at a ready prompt while it streams output), so
/// the debounce rejects the transient boundary misread that abandoned a live
/// turn, while a genuinely orphaned/superseded cycle stays stalled every poll
/// and still resolves after the confirm window (~10s at 20 ticks).
pub const STALLED_CYCLE_RESOLVE_CONFIRM_TICKS: u32 = 20;

impl CycleState {
    pub fn is_open(&self) -> bool {
        !matches!(self.phase, CyclePhase::Committed | CyclePhase::Abandoned)
    }

    /// `#suprecyclespin` — whether this open cycle has stalled past
    /// `deadline_secs`: still open, no IPC advisory connection in flight, and untouched
    /// (`now_secs - updated_at > deadline_secs`). A stalled open cycle is an
    /// abandoned older turn that a newer committed cycle has superseded; the
    /// supervisor recycle-defer path force-closes such a cycle (`mark_abandoned`)
    /// so the `execve` recycle reaches its boundary instead of deferring forever.
    /// Pure so the deadline policy is unit-testable without the live idle watch.
    pub fn open_stalled(&self, inflight: u64, now_secs: u64, deadline_secs: u64) -> bool {
        self.is_open() && inflight == 0 && now_secs.saturating_sub(self.updated_at) > deadline_secs
    }

    /// `#suprecyclespin`: whether a stalled open cycle may be force-abandoned at
    /// a supervisor recycle boundary. Only cycles before response capture are disposable.
    /// Once a response has been captured or written, the open cycle is durable
    /// recovery evidence and must survive recycle so the fresh process can retry
    /// or reconcile it instead of silently losing the response.
    pub fn stalled_before_response_capture_cycle(
        &self,
        inflight: u64,
        now_secs: u64,
        deadline_secs: u64,
    ) -> bool {
        self.open_stalled(inflight, now_secs, deadline_secs)
            && matches!(self.phase, CyclePhase::PreflightStarted)
            && self.capture_id.is_none()
            && self.response_sha256.is_none()
    }

    /// `#preflightinbinary` — should a preflight invocation **reuse** this open
    /// cycle instead of minting a new one?
    ///
    /// Once preflight runs in the binary on the trigger prompt, two preflights
    /// can legitimately reach the same turn: the hook's, and an agent (or a
    /// harness without the hook) that also runs one. `start_preflight_with_task`
    /// unconditionally minted `cycle-{now_millis}`, so the second silently
    /// replaced the first — discarding its identity along with
    /// `active_queue_heads`, the carried semantic-merge acks, and the skipped-head
    /// accumulator. Nothing downstream could tell it had happened.
    ///
    /// A cycle qualifies for reuse when it is still `PreflightStarted` with no
    /// response artifact — i.e. a turn is open and nothing has been captured
    /// against it yet, so re-running preflight is a *re-entry*, not a new turn.
    /// Deliberately does not compare content hashes: preflight itself mutates the
    /// document (queue convergence, baseline checkpoint), so the hook's run and a
    /// later run in the same turn legitimately see different bytes.
    ///
    /// A **stalled** open cycle is excluded: that is a crashed or superseded
    /// older turn, and a genuinely new turn must not inherit its identity. The
    /// same `stalled_before_response_capture_cycle` policy the supervisor uses to
    /// force-abandon one decides it here, so the two cannot drift.
    pub fn reusable_by_reentrant_preflight(
        &self,
        inflight: u64,
        now_secs: u64,
        deadline_secs: u64,
    ) -> bool {
        matches!(self.phase, CyclePhase::PreflightStarted)
            && self.capture_id.is_none()
            && self.response_sha256.is_none()
            && !self.stalled_before_response_capture_cycle(inflight, now_secs, deadline_secs)
    }

    /// Derive the live finalize-pipeline view (`#fm-run-id-step` / `#fmrunid-wire`)
    /// from the authoritative cycle-state fields: `run_id` = cycle id, `step` =
    /// lowercase phase, plus the recorded `turn_id` / `queue_task_id`.
    ///
    /// This is the read-side mirror — preflight surfaces it so any invocation or
    /// editor plugin can see where the cycle is without replaying the state ledger.
    /// Cycle-state stays authoritative; the document `agent_doc_pipeline:` block
    /// is only a fallback hint when no live cycle-state exists.
    pub fn to_pipeline(&self) -> agent_doc_frontmatter::frontmatter::AgentDocPipeline {
        agent_doc_frontmatter::frontmatter::AgentDocPipeline {
            run_id: Some(self.cycle_id.clone()),
            step: Some(self.phase.as_str().to_string()),
            turn_id: self.turn_id.clone(),
            queue_task_id: self.queue_task_id.clone(),
        }
    }
}

pub fn load(file: &Path) -> Result<Option<CycleState>> {
    let Some(document) = load_document_projection(file)? else {
        return Ok(None);
    };
    reconstruct_cycle_state(&document)
}

/// Reconstruct the authoritative `CycleState` from a document projection: the
/// last `turn_intent_checkpoint` rebased onto the closeout phase projection.
/// Pure (no I/O), so it is shared by the SQLite cold-start [`load`] and by the
/// controller's live in-memory read (`#lzdurablesink` — a client transition no
/// longer replays `state.db`, the actor reconstructs from its own projection).
pub fn reconstruct_cycle_state(
    document: &agent_doc_state_backbone::DocumentStateProjection,
) -> Result<Option<CycleState>> {
    let projection = ProjectedCloseoutState::from(document.closeout.clone());

    let mut state = if let Some(checkpoint) = &document.closeout.turn_intent_checkpoint {
        let actual_sha256 = agent_doc_hash::content_hash(&checkpoint.state_json);
        anyhow::ensure!(
            actual_sha256 == checkpoint.state_sha256,
            "turn-intent checkpoint hash mismatch for cycle {}",
            checkpoint.cycle_id
        );
        let decoded: CycleState = serde_json::from_str(&checkpoint.state_json)
            .context("decode turn-intent checkpoint from state ledger")?;
        anyhow::ensure!(
            decoded.cycle_id == checkpoint.cycle_id,
            "turn-intent checkpoint cycle mismatch: event={} payload={}",
            checkpoint.cycle_id,
            decoded.cycle_id
        );
        decoded
    } else {
        return Ok(None);
    };

    if projection.matches_cycle(&state.cycle_id) {
        apply_closeout_projection_to_cycle_state(&mut state, &projection);
        state.semantic_merge_conflict_advisories =
            projection.semantic_merge_conflict_advisories.clone();
    }
    Ok(Some(state))
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectedCloseoutState {
    pub cycle_id: Option<String>,
    pub session_id: Option<String>,
    pub phase: Option<CyclePhase>,
    pub turn_intent_checkpoint_sequence: Option<u64>,
    pub capture_id: Option<String>,
    pub response_sha256: Option<String>,
    pub captured_response: Option<ProjectedCapturedResponse>,
    pub captured_response_retired_reason: Option<String>,
    pub response_draft: Option<ProjectedResponseDraft>,
    pub capture_replayed: bool,
    pub patch_id: Option<String>,
    pub response_cell: Option<ProjectedResponseCell>,
    pub commit: Option<String>,
    pub session_check_passed: bool,
    pub tracked_work_maintenance_required: Option<bool>,
    pub abandoned_reason: Option<String>,
    pub semantic_merge_conflict_advisories: Vec<SemanticMergeConflictAdvisory>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectedResponseCell {
    pub operation_id: String,
    pub cell_id: String,
    pub response_sha256: String,
    pub content_hash: String,
    pub applied: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectedCapturedResponse {
    pub cycle_id: String,
    pub capture_id: String,
    pub response_sha256: String,
    pub response_body: String,
    pub intent_body: Option<String>,
    pub mutation_plan_json: Option<String>,
    pub file_hash: Option<String>,
    pub snapshot_hash: Option<String>,
    pub baseline_content: Option<String>,
}

impl From<agent_doc_state_backbone::CapturedResponseProjection> for ProjectedCapturedResponse {
    fn from(capture: agent_doc_state_backbone::CapturedResponseProjection) -> Self {
        Self {
            cycle_id: capture.cycle_id,
            capture_id: capture.capture_id,
            response_sha256: capture.response_sha256,
            response_body: capture.response_body,
            intent_body: capture.intent_body,
            mutation_plan_json: capture.mutation_plan_json,
            file_hash: capture.file_hash,
            snapshot_hash: capture.snapshot_hash,
            baseline_content: capture.baseline_content,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectedResponseDraft {
    pub cycle_id: String,
    pub checkpoint_id: String,
    pub checkpoint_count: u64,
    pub response_sha256: String,
    pub response_body: String,
    pub file_hash: Option<String>,
}

/// One content-bearing response checkpoint from the append-only Lazily
/// ledger. Unlike [`ProjectedCapturedResponse`], this retains history order so
/// structural recovery can inspect an older valid checkpoint after a newer
/// whole-document materialization was corrupted.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CapturedResponseCheckpoint {
    pub sequence: u64,
    pub cycle_id: String,
    pub capture_id: String,
    pub response_sha256: String,
    pub response_body: String,
    pub file_hash: Option<String>,
    pub snapshot_hash: Option<String>,
    pub baseline_content: Option<String>,
}

impl ProjectedCloseoutState {
    pub fn phase_is_open(&self) -> Option<bool> {
        self.phase.map(CyclePhase::is_open)
    }

    pub fn matches_cycle(&self, cycle_id: &str) -> bool {
        self.cycle_id.as_deref() == Some(cycle_id)
    }

    pub fn event_label(&self, phase: CyclePhase) -> String {
        match phase {
            CyclePhase::PreflightStarted => "state_backbone_preflight_started".to_string(),
            CyclePhase::ResponseCaptured => self
                .capture_id
                .as_deref()
                .map(|capture_id| {
                    format!("state_backbone_response_captured capture_id={capture_id}")
                })
                .unwrap_or_else(|| "state_backbone_response_captured".to_string()),
            CyclePhase::WriteApplied => self
                .patch_id
                .as_deref()
                .map(|patch_id| format!("state_backbone_write_applied patch_id={patch_id}"))
                .unwrap_or_else(|| "state_backbone_write_applied".to_string()),
            CyclePhase::Committed => self
                .commit
                .as_deref()
                .map(|commit| format!("state_backbone_commit_observed commit={commit}"))
                .unwrap_or_else(|| "state_backbone_commit_observed".to_string()),
            CyclePhase::Abandoned => self
                .abandoned_reason
                .as_deref()
                .map(|reason| format!("state_backbone_cycle_abandoned reason={reason}"))
                .unwrap_or_else(|| "state_backbone_cycle_abandoned".to_string()),
        }
    }
}

impl From<agent_doc_state_backbone::CloseoutProjection> for ProjectedCloseoutState {
    fn from(projection: agent_doc_state_backbone::CloseoutProjection) -> Self {
        Self {
            cycle_id: projection.cycle_id,
            session_id: projection.session_id,
            phase: projection.phase,
            turn_intent_checkpoint_sequence: projection
                .turn_intent_checkpoint
                .as_ref()
                .map(|checkpoint| checkpoint.checkpoint_sequence),
            capture_id: projection.capture_id,
            response_sha256: projection.response_sha256,
            captured_response: projection
                .captured_response
                .map(ProjectedCapturedResponse::from),
            captured_response_retired_reason: projection.captured_response_retired_reason,
            response_draft: projection
                .response_draft
                .map(|draft| ProjectedResponseDraft {
                    cycle_id: draft.cycle_id,
                    checkpoint_id: draft.checkpoint_id,
                    checkpoint_count: draft.checkpoint_count,
                    response_sha256: draft.response_sha256,
                    response_body: draft.response_body,
                    file_hash: draft.file_hash,
                }),
            capture_replayed: projection.capture_replayed,
            patch_id: projection.patch_id,
            response_cell: projection.response_cell.map(|cell| ProjectedResponseCell {
                operation_id: cell.operation_id,
                cell_id: cell.cell_id,
                response_sha256: cell.response_sha256,
                content_hash: cell.content_hash,
                applied: cell.applied,
            }),
            commit: projection.commit,
            session_check_passed: projection.session_check_passed,
            tracked_work_maintenance_required: projection.tracked_work_maintenance_required,
            abandoned_reason: projection.abandoned_reason,
            semantic_merge_conflict_advisories: projection
                .semantic_merge_conflict_advisories
                .into_iter()
                .map(|advisory| SemanticMergeConflictAdvisory {
                    component: advisory.component,
                    id: advisory.id,
                    reason: advisory.reason,
                    detail: advisory.detail,
                    recorded_cycle_id: advisory.recorded_cycle_id,
                })
                .collect(),
        }
    }
}

pub fn load_closeout_projection(file: &Path) -> Result<Option<ProjectedCloseoutState>> {
    Ok(load_document_projection(file)?
        .map(|document| ProjectedCloseoutState::from(document.closeout)))
}

pub fn load_projected_captured_response(
    file: &Path,
    capture_id: &str,
) -> Result<Option<ProjectedCapturedResponse>> {
    Ok(load_closeout_projection(file)?
        .and_then(|projection| projection.captured_response)
        .filter(|capture| capture.capture_id == capture_id))
}

/// Load the response continuation pinned to the active retained Base → Target
/// transition. This identity survives later closeout-cycle recycling and is
/// therefore authoritative for reactive retained-finalize resumption.
pub fn load_projected_retained_captured_response(
    file: &Path,
) -> Result<Option<ProjectedCapturedResponse>> {
    Ok(load_document_projection(file)?
        .and_then(|document| document.document.pending_write)
        .and_then(|intent| intent.continuation)
        .map(ProjectedCapturedResponse::from))
}

pub fn load_projected_response_draft(
    file: &Path,
    cycle_id: &str,
) -> Result<Option<ProjectedResponseDraft>> {
    Ok(load_closeout_projection(file)?
        .and_then(|projection| projection.response_draft)
        .filter(|draft| draft.cycle_id == cycle_id))
}

/// Load a bounded newest-first history of content-bearing response captures.
///
/// A malformed latest baseline can hide the valid checkpoint directly before
/// it. This bounded state.db query keeps Lazily authoritative without replaying
/// the full event ledger on every repair attempt.
pub fn load_recent_captured_response_checkpoints(
    file: &Path,
    limit: usize,
) -> Result<Vec<CapturedResponseCheckpoint>> {
    let canonical = file.canonicalize().unwrap_or_else(|_| file.to_path_buf());
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(Vec::new());
    };
    let project_root = agent_doc_project_root_io::project_root_or_file_parent(&canonical)?;
    if !agent_doc_sqlite::state_store::state_db_path(&project_root).exists() {
        return Ok(Vec::new());
    }
    let conn = agent_doc_sqlite::state_store::open_state_db(&project_root)?;
    let rows = agent_doc_sqlite::state_store::load_recent_state_events_by_fact_type_from_db(
        &conn,
        &document_hash,
        "response_captured",
        limit,
    )?;
    let mut checkpoints = Vec::new();
    for row in rows {
        let event: agent_doc_state_backbone::StateEvent =
            serde_json::from_str(&row.payload_json)
                .with_context(|| format!("decode state event {}", row.event_id))?;
        if let agent_doc_state_backbone::StateFact::ResponseCaptured {
            cycle_id,
            capture_id,
            response_sha256,
            response_body: Some(response_body),
            file_hash,
            snapshot_hash,
            baseline_content,
            ..
        } = event.fact
        {
            checkpoints.push(CapturedResponseCheckpoint {
                sequence: row.sequence,
                cycle_id,
                capture_id,
                response_sha256,
                response_body,
                file_hash,
                snapshot_hash,
                baseline_content,
            });
        }
    }
    Ok(checkpoints)
}

/// Load the authoritative per-document projection.
///
/// A live controller serves its in-memory reactive projection. Cold durable
/// replay is reserved for actorless/bootstrap use when no controller is
/// reachable.
pub fn load_document_projection(
    file: &Path,
) -> Result<Option<agent_doc_state_backbone::DocumentStateProjection>> {
    let canonical = file.canonicalize().unwrap_or_else(|_| file.to_path_buf());
    let project_root = agent_doc_project_root_io::project_root_or_file_parent(&canonical)?;
    // `#lazily-hot-path`: prefer the controller's live in-memory projection (the
    // authority) when a controller is live; replay cold `state.db` only for the
    // actorless/bootstrap boundary. Read-only — no fact is emitted.
    let controller_socket = agent_doc_controller::paths::socket_path(&project_root);
    if controller_socket.exists() && !in_controller_request() {
        let request = serde_json::json!({
            "command": "document_state_projection",
            "file": canonical,
        });
        match agent_doc_state_wire::send_ndjson_request_to_actor(
            &controller_socket,
            &request,
            std::time::Duration::from_secs(5),
        ) {
            Ok(raw) => {
                #[derive(serde::Deserialize)]
                #[allow(dead_code)]
                struct ProjectionEnvelope {
                    ok: bool,
                    data: Option<agent_doc_state_backbone::DocumentStateProjection>,
                    error: Option<String>,
                }
                let envelope: ProjectionEnvelope = serde_json::from_str(&raw)
                    .context("decode document_state_projection response")?;
                if !envelope.ok {
                    anyhow::bail!(
                        "document_state_projection rejected by controller: {}",
                        envelope.error.as_deref().unwrap_or("unknown error")
                    );
                }
                return Ok(envelope.data);
            }
            Err(agent_doc_state_wire::ActorRequestError::Connect(_))
            | Err(agent_doc_state_wire::ActorRequestError::Timeout(_)) => {
                // The socket file may be stale (controller exited/crashed) or
                // absent (cold start), or the live controller accepted the
                // connection but did not answer this read within the timeout
                // (a wedged/overloaded actor surfaces `EAGAIN` — "Resource
                // temporarily unavailable"). All are momentary unreachability:
                // fall back to `state.db` rather than erroring — a read must not
                // fail just because the live controller is slow or momentarily
                // unreachable. (Hot-path *writes*/commands below still fail
                // closed on `Timeout` via their catch-all arm.)
            }
            Err(err) => {
                return Err(err)
                    .context("query live document_state_projection through the Lazily controller");
            }
        }
    }

    // Actorless fallback: replay cold `state.db`.
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(None);
    };
    if !agent_doc_sqlite::state_store::state_db_path(&project_root).exists() {
        return Ok(None);
    }
    let conn = agent_doc_sqlite::state_store::open_state_db(&project_root)?;
    let rows = agent_doc_sqlite::state_store::load_state_events_for_cycle_projection_from_db(
        &conn,
        &document_hash,
    )?;
    if rows.is_empty() {
        return Ok(None);
    }

    let mut ledger = agent_doc_state_backbone::EventLedger::new();
    for row in rows {
        let event: agent_doc_state_backbone::StateEvent =
            serde_json::from_str(&row.payload_json)
                .with_context(|| format!("decode state event {}", row.event_id))?;
        ledger.append(event);
    }

    Ok(ledger.project_document(&document_hash))
}

/// Load the newest durable agent-doc disk-write observation for `file`.
///
/// This is the watcher-facing state-machine seam for self-write echo
/// filtering. It reads the single `state.db` lifecycle authority.
pub fn load_document_disk_write(
    file: &Path,
) -> Result<Option<agent_doc_state_backbone::DocumentDiskWriteProjection>> {
    Ok(
        load_document_projection(file)?
            .and_then(|projection| projection.document.latest_disk_write),
    )
}

/// Load the exact target hash of the newest converged binary-owned pending write.
///
/// Commit recovery uses this durable state-plane proof when an older binary
/// settled a retained editor delivery before it learned to preserve
/// `pending_only_commit_target_hash`. The projection is controller-live when
/// available and cold-replayed only at the actorless boundary.
pub fn load_latest_converged_pending_write_hash(file: &Path) -> Result<Option<String>> {
    Ok(load_document_projection(file)?
        .and_then(|projection| projection.document.latest_converged_write)
        .and_then(|write| {
            (write.source
                == agent_doc_state_backbone::write_source::DocumentWriteSource::PendingWrite)
                .then_some(write.target_hash)
        }))
}

pub fn apply_closeout_projection_to_cycle_state(
    state: &mut CycleState,
    projection: &ProjectedCloseoutState,
) -> bool {
    let Some(projected_phase) = projection.phase else {
        return false;
    };
    if !projection.matches_cycle(&state.cycle_id) {
        return false;
    }

    let resolution = agent_doc_turn::resolve_closeout_phase(state.phase, projected_phase);
    let preserve_noop_commit_event = state.phase == projected_phase
        && agent_doc_turn::cycle_policy::is_noop_commit_event(&state.last_event);
    let changed = matches!(
        resolution,
        agent_doc_turn::CloseoutPhaseResolution::AdoptProjected
    );
    if changed {
        state.phase = projected_phase;
    }
    // A complete turn-intent checkpoint retains the precise transition event.
    // Projection labels are only needed when reconstructing a pre-cutover
    // ledger that has typed closeout facts but no checkpoint.
    if (changed || state.last_event == "synthetic_state") && !preserve_noop_commit_event {
        state.last_event = projection.event_label(projected_phase);
    }
    // The content-bearing capture fact can be appended after the turn-intent
    // checkpoint (for example when a retry differs only by terminal newline).
    // While the cycle is still open, the newest same-cycle ledger capture is
    // authoritative. Keeping the checkpoint's older hash strands recovery:
    // the projected body and cycle identity can no longer match each other.
    if state.phase.is_open() && projection.capture_id.is_some() {
        state.capture_id = projection.capture_id.clone();
        state.response_sha256 = projection.response_sha256.clone();
    } else {
        if state.capture_id.is_none() {
            state.capture_id = projection.capture_id.clone();
        }
        if state.response_sha256.is_none() {
            state.response_sha256 = projection.response_sha256.clone();
        }
    }
    state.tracked_work_maintenance_required_at_preflight =
        projection.tracked_work_maintenance_required;
    changed
}

pub fn load_with_closeout_projection(file: &Path) -> Result<Option<CycleState>> {
    load(file)
}

#[derive(Debug, Clone, Copy)]
pub struct TerminalCloseoutProofInput<'a> {
    pub cycle_id: &'a str,
    pub last_event: &'a str,
    pub did_commit: bool,
    pub file_hash: &'a str,
    pub snapshot_hash: &'a str,
    pub head_hash: &'a str,
    pub state_file_hash_matches: bool,
    pub state_snapshot_hash_matches: bool,
    pub agreement: agent_doc_state_backbone::TerminalCloseoutAgreement,
    pub response_materialized_in_head: bool,
    pub capture_id: Option<&'a str>,
    pub response_sha256: Option<&'a str>,
    pub recorded_at_ms: u64,
}

pub fn append_terminal_closeout_proof(
    file: &Path,
    proof: TerminalCloseoutProofInput<'_>,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let event_id = format!(
        "terminal-closeout-proof:{document_hash}:{}:{}:{}:{}:{}",
        proof.cycle_id, proof.file_hash, proof.snapshot_hash, proof.head_hash, proof.recorded_at_ms
    );
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::TerminalCloseoutProofRecorded {
            document_hash,
            cycle_id: proof.cycle_id.to_string(),
            last_event: proof.last_event.to_string(),
            did_commit: proof.did_commit,
            file_hash: proof.file_hash.to_string(),
            snapshot_hash: proof.snapshot_hash.to_string(),
            head_hash: proof.head_hash.to_string(),
            state_file_hash_matches: proof.state_file_hash_matches,
            state_snapshot_hash_matches: proof.state_snapshot_hash_matches,
            agreement: proof.agreement,
            response_materialized_in_head: proof.response_materialized_in_head,
            capture_id: proof.capture_id.map(str::to_string),
            response_sha256: proof.response_sha256.map(str::to_string),
            recorded_at_ms: proof.recorded_at_ms,
        },
    )
}

pub fn load_latest_terminal_closeout_proof(
    file: &Path,
) -> Result<Option<agent_doc_state_backbone::TerminalCloseoutProofProjection>> {
    let Some(document) = load_document_projection(file)? else {
        return Ok(None);
    };
    let Some(cycle_id) = document.proof.latest_terminal_closeout_cycle_id else {
        return Ok(None);
    };
    Ok(document.proof.terminal_closeouts.get(&cycle_id).cloned())
}

pub struct CloseoutRecoveryEvidenceInput<'a> {
    pub visible_markdown_hash: &'a str,
    pub snapshot_hash: Option<&'a str>,
    pub active_cycle_id: Option<&'a str>,
    pub active_cycle_phase: Option<CyclePhase>,
    pub active_capture_id: Option<&'a str>,
    pub active_capture_cycle_id: Option<&'a str>,
    pub active_capture_state: Option<&'a str>,
    pub active_capture_response_sha256: Option<&'a str>,
    pub response_body: agent_doc_state_backbone::CloseoutRecoveryResponseBodyEvidence,
    pub queue_only_drift: Option<agent_doc_state_backbone::CloseoutRecoveryQueueOnlyDriftEvidence>,
    pub snapshot_head_drift: Option<agent_doc_state_backbone::CloseoutRecoveryDriftEvidence>,
    pub snapshot_visible_drift: Option<agent_doc_state_backbone::CloseoutRecoveryDriftEvidence>,
    pub editor_ipc: agent_doc_state_backbone::CloseoutRecoveryEditorIpcEvidence,
    pub binary_freshness: agent_doc_state_backbone::CloseoutRecoveryBinaryFreshnessEvidence,
    pub recorded_at_ms: u64,
}

pub fn append_closeout_recovery_evidence(
    file: &Path,
    evidence: CloseoutRecoveryEvidenceInput<'_>,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let evidence_key = closeout_recovery_evidence_key(&evidence)?;
    let event_id = format!("closeout-recovery-evidence:{document_hash}:{evidence_key}");
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::CloseoutRecoveryEvidenceRecorded {
            document_hash,
            evidence_key,
            visible_markdown_hash: evidence.visible_markdown_hash.to_string(),
            snapshot_hash: evidence.snapshot_hash.map(str::to_string),
            active_cycle_id: evidence.active_cycle_id.map(str::to_string),
            active_cycle_phase: evidence.active_cycle_phase,
            active_capture_id: evidence.active_capture_id.map(str::to_string),
            active_capture_cycle_id: evidence.active_capture_cycle_id.map(str::to_string),
            active_capture_state: evidence.active_capture_state.map(str::to_string),
            active_capture_response_sha256: evidence
                .active_capture_response_sha256
                .map(str::to_string),
            response_body: evidence.response_body,
            queue_only_drift: evidence.queue_only_drift,
            snapshot_head_drift: evidence.snapshot_head_drift,
            snapshot_visible_drift: evidence.snapshot_visible_drift,
            editor_ipc: evidence.editor_ipc,
            binary_freshness: evidence.binary_freshness,
            recorded_at_ms: evidence.recorded_at_ms,
        },
    )
}

pub fn load_latest_closeout_recovery_evidence(
    file: &Path,
) -> Result<Option<agent_doc_state_backbone::CloseoutRecoveryEvidenceProjection>> {
    let Some(document) = load_document_projection(file)? else {
        return Ok(None);
    };
    let Some(key) = document.proof.latest_closeout_recovery_evidence_key else {
        return Ok(None);
    };
    Ok(document.proof.closeout_recovery_evidence.get(&key).cloned())
}

fn closeout_recovery_evidence_key(evidence: &CloseoutRecoveryEvidenceInput<'_>) -> Result<String> {
    let payload = serde_json::to_string(&serde_json::json!({
        "visible_markdown_hash": evidence.visible_markdown_hash,
        "snapshot_hash": evidence.snapshot_hash,
        "active_cycle_id": evidence.active_cycle_id,
        "active_cycle_phase": evidence.active_cycle_phase,
        "active_capture_id": evidence.active_capture_id,
        "active_capture_cycle_id": evidence.active_capture_cycle_id,
        "active_capture_state": evidence.active_capture_state,
        "active_capture_response_sha256": evidence.active_capture_response_sha256,
        "response_body": &evidence.response_body,
        "queue_only_drift": &evidence.queue_only_drift,
        "snapshot_head_drift": &evidence.snapshot_head_drift,
        "snapshot_visible_drift": &evidence.snapshot_visible_drift,
        "editor_ipc": &evidence.editor_ipc,
        "binary_freshness": &evidence.binary_freshness,
    }))
    .context("serialize closeout recovery evidence key")?;
    Ok(agent_doc_hash::content_hash(&payload))
}

pub fn admit_with_current_resolver<R, S, L>(
    file: &Path,
    mut resolve_current: R,
    mut load_snapshot: S,
    mut log_admission: L,
) -> Result<AdmitOutput>
where
    R: FnMut(&Path) -> Result<String>,
    S: FnMut(&Path) -> Result<Option<String>>,
    L: FnMut(&Path, &str),
{
    if !file.exists() {
        anyhow::bail!("file not found: {}", file.display());
    }

    let current = resolve_current(file)
        .with_context(|| format!("failed to resolve current document for {}", file.display()))?;
    let snapshot = load_snapshot(file)
        .with_context(|| format!("failed to load snapshot for {}", file.display()))?;

    let state = start_preflight(file, snapshot.as_deref(), Some(&current))?;
    let phase = state.phase.as_str().to_string();
    log_admission(
        file,
        &format!(
            "realtime_admit file={} cycle_id={} phase={} source=admit action=accepted maintenance_required=false preflight_required=false",
            file.display(),
            state.cycle_id,
            phase
        ),
    );

    Ok(AdmitOutput {
        admitted: true,
        file: file
            .canonicalize()
            .unwrap_or_else(|_| file.to_path_buf())
            .display()
            .to_string(),
        cycle_id: state.cycle_id,
        cycle_phase: phase,
        last_event: state.last_event,
        source: "admit".to_string(),
        maintenance_required: false,
        preflight_required: false,
        snapshot_hash: state.snapshot_hash,
        file_hash: state.file_hash,
    })
}

pub fn start_preflight(
    file: &Path,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> Result<CycleState> {
    start_preflight_with_task(file, snapshot_content, file_content, None, None)
}

/// (#reentrant-finalize Phase 5) Start preflight with optional queue task
/// identifiers. `queue_task_id` is the backlog ID (e.g. `#reentrant-phase2`).
/// `turn_id` is derived from the backlog ID or auto-generated. Both are stored
/// in cycle state so crash recovery can correlate the pipeline step with the
/// original task.
pub fn start_preflight_with_task(
    file: &Path,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
    queue_task_id: Option<&str>,
    turn_id: Option<&str>,
) -> Result<CycleState> {
    let now = now_secs();
    let canonical = file.canonicalize().unwrap_or_else(|_| file.to_path_buf());
    // `#queueskip`: carry forward the skipped-head accumulator so a head marked
    // skippable in a prior cycle stays skipped until it is consumed or removed
    // (preflight recomputes/clears it each cycle). Without this the flag would
    // reset every cycle and the dead head would be re-dispatched on alternating
    // cycles instead of staying skipped.
    let carried_skipped_queue_head_ids = load(file)
        .ok()
        .flatten()
        .map(|prior| prior.skipped_queue_head_ids)
        .unwrap_or_default();
    // `#preflightinbinary`: a second preflight inside the same open turn is a
    // re-entry, not a new turn — see `reusable_by_reentrant_preflight`. Keep the
    // turn's IDENTITY (cycle id and start time) and recompute everything else,
    // rather than returning the stored state: the contract must reflect facts
    // recorded since the turn opened (acks, queue heads, hashes), so a re-entry
    // that replayed the opening snapshot would report a stale document.
    let reentrant = load(file)
        .ok()
        .flatten()
        .filter(|open| open.reusable_by_reentrant_preflight(0, now, STALLED_CYCLE_RESOLVE_SECS));
    if let Some(open) = reentrant.as_ref() {
        eprintln!(
            "[preflight] re-entrant preflight kept open cycle {} for {}",
            open.cycle_id,
            file.display()
        );
    }
    let phase = CyclePhaseMachine::transition(CyclePhase::Committed, CycleEvent::StartPreflight)
        .unwrap_or(CyclePhase::PreflightStarted);
    let state = CycleState {
        cycle_id: reentrant
            .as_ref()
            .map(|open| open.cycle_id.clone())
            .unwrap_or_else(|| format!("cycle-{}", now_millis())),
        file: canonical.display().to_string(),
        phase,
        last_event: "preflight_started".to_string(),
        started_at: reentrant.as_ref().map_or(now, |open| open.started_at),
        updated_at: now,
        snapshot_hash: snapshot_content.map(agent_doc_hash::content_hash),
        file_hash: file_content.map(agent_doc_hash::content_hash),
        normalized_snapshot_hash: snapshot_content.map(replay_content_hash),
        normalized_file_hash: file_content.map(replay_content_hash),
        capture_id: None,
        response_sha256: None,
        had_pending_mutations: false,
        requires_backlog_capture: false,
        required_backlog_targets: Vec::new(),
        required_explicit_backlog_item_count: 0,
        required_plan_reference_count: 0,
        prompt_targets: Vec::new(),
        queue_task_id: queue_task_id.map(|s| s.to_string()),
        turn_id: turn_id.map(|s| s.to_string()),
        recycle_resume_consumed: false,
        pending_done_ids: Vec::new(),
        pending_only_commit_target_hash: None,
        pending_kept_open_ids: Vec::new(),
        reaped_pending_ids: Vec::new(),
        expect_done_or_gate_ids: Vec::new(),
        pending_gated_ids: Vec::new(),
        pending_added_this_cycle: false,
        pending_added_ids: Vec::new(),
        requested_done_ids: Vec::new(),
        requested_added_ids: Vec::new(),
        pending_actionable_ids: Vec::new(),
        pending_anchored_ids: Vec::new(),
        tracked_work_maintenance_required_at_preflight: file_content
            .map(agent_doc_document::tracked_work_projection::tracked_work_maintenance_required),
        ipc_snapshot_adoption_blocked: false,
        dropped_exchange_prompts: Vec::new(),
        dropped_queue_prompts: Vec::new(),
        active_queue_heads: file_content
            .map(agent_doc_queue::queue_heads::active_queue_heads)
            .unwrap_or_default(),
        active_free_text_queue_heads: file_content
            .map(agent_doc_queue::queue_heads::active_free_text_queue_heads)
            .unwrap_or_default(),
        semantic_merge_conflict_advisories: reentrant
            .as_ref()
            .map(|open| open.semantic_merge_conflict_advisories.clone())
            .unwrap_or_default(),
        blocked_closeout: None,
        skipped_queue_head_ids: carried_skipped_queue_head_ids,
    };
    save(file, &state)?;
    append_closeout_projection_event(file, &state, CloseoutProjectionEvent::PreflightStarted)?;
    append_phase_event_to_session_log(file, &state, file_content);
    Ok(state)
}

/// Record (or overwrite) the runnable `agent:queue` heads that were active at
/// the start of the cycle (`#queue-clear-unrun-items`). Normally populated by
/// `start_preflight` from the visible document; exposed so a caller that only
/// has the pre-cycle queue text later can backfill the proof anchor.
pub fn record_active_queue_heads(file: &Path, heads: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let normalized: Vec<String> = heads
        .iter()
        .map(|head| head.trim().to_string())
        .filter(|head| !head.is_empty())
        .collect();
    if state.active_queue_heads != normalized {
        state.active_queue_heads = normalized;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#manual-queue-head-loss`: union the live working-tree `agent:queue` directive
/// heads into the recorded removal-proof anchor. `start_preflight` records
/// `active_queue_heads` once, from the document as it stood at preflight. A user
/// who types a fresh `do [#id]` into `agent:queue` AFTER that point — e.g. while a
/// dispatch attempt is stalled on owner-pane recursion or a busy authoritative
/// actor — is invisible to the `#queue-clear-unrun-items` removal guard, so a
/// later closeout convergence can silently drop that runnable manual head while
/// its backlog item stays open. Observing the live heads at the write/commit
/// boundary (before any pending mutation or queue convergence) extends the same
/// durable-proof requirement to manually inserted heads. Only ADDS heads; never
/// removes an already-recorded head, so a legitimately consumed head still
/// resolves through the existing done/gate/reap proof in the removal guard.
pub fn observe_live_queue_heads(file: &Path, doc: &str) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let mut changed = false;
    for head in agent_doc_queue::queue_heads::active_queue_heads(doc) {
        let head = head.trim().to_string();
        if head.is_empty() {
            continue;
        }
        if !state
            .active_queue_heads
            .iter()
            .any(|existing| existing == &head)
        {
            state.active_queue_heads.push(head);
            changed = true;
        }
    }
    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#durablerecycle`: persist the prompt-facing turn checkpoint once preflight
/// has computed the stable merge baseline and prompt target identity. The
/// initial `start_preflight` record opens early; this fills the restart-critical
/// fields that are only known after queue and diff analysis.
pub fn record_turn_checkpoint(
    file: &Path,
    prompt_targets: &[String],
    queue_task_id: Option<&str>,
    turn_id: Option<&str>,
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.is_open() {
        return Ok(Some(state));
    }

    let normalized_prompt_targets = normalize_checkpoint_text_list(prompt_targets);
    let normalized_queue_task_id = queue_task_id
        .map(normalize_checkpoint_task_id)
        .filter(|id| !id.is_empty());
    let normalized_turn_id = turn_id
        .map(normalize_checkpoint_task_id)
        .filter(|id| !id.is_empty());

    let mut changed = false;
    if state.prompt_targets != normalized_prompt_targets {
        state.prompt_targets = normalized_prompt_targets;
        changed = true;
    }
    if state.queue_task_id != normalized_queue_task_id {
        state.queue_task_id = normalized_queue_task_id;
        changed = true;
    }
    if state.turn_id != normalized_turn_id {
        state.turn_id = normalized_turn_id;
        changed = true;
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#midturn-recycle-resume` Phase B: latch that a fresh supervisor boot has
/// re-dispatched this open cycle's interrupted turn from the `#durablerecycle`
/// checkpoint (the harness child died across the `execve` recycle). Idempotency
/// guard for the boot-resume path — a later boot reading the same still-open
/// checkpoint sees the latch and does NOT re-dispatch the turn again. A committed
/// cycle is never open so this is a no-op there. Returns the (possibly updated)
/// state, or `Ok(None)` when no cycle state exists.
pub fn mark_recycle_resume_consumed(file: &Path) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.recycle_resume_consumed {
        state.recycle_resume_consumed = true;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// Pure decision core of `mark_write_applied`: advance the phase and produce the
/// mutated state. Returns `(state, transitioned)`; `transitioned = false` means
/// the phase could not advance (idempotent no-op) and the caller MUST emit no
/// fact. Pure (no I/O), so it is shared by [`mark_write_applied`] and the
/// controller's live authority path on the command plane (`#lzdurablesink`).
pub fn decide_write_applied(
    current: Option<CycleState>,
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> (CycleState, bool) {
    let mut state = current.unwrap_or_else(|| synthetic_state(file, CyclePhase::PreflightStarted));
    let Some(next_phase) = CyclePhaseMachine::transition(state.phase, CycleEvent::WriteApplied)
    else {
        return (state, false);
    };
    state.phase = next_phase;
    state.last_event = event.to_string();
    state.updated_at = now_secs();
    state.snapshot_hash = snapshot_content.map(agent_doc_hash::content_hash);
    state.file_hash = file_content.map(agent_doc_hash::content_hash);
    state.normalized_snapshot_hash = snapshot_content.map(replay_content_hash);
    state.normalized_file_hash = file_content.map(replay_content_hash);
    (state, true)
}

pub fn mark_write_applied(
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> Result<CycleState> {
    // `#lazily-hot-path`: when a controller is live, the transition is decided
    // from the live Lazily projection over the command plane (not this client's
    // cold `state.db` read) and the phase fact(s) are sunk in the controller.
    // When no controller exists (actorless bootstrap / cold start), keep the
    // local `load→decide→save→append` path as the compatibility boundary.
    if submit_closeout_advance_via_socket(
        file,
        command_plane::CloseoutPhaseEvent::WriteApplied,
        None,
        None,
        snapshot_content,
        file_content,
        None,
        None,
    )? {
        let state = load(file)?.context(
            "closeout_advance WriteApplied applied through the command plane but no cycle state \
             was read back",
        )?;
        append_phase_event_to_session_log(file, &state, file_content);
        return Ok(state);
    }
    let (state, transitioned) =
        decide_write_applied(load(file)?, file, event, snapshot_content, file_content);
    if transitioned {
        save(file, &state)?;
        append_closeout_projection_event(file, &state, CloseoutProjectionEvent::WriteApplied)?;
        append_phase_event_to_session_log(file, &state, file_content);
    }
    Ok(state)
}

/// Pure decision core of `mark_response_captured`. See [`decide_write_applied`]
/// for the contract; `transitioned = false` means emit no fact. Shared by
/// [`mark_response_captured`] and the controller authority (`#lzdurablesink`).
pub fn decide_response_captured(
    current: Option<CycleState>,
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
    response_sha256: &str,
    cycle_id_hint: Option<&str>,
) -> (CycleState, bool) {
    let mut state = current.unwrap_or_else(|| {
        synthetic_state_with_id(file, CyclePhase::PreflightStarted, cycle_id_hint)
    });
    let Some(next_phase) = CyclePhaseMachine::transition(state.phase, CycleEvent::ResponseCaptured)
    else {
        return (state, false);
    };
    state.phase = next_phase;
    state.last_event = event.to_string();
    state.updated_at = now_secs();
    state.snapshot_hash = snapshot_content.map(agent_doc_hash::content_hash);
    state.file_hash = file_content.map(agent_doc_hash::content_hash);
    state.normalized_snapshot_hash = snapshot_content.map(replay_content_hash);
    state.normalized_file_hash = file_content.map(replay_content_hash);
    state.capture_id = Some(state.cycle_id.clone());
    state.response_sha256 = Some(response_sha256.to_string());
    (state, true)
}

pub fn mark_response_captured(
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
    response_sha256: &str,
    cycle_id_hint: Option<&str>,
) -> Result<CycleState> {
    // `#lazily-hot-path`: submit over the command plane when a controller is
    // live (decide from the live projection); fall back to the local path for
    // the actorless/bootstrap boundary.
    if submit_closeout_advance_via_socket(
        file,
        command_plane::CloseoutPhaseEvent::ResponseCaptured,
        None,
        None,
        snapshot_content,
        file_content,
        Some(response_sha256),
        cycle_id_hint,
    )? {
        let state = load(file)?.context(
            "closeout_advance ResponseCaptured applied through the command plane but no cycle \
             state was read back",
        )?;
        append_phase_event_to_session_log(file, &state, file_content);
        return Ok(state);
    }
    let (state, transitioned) = decide_response_captured(
        load(file)?,
        file,
        event,
        snapshot_content,
        file_content,
        response_sha256,
        cycle_id_hint,
    );
    if transitioned {
        save(file, &state)?;
        append_closeout_projection_event(file, &state, CloseoutProjectionEvent::ResponseCaptured)?;
        append_phase_event_to_session_log(file, &state, file_content);
        // NB: intentionally do NOT mirror the pipeline block here. A captured response
        // is the complete final payload, but capture durability is not visible document
        // authority. The mirror runs at `write_applied` only after final placement is
        // proven. Recovery-only partial checkpoints never call this transition (#22a8).
    }
    Ok(state)
}

pub fn mark_pending_mutations(file: &Path) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.had_pending_mutations {
        state.had_pending_mutations = true;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_pending_done_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .pending_done_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.pending_done_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_pending_only_commit_target(
    file: &Path,
    target_hash: &str,
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let target_hash = target_hash.trim();
    if target_hash.is_empty() {
        anyhow::bail!("pending-only commit target hash must not be empty");
    }
    if state.pending_only_commit_target_hash.as_deref() != Some(target_hash) {
        state.pending_only_commit_target_hash = Some(target_hash.to_string());
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn clear_pending_only_commit_target(file: &Path, expected_target_hash: &str) -> Result<bool> {
    let Some(mut state) = load(file)? else {
        return Ok(false);
    };
    if state.pending_only_commit_target_hash.as_deref() != Some(expected_target_hash) {
        return Ok(false);
    }
    state.pending_only_commit_target_hash = None;
    state.updated_at = now_secs();
    save(file, &state)?;
    Ok(true)
}

/// Retire the whole retained pending-only continuation after it either commits
/// or is superseded by a newer editor-authoritative cut.
///
/// `pending_done_ids` belong to the target identity: retaining them after the
/// target is rejected could re-authorize the same stale deletion through the
/// legacy migration path.
pub fn clear_pending_only_commit_intent(file: &Path, expected_target_hash: &str) -> Result<bool> {
    let Some(mut state) = load(file)? else {
        return Ok(false);
    };
    if state.pending_only_commit_target_hash.as_deref() != Some(expected_target_hash) {
        return Ok(false);
    }
    state.pending_only_commit_target_hash = None;
    state.pending_done_ids.clear();
    state.updated_at = now_secs();
    save(file, &state)?;
    Ok(true)
}

pub fn record_pending_kept_open_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .pending_kept_open_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.pending_kept_open_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_reaped_pending_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .reaped_pending_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.reaped_pending_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#queueskip`: overwrite the skipped-head accumulator with the fully-recomputed
/// set for this cycle (carry-forward + newly-stalled heads, minus any that were
/// consumed or are no longer present). Preflight owns the recompute, so this is a
/// replace, not an append.
pub fn set_skipped_queue_head_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let mut seen = std::collections::HashSet::new();
    let normalized: Vec<String> = ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty() && seen.insert(id.clone()))
        .collect();
    if state.skipped_queue_head_ids != normalized {
        state.skipped_queue_head_ids = normalized;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_expect_done_or_gate_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .expect_done_or_gate_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.expect_done_or_gate_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#blocked-closeout-followup-capture`: record tracked ids gated this cycle.
pub fn record_pending_gated_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .pending_gated_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.pending_gated_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#opsproof-samecycle-add`: record the ids of tracked-work items added this
/// cycle (via `--pending-add` / `--pending-add-gated` / `--review-add`) so the
/// opportunistic ops-proof auto-completion can exclude brand-new same-cycle adds.
/// Record the tracked-work mutations this cycle REQUESTED, before the response
/// cell is published (`#mutprovenancepreresponse`).
pub fn record_requested_tracked_work(
    file: &Path,
    done_ids: &[String],
    added_ids: &[String],
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let mut changed = false;
    for (ids, slot) in [
        (done_ids, &mut state.requested_done_ids),
        (added_ids, &mut state.requested_added_ids),
    ] {
        for id in ids
            .iter()
            .map(|id| normalize_pending_id(id))
            .filter(|id| !id.is_empty())
        {
            if !slot.iter().any(|existing| existing == &id) {
                slot.push(id);
                changed = true;
            }
        }
    }
    if !changed {
        return Ok(Some(state));
    }
    save(file, &state)?;
    Ok(Some(state))
}

pub fn record_pending_added_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .pending_added_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.pending_added_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#opsproof-samecycle-add`: ids of tracked-work items added this cycle. Empty
/// when no cycle state exists. Used to exclude brand-new same-cycle adds from
/// opportunistic ops-proof auto-completion.
pub fn pending_added_ids(file: &Path) -> std::collections::HashSet<String> {
    load(file)
        .ok()
        .flatten()
        .map(|state| state.pending_added_ids.into_iter().collect())
        .unwrap_or_default()
}

/// `#backlogqueuepopulation`: record ids that became actionable during the
/// current closeout so queue reconciliation mirrors only explicit mutation
/// intent. This avoids resurrecting operator-deleted queue entries.
pub fn record_pending_actionable_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .pending_actionable_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.pending_actionable_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#backlogqueuepopulation`: ids made actionable by an add or ungate in this
/// cycle. Empty when no cycle state exists.
pub fn pending_actionable_ids(file: &Path) -> std::collections::HashSet<String> {
    load(file)
        .ok()
        .flatten()
        .map(|state| state.pending_actionable_ids.into_iter().collect())
        .unwrap_or_default()
}

/// `#queuemirrororder`: record ids filed against an explicit backlog anchor, so
/// the queue mirror can follow that anchor instead of the placement edge.
pub fn record_pending_anchored_ids(file: &Path, ids: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };

    let mut changed = false;
    for id in ids
        .iter()
        .map(|id| normalize_pending_id(id))
        .filter(|id| !id.is_empty())
    {
        if !state
            .pending_anchored_ids
            .iter()
            .any(|existing| existing == &id)
        {
            state.pending_anchored_ids.push(id);
            changed = true;
        }
    }

    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#queuemirrororder`: ids filed with an explicit `--backlog-add-after` /
/// `--backlog-add-before` anchor this cycle. Empty when no cycle state exists.
pub fn pending_anchored_ids(file: &Path) -> std::collections::HashSet<String> {
    load(file)
        .ok()
        .flatten()
        .map(|state| state.pending_anchored_ids.into_iter().collect())
        .unwrap_or_default()
}

/// `#blocked-closeout-followup-capture`: mark that this cycle added at least one
/// follow-up backlog item via a `--pending-add*` primitive.
pub fn mark_pending_added(file: &Path) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.pending_added_this_cycle {
        state.pending_added_this_cycle = true;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn resolved_pending_ids(file: &Path) -> Result<std::collections::HashSet<String>> {
    let Some(state) = load(file)? else {
        return Ok(std::collections::HashSet::new());
    };

    Ok(state
        .pending_done_ids
        .into_iter()
        .chain(state.reaped_pending_ids)
        .collect())
}

pub fn record_backlog_capture_requirement(
    file: &Path,
    required: bool,
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if state.requires_backlog_capture != required {
        state.requires_backlog_capture = required;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_backlog_target_requirements(
    file: &Path,
    requirements: &[BacklogTargetRequirement],
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if state.required_backlog_targets != requirements {
        state.required_backlog_targets = requirements.to_vec();
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_required_explicit_backlog_item_count(
    file: &Path,
    count: usize,
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if state.required_explicit_backlog_item_count != count {
        state.required_explicit_backlog_item_count = count;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_required_plan_reference_count(
    file: &Path,
    count: usize,
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if state.required_plan_reference_count != count {
        state.required_plan_reference_count = count;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn mark_recoverable_preflight_timeout(file: &Path, event: &str) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.is_open() {
        return Ok(Some(state));
    }
    let Some(next_phase) =
        CyclePhaseMachine::transition(state.phase, CycleEvent::RecoverablePreflightTimeout)
    else {
        return Ok(Some(state));
    };
    state.phase = next_phase;
    state.last_event = event.to_string();
    state.updated_at = now_secs();
    save(file, &state)?;
    append_phase_event_to_session_log(file, &state, None);
    Ok(Some(state))
}

pub fn record_open_cycle_progress(file: &Path, event: &str) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.is_open() {
        return Ok(Some(state));
    }
    state.last_event = event.to_string();
    state.updated_at = now_secs();
    save(file, &state)?;
    append_phase_event_to_session_log(file, &state, None);
    Ok(Some(state))
}

pub fn record_ipc_snapshot_adoption_blocked(file: &Path) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.ipc_snapshot_adoption_blocked {
        state.ipc_snapshot_adoption_blocked = true;
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#exchange-prompt-dropped-on-merge`: record user-authored exchange prompt
/// line(s) dropped when `content_ours` was adopted over a divergent IPC
/// candidate, so `session-check` can fail closed even after the editor
/// overwrites the disk prompt. Appends only previously-unseen prompt lines.
pub fn record_dropped_exchange_prompts(
    file: &Path,
    prompts: &[String],
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let mut changed = false;
    for prompt in prompts {
        let trimmed = prompt.trim();
        if trimmed.is_empty() {
            continue;
        }
        if !state
            .dropped_exchange_prompts
            .iter()
            .any(|existing| existing == trimmed)
        {
            state.dropped_exchange_prompts.push(trimmed.to_string());
            changed = true;
        }
    }
    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// Clear the recorded dropped-prompt markers once they are resolved (the prompt
/// reached the committed document on a later cycle).
pub fn clear_dropped_exchange_prompts(file: &Path) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.dropped_exchange_prompts.is_empty() {
        state.dropped_exchange_prompts.clear();
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// `#queue-user-edit-overwrite`: record user-authored `agent:queue` prompt
/// line(s) dropped when `content_ours` was adopted over a divergent IPC
/// candidate, so `session-check` can fail closed if a user queue edit was
/// silently deleted instead of consumed. Appends only previously-unseen lines.
pub fn record_dropped_queue_prompts(file: &Path, prompts: &[String]) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let mut changed = false;
    for prompt in prompts {
        let trimmed = prompt.trim();
        if trimmed.is_empty() {
            continue;
        }
        if !state
            .dropped_queue_prompts
            .iter()
            .any(|existing| existing == trimmed)
        {
            state.dropped_queue_prompts.push(trimmed.to_string());
            changed = true;
        }
    }
    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// Project node-keyed semantic-merge conflicts into the current cycle.
///
/// These are reactive diagnostics, not future-turn requests. A new preflight
/// clears them with the rest of the prior cycle projection.
pub fn record_semantic_merge_conflict_advisories(
    file: &Path,
    advisories: &[agent_doc_merge::document_cell_merge::MergeConflictAdvisory],
) -> Result<Option<CycleState>> {
    let Some(mut state) = load_with_closeout_projection(file)? else {
        if let Some(projection) = load_closeout_projection(file)?
            && let Some(cycle_id) = projection.cycle_id.as_deref()
        {
            for advisory in advisories {
                append_semantic_merge_conflict_advisory_recorded_event(
                    file,
                    cycle_id,
                    &SemanticMergeConflictAdvisory {
                        component: advisory.component.clone(),
                        id: advisory.id.clone(),
                        reason: advisory.reason.token().to_string(),
                        detail: advisory.detail.clone(),
                        recorded_cycle_id: Some(cycle_id.to_string()),
                    },
                )?;
            }
        }
        return Ok(None);
    };
    let cycle_id = state.cycle_id.clone();
    let mut changed = false;
    for advisory in advisories {
        let reason = advisory.reason.token().to_string();
        let projected_advisory = SemanticMergeConflictAdvisory {
            component: advisory.component.clone(),
            id: advisory.id.clone(),
            reason: reason.clone(),
            detail: advisory.detail.clone(),
            recorded_cycle_id: Some(cycle_id.clone()),
        };
        append_semantic_merge_conflict_advisory_recorded_event(
            file,
            &cycle_id,
            &projected_advisory,
        )?;
        if !state
            .semantic_merge_conflict_advisories
            .iter()
            .any(|existing| {
                existing.component == advisory.component
                    && existing.id == advisory.id
                    && existing.reason == reason
            })
        {
            state
                .semantic_merge_conflict_advisories
                .push(projected_advisory);
            changed = true;
        }
    }
    if changed {
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

/// Clear the recorded dropped-queue markers once they are resolved (the queue
/// edit reached the committed document or was legitimately consumed).
pub fn clear_dropped_queue_prompts(file: &Path) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    if !state.dropped_queue_prompts.is_empty() {
        state.dropped_queue_prompts.clear();
        state.updated_at = now_secs();
        save(file, &state)?;
    }
    Ok(Some(state))
}

pub fn record_editor_convergence_required(
    file: &Path,
    source: &str,
    reason: &str,
    patch_id: Option<&str>,
    detail: Option<&str>,
) -> Result<Option<CycleState>> {
    let Some(mut state) = load(file)? else {
        return Ok(None);
    };
    let blocked = BlockedCloseout {
        kind: "editor_convergence_required".to_string(),
        reason: reason.to_string(),
        source: source.to_string(),
        patch_id: patch_id.map(str::to_string),
        recovery: Some("retry_without_disk_write".to_string()),
        // A captured response is resumed by the keyed supervisor worker. A
        // manual closeout command here caused agents to stack another write on
        // top of the still-live capture.
        recovery_command: None,
        detail: detail.map(str::to_string),
    };
    if state.blocked_closeout.as_ref() != Some(&blocked) {
        state.blocked_closeout = Some(blocked);
        state.updated_at = now_secs();
        save(file, &state)?;
        append_phase_event_to_session_log(file, &state, None);
    }
    Ok(Some(state))
}

/// A pure closeout transition decision richer than `(state, transitioned)`: the
/// resulting state plus which durable facts to emit. `checkpoint` requests a
/// `turn_intent_checkpoint` fact; `facts` are the phase facts; `session_log` is a
/// client-side log signal (the authority ignores it). An empty decision
/// (`checkpoint=false`, no `facts`) is an idempotent no-op that emits nothing.
#[derive(Debug, Clone)]
pub struct CloseoutDecision {
    pub state: CycleState,
    pub checkpoint: bool,
    pub facts: Vec<CloseoutProjectionEvent>,
    pub session_log: bool,
}

/// Pure decision core of `mark_committed`. The committed transition is tri-state:
/// (1) already committed + matching/stable event → refresh content hashes
/// (checkpoint only if a hash changed) and re-emit the `Committed` fact; (2) the
/// phase machine cannot advance → no-op; (3) a real advance → checkpoint +
/// `Committed` fact. Shared by [`mark_committed`] and the authority.
pub fn decide_committed(
    current: Option<CycleState>,
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> CloseoutDecision {
    let mut state = match current {
        Some(state) if state.phase == CyclePhase::Abandoned => {
            synthetic_state(file, CyclePhase::WriteApplied)
        }
        Some(state) => state,
        None => synthetic_state(file, CyclePhase::WriteApplied),
    };
    if matches!(state.phase, CyclePhase::Committed)
        && (state.last_event == event || is_stable_commit_event(&state.last_event))
    {
        let mut refreshed = false;
        // A verified already-current commit is stronger terminal evidence than
        // the earlier commit transition: it proves the present projection was
        // reviewed without a new binary-owned response write. Persist that
        // observation so session-check cannot reconstruct the older
        // `commit_success` label and repeatedly fire the missing-response guard.
        if state.last_event != event && is_noop_commit_event(event) {
            state.last_event = event.to_string();
            refreshed = true;
        }
        if let Some(snapshot) = snapshot_content {
            let hash = agent_doc_hash::content_hash(snapshot);
            let normalized_hash = replay_content_hash(snapshot);
            refreshed |= state.snapshot_hash.as_deref() != Some(hash.as_str())
                || state.normalized_snapshot_hash.as_deref() != Some(normalized_hash.as_str());
            state.snapshot_hash = Some(hash);
            state.normalized_snapshot_hash = Some(normalized_hash);
        }
        if let Some(content) = file_content {
            let hash = agent_doc_hash::content_hash(content);
            let normalized_hash = replay_content_hash(content);
            refreshed |= state.file_hash.as_deref() != Some(hash.as_str())
                || state.normalized_file_hash.as_deref() != Some(normalized_hash.as_str());
            state.file_hash = Some(hash);
            state.normalized_file_hash = Some(normalized_hash);
        }
        if refreshed {
            state.updated_at = now_secs();
        }
        return CloseoutDecision {
            state,
            checkpoint: refreshed,
            facts: vec![CloseoutProjectionEvent::Committed],
            session_log: false,
        };
    }
    let Some(next_phase) = CyclePhaseMachine::transition(state.phase, CycleEvent::Committed) else {
        return CloseoutDecision {
            state,
            checkpoint: false,
            facts: vec![],
            session_log: false,
        };
    };
    state.phase = next_phase;
    state.last_event = event.to_string();
    state.updated_at = now_secs();
    state.blocked_closeout = None;
    if let Some(snapshot) = snapshot_content {
        state.snapshot_hash = Some(agent_doc_hash::content_hash(snapshot));
        state.normalized_snapshot_hash = Some(replay_content_hash(snapshot));
    }
    if let Some(content) = file_content {
        state.file_hash = Some(agent_doc_hash::content_hash(content));
        state.normalized_file_hash = Some(replay_content_hash(content));
    }
    CloseoutDecision {
        state,
        checkpoint: true,
        facts: vec![CloseoutProjectionEvent::Committed],
        session_log: true,
    }
}

pub fn mark_committed(
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> Result<CycleState> {
    let observation = command_plane::commit_observation_from_event_label(event);
    if submit_closeout_advance_via_socket(
        file,
        command_plane::CloseoutPhaseEvent::Committed(observation),
        Some(event),
        None,
        snapshot_content,
        file_content,
        None,
        None,
    )? {
        let state = load(file)?.context(
            "closeout_advance Committed applied through the command plane but no cycle state was \
             read back",
        )?;
        append_phase_event_to_session_log(file, &state, file_content);
        return Ok(state);
    }
    let decision = decide_committed(load(file)?, file, event, snapshot_content, file_content);
    if decision.checkpoint {
        save(file, &decision.state)?;
    }
    for fact in &decision.facts {
        append_closeout_projection_event(file, &decision.state, *fact)?;
    }
    if decision.session_log {
        append_phase_event_to_session_log(file, &decision.state, file_content);
    }
    Ok(decision.state)
}

/// Pure decision core of `mark_abandoned` (see [`CloseoutDecision`]). Tri-state:
/// (1) already abandoned → re-emit the `Abandoned` fact only; (2) not an open
/// cycle → no-op; (3) real abandon → checkpoint + `Abandoned` fact. Shared by
/// [`mark_abandoned`] and the authority.
pub fn decide_abandoned(
    current: Option<CycleState>,
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> CloseoutDecision {
    let mut state = current.unwrap_or_else(|| synthetic_state(file, CyclePhase::PreflightStarted));
    if state.phase == CyclePhase::Abandoned {
        return CloseoutDecision {
            state,
            checkpoint: false,
            facts: vec![CloseoutProjectionEvent::Abandoned],
            session_log: false,
        };
    }
    if !state.is_open() {
        return CloseoutDecision {
            state,
            checkpoint: false,
            facts: vec![],
            session_log: false,
        };
    }
    let Some(next_phase) = CyclePhaseMachine::transition(state.phase, CycleEvent::Abandoned) else {
        return CloseoutDecision {
            state,
            checkpoint: false,
            facts: vec![],
            session_log: false,
        };
    };
    state.phase = next_phase;
    state.last_event = event.to_string();
    state.updated_at = now_secs();
    if let Some(snapshot) = snapshot_content {
        state.snapshot_hash = Some(agent_doc_hash::content_hash(snapshot));
        state.normalized_snapshot_hash = Some(replay_content_hash(snapshot));
    }
    if let Some(content) = file_content {
        state.file_hash = Some(agent_doc_hash::content_hash(content));
        state.normalized_file_hash = Some(replay_content_hash(content));
    }
    CloseoutDecision {
        state,
        checkpoint: true,
        facts: vec![CloseoutProjectionEvent::Abandoned],
        session_log: true,
    }
}

pub fn mark_abandoned(
    file: &Path,
    event: &str,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
) -> Result<CycleState> {
    // `#lazily-hot-path`: submit over the command plane when a controller is
    // live. The abandon `event` is descriptive, so it rides the command-plane
    // `reason` field (the typed event is `Abandoned`); the authority stamps it
    // back as the cycle's `last_event`, matching the local path.
    if submit_closeout_advance_via_socket(
        file,
        command_plane::CloseoutPhaseEvent::Abandoned,
        None,
        Some(event),
        snapshot_content,
        file_content,
        None,
        None,
    )? {
        let state = load(file)?.context(
            "closeout_advance Abandoned applied through the command plane but no cycle state was \
             read back",
        )?;
        append_phase_event_to_session_log(file, &state, file_content);
        return Ok(state);
    }
    let decision = decide_abandoned(load(file)?, file, event, snapshot_content, file_content);
    if decision.checkpoint {
        save(file, &decision.state)?;
    }
    for fact in &decision.facts {
        append_closeout_projection_event(file, &decision.state, *fact)?;
    }
    if decision.session_log {
        append_phase_event_to_session_log(file, &decision.state, file_content);
    }
    Ok(decision.state)
}

/// Restore the exact cycle that was incorrectly abandoned by the
/// `retire_superseded_captured_only_orphan` repair while a retained document
/// write was still in flight. This is deliberately narrower than an ordinary
/// abandoned-cycle reopen: both durable identities and the precise retirement
/// event must match.
pub fn reactivate_false_stale_capture_retirement(
    file: &Path,
    capture_id: &str,
    response_sha256: &str,
) -> Result<bool> {
    let retained_capture = load_projected_captured_response(file, capture_id)?;
    let Some(mut state) = load(file)? else {
        return Ok(false);
    };
    let retained_capture_matches = retained_capture.as_ref().is_some_and(|capture| {
        capture.cycle_id == state.cycle_id
            && capture.capture_id == capture_id
            && capture.response_sha256 == response_sha256
    });
    if state.phase == CyclePhase::ResponseCaptured
        && state.capture_id.as_deref() == Some(capture_id)
        && state.response_sha256.as_deref() == Some(response_sha256)
        && retained_capture_matches
    {
        reactivate_projected_captured_response(file, &state.cycle_id, capture_id)?;
        return Ok(true);
    }
    if state.phase != CyclePhase::Abandoned
        || state.last_event != "repair_retire_superseded_captured_only_orphan"
        || state.cycle_id != capture_id
        || state.capture_id.as_deref() != Some(capture_id)
        || state.response_sha256.as_deref() != Some(response_sha256)
        || !retained_capture_matches
    {
        return Ok(false);
    }

    state.phase = CyclePhase::ResponseCaptured;
    state.last_event = "session_check_reactivated_false_stale_capture_retirement".to_string();
    state.updated_at = now_secs();
    state.blocked_closeout = None;
    save(file, &state)?;
    append_closeout_projection_event(file, &state, CloseoutProjectionEvent::FalseStaleReactivated)?;
    reactivate_projected_captured_response(file, &state.cycle_id, capture_id)?;
    append_phase_event_to_session_log(file, &state, None);
    Ok(true)
}

fn append_phase_event_to_session_log(file: &Path, state: &CycleState, file_content: Option<&str>) {
    let Some(content) = file_content else {
        return;
    };
    let Ok((fm, _)) = agent_doc_frontmatter::frontmatter::parse(content) else {
        return;
    };
    let Some(session_id) = fm
        .session
        .as_deref()
        .map(str::trim)
        .filter(|id| !id.is_empty())
    else {
        return;
    };

    let mut event = format!(
        "document_cycle phase={} cycle={} event={}",
        state.phase.as_str(),
        state.cycle_id,
        state.last_event
    );
    if let Some(capture_id) = state.capture_id.as_deref() {
        event.push_str(&format!(" capture_id={capture_id}"));
    }
    let _ =
        agent_doc_supervisor_io::startup_miss::append_session_log_event(file, session_id, &event);
}

/// Build the `turn_intent_checkpoint` fact purely from `(document_hash,
/// checkpoint_sequence, state)`; `checkpoint_sequence` is the NEXT sequence
/// (caller passes current+1). Pure: no I/O. Shared by [`save`] and the
/// controller's in-process append+apply (`#lzdurablesink`), so the actor emits
/// the identical idempotency-keyed checkpoint fact in either process.
pub fn build_turn_intent_checkpoint_event(
    document_hash: &str,
    checkpoint_sequence: u64,
    state: &CycleState,
) -> Result<agent_doc_state_backbone::StateEvent> {
    let state_json = serde_json::to_string(state)?;
    let state_sha256 = agent_doc_hash::content_hash(&state_json);
    let event_id = format!(
        "turn-intent-checkpoint:{document_hash}:{}:{checkpoint_sequence}:{state_sha256}",
        state.cycle_id,
    );
    Ok(agent_doc_state_backbone::StateEvent::new(
        event_id,
        agent_doc_state_backbone::StateFact::TurnIntentCheckpointed {
            document_hash: document_hash.to_string(),
            cycle_id: state.cycle_id.clone(),
            checkpoint_sequence,
            state_sha256,
            state_json,
        },
    ))
}

fn save(file: &Path, state: &CycleState) -> Result<()> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(());
    };
    let checkpoint_sequence = load_closeout_projection(file)?
        .and_then(|projection| projection.turn_intent_checkpoint_sequence)
        .unwrap_or(0)
        .saturating_add(1);
    let event = build_turn_intent_checkpoint_event(&document_hash, checkpoint_sequence, state)?;
    append_state_fact(file, event.event_id, event.fact)?;
    Ok(())
}

#[cfg(feature = "test-support")]
#[doc(hidden)]
pub fn age_current_cycle_for_tests(file: &Path, age_secs: u64) -> Result<()> {
    let mut state = load(file)?.context("cycle state")?;
    state.started_at = state.started_at.saturating_sub(age_secs);
    state.updated_at = state.updated_at.saturating_sub(age_secs);
    save(file, &state)
}

#[derive(Debug, Clone, Copy)]
pub enum CloseoutProjectionEvent {
    PreflightStarted,
    ResponseCaptured,
    WriteApplied,
    Committed,
    Abandoned,
    FalseStaleReactivated,
}

/// Build the closeout phase fact for `event` purely from `(document_hash,
/// state)`, returning `None` when the event carries no emittable fact (e.g. a
/// capture transition without a capture id). Pure: no I/O. Shared by the
/// file-based [`append_closeout_projection_event`] and the controller's
/// in-process append+apply (`#lzdurablesink`), so the actor emits the identical
/// idempotency-keyed fact whether it runs in a client process or in the
/// controller.
pub fn build_closeout_projection_event(
    document_hash: &str,
    state: &CycleState,
    event: CloseoutProjectionEvent,
) -> Option<agent_doc_state_backbone::StateEvent> {
    let fact = match event {
        CloseoutProjectionEvent::PreflightStarted => {
            agent_doc_state_backbone::StateFact::PreflightStarted {
                document_hash: document_hash.to_string(),
                cycle_id: state.cycle_id.clone(),
                session_id: None,
                tracked_work_maintenance_required: state
                    .tracked_work_maintenance_required_at_preflight,
            }
        }
        CloseoutProjectionEvent::ResponseCaptured => {
            let capture_id = state.capture_id.clone()?;
            let response_sha256 = state.response_sha256.clone()?;
            agent_doc_state_backbone::StateFact::ResponseCaptured {
                document_hash: document_hash.to_string(),
                cycle_id: state.cycle_id.clone(),
                capture_id,
                response_sha256,
                response_body: None,
                intent_body: None,
                mutation_plan_json: None,
                file_hash: None,
                snapshot_hash: None,
                baseline_content: None,
            }
        }
        CloseoutProjectionEvent::WriteApplied => {
            agent_doc_state_backbone::StateFact::WriteApplied {
                document_hash: document_hash.to_string(),
                cycle_id: state.cycle_id.clone(),
                patch_id: None,
                file_hash: state.file_hash.clone(),
                snapshot_hash: state.snapshot_hash.clone(),
            }
        }
        CloseoutProjectionEvent::Committed => agent_doc_state_backbone::StateFact::CommitObserved {
            document_hash: document_hash.to_string(),
            cycle_id: state.cycle_id.clone(),
            commit: state
                .file_hash
                .as_ref()
                .map(|hash| format!("content:{hash}"))
                .unwrap_or_else(|| format!("event:{}", state.last_event)),
            file_hash: state.file_hash.clone(),
            snapshot_hash: state.snapshot_hash.clone(),
        },
        CloseoutProjectionEvent::Abandoned => agent_doc_state_backbone::StateFact::CycleAbandoned {
            document_hash: document_hash.to_string(),
            cycle_id: state.cycle_id.clone(),
            reason: state.last_event.clone(),
        },
        CloseoutProjectionEvent::FalseStaleReactivated => {
            let capture_id = state.capture_id.clone()?;
            let response_sha256 = state.response_sha256.clone()?;
            agent_doc_state_backbone::StateFact::FalseStaleCaptureReactivated {
                document_hash: document_hash.to_string(),
                cycle_id: state.cycle_id.clone(),
                capture_id,
                response_sha256,
                retirement_reason: "repair_retire_superseded_captured_only_orphan".to_string(),
            }
        }
    };
    let event_id = closeout_projection_event_id(document_hash, state, event);
    Some(agent_doc_state_backbone::StateEvent::new(event_id, fact))
}

fn append_closeout_projection_event(
    file: &Path,
    state: &CycleState,
    event: CloseoutProjectionEvent,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let Some(state_event) = build_closeout_projection_event(&document_hash, state, event) else {
        return Ok(false);
    };
    append_state_fact(file, state_event.event_id, state_event.fact)
}

pub struct CapturedResponseFactInput<'a> {
    pub cycle_id: &'a str,
    pub capture_id: &'a str,
    pub response_sha256: &'a str,
    pub response_body: &'a str,
    pub intent_body: Option<&'a str>,
    pub mutation_plan_json: Option<&'a str>,
    pub file_hash: Option<&'a str>,
    pub snapshot_hash: Option<&'a str>,
    pub baseline_content: Option<&'a str>,
}

pub fn append_response_captured_body(
    file: &Path,
    input: CapturedResponseFactInput<'_>,
) -> Result<bool> {
    let CapturedResponseFactInput {
        cycle_id,
        capture_id,
        response_sha256,
        response_body,
        intent_body,
        mutation_plan_json,
        file_hash,
        snapshot_hash,
        baseline_content,
    } = input;
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let projection_hash = agent_doc_hash::content_hash(&format!(
        "file={:?}\nsnapshot={:?}\nbaseline={:?}\nintent={:?}\nmutation_plan={:?}",
        file_hash,
        snapshot_hash,
        baseline_content,
        intent_body.map(agent_doc_hash::content_hash),
        mutation_plan_json.map(agent_doc_hash::content_hash),
    ));
    let event_id = format!(
        "closeout-response-captured-body:v3:{document_hash}:{cycle_id}:{capture_id}:{response_sha256}:{projection_hash}"
    );
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::ResponseCaptured {
            document_hash,
            cycle_id: cycle_id.to_string(),
            capture_id: capture_id.to_string(),
            response_sha256: response_sha256.to_string(),
            response_body: Some(response_body.to_string()),
            intent_body: intent_body.map(str::to_string),
            mutation_plan_json: mutation_plan_json.map(str::to_string),
            file_hash: file_hash.map(str::to_string),
            snapshot_hash: snapshot_hash.map(str::to_string),
            baseline_content: baseline_content.map(str::to_string),
        },
    )
}

/// Retire a captured-response ledger projection after another durable surface
/// (such as a compact archive referenced by HEAD) proves it remains materialized.
pub fn retire_projected_captured_response(
    file: &Path,
    cycle_id: &str,
    capture_id: &str,
    reason: &str,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let event_id = format!(
        "captured-response-retired:{document_hash}:{cycle_id}:{capture_id}:{}",
        agent_doc_hash::content_hash(reason)
    );
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::CapturedResponseRetired {
            document_hash,
            cycle_id: cycle_id.to_string(),
            capture_id: capture_id.to_string(),
            reason: reason.to_string(),
        },
    )
}

pub fn reactivate_projected_captured_response(
    file: &Path,
    cycle_id: &str,
    capture_id: &str,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    append_state_fact(
        file,
        format!("captured-response-reactivated:{document_hash}:{cycle_id}:{capture_id}"),
        agent_doc_state_backbone::StateFact::CapturedResponseReactivated {
            document_hash,
            cycle_id: cycle_id.to_string(),
            capture_id: capture_id.to_string(),
        },
    )
}

pub fn checkpoint_response_draft(
    file: &Path,
    cycle_id: &str,
    checkpoint_id: &str,
    checkpoint_count: u64,
    response_sha256: &str,
    response_body: &str,
    file_hash: Option<&str>,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let event_id =
        format!("response-draft:{document_hash}:{cycle_id}:{checkpoint_count}:{response_sha256}");
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::ResponseDraftCheckpointed {
            document_hash,
            cycle_id: cycle_id.to_string(),
            checkpoint_id: checkpoint_id.to_string(),
            checkpoint_count,
            response_sha256: response_sha256.to_string(),
            response_body: response_body.to_string(),
            file_hash: file_hash.map(str::to_string),
        },
    )
}

pub fn clear_response_draft(file: &Path, cycle_id: &str, reason: &str) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let event_id = format!(
        "response-draft-cleared:{document_hash}:{cycle_id}:{}",
        agent_doc_hash::content_hash(reason)
    );
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::ResponseDraftCleared {
            document_hash,
            cycle_id: cycle_id.to_string(),
            reason: reason.to_string(),
        },
    )
}

pub fn record_response_replay(file: &Path, cycle_id: &str, capture_id: &str) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    append_state_fact(
        file,
        format!("response-replayed:{document_hash}:{cycle_id}:{capture_id}"),
        agent_doc_state_backbone::StateFact::ResponseReplayObserved {
            document_hash,
            cycle_id: cycle_id.to_string(),
            capture_id: capture_id.to_string(),
        },
    )
}

/// Persist the atomic assistant-response CRDT operation and advance closeout to
/// `write_applied` in one idempotent backbone fact.
pub fn append_response_cell_added(
    file: &Path,
    cycle_id: &str,
    operation_id: &str,
    cell_id: &str,
    response_sha256: &str,
    content_hash: &str,
    applied: bool,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let event_id =
        format!("response-cell-added:{document_hash}:{cycle_id}:{operation_id}:{cell_id}");
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::ResponseCellAdded {
            document_hash,
            cycle_id: cycle_id.to_string(),
            operation_id: operation_id.to_string(),
            cell_id: cell_id.to_string(),
            response_sha256: response_sha256.to_string(),
            content_hash: content_hash.to_string(),
            applied,
        },
    )
}

fn append_state_fact(
    file: &Path,
    event_id: String,
    fact: agent_doc_state_backbone::StateFact,
) -> Result<bool> {
    let canonical = file.canonicalize().unwrap_or_else(|_| file.to_path_buf());
    let project_root = agent_doc_project_root_io::project_root_or_file_parent(&canonical)?;
    let fact_label = fact.label();
    let event = agent_doc_state_backbone::StateEvent::new(event_id, fact);
    let payload_json = serde_json::to_string(&event).context("serialize closeout state event")?;
    let controller_socket = agent_doc_controller::paths::socket_path(&project_root);
    #[derive(serde::Deserialize)]
    struct ActorAppendEnvelope {
        ok: bool,
        data: Option<bool>,
        error: Option<String>,
    }

    let request = serde_json::json!({
        "command": "state_event_append",
        "file": canonical,
        "caller": "cycle_state",
        "diagnostic_payload": payload_json.clone(),
    });
    match agent_doc_state_wire::send_ndjson_request_to_actor(
        &controller_socket,
        &request,
        std::time::Duration::from_secs(5),
    ) {
        Ok(raw) => {
            let envelope: ActorAppendEnvelope =
                serde_json::from_str(&raw).context("decode Lazily state append response")?;
            if !envelope.ok {
                anyhow::bail!(
                    "Lazily state actor rejected closeout fact append: {}",
                    envelope.error.as_deref().unwrap_or("unknown error")
                );
            }
            return envelope
                .data
                .context("Lazily state actor returned no append result");
        }
        Err(agent_doc_state_wire::ActorRequestError::Connect(_)) if !controller_socket.exists() => {
        }
        Err(err) => {
            return Err(err).context("append closeout state event through Lazily document actor");
        }
    }

    // Cold-start compatibility: before a project controller exists, seed the
    // durable ledger so the actor can hydrate it on launch. Once the actor socket
    // exists, failures above fail closed instead of bypassing actor authority.
    let conn = agent_doc_sqlite::state_store::open_state_db(&project_root)?;
    agent_doc_sqlite::state_store::insert_state_event_in_db(
        &conn,
        &agent_doc_sqlite::state_store::StateEventInsert {
            event_id: &event.event_id,
            document_hash: event.document_hash(),
            domain: event.domain().label(),
            fact_type: fact_label,
            payload_json: &payload_json,
        },
    )
}

/// Submit a `closeout_advance` command to the controller authority over the
/// lazily command plane (`#lzdurablesink`, `command-plane-v1`). When a controller
/// is live, the transition is decided from the live in-memory projection — not
/// this client's cold `state.db` read — and the phase fact(s) are sunk in the
/// controller. This is the `#lazily-hot-path` authority rule: a hot-path
/// transition must not reload storage to arbitrate itself.
///
/// Returns:
/// - `Ok(true)` — a controller was live and the terminal `CausalReceipt` was
///   `applied` (an idempotent no-op at the current phase folds as `applied`).
/// - `Ok(false)` — no controller socket exists (actorless bootstrap / cold
///   start); the caller must use the local `load→decide→save→append` fallback.
/// - `Err(_)` — the controller was live but the transport/decode failed or the
///   terminal receipt was `rejected` (fail closed; never silently fall back).
#[allow(clippy::too_many_arguments)]
fn submit_closeout_advance_via_socket(
    file: &Path,
    event: command_plane::CloseoutPhaseEvent,
    event_label: Option<&str>,
    reason: Option<&str>,
    snapshot_content: Option<&str>,
    file_content: Option<&str>,
    response_sha256: Option<&str>,
    cycle_id_hint: Option<&str>,
) -> Result<bool> {
    use command_plane::{CloseoutAdvancePayload, build_closeout_advance_submit};

    let canonical = file.canonicalize().unwrap_or_else(|_| file.to_path_buf());
    let project_root = agent_doc_project_root_io::project_root_or_file_parent(&canonical)?;
    let controller_socket = agent_doc_controller::paths::socket_path(&project_root);
    if !controller_socket.exists() {
        return Ok(false);
    }
    let document_hash = agent_doc_hash::document_id_for_path(&canonical);
    let payload = CloseoutAdvancePayload {
        document_path: canonical.to_string_lossy().to_string(),
        event,
        event_label: event_label.map(str::to_string),
        reason: reason.map(str::to_string),
        snapshot_content: snapshot_content.map(str::to_string),
        file_content: file_content.map(str::to_string),
        response_sha256: response_sha256.map(str::to_string),
        cycle_id_hint: cycle_id_hint.map(str::to_string),
    };
    // Stable, replay-safe command id / idempotency key derived from
    // (document, event, content) so a duplicate advance dedupes onto the same
    // command — mirroring the durable sink's idempotent re-delivery.
    let event_label = payload.last_event_label();
    let content_key = format!(
        "{}:{}:{}",
        snapshot_content.unwrap_or(""),
        file_content.unwrap_or(""),
        response_sha256.unwrap_or("")
    );
    let command_id = format!(
        "closeout-advance:{document_hash}:{event_label}:{}",
        agent_doc_hash::content_hash(&content_key)
    );
    let submit =
        build_closeout_advance_submit(&command_id, "cycle_state", &command_id, 0, payload)?;
    let payload_json = serde_json::to_string(&submit).context("encode command_plane_submit")?;
    let request = serde_json::json!({
        "command": "command_plane_submit",
        "file": canonical,
        "caller": "cycle_state",
        "diagnostic_payload": payload_json,
    });
    #[derive(serde::Deserialize)]
    #[allow(dead_code)]
    struct ReceiptEnvelope {
        ok: bool,
        data: Option<lazily::CausalReceipt>,
        error: Option<String>,
    }
    let raw = agent_doc_state_wire::send_ndjson_request_to_actor(
        &controller_socket,
        &request,
        std::time::Duration::from_secs(5),
    )
    .context("submit closeout_advance command through the Lazily controller")?;
    let envelope: ReceiptEnvelope =
        serde_json::from_str(&raw).context("decode closeout_advance command-plane response")?;
    if !envelope.ok {
        anyhow::bail!(
            "closeout_advance command rejected by controller: {}",
            envelope.error.as_deref().unwrap_or("unknown error")
        );
    }
    let receipt = envelope
        .data
        .context("closeout_advance command returned no terminal receipt")?;
    if receipt.outcome != lazily::ReceiptOutcome::Applied {
        anyhow::bail!(
            "closeout_advance command not applied: {:?}{}",
            receipt.outcome,
            receipt
                .reason
                .as_deref()
                .map(|r| format!(" ({r})"))
                .unwrap_or_default()
        );
    }
    Ok(true)
}

fn append_semantic_merge_conflict_advisory_recorded_event(
    file: &Path,
    cycle_id: &str,
    advisory: &SemanticMergeConflictAdvisory,
) -> Result<bool> {
    let Some(document_hash) = cycle_document_hash(file)? else {
        return Ok(false);
    };
    let event_id = format!(
        "semantic-merge-conflict-observed:{document_hash}:{cycle_id}:{}:{}:{}",
        advisory.component, advisory.id, advisory.reason
    );
    append_state_fact(
        file,
        event_id,
        agent_doc_state_backbone::StateFact::DocumentCellMergeConflictObserved {
            document_hash,
            cycle_id: cycle_id.to_string(),
            component: advisory.component.clone(),
            id: advisory.id.clone(),
            reason: advisory.reason.clone(),
            detail: advisory.detail.clone(),
        },
    )
}

/// Resolve the ledger key for `file`, or `None` when the document has no
/// resolvable path at all.
///
/// `#ay80ci`: this used to additionally require an existing `.agent-doc`
/// ancestor and silently return `None` otherwise, which made every cycle-state
/// write a no-op for a document whose project root had not been materialized
/// yet. That is not what the surrounding code does: [`append_state_fact`]
/// resolves its destination with `project_root_or_file_parent` (which falls
/// back to the document's parent) and `open_state_db` creates `.agent-doc`
/// there, and the reader
/// (`agent_doc_repair_io::load_pending_response_state`) resolves the same way.
/// So the guard did not prevent the directory from being created — it only made
/// the write side disagree with the read side, and it swallowed the write.
///
/// The observable failure was CI-only because whether `.agent-doc` already
/// exists by capture time depends on the environment: locally a global config
/// or a live controller materializes the root earlier in the run, so the guard
/// passed. On a clean runner nothing did, so `finalize` captured a response,
/// silently dropped the durability write, then failed reading it back with
/// `captured turn intent <id> was not projected as active`.
fn cycle_document_hash(file: &Path) -> Result<Option<String>> {
    let canonical = match file.canonicalize() {
        Ok(path) => path,
        Err(_) => return Ok(None),
    };
    Ok(Some(agent_doc_fs::document_state_hash(&canonical)?))
}

fn closeout_projection_event_id(
    document_hash: &str,
    state: &CycleState,
    event: CloseoutProjectionEvent,
) -> String {
    match event {
        CloseoutProjectionEvent::PreflightStarted => {
            format!(
                "closeout-preflight-started:{document_hash}:{}",
                state.cycle_id
            )
        }
        CloseoutProjectionEvent::ResponseCaptured => format!(
            "closeout-response-captured:{document_hash}:{}:{}",
            state.cycle_id,
            state.response_sha256.as_deref().unwrap_or("missing")
        ),
        CloseoutProjectionEvent::WriteApplied => format!(
            "closeout-write-applied:{document_hash}:{}:{}",
            state.cycle_id,
            state.file_hash.as_deref().unwrap_or("missing")
        ),
        CloseoutProjectionEvent::Committed => format!(
            "closeout-committed:{document_hash}:{}:{}",
            state.cycle_id,
            state.file_hash.as_deref().unwrap_or("missing")
        ),
        CloseoutProjectionEvent::Abandoned => format!(
            "closeout-abandoned:{document_hash}:{}:{}",
            state.cycle_id, state.last_event
        ),
        CloseoutProjectionEvent::FalseStaleReactivated => format!(
            "closeout-false-stale-reactivated:{document_hash}:{}:{}",
            state.cycle_id,
            state.response_sha256.as_deref().unwrap_or("missing")
        ),
    }
}

fn synthetic_state(file: &Path, phase: CyclePhase) -> CycleState {
    synthetic_state_with_id(file, phase, None)
}

fn synthetic_state_with_id(
    file: &Path,
    phase: CyclePhase,
    cycle_id_hint: Option<&str>,
) -> CycleState {
    let now = now_secs();
    CycleState {
        cycle_id: cycle_id_hint
            .map(|s| s.to_string())
            .unwrap_or_else(|| format!("synthetic-{}", now_millis())),
        file: file.display().to_string(),
        phase,
        last_event: "synthetic_state".to_string(),
        started_at: now,
        updated_at: now,
        snapshot_hash: None,
        file_hash: None,
        normalized_snapshot_hash: None,
        normalized_file_hash: None,
        capture_id: None,
        response_sha256: None,
        had_pending_mutations: false,
        requires_backlog_capture: false,
        required_backlog_targets: Vec::new(),
        required_explicit_backlog_item_count: 0,
        required_plan_reference_count: 0,
        prompt_targets: Vec::new(),
        queue_task_id: None,
        turn_id: None,
        recycle_resume_consumed: false,
        pending_done_ids: Vec::new(),
        pending_only_commit_target_hash: None,
        pending_kept_open_ids: Vec::new(),
        reaped_pending_ids: Vec::new(),
        expect_done_or_gate_ids: Vec::new(),
        pending_gated_ids: Vec::new(),
        pending_added_this_cycle: false,
        pending_added_ids: Vec::new(),
        requested_done_ids: Vec::new(),
        requested_added_ids: Vec::new(),
        pending_actionable_ids: Vec::new(),
        pending_anchored_ids: Vec::new(),
        // A synthetic cycle did not observe a preflight document. Preserve
        // that distinction instead of asserting a false preflight fact that
        // the state projection cannot reproduce.
        tracked_work_maintenance_required_at_preflight: None,
        ipc_snapshot_adoption_blocked: false,
        dropped_exchange_prompts: Vec::new(),
        dropped_queue_prompts: Vec::new(),
        active_queue_heads: Vec::new(),
        active_free_text_queue_heads: Vec::new(),
        semantic_merge_conflict_advisories: Vec::new(),
        blocked_closeout: None,
        skipped_queue_head_ids: Vec::new(),
    }
}

fn is_zero(value: &usize) -> bool {
    *value == 0
}

fn is_false(value: &bool) -> bool {
    !*value
}

fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

fn now_millis() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
}

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

    fn setup_project() -> tempfile::TempDir {
        let dir = tempfile::TempDir::new().unwrap();
        fs::create_dir_all(dir.path().join(".agent-doc/snapshots")).unwrap();
        dir
    }

    #[test]
    fn open_stalled_resolves_only_abandoned_older_turns() {
        // `#suprecyclespin`: the recycle-defer gate must distinguish a live cycle
        // (keep deferring) from a stalled/superseded older turn (force-close).
        let mut state = synthetic_state(Path::new("/tmp/doc.md"), CyclePhase::PreflightStarted);
        let deadline = STALLED_CYCLE_RESOLVE_SECS;
        // updated_at far in the past, no IPC inflight → stalled (resolvable).
        state.updated_at = 1_000;
        assert!(state.open_stalled(0, 1_000 + deadline + 1, deadline));
        assert!(state.stalled_before_response_capture_cycle(0, 1_000 + deadline + 1, deadline));
        // Exactly at the deadline is NOT yet stalled (strict `>`).
        assert!(!state.open_stalled(0, 1_000 + deadline, deadline));
        assert!(!state.stalled_before_response_capture_cycle(0, 1_000 + deadline, deadline));
        // Fresh `updated_at` → a live cycle, never force-closed.
        state.updated_at = 1_000 + deadline + 1;
        assert!(!state.open_stalled(0, 1_000 + deadline + 5, deadline));
        assert!(!state.stalled_before_response_capture_cycle(0, 1_000 + deadline + 5, deadline));
        // IPC advisory connection in flight → finalize is live, never force-closed even
        // if the cycle has been open a long time.
        state.updated_at = 1_000;
        assert!(!state.open_stalled(1, 1_000 + deadline + 100, deadline));
        assert!(!state.stalled_before_response_capture_cycle(1, 1_000 + deadline + 100, deadline));
        // Once a response is captured, the stalled open cycle is durable recovery
        // evidence. It may keep the recycle gate open until escalation, but must
        // not be abandoned as disposable state before response capture.
        let mut captured = synthetic_state(Path::new("/tmp/doc.md"), CyclePhase::ResponseCaptured);
        captured.updated_at = 1_000;
        captured.capture_id = Some("cycle-1".to_string());
        captured.response_sha256 = Some("abc".to_string());
        assert!(captured.open_stalled(0, 1_000 + deadline + 1, deadline));
        assert!(!captured.stalled_before_response_capture_cycle(0, 1_000 + deadline + 1, deadline));
        // A committed cycle is not open, so it is never "stalled".
        let committed = synthetic_state(Path::new("/tmp/doc.md"), CyclePhase::Committed);
        assert!(!committed.open_stalled(0, u64::MAX, deadline));
        assert!(!committed.stalled_before_response_capture_cycle(0, u64::MAX, deadline));
    }

    #[test]
    fn stalled_resolve_deadline_survives_slow_first_response() {
        // `#suprecyclespin-falseabandon` regression: a live preflight cycle whose
        // harness is still generating its first response sits untouched with
        // `inflight == 0` and an unadvanced `updated_at`. At an observed 46s the
        // old 45s deadline abandoned this live turn; the bumped deadline must
        // clear normal first-response latency.
        let mut state = synthetic_state(Path::new("/tmp/doc.md"), CyclePhase::PreflightStarted);
        state.updated_at = 1_000;
        let incident_stalled_secs = 46; // observed stalled_secs that falsely abandoned a live turn
        assert!(
            !state.stalled_before_response_capture_cycle(
                0,
                1_000 + incident_stalled_secs,
                STALLED_CYCLE_RESOLVE_SECS,
            ),
            "a 46s-stale live preflight cycle must not be resolvable after the deadline bump"
        );
        // Still bounded: a genuinely orphaned cycle past the (larger) deadline
        // remains resolvable so the recycle spin cannot wedge forever.
        assert!(state.stalled_before_response_capture_cycle(
            0,
            1_000 + STALLED_CYCLE_RESOLVE_SECS + 1,
            STALLED_CYCLE_RESOLVE_SECS,
        ));
        // The consecutive-tick debounce is a second, independent gate on top of
        // the deadline — a stale observation cannot abandon a live turn on a
        // single transiently-misread boundary poll.
        const { assert!(STALLED_CYCLE_RESOLVE_CONFIRM_TICKS > 0) };
    }

    #[test]
    fn save_persists_ledger_checkpoint() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let _ = start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = mark_committed(&doc, "evt", Some("snap"), Some("body")).unwrap();
        assert_eq!(state.phase, CyclePhase::Committed);
        assert_eq!(load(&doc).unwrap().unwrap().phase, CyclePhase::Committed);
    }

    #[test]
    fn start_preflight_persists_open_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let state = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        assert_eq!(state.phase, CyclePhase::PreflightStarted);
        assert_eq!(
            load(&doc).unwrap().unwrap().phase,
            CyclePhase::PreflightStarted
        );
        assert!(load(&doc).unwrap().unwrap().is_open());
    }

    #[test]
    fn admit_with_current_resolver_opens_cycle_without_preflight_maintenance() {
        let dir = setup_project();
        let doc = dir.path().join("session.md");
        let original = "# Session\n\nOperator prompt.\n";
        fs::write(&doc, original).unwrap();
        let mut logs = Vec::new();

        let output = admit_with_current_resolver(
            &doc,
            |file| std::fs::read_to_string(file).context("test resolver should read document"),
            |_file| Ok(None),
            |file, message| logs.push((file.display().to_string(), message.to_string())),
        )
        .unwrap();

        assert!(output.admitted);
        assert_eq!(output.source, "admit");
        assert!(!output.maintenance_required);
        assert!(!output.preflight_required);
        assert_eq!(output.cycle_phase, "preflight_started");
        assert_eq!(fs::read_to_string(&doc).unwrap(), original);

        let state = load(&doc).unwrap().unwrap();
        assert_eq!(state.cycle_id, output.cycle_id);
        assert_eq!(state.phase, CyclePhase::PreflightStarted);
        assert_eq!(state.last_event, "preflight_started");

        assert_eq!(logs.len(), 1);
        let log = &logs[0].1;
        assert!(log.contains("realtime_admit"), "admission log:\n{log}");
        assert!(
            log.contains("maintenance_required=false"),
            "admission log:\n{log}"
        );
        assert!(
            log.contains("preflight_required=false"),
            "admission log:\n{log}"
        );
        assert!(
            !log.contains("preflight_diff_start"),
            "admit must not run preflight diff start:\n{log}"
        );
        assert!(
            !log.contains("deprecated_queue_active_line_dropped"),
            "admit must not run queue maintenance:\n{log}"
        );
        assert!(
            !log.contains("layout repair"),
            "admit must not run preflight layout repair:\n{log}"
        );
    }

    fn advisory(
        component: &str,
        id: &str,
        reason: agent_doc_merge::document_cell_merge::MergeConflictReason,
    ) -> agent_doc_merge::document_cell_merge::MergeConflictAdvisory {
        agent_doc_merge::document_cell_merge::MergeConflictAdvisory {
            component: component.to_string(),
            id: id.to_string(),
            reason,
            detail: format!("{component}:{id} detail"),
        }
    }

    #[test]
    fn record_semantic_merge_conflict_advisories_tags_current_cycle_and_dedupes() {
        use agent_doc_merge::document_cell_merge::MergeConflictReason;
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        record_semantic_merge_conflict_advisories(
            &doc,
            &[advisory(
                "exchange",
                "a",
                MergeConflictReason::SameNodeOperatorOverride,
            )],
        )
        .unwrap();
        // Re-recording the same (component, id, reason) is a no-op.
        record_semantic_merge_conflict_advisories(
            &doc,
            &[advisory(
                "exchange",
                "a",
                MergeConflictReason::SameNodeOperatorOverride,
            )],
        )
        .unwrap();

        let state = load(&doc).unwrap().unwrap();
        assert_eq!(state.semantic_merge_conflict_advisories.len(), 1);
        let recorded = &state.semantic_merge_conflict_advisories[0];
        assert_eq!(recorded.component, "exchange");
        assert_eq!(recorded.id, "a");
        assert_eq!(recorded.reason, "same_node_operator_override");
        assert_eq!(
            recorded.recorded_cycle_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
    }

    #[test]
    fn a_reentrant_preflight_keeps_the_turn_identity_and_refreshes_the_contract() {
        use agent_doc_merge::document_cell_merge::MergeConflictReason;
        // #preflightinbinary: with preflight running in the binary on the
        // trigger prompt, a second preflight can reach the same turn (an agent
        // or a hookless harness also runs one). It used to mint
        // `cycle-{now_millis}` over the first, silently discarding its identity
        // and accumulators. Re-entry must keep the identity — and must still
        // report facts recorded since the turn opened, or the agent acts on a
        // stale contract.
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let first = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        record_semantic_merge_conflict_advisories(
            &doc,
            &[advisory(
                "exchange",
                "a",
                MergeConflictReason::SameNodeOperatorOverride,
            )],
        )
        .unwrap();

        let reentrant = start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        assert_eq!(
            reentrant.cycle_id, first.cycle_id,
            "a re-entrant preflight must not open a second cycle over the open turn"
        );
        assert_eq!(reentrant.started_at, first.started_at);
        assert_eq!(
            reentrant.semantic_merge_conflict_advisories.len(),
            1,
            "the refreshed contract must carry facts recorded since the turn opened"
        );
    }

    #[test]
    fn a_committed_turn_still_opens_a_new_cycle() {
        // The boundary that must survive: re-entrancy must not merge two real
        // turns into one.
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let first = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_committed(&doc, "commit_success", Some("snap"), Some("body")).unwrap();
        let second = start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        assert_ne!(
            second.cycle_id, first.cycle_id,
            "a committed turn is a real boundary"
        );
    }

    #[test]
    fn semantic_merge_conflict_advisories_are_cycle_local_reactive_state() {
        use agent_doc_merge::document_cell_merge::MergeConflictReason;
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        record_semantic_merge_conflict_advisories(
            &doc,
            &[advisory(
                "exchange",
                "a",
                MergeConflictReason::SameNodeOperatorOverride,
            )],
        )
        .unwrap();
        let current = load(&doc).unwrap().unwrap();
        assert_eq!(current.semantic_merge_conflict_advisories.len(), 1);

        mark_committed(&doc, "commit_success", Some("snap"), Some("body")).unwrap();
        let next = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        assert!(
            next.semantic_merge_conflict_advisories.is_empty(),
            "a conflict observation must not become a future-turn courtesy request"
        );
    }

    #[test]
    fn mark_write_applied_advances_existing_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = mark_write_applied(&doc, "write_template", Some("new"), Some("new")).unwrap();
        assert_eq!(state.phase, CyclePhase::WriteApplied);
        assert_eq!(state.last_event, "write_template");
        assert!(state.snapshot_hash.is_some());
    }

    #[test]
    fn record_ipc_snapshot_adoption_blocked_sets_cycle_flag() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_ipc_snapshot_adoption_blocked(&doc)
            .unwrap()
            .expect("state should exist");

        assert!(state.ipc_snapshot_adoption_blocked);
        assert!(
            load(&doc)
                .unwrap()
                .expect("state should persist")
                .ipc_snapshot_adoption_blocked
        );
    }

    #[test]
    fn record_and_clear_dropped_exchange_prompts() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        record_dropped_exchange_prompts(&doc, &["go".to_string(), "go".to_string()]).unwrap();
        record_dropped_exchange_prompts(&doc, &["go".to_string(), "do #x".to_string()]).unwrap();
        let state = load(&doc).unwrap().expect("state");
        // De-duplicated across calls.
        assert_eq!(state.dropped_exchange_prompts, vec!["go", "do #x"]);

        clear_dropped_exchange_prompts(&doc).unwrap();
        assert!(
            load(&doc)
                .unwrap()
                .expect("state")
                .dropped_exchange_prompts
                .is_empty()
        );
    }

    #[test]
    fn mark_response_captured_sets_capture_metadata() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "abc",
            None,
        )
        .unwrap();
        assert_eq!(state.phase, CyclePhase::ResponseCaptured);
        assert_eq!(state.capture_id.as_deref(), Some(state.cycle_id.as_str()));
        assert_eq!(state.response_sha256.as_deref(), Some("abc"));
    }

    #[test]
    fn mark_pending_mutations_sets_flag() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = mark_pending_mutations(&doc).unwrap().unwrap();
        assert!(state.had_pending_mutations);
        assert!(load(&doc).unwrap().unwrap().had_pending_mutations);
    }

    #[test]
    fn record_pending_done_ids_persists_normalized_ids() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_pending_done_ids(
            &doc,
            &["#AbC1".to_string(), "abc1".to_string(), "z9".to_string()],
        )
        .unwrap()
        .unwrap();

        assert_eq!(
            state.pending_done_ids,
            vec!["abc1".to_string(), "z9".to_string()]
        );
        assert_eq!(
            load(&doc).unwrap().unwrap().pending_done_ids,
            vec!["abc1".to_string(), "z9".to_string()]
        );
    }

    #[test]
    fn clearing_pending_only_commit_intent_retires_done_ids_with_target() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        record_pending_done_ids(&doc, &["done1".to_string()]).unwrap();
        record_pending_only_commit_target(&doc, "target-hash").unwrap();

        assert!(clear_pending_only_commit_intent(&doc, "target-hash").unwrap());
        let state = load(&doc).unwrap().unwrap();
        assert_eq!(state.pending_only_commit_target_hash, None);
        assert!(state.pending_done_ids.is_empty());
    }

    #[test]
    fn record_pending_kept_open_ids_persists_normalized_ids() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_pending_kept_open_ids(
            &doc,
            &["#AbC1".to_string(), "abc1".to_string(), "z9".to_string()],
        )
        .unwrap()
        .unwrap();

        assert_eq!(
            state.pending_kept_open_ids,
            vec!["abc1".to_string(), "z9".to_string()]
        );
        assert_eq!(
            load(&doc).unwrap().unwrap().pending_kept_open_ids,
            vec!["abc1".to_string(), "z9".to_string()]
        );
    }

    #[test]
    fn record_pending_actionable_ids_is_normalized_and_distinct_from_add_proof() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_pending_actionable_ids(
            &doc,
            &[
                "#PhaseB".to_string(),
                "phaseb".to_string(),
                "phasea".to_string(),
            ],
        )
        .unwrap()
        .unwrap();

        assert_eq!(
            state.pending_actionable_ids,
            vec!["phaseb".to_string(), "phasea".to_string()]
        );
        assert!(
            state.pending_added_ids.is_empty(),
            "ungated queue candidates are actionable without pretending they were newly added"
        );
        assert_eq!(
            pending_actionable_ids(&doc),
            std::collections::HashSet::from(["phasea".to_string(), "phaseb".to_string()])
        );
    }

    #[test]
    fn record_reaped_pending_ids_persists_normalized_ids() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_reaped_pending_ids(
            &doc,
            &["#AbC1".to_string(), "abc1".to_string(), "z9".to_string()],
        )
        .unwrap()
        .unwrap();

        assert_eq!(
            state.reaped_pending_ids,
            vec!["abc1".to_string(), "z9".to_string()]
        );
        assert_eq!(
            load(&doc).unwrap().unwrap().reaped_pending_ids,
            vec!["abc1".to_string(), "z9".to_string()]
        );
    }

    #[test]
    fn record_backlog_capture_requirement_sets_flag() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_backlog_capture_requirement(&doc, true)
            .unwrap()
            .unwrap();
        assert!(state.requires_backlog_capture);
        assert!(load(&doc).unwrap().unwrap().requires_backlog_capture);
    }

    #[test]
    fn record_backlog_target_requirements_persists_targets() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let requirements = vec![BacklogTargetRequirement {
            path: dir.path().join("tasks/bugs.md").display().to_string(),
            component: Some("backlog".to_string()),
            baseline_hash: Some("abc".to_string()),
            baseline_item_ids: vec!["bug1".to_string()],
        }];

        let state = record_backlog_target_requirements(&doc, &requirements)
            .unwrap()
            .unwrap();
        assert_eq!(state.required_backlog_targets, requirements);
        assert_eq!(
            load(&doc).unwrap().unwrap().required_backlog_targets,
            requirements
        );
    }

    #[test]
    fn record_required_explicit_backlog_item_count_persists_count() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_required_explicit_backlog_item_count(&doc, 3)
            .unwrap()
            .unwrap();
        assert_eq!(state.required_explicit_backlog_item_count, 3);
        assert_eq!(
            load(&doc)
                .unwrap()
                .unwrap()
                .required_explicit_backlog_item_count,
            3
        );
    }

    #[test]
    fn record_required_plan_reference_count_persists_count() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state = record_required_plan_reference_count(&doc, 2)
            .unwrap()
            .unwrap();
        assert_eq!(state.required_plan_reference_count, 2);
        assert_eq!(
            load(&doc).unwrap().unwrap().required_plan_reference_count,
            2
        );
    }

    #[test]
    fn mark_committed_closes_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_write_applied(&doc, "write_template", Some("new"), Some("new")).unwrap();

        let state = mark_committed(&doc, "commit", Some("new"), Some("new")).unwrap();
        assert_eq!(state.phase, CyclePhase::Committed);
        assert!(!state.is_open());
    }

    #[test]
    fn cycle_transitions_feed_state_backbone_closeout_projection() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "response-sha",
            None,
        )
        .unwrap();
        mark_write_applied(&doc, "write_applied", Some("written"), Some("written")).unwrap();
        mark_committed(&doc, "commit_success", Some("written"), Some("written")).unwrap();

        let conn = agent_doc_sqlite::state_store::open_state_db(dir.path()).unwrap();
        let mut ledger = agent_doc_state_backbone::EventLedger::new();
        for row in agent_doc_sqlite::state_store::load_state_events_from_db(&conn, None).unwrap() {
            let event: agent_doc_state_backbone::StateEvent =
                serde_json::from_str(&row.payload_json).unwrap();
            ledger.append(event);
        }
        let document_hash = agent_doc_fs::document_state_hash(&doc).unwrap();
        let closeout = ledger
            .project_document(&document_hash)
            .expect("cycle transition events should project document state")
            .closeout;
        assert_eq!(
            closeout.cycle_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
        assert_eq!(closeout.phase, Some(CyclePhase::Committed));
        assert_eq!(
            closeout.capture_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
        assert_eq!(closeout.response_sha256.as_deref(), Some("response-sha"));
        assert!(
            closeout
                .commit
                .as_deref()
                .is_some_and(|commit| commit.starts_with("content:"))
        );
    }

    #[test]
    fn preflight_tracks_pending_maintenance_need_in_closeout_projection() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        let content = concat!(
            "<!-- agent:backlog -->\n",
            "- [x] [#done1] Reap me\n",
            "- [ ] [#keep1] Keep me\n",
            "<!-- /agent:backlog -->\n",
        );
        fs::write(&doc, content).unwrap();

        let started = start_preflight(&doc, Some(content), Some(content)).unwrap();
        assert_eq!(
            started.tracked_work_maintenance_required_at_preflight,
            Some(true)
        );

        let projected = load_closeout_projection(&doc)
            .unwrap()
            .expect("preflight should feed closeout projection");
        assert_eq!(projected.tracked_work_maintenance_required, Some(true));

        let loaded = load_with_closeout_projection(&doc)
            .unwrap()
            .expect("cycle state should load");
        assert_eq!(
            loaded.tracked_work_maintenance_required_at_preflight,
            Some(true)
        );
    }

    #[test]
    fn response_capture_body_feeds_closeout_projection() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "response-sha",
            None,
        )
        .unwrap();

        append_response_captured_body(
            &doc,
            CapturedResponseFactInput {
                cycle_id: &started.cycle_id,
                capture_id: &started.cycle_id,
                response_sha256: "response-sha",
                response_body: "### Re: topic - gpt-5\n\nDone.\n",
                intent_body: None,
                mutation_plan_json: None,
                file_hash: Some("file-sha"),
                snapshot_hash: Some("snapshot-sha"),
                baseline_content: Some("body"),
            },
        )
        .unwrap();

        let projected = load_projected_captured_response(&doc, &started.cycle_id)
            .unwrap()
            .expect("captured response projection");
        assert_eq!(projected.cycle_id, started.cycle_id);
        assert_eq!(projected.capture_id, started.cycle_id);
        assert_eq!(projected.response_sha256, "response-sha");
        assert_eq!(projected.file_hash.as_deref(), Some("file-sha"));
        assert_eq!(projected.snapshot_hash.as_deref(), Some("snapshot-sha"));
        assert_eq!(projected.baseline_content.as_deref(), Some("body"));
        assert_eq!(projected.response_body, "### Re: topic - gpt-5\n\nDone.\n");
    }

    #[test]
    fn open_cycle_adopts_newest_same_cycle_capture_identity() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "old-response-sha",
            None,
        )
        .unwrap();
        append_response_captured_body(
            &doc,
            CapturedResponseFactInput {
                cycle_id: &started.cycle_id,
                capture_id: &started.cycle_id,
                response_sha256: "old-response-sha",
                response_body: "### Re: topic - gpt-5\n\nDone.\n",
                intent_body: None,
                mutation_plan_json: None,
                file_hash: Some("file-sha"),
                snapshot_hash: Some("snapshot-sha"),
                baseline_content: Some("body"),
            },
        )
        .unwrap();
        mark_write_applied(&doc, "write_template", Some("new"), Some("new")).unwrap();

        append_response_captured_body(
            &doc,
            CapturedResponseFactInput {
                cycle_id: &started.cycle_id,
                capture_id: &started.cycle_id,
                response_sha256: "new-response-sha",
                response_body: "### Re: topic - gpt-5\n\nCorrected.\n",
                intent_body: None,
                mutation_plan_json: None,
                file_hash: Some("file-sha"),
                snapshot_hash: Some("snapshot-sha"),
                baseline_content: Some("body"),
            },
        )
        .unwrap();

        let loaded = load_with_closeout_projection(&doc)
            .unwrap()
            .expect("open cycle should load");
        assert_eq!(loaded.phase, CyclePhase::WriteApplied);
        assert_eq!(
            loaded.capture_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
        assert_eq!(loaded.response_sha256.as_deref(), Some("new-response-sha"));
        let projected = load_projected_captured_response(&doc, &started.cycle_id)
            .unwrap()
            .expect("newest capture projection");
        assert_eq!(projected.response_sha256, "new-response-sha");
        assert!(projected.response_body.contains("Corrected."));
    }

    #[test]
    fn recent_response_capture_history_is_bounded_and_newest_first() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        for (response_sha256, response_body, baseline) in [
            ("response-1", "first response", "baseline-1"),
            ("response-2", "second response", "baseline-2"),
            ("response-3", "third response", "baseline-3"),
        ] {
            append_response_captured_body(
                &doc,
                CapturedResponseFactInput {
                    cycle_id: &started.cycle_id,
                    capture_id: &started.cycle_id,
                    response_sha256,
                    response_body,
                    intent_body: None,
                    mutation_plan_json: None,
                    file_hash: Some(response_sha256),
                    snapshot_hash: None,
                    baseline_content: Some(baseline),
                },
            )
            .unwrap();
        }

        let history = load_recent_captured_response_checkpoints(&doc, 2).unwrap();
        assert_eq!(history.len(), 2);
        assert_eq!(history[0].response_sha256, "response-3");
        assert_eq!(history[0].baseline_content.as_deref(), Some("baseline-3"));
        assert_eq!(history[1].response_sha256, "response-2");
        assert!(history[0].sequence > history[1].sequence);
    }

    #[test]
    fn terminal_closeout_proof_feeds_state_backbone_projection() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();

        append_terminal_closeout_proof(
            &doc,
            TerminalCloseoutProofInput {
                cycle_id: &started.cycle_id,
                last_event: "commit_success",
                did_commit: true,
                file_hash: "file-sha",
                snapshot_hash: "head-sha",
                head_hash: "head-sha",
                state_file_hash_matches: false,
                state_snapshot_hash_matches: true,
                agreement:
                    agent_doc_state_backbone::TerminalCloseoutAgreement::SnapshotHeadVisibleDrift,
                response_materialized_in_head: true,
                capture_id: Some("capture-1"),
                response_sha256: Some("response-sha"),
                recorded_at_ms: 42,
            },
        )
        .unwrap();

        let proof = load_latest_terminal_closeout_proof(&doc)
            .unwrap()
            .expect("terminal closeout proof");
        assert_eq!(proof.cycle_id, started.cycle_id);
        assert_eq!(proof.file_hash, "file-sha");
        assert_eq!(proof.snapshot_hash, "head-sha");
        assert_eq!(proof.head_hash, "head-sha");
        assert_eq!(
            proof.agreement,
            agent_doc_state_backbone::TerminalCloseoutAgreement::SnapshotHeadVisibleDrift
        );
        assert!(proof.response_materialized_in_head);
        assert_eq!(proof.capture_id.as_deref(), Some("capture-1"));
        assert_eq!(proof.response_sha256.as_deref(), Some("response-sha"));
        assert_eq!(proof.recorded_at_ms, 42);
    }

    #[test]
    fn closeout_recovery_evidence_feeds_state_backbone_projection() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        append_closeout_recovery_evidence(
            &doc,
            CloseoutRecoveryEvidenceInput {
                visible_markdown_hash: "visible-sha",
                snapshot_hash: Some("snapshot-sha"),
                active_cycle_id: Some(&started.cycle_id),
                active_cycle_phase: Some(CyclePhase::ResponseCaptured),
                active_capture_id: Some("capture-1"),
                active_capture_cycle_id: Some(&started.cycle_id),
                active_capture_state: Some("captured"),
                active_capture_response_sha256: Some("response-sha"),
                response_body: agent_doc_state_backbone::CloseoutRecoveryResponseBodyEvidence::PresentInVisible {
                    capture_id: "capture-1".into(),
                },
                queue_only_drift: Some(
                    agent_doc_state_backbone::CloseoutRecoveryQueueOnlyDriftEvidence {
                        file_hash_mismatch: true,
                        snapshot_hash_mismatch: false,
                        proven_queue_only: true,
                    },
                ),
                snapshot_head_drift: Some(
                    agent_doc_state_backbone::CloseoutRecoveryDriftEvidence::MetadataOnly,
                ),
                snapshot_visible_drift: Some(
                    agent_doc_state_backbone::CloseoutRecoveryDriftEvidence::BoundaryOnly,
                ),
                editor_ipc: agent_doc_state_backbone::CloseoutRecoveryEditorIpcEvidence::FreshLiveBuffer {
                    live_buffer_count: 1,
                    socket_degraded: false,
                },
                binary_freshness: agent_doc_state_backbone::CloseoutRecoveryBinaryFreshnessEvidence::NoStaleWarning,
                recorded_at_ms: 42,
            },
        )
        .unwrap();

        let evidence = load_latest_closeout_recovery_evidence(&doc)
            .unwrap()
            .expect("closeout recovery evidence projection");
        assert_eq!(evidence.visible_markdown_hash, "visible-sha");
        assert_eq!(evidence.snapshot_hash.as_deref(), Some("snapshot-sha"));
        assert_eq!(
            evidence.active_cycle_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
        assert_eq!(
            evidence.active_cycle_phase,
            Some(CyclePhase::ResponseCaptured)
        );
        assert_eq!(evidence.active_capture_id.as_deref(), Some("capture-1"));
        assert_eq!(
            evidence.active_capture_response_sha256.as_deref(),
            Some("response-sha")
        );
        assert_eq!(
            evidence
                .queue_only_drift
                .as_ref()
                .map(|drift| drift.proven_queue_only),
            Some(true)
        );
        assert_eq!(
            evidence.snapshot_head_drift,
            Some(agent_doc_state_backbone::CloseoutRecoveryDriftEvidence::MetadataOnly)
        );
        assert_eq!(
            evidence.snapshot_visible_drift,
            Some(agent_doc_state_backbone::CloseoutRecoveryDriftEvidence::BoundaryOnly)
        );
        assert_eq!(evidence.recorded_at_ms, 42);
    }

    #[test]
    fn load_closeout_projection_absent_store_does_not_create_state_db() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let state_db = agent_doc_sqlite::state_store::state_db_path(dir.path());
        assert!(!state_db.exists());

        assert_eq!(load_closeout_projection(&doc).unwrap(), None);
        assert!(
            !state_db.exists(),
            "read-only projection lookup must not initialize state.db"
        );
    }

    #[test]
    fn load_closeout_projection_replays_state_backbone_events() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "response-sha",
            None,
        )
        .unwrap();
        mark_write_applied(&doc, "write_applied", Some("written"), Some("written")).unwrap();
        mark_committed(&doc, "commit_success", Some("written"), Some("written")).unwrap();

        let projected = load_closeout_projection(&doc)
            .unwrap()
            .expect("closeout projection should replay from state backbone");
        assert_eq!(
            projected.cycle_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
        assert_eq!(projected.phase, Some(CyclePhase::Committed));
        assert_eq!(
            projected.capture_id.as_deref(),
            Some(started.cycle_id.as_str())
        );
        assert_eq!(projected.response_sha256.as_deref(), Some("response-sha"));
        assert!(
            projected
                .commit
                .as_deref()
                .is_some_and(|commit| commit.starts_with("content:"))
        );
    }

    #[test]
    fn load_with_closeout_projection_never_regresses_newer_checkpoint_phase() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("body"), Some("body")).unwrap();
        mark_response_captured(
            &doc,
            "response_captured",
            Some("body"),
            Some("body"),
            "response-sha",
            None,
        )
        .unwrap();

        // Simulate a crash after the local phase advanced but before the
        // matching backbone fact was appended.
        let mut checkpoint = load(&doc).unwrap().unwrap();
        checkpoint.phase = CyclePhase::WriteApplied;
        checkpoint.last_event = "capture_write_applied_proof".to_string();
        save(&doc, &checkpoint).unwrap();

        let projected = load_with_closeout_projection(&doc)
            .unwrap()
            .expect("cycle state");
        assert_eq!(projected.phase, CyclePhase::WriteApplied);
        assert_eq!(projected.last_event, "capture_write_applied_proof");
    }

    #[test]
    fn load_with_closeout_projection_preserves_noop_commit_event_on_matching_phase() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("body"), Some("body")).unwrap();
        mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();
        mark_committed(&doc, "commit_already_current", Some("body"), Some("body")).unwrap();

        let projected = load_with_closeout_projection(&doc)
            .unwrap()
            .expect("cycle state");
        assert_eq!(projected.phase, CyclePhase::Committed);
        assert_eq!(projected.last_event, "commit_already_current");
    }

    #[test]
    fn committed_reentry_repairs_missing_terminal_closeout_projection() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "response-sha",
            None,
        )
        .unwrap();
        mark_write_applied(&doc, "write_applied", Some("written"), Some("written")).unwrap();
        mark_committed(&doc, "commit_success", Some("written"), Some("written")).unwrap();

        let conn = agent_doc_sqlite::state_store::open_state_db(dir.path()).unwrap();
        let deleted = conn
            .execute(
                "DELETE FROM state_events WHERE fact_type = 'commit_observed'",
                [],
            )
            .unwrap();
        assert_eq!(deleted, 1);

        mark_committed(&doc, "commit_success", Some("written"), Some("written")).unwrap();
        mark_committed(&doc, "commit_success", Some("written"), Some("written")).unwrap();

        let rows = agent_doc_sqlite::state_store::load_state_events_from_db(&conn, None).unwrap();
        assert_eq!(
            rows.iter()
                .filter(|row| row.fact_type == "commit_observed")
                .count(),
            1,
            "terminal re-entry should restore the missing commit fact exactly once"
        );
        let mut ledger = agent_doc_state_backbone::EventLedger::new();
        for row in rows {
            let event: agent_doc_state_backbone::StateEvent =
                serde_json::from_str(&row.payload_json).unwrap();
            ledger.append(event);
        }
        let document_hash = agent_doc_fs::document_state_hash(&doc).unwrap();
        let closeout = ledger
            .project_document(&document_hash)
            .expect("repaired terminal event should project document state")
            .closeout;
        assert_eq!(closeout.phase, Some(CyclePhase::Committed));
        assert!(
            closeout
                .commit
                .as_deref()
                .is_some_and(|commit| commit.starts_with("content:"))
        );
    }

    #[test]
    fn mark_recycle_resume_consumed_latches_and_is_idempotent() {
        // `#midturn-recycle-resume` Phase B: the boot-resume consume latch flips once
        // and stays latched, so a second boot reading the same still-open checkpoint
        // does not re-dispatch the interrupted turn again.
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        // Fresh open cycle: not yet consumed.
        let opened = load(&doc).unwrap().unwrap();
        assert!(opened.is_open());
        assert!(!opened.recycle_resume_consumed);

        let consumed = mark_recycle_resume_consumed(&doc).unwrap().unwrap();
        assert!(consumed.recycle_resume_consumed);
        assert!(consumed.is_open(), "consuming does not close the cycle");
        // Persisted.
        assert!(load(&doc).unwrap().unwrap().recycle_resume_consumed);

        // Idempotent: a second consume keeps the latch set (no panic, no reset).
        let again = mark_recycle_resume_consumed(&doc).unwrap().unwrap();
        assert!(again.recycle_resume_consumed);

        // No cycle state → Ok(None), never an error.
        let other = dir.path().join("other.md");
        fs::write(&other, "x").unwrap();
        assert!(mark_recycle_resume_consumed(&other).unwrap().is_none());
    }

    #[test]
    fn mark_committed_refreshes_terminal_content_proof_idempotently() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let committed = mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();
        let refreshed = mark_committed(&doc, "repair_applied", Some("new"), Some("new")).unwrap();
        let replay = mark_committed(&doc, "repair_applied", Some("new"), Some("new")).unwrap();

        assert_eq!(refreshed.phase, committed.phase);
        assert_eq!(refreshed.last_event, committed.last_event);
        assert_eq!(
            refreshed.file_hash.as_deref(),
            Some(agent_doc_hash::content_hash("new").as_str())
        );
        assert_eq!(replay, refreshed);
        assert_eq!(load(&doc).unwrap().unwrap(), refreshed);
    }

    #[test]
    fn mark_committed_supersedes_abandoned_cycle_with_synthetic_committed_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        let abandoned =
            mark_abandoned(&doc, "stalled_preflight", Some("snap"), Some("body")).unwrap();

        let committed = mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();

        assert_eq!(committed.phase, CyclePhase::Committed);
        assert_eq!(committed.last_event, "commit_success");
        assert_ne!(committed.cycle_id, abandoned.cycle_id);
        assert!(committed.cycle_id.starts_with("synthetic-"));
        assert_eq!(load(&doc).unwrap().unwrap(), committed);
    }

    #[test]
    fn abandoned_and_timeout_bookkeeping_do_not_reopen_committed_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let committed = mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();
        let abandoned = mark_abandoned(&doc, "stale_empty", Some("new"), Some("new")).unwrap();
        let timeout = mark_recoverable_preflight_timeout(&doc, "recoverable_timeout")
            .unwrap()
            .unwrap();

        assert_eq!(abandoned, committed);
        assert_eq!(timeout, committed);
        assert_eq!(load(&doc).unwrap().unwrap(), committed);
    }

    #[test]
    fn mark_abandoned_closes_cycle_without_commit() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let state =
            mark_abandoned(&doc, "abandon_empty_preflight", Some("snap"), Some("body")).unwrap();
        assert_eq!(state.phase, CyclePhase::Abandoned);
        assert_eq!(state.last_event, "abandon_empty_preflight");
        assert!(!state.is_open());
    }

    #[test]
    fn exact_false_stale_retirement_can_reactivate_same_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        let started = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        let captured = mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some("body"),
            "response-sha",
            Some(&started.cycle_id),
        )
        .unwrap();
        append_response_captured_body(
            &doc,
            CapturedResponseFactInput {
                cycle_id: &captured.cycle_id,
                capture_id: captured.capture_id.as_deref().unwrap(),
                response_sha256: "response-sha",
                response_body: "response body",
                intent_body: None,
                mutation_plan_json: None,
                file_hash: captured.file_hash.as_deref(),
                snapshot_hash: captured.snapshot_hash.as_deref(),
                baseline_content: Some("body"),
            },
        )
        .unwrap();
        retire_projected_captured_response(
            &doc,
            &captured.cycle_id,
            captured.capture_id.as_deref().unwrap(),
            "capture_discarded",
        )
        .unwrap();
        mark_abandoned(
            &doc,
            "repair_retire_superseded_captured_only_orphan",
            Some("snap"),
            Some("body"),
        )
        .unwrap();

        assert!(
            reactivate_false_stale_capture_retirement(
                &doc,
                captured.capture_id.as_deref().unwrap(),
                "response-sha",
            )
            .unwrap()
        );
        let reactivated = load(&doc).unwrap().unwrap();
        assert_eq!(reactivated.cycle_id, started.cycle_id);
        assert_eq!(reactivated.phase, CyclePhase::ResponseCaptured);
        assert_eq!(
            reactivated.last_event,
            "session_check_reactivated_false_stale_capture_retirement"
        );
        assert!(
            reactivate_false_stale_capture_retirement(
                &doc,
                captured.capture_id.as_deref().unwrap(),
                "response-sha",
            )
            .unwrap(),
            "recovery must remain idempotent after the cycle projection advances first"
        );
    }

    #[test]
    fn mark_write_applied_creates_synthetic_cycle_when_missing() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let state = mark_write_applied(&doc, "recover_apply", Some("body"), Some("body")).unwrap();
        assert_eq!(state.phase, CyclePhase::WriteApplied);
        assert!(state.cycle_id.starts_with("synthetic-"));
    }

    #[test]
    fn mark_write_applied_does_not_regress_committed_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();

        let state = mark_write_applied(&doc, "repair_applied", Some("new"), Some("new")).unwrap();
        let body_hash = agent_doc_hash::content_hash("body");
        assert_eq!(state.phase, CyclePhase::Committed);
        assert_eq!(state.last_event, "commit_success");
        assert_eq!(state.snapshot_hash.as_deref(), Some(body_hash.as_str()));
        assert_eq!(state.file_hash.as_deref(), Some(body_hash.as_str()));
    }

    #[test]
    fn mark_response_captured_does_not_regress_committed_cycle() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        let committed = mark_committed(&doc, "commit_success", Some("body"), Some("body")).unwrap();

        let state = mark_response_captured(
            &doc,
            "response_captured",
            Some("new"),
            Some("new"),
            "abc",
            Some(&committed.cycle_id),
        )
        .unwrap();
        assert_eq!(state.phase, CyclePhase::Committed);
        assert_eq!(state.last_event, "commit_success");
        assert_eq!(state.capture_id, committed.capture_id);
        assert_eq!(state.response_sha256, committed.response_sha256);
    }

    #[test]
    fn cycle_phase_transitions_append_to_session_log() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        let doc_content = "---\nagent_doc_session: sess-123\n---\n\nbody\n";
        fs::write(&doc, doc_content).unwrap();

        let started = start_preflight(&doc, Some("snap"), Some(doc_content)).unwrap();
        let captured = mark_response_captured(
            &doc,
            "response_captured",
            Some("snap"),
            Some(doc_content),
            "abc123",
            Some(&started.cycle_id),
        )
        .unwrap();
        let written =
            mark_write_applied(&doc, "write_template", Some(doc_content), Some(doc_content))
                .unwrap();
        let committed =
            mark_committed(&doc, "commit_success", Some(doc_content), Some(doc_content)).unwrap();

        let log = fs::read_to_string(dir.path().join(".agent-doc/logs/sess-123.log")).unwrap();
        assert!(log.contains(&format!(
            "document_cycle phase=preflight_started cycle={} event=preflight_started",
            started.cycle_id
        )));
        assert!(log.contains(&format!(
            "document_cycle phase=response_captured cycle={} event=response_captured capture_id={}",
            captured.cycle_id, captured.cycle_id
        )));
        assert!(log.contains(&format!(
            "document_cycle phase=write_applied cycle={} event=write_template",
            written.cycle_id
        )));
        assert!(log.contains(&format!(
            "document_cycle phase=committed cycle={} event=commit_success",
            committed.cycle_id
        )));
    }

    #[test]
    fn cycle_phase_transitions_skip_session_log_without_session_id() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body\n").unwrap();

        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        assert!(
            !dir.path().join(".agent-doc/logs").exists(),
            "plain documents without a session id should not create session logs"
        );
    }

    #[test]
    fn start_preflight_with_task_stores_queue_task_id_and_turn_id() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let state = start_preflight_with_task(
            &doc,
            Some("snap"),
            Some("body"),
            Some("#reentrant-phase2"),
            Some("#reentrant-phase2"),
        )
        .unwrap();
        assert_eq!(state.queue_task_id.as_deref(), Some("#reentrant-phase2"));
        assert_eq!(state.turn_id.as_deref(), Some("#reentrant-phase2"));

        let loaded = load(&doc).unwrap().expect("state should persist");
        assert_eq!(loaded.queue_task_id, state.queue_task_id);
        assert_eq!(loaded.turn_id, state.turn_id);
    }

    #[test]
    fn record_turn_checkpoint_persists_resume_identity_and_baseline() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();
        start_preflight(&doc, Some("snap"), Some("body")).unwrap();

        let prompts = vec![
            "  do [#DurableRecycle]  ".to_string(),
            "do [#DurableRecycle]".to_string(),
            "free text prompt".to_string(),
        ];
        let state = record_turn_checkpoint(
            &doc,
            &prompts,
            Some("#DurableRecycle"),
            Some("#DurableRecycle"),
        )
        .unwrap()
        .expect("state should exist");

        assert_eq!(
            state.prompt_targets,
            vec![
                "do [#DurableRecycle]".to_string(),
                "free text prompt".to_string()
            ]
        );
        assert_eq!(state.queue_task_id.as_deref(), Some("#durablerecycle"));
        assert_eq!(state.turn_id.as_deref(), Some("#durablerecycle"));

        let loaded = load(&doc).unwrap().unwrap();
        assert_eq!(loaded.prompt_targets, state.prompt_targets);
        assert_eq!(loaded.queue_task_id, state.queue_task_id);
        assert_eq!(loaded.turn_id, state.turn_id);
    }

    #[test]
    fn start_preflight_without_task_has_none_ids() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let state = start_preflight(&doc, Some("snap"), Some("body")).unwrap();
        assert!(state.queue_task_id.is_none());
        assert!(state.turn_id.is_none());
    }

    #[test]
    fn to_pipeline_mirrors_cycle_state_fields_and_tracks_phase() {
        let dir = setup_project();
        let doc = dir.path().join("doc.md");
        fs::write(&doc, "body").unwrap();

        let state = start_preflight_with_task(
            &doc,
            Some("snap"),
            Some("body"),
            Some("#fmrunid-wire"),
            Some("#fmrunid-wire"),
        )
        .unwrap();

        let pipeline = state.to_pipeline();
        assert_eq!(pipeline.run_id.as_deref(), Some(state.cycle_id.as_str()));
        assert_eq!(pipeline.step.as_deref(), Some("preflight_started"));
        assert_eq!(pipeline.turn_id.as_deref(), Some("#fmrunid-wire"));
        assert_eq!(pipeline.queue_task_id.as_deref(), Some("#fmrunid-wire"));
        assert!(!pipeline.is_empty());

        // `step` follows the authoritative phase transition.
        let captured =
            mark_response_captured(&doc, "captured", Some("snap"), Some("body"), "abc", None)
                .unwrap();
        assert_eq!(
            captured.to_pipeline().step.as_deref(),
            Some("response_captured")
        );
        // run_id stays stable across transitions (same cycle).
        assert_eq!(captured.to_pipeline().run_id, pipeline.run_id);
    }
}

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

    fn state(phase: CyclePhase, updated_at: u64) -> CycleState {
        let mut state = synthetic_state(Path::new("/tmp/plan.md"), phase);
        state.updated_at = updated_at;
        state
    }

    /// The case the hook creates: preflight already opened this turn, and a
    /// second preflight arrives before anything was captured.
    #[test]
    fn a_second_preflight_in_the_same_open_turn_reuses_the_cycle() {
        let open = state(CyclePhase::PreflightStarted, 1_000);
        assert!(open.reusable_by_reentrant_preflight(0, 1_005, STALLED_CYCLE_RESOLVE_SECS));
    }

    /// A committed turn is a real boundary — the next preflight is a new turn
    /// and must mint a new id.
    #[test]
    fn a_committed_cycle_is_not_reused() {
        let committed = state(CyclePhase::Committed, 1_000);
        assert!(!committed.reusable_by_reentrant_preflight(0, 1_005, STALLED_CYCLE_RESOLVE_SECS));
    }

    /// Once a response exists, the cycle is durable recovery evidence. Reusing
    /// it would let a fresh turn inherit another turn's captured response.
    #[test]
    fn a_cycle_with_a_captured_response_is_not_reused() {
        let mut captured = state(CyclePhase::PreflightStarted, 1_000);
        captured.capture_id = Some("capture-1".to_string());
        assert!(!captured.reusable_by_reentrant_preflight(0, 1_005, STALLED_CYCLE_RESOLVE_SECS));

        let mut written = state(CyclePhase::PreflightStarted, 1_000);
        written.response_sha256 = Some("deadbeef".to_string());
        assert!(!written.reusable_by_reentrant_preflight(0, 1_005, STALLED_CYCLE_RESOLVE_SECS));
    }

    /// A crashed or superseded older turn must not be inherited by a new one.
    /// Uses the same staleness policy the supervisor force-abandons on, so the
    /// two cannot drift.
    #[test]
    fn a_stalled_open_cycle_is_not_inherited_by_a_new_turn() {
        let stalled = state(CyclePhase::PreflightStarted, 1_000);
        let now = 1_000 + STALLED_CYCLE_RESOLVE_SECS + 1;
        assert!(stalled.stalled_before_response_capture_cycle(0, now, STALLED_CYCLE_RESOLVE_SECS));
        assert!(!stalled.reusable_by_reentrant_preflight(0, now, STALLED_CYCLE_RESOLVE_SECS));
    }

    /// Inflight IPC means the cycle is live, so it is not stalled and stays
    /// reusable even well past the deadline.
    #[test]
    fn an_inflight_cycle_stays_reusable_past_the_deadline() {
        let live = state(CyclePhase::PreflightStarted, 1_000);
        let now = 1_000 + STALLED_CYCLE_RESOLVE_SECS + 1;
        assert!(live.reusable_by_reentrant_preflight(1, now, STALLED_CYCLE_RESOLVE_SECS));
    }
}

#[cfg(test)]
mod project_root_symmetry_tests {
    use super::*;
    use std::fs;

    /// `#ay80ci`: a cycle-state write must land for a document whose project
    /// root has not been materialized yet.
    ///
    /// `append_state_fact` resolves its destination with
    /// `project_root_or_file_parent` and lets `open_state_db` create
    /// `.agent-doc` there, and the repair-side reader resolves the same way. If
    /// the key derivation is stricter than either of them, the write is silently
    /// skipped and the read-back fails — which is how CI kept losing
    /// `finalize_skips_ignored_untracked_session_doc`.
    #[test]
    fn capture_write_lands_without_a_preexisting_agent_doc_ancestor() {
        let dir = tempfile::TempDir::new().unwrap();
        // Deliberately NO `.agent-doc` anywhere: this is a document that
        // `finalize` is legitimately operating on before the root exists.
        fs::create_dir_all(dir.path().join("scratch")).unwrap();
        let doc = dir.path().join("scratch/session.md");
        fs::write(&doc, "---\nagent_doc_format: template\n---\n\nbody\n").unwrap();

        let wrote = append_response_captured_body(
            &doc,
            CapturedResponseFactInput {
                cycle_id: "c1",
                capture_id: "c1",
                response_sha256: "abc",
                response_body: "captured body",
                intent_body: Some("captured body"),
                mutation_plan_json: None,
                file_hash: None,
                snapshot_hash: None,
                baseline_content: None,
            },
        )
        .unwrap();
        assert!(
            wrote,
            "the captured-response fact must be durably written, not silently skipped",
        );

        // And it must be readable back through the same root resolution the
        // repair path uses — write and read have to agree on the root.
        let root = agent_doc_project_root_io::project_root_or_file_parent(&doc).unwrap();
        assert!(
            agent_doc_sqlite::state_store::state_db_path(&root).exists(),
            "the write must have materialized the state db at the root the reader resolves",
        );
    }
}
