zynk 1.0.1

use crate::{CliError, CliResult};
use clap::{Args, Subcommand};
use rand::{rngs::OsRng, Rng};
use rusqlite::{
    params, Connection, OpenFlags, OptionalExtension, Transaction, TransactionBehavior,
};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fs;
use std::io::Read;
use std::path::{Path, PathBuf};
use std::time::Duration;

const CURRENT_SCHEMA_VERSION: i64 = 9;

/// SQLite GLOB matching the canonical `YYYY-MM-DDTHH:MM:SSZ` timestamp form.
/// `?` matches exactly one character, so offset forms like `+07:00` (length 25)
/// never match. See ADR 026 D1.
const CANONICAL_TS_GLOB: &str = "????-??-??T??:??:??Z";
const BUSY_TIMEOUT_MS: u64 = 5000;

#[derive(Debug, Args)]
pub struct DbArgs {
    #[arg(
        long,
        default_value = ".zynk/zynk.db",
        help = "SQLite database path for live zynk state"
    )]
    pub db: PathBuf,
    #[command(subcommand)]
    pub command: DbCommand,
}

#[derive(Debug, Subcommand)]
pub enum DbCommand {
    /// Create or migrate the live-state SQLite database.
    Init,
    /// Append audit records to the live-state SQLite database.
    Audit(Box<DbAuditCommand>),
    /// Import human-readable artifacts into the live-state SQLite database.
    Import(Box<DbImportCommand>),
    /// Export live-state SQLite data into human-readable artifacts.
    Export(Box<DbExportCommand>),
    /// Serve the DB-backed local dashboard console.
    Serve(crate::db_dashboard::DbServeArgs),
}

#[derive(Debug, Args)]
pub struct DbAuditCommand {
    #[command(subcommand)]
    pub command: DbAuditSubcommand,
}

#[derive(Debug, Subcommand)]
pub enum DbAuditSubcommand {
    /// Append one ADR 023 audit record transactionally.
    Append(DbAuditAppendArgs),
}

#[derive(Debug, Args)]
pub struct DbAuditAppendArgs {
    #[arg(long)]
    pub session_id: String,
    #[arg(long)]
    pub audit_id: Option<String>,
    #[arg(long)]
    pub timestamp: String,
    #[arg(long)]
    pub source_agent_id: Option<String>,
    #[arg(long)]
    pub target_agent_id: Option<String>,
    #[arg(long)]
    pub source_address: String,
    #[arg(long)]
    pub target_address: String,
    #[arg(long)]
    pub transport: String,
    #[arg(long)]
    pub transport_thread_id: Option<String>,
    #[arg(long)]
    pub workspace_id: String,
    #[arg(long)]
    pub mid: String,
    #[arg(long = "type")]
    pub record_type: String,
    #[arg(long)]
    pub mode: Option<String>,
    #[arg(long)]
    pub r#ref: Option<String>,
    #[arg(long)]
    pub re: Option<String>,
    #[arg(long, value_parser = ["agent", "operator", "helper-tool", "unknown"])]
    pub command_origin: String,
    #[arg(long)]
    pub payload: Option<String>,
    #[arg(long)]
    pub payload_file: Option<PathBuf>,
    #[arg(long, default_value = "hash-only")]
    pub payload_redaction_policy: String,
    #[arg(long, default_value_t = 12)]
    pub excerpt_chars: usize,
    #[arg(long)]
    pub delivery_status: String,
    #[arg(long)]
    pub observed_by: String,
    #[arg(long, value_parser = ["transport", "agent", "operator", "helper-tool"])]
    pub verified_by: String,
}

#[derive(Debug, Args)]
pub struct DbImportCommand {
    #[command(subcommand)]
    pub command: DbImportSubcommand,
}

#[derive(Debug, Subcommand)]
pub enum DbImportSubcommand {
    /// Import v0.1 outputs/sessions artifacts.
    Outputs(DbImportOutputsArgs),
}

#[derive(Debug, Args)]
pub struct DbExportCommand {
    #[command(subcommand)]
    pub command: DbExportSubcommand,
}

#[derive(Debug, Subcommand)]
pub enum DbExportSubcommand {
    /// Export one session to outputs/sessions artifacts.
    Outputs(DbExportOutputsArgs),
}

#[derive(Debug, Args)]
#[command(
    after_help = "v0.2 exports status.md and audit.md only; summary.md is not exported because the DB schema does not store summary body content."
)]
pub struct DbExportOutputsArgs {
    #[arg(
        long,
        default_value = "outputs",
        help = "runtime outputs root; writes <root>/sessions/<session-id>"
    )]
    pub root: PathBuf,
    #[arg(long)]
    pub session_id: String,
}

#[derive(Debug, Args)]
#[command(
    after_help = "Legacy DB recovery: if an upgraded database warns about non-canonical \
timestamps, rebuild it from artifacts with: rm .zynk/zynk.db && zynk db init && \
zynk db import outputs --root outputs"
)]
pub struct DbImportOutputsArgs {
    #[arg(
        long,
        default_value = "outputs",
        help = "runtime outputs root; reads <root>/sessions/*"
    )]
    pub root: PathBuf,
    #[arg(long)]
    pub session_id: Option<String>,
    #[arg(long)]
    pub timestamp: Option<String>,
}

pub fn run(args: DbArgs) -> CliResult<()> {
    match args.command {
        DbCommand::Init => {
            initialize(&args.db)?;
            println!("{}", display_path(&args.db)?.display());
            Ok(())
        }
        DbCommand::Audit(command) => run_audit_command(&args.db, *command),
        DbCommand::Import(command) => run_import_command(&args.db, *command),
        DbCommand::Export(command) => run_export_command(&args.db, *command),
        DbCommand::Serve(serve_args) => crate::db_dashboard::serve(&args.db, serve_args),
    }
}

fn run_audit_command(path: &Path, command: DbAuditCommand) -> CliResult<()> {
    match command.command {
        DbAuditSubcommand::Append(args) => {
            let audit_id = append_audit_record(path, &args)?;
            println!("{audit_id}");
            Ok(())
        }
    }
}

fn run_import_command(path: &Path, command: DbImportCommand) -> CliResult<()> {
    match command.command {
        DbImportSubcommand::Outputs(args) => {
            let summary = import_outputs(path, &args)?;
            println!(
                "sessions imported: {}; warnings: {}",
                summary.sessions_seen, summary.warning_count
            );
            Ok(())
        }
    }
}

fn run_export_command(path: &Path, command: DbExportCommand) -> CliResult<()> {
    match command.command {
        DbExportSubcommand::Outputs(args) => {
            let summary = export_outputs(path, &args)?;
            println!(
                "session exported: {}; written: {}; unchanged: {}",
                args.session_id, summary.written, summary.unchanged
            );
            Ok(())
        }
    }
}

fn display_path(path: &Path) -> CliResult<PathBuf> {
    if path.is_absolute() {
        return Ok(path.to_path_buf());
    }
    std::env::current_dir()
        .map(|cwd| cwd.join(path))
        .map_err(|error| CliError::failure(format!("failed to read current directory: {error}")))
}

fn initialize(path: &Path) -> CliResult<()> {
    // ADR 028 C1: db init at a `.zynk` default path also self-protects.
    ensure_zynk_gitignore(path)?;
    open_database(path).map(|_| ())
}

pub(crate) fn open_database(path: &Path) -> CliResult<Connection> {
    if let Some(parent) = path
        .parent()
        .filter(|parent| !parent.as_os_str().is_empty())
    {
        fs::create_dir_all(parent).map_err(|error| {
            CliError::failure(format!("failed to create {}: {error}", parent.display()))
        })?;
    }

    let mut connection = Connection::open(path).map_err(|error| {
        CliError::failure(format!("failed to open {}: {error}", path.display()))
    })?;
    configure_connection(&connection)?;
    migrate(&mut connection)?;
    Ok(connection)
}

fn configure_connection(connection: &Connection) -> CliResult<()> {
    configure_read_connection(connection)?;
    connection
        .pragma_update(None, "journal_mode", "WAL")
        .map_err(|error| CliError::failure(format!("failed to enable SQLite WAL mode: {error}")))?;
    Ok(())
}

pub(crate) fn open_read_database(path: &Path) -> CliResult<Connection> {
    let connection =
        Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY).map_err(|error| {
            CliError::failure(format!("failed to open {}: {error}", path.display()))
        })?;
    configure_read_connection(&connection)?;
    Ok(connection)
}

fn configure_read_connection(connection: &Connection) -> CliResult<()> {
    connection
        .busy_timeout(Duration::from_millis(BUSY_TIMEOUT_MS))
        .map_err(|error| {
            CliError::failure(format!("failed to set SQLite busy timeout: {error}"))
        })?;
    connection
        .pragma_update(None, "foreign_keys", "ON")
        .map_err(|error| {
            CliError::failure(format!("failed to enable SQLite foreign keys: {error}"))
        })?;
    Ok(())
}

fn migrate(connection: &mut Connection) -> CliResult<()> {
    let version: i64 = connection
        .pragma_query_value(None, "user_version", |row| row.get(0))
        .map_err(|error| {
            CliError::failure(format!("failed to read SQLite schema version: {error}"))
        })?;
    if version > CURRENT_SCHEMA_VERSION {
        return Err(CliError::failure(format!(
            "database schema version {version} is newer than this zynk binary supports ({CURRENT_SCHEMA_VERSION}); upgrade zynk"
        )));
    }
    // T9: nothing to apply at the current version; avoid opening an empty
    // migration transaction on every DB open.
    if version == CURRENT_SCHEMA_VERSION {
        return Ok(());
    }
    let transaction = connection
        .transaction()
        .map_err(|error| CliError::failure(format!("failed to start SQLite migration: {error}")))?;
    create_schema_migrations(&transaction)?;
    if version < 1 {
        apply_v1(&transaction)?;
        set_user_version(&transaction, 1)?;
    }
    if version < 2 {
        apply_v2(&transaction)?;
        set_user_version(&transaction, 2)?;
    }
    if version < 3 {
        apply_v3(&transaction)?;
        set_user_version(&transaction, 3)?;
    }
    if version < 4 {
        apply_v4(&transaction)?;
        set_user_version(&transaction, 4)?;
    }
    if version < 5 {
        apply_v5(&transaction)?;
        set_user_version(&transaction, 5)?;
    }
    if version < 6 {
        apply_v6(&transaction)?;
        set_user_version(&transaction, 6)?;
    }
    if version < 7 {
        apply_v7(&transaction)?;
        set_user_version(&transaction, 7)?;
    }
    if version < 8 {
        apply_v8(&transaction)?;
        set_user_version(&transaction, 8)?;
    }
    if version < 9 {
        apply_v9(&transaction)?;
        set_user_version(&transaction, 9)?;
    }
    transaction.commit().map_err(|error| {
        CliError::failure(format!("failed to commit SQLite migration: {error}"))
    })?;
    // ADR 026 D1/C3: after upgrading a legacy DB, the timestamp triggers only
    // constrain future writes. Pre-existing non-canonical rows remain and would
    // still sort incorrectly, so emit a visible one-time warning with the
    // recovery recipe rather than relying on --help text alone.
    if version < 4 {
        warn_if_noncanonical_timestamps(connection)?;
    }
    Ok(())
}

fn warn_if_noncanonical_timestamps(connection: &Connection) -> CliResult<()> {
    let count = noncanonical_timestamp_count(connection)?;
    if count > 0 {
        eprintln!(
            "warning: {count} pre-existing row(s) carry non-canonical timestamps and will sort \
incorrectly; this upgrade does not rewrite them. Recover with: \
rm .zynk/zynk.db && zynk db init && zynk db import outputs --root outputs"
        );
    }
    Ok(())
}

/// Count rows in sort-critical columns whose timestamp is not canonical UTC-Z.
fn noncanonical_timestamp_count(connection: &Connection) -> CliResult<i64> {
    let glob = CANONICAL_TS_GLOB;
    let sql = format!(
        "SELECT
            (SELECT COUNT(*) FROM audit_records WHERE timestamp NOT GLOB '{glob}')
          + (SELECT COUNT(*) FROM status_events WHERE timestamp NOT GLOB '{glob}')
          + (SELECT COUNT(*) FROM messages WHERE timestamp NOT GLOB '{glob}')
          + (SELECT COUNT(*) FROM sessions
                WHERE created_at NOT GLOB '{glob}'
                   OR updated_at NOT GLOB '{glob}'
                   OR (completed_at IS NOT NULL AND completed_at NOT GLOB '{glob}'))"
    );
    connection
        .query_row(&sql, [], |row| row.get(0))
        .map_err(|error| CliError::failure(format!("failed to scan timestamps: {error}")))
}

fn set_user_version(connection: &Connection, version: i64) -> CliResult<()> {
    connection
        .pragma_update(None, "user_version", version)
        .map_err(|error| {
            CliError::failure(format!("failed to set SQLite schema version: {error}"))
        })?;
    Ok(())
}

fn create_schema_migrations(connection: &Connection) -> CliResult<()> {
    connection
        .execute_batch(
            "CREATE TABLE IF NOT EXISTS schema_migrations (
                version INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
            );",
        )
        .map_err(|error| {
            CliError::failure(format!("failed to create schema_migrations: {error}"))
        })?;
    Ok(())
}

