zynk 1.4.1

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::collections::BTreeMap;
use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Output};

#[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).
    dispatch_herdr_message(&args, &composed.message)?;

    // 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(())
}

/// ADR 041 D2/D3/D3a: a snapshot of what the receiver pane shows — the count of
/// the exact rendered header marker (a delivered message), and the layout-aware
/// classification of the active input box.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReceiverState {
    pub marker_count: usize,
    pub input: InputClass,
}

/// ADR 041 D3 / D3a: the layout-aware classification of the receiver's input box.
/// `Recognized(_)` is a *confidently* recognized Claude-Code input box (the bottom
/// rule-pair structure WITH a `❯` (U+276F) prompt line) — D3 applies exactly.
/// `Unrecognized` is any other layout (Codex/GPT's `›`-after-the-final-rule + ghost
/// placeholder, shells, other TUIs): UNINSPECTABLE, not "safe" — zynk neither
/// preflight-aborts on it nor ever types into it; only the rendered marker gates
/// `sent` (D2).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InputClass {
    Recognized(InputState),
    Unrecognized,
}

/// ADR 041 D3: the classified content of a RECOGNIZED Claude-Code input box.
/// `Empty` = nothing pending; `OneChip` = exactly one active paste chip and nothing
/// else (zynk's own chip — safe to submit with one Enter only when the preflight
/// state was Empty); `Queued` = the receiver TUI has accepted messages into its
/// own queue ("Messages to be submitted after next tool call" / "Press up to edit
/// queued messages") — delivery can be verified by the marker, but zynk must not
/// press Enter into that queue; `Other` = ambiguous/non-empty input (typed text, a
/// chip plus other text, or two+ chips) — never auto-submit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InputState {
    Empty,
    OneChip,
    Queued,
    Other,
}

/// Parse a rendered `[herdr ...]` header into field/value pairs. Header values are
/// protocol routing tokens and contain no whitespace; if a terminal hard-wrap
/// splits a token, callers must join that wrap before parsing.
fn parse_herdr_header_fields(header: &str) -> Option<BTreeMap<String, String>> {
    let inner = header.trim().strip_prefix("[herdr ")?.strip_suffix(']')?;
    let mut fields = BTreeMap::new();
    for token in inner.split_whitespace() {
        let (key, value) = token.split_once('=')?;
        if key.is_empty() || value.is_empty() {
            return None;
        }
        if fields.insert(key.to_string(), value.to_string()).is_some() {
            return None;
        }
    }
    Some(fields)
}

fn collapse_header_whitespace(header: &str) -> String {
    header.split_whitespace().collect::<Vec<_>>().join(" ")
}

fn join_hard_wrapped_header(header: &str) -> String {
    let mut joined = String::new();
    for (idx, line) in header.lines().enumerate() {
        if idx == 0 {
            joined.push_str(line);
        } else {
            joined.push_str(line.trim_start());
        }
    }
    if joined.is_empty() {
        header.to_string()
    } else {
        joined
    }
}

fn header_fields_match(
    candidate: &BTreeMap<String, String>,
    expected: &BTreeMap<String, String>,
) -> bool {
    for required in ["from", "to", "mid", "type"] {
        if candidate.get(required) != expected.get(required) {
            return false;
        }
    }
    expected
        .iter()
        .all(|(key, value)| candidate.get(key) == Some(value))
}

fn header_candidate_matches(candidate: &str, expected: &BTreeMap<String, String>) -> bool {
    let variants = [
        candidate.to_string(),
        collapse_header_whitespace(candidate),
        join_hard_wrapped_header(candidate),
    ];
    variants.iter().any(|variant| {
        parse_herdr_header_fields(variant)
            .as_ref()
            .is_some_and(|fields| header_fields_match(fields, expected))
    })
}

