use std::cmp::Ordering;
use std::fs;
use std::io::{Read as _, Write as _};
use std::path::{Component, Path, PathBuf};
use std::process::{Command, ExitCode};

use anyhow::{bail, Context, Result};
use serde::Serialize;
use serde_json::{json, Value as JsonValue};

use crate::output::CommandReport;
use crate::paths::state::StateLayout;
use crate::paths::state::ThreadSyncConfig;
use crate::profile;
use crate::state::machine_registry;
use crate::state::session as session_state;
use crate::state::thread_transfer::{self, ThreadDiagnostic};

const SYNC_SCHEMA_VERSION: u32 = 1;
const CAPSULE_KIND: &str = "ccd.thread_transfer";
const CAPSULE_SCHEMA_VERSION: u32 = 2;
const INDEX_KIND: &str = "ccd.thread_sync_index";
const IDENTITY_KIND: &str = "ccd.thread_sync_identity";
const ENV_SYNC_DIR: &str = "CCD_THREAD_SYNC_DIR";
const MAX_CAPSULES_PER_THREAD: usize = 20;
const MAX_CAPSULE_FILES_PER_THREAD: usize = 80;
const MAX_CAPSULE_SCAN_FILES: usize = 200;
const MAX_SYNC_CAPSULE_BYTES: u64 = 512 * 1024;

#[derive(Serialize)]
pub struct ThreadSyncReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    project_id: String,
    locality_id: String,
    sync: ThreadSyncView,
}

#[derive(Serialize)]
struct ThreadSyncView {
    backend: &'static str,
    schema_version: u32,
    configured_from: &'static str,
    sync_dir: String,
    thread_dir: String,
    status: &'static str,
    sync_project_id: String,
    identity_source: &'static str,
    identity_material_digest: String,
    identity_portable: bool,
    thread_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    latest: Option<CapsuleSummary>,
    #[serde(skip_serializing_if = "Option::is_none")]
    capsule_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    latest_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    index_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    identity_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    retention: Option<RetentionView>,
    diagnostics: Vec<ThreadDiagnostic>,
    #[serde(skip_serializing_if = "Option::is_none")]
    import: Option<JsonValue>,
}

#[derive(Clone, Serialize)]
struct CapsuleSummary {
    path: String,
    digest: String,
    created_at_epoch_s: u64,
    thread_id: String,
    source_profile: String,
    source_project_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    source_workspace: Option<String>,
    attachment_count: usize,
}

#[derive(Serialize)]
struct RetentionView {
    max_capsules: usize,
    max_capsule_files: usize,
    removed_count: usize,
}

#[derive(Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum RemoteResumeStatus {
    Available,
    UntrustedMachine,
}

#[derive(Serialize)]
pub(crate) struct RemoteResumeAdvisoryView {
    pub(crate) status: RemoteResumeStatus,
    pub(crate) source: String,
    pub(crate) capsule_path: String,
    pub(crate) digest: String,
    pub(crate) created_at_epoch_s: u64,
}

struct ThreadSyncContext {
    repo_root: PathBuf,
    profile: String,
    locality_id: String,
    sync_project_id: String,
    sync_identity_source: &'static str,
    sync_identity_material_digest: String,
    sync_identity_portable: bool,
    thread_id: String,
    sync_dir: PathBuf,
    configured_from: &'static str,
    thread_dir: PathBuf,
    max_capsules_per_thread: usize,
    max_capsule_files_per_thread: usize,
    config_warning: Option<String>,
    source_machine_id: Option<String>,
    source_machine_slug: String,
}

struct SyncIdentity {
    project_id: String,
    source: &'static str,
    material_digest: String,
    portable: bool,
}

struct ResolvedSyncDir {
    path: PathBuf,
    configured_from: &'static str,
}

struct CapsulePathList {
    paths: Vec<PathBuf>,
    limit_exceeded: bool,
}

#[derive(Clone)]
struct CapsuleCandidate {
    path: PathBuf,
    summary: CapsuleSummary,
}

#[derive(Clone, Copy)]
enum CandidateSource {
    Latest,
    Capsule,
}

struct CapsuleValidationError {
    code: &'static str,
    message: String,
}

struct Discovery {
    selected: Option<CapsuleCandidate>,
    status: &'static str,
    diagnostics: Vec<ThreadDiagnostic>,
}

impl CommandReport for ThreadSyncReport {
    fn exit_code(&self) -> ExitCode {
        if self.ok {
            ExitCode::SUCCESS
        } else {
            ExitCode::from(1)
        }
    }

    fn render_text(&self) {
        match self.command {
            "thread-status" => {
                println!(
                    "Thread sync `{}` for profile `{}` at {}.",
                    self.sync.status, self.profile, self.sync.sync_dir
                );
                if let Some(latest) = &self.sync.latest {
                    println!(
                        "Latest capsule: {} (source thread {}, created_at {}).",
                        latest.path, latest.thread_id, latest.created_at_epoch_s
                    );
                }
            }
            "thread-export-sync" => {
                println!(
                    "Published thread capsule for profile `{}` to {}.",
                    self.profile,
                    self.sync
                        .capsule_path
                        .as_deref()
                        .unwrap_or(self.sync.thread_dir.as_str())
                );
            }
            "thread-import-sync" => {
                println!(
                    "Thread sync import `{}` for profile `{}` from {}.",
                    self.sync.status, self.profile, self.sync.sync_dir
                );
            }
            _ => println!("Thread sync status: {}.", self.sync.status),
        }
        for diagnostic in &self.sync.diagnostics {
            println!("- {}: {}", diagnostic.code, diagnostic.message);
        }
    }
}

