zynk 1.1.0

use crate::audit::{self, AuditArgs};
use crate::compose::{build_message, ComposeArgs, ComposedMessage};
use crate::profile::load_profile;
use crate::{CliError, CliResult};
use clap::Args;
use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::process::Command;

#[derive(Debug, Args)]
pub struct SendHerdrArgs {
    #[arg(long, help = "Target herdr pane id.")]
    pub pane: String,
    #[arg(long, help = "Print composed message without sending.")]
    pub dry_run: bool,
    #[arg(long, default_value = "herdr", help = "herdr executable path.")]
    pub herdr_bin: String,
    #[arg(
        long,
        help = "session id; on a successful send this records the sender audit + message corpus automatically (ADR 029) — do not also run zynk audit for the same sent message. Omit for an ad-hoc, unaudited send."
    )]
    pub session_id: Option<String>,
    #[arg(
        long,
        help = "opt out of the audited send: send the message but write no audit/corpus record (ADR 029)."
    )]
    pub no_audit: bool,
    #[arg(
        long,
        default_value = "outputs",
        help = "audit artifact root (independent of --db); the record is written under <root>/sessions/<session-id>/audit.md."
    )]
    pub root: PathBuf,
    #[arg(
        long,
        default_value = "agent",
        value_parser = ["agent", "operator", "helper-tool", "unknown"],
        help = "who originated this send, for the audit record (ADR 029 C5)."
    )]
    pub command_origin: String,
    #[arg(
        long,
        default_value = "full",
        help = "redaction policy for the audited payload; defaults to full so the corpus is queryable (ADR 029 decision 7)."
    )]
    pub payload_redaction_policy: String,
    #[arg(
        long,
        help = "sensitive category; a profile force_hash_only category forces hash-only regardless of --payload-redaction-policy."
    )]
    pub sensitive_category: Option<String>,
    #[arg(
        long,
        help = "live DB path for the audited send's projection; defaults to the cwd .zynk/zynk.db (ADR 028); --no-db forces file-only."
    )]
    pub db: Option<PathBuf>,
    #[arg(
        long,
        help = "force the audited send file-only; skip the DB projection."
    )]
    pub no_db: bool,
    #[arg(
        long,
        help = "ADR 034 D7: ALSO retain the full sent message as recoverable ciphertext in the DB custody_vault (additive to the redacted corpus); requires an audited send (--session-id, not --no-audit/--no-db). A retention failure is LOUD (nonzero), never a silent no-retain."
    )]
    pub retain_custody: bool,
    #[arg(
        long,
        help = "path to the operator-owned custody key file (default <db-dir>/custody.key or $ZYNK_CUSTODY_KEY_FILE); only used with --retain-custody."
    )]
    pub custody_key_file: Option<PathBuf>,
    #[command(flatten)]
    pub compose: ComposeArgs,
}

