lifeloop-cli 0.5.0

//! `lifeloop continuation` — cross-restart key/blob store per `CONTINUATION-001`.
//!
//! Five subcommands:
//! - `put` — write a blob (bytes on stdin) + metadata
//! - `get` — read a blob (bytes on stdout, meta on stderr); honors TTL
//! - `drop` — delete a blob (idempotent — exit 0 even if absent)
//! - `list` — list all blob keys for a thread
//! - `drop-thread` — delete all blobs for a thread
//!
//! ## Storage layout (per spec amendment)
//!
//! Each entry is a single combined file:
//!
//! ```text
//! $XDG_STATE_HOME/lifeloop/continuation/<thread-id>/<key>
//!   ↑
//!   [ 4 bytes u32 BE: META_LEN ][ META_LEN bytes meta JSON ][ blob bytes... ]
//! ```
//!
//! Single-file framing exists because the spec's "prior pair or no
//! pair, never partial" atomicity guarantee is unachievable with two
//! files (POSIX `rename(2)` is atomic per-path, not pair-atomic).
//! Combining both into one file makes the atomic operation a single
//! `rename(2)`. See the spec body for the full rationale.
//!
//! The CLI surface is unchanged: `get` splits the framing back into
//! blob bytes (stdout) and meta JSON (stderr) — callers never observe
//! the on-disk framing.

use std::fs;
use std::io::{Read, Write};
#[cfg(unix)]
use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};
use serde_json::json;

use super::CliError;

/// Schema version pinned to the metadata JSON (per spec).
const SCHEMA_VERSION: &str = "lifeloop.continuation.v0.1";

/// Default blob size cap. Configurable via `$LIFELOOP_CONTINUATION_MAX_BYTES`
/// if an operator needs to tune. 16 MiB matches the stdin cap used by other
/// lifeloop subcommands (see `cli::MAX_STDIN_BYTES`).
const DEFAULT_BLOB_MAX_BYTES: u64 = 16 * 1024 * 1024;

/// Maximum allowed identifier length. Long enough to fit any realistic
/// thread/key (ULIDs are 26 chars; UUIDs are 36); short enough to keep
/// path lengths reasonable.
const MAX_IDENTIFIER_LEN: usize = 128;

// ---------------------------------------------------------------------------
// Public dispatch
// ---------------------------------------------------------------------------

pub fn run<I: Iterator<Item = String>>(mut args: I) -> Result<(), CliError> {
    let action = args.next().ok_or_else(|| {
        CliError::Usage(
            "continuation requires a subcommand: put | get | drop | list | drop-thread".into(),
        )
    })?;
    match action.as_str() {
        "put" => run_put(args),
        "get" => run_get(args),
        "drop" => run_drop(args),
        "list" => run_list(args),
        "drop-thread" => run_drop_thread(args),
        other => Err(CliError::Usage(format!(
            "continuation: unknown subcommand `{other}`; want put|get|drop|list|drop-thread"
        ))),
    }
}

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

/// On-disk metadata schema. `compose_framed` is the sole writer and
/// always serializes `ttl_s` (the field has no `skip_serializing_if`,
/// so the key is emitted as `<u64-or-null>` on every write). On read,
/// serde maps a missing-or-null `ttl_s` to `None`. A key-absent file
/// can only arise from hand-corruption, which `parse_framed` already
/// surfaces as `storage_failure` through its truncation / JSON-parse
/// guards — so the plain `Option<u64>` shape is sufficient.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
struct Meta {
    schema_version: String,
    client_id: String,
    written_at_epoch_s: u64,
    #[serde(default)]
    ttl_s: Option<u64>,
}

#[derive(Debug)]
struct CommonArgs {
    thread: String,
    key: String,
}

// ---------------------------------------------------------------------------
// Arg parsing
// ---------------------------------------------------------------------------

fn parse_thread_and_key<I: Iterator<Item = String>>(
    args: &mut I,
    subcommand: &str,
) -> Result<(CommonArgs, Vec<String>), CliError> {
    let mut thread: Option<String> = None;
    let mut key: Option<String> = None;
    let mut leftover: Vec<String> = Vec::new();
    while let Some(arg) = args.next() {
        match arg.as_str() {
            "--thread" => {
                thread = Some(require_value(&arg, args.next())?);
            }
            "--key" => {
                key = Some(require_value(&arg, args.next())?);
            }
            _ => leftover.push(arg),
        }
    }
    let thread = thread.ok_or_else(|| {
        CliError::Usage(format!(
            "continuation {subcommand}: missing required --thread"
        ))
    })?;
    let key = key.ok_or_else(|| {
        CliError::Usage(format!("continuation {subcommand}: missing required --key"))
    })?;
    Ok((CommonArgs { thread, key }, leftover))
}

fn require_value(flag: &str, value: Option<String>) -> Result<String, CliError> {
    value.ok_or_else(|| CliError::Usage(format!("flag `{flag}` requires a value")))
}

fn parse_opt_string(leftover: &mut Vec<String>, flag: &str) -> Result<Option<String>, CliError> {
    let mut found: Option<String> = None;
    let mut consumed: Vec<usize> = Vec::new();
    let mut i = 0;
    while i < leftover.len() {
        if leftover[i] == flag {
            if i + 1 >= leftover.len() {
                return Err(CliError::Usage(format!("flag `{flag}` requires a value")));
            }
            found = Some(leftover[i + 1].clone());
            consumed.push(i);
            consumed.push(i + 1);
            i += 2;
        } else {
            i += 1;
        }
    }
    for idx in consumed.into_iter().rev() {
        leftover.remove(idx);
    }
    Ok(found)
}

fn parse_opt_u64(leftover: &mut Vec<String>, flag: &str) -> Result<Option<u64>, CliError> {
    parse_opt_string(leftover, flag)?
        .map(|v| {
            v.parse::<u64>().map_err(|_| {
                CliError::Usage(format!("flag `{flag}` requires a non-negative integer"))
            })
        })
        .transpose()
}