fn apply_v1(connection: &Connection) -> CliResult<()> {
    // T4: no inner idempotency guard — migrate() only calls this when
    // user_version < 1, the single source of truth for "run v1".
    connection
        .execute_batch(
            "CREATE TABLE projects (
                project_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                root_path TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            CREATE TABLE agents (
                agent_id TEXT PRIMARY KEY,
                display_name TEXT NOT NULL,
                agent_kind TEXT NOT NULL,
                current_transport TEXT,
                current_address TEXT,
                -- D4: nullable by design. An agent with no active session
                -- legitimately has no current status; session_agents.agent_status
                -- (below) is NOT NULL because an agent in a session always has one.
                current_agent_status TEXT CHECK (
                    current_agent_status IS NULL
                    OR current_agent_status IN ('idle', 'working', 'blocked', 'done', 'unknown')
                ),
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            CREATE TABLE sessions (
                session_id TEXT PRIMARY KEY,
                project_id TEXT NOT NULL REFERENCES projects(project_id),
                title TEXT NOT NULL,
                phase TEXT NOT NULL,
                mode TEXT NOT NULL,
                workflow_status TEXT NOT NULL CHECK (
                    workflow_status IN ('idle', 'working', 'blocked', 'waiting-for-operator', 'done')
                ),
                lead_agent_id TEXT REFERENCES agents(agent_id),
                artifact_ref TEXT,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                completed_at TEXT
            );
            CREATE TABLE agent_addresses (
                agent_id TEXT NOT NULL REFERENCES agents(agent_id),
                transport TEXT NOT NULL,
                workspace_id TEXT,
                address TEXT NOT NULL,
                first_seen_at TEXT NOT NULL,
                last_seen_at TEXT NOT NULL,
                PRIMARY KEY (agent_id, transport, address, first_seen_at)
            );
            CREATE TABLE session_agents (
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                agent_id TEXT NOT NULL REFERENCES agents(agent_id),
                role TEXT NOT NULL,
                -- D4: NOT NULL — an agent within a session always has a status
                -- (the Herdr agent_status domain, distinct from sessions.workflow_status).
                agent_status TEXT NOT NULL CHECK (
                    agent_status IN ('idle', 'working', 'blocked', 'done', 'unknown')
                ),
                last_seen_at TEXT NOT NULL,
                PRIMARY KEY (session_id, agent_id)
            );
            CREATE TABLE audit_records (
                audit_id TEXT PRIMARY KEY,
                previous_audit_id TEXT REFERENCES audit_records(audit_id),
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                source_agent_id TEXT,
                target_agent_id TEXT,
                source_address TEXT NOT NULL,
                target_address TEXT NOT NULL,
                transport TEXT NOT NULL,
                workspace_id TEXT NOT NULL,
                mid TEXT NOT NULL,
                record_type TEXT NOT NULL,
                command_origin TEXT NOT NULL CHECK (
                    command_origin IN ('agent', 'operator', 'helper-tool', 'unknown')
                ),
                mode TEXT,
                ref TEXT,
                re TEXT,
                payload_hash TEXT NOT NULL,
                payload_redaction_policy TEXT NOT NULL CHECK (
                    payload_redaction_policy IN ('hash-only', 'excerpt', 'full')
                ),
                content_size INTEGER NOT NULL,
                delivery_status TEXT NOT NULL CHECK (
                    delivery_status IN ('drafted', 'sent', 'observed', 'failed', 'unknown')
                ),
                observed_by TEXT NOT NULL,
                verified_by TEXT NOT NULL CHECK (
                    verified_by IN ('transport', 'agent', 'operator', 'helper-tool')
                ),
                timestamp TEXT NOT NULL
            );
            CREATE TABLE messages (
                message_id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                mid TEXT NOT NULL,
                source_agent_id TEXT REFERENCES agents(agent_id),
                target_agent_id TEXT REFERENCES agents(agent_id),
                message_type TEXT NOT NULL,
                mode TEXT,
                ref TEXT,
                transport TEXT NOT NULL,
                transport_thread_id TEXT,
                payload_redaction_policy TEXT NOT NULL CHECK (
                    payload_redaction_policy IN ('hash-only', 'excerpt', 'full')
                ),
                payload_excerpt TEXT,
                payload_hash TEXT NOT NULL,
                latest_delivery_status TEXT NOT NULL CHECK (
                    latest_delivery_status IN ('drafted', 'sent', 'observed', 'failed', 'unknown')
                ),
                latest_verified_by TEXT NOT NULL CHECK (
                    latest_verified_by IN ('transport', 'agent', 'operator', 'helper-tool')
                ),
                latest_audit_id TEXT REFERENCES audit_records(audit_id),
                timestamp TEXT NOT NULL,
                UNIQUE (session_id, mid)
            );
            CREATE TABLE status_events (
                status_event_id INTEGER PRIMARY KEY,
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                timestamp TEXT NOT NULL,
                phase TEXT NOT NULL,
                mode TEXT NOT NULL,
                workflow_status TEXT NOT NULL CHECK (
                    workflow_status IN ('idle', 'working', 'blocked', 'waiting-for-operator', 'done')
                ),
                completed_since_last_update TEXT NOT NULL,
                in_progress TEXT NOT NULL,
                next_action TEXT NOT NULL,
                blockers TEXT NOT NULL,
                asks_for_zevs TEXT NOT NULL,
                risk_or_residual_uncertainty TEXT NOT NULL,
                expected_wait TEXT NOT NULL
            );
            CREATE TABLE artifacts (
                artifact_id INTEGER PRIMARY KEY,
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                kind TEXT NOT NULL,
                path TEXT NOT NULL,
                title TEXT,
                content_hash TEXT,
                updated_at TEXT NOT NULL,
                UNIQUE (session_id, path)
            );
            CREATE TABLE imports (
                import_id INTEGER PRIMARY KEY,
                source_kind TEXT NOT NULL CHECK (source_kind IN ('outputs', 'agent-loop')),
                source_path TEXT NOT NULL,
                source_hash TEXT NOT NULL,
                imported_at TEXT NOT NULL,
                result TEXT NOT NULL,
                warning_summary TEXT NOT NULL
            );
            INSERT OR IGNORE INTO schema_migrations (version, name) VALUES (1, 'initial-live-state');",
        )
        .map_err(|error| {
            CliError::failure(format!("failed to apply SQLite migration 1: {error}"))
        })?;
    Ok(())
}

fn apply_v2(connection: &Connection) -> CliResult<()> {
    if !table_has_column(connection, "audit_records", "command_origin")? {
        connection
            .execute_batch(
                "ALTER TABLE audit_records
                    ADD COLUMN command_origin TEXT NOT NULL DEFAULT 'unknown' CHECK (
                        command_origin IN ('agent', 'operator', 'helper-tool', 'unknown')
                    );",
            )
            .map_err(|error| {
                CliError::failure(format!(
                    "failed to add audit_records.command_origin: {error}"
                ))
            })?;
    }

    connection
        .execute_batch(
            "CREATE UNIQUE INDEX IF NOT EXISTS audit_records_one_child_per_previous
                ON audit_records(session_id, previous_audit_id)
                WHERE previous_audit_id IS NOT NULL;
            CREATE TRIGGER IF NOT EXISTS audit_records_no_sent_agent_insert
                BEFORE INSERT ON audit_records
                WHEN NEW.delivery_status = 'sent' AND NEW.verified_by = 'agent'
            BEGIN
                SELECT RAISE(ABORT, 'delivery_status=sent requires transport, helper-tool, or operator verification');
            END;
            CREATE TRIGGER IF NOT EXISTS audit_records_no_update
                BEFORE UPDATE ON audit_records
            BEGIN
                SELECT RAISE(ABORT, 'audit_records are append-only');
            END;
            CREATE TRIGGER IF NOT EXISTS audit_records_no_delete
                BEFORE DELETE ON audit_records
            BEGIN
                SELECT RAISE(ABORT, 'audit_records are append-only');
            END;
            CREATE TRIGGER IF NOT EXISTS messages_no_sent_agent_insert
                BEFORE INSERT ON messages
                WHEN NEW.latest_delivery_status = 'sent' AND NEW.latest_verified_by = 'agent'
            BEGIN
                SELECT RAISE(ABORT, 'delivery_status=sent requires transport, helper-tool, or operator verification');
            END;
            CREATE TRIGGER IF NOT EXISTS messages_no_sent_agent_update
                BEFORE UPDATE ON messages
                WHEN NEW.latest_delivery_status = 'sent' AND NEW.latest_verified_by = 'agent'
            BEGIN
                SELECT RAISE(ABORT, 'delivery_status=sent requires transport, helper-tool, or operator verification');
            END;
            INSERT OR IGNORE INTO schema_migrations (version, name) VALUES (2, 'audit-chain-constraints');",
        )
        .map_err(|error| {
            CliError::failure(format!("failed to apply SQLite migration 2: {error}"))
        })?;
    Ok(())
}

fn apply_v3(connection: &Connection) -> CliResult<()> {
    if !table_has_column(connection, "status_events", "event_text")? {
        connection
            .execute_batch(
                "ALTER TABLE status_events
                    ADD COLUMN event_text TEXT NOT NULL DEFAULT '';",
            )
            .map_err(|error| {
                CliError::failure(format!("failed to add status_events.event_text: {error}"))
            })?;
    }
    connection
        .execute(
            "INSERT OR IGNORE INTO schema_migrations (version, name)
             VALUES (3, 'status-event-text')",
            [],
        )
        .map_err(|error| {
            CliError::failure(format!("failed to apply SQLite migration 3: {error}"))
        })?;
    Ok(())
}

// ADR 026 v0.2.1 hardening: status_events.content_hash (+ backfill) and
// canonical-timestamp enforcement triggers on sort-critical columns.
fn apply_v4(connection: &Connection) -> CliResult<()> {
    if !table_has_column(connection, "status_events", "content_hash")? {
        connection
            .execute_batch(
                "ALTER TABLE status_events
                    ADD COLUMN content_hash TEXT NOT NULL DEFAULT '';",
            )
            .map_err(|error| {
                CliError::failure(format!("failed to add status_events.content_hash: {error}"))
            })?;
    }
    // C5: backfill existing rows with the same helper the import path uses, so a
    // legacy row's hash equals what a re-import computes (preserves idempotency).
    backfill_status_content_hash(connection)?;

    let glob = CANONICAL_TS_GLOB;
    connection
        .execute_batch(&format!(
            "CREATE TRIGGER IF NOT EXISTS audit_records_ts_canonical_insert
                BEFORE INSERT ON audit_records
                WHEN NEW.timestamp NOT GLOB '{glob}'
             BEGIN SELECT RAISE(ABORT, 'audit_records.timestamp must be RFC3339 UTC seconds (Z)'); END;
             CREATE TRIGGER IF NOT EXISTS status_events_ts_canonical_insert
                BEFORE INSERT ON status_events
                WHEN NEW.timestamp NOT GLOB '{glob}'
             BEGIN SELECT RAISE(ABORT, 'status_events.timestamp must be RFC3339 UTC seconds (Z)'); END;
             CREATE TRIGGER IF NOT EXISTS messages_ts_canonical_insert
                BEFORE INSERT ON messages
                WHEN NEW.timestamp NOT GLOB '{glob}'
             BEGIN SELECT RAISE(ABORT, 'messages.timestamp must be RFC3339 UTC seconds (Z)'); END;
             CREATE TRIGGER IF NOT EXISTS messages_ts_canonical_update
                BEFORE UPDATE ON messages
                WHEN NEW.timestamp NOT GLOB '{glob}'
             BEGIN SELECT RAISE(ABORT, 'messages.timestamp must be RFC3339 UTC seconds (Z)'); END;
             CREATE TRIGGER IF NOT EXISTS sessions_ts_canonical_insert
                BEFORE INSERT ON sessions
                WHEN NEW.created_at NOT GLOB '{glob}'
                  OR NEW.updated_at NOT GLOB '{glob}'
                  OR (NEW.completed_at IS NOT NULL AND NEW.completed_at NOT GLOB '{glob}')
             BEGIN SELECT RAISE(ABORT, 'sessions timestamps must be RFC3339 UTC seconds (Z)'); END;
             CREATE TRIGGER IF NOT EXISTS sessions_ts_canonical_update
                BEFORE UPDATE ON sessions
                WHEN NEW.created_at NOT GLOB '{glob}'
                  OR NEW.updated_at NOT GLOB '{glob}'
                  OR (NEW.completed_at IS NOT NULL AND NEW.completed_at NOT GLOB '{glob}')
             BEGIN SELECT RAISE(ABORT, 'sessions timestamps must be RFC3339 UTC seconds (Z)'); END;"
        ))
        .map_err(|error| {
            CliError::failure(format!("failed to create v4 timestamp triggers: {error}"))
        })?;

    connection
        .execute(
            "INSERT OR IGNORE INTO schema_migrations (version, name)
             VALUES (4, 'timestamp-canonical-and-content-hash')",
            [],
        )
        .map_err(|error| {
            CliError::failure(format!("failed to apply SQLite migration 4: {error}"))
        })?;
    Ok(())
}

// ADR 027: additive nullable audit_records.due + extend the canonical-timestamp
// shape guard to cover a non-null due.
fn apply_v5(connection: &Connection) -> CliResult<()> {
    // ADD COLUMN does not rewrite rows and coexists with the append-only
    // UPDATE/DELETE triggers; existing rows read due=NULL ("no deadline").
    if !table_has_column(connection, "audit_records", "due")? {
        connection
            .execute_batch("ALTER TABLE audit_records ADD COLUMN due TEXT;")
            .map_err(|error| {
                CliError::failure(format!("failed to add audit_records.due: {error}"))
            })?;
    }
    // Extend the v4 canonical-timestamp shape guard to also reject a non-null due
    // that is not canonical UTC-Z. CREATE TRIGGER IF NOT EXISTS would not replace
    // the existing trigger, so drop then recreate. This runs inside the migration
    // transaction, so there is no window where the invariant is unenforced. The
    // GLOB is defense-in-depth only; real RFC3339 validation happens at the CLI /
    // import boundary (ADR 027 C4).
    let glob = CANONICAL_TS_GLOB;
    connection
        .execute_batch(&format!(
            "DROP TRIGGER IF EXISTS audit_records_ts_canonical_insert;
             CREATE TRIGGER audit_records_ts_canonical_insert
                BEFORE INSERT ON audit_records
                WHEN NEW.timestamp NOT GLOB '{glob}'
                  OR (NEW.due IS NOT NULL AND NEW.due NOT GLOB '{glob}')
             BEGIN SELECT RAISE(ABORT, 'audit_records.timestamp and due must be RFC3339 UTC seconds (Z)'); END;"
        ))
        .map_err(|error| {
            CliError::failure(format!("failed to update v5 due trigger: {error}"))
        })?;
    connection
        .execute(
            "INSERT OR IGNORE INTO schema_migrations (version, name)
             VALUES (5, 'audit-due-and-db-canonical-write-path')",
            [],
        )
        .map_err(|error| {
            CliError::failure(format!("failed to apply SQLite migration 5: {error}"))
        })?;
    Ok(())
}

// ADR 033 M1: additive typed work-telemetry store. One row per work event; the
// payload column holds the validated, serialized typed value (serde_norway), not
// an untyped blob. The UNIQUE(session_id, kind, content_hash) backs the
// INSERT OR IGNORE content-dedup in `project_work_event`.
fn apply_v6(connection: &Connection) -> CliResult<()> {
    connection
        .execute_batch(
            "CREATE TABLE work_events (
                work_event_id INTEGER PRIMARY KEY,
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                actor_agent_id TEXT NOT NULL,
                kind TEXT NOT NULL CHECK (kind IN
                    ('think','tool','diff','plan','artifact','usage','system','gate','conflict')),
                timestamp TEXT NOT NULL,
                payload TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                created_at TEXT NOT NULL,
                UNIQUE (session_id, kind, content_hash)
            );
            CREATE INDEX work_events_session_ts ON work_events (session_id, timestamp);
            INSERT OR IGNORE INTO schema_migrations (version, name) VALUES (6, 'work-events');",
        )
        .map_err(|error| CliError::failure(format!("failed to apply schema v6: {error}")))?;
    Ok(())
}

// ADR 033 M2 (D4): typed operator-decision projection. audit_records holds the
// immutable proof; this table is the typed query/effect surface keyed by audit_id.
fn apply_v7(connection: &Connection) -> CliResult<()> {
    connection
        .execute_batch(
            "CREATE TABLE operator_decisions (
                audit_id TEXT PRIMARY KEY REFERENCES audit_records(audit_id),
                session_id TEXT NOT NULL REFERENCES sessions(session_id),
                decision_type TEXT NOT NULL CHECK (decision_type IN
                    ('gate-decision','conflict-resolve','mode-switch','interrupt','redirect')),
                target_work_event_id INTEGER REFERENCES work_events(work_event_id),
                verdict TEXT,
                resolution TEXT,
                mode_to TEXT,
                target_agent TEXT,
                reason TEXT,
                note TEXT,
                decision_status TEXT NOT NULL DEFAULT 'decided'
                    CHECK (decision_status IN ('decided')),
                notification_status TEXT NOT NULL DEFAULT 'not-requested'
                    CHECK (notification_status IN ('not-requested','sent','failed')),
                notification_audit_id TEXT REFERENCES audit_records(audit_id),
                notification_mid TEXT,
                notification_error TEXT,
                created_at TEXT NOT NULL
            );
            CREATE INDEX operator_decisions_session ON operator_decisions (session_id, created_at);
            CREATE INDEX operator_decisions_target ON operator_decisions (target_work_event_id);
            INSERT OR IGNORE INTO schema_migrations (version, name) VALUES (7, 'operator-decisions');",
        )
        .map_err(|error| CliError::failure(format!("failed to apply schema v7: {error}")))?;
    Ok(())
}

// ADR 034 M3: the local encrypted custody vault. One row per RETAINED record;
// ciphertext+nonce keyed by audit_id; the KEY lives outside the DB (the 0600 key
// file). cipher_id/key_version give crypto agility (an unsupported value fails loud
// on reveal).
fn apply_v8(connection: &Connection) -> CliResult<()> {
    connection
        .execute_batch(
            "CREATE TABLE custody_vault (
                audit_id TEXT PRIMARY KEY REFERENCES audit_records(audit_id),
                ciphertext BLOB NOT NULL,
                nonce BLOB NOT NULL,
                cipher_id TEXT NOT NULL,
                key_version INTEGER NOT NULL,
                created_at TEXT NOT NULL
            );
            INSERT OR IGNORE INTO schema_migrations (version, name) VALUES (8, 'custody-vault');",
        )
        .map_err(|error| CliError::failure(format!("failed to apply schema v8: {error}")))?;
    Ok(())
}

// ADR 024 §1a / ADR 034 D8 (v1 M3a R3 P4): the AUTHORITATIVE DB-layer guard that a
// non-transport operator proof (any decision record_type, plus `reveal`) must never
// be `delivery_status=sent` — it is an observed/operator event, not a delivered
// message. Mirrors the existing `audit_records_no_sent_agent_insert` (v2) trigger and
// the CLI pre-check in `append_audit_record`/`validate_audit_args`, but as its own v9
// migration so existing v8 DBs gain it on upgrade and covers EVERY producer that
// bypasses the CLI (`db audit append`, `db import`, any direct insert). The hardcoded
// 6-type list = `crate::decision::DECISION_RECORD_TYPES` (the 5 decision types) +
// `reveal`; it matches `is_non_transport_proof_record_type`.
fn apply_v9(connection: &Connection) -> CliResult<()> {
    connection
        .execute_batch(
            "CREATE TRIGGER IF NOT EXISTS audit_records_no_sent_nontransport_proof_insert
                BEFORE INSERT ON audit_records
                WHEN NEW.delivery_status = 'sent'
                 AND NEW.record_type IN ('gate-decision','conflict-resolve','mode-switch','interrupt','redirect','reveal')
            BEGIN
                SELECT RAISE(ABORT, 'a non-transport operator proof (decision/reveal) must not be delivery_status=sent (ADR 024 / ADR 034 D8)');
            END;
            INSERT OR IGNORE INTO schema_migrations (version, name) VALUES (9, 'reject-sent-nontransport-proof');",
        )
        .map_err(|error| CliError::failure(format!("failed to apply schema v9: {error}")))?;
    Ok(())
}

/// A retained `custody_vault` row (ADR 034 M3): the encrypted payload + its nonce and
/// the crypto identity. The key lives OUTSIDE the DB (the 0600 key file); reveal
/// (M3a T5) decrypts `ciphertext` under that key and re-verifies the plaintext hash.
/// `created_at` is read on demand from the row, not carried here.
#[derive(Debug)]
pub(crate) struct CustodyRow {
    pub ciphertext: Vec<u8>,
    pub nonce: Vec<u8>,
    pub cipher_id: String,
    pub key_version: i64,
}

/// Open the projection DB (auto-create + migrate, like `project_audit`) and insert
/// the retained ciphertext into `custody_vault`. This is the `pub(crate)` entry the
/// `--retain-custody` audited write (M3a T4) calls — it keeps `open_projection_database`
/// private while reusing the same opener every projection uses. ANY error propagates
/// so the caller can fail LOUD (ADR 034 D7: a requested retention never silently
/// degrades to no-retain). Never logs key material or plaintext.
pub(crate) fn project_custody_vault(
    db_path: &Path,
    audit_id: &str,
    ciphertext: &[u8],
    nonce: &[u8],
    cipher_id: &str,
    key_version: i64,
    created_at: &str,
) -> CliResult<()> {
    let connection = open_projection_database(db_path)?;
    insert_custody_vault(
        &connection,
        audit_id,
        ciphertext,
        nonce,
        cipher_id,
        key_version,
        created_at,
    )
}

/// Insert a retained ciphertext into `custody_vault`, keyed by `audit_id` (the PK,
/// which FKs `audit_records`). `INSERT OR IGNORE` makes it idempotent: a re-run for
/// an already-retained record is a silent no-op (never an error, never a duplicate),
/// so a retried `--retain-custody` write can't conflict. BLOBs bind as `&[u8]`.
/// Consumed by the `--retain-custody` audited write in M3a T4.
pub(crate) fn insert_custody_vault(
    conn: &Connection,
    audit_id: &str,
    ciphertext: &[u8],
    nonce: &[u8],
    cipher_id: &str,
    key_version: i64,
    created_at: &str,
) -> CliResult<()> {
    conn.execute(
        "INSERT OR IGNORE INTO custody_vault
            (audit_id, ciphertext, nonce, cipher_id, key_version, created_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
        params![
            audit_id,
            ciphertext,
            nonce,
            cipher_id,
            key_version,
            created_at
        ],
    )
    .map_err(|error| CliError::failure(format!("failed to insert custody_vault row: {error}")))?;
    Ok(())
}

/// Read a retained `custody_vault` row by `audit_id`. `None` when the record was
/// never retained (not revealable). BLOB columns read back as `Vec<u8>`. Consumed by
/// `zynk reveal` (M3a T5) to fetch the ciphertext+nonce+crypto identity.
pub(crate) fn read_custody_vault(
    conn: &Connection,
    audit_id: &str,
) -> CliResult<Option<CustodyRow>> {
    conn.query_row(
        "SELECT ciphertext, nonce, cipher_id, key_version
         FROM custody_vault WHERE audit_id = ?1",
        params![audit_id],
        |row| {
            Ok(CustodyRow {
                ciphertext: row.get(0)?,
                nonce: row.get(1)?,
                cipher_id: row.get(2)?,
                key_version: row.get(3)?,
            })
        },
    )
    .optional()
    .map_err(|error| CliError::failure(format!("failed to read custody_vault row: {error}")))
}

/// Read BOTH the `session_id` and `payload_hash` of an audit record by `audit_id`,
/// `None` when no such record. Consumed by `zynk reveal` (M3a T5): the reveal proof
/// lands in the SAME session as the revealed record (chained/visible there), and the
/// `payload_hash` is the disclosure contract the recomputed plaintext hash is checked
/// against. One read so the two values can never diverge.
pub(crate) fn read_reveal_target(
    conn: &Connection,
    audit_id: &str,
) -> CliResult<Option<(String, String)>> {
    conn.query_row(
        "SELECT session_id, payload_hash FROM audit_records WHERE audit_id = ?1",
        params![audit_id],
        |row| Ok((row.get(0)?, row.get(1)?)),
    )
    .optional()
    .map_err(|error| CliError::failure(format!("failed to read audit record for reveal: {error}")))
}

fn backfill_status_content_hash(connection: &Connection) -> CliResult<()> {
    let mut select = connection
        .prepare(
            "SELECT status_event_id, phase, mode, workflow_status,
                    completed_since_last_update, in_progress, next_action, blockers,
                    asks_for_zevs, risk_or_residual_uncertainty, expected_wait, event_text
             FROM status_events
             WHERE content_hash = ''",
        )
        .map_err(|error| {
            CliError::failure(format!(
                "failed to read status_events for backfill: {error}"
            ))
        })?;
    let rows = select
        .query_map([], |row| {
            let id: i64 = row.get(0)?;
            let fields = (1..=11)
                .map(|i| row.get::<_, String>(i))
                .collect::<Result<Vec<_>, _>>()?;
            Ok((id, fields))
        })
        .map_err(|error| {
            CliError::failure(format!(
                "failed to read status_events for backfill: {error}"
            ))
        })?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| {
            CliError::failure(format!(
                "failed to read status_events for backfill: {error}"
            ))
        })?;
    for (id, fields) in rows {
        let refs: Vec<&str> = fields.iter().map(String::as_str).collect();
        let hash = status_content_hash(&refs);
        connection
            .execute(
                "UPDATE status_events SET content_hash = ?1 WHERE status_event_id = ?2",
                params![hash, id],
            )
            .map_err(|error| {
                CliError::failure(format!("failed to backfill content_hash: {error}"))
            })?;
    }
    Ok(())
}

/// Canonical content hash for status-event dedup (ADR 026 D3). Order-fixed over
/// the 11 content fields (NOT session_id/timestamp). Used by both the v4
/// backfill and the import dedup so they agree.
fn status_content_hash(fields: &[&str]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(fields.join("\n").as_bytes());
    format!("sha256:{:x}", hasher.finalize())
}

#[derive(Default)]
struct ImportSummary {
    sessions_seen: usize,
    warning_count: usize,
}

#[derive(Debug)]
pub(crate) struct ImportedStatus {
    pub(crate) session_id: String,
    pub(crate) last_update: String,
    pub(crate) lead_agent: String,
    pub(crate) phase: String,
    pub(crate) mode: String,
    pub(crate) artifact_ref: String,
    pub(crate) workflow_status: String,
    pub(crate) completed_since_last_update: String,
    pub(crate) in_progress: String,
    pub(crate) next_action: String,
    pub(crate) blockers: String,
    pub(crate) asks_for_zevs: String,
    pub(crate) risk_or_residual_uncertainty: String,
    pub(crate) expected_wait: String,
    pub(crate) rolling_events: Vec<String>,
}

#[derive(Debug, Clone)]
pub(crate) struct ImportedAuditRecord {
    pub(crate) audit_id: String,
    pub(crate) previous_audit_id: Option<String>,
    pub(crate) timestamp: String,
    pub(crate) source_agent_id: Option<String>,
    pub(crate) source_address: String,
    pub(crate) target_agent_id: Option<String>,
    pub(crate) target_address: String,
    pub(crate) transport: String,
    pub(crate) workspace_id: String,
    pub(crate) session_id: String,
    pub(crate) mid: String,
    pub(crate) record_type: String,
    pub(crate) command_origin: String,
    pub(crate) mode: Option<String>,
    pub(crate) r#ref: Option<String>,
    pub(crate) re: Option<String>,
    pub(crate) payload_hash: String,
    pub(crate) payload_redaction_policy: String,
    pub(crate) content_size: i64,
    pub(crate) delivery_status: String,
    pub(crate) observed_by: String,
    pub(crate) verified_by: String,
    pub(crate) due: Option<String>,
    /// ADR 029 C6: the readable corpus excerpt (None for hash-only), carried into
    /// `messages.payload_excerpt` so the file-first path populates corpus content,
    /// not just the file artifact.
    pub(crate) payload_excerpt: Option<String>,
}

