shepherd-cli 6.6.1

//! Native, two-phase Planning orientation custody.
//!
//! Engineer-authored phase0 Markdown and native dispatch records are the only
//! inputs. The current published Engineer remains active while its completed
//! children supply results. Native pending records bind each issued path.
//! This module validates bytes through one no-follow descriptor and
//! publishes the pre/post verification artifacts under the selected run.

use std::{
    collections::{BTreeMap, BTreeSet},
    path::Path,
};

use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
// Digests come from the engine. This file used to carry its own copy of the
// sha256/hex helper; `scripts/check-rust-duplicates.py` counted fourteen of
// them across cli, render and registry, and asserts one home in shepherd-core.
use shepherd::digest::{format_digest, sha256_hex};
use shepherd::run::{LaneStatus, RunStatus};
use shepherd::{
    RunState,
    dispatch::{
        DispatchRecord, DispatchState, PendingDispatch, PendingLaunchState, ProjectId,
        ReviewCustodyState, Role, RunId, WorkKind,
    },
    plan::PlanTopology,
    registry::{OpenMode, Registry, SingletonPublicationState},
    run::LaneState,
};

use crate::{
    ContextInputs, DispatchService, DispatchStore, DispatchStoreError, ExecutionContext, RunStore,
    RunStoreError,
    cmd::planning::{
        PlanReadiness, reverify_plan_readiness, verify_plan_binding, verify_plan_readiness,
    },
    interface::CliError,
    native_authority::{
        self, ArchivedInput, PostCheckpoint, PreCheckpoint, RetiredPreCheckpoint, RunAuthority,
    },
    orientation_fs::{AnchoredFiles, OpenedBytes, read_snapshot},
    run_store::RunAccess,
};