pub fn status(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    explicit_sync_dir: Option<&Path>,
) -> Result<ThreadSyncReport> {
    let context = resolve_context(
        repo_root,
        explicit_profile,
        explicit_sync_dir,
        "thread status",
    )?;
    let discovery = discover_latest(&context)?;
    Ok(report(
        "thread-status",
        discovery.status != "invalid",
        &context,
        discovery.status,
        discovery.selected.map(|candidate| candidate.summary),
        None,
        Some(latest_path(&context).display().to_string()),
        Some(index_path(&context).display().to_string()),
        Some(identity_path(&context).display().to_string()),
        None,
        discovery.diagnostics,
        None,
    ))
}

/// Boundary events that may trigger `thread_sync` auto-publish.
///
/// Phase 7 inlined this surface from the deleted `src/extensions/` cluster.
/// Only the three boundaries CCD actually triggers post-slim are retained;
/// `agent_end` strings in operator `publish_on = [...]` config are simply
/// ignored (no ccd command publishes on that boundary today).
#[derive(Clone, Copy, Debug)]
pub(crate) enum AutoPublishBoundary {
    Checkpoint,
    Handover,
    SessionEnd,
}

impl AutoPublishBoundary {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::Checkpoint => "checkpoint",
            Self::Handover => "handover",
            Self::SessionEnd => "session_end",
        }
    }
}

/// Config-gated auto-publish helper invoked from boundary ceremonies
/// (checkpoint, handover, session_clear). Returns silently when the
/// workspace is not linked, the profile root is missing, or
/// `[thread_sync]` does not opt in to publishing this boundary.
///
/// Push failures surface as stderr `warning:` lines unless `strict_publish`
/// is on, in which case the prefix is `error:`. Failures never propagate —
/// the host CLI command stays successful (preserves Phase-6 behavior).
pub(crate) fn publish_on_boundary(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    boundary: AutoPublishBoundary,
) {
    let Ok(profile_name) = profile::resolve(explicit_profile) else {
        return;
    };
    let Ok(layout) = StateLayout::resolve(repo_root, profile_name) else {
        return;
    };
    if !layout.profile_root().is_dir() {
        return;
    }
    let Ok(locality_id) = thread_transfer::ensure_repo_linked(repo_root) else {
        return;
    };
    let cfg = match layout.effective_thread_sync_config(&locality_id) {
        Ok(cfg) => cfg,
        Err(error) => {
            eprintln!(
                "warning: thread-sync auto-publish skipped on {}: failed to load [thread_sync] config: {error}",
                boundary.as_str()
            );
            return;
        }
    };
    if cfg.auto_publish != Some(true) {
        return;
    }
    let needle = boundary.as_str();
    if !cfg.publish_on.iter().any(|entry| entry == needle) {
        return;
    }
    let strict = cfg.strict_publish == Some(true);
    if let Err(error) = push(repo_root, explicit_profile, None) {
        let prefix = if strict { "error" } else { "warning" };
        eprintln!(
            "{prefix}: thread-sync auto-publish on {} was rejected: {error}",
            boundary.as_str()
        );
    }
}

pub fn push(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    explicit_sync_dir: Option<&Path>,
) -> Result<ThreadSyncReport> {
    let context = resolve_context(
        repo_root,
        explicit_profile,
        explicit_sync_dir,
        "thread export --to-sync-dir",
    )?;
    create_dir_all_no_symlink(&context.sync_dir, &capsules_dir(&context))?;

    let export_report = thread_transfer::export(repo_root, explicit_profile)?;
    let export_value = serde_json::to_value(&export_report)?;
    let capsule = export_value
        .pointer("/export/capsule")
        .cloned()
        .ok_or_else(|| anyhow::anyhow!("thread export report did not contain export.capsule"))?;
    let capsule_bytes = serde_json::to_vec_pretty(&capsule)?;
    if capsule_bytes.len() as u64 > MAX_SYNC_CAPSULE_BYTES {
        return Ok(report(
            "thread-export-sync",
            false,
            &context,
            "publish_rejected",
            None,
            None,
            Some(latest_path(&context).display().to_string()),
            Some(index_path(&context).display().to_string()),
            Some(identity_path(&context).display().to_string()),
            None,
            vec![diagnostic(
                "error",
                "synced_capsule_too_large",
                format!(
                    "Exported thread capsule is {} bytes, above the {} byte filesystem sync limit.",
                    capsule_bytes.len(),
                    MAX_SYNC_CAPSULE_BYTES
                ),
            )],
            None,
        ));
    }
    let capsule_text = std::str::from_utf8(&capsule_bytes)
        .context("thread transfer capsule JSON was not valid UTF-8")?;
    let digest = format!("sha256:{}", thread_transfer::sha256_hex(capsule_text));
    let summary = validate_capsule_summary(&capsule, &digest, &context, None).map_err(|error| {
        anyhow::anyhow!(
            "exported thread capsule is not sync-safe: {}",
            error.message
        )
    })?;
    let file_name = capsule_file_name(
        summary.created_at_epoch_s,
        Some(context.source_machine_slug.as_str()),
        &digest,
    );
    let capsule_path = capsules_dir(&context).join(file_name);
    let latest_path = latest_path(&context);
    let index_path = index_path(&context);
    let identity_path = identity_path(&context);

    atomic_write_bytes(&context, &capsule_path, &capsule_bytes)?;
    atomic_write_bytes(&context, &latest_path, &capsule_bytes)?;
    let index = json!({
        "kind": INDEX_KIND,
        "schema_version": SYNC_SCHEMA_VERSION,
        "backend": "filesystem",
        "profile": &context.profile,
        "project_id": &context.locality_id,
        "sync_project_id": &context.sync_project_id,
        "identity_source": context.sync_identity_source,
        "identity_material_digest": &context.sync_identity_material_digest,
        "thread_id": &context.thread_id,
        "latest": {
            "path": relative_capsule_path(&capsule_path, &context.thread_dir),
            "digest": digest,
            "created_at_epoch_s": summary.created_at_epoch_s
        },
        "source": {
            "workspace": context.repo_root.display().to_string(),
            "machine_id": &context.source_machine_id
        }
    });
    atomic_write_json(&context, &index_path, &index)?;
    atomic_write_json(&context, &identity_path, &identity_json(&context)?)?;
    let retention = prune_old_capsules(&context, &capsule_path)?;
    let mut diagnostics = Vec::new();
    if context.source_machine_id.is_none() {
        diagnostics.push(diagnostic(
            "warning",
            "machine_identity_unavailable",
            "No CCD machine identity is registered for this workspace; capsule filenames use a locality-derived fallback slug."
                .to_owned(),
        ));
    }

    let latest = read_candidate(&capsule_path, &context, CandidateSource::Capsule)
        .map(|candidate| candidate.summary)
        .map_err(|diagnostic| anyhow::anyhow!(diagnostic.message))?;
    Ok(report(
        "thread-export-sync",
        true,
        &context,
        "published",
        Some(latest),
        Some(capsule_path.display().to_string()),
        Some(latest_path.display().to_string()),
        Some(index_path.display().to_string()),
        Some(identity_path.display().to_string()),
        Some(retention),
        diagnostics,
        None,
    ))
}