/// v1 M1 R1 P2 (ADR 027 file-as-authority recovery): one parsed `work-event` block
/// from a session's `work.md`, ready to project back into `work_events` via the
/// SAME `project_work_event` the live path uses. The payload is the typed enum
/// (round-tripped through `WorkEventPayload::from_storage`), never a raw blob.
#[derive(Debug, Clone)]
pub(crate) struct ImportedWorkEvent {
    pub(crate) actor: String,
    pub(crate) timestamp: String,
    /// v1 M1 R2 P4: the file-rendered per-session ordinal (`seq=` header), part of the
    /// event identity so two same-second identical events stay distinct rows and a
    /// re-import is a stable-hash no-op.
    pub(crate) seq: usize,
    pub(crate) payload: crate::work_event::WorkEventPayload,
}

enum ImportedAuditInsert {
    Inserted,
    Existing,
    RebasedToRoot { original_error: String },
}

#[derive(Default)]
struct ExportSummary {
    written: usize,
    unchanged: usize,
}

#[derive(Debug, Clone)]
struct ExportedAuditRecord {
    audit_id: String,
    previous_audit_id: Option<String>,
    timestamp: String,
    source_agent: String,
    source_address: String,
    target_agent: String,
    target_address: String,
    transport: String,
    workspace_id: String,
    session_id: String,
    mid: String,
    record_type: String,
    command_origin: String,
    mode: Option<String>,
    r#ref: Option<String>,
    re: Option<String>,
    payload_hash: String,
    payload_redaction_policy: String,
    content_size: i64,
    delivery_status: String,
    observed_by: String,
    verified_by: String,
    due: Option<String>,
}

fn export_outputs(path: &Path, args: &DbExportOutputsArgs) -> CliResult<ExportSummary> {
    let connection = open_read_database(path)?;
    let status = load_export_status(&connection, &args.session_id)?;
    let events = load_export_status_events(&connection, &args.session_id)?;
    let audit_records = load_export_audit_records(&connection, &args.session_id)?;
    let session_dir = args.root.join("sessions").join(&args.session_id);
    let exports = [
        (
            session_dir.join("status.md"),
            render_export_status(&status, &events),
        ),
        (
            session_dir.join("audit.md"),
            render_export_audit(&args.session_id, &audit_records),
        ),
    ];
    let mut summary = ExportSummary::default();
    for (path, content) in exports {
        if write_if_changed(&path, &content)? {
            summary.written += 1;
        } else {
            summary.unchanged += 1;
        }
    }
    Ok(summary)
}

fn load_export_status(connection: &Connection, session_id: &str) -> CliResult<ImportedStatus> {
    connection
        .query_row(
            "SELECT s.session_id,
                    COALESCE(se.timestamp, s.updated_at),
                    COALESCE(s.lead_agent_id, 'unknown'),
                    COALESCE(se.phase, s.phase),
                    -- ADR 033 M2b T3 (C3=b): the EXPORT intentionally stays status-only
                    -- (NO latest-writer across operator decisions, unlike the live/snapshot
                    -- dashboard reads). `status.md` is the agent-authored status artifact;
                    -- folding a decision-derived mode in here would fake the operator's
                    -- mode-switch into the agent status timeline — the exact provenance
                    -- conflation C3=b's rationale rejected. The operator decision is
                    -- exported separately via the audit chain, so the export's mode is by
                    -- design the status-authored mode.
                    COALESCE(se.mode, s.mode),
                    COALESCE(s.artifact_ref, 'unknown'),
                    COALESCE(se.workflow_status, s.workflow_status),
                    COALESCE(se.completed_since_last_update, 'unknown'),
                    COALESCE(se.in_progress, 'unknown'),
                    COALESCE(se.next_action, 'unknown'),
                    COALESCE(se.blockers, 'unknown'),
                    COALESCE(se.asks_for_zevs, 'unknown'),
                    COALESCE(se.risk_or_residual_uncertainty, 'unknown'),
                    COALESCE(se.expected_wait, 'unknown')
             FROM sessions AS s
             LEFT JOIN status_events AS se
               ON se.status_event_id = (
                    SELECT status_event_id
                    FROM status_events
                    WHERE session_id = s.session_id
                    ORDER BY timestamp DESC, status_event_id DESC
                    LIMIT 1
               )
             WHERE s.session_id = ?1",
            [session_id],
            |row| {
                Ok(ImportedStatus {
                    session_id: row.get(0)?,
                    last_update: row.get(1)?,
                    lead_agent: row.get(2)?,
                    phase: row.get(3)?,
                    mode: row.get(4)?,
                    artifact_ref: row.get(5)?,
                    workflow_status: row.get(6)?,
                    completed_since_last_update: row.get(7)?,
                    in_progress: row.get(8)?,
                    next_action: row.get(9)?,
                    blockers: row.get(10)?,
                    asks_for_zevs: row.get(11)?,
                    risk_or_residual_uncertainty: row.get(12)?,
                    expected_wait: row.get(13)?,
                    rolling_events: Vec::new(),
                })
            },
        )
        .optional()
        .map_err(|error| CliError::failure(format!("failed to load session for export: {error}")))?
        .ok_or_else(|| CliError::failure(format!("session not found: {session_id}")))
}

fn load_export_status_events(connection: &Connection, session_id: &str) -> CliResult<Vec<String>> {
    let mut statement = connection
        .prepare(
            "SELECT timestamp, workflow_status, next_action, event_text
             FROM status_events
             WHERE session_id = ?1
             ORDER BY timestamp DESC, status_event_id DESC
             LIMIT 10",
        )
        .map_err(|error| CliError::failure(format!("failed to load status events: {error}")))?;
    let events = statement
        .query_map([session_id], |row| {
            let timestamp: String = row.get(0)?;
            let workflow_status: String = row.get(1)?;
            let next_action: String = row.get(2)?;
            let event_text: String = row.get(3)?;
            let events = if event_text.trim().is_empty() {
                vec![format!(
                    "{timestamp} - status={workflow_status}; next={next_action}"
                )]
            } else {
                event_text
                    .lines()
                    .map(str::trim)
                    .filter(|line| !line.is_empty())
                    .map(str::to_string)
                    .collect()
            };
            Ok(events)
        })
        .map_err(|error| CliError::failure(format!("failed to load status events: {error}")))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| CliError::failure(format!("failed to read status events: {error}")))?;
    let mut events = events.into_iter().flatten().collect::<Vec<_>>();
    events.truncate(10);
    Ok(events)
}

fn load_export_audit_records(
    connection: &Connection,
    session_id: &str,
) -> CliResult<Vec<ExportedAuditRecord>> {
    let mut statement = connection
        .prepare(
            "SELECT audit_id, previous_audit_id, timestamp,
                    COALESCE(source_agent_id, 'unknown'), source_address,
                    COALESCE(target_agent_id, 'unknown'), target_address,
                    transport, workspace_id, session_id, mid, record_type,
                    command_origin, mode, ref, re, payload_hash,
                    payload_redaction_policy, content_size, delivery_status,
                    observed_by, verified_by, due
             FROM audit_records
             WHERE session_id = ?1
             ORDER BY timestamp, audit_id",
        )
        .map_err(|error| CliError::failure(format!("failed to load audit records: {error}")))?;
    let records = statement
        .query_map([session_id], |row| {
            Ok(ExportedAuditRecord {
                audit_id: row.get(0)?,
                previous_audit_id: row.get(1)?,
                timestamp: row.get(2)?,
                source_agent: row.get(3)?,
                source_address: row.get(4)?,
                target_agent: row.get(5)?,
                target_address: row.get(6)?,
                transport: row.get(7)?,
                workspace_id: row.get(8)?,
                session_id: row.get(9)?,
                mid: row.get(10)?,
                record_type: row.get(11)?,
                command_origin: row.get(12)?,
                mode: row.get(13)?,
                r#ref: row.get(14)?,
                re: row.get(15)?,
                payload_hash: row.get(16)?,
                payload_redaction_policy: row.get(17)?,
                content_size: row.get(18)?,
                delivery_status: row.get(19)?,
                observed_by: row.get(20)?,
                verified_by: row.get(21)?,
                due: row.get(22)?,
            })
        })
        .map_err(|error| CliError::failure(format!("failed to load audit records: {error}")))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| CliError::failure(format!("failed to read audit records: {error}")))?;
    Ok(order_export_audit_records(records))
}

fn order_export_audit_records(records: Vec<ExportedAuditRecord>) -> Vec<ExportedAuditRecord> {
    let mut roots = Vec::new();
    let mut children: HashMap<String, Vec<ExportedAuditRecord>> = HashMap::new();
    for record in records {
        if let Some(previous_audit_id) = &record.previous_audit_id {
            children
                .entry(previous_audit_id.clone())
                .or_default()
                .push(record);
        } else {
            roots.push(record);
        }
    }
    sort_audit_records(&mut roots);

    let mut ordered = Vec::new();
    let mut leftover = Vec::new();
    for root in roots {
        let mut current = Some(root);
        while let Some(record) = current {
            let audit_id = record.audit_id.clone();
            ordered.push(record);
            current = children.remove(&audit_id).and_then(|mut child_records| {
                sort_audit_records(&mut child_records);
                let mut child_records = child_records.into_iter();
                let next = child_records.next();
                leftover.extend(child_records);
                next
            });
        }
    }
    leftover.extend(children.into_values().flatten());
    sort_audit_records(&mut leftover);
    ordered.extend(leftover);
    ordered
}

fn sort_audit_records(records: &mut [ExportedAuditRecord]) {
    records.sort_by(|left, right| {
        left.timestamp
            .cmp(&right.timestamp)
            .then_with(|| left.audit_id.cmp(&right.audit_id))
    });
}

fn render_export_status(status: &ImportedStatus, events: &[String]) -> String {
    let event_lines = if events.is_empty() {
        "No events recorded.".to_string()
    } else {
        events
            .iter()
            .enumerate()
            .map(|(index, event)| format!("{}. {event}", index + 1))
            .collect::<Vec<_>>()
            .join("\n")
    };

    format!(
        "# Session Status: {session_id}\n\n\
session_id: {session_id}\n\
last_update: {timestamp}\n\
lead_agent: {lead_agent}\n\
status: {status}\n\n\
## Current State\n\n\
- phase: {phase}\n\
- mode: {mode}\n\
- artifact_ref: {artifact_ref}\n\
- completed_since_last_update: {completed}\n\
- in_progress: {in_progress}\n\
- next_action: {next_action}\n\
- blockers: {blockers}\n\
- asks_for_Zevs: {asks_for_zevs}\n\
- risk_or_residual_uncertainty: {risk}\n\
- expected_wait: {expected_wait}\n\n\
## Rolling Events\n\n\
{event_lines}\n",
        session_id = status.session_id,
        timestamp = status.last_update,
        lead_agent = status.lead_agent,
        status = status.workflow_status,
        phase = status.phase,
        mode = status.mode,
        artifact_ref = status.artifact_ref,
        completed = status.completed_since_last_update,
        in_progress = status.in_progress,
        next_action = status.next_action,
        blockers = status.blockers,
        asks_for_zevs = status.asks_for_zevs,
        risk = status.risk_or_residual_uncertainty,
        expected_wait = status.expected_wait,
    )
}

fn render_export_audit(session_id: &str, records: &[ExportedAuditRecord]) -> String {
    let record_text = records
        .iter()
        .map(render_export_audit_record)
        .collect::<Vec<_>>()
        .join("\n");
    format!(
        "# Audit Trail: {session_id}\n\nsession_id: {session_id}\n\n## Records\n\n{record_text}\n"
    )
}

fn render_export_audit_record(record: &ExportedAuditRecord) -> String {
    let mut fields = vec![
        ("audit_id", record.audit_id.clone()),
        (
            "previous_audit_id",
            record
                .previous_audit_id
                .clone()
                .unwrap_or_else(|| "none".to_string()),
        ),
        ("timestamp", record.timestamp.clone()),
        ("source_agent", record.source_agent.clone()),
        ("source_address", record.source_address.clone()),
        ("target_agent", record.target_agent.clone()),
        ("target_address", record.target_address.clone()),
        ("transport", record.transport.clone()),
        ("workspace_id", record.workspace_id.clone()),
        ("session_id", record.session_id.clone()),
        ("mid", record.mid.clone()),
        ("type", record.record_type.clone()),
        ("command_origin", record.command_origin.clone()),
        ("payload_hash", record.payload_hash.clone()),
        (
            "payload_redaction_policy",
            record.payload_redaction_policy.clone(),
        ),
        ("content_size", record.content_size.to_string()),
        ("delivery_status", record.delivery_status.clone()),
        ("observed_by", record.observed_by.clone()),
        ("verified_by", record.verified_by.clone()),
    ];
    for (key, value) in [
        ("mode", record.mode.as_ref()),
        ("ref", record.r#ref.as_ref()),
        ("re", record.re.as_ref()),
        ("due", record.due.as_ref()),
    ] {
        if let Some(value) = value {
            fields.push((key, value.clone()));
        }
    }
    let fields = fields
        .into_iter()
        .map(|(key, value)| format!("{key}={value}"))
        .collect::<Vec<_>>()
        .join("\n");
    format!("```text\n{fields}\n```")
}

fn write_if_changed(path: &Path, content: &str) -> CliResult<bool> {
    if path.exists() {
        let existing = fs::read_to_string(path).map_err(|error| {
            CliError::failure(format!("failed to read {}: {error}", path.display()))
        })?;
        if existing == content {
            return Ok(false);
        }
    }
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(|error| {
            CliError::failure(format!("failed to create {}: {error}", parent.display()))
        })?;
    }
    fs::write(path, content).map_err(|error| {
        CliError::failure(format!("failed to write {}: {error}", path.display()))
    })?;
    Ok(true)
}

/// v0.2.2: opt-in artifact-follow import used by `db serve --auto-import`.
/// Reuses the idempotent import path (append-only audit, content-hash status
/// dedup) so re-importing before each render cannot duplicate. This is a
/// convenience for live dashboards; it does NOT change ADR 025's default
/// (serve stays DB-read-only unless `--auto-import` is set).
pub(crate) fn import_outputs_root(db_path: &Path, root: &Path) -> CliResult<()> {
    import_outputs(
        db_path,
        &DbImportOutputsArgs {
            root: root.to_path_buf(),
            session_id: None,
            timestamp: None,
        },
    )
    .map(|_summary| ())
}

fn import_outputs(path: &Path, args: &DbImportOutputsArgs) -> CliResult<ImportSummary> {
    let sessions_dir = args.root.join("sessions");
    if !sessions_dir.exists() {
        return Ok(ImportSummary::default());
    }

    let timestamp = match &args.timestamp {
        Some(value) => normalize_arg_timestamp(value)?,
        None => crate::timestamp::now_utc_seconds(),
    };
    let mut connection = open_database(path)?;
    let mut summary = ImportSummary::default();

    for entry in fs::read_dir(&sessions_dir).map_err(|error| {
        CliError::failure(format!(
            "failed to read {}: {error}",
            sessions_dir.display()
        ))
    })? {
        let entry = entry
            .map_err(|error| CliError::failure(format!("failed to read session entry: {error}")))?;
        let session_dir = entry.path();
        if !session_dir.is_dir() {
            continue;
        }
        let Some(session_id) = session_dir
            .file_name()
            .and_then(|name| name.to_str())
            .map(str::to_string)
        else {
            continue;
        };
        if args
            .session_id
            .as_ref()
            .is_some_and(|requested| requested != &session_id)
        {
            continue;
        }

        summary.sessions_seen += 1;
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(|error| {
                CliError::failure(format!(
                    "failed to start SQLite import transaction: {error}"
                ))
            })?;
        let warnings = import_outputs_session(&transaction, &args.root, &session_dir, &timestamp)?;
        summary.warning_count += warnings.len();
        insert_import_row(
            &transaction,
            &session_dir,
            &source_hash_for_session_dir(&session_dir)?,
            &timestamp,
            &warnings,
        )?;
        transaction.commit().map_err(|error| {
            CliError::failure(format!(
                "failed to commit outputs import transaction: {error}"
            ))
        })?;
    }

    Ok(summary)
}

fn import_outputs_session(
    transaction: &Transaction<'_>,
    root: &Path,
    session_dir: &Path,
    import_timestamp: &str,
) -> CliResult<Vec<String>> {
    let mut warnings = Vec::new();
    let fallback_session_id = session_dir
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("unknown")
        .to_string();
    let status_path = session_dir.join("status.md");
    let mut status = if status_path.exists() {
        Some(parse_status_artifact(&status_path, &fallback_session_id)?)
    } else {
        None
    };
    // V1-C4: validate the status timestamp with real RFC3339 parsing and fail
    // cleanly on garbage rather than letting it reach the DB / a shape-only check.
    if let Some(status) = status.as_mut() {
        status.last_update =
            crate::timestamp::canonicalize(&status.last_update).ok_or_else(|| {
                CliError::failure(format!(
                    "invalid status timestamp in {}: {}",
                    status_path.display(),
                    status.last_update
                ))
            })?;
    }
    let session_id = status
        .as_ref()
        .map(|status| status.session_id.clone())
        .unwrap_or(fallback_session_id);
    let project_id = import_project_id(root);

    ensure_project(transaction, &project_id, root, import_timestamp)?;
    if let Some(status) = &status {
        if status.lead_agent != "unknown" {
            ensure_agent(transaction, &status.lead_agent, import_timestamp)?;
        }
        let inserted = transaction
            .execute(
                "INSERT OR IGNORE INTO sessions (
                    session_id, project_id, title, phase, mode, workflow_status,
                    lead_agent_id, artifact_ref, created_at, updated_at
                 )
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)",
                params![
                    status.session_id,
                    project_id,
                    status.session_id,
                    status.phase,
                    status.mode,
                    status.workflow_status,
                    nullable_unknown(&status.lead_agent),
                    status.artifact_ref,
                    status.last_update,
                ],
            )
            .map_err(|error| CliError::failure(format!("failed to import session: {error}")))?;
        if inserted == 0 {
            warnings.push(format!("skipped existing session {}", status.session_id));
        }
        insert_status_event_if_missing(transaction, status)?;
    } else {
        transaction
            .execute(
                "INSERT OR IGNORE INTO sessions (
                    session_id, project_id, title, phase, mode, workflow_status,
                    created_at, updated_at
                 )
                 VALUES (?1, ?2, ?1, 'import', 'validate', 'idle', ?3, ?3)",
                params![session_id, project_id, import_timestamp],
            )
            .map_err(|error| CliError::failure(format!("failed to import session: {error}")))?;
    }

    // v1 M2a R2 P4 (Codex): accepted decision audit records, collected during the audit
    // loop but rebuilt into the typed `operator_decisions` table AFTER the work.md import
    // below — so a fully-file-first gate/conflict (work_event + decision both `--no-db`)
    // finds its referenced `work_events` row already present and the v7 FK is satisfied.
    // Rebuilding inline (R1 P1) ran before work events existed and FK-aborted the whole
    // session import. Still the SAME import transaction, just ordered after work events.
    let mut decision_records: Vec<ImportedAuditRecord> = Vec::new();

    let audit_path = session_dir.join("audit.md");
    if audit_path.exists() {
        let mut new_messages = HashSet::new();
        for mut record in parse_audit_artifact(&audit_path, &session_id)? {
            // ADR 026 D2: fail loud rather than fabricate a dangerous `agent`
            // default for a missing verifier.
            if record.verified_by.is_empty() {
                warnings.push(format!(
                    "skipped audit_id {}: missing verified_by",
                    record.audit_id
                ));
                continue;
            }
            // V1-C4: validate the timestamp with real RFC3339 parsing BEFORE any
            // side-effecting insert (ensure_agent), so a bad value cannot pollute
            // agents or slip through the shape-only trigger.
            match crate::timestamp::canonicalize(&record.timestamp) {
                Some(canonical) => record.timestamp = canonical,
                None => {
                    warnings.push(format!(
                        "skipped audit_id {}: invalid timestamp {}",
                        record.audit_id, record.timestamp
                    ));
                    continue;
                }
            }
            // ADR 027 C4/R1 P2: due validation is fail-loud. Canonicalize when
            // present; on a malformed value SKIP the record with a warning rather
            // than silently losing the deadline by nulling due.
            if let Some(due) = record.due.clone() {
                match crate::timestamp::canonicalize(&due) {
                    Some(canonical) => record.due = Some(canonical),
                    None => {
                        warnings.push(format!(
                            "skipped audit_id {}: invalid due {due}",
                            record.audit_id
                        ));
                        continue;
                    }
                }
            }
            // V2-C1: create agents only for records the DB actually accepts.
            // ensure_agent used to run before the insert, so a record rejected by
            // the ADR 024 trigger (sent+agent) still polluted `agents`. Doing it
            // inside the accept arms means no side effects precede acceptance.
            // A decision audit row was ACCEPTED in any of the three accept arms below;
            // collect it for the deferred operator_decisions rebuild (after work.md).
            let is_decision =
                crate::decision::DECISION_RECORD_TYPES.contains(&record.record_type.as_str());
            match insert_imported_audit_record(transaction, &record) {
                Ok(ImportedAuditInsert::Inserted) => {
                    ensure_record_agents(transaction, &record)?;
                    insert_or_update_imported_message(transaction, &record, &mut new_messages)?;
                    if is_decision {
                        decision_records.push(record);
                    }
                }
                Ok(ImportedAuditInsert::Existing) => {
                    warnings.push(format!("skipped existing audit_id {}", record.audit_id));
                    // The audit row already exists, but a prior file-first import (or a
                    // soft-degrade) may have left the typed operator_decisions row absent.
                    // Still rebuild (deferred) — INSERT OR IGNORE keeps it a safe no-op
                    // when the typed row is already present.
                    if is_decision {
                        decision_records.push(record);
                    }
                }
                Ok(ImportedAuditInsert::RebasedToRoot { original_error }) => {
                    warnings.push(format!(
                        "rebased audit_id {} to chain root after insert failed: {}",
                        record.audit_id, original_error
                    ));
                    ensure_record_agents(transaction, &record)?;
                    insert_or_update_imported_message(transaction, &record, &mut new_messages)?;
                    if is_decision {
                        decision_records.push(record);
                    }
                }
                Err(error) => {
                    warnings.push(format!("skipped audit_id {}: {error}", record.audit_id))
                }
            }
        }
    }

    // v1 M1 R1 P2: reconcile file-first work events from `work.md` back into
    // `work_events` (the `--no-db`/soft-degrade recovery gap ADR 027 promises). For
    // each parsed event: canonicalize the timestamp (warn+skip on None, exactly like
    // the audit import above), `validate()` (warn+skip on error), then project via
    // the SAME `project_work_event` the live path uses (INSERT OR IGNORE on the
    // event-identity content_hash → append-only + idempotent re-import). An
    // `Artifact` payload ALSO dual-projects one `artifacts` row per file, mirroring
    // exactly what `project_report` does — but inside this import transaction.
    let work_path = session_dir.join("work.md");
    if work_path.exists() {
        let (work_events, parse_warnings) = parse_work_artifact(&work_path, &session_id)?;
        warnings.extend(parse_warnings);
        for event in work_events {
            let canonical_ts = match crate::timestamp::canonicalize(&event.timestamp) {
                Some(value) => value,
                None => {
                    warnings.push(format!(
                        "skipped work event: invalid timestamp {}",
                        event.timestamp
                    ));
                    continue;
                }
            };
            if let Err(error) = event.payload.validate() {
                warnings.push(format!(
                    "skipped work event at {canonical_ts}: {}",
                    error.message
                ));
                continue;
            }
            project_work_event(
                transaction,
                &session_id,
                &event.actor,
                &canonical_ts,
                event.seq,
                &event.payload,
            )?;
            if let crate::work_event::WorkEventPayload::Artifact { files } = &event.payload {
                for file in files {
                    upsert_artifact(transaction, &session_id, &file.path, &canonical_ts)?;
                }
            }
        }
    }

    // v1 M2a R2 P4: rebuild the typed operator_decisions rows NOW — work_events from
    // work.md are in this transaction, so a gate/conflict ref resolves and the v7 FK is
    // satisfied. Each rebuild validates its ref against the in-transaction work_events
    // (warn+skip on missing/wrong-kind, no rollback); INSERT OR IGNORE keeps it
    // idempotent. Still atomic with the audit/work rows (same `transaction`).
    for record in &decision_records {
        if let Some(warning) = rebuild_imported_operator_decision(transaction, &session_id, record)?
        {
            warnings.push(warning);
        }
    }

    Ok(warnings)
}