const MANIFEST_FILE: &str = "orientation-manifest.json";
const PRE_FILE: &str = "orientation-pre.json";
const POST_FILE: &str = "orientation-post.json";
const REPORT_SCHEMA: &str = "shepherd.orientation-report/1";
const MANIFEST_SCHEMA: &str = "shepherd.orientation-manifest/2";
const PRE_SCHEMA: &str = "shepherd.orientation-pre/1";
const POST_SCHEMA: &str = "shepherd.orientation-post/1";
const PHASE0_HEADINGS: [&str; 8] = [
    "# Orientation",
    "## Run and seed",
    "## Auditor briefs",
    "## Discovery briefs",
    "## Assumptions and decisions",
    "## Coverage map",
    "## Self-review",
    "## Critic loop",
];

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct OrientationSource {
    id: String,
    kind: String,
    role: String,
    path: String,
    sha256: String,
    read_scope: Vec<String>,
    write_scope: Vec<String>,
    project_id: String,
    session_id: String,
    incarnation: String,
    revision: u64,
    started_at: i64,
    lease_expires_at: i64,
    stopped_at: Option<i64>,
    nonce: String,
    dispatch_path: String,
    dispatch_sha256: String,
    pending_path: String,
    pending_sha256: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct OrientationManifest {
    schema: String,
    phase: String,
    run: String,
    run_dir: String,
    project_root: String,
    incarnation: String,
    orientation_epoch: u64,
    created_at: i64,
    sources: Vec<OrientationSource>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct OrientationPre {
    schema: String,
    phase: String,
    run: String,
    run_dir: String,
    project_root: String,
    incarnation: String,
    orientation_epoch: u64,
    created_at: i64,
    manifest_sha256: String,
    accepted: bool,
    sources: Vec<OrientationSource>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct OrientationPost {
    schema: String,
    phase: String,
    run: String,
    run_dir: String,
    project_root: String,
    incarnation: String,
    orientation_epoch: u64,
    created_at: i64,
    manifest_sha256: String,
    pre_sha256: String,
    accepted: bool,
    sources: Vec<OrientationSource>,
    verdict: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    planning: Option<PlanReadiness>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct OrientationArchive {
    schema: String,
    run: String,
    epoch: u64,
    inputs: BTreeMap<String, ArchivedInput>,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct NamedField {
    id: String,
    statement: String,
}

#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd)]
#[serde(deny_unknown_fields)]
struct Evidence {
    path: String,
    sha256: String,
    line: usize,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct Claim {
    id: String,
    statement: String,
    evidence: Vec<Evidence>,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct OrientationReport {
    schema: String,
    run: String,
    result_id: String,
    kind: String,
    role: String,
    read_scope: Vec<String>,
    status: String,
    summary: String,
    assumptions: Vec<NamedField>,
    claims: Vec<Claim>,
    evidence: Vec<Evidence>,
    caveats: Vec<String>,
    #[serde(default)]
    orientation_pre_sha256: Option<String>,
    #[serde(default)]
    verdict: Option<Value>,
}

#[derive(Clone, Debug)]
struct ChildEvidence {
    source: OrientationSource,
    report: OrientationReport,
    result_bytes: Vec<u8>,
}

#[derive(Clone, Debug)]
struct PreContext {
    manifest: OrientationManifest,
    manifest_bytes: Vec<u8>,
    pre_bytes: Vec<u8>,
    children: Vec<ChildEvidence>,
    phase0_bytes: Vec<u8>,
    claims: BTreeMap<String, String>,
    assumptions: BTreeSet<String>,
}

struct CollectedPre {
    phase0: OrientationSource,
    children: Vec<ChildEvidence>,
    phase0_bytes: Vec<u8>,
    claims: BTreeMap<String, String>,
    assumptions: BTreeSet<String>,
    engineer_nonce: String,
}

fn archive_bytes(
    files: &AnchoredFiles<'_>,
    epoch: u64,
    bytes: &[u8],
) -> Result<ArchivedInput, CliError> {
    if bytes.len() > 128 * 1024 {
        return Err(error(
            "orientation input archive exceeds its bounded reader",
        ));
    }
    let digest = sha256_hex(bytes);
    let path = format!(".orientation-epoch-{epoch}-{digest}.json");
    if let Err(source) = files.write_new(&path, bytes) {
        // Content-addressed replay is allowed, overwrite is never allowed.
        if files.read_file(&path, "immutable orientation input")?.bytes != bytes {
            return Err(source);
        }
    }
    Ok(ArchivedInput {
        path,
        sha256: digest,
    })
}

fn archive_orientation_inputs<'a>(
    files: &AnchoredFiles<'_>,
    run: &str,
    epoch: u64,
    sources: &[OrientationSource],
    reports: impl IntoIterator<Item = &'a ChildEvidence>,
    generated: impl IntoIterator<Item = (&'a str, &'a [u8])>,
) -> Result<ArchivedInput, CliError> {
    let mut inputs = BTreeMap::new();
    let mut insert = |key: String, bytes: &[u8], expected: Option<&str>| -> Result<(), CliError> {
        let value = archive_bytes(files, epoch, bytes)?;
        if expected.is_some_and(|digest| digest != value.sha256) {
            return Err(error("orientation input changed before immutable archival"));
        }
        if inputs.get(&key).is_some_and(|previous| previous != &value) {
            return Err(error(
                "orientation archive has conflicting input identities",
            ));
        }
        inputs.insert(key, value);
        Ok(())
    };
    for (name, bytes) in generated {
        insert(format!("native/{name}"), bytes, None)?;
    }
    for source in sources {
        let opened = files.read_authoring_file(&source.path, "orientation source archive")?;
        insert(
            format!("authoring/{}", source.path),
            &opened.bytes,
            Some(&source.sha256),
        )?;
        for (relative, digest) in [
            (&source.dispatch_path, &source.dispatch_sha256),
            (&source.pending_path, &source.pending_sha256),
        ] {
            let opened = files.read_file(relative, "orientation native input archive")?;
            insert(format!("native/{relative}"), &opened.bytes, Some(digest))?;
        }
    }
    for child in reports {
        for evidence in &child.report.evidence {
            let opened =
                files.read_project_file(&evidence.path, "orientation cited input archive")?;
            insert(
                format!("project/{}", evidence.path),
                &opened.bytes,
                Some(&evidence.sha256),
            )?;
        }
    }
    archive_bytes(
        files,
        epoch,
        &encode_json(&OrientationArchive {
            schema: "shepherd.orientation-input-archive/1".into(),
            run: run.into(),
            epoch,
            inputs,
        })?,
    )
}

fn read_archive_input(
    files: &AnchoredFiles<'_>,
    epoch: u64,
    input: &ArchivedInput,
) -> Result<Vec<u8>, CliError> {
    if input.path != format!(".orientation-epoch-{epoch}-{}.json", input.sha256) {
        return Err(error(
            "orientation archive path is not its exact content identity",
        ));
    }
    let opened = files.read_file(&input.path, "immutable orientation input")?;
    if opened.sha256 != input.sha256 {
        return Err(error("immutable orientation input archive changed"));
    }
    Ok(opened.bytes)
}

fn verify_input_archive(
    files: &AnchoredFiles<'_>,
    run: &str,
    pre: &PreCheckpoint,
    post: Option<&PostCheckpoint>,
    reference: &ArchivedInput,
) -> Result<(), CliError> {
    let archive: OrientationArchive = decode_json(
        &read_archive_input(files, pre.epoch, reference)?,
        "orientation input archive",
    )?;
    if archive.schema != "shepherd.orientation-input-archive/1"
        || archive.run != run
        || archive.epoch != pre.epoch
    {
        return Err(error(
            "orientation input archive identity differs from its checkpoint",
        ));
    }
    let mut bytes = BTreeMap::new();
    for (key, reference) in &archive.inputs {
        bytes.insert(
            key.clone(),
            read_archive_input(files, pre.epoch, reference)?,
        );
    }
    let mut required = BTreeSet::new();
    let mut check = |key: String, digest: &str| -> Result<&Vec<u8>, CliError> {
        required.insert(key.clone());
        let value = bytes
            .get(&key)
            .ok_or_else(|| error(format!("orientation archive omits {key}")))?;
        if sha256_hex(value) != digest {
            return Err(error(
                "orientation archive input differs from checkpoint sources",
            ));
        }
        Ok(value)
    };
    let manifest: OrientationManifest = decode_json(
        check(format!("native/{MANIFEST_FILE}"), &pre.manifest_sha256)?,
        "archived manifest",
    )?;
    let archived_pre: OrientationPre = decode_json(
        check(format!("native/{PRE_FILE}"), &pre.pre_sha256)?,
        "archived pre",
    )?;
    if manifest.run != run
        || manifest.orientation_epoch != pre.epoch
        || manifest.sources != archived_pre.sources
    {
        return Err(error("archived pre source identity mismatch"));
    }
    let sources = if let Some(post) = post {
        let archived_post: OrientationPost = decode_json(
            check(format!("native/{POST_FILE}"), &post.post_sha256)?,
            "archived post",
        )?;
        if archived_post.run != run || archived_post.orientation_epoch != pre.epoch {
            return Err(error("archived post identity mismatch"));
        }
        archived_post.sources
    } else {
        manifest.sources
    };
    for source in &sources {
        let artifact = check(format!("authoring/{}", source.path), &source.sha256)?;
        let report = if source.kind == "phase0" {
            None
        } else {
            Some(decode_json::<OrientationReport>(
                artifact,
                "archived child report",
            )?)
        };
        check(
            format!("native/{}", source.dispatch_path),
            &source.dispatch_sha256,
        )?;
        check(
            format!("native/{}", source.pending_path),
            &source.pending_sha256,
        )?;
        if let Some(report) = report {
            for evidence in &report.evidence {
                check(format!("project/{}", evidence.path), &evidence.sha256)?;
            }
        }
    }
    if required != archive.inputs.keys().cloned().collect() {
        return Err(error(
            "orientation archive has incomplete or unexpected input inventory",
        ));
    }
    Ok(())
}

fn verify_checkpoint_archives(
    files: &AnchoredFiles<'_>,
    run: &str,
    pre: &PreCheckpoint,
    post: Option<&PostCheckpoint>,
) -> Result<(), CliError> {
    let reference = pre.input_archive.as_ref().ok_or_else(|| {
        error("pre checkpoint has no immutable input archive; replacement cannot reset it")
    })?;
    verify_input_archive(files, run, pre, None, reference)?;
    if let Some(post) = post {
        let reference = post
            .input_archive
            .as_ref()
            .ok_or_else(|| error("post checkpoint has no immutable input archive"))?;
        verify_input_archive(files, run, pre, Some(post), reference)?;
    }
    Ok(())
}

fn retire_replaced_pre(
    context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    native: &PlanningDispatches,
) -> Result<RunAuthority, CliError> {
    let pre = authority
        .pre
        .as_ref()
        .ok_or_else(|| error("pre checkpoint is absent"))?;
    let pending = exact_pending(&native.pending, &native.engineer)?;
    let replaced = pending
        .replaces_agent_id
        .as_ref()
        .ok_or_else(|| error("new pre requires exact root-authorized replacement lineage"))?;
    let original = native
        .records
        .iter()
        .find(|record| record.agent_id == *replaced)
        .ok_or_else(|| error("replaced Engineer record is absent"))?;
    let originals = native
        .pending
        .iter()
        .filter(|pending| pending.expected_attachment.agent_id == *replaced)
        .collect::<Vec<_>>();
    let [original_pending] = originals.as_slice() else {
        return Err(error(
            "replaced Engineer has no unique native pending record",
        ));
    };
    let custody = DispatchStore::new(&context.runs_root)
        .with_run_access(&native.engineer.run, files.access, |held| {
            held.load_review_custody(replaced)
        })
        .map_err(|source| error(source.to_string()))?;
    custody
        .validate()
        .map_err(|source| error(source.to_string()))?;
    if original.role != Role::Engineer
        || original.state != DispatchState::Malignant
        || original.nonce != pre.engineer_nonce
        || original.run_incarnation != authority.run_incarnation
        || original.project_id != native.engineer.project_id
        || original.run != native.engineer.run
        || original.root_session_id != native.engineer.root_session_id
        || original_pending.launch_state != PendingLaunchState::Quarantined
        || !crate::dispatch_service::same_replacement_contract(original_pending, pending)
        || custody.state != ReviewCustodyState::Replaced
        || custody.subject_agent_id != *replaced
        || custody.subject_role != Role::Engineer
        || custody.subject_session_id != original.session_id
        || custody.root_session_id != original.root_session_id
        || custody.project_id != original.project_id
        || custody.run != original.run
        || custody.pending_launch_id_hash != original_pending.launch_id_hash
        || custody.task_sha256 != original_pending.task_sha256
        || custody.replacement_agent_id.as_ref() != Some(&native.engineer.agent_id)
        || !custody.claim_revoked
        || !custody.write_revoked
        || !custody.session_quarantined
        || custody.updated_at > context.now_unix_millis()
    {
        return Err(error(
            "new pre does not match exact malignant replacement custody",
        ));
    }
    verify_checkpoint_archives(files, run, pre, authority.post.as_ref())?;
    if files.read_file(MANIFEST_FILE, "retired manifest")?.sha256 != pre.manifest_sha256
        || files.read_file(PRE_FILE, "retired pre")?.sha256 != pre.pre_sha256
        || authority.post.as_ref().is_some_and(|post| {
            files
                .read_file(POST_FILE, "retired post")
                .map_or(true, |value| value.sha256 != post.post_sha256)
        })
    {
        return Err(error("native pre/post changed before lineage retirement"));
    }
    native_authority::retire_pre(
        context,
        run,
        authority,
        RetiredPreCheckpoint {
            pre: pre.clone(),
            post: authority.post.clone(),
            replaced_agent_id: replaced.to_string(),
            replacement_agent_id: native.engineer.agent_id.to_string(),
            replacement_nonce: native.engineer.nonce.clone(),
        },
    )
}

pub(crate) fn create_pre(
    context: &ExecutionContext,
    run: &str,
    json_output: bool,
) -> Result<Value, CliError> {
    let _ = json_output;
    let path = context.runs_root.join(run).join("run.json");
    RunStore::new(path)
        .update_with_access(|state, access| create_pre_locked(context, run, state, access))
        .map_err(|source| error(source.to_string()))
}

fn create_pre_locked(
    context: &ExecutionContext,
    run: &str,
    state: &mut RunState,
    access: &RunAccess<'_>,
) -> Result<Value, RunStoreError> {
    ensure_planning_state(state, run)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let files = AnchoredFiles::open(context, run, access)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let mut authority =
        native_authority::validate_and_bind_pre_state(context, run, state, project_id.as_str())
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let mut replacement_sources = None;
    if let Some(checkpoint) = authority.pre.as_ref() {
        let native = planning_dispatches(context, run, &files, &authority)
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
        if native.engineer.nonce == checkpoint.engineer_nonce {
            return recover_committed_pre(context, run, state, &files, &authority, checkpoint)
                .map_err(|source| RunStoreError::mutation(cli_message(source)));
        }
        // Require complete fresh input before retiring anything. An authorized
        // replacement without its own children cannot clear the prior pre.
        replacement_sources = Some(
            collect_pre_sources(context, run, state, &files, &authority, true)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?,
        );
        authority = retire_replaced_pre(context, run, &files, &authority, &native)
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    }
    if authority.pre.is_none() && authority.orientation_epoch > 0 {
        let native = planning_dispatches(context, run, &files, &authority)
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
        let retired = authority
            .pre_history
            .last()
            .ok_or_else(|| RunStoreError::mutation("missing native pre retirement history"))?;
        if retired.pre.epoch != authority.orientation_epoch
            || retired.replacement_nonce != native.engineer.nonce
        {
            return Err(RunStoreError::mutation(
                "native pre retirement belongs to another Engineer",
            ));
        }
        verify_checkpoint_archives(&files, run, &retired.pre, retired.post.as_ref())
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
        files
            .unlink(POST_FILE)
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    }
    files
        .unlink(PRE_FILE)
        .and_then(|_| files.unlink(MANIFEST_FILE))
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    files
        .reject_unowned_orientation_artifacts()
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let collected = if let Some(collected) = replacement_sources {
        collected
    } else {
        collect_pre_sources(context, run, state, &files, &authority, true)
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?
    };
    let sources = std::iter::once(collected.phase0)
        .chain(collected.children.iter().map(|child| child.source.clone()))
        .collect();
    let manifest = OrientationManifest {
        schema: MANIFEST_SCHEMA.into(),
        phase: "pre".into(),
        run: run.into(),
        run_dir: files.run_dir.display().to_string(),
        project_root: files.project_root.display().to_string(),
        incarnation: authority.run_incarnation.clone(),
        orientation_epoch: authority.orientation_epoch + 1,
        created_at: context.now_unix_millis(),
        sources,
    };
    let manifest_bytes =
        encode_json(&manifest).map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let pre = OrientationPre {
        schema: PRE_SCHEMA.into(),
        phase: "pre".into(),
        run: run.into(),
        run_dir: manifest.run_dir.clone(),
        project_root: manifest.project_root.clone(),
        incarnation: authority.run_incarnation.clone(),
        orientation_epoch: manifest.orientation_epoch,
        created_at: manifest.created_at,
        manifest_sha256: sha256_hex(&manifest_bytes),
        accepted: true,
        sources: manifest.sources.clone(),
    };
    let pre_bytes =
        encode_json(&pre).map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    files
        .write_new(MANIFEST_FILE, &manifest_bytes)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    if let Err(source) = files.write_new(PRE_FILE, &pre_bytes) {
        let _ = files.unlink(MANIFEST_FILE);
        return Err(RunStoreError::mutation(cli_message(source)));
    }
    let checkpoint = PreCheckpoint {
        epoch: manifest.orientation_epoch,
        created_at: manifest.created_at,
        run_revision: state.orientation_epoch + 1,
        manifest_sha256: sha256_hex(&manifest_bytes),
        pre_sha256: sha256_hex(&pre_bytes),
        engineer_nonce: collected.engineer_nonce,
        input_archive: Some(
            archive_orientation_inputs(
                &files,
                run,
                manifest.orientation_epoch,
                &manifest.sources,
                collected.children.iter(),
                [
                    (MANIFEST_FILE, manifest_bytes.as_slice()),
                    (PRE_FILE, pre_bytes.as_slice()),
                ],
            )
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?,
        ),
    };
    authority = match native_authority::record_pre(context, run, &authority, checkpoint) {
        Ok(authority) => authority,
        Err(source) => {
            let _ = files.unlink(PRE_FILE);
            let _ = files.unlink(MANIFEST_FILE);
            return Err(RunStoreError::mutation(cli_message(source)));
        }
    };
    state.orientation_epoch = authority.orientation_epoch;
    verification_json(
        "pre",
        true,
        run,
        &manifest,
        sha256_hex(&pre_bytes),
        None,
        &collected.claims,
    )
    .map_err(|source| RunStoreError::mutation(cli_message(source)))
}

fn recover_committed_pre(
    context: &ExecutionContext,
    run: &str,
    state: &mut RunState,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    checkpoint: &PreCheckpoint,
) -> Result<Value, CliError> {
    verify_checkpoint_archives(files, run, checkpoint, authority.post.as_ref())?;
    let manifest_opened = files.read_file(MANIFEST_FILE, "orientation manifest")?;
    let pre_opened = files.read_file(PRE_FILE, "orientation pre")?;
    let manifest: OrientationManifest =
        decode_json(&manifest_opened.bytes, "orientation manifest")?;
    let pre: OrientationPre = decode_json(&pre_opened.bytes, "orientation pre")?;
    let manifest_sha256 = sha256_hex(&manifest_opened.bytes);
    let pre_sha256 = sha256_hex(&pre_opened.bytes);
    if manifest.schema != MANIFEST_SCHEMA
        || manifest.phase != "pre"
        || pre.schema != PRE_SCHEMA
        || pre.phase != "pre"
        || !pre.accepted
        || manifest.run != run
        || pre.run != run
        || manifest.incarnation != authority.run_incarnation
        || pre.incarnation != authority.run_incarnation
        || manifest.orientation_epoch != checkpoint.epoch
        || pre.orientation_epoch != checkpoint.epoch
        || manifest.created_at != checkpoint.created_at
        || pre.created_at != checkpoint.created_at
        || pre.manifest_sha256 != manifest_sha256
        || checkpoint.manifest_sha256 != manifest_sha256
        || checkpoint.pre_sha256 != pre_sha256
        || authority.orientation_epoch != checkpoint.epoch
        || checkpoint.run_revision != checkpoint.epoch
    {
        return Err(error(
            "committed native pre checkpoint does not match its artifacts",
        ));
    }
    if state.orientation_epoch > checkpoint.run_revision
        || state.orientation_epoch + 1 < checkpoint.run_revision
    {
        return Err(error(
            "run orientation epoch cannot recover the committed pre checkpoint",
        ));
    }
    let collected = collect_pre_sources(context, run, state, files, authority, true)?;
    let sources: Vec<OrientationSource> = std::iter::once(collected.phase0)
        .chain(collected.children.iter().map(|child| child.source.clone()))
        .collect();
    if manifest.sources != sources
        || pre.sources != sources
        || checkpoint.engineer_nonce != collected.engineer_nonce
    {
        return Err(error(
            "committed native pre sources changed during recovery",
        ));
    }
    state.orientation_epoch = checkpoint.run_revision;
    verification_json(
        "pre",
        true,
        run,
        &manifest,
        pre_sha256,
        None,
        &collected.claims,
    )
}

pub(crate) fn create_post(
    context: &ExecutionContext,
    run: &str,
    _json_output: bool,
) -> Result<Value, CliError> {
    let path = context.runs_root.join(run).join("run.json");
    RunStore::new(path)
        .update_with_access(|state, access| create_post_locked(context, run, state, access))
        .map_err(|source| error(source.to_string()))
}

fn create_post_locked(
    context: &ExecutionContext,
    run: &str,
    state: &mut RunState,
    access: &RunAccess<'_>,
) -> Result<Value, RunStoreError> {
    ensure_planning_state(state, run)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let files = AnchoredFiles::open(context, run, access)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let authority =
        native_authority::validate_and_bind_state(context, run, state, project_id.as_str())
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let pre_context = read_pre_context(context, run, state, &files, &authority)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    let post = recompute_post(context, run, state, &files, &authority, &pre_context, None)
        .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
    verification_json(
        "post",
        post.accepted,
        run,
        &pre_context.manifest,
        sha256_hex(&pre_context.pre_bytes),
        Some(&post.verdict),
        &pre_context.claims,
    )
    .map_err(|source| RunStoreError::mutation(cli_message(source)))
}

/// Validate and commit the planted -> planned transition while one native
/// run lock covers the proposed state, current records, post publication, and
/// final state write. Caller-supplied post bytes are never the authority.
pub(crate) fn transition_planned(context: &ExecutionContext, run: &str) -> Result<(), CliError> {
    let path = context.runs_root.join(run).join("run.json");
    RunStore::new(path)
        .update_with_access(|state, access| {
            ensure_planning_state(state, run)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            let files = AnchoredFiles::open(context, run, access)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            let authority =
                native_authority::validate_and_bind_state(context, run, state, project_id.as_str())
                    .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            let pre_context = read_pre_context(context, run, state, &files, &authority)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            let (_, planning) = verify_plan_readiness(context, state)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            if planning.run_artifacts.get("phase0.md")
                != Some(&sha256_hex(&pre_context.phase0_bytes))
            {
                return Err(RunStoreError::mutation(
                    "workspace planning evidence differs from accepted orientation",
                ));
            }
            let seed = files
                .read_project_file(&state.seed, "planned seed verification")
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            if planning.run_artifacts.get("seed.md") != Some(&seed.sha256) {
                return Err(RunStoreError::mutation(
                    "workspace seed differs from accepted orientation",
                ));
            }
            let post = recompute_post(
                context,
                run,
                state,
                &files,
                &authority,
                &pre_context,
                Some(planning),
            )
            .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            if !post.accepted || post.verdict != "GREEN" {
                return Err(RunStoreError::mutation(
                    "orientation post is not accepted GREEN",
                ));
            }
            let post_opened = files
                .read_file(POST_FILE, "orientation post verification")
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            let decoded: OrientationPost =
                decode_json(&post_opened.bytes, "orientation post verification")
                    .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            validate_post_shape(&decoded, context, run, &files, &authority)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            if decoded != post {
                return Err(RunStoreError::mutation(
                    "native post artifact changed during planned transition",
                ));
            }
            let planning = post.planning.as_ref().ok_or_else(|| {
                RunStoreError::mutation("planned checkpoint has no plan-v2 binding")
            })?;
            verify_plan_binding(context, state, planning)
                .map_err(|source| RunStoreError::mutation(cli_message(source)))?;
            state.plan = format!(".shepherd/runs/{run}/plan.md");
            state.status = RunStatus::Planned.into();
            Ok(())
        })
        .map_err(|source| error(source.to_string()))
}

/// Revalidate the native Planning checkpoint and atomically open execution.
/// A caller-authored `status: planned` value is not authority: the run
/// incarnation, pre/post checkpoints, source ledger, Critic verdict, and
/// persisted post bytes must all still agree while the run lock is held.
pub(crate) fn transition_executing(context: &ExecutionContext, run: &str) -> Result<(), CliError> {
    let path = context.runs_root.join(run).join("run.json");
    RunStore::new(path)
        .update_with_access(|state, access| {
            let invalid = |detail: String| {
                RunStoreError::mutation(format!(
                    "sprint open requires native verified planned state: {detail}"
                ))
            };
            if state.run != run || !state.status.is(RunStatus::Planned) {
                return Err(invalid(format!("run {run} has status {}", state.status)));
            }
            if state.plan != format!(".shepherd/runs/{run}/plan.md") {
                return Err(invalid(
                    "run plan pointer differs from the native planned checkpoint".into(),
                ));
            }
            let files = AnchoredFiles::open(context, run, access)
                .map_err(|source| invalid(cli_message(source)))?;
            let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)
                .map_err(|source| invalid(cli_message(source)))?;
            let authority =
                native_authority::validate_state(context, run, state, Some(project_id.as_str()))
                    .map_err(|source| invalid(cli_message(source)))?;
            let pre_checkpoint = authority
                .pre
                .as_ref()
                .ok_or_else(|| invalid("native pre checkpoint is absent".into()))?;
            let post_checkpoint = authority
                .post
                .as_ref()
                .ok_or_else(|| invalid("native post checkpoint is absent".into()))?;
            if authority.orientation_epoch == 0
                || pre_checkpoint.epoch != authority.orientation_epoch
                || pre_checkpoint.run_revision != state.orientation_epoch
                || post_checkpoint.epoch != authority.orientation_epoch
                || post_checkpoint.run_revision != state.orientation_epoch
            {
                return Err(invalid(
                    "native checkpoint epochs do not match the planned run".into(),
                ));
            }

            let pre_context = read_pre_context(context, run, state, &files, &authority)
                .map_err(|source| invalid(cli_message(source)))?;
            if pre_checkpoint.manifest_sha256 != sha256_hex(&pre_context.manifest_bytes)
                || pre_checkpoint.pre_sha256 != sha256_hex(&pre_context.pre_bytes)
            {
                return Err(invalid(
                    "native pre checkpoint digests changed after Planning".into(),
                ));
            }
            let critic = collect_final_critic(context, run, &files, &authority, &pre_context)
                .map_err(|source| invalid(cli_message(source)))?;
            let decision = validate_critic(
                &critic.report,
                &critic.source,
                &critic.result_bytes,
                &pre_context,
            )
            .map_err(|source| invalid(cli_message(source)))?;
            if decision != "GREEN" || critic.source.nonce != post_checkpoint.critic_nonce {
                return Err(invalid(
                    "native Critic verdict does not match the planned checkpoint".into(),
                ));
            }

            let post_opened = files
                .read_file(POST_FILE, "planned execution verification")
                .map_err(|source| invalid(cli_message(source)))?;
            if post_opened.sha256 != post_checkpoint.post_sha256 {
                return Err(invalid(
                    "native post checkpoint digest changed after Planning".into(),
                ));
            }
            let post: OrientationPost =
                decode_json(&post_opened.bytes, "planned execution verification")
                    .map_err(|source| invalid(cli_message(source)))?;
            validate_post_shape(&post, context, run, &files, &authority)
                .map_err(|source| invalid(cli_message(source)))?;
            let expected_sources = pre_context
                .manifest
                .sources
                .iter()
                .cloned()
                .chain(std::iter::once(critic.source))
                .collect::<Vec<_>>();
            if !post.accepted
                || post.verdict != "GREEN"
                || post.created_at != post_checkpoint.created_at
                || post.manifest_sha256 != sha256_hex(&pre_context.manifest_bytes)
                || post.pre_sha256 != sha256_hex(&pre_context.pre_bytes)
                || post.sources != expected_sources
            {
                return Err(invalid(
                    "native post artifact no longer proves the planned run".into(),
                ));
            }
            let binding = post
                .planning
                .as_ref()
                .ok_or_else(|| invalid("native post lacks plan-v2 readiness".into()))?;
            verify_critic_planning(&critic.report, binding)
                .map_err(|source| invalid(cli_message(source)))?;
            let execution_head = crate::cmd::planning::execution_head(context)
                .map_err(|source| invalid(cli_message(source)))?;
            let (topology, planning) = reverify_plan_readiness(context, state, binding)
                .map_err(|source| invalid(cli_message(source)))?;
            if post.planning.as_ref() != Some(&planning) {
                return Err(invalid(
                    "plan-v2 inputs changed or have no native planning checkpoint".into(),
                ));
            }
            let lanes =
                execution_lanes(state, &topology).map_err(|source| invalid(cli_message(source)))?;
            if crate::cmd::planning::execution_head(context)
                .map_err(|source| invalid(cli_message(source)))?
                != execution_head
            {
                return Err(invalid(
                    "Git HEAD changed while opening the verified plan".into(),
                ));
            }
            state.lanes = lanes;
            state
                .extra
                .insert("planning_execution_head".into(), execution_head.into());
            state.status = RunStatus::Executing.into();
            Ok(())
        })
        .map_err(|source| error(source.to_string()))
}

fn execution_lanes(state: &RunState, topology: &PlanTopology) -> Result<Vec<LaneState>, CliError> {
    if topology.lanes.is_empty() {
        return Err(error("verified execution topology has no lanes"));
    }
    let expected = topology
        .lanes
        .iter()
        .map(|lane| lane.id.as_str())
        .collect::<BTreeSet<_>>();
    let mut registered = BTreeMap::new();
    for lane in &state.lanes {
        if !expected.contains(lane.id.as_str()) {
            return Err(error(format!(
                "pre-execution lane `{}` is absent from the verified topology",
                lane.id
            )));
        }
        if registered.insert(lane.id.as_str(), lane).is_some() {
            return Err(error(format!("duplicate pre-execution lane `{}`", lane.id)));
        }
        if lane.plan != format!("lanes/{}/plan.md", lane.id)
            || !lane.state.is(LaneStatus::Pending)
            || lane.accepted_commit.is_some()
            || lane.merged
        {
            return Err(error(format!(
                "pre-execution lane `{}` is not a matching pristine pending registration",
                lane.id
            )));
        }
    }
    Ok(topology
        .lanes
        .iter()
        .map(|lane| {
            registered.get(lane.id.as_str()).map_or_else(
                || LaneState {
                    id: lane.id.clone(),
                    plan: format!("lanes/{}/plan.md", lane.id),
                    worktree: String::new(),
                    branch: String::new(),
                    state: LaneStatus::Pending.into(),
                    accepted_commit: None,
                    merged: false,
                    updated_at: 0,
                    extra: BTreeMap::new(),
                },
                |existing| (*existing).clone(),
            )
        })
        .collect())
}

fn recompute_post(
    context: &ExecutionContext,
    run: &str,
    state: &RunState,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    pre_context: &PreContext,
    planning: Option<PlanReadiness>,
) -> Result<OrientationPost, CliError> {
    let critic = collect_final_critic(context, run, files, authority, pre_context)?;
    let decision = validate_critic(
        &critic.report,
        &critic.source,
        &critic.result_bytes,
        pre_context,
    )?;
    if let Some(binding) = &planning {
        verify_critic_planning(&critic.report, binding)?;
    }
    let created_at = context.now_unix_millis();
    let post = OrientationPost {
        schema: POST_SCHEMA.into(),
        phase: "post".into(),
        run: run.into(),
        run_dir: files.run_dir.display().to_string(),
        project_root: files.project_root.display().to_string(),
        incarnation: authority.run_incarnation.clone(),
        orientation_epoch: authority.orientation_epoch,
        created_at,
        manifest_sha256: sha256_hex(&pre_context.manifest_bytes),
        pre_sha256: sha256_hex(&pre_context.pre_bytes),
        accepted: decision == "GREEN",
        sources: pre_context
            .manifest
            .sources
            .iter()
            .cloned()
            .chain(std::iter::once(critic.source.clone()))
            .collect(),
        verdict: decision,
        planning,
    };
    let post_bytes = encode_json(&post)?;
    let input_archive = archive_orientation_inputs(
        files,
        run,
        authority.orientation_epoch,
        &post.sources,
        pre_context.children.iter().chain(std::iter::once(&critic)),
        [
            (MANIFEST_FILE, pre_context.manifest_bytes.as_slice()),
            (PRE_FILE, pre_context.pre_bytes.as_slice()),
            (POST_FILE, post_bytes.as_slice()),
        ],
    )?;
    files.write_replace(POST_FILE, &post_bytes)?;
    native_authority::record_post(
        context,
        run,
        authority,
        PostCheckpoint {
            epoch: authority.orientation_epoch,
            created_at,
            run_revision: state.orientation_epoch,
            post_sha256: sha256_hex(&post_bytes),
            critic_nonce: critic.source.nonce.clone(),
            input_archive: Some(input_archive),
        },
    )?;
    Ok(post)
}

fn verify_critic_planning(
    critic: &OrientationReport,
    planning: &PlanReadiness,
) -> Result<(), CliError> {
    for (relative, digest) in &planning.run_artifacts {
        let path = format!(".shepherd/runs/{}/{relative}", critic.run);
        if !critic
            .evidence
            .iter()
            .any(|entry| entry.path == path && entry.sha256 == *digest)
        {
            return Err(error(format!(
                "Critic did not bind exact planning artifact: {relative}"
            )));
        }
    }
    Ok(())
}

/// Dispatch-time plan custody. This does not rerun environment commands or
/// compare implementation files to their pre-execution snapshots, and it never
/// acquires RunStore's lock recursively.
pub(crate) fn verified_execution_topology(
    context: &ExecutionContext,
    state: &RunState,
) -> Result<shepherd::plan::PlanTopology, CliError> {
    if !state.status.is(RunStatus::Executing) {
        return Err(error(
            "dispatch planning checkpoint requires an executing run",
        ));
    }
    let binding = read_planning_checkpoint(context, state)?;
    verify_plan_binding(context, state, &binding)
}

/// Diagnostic verification of a planned/executing run uses the seed already
/// authenticated by its native checkpoint, then re-runs the current probes.
pub(crate) fn verify_checkpointed_plan(
    context: &ExecutionContext,
    state: &RunState,
) -> Result<(shepherd::plan::PlanTopology, PlanReadiness), CliError> {
    let binding = read_planning_checkpoint(context, state)?;
    reverify_plan_readiness(context, state, &binding)
}

fn read_planning_checkpoint(
    context: &ExecutionContext,
    state: &RunState,
) -> Result<PlanReadiness, CliError> {
    if !matches!(state.status.as_str(), "planned" | "executing")
        || state.plan != format!(".shepherd/runs/{}/plan.md", state.run)
    {
        return Err(error(
            "native planning checkpoint requires the canonical planned/executing run",
        ));
    }
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)?;
    let authority =
        native_authority::validate_state(context, &state.run, state, Some(project_id.as_str()))?;
    let checkpoint = authority
        .post
        .ok_or_else(|| error("native planning checkpoint is absent"))?;
    let bytes =
        crate::cmd::planning::read_regular(&context.runs_root.join(&state.run).join(POST_FILE))?;
    if sha256_hex(&bytes) != checkpoint.post_sha256 {
        return Err(error("native planning checkpoint digest changed"));
    }
    let post: OrientationPost = decode_json(&bytes, "native planning checkpoint")?;
    if post.schema != POST_SCHEMA
        || post.run != state.run
        || post.incarnation != state.run_incarnation
        || post.orientation_epoch != state.orientation_epoch
        || !post.accepted
        || post.verdict != "GREEN"
    {
        return Err(error(
            "native planning checkpoint does not match the executing run",
        ));
    }
    post.planning
        .ok_or_else(|| error("native planning checkpoint has no plan-v2 binding"))
}

fn ensure_planning_state(state: &RunState, run: &str) -> Result<(), CliError> {
    if state.run != run {
        return Err(error(format!(
            "run state belongs to `{}`, expected `{run}`",
            state.run
        )));
    }
    if !state.status.is(RunStatus::Planted) {
        return Err(error(format!(
            "orientation requires planted run {run}, found {}",
            state.status
        )));
    }
    if state.seed.is_empty() {
        return Err(error("orientation requires run.json seed path"));
    }
    safe_relative(&state.seed).map_err(error)?;
    Ok(())
}

struct PlanningDispatches {
    engineer: DispatchRecord,
    pending: Vec<PendingDispatch>,
    records: Vec<DispatchRecord>,
}

/// The caller supplies the current native parent/pending snapshot while holding
/// its dispatch lock. Never reacquire that lock here: prepare, claim and activate
/// all use this same no-mutation admission boundary.
pub(crate) fn verify_critic_pre(
    workspace: &Path,
    runs_root: &Path,
    state: &RunState,
    engineer: &DispatchRecord,
    pending: &PendingDispatch,
) -> Result<(), CliError> {
    let inputs =
        ContextInputs::from_environment(workspace).map_err(|source| error(source.to_string()))?;
    let context = ExecutionContext::discover(inputs).map_err(|source| error(source.to_string()))?;
    if context.workspace_root != workspace
        || context.runs_root != runs_root
        || !state.status.is(RunStatus::Planted)
        || state.run != engineer.run.as_str()
        || engineer.role != Role::Engineer
        || engineer.state != DispatchState::Active
        || engineer.lane.is_some()
        || engineer.parent_agent_id.is_some()
    {
        return Err(error(
            "Critic requires the exact active planning Engineer and workspace",
        ));
    }
    exact_pending(std::slice::from_ref(pending), engineer)?;
    let workspace_id = DispatchService::with_project_root(
        DispatchStore::new(&context.runs_root),
        engineer.project_id.clone(),
        workspace,
    )
    .project_filesystem_id()
    .map_err(|source| error(source.to_string()))?;
    if pending.project_filesystem_id != workspace_id
        || pending.work_kind != WorkKind::Planning
        || pending.caller_role != Role::Shepherd
        || !pending.run_status.is(RunStatus::Planted)
    {
        return Err(error(
            "Critic parent pending launch belongs to another planning context",
        ));
    }
    let authority = native_authority::validate_state(
        &context,
        &state.run,
        state,
        Some(engineer.project_id.as_str()),
    )?;
    let checkpoint = authority
        .pre
        .as_ref()
        .ok_or_else(|| error("Critic dispatch requires accepted native orientation pre"))?;
    if checkpoint.engineer_nonce != engineer.nonce
        || engineer.run_incarnation != authority.run_incarnation
    {
        return Err(error("Critic pre belongs to another Engineer incarnation"));
    }
    let run_root = context.runs_root.join(&state.run);
    let manifest_opened = read_snapshot(&run_root, MANIFEST_FILE, "Critic orientation manifest")?;
    let pre_opened = read_snapshot(&run_root, PRE_FILE, "Critic orientation pre")?;
    let manifest: OrientationManifest =
        decode_json(&manifest_opened.bytes, "Critic orientation manifest")?;
    let pre: OrientationPre = decode_json(&pre_opened.bytes, "Critic orientation pre")?;
    if manifest_opened.sha256 != checkpoint.manifest_sha256
        || pre_opened.sha256 != checkpoint.pre_sha256
        || manifest.schema != MANIFEST_SCHEMA
        || pre.schema != PRE_SCHEMA
        || manifest.phase != "pre"
        || pre.phase != "pre"
        || !pre.accepted
        || manifest.run != state.run
        || pre.run != state.run
        || manifest.incarnation != authority.run_incarnation
        || pre.incarnation != authority.run_incarnation
        || manifest.orientation_epoch != authority.orientation_epoch
        || pre.orientation_epoch != authority.orientation_epoch
        || checkpoint.epoch != authority.orientation_epoch
        || manifest.created_at != checkpoint.created_at
        || pre.created_at != checkpoint.created_at
        || pre.manifest_sha256 != manifest_opened.sha256
        || pre.sources != manifest.sources
        || manifest.run_dir != run_root.display().to_string()
        || pre.run_dir != manifest.run_dir
        || manifest.project_root != context.primary_root.display().to_string()
        || pre.project_root != manifest.project_root
    {
        return Err(error(
            "Critic pre is not the exact accepted native checkpoint",
        ));
    }
    let phase0 = manifest
        .sources
        .iter()
        .filter(|source| source.kind == "phase0")
        .collect::<Vec<_>>();
    if phase0.len() != 1
        || phase0[0].id != engineer.agent_id.as_str()
        || phase0[0].nonce != engineer.nonce
        || phase0[0].revision != engineer.revision
        || phase0[0].session_id != engineer.session_id.as_str()
        || phase0[0].stopped_at.is_some()
    {
        return Err(error(
            "Critic pre does not bind the current published Engineer",
        ));
    }
    for source in &manifest.sources {
        let artifact = read_snapshot(
            workspace,
            &format!(".shepherd/runs/{}/{}", state.run, source.path),
            "Critic pre source",
        )?;
        let record = read_snapshot(&run_root, &source.dispatch_path, "Critic pre dispatch")?;
        let pending = read_snapshot(&run_root, &source.pending_path, "Critic pre pending")?;
        if artifact.sha256 != source.sha256
            || record.sha256 != source.dispatch_sha256
            || pending.sha256 != source.pending_sha256
        {
            return Err(error("Critic pre source changed after native acceptance"));
        }
    }
    Ok(())
}

/// A singleton filename or role label is not planning authority. Match the
/// current published registry claim, active native launch, root, and workspace.
fn planning_dispatches(
    context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
) -> Result<PlanningDispatches, CliError> {
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)?;
    let registry = Registry::open(&context.registry_path, OpenMode::ReadOnly)
        .map_err(|source| error(format!("open native Engineer registry: {source}")))?;
    let claim = registry
        .load_dispatch_singleton(project_id.as_str(), run, "engineer", "__run__")
        .map_err(|source| error(format!("read native Engineer singleton: {source}")))?
        .ok_or_else(|| error("orientation requires the current published Engineer singleton"))?;
    if claim.publication_state != Some(SingletonPublicationState::Published) {
        return Err(error("Engineer singleton publication is not committed"));
    }
    let run_id = RunId::new(run).map_err(|source| error(source.to_string()))?;
    let store = DispatchStore::new(&context.runs_root);
    let (inventory, engineer, root, custody) = store
        .with_run_access(&run_id, files.access, |native| {
            let inventory = native.inventory()?;
            let engineer = inventory
                .records
                .iter()
                .find(|record| {
                    record.role == Role::Engineer && record.agent_id.as_str() == claim.agent_id
                })
                .cloned()
                .ok_or_else(|| {
                    DispatchStoreError::Reconciliation("published Engineer record is absent".into())
                })?;
            let root = native.load_root_binding(&engineer.root_session_id)?;
            let custody = match native.load_review_custody(&engineer.agent_id) {
                Ok(custody) => Some(custody),
                Err(DispatchStoreError::Io { source, .. })
                    if source.kind() == std::io::ErrorKind::NotFound =>
                {
                    None
                }
                Err(source) => return Err(source),
            };
            Ok((inventory, engineer, root, custody))
        })
        .map_err(|source| error(source.to_string()))?;
    let current_root = store
        .load_current_root_binding(&engineer.root_session_id)
        .map_err(|source| error(source.to_string()))?;
    let workspace =
        DispatchService::with_project_root(store, project_id.clone(), &context.workspace_root)
            .project_filesystem_id()
            .map_err(|source| error(source.to_string()))?;
    let now = context.now_unix_millis();
    if root != current_root
        || root.project_id != project_id
        || root.run != run_id
        || root.role != Role::Shepherd
        || !root.mode.is_planting()
        || root.harness != engineer.harness
        || root.project_filesystem_id.as_deref() != Some(workspace.as_str())
        || now < root.bound_at
        || now >= root.expires_at
        || engineer.run_incarnation != authority.run_incarnation
        || engineer.project_id != project_id
        || engineer.run != run_id
        || engineer.lane.is_some()
        || engineer.parent_agent_id.is_some()
        || engineer.state != DispatchState::Active
        || engineer.stopped_at.is_some()
        || now < engineer.started_at
        || now >= engineer.lease_expires_at
        || custody
            .as_ref()
            .is_some_and(|custody| custody.state != ReviewCustodyState::Active)
        || inventory.records.iter().any(|record| {
            record.role == Role::Engineer
                && record.state == DispatchState::Active
                && record.agent_id != engineer.agent_id
        })
    {
        return Err(error(
            "Engineer is not the exact live planning root's current singleton",
        ));
    }
    let dispatch_path = format!("dispatch/{}.json", engineer.agent_id);
    let opened = files.read_file(&dispatch_path, "published Engineer dispatch")?;
    let decoded: DispatchRecord = decode_json(&opened.bytes, "published Engineer dispatch")?;
    if decoded != engineer
        || claim.agent_id != engineer.agent_id.as_str()
        || claim.project_id != project_id.as_str()
        || claim.run_id != run
        || claim.role != "engineer"
        || claim.lane_id.is_some()
        || claim.parent_agent_id.is_some()
        || claim.session_id != engineer.session_id.as_str()
        || claim.write_scope != engineer.write_scope
        || claim.harness != engineer.harness.to_string()
        || claim.agent_type != engineer.agent_type.as_str()
        || claim.publication_nonce.as_deref() != Some(&engineer.nonce)
        || claim.record_sha256.as_deref() != Some(&opened.sha256)
        || claim.record_path.as_deref() != Some(format!("{run}/{dispatch_path}").as_str())
    {
        return Err(error(
            "Engineer record differs from its committed native singleton",
        ));
    }
    let pending = exact_pending(&inventory.pending, &engineer)?;
    if !pending.run_status.is(RunStatus::Planted)
        || pending.work_kind != WorkKind::Planning
        || pending.caller_role != Role::Shepherd
        || pending.parent_dispatch_id.is_some()
        || pending.project_filesystem_id != workspace
        || !engineer
            .write_scope
            .contains(&format!(".shepherd/runs/{run}/phase0.md"))
        || !engineer
            .write_scope
            .contains(&format!(".shepherd/runs/{run}/plan.md"))
    {
        return Err(error(
            "Engineer lacks its native planning launch and exact phase0/plan scope",
        ));
    }
    Ok(PlanningDispatches {
        engineer,
        pending: inventory.pending,
        records: inventory.records,
    })
}

fn exact_pending<'a>(
    pending: &'a [PendingDispatch],
    record: &DispatchRecord,
) -> Result<&'a PendingDispatch, CliError> {
    let mut matches = pending
        .iter()
        .filter(|pending| pending.expected_attachment.agent_id == record.agent_id);
    let pending = matches
        .next()
        .ok_or_else(|| error("orientation source has no native pending launch"))?;
    if matches.next().is_some()
        || pending.launch_state != PendingLaunchState::Active
        || pending.project_id != record.project_id
        || pending.run != record.run
        || pending.root_session_id != record.root_session_id
        || pending.role != record.role
        || pending.lane != record.lane
        || pending.expected_attachment.target != record.harness
        || pending.expected_child_session_id != record.session_id
        || pending.parent_dispatch_id.as_ref().map(|id| id.as_str())
            != record.parent_agent_id.as_ref().map(|id| id.as_str())
        || pending.activated_at != Some(record.started_at)
        || pending.expires_at != record.lease_expires_at
        || record.result_artifact.as_deref() != Some(pending.result_artifact.as_str())
        || pending
            .write_scope
            .iter()
            .map(|scope| scope.as_str())
            .collect::<Vec<_>>()
            != record
                .write_scope
                .iter()
                .map(String::as_str)
                .collect::<Vec<_>>()
        || record.attachment_nonce.as_deref() != Some(format_digest(pending.nonce_sha256).as_str())
        || record.startup_attachment.as_ref().is_none_or(|attachment| {
            attachment.skill != pending.expected_attachment.startup_skill
                || attachment.bundle_digest
                    != format_digest(pending.expected_attachment.skill_bundle_sha256)
        })
    {
        return Err(error(
            "orientation source differs from its exact active native pending launch",
        ));
    }
    Ok(pending)
}

fn pending_source(
    files: &AnchoredFiles<'_>,
    pending: &PendingDispatch,
) -> Result<(String, String), CliError> {
    let path = format!(
        "dispatch/pending-{}.json",
        format_digest(pending.launch_id_hash)
    );
    let opened = files.read_file(&path, "native orientation pending")?;
    let decoded: PendingDispatch = decode_json(&opened.bytes, "native orientation pending")?;
    if decoded != *pending {
        return Err(error(
            "native pending changed during orientation collection",
        ));
    }
    Ok((path, opened.sha256))
}

fn run_artifact<'a>(run: &str, path: &'a str) -> Result<&'a str, CliError> {
    let relative = path
        .strip_prefix(&format!(".shepherd/runs/{run}/"))
        .ok_or_else(|| error("native orientation artifact is outside its canonical run prefix"))?;
    safe_relative(relative).map_err(error)?;
    Ok(relative)
}