/// Count rendered header-marker occurrences anywhere in a
/// `--source recent-unwrapped` dump (ADR 041 D2). The delivered header lives in
/// the conversation scrollback, so it is counted across the whole transcript.
///
/// Herdr/terminal reads may hard-wrap a long `[herdr ...]` header across lines
/// even under `recent-unwrapped` (for example splitting `re=...-1` into
/// `re=...-\n  1`). Matching the raw marker string would false-negative on a
/// delivered message. Keep the proof as the structured `[herdr ...]` header, not
/// `mid` alone: each candidate header must match all expected header fields, with
/// `from`/`to`/`mid`/`type` required and all optional fields present in the sent
/// marker also equal.
fn parse_marker_count(recent: &str, marker: &str) -> usize {
    if marker.is_empty() {
        return 0;
    }
    let Some(expected) = parse_herdr_header_fields(marker) else {
        return 0;
    };

    let mut count = 0;
    let mut cursor = 0;
    while let Some(relative_start) = recent[cursor..].find("[herdr ") {
        let start = cursor + relative_start;
        let Some(relative_end) = recent[start..].find(']') else {
            break;
        };
        let end = start + relative_end + 1;
        if header_candidate_matches(&recent[start..end], &expected) {
            count += 1;
        }
        cursor = end;
    }
    count
}

/// Is this line a Claude-TUI input-box horizontal rule — a run of U+2500 `─`
/// box-drawing chars (≥8), and nothing but `─`/spaces?
fn is_input_box_rule(line: &str) -> bool {
    let trimmed = line.trim();
    !trimmed.is_empty()
        && trimmed.chars().all(|c| c == '─' || c == ' ')
        && trimmed.chars().filter(|&c| c == '─').count() >= 8
}

/// Does this line's first non-whitespace char equal the Claude-Code prompt glyph
/// `❯` (U+276F)? This is the D3a recognition signal.
fn starts_with_claude_prompt(line: &str) -> bool {
    line.trim_start().starts_with('\u{276f}')
}

fn is_queued_message_notice(text: &str) -> bool {
    text.contains("Messages to be submitted after next tool call")
        || text.contains("Press up to edit queued messages")
        || region_renders_herdr_header(text)
}

/// Does the cleaned input region render a real `[herdr ...]` protocol header — the
/// `[herdr ` opener followed by all four required routing keys (`from`/`to`/`mid`/
/// `type`)? When a working receiver accepts a message into its own queue, the
/// active region can briefly show the rendered protocol message before the
/// "Press up.../Messages to be submitted..." notice stabilizes. Keying on the real
/// header — not a loose `[from-` / `via herdr]` substring triple — is robust to the
/// header hard-wrapping, to leading transcript text, and to the `[from-… via herdr]`
/// envelope scrolling off, while rejecting prose that merely mentions `[herdr`.
/// NOTE: still best-effort `--source visible` scraping; the durable fix is Herdr
/// exposing structured active-input state (ADR 041 D8).
fn region_renders_herdr_header(text: &str) -> bool {
    let Some(start) = text.find("[herdr ") else {
        return false;
    };
    let header = &text[start..];
    header.contains("from=")
        && header.contains("to=")
        && header.contains("mid=")
        && header.contains("type=")
}