fn parse_status_artifact(path: &Path, fallback_session_id: &str) -> CliResult<ImportedStatus> {
    let content = fs::read_to_string(path).map_err(|error| {
        CliError::failure(format!("failed to read {}: {error}", path.display()))
    })?;
    let values = parse_markdown_key_values(&content);
    Ok(ImportedStatus {
        session_id: value_or(&values, "session_id", fallback_session_id),
        last_update: value_or(&values, "last_update", "1970-01-01T00:00:00Z"),
        lead_agent: value_or(&values, "lead_agent", "unknown"),
        phase: value_or(&values, "phase", "import"),
        mode: value_or(&values, "mode", "validate"),
        artifact_ref: value_or(&values, "artifact_ref", "unknown"),
        workflow_status: value_or(&values, "status", "idle"),
        completed_since_last_update: value_or(&values, "completed_since_last_update", "unknown"),
        in_progress: value_or(&values, "in_progress", "unknown"),
        next_action: value_or(&values, "next_action", "unknown"),
        blockers: value_or(&values, "blockers", "unknown"),
        asks_for_zevs: value_or(&values, "asks_for_Zevs", "unknown"),
        risk_or_residual_uncertainty: value_or(&values, "risk_or_residual_uncertainty", "unknown"),
        expected_wait: value_or(&values, "expected_wait", "unknown"),
        rolling_events: parse_rolling_events(&content),
    })
}

fn parse_audit_artifact(
    path: &Path,
    fallback_session_id: &str,
) -> CliResult<Vec<ImportedAuditRecord>> {
    let content = fs::read_to_string(path).map_err(|error| {
        CliError::failure(format!("failed to read {}: {error}", path.display()))
    })?;
    let mut records = Vec::new();
    for values in parse_fenced_key_value_blocks(&content) {
        let audit_id = value_or(&values, "audit_id", "");
        if audit_id.is_empty() {
            continue;
        }
        records.push(ImportedAuditRecord {
            audit_id,
            previous_audit_id: optional_audit_id(&value_or(&values, "previous_audit_id", "none")),
            timestamp: value_or(&values, "timestamp", "1970-01-01T00:00:00Z"),
            source_agent_id: nullable_unknown(&value_or(&values, "source_agent", "unknown")),
            source_address: value_or(&values, "source_address", "unknown"),
            target_agent_id: nullable_unknown(&value_or(&values, "target_agent", "unknown")),
            target_address: value_or(&values, "target_address", "unknown"),
            transport: value_or(&values, "transport", "unknown"),
            workspace_id: value_or(&values, "workspace_id", "unknown"),
            session_id: value_or(&values, "session_id", fallback_session_id),
            mid: value_or(&values, "mid", "unknown"),
            record_type: value_or(&values, "type", "unknown"),
            command_origin: value_or(&values, "command_origin", "unknown"),
            mode: optional_text(&value_or(&values, "mode", "")),
            r#ref: optional_text(&value_or(&values, "ref", "")),
            re: optional_text(&value_or(&values, "re", "")),
            payload_hash: value_or(&values, "payload_hash", "sha256:unknown"),
            payload_redaction_policy: value_or(&values, "payload_redaction_policy", "hash-only"),
            content_size: value_or(&values, "content_size", "0").parse().unwrap_or(0),
            delivery_status: value_or(&values, "delivery_status", "unknown"),
            observed_by: value_or(&values, "observed_by", "unknown"),
            verified_by: value_or(&values, "verified_by", ""),
            due: optional_text(&value_or(&values, "due", "")),
            payload_excerpt: optional_text(&value_or(&values, "payload_excerpt", "")),
        });
    }
    Ok(records)
}

/// v1 M1 R1 P2 (+ R2 P4/P5): parse a session's `work.md` into typed
/// `ImportedWorkEvent`s. The file is a sequence of ```` ```work-event ```` … ```` ```
/// ```` fences (see `report::run`); blocks are split by the shared `split_work_blocks`
/// using EXACT column-0 fences (R2 P5), so an INDENTED payload fence is kept as payload,
/// not a false close. Inside each fence the block is written POSITIONALLY by the writer:
/// line 1 `kind={k}`, line 2 `actor={a}`, line 3 `timestamp={t}`, line 4 `seq={n}`, and
/// line 5 onward is the stored typed payload (serde_norway, possibly multiple lines).
/// The split is positional — NOT a per-line content heuristic — so a hand-edited /
/// corrupted block with a MISSING or reordered header (the import path is the ADR 027
/// file-as-authority recovery path and faces edited files) is a LOUD malformed skip,
/// never silently promoting a payload line into the header map. A block with fewer than
/// five inner lines, or whose first four lines do not start with the required
/// `kind=`/`actor=`/`timestamp=`/`seq=` prefixes in order (with `seq=` parsing as a
/// non-negative integer), or whose stored payload fails `from_storage`, or whose header
/// `kind` disagrees with the payload `kind()`, is a malformed record: it is collected as
/// a warning and SKIPPED, never silently dropped or imported as garbage — mirroring the
/// audit/status import discipline. An unterminated trailing block also warns (R2 P5).
/// The caller (`import_outputs_session`) canonicalizes the timestamp and validates the
/// payload before projecting, exactly as the audit import does.
fn parse_work_artifact(
    path: &Path,
    session_id: &str,
) -> CliResult<(Vec<ImportedWorkEvent>, Vec<String>)> {
    let content = fs::read_to_string(path).map_err(|error| {
        CliError::failure(format!("failed to read {}: {error}", path.display()))
    })?;
    let mut events = Vec::new();
    let mut warnings = Vec::new();
    // v1 M1 R2 P5: split on EXACT (untrimmed) column-0 fences via the shared
    // `split_work_blocks` so an INDENTED payload fence (a serde_norway block scalar
    // whose value itself contains a ```` ``` ````) is treated as payload, not a false
    // structural close. An unterminated trailing block warns (fail-loud) instead of
    // being silently dropped.
    let blocks = split_work_blocks(&content);
    for (index, block) in blocks.blocks.iter().enumerate() {
        let block_index = index + 1; // 1-based for the warning message
        match parse_work_block(block) {
            Ok(event) => events.push(event),
            Err(reason) => warnings.push(format!(
                "skipped work event in {} (block {block_index}, session {session_id}): {reason}",
                path.display()
            )),
        }
    }
    if blocks.unterminated {
        warnings.push(format!(
            "unterminated work-event block in {} (session {session_id}): \
             opening ```work-event fence at column 0 with no matching closing fence \
             before EOF — block skipped",
            path.display()
        ));
    }
    Ok((events, warnings))
}

/// The blocks recovered from a `work.md` content string by `split_work_blocks`.
struct WorkBlocks {
    /// Each terminated `work-event` block's inner lines (verbatim, fences excluded).
    blocks: Vec<Vec<String>>,
    /// True when an opening fence had no matching close before EOF (fail-loud).
    unterminated: bool,
}

/// v1 M1 R2 P5: split a `work.md` `content` into `work-event` blocks using EXACT
/// (untrimmed) COLUMN-0 fences. `report::run` always emits the structural fences at
/// column 0 (`\n```work-event\n…\n{stored}```\n`), and a serde_norway block-scalar
/// payload's content is always INDENTED (≥1 space) relative to its key — so a payload
/// line that happens to be a Markdown fence (e.g. `  ```` ```` `) can never appear at
/// column 0 and is correctly kept as payload. Detecting the fences by `.trim()` (the
/// old behavior) wrongly promoted such an indented fence to a structural close,
/// truncating the block. A line that is EXACTLY `` ```work-event `` opens a block; a
/// line that is EXACTLY `` ``` `` closes the open block; every other line (including an
/// indented fence) is block content. An opening fence with no matching close before EOF
/// is reported via `unterminated` so the caller can warn (fail-loud), not silently drop.
/// This is the single boundary authority shared by `parse_work_artifact` (import) and
/// the producer's `seq` count (`report::run`, via `count_work_blocks`), so both use
/// identical block boundaries.
fn split_work_blocks(content: &str) -> WorkBlocks {
    let mut blocks: Vec<Vec<String>> = Vec::new();
    let mut current: Option<Vec<String>> = None; // inner block lines, verbatim
    for line in content.lines() {
        if current.is_none() {
            if line == "```work-event" {
                current = Some(Vec::new());
            }
            // Lines outside any block (prose, blank separators) are ignored.
            continue;
        }
        // Inside an open block: only a column-0 exact ``` closes it.
        if line == "```" {
            blocks.push(current.take().expect("block is open"));
            continue;
        }
        current
            .as_mut()
            .expect("block is open")
            .push(line.to_string());
    }
    let unterminated = current.is_some();
    WorkBlocks {
        blocks,
        unterminated,
    }
}

/// v1 M1 R2 P4: count the existing `work-event` blocks in a session's `work.md`,
/// reusing the SAME `split_work_blocks` boundary authority so the producer's
/// file-rendered `seq` ordinal matches what the importer will re-derive. Returns 0
/// when the file is absent or empty. An unterminated trailing block (a partial prior
/// write) is NOT counted — it will not import as an event, so the next event's `seq`
/// must not reserve an ordinal for it.
pub(crate) fn count_work_blocks(content: &str) -> usize {
    split_work_blocks(content).blocks.len()
}

/// Parse one `work-event` block's inner lines into a typed `ImportedWorkEvent` via a
/// POSITIONAL split that mirrors the writer (`report::run`): line 0 `kind={k}`, line 1
/// `actor={a}`, line 2 `timestamp={t}`, line 3 `seq={n}`, then the stored payload from
/// line 4 on (v1 M1 R2 P4). A missing/reordered header (any of the first four lines not
/// starting with its required prefix), a `seq=` that is not a non-negative integer, or
/// fewer than five lines, is MALFORMED — returns a human-readable reason so the caller
/// can warn+skip rather than silently importing a corrupted event.
fn parse_work_block(lines: &[String]) -> Result<ImportedWorkEvent, String> {
    // Positional, strict-prefix split: the writer always emits the four headers
    // first, in order, then the payload — so a deviation is a corrupted/edited file.
    let header_kind = lines
        .first()
        .and_then(|l| l.strip_prefix("kind="))
        .ok_or_else(|| "missing or misplaced kind= header (expected line 1)".to_string())?
        .trim()
        .to_string();
    let actor = lines
        .get(1)
        .and_then(|l| l.strip_prefix("actor="))
        .ok_or_else(|| "missing or misplaced actor= header (expected line 2)".to_string())?
        .trim()
        .to_string();
    if actor.is_empty() {
        return Err("empty actor= header".to_string());
    }
    let timestamp = lines
        .get(2)
        .and_then(|l| l.strip_prefix("timestamp="))
        .ok_or_else(|| "missing or misplaced timestamp= header (expected line 3)".to_string())?
        .trim()
        .to_string();
    if timestamp.is_empty() {
        return Err("empty timestamp= header".to_string());
    }
    let seq_raw = lines
        .get(3)
        .and_then(|l| l.strip_prefix("seq="))
        .ok_or_else(|| "missing or misplaced seq= header (expected line 4)".to_string())?
        .trim();
    let seq: usize = seq_raw
        .parse()
        .map_err(|_| format!("seq= header must be a non-negative integer (got {seq_raw:?})"))?;
    if lines.len() < 5 {
        return Err("missing stored payload (block has no body after headers)".to_string());
    }
    // The stored payload is everything from line 4 on, reconstructed VERBATIM with the
    // trailing newline that `to_storage` (serde_norway) emits — so writer-produced
    // blocks round-trip identically through `from_storage`.
    let mut payload_body = String::new();
    for line in &lines[4..] {
        payload_body.push_str(line);
        payload_body.push('\n');
    }
    let payload = crate::work_event::WorkEventPayload::from_storage(&payload_body)
        .map_err(|error| format!("unparseable payload: {}", error.message))?;
    // The header `kind=` must agree with the typed payload kind — the same
    // typed-not-blob discipline the read-model enforces.
    if header_kind != payload.kind() {
        return Err(format!(
            "kind header {header_kind:?} disagrees with payload kind {:?}",
            payload.kind()
        ));
    }
    Ok(ImportedWorkEvent {
        actor,
        timestamp,
        seq,
        payload,
    })
}

fn parse_markdown_key_values(content: &str) -> BTreeMap<String, String> {
    let mut values = BTreeMap::new();
    for line in content.lines() {
        let line = line.trim();
        if let Some((key, value)) = line
            .strip_prefix("- ")
            .and_then(|line| line.split_once(": "))
        {
            values.insert(key.to_string(), value.to_string());
        } else if let Some((key, value)) = line.split_once(": ") {
            values.insert(key.to_string(), value.to_string());
        }
    }
    values
}

fn parse_rolling_events(content: &str) -> Vec<String> {
    let Some((_, events_text)) = content.split_once("## Rolling Events") else {
        return Vec::new();
    };
    events_text
        .lines()
        .filter_map(|line| {
            let line = line.trim();
            let (prefix, entry) = line.split_once(". ")?;
            prefix.parse::<usize>().ok()?;
            Some(entry.to_string())
        })
        .collect()
}

/// Parse the ```` ```text ```` `key=value` blocks of an audit artifact into per-record
/// maps. v1 M2a R1 P1 (approach A): `payload_excerpt` is a TERMINAL field — the writer
/// renders it LAST (`render_record`, src/audit.rs), so for `full` redaction a multi-line
/// serde_norway payload (e.g. a typed `Decision`) lands as `payload_excerpt=<line1>` plus
/// continuation lines with NO `key=`. The pre-fix single-line `split_once('=')` dropped
/// those continuation lines, truncating the payload to its first YAML line and breaking
/// `Decision::from_storage` on import. Now: once `payload_excerpt` is seen, its first-line
/// remainder PLUS every subsequent raw line up to the block's closing fence is captured
/// VERBATIM (no per-line trim — serde_norway needs the original indentation/blank lines)
/// and joined with `\n`. While capturing the terminal payload, the closing fence is matched
/// EXACTLY on the untrimmed column-0 `` ``` `` (the writer always emits the close at column
/// 0), so an INDENTED `` ``` `` that happens to appear inside the payload is kept as payload,
/// not mistaken for the block close — the same exact-fence discipline as the M1 work.md P5
/// fix. A column-0 `` ``` `` literally INSIDE a payload remains an accepted limitation (no
/// escaping); serde_norway block scalars indent their content so it cannot occur in
/// practice. Non-payload fields keep the trimmed `key=value` parse, so blocks without a
/// `payload_excerpt` (and single-line payload_excerpt blocks) parse identically to before.
fn parse_fenced_key_value_blocks(content: &str) -> Vec<BTreeMap<String, String>> {
    const PAYLOAD_KEY: &str = "payload_excerpt";
    let mut blocks = Vec::new();
    let mut current: Option<BTreeMap<String, String>> = None;
    // Some(lines) once the terminal `payload_excerpt` field is open: subsequent raw
    // lines are appended verbatim until the exact column-0 closing fence.
    let mut payload_capture: Option<Vec<String>> = None;
    for raw_line in content.lines() {
        // While capturing the terminal payload, ONLY an exact column-0 fence closes the
        // block (untrimmed). Check this before the trimmed/key-value handling below.
        if payload_capture.is_some() {
            if raw_line == "```" {
                if let (Some(values), Some(lines)) = (current.as_mut(), payload_capture.take()) {
                    values.insert(PAYLOAD_KEY.to_string(), lines.join("\n"));
                }
                if let Some(values) = current.take() {
                    blocks.push(values);
                }
                continue;
            }
            if let Some(lines) = payload_capture.as_mut() {
                lines.push(raw_line.to_string());
            }
            continue;
        }

        let line = raw_line.trim();
        if line == "```text" {
            current = Some(BTreeMap::new());
            continue;
        }
        if line == "```" {
            if let Some(values) = current.take() {
                blocks.push(values);
            }
            continue;
        }
        if current.is_some() {
            if let Some((key, value)) = line.split_once('=') {
                if key == PAYLOAD_KEY {
                    // Open the terminal capture with this line's remainder as line 1.
                    payload_capture = Some(vec![value.to_string()]);
                } else {
                    if let Some(values) = current.as_mut() {
                        values.insert(key.to_string(), value.to_string());
                    }
                }
            }
        }
    }
    blocks
}