fn collect_pre_sources(
    context: &ExecutionContext,
    run: &str,
    state: &RunState,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    reject_critics: bool,
) -> Result<CollectedPre, CliError> {
    let phase0 = files.read_authoring_file("phase0.md", "phase0.md")?;
    let seed = files.read_project_file(&state.seed, "verified seed")?;
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)?;
    let mut children = Vec::new();
    let mut critics = Vec::new();
    let native = planning_dispatches(context, run, files, authority)?;
    let engineer = &native.engineer;
    for record in &native.records {
        if record.parent_agent_id.as_ref() != Some(&engineer.agent_id) {
            continue;
        }
        match record.role {
            Role::Auditor | Role::Discovery => {
                children.push(read_child(
                    context,
                    run,
                    files,
                    authority,
                    record,
                    exact_pending(&native.pending, record)?,
                    &project_id,
                )?);
            }
            Role::Critic => critics.push(record),
            _ => {}
        }
    }
    if reject_critics && !critics.is_empty() {
        return Err(error("Critic dispatch exists before native pre acceptance"));
    }
    let engineer_dispatch_relative = format!("dispatch/{}.json", engineer.agent_id.as_str());
    let engineer_dispatch = files.read_file(&engineer_dispatch_relative, "Engineer dispatch")?;
    let decoded: DispatchRecord = decode_json(&engineer_dispatch.bytes, "Engineer dispatch")?;
    if decoded != *engineer {
        return Err(error(
            "Engineer dispatch changed while orientation was collecting it",
        ));
    }
    if !children.iter().any(|child| child.source.kind == "auditor") {
        return Err(error("orientation pre is missing a stopped Auditor result"));
    }
    if !children
        .iter()
        .any(|child| child.source.kind == "discovery")
    {
        return Err(error(
            "orientation pre is missing a stopped Discovery result",
        ));
    }
    children.sort_by(|left, right| left.source.id.cmp(&right.source.id));
    let mut source_ids = BTreeSet::new();
    let mut source_paths = BTreeSet::new();
    let mut result_hashes = BTreeSet::new();
    for child in &children {
        if !source_ids.insert(child.source.id.as_str())
            || !source_paths.insert(child.source.path.as_str())
            || !result_hashes.insert(child.source.sha256.as_str())
        {
            return Err(error(
                "orientation pre contains duplicate native source identity, path, or hash",
            ));
        }
    }
    validate_phase0(&phase0.bytes, run, &state.seed, &seed.sha256, &children)?;
    let (pending_path, pending_sha256) =
        pending_source(files, exact_pending(&native.pending, engineer)?)?;
    let phase0_source = source_from_record(
        engineer,
        "phase0",
        "engineer",
        "phase0.md".into(),
        phase0.sha256.clone(),
        Vec::new(),
        engineer_dispatch_relative,
        engineer_dispatch.sha256,
        pending_path,
        pending_sha256,
    );
    let mut claims = BTreeMap::new();
    let mut assumptions = BTreeSet::new();
    for child in &children {
        add_claims(&mut claims, child)?;
        add_assumptions(&mut assumptions, child)?;
    }
    Ok(CollectedPre {
        phase0: phase0_source,
        children,
        phase0_bytes: phase0.bytes,
        claims,
        assumptions,
        engineer_nonce: engineer.nonce.clone(),
    })
}