/// Layout-aware classification of the ACTIVE input box of a `--source visible` dump
/// (ADR 041 D3a). The box is RECOGNIZED only when it is *confidently* a Claude-Code
/// input box: there are ≥2 horizontal-rule lines AND the region strictly between
/// the LAST TWO rules contains a line whose first non-whitespace char is `❯`
/// (U+276F). Any other layout (fewer than two rules — e.g. a shell — OR no
/// `❯`-prompt line in the region, e.g. Codex/GPT's `›`-after-the-final-rule) is
/// `Unrecognized`: zynk treats it as UNINSPECTABLE (never typed into; only the
/// marker gates `sent`), NOT as "safe/empty".
///
/// For a recognized box, the region is cleaned by stripping the prompt glyph(s)
/// (U+276F `❯`, and a leading ASCII `>`) and surrounding whitespace, then
/// classified: empty → `Empty`; receiver-owned queue notices → `Queued`; exactly
/// one `[Pasted text…]` chip and nothing else → `OneChip`; anything else (typed
/// text, chip + other text, two+ chips) → `Other`.
fn classify_input(visible: &str) -> InputClass {
    let lines: Vec<&str> = visible.lines().collect();
    let rules: Vec<usize> = lines
        .iter()
        .enumerate()
        .filter(|(_, l)| is_input_box_rule(l))
        .map(|(i, _)| i)
        .collect();
    if rules.len() < 2 {
        return InputClass::Unrecognized;
    }
    let (top, bottom) = (rules[rules.len() - 2], rules[rules.len() - 1]);
    let region_lines = &lines[top + 1..bottom];

    // D3a recognition: the region must carry a Claude-Code `❯` prompt line.
    if !region_lines.iter().any(|l| starts_with_claude_prompt(l)) {
        return InputClass::Unrecognized;
    }
    let region = region_lines.join(" ");

    // Strip the prompt glyph(s) and all surrounding whitespace.
    let cleaned = region
        .replace('\u{276f}', " ") // ❯
        .trim()
        .trim_start_matches('>')
        .trim()
        .to_string();

    if cleaned.is_empty() {
        return InputClass::Recognized(InputState::Empty);
    }
    if is_queued_message_notice(&cleaned) {
        return InputClass::Recognized(InputState::Queued);
    }
    // Exactly one paste chip and nothing else → OneChip. The chip token runs from
    // "[Pasted text" through its next ']'; if removing it leaves only whitespace,
    // the sole pending artifact is that one chip.
    if let Some(rest) = cleaned.strip_prefix("[Pasted text") {
        if let Some(close) = rest.find(']') {
            if rest[close + 1..].trim().is_empty() {
                return InputClass::Recognized(InputState::OneChip);
            }
        }
    }
    InputClass::Recognized(InputState::Other)
}

/// Extract the exact rendered `[herdr ...]` header segment of the composed
/// message — the delivery marker (ADR 041 D2), unique by mid. The composed wire
/// message is `[from-X via herdr] [herdr from=… to=… mid=… type=…] BODY: …`; the
/// marker is the whole `[herdr …]` segment, NOT the mid alone.
fn delivery_marker(composed_message: &str) -> CliResult<String> {
    let start = composed_message
        .find("[herdr ")
        .ok_or_else(|| CliError::failure("composed message has no [herdr ...] header to verify"))?;
    let rel_end = composed_message[start..]
        .find(']')
        .ok_or_else(|| CliError::failure("composed message [herdr ...] header is unterminated"))?;
    Ok(composed_message[start..=start + rel_end].to_string())
}

fn dispatch_herdr_message(args: &SendHerdrArgs, message: &str) -> CliResult<()> {
    // ADR 041 D1: verify receiver-side submission BEFORE run() writes the `sent`
    // audit. verify_delivery owns the preflight → pane run → verify state machine;
    // an Err here returns before run()'s audit write, so `delivery_status=sent` is
    // never recorded on `pane run` exit alone.
    verify_delivery(args, message)
}

/// Read an unsigned millisecond knob from the environment, else the default.
/// Lets tests crank the verify timeout/poll down so the timeout path is fast.
fn env_u64(var: &str, default: u64) -> u64 {
    std::env::var(var)
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(default)
}

