//! Radar module: snapshot builder, context-check decision tree, session
//! boundary recommendation, behavioral-drift evaluator, candidate-update
//! generation, and approval-step assembly.
//!
//! The module is decomposed across submodules by responsibility:
//!
//! - [`evaluation`] — `build_evaluation`, evidence helpers
//! - [`context_health`] — `build_context_health`, env-signal helpers
//! - [`context_check`] — `build_context_check_decision`, digest/delta/risk
//!   summary builders
//! - [`session_boundary`] — `build_session_boundary`, `build_approval_steps`
//! - [`behavioral_drift`] — `build_behavioral_drift`, individual signal
//!   builders, session-commit scanning
//! - [`candidates`] — `build_candidate_updates`, memory-candidate action
//!   builders
//!
//! The report types (`RadarStateReport`, `ContextCheckReport`,
//! `CheckpointResult`, `HandoverResult`) and the public entry points
//! (`checkpoint`, `evaluate`, `handover`, `run_context_check`) live here,
//! along with the shared struct definitions used across submodules.

mod behavioral_drift;
mod candidates;
mod context_check;
mod context_health;
mod evaluation;
mod session_boundary;
mod status;

pub(crate) use status::{MemoryStatus, NextStepStatus, WrapUpStatus};

use std::collections::BTreeMap;
use std::path::Path;
use std::process::ExitCode;

use anyhow::{bail, Result};
use serde::{Deserialize, Serialize};

use crate::commands::check;
use crate::db;
use crate::handoff::{self, BranchMode, GitState};
use crate::memory::entries as memory_entries;
use crate::output::CommandReport;
use crate::paths::state::StateLayout;
use crate::profile::{self, DEFAULT_PROFILE};
use crate::repo::marker as repo_marker;
use crate::session_boundary::SessionBoundaryRecommendation;
use crate::state::compiled as compiled_state;
use crate::state::consistency;
use crate::state::escalation as escalation_state;
use crate::state::projection_metadata;
use crate::state::runtime as runtime_state;
use crate::state::session as session_state;
use crate::state::session_gates;
use crate::state::work_stream_decay;
use crate::telemetry::{cost as telemetry_cost, host as host_telemetry};
use tracing::info_span;

pub(crate) const HANDOVER_SCHEMA_VERSION: u32 = 23;
// Checkpoint shares the bumped version: its payload shape also changed in
// this PR — handover-only fields (evaluation, approval_*, repo_native_checks)
// and the nested candidate_updates.memory subfield are now absent rather
// than emitted empty. Consumers keying on schema_version need a distinct
// signal for the strict-subset contract.
pub(crate) const CHECKPOINT_SCHEMA_VERSION: u32 = 23;
const CONTEXT_CHECK_SCHEMA_VERSION: u32 = 1;

/// How long a low-value context-check result is considered fresh.  Subsequent
/// low-value triggers (interval, supervisor-poll) that arrive inside this
/// window return a suppressed/throttled response without rebuilding checkpoint
/// or radar state (#627).
const CONTEXT_CHECK_THROTTLE_WINDOW_SECONDS: u64 = 120;

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ContextCheckTrigger {
    Interval,
    Manual,
    PreCompaction,
    IdleReset,
    Resume,
    SupervisorPoll,
}

impl ContextCheckTrigger {
    fn as_str(self) -> &'static str {
        match self {
            Self::Interval => "interval",
            Self::Manual => "manual",
            Self::PreCompaction => "pre_compaction",
            Self::IdleReset => "idle_reset",
            Self::Resume => "resume",
            Self::SupervisorPoll => "supervisor_poll",
        }
    }

    /// Returns `true` for triggers that are cheap refresh polls and can
    /// use the lightweight checkpoint-only codepath, skipping handover-only
    /// evaluation (repo-native checks, memory candidate generation, approval
    /// steps). Pre-compaction, manual, resume, and idle-reset always get the
    /// full path because they need fresh evidence.
    fn is_low_value(self) -> bool {
        matches!(self, Self::Interval | Self::SupervisorPoll)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ContextCheckAction {
    None,
    InjectDelta,
    CheckpointNow,
    FlushForCompaction,
    WrapUpRequired,
    Escalate,
}

impl ContextCheckAction {
    fn as_str(self) -> &'static str {
        match self {
            Self::None => "none",
            Self::InjectDelta => "inject_delta",
            Self::CheckpointNow => "checkpoint_now",
            Self::FlushForCompaction => "flush_for_compaction",
            Self::WrapUpRequired => "wrap_up_required",
            Self::Escalate => "escalate",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ContextCheckUrgency {
    Low,
    Moderate,
    High,
    Critical,
}

impl ContextCheckUrgency {
    fn as_str(self) -> &'static str {
        match self {
            Self::Low => "low",
            Self::Moderate => "moderate",
            Self::High => "high",
            Self::Critical => "critical",
        }
    }
}

#[derive(Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DriftAggregateStatus {
    Aligned,
    NeedsRecalibration,
    NoSignal,
}

#[derive(Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DriftSignalStatus {
    Aligned,
    Drift,
    NoSignal,
}

#[derive(Serialize)]
pub struct RadarStateReport {
    command: &'static str,
    mode: &'static str,
    schema_version: u32,
    ok: bool,
    decisions: ReportDecisions,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    handoff: HandoffState,
    execution_gates: session_gates::ExecutionGatesView,
    effective_memory: MemoryState,
    repo_native_checks: RepoNativeChecksState,
    git: Option<GitState>,
    checkout_state: Option<handoff::CheckoutStateView>,
    session_state: RadarSessionStateView,
    recovery: runtime_state::RuntimeRecoveryView,
    escalation: escalation_state::EscalationView,
    projection_telemetry: ProjectionTelemetryView,
    work_stream_decay: work_stream_decay::WorkStreamDecayView,
    context_health: ContextHealth,
    session_boundary: SessionBoundaryRecommendation,
    behavioral_drift: BehavioralDriftState,
    evaluation: RadarEvaluation,
    candidate_updates: CandidateUpdates,
    approval_intents: Vec<&'static str>,
    approval_steps: Vec<ApprovalStep>,
}

#[derive(Serialize)]
pub struct ContextCheckReport {
    command: &'static str,
    schema_version: u32,
    ok: bool,
    decisions: ReportDecisions,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    trigger: ContextCheckTrigger,
    action: ContextCheckAction,
    reason: &'static str,
    recommendation: String,
    urgency: ContextCheckUrgency,
    throttle: ContextCheckThrottle,
    /// Projection source fingerprint for cache-hit detection (#627).
    /// Callers can compare this to a previously seen value to determine
    /// whether before-prompt-build context has changed.
    source_fingerprint: String,
    payload: ContextCheckPayload,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    evidence: Vec<String>,
    session_state: RadarSessionStateView,
    execution_gates: session_gates::ExecutionGatesView,
    recovery: runtime_state::RuntimeRecoveryView,
    escalation: escalation_state::EscalationView,
    context_health: ContextHealth,
    session_boundary: SessionBoundaryRecommendation,
    behavioral_drift: BehavioralDriftState,
}

#[derive(Serialize)]
pub(crate) struct ContextCheckThrottle {
    suppressed: bool,
    next_interval_hint_seconds: u64,
}

#[derive(Serialize)]
pub(crate) struct ContextCheckPayload {
    policy_digest: String,
    focus_digest: String,
    state_delta: String,
    pub(crate) risk_summary: String,
}

#[derive(Serialize)]
pub(crate) struct HandoffState {
    path: String,
    pub(crate) title: String,
    pub(crate) immediate_actions: Vec<String>,
    current_system_state: Vec<String>,
    pub(crate) key_files: Vec<String>,
    pub(crate) definition_of_done: Vec<String>,
}

#[derive(Serialize)]
pub(crate) struct RadarSessionStateView {
    path: String,
    status: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    session_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) mode: Option<session_state::SessionMode>,
    #[serde(skip_serializing_if = "Option::is_none")]
    start_count: Option<u32>,
    #[serde(flatten)]
    lifecycle: session_state::SessionLifecycleProjection,
}

#[derive(Serialize)]
pub(crate) struct ProjectionTelemetryView {
    source_fingerprint: String,
    delta_status: &'static str,
    delta: compiled_state::ProjectionDeltaStats,
    token_counts: compiled_state::ProjectionTokenTelemetry,
    cadence: ProjectionCadenceView,
}

#[derive(Serialize)]
struct ProjectionCadenceView {
    target: &'static str,
    current: BTreeMap<String, compiled_state::ProjectionFormatCadenceCurrent>,
    history: BTreeMap<String, compiled_state::ProjectionFormatCadenceSummary>,
    #[serde(skip_serializing_if = "Option::is_none")]
    provider_cache_usage: Option<ProjectionProviderCacheUsageView>,
}

#[derive(Serialize)]
struct ProjectionProviderCacheUsageView {
    #[serde(skip_serializing_if = "Option::is_none")]
    cache_creation_input_tokens: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cache_read_input_tokens: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    blended_total_tokens: Option<u64>,
}

#[derive(Serialize)]
pub(crate) struct MemoryState {
    status: &'static str,
    profile_path: String,
    project_path: String,
    locality_path: String,
    work_stream_path: String,
    branch_path: String,
    workspace_path: String,
    clone_path: String,
    pub(crate) content: String,
    pub(crate) structured: memory_entries::StructuredMemoryView,
    #[serde(skip_serializing)]
    pub(crate) contributing_paths: Vec<String>,
}