pub fn run(args: SendHerdrArgs) -> CliResult<()> {
    let profile = load_profile(args.compose.profile.as_deref())?;
    let composed = build_message(&args.compose, &profile)?;

    // ADR 029: --session-id is the tracking signal; --no-audit is the opt-out.
    let audited = args.session_id.is_some() && !args.no_audit;
    if args.db.is_some() && !audited {
        eprintln!(
            "warning: --db has no effect without an audited send (--session-id, and not --no-audit)"
        );
    }

    // ADR 034 D7 + Codex C5: validate --retain-custody EARLY — before the
    // irreversible transport — so a misconfigured retention never sends first.
    if args.retain_custody {
        // C5(1): retention needs an audit record (and thus the DB) to retain
        // against, so it requires an audited send: a --session-id, not --no-audit.
        if !audited {
            return Err(CliError::usage(
                "--retain-custody requires an audited send (a --session-id, not --no-audit)",
            ));
        }
        // C5(2): the vault lives in the DB, so --no-db is incoherent with retention.
        // Reject before transport — never send-then-fail-to-retain.
        if args.no_db {
            return Err(CliError::usage(
                "--retain-custody cannot be combined with --no-db: custody is retained in the DB vault",
            ));
        }
    }

    // --dry-run is unchanged: no send, no audit, no retain (C5(3): nothing was
    // sent/recorded) (ADR 029 how-to-apply step 5).
    if args.dry_run {
        eprintln!("DRY RUN: message was not sent; do not record delivery_status=sent.");
        println!("{}", composed.message);
        return Ok(());
    }

    // ADR 029 hard part 1: build + validate the audit BEFORE the irreversible
    // send. If the audit cannot be formed/validated, fail before sending — we
    // never send a message we cannot then record.
    let audit_args = if audited {
        let session_id = args
            .session_id
            .clone()
            .expect("audited implies --session-id is present");
        let prepared = prepare_audit(&args, &composed, session_id)?;
        audit::validate_audit_args(&prepared, &profile)?;
        Some(prepared)
    } else {
        None
    };

    // Send (the irreversible transport act).
    let output = match Command::new(&args.herdr_bin)
        .args(["pane", "run", &args.pane, &composed.message])
        .output()
    {
        Ok(output) => output,
        Err(error) if error.kind() == io::ErrorKind::NotFound => {
            return Err(CliError::with_code(
                127,
                format!("herdr CLI not found at {}", args.herdr_bin),
            ))
        }
        Err(error) => return Err(CliError::failure(format!("failed to run herdr: {error}"))),
    };
    io::stdout()
        .write_all(&output.stdout)
        .map_err(|error| CliError::failure(format!("failed to write stdout: {error}")))?;
    io::stderr()
        .write_all(&output.stderr)
        .map_err(|error| CliError::failure(format!("failed to write stderr: {error}")))?;

    // ADR 029 hard part 2: transport failure → no `sent` audit (it was not sent).
    if !output.status.success() {
        return Err(CliError::with_code(
            output.status.code().unwrap_or(1),
            "herdr pane run failed",
        ));
    }

    // Transport succeeded. Write the sender audit (file-first then project).
    if let Some(audit_args) = audit_args {
        match audit::write_audit_file(&audit_args, &profile) {
            Ok((audit_path, record)) => {
                audit::project_record(
                    audit_args.db.as_deref(),
                    audit_args.no_db,
                    &audit_args.root,
                    &record,
                )?;
                // ADR 034 D7 + Codex C5(4): the record is durable — now ALSO retain
                // the full sent message as ciphertext in the DB custody_vault. Reuse
                // the EXACT bytes the audit hashed (composed.message — `payload` in
                // prepare_audit), never a re-read/re-render, so the recorded
                // payload_hash and the ciphertext bind identical bytes. A retention
                // failure is ALWAYS LOUD (nonzero): the record is written but custody
                // is NOT retained — never a silent no-retain.
                if args.retain_custody {
                    crate::audit::retain_custody_for(
                        args.db.as_deref(),
                        args.no_db,
                        args.custody_key_file.as_deref(),
                        &record,
                        composed.message.as_bytes(),
                    )
                    .map_err(|error| {
                        CliError::failure(format!(
                            "{}; record written but custody NOT retained: {}",
                            audit_path.display(),
                            error.message
                        ))
                    })?;
                }
            }
            // ADR 029 hard part 3: the message left but the durable record did
            // not land — the one non-atomic seam. Fail loud, never silently 0.
            Err(error) => return Err(integrity_gap(&audit_args, &composed.message, error)),
        }
    }
    Ok(())
}