/// Read the receiver pane into a `ReceiverState` (ADR 041 D2/D3). Two reads with
/// different scopes:
/// - marker_count from `--source recent-unwrapped --lines 400`: the delivered
///   header lives in the conversation scrollback, and the wide window is needed to
///   scrape a submitted Claude-TUI header (40 lines missed it, 300 found it).
/// - input from `--source visible`: the ACTIVE input box is classified ONLY there
///   — never from scrollback prose that merely mentions "[Pasted text" (Codex R1
///   P2), and including ordinary typed text, not just paste chips (Codex R2 P1).
fn read_state(herdr_bin: &str, pane: &str, marker: &str) -> CliResult<ReceiverState> {
    let recent = run_herdr_command(
        herdr_bin,
        &[
            "pane",
            "read",
            pane,
            "--source",
            "recent-unwrapped",
            "--lines",
            "400",
        ],
    )?;
    ensure_herdr_success(&recent, "herdr pane read failed")?;
    let marker_count = parse_marker_count(&String::from_utf8_lossy(&recent.stdout), marker);

    let visible = run_herdr_command(herdr_bin, &["pane", "read", pane, "--source", "visible"])?;
    ensure_herdr_success(&visible, "herdr pane read failed")?;
    let input = classify_input(&String::from_utf8_lossy(&visible.stdout));

    Ok(ReceiverState {
        marker_count,
        input,
    })
}