fn ensure_project(
    transaction: &Transaction<'_>,
    project_id: &str,
    root: &Path,
    timestamp: &str,
) -> CliResult<()> {
    transaction
        .execute(
            "INSERT OR IGNORE INTO projects (project_id, name, root_path, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?4)",
            params![
                project_id,
                root.file_name()
                    .and_then(|name| name.to_str())
                    .unwrap_or("outputs"),
                root.display().to_string(),
                timestamp,
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to import project: {error}")))?;
    Ok(())
}

fn ensure_agent(transaction: &Transaction<'_>, agent_id: &str, timestamp: &str) -> CliResult<()> {
    transaction
        .execute(
            "INSERT OR IGNORE INTO agents (
                agent_id, display_name, agent_kind, created_at, updated_at
             )
             VALUES (?1, ?1, 'imported', ?2, ?2)",
            params![agent_id, timestamp],
        )
        .map_err(|error| CliError::failure(format!("failed to import agent: {error}")))?;
    Ok(())
}

/// Ensure the source/target agents of an accepted audit record exist. Called
/// only after the record is accepted (V2-C1), so rejected records leave no
/// agent rows behind.
fn ensure_record_agents(
    transaction: &Transaction<'_>,
    record: &ImportedAuditRecord,
) -> CliResult<()> {
    if let Some(agent_id) = &record.source_agent_id {
        ensure_agent(transaction, agent_id, &record.timestamp)?;
    }
    if let Some(agent_id) = &record.target_agent_id {
        ensure_agent(transaction, agent_id, &record.timestamp)?;
    }
    Ok(())
}

fn insert_status_event_if_missing(
    transaction: &Transaction<'_>,
    status: &ImportedStatus,
) -> CliResult<()> {
    // ADR 026 D3: dedup on content, not just (session, timestamp, phase, mode,
    // status). The 11-field order here MUST match backfill_status_content_hash.
    let event_text = status.rolling_events.join("\n");
    let content_hash = status_content_hash(&[
        &status.phase,
        &status.mode,
        &status.workflow_status,
        &status.completed_since_last_update,
        &status.in_progress,
        &status.next_action,
        &status.blockers,
        &status.asks_for_zevs,
        &status.risk_or_residual_uncertainty,
        &status.expected_wait,
        &event_text,
    ]);

    let exists: Option<i64> = transaction
        .query_row(
            "SELECT status_event_id
             FROM status_events
             WHERE session_id = ?1 AND timestamp = ?2 AND content_hash = ?3
             LIMIT 1",
            params![status.session_id, status.last_update, content_hash],
            |row| row.get(0),
        )
        .optional()
        .map_err(|error| CliError::failure(format!("failed to read status events: {error}")))?;
    if exists.is_some() {
        return Ok(());
    }

    transaction
        .execute(
            "INSERT INTO status_events (
                session_id, timestamp, phase, mode, workflow_status,
                completed_since_last_update, in_progress, next_action, blockers,
                asks_for_zevs, risk_or_residual_uncertainty, expected_wait, event_text,
                content_hash
             )
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
            params![
                status.session_id,
                status.last_update,
                status.phase,
                status.mode,
                status.workflow_status,
                status.completed_since_last_update,
                status.in_progress,
                status.next_action,
                status.blockers,
                status.asks_for_zevs,
                status.risk_or_residual_uncertainty,
                status.expected_wait,
                event_text,
                content_hash,
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to import status event: {error}")))?;
    Ok(())
}

fn insert_imported_audit_record(
    transaction: &Transaction<'_>,
    record: &ImportedAuditRecord,
) -> Result<ImportedAuditInsert, String> {
    let exists: Option<String> = transaction
        .query_row(
            "SELECT audit_id FROM audit_records WHERE audit_id = ?1",
            [&record.audit_id],
            |row| row.get(0),
        )
        .optional()
        .map_err(|error| error.to_string())?;
    if exists.is_some() {
        return Ok(ImportedAuditInsert::Existing);
    }

    match execute_imported_audit_insert(transaction, record, record.previous_audit_id.as_deref()) {
        Ok(_) => Ok(ImportedAuditInsert::Inserted),
        Err(error) if record.previous_audit_id.is_some() => {
            let original_error = error.to_string();
            match execute_imported_audit_insert(transaction, record, None) {
                Ok(_) => Ok(ImportedAuditInsert::RebasedToRoot { original_error }),
                Err(fallback_error) => Err(format!(
                    "{original_error}; root fallback failed: {fallback_error}"
                )),
            }
        }
        Err(error) => Err(error.to_string()),
    }
}

fn execute_imported_audit_insert(
    transaction: &Transaction<'_>,
    record: &ImportedAuditRecord,
    previous_audit_id: Option<&str>,
) -> rusqlite::Result<usize> {
    transaction.execute(
        "INSERT INTO audit_records (
            audit_id, previous_audit_id, session_id, source_agent_id, target_agent_id,
            source_address, target_address, transport, workspace_id, mid, record_type,
            command_origin, mode, ref, re, payload_hash, payload_redaction_policy,
            content_size, delivery_status, observed_by, verified_by, timestamp, due
         )
         VALUES (
            ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11,
            ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23
         )",
        params![
            record.audit_id,
            previous_audit_id,
            record.session_id,
            record.source_agent_id,
            record.target_agent_id,
            record.source_address,
            record.target_address,
            record.transport,
            record.workspace_id,
            record.mid,
            record.record_type,
            record.command_origin,
            record.mode,
            record.r#ref,
            record.re,
            record.payload_hash,
            record.payload_redaction_policy,
            record.content_size,
            record.delivery_status,
            record.observed_by,
            record.verified_by,
            record.timestamp,
            record.due,
        ],
    )
}

fn insert_or_update_imported_message(
    transaction: &Transaction<'_>,
    record: &ImportedAuditRecord,
    new_messages: &mut HashSet<String>,
) -> CliResult<()> {
    let message_id = message_id(&record.session_id, &record.mid);
    let inserted = transaction
        .execute(
            "INSERT OR IGNORE INTO messages (
                message_id, session_id, mid, source_agent_id, target_agent_id, message_type,
                mode, ref, transport, payload_redaction_policy, payload_hash,
                latest_delivery_status, latest_verified_by, latest_audit_id, timestamp,
                payload_excerpt
             )
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)",
            params![
                message_id,
                record.session_id,
                record.mid,
                record.source_agent_id,
                record.target_agent_id,
                record.record_type,
                record.mode,
                record.r#ref,
                record.transport,
                record.payload_redaction_policy,
                record.payload_hash,
                record.delivery_status,
                record.verified_by,
                record.audit_id,
                record.timestamp,
                corpus_payload_excerpt(record),
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to import message: {error}")))?;
    if inserted == 1 {
        new_messages.insert(message_id);
        return Ok(());
    }
    if !new_messages.contains(&message_id) {
        return Ok(());
    }

    transaction
        .execute(
            "UPDATE messages
             SET source_agent_id = ?2,
                 target_agent_id = ?3,
                 message_type = ?4,
                 mode = ?5,
                 ref = ?6,
                 transport = ?7,
                 payload_redaction_policy = ?8,
                 payload_hash = ?9,
                 latest_delivery_status = ?10,
                 latest_verified_by = ?11,
                 latest_audit_id = ?12,
                 timestamp = ?13,
                 payload_excerpt = ?14
             WHERE message_id = ?1",
            params![
                message_id,
                record.source_agent_id,
                record.target_agent_id,
                record.record_type,
                record.mode,
                record.r#ref,
                record.transport,
                record.payload_redaction_policy,
                record.payload_hash,
                record.delivery_status,
                record.verified_by,
                record.audit_id,
                record.timestamp,
                corpus_payload_excerpt(record),
            ],
        )
        .map_err(|error| {
            CliError::failure(format!("failed to update imported message: {error}"))
        })?;
    Ok(())
}

// ===== ADR 027 + ADR 028: live DB projection (write path keeps the DB canonical) =====

/// Where a status/audit write should project (ADR 028).
pub(crate) enum ProjectionTarget {
    /// `--no-db`: file-only, never touch a DB.
    None,
    /// Default cwd DB: auto-create+migrate; a projection failure soft-degrades.
    Default(PathBuf),
    /// Explicit `--db <path>`: a projection failure hard-fails (file already written).
    Explicit(PathBuf),
}

impl ProjectionTarget {
    /// The path to project into and whether it was explicit (`--db`), or None for
    /// `--no-db`. Explicit targets hard-fail on projection error; default targets
    /// soft-degrade (the file is the durable record either way).
    pub(crate) fn into_path_and_mode(self) -> Option<(PathBuf, bool)> {
        match self {
            ProjectionTarget::None => None,
            ProjectionTarget::Default(path) => Some((path, false)),
            ProjectionTarget::Explicit(path) => Some((path, true)),
        }
    }

    /// Borrowing view of `into_path_and_mode` (the path + explicit flag) so a
    /// caller can consult the target without consuming it — `zynk decide` uses
    /// this for the pre-write ref-kind check, then `into_path_and_mode` for the
    /// post-write projection.
    pub(crate) fn path_and_mode(&self) -> Option<(&Path, bool)> {
        match self {
            ProjectionTarget::None => None,
            ProjectionTarget::Default(path) => Some((path.as_path(), false)),
            ProjectionTarget::Explicit(path) => Some((path.as_path(), true)),
        }
    }
}

/// Resolve the projection target from `--db`/`--no-db` (ADR 028): `--no-db` =>
/// None; `--db <p>` => Explicit(p); neither => Default(`<cwd>/.zynk/zynk.db`),
/// which is auto-created on use — no presence gate.
pub(crate) fn resolve_projection_target(db: Option<&Path>, no_db: bool) -> ProjectionTarget {
    if no_db {
        return ProjectionTarget::None;
    }
    match db {
        Some(path) => ProjectionTarget::Explicit(path.to_path_buf()),
        None => ProjectionTarget::Default(PathBuf::from(".zynk/zynk.db")),
    }
}

/// Path-keyed self-protection (ADR 028 C1/C7): if the DB lives in a `.zynk`
/// directory, ensure a `.zynk/.gitignore` of `*` so the runtime DB is never
/// accidentally committed, even in a repo that never configured zynk. Idempotent;
/// also protects a pre-existing v0.3 DB and `db init` at the default path.
fn ensure_zynk_gitignore(path: &Path) -> CliResult<()> {
    let in_zynk_dir = path
        .parent()
        .and_then(|parent| parent.file_name())
        .and_then(|name| name.to_str())
        == Some(".zynk");
    if !in_zynk_dir {
        return Ok(());
    }
    let Some(dir) = path.parent() else {
        return Ok(());
    };
    fs::create_dir_all(dir).map_err(|error| {
        CliError::failure(format!("failed to create {}: {error}", dir.display()))
    })?;
    let gitignore = dir.join(".gitignore");
    let has_star = |content: &str| content.lines().any(|line| line.trim() == "*");
    // ADR 028 R2 P4: this is FALLIBLE and the success condition is "a `*` rule is
    // actually present" — never claim self-ignore without proof, so the caller can
    // refuse to create/use the DB when protection can't be guaranteed.
    match fs::read_to_string(&gitignore) {
        Ok(content) if has_star(&content) => return Ok(()),
        Ok(content) => {
            // P1: append `*`, preserving existing content.
            let mut updated = content;
            if !updated.is_empty() && !updated.ends_with('\n') {
                updated.push('\n');
            }
            updated.push_str("*\n");
            fs::write(&gitignore, updated).map_err(|error| {
                CliError::failure(format!("failed to update {}: {error}", gitignore.display()))
            })?;
        }
        Err(_) => {
            // Missing or unreadable (e.g. `.gitignore` is itself a directory).
            fs::write(&gitignore, "*\n").map_err(|error| {
                CliError::failure(format!(
                    "failed to write {} (is it a directory?): {error}",
                    gitignore.display()
                ))
            })?;
        }
    }
    let final_content = fs::read_to_string(&gitignore).map_err(|error| {
        CliError::failure(format!("failed to verify {}: {error}", gitignore.display()))
    })?;
    if has_star(&final_content) {
        Ok(())
    } else {
        Err(CliError::failure(format!(
            "could not ensure a `*` rule in {}",
            gitignore.display()
        )))
    }
}

/// Open the projection DB, auto-creating + migrating it. Ensures the path-keyed
/// `.zynk/.gitignore` and, only on first creation of a `.zynk` DB, prints a
/// one-time notice. An explicit `--db` outside `.zynk/` is the user's path.
fn open_projection_database(path: &Path) -> CliResult<Connection> {
    let existed = path.exists();
    ensure_zynk_gitignore(path)?;
    let connection = open_database(path)?;
    let in_zynk_dir = path
        .parent()
        .and_then(|parent| parent.file_name())
        .and_then(|name| name.to_str())
        == Some(".zynk");
    if !existed && in_zynk_dir {
        eprintln!(
            "note: created {} for live state (+ .zynk/.gitignore); pass --no-db for file-only",
            path.display()
        );
    }
    Ok(connection)
}

/// Project a status write into the live DB (open + migrate + IMMEDIATE tx).
/// The status file remains the durable record; this keeps the DB current.
pub(crate) fn project_status(
    db_path: &Path,
    root: &Path,
    status: &ImportedStatus,
) -> CliResult<()> {
    let mut connection = open_projection_database(db_path)?;
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|error| {
            CliError::failure(format!("failed to start status projection: {error}"))
        })?;
    project_live_status(&transaction, root, status)?;
    transaction.commit().map_err(|error| {
        CliError::failure(format!("failed to commit status projection: {error}"))
    })?;
    Ok(())
}

/// Live status projection: a status write OWNS the session's current state, so
/// UPSERT it (unlike import's INSERT OR IGNORE) and append the status event.
fn project_live_status(
    transaction: &Transaction<'_>,
    root: &Path,
    status: &ImportedStatus,
) -> CliResult<()> {
    let project_id = import_project_id(root);
    ensure_project(transaction, &project_id, root, &status.last_update)?;
    if status.lead_agent != "unknown" {
        ensure_agent(transaction, &status.lead_agent, &status.last_update)?;
    }
    transaction
        .execute(
            "INSERT INTO sessions (
                session_id, project_id, title, phase, mode, workflow_status,
                lead_agent_id, artifact_ref, created_at, updated_at
             )
             VALUES (?1, ?2, ?1, ?3, ?4, ?5, ?6, ?7, ?8, ?8)
             ON CONFLICT(session_id) DO UPDATE SET
                phase = excluded.phase,
                mode = excluded.mode,
                workflow_status = excluded.workflow_status,
                lead_agent_id = excluded.lead_agent_id,
                artifact_ref = excluded.artifact_ref,
                updated_at = excluded.updated_at",
            params![
                status.session_id,
                project_id,
                status.phase,
                status.mode,
                status.workflow_status,
                nullable_unknown(&status.lead_agent),
                status.artifact_ref,
                status.last_update,
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to project session: {error}")))?;
    insert_status_event_if_missing(transaction, status)?;
    Ok(())
}

/// Outcome of projecting one audit record into the live DB (ADR 027 C6).
#[derive(Debug, PartialEq)]
pub(crate) enum LiveAuditProjection {
    Projected,
    AlreadyPresent,
    /// `previous_audit_id` is missing in the DB, or the insert otherwise
    /// conflicts (e.g. one-child-per-previous). The file is authoritative; we
    /// never live-rebase — `db import` reconciles the chain later.
    Gap {
        reason: String,
    },
}

/// Project an audit write into the live DB (open + migrate + IMMEDIATE tx).
pub(crate) fn project_audit(
    db_path: &Path,
    root: &Path,
    record: &ImportedAuditRecord,
) -> CliResult<LiveAuditProjection> {
    let mut connection = open_projection_database(db_path)?;
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|error| CliError::failure(format!("failed to start audit projection: {error}")))?;
    let outcome = project_live_audit_record(&transaction, root, record)?;
    transaction.commit().map_err(|error| {
        CliError::failure(format!("failed to commit audit projection: {error}"))
    })?;
    Ok(outcome)
}

/// v1 M2a R1 P1 / R2 P4: rebuild the typed `operator_decisions` row from an imported
/// decision audit, in the caller's import transaction (atomic with the audit row). The
/// ADR 027 file-first recovery path: a `decide --no-db` (or a default-projection
/// soft-degrade) leaves a durable decision audit but NO typed row; `db import outputs`
/// re-derives the audit row and must also reconstruct the typed query/effect surface —
/// the same class as the M1 work.md import gap.
///
/// R2 P4: this runs AFTER the work.md import (the caller defers it), so a fully-file-first
/// gate/conflict finds its `work_events` row already in the transaction. For a bound
/// gate/conflict the `--ref` is re-validated against the IN-TRANSACTION `work_events`
/// (`decide --no-db` had no DB to check it at write time): query the `transaction`
/// DIRECTLY — NOT `work_event_kind`, which opens a separate connection that cannot see the
/// uncommitted work rows. A missing ref or a kind mismatch (gate↔"gate", conflict↔"conflict")
/// is a warn+skip of the typed row, never an FK abort of the whole import.
///
/// Only decision record_types (`DECISION_RECORD_TYPES`) are decisions; for everything else
/// this is a no-op (`Ok(None)`). The full serde_norway `Decision` lives in the record's
/// `payload_excerpt` (the `decide` path writes `payload_redaction_policy=full`, so the
/// importer — with the P1 multi-line parser fix — carries it untruncated). On a missing or
/// undeserializable payload (e.g. a redacted/hand-edited/truncated artifact), a payload that
/// fails `validate()`, or a dangling/wrong-kind ref, this returns `Ok(Some(warning))`: the
/// caller warns and CONTINUES — it never fails the whole import and never inserts a garbage
/// row (fail-loud-consistent with the audit/work import skip discipline). A genuine DB error
/// from the typed insert propagates as `Err` so the import transaction rolls back.
/// `insert_operator_decision` uses INSERT OR IGNORE, so a re-import (or
/// accept-then-already-present) stays a clean no-op.
fn rebuild_imported_operator_decision(
    transaction: &Transaction<'_>,
    session_id: &str,
    record: &ImportedAuditRecord,
) -> CliResult<Option<String>> {
    use crate::decision::Decision;
    if !crate::decision::DECISION_RECORD_TYPES.contains(&record.record_type.as_str()) {
        return Ok(None);
    }
    let Some(payload) = record.payload_excerpt.as_deref() else {
        return Ok(Some(format!(
            "skipped operator_decisions rebuild for audit_id {}: no decision payload in artifact",
            record.audit_id
        )));
    };
    let decision = match Decision::from_storage(payload) {
        Ok(decision) => decision,
        Err(error) => {
            return Ok(Some(format!(
                "skipped operator_decisions rebuild for audit_id {}: {}",
                record.audit_id, error.message
            )));
        }
    };
    if let Err(error) = decision.validate() {
        return Ok(Some(format!(
            "skipped operator_decisions rebuild for audit_id {}: {}",
            record.audit_id, error.message
        )));
    }
    // For a bound gate/conflict, re-validate the ref against the in-transaction
    // work_events (the `decide --no-db` write path could not). Query the transaction
    // directly so it sees the work rows just imported above.
    let expected_kind = match &decision {
        Decision::Gate { .. } => Some("gate"),
        Decision::Conflict { .. } => Some("conflict"),
        Decision::Mode { .. } | Decision::Interrupt { .. } | Decision::Redirect { .. } => None,
    };
    if let (Some(expected_kind), Some(wid)) = (expected_kind, decision.target_work_event_id()) {
        let actual_kind: Option<String> = transaction
            .query_row(
                "SELECT kind FROM work_events WHERE work_event_id = ?1 AND session_id = ?2",
                params![wid, session_id],
                |row| row.get(0),
            )
            .optional()
            .map_err(|error| {
                CliError::failure(format!("failed to read work_event kind on import: {error}"))
            })?;
        match actual_kind.as_deref() {
            Some(kind) if kind == expected_kind => {}
            Some(other) => {
                return Ok(Some(format!(
                    "skipped operator_decisions rebuild for audit_id {}: --ref {wid} is a \
                     {other} work-event, not a {expected_kind}, in session",
                    record.audit_id
                )));
            }
            None => {
                return Ok(Some(format!(
                    "skipped operator_decisions rebuild for audit_id {}: --ref {wid} missing \
                     in session",
                    record.audit_id
                )));
            }
        }
    }
    insert_operator_decision(transaction, record, &decision)?;
    Ok(None)
}