fn reject_extra_args(leftover: &[String], subcommand: &str) -> Result<(), CliError> {
    if let Some(extra) = leftover.first() {
        return Err(CliError::Usage(format!(
            "continuation {subcommand}: unexpected argument `{extra}`"
        )));
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Identifier validation (per spec §"Storage Layout")
// ---------------------------------------------------------------------------

/// Reject filesystem-unsafe identifiers with the spec's
/// `{"error":"invalid_identifier"}` shape.
///
/// Accepts ONLY: ASCII alphanumerics + `_`, `-`, `.` (when not leading).
/// Rejects: empty, oversize (> [`MAX_IDENTIFIER_LEN`]), leading dot,
/// and anything outside the allow-list (spaces, Unicode, Windows-
/// reserved characters like `<>:"|?*`, control characters, NUL, path
/// separators).
///
/// The allow-list is intentionally narrow because identifiers become
/// path components on disk; the spec's "non-portable characters"
/// language motivates restricting to a portable subset (POSIX
/// "Portable Filename Character Set") rather than leaving the rejection
/// rules implicit.
fn validate_identifier(value: &str, field: &str) -> Result<(), CliError> {
    if value.is_empty() {
        return Err(invalid_identifier_error(field, "must not be empty"));
    }
    if value.len() > MAX_IDENTIFIER_LEN {
        return Err(invalid_identifier_error(
            field,
            &format!("exceeds max length {MAX_IDENTIFIER_LEN}"),
        ));
    }
    if value.starts_with('.') {
        return Err(invalid_identifier_error(
            field,
            "must not start with `.` (hidden-file convention)",
        ));
    }
    for ch in value.chars() {
        if !(ch.is_ascii_alphanumeric() || ch == '_' || ch == '-' || ch == '.') {
            return Err(invalid_identifier_error(
                field,
                "must contain only ASCII alphanumerics, `_`, `-`, or `.`",
            ));
        }
    }
    Ok(())
}

/// Build the spec-conformant `{"error":"invalid_identifier"}` JSON
/// envelope and surface it as an `Input` error so the CLI prints it
/// to stderr with exit code 3.
fn invalid_identifier_error(field: &str, detail: &str) -> CliError {
    let envelope = json!({
        "error": "invalid_identifier",
        "field": field,
        "detail": detail,
    });
    CliError::Input(envelope.to_string())
}

/// Build the spec-conformant `{"error":"storage_failure","reason":"..."}`
/// JSON envelope per spec §"Error Surfaces". Use this for ANY
/// filesystem-side failure (I/O error, permission denied, disk full,
/// serialization fault) — those failures should surface as a structured
/// wire envelope rather than free-form text so machine callers can
/// branch on `.error == "storage_failure"`.
///
/// Reserved for transient/operational failures. Logic errors that
/// indicate caller misuse (e.g., missing required flag) stay as
/// `CliError::Usage`; client-visible miss/mismatch conditions
/// (`not_found`, `client_id_mismatch`, `invalid_identifier`,
/// `blob_too_large`) have their own envelopes.
fn storage_failure_error(reason: impl AsRef<str>) -> CliError {
    let envelope = json!({
        "error": "storage_failure",
        "reason": reason.as_ref(),
    });
    CliError::Input(envelope.to_string())
}

// ---------------------------------------------------------------------------
// Path resolution
// ---------------------------------------------------------------------------

/// `$XDG_STATE_HOME/lifeloop/continuation/` or the XDG-spec fallback
/// `$HOME/.local/state/lifeloop/continuation/`. Allows
/// `$LIFELOOP_CONTINUATION_ROOT` as a hard override for tests.
///
/// Note: the fallback is `.local/state`, NOT `.local/share`. Per the
/// XDG Base Directory spec, `XDG_STATE_HOME` defaults to
/// `~/.local/state` ("state data such as logs, session restore that
/// should persist between application restarts"). `~/.local/share` is
/// `XDG_DATA_HOME`'s default — user-created data that would be
/// backed up. Continuation-store entries are transient cross-restart
/// state, so `state/` is the semantic match.
fn continuation_root() -> Result<PathBuf, CliError> {
    if let Some(override_path) = env_var_present("LIFELOOP_CONTINUATION_ROOT") {
        return Ok(PathBuf::from(override_path));
    }
    if let Some(xdg) = env_var_present("XDG_STATE_HOME") {
        return Ok(PathBuf::from(xdg).join("lifeloop").join("continuation"));
    }
    let home = env_var_present("HOME").ok_or_else(|| {
        storage_failure_error("cannot resolve storage root — neither $XDG_STATE_HOME nor $HOME set")
    })?;
    Ok(PathBuf::from(home)
        .join(".local")
        .join("state")
        .join("lifeloop")
        .join("continuation"))
}

fn env_var_present(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|s| !s.is_empty())
}

fn thread_dir(common: &CommonArgs) -> Result<PathBuf, CliError> {
    Ok(continuation_root()?.join(&common.thread))
}

fn entry_path(common: &CommonArgs) -> Result<PathBuf, CliError> {
    Ok(thread_dir(common)?.join(&common.key))
}

// ---------------------------------------------------------------------------
// Framing read/write (single-file pair-atomicity)
// ---------------------------------------------------------------------------

/// Compose the framed bytes `[u32 BE meta_len][meta JSON][blob]`.
fn compose_framed(meta: &Meta, blob: &[u8]) -> Result<Vec<u8>, CliError> {
    let meta_bytes = serde_json::to_vec(meta)
        .map_err(|err| storage_failure_error(format!("serialize meta: {err}")))?;
    let meta_len: u32 = meta_bytes.len().try_into().map_err(|_| {
        storage_failure_error(format!(
            "meta JSON exceeds u32 length ({} bytes)",
            meta_bytes.len()
        ))
    })?;
    let mut framed = Vec::with_capacity(4 + meta_bytes.len() + blob.len());
    framed.extend_from_slice(&meta_len.to_be_bytes());
    framed.extend_from_slice(&meta_bytes);
    framed.extend_from_slice(blob);
    Ok(framed)
}

/// Split framed bytes back into (meta, blob). Corrupt on-disk data
/// surfaces as `storage_failure` — the file existed but its content
/// is unreadable, which is operationally indistinguishable from a
/// disk-side fault.
fn parse_framed(bytes: &[u8]) -> Result<(Meta, Vec<u8>), CliError> {
    if bytes.len() < 4 {
        return Err(storage_failure_error(format!(
            "framed file is truncated ({} bytes; need at least 4 for header)",
            bytes.len()
        )));
    }
    let meta_len = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize;
    let meta_start = 4usize;
    let blob_start = meta_start
        .checked_add(meta_len)
        .ok_or_else(|| storage_failure_error("meta_len overflow"))?;
    if blob_start > bytes.len() {
        return Err(storage_failure_error(format!(
            "framed file is truncated (declared meta_len={meta_len} but file is {} bytes)",
            bytes.len()
        )));
    }
    let meta: Meta = serde_json::from_slice(&bytes[meta_start..blob_start])
        .map_err(|err| storage_failure_error(format!("meta JSON parse: {err}")))?;
    let blob = bytes[blob_start..].to_vec();
    Ok((meta, blob))
}

/// Atomic write: tempfile in same dir + rename(2). Same-filesystem
/// rename is required for atomicity.
///
/// All filesystem-side failures surface via `storage_failure_error`
/// (spec `{"error":"storage_failure","reason":"..."}` envelope) — the
/// reason text retains the operation + path so operators can debug,
/// while machine callers can branch on the structured error code.
fn atomic_write(target: &Path, contents: &[u8]) -> Result<(), CliError> {
    let parent = target.parent().ok_or_else(|| {
        storage_failure_error(format!("target path has no parent: {}", target.display()))
    })?;
    // Owner-only directory perms on Unix. Continuation blobs hold
    // client-owned cross-restart state (renewal-token-like payloads);
    // a permissive umask would leave them readable to other local
    // users. Codex finding on PR #65.
    create_dir_all_owner_only(parent)?;
    let temp_name = format!(
        ".{}.tmp.{}",
        target
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("entry"),
        unique_suffix()
    );
    let temp_path = parent.join(temp_name);

    // Open the tempfile with `create_new(true) + mode(0o600)` on Unix
    // so the file is owner-only from the moment it exists (closes the
    // umask-window where another process could open it for reading).
    // Clean up the tempfile on EVERY failure path (create / write /
    // sync / rename), not only rename — Codex finding on PR #65.
    let open_result = {
        let mut opts = fs::OpenOptions::new();
        opts.write(true).create_new(true);
        #[cfg(unix)]
        opts.mode(0o600);
        opts.open(&temp_path)
    };
    let mut f = open_result.map_err(|err| {
        storage_failure_error(format!("create tempfile {}: {err}", temp_path.display()))
    })?;
    if let Err(err) = f.write_all(contents) {
        let _ = fs::remove_file(&temp_path);
        return Err(storage_failure_error(format!(
            "write tempfile {}: {err}",
            temp_path.display()
        )));
    }
    if let Err(err) = f.sync_all() {
        let _ = fs::remove_file(&temp_path);
        return Err(storage_failure_error(format!(
            "fsync tempfile {}: {err}",
            temp_path.display()
        )));
    }
    drop(f);
    fs::rename(&temp_path, target).map_err(|err| {
        let _ = fs::remove_file(&temp_path);
        storage_failure_error(format!(
            "atomic rename {} -> {}: {err}",
            temp_path.display(),
            target.display()
        ))
    })?;
    Ok(())
}

/// Create a directory (and parents) with owner-only permissions on
/// Unix. On Windows, falls back to `fs::create_dir_all` (which honors
/// the default ACL).
fn create_dir_all_owner_only(path: &Path) -> Result<(), CliError> {
    #[cfg(unix)]
    {
        // `DirBuilder::recursive(true).mode(0o700)` walks parents but
        // applies the mode ONLY to directories it actually creates —
        // pre-existing dirs keep their existing perms, matching the
        // semantics of `mkdir -p -m 700`.
        let mut builder = fs::DirBuilder::new();
        builder.recursive(true).mode(0o700);
        builder.create(path).map_err(|err| {
            storage_failure_error(format!(
                "create parent dir {} with 0o700: {err}",
                path.display()
            ))
        })
    }
    #[cfg(not(unix))]
    {
        fs::create_dir_all(path).map_err(|err| {
            storage_failure_error(format!("create parent dir {}: {err}", path.display()))
        })
    }
}

fn unique_suffix() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let pid = std::process::id();
    let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
    let ts = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!("{pid}-{seq}-{ts}")
}