#[derive(Serialize)]
pub(crate) struct RepoNativeChecksState {
    surface: check::CheckSurfaceView,
    pub(crate) failures: usize,
    pub(crate) checks: Vec<RepoNativeCheckResult>,
}

#[derive(Serialize)]
pub(crate) struct RepoNativeCheckResult {
    pub(crate) id: String,
    path: String,
    pub(crate) status: &'static str,
    pub(crate) severity: &'static str,
    pub(crate) message: String,
    duration_ms: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    exit_code: Option<i32>,
}

#[derive(Serialize)]
pub(crate) struct ContextHealth {
    source: &'static str,
    context_used_pct: Option<u8>,
    degradation_risk_pct: Option<u8>,
    pub(crate) band: &'static str,
    confidence: &'static str,
    signals: ContextSignals,
    cost: telemetry_cost::TelemetryCostView,
    pub(crate) recommendation: &'static str,
}

#[derive(Serialize)]
pub(crate) struct ContextSignals {
    pub(crate) compacted: Option<bool>,
    pub(crate) session_minutes: Option<u64>,
    pub(crate) ccd_start_cycles_in_conversation: Option<u32>,
    pub(crate) host_total_tokens: Option<u64>,
    pub(crate) host_context_window_tokens: Option<u64>,
    pub(crate) tool_output_kb: Option<u64>,
    pub(crate) operator_symptoms: Vec<String>,
}

#[derive(Serialize)]
pub(crate) struct ReportDecisions {
    requires_human_choice: bool,
    requires_doctor: bool,
    can_clear_session: bool,
    safe_clean_close: bool,
    needs_full_handover: bool,
    provider_live_check_needed: bool,
    prompt_context_unchanged: bool,
    recommended_next_command: &'static str,
}

#[derive(Serialize)]
pub(crate) struct BehavioralDriftState {
    pub(crate) status: DriftAggregateStatus,
    pub(crate) summary: String,
    pub(crate) evidence: Vec<String>,
    recommended_corrections: Vec<String>,
    signals: Vec<BehavioralDriftSignal>,
}

#[derive(Serialize)]
pub(crate) struct BehavioralDriftSignal {
    pub(crate) id: &'static str,
    pub(crate) status: DriftSignalStatus,
    pub(crate) summary: String,
    pub(crate) evidence: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) recommended_correction: Option<String>,
}

pub(crate) struct BehavioralDriftInputs<'a> {
    pub(crate) layout: &'a StateLayout,
    pub(crate) locality_id: &'a str,
    pub(crate) git: Option<&'a GitState>,
    pub(crate) handoff: &'a HandoffState,
    pub(crate) handoff_contents: &'a str,
    pub(crate) runtime: &'a runtime_state::LoadedRuntimeState,
    pub(crate) runtime_handoff: &'a runtime_state::RuntimeHandoffState,
    pub(crate) tracked_session: Option<&'a session_state::SessionStateFile>,
    pub(crate) consistency_axes: Vec<consistency::ConsistencyAxis>,
}

pub(crate) struct EvaluationInputs<'a> {
    pub(crate) git: Option<&'a GitState>,
    pub(crate) candidates: &'a CandidateUpdates,
    pub(crate) execution_gates: &'a session_gates::ExecutionGatesView,
    pub(crate) effective_memory: &'a MemoryState,
    pub(crate) repo_native_checks: &'a RepoNativeChecksState,
    pub(crate) behavioral_drift: &'a BehavioralDriftState,
    pub(crate) escalation_view: &'a escalation_state::EscalationView,
}

struct LoadedHandoffState {
    handoff: HandoffState,
    contents: String,
}

/// Strict §5.2 checkpoint contract: shared foundation that `evaluate()` and
/// `handover()` build on. Serialized fields match the CLI checkpoint subset
/// exactly; the `#[serde(skip)]` fields carry state forward so `evaluate()`
/// does not re-load anything.
#[derive(Serialize)]
pub struct CheckpointResult {
    command: &'static str,
    mode: &'static str,
    schema_version: u32,
    ok: bool,
    pub(crate) decisions: ReportDecisions,
    pub(crate) path: String,
    pub(crate) profile: String,
    pub(crate) project_id: String,
    pub(crate) locality_id: String,
    pub(crate) handoff: HandoffState,
    pub(crate) execution_gates: session_gates::ExecutionGatesView,
    pub(crate) effective_memory: MemoryState,
    pub(crate) git: Option<GitState>,
    pub(crate) checkout_state: Option<handoff::CheckoutStateView>,
    pub(crate) session_state: RadarSessionStateView,
    pub(crate) recovery: runtime_state::RuntimeRecoveryView,
    pub(crate) escalation: escalation_state::EscalationView,
    pub(crate) projection_telemetry: ProjectionTelemetryView,
    pub(crate) work_stream_decay: work_stream_decay::WorkStreamDecayView,
    pub(crate) context_health: ContextHealth,
    pub(crate) session_boundary: ProvisionalSessionBoundary,
    pub(crate) behavioral_drift: BehavioralDriftState,
    pub(crate) candidate_updates: CheckpointCandidateUpdates,
    // State carried forward to evaluate()/handover() without serialization.
    #[serde(skip)]
    pub(crate) layout: StateLayout,
    #[serde(skip)]
    pub(crate) tracked_session: Option<session_state::SessionStateFile>,
    #[serde(skip)]
    pub(crate) projection_digests: compiled_state::ProjectionDigests,
    #[serde(skip)]
    pub(crate) policy_digest: String,
}

/// Strict §5.2 subset of `CandidateUpdates`: checkpoint exposes only the
/// handoff-refresh candidates (the auto-lifecycle trigger input); memory
/// candidates are an evaluate-only concern and are absent from the wire
/// shape rather than emitted as an empty array.
#[derive(Serialize)]
pub struct CheckpointCandidateUpdates {
    pub(crate) handoff: Vec<CandidateUpdate>,
}

/// Wraps `SessionBoundaryRecommendation` with a top-level `provisional: true`
/// flag. Used only in checkpoint output; handover serializes the plain
/// `SessionBoundaryRecommendation` (no `provisional` key).
#[derive(Serialize)]
pub struct ProvisionalSessionBoundary {
    #[serde(flatten)]
    pub(crate) inner: SessionBoundaryRecommendation,
    pub(crate) provisional: bool,
}

/// Evaluate-only extension added on top of a `CheckpointResult`: the fields
/// the handover contract exposes that the checkpoint contract deliberately
/// omits, plus the refined (non-provisional) session boundary.
pub struct EvaluationResult {
    pub(crate) repo_native_checks: RepoNativeChecksState,
    pub(crate) evaluation: RadarEvaluation,
    pub(crate) approval_steps: Vec<ApprovalStep>,
    pub(crate) memory_candidates: Vec<CandidateUpdate>,
    pub(crate) session_boundary: SessionBoundaryRecommendation,
}

/// Return type of `handover()`: the full composed report plus the projection
/// digests callers need for `--since-session` delta comparisons.
pub struct HandoverResult {
    pub(crate) report: RadarStateReport,
    pub(crate) projection_digests: compiled_state::ProjectionDigests,
    pub(crate) policy_digest: String,
}

#[derive(Serialize)]
pub(crate) struct RadarEvaluation {
    pub(crate) next_step: EvaluationBucket<NextStepStatus>,
    pub(crate) memory: EvaluationBucket<MemoryStatus>,
    pub(crate) wrap_up: EvaluationBucket<WrapUpStatus>,
}

#[derive(Serialize)]
pub(crate) struct EvaluationBucket<S: Serialize> {
    pub(crate) status: S,
    pub(crate) summary: String,
    pub(crate) evidence: Vec<String>,
}

#[derive(Serialize)]
pub(crate) struct CandidateUpdates {
    pub(crate) handoff: Vec<CandidateUpdate>,
    pub(crate) memory: Vec<CandidateUpdate>,
}

#[derive(Clone, Serialize)]
pub(crate) struct CandidateUpdate {
    pub(crate) kind: &'static str,
    pub(crate) summary: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) section: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) replacement_lines: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) entry_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) source_scope: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) suggested_destination: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) action: Option<CandidateAction>,
}

#[derive(Clone, Serialize)]
pub(crate) struct CandidateAction {
    pub(crate) kind: &'static str,
    pub(crate) argv: Vec<String>,
    pub(crate) note: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) apply: Option<CandidateActionApply>,
}

#[derive(Clone, Serialize)]
pub(crate) struct CandidateActionApply {
    pub(crate) argv: Vec<String>,
    pub(crate) requirements: Vec<CandidateActionRequirement>,
    pub(crate) note: &'static str,
}

#[derive(Clone, Serialize)]
pub(crate) struct CandidateActionRequirement {
    pub(crate) id: &'static str,
    pub(crate) flag: &'static str,
    pub(crate) value_type: &'static str,
    pub(crate) required: bool,
    pub(crate) allowed_values: Vec<&'static str>,
    pub(crate) note: &'static str,
}

#[derive(Serialize)]
pub(crate) struct ApprovalStep {
    pub(crate) id: &'static str,
    pub(crate) question: &'static str,
    pub(crate) recommended_answer: &'static str,
    pub(crate) recommendation: String,
    pub(crate) evidence: Vec<String>,
}