/// Project an operator decision (ADR 033 D4 / M2a): the immutable audit proof row
/// AND the typed `operator_decisions` projection in ONE immediate transaction, so
/// the proof and the typed surface never diverge (they commit together or roll
/// back together). Mirrors `project_audit` but adds the decision insert. The audit
/// row keeps its file-rendered identity — `project_live_audit_record` never
/// live-rebases (ADR 027 C6). Because the audit row is inserted first in the same
/// transaction, the `operator_decisions.audit_id` FK is satisfied.
///
/// T4 (T3-review hardening): if the audit row GAPS (non-canonical timestamp, a
/// missing/conflicting `previous_audit_id`), `project_live_audit_record` returns
/// `Gap` WITHOUT inserting the row, so the typed insert below would otherwise hit
/// an opaque "FOREIGN KEY constraint failed" on `operator_decisions.audit_id`.
/// Capture the projection outcome and fail loud with the gap REASON before the
/// typed insert (the transaction is dropped → rolled back; no decision row leaks).
/// The normal `decide` path pins `previous_audit_id=None` + a canonical timestamp,
/// so it projects cleanly — this guard is for the edge.
pub(crate) fn project_decision(
    db_path: &Path,
    root: &Path,
    record: &ImportedAuditRecord,
    decision: &crate::decision::Decision,
) -> CliResult<()> {
    let mut connection = open_projection_database(db_path)?;
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|error| {
            CliError::failure(format!("failed to start decision projection: {error}"))
        })?;
    match project_live_audit_record(&transaction, root, record)? {
        LiveAuditProjection::Projected | LiveAuditProjection::AlreadyPresent => {}
        // The audit row did NOT land — inserting the typed row would FK-fail on a
        // dangling audit_id. Surface the gap reason instead (tx rolls back on drop).
        LiveAuditProjection::Gap { reason } => {
            return Err(CliError::failure(format!(
                "decision audit did not project (gap): {reason}"
            )))
        }
    }
    insert_operator_decision(&transaction, record, decision)?;
    transaction.commit().map_err(|error| {
        CliError::failure(format!("failed to commit decision projection: {error}"))
    })?;
    Ok(())
}

/// Insert the typed `operator_decisions` row keyed by the audit_id. `decision_status`
/// is always 'decided'; `notification_status` starts 'not-requested' (a later
/// `update_decision_notification` records the notify outcome, M2a T5). The variant
/// determines which of the optional columns carry a value.
fn insert_operator_decision(
    transaction: &Transaction<'_>,
    record: &ImportedAuditRecord,
    decision: &crate::decision::Decision,
) -> CliResult<()> {
    use crate::decision::Decision;
    let (verdict, resolution, mode_to, target_agent, reason, note) = match decision {
        Decision::Gate { verdict, note, .. } => {
            (Some(verdict.clone()), None, None, None, None, note.clone())
        }
        Decision::Conflict {
            resolution, note, ..
        } => (
            None,
            Some(resolution.clone()),
            None,
            None,
            None,
            note.clone(),
        ),
        Decision::Mode { mode_to } => (None, None, Some(mode_to.clone()), None, None, None),
        Decision::Interrupt { reason } => (None, None, None, None, reason.clone(), None),
        Decision::Redirect {
            target_agent,
            reason,
        } => (
            None,
            None,
            None,
            Some(target_agent.clone()),
            reason.clone(),
            None,
        ),
    };
    // `created_at` must match the canonical UTC-`Z` form the audit proof row gets in
    // `project_live_audit_record`, so the typed decision row and its proof stay
    // consistent and `ORDER BY created_at` is chronological. `record.timestamp` is
    // already validated RFC3339 upstream, so canonicalize won't be None in practice;
    // the fallback is defensive only.
    let created_at = crate::timestamp::canonicalize(&record.timestamp)
        .unwrap_or_else(|| record.timestamp.clone());
    // INSERT OR IGNORE on the audit_id PK: idempotent so a re-import (or an
    // accept-then-already-present rebuild on `db import`, v1 M2a R1 P1) never errors
    // or duplicates the typed row. The live `project_decision` path is unaffected —
    // its audit_id is always freshly minted, so the IGNORE clause never fires there.
    transaction
        .execute(
            "INSERT OR IGNORE INTO operator_decisions
                (audit_id, session_id, decision_type, target_work_event_id,
                 verdict, resolution, mode_to, target_agent, reason, note,
                 decision_status, notification_status, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'decided', 'not-requested', ?11)",
            params![
                record.audit_id,
                record.session_id,
                decision.record_type(),
                decision.target_work_event_id(),
                verdict,
                resolution,
                mode_to,
                target_agent,
                reason,
                note,
                created_at,
            ],
        )
        .map_err(|error| {
            CliError::failure(format!("failed to insert operator_decision: {error}"))
        })?;
    Ok(())
}

/// Record the OPTIONAL notification outcome on a durable decision row (ADR 033 D4 /
/// C4=b, M2a T5). The decision is ALREADY durable; this only updates the
/// notification columns of its `operator_decisions` row.
///
/// On `status="sent"` with a `notification_mid`, resolve the REAL sent audit the
/// audited send (ADR 029) just wrote — the row keyed by that mid with
/// `delivery_status='sent'` — and link it as `notification_audit_id`. On `failed`,
/// `notification_audit_id` stays NULL and `notification_error` carries the reason:
/// the honesty core is that a failed notify NEVER fabricates a `sent` proof and
/// NEVER writes any audit row; the failure lives on the decision row alone.
pub(crate) fn update_decision_notification(
    db_path: &Path,
    decision_audit_id: &str,
    status: &str,
    notification_mid: Option<&str>,
    notification_error: Option<&str>,
) -> CliResult<()> {
    let connection = open_projection_database(db_path)?;
    // On success, resolve the audit_id of the sent notification the audited send
    // wrote (keyed by the notification mid). Most-recent sent row by that mid wins.
    let notification_audit_id: Option<String> = match (status, notification_mid) {
        ("sent", Some(mid)) => connection
            .query_row(
                "SELECT audit_id FROM audit_records
                 WHERE mid = ?1 AND delivery_status = 'sent'
                 ORDER BY rowid DESC LIMIT 1",
                params![mid],
                |row| row.get(0),
            )
            .optional()
            .map_err(|error| {
                CliError::failure(format!("failed to resolve notification audit: {error}"))
            })?,
        _ => None,
    };
    let updated = connection
        .execute(
            "UPDATE operator_decisions
                SET notification_status = ?1,
                    notification_mid = ?2,
                    notification_audit_id = ?3,
                    notification_error = ?4
              WHERE audit_id = ?5",
            params![
                status,
                notification_mid,
                notification_audit_id,
                notification_error,
                decision_audit_id,
            ],
        )
        .map_err(|error| {
            CliError::failure(format!("failed to update decision notification: {error}"))
        })?;
    if updated == 0 {
        return Err(CliError::failure(format!(
            "no operator_decisions row for audit_id {decision_audit_id} to record notification"
        )));
    }
    Ok(())
}

/// Read a `work_event`'s kind for ref validation (gate/conflict decisions bind a
/// target work-event). Returns None when the id is absent in the session. Consumed
/// by `zynk decide` ref validation in M2a T4.
pub(crate) fn work_event_kind(
    db_path: &Path,
    session_id: &str,
    work_event_id: i64,
) -> CliResult<Option<String>> {
    let connection = open_projection_database(db_path)?;
    let kind: Option<String> = connection
        .query_row(
            "SELECT kind FROM work_events WHERE work_event_id = ?1 AND session_id = ?2",
            params![work_event_id, session_id],
            |row| row.get(0),
        )
        .optional()
        .map_err(|error| CliError::failure(format!("failed to read work_event kind: {error}")))?;
    Ok(kind)
}

/// ADR 033 M2b T3 (C3=b): the LATEST `mode-switch` operator decision for a session —
/// `(audit timestamp, mode_to)` of the most-recent decision by `audit_records.timestamp`,
/// or `None` if the session has no mode-switch.
///
/// This exists so the current-state read can make `mode` latest-writer across BOTH
/// the agent `status_events.mode` AND an operator `mode-switch` decision's `mode_to`.
/// Provenance stays clean: the operator decision keeps living in its typed
/// `operator_decisions` table joined to its `audit_records` proof row — it is NOT
/// synthesized into a fake `status_event` (the rejected alternative, which would
/// claim a phase/next-action/workflow_status the operator never authored). ONLY
/// `mode` participates cross-source; interrupt/redirect carry no `mode_to` and are
/// excluded by the `decision_type='mode-switch'` predicate, so they never touch the
/// current-state read. Takes a borrowed `&Connection` because the dashboard read
/// already holds one.
pub(crate) fn latest_mode_decision(
    connection: &Connection,
    session_id: &str,
) -> CliResult<Option<(String, String)>> {
    connection
        .query_row(
            "SELECT a.timestamp, od.mode_to
             FROM operator_decisions AS od
             JOIN audit_records AS a ON a.audit_id = od.audit_id
             WHERE od.session_id = ?1 AND od.decision_type = 'mode-switch'
             ORDER BY a.timestamp DESC, od.audit_id DESC
             LIMIT 1",
            [session_id],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
        )
        .optional()
        .map_err(|error| {
            CliError::failure(format!(
                "failed to read latest mode-switch decision: {error}"
            ))
        })
}

/// Project a validated work event into the live DB (ADR 033 M1). Validate before
/// any write (ADR 029 discipline — never store-bad), serialize to the typed
/// storage form, then content-dedup via INSERT OR IGNORE on the
/// UNIQUE(session_id, kind, content_hash) key so re-projecting the same event is
/// a no-op.
pub(crate) fn project_work_event(
    connection: &Connection,
    session_id: &str,
    actor: &str,
    timestamp: &str,
    seq: usize,
    payload: &crate::work_event::WorkEventPayload,
) -> CliResult<()> {
    payload.validate()?; // never store-bad (ADR 029 discipline)
    let stored = payload.to_storage()?;
    // The content hash is the EVENT IDENTITY: seq + actor + timestamp + payload (NOT
    // the payload alone). `work_events` is an event log, so two genuinely distinct
    // events must stay distinct rows. The file-rendered per-session `seq` ordinal
    // (v1 M1 R2 P4) disambiguates events that share an actor + second-precision
    // timestamp + payload — wrappers legitimately emit repeated identical
    // tool/usage/system events within one second, and those are real distinct events,
    // not a dedup. `seq` is generated before the file write and imported VERBATIM from
    // work.md, so re-projecting / re-importing the SAME block (same seq+actor+ts+
    // payload) is still an INSERT OR IGNORE no-op (idempotent). Reuses the sha256
    // convention (see `status_content_hash`).
    let content_hash = format!(
        "sha256:{:x}",
        Sha256::digest(format!("{seq}\n{actor}\n{timestamp}\n{stored}").as_bytes())
    );
    connection
        .execute(
            "INSERT OR IGNORE INTO work_events
                (session_id, actor_agent_id, kind, timestamp, payload, content_hash, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?4)",
            params![
                session_id,
                actor,
                payload.kind(),
                timestamp,
                stored,
                content_hash
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to project work event: {error}")))?;
    Ok(())
}

/// Identity-preserving live audit projection (ADR 027 C6): insert the
/// file-rendered `audit_id`/`previous_audit_id` VERBATIM (never DB-tail-derived,
/// which is what `db audit append` does). A missing previous or other insert
/// conflict is a projection Gap, not a live rebase — rebasing is `db import`'s
/// job. The timestamp is canonicalized for the DB exactly as import does; the
/// file keeps the raw form.
fn project_live_audit_record(
    transaction: &Transaction<'_>,
    root: &Path,
    record: &ImportedAuditRecord,
) -> CliResult<LiveAuditProjection> {
    let canonical_ts = match crate::timestamp::canonicalize(&record.timestamp) {
        Some(value) => value,
        None => {
            return Ok(LiveAuditProjection::Gap {
                reason: format!("non-canonical timestamp {}", record.timestamp),
            })
        }
    };

    // Idempotent: re-projecting the same audit_id is a no-op.
    let exists: Option<String> = transaction
        .query_row(
            "SELECT audit_id FROM audit_records WHERE audit_id = ?1",
            [&record.audit_id],
            |row| row.get(0),
        )
        .optional()
        .map_err(|error| CliError::failure(format!("failed to read audit_records: {error}")))?;
    if exists.is_some() {
        return Ok(LiveAuditProjection::AlreadyPresent);
    }

    // Ensure project + a minimal session row so the audit FK holds even when the
    // audit is the first DB write for this session (a later status projection
    // UPSERTs the real current-state; import never overwrites it).
    let project_id = import_project_id(root);
    ensure_project(transaction, &project_id, root, &canonical_ts)?;
    ensure_session_min(transaction, &project_id, &record.session_id, &canonical_ts)?;

    let mut canonical_record = record.clone();
    canonical_record.timestamp = canonical_ts;

    match execute_imported_audit_insert(
        transaction,
        &canonical_record,
        canonical_record.previous_audit_id.as_deref(),
    ) {
        Ok(_) => {
            // V2-C1: create agents only after the record is accepted.
            ensure_record_agents(transaction, &canonical_record)?;
            upsert_message_latest(transaction, &canonical_record)?;
            Ok(LiveAuditProjection::Projected)
        }
        // No live rebase (ADR 027 C6): leave the file authoritative.
        Err(error) => Ok(LiveAuditProjection::Gap {
            reason: error.to_string(),
        }),
    }
}

/// Ensure a minimal session row exists for FK integrity. Placeholder
/// current-state is corrected by a live status projection.
fn ensure_session_min(
    transaction: &Transaction<'_>,
    project_id: &str,
    session_id: &str,
    timestamp: &str,
) -> CliResult<()> {
    transaction
        .execute(
            "INSERT OR IGNORE INTO sessions (
                session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at
             )
             VALUES (?1, ?2, ?1, 'live', 'unknown', 'idle', ?3, ?3)",
            params![session_id, project_id, timestamp],
        )
        .map_err(|error| CliError::failure(format!("failed to ensure session: {error}")))?;
    Ok(())
}

/// Project a `zynk report` work event into the live DB (ADR 027/028): open the
/// projection DB the SAME way `status`/`audit` do — through
/// `open_projection_database`, so the default `.zynk/` DB auto-creates WITH its
/// self-ignoring `.zynk/.gitignore` (+ one-time notice). One immediate
/// transaction seeds the project + session FK rows (mirroring the
/// `ensure_project`/`ensure_session_min` pair import uses) before inserting the
/// validated, content-deduped work-event row.
pub(crate) fn project_report(
    db_path: &Path,
    root: &Path,
    session_id: &str,
    actor: &str,
    timestamp: &str,
    seq: usize,
    payload: &crate::work_event::WorkEventPayload,
) -> CliResult<()> {
    let mut connection = open_projection_database(db_path)?;
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|error| {
            CliError::failure(format!("failed to start report projection: {error}"))
        })?;
    let project_id = import_project_id(root);
    ensure_project(&transaction, &project_id, root, timestamp)?;
    ensure_session_min(&transaction, &project_id, session_id, timestamp)?;
    // `project_work_event` runs on the connection; a `Transaction` derefs to it,
    // so this insert participates in the same immediate transaction. The producer-
    // computed `seq` (v1 M1 R2 P4) matches what `db import` re-derives from work.md,
    // so the live and recovery paths agree on the event identity.
    project_work_event(&transaction, session_id, actor, timestamp, seq, payload)?;
    // T4 dual-projection: an `artifact` work event ALSO upserts one row per file
    // into the `artifacts` table — in the SAME immediate transaction as the
    // work_event insert, so the two views never diverge on a partial failure.
    if let crate::work_event::WorkEventPayload::Artifact { files } = payload {
        for file in files {
            upsert_artifact(&transaction, session_id, &file.path, timestamp)?;
        }
    }
    transaction.commit().map_err(|error| {
        CliError::failure(format!("failed to commit report projection: {error}"))
    })?;
    Ok(())
}

/// T4: upsert one `artifacts` row per reported file path. The artifacts table is
/// the live "what files moved" view (UNIQUE(session_id, path)); re-reporting the
/// same path is an idempotent `updated_at` bump, not a duplicate row. Called from
/// within `project_report`'s immediate transaction so it commits atomically with
/// the `work_events` insert.
pub(crate) fn upsert_artifact(
    connection: &Connection,
    session_id: &str,
    path: &str,
    ts: &str,
) -> CliResult<()> {
    connection
        .execute(
            "INSERT INTO artifacts (session_id, kind, path, updated_at) VALUES (?1, 'file', ?2, ?3)
             ON CONFLICT(session_id, path) DO UPDATE SET updated_at=excluded.updated_at",
            params![session_id, path, ts],
        )
        .map_err(|e| CliError::failure(format!("failed to upsert artifact: {e}")))?;
    Ok(())
}

/// Upsert the message row's latest delivery state for a live audit projection
/// (ON CONFLICT updates latest_*, mirroring `db audit append`).
/// ADR 029 C6 (R1 P1): the corpus excerpt to persist for a record. The redaction
/// policy is the authority — content is stored only for `full`/`excerpt`, never
/// for `hash-only`, regardless of what a parsed/imported artifact carries.
/// Enforcing this at the corpus-write boundary closes the import-trust leak where
/// a stale/forged `payload_excerpt=` line on a hash-only record could reach the
/// live corpus.
fn corpus_payload_excerpt(record: &ImportedAuditRecord) -> Option<&str> {
    if record.payload_redaction_policy == "hash-only" {
        None
    } else {
        record.payload_excerpt.as_deref()
    }
}

fn upsert_message_latest(
    transaction: &Transaction<'_>,
    record: &ImportedAuditRecord,
) -> CliResult<()> {
    let message_id = message_id(&record.session_id, &record.mid);
    transaction
        .execute(
            "INSERT INTO messages (
                message_id, session_id, mid, source_agent_id, target_agent_id, message_type,
                mode, ref, transport, payload_redaction_policy, payload_hash,
                latest_delivery_status, latest_verified_by, latest_audit_id, timestamp,
                payload_excerpt
             )
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)
             ON CONFLICT(session_id, mid) DO UPDATE SET
                source_agent_id = excluded.source_agent_id,
                target_agent_id = excluded.target_agent_id,
                message_type = excluded.message_type,
                mode = excluded.mode,
                ref = excluded.ref,
                transport = excluded.transport,
                payload_redaction_policy = excluded.payload_redaction_policy,
                payload_hash = excluded.payload_hash,
                latest_delivery_status = excluded.latest_delivery_status,
                latest_verified_by = excluded.latest_verified_by,
                latest_audit_id = excluded.latest_audit_id,
                timestamp = excluded.timestamp,
                payload_excerpt = excluded.payload_excerpt",
            params![
                message_id,
                record.session_id,
                record.mid,
                record.source_agent_id,
                record.target_agent_id,
                record.record_type,
                record.mode,
                record.r#ref,
                record.transport,
                record.payload_redaction_policy,
                record.payload_hash,
                record.delivery_status,
                record.verified_by,
                record.audit_id,
                record.timestamp,
                corpus_payload_excerpt(record),
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to project message: {error}")))?;
    Ok(())
}