/// Recognize tempfile/claim-file names that THIS module produces:
///
/// - `.{key}.tmp.{pid}-{seq}-{nanos}`         — in-flight `put` tempfile
/// - `.{key}.delete-claim.{pid}-{seq}-{nanos}` — in-flight guarded `drop` claim
///
/// Used by `drop-thread` to clean up artifacts from interrupted writes
/// or aborted guarded-drops (Codex finding on PR #65 — without this,
/// `drop-thread` would report success while retaining client blob bytes
/// in the thread directory). Foreign hidden files (anything that
/// doesn't match our patterns) are LEFT in place to avoid clobbering
/// non-continuation-store data an operator may have placed there.
///
/// We do a STRICT format check, not a substring `contains` (Copilot
/// finding on PR #65). The relevant adversarial cases:
///
/// - `.tmp.swp` (vim swap file) — has `.tmp.` substring but suffix is
///   `swp`, not our `{pid}-{seq}-{nanos}` format → preserved.
/// - `.foo.delete-claim.backup` (operator backup) — has `.delete-claim.`
///   substring but suffix is `backup`, not numeric → preserved.
/// - `.bar.tmp.1-2-3` — matches our format exactly → cleaned up
///   (correct: this IS one of our tempfiles, by construction).
///
/// The strict format binds `drop-thread`'s cleanup to the exact filename
/// shape `unique_suffix()` produces. Any future writer to a thread dir
/// MUST use this format or `drop-thread` will preserve their orphans.
fn is_owned_tempfile(name: &str) -> bool {
    // All owned tempfile names start with '.'. Anything that doesn't
    // is operator-visible and handled separately.
    if !name.starts_with('.') {
        return false;
    }
    let stem = &name[1..]; // strip the leading '.'
    // The infix is either `.tmp.` or `.delete-claim.`. After it, the
    // remaining suffix must match `{pid}-{seq}-{nanos}` — three
    // non-empty all-digit groups separated by dashes.
    let suffix = if let Some(rest) = stem.rsplit_once(".tmp.").map(|(_, s)| s) {
        rest
    } else if let Some(rest) = stem.rsplit_once(".delete-claim.").map(|(_, s)| s) {
        rest
    } else {
        return false;
    };
    let mut parts = suffix.split('-');
    let p1 = parts.next();
    let p2 = parts.next();
    let p3 = parts.next();
    if parts.next().is_some() {
        return false; // more than 3 segments
    }
    matches!(
        (p1, p2, p3),
        (Some(a), Some(b), Some(c))
            if !a.is_empty()
            && !b.is_empty()
            && !c.is_empty()
            && a.bytes().all(|c| c.is_ascii_digit())
            && b.bytes().all(|c| c.is_ascii_digit())
            && c.bytes().all(|c| c.is_ascii_digit())
    )
}

#[cfg(test)]
mod is_owned_tempfile_tests {
    use super::is_owned_tempfile;

    #[test]
    fn accepts_our_tmp_pattern() {
        assert!(is_owned_tempfile(".foo.tmp.99999-0-1234567890"));
        assert!(is_owned_tempfile(".key.with.dots.tmp.1-2-3"));
    }

    #[test]
    fn accepts_our_delete_claim_pattern() {
        assert!(is_owned_tempfile(".foo.delete-claim.99999-0-1234567890"));
    }

    #[test]
    fn rejects_vim_swapfile_lookalike() {
        // `.tmp.swp` has the `.tmp.` substring but `swp` is not a
        // numeric triple — must NOT be claimed as ours.
        assert!(!is_owned_tempfile(".tmp.swp"));
        assert!(!is_owned_tempfile(".foo.tmp.swp"));
    }

    #[test]
    fn rejects_operator_backup_lookalike() {
        // `.delete-claim.backup` has the infix but `backup` is not a
        // numeric triple — preserved.
        assert!(!is_owned_tempfile(".foo.delete-claim.backup"));
    }

    #[test]
    fn rejects_non_dotfiles() {
        assert!(!is_owned_tempfile("foo.tmp.1-2-3"));
        assert!(!is_owned_tempfile("plain_key"));
    }

    #[test]
    fn rejects_partial_numeric_triples() {
        assert!(!is_owned_tempfile(".foo.tmp.1-2")); // only 2 segments
        assert!(!is_owned_tempfile(".foo.tmp.1-2-3-4")); // 4 segments
        assert!(!is_owned_tempfile(".foo.tmp.--")); // empty segments
        assert!(!is_owned_tempfile(".foo.tmp.1-a-3")); // non-digit
    }
}

// ---------------------------------------------------------------------------
// `put`
// ---------------------------------------------------------------------------

fn run_put<I: Iterator<Item = String>>(args: I) -> Result<(), CliError> {
    let mut iter = args.collect::<Vec<_>>().into_iter();
    let (common, mut leftover) = parse_thread_and_key(&mut iter, "put")?;
    let client_id =
        parse_opt_string(&mut leftover, "--client-id")?.unwrap_or_else(|| "unknown".into());
    let ttl_s = parse_opt_u64(&mut leftover, "--ttl-s")?;
    reject_extra_args(&leftover, "put")?;
    validate_identifier(&common.thread, "thread")?;
    validate_identifier(&common.key, "key")?;
    // client_id MUST satisfy the same allow-list / length cap as
    // identifier fields. Without this, a caller could supply a huge
    // unbounded `--client-id` that inflates the on-disk Meta JSON past
    // MAX_META_OVERHEAD_BYTES → the entry would write fine but be
    // permanently unreadable (bounded-read trips on read/list/drop).
    // Copilot finding on PR #67 (round 2, #3 + #4).
    validate_identifier(&client_id, "client_id")?;

    let max_bytes = configured_max_bytes()?;
    let blob = read_bounded_stdin_bytes(max_bytes)?;
    if blob.len() as u64 > max_bytes {
        let envelope = json!({
            "error": "blob_too_large",
            "max_bytes": max_bytes,
        });
        return Err(CliError::Input(envelope.to_string()));
    }

    let now = epoch_s();
    let meta = Meta {
        schema_version: SCHEMA_VERSION.to_owned(),
        client_id,
        written_at_epoch_s: now,
        ttl_s,
    };
    let framed = compose_framed(&meta, &blob)?;
    let target = entry_path(&common)?;
    atomic_write(&target, &framed)?;

    println!("{}", json!({ "status": "ok", "written_at_epoch_s": now }));
    Ok(())
}