/// Build the audit record derived from a `send --session-id` invocation
/// (ADR 029 derived-fields table). Pre-send validation lives here so a bad
/// invocation fails before the irreversible send.
fn prepare_audit(
    args: &SendHerdrArgs,
    composed: &ComposedMessage,
    session_id: String,
) -> CliResult<AuditArgs> {
    let from = args.compose.from.as_deref().ok_or_else(|| {
        CliError::usage("audited send (--session-id) requires --from agent:address")
    })?;
    let to = args.compose.to.as_deref().ok_or_else(|| {
        CliError::usage("audited send (--session-id) requires --to agent:address")
    })?;
    let (source_agent, source_address) = split_agent_address(from, "--from")?;
    let (target_agent, target_address) = split_agent_address(to, "--to")?;

    // ADR 029 C2: the audit's target_address MUST reflect the real transport
    // destination, otherwise the record would claim a target that never got it.
    if target_address != args.pane {
        return Err(CliError::usage(format!(
            "--to address ({target_address}) must equal --pane ({}) so the audit records the real destination (ADR 029 C2)",
            args.pane
        )));
    }
    let workspace_id = workspace_from_pane(&args.pane)?;
    let mid = args
        .compose
        .mid
        .clone()
        .ok_or_else(|| CliError::usage("audited send (--session-id) requires --mid"))?;

    // ADR 029 C1: the free-form header --due rides in the payload and is NEVER
    // rejected; audit_records.due is populated only when it parses as RFC3339.
    let due = args
        .compose
        .due
        .as_deref()
        .and_then(crate::timestamp::canonicalize);

    Ok(AuditArgs {
        profile: args.compose.profile.clone(),
        root: args.root.clone(),
        session_id,
        audit_id: None,
        previous_audit_id: None,
        timestamp: None,
        due,
        source_agent: source_agent.clone(),
        source_address,
        target_agent,
        target_address,
        transport: "herdr".to_string(),
        workspace_id,
        transport_thread_id: None,
        mid,
        record_type: composed.message_type.clone(),
        mode: composed.mode.clone(),
        r#ref: args.compose.r#ref.clone(),
        re: args.compose.re.clone(),
        command_origin: args.command_origin.clone(),
        // C4: the audited payload is the exact rendered wire message that was sent.
        payload: Some(composed.message.clone()),
        payload_file: None,
        payload_redaction_policy: args.payload_redaction_policy.clone(),
        payload_ref: None,
        sensitive_category: args.sensitive_category.clone(),
        excerpt_chars: 12,
        // ADR 024: zynk dispatched the transport, so delivery_status=sent is
        // proven by verified_by=helper-tool, never agent self-attestation.
        delivery_status: "sent".to_string(),
        observed_by: source_agent,
        verified_by: "helper-tool".to_string(),
        db: args.db.clone(),
        no_db: args.no_db,
        // ADR 034 D7: the audited send does not expose `--retain-custody` (M3a wires
        // custody only on the `zynk audit` path); the redacted corpus is unchanged.
        retain_custody: false,
        custody_key_file: None,
    })
}

/// Split an `agent:address` value, rejecting empty halves.
fn split_agent_address(value: &str, flag: &str) -> CliResult<(String, String)> {
    match value.split_once(':') {
        Some((agent, address)) if !agent.is_empty() && !address.is_empty() => {
            Ok((agent.to_string(), address.to_string()))
        }
        _ => Err(CliError::usage(format!(
            "{flag} must be agent:address (got {value:?})"
        ))),
    }
}

/// Derive the workspace id from a herdr pane id (`<workspace>-<n>` → `<workspace>`,
/// ADR 029). Doubles as a pane shape check.
fn workspace_from_pane(pane: &str) -> CliResult<String> {
    match pane.rsplit_once('-') {
        Some((workspace, index))
            if !workspace.is_empty()
                && !index.is_empty()
                && index.chars().all(|c| c.is_ascii_digit()) =>
        {
            Ok(workspace.to_string())
        }
        _ => Err(CliError::usage(format!(
            "--pane must look like <workspace>-<n> to derive workspace_id (got {pane:?})"
        ))),
    }
}

/// ADR 029 hard part 3: the message was sent but the audit file write failed.
/// Preserve the exact wire message to a recovery file and reprint the `zynk
/// audit … --payload-file <recovery>` command (same --root) to reconcile.
fn integrity_gap(audit_args: &AuditArgs, message: &str, cause: CliError) -> CliError {
    let recovery = std::env::current_dir()
        .unwrap_or_else(|_| PathBuf::from("."))
        .join(format!(
            "zynk-recovery-{}-{}.txt",
            audit_args.session_id, audit_args.mid
        ));
    let recovery_note = match fs::write(&recovery, message) {
        Ok(()) => format!("preserved the sent message to {}", recovery.display()),
        Err(error) => {
            format!("FAILED to preserve the sent message ({error}); message body below:\n{message}")
        }
    };
    CliError::with_code(
        1,
        format!(
            "INTEGRITY GAP: message was SENT but the audit file write failed: {}\n{}\nreconcile the record with:\n  {}",
            cause.message,
            recovery_note,
            reconcile_command(audit_args, &recovery)
        ),
    )
}