fn read_pre_context(
    context: &ExecutionContext,
    run: &str,
    state: &RunState,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
) -> Result<PreContext, CliError> {
    let checkpoint = authority
        .pre
        .as_ref()
        .ok_or_else(|| error("native pre checkpoint is absent"))?;
    verify_checkpoint_archives(files, run, checkpoint, authority.post.as_ref())?;
    let manifest_opened = files.read_file(MANIFEST_FILE, "orientation manifest")?;
    let manifest: OrientationManifest =
        decode_json(&manifest_opened.bytes, "orientation manifest")?;
    validate_manifest_shape(&manifest, context, run, files, authority)?;
    let pre_opened = files.read_file(PRE_FILE, "orientation pre verification")?;
    let pre: OrientationPre = decode_json(&pre_opened.bytes, "orientation pre verification")?;
    validate_pre_shape(
        &pre,
        &manifest,
        context,
        run,
        files,
        authority,
        &manifest_opened,
    )?;
    let collected = collect_pre_sources(context, run, state, files, authority, false)?;
    let expected_sources: Vec<_> = std::iter::once(collected.phase0)
        .chain(collected.children.iter().map(|child| child.source.clone()))
        .collect();
    if manifest.sources != expected_sources || pre.sources != expected_sources {
        return Err(error(
            "native orientation sources changed after pre acceptance",
        ));
    }
    Ok(PreContext {
        manifest,
        manifest_bytes: manifest_opened.bytes,
        pre_bytes: pre_opened.bytes,
        children: collected.children,
        phase0_bytes: collected.phase0_bytes,
        claims: collected.claims,
        assumptions: collected.assumptions,
    })
}