/// Read up to `max_bytes + 1` bytes so we can detect overflow. Returns
/// the full buffer; caller checks `.len() > max_bytes` to surface the
/// `blob_too_large` envelope.
fn read_bounded_stdin_bytes(max_bytes: u64) -> Result<Vec<u8>, CliError> {
    let mut buf = Vec::new();
    std::io::stdin()
        .lock()
        // saturating_add(1) so a pathological env var setting
        // $LIFELOOP_CONTINUATION_MAX_BYTES=u64::MAX can't overflow
        // here (panic in debug; wrap in release defeats the cap).
        // Copilot finding on PR #67 (round 2, #2).
        .take(max_bytes.saturating_add(1))
        .read_to_end(&mut buf)
        .map_err(|err| storage_failure_error(format!("put: read stdin: {err}")))?;
    Ok(buf)
}

fn configured_max_bytes() -> Result<u64, CliError> {
    if let Some(v) = env_var_present("LIFELOOP_CONTINUATION_MAX_BYTES") {
        return v.parse::<u64>().map_err(|_| {
            // Misconfigured env var is a usage error (operator action
            // required), not a storage failure — keeps the wire envelope
            // honest about who needs to act.
            CliError::Usage(
                "continuation: $LIFELOOP_CONTINUATION_MAX_BYTES must be a non-negative integer"
                    .into(),
            )
        });
    }
    Ok(DEFAULT_BLOB_MAX_BYTES)
}

// ---------------------------------------------------------------------------
// `get`
// ---------------------------------------------------------------------------

fn run_get<I: Iterator<Item = String>>(args: I) -> Result<(), CliError> {
    let mut iter = args.collect::<Vec<_>>().into_iter();
    let (common, mut leftover) = parse_thread_and_key(&mut iter, "get")?;
    let require_client_id = parse_opt_string(&mut leftover, "--require-client-id")?;
    reject_extra_args(&leftover, "get")?;
    validate_identifier(&common.thread, "thread")?;
    validate_identifier(&common.key, "key")?;
    // Symmetric validation with `--client-id` on put — guarded reads
    // must use the same allow-list / length cap.
    if let Some(ref rc) = require_client_id {
        validate_identifier(rc, "client_id")?;
    }

    let path = entry_path(&common)?;
    // Bounded read: cap file size at (max_blob + framing overhead) so
    // a foreign/corrupt oversized file in the thread dir cannot force
    // unbounded allocation (local DoS). The cap is configured-blob-max
    // plus headroom for the 4-byte length prefix and the metadata JSON
    // (we don't enforce a separate meta size — meta is bounded by the
    // u32 length prefix to 4 GiB, but in practice it's a fixed-shape
    // JSON object well under 1 KiB). Copilot finding on PR #65.
    let max_bytes = configured_max_bytes()?;
    let max_file_bytes = max_bytes.saturating_add(MAX_META_OVERHEAD_BYTES);
    let bytes = match read_bounded_file(&path, max_file_bytes) {
        Ok(b) => b,
        Err(BoundedReadError::NotFound) => return Err(not_found_error()),
        Err(BoundedReadError::TooLarge { actual }) => {
            return Err(storage_failure_error(format!(
                "entry exceeds bounded-read cap ({actual} > {max_file_bytes} bytes): {}",
                path.display()
            )));
        }
        Err(BoundedReadError::Io(err)) => {
            return Err(storage_failure_error(format!(
                "read {}: {err}",
                path.display()
            )));
        }
    };
    let (meta, blob) = parse_framed(&bytes)?;

    // TTL check: spec says expired entries return not_found.
    // `checked_add` defends against corrupt on-disk meta — if the
    // expiry overflows u64, treat as expired (failure-closed for
    // adversarial data). Copilot finding on PR #65.
    if let Some(ttl) = meta.ttl_s
        && meta
            .written_at_epoch_s
            .checked_add(ttl)
            .is_none_or(|exp| epoch_s() >= exp)
    {
        return Err(not_found_error());
    }

    // --require-client-id check (post-TTL — a client must not learn
    // that a foreign-owned key existed by hitting client_id_mismatch
    // when the entry is expired anyway; the spec doesn't pin this
    // order but expiry-shadows-mismatch is the safer default).
    if let Some(required) = require_client_id
        && meta.client_id != required
    {
        let envelope = json!({
            "error": "client_id_mismatch",
            "required": required,
            "actual": meta.client_id,
        });
        return Err(CliError::Input(envelope.to_string()));
    }

    // Write meta to stderr (machine-readable), blob to stdout (raw bytes).
    // The stderr shape is the spec-documented 3-field subset
    // `{client_id, written_at_epoch_s, ttl_s}` — NOT the full on-disk
    // Meta (which also includes schema_version). Emitting schema_version
    // here would break strict clients per the spec's wire-surface table
    // at body.md line 45. Copilot finding on PR #67 (round 2, #5).
    let stderr_meta = json!({
        "client_id": meta.client_id,
        "written_at_epoch_s": meta.written_at_epoch_s,
        "ttl_s": meta.ttl_s,
    });
    eprintln!("{stderr_meta}");
    std::io::stdout()
        .lock()
        .write_all(&blob)
        .map_err(|err| storage_failure_error(format!("write blob to stdout: {err}")))?;
    Ok(())
}

fn not_found_error() -> CliError {
    CliError::Input(json!({ "error": "not_found" }).to_string())
}

/// Headroom added to the configured blob-max when computing the
/// bounded-read cap on `get`/`list`/`drop --require-client-id`.
/// Covers the 4-byte length prefix plus up to 4 KiB of metadata JSON.
///
/// Meta size is bounded by construction: `schema_version` is a
/// compile-time constant, `written_at_epoch_s` and `ttl_s` are
/// fixed-width numbers, and `client_id` is required to satisfy
/// [`validate_identifier`] (capped at [`MAX_IDENTIFIER_LEN`] = 128
/// bytes, ASCII-portable charset). Total JSON envelope worst-case is
/// well under 1 KiB; the 4 KiB cap leaves headroom for future
/// fields. Without the `validate_identifier` cap on `client_id` (added
/// in Copilot #67 round-2 fix #3+#4), a caller could write a Meta
/// larger than this cap → the entry would write fine but be
/// permanently unreadable, violating the bounded-read invariant.
const MAX_META_OVERHEAD_BYTES: u64 = 4 + 4 * 1024;

#[derive(Debug)]
enum BoundedReadError {
    NotFound,
    TooLarge { actual: u64 },
    Io(std::io::Error),
}