impl RadarEvaluation {
    fn default_empty() -> Self {
        Self {
            next_step: EvaluationBucket {
                status: NextStepStatus::NoChange,
                summary: String::new(),
                evidence: Vec::new(),
            },
            memory: EvaluationBucket {
                status: MemoryStatus::NoChange,
                summary: String::new(),
                evidence: Vec::new(),
            },
            wrap_up: EvaluationBucket {
                status: WrapUpStatus::NoChange,
                summary: String::new(),
                evidence: Vec::new(),
            },
        }
    }
}

fn checkpoint_decisions() -> ReportDecisions {
    ReportDecisions {
        requires_human_choice: false,
        requires_doctor: false,
        can_clear_session: false,
        safe_clean_close: false,
        needs_full_handover: true,
        provider_live_check_needed: false,
        prompt_context_unchanged: false,
        recommended_next_command: "ccd --output json handover",
    }
}

fn handover_decisions(
    checkpoint: &CheckpointResult,
    evaluation: &EvaluationResult,
) -> ReportDecisions {
    let requires_doctor = evaluation.repo_native_checks.failures > 0;
    let has_candidates = !checkpoint.candidate_updates.handoff.is_empty()
        || !evaluation.memory_candidates.is_empty();
    let has_unfinished_gates = checkpoint.execution_gates.unfinished_count > 0;
    let has_review_required = evaluation.evaluation.next_step.status
        == NextStepStatus::ReviewRequired
        || evaluation.evaluation.memory.status == MemoryStatus::ReviewRequired
        || evaluation.evaluation.wrap_up.status == WrapUpStatus::NeedsReview;
    let safe_clean_close =
        !requires_doctor && !has_candidates && !has_unfinished_gates && !has_review_required;
    let requires_human_choice =
        requires_doctor || has_candidates || has_unfinished_gates || has_review_required;
    let recommended_next_command = if safe_clean_close {
        "ccd --output json handover --close-clean"
    } else if requires_doctor {
        "ccd doctor"
    } else {
        "resolve_human_choice"
    };

    ReportDecisions {
        requires_human_choice,
        requires_doctor,
        can_clear_session: safe_clean_close,
        safe_clean_close,
        needs_full_handover: requires_human_choice,
        provider_live_check_needed: false,
        prompt_context_unchanged: false,
        recommended_next_command,
    }
}

fn context_check_decisions(
    action: ContextCheckAction,
    reason: &str,
    recommendation: &str,
) -> ReportDecisions {
    let prompt_context_unchanged = reason == "throttled_unchanged";
    let requires_human_choice = matches!(action, ContextCheckAction::Escalate);
    ReportDecisions {
        requires_human_choice,
        requires_doctor: false,
        can_clear_session: matches!(action, ContextCheckAction::WrapUpRequired),
        safe_clean_close: false,
        needs_full_handover: matches!(action, ContextCheckAction::CheckpointNow),
        provider_live_check_needed: false,
        prompt_context_unchanged,
        recommended_next_command: match recommendation {
            "checkpoint_now" => "ccd --output json checkpoint",
            "capture_before_compaction" => "ccd checkpoint",
            "wrap_up_and_clear" => "ccd --output json handover --close-clean",
            "continue" => "continue",
            _ => action.as_str(),
        },
    }
}

impl CommandReport for CheckpointResult {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        render_handoff_intro(&self.handoff);
        render_workspace_view(self.git.as_ref(), self.checkout_state.as_ref());
        render_context_health(&self.context_health);
        // Checkpoint's session boundary is computed before evaluate() runs,
        // so a later handover can refine it. The text output makes that
        // provisional status visible; the structured contract carries the
        // same signal as `session_boundary.provisional: true`.
        println!(
            "Recommendation (provisional): {} - {}",
            self.session_boundary.inner.action.as_str(),
            self.session_boundary.inner.summary
        );
        let findings = radar_material_findings_from_parts(RadarFindingsParts {
            execution_gates: &self.execution_gates,
            repo_native_checks: None,
            behavioral_drift: &self.behavioral_drift,
            work_stream_decay: &self.work_stream_decay,
        });
        if !findings.is_empty() {
            println!("Material findings:");
            for finding in findings {
                println!("- {finding}");
            }
        }
    }
}

impl CommandReport for RadarStateReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        render_handoff_intro(&self.handoff);
        render_workspace_view(self.git.as_ref(), self.checkout_state.as_ref());
        render_context_health(&self.context_health);
        if !self.evaluation.wrap_up.summary.is_empty() {
            println!(
                "Wrap-up: {} [{}]",
                self.evaluation.wrap_up.summary,
                self.evaluation.wrap_up.status.as_str()
            );
        }
        println!(
            "Recommendation: {} - {}",
            self.session_boundary.action.as_str(),
            self.session_boundary.summary
        );

        let findings = radar_material_findings(self);
        if !findings.is_empty() {
            println!("Material findings:");
            for finding in findings {
                println!("- {finding}");
            }
        }
    }
}

fn radar_git_summary(git: &GitState) -> String {
    let tracking = match (git.upstream.as_deref(), git.ahead, git.behind) {
        (Some(upstream), 0, 0) => format!("tracking {upstream} in sync"),
        (Some(upstream), ahead, 0) => format!("tracking {upstream}, ahead {ahead}"),
        (Some(upstream), 0, behind) => format!("tracking {upstream}, behind {behind}"),
        (Some(upstream), ahead, behind) => {
            format!("tracking {upstream}, ahead {ahead}, behind {behind}")
        }
        (None, _, _) => "no upstream".to_owned(),
    };
    let cleanliness = if git.clean { "clean" } else { "dirty" };
    format!("{} at {} ({tracking}, {cleanliness})", git.branch, git.head)
}

fn radar_git_changes(git: &GitState) -> Vec<String> {
    let mut lines = Vec::new();
    if !git.staged_files.is_empty() {
        lines.push(format!("staged: {}", git.staged_files.join(", ")));
    }
    if !git.unstaged_files.is_empty() {
        lines.push(format!("unstaged: {}", git.unstaged_files.join(", ")));
    }
    if !git.untracked_files.is_empty() {
        lines.push(format!("untracked: {}", git.untracked_files.join(", ")));
    }
    lines
}

fn radar_material_findings(report: &RadarStateReport) -> Vec<String> {
    radar_material_findings_from_parts(RadarFindingsParts {
        execution_gates: &report.execution_gates,
        repo_native_checks: Some(&report.repo_native_checks),
        behavioral_drift: &report.behavioral_drift,
        work_stream_decay: &report.work_stream_decay,
    })
}

struct RadarFindingsParts<'a> {
    execution_gates: &'a session_gates::ExecutionGatesView,
    repo_native_checks: Option<&'a RepoNativeChecksState>,
    behavioral_drift: &'a BehavioralDriftState,
    work_stream_decay: &'a work_stream_decay::WorkStreamDecayView,
}

fn radar_material_findings_from_parts(parts: RadarFindingsParts<'_>) -> Vec<String> {
    let mut findings = Vec::new();
    if let Some(anchor) = &parts.execution_gates.attention_anchor {
        findings.push(format!(
            "Execution gate [{} #{}/{}]: {}",
            anchor.status.as_str(),
            anchor.index,
            parts.execution_gates.total_count,
            anchor.text
        ));
    } else if parts.execution_gates.total_count > 0 {
        if parts.execution_gates.unfinished_count > 0 {
            findings.push(format!(
                "Execution gates: {} total, {} unfinished.",
                parts.execution_gates.total_count, parts.execution_gates.unfinished_count
            ));
        } else {
            findings.push(format!(
                "Execution gates: {} total.",
                parts.execution_gates.total_count
            ));
        }
    }
    if let Some(repo_native_checks) = parts.repo_native_checks {
        if let Some(note) = &repo_native_checks.surface.note {
            findings.push(note.clone());
        }
        for check in &repo_native_checks.checks {
            findings.push(format!(
                "{} [{}]: {}",
                check.id, check.status, check.message
            ));
        }
    }
    if !matches!(parts.behavioral_drift.status, DriftAggregateStatus::Aligned) {
        findings.push(parts.behavioral_drift.summary.clone());
    }
    if parts.work_stream_decay.status != "missing"
        && parts.work_stream_decay.consecutive_no_progress > 0
    {
        findings.push(format!(
            "Work-stream decay: {} consecutive no-progress attempt(s).",
            parts.work_stream_decay.consecutive_no_progress
        ));
    }
    findings
}

fn render_handoff_intro(handoff: &HandoffState) {
    println!("Handoff: {}", handoff.title);
    if !handoff.immediate_actions.is_empty() {
        println!("Immediate actions:");
        for (index, step) in handoff.immediate_actions.iter().enumerate() {
            println!("{}. {step}", index + 1);
        }
    }
}

fn render_workspace_view(
    git: Option<&GitState>,
    checkout_state: Option<&handoff::CheckoutStateView>,
) {
    if let Some(git) = git {
        println!("Workspace: {}", radar_git_summary(git));
        if !git.staged_files.is_empty()
            || !git.unstaged_files.is_empty()
            || !git.untracked_files.is_empty()
        {
            println!("Workspace changes:");
            for line in radar_git_changes(git) {
                println!("- {line}");
            }
        }
    } else {
        println!("Workspace: unavailable by design for directory-substrate workspaces");
    }
    if let Some(checkout_state) = checkout_state.filter(|state| state.advisory) {
        println!("Workspace note: {}", checkout_state.summary);
    }
}