fn collect_final_critic(
    context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    pre_context: &PreContext,
) -> Result<ChildEvidence, CliError> {
    let native = planning_dispatches(context, run, files, authority)?;
    let mut critics = native
        .records
        .iter()
        .filter(|record| {
            record.role == Role::Critic
                && record.parent_agent_id.as_ref() == Some(&native.engineer.agent_id)
        })
        .collect::<Vec<_>>();
    if critics.is_empty() {
        return Err(error("orientation post requires a native Critic result"));
    }
    if critics.iter().any(|record| {
        record.state != DispatchState::Stopped
            || record.run_incarnation != authority.run_incarnation
    }) {
        return Err(error(
            "orientation post found an active or stale Critic dispatch",
        ));
    }
    critics.sort_by_key(|record| record.started_at);
    if critics
        .windows(2)
        .any(|pair| pair[0].started_at == pair[1].started_at)
    {
        return Err(error("Critic dispatch timestamps are ambiguous"));
    }
    let critic = critics.last().expect("non-empty critics");
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)?;
    let evidence = read_child(
        context,
        run,
        files,
        authority,
        critic,
        exact_pending(&native.pending, critic)?,
        &project_id,
    )?;
    if evidence.report.orientation_pre_sha256.as_deref()
        != Some(&sha256_hex(&pre_context.pre_bytes))
    {
        return Err(error("Critic result is not bound to the accepted pre hash"));
    }
    if critic.started_at <= authority.pre.as_ref().expect("pre checkpoint").created_at {
        return Err(error(
            "Critic dispatch did not start after native pre checkpoint",
        ));
    }
    Ok(evidence)
}