/// Read a file into a byte buffer after pre-flighting its
/// `metadata().len()` against `max_bytes`. Returns `TooLarge` if the
/// file's declared size exceeds the cap, so we never allocate buffers
/// larger than the cap.
///
/// This is the defense against a foreign/corrupt oversized file in a
/// continuation thread directory triggering unbounded `fs::read`
/// allocation (local DoS — Copilot finding on PR #65).
///
/// The read is strict against concurrent grow (Copilot finding on
/// PR #67): we `take(max_bytes + 1)` and bail with `TooLarge` if the
/// read produces more than `max_bytes`. A naive `take(max_bytes)`
/// would silently truncate, letting a concurrent appender's
/// post-stat growth slip past unnoticed — the caller would then parse
/// a valid-looking framing while ignoring appended bytes.
fn read_bounded_file(path: &Path, max_bytes: u64) -> Result<Vec<u8>, BoundedReadError> {
    let meta = match fs::metadata(path) {
        Ok(m) => m,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            return Err(BoundedReadError::NotFound);
        }
        Err(err) => return Err(BoundedReadError::Io(err)),
    };
    let len = meta.len();
    if len > max_bytes {
        return Err(BoundedReadError::TooLarge { actual: len });
    }
    let mut f = match fs::File::open(path) {
        Ok(f) => f,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            return Err(BoundedReadError::NotFound);
        }
        Err(err) => return Err(BoundedReadError::Io(err)),
    };
    // Read up to (max_bytes + 1) so that a concurrent appender's
    // growth past max_bytes between stat and read is detected
    // (rather than silently truncated). The +1 sentinel byte means:
    // if we got exactly max_bytes + 1, the file grew past the cap.
    let cap_with_sentinel = max_bytes.saturating_add(1);
    let mut buf = Vec::with_capacity(len as usize);
    std::io::Read::take(&mut f, cap_with_sentinel)
        .read_to_end(&mut buf)
        .map_err(BoundedReadError::Io)?;
    if buf.len() as u64 > max_bytes {
        return Err(BoundedReadError::TooLarge {
            actual: buf.len() as u64,
        });
    }
    Ok(buf)
}

// ---------------------------------------------------------------------------
// `drop`
// ---------------------------------------------------------------------------

fn run_drop<I: Iterator<Item = String>>(args: I) -> Result<(), CliError> {
    let mut iter = args.collect::<Vec<_>>().into_iter();
    let (common, mut leftover) = parse_thread_and_key(&mut iter, "drop")?;
    let require_client_id = parse_opt_string(&mut leftover, "--require-client-id")?;
    reject_extra_args(&leftover, "drop")?;
    validate_identifier(&common.thread, "thread")?;
    validate_identifier(&common.key, "key")?;
    // Symmetric validation with `--client-id` on put — guarded drops
    // must use the same allow-list / length cap.
    if let Some(ref rc) = require_client_id {
        validate_identifier(rc, "client_id")?;
    }

    let path = entry_path(&common)?;

    // `--require-client-id` requires a race-free read-validate-delete
    // sequence. The naive shape (`fs::read` → check → `fs::remove_file`)
    // has a TOCTOU window: between the validation and the unlink, a
    // concurrent `put` from another client can replace the file via
    // rename, and the guarded drop then deletes the new (foreign)
    // entry that would NOT have passed the check. Codex finding on
    // PR #65.
    //
    // The race-safe pattern is **rename-to-claim with post-rename
    // verification**:
    //
    //   1. Read meta at `path`, validate client_id (early-out for
    //      mismatch + idempotent NotFound).
    //   2. `rename(path, claim_path)` — atomic. Either we grab the
    //      file at `path` (whatever its current content is) or rename
    //      fails with NotFound (race: someone else dropped it; that's
    //      idempotent success).
    //   3. Re-read meta from the CLAIM path. If the claim's client_id
    //      still matches `required`, the file is ours to delete →
    //      unlink. If a concurrent `put` raced in between step 1's
    //      read and step 2's rename, the claim now contains the
    //      foreign-client entry — leave the claim file as forensic
    //      evidence and surface `client_id_mismatch`. We deliberately
    //      do NOT rename it back (that would create a fresh race with
    //      a subsequent put that could clobber the new entry).
    //
    // The unguarded path (no `--require-client-id`) does a plain
    // `remove_file` — the spec says drop is idempotent, and without
    // a guard there's no foreign-client concern to defend against.
    if let Some(required) = require_client_id {
        return run_drop_guarded(&path, required);
    }

    // Unguarded: delete (idempotent — NotFound is success per spec).
    match fs::remove_file(&path) {
        Ok(()) => {}
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
        Err(err) => {
            return Err(storage_failure_error(format!(
                "remove {}: {err}",
                path.display()
            )));
        }
    }
    println!("{}", json!({ "status": "ok" }));
    Ok(())
}