pub fn pull(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    explicit_sync_dir: Option<&Path>,
) -> Result<ThreadSyncReport> {
    let context = resolve_context(
        repo_root,
        explicit_profile,
        explicit_sync_dir,
        "thread import --from-sync-dir",
    )?;
    let mut discovery = discover_latest(&context)?;
    let Some(selected) = discovery.selected else {
        return Ok(report(
            "thread-import-sync",
            false,
            &context,
            if discovery.status == "invalid" {
                "invalid_capsule"
            } else {
                "empty"
            },
            None,
            None,
            Some(latest_path(&context).display().to_string()),
            Some(index_path(&context).display().to_string()),
            Some(identity_path(&context).display().to_string()),
            None,
            discovery.diagnostics,
            None,
        ));
    };

    match thread_transfer::import_preview(repo_root, explicit_profile, &selected.path) {
        Ok(import_report) => {
            let import = serde_json::to_value(&import_report)?;
            let import_ok = import
                .get("ok")
                .and_then(JsonValue::as_bool)
                .unwrap_or(false);
            let status = if import_ok {
                "envelope_valid"
            } else {
                "import_rejected"
            };
            Ok(report(
                "thread-import-sync",
                import_ok,
                &context,
                status,
                Some(selected.summary),
                Some(selected.path.display().to_string()),
                Some(latest_path(&context).display().to_string()),
                Some(index_path(&context).display().to_string()),
                Some(identity_path(&context).display().to_string()),
                None,
                discovery.diagnostics,
                Some(import),
            ))
        }
        Err(error) => {
            discovery.diagnostics.push(diagnostic(
                "error",
                "invalid_synced_capsule",
                format!(
                    "Synced capsule {} could not be imported: {error}",
                    selected.path.display()
                ),
            ));
            Ok(report(
                "thread-import-sync",
                false,
                &context,
                "invalid_capsule",
                Some(selected.summary),
                Some(selected.path.display().to_string()),
                Some(latest_path(&context).display().to_string()),
                Some(index_path(&context).display().to_string()),
                Some(identity_path(&context).display().to_string()),
                None,
                discovery.diagnostics,
                None,
            ))
        }
    }
}

/// Read-side advisory for `ccd start` / `ccd status` when
/// `[thread_sync] auto_discover = true`. Best-effort and read-only: any
/// error resolving sync state collapses to `None`. Never mutates local
/// continuity; explicit `ccd thread import --from-sync-dir` is still the
/// only path that applies a synced capsule.
pub fn discover_remote_resume(
    repo_root: &Path,
    explicit_profile: Option<&str>,
) -> Option<RemoteResumeAdvisoryView> {
    let profile_name = profile::resolve(explicit_profile).ok()?;
    let layout = StateLayout::resolve(repo_root, profile_name).ok()?;
    let locality_id = thread_transfer::ensure_repo_linked(repo_root).ok()?;
    let cfg = layout.effective_thread_sync_config(&locality_id).ok()?;
    if cfg.auto_discover != Some(true) {
        return None;
    }
    let context = resolve_context(repo_root, explicit_profile, None, "thread status").ok()?;
    let discovery = discover_latest(&context).ok()?;
    let selected = discovery.selected?;
    let source = parse_source_slug(&selected.path)?;
    // The capsule filename embeds `safe_path_segment(machine_id)`, so the
    // raw `source_machine_slug` must be sanitized the same way before
    // comparison. Otherwise a machine_id containing characters outside
    // `[A-Za-z0-9._-]` (legal at registration time) compares unequal and the
    // local machine's own capsule leaks back as a remote-resume advisory.
    if source == safe_path_segment(&context.source_machine_slug) {
        return None;
    }
    // Strict-match `trusted_machines` against the on-disk slug. The slug is
    // `safe_path_segment(machine_id)`, so operators must enter the sanitized
    // form. We deliberately do NOT sanitize config entries here — doing so
    // can create silent collisions (e.g. `evil#machine` and `evil_machine`
    // both compare equal to the same on-disk slug, granting false trust).
    let status = if cfg.trusted_machines.is_empty()
        || cfg.trusted_machines.iter().any(|entry| entry == &source)
    {
        RemoteResumeStatus::Available
    } else {
        RemoteResumeStatus::UntrustedMachine
    };
    Some(RemoteResumeAdvisoryView {
        status,
        source,
        capsule_path: selected.path.display().to_string(),
        digest: selected.summary.digest,
        created_at_epoch_s: selected.summary.created_at_epoch_s,
    })
}