#[allow(clippy::too_many_arguments)]
fn source_from_record(
    record: &DispatchRecord,
    kind: &str,
    role: &str,
    path: String,
    sha256: String,
    read_scope: Vec<String>,
    dispatch_path: String,
    dispatch_sha256: String,
    pending_path: String,
    pending_sha256: String,
) -> OrientationSource {
    OrientationSource {
        id: record.agent_id.to_string(),
        kind: kind.into(),
        role: role.into(),
        path,
        sha256,
        read_scope,
        write_scope: record.write_scope.clone(),
        project_id: record.project_id.to_string(),
        session_id: record.session_id.to_string(),
        incarnation: record.run_incarnation.clone(),
        revision: record.revision,
        started_at: record.started_at,
        lease_expires_at: record.lease_expires_at,
        stopped_at: record.stopped_at,
        nonce: record.nonce.clone(),
        dispatch_path,
        dispatch_sha256,
        pending_path,
        pending_sha256,
    }
}

fn read_child(
    context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    record: &DispatchRecord,
    pending: &PendingDispatch,
    expected_project_id: &ProjectId,
) -> Result<ChildEvidence, CliError> {
    let kind = record.role.as_str();
    if !matches!(record.role, Role::Auditor | Role::Discovery | Role::Critic) {
        return Err(error("non-orientation record reached orientation reader"));
    }
    if record.run.as_str() != run {
        return Err(error("dispatch record belongs to the wrong run"));
    }
    if record.project_id != *expected_project_id {
        return Err(error("dispatch record belongs to the wrong project"));
    }
    if record.run_incarnation != authority.run_incarnation {
        return Err(error(
            "dispatch record belongs to the wrong native run incarnation",
        ));
    }
    if record.state != DispatchState::Stopped {
        return Err(error(format!("{kind} dispatch is not stopped")));
    }
    if !pending.run_status.is(RunStatus::Planted)
        || pending.caller_role != Role::Engineer
        || pending.parent_dispatch_id.is_none()
        || pending.lane.is_some()
        || pending.work_kind
            != if record.role == Role::Discovery {
                WorkKind::Research
            } else {
                WorkKind::Review
            }
        || !record.write_scope.is_empty()
    {
        return Err(error(
            "orientation child lacks its exact read-only Engineer-owned launch",
        ));
    }
    let result_path = record
        .result_artifact
        .as_deref()
        .ok_or_else(|| error(format!("{kind} dispatch has no result artifact")))?;
    let result_path = run_artifact(run, result_path)?;
    let dispatch_relative = format!("dispatch/{}.json", record.agent_id.as_str());
    let dispatch_opened = files.read_file(&dispatch_relative, "dispatch record")?;
    let decoded: DispatchRecord = decode_json(&dispatch_opened.bytes, "dispatch record")?;
    if decoded != *record {
        return Err(error(
            "dispatch record changed while orientation was collecting it",
        ));
    }
    let result_opened = files.read_authoring_file(result_path, "orientation result")?;
    let report: OrientationReport = decode_json(&result_opened.bytes, "orientation result")?;
    if report.read_scope.iter().any(|path| {
        !pending
            .read_scope
            .iter()
            .any(|scope| scope.contains(path).unwrap_or(false))
    }) {
        return Err(error(
            "orientation report read scope exceeds its native issued scope",
        ));
    }
    let (pending_path, pending_sha256) = pending_source(files, pending)?;
    let source = source_from_record(
        record,
        kind,
        kind,
        result_path.into(),
        result_opened.sha256,
        report.read_scope.clone(),
        dispatch_relative,
        dispatch_opened.sha256,
        pending_path,
        pending_sha256,
    );
    validate_report(&report, &source, files, context, run)?;
    Ok(ChildEvidence {
        source,
        report,
        result_bytes: result_opened.bytes,
    })
}