/// Race-safe guarded drop (see `run_drop` for the full rationale).
///
/// On `client_id_mismatch` we leave the entry intact (read-only check
/// — no rename happened). On a concurrent-put race we leave the claim
/// file in place rather than renaming back; the cost is a small piece
/// of garbage that operators can clean up via `drop-thread` or a
/// future GC sweep. The forensic path is logged to stderr (not embedded
/// in the wire envelope) so the `client_id_mismatch` JSON stays
/// strictly spec-conformant (Copilot finding on PR #65: extra fields
/// on the error envelope would break `deny_unknown_fields` decoders).
fn run_drop_guarded(path: &Path, required: String) -> Result<(), CliError> {
    // Bounded read cap (same defense as `get`/`list`). Copilot finding
    // on PR #67: guarded drop initially used `fs::read` without size
    // cap, reintroducing the local-DoS via a crafted oversized file.
    let max_bytes = configured_max_bytes()?;
    let max_file_bytes = max_bytes.saturating_add(MAX_META_OVERHEAD_BYTES);

    // Step 1: read-and-validate. Cheap early-out — if the file isn't
    // ours, fail closed without touching anything.
    let initial_bytes = match read_bounded_file(path, max_file_bytes) {
        Ok(bytes) => bytes,
        Err(BoundedReadError::NotFound) => {
            // Idempotent: drop on absent is success per spec.
            println!("{}", json!({ "status": "ok" }));
            return Ok(());
        }
        Err(BoundedReadError::TooLarge { actual }) => {
            return Err(storage_failure_error(format!(
                "entry exceeds bounded-read cap ({actual} > {max_file_bytes} bytes): {}",
                path.display()
            )));
        }
        Err(BoundedReadError::Io(err)) => {
            return Err(storage_failure_error(format!(
                "read {}: {err}",
                path.display()
            )));
        }
    };
    let (initial_meta, _) = parse_framed(&initial_bytes)?;
    // Expiry-shadows-mismatch: a TTL-expired entry is treated as absent
    // (idempotent `status:ok`) BEFORE the client_id guard, matching the
    // ordering `get` deliberately uses. Without this, a guarded drop
    // against a foreign-owned but expired entry would leak that foreign
    // ownership via `client_id_mismatch` instead of behaving like the
    // drop of a truly-absent key. (Same `checked_add` overflow-as-
    // expired defense as `get`/`list`.)
    if let Some(ttl) = initial_meta.ttl_s
        && initial_meta
            .written_at_epoch_s
            .checked_add(ttl)
            .is_none_or(|exp| epoch_s() >= exp)
    {
        println!("{}", json!({ "status": "ok" }));
        return Ok(());
    }
    if initial_meta.client_id != required {
        let envelope = json!({
            "error": "client_id_mismatch",
            "required": required,
            "actual": initial_meta.client_id,
        });
        return Err(CliError::Input(envelope.to_string()));
    }

    // Step 2: atomic rename-to-claim. POSIX rename(2) is atomic per
    // path, so this either grabs whatever's currently at `path` or
    // fails with NotFound (someone else dropped it — that's
    // idempotent ok).
    let parent = path.parent().ok_or_else(|| {
        storage_failure_error(format!("target has no parent: {}", path.display()))
    })?;
    let claim_name = format!(
        ".{}.delete-claim.{}",
        path.file_name().and_then(|n| n.to_str()).unwrap_or("entry"),
        unique_suffix()
    );
    let claim_path = parent.join(claim_name);
    match fs::rename(path, &claim_path) {
        Ok(()) => {}
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            println!("{}", json!({ "status": "ok" }));
            return Ok(());
        }
        Err(err) => {
            return Err(storage_failure_error(format!(
                "claim rename {} -> {}: {err}",
                path.display(),
                claim_path.display()
            )));
        }
    }

    // Step 3: re-read meta from the CLAIM (bounded — same cap as the
    // initial read). If the claim's client_id still matches `required`,
    // the file we grabbed is ours — unlink. If a concurrent `put`
    // raced in between step 1 and step 2, the claim now holds a
    // foreign-client entry. We attempt to restore it by renaming the
    // claim back to `path` (Copilot finding on PR #67 round 2 #7 —
    // the original behavior left foreign data DISPLACED at a hidden
    // claim path, breaking last-writer-wins for the foreign client's
    // subsequent reads). If the rename-back fails (a NEWER put has
    // landed at `path` in the meantime — a 3-way race), we leave the
    // claim file at `claim_path` as forensic evidence; this preserves
    // the newer write at the expense of leaving the displaced foreign
    // entry orphaned. Either way the wire envelope stays spec-strict
    // (forensic detail to stderr only).
    let claim_bytes = match read_bounded_file(&claim_path, max_file_bytes) {
        Ok(bytes) => bytes,
        Err(BoundedReadError::NotFound) => {
            // Claim file vanished between rename and read? Treat as
            // success (the file is gone, which is what drop wanted).
            // Highly unlikely but plausible under aggressive cleanup.
            println!("{}", json!({ "status": "ok" }));
            return Ok(());
        }
        Err(BoundedReadError::TooLarge { actual }) => {
            return Err(storage_failure_error(format!(
                "claim entry exceeds bounded-read cap ({actual} > {max_file_bytes} bytes): {}",
                claim_path.display()
            )));
        }
        Err(BoundedReadError::Io(err)) => {
            return Err(storage_failure_error(format!(
                "read claim {}: {err}",
                claim_path.display()
            )));
        }
    };
    let (claim_meta, _) = parse_framed(&claim_bytes)?;
    if claim_meta.client_id != required {
        // Concurrent foreign-client put raced in between our check
        // and our claim. Try to restore the foreign entry to its
        // expected `path` so subsequent `get`/`list` calls see it
        // (preserves last-writer-wins semantics for the foreign
        // client). Copilot finding on PR #67 (round 2, #7).
        //
        // We use `hard_link(claim_path, path)` + `remove_file(claim_path)`
        // — NOT `rename(claim_path, path)` — because POSIX `rename(2)`
        // SILENTLY REPLACES the destination if it exists (Copilot
        // finding on PR #67 round 3, #4). If a NEWER put has landed at
        // `path` between our claim and our restore, rename would
        // clobber it, breaking last-writer-wins. `link(2)` (via
        // `fs::hard_link`) fails atomically with `AlreadyExists` when
        // the destination exists, so we can detect the 3-way race and
        // leave the newer entry intact.
        let link_result = fs::hard_link(&claim_path, path);
        match link_result {
            Ok(()) => {
                // Link succeeded: path now points to the foreign data's
                // inode. Remove the (now-redundant) claim file. Both
                // failure modes here are operationally fine — the
                // restore succeeded; the leftover dot-file is just a
                // hygiene wart.
                if let Err(e) = fs::remove_file(&claim_path) {
                    eprintln!(
                        "{}",
                        json!({
                            "diagnostic": "foreign_entry_restored_with_residual_claim",
                            "path": path.display().to_string(),
                            "claim_path": claim_path.display().to_string(),
                            "residual_unlink_error": e.to_string(),
                            "reason": "hard_link succeeded but claim unlink failed; foreign entry is visible at path, claim_path is a hygiene leftover",
                        })
                    );
                } else {
                    eprintln!(
                        "{}",
                        json!({
                            "diagnostic": "foreign_entry_restored",
                            "path": path.display().to_string(),
                            "reason": "concurrent foreign-client put raced between validate and claim; restored to expected path via hard_link",
                        })
                    );
                }
            }
            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
                eprintln!(
                    "{}",
                    json!({
                        "diagnostic": "claim_file_left_in_place",
                        "claim_path": claim_path.display().to_string(),
                        "reason": "3-way race — newer put landed at target path; foreign data orphaned at claim_path (newer entry preserved per last-writer-wins)",
                    })
                );
            }
            Err(err) => {
                // Other error (permission, filesystem-cross, etc).
                // Same outcome as AlreadyExists from the wire's
                // perspective: don't try to recover, leave claim.
                eprintln!(
                    "{}",
                    json!({
                        "diagnostic": "claim_file_left_in_place",
                        "claim_path": claim_path.display().to_string(),
                        "hard_link_error": err.to_string(),
                        "reason": "hard_link failed for non-AlreadyExists reason; foreign data orphaned at claim_path",
                    })
                );
            }
        }
        // Wire envelope stays strictly spec-shaped (Copilot finding on
        // PR #65 — `deny_unknown_fields` decoders would reject extras).
        let envelope = json!({
            "error": "client_id_mismatch",
            "required": required,
            "actual": claim_meta.client_id,
        });
        return Err(CliError::Input(envelope.to_string()));
    }

    // Step 4: unlink the validated claim.
    if let Err(err) = fs::remove_file(&claim_path) {
        return Err(storage_failure_error(format!(
            "unlink claim {}: {err}",
            claim_path.display()
        )));
    }
    println!("{}", json!({ "status": "ok" }));
    Ok(())
}

// ---------------------------------------------------------------------------
// `list`
// ---------------------------------------------------------------------------