fn render_context_health(context_health: &ContextHealth) {
    println!("Context health: {}", context_health.band);
    if let Some(cost_summary) = radar_cost_summary(&context_health.cost) {
        println!("Cost: {cost_summary}");
    }
}

fn radar_cost_summary(cost: &telemetry_cost::TelemetryCostView) -> Option<String> {
    let session_cost = cost.session_estimate_usd?;
    let mut parts = vec![format!(
        "session ${}",
        crate::telemetry::cost::format_usd(session_cost)
    )];
    if let Some(model) = &cost.model {
        parts.push(format!("model {model}"));
    }
    Some(parts.join(", "))
}

impl CommandReport for ContextCheckReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        println!(
            "Context check: {} ({}, trigger {}, reason {})",
            self.action.as_str(),
            self.urgency.as_str(),
            self.trigger.as_str(),
            self.reason
        );
        println!("{}", self.payload.risk_summary);
        if self.action == ContextCheckAction::InjectDelta {
            println!("Policy digest: {}", self.payload.policy_digest);
            println!("Focus digest: {}", self.payload.focus_digest);
            println!("State delta: {}", self.payload.state_delta);
        }
        if self.throttle.suppressed {
            println!(
                "Throttle: suppressed; next interval hint {}s",
                self.throttle.next_interval_hint_seconds
            );
        }
    }
}

/// Strict §5.2 checkpoint: the shared foundation used by the auto-lifecycle
/// trigger path and by handover composition. Self-contained — includes
/// `consistency_axes` (so behavioral-drift runs the full axis set) and a
/// provisional `session_boundary` that `evaluate()` later refines.
pub fn checkpoint(repo_root: &Path, explicit_profile: Option<&str>) -> Result<CheckpointResult> {
    let _span = info_span!("radar_checkpoint").entered();
    let profile = profile::resolve(explicit_profile)?;
    let profile_name = profile.to_string();
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    if !layout.profile_root().is_dir() {
        bail!(
            "profile `{}` does not exist at {}; bootstrap it with `ccd attach` before using `ccd checkpoint`",
            layout.profile(),
            layout.profile_root().display()
        );
    }

    let marker = repo_marker::load(repo_root)?.ok_or_else(|| {
        anyhow::anyhow!(
            "repo is not linked: {} is missing; bootstrap this clone with `ccd attach` or `ccd link` first",
            repo_root.join(repo_marker::MARKER_FILE).display()
        )
    })?;

    let git = if layout.resolved_substrate().is_git() {
        Some(handoff::read_git_state(
            repo_root,
            BranchMode::AllowDetachedHead,
        )?)
    } else {
        None
    };
    let checkout_state = git
        .as_ref()
        .map(|git| handoff::checkout_state_view(repo_root, git));
    let runtime = runtime_state::load_runtime_state(repo_root, &layout, &marker.locality_id)?;
    let shared_db = db::StateDb::try_open_for_layout(&layout)?;
    let (tracked_session, tracked_activity, active_session_id) = match shared_db.as_ref() {
        Some(db) => (
            session_state::load_from_db(Some(db))?,
            session_state::load_activity_from_db(Some(db))?,
            session_state::load_session_id_from_db(Some(db))?,
        ),
        None => (
            session_state::load_for_layout(&layout)?,
            session_state::load_activity_for_layout(&layout)?,
            session_state::load_session_id(&layout)?,
        ),
    };
    let work_stream_decay =
        work_stream_decay::load_view_from_db(shared_db.as_ref(), active_session_id.as_deref())?;
    let host_snapshot = host_telemetry::current(repo_root).ok().flatten();
    let session_state_view =
        build_session_state_view(&layout, tracked_session.as_ref(), tracked_activity.as_ref());
    let loaded_handoff = read_handoff(
        repo_root,
        &layout,
        &runtime,
        git.as_ref(),
        active_session_id.as_deref(),
        &session_state_view,
    )?;
    let execution_gates = read_execution_gates(&runtime);
    let effective_memory = read_effective_memory(&runtime);
    let escalation_entries = if let Some(db) = shared_db.as_ref() {
        escalation_state::load_from_db(Some(db))?
    } else {
        escalation_state::load_for_layout(&layout)?
    };
    let escalation_view = escalation_state::build_view(&layout, &escalation_entries);
    let recovery = runtime_state::recovery_view(&runtime.recovery);
    let (projection_telemetry, consistency_current_digests, projection_digests) =
        build_projection_telemetry(
            &layout,
            shared_db.as_ref(),
            &runtime,
            &execution_gates,
            &escalation_view,
            &recovery,
            &git,
            &session_state_view,
            active_session_id.as_deref(),
            host_snapshot.as_ref(),
            compiled_state::ProjectionTarget::Session,
        )?;

    // consistency_axes is part of the checkpoint contract: it feeds
    // behavioral_drift's consistency-family signals and is consumed by
    // auto-lifecycle triggers in PR-3.
    //
    // Consistency evaluation reads on-disk runtime-state ledgers that can be
    // corrupt or schema-skewed. `ccd consistency` and `ccd doctor` surface
    // those failures explicitly, but `radar::checkpoint` also runs inside
    // retry-sensitive hook paths (`on-agent-end`) where `handle_agent_end`
    // has already persisted the just-recorded attempt outcome before
    // building the payload. Propagating a consistency failure here would
    // double-count decay on retry. We degrade gracefully instead: empty axes
    // feed behavioral_drift as "no_signal" while standalone consistency and
    // doctor still report the underlying corruption.
    let latest_projection_observation = consistency::load_latest_projection_observation(&layout)?;
    let consistency_axes = match consistency::evaluate_with_inputs(consistency::ConsistencyInputs {
        repo_root,
        layout: &layout,
        locality_id: &marker.locality_id,
        handoff: &runtime.state.handoff,
        checkpoint: runtime.recovery.state.checkpoint.as_ref(),
        include_checkpoint_references: false,
        source_fingerprint: &projection_telemetry.source_fingerprint,
        current_digests: &consistency_current_digests,
        latest_observation: latest_projection_observation.as_ref(),
    }) {
        Ok(report) => report.axes,
        Err(_) => Vec::new(),
    };

    let context_health = context_health::build_context_health(
        &layout,
        &marker.locality_id,
        tracked_session.as_ref(),
        active_session_id.as_deref(),
        host_snapshot.as_ref(),
        &runtime.state.handoff,
        runtime.execution_gates.view.attention_anchor.as_ref(),
        &work_stream_decay,
    )?;
    let behavioral_drift = behavioral_drift::build_behavioral_drift(
        repo_root,
        BehavioralDriftInputs {
            layout: &layout,
            locality_id: &marker.locality_id,
            git: git.as_ref(),
            handoff: &loaded_handoff.handoff,
            handoff_contents: &loaded_handoff.contents,
            runtime: &runtime,
            runtime_handoff: &runtime.state.handoff,
            tracked_session: tracked_session.as_ref(),
            consistency_axes,
        },
    );

    let handoff_config = layout.effective_handoff_config(&marker.locality_id)?;
    let handoff_candidate_refresh_enabled = handoff_config.candidate_refresh_enabled();
    let candidate_updates = CheckpointCandidateUpdates {
        handoff: candidates::build_handoff_refresh_candidates(
            repo_root,
            &loaded_handoff.handoff,
            tracked_session.as_ref(),
            handoff_candidate_refresh_enabled,
        ),
    };

    let provisional_inner = session_boundary::build_session_boundary(
        checkout_state.as_ref(),
        &recovery,
        &work_stream_decay,
        &context_health,
        &behavioral_drift,
        &RadarEvaluation::default_empty(),
        &escalation_view,
    );

    let policy_digest = context_check::build_policy_digest(&runtime);

    Ok(CheckpointResult {
        command: "checkpoint",
        mode: "checkpoint",
        schema_version: CHECKPOINT_SCHEMA_VERSION,
        ok: true,
        decisions: checkpoint_decisions(),
        path: repo_root.display().to_string(),
        profile: profile_name,
        project_id: marker.locality_id.clone(),
        locality_id: marker.locality_id,
        handoff: loaded_handoff.handoff,
        execution_gates,
        effective_memory,
        git,
        checkout_state,
        session_state: session_state_view,
        recovery,
        escalation: escalation_view,
        projection_telemetry,
        work_stream_decay,
        context_health,
        session_boundary: ProvisionalSessionBoundary {
            inner: provisional_inner,
            provisional: true,
        },
        behavioral_drift,
        candidate_updates,
        layout,
        tracked_session,
        projection_digests,
        policy_digest,
    })
}