fn parse_source_slug(path: &Path) -> Option<String> {
    let stem = path.file_stem()?.to_str()?;
    let (_epoch, rest) = stem.split_once('-')?;
    let (slug, _digest) = rest.rsplit_once('-')?;
    Some(slug.to_owned())
}

fn report(
    command: &'static str,
    ok: bool,
    context: &ThreadSyncContext,
    status: &'static str,
    latest: Option<CapsuleSummary>,
    capsule_path: Option<String>,
    latest_path: Option<String>,
    index_path: Option<String>,
    identity_path: Option<String>,
    retention: Option<RetentionView>,
    mut diagnostics: Vec<ThreadDiagnostic>,
    import: Option<JsonValue>,
) -> ThreadSyncReport {
    if let Some(message) = &context.config_warning {
        diagnostics.insert(
            0,
            diagnostic("warning", "thread_sync_config_unavailable", message.clone()),
        );
    }
    ThreadSyncReport {
        command,
        ok,
        path: context.repo_root.display().to_string(),
        profile: context.profile.clone(),
        project_id: context.locality_id.clone(),
        locality_id: context.locality_id.clone(),
        sync: ThreadSyncView {
            backend: "filesystem",
            schema_version: SYNC_SCHEMA_VERSION,
            configured_from: context.configured_from,
            sync_dir: context.sync_dir.display().to_string(),
            thread_dir: context.thread_dir.display().to_string(),
            status,
            sync_project_id: context.sync_project_id.clone(),
            identity_source: context.sync_identity_source,
            identity_material_digest: context.sync_identity_material_digest.clone(),
            identity_portable: context.sync_identity_portable,
            thread_id: context.thread_id.clone(),
            latest,
            capsule_path,
            latest_path,
            index_path,
            identity_path,
            retention,
            diagnostics,
            import,
        },
    }
}

fn resolve_context(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    explicit_sync_dir: Option<&Path>,
    command_name: &str,
) -> Result<ThreadSyncContext> {
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    thread_transfer::ensure_profile_exists(&layout, repo_root, command_name)?;
    let locality_id = thread_transfer::ensure_repo_linked(repo_root)?;
    let sync_identity = resolve_sync_identity(repo_root, &locality_id);
    let thread_id = thread_transfer::stable_thread_id(profile.as_str(), &sync_identity.project_id);
    let resolved_config = resolve_context_config(explicit_sync_dir, &layout, &locality_id)?;
    let config = resolved_config.config;
    let resolved_sync_dir = resolve_sync_dir(explicit_sync_dir, &config)?;
    let retention = resolve_retention_limits(&config)?;
    let thread_dir = resolved_sync_dir
        .path
        .join("thread-sync")
        .join("v1")
        .join("projects")
        .join(&sync_identity.project_id)
        .join("profiles")
        .join(profile.as_str())
        .join("threads")
        .join(&thread_id);
    let source_machine_id = machine_registry::resolve_machine_identity_view(&layout)
        .ok()
        .and_then(|view| view.id);
    let source_machine_slug = source_machine_id.clone().unwrap_or_else(|| {
        let digest = thread_transfer::sha256_hex(&locality_id);
        format!("unregistered-{}", &digest[..12])
    });

    Ok(ThreadSyncContext {
        repo_root: repo_root.to_path_buf(),
        profile: profile.to_string(),
        locality_id,
        sync_project_id: sync_identity.project_id,
        sync_identity_source: sync_identity.source,
        sync_identity_material_digest: sync_identity.material_digest,
        sync_identity_portable: sync_identity.portable,
        thread_id,
        sync_dir: resolved_sync_dir.path,
        configured_from: resolved_sync_dir.configured_from,
        thread_dir,
        max_capsules_per_thread: retention.max_capsules,
        max_capsule_files_per_thread: retention.max_capsule_files,
        config_warning: resolved_config.warning,
        source_machine_id,
        source_machine_slug,
    })
}

struct ResolvedThreadSyncConfig {
    config: ThreadSyncConfig,
    warning: Option<String>,
}

struct RetentionLimits {
    max_capsules: usize,
    max_capsule_files: usize,
}

fn resolve_context_config(
    explicit_sync_dir: Option<&Path>,
    layout: &StateLayout,
    locality_id: &str,
) -> Result<ResolvedThreadSyncConfig> {
    let config_required_for_sync_dir = explicit_sync_dir.is_none()
        && match std::env::var_os(ENV_SYNC_DIR) {
            Some(value) => value.is_empty(),
            None => true,
        };
    match layout.effective_thread_sync_config(locality_id) {
        Ok(config) => Ok(ResolvedThreadSyncConfig {
            config,
            warning: None,
        }),
        Err(error) if !config_required_for_sync_dir => Ok(ResolvedThreadSyncConfig {
            config: ThreadSyncConfig::default(),
            warning: Some(format!(
                "Effective [thread_sync] config could not be loaded, so the explicit sync directory is using default thread-sync retention limits: {error}"
            )),
        }),
        Err(error) => Err(error).context("failed to load effective thread-sync config"),
    }
}

fn resolve_retention_limits(config: &ThreadSyncConfig) -> Result<RetentionLimits> {
    let max_capsules = resolve_retention_limit(
        config.max_capsules_per_thread,
        MAX_CAPSULES_PER_THREAD,
        "max_capsules_per_thread",
    )?;
    let max_capsule_files = resolve_retention_limit(
        config.max_capsule_files_per_thread,
        MAX_CAPSULE_FILES_PER_THREAD,
        "max_capsule_files_per_thread",
    )?;
    if max_capsule_files < max_capsules {
        bail!(
            "[thread_sync] max_capsule_files_per_thread ({max_capsule_files}) must be greater than or equal to max_capsules_per_thread ({max_capsules})"
        );
    }
    Ok(RetentionLimits {
        max_capsules,
        max_capsule_files,
    })
}