/// ADR 041 D1–D5 + D3a: prove receiver-side delivery before the caller records
/// `sent`. Detection is LAYOUT-AWARE (D3a): the receiver's input box is classified
/// once from the baseline read (it is stable), and the flow branches on it.
///
/// RECOGNIZED (a confidently-recognized Claude-Code `❯` input box) — D3 exactly,
/// with receiver-owned queues separated from mutable input: preflight aborts if
/// the baseline box holds a pre-existing paste chip OR ordinary typed text. A
/// queued-message notice is allowed because the receiver TUI has already accepted
/// inbound messages into its own queue; a new marker verifies queued delivery, but
/// zynk must not press Enter into that queue. After `pane run`, a new marker
/// verifies with `Empty` OR `Queued` (D2/D5); a single active chip (zynk's own) is
/// submitted with exactly ONE Enter ONLY when the preflight state was `Empty` (D3);
/// any other non-empty input → fail loud with NO Enter (D3); timeout → fail (D4).
///
/// UNRECOGNIZED (Codex/GPT, shells, other TUIs — D3a) — UNINSPECTABLE, not "safe":
/// NO preflight abort, NO input-empty requirement, and zynk NEVER presses an
/// Enter/send-keys into it. After `pane run`, poll for the marker only: a new
/// marker beyond baseline → `sent` (D2); timeout with no new marker → fail (D4).
///
/// In BOTH branches `sent` is written ONLY on a verified new marker, and any Err
/// returns before the audit so a failed verification writes no audit/corpus (D4).
fn verify_delivery(args: &SendHerdrArgs, message: &str) -> CliResult<()> {
    let marker = delivery_marker(message)?;

    // Fail-closed on a bad multiline payload BEFORE any herdr interaction: the
    // ESC/C1 CSI guard is a pure check on the message, so a rejected payload never
    // touches the receiver (no preflight read, no transport).
    if message.contains('\n') {
        validate_bracketed_paste_payload(message)?;
    }

    let baseline = read_state(&args.herdr_bin, &args.pane, &marker)?;

    // D3a: the layout is classified once from the baseline (it is stable) and gates
    // whether zynk is allowed to inspect/auto-submit the input box at all.
    let recognized = matches!(baseline.input, InputClass::Recognized(_));

    // D3 preflight (RECOGNIZED layouts only): refuse to send into pre-existing
    // mutable input — a paste chip OR ordinary typed operator text (Codex R2 P1).
    // A receiver-owned queued-message notice is not mutable input; zynk may send
    // into that queue but must never auto-Enter it. UNRECOGNIZED layouts are
    // uninspectable (D3a): no preflight abort, no input-empty requirement.
    let baseline_empty = baseline.input == InputClass::Recognized(InputState::Empty);
    let baseline_queued = baseline.input == InputClass::Recognized(InputState::Queued);
    if recognized && !baseline_empty && !baseline_queued {
        return Err(CliError::failure(format!(
            "receiver pane {} has pre-existing pending input; refusing to send (ADR 041 D3)",
            args.pane
        )));
    }

    let run = run_herdr_command(&args.herdr_bin, &["pane", "run", &args.pane, message])?;
    write_child_output(&run)?;
    ensure_herdr_success(&run, "herdr pane run failed")?;

    // D2/D3/D5 verify.
    let timeout = std::time::Duration::from_millis(env_u64("ZYNK_VERIFY_TIMEOUT_MS", 4000));
    let poll = std::time::Duration::from_millis(env_u64("ZYNK_VERIFY_POLL_MS", 200));
    let deadline = std::time::Instant::now() + timeout;
    let mut entered = false;
    loop {
        let state = read_state(&args.herdr_bin, &args.pane, &marker)?;
        let new_marker = state.marker_count > baseline.marker_count;

        if recognized {
            // RECOGNIZED (D3/D5): a new marker verifies delivery when the input
            // box is empty OR the receiver has accepted the message into its own
            // queued-message area. A marker that coexists with mutable input
            // (Codex R1 P1 / R2 P1) is NOT delivered yet.
            if new_marker
                && matches!(
                    state.input,
                    InputClass::Recognized(InputState::Empty | InputState::Queued)
                )
            {
                return Ok(());
            }
            match state.input {
                // D3: exactly one active paste chip (zynk's own, since the preflight
                // found the box empty) — submit it with exactly ONE Enter, at most once.
                InputClass::Recognized(InputState::OneChip) if baseline_empty && !entered => {
                    let keyed = run_herdr_command(
                        &args.herdr_bin,
                        &["pane", "send-keys", &args.pane, "Enter"],
                    )?;
                    ensure_herdr_success(&keyed, "herdr pane send-keys failed")?;
                    entered = true;
                }
                InputClass::Recognized(InputState::OneChip) if !baseline_empty => {
                    return Err(CliError::failure(format!(
                        "receiver pane {} shows a pending paste chip that is not safe to auto-submit (ADR 041 D3)",
                        args.pane
                    )));
                }
                // D3 + queue transient (operator-confirmed live): a post-run Other can
                // be a MOMENTARY ambiguous state while a working receiver ingests zynk's
                // paste into its own message queue — the message is being queued
                // (delivered), not a mutable operator draft and not a failed send. zynk
                // never auto-Enters an Other state, so there is no "submit the wrong
                // thing" risk in waiting; keep polling until it settles to a verified
                // delivery (a new marker + Empty|Queued) or the D4 timeout. A genuinely
                // stable ambiguous input simply times out (fail loud, NO audit/corpus,
                // NO Enter) rather than failing on the transient.
                InputClass::Recognized(InputState::Other) => {}
                _ => {}
            }
        } else {
            // UNRECOGNIZED (D3a): zynk never inspects or types into the layout — the
            // rendered marker is the sole proof. A new marker beyond baseline = sent.
            if new_marker {
                return Ok(());
            }
        }

        if std::time::Instant::now() >= deadline {
            // D4/D5: no verified delivery within the timeout — do NOT record sent.
            return Err(CliError::failure(format!(
                "delivery to pane {} unverified within timeout; not recording sent (ADR 041 D1/D4)",
                args.pane
            )));
        }
        std::thread::sleep(poll);
    }
}

fn validate_bracketed_paste_payload(message: &str) -> CliResult<()> {
    if message.contains('\u{1b}') || message.contains('\u{009b}') {
        return Err(CliError::failure(
            "multiline Herdr messages may not contain terminal control bytes (ESC or C1 CSI)",
        ));
    }
    Ok(())
}

fn run_herdr_command(herdr_bin: &str, args: &[&str]) -> CliResult<Output> {
    match Command::new(herdr_bin).args(args).output() {
        Ok(output) => Ok(output),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Err(CliError::with_code(
            127,
            format!("herdr CLI not found at {herdr_bin}"),
        )),
        Err(error) => Err(CliError::failure(format!("failed to run herdr: {error}"))),
    }
}