fn validate_report(
    report: &OrientationReport,
    source: &OrientationSource,
    files: &AnchoredFiles<'_>,
    _context: &ExecutionContext,
    run: &str,
) -> Result<(), CliError> {
    if report.schema != REPORT_SCHEMA
        || report.run != run
        || report.result_id != source.id
        || report.kind != source.kind
        || report.role != source.role
    {
        return Err(error(format!(
            "result {} has the wrong typed identity",
            source.id
        )));
    }
    if report.status != "complete" || report.summary.trim().is_empty() {
        return Err(error(format!("result {} is not complete", source.id)));
    }
    if report.read_scope.is_empty()
        || report.read_scope != sorted_unique(&report.read_scope)
        || report
            .read_scope
            .iter()
            .any(|scope| safe_relative(scope).is_err())
    {
        return Err(error(format!(
            "result {} has an invalid read scope",
            source.id
        )));
    }
    if source.kind != "critic"
        && (report.orientation_pre_sha256.is_some() || report.verdict.is_some())
    {
        return Err(error(format!(
            "result {} has Critic-only fields",
            source.id
        )));
    }
    if report.evidence.is_empty() || report.claims.is_empty() {
        return Err(error(format!(
            "result {} must contain claims and evidence",
            source.id
        )));
    }
    let mut evidence = BTreeSet::new();
    for item in &report.evidence {
        if !evidence.insert(item.clone()) {
            return Err(error(format!(
                "result {} repeats an evidence citation",
                source.id
            )));
        }
        safe_relative(&item.path).map_err(error)?;
        if !inside_scope(&item.path, &report.read_scope) {
            return Err(error(format!(
                "result {} evidence escapes its read scope",
                source.id
            )));
        }
        let opened = files.read_project_file(&item.path, "orientation evidence")?;
        if item.sha256 != opened.sha256 {
            return Err(error(format!(
                "result {} evidence hash is stale",
                source.id
            )));
        }
        let lines = line_count(&opened.bytes).map_err(error)?;
        if item.line == 0 || item.line > lines {
            return Err(error(format!(
                "result {} evidence line is outside opened bytes",
                source.id
            )));
        }
        evidence.insert(item.clone());
    }
    let mut claim_ids = BTreeSet::new();
    let mut assumption_ids = BTreeSet::new();
    for assumption in &report.assumptions {
        if assumption.id.trim().is_empty()
            || assumption.statement.trim().is_empty()
            || !assumption_ids.insert(&assumption.id)
        {
            return Err(error(format!(
                "result {} has duplicate or empty assumption ids",
                source.id
            )));
        }
    }
    for claim in &report.claims {
        if claim.id.trim().is_empty()
            || claim.statement.trim().is_empty()
            || !claim_ids.insert(&claim.id)
        {
            return Err(error(format!(
                "result {} has duplicate or empty claim ids",
                source.id
            )));
        }
        if claim.evidence.is_empty() || claim.evidence.iter().any(|item| !evidence.contains(item)) {
            return Err(error(format!(
                "result {} claim has unbound evidence",
                source.id
            )));
        }
    }
    if report.caveats.iter().any(|caveat| caveat.trim().is_empty()) {
        return Err(error(format!("result {} has an empty caveat", source.id)));
    }
    Ok(())
}

fn add_claims(
    claims: &mut BTreeMap<String, String>,
    child: &ChildEvidence,
) -> Result<(), CliError> {
    for claim in &child.report.claims {
        if claims
            .insert(claim.id.clone(), child.source.id.clone())
            .is_some()
        {
            return Err(error(format!(
                "duplicate orientation claim id: {}",
                claim.id
            )));
        }
    }
    Ok(())
}

fn add_assumptions(
    assumptions: &mut BTreeSet<String>,
    child: &ChildEvidence,
) -> Result<(), CliError> {
    for assumption in &child.report.assumptions {
        if !assumptions.insert(assumption.id.clone()) {
            return Err(error(format!(
                "duplicate orientation assumption id: {}",
                assumption.id
            )));
        }
    }
    Ok(())
}

fn validate_critic(
    report: &OrientationReport,
    source: &OrientationSource,
    result_bytes: &[u8],
    pre_context: &PreContext,
) -> Result<String, CliError> {
    if report.orientation_pre_sha256.as_deref() != Some(&sha256_hex(&pre_context.pre_bytes)) {
        return Err(error("Critic result did not receive the accepted pre hash"));
    }
    let mut critic_claim_ids = BTreeSet::new();
    for claim in &report.claims {
        if pre_context.claims.contains_key(&claim.id) || !critic_claim_ids.insert(&claim.id) {
            return Err(error(format!(
                "Critic result repeats an orientation claim id: {}",
                claim.id
            )));
        }
    }
    if report
        .assumptions
        .iter()
        .any(|assumption| pre_context.assumptions.contains(&assumption.id))
    {
        return Err(error("Critic result repeats an orientation assumption id"));
    }
    let verdict = report
        .verdict
        .as_ref()
        .and_then(Value::as_object)
        .ok_or_else(|| error("Critic result has no typed verdict"))?;
    let expected = [
        "decision",
        "findings",
        "corrections",
        "blockers",
        "citations",
    ];
    if verdict.len() != expected.len() || expected.iter().any(|key| !verdict.contains_key(*key)) {
        return Err(error("Critic verdict has non-canonical fields"));
    }
    let decision = verdict
        .get("decision")
        .and_then(Value::as_str)
        .ok_or_else(|| error("Critic decision is not a string"))?;
    if !matches!(decision, "GREEN" | "RED") {
        return Err(error("Critic decision must be GREEN or RED"));
    }
    for key in ["findings", "corrections", "blockers"] {
        if !verdict.get(key).is_some_and(Value::is_array) {
            return Err(error(format!("Critic {key} is not a list")));
        }
    }
    let citations = verdict
        .get("citations")
        .and_then(Value::as_array)
        .ok_or_else(|| error("Critic citations are not a list"))?;
    if citations.is_empty() {
        return Err(error("Critic citations are empty"));
    }
    let mut seen = BTreeSet::new();
    let mut sources = BTreeMap::new();
    let phase0_source = pre_context
        .manifest
        .sources
        .first()
        .ok_or_else(|| error("orientation manifest has no phase0 source"))?;
    sources.insert(
        phase0_source.id.clone(),
        (phase0_source, pre_context.phase0_bytes.as_slice()),
    );
    for child in &pre_context.children {
        sources.insert(
            child.source.id.clone(),
            (&child.source, child.result_bytes.as_slice()),
        );
    }
    let critic_key = source.id.clone();
    sources.insert(critic_key.clone(), (source, result_bytes));
    for citation in citations {
        let object = citation
            .as_object()
            .ok_or_else(|| error("Critic citation is not an object"))?;
        let identity = if let Some(claim_id) = object.get("claim_id").and_then(Value::as_str) {
            if object.len() != 1
                || !pre_context.claims.contains_key(claim_id)
                    && !report.claims.iter().any(|claim| claim.id == claim_id)
            {
                return Err(error("Critic citation has an unresolved claim id"));
            }
            if report.claims.iter().any(|claim| claim.id == claim_id) {
                return Err(error("Critic citation cannot cite its own claim"));
            }
            format!("claim:{claim_id}")
        } else {
            if object.len() != 4
                || !object.contains_key("source_id")
                || !object.contains_key("path")
                || !object.contains_key("sha256")
                || !object.contains_key("line")
            {
                return Err(error("Critic source citation is not exact"));
            }
            let source_id = object["source_id"]
                .as_str()
                .ok_or_else(|| error("Critic citation source id is not a string"))?;
            if source_id == source.id {
                return Err(error("Critic citation cannot cite its own result"));
            }
            let (bound, bytes) = sources
                .get(source_id)
                .ok_or_else(|| error("Critic citation names an unbound source"))?;
            if object["path"] != bound.path || object["sha256"] != bound.sha256 {
                return Err(error("Critic citation substituted a bound source"));
            }
            let line = object["line"]
                .as_u64()
                .ok_or_else(|| error("Critic citation line is not an integer"))?;
            let line_count = line_count(bytes).map_err(error)? as u64;
            if line == 0 || line > line_count {
                return Err(error("Critic citation line is outside opened bytes"));
            }
            format!("source:{source_id}:{line}")
        };
        if !seen.insert(identity) {
            return Err(error("Critic citation is duplicated"));
        }
    }
    Ok(decision.into())
}