fn resolve_retention_limit(
    configured: Option<usize>,
    default: usize,
    field: &'static str,
) -> Result<usize> {
    match configured {
        Some(0) => bail!("[thread_sync] {field} must be at least 1"),
        Some(value) => Ok(value),
        None => Ok(default),
    }
}

fn resolve_sync_identity(repo_root: &Path, locality_id: &str) -> SyncIdentity {
    let origin_urls = git_lines(repo_root, &["remote", "get-url", "origin"]);
    if let Some(origin_url) = origin_urls.first() {
        return portable_sync_identity("git_remote_origin", std::slice::from_ref(origin_url));
    }

    let mut root_commits = git_lines(repo_root, &["rev-list", "--max-parents=0", "HEAD"]);
    root_commits.sort();
    root_commits.dedup();
    if !root_commits.is_empty() {
        return portable_sync_identity("git_root_commits", &root_commits);
    }

    let mut remotes = git_lines(repo_root, &["remote"])
        .into_iter()
        .flat_map(|remote| git_lines(repo_root, &["remote", "get-url", &remote]))
        .collect::<Vec<_>>();
    remotes.sort();
    remotes.dedup();
    if !remotes.is_empty() {
        return portable_sync_identity("git_remote_urls", &remotes);
    }

    SyncIdentity {
        project_id: locality_id.to_owned(),
        source: "locality_id",
        material_digest: thread_transfer::sha256_hex(locality_id),
        portable: false,
    }
}

fn portable_sync_identity(source: &'static str, values: &[String]) -> SyncIdentity {
    let mut material = String::from(source);
    for value in values {
        material.push('\0');
        material.push_str(value);
    }
    let digest = thread_transfer::sha256_hex(&material);
    SyncIdentity {
        project_id: format!("syncproj_{}", &digest[..32]),
        source,
        material_digest: digest,
        portable: true,
    }
}

fn git_lines(repo_root: &Path, args: &[&str]) -> Vec<String> {
    let output = Command::new("git")
        .args(args)
        .current_dir(repo_root)
        .output();
    let Ok(output) = output else {
        return Vec::new();
    };
    if !output.status.success() {
        return Vec::new();
    }
    String::from_utf8_lossy(&output.stdout)
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .map(str::to_owned)
        .collect()
}

fn resolve_sync_dir(
    explicit_sync_dir: Option<&Path>,
    config: &ThreadSyncConfig,
) -> Result<ResolvedSyncDir> {
    if let Some(path) = explicit_sync_dir {
        return Ok(ResolvedSyncDir {
            path: absolutize_configured_dir(path)?,
            configured_from: "flag",
        });
    }
    if let Some(value) = std::env::var_os(ENV_SYNC_DIR) {
        if !value.is_empty() {
            return Ok(ResolvedSyncDir {
                path: absolutize_configured_dir(Path::new(&value))?,
                configured_from: "env",
            });
        }
    }
    if let Some(raw) = config.sync_dir.as_deref() {
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            bail!(
                "[thread_sync] sync_dir in effective config is empty; remove the field or set a non-empty path"
            );
        }
        return Ok(ResolvedSyncDir {
            path: absolutize_configured_dir(Path::new(trimmed))?,
            configured_from: "config",
        });
    }
    bail!(
        "thread sync directory is not configured; pass a sync directory flag, set {ENV_SYNC_DIR}, or set [thread_sync] sync_dir in the profile config"
    );
}

fn absolutize_configured_dir(path: &Path) -> Result<PathBuf> {
    let raw = path
        .to_str()
        .ok_or_else(|| anyhow::anyhow!("unsafe sync directory value: paths must be valid UTF-8"))?;
    if raw.chars().any(char::is_control) {
        bail!("unsafe sync directory value: control characters are not allowed");
    }
    if path
        .components()
        .any(|component| matches!(component, Component::ParentDir))
    {
        bail!("unsafe sync directory value: parent-directory traversal (`..`) is not allowed");
    }
    if path.is_absolute() {
        Ok(path.to_path_buf())
    } else {
        Ok(std::env::current_dir()?.join(path))
    }
}

fn discover_latest(context: &ThreadSyncContext) -> Result<Discovery> {
    let mut candidates = Vec::new();
    let mut diagnostics = Vec::new();
    let mut invalid_count = 0usize;
    let latest_path = latest_path(context);

    let latest_candidate = match fs::symlink_metadata(&latest_path) {
        Ok(_) => match read_candidate(&latest_path, context, CandidateSource::Latest) {
            Ok(candidate) => Some(candidate),
            Err(diagnostic) => {
                invalid_count += 1;
                diagnostics.push(diagnostic);
                None
            }
        },
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
        Err(error) => {
            invalid_count += 1;
            diagnostics.push(diagnostic(
                "warning",
                "malformed_latest",
                format!(
                    "Failed to inspect latest.json {}: {error}",
                    latest_path.display()
                ),
            ));
            None
        }
    };

    let capsules = read_capsule_paths(&capsules_dir(context))?;
    if capsules.limit_exceeded {
        diagnostics.push(diagnostic(
            "error",
            "capsule_scan_limit_exceeded",
            format!(
                "Refusing to scan more than {MAX_CAPSULE_SCAN_FILES} capsule file(s) under {}; run `ccd thread export --to-sync-dir` from a trusted checkout to prune the capsule directory.",
                capsules_dir(context).display()
            ),
        ));
        return Ok(Discovery {
            selected: None,
            status: "invalid",
            diagnostics,
        });
    }

    for path in capsules.paths {
        match read_candidate(&path, context, CandidateSource::Capsule) {
            Ok(candidate) => candidates.push(candidate),
            Err(diagnostic) => {
                invalid_count += 1;
                diagnostics.push(diagnostic);
            }
        }
    }

    candidates.sort_by(compare_candidates);
    let selected = candidates.pop();
    if let (Some(latest), Some(selected)) = (&latest_candidate, &selected) {
        if selected.summary.digest != latest.summary.digest {
            let code = if selected.summary.created_at_epoch_s >= latest.summary.created_at_epoch_s {
                "latest_index_stale"
            } else {
                "latest_without_capsule"
            };
            diagnostics.push(diagnostic(
                "warning",
                code,
                format!(
                    "latest.json points at digest {}, but the selected complete capsule is {}.",
                    latest.summary.digest, selected.summary.digest
                ),
            ));
        }
    } else if latest_candidate.is_none() && selected.is_some() {
        diagnostics.push(diagnostic(
            "warning",
            "latest_index_missing",
            "latest.json is missing; selected the newest valid capsule file.".to_owned(),
        ));
    } else if latest_candidate.is_some() && selected.is_none() {
        diagnostics.push(diagnostic(
            "warning",
            "latest_without_capsule",
            "latest.json exists but no complete capsule file is available; ignoring latest.json until a capsule appears under capsules/."
                .to_owned(),
        ));
    }

    let status = if selected.is_some() {
        "available"
    } else if invalid_count > 0 || latest_candidate.is_some() {
        "invalid"
    } else {
        "empty"
    };

    Ok(Discovery {
        selected,
        status,
        diagnostics,
    })
}