fn write_child_output(output: &Output) -> CliResult<()> {
    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}")))?;
    Ok(())
}

fn ensure_herdr_success(output: &Output, message: &str) -> CliResult<()> {
    if output.status.success() {
        Ok(())
    } else {
        Err(CliError::with_code(
            output.status.code().unwrap_or(1),
            message,
        ))
    }
}

/// 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::{
        classify_input, delivery_marker, parse_marker_count, shell_quote, InputClass, InputState,
    };

    /// A real `herdr pane read --source visible` capture of a live Claude-Code pane
    /// (ground truth): its scrollback prose mentions "[Pasted text" twice, but the
    /// `❯`-box between the last two `─` rules is empty → Recognized(Empty).
    const REAL_CLAUDE_VISIBLE: &str =
        include_str!("../tests/fixtures/adr041-claude-pane-visible.txt");

    /// A real `herdr pane read --source visible` capture of a live Codex/GPT pane
    /// (ADR 041 D3a ground truth): it has section rules INSIDE the transcript and
    /// its real prompt is `›` (U+203A) AFTER the final rule with NO `❯`, so the
    /// region between the last two rules is transcript content — Unrecognized.
    const REAL_CODEX_VISIBLE: &str =
        include_str!("../tests/fixtures/adr041-codex-pane-visible.txt");

    // ADR 041 D2: parse_marker_count counts exact-marker occurrences (a delivered
    // header beyond baseline) anywhere in the recent-unwrapped scrollback. An empty
    // marker never matches.
    #[test]
    fn parse_marker_count_counts_header_occurrences() {
        let marker = "[herdr from=a to=b mid=m1 type=ack]";
        assert_eq!(parse_marker_count("", marker), 0);
        let twice = format!("line\n{marker}\nmiddle\n{marker}\ntail\n");
        assert_eq!(parse_marker_count(&twice, marker), 2);
        // A chip in scrollback never contributes to the marker count.
        assert_eq!(
            parse_marker_count("[Pasted text #1 +9 lines]\nnothing here\n", marker),
            0
        );
        // An empty marker never matches (guards the "no header" case).
        assert_eq!(parse_marker_count("anything", ""), 0);
    }

    #[test]
    fn parse_marker_count_matches_hard_wrapped_header_value() {
        let marker = "[herdr from=claude:w652eed593568a3-3 to=codex:w652eed593568a3-1 mid=smscode-claude-zynk-v14-ack type=ack ref=zynk-v1.4.0-release re=smscode-zynk-v14-ack-claude-1 mode=status]";
        let wrapped = "\
› [from-claude via herdr] [herdr from=claude:w652eed593568a3-3 to=codex:w652eed593568a3-1 mid=smscode-claude-zynk-v14-ack type=ack ref=zynk-v1.4.0-release re=smscode-zynk-v14-ack-claude-
  1 mode=status] BODY: delivered text
";
        assert_eq!(
            parse_marker_count(wrapped, marker),
            1,
            "a terminal hard-wrap inside a header value must not hide a delivered marker"
        );
    }

    #[test]
    fn parse_marker_count_matches_header_wrapped_between_fields() {
        let marker = "[herdr from=a to=b mid=m1 type=ack ref=topic re=parent mode=review]";
        let wrapped = "\
[from-a via herdr] [herdr from=a to=b mid=m1 type=ack
  ref=topic re=parent mode=review] BODY: ok