fn run_list<I: Iterator<Item = String>>(args: I) -> Result<(), CliError> {
    let mut iter = args.collect::<Vec<_>>().into_iter();
    let mut thread: Option<String> = None;
    while let Some(arg) = iter.next() {
        match arg.as_str() {
            "--thread" => thread = Some(require_value(&arg, iter.next())?),
            other => {
                return Err(CliError::Usage(format!(
                    "continuation list: unexpected argument `{other}`"
                )));
            }
        }
    }
    let thread = thread
        .ok_or_else(|| CliError::Usage("continuation list: missing required --thread".into()))?;
    validate_identifier(&thread, "thread")?;

    let dir = continuation_root()?.join(&thread);
    let max_bytes = configured_max_bytes()?;
    let max_file_bytes = max_bytes.saturating_add(MAX_META_OVERHEAD_BYTES);
    let mut keys: Vec<serde_json::Value> = Vec::new();
    match fs::read_dir(&dir) {
        Ok(rd) => {
            for entry in rd {
                let entry = entry.map_err(|err| {
                    storage_failure_error(format!("iterate {}: {err}", dir.display()))
                })?;
                let name = entry.file_name();
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue, // non-UTF8 → skip (shouldn't happen given identifier rules)
                };
                // Skip tempfiles (`.foo.tmp.*`).
                if name_str.starts_with('.') {
                    continue;
                }
                let path = entry.path();
                // Bounded read: same DoS-defense rationale as `get` —
                // a crafted oversized file in the thread dir must not
                // force unbounded allocation. Copilot finding on PR #65.
                //
                // `list` is BEST-EFFORT per spec: malformed/oversized/
                // unreadable entries are SKIPPED rather than aborting
                // the listing (so one corrupt blob can't blind the
                // operator to the rest of the thread). The skipped
                // entries are logged to stderr as diagnostic JSON so
                // operators can spot corruption without parsing wire
                // output. Copilot finding on PR #67 round 3, #1.
                let bytes = match read_bounded_file(&path, max_file_bytes) {
                    Ok(b) => b,
                    Err(BoundedReadError::NotFound) => continue, // race with concurrent drop; skip silently
                    Err(BoundedReadError::TooLarge { actual }) => {
                        eprintln!(
                            "{}",
                            json!({
                                "diagnostic": "list_skipped_entry",
                                "path": path.display().to_string(),
                                "reason": "bounded_read_too_large",
                                "actual_bytes": actual,
                            })
                        );
                        continue;
                    }
                    Err(BoundedReadError::Io(err)) => {
                        eprintln!(
                            "{}",
                            json!({
                                "diagnostic": "list_skipped_entry",
                                "path": path.display().to_string(),
                                "reason": "io_error",
                                "error": err.to_string(),
                            })
                        );
                        continue;
                    }
                };
                let (meta, _) = match parse_framed(&bytes) {
                    Ok(pair) => pair,
                    Err(err) => {
                        eprintln!(
                            "{}",
                            json!({
                                "diagnostic": "list_skipped_entry",
                                "path": path.display().to_string(),
                                "reason": "malformed_meta",
                                "error": err.message(),
                            })
                        );
                        continue;
                    }
                };
                // Expiry check: list omits TTL-expired entries (consistent
                // with `get`). `checked_add` defends against corrupt
                // on-disk meta — overflow → treat as expired (omit).
                // Copilot finding on PR #65.
                if let Some(ttl) = meta.ttl_s
                    && meta
                        .written_at_epoch_s
                        .checked_add(ttl)
                        .is_none_or(|exp| epoch_s() >= exp)
                {
                    continue;
                }
                keys.push(json!({
                    "key": name_str,
                    "client_id": meta.client_id,
                    "written_at_epoch_s": meta.written_at_epoch_s,
                    "ttl_s": meta.ttl_s,
                }));
            }
        }
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            // No thread dir = empty list.
        }
        Err(err) => {
            return Err(storage_failure_error(format!(
                "open {}: {err}",
                dir.display()
            )));
        }
    }
    println!("{}", json!({ "thread": thread, "keys": keys }));
    Ok(())
}

// ---------------------------------------------------------------------------
// `drop-thread`
// ---------------------------------------------------------------------------

fn run_drop_thread<I: Iterator<Item = String>>(args: I) -> Result<(), CliError> {
    let mut iter = args.collect::<Vec<_>>().into_iter();
    let mut thread: Option<String> = None;
    while let Some(arg) = iter.next() {
        match arg.as_str() {
            "--thread" => thread = Some(require_value(&arg, iter.next())?),
            other => {
                return Err(CliError::Usage(format!(
                    "continuation drop-thread: unexpected argument `{other}`"
                )));
            }
        }
    }
    let thread = thread.ok_or_else(|| {
        CliError::Usage("continuation drop-thread: missing required --thread".into())
    })?;
    validate_identifier(&thread, "thread")?;

    let dir = continuation_root()?.join(&thread);
    // `dropped_count` is wire-typed u32 per spec; track as u32 so
    // we never silently overflow into a value the spec can't carry.
    // Saturating-add caps at u32::MAX (4B+ files in one thread dir is
    // already a pathological state; the cap is purely defensive).
    // Copilot finding on PR #67 (spec/impl type mismatch).
    let mut dropped: u32 = 0;
    match fs::read_dir(&dir) {
        Ok(rd) => {
            for entry in rd {
                let entry = entry.map_err(|err| {
                    storage_failure_error(format!("iterate {}: {err}", dir.display()))
                })?;
                let path = entry.path();
                let name = entry.file_name();
                // Non-UTF8 filenames are foreign-by-construction
                // (our identifier validator only emits ASCII-portable
                // strings). Skip them so `.to_str().unwrap_or("")`
                // defaulting to empty doesn't accidentally treat them
                // as visible entries (which would delete them and
                // bump dropped_count). Copilot finding on PR #67.
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue,
                };
                // Skip dot-prefixed entries that aren't OUR tempfile
                // patterns — they could be operator-placed files we
                // shouldn't touch. Our own tempfiles (matching the
                // strict `is_owned_tempfile` format check) DO get
                // cleaned up, because leaving them behind would make
                // `drop-thread` retain client-owned blob bytes on
                // disk (Codex finding on PR #65). The strict format
                // check defends against accidentally clobbering
                // operator dotfiles like `.tmp.swp` or
                // `.foo.delete-claim.backup` (Copilot finding on PR #65).
                if name_str.starts_with('.') && !is_owned_tempfile(name_str) {
                    continue;
                }
                match fs::remove_file(&path) {
                    Ok(()) => {
                        // Only count operator-visible entries
                        // (non-dot-prefixed). Our internal tempfile
                        // cleanups don't bump the user-facing count
                        // — they're forensic artifacts of in-flight
                        // failures, not "entries that existed."
                        if !name_str.starts_with('.') {
                            dropped = dropped.saturating_add(1);
                        }
                    }
                    Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
                    Err(err) => {
                        return Err(storage_failure_error(format!(
                            "remove {}: {err}",
                            path.display()
                        )));
                    }
                }
            }
            // Best-effort remove the now-empty dir; ignore errors (race with concurrent put).
            let _ = fs::remove_dir(&dir);
        }
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            // Already absent — idempotent.
        }
        Err(err) => {
            return Err(storage_failure_error(format!(
                "open {}: {err}",
                dir.display()
            )));
        }
    }
    println!("{}", json!({ "status": "ok", "dropped_count": dropped }));
    Ok(())
}

// ---------------------------------------------------------------------------
// Time
// ---------------------------------------------------------------------------