fn read_capsule_paths(root: &Path) -> Result<CapsulePathList> {
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            return Ok(CapsulePathList {
                paths: Vec::new(),
                limit_exceeded: false,
            });
        }
        Err(error) => {
            return Err(error).with_context(|| format!("failed to read {}", root.display()));
        }
    };
    let mut paths = Vec::new();
    for entry in entries {
        let path = entry?.path();
        if path.extension().and_then(|value| value.to_str()) == Some("json") {
            if paths.len() >= MAX_CAPSULE_SCAN_FILES {
                return Ok(CapsulePathList {
                    paths,
                    limit_exceeded: true,
                });
            }
            paths.push(path);
        }
    }
    Ok(CapsulePathList {
        paths,
        limit_exceeded: false,
    })
}

fn list_capsule_paths_unbounded(root: &Path) -> Result<Vec<PathBuf>> {
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(error) => {
            return Err(error).with_context(|| format!("failed to read {}", root.display()));
        }
    };
    let mut paths = Vec::new();
    for entry in entries {
        let path = entry?.path();
        if path.extension().and_then(|value| value.to_str()) == Some("json") {
            paths.push(path);
        }
    }
    Ok(paths)
}

fn read_candidate(
    path: &Path,
    context: &ThreadSyncContext,
    source: CandidateSource,
) -> std::result::Result<CapsuleCandidate, ThreadDiagnostic> {
    let metadata = fs::symlink_metadata(path).map_err(|error| {
        diagnostic(
            "warning",
            diagnostic_code(source),
            format!(
                "Failed to inspect synced capsule {}: {error}",
                path.display()
            ),
        )
    })?;
    if metadata.file_type().is_symlink() {
        return Err(diagnostic(
            "warning",
            diagnostic_code(source),
            format!(
                "Rejected synced capsule {} because it is a symbolic link.",
                path.display()
            ),
        ));
    }
    if !metadata.is_file() {
        return Err(diagnostic(
            "warning",
            diagnostic_code(source),
            format!(
                "Rejected synced capsule {} because it is not a regular file.",
                path.display()
            ),
        ));
    }
    if metadata.len() > MAX_SYNC_CAPSULE_BYTES {
        return Err(diagnostic(
            "warning",
            "synced_capsule_too_large",
            format!(
                "Rejected synced capsule {} because it is {} bytes, above the {} byte sync limit.",
                path.display(),
                metadata.len(),
                MAX_SYNC_CAPSULE_BYTES
            ),
        ));
    }

    let file = fs::File::open(path).map_err(|error| {
        diagnostic(
            "warning",
            diagnostic_code(source),
            format!("Failed to open synced capsule {}: {error}", path.display()),
        )
    })?;
    let mut contents = String::new();
    file.take(MAX_SYNC_CAPSULE_BYTES + 1)
        .read_to_string(&mut contents)
        .map_err(|error| {
            diagnostic(
                "warning",
                diagnostic_code(source),
                format!("Failed to read synced capsule {}: {error}", path.display()),
            )
        })?;
    if contents.len() as u64 > MAX_SYNC_CAPSULE_BYTES {
        return Err(diagnostic(
            "warning",
            "synced_capsule_too_large",
            format!(
                "Rejected synced capsule {} because it exceeded the {} byte sync limit while reading.",
                path.display(),
                MAX_SYNC_CAPSULE_BYTES
            ),
        ));
    }

    let digest = format!("sha256:{}", thread_transfer::sha256_hex(&contents));
    let value: JsonValue = serde_json::from_str(&contents).map_err(|error| {
        diagnostic(
            "warning",
            diagnostic_code(source),
            format!("Failed to parse synced capsule {}: {error}", path.display()),
        )
    })?;
    let summary =
        validate_capsule_summary(&value, &digest, context, Some(path)).map_err(|error| {
            diagnostic(
                "warning",
                error.code,
                format!(
                    "Rejected synced capsule {}: {}",
                    path.display(),
                    error.message
                ),
            )
        })?;
    Ok(CapsuleCandidate {
        path: path.to_path_buf(),
        summary,
    })
}