/// Evaluate-only extension: runs `repo_native_checks`, computes memory
/// candidates, builds the full `RadarEvaluation`, assembles approval steps,
/// and refines the session boundary (no `provisional` flag).
///
/// Derives `repo_root` from `checkpoint.path` so there is no way for a caller
/// to compose a checkpoint built against one checkout with evaluation data
/// from another. This makes the checkpoint→evaluate handoff self-contained
/// and removes a silent-mismatch footgun.
pub fn evaluate(checkpoint: &CheckpointResult) -> Result<EvaluationResult> {
    let _span = info_span!("radar_evaluate").entered();
    let repo_root = Path::new(&checkpoint.path);

    let repo_native_checks = read_repo_native_checks(repo_root);
    let branch_memory_available = runtime_state::resolve_active_branch_memory_path(
        repo_root,
        &checkpoint.layout,
        &checkpoint.locality_id,
    )?
    .is_some();
    let handoff_config = checkpoint
        .layout
        .effective_handoff_config(&checkpoint.locality_id)?;
    let handoff_candidate_refresh_enabled = handoff_config.candidate_refresh_enabled();
    let full_candidate_updates = candidates::build_candidate_updates(
        repo_root,
        &checkpoint.profile,
        &checkpoint.handoff,
        &checkpoint.effective_memory,
        checkpoint.tracked_session.as_ref(),
        branch_memory_available,
        handoff_candidate_refresh_enabled,
    );
    let memory_candidates = full_candidate_updates.memory;

    let combined_candidates = CandidateUpdates {
        handoff: checkpoint.candidate_updates.handoff.clone(),
        memory: memory_candidates.clone(),
    };

    let evaluation = evaluation::build_evaluation(EvaluationInputs {
        git: checkpoint.git.as_ref(),
        candidates: &combined_candidates,
        execution_gates: &checkpoint.execution_gates,
        effective_memory: &checkpoint.effective_memory,
        repo_native_checks: &repo_native_checks,
        behavioral_drift: &checkpoint.behavioral_drift,
        escalation_view: &checkpoint.escalation,
    });
    let approval_steps = session_boundary::build_approval_steps(
        &evaluation,
        &combined_candidates,
        &checkpoint.behavioral_drift,
    );

    let refined_session_boundary = session_boundary::build_session_boundary(
        checkpoint.checkout_state.as_ref(),
        &checkpoint.recovery,
        &checkpoint.work_stream_decay,
        &checkpoint.context_health,
        &checkpoint.behavioral_drift,
        &evaluation,
        &checkpoint.escalation,
    );

    Ok(EvaluationResult {
        repo_native_checks,
        evaluation,
        approval_steps,
        memory_candidates,
        session_boundary: refined_session_boundary,
    })
}

/// Full handover composition: `checkpoint()` + `evaluate()` merged into the
/// legacy `RadarStateReport` shape expected by `ccd handover` consumers.
pub fn handover(repo_root: &Path, explicit_profile: Option<&str>) -> Result<HandoverResult> {
    let checkpoint_result = checkpoint(repo_root, explicit_profile)?;
    let evaluation_result = evaluate(&checkpoint_result)?;
    let decisions = handover_decisions(&checkpoint_result, &evaluation_result);

    let report = RadarStateReport {
        command: "handover",
        mode: "handover",
        schema_version: HANDOVER_SCHEMA_VERSION,
        ok: true,
        decisions,
        path: checkpoint_result.path,
        profile: checkpoint_result.profile,
        project_id: checkpoint_result.project_id,
        locality_id: checkpoint_result.locality_id,
        handoff: checkpoint_result.handoff,
        execution_gates: checkpoint_result.execution_gates,
        effective_memory: checkpoint_result.effective_memory,
        repo_native_checks: evaluation_result.repo_native_checks,
        git: checkpoint_result.git,
        checkout_state: checkpoint_result.checkout_state,
        session_state: checkpoint_result.session_state,
        recovery: checkpoint_result.recovery,
        escalation: checkpoint_result.escalation,
        projection_telemetry: checkpoint_result.projection_telemetry,
        work_stream_decay: checkpoint_result.work_stream_decay,
        context_health: checkpoint_result.context_health,
        session_boundary: evaluation_result.session_boundary,
        behavioral_drift: checkpoint_result.behavioral_drift,
        evaluation: evaluation_result.evaluation,
        candidate_updates: CandidateUpdates {
            handoff: checkpoint_result.candidate_updates.handoff,
            memory: evaluation_result.memory_candidates,
        },
        approval_intents: vec![
            "approve",
            "approve_and_reuse_when_supported",
            "reject_and_explain",
        ],
        approval_steps: evaluation_result.approval_steps,
    };

    Ok(HandoverResult {
        report,
        projection_digests: checkpoint_result.projection_digests,
        policy_digest: checkpoint_result.policy_digest,
    })
}

pub fn run_context_check(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    trigger: ContextCheckTrigger,
) -> Result<ContextCheckReport> {
    let _span = info_span!("context_check").entered();

    // Low-value triggers (interval, supervisor-poll) use the lightweight
    // checkpoint-only path — skipping evaluate() which runs expensive
    // repo-native checks, memory candidate generation, and approval
    // step assembly (#627).
    if trigger.is_low_value() {
        return run_context_check_lightweight(repo_root, explicit_profile, trigger);
    }

    let handover_result = handover(repo_root, explicit_profile)?;
    let radar_report = handover_result.report;
    let source_fingerprint = radar_report.projection_telemetry.source_fingerprint.clone();
    let inputs = context_check::ContextCheckInputs::from_handover(&radar_report);

    let payload = ContextCheckPayload {
        policy_digest: handover_result.policy_digest,
        focus_digest: context_check::build_focus_digest_from_inputs(&inputs),
        state_delta: context_check::build_state_delta_from_inputs(&inputs),
        risk_summary: context_check::build_risk_summary_from_inputs(trigger, &inputs),
    };
    let decision = context_check::decide(trigger, &inputs, &payload);
    let decisions = context_check_decisions(
        decision.action,
        decision.reason,
        decision.recommendation.as_str(),
    );

    Ok(ContextCheckReport {
        command: "context-check",
        schema_version: CONTEXT_CHECK_SCHEMA_VERSION,
        ok: true,
        decisions,
        path: radar_report.path,
        profile: radar_report.profile,
        project_id: radar_report.project_id,
        locality_id: radar_report.locality_id,
        trigger,
        action: decision.action,
        reason: decision.reason,
        recommendation: decision.recommendation,
        urgency: decision.urgency,
        throttle: ContextCheckThrottle {
            suppressed: decision.suppressed,
            next_interval_hint_seconds: decision.next_interval_hint_seconds,
        },
        source_fingerprint,
        payload,
        evidence: decision.evidence,
        session_state: radar_report.session_state,
        execution_gates: radar_report.execution_gates,
        recovery: radar_report.recovery,
        escalation: radar_report.escalation,
        context_health: radar_report.context_health,
        session_boundary: radar_report.session_boundary,
        behavioral_drift: radar_report.behavioral_drift,
    })
}

/// Persisted metadata for the source-fingerprint/time-based throttle.
/// Written after each non-throttled low-value context check and read back
/// on the next one to decide whether expensive state construction can be
/// skipped entirely (#627).
#[derive(Serialize, Deserialize)]
struct ContextCheckCache {
    source_fingerprint: String,
    epoch_s: u64,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
}

fn context_check_cache_path(layout: &StateLayout) -> std::path::PathBuf {
    layout
        .state_db_path()
        .with_file_name("context_check_cache.json")
}

fn read_context_check_cache(layout: &StateLayout) -> Option<ContextCheckCache> {
    let path = context_check_cache_path(layout);
    let data = std::fs::read_to_string(&path).ok()?;
    serde_json::from_str(&data).ok()
}

fn write_context_check_cache(layout: &StateLayout, cache: &ContextCheckCache) {
    let path = context_check_cache_path(layout);
    let _ = serde_json::to_string(cache)
        .ok()
        .and_then(|json| std::fs::write(&path, json).ok());
}