";
        assert_eq!(parse_marker_count(wrapped, marker), 1);
    }

    #[test]
    fn parse_marker_count_requires_structured_field_match_not_mid_only() {
        let marker = "[herdr from=a to=b mid=m1 type=ack ref=topic]";
        assert_eq!(
            parse_marker_count("[herdr from=x to=b mid=m1 type=ack ref=topic]", marker),
            0,
            "wrong sender must not verify"
        );
        assert_eq!(
            parse_marker_count("[herdr from=a to=b mid=m1 type=ack]", marker),
            0,
            "missing expected optional fields must not verify"
        );
        assert_eq!(
            parse_marker_count(
                "[herdr from=x from=a to=b mid=m1 type=ack ref=topic]",
                marker
            ),
            0,
            "malformed duplicate fields must not verify"
        );
        assert_eq!(
            parse_marker_count("plain prose mentions mid=m1 but has no header", marker),
            0,
            "mid-only prose is not delivery proof"
        );
    }

    // ADR 041 D3a ground-truth (two-fixture recognition proof): the REAL Claude-Code
    // pane is Recognized(Empty) — its `❯` box is empty despite historical "[Pasted
    // text" prose in scrollback.
    #[test]
    fn classify_input_recognizes_real_claude_box_as_empty() {
        assert!(
            REAL_CLAUDE_VISIBLE.matches("[Pasted text").count() >= 2,
            "the fixture must actually contain historical prose chips to be a real test"
        );
        assert_eq!(
            classify_input(REAL_CLAUDE_VISIBLE),
            InputClass::Recognized(InputState::Empty),
            "the real Claude-Code `❯` input box is recognized and empty"
        );
    }

    // ADR 041 D3a ground-truth: the REAL Codex/GPT pane is Unrecognized — no
    // `❯`-prompt line in the region between the last two rules (its prompt is `›`
    // after the final rule). zynk must NOT false-abort on it.
    #[test]
    fn classify_input_unrecognized_for_real_codex_pane() {
        assert!(
            !REAL_CODEX_VISIBLE.contains('\u{276f}'),
            "the Codex fixture must have no `❯` glyph to be a real Unrecognized test"
        );
        assert_eq!(
            classify_input(REAL_CODEX_VISIBLE),
            InputClass::Unrecognized,
            "a Codex/GPT layout (no `❯`-prompt in the rule region) is Unrecognized, not a false abort"
        );
    }

    // A recognized Claude-Code box with a bare `❯` prompt → Recognized(Empty).
    #[test]
    fn classify_input_recognized_empty_for_bare_prompt() {
        let visible = "\
some scrollback
────────────────────────
❯
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Empty)
        );
    }

    // Synthetic Claude-Code Recognized(OneChip): exactly one paste chip, nothing else.
    #[test]
    fn classify_input_recognized_one_chip_for_sole_paste_chip() {
        let visible = "\
some scrollback
────────────────────────
❯ [Pasted text #1 +9 lines]
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::OneChip)
        );
    }

    #[test]
    fn classify_input_recognized_queued_for_press_up_notice() {
        let visible = "\
prior
────────────────────────
❯ Press up to edit queued messages
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Queued)
        );
    }

    #[test]
    fn classify_input_recognized_queued_for_next_tool_call_notice() {
        let visible = "\
prior
────────────────────────
❯ Messages to be submitted after next tool call (press esc to interrupt and send immediately)
  ↳ LIKE THIS
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Queued)
        );
    }

    #[test]
    fn classify_input_recognized_queued_for_rendered_herdr_message() {
        let visible = "\
prior
────────────────────────
❯ [from-codex via herdr] [herdr from=codex:w1 to=claude:w2 mid=m1 type=status-update ref=r1 re=p1 mode=validate] BODY: queued while working
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Queued)
        );
    }

    #[test]
    fn classify_input_recognized_queued_for_rendered_herdr_message_after_transcript() {
        let visible = "\
prior
────────────────────────
· v1.4.1 review in progress
  ⎿  details from the active turn
❯ [from-codex via herdr] [herdr from=codex:w1 to=claude:w2 mid=m1 type=status-update ref=r1 re=p1 mode=validate] BODY: queued while working
❯ Press up to edit queued messages
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Queued)
        );
    }

    // Synthetic Claude-Code Recognized(Other): plain typed operator text (must abort).
    #[test]
    fn classify_input_recognized_other_for_plain_typed_text() {
        let visible = "\
prior
────────────────────────
❯ draft text
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Other)
        );
    }

    // Precision: typed/pasted PROSE that merely mentions `[herdr` (no real protocol
    // header — missing the routing keys) must stay Other (abort), not be mistaken
    // for a receiver-owned queue. Tighter than the prior `[from-`/`via herdr]`
    // substring triple, which a review discussion like this could trip.
    #[test]
    fn classify_input_recognized_other_for_prose_mentioning_herdr_without_header() {
        let visible = "\
prior
────────────────────────
❯ remember to check the [herdr ...] marker in parse_marker_count and from= handling
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Other)
        );
    }

    // A chip PLUS extra typed text → Recognized(Other) (the one Enter could submit
    // the wrong thing).
    #[test]
    fn classify_input_recognized_other_for_chip_plus_text() {
        let visible = "\
prior
────────────────────────
❯ [Pasted text #1 +9 lines] and typed text
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Other)
        );
    }

    // Two active chips → Recognized(Other) (ambiguous; subsumes the old >1 case).
    #[test]
    fn classify_input_recognized_other_for_two_chips() {
        let visible = "\
prior
────────────────────────
❯ [Pasted text #1 +9 lines]
  [Pasted text #2 +3 lines]
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Other)
        );
    }

    // A shell receiver with no TUI input box (fewer than two `─` rule lines) →
    // Unrecognized (D3a: uninspectable, not "safe"); only the marker gates sent.
    #[test]
    fn classify_input_unrecognized_without_input_box() {
        let visible = "$ ls\nfile1 file2\n[Pasted text #1 +9 lines] mentioned in output\n";
        assert_eq!(classify_input(visible), InputClass::Unrecognized);
    }

    // A box that HAS two rules but whose prompt is `›` (U+203A, Codex-style) and no
    // `❯` in the region → Unrecognized.
    #[test]
    fn classify_input_unrecognized_for_non_claude_prompt_in_box() {
        let visible = "\
prior
────────────────────────
› ghost placeholder text
────────────────────────
";
        assert_eq!(classify_input(visible), InputClass::Unrecognized);
    }

    // marker + historical prose in a recognized box: the marker is counted from
    // recent-unwrapped, while the same scrollback "[Pasted text" never blocks
    // delivery because the `❯` input box is Recognized(Empty).
    #[test]
    fn marker_counted_while_historical_prose_does_not_block() {
        let marker = "[herdr from=a to=b mid=m1 type=ack]";
        let recent = format!("{marker}\nold: [Pasted text #1 +9 lines] discussed\n");
        assert_eq!(parse_marker_count(&recent, marker), 1);
        // The visible read has prose chips only outside the (empty) input box.
        let visible = "\
talk about [Pasted text #1 +9 lines]
────────────────────────
❯
────────────────────────
";
        assert_eq!(
            classify_input(visible),
            InputClass::Recognized(InputState::Empty)
        );
    }

    // ADR 041 D2: the delivery marker is the exact `[herdr ...]` header segment of
    // the composed message — NOT mid alone.
    #[test]
    fn delivery_marker_extracts_the_herdr_header_segment() {
        let composed = "[from-claude via herdr] [herdr from=a to=b mid=m1 type=ack] BODY: hi";
        assert_eq!(
            delivery_marker(composed).unwrap(),
            "[herdr from=a to=b mid=m1 type=ack]"
        );
    }

    #[test]
    fn delivery_marker_errors_without_a_header() {
        let err = delivery_marker("no header here BODY: hi").unwrap_err();
        assert!(
            err.message.contains("no [herdr"),
            "missing header must fail loud: {}",
            err.message
        );
    }

    #[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'");
    }
}