fn epoch_s() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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

    /// The three `ttl_s` input cases (missing key, null value, u64
    /// value) and their expected outcomes. `compose_framed` always emits
    /// the key, so a missing key only arises from hand-corruption; serde
    /// maps missing-or-null to `None`.
    #[test]
    fn parse_framed_maps_missing_ttl_s_key_to_none() {
        // A framed file with a Meta missing the ttl_s key parses with
        // ttl_s = None (serde default for a missing `Option`).
        let meta = r#"{"schema_version":"lifeloop.continuation.v0.1","client_id":"x","written_at_epoch_s":1}"#;
        let meta_bytes = meta.as_bytes();
        let mut framed = Vec::with_capacity(4 + meta_bytes.len());
        framed.extend_from_slice(&(meta_bytes.len() as u32).to_be_bytes());
        framed.extend_from_slice(meta_bytes);
        framed.extend_from_slice(b"blob");

        let (meta, _) = parse_framed(&framed).expect("missing ttl_s → None");
        assert!(meta.ttl_s.is_none(), "missing ttl_s key → None (no TTL)");
    }

    #[test]
    fn parse_framed_accepts_ttl_s_null() {
        let good_meta = r#"{"schema_version":"lifeloop.continuation.v0.1","client_id":"x","written_at_epoch_s":1,"ttl_s":null}"#;
        let meta_bytes = good_meta.as_bytes();
        let mut framed = Vec::with_capacity(4 + meta_bytes.len());
        framed.extend_from_slice(&(meta_bytes.len() as u32).to_be_bytes());
        framed.extend_from_slice(meta_bytes);
        framed.extend_from_slice(b"blob");

        let (meta, blob) = parse_framed(&framed).expect("ttl_s null is valid");
        assert!(meta.ttl_s.is_none(), "ttl_s null → None (no TTL)");
        assert_eq!(blob, b"blob");
    }

    #[test]
    fn parse_framed_accepts_ttl_s_u64() {
        let good_meta = r#"{"schema_version":"lifeloop.continuation.v0.1","client_id":"x","written_at_epoch_s":1,"ttl_s":300}"#;
        let meta_bytes = good_meta.as_bytes();
        let mut framed = Vec::with_capacity(4 + meta_bytes.len());
        framed.extend_from_slice(&(meta_bytes.len() as u32).to_be_bytes());
        framed.extend_from_slice(meta_bytes);
        framed.extend_from_slice(b"blob");

        let (meta, _) = parse_framed(&framed).expect("ttl_s u64 is valid");
        assert_eq!(meta.ttl_s, Some(300));
    }

    #[test]
    fn validate_identifier_accepts_normal_names() {
        assert!(validate_identifier("thread-abc", "thread").is_ok());
        assert!(validate_identifier("01H8K7Q2RXJG7HBPV5MDT9NS3R", "thread").is_ok());
        assert!(validate_identifier("renewal-state", "key").is_ok());
    }

    #[test]
    fn validate_identifier_rejects_path_separators() {
        // Use a non-dot-prefixed input so the allow-list branch fires
        // before the leading-dot branch. The strict allow-list (Copilot
        // finding on PR #65) rejects path separators alongside every
        // other non-portable character with a single uniform message
        // referring to the allow-list.
        let err = validate_identifier("foo/bar", "thread").unwrap_err();
        assert!(err.message().contains("invalid_identifier"));
        assert!(err.message().contains("ASCII alphanumerics"));

        let err = validate_identifier("foo\\bar", "thread").unwrap_err();
        assert!(err.message().contains("ASCII alphanumerics"));
    }

    #[test]
    fn validate_identifier_leading_dot_takes_precedence_over_separators() {
        // `../escape` is the classic path-traversal payload; spec rejects
        // it via the leading-dot branch (broader-catching: rejects any
        // hidden file, not just traversals).
        let err = validate_identifier("../escape", "thread").unwrap_err();
        assert!(err.message().contains("invalid_identifier"));
        assert!(err.message().contains("hidden-file"));
    }

    #[test]
    fn validate_identifier_rejects_leading_dot() {
        let err = validate_identifier(".hidden", "key").unwrap_err();
        assert!(err.message().contains("invalid_identifier"));
        assert!(err.message().contains("hidden-file"));
    }

    #[test]
    fn validate_identifier_rejects_empty() {
        let err = validate_identifier("", "thread").unwrap_err();
        assert!(err.message().contains("invalid_identifier"));
    }

    #[test]
    fn validate_identifier_rejects_control_chars() {
        let err = validate_identifier("with\nnewline", "key").unwrap_err();
        assert!(err.message().contains("invalid_identifier"));
    }

    #[test]
    fn validate_identifier_rejects_oversized() {
        let big = "a".repeat(MAX_IDENTIFIER_LEN + 1);
        let err = validate_identifier(&big, "key").unwrap_err();
        assert!(err.message().contains("invalid_identifier"));
    }

    #[test]
    fn framing_roundtrip_preserves_meta_and_blob() {
        // Use a generic client_id placeholder — lifeloop core does not
        // name specific clients (per the kernel-purity gate).
        let meta = Meta {
            schema_version: SCHEMA_VERSION.to_owned(),
            client_id: "test-client".to_owned(),
            written_at_epoch_s: 1716385200,
            ttl_s: Some(600),
        };
        let blob = b"hello world\x00\x01\x02 binary content";
        let framed = compose_framed(&meta, blob).unwrap();
        let (m2, b2) = parse_framed(&framed).unwrap();
        assert_eq!(meta, m2);
        assert_eq!(blob.as_slice(), b2.as_slice());
    }

    #[test]
    fn framing_rejects_truncated_header() {
        // 3 bytes — less than 4-byte header
        let err = parse_framed(&[0u8, 0, 0]).unwrap_err();
        assert!(err.message().contains("truncated"));
    }

    #[test]
    fn framing_rejects_truncated_meta() {
        // Header says meta_len=100 but file is only 4+10 bytes
        let mut bytes = vec![0u8, 0, 0, 100];
        bytes.extend_from_slice(&[0u8; 10]);
        let err = parse_framed(&bytes).unwrap_err();
        assert!(err.message().contains("truncated"));
    }

    #[test]
    fn invalid_identifier_envelope_shape() {
        let err = validate_identifier(".hidden", "key").unwrap_err();
        let parsed: serde_json::Value = serde_json::from_str(err.message()).unwrap();
        assert_eq!(parsed["error"], "invalid_identifier");
        assert_eq!(parsed["field"], "key");
        assert!(parsed["detail"].is_string());
    }

    /// Finding #15: `run_drop_guarded` against a TTL-expired entry owned
    /// by ANOTHER client must behave like dropping an absent key
    /// (idempotent `Ok`), NOT leak the foreign owner via
    /// `client_id_mismatch`. This mirrors the expiry-shadows-mismatch
    /// ordering `get` uses.
    #[test]
    fn guarded_drop_on_expired_foreign_entry_is_idempotent_ok() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("entry");
        // Foreign-owned (`client-a`) entry written 1000s ago with a
        // 1s TTL → long expired.
        let meta = Meta {
            schema_version: SCHEMA_VERSION.to_owned(),
            client_id: "client-a".to_owned(),
            written_at_epoch_s: epoch_s().saturating_sub(1000),
            ttl_s: Some(1),
        };
        let framed = compose_framed(&meta, b"blob").unwrap();
        fs::write(&path, &framed).unwrap();

        // Guarded drop requiring a DIFFERENT client must NOT surface
        // client_id_mismatch — the expired entry is treated as absent.
        let result = run_drop_guarded(&path, "client-b".to_owned());
        assert!(
            result.is_ok(),
            "expired foreign entry must drop idempotently, got: {result:?}"
        );
    }

    /// Counterpart to the above: a LIVE foreign-owned entry must still
    /// surface `client_id_mismatch` (the early-out only fires on
    /// expiry).
    #[test]
    fn guarded_drop_on_live_foreign_entry_still_mismatches() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("entry");
        let meta = Meta {
            schema_version: SCHEMA_VERSION.to_owned(),
            client_id: "client-a".to_owned(),
            written_at_epoch_s: epoch_s(),
            ttl_s: Some(3600),
        };
        let framed = compose_framed(&meta, b"blob").unwrap();
        fs::write(&path, &framed).unwrap();

        let err = run_drop_guarded(&path, "client-b".to_owned()).unwrap_err();
        assert!(err.message().contains("client_id_mismatch"));
    }
}