/// Lightweight context-check path for low-value triggers (interval,
/// supervisor-poll). Uses checkpoint() only — skips evaluate() which
/// runs expensive repo-native checks, memory candidate generation, and
/// approval step assembly. The decision tree still runs with full
/// fidelity via ContextCheckInputs::from_checkpoint (#627).
///
/// Before calling checkpoint(), the function reads a persisted
/// source-fingerprint/timestamp cache.  If the cached entry is inside
/// the throttle window, a quick source fingerprint is computed from the
/// raw runtime surfaces only.  When the fingerprint matches, the entire
/// checkpoint + decision tree is skipped and a compact throttled report
/// is returned (#627 short-circuit requirement).
fn run_context_check_lightweight(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    trigger: ContextCheckTrigger,
) -> Result<ContextCheckReport> {
    // --- pre-check: time + fingerprint based throttle (#627) ---
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile)?;

    if let Some(cached) = read_context_check_cache(&layout) {
        let now = session_state::now_epoch_s().unwrap_or(0);
        if now.saturating_sub(cached.epoch_s) < CONTEXT_CHECK_THROTTLE_WINDOW_SECONDS {
            // Inside throttle window — compute a quick source fingerprint
            // from raw runtime surfaces (reads ~6 small files, no DB, no
            // git state, no session state) to confirm nothing changed.
            let marker = repo_marker::load(repo_root)?
                .ok_or_else(|| anyhow::anyhow!("repo is not linked"))?;
            let pod_active =
                crate::state::machine_registry::resolve_active_identity(&layout)?.is_some();
            let raw = runtime_state::load_raw_start_runtime_sources(
                repo_root,
                &layout,
                &marker.locality_id,
                pod_active,
            )?;
            let surfaces = runtime_state::RuntimeSourceSurfaces {
                profile_memory: raw.profile_memory,
                locality_memory: raw.locality_memory,
                branch_memory: raw.branch_memory,
                clone_memory: raw.clone_memory,
                execution_gates: raw.execution_gates.source,
                handoff: raw.handoff,
            };
            let current_fp = compiled_state::fingerprint_for_inputs(&surfaces, pod_active)?;
            if current_fp == cached.source_fingerprint {
                // Sources unchanged and inside throttle window —
                // short-circuit before expensive state construction.
                return Ok(build_throttled_context_check_report(
                    trigger, &cached, &layout,
                ));
            }
        }
    }

    // --- normal path: full checkpoint + decision tree ---
    let checkpoint_result = checkpoint(repo_root, explicit_profile)?;
    let source_fingerprint = checkpoint_result
        .projection_telemetry
        .source_fingerprint
        .clone();
    let inputs = context_check::ContextCheckInputs::from_checkpoint(&checkpoint_result);

    let payload = ContextCheckPayload {
        policy_digest: checkpoint_result.policy_digest.clone(),
        focus_digest: context_check::build_focus_digest_from_inputs(&inputs),
        state_delta: context_check::build_state_delta_from_inputs(&inputs),
        risk_summary: context_check::build_risk_summary_from_inputs(trigger, &inputs),
    };
    let decision = context_check::decide(trigger, &inputs, &payload);
    let decisions = context_check_decisions(
        decision.action,
        decision.reason,
        decision.recommendation.as_str(),
    );

    // Persist cache for future throttle pre-checks.
    write_context_check_cache(
        &checkpoint_result.layout,
        &ContextCheckCache {
            source_fingerprint: source_fingerprint.clone(),
            epoch_s: session_state::now_epoch_s().unwrap_or(0),
            path: checkpoint_result.path.clone(),
            profile: checkpoint_result.profile.clone(),
            project_id: checkpoint_result.project_id.clone(),
            locality_id: checkpoint_result.locality_id.clone(),
        },
    );

    Ok(ContextCheckReport {
        command: "context-check",
        schema_version: CONTEXT_CHECK_SCHEMA_VERSION,
        ok: true,
        decisions,
        path: checkpoint_result.path,
        profile: checkpoint_result.profile,
        project_id: checkpoint_result.project_id,
        locality_id: checkpoint_result.locality_id,
        trigger,
        action: decision.action,
        reason: decision.reason,
        recommendation: decision.recommendation,
        urgency: decision.urgency,
        throttle: ContextCheckThrottle {
            suppressed: decision.suppressed,
            next_interval_hint_seconds: decision.next_interval_hint_seconds,
        },
        source_fingerprint,
        payload,
        evidence: decision.evidence,
        session_state: checkpoint_result.session_state,
        execution_gates: checkpoint_result.execution_gates,
        recovery: checkpoint_result.recovery,
        escalation: checkpoint_result.escalation,
        context_health: checkpoint_result.context_health,
        session_boundary: checkpoint_result.session_boundary.inner,
        behavioral_drift: checkpoint_result.behavioral_drift,
    })
}

/// Build a compact throttled context-check report from cached metadata.
/// Returned when the source fingerprint is unchanged and the last check
/// is inside the throttle window — skips the entire checkpoint/decision
/// pipeline (#627).
fn build_throttled_context_check_report(
    trigger: ContextCheckTrigger,
    cached: &ContextCheckCache,
    layout: &StateLayout,
) -> ContextCheckReport {
    let path = layout.state_db_path().display().to_string();
    ContextCheckReport {
        command: "context-check",
        schema_version: CONTEXT_CHECK_SCHEMA_VERSION,
        ok: true,
        decisions: context_check_decisions(
            ContextCheckAction::None,
            "throttled_unchanged",
            "continue",
        ),
        path: cached.path.clone(),
        profile: cached.profile.clone(),
        project_id: cached.project_id.clone(),
        locality_id: cached.locality_id.clone(),
        trigger,
        action: ContextCheckAction::None,
        reason: "throttled_unchanged",
        recommendation: "continue".to_owned(),
        urgency: ContextCheckUrgency::Low,
        throttle: ContextCheckThrottle {
            suppressed: true,
            next_interval_hint_seconds: CONTEXT_CHECK_THROTTLE_WINDOW_SECONDS,
        },
        source_fingerprint: cached.source_fingerprint.clone(),
        payload: ContextCheckPayload {
            policy_digest: String::new(),
            focus_digest: String::new(),
            state_delta: String::new(),
            risk_summary: format!(
                "throttled: source fingerprint unchanged within {}s window",
                CONTEXT_CHECK_THROTTLE_WINDOW_SECONDS
            ),
        },
        evidence: Vec::new(),
        session_state: RadarSessionStateView {
            path: path.clone(),
            status: "throttled",
            session_id: None,
            mode: None,
            start_count: None,
            lifecycle: session_state::SessionLifecycleProjection::missing(),
        },
        execution_gates: session_gates::ExecutionGatesView {
            path: path.clone(),
            status: "throttled",
            seeded_from: None,
            revision: None,
            total_count: 0,
            unfinished_count: 0,
            attention_anchor: None,
            gates: Vec::new(),
        },
        recovery: runtime_state::RuntimeRecoveryView {
            status: "throttled",
            checkpoint: runtime_state::RuntimeCheckpointView {
                path: String::new(),
                status: "throttled",
                origin: None,
                captured_at_epoch_s: None,
                session_started_at_epoch_s: None,
                summary: None,
                immediate_actions: Vec::new(),
                key_files: Vec::new(),
            },
            working_buffer: runtime_state::RuntimeWorkingBufferView {
                path: String::new(),
                status: "throttled",
                origin: None,
                captured_at_epoch_s: None,
                session_started_at_epoch_s: None,
                summary_lines: Vec::new(),
            },
        },
        escalation: escalation_state::EscalationView {
            path,
            status: "throttled",
            blocking_count: 0,
            non_blocking_count: 0,
            entries: Vec::new(),
        },
        context_health: ContextHealth {
            source: "throttled",
            context_used_pct: None,
            degradation_risk_pct: None,
            band: "healthy",
            confidence: "low",
            signals: ContextSignals {
                compacted: None,
                session_minutes: None,
                ccd_start_cycles_in_conversation: None,
                host_total_tokens: None,
                host_context_window_tokens: None,
                tool_output_kb: None,
                operator_symptoms: Vec::new(),
            },
            cost: telemetry_cost::TelemetryCostView::unavailable(
                "throttled",
                "context-check throttled; cost not evaluated",
            ),
            recommendation: "continue",
        },
        session_boundary: SessionBoundaryRecommendation::new(
            crate::session_boundary::SessionBoundaryAction::Continue,
            "throttled: no fresh boundary evaluation",
            Vec::new(),
        ),
        behavioral_drift: BehavioralDriftState {
            status: DriftAggregateStatus::NoSignal,
            summary: "throttled: no fresh drift evaluation".to_owned(),
            evidence: Vec::new(),
            recommended_corrections: Vec::new(),
            signals: Vec::new(),
        },
    }
}

fn read_handoff(
    repo_root: &Path,
    layout: &StateLayout,
    runtime: &runtime_state::LoadedRuntimeState,
    git: Option<&GitState>,
    active_session_id: Option<&str>,
    session_view: &RadarSessionStateView,
) -> Result<LoadedHandoffState> {
    let path = layout.state_db_path();
    if runtime.sources.handoff.is_missing() {
        let remediation = handoff_write_command(repo_root, layout);
        let rerun = radar_state_command(repo_root, layout);
        bail!(
            "ccd handover cannot evaluate continuity or wrap-up readiness because the canonical workspace-local handoff is uninitialized at {}. Create it with `{remediation}` and rerun `{rerun}`.",
            path.display(),
        );
    }

    let mut current_system_state = handoff::current_system_state_lines(git, active_session_id);
    if let Some(mode) = session_view.mode {
        if session_view.status == "active" && mode != session_state::SessionMode::General {
            current_system_state.push(format!("Session mode: `{}`", mode.as_str()));
        }
    }

    let contents = runtime.sources.handoff.content.clone();
    Ok(LoadedHandoffState {
        handoff: HandoffState {
            path: path.display().to_string(),
            title: handoff::extract_title(&contents)?,
            immediate_actions: handoff::extract_numbered_section(&contents, "Immediate Actions"),
            current_system_state,
            key_files: handoff::extract_numbered_section(
                &contents,
                handoff::KEY_FILES_SECTION_TITLE,
            ),
            definition_of_done: handoff::extract_numbered_section(&contents, "Definition of Done"),
        },
        contents,
    })
}

fn handoff_write_command(repo_root: &Path, layout: &StateLayout) -> String {
    let base = format!("ccd handoff write --path {}", repo_root.display());
    if layout.profile().as_str() == DEFAULT_PROFILE {
        base
    } else {
        format!("{base} --profile {}", layout.profile().as_str())
    }
}

fn radar_state_command(repo_root: &Path, layout: &StateLayout) -> String {
    let base = format!("ccd handover --path {}", repo_root.display());
    if layout.profile().as_str() == DEFAULT_PROFILE {
        base
    } else {
        format!("{base} --profile {}", layout.profile().as_str())
    }
}