fn validate_capsule_summary(
    value: &JsonValue,
    digest: &str,
    context: &ThreadSyncContext,
    path: Option<&Path>,
) -> std::result::Result<CapsuleSummary, CapsuleValidationError> {
    let kind = string_at(value, "/kind")
        .ok_or_else(|| validation_error("invalid_capsule", "missing kind"))?;
    if kind != CAPSULE_KIND {
        return Err(validation_error(
            "invalid_capsule",
            format!("unsupported kind `{kind}`; expected `{CAPSULE_KIND}`"),
        ));
    }
    let schema_version = value
        .pointer("/schema_version")
        .and_then(JsonValue::as_u64)
        .ok_or_else(|| validation_error("invalid_capsule", "missing schema_version"))?;
    if schema_version > CAPSULE_SCHEMA_VERSION as u64 {
        return Err(validation_error(
            "peer_schema_ahead",
            format!(
                "synced capsule schema_version {schema_version} is newer than supported {CAPSULE_SCHEMA_VERSION}; upgrade CCD before importing it"
            ),
        ));
    }
    if schema_version != CAPSULE_SCHEMA_VERSION as u64 {
        return Err(validation_error(
            "invalid_capsule",
            format!(
                "unsupported schema_version {schema_version}; expected {CAPSULE_SCHEMA_VERSION}"
            ),
        ));
    }
    let thread_id = string_at(value, "/thread/thread_id")
        .or_else(|| string_at(value, "/thread/id"))
        .ok_or_else(|| validation_error("invalid_capsule", "missing thread.thread_id"))?;
    let source_profile = string_at(value, "/thread/source_profile")
        .ok_or_else(|| validation_error("invalid_capsule", "missing thread.source_profile"))?;
    if source_profile != context.profile.as_str() {
        return Err(validation_error(
            "profile_mismatch",
            format!(
                "source_profile `{source_profile}` does not match profile `{}`",
                context.profile
            ),
        ));
    }
    let source_project_id = string_at(value, "/thread/source_project_id")
        .or_else(|| string_at(value, "/thread/source_locality_id"))
        .ok_or_else(|| validation_error("invalid_capsule", "missing thread.source_project_id"))?;
    let created_at_epoch_s = value
        .pointer("/thread/created_at_epoch_s")
        .and_then(JsonValue::as_u64)
        .ok_or_else(|| validation_error("invalid_capsule", "missing thread.created_at_epoch_s"))?;
    let attachment_count = value
        .pointer("/attachments")
        .and_then(JsonValue::as_array)
        .map(Vec::len)
        .ok_or_else(|| validation_error("invalid_capsule", "missing attachments array"))?;
    Ok(CapsuleSummary {
        path: path
            .map(|path| path.display().to_string())
            .unwrap_or_default(),
        digest: digest.to_owned(),
        created_at_epoch_s,
        thread_id,
        source_profile,
        source_project_id,
        source_workspace: string_at(value, "/thread/source_workspace"),
        attachment_count,
    })
}

fn compare_candidates(a: &CapsuleCandidate, b: &CapsuleCandidate) -> Ordering {
    a.summary
        .created_at_epoch_s
        .cmp(&b.summary.created_at_epoch_s)
        .then_with(|| a.path.cmp(&b.path))
}

fn diagnostic_code(source: CandidateSource) -> &'static str {
    match source {
        CandidateSource::Latest => "malformed_latest",
        CandidateSource::Capsule => "malformed_capsule",
    }
}

fn prune_old_capsules(context: &ThreadSyncContext, protected_path: &Path) -> Result<RetentionView> {
    let mut all_paths = list_capsule_paths_unbounded(&capsules_dir(context))?;
    all_paths.sort_by(|a, b| b.file_name().cmp(&a.file_name()).then_with(|| b.cmp(a)));
    let mut removed_count = 0usize;
    let mut kept_paths = Vec::new();
    let mut other_paths = Vec::new();
    for path in all_paths {
        if path == protected_path {
            kept_paths.push(path);
        } else {
            other_paths.push(path);
        }
    }
    for path in other_paths {
        if kept_paths.len() < context.max_capsule_files_per_thread {
            kept_paths.push(path);
        } else {
            match fs::remove_file(&path) {
                Ok(()) => removed_count += 1,
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => {
                    return Err(error)
                        .with_context(|| format!("failed to prune capsule {}", path.display()));
                }
            }
        }
    }

    let mut protected_candidates = Vec::new();
    let mut other_candidates = Vec::new();
    for path in kept_paths {
        let Some(candidate) = read_candidate(&path, context, CandidateSource::Capsule)
            .ok()
            .map(|candidate| {
                (
                    candidate.path,
                    candidate.summary.created_at_epoch_s,
                    candidate.summary.digest,
                )
            })
        else {
            continue;
        };
        if candidate.0 == protected_path {
            protected_candidates.push(candidate);
        } else {
            other_candidates.push(candidate);
        }
    }
    protected_candidates.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| b.2.cmp(&a.2)));
    other_candidates.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| b.2.cmp(&a.2)));
    let protected_count = protected_candidates.len().min(1);
    let keep_other_count = context
        .max_capsules_per_thread
        .saturating_sub(protected_count);
    for (path, _, _) in protected_candidates
        .into_iter()
        .skip(protected_count)
        .chain(other_candidates.into_iter().skip(keep_other_count))
    {
        match fs::remove_file(&path) {
            Ok(()) => removed_count += 1,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => {
                return Err(error)
                    .with_context(|| format!("failed to prune old capsule {}", path.display()));
            }
        }
    }
    Ok(RetentionView {
        max_capsules: context.max_capsules_per_thread,
        max_capsule_files: context.max_capsule_files_per_thread,
        removed_count,
    })
}

fn atomic_write_json(context: &ThreadSyncContext, path: &Path, value: &JsonValue) -> Result<()> {
    atomic_write_bytes(context, path, &serde_json::to_vec_pretty(value)?)
}