fn validate_manifest_shape(
    manifest: &OrientationManifest,
    _context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
) -> Result<(), CliError> {
    if manifest.schema != MANIFEST_SCHEMA
        || manifest.phase != "pre"
        || manifest.run != run
        || manifest.incarnation != authority.run_incarnation
        || manifest.orientation_epoch != authority.orientation_epoch
        || manifest.created_at <= 0
    {
        return Err(error("orientation manifest has the wrong native identity"));
    }
    validate_roots(&manifest.run_dir, &manifest.project_root, files)?;
    Ok(())
}

fn validate_pre_shape(
    pre: &OrientationPre,
    manifest: &OrientationManifest,
    _context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
    manifest_opened: &OpenedBytes,
) -> Result<(), CliError> {
    if pre.schema != PRE_SCHEMA
        || pre.phase != "pre"
        || pre.run != run
        || !pre.accepted
        || pre.incarnation != authority.run_incarnation
        || pre.orientation_epoch != authority.orientation_epoch
        || pre.created_at != manifest.created_at
        || pre.manifest_sha256 != manifest_opened.sha256
        || pre.sources != manifest.sources
    {
        return Err(error(
            "orientation pre verification is stale or substituted",
        ));
    }
    validate_roots(&pre.run_dir, &pre.project_root, files)
}

fn validate_post_shape(
    post: &OrientationPost,
    _context: &ExecutionContext,
    run: &str,
    files: &AnchoredFiles<'_>,
    authority: &RunAuthority,
) -> Result<(), CliError> {
    if post.schema != POST_SCHEMA
        || post.phase != "post"
        || post.run != run
        || post.incarnation != authority.run_incarnation
        || post.orientation_epoch != authority.orientation_epoch
        || post.created_at <= 0
    {
        return Err(error(
            "orientation post verification has the wrong native identity",
        ));
    }
    validate_roots(&post.run_dir, &post.project_root, files)
}

fn validate_roots(
    run_dir: &str,
    project_root: &str,
    files: &AnchoredFiles<'_>,
) -> Result<(), CliError> {
    if run_dir != files.run_dir.display().to_string()
        || project_root != files.project_root.display().to_string()
    {
        return Err(error(
            "orientation artifact changed its canonical run or project root",
        ));
    }
    Ok(())
}

fn validate_phase0(
    bytes: &[u8],
    run: &str,
    seed: &str,
    seed_sha256: &str,
    children: &[ChildEvidence],
) -> Result<(), CliError> {
    let text = std::str::from_utf8(bytes)
        .map_err(|source| error(format!("phase0.md is not UTF-8: {source}")))?;
    let lines = text.lines().collect::<Vec<_>>();
    let positions = PHASE0_HEADINGS
        .iter()
        .map(|heading| {
            let matches = lines.iter().filter(|line| *line == heading).count();
            if matches != 1 {
                return Err(error(format!(
                    "phase0.md must contain exactly one required heading: {heading}"
                )));
            }
            lines
                .iter()
                .position(|line| line == heading)
                .ok_or_else(|| error(format!("phase0.md is missing required heading: {heading}")))
        })
        .collect::<Result<Vec<_>, _>>()?;
    if positions.windows(2).any(|pair| pair[0] >= pair[1])
        || lines.first().copied() != Some(PHASE0_HEADINGS[0])
    {
        return Err(error("phase0.md headings are not canonical"));
    }
    for pair in positions[1..].windows(2) {
        if !lines[pair[0] + 1..pair[1]]
            .iter()
            .any(|line| !line.trim().is_empty())
        {
            return Err(error("phase0.md has an empty required section"));
        }
    }
    let last_start = *positions.last().expect("headings are non-empty");
    if !lines[last_start + 1..]
        .iter()
        .any(|line| !line.trim().is_empty())
    {
        return Err(error("phase0.md has an empty Critic loop section"));
    }
    let run_section = &lines[positions[1] + 1..positions[2]];
    let run_lines = run_section
        .iter()
        .filter_map(|line| line.strip_prefix("- run: "))
        .collect::<Vec<_>>();
    if run_lines.len() != 1 || run_lines[0] != run {
        return Err(error("phase0.md has no exact run line"));
    }
    let seed_path_lines = run_section
        .iter()
        .filter_map(|line| line.strip_prefix("- verified seed path: "))
        .collect::<Vec<_>>();
    if seed_path_lines.len() != 1 || seed_path_lines[0] != seed {
        return Err(error("phase0.md has no exact verified seed path line"));
    }
    let seed_hashes = run_section
        .iter()
        .filter_map(|line| line.strip_prefix("- seed hash: "))
        .collect::<Vec<_>>();
    if seed_hashes.len() != 1
        || seed_hashes[0].len() != 64
        || !seed_hashes[0]
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        || seed_hashes[0] != seed_sha256
    {
        return Err(error("phase0.md seed hash is absent, invalid, or stale"));
    }
    let planted_lines = run_section
        .iter()
        .filter_map(|line| line.strip_prefix("- planted observation: "))
        .collect::<Vec<_>>();
    if planted_lines.len() != 1 || planted_lines[0].trim().is_empty() {
        return Err(error("phase0.md has no exact planted observation line"));
    }
    for child in children {
        let heading = format!("### {}", child.source.id);
        let output = format!("- output path: {}", child.source.path);
        let scopes = child
            .source
            .read_scope
            .iter()
            .map(|scope| format!("- exact read scope: {scope}"))
            .collect::<Vec<_>>();
        if !lines.iter().any(|line| *line == heading)
            || !lines.iter().any(|line| *line == output)
            || scopes
                .iter()
                .any(|scope| !lines.iter().any(|line| line == scope))
        {
            return Err(error(format!(
                "phase0.md omits native brief {}",
                child.source.id
            )));
        }
    }
    Ok(())
}

fn verification_json(
    phase: &str,
    accepted: bool,
    run: &str,
    manifest: &OrientationManifest,
    pre_sha256: String,
    decision: Option<&String>,
    claims: &BTreeMap<String, String>,
) -> Result<Value, CliError> {
    let sources = manifest
        .sources
        .iter()
        .map(|source| json!({"id": source.id, "path": source.path, "sha256": source.sha256}))
        .collect::<Vec<_>>();
    Ok(json!({
        "accepted": accepted,
        "phase": phase,
        "run": run,
        "manifest_sha256": sha256_hex(&encode_json(manifest)?),
        "pre_sha256": pre_sha256,
        "critic_decision": decision,
        "claim_count": claims.len(),
        "sources": sources,
    }))
}

fn encode_json<T: Serialize>(value: &T) -> Result<Vec<u8>, CliError> {
    let mut bytes = serde_json::to_vec_pretty(value).map_err(|source| error(source.to_string()))?;
    bytes.push(b'\n');
    Ok(bytes)
}

fn decode_json<T: for<'de> Deserialize<'de>>(bytes: &[u8], label: &str) -> Result<T, CliError> {
    serde_json::from_slice(bytes)
        .map_err(|source| error(format!("{label} is invalid JSON: {source}")))
}

fn line_count(bytes: &[u8]) -> Result<usize, String> {
    let text = std::str::from_utf8(bytes)
        .map_err(|error| format!("evidence is not UTF-8 text: {error}"))?;
    Ok(text.lines().count())
}

fn sorted_unique(values: &[String]) -> Vec<String> {
    let mut result = values.to_vec();
    result.sort();
    result.dedup();
    result
}

fn inside_scope(path: &str, scopes: &[String]) -> bool {
    scopes.iter().any(|scope| {
        path == scope
            || (scope.ends_with("/**")
                && path.starts_with(&format!("{}/", &scope[..scope.len() - 3])))
    })
}

fn safe_relative(value: &str) -> Result<(), String> {
    if value.is_empty() || value.starts_with('/') || value.contains('\\') || value.contains('\0') {
        return Err(format!("unsafe relative path: {value:?}"));
    }
    if value
        .split('/')
        .any(|part| part.is_empty() || part == "." || part == "..")
    {
        return Err(format!("unsafe relative path: {value:?}"));
    }
    Ok(())
}

fn error(message: impl Into<String>) -> CliError {
    CliError::message(message.into())
}

fn cli_message(error: CliError) -> String {
    error
        .message_text()
        .unwrap_or("orientation validation failed")
        .into()
}

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

    #[test]
    fn empty_text_has_zero_lines() {
        assert_eq!(line_count(b"").expect("line count"), 0);
        assert_eq!(line_count(b"one\n").expect("line count"), 1);
    }

    #[test]
    fn relative_paths_reject_parent_components() {
        assert!(safe_relative("reports/result.json").is_ok());
        assert!(safe_relative("reports/../result.json").is_err());
    }
}