pub(crate) fn build_session_state_view(
    layout: &StateLayout,
    tracked_session: Option<&session_state::SessionStateFile>,
    tracked_activity: Option<&session_state::SessionActivityState>,
) -> RadarSessionStateView {
    let path = layout.state_db_path().display().to_string();
    match tracked_session {
        Some(state) => {
            let status = session_state_status(state, session_state::now_epoch_s().ok());
            RadarSessionStateView {
                path,
                status,
                session_id: state.session_id.clone(),
                mode: Some(state.mode),
                start_count: Some(state.start_count),
                lifecycle: session_state::lifecycle_projection(
                    state,
                    session_state::now_epoch_s().unwrap_or_default(),
                    None,
                    tracked_activity,
                ),
            }
        }
        None => RadarSessionStateView {
            path,
            status: "missing",
            session_id: None,
            mode: None,
            start_count: None,
            lifecycle: session_state::SessionLifecycleProjection::missing(),
        },
    }
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn build_projection_telemetry(
    layout: &StateLayout,
    shared_db: Option<&db::StateDb>,
    runtime: &runtime_state::LoadedRuntimeState,
    execution_gates: &session_gates::ExecutionGatesView,
    escalation_view: &escalation_state::EscalationView,
    recovery: &runtime_state::RuntimeRecoveryView,
    git: &Option<GitState>,
    session_state_view: &RadarSessionStateView,
    active_session_id: Option<&str>,
    host_snapshot: Option<&host_telemetry::HostContextSnapshot>,
    target: compiled_state::ProjectionTarget,
) -> Result<(
    ProjectionTelemetryView,
    compiled_state::ProjectionDigests,
    compiled_state::ProjectionDigests,
)> {
    let source_fingerprint =
        compiled_state::fingerprint_for_inputs(&runtime.sources, runtime.pod_identity_active)?;
    let narrative = compiled_state::preview_with_precomputed_fingerprint(
        layout,
        shared_db,
        runtime,
        target,
        &source_fingerprint,
        true,
    )?;
    let symbolic = compiled_state::preview_symbolic_with_precomputed_fingerprint(
        layout,
        shared_db,
        runtime,
        target,
        &source_fingerprint,
    )?;
    let bundle = compiled_state::preview_bundle_with_precomputed_fingerprint(
        layout,
        shared_db,
        runtime,
        target,
        &source_fingerprint,
    )?;

    let execution_gates_json = serde_json::to_string(execution_gates).unwrap_or_default();
    let escalation_json = serde_json::to_string(escalation_view).unwrap_or_default();
    let recovery_json = serde_json::to_string(recovery).unwrap_or_default();
    let git_state_json = serde_json::to_string(git).unwrap_or_default();
    let session_state_json = serde_json::to_string(session_state_view).unwrap_or_default();

    let base_digests = compiled_state::compute_projection_digests(&narrative.value);
    let projection_digests = compiled_state::compute_extended_digests(
        &base_digests,
        &execution_gates_json,
        &escalation_json,
        &recovery_json,
        &git_state_json,
        &session_state_json,
    );

    let (delta_status, delta) = if let Some(session_id) = active_session_id {
        // Reuse the shared handle when the caller provided one via the
        // shared-db-with-migration helper so trees where `state.db` exists
        // but `projection_metadata.json` has not yet been migrated still
        // resolve the baseline correctly. On trees with no state.db yet
        // (shared_db is None) fall back to the per-module probe, which
        // runs migration through its own opener.
        let baseline_lookup = match shared_db {
            Some(db) => projection_metadata::load_baseline_for_session_from_shared_db(
                db, layout, session_id,
            )?,
            None => projection_metadata::load_baseline_for_session(layout, session_id)?,
        };
        match baseline_lookup {
            Some(observation) => match observation.projection_digests {
                Some(baseline) => (
                    "active_session",
                    compiled_state::compute_projection_delta_stats(&baseline, &projection_digests),
                ),
                None => ("baseline_missing", default_projection_delta_stats()),
            },
            None => ("baseline_missing", default_projection_delta_stats()),
        }
    } else {
        ("no_active_session", default_projection_delta_stats())
    };

    let token_counts = compiled_state::ProjectionTokenTelemetry {
        surface_tokens: compiled_state::compute_surface_token_counts(
            &narrative.value,
            &execution_gates_json,
            &escalation_json,
            &recovery_json,
            &git_state_json,
            &session_state_json,
        ),
        format_tokens: compiled_state::compute_format_token_counts(
            &narrative.value,
            &symbolic.value,
            &bundle.value,
        ),
    };
    let source_fingerprint = narrative.value.source_fingerprint.clone();

    let mut current = BTreeMap::new();
    current.insert(
        compiled_state::ProjectionFormat::Narrative
            .as_str()
            .to_owned(),
        narrative.cadence,
    );
    current.insert(
        compiled_state::ProjectionFormat::Symbolic
            .as_str()
            .to_owned(),
        symbolic.cadence,
    );
    current.insert(
        compiled_state::ProjectionFormat::Bundle.as_str().to_owned(),
        bundle.cadence,
    );

    Ok((
        ProjectionTelemetryView {
            source_fingerprint,
            delta_status,
            delta,
            token_counts,
            cadence: ProjectionCadenceView {
                target: target.as_str(),
                current,
                history: compiled_state::cadence_summary_for_target(layout, target, 128)?,
                provider_cache_usage: provider_cache_usage_view(host_snapshot),
            },
        },
        base_digests,
        projection_digests,
    ))
}

fn default_projection_delta_stats() -> compiled_state::ProjectionDeltaStats {
    compiled_state::ProjectionDeltaStats {
        skipped: 0,
        refreshed: compiled_state::SESSION_RENDER_SURFACE_ORDER.len() as u32,
        total: compiled_state::SESSION_RENDER_SURFACE_ORDER.len() as u32,
    }
}

fn provider_cache_usage_view(
    host_snapshot: Option<&host_telemetry::HostContextSnapshot>,
) -> Option<ProjectionProviderCacheUsageView> {
    let usage = host_snapshot.map(|snapshot| &snapshot.cost_usage)?;
    if usage.is_empty() {
        return None;
    }
    Some(ProjectionProviderCacheUsageView {
        cache_creation_input_tokens: (usage.cache_creation_input_tokens > 0)
            .then_some(usage.cache_creation_input_tokens),
        cache_read_input_tokens: (usage.cache_read_input_tokens > 0)
            .then_some(usage.cache_read_input_tokens),
        blended_total_tokens: usage.blended_total_tokens,
    })
}

fn session_state_status(
    state: &session_state::SessionStateFile,
    now_epoch_s: Option<u64>,
) -> &'static str {
    if state.session_id.is_none() {
        return "stale";
    }
    match now_epoch_s {
        Some(now) if session_state::is_stale(state, now) => "stale",
        _ => "active",
    }
}

fn read_execution_gates(
    runtime: &runtime_state::LoadedRuntimeState,
) -> session_gates::ExecutionGatesView {
    runtime.execution_gates.view.clone()
}

fn read_effective_memory(runtime: &runtime_state::LoadedRuntimeState) -> MemoryState {
    let structured = memory_entries::inspect_sources_with_branch_and_clone(
        some_if_loaded(&runtime.sources.profile_memory),
        some_if_loaded(&runtime.sources.locality_memory),
        some_if_loaded(&runtime.sources.branch_memory),
        some_if_loaded(&runtime.sources.clone_memory),
    );

    let mut segments = Vec::new();
    let mut contributing_paths = Vec::new();
    if let Some(contents) = some_if_loaded(&runtime.sources.profile_memory) {
        if !contents.is_empty() {
            segments.push(contents.to_owned());
            contributing_paths.push(runtime.sources.profile_memory.path.display().to_string());
        }
    }
    if let Some(contents) = some_if_loaded(&runtime.sources.locality_memory) {
        if !contents.is_empty() {
            segments.push(contents.to_owned());
            contributing_paths.push(runtime.sources.locality_memory.path.display().to_string());
        }
    }
    if let Some(contents) = some_if_loaded(&runtime.sources.branch_memory) {
        if !contents.is_empty() {
            segments.push(contents.to_owned());
            contributing_paths.push(runtime.sources.branch_memory.path.display().to_string());
        }
    }
    if let Some(contents) = some_if_loaded(&runtime.sources.clone_memory) {
        if !contents.is_empty() {
            segments.push(contents.to_owned());
            contributing_paths.push(runtime.sources.clone_memory.path.display().to_string());
        }
    }

    let content = segments.join("\n\n");
    let status = if content.is_empty() {
        "empty"
    } else {
        "loaded"
    };

    MemoryState {
        status,
        profile_path: runtime.sources.profile_memory.path.display().to_string(),
        project_path: runtime.sources.locality_memory.path.display().to_string(),
        locality_path: runtime.sources.locality_memory.path.display().to_string(),
        work_stream_path: runtime.sources.branch_memory.path.display().to_string(),
        branch_path: runtime.sources.branch_memory.path.display().to_string(),
        workspace_path: runtime.sources.clone_memory.path.display().to_string(),
        clone_path: runtime.sources.clone_memory.path.display().to_string(),
        content,
        structured,
        contributing_paths,
    }
}