/// POSIX single-quote a value so the reprinted reconcile command is copy/paste
/// safe even when a value (path, etc.) contains spaces or shell metacharacters
/// (ADR 029 hard part 3 — the recovery command must be exact and runnable).
/// Values made only of shell-safe characters pass through unquoted.
fn shell_quote(value: &str) -> String {
    if !value.is_empty()
        && value
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b"-_./=:+,@%".contains(&b))
    {
        value.to_string()
    } else {
        format!("'{}'", value.replace('\'', "'\\''"))
    }
}

/// Reprint the equivalent `zynk audit` command (with the derived fields, same
/// --root) so the operator can reconcile a gapped send from the recovery file.
/// Every interpolated value is shell-quoted so the command is runnable verbatim.
fn reconcile_command(a: &AuditArgs, recovery: &Path) -> String {
    let mut parts = vec![
        "zynk audit".to_string(),
        format!("--root {}", shell_quote(&a.root.display().to_string())),
        format!("--session-id {}", shell_quote(&a.session_id)),
        format!("--source-agent {}", shell_quote(&a.source_agent)),
        format!("--source-address {}", shell_quote(&a.source_address)),
        format!("--target-agent {}", shell_quote(&a.target_agent)),
        format!("--target-address {}", shell_quote(&a.target_address)),
        format!("--transport {}", shell_quote(&a.transport)),
        format!("--workspace-id {}", shell_quote(&a.workspace_id)),
        format!("--mid {}", shell_quote(&a.mid)),
        format!("--type {}", shell_quote(&a.record_type)),
        format!("--command-origin {}", shell_quote(&a.command_origin)),
        format!(
            "--payload-redaction-policy {}",
            shell_quote(&a.payload_redaction_policy)
        ),
        "--delivery-status sent".to_string(),
        format!("--observed-by {}", shell_quote(&a.observed_by)),
        "--verified-by helper-tool".to_string(),
        format!(
            "--payload-file {}",
            shell_quote(&recovery.display().to_string())
        ),
    ];
    // Preserve the original projection mode so recovery lands where the send
    // intended (R2 P3): --no-db stays file-only, an explicit --db targets the
    // same DB, default is unchanged. Mirrors resolve_projection_target precedence
    // (--no-db wins over --db).
    if a.no_db {
        parts.push("--no-db".to_string());
    } else if let Some(db) = &a.db {
        parts.push(format!("--db {}", shell_quote(&db.display().to_string())));
    }
    if let Some(value) = &a.re {
        parts.push(format!("--re {}", shell_quote(value)));
    }
    if let Some(value) = &a.r#ref {
        parts.push(format!("--ref {}", shell_quote(value)));
    }
    if let Some(value) = &a.mode {
        parts.push(format!("--mode {}", shell_quote(value)));
    }
    if let Some(value) = &a.due {
        parts.push(format!("--due {}", shell_quote(value)));
    }
    parts.join(" ")
}

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

    #[test]
    fn shell_quote_passes_safe_values_and_quotes_the_rest() {
        // Shell-safe values pass through unquoted for readable output.
        assert_eq!(shell_quote("outputs"), "outputs");
        assert_eq!(shell_quote("/a/b-c_d.e:1"), "/a/b-c_d.e:1");
        // Empty and space/metachar values are single-quoted.
        assert_eq!(shell_quote(""), "''");
        assert_eq!(shell_quote("with space"), "'with space'");
        assert_eq!(shell_quote("a;rm -rf b"), "'a;rm -rf b'");
        // An embedded single quote is closed, escaped, and reopened: ' -> '\''
        assert_eq!(shell_quote("a'b"), "'a'\\''b'");
    }
}