fn insert_import_row(
    transaction: &Transaction<'_>,
    session_dir: &Path,
    source_hash: &str,
    imported_at: &str,
    warnings: &[String],
) -> CliResult<()> {
    let warning_summary = if warnings.is_empty() {
        "none".to_string()
    } else {
        warnings.join("; ")
    };
    let result = if warnings.is_empty() {
        "imported"
    } else {
        "imported-with-warnings"
    };
    transaction
        .execute(
            "INSERT INTO imports (
                source_kind, source_path, source_hash, imported_at, result, warning_summary
             )
             VALUES ('outputs', ?1, ?2, ?3, ?4, ?5)",
            params![
                session_dir.display().to_string(),
                source_hash,
                imported_at,
                result,
                warning_summary,
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to record import: {error}")))?;
    Ok(())
}

/// Normalize an explicit CLI `--timestamp` argument to canonical UTC-`Z` before
/// it reaches the DB (ADR 026 D1 / Codex V1-C1). Valid RFC3339 offsets are
/// converted; invalid input fails cleanly here rather than as a trigger abort.
fn normalize_arg_timestamp(value: &str) -> CliResult<String> {
    crate::timestamp::canonicalize(value).ok_or_else(|| {
        CliError::usage(format!(
            "--timestamp must be RFC3339 (e.g. 2026-05-29T02:00:00Z): {value}"
        ))
    })
}

fn append_audit_record(path: &Path, args: &DbAuditAppendArgs) -> CliResult<String> {
    if args.payload.is_some() == args.payload_file.is_some() {
        return Err(CliError::usage(
            "exactly one of --payload or --payload-file is required",
        ));
    }
    if args.excerpt_chars < 1 {
        return Err(CliError::usage("--excerpt-chars must be >= 1"));
    }
    if !matches!(
        args.payload_redaction_policy.as_str(),
        "hash-only" | "excerpt" | "full"
    ) {
        return Err(CliError::usage(format!(
            "invalid redaction policy: {}",
            args.payload_redaction_policy
        )));
    }
    if !matches!(
        args.delivery_status.as_str(),
        "drafted" | "sent" | "observed" | "failed" | "unknown"
    ) {
        return Err(CliError::usage(format!(
            "invalid delivery status: {}",
            args.delivery_status
        )));
    }
    // ADR 024 §1a / ADR 034 D8 (v1 M3a R3 P4): `db audit append` is a SEPARATE audit
    // producer from `zynk audit` (it never calls `validate_audit_args`), so it needs
    // its OWN §1a guard. A non-transport operator proof — any decision record_type or
    // `reveal` — is observed/operator, never a delivered message. Fail fast with a
    // clear message; the v9 BEFORE INSERT trigger is the authoritative backstop for
    // producers that bypass this CLI path (import / direct insert).
    if crate::decision::is_non_transport_proof_record_type(&args.record_type)
        && args.delivery_status == "sent"
    {
        return Err(CliError::usage(
            "a non-transport operator proof (a decision record_type or `reveal`) must not be delivery_status=sent (ADR 024 §1a / ADR 034 D8: an observed/operator event, not a delivered message)",
        ));
    }

    let timestamp = normalize_arg_timestamp(&args.timestamp)?;
    let payload = payload_from_args(args)?;
    let payload_bytes = payload.as_bytes();
    let payload_hash = format!("sha256:{:x}", Sha256::digest(payload_bytes));
    let payload_excerpt =
        payload_excerpt(&payload, args.excerpt_chars, &args.payload_redaction_policy);
    let message_id = message_id(&args.session_id, &args.mid);
    let mut connection = open_database(path)?;
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|error| {
            CliError::failure(format!("failed to start SQLite audit transaction: {error}"))
        })?;
    let previous_audit_id = tail_audit_id(&transaction, &args.session_id)?;
    let audit_id = match &args.audit_id {
        Some(audit_id) => audit_id.clone(),
        None => generate_audit_id(&transaction)?,
    };

    transaction
        .execute(
            "INSERT INTO audit_records (
                audit_id, previous_audit_id, session_id, source_agent_id, target_agent_id,
                source_address, target_address, transport, workspace_id, mid, record_type,
                command_origin, mode, ref, re, payload_hash, payload_redaction_policy,
                content_size, delivery_status, observed_by, verified_by, timestamp
             )
             VALUES (
                ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11,
                ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22
             )",
            params![
                audit_id,
                previous_audit_id,
                args.session_id,
                args.source_agent_id,
                args.target_agent_id,
                args.source_address,
                args.target_address,
                args.transport,
                args.workspace_id,
                args.mid,
                args.record_type,
                args.command_origin,
                args.mode,
                args.r#ref,
                args.re,
                payload_hash,
                args.payload_redaction_policy,
                payload_bytes.len() as i64,
                args.delivery_status,
                args.observed_by,
                args.verified_by,
                timestamp,
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to append audit record: {error}")))?;

    transaction
        .execute(
            "INSERT INTO messages (
                message_id, session_id, mid, source_agent_id, target_agent_id, message_type,
                mode, ref, transport, transport_thread_id, payload_redaction_policy,
                payload_excerpt, payload_hash, latest_delivery_status, latest_verified_by,
                latest_audit_id, timestamp
             )
             VALUES (
                ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17
             )
             ON CONFLICT(session_id, mid) DO UPDATE SET
                source_agent_id = excluded.source_agent_id,
                target_agent_id = excluded.target_agent_id,
                message_type = excluded.message_type,
                mode = excluded.mode,
                ref = excluded.ref,
                transport = excluded.transport,
                transport_thread_id = excluded.transport_thread_id,
                payload_redaction_policy = excluded.payload_redaction_policy,
                payload_excerpt = excluded.payload_excerpt,
                payload_hash = excluded.payload_hash,
                latest_delivery_status = excluded.latest_delivery_status,
                latest_verified_by = excluded.latest_verified_by,
                latest_audit_id = excluded.latest_audit_id,
                timestamp = excluded.timestamp",
            params![
                message_id,
                args.session_id,
                args.mid,
                args.source_agent_id,
                args.target_agent_id,
                args.record_type,
                args.mode,
                args.r#ref,
                args.transport,
                args.transport_thread_id,
                args.payload_redaction_policy,
                payload_excerpt,
                payload_hash,
                args.delivery_status,
                args.verified_by,
                audit_id,
                timestamp,
            ],
        )
        .map_err(|error| CliError::failure(format!("failed to update message state: {error}")))?;

    transaction.commit().map_err(|error| {
        CliError::failure(format!("failed to commit audit transaction: {error}"))
    })?;
    Ok(audit_id)
}

fn payload_from_args(args: &DbAuditAppendArgs) -> CliResult<String> {
    if let Some(path) = &args.payload_file {
        let mut payload = String::new();
        fs::File::open(path)
            .and_then(|mut file| file.read_to_string(&mut payload))
            .map_err(|error| {
                CliError::failure(format!(
                    "failed to read payload file {}: {error}",
                    path.display()
                ))
            })?;
        return Ok(payload);
    }
    Ok(args.payload.clone().unwrap_or_default())
}

fn payload_excerpt(payload: &str, chars: usize, policy: &str) -> Option<String> {
    match policy {
        "hash-only" => None,
        "full" => Some(payload.to_string()),
        _ => {
            let count = payload.chars().count();
            if count <= chars * 2 {
                Some(payload.to_string())
            } else {
                let start = payload.chars().take(chars).collect::<String>();
                let end = payload
                    .chars()
                    .rev()
                    .take(chars)
                    .collect::<String>()
                    .chars()
                    .rev()
                    .collect::<String>();
                Some(format!("{start}...{end}"))
            }
        }
    }
}

fn tail_audit_id(transaction: &Transaction<'_>, session_id: &str) -> CliResult<Option<String>> {
    transaction
        .query_row(
            "SELECT record.audit_id
             FROM audit_records AS record
             WHERE record.session_id = ?1
               AND NOT EXISTS (
                 SELECT 1
                 FROM audit_records AS child
                 WHERE child.session_id = record.session_id
                   AND child.previous_audit_id = record.audit_id
               )
             ORDER BY record.timestamp DESC, record.audit_id DESC
             LIMIT 1",
            [session_id],
            |row| row.get(0),
        )
        .optional()
        .map_err(|error| CliError::failure(format!("failed to read audit chain tail: {error}")))
}

fn generate_audit_id(transaction: &Transaction<'_>) -> CliResult<String> {
    let mut statement = transaction
        .prepare("SELECT audit_id FROM audit_records")
        .map_err(|error| CliError::failure(format!("failed to read audit ids: {error}")))?;
    let existing_ids = statement
        .query_map([], |row| row.get::<_, String>(0))
        .map_err(|error| CliError::failure(format!("failed to read audit ids: {error}")))?
        .collect::<Result<HashSet<_>, _>>()
        .map_err(|error| CliError::failure(format!("failed to read audit ids: {error}")))?;

    const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789";
    let mut rng = OsRng;
    for _ in 0..100 {
        let audit_id = (0..6)
            .map(|_| {
                let index = rng.gen_range(0..ALPHABET.len());
                ALPHABET[index] as char
            })
            .collect::<String>();
        if !existing_ids.contains(&audit_id) {
            return Ok(audit_id);
        }
    }
    Err(CliError::usage(
        "could not generate unique audit_id after 100 attempts",
    ))
}

fn message_id(session_id: &str, mid: &str) -> String {
    let digest = Sha256::digest(format!("{session_id}\0{mid}").as_bytes());
    format!("msg-{digest:x}")
}

fn import_project_id(root: &Path) -> String {
    let digest = Sha256::digest(root.display().to_string().as_bytes());
    let hex = format!("{digest:x}");
    format!("project-{}", &hex[..16])
}

fn source_hash_for_session_dir(session_dir: &Path) -> CliResult<String> {
    let mut hasher = Sha256::new();
    for file_name in ["status.md", "audit.md", "summary.md"] {
        let path = session_dir.join(file_name);
        if path.exists() {
            hasher.update(file_name.as_bytes());
            let content = fs::read(&path).map_err(|error| {
                CliError::failure(format!("failed to read {}: {error}", path.display()))
            })?;
            hasher.update(content);
        }
    }
    Ok(format!("sha256:{:x}", hasher.finalize()))
}

fn value_or(values: &BTreeMap<String, String>, key: &str, fallback: &str) -> String {
    values
        .get(key)
        .filter(|value| !value.is_empty())
        .cloned()
        .unwrap_or_else(|| fallback.to_string())
}

fn optional_text(value: &str) -> Option<String> {
    if value.is_empty() || value == "unknown" {
        None
    } else {
        Some(value.to_string())
    }
}

fn optional_audit_id(value: &str) -> Option<String> {
    if matches!(value, "" | "none" | "unknown") {
        None
    } else {
        Some(value.to_string())
    }
}

fn nullable_unknown(value: &str) -> Option<String> {
    if value.is_empty() || value == "unknown" {
        None
    } else {
        Some(value.to_string())
    }
}