fn read_repo_native_checks(repo_root: &Path) -> RepoNativeChecksState {
    match check::run(repo_root) {
        Ok(report) => RepoNativeChecksState {
            surface: report.surface,
            failures: report.failures,
            checks: report
                .checks
                .into_iter()
                .map(|check| RepoNativeCheckResult {
                    id: check.id,
                    path: check.path,
                    status: check.status,
                    severity: check.severity,
                    message: check.message,
                    duration_ms: check.duration_ms,
                    exit_code: check.exit_code,
                })
                .collect(),
        },
        Err(error) => {
            let surface_path = repo_root.join(check::COMMANDS_DIR);
            let message = format!("Failed to inspect repo-native checks: {error:#}");

            RepoNativeChecksState {
                surface: check::CheckSurfaceView {
                    path: surface_path.display().to_string(),
                    status: "error",
                    prefix: check::CHECK_PREFIX,
                    entries: Vec::new(),
                    note: Some(message.clone()),
                },
                failures: 1,
                checks: vec![RepoNativeCheckResult {
                    id: "surface".to_owned(),
                    path: surface_path.display().to_string(),
                    status: "fail",
                    severity: "error",
                    message,
                    duration_ms: 0,
                    exit_code: None,
                }],
            }
        }
    }
}

fn some_if_loaded(surface: &runtime_state::RuntimeTextSurface) -> Option<&str> {
    if surface.is_missing() {
        None
    } else {
        Some(&surface.content)
    }
}

fn active_session_start_count(
    tracked_session: Option<&session_state::SessionStateFile>,
) -> Option<u64> {
    let tracked_session = tracked_session?;
    if session_state_status(tracked_session, session_state::now_epoch_s().ok()) != "active" {
        return None;
    }
    Some(u64::from(tracked_session.start_count))
}

// ---------------------------------------------------------------------------
// apply_delta_filter: replace unchanged source-data surfaces with stubs
// ---------------------------------------------------------------------------

/// Replace unchanged source-data surfaces with `{"status": "unchanged", "digest": "<hash>"}`.
/// Computed surfaces (context_health, behavioral_drift, evaluation, etc.) are always emitted in full.
pub(crate) fn apply_delta_filter(
    mut value: serde_json::Value,
    baseline: &compiled_state::ProjectionDigests,
    current: &compiled_state::ProjectionDigests,
) -> serde_json::Value {
    // Pairs of (JSON field name in the report, baseline digest, current digest)
    let pairs: &[(&str, &str, &str)] = &[
        ("handoff", &baseline.handoff, &current.handoff),
        (
            "effective_memory",
            &baseline.effective_memory,
            &current.effective_memory,
        ),
        (
            "execution_gates",
            &baseline.execution_gates,
            &current.execution_gates,
        ),
        ("escalation", &baseline.escalation, &current.escalation),
        ("recovery", &baseline.recovery, &current.recovery),
        ("git", &baseline.git_state, &current.git_state),
        (
            "session_state",
            &baseline.session_state,
            &current.session_state,
        ),
    ];

    if let Some(obj) = value.as_object_mut() {
        for &(field, base_digest, curr_digest) in pairs {
            if !base_digest.is_empty() && base_digest == curr_digest {
                // Only collapse fields that are present and non-null; replacing
                // null with an object stub would change the JSON type for
                // consumers that distinguish null vs structured values.
                let is_present_and_non_null = obj.get(field).is_some_and(|v| !v.is_null());
                if is_present_and_non_null {
                    obj.insert(
                        field.to_owned(),
                        serde_json::json!({
                            "status": "unchanged",
                            "digest": curr_digest
                        }),
                    );
                }
            }
        }
    }

    value
}

// ---------------------------------------------------------------------------
// commit_writes: batch handoff + recovery + session-clear in one call
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
pub(crate) struct CommitPayload {
    #[serde(default)]
    pub(crate) handoff: Option<crate::handoff::write::HandoffWriteInput>,
    #[serde(default)]
    pub(crate) recovery: Option<crate::recovery::RecoveryWriteInput>,
    #[serde(default)]
    pub(crate) clear_session: bool,
}

#[derive(Serialize)]
pub(crate) struct CommitResult {
    pub(crate) ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) handoff: Option<CommitWriteOutcome>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) recovery: Option<CommitWriteOutcome>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) session_cleared: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) error: Option<String>,
}

#[derive(Serialize)]
pub(crate) struct CommitWriteOutcome {
    pub(crate) ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) error: Option<String>,
}

pub(crate) fn commit_writes(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    payload: CommitPayload,
) -> CommitResult {
    let write_options = crate::state::protected_write::ExclusiveWriteOptions::default();

    let mut result = CommitResult {
        ok: true,
        handoff: None,
        recovery: None,
        session_cleared: None,
        error: None,
    };

    if let Some(handoff_input) = payload.handoff {
        match crate::handoff::write::run_with_input(
            repo_root,
            explicit_profile,
            write_options.clone(),
            handoff_input,
            // Radar's auto-commit path drafts handoffs from freshly
            // inspected repo state; reference audits there are already
            // covered by the read-time `handoff_references_resolved`
            // signal. Keep the default write-time audit on so the commit
            // fails closed the same way an interactive write would.
            crate::handoff::write::WriteOptions::default(),
        ) {
            Ok(_) => {
                result.handoff = Some(CommitWriteOutcome {
                    ok: true,
                    error: None,
                });
            }
            Err(e) => {
                result.ok = false;
                result.handoff = Some(CommitWriteOutcome {
                    ok: false,
                    error: Some(format!("{e:#}")),
                });
                result.error = Some(format!("handoff write failed: {e:#}"));
                return result;
            }
        }
    }

    if let Some(recovery_input) = payload.recovery {
        match crate::recovery::write_with_input(
            repo_root,
            explicit_profile,
            recovery_input,
            write_options,
        ) {
            Ok(_) => {
                result.recovery = Some(CommitWriteOutcome {
                    ok: true,
                    error: None,
                });
            }
            Err(e) => {
                result.ok = false;
                result.recovery = Some(CommitWriteOutcome {
                    ok: false,
                    error: Some(format!("{e:#}")),
                });
                result.error = Some(format!("recovery write failed: {e:#}"));
                return result;
            }
        }
    }

    if payload.clear_session {
        let locality_id = crate::repo::marker::load(repo_root)
            .ok()
            .flatten()
            .map(|m| m.locality_id);
        match crate::state::session::clear(
            repo_root,
            explicit_profile,
            locality_id.as_deref(),
            crate::state::session::SessionClearOptions::default(),
        ) {
            Ok(_) => {
                result.session_cleared = Some(true);
            }
            Err(e) => {
                result.ok = false;
                result.session_cleared = Some(false);
                result.error = Some(format!("session clear failed: {e:#}"));
                return result;
            }
        }
    }

    result
}

#[cfg(test)]
mod tests {
    use super::behavioral_drift::{build_handoff_issue_references_signal, collect_session_commits};
    use super::{session_state_status, DriftSignalStatus};
    use crate::state::runtime::{RuntimeHandoffItem, RuntimeHandoffState, RuntimeLifecycle};
    use crate::state::session::{SessionMode, SessionOwnerKind, SessionStateFile};
    use std::path::Path;

    #[test]
    fn collect_session_commits_returns_err_on_non_git_path() {
        // Fail-closed precondition for ccd#570: when `git log` cannot
        // walk the target directory, the helper must surface the
        // failure rather than swallow it as an empty commit list.
        let result = collect_session_commits(Path::new("/nonexistent-path-for-test"), 1);
        assert!(result.is_err(), "expected Err for non-git path");
    }

    #[test]
    fn handoff_issue_references_signal_fails_closed_on_git_log_failure() {
        // ccd#570 adversarial-review: when the handoff names issue
        // tokens but commit history cannot be read, the detector must
        // return `drift` so the fast-path rule forces a handoff
        // rewrite. A `no_signal` return here would let the ccd-radar
        // fast path declare the handoff clean without the one check
        // meant to catch stale issue references ever running.
        let handoff = RuntimeHandoffState {
            title: "Next: continue ccd#999".to_owned(),
            immediate_actions: vec![RuntimeHandoffItem {
                text: "Work on #999.".to_owned(),
                lifecycle: RuntimeLifecycle::Active,
            }],
            completed_state: Vec::new(),
            operational_guardrails: Vec::new(),
            key_files: Vec::new(),
            definition_of_done: Vec::new(),
        };
        let session = SessionStateFile {
            schema_version: 3,
            started_at_epoch_s: 1,
            last_started_at_epoch_s: 1,
            start_count: 1,
            session_id: Some("ses_test".to_owned()),
            mode: SessionMode::General,
            owner_kind: SessionOwnerKind::Interactive,
            owner_id: None,
            supervisor_id: None,
            lease_ttl_secs: None,
            last_heartbeat_at_epoch_s: None,
            revision: 1,
        };

        let signal = build_handoff_issue_references_signal(
            Path::new("/nonexistent-path-for-test"),
            &handoff,
            Some(&session),
        );

        assert!(
            signal.status == DriftSignalStatus::Drift,
            "git-log read failure with handoff issue refs present must fail closed; got summary={}",
            signal.summary
        );
        assert!(
            signal.evidence.iter().any(|line| line.contains("#999")),
            "evidence must surface the unverified issue tokens, got: {:?}",
            signal.evidence
        );
    }

    #[test]
    fn session_state_status_treats_missing_session_id_as_stale_without_clock() {
        let state = SessionStateFile {
            schema_version: 3,
            started_at_epoch_s: 10,
            last_started_at_epoch_s: 10,
            start_count: 1,
            session_id: None,
            mode: SessionMode::Research,
            owner_kind: SessionOwnerKind::Interactive,
            owner_id: None,
            supervisor_id: None,
            lease_ttl_secs: None,
            last_heartbeat_at_epoch_s: None,
            revision: 0,
        };

        assert_eq!(session_state_status(&state, None), "stale");
    }
}