fn atomic_write_bytes(context: &ThreadSyncContext, path: &Path, bytes: &[u8]) -> Result<()> {
    let parent = path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("cannot write path without parent: {}", path.display()))?;
    create_dir_all_no_symlink(&context.sync_dir, parent)?;
    let tmp_dir = parent.join("tmp");
    create_dir_all_no_symlink(&context.sync_dir, &tmp_dir)?;
    let tmp = tmp_dir.join(format!(
        "{}-{}-{}.tmp",
        safe_path_segment(
            path.file_name()
                .and_then(|value| value.to_str())
                .unwrap_or("thread-sync")
        ),
        std::process::id(),
        session_state::now_epoch_s()?
    ));
    {
        let mut file = fs::File::create(&tmp)
            .with_context(|| format!("failed to create {}", tmp.display()))?;
        file.write_all(bytes)
            .with_context(|| format!("failed to write {}", tmp.display()))?;
        file.sync_all()
            .with_context(|| format!("failed to sync {}", tmp.display()))?;
    }
    fs::rename(&tmp, path)
        .with_context(|| format!("failed to rename {} to {}", tmp.display(), path.display()))?;
    Ok(())
}

fn create_dir_all_no_symlink(sync_root: &Path, path: &Path) -> Result<()> {
    if !path.starts_with(sync_root) {
        bail!(
            "unsafe sync path {}; expected path under configured sync root {}",
            path.display(),
            sync_root.display()
        );
    }

    fs::create_dir_all(sync_root)
        .with_context(|| format!("failed to create {}", sync_root.display()))?;
    let root_metadata = fs::symlink_metadata(sync_root)
        .with_context(|| format!("failed to inspect {}", sync_root.display()))?;
    if root_metadata.file_type().is_symlink() {
        bail!(
            "unsafe sync path {}; refusing to write through symbolic link {}",
            path.display(),
            sync_root.display()
        );
    }
    if !root_metadata.is_dir() {
        bail!(
            "unsafe sync path {}; {} exists but is not a directory",
            path.display(),
            sync_root.display()
        );
    }

    let relative = path.strip_prefix(sync_root).with_context(|| {
        format!(
            "failed to resolve {} relative to sync root {}",
            path.display(),
            sync_root.display()
        )
    })?;
    let mut current = sync_root.to_path_buf();
    for component in relative.components() {
        match component {
            Component::Prefix(_) | Component::RootDir => {
                bail!(
                    "unsafe sync path {}; absolute child path is not allowed",
                    path.display()
                );
            }
            Component::CurDir => continue,
            Component::ParentDir => bail!(
                "unsafe sync path {}; parent-directory traversal (`..`) is not allowed",
                path.display()
            ),
            Component::Normal(part) => current.push(part),
        }
        if current.as_os_str().is_empty() {
            continue;
        }
        match fs::symlink_metadata(&current) {
            Ok(metadata) if metadata.file_type().is_symlink() => {
                bail!(
                    "unsafe sync path {}; refusing to write through symbolic link {}",
                    path.display(),
                    current.display()
                );
            }
            Ok(metadata) if metadata.is_dir() => {}
            Ok(_) => {
                bail!(
                    "unsafe sync path {}; {} exists but is not a directory",
                    path.display(),
                    current.display()
                );
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                fs::create_dir(&current)
                    .with_context(|| format!("failed to create {}", current.display()))?;
            }
            Err(error) => {
                return Err(error)
                    .with_context(|| format!("failed to inspect {}", current.display()));
            }
        }
    }
    Ok(())
}

fn capsules_dir(context: &ThreadSyncContext) -> PathBuf {
    context.thread_dir.join("capsules")
}

fn latest_path(context: &ThreadSyncContext) -> PathBuf {
    context.thread_dir.join("latest.json")
}

fn index_path(context: &ThreadSyncContext) -> PathBuf {
    context.thread_dir.join("index.json")
}

fn identity_path(context: &ThreadSyncContext) -> PathBuf {
    context.thread_dir.join("identity.json")
}

fn identity_json(context: &ThreadSyncContext) -> Result<JsonValue> {
    Ok(json!({
        "kind": IDENTITY_KIND,
        "schema_version": SYNC_SCHEMA_VERSION,
        "backend": "filesystem",
        "sync_project_id": &context.sync_project_id,
        "identity_source": context.sync_identity_source,
        "identity_material_digest": &context.sync_identity_material_digest,
        "identity_portable": context.sync_identity_portable,
        "profile": &context.profile,
        "thread_id": &context.thread_id,
        "source": {
            "locality_id": &context.locality_id,
            "workspace": context.repo_root.display().to_string(),
            "machine_id": &context.source_machine_id,
            "machine_slug": &context.source_machine_slug,
            "updated_at_epoch_s": session_state::now_epoch_s()?
        }
    }))
}

fn capsule_file_name(created_at_epoch_s: u64, machine_id: Option<&str>, digest: &str) -> String {
    let digest_prefix = digest
        .strip_prefix("sha256:")
        .unwrap_or(digest)
        .chars()
        .take(16)
        .collect::<String>();
    let machine = safe_path_segment(machine_id.unwrap_or("unknown-machine"));
    format!("{created_at_epoch_s}-{machine}-{digest_prefix}.json")
}

fn relative_capsule_path(path: &Path, thread_dir: &Path) -> String {
    path.strip_prefix(thread_dir)
        .unwrap_or(path)
        .display()
        .to_string()
}

fn string_at(value: &JsonValue, pointer: &str) -> Option<String> {
    value
        .pointer(pointer)
        .and_then(JsonValue::as_str)
        .map(str::to_owned)
}

fn safe_path_segment(value: &str) -> String {
    let mut segment = value
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
                ch
            } else {
                '_'
            }
        })
        .collect::<String>();
    if segment.is_empty() {
        segment = "unknown".to_owned();
    }
    segment.truncate(64);
    segment
}

fn validation_error(code: &'static str, message: impl Into<String>) -> CapsuleValidationError {
    CapsuleValidationError {
        code,
        message: message.into(),
    }
}

fn diagnostic(severity: &'static str, code: &'static str, message: String) -> ThreadDiagnostic {
    ThreadDiagnostic {
        severity,
        code,
        message,
    }
}