fn table_has_column(connection: &Connection, table: &str, column: &str) -> CliResult<bool> {
    let mut statement = connection
        .prepare(&format!("PRAGMA table_info({table})"))
        .map_err(|error| {
            CliError::failure(format!("failed to inspect SQLite table {table}: {error}"))
        })?;
    let columns = statement
        .query_map([], |row| row.get::<_, String>(1))
        .map_err(|error| {
            CliError::failure(format!("failed to inspect SQLite table {table}: {error}"))
        })?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| {
            CliError::failure(format!("failed to inspect SQLite table {table}: {error}"))
        })?;
    Ok(columns.iter().any(|name| name == column))
}

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

    // Build a v3-state database (pre-content_hash) the way a v0.2.0 binary would.
    fn open_v3(connection: &Connection) {
        create_schema_migrations(connection).unwrap();
        apply_v1(connection).unwrap();
        apply_v2(connection).unwrap();
        apply_v3(connection).unwrap();
        connection.pragma_update(None, "user_version", 3).unwrap();
    }

    // ADR 033 M1: a freshly opened DB is at schema v6 with a work_events table,
    // and project_work_event inserts a row that reads back as the typed payload.
    #[test]
    fn v6_work_events_insert_and_read() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("z.db");
        let conn = open_database(&db).unwrap();
        // seed a session row (FK). Timestamps must be canonical RFC3339 UTC-Z to
        // satisfy the v4 sessions_ts_canonical trigger.
        let ts = "2026-05-31T00:00:00Z";
        conn.execute(
            "INSERT INTO projects (project_id,name,root_path,created_at,updated_at) VALUES ('p','p','/tmp',?1,?1)",
            params![ts],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO sessions (session_id,project_id,title,phase,mode,workflow_status,created_at,updated_at) VALUES ('s1','p','s1','live','unknown','idle',?1,?1)",
            params![ts],
        )
        .unwrap();
        let payload = crate::work_event::WorkEventPayload::Think {
            text: "weighing options".into(),
        };
        project_work_event(&conn, "s1", "claude", "2026-05-31T00:00:00Z", 0, &payload).unwrap();
        let (kind, stored): (String, String) = conn
            .query_row(
                "SELECT kind, payload FROM work_events WHERE session_id='s1'",
                [],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )
            .unwrap();
        assert_eq!(kind, "think");
        assert_eq!(
            crate::work_event::WorkEventPayload::from_storage(&stored).unwrap(),
            payload
        );
    }

    // ADR 026 C5: a legacy status_events row predating content_hash must be
    // backfilled by apply_v4 with the same hash the import path computes, so a
    // later re-import dedups instead of inserting a duplicate.
    #[test]
    fn v4_backfills_content_hash_for_legacy_status_event() {
        let connection = Connection::open_in_memory().unwrap();
        open_v3(&connection);
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/p', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES ('s', 'p', 's', 'tooling', 'review', 'working', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
        // v3 status_events has no content_hash column yet.
        connection
            .execute(
                "INSERT INTO status_events (session_id, timestamp, phase, mode, workflow_status,
                    completed_since_last_update, in_progress, next_action, blockers,
                    asks_for_zevs, risk_or_residual_uncertainty, expected_wait, event_text)
                 VALUES ('s', '2026-05-29T01:00:00Z', 'tooling', 'review', 'working',
                    'c', 'p', 'next-A', 'none', 'none', 'none', 'unknown', 'event one')",
                [],
            )
            .unwrap();

        apply_v4(&connection).unwrap();

        let stored: String = connection
            .query_row(
                "SELECT content_hash FROM status_events WHERE session_id='s'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let expected = status_content_hash(&[
            "tooling",
            "review",
            "working",
            "c",
            "p",
            "next-A",
            "none",
            "none",
            "none",
            "unknown",
            "event one",
        ]);
        assert_eq!(
            stored, expected,
            "legacy row must be backfilled with the canonical hash"
        );
        assert!(stored.starts_with("sha256:"));
    }

    // ADR 026 C3: scan counts pre-existing non-canonical timestamps so migrate()
    // can warn. Rows inserted before the v4 triggers exist can be non-canonical.
    #[test]
    fn noncanonical_scan_counts_legacy_offset_rows() {
        let connection = Connection::open_in_memory().unwrap();
        open_v3(&connection);
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/p', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
        // sessions.updated_at carries a legacy offset form (pre-v4, no trigger).
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES ('s', 'p', 's', 'x', 'review', 'working', '2026-05-29T00:00:00Z', '2026-05-28T18:00:00+07:00')",
                [],
            )
            .unwrap();
        assert_eq!(noncanonical_timestamp_count(&connection).unwrap(), 1);

        apply_v4(&connection).unwrap();
        // apply_v4 does not rewrite existing rows, so the legacy row is still counted.
        assert_eq!(noncanonical_timestamp_count(&connection).unwrap(), 1);
    }

    // ADR 026 C5 / Codex V1-C2: the full migrated path — a v3 legacy row with
    // content_hash='' is backfilled by v4, and a re-import of the SAME content
    // must dedup (not duplicate) because backfill and import share the helper.
    #[test]
    fn migrated_legacy_status_row_dedups_on_reimport() {
        let mut connection = Connection::open_in_memory().unwrap();
        open_v3(&connection);
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/p', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES ('s', 'p', 's', 'tooling', 'review', 'working', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
        // legacy v3 row: content_hash column does not exist yet
        connection
            .execute(
                "INSERT INTO status_events (session_id, timestamp, phase, mode, workflow_status,
                    completed_since_last_update, in_progress, next_action, blockers,
                    asks_for_zevs, risk_or_residual_uncertainty, expected_wait, event_text)
                 VALUES ('s', '2026-05-29T01:00:00Z', 'tooling', 'review', 'working',
                    'c', 'p', 'next-A', 'none', 'none', 'none', 'unknown', 'event one')",
                [],
            )
            .unwrap();

        apply_v4(&connection).unwrap();

        // Re-import the identical content through the importer dedup path.
        let status = ImportedStatus {
            session_id: "s".to_string(),
            last_update: "2026-05-29T01:00:00Z".to_string(),
            lead_agent: "codex".to_string(),
            phase: "tooling".to_string(),
            mode: "review".to_string(),
            artifact_ref: "tools/x".to_string(),
            workflow_status: "working".to_string(),
            completed_since_last_update: "c".to_string(),
            in_progress: "p".to_string(),
            next_action: "next-A".to_string(),
            blockers: "none".to_string(),
            asks_for_zevs: "none".to_string(),
            risk_or_residual_uncertainty: "none".to_string(),
            expected_wait: "unknown".to_string(),
            rolling_events: vec!["event one".to_string()],
        };
        let tx = connection.transaction().unwrap();
        insert_status_event_if_missing(&tx, &status).unwrap();
        tx.commit().unwrap();

        let count: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM status_events WHERE session_id='s' AND timestamp='2026-05-29T01:00:00Z'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            count, 1,
            "re-import of a backfilled legacy row must dedup, not duplicate"
        );
    }

    // ADR 026 D5 (+ Codex V2-C2): the schema CHECK enums must EXACTLY match the
    // canonical per-column value lists. Asserting the exact `col IN (...)` literal
    // (one line in the schema) catches missing values, EXTRA values, reordering,
    // and column-specific drift (a value in the wrong column's CHECK) — not just
    // presence somewhere in the table. The fragments here are the spec.
    #[test]
    fn schema_check_enums_match_canonical_values() {
        let connection = Connection::open_in_memory().unwrap();
        open_v3(&connection);
        apply_v4(&connection).unwrap();
        let table_sql = |name: &str| -> String {
            connection
                .query_row(
                    "SELECT sql FROM sqlite_master WHERE type='table' AND name=?1",
                    [name],
                    |row| row.get(0),
                )
                .unwrap()
        };
        // (table, exact CHECK IN-list fragment as it appears on one schema line)
        let expectations: &[(&str, &str)] = &[
            (
                "sessions",
                "workflow_status IN ('idle', 'working', 'blocked', 'waiting-for-operator', 'done')",
            ),
            (
                "status_events",
                "workflow_status IN ('idle', 'working', 'blocked', 'waiting-for-operator', 'done')",
            ),
            (
                "agents",
                "current_agent_status IN ('idle', 'working', 'blocked', 'done', 'unknown')",
            ),
            (
                "session_agents",
                "agent_status IN ('idle', 'working', 'blocked', 'done', 'unknown')",
            ),
            (
                "audit_records",
                "command_origin IN ('agent', 'operator', 'helper-tool', 'unknown')",
            ),
            (
                "audit_records",
                "payload_redaction_policy IN ('hash-only', 'excerpt', 'full')",
            ),
            (
                "audit_records",
                "delivery_status IN ('drafted', 'sent', 'observed', 'failed', 'unknown')",
            ),
            (
                "audit_records",
                "verified_by IN ('transport', 'agent', 'operator', 'helper-tool')",
            ),
            (
                "messages",
                "payload_redaction_policy IN ('hash-only', 'excerpt', 'full')",
            ),
            (
                "messages",
                "latest_delivery_status IN ('drafted', 'sent', 'observed', 'failed', 'unknown')",
            ),
            (
                "messages",
                "latest_verified_by IN ('transport', 'agent', 'operator', 'helper-tool')",
            ),
        ];
        for (table, fragment) in expectations {
            let sql = table_sql(table);
            assert!(
                sql.contains(fragment),
                "{table} CHECK drifted from canonical enum; expected exact fragment: {fragment}"
            );
        }
    }

    // ===== ADR 027 schema v5: audit_records.due =====

    // Build a v4-state database (pre-`due`) the way a v0.2.2 binary would.
    fn open_v4(connection: &Connection) {
        create_schema_migrations(connection).unwrap();
        apply_v1(connection).unwrap();
        apply_v2(connection).unwrap();
        apply_v3(connection).unwrap();
        apply_v4(connection).unwrap();
        connection.pragma_update(None, "user_version", 4).unwrap();
    }

    fn seed_audit_session(connection: &Connection) {
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/p', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES ('s', 'p', 's', 'x', 'review', 'working', '2026-05-29T00:00:00Z', '2026-05-29T00:00:00Z')",
                [],
            )
            .unwrap();
    }

    // Insert an audit row carrying an explicit `due` (None => SQL NULL). Requires v5.
    fn insert_audit_with_due(
        connection: &Connection,
        audit_id: &str,
        due: Option<&str>,
    ) -> rusqlite::Result<usize> {
        connection.execute(
            "INSERT INTO audit_records (audit_id, previous_audit_id, session_id, source_address,
                target_address, transport, workspace_id, mid, record_type, command_origin,
                payload_hash, payload_redaction_policy, content_size, delivery_status, observed_by,
                verified_by, timestamp, due)
             VALUES (?1, NULL, 's', 'src', 'tgt', 'herdr', 'w', ?1, 'status-update', 'agent',
                'sha256:x', 'hash-only', 0, 'observed', 'codex', 'helper-tool',
                '2026-05-29T01:00:00Z', ?2)",
            params![audit_id, due],
        )
    }

    // v5 adds the column and does not disturb existing rows (new column reads NULL).
    #[test]
    fn v5_adds_due_column_and_preserves_existing_audit_rows() {
        let connection = Connection::open_in_memory().unwrap();
        open_v4(&connection);
        seed_audit_session(&connection);
        connection
            .execute(
                "INSERT INTO audit_records (audit_id, previous_audit_id, session_id, source_address,
                    target_address, transport, workspace_id, mid, record_type, command_origin,
                    payload_hash, payload_redaction_policy, content_size, delivery_status,
                    observed_by, verified_by, timestamp)
                 VALUES ('a1', NULL, 's', 'src', 'tgt', 'herdr', 'w', 'm1', 'status-update',
                    'agent', 'sha256:x', 'hash-only', 0, 'observed', 'codex', 'helper-tool',
                    '2026-05-29T01:00:00Z')",
                [],
            )
            .unwrap();
        assert!(!table_has_column(&connection, "audit_records", "due").unwrap());

        apply_v5(&connection).unwrap();

        assert!(table_has_column(&connection, "audit_records", "due").unwrap());
        let due: Option<String> = connection
            .query_row(
                "SELECT due FROM audit_records WHERE audit_id='a1'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert!(
            due.is_none(),
            "existing audit row must read due=NULL after v5"
        );
    }

    // The v5 shape trigger is defense-in-depth: NULL and canonical `due` are
    // accepted; a shape-violating `due` is rejected at insert.
    #[test]
    fn v5_due_shape_trigger_accepts_null_and_canonical_rejects_malformed() {
        let connection = Connection::open_in_memory().unwrap();
        open_v4(&connection);
        apply_v5(&connection).unwrap();
        seed_audit_session(&connection);

        insert_audit_with_due(&connection, "a1", None).expect("NULL due must be accepted");
        insert_audit_with_due(&connection, "a2", Some("2026-06-01T12:00:00Z"))
            .expect("canonical due must be accepted");
        assert!(
            insert_audit_with_due(&connection, "a3", Some("2026-06-01 12:00:00")).is_err(),
            "shape-violating due (no T/Z) must be rejected by the trigger"
        );
    }

    // End-to-end: migrate() advances a v4 DB to the current head (>= v5), adds the
    // v5 due column, and re-open is a no-op. The terminal version tracks
    // CURRENT_SCHEMA_VERSION (now v8 after the additive operator_decisions (v7) and
    // custody_vault (v8) migrations); the v5 behavior (due column present) is unchanged.
    #[test]
    fn migrate_advances_v4_db_to_v5_with_due_column() {
        let mut connection = Connection::open_in_memory().unwrap();
        open_v4(&connection);
        assert!(!table_has_column(&connection, "audit_records", "due").unwrap());

        migrate(&mut connection).unwrap();

        let version: i64 = connection
            .pragma_query_value(None, "user_version", |row| row.get(0))
            .unwrap();
        assert_eq!(version, CURRENT_SCHEMA_VERSION);
        assert!(table_has_column(&connection, "audit_records", "due").unwrap());

        // The v7->v8 step creates the custody_vault table; assert it exists post-migration
        // so the v4->v8 ladder's terminal step has explicit coverage (matches the
        // `v8_custody_vault_table_present` integration check).
        let custody_vault_present: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
                ["custody_vault"],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            custody_vault_present, 1,
            "custody_vault table must exist after a v4->v8 migration"
        );

        // v1 M3a R3 P4: the v8->v9 step adds the non-transport-proof sent-guard
        // trigger; assert it exists after the v4->v9 migration so an upgraded
        // legacy DB rejects forged `sent` decision/reveal rows (ADR 024 / ADR 034 D8).
        let nontransport_trigger_present: i64 = connection
            .query_row(
                "SELECT count(*) FROM sqlite_master
                 WHERE type = 'trigger' AND name = ?1",
                ["audit_records_no_sent_nontransport_proof_insert"],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            nontransport_trigger_present, 1,
            "audit_records_no_sent_nontransport_proof_insert trigger must exist after a v4->v9 migration"
        );

        // T9: re-running migrate at the current version is a no-op.
        migrate(&mut connection).unwrap();
        let version_again: i64 = connection
            .pragma_query_value(None, "user_version", |row| row.get(0))
            .unwrap();
        assert_eq!(version_again, CURRENT_SCHEMA_VERSION);
    }

    // ADR 027 slice 4a: a live status projection OWNS session current-state, so a
    // second write UPDATEs the session row (unlike import's INSERT OR IGNORE) and
    // appends a distinct status event.
    #[test]
    fn live_status_projection_upserts_session_current_state() {
        let mut connection = Connection::open_in_memory().unwrap();
        open_v4(&connection);
        apply_v5(&connection).unwrap();
        let root = Path::new("outputs");
        let make =
            |phase: &str, status: &str, lead: &str, artifact: &str, ts: &str| ImportedStatus {
                session_id: "s".to_string(),
                last_update: ts.to_string(),
                lead_agent: lead.to_string(),
                phase: phase.to_string(),
                mode: "build".to_string(),
                artifact_ref: artifact.to_string(),
                workflow_status: status.to_string(),
                completed_since_last_update: "c".to_string(),
                in_progress: "p".to_string(),
                next_action: "n".to_string(),
                blockers: "none".to_string(),
                asks_for_zevs: "none".to_string(),
                risk_or_residual_uncertainty: "none".to_string(),
                expected_wait: "unknown".to_string(),
                rolling_events: vec![format!("event-{phase}")],
            };
        {
            let tx = connection.transaction().unwrap();
            project_live_status(
                &tx,
                root,
                &make("p1", "working", "claude", "a1", "2026-05-29T01:00:00Z"),
            )
            .unwrap();
            tx.commit().unwrap();
        }
        {
            let tx = connection.transaction().unwrap();
            project_live_status(
                &tx,
                root,
                &make("p2", "blocked", "codex", "a2", "2026-05-29T02:00:00Z"),
            )
            .unwrap();
            tx.commit().unwrap();
        }

        let (phase, status, lead, artifact): (String, String, Option<String>, Option<String>) =
            connection
                .query_row(
                    "SELECT phase, workflow_status, lead_agent_id, artifact_ref
                     FROM sessions WHERE session_id='s'",
                    [],
                    |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
                )
                .unwrap();
        assert_eq!(phase, "p2", "session phase must reflect the latest write");
        assert_eq!(status, "blocked");
        assert_eq!(lead.as_deref(), Some("codex"));
        assert_eq!(artifact.as_deref(), Some("a2"));

        let events: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM status_events WHERE session_id='s'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(events, 2, "two distinct status writes append two events");
    }

    // v5 with foreign keys ON, matching how open_database configures a real DB —
    // required for the missing-previous FK gap to fire.
    fn open_v5(connection: &Connection) {
        open_v4(connection);
        apply_v5(connection).unwrap();
        connection
            .pragma_update(None, "foreign_keys", "ON")
            .unwrap();
    }

    fn live_audit_record(audit_id: &str, previous: Option<&str>, ts: &str) -> ImportedAuditRecord {
        ImportedAuditRecord {
            audit_id: audit_id.to_string(),
            previous_audit_id: previous.map(str::to_string),
            timestamp: ts.to_string(),
            source_agent_id: Some("claude".to_string()),
            source_address: "w-1".to_string(),
            target_agent_id: Some("codex".to_string()),
            target_address: "w-2".to_string(),
            transport: "herdr".to_string(),
            workspace_id: "w".to_string(),
            session_id: "s".to_string(),
            mid: audit_id.to_string(),
            record_type: "status-update".to_string(),
            command_origin: "agent".to_string(),
            mode: None,
            r#ref: None,
            re: None,
            payload_hash: "sha256:x".to_string(),
            payload_redaction_policy: "hash-only".to_string(),
            content_size: 0,
            delivery_status: "observed".to_string(),
            observed_by: "codex".to_string(),
            verified_by: "helper-tool".to_string(),
            due: None,
            payload_excerpt: None,
        }
    }

    fn project_one_audit(
        connection: &mut Connection,
        record: &ImportedAuditRecord,
    ) -> LiveAuditProjection {
        let tx = connection.transaction().unwrap();
        let outcome = project_live_audit_record(&tx, Path::new("outputs"), record).unwrap();
        tx.commit().unwrap();
        outcome
    }

    // ADR 027 C6: the live audit projection inserts the file-rendered chain
    // identity verbatim — previous is the file value, NOT DB-tail-derived.
    #[test]
    fn live_audit_projection_preserves_file_chain_identity() {
        let mut connection = Connection::open_in_memory().unwrap();
        open_v5(&connection);
        assert_eq!(
            project_one_audit(
                &mut connection,
                &live_audit_record("a1", None, "2026-05-29T01:00:00Z")
            ),
            LiveAuditProjection::Projected
        );
        assert_eq!(
            project_one_audit(
                &mut connection,
                &live_audit_record("a2", Some("a1"), "2026-05-29T02:00:00Z")
            ),
            LiveAuditProjection::Projected
        );
        let previous: Option<String> = connection
            .query_row(
                "SELECT previous_audit_id FROM audit_records WHERE audit_id='a2'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            previous.as_deref(),
            Some("a1"),
            "previous must be the file value verbatim, not DB-tail-derived"
        );
    }

    // ADR 027 C6: a missing previous is a projection Gap — the record is NOT
    // inserted and NOT rebased to root. db import reconciles later.
    #[test]
    fn live_audit_projection_gap_on_missing_previous_without_rebase() {
        let mut connection = Connection::open_in_memory().unwrap();
        open_v5(&connection);
        let outcome = project_one_audit(
            &mut connection,
            &live_audit_record("a9", Some("ghost"), "2026-05-29T03:00:00Z"),
        );
        assert!(
            matches!(outcome, LiveAuditProjection::Gap { .. }),
            "missing previous must be a Gap, got {outcome:?}"
        );
        let count: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM audit_records WHERE audit_id='a9'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            count, 0,
            "a gapped record must not be inserted (no live rebase)"
        );
    }

    #[test]
    fn live_audit_projection_is_idempotent_on_reprojection() {
        let mut connection = Connection::open_in_memory().unwrap();
        open_v5(&connection);
        assert_eq!(
            project_one_audit(
                &mut connection,
                &live_audit_record("a1", None, "2026-05-29T01:00:00Z")
            ),
            LiveAuditProjection::Projected
        );
        assert_eq!(
            project_one_audit(
                &mut connection,
                &live_audit_record("a1", None, "2026-05-29T01:00:00Z")
            ),
            LiveAuditProjection::AlreadyPresent
        );
        let count: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM audit_records WHERE audit_id='a1'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    /// Seed a project + session + one `gate` work_event in a freshly migrated DB.
    /// The first inserted `work_events` row gets `work_event_id = 1` (INTEGER
    /// PRIMARY KEY), so the returned gate decision can reference it via FK.
    fn seed_session_and_gate_work_event(db: &Path, session_id: &str) {
        let connection = open_projection_database(db).unwrap();
        let ts = "2026-05-31T00:00:00Z";
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/tmp', ?1, ?1)",
                params![ts],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES (?1, 'p', ?1, 'live', 'unknown', 'idle', ?2, ?2)",
                params![session_id, ts],
            )
            .unwrap();
        let payload = crate::work_event::WorkEventPayload::Gate {
            title: "merge?".into(),
            summary: "approve the merge".into(),
            proposer: "claude".into(),
            actions: vec!["merge".into()],
        };
        project_work_event(&connection, session_id, "claude", ts, 0, &payload).unwrap();
    }

    /// An `ImportedAuditRecord` carrying the decision conventions (ADR 033 D4):
    /// operator-originated, observed + operator-verified, transport=none, the
    /// `gate-decision` record_type, and `ref` pointing at the target work-event.
    /// `previous_audit_id: None` makes this a clean chain root (projects, not Gap).
    fn decision_audit_record_fixture(session_id: &str, audit_id: &str) -> ImportedAuditRecord {
        ImportedAuditRecord {
            audit_id: audit_id.to_string(),
            previous_audit_id: None,
            timestamp: "2026-05-31T01:00:00Z".to_string(),
            source_agent_id: Some("operator".to_string()),
            source_address: "operator".to_string(),
            target_agent_id: None,
            target_address: "none".to_string(),
            transport: "none".to_string(),
            workspace_id: "w".to_string(),
            session_id: session_id.to_string(),
            mid: audit_id.to_string(),
            record_type: "gate-decision".to_string(),
            command_origin: "operator".to_string(),
            mode: None,
            r#ref: Some("1".to_string()),
            re: None,
            payload_hash: "sha256:x".to_string(),
            payload_redaction_policy: "hash-only".to_string(),
            content_size: 0,
            delivery_status: "observed".to_string(),
            observed_by: "operator".to_string(),
            verified_by: "operator".to_string(),
            due: None,
            payload_excerpt: None,
        }
    }

    // ADR 033 D4 / M2a T3: project_decision writes the immutable audit proof row
    // AND the typed operator_decisions row in ONE immediate transaction (atomic).
    #[test]
    fn project_decision_writes_audit_and_typed_row() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("zynk.db");
        open_projection_database(&db).unwrap(); // auto-creates at CURRENT_SCHEMA_VERSION (v7)
        seed_session_and_gate_work_event(&db, "s1");
        let record = decision_audit_record_fixture("s1", "dec-aud-1");
        let decision = crate::decision::Decision::Gate {
            target_work_event_id: 1,
            verdict: "approve".into(),
            note: None,
        };
        project_decision(&db, Path::new("outputs"), &record, &decision).unwrap();

        let connection = open_database(&db).unwrap();
        let audit_rows: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM audit_records WHERE audit_id='dec-aud-1'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            audit_rows, 1,
            "the decision audit proof row must be present"
        );

        let (dt, st, ns, twe, verdict): (String, String, String, i64, String) = connection
            .query_row(
                "SELECT decision_type, decision_status, notification_status,
                        target_work_event_id, verdict
                 FROM operator_decisions WHERE audit_id='dec-aud-1'",
                [],
                |row| {
                    Ok((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(dt, "gate-decision");
        assert_eq!(st, "decided");
        assert_eq!(ns, "not-requested");
        assert_eq!(twe, 1, "the gate decision binds the target work-event");
        assert_eq!(verdict, "approve");
    }

    // ADR 033 D4 / M2a T3: the audit row and the typed row are ONE transaction. If
    // the typed insert fails (here: a dangling target_work_event_id FK), the whole
    // transaction rolls back — the audit proof row must NOT be left committed alone.
    #[test]
    fn project_decision_rolls_back_audit_row_when_typed_insert_fails() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("zynk.db");
        open_projection_database(&db).unwrap();
        // Seed session ONLY — no work_events, so target_work_event_id=1 dangles.
        let connection = open_projection_database(&db).unwrap();
        let ts = "2026-05-31T00:00:00Z";
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/tmp', ?1, ?1)",
                params![ts],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES ('s1', 'p', 's1', 'live', 'unknown', 'idle', ?1, ?1)",
                params![ts],
            )
            .unwrap();
        drop(connection);

        let record = decision_audit_record_fixture("s1", "dec-aud-2");
        let decision = crate::decision::Decision::Gate {
            target_work_event_id: 1, // no such work_event -> FK violation
            verdict: "approve".into(),
            note: None,
        };
        let result = project_decision(&db, Path::new("outputs"), &record, &decision);
        assert!(
            result.is_err(),
            "a dangling target_work_event_id FK must fail the projection"
        );

        let connection = open_database(&db).unwrap();
        let audit_rows: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM audit_records WHERE audit_id='dec-aud-2'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            audit_rows, 0,
            "the audit proof row must roll back with the failed typed insert (atomic)"
        );
    }

    // ADR 033 D4 / M2a T4 (T3-review hardening): if the decision audit row GAPS
    // (e.g. a non-canonical timestamp), `project_decision` must surface a CLEAR
    // error naming the gap reason BEFORE attempting the typed insert — never an
    // opaque "FOREIGN KEY constraint failed" — and write NO operator_decisions row.
    #[test]
    fn project_decision_gap_audit_yields_clear_error_no_typed_row() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("zynk.db");
        open_projection_database(&db).unwrap();
        seed_session_and_gate_work_event(&db, "s1");

        // A deliberately non-canonical timestamp forces project_live_audit_record
        // to return Gap (the normal decide path uses a canonical ts, so it projects).
        let mut record = decision_audit_record_fixture("s1", "dec-aud-gap");
        record.timestamp = "not-a-timestamp".to_string();
        let decision = crate::decision::Decision::Gate {
            target_work_event_id: 1,
            verdict: "approve".into(),
            note: None,
        };
        let err = project_decision(&db, Path::new("outputs"), &record, &decision)
            .expect_err("a gapping decision audit must fail the projection");
        assert!(
            err.message.contains("gap"),
            "the error must name the gap, not surface an opaque FK error: {}",
            err.message
        );

        let connection = open_database(&db).unwrap();
        let decisions: i64 = connection
            .query_row("SELECT COUNT(*) FROM operator_decisions", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(
            decisions, 0,
            "no operator_decisions row may be written when the audit gaps"
        );
        let audit_rows: i64 = connection
            .query_row(
                "SELECT COUNT(*) FROM audit_records WHERE audit_id='dec-aud-gap'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(audit_rows, 0, "a gapping audit row is not projected");
    }

    /// Seed a minimal, valid `audit_records` row for custody tests: a project + a
    /// session (the FK chain) then a NON-sent, NON-decision audit row inserted via a
    /// direct parameterized INSERT. `delivery_status='drafted'` + `verified_by='agent'`
    /// deliberately does NOT trip the ADR 024 sent+agent trigger (which fires only on
    /// `delivery_status='sent'`), so the fixture lands without any decision/work-event.
    fn seed_session_and_audit_record(
        db: &Path,
        session_id: &str,
        audit_id: &str,
        payload_hash: &str,
    ) {
        let connection = open_projection_database(db).unwrap();
        let ts = "2026-05-31T00:00:00Z";
        connection
            .execute(
                "INSERT INTO projects (project_id, name, root_path, created_at, updated_at)
                 VALUES ('p', 'p', '/tmp', ?1, ?1)",
                params![ts],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO sessions (session_id, project_id, title, phase, mode, workflow_status, created_at, updated_at)
                 VALUES (?1, 'p', ?1, 'live', 'unknown', 'idle', ?2, ?2)",
                params![session_id, ts],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO audit_records (
                    audit_id, previous_audit_id, session_id, source_agent_id,
                    target_agent_id, source_address, target_address, transport,
                    workspace_id, mid, record_type, command_origin, mode, ref, re,
                    payload_hash, payload_redaction_policy, content_size,
                    delivery_status, observed_by, verified_by, timestamp
                 )
                 VALUES (?1, NULL, ?2, 'claude', NULL, 'claude', 'none', 'none',
                         'w', ?1, 'note', 'agent', NULL, NULL, NULL,
                         ?3, 'hash-only', 0, 'drafted', 'none', 'agent', ?4)",
                params![audit_id, session_id, payload_hash, ts],
            )
            .unwrap();
    }

    // ADR 034 M3a T3: the custody_vault projection helpers round-trip a ciphertext
    // BLOB keyed by audit_id, and `read_reveal_target` reads the audit record's
    // (session_id, payload_hash) — the reveal proof's session + the disclosure-contract
    // hash that the decrypted plaintext is re-verified against in T5 (one read so the
    // two can never diverge).
    #[test]
    fn custody_vault_insert_and_read() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("zynk.db");
        open_projection_database(&db).unwrap();
        seed_session_and_audit_record(&db, "s1", "aud-1", "sha256:ab");

        let conn = open_projection_database(&db).unwrap();
        insert_custody_vault(
            &conn,
            "aud-1",
            b"CIPHER",
            b"NONCE-24-BYTES-FILLER-XX",
            "xchacha20poly1305",
            1,
            "2026-05-31T00:00:00Z",
        )
        .unwrap();

        let row = read_custody_vault(&conn, "aud-1")
            .unwrap()
            .expect("vault row");
        assert_eq!(row.ciphertext, b"CIPHER");
        assert_eq!(row.nonce, b"NONCE-24-BYTES-FILLER-XX");
        assert_eq!(row.cipher_id, "xchacha20poly1305");
        assert_eq!(row.key_version, 1);

        // read_reveal_target returns BOTH the session_id and the payload_hash for a
        // present row (T5 reveal consumes this exact pair).
        assert_eq!(
            read_reveal_target(&conn, "aud-1").unwrap(),
            Some(("s1".to_string(), "sha256:ab".to_string()))
        );

        // Idempotent on the audit_id PK: a second insert with DIFFERENT bytes is
        // ignored (INSERT OR IGNORE), leaving the original row intact.
        insert_custody_vault(
            &conn,
            "aud-1",
            b"DIFFERENT",
            b"OTHER-NONCE-24-BYTES-FILL",
            "xchacha20poly1305",
            1,
            "2026-05-31T01:00:00Z",
        )
        .unwrap();
        let again = read_custody_vault(&conn, "aud-1").unwrap().expect("row");
        assert_eq!(
            again.ciphertext, b"CIPHER",
            "INSERT OR IGNORE keeps the original"
        );

        // Absent audit_id -> None for both reads.
        assert!(read_custody_vault(&conn, "missing").unwrap().is_none());
        assert!(read_reveal_target(&conn, "missing").unwrap().is_none());
    }
}