use std::{
    borrow::Cow,
    collections::{BTreeMap, HashSet},
    path::{Path, PathBuf},
    sync::OnceLock,
};

use miette::{Result, miette};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use tokio::{
    fs,
    fs::OpenOptions,
    io::{AsyncBufReadExt, BufReader},
};

use crate::{
    daat_locus_paths::daat_locus_paths,
    persistence::{PersistenceFileMode, append_bytes_durable, write_bytes_atomic},
    workspace_app::paths::resolve_runtime_workspace_dir,
};

const MAX_SUMMARY_ITEMS: usize = 12;
const MAX_COMPACT_SUMMARY_CHARS: usize = 180;
const WORKFLOWS_DIR_NAME: &str = "workflows";
const PRIMITIVE_RUN_RECORDS_FILE_NAME: &str = "run_records.jsonl";
static PRIMITIVE_RUN_RECORDS_IO_LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();

mod builtin_primitive_bindings {
    include!(concat!(env!("OUT_DIR"), "/builtin_workflows.rs"));
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum PrimitiveOrigin {
    Builtin,
    Workspace,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveSpec {
    pub id: String,
    #[serde(default)]
    pub when_to_use: Vec<String>,
    #[serde(default)]
    pub preconditions: Vec<String>,
    #[serde(default)]
    pub primitive_steps: Vec<String>,
    #[serde(default)]
    pub done_criteria: Vec<String>,
    #[serde(default)]
    pub recovery: Vec<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveComposition {
    pub composition_id: String,
    pub primitive_ids: Vec<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum PrimitiveActivation {
    Single { primitive: PrimitiveSpec },
    Composition { composition: PrimitiveComposition },
}

impl PrimitiveSpec {
    fn normalize(mut self) -> Result<Self> {
        self.id = normalize_identifier(&self.id);
        self.when_to_use = normalize_string_list(self.when_to_use);
        self.preconditions = normalize_string_list(self.preconditions);
        self.primitive_steps = normalize_string_list(self.primitive_steps);
        self.done_criteria = normalize_string_list(self.done_criteria);
        self.recovery = normalize_string_list(self.recovery);

        if self.id.is_empty() {
            return Err(miette!("workflow.id cannot be empty"));
        }
        Ok(self)
    }

    pub fn primitive_summary(&self) -> PrimitiveSummary {
        PrimitiveSummary {
            id: self.id.clone(),
            origin: PrimitiveOrigin::Workspace,
            capability_summary: compact_list_summary(&self.primitive_steps, 2),
            inputs_summary: compact_list_summary(&self.preconditions, 2),
            outputs_summary: compact_list_summary(&self.done_criteria, 2),
            when_to_use_summary: compact_list_summary(&self.when_to_use, 1),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveSummary {
    pub id: String,
    pub origin: PrimitiveOrigin,
    #[serde(default)]
    pub capability_summary: String,
    #[serde(default)]
    pub inputs_summary: String,
    #[serde(default)]
    pub outputs_summary: String,
    #[serde(default)]
    pub when_to_use_summary: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveRoutingCatalog {
    pub primitive_ids: Vec<PrimitiveId>,
    pub relevant_primitives: Vec<PrimitiveSummary>,
    pub total_count: usize,
    pub relevant_omitted_count: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveId {
    pub id: String,
    pub origin: PrimitiveOrigin,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct NewPrimitiveSpec {
    pub id: String,
    #[serde(default)]
    pub when_to_use: Vec<String>,
    #[serde(default)]
    pub preconditions: Vec<String>,
    #[serde(default)]
    pub primitive_steps: Vec<String>,
    #[serde(default)]
    pub done_criteria: Vec<String>,
    #[serde(default)]
    pub recovery: Vec<String>,
}

impl NewPrimitiveSpec {
    pub fn into_workflow_spec(self) -> PrimitiveSpec {
        PrimitiveSpec {
            id: self.id,
            when_to_use: self.when_to_use,
            preconditions: self.preconditions,
            primitive_steps: self.primitive_steps,
            done_criteria: self.done_criteria,
            recovery: self.recovery,
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveRunRecord {
    pub run_id: String,
    pub workflow_id: String,
    pub started_at_ms: i64,
    pub ended_at_ms: i64,
    pub origin: String,
    pub outcome: PrimitiveRunOutcome,
    pub turn_count: usize,
    pub tool_action_count: usize,
    pub manual_fix_detected: bool,
    pub rollback_detected: bool,
    #[serde(default)]
    pub failure_types: Vec<String>,
    pub final_summary: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum PrimitiveRunOutcome {
    Completed,
    Blocked,
    Abandoned,
    Superseded,
    NoProgress,
}

pub struct PrimitiveRunBatch {
    pub records: Vec<PrimitiveRunRecord>,
    pub unread_record_count: usize,
    pub next_offset: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct PrimitiveSpecPatch {
    pub workflow_id: String,
    #[serde(default)]
    pub when_to_use_additions: Vec<String>,
    #[serde(default)]
    pub precondition_additions: Vec<String>,
    #[serde(default)]
    pub workflow_step_additions: Vec<String>,
    #[serde(default)]
    pub done_criteria_additions: Vec<String>,
    #[serde(default)]
    pub recovery_additions: Vec<String>,
}

struct StoredPrimitive {
    spec: PrimitiveSpec,
    path: Option<PathBuf>,
    origin: PrimitiveOrigin,
}

pub struct PrimitiveStore {
    workflow_dir: PathBuf,
    workflows: BTreeMap<String, StoredPrimitive>,
}

impl PrimitiveStore {
    pub async fn new() -> Self {
        let workflow_dir = resolve_runtime_workspace_dir()
            .unwrap()
            .join(WORKFLOWS_DIR_NAME);
        Self::open_scoped(workflow_dir).await
    }

    pub(crate) async fn open_scoped(workflow_dir: PathBuf) -> Self {
        let mut store = Self {
            workflow_dir,
            workflows: load_builtin_primitives(),
        };
        store.load_from_disk().await;
        store
    }

    pub fn get(&self, workflow_id: &str) -> Option<&PrimitiveSpec> {
        self.workflows.get(workflow_id).map(|stored| &stored.spec)
    }

    pub fn workflow_origin(&self, workflow_id: &str) -> Option<PrimitiveOrigin> {
        self.workflows.get(workflow_id).map(|stored| stored.origin)
    }

    pub fn workflow_path(&self, workflow_id: &str) -> Option<&Path> {
        self.workflows
            .get(workflow_id)
            .and_then(|stored| stored.path.as_deref())
    }

    pub fn workspace_list(&self) -> Vec<PrimitiveSpec> {
        self.workflows
            .values()
            .filter(|stored| stored.origin == PrimitiveOrigin::Workspace)
            .map(|stored| stored.spec.clone())
            .collect()
    }

    pub fn primitive_routing_catalog(&self, query: &str, limit: usize) -> PrimitiveRoutingCatalog {
        let query_terms = workflow_relevance_terms(query);
        let mut primitive_ids = self
            .workflows
            .values()
            .map(|stored| PrimitiveId {
                id: stored.spec.id.clone(),
                origin: stored.origin,
            })
            .collect::<Vec<_>>();
        primitive_ids.sort_by(|left, right| {
            workflow_origin_sort_key(left.origin)
                .cmp(&workflow_origin_sort_key(right.origin))
                .then_with(|| left.id.cmp(&right.id))
        });

        let mut scored = self
            .workflows
            .values()
            .filter_map(|stored| {
                if query_terms.is_empty() {
                    return None;
                }
                let score = workflow_route_score(&stored.spec, &query_terms);
                if score == 0 {
                    return None;
                }
                let mut summary = stored.spec.primitive_summary();
                summary.origin = stored.origin;
                Some((score, summary))
            })
            .collect::<Vec<_>>();
        scored.sort_by(|(left_score, left), (right_score, right)| {
            right_score
                .cmp(left_score)
                .then_with(|| {
                    workflow_origin_sort_key(left.origin)
                        .cmp(&workflow_origin_sort_key(right.origin))
                })
                .then_with(|| left.id.cmp(&right.id))
        });

        let total_count = self.workflows.len();
        let take_limit = limit.min(MAX_SUMMARY_ITEMS);
        let scored_count = scored.len();
        let relevant_primitives = scored
            .into_iter()
            .take(take_limit)
            .map(|(_, summary)| summary)
            .collect::<Vec<_>>();
        let relevant_omitted_count = scored_count.saturating_sub(relevant_primitives.len());
        PrimitiveRoutingCatalog {
            primitive_ids,
            relevant_primitives,
            total_count,
            relevant_omitted_count,
        }
    }

    pub fn activate_composed_primitive(&self, primitive_id: &str) -> Result<PrimitiveActivation> {
        let primitive_id = primitive_id.trim();
        if primitive_id.is_empty() {
            return Err(miette!(
                "activate_composed_primitive requires non-empty primitive_id"
            ));
        }
        if let Some(stored) = self.workflows.get(primitive_id) {
            return Ok(PrimitiveActivation::Single {
                primitive: stored.spec.clone(),
            });
        }

        let composition = self.compose_primitives(primitive_id)?;
        Ok(PrimitiveActivation::Composition { composition })
    }

    pub fn compose_primitives(&self, composition: &str) -> Result<PrimitiveComposition> {
        let primitive_ids = parse_primitive_composition(composition, |candidate| {
            self.workflows.contains_key(candidate)
        })?;
        Ok(PrimitiveComposition {
            composition_id: primitive_ids.join("-"),
            primitive_ids,
        })
    }

    pub async fn create_workflow(&mut self, draft: NewPrimitiveSpec) -> Result<PrimitiveSpec> {
        if draft.id.trim().is_empty() {
            return Err(miette!("create_primitive_spec requires non-empty id"));
        }
        if !is_valid_primitive_id(draft.id.trim()) {
            return Err(miette!(
                "primitive_id `{}` is invalid; primitive ids may only use lowercase a-z and '-'",
                draft.id
            ));
        }
        if draft.when_to_use.is_empty() {
            return Err(miette!(
                "create_primitive_spec requires at least one when_to_use item"
            ));
        }
        if draft.done_criteria.is_empty() {
            return Err(miette!(
                "create_primitive_spec requires at least one done_criteria item"
            ));
        }

        let spec = draft.into_workflow_spec().normalize()?;
        if self.workflows.contains_key(&spec.id) {
            return Err(miette!("primitive_id `{}` already exists", spec.id));
        }
        let path = self.workflow_dir.join(format!("{}.md", spec.id));
        write_workflow_file(&path, &spec).await?;
        self.workflows.insert(
            spec.id.clone(),
            StoredPrimitive {
                spec: spec.clone(),
                path: Some(path),
                origin: PrimitiveOrigin::Workspace,
            },
        );
        Ok(spec)
    }

    pub async fn apply_patch(&mut self, patch: PrimitiveSpecPatch) -> Result<PrimitiveSpec> {
        let stored = self
            .workflows
            .get_mut(&patch.workflow_id)
            .ok_or_else(|| miette!("unknown workflow_id `{}`", patch.workflow_id))?;
        if stored.origin != PrimitiveOrigin::Workspace {
            return Err(miette!(
                "builtin primitive `{}` is read-only and cannot be patched",
                patch.workflow_id
            ));
        }
        let path = stored.path.clone().ok_or_else(|| {
            miette!(
                "workspace primitive `{}` is missing backing path",
                patch.workflow_id
            )
        })?;

        let before = stored.spec.clone();
        extend_unique(
            &mut stored.spec.when_to_use,
            normalize_string_list(patch.when_to_use_additions),
        );
        extend_unique(
            &mut stored.spec.preconditions,
            normalize_string_list(patch.precondition_additions),
        );
        extend_unique(
            &mut stored.spec.primitive_steps,
            normalize_string_list(patch.workflow_step_additions),
        );
        extend_unique(
            &mut stored.spec.done_criteria,
            normalize_string_list(patch.done_criteria_additions),
        );
        extend_unique(
            &mut stored.spec.recovery,
            normalize_string_list(patch.recovery_additions),
        );

        stored.spec = stored.spec.clone().normalize()?;
        if !workflow_content_equal(&before, &stored.spec) {
            write_workflow_file(&path, &stored.spec).await?;
        }

        Ok(stored.spec.clone())
    }

    pub async fn replace_workspace_workflow(
        &mut self,
        workflow_id: &str,
        replacement: PrimitiveSpec,
    ) -> Result<PrimitiveSpec> {
        let workflow_id = workflow_id.trim().to_string();
        if workflow_id.is_empty() {
            return Err(miette!("replace_workspace_workflow requires non-empty id"));
        }
        if !is_valid_primitive_id(&workflow_id) {
            return Err(miette!(
                "primitive_id `{workflow_id}` is invalid; primitive ids may only use lowercase a-z and '-'"
            ));
        }

        let stored = self
            .workflows
            .get_mut(&workflow_id)
            .ok_or_else(|| miette!("unknown primitive_id `{workflow_id}`"))?;
        if stored.origin != PrimitiveOrigin::Workspace {
            return Err(miette!(
                "builtin primitive `{workflow_id}` is read-only and cannot be updated"
            ));
        }
        let path = stored.path.clone().ok_or_else(|| {
            miette!("workspace primitive `{workflow_id}` is missing backing path")
        })?;

        if !is_valid_primitive_id(replacement.id.trim()) {
            return Err(miette!(
                "replacement primitive id `{}` is invalid; primitive ids may only use lowercase a-z and '-'",
                replacement.id
            ));
        }
        let replacement = replacement.normalize()?;
        if replacement.id != workflow_id {
            return Err(miette!(
                "replacement primitive id `{}` does not match target primitive_id `{workflow_id}`",
                replacement.id
            ));
        }
        if replacement.when_to_use.is_empty() {
            return Err(miette!(
                "update_primitive_spec requires at least one when_to_use item"
            ));
        }
        if replacement.primitive_steps.is_empty() {
            return Err(miette!(
                "update_primitive_spec requires at least one primitive_steps item"
            ));
        }
        if replacement.done_criteria.is_empty() {
            return Err(miette!(
                "update_primitive_spec requires at least one done_criteria item"
            ));
        }

        let before = stored.spec.clone();
        stored.spec = replacement;
        if !workflow_content_equal(&before, &stored.spec) {
            write_workflow_file(&path, &stored.spec).await?;
        }

        Ok(stored.spec.clone())
    }

    pub async fn merge_workflows(
        &mut self,
        target_workflow_id: &str,
        source_workflow_ids: &[String],
        _reason: Option<String>,
    ) -> Result<PrimitiveSpec> {
        if !self.workflows.contains_key(target_workflow_id) {
            return Err(miette!("unknown target workflow_id `{target_workflow_id}`"));
        }
        if self.workflow_origin(target_workflow_id) != Some(PrimitiveOrigin::Workspace) {
            return Err(miette!(
                "builtin primitive `{target_workflow_id}` is read-only and cannot be merged"
            ));
        }

        let source_ids = source_workflow_ids
            .iter()
            .map(|item| normalize_identifier(item))
            .filter(|item| !item.is_empty() && item != target_workflow_id)
            .collect::<Vec<_>>();
        if source_ids.is_empty() {
            return Err(miette!(
                "merge_workflows requires at least one source workflow"
            ));
        }

        let sources = source_ids
            .iter()
            .map(|source_id| {
                self.workflows
                    .get(source_id)
                    .filter(|stored| stored.origin == PrimitiveOrigin::Workspace)
                    .map(|stored| stored.spec.clone())
                    .ok_or_else(|| miette!("unknown source workflow_id `{source_id}`"))
            })
            .collect::<Result<Vec<_>>>()?;

        let target = self
            .workflows
            .get_mut(target_workflow_id)
            .ok_or_else(|| miette!("unknown target workflow_id `{target_workflow_id}`"))?;
        let target_path = target.path.clone().ok_or_else(|| {
            miette!("workspace primitive `{target_workflow_id}` is missing backing path")
        })?;

        for source in &sources {
            extend_unique(
                &mut target.spec.when_to_use,
                normalize_string_list(source.when_to_use.clone()),
            );
            extend_unique(
                &mut target.spec.preconditions,
                normalize_string_list(source.preconditions.clone()),
            );
            extend_unique(
                &mut target.spec.primitive_steps,
                normalize_string_list(source.primitive_steps.clone()),
            );
            extend_unique(
                &mut target.spec.done_criteria,
                normalize_string_list(source.done_criteria.clone()),
            );
            extend_unique(
                &mut target.spec.recovery,
                normalize_string_list(source.recovery.clone()),
            );
        }

        target.spec = target.spec.clone().normalize()?;
        write_workflow_file(&target_path, &target.spec).await?;

        for source_id in &source_ids {
            if let Some(stored) = self.workflows.remove(source_id)
                && let Some(path) = stored.path
            {
                let _ = tokio::fs::remove_file(path).await;
            }
        }

        self.workflows
            .get(target_workflow_id)
            .map(|stored| stored.spec.clone())
            .ok_or_else(|| miette!("unknown target workflow_id `{target_workflow_id}`"))
    }

    pub async fn shutdown(self) {}

    async fn load_from_disk(&mut self) {
        let _ = tokio::fs::create_dir_all(&self.workflow_dir).await;
        let Ok(mut entries) = tokio::fs::read_dir(&self.workflow_dir).await else {
            return;
        };
        while let Ok(Some(entry)) = entries.next_entry().await {
            let path = entry.path();
            if path.extension().and_then(|ext| ext.to_str()) != Some("md") {
                continue;
            }
            let Some(file_id) = primitive_id_from_path(&path) else {
                tracing::warn!(
                    "workspace primitive filename `{}` is invalid; primitive ids may only use lowercase a-z and '-'",
                    path.display()
                );
                continue;
            };
            let Ok(content) = tokio::fs::read_to_string(&path).await else {
                continue;
            };
            match parse_workflow_file(&content, Some(&file_id)) {
                Ok(spec) => {
                    if self.workflows.contains_key(&spec.id) {
                        tracing::warn!(
                            "workspace primitive id `{}` conflicts with existing builtin/workspace definition at {}; skipping",
                            spec.id,
                            path.display()
                        );
                        continue;
                    }
                    self.workflows.insert(
                        spec.id.clone(),
                        StoredPrimitive {
                            spec,
                            path: Some(path),
                            origin: PrimitiveOrigin::Workspace,
                        },
                    );
                }
                Err(err) => {
                    tracing::warn!(
                        "failed to parse primitive spec file {}: {err:?}",
                        path.display()
                    );
                }
            }
        }
    }
}

fn load_builtin_primitives() -> BTreeMap<String, StoredPrimitive> {
    let mut workflows = BTreeMap::new();
    for (source_name, content) in builtin_primitive_bindings::BUILTIN_PRIMITIVE_SOURCES {
        if !is_valid_primitive_id(source_name) {
            tracing::warn!(
                "builtin primitive filename `{source_name}` is invalid; primitive ids may only use lowercase a-z and '-'"
            );
            continue;
        }
        match parse_workflow_file(content, Some(source_name)) {
            Ok(spec) => {
                if workflows.contains_key(&spec.id) {
                    tracing::warn!(
                        "duplicate builtin primitive id `{}` detected in source {}; keeping first definition",
                        spec.id,
                        source_name
                    );
                    continue;
                }
                workflows.insert(
                    spec.id.clone(),
                    StoredPrimitive {
                        spec,
                        path: None,
                        origin: PrimitiveOrigin::Builtin,
                    },
                );
            }
            Err(err) => {
                tracing::warn!(
                    "failed to parse builtin primitive source {}: {err:?}",
                    source_name
                );
            }
        }
    }
    workflows
}

pub async fn load_primitive_run_batch() -> Result<PrimitiveRunBatch> {
    let workflow_run_records_io_guard = workflow_run_records_io_lock().lock().await;
    let path = workflow_run_records_file_path().await;
    let bytes = match fs::read(&path).await {
        Ok(bytes) => bytes,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            drop(workflow_run_records_io_guard);
            return Ok(PrimitiveRunBatch {
                records: Vec::new(),
                unread_record_count: 0,
                next_offset: 0,
            });
        }
        Err(err) => {
            drop(workflow_run_records_io_guard);
            return Err(miette!(
                "failed to read workflow run records {}: {err}",
                path.display()
            ));
        }
    };

    let mut offset = 0u64;
    let mut records = Vec::new();
    for chunk in bytes.split_inclusive(|byte| *byte == b'\n') {
        offset += chunk.len() as u64;
        let line = std::str::from_utf8(chunk)
            .map(str::trim)
            .unwrap_or_default();
        if line.is_empty() {
            continue;
        }
        let record: PrimitiveRunRecord = serde_json::from_str(line).map_err(|err| {
            miette!(
                "failed to parse workflow run record from {}: {err}",
                path.display()
            )
        })?;
        records.push(record);
    }
    let unread_record_count = records.len();
    drop(workflow_run_records_io_guard);
    Ok(PrimitiveRunBatch {
        records,
        unread_record_count,
        next_offset: offset,
    })
}

pub async fn primitive_run_record_count() -> Result<usize> {
    let workflow_run_records_io_guard = workflow_run_records_io_lock().lock().await;
    let path = workflow_run_records_file_path().await;
    let file = match OpenOptions::new().read(true).open(&path).await {
        Ok(file) => file,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            drop(workflow_run_records_io_guard);
            return Ok(0);
        }
        Err(err) => {
            drop(workflow_run_records_io_guard);
            return Err(miette!(
                "failed to open workflow run records {}: {err}",
                path.display()
            ));
        }
    };
    let mut lines = BufReader::new(file).lines();
    let mut records = 0usize;
    while let Some(line) = lines.next_line().await.map_err(|err| {
        miette!(
            "failed to read workflow run records {}: {err}",
            path.display()
        )
    })? {
        if !line.trim().is_empty() {
            records += 1;
        }
    }
    drop(workflow_run_records_io_guard);
    Ok(records)
}

pub async fn append_primitive_run_records(records: &[PrimitiveRunRecord]) -> Result<usize> {
    if records.is_empty() {
        return Ok(0);
    }

    let workflow_run_records_io_guard = workflow_run_records_io_lock().lock().await;
    let path = workflow_run_records_file_path().await;
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).await.map_err(|err| {
            miette!(
                "failed to create workflow run record directory {}: {err}",
                parent.display()
            )
        })?;
    }

    let mut existing_ids = HashSet::new();
    if let Ok(file) = OpenOptions::new().read(true).open(&path).await {
        let mut lines = BufReader::new(file).lines();
        while let Some(line) = lines.next_line().await.map_err(|err| {
            miette!(
                "failed to read workflow run records for dedupe {}: {err}",
                path.display()
            )
        })? {
            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }
            let record: PrimitiveRunRecord = serde_json::from_str(trimmed).map_err(|err| {
                miette!(
                    "failed to parse workflow run record during dedupe {}: {err}",
                    path.display()
                )
            })?;
            existing_ids.insert(record.run_id);
        }
    }

    let mut appended = 0usize;
    let mut batch = Vec::new();
    for record in records {
        if !existing_ids.insert(record.run_id.clone()) {
            continue;
        }
        let mut bytes = serde_json::to_vec(record)
            .map_err(|err| miette!("failed to serialize workflow run record: {err}"))?;
        bytes.push(b'\n');
        batch.extend(bytes);
        appended += 1;
    }
    if !batch.is_empty() {
        append_bytes_durable(path.clone(), batch)
            .await
            .map_err(|err| {
                miette!(
                    "failed to append workflow run record {}: {err}",
                    path.display()
                )
            })?;
    }
    drop(workflow_run_records_io_guard);
    Ok(appended)
}

pub async fn compact_workflow_run_record_file(consumed_offset: u64) -> Result<()> {
    let workflow_run_records_io_guard = workflow_run_records_io_lock().lock().await;
    let path = workflow_run_records_file_path().await;
    let bytes = match fs::read(&path).await {
        Ok(bytes) => bytes,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            drop(workflow_run_records_io_guard);
            return Ok(());
        }
        Err(err) => {
            drop(workflow_run_records_io_guard);
            return Err(miette!(
                "failed to read workflow run records {} for compaction: {err}",
                path.display()
            ));
        }
    };
    let keep_from = (consumed_offset as usize).min(bytes.len());
    write_bytes_atomic(
        path.clone(),
        bytes[keep_from..].to_vec(),
        PersistenceFileMode::Default,
    )
    .await
    .map_err(|err| {
        miette!(
            "failed to rewrite workflow run records {} during compaction: {err}",
            path.display()
        )
    })?;
    drop(workflow_run_records_io_guard);
    Ok(())
}

fn parse_workflow_file(content: &str, expected_id: Option<&str>) -> Result<PrimitiveSpec> {
    let content = normalize_workflow_line_endings(content);
    let body = strip_optional_frontmatter(content.as_ref());
    let sections = parse_markdown_sections(body);
    let Some(expected_id) = expected_id else {
        return Err(miette!(
            "parse_workflow_file requires filename primitive id"
        ));
    };
    let spec = PrimitiveSpec {
        id: expected_id.to_string(),
        when_to_use: parse_markdown_list(sections.get("When To Use")),
        preconditions: parse_markdown_list(sections.get("Preconditions")),
        primitive_steps: parse_markdown_list(sections.get("Workflow")),
        done_criteria: parse_markdown_list(sections.get("Done Criteria")),
        recovery: parse_markdown_list(sections.get("Recovery")),
    }
    .normalize()?;
    if spec.id != expected_id {
        return Err(miette!(
            "primitive filename id `{expected_id}` normalizes to `{}`, but primitive filenames must already use lowercase a-z and '-' only",
            spec.id
        ));
    }
    Ok(spec)
}

fn strip_optional_frontmatter(content: &str) -> &str {
    let Some(rest) = content.strip_prefix("---\n") else {
        return content;
    };
    rest.split_once("\n---\n")
        .map(|(_, body)| body)
        .unwrap_or(content)
}

fn normalize_workflow_line_endings(content: &str) -> Cow<'_, str> {
    if content.contains('\r') {
        Cow::Owned(content.replace("\r\n", "\n").replace('\r', "\n"))
    } else {
        Cow::Borrowed(content)
    }
}

async fn write_workflow_file(path: &Path, spec: &PrimitiveSpec) -> Result<()> {
    if let Some(parent) = path.parent() {
        tokio::fs::create_dir_all(parent).await.map_err(|err| {
            miette!(
                "create primitive spec directory {} failed: {err}",
                parent.display()
            )
        })?;
    }

    let body = render_workflow_markdown_body(spec);
    write_bytes_atomic(
        path.to_path_buf(),
        body.into_bytes(),
        PersistenceFileMode::Default,
    )
    .await
    .map_err(|err| miette!("write primitive spec file {} failed: {err}", path.display()))
}

fn render_workflow_markdown_body(spec: &PrimitiveSpec) -> String {
    [
        render_section("When To Use", &spec.when_to_use, false),
        render_section("Preconditions", &spec.preconditions, false),
        render_section("Workflow", &spec.primitive_steps, true),
        render_section("Done Criteria", &spec.done_criteria, false),
        render_section("Recovery", &spec.recovery, false),
    ]
    .join("\n\n")
}

fn render_section(title: &str, items: &[String], ordered: bool) -> String {
    let mut lines = vec![format!("## {title}")];
    if items.is_empty() {
        lines.push("- <empty>".to_string());
    } else if ordered {
        lines.extend(
            items
                .iter()
                .enumerate()
                .map(|(index, item)| format!("{}. {}", index + 1, item)),
        );
    } else {
        lines.extend(items.iter().map(|item| format!("- {item}")));
    }
    lines.join("\n")
}

fn parse_markdown_sections(body: &str) -> BTreeMap<String, String> {
    let mut sections = BTreeMap::new();
    let mut current: Option<String> = None;
    let mut current_lines = Vec::new();

    for line in body.lines() {
        if let Some(title) = line.trim().strip_prefix("## ") {
            if let Some(current_title) = current.replace(title.trim().to_string()) {
                sections.insert(current_title, current_lines.join("\n"));
                current_lines.clear();
            }
            continue;
        }
        if current.is_some() {
            current_lines.push(line.to_string());
        }
    }

    if let Some(current_title) = current {
        sections.insert(current_title, current_lines.join("\n"));
    }
    sections
}

fn parse_markdown_list(section: Option<&String>) -> Vec<String> {
    let Some(section) = section else {
        return Vec::new();
    };
    section
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .filter_map(|line| {
            if let Some(item) = line.strip_prefix("- ") {
                return Some(item.trim().to_string());
            }
            if let Some((prefix, rest)) = line.split_once(". ")
                && prefix.chars().all(|ch| ch.is_ascii_digit())
            {
                return Some(rest.trim().to_string());
            }
            None
        })
        .filter(|item| item != "<empty>")
        .collect()
}

fn workflow_content_equal(left: &PrimitiveSpec, right: &PrimitiveSpec) -> bool {
    left.id == right.id
        && left.when_to_use == right.when_to_use
        && left.preconditions == right.preconditions
        && left.primitive_steps == right.primitive_steps
        && left.done_criteria == right.done_criteria
        && left.recovery == right.recovery
}

fn compact_list_summary(items: &[String], limit: usize) -> String {
    if items.is_empty() || limit == 0 {
        return String::new();
    }
    let mut parts = items
        .iter()
        .take(limit)
        .map(|item| compact_summary_text(item))
        .collect::<Vec<_>>();
    if items.len() > limit {
        parts.push(format!("+{} more", items.len() - limit));
    }
    parts.join("; ")
}

fn compact_summary_text(text: &str) -> String {
    let compact = text.split_whitespace().collect::<Vec<_>>().join(" ");
    if compact.chars().count() <= MAX_COMPACT_SUMMARY_CHARS {
        return compact;
    }
    let head = compact
        .chars()
        .take(MAX_COMPACT_SUMMARY_CHARS)
        .collect::<String>();
    format!("{head}...")
}

fn workflow_route_score(spec: &PrimitiveSpec, query_terms: &[String]) -> usize {
    if query_terms.is_empty() {
        return 0;
    }
    let candidate = workflow_relevance_text(spec);
    query_terms
        .iter()
        .filter(|term| candidate.contains(term.as_str()))
        .count()
}

fn workflow_relevance_text(spec: &PrimitiveSpec) -> String {
    let mut parts = vec![spec.id.replace(['-', '_'], " ")];
    parts.extend(spec.when_to_use.iter().cloned());
    parts.extend(spec.preconditions.iter().cloned());
    parts.extend(spec.primitive_steps.iter().cloned());
    parts.extend(spec.done_criteria.iter().cloned());
    parts.extend(spec.recovery.iter().cloned());
    parts.join("\n").to_lowercase()
}

fn workflow_relevance_terms(query: &str) -> Vec<String> {
    let mut terms = HashSet::new();
    let mut ascii_run = String::new();
    let mut cjk_run = String::new();

    for ch in query.to_lowercase().chars() {
        if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
            flush_cjk_terms(&mut terms, &mut cjk_run);
            ascii_run.push(ch);
        } else if is_cjk_char(ch) {
            flush_ascii_term(&mut terms, &mut ascii_run);
            cjk_run.push(ch);
        } else {
            flush_ascii_term(&mut terms, &mut ascii_run);
            flush_cjk_terms(&mut terms, &mut cjk_run);
        }
    }
    flush_ascii_term(&mut terms, &mut ascii_run);
    flush_cjk_terms(&mut terms, &mut cjk_run);

    let mut terms = terms.into_iter().collect::<Vec<_>>();
    terms.sort();
    terms
}

fn flush_ascii_term(terms: &mut HashSet<String>, current: &mut String) {
    if current.chars().count() >= 2 && !is_stop_term(current) {
        terms.insert(current.clone());
    }
    current.clear();
}

fn flush_cjk_terms(terms: &mut HashSet<String>, current: &mut String) {
    let chars = current.chars().collect::<Vec<_>>();
    match chars.len() {
        0 => {}
        1 => {
            let term = chars[0].to_string();
            if !is_stop_term(&term) {
                terms.insert(term);
            }
        }
        _ => {
            let full = chars.iter().collect::<String>();
            if !is_stop_term(&full) {
                terms.insert(full);
            }
            for pair in chars.windows(2) {
                let term = pair.iter().collect::<String>();
                if !is_stop_term(&term) {
                    terms.insert(term);
                }
            }
        }
    }
    current.clear();
}

fn is_cjk_char(ch: char) -> bool {
    ('\u{4E00}'..='\u{9FFF}').contains(&ch)
        || ('\u{3400}'..='\u{4DBF}').contains(&ch)
        || ('\u{F900}'..='\u{FAFF}').contains(&ch)
}

fn is_stop_term(term: &str) -> bool {
    matches!(
        term,
        "a" | "an"
            | "and"
            | "are"
            | "as"
            | "for"
            | "in"
            | "is"
            | "it"
            | "of"
            | "on"
            | "or"
            | "the"
            | "to"
            | "user"
            | "with"
            | "好"
            | "的"
            | "了"
            | "就"
            | "按"
            | "此"
    )
}

fn workflow_origin_sort_key(origin: PrimitiveOrigin) -> u8 {
    match origin {
        PrimitiveOrigin::Builtin => 0,
        PrimitiveOrigin::Workspace => 1,
    }
}

fn normalize_identifier(value: &str) -> String {
    let mut normalized = String::new();
    let mut previous_was_dash = true;
    for ch in value.trim().chars() {
        if ch.is_ascii_alphabetic() {
            normalized.push(ch.to_ascii_lowercase());
            previous_was_dash = false;
        } else if (ch == '-' || ch.is_whitespace() || ch == '_') && !previous_was_dash {
            normalized.push('-');
            previous_was_dash = true;
        }
    }
    if previous_was_dash {
        normalized.pop();
    }
    normalized
}

fn is_valid_primitive_id(value: &str) -> bool {
    !value.is_empty()
        && !value.starts_with('-')
        && !value.ends_with('-')
        && value.chars().all(|ch| ch.is_ascii_lowercase() || ch == '-')
}

fn primitive_id_from_path(path: &Path) -> Option<String> {
    let file_id = path.file_stem()?.to_str()?;
    is_valid_primitive_id(file_id).then(|| file_id.to_string())
}

fn parse_primitive_composition(
    composition: &str,
    exists: impl Fn(&str) -> bool,
) -> Result<Vec<String>> {
    let composition = composition.trim();
    if composition.is_empty() {
        return Err(miette!(
            "activate_composed_primitive requires non-empty primitive_id"
        ));
    }
    if !is_valid_primitive_id(composition) {
        return Err(miette!(
            "primitive composition `{composition}` is invalid; use only lowercase a-z and '-'"
        ));
    }

    fn segment<F: Fn(&str) -> bool>(
        parts: &[&str],
        index: usize,
        exists: &F,
    ) -> Option<Vec<String>> {
        if index == parts.len() {
            return Some(Vec::new());
        }

        for end in (index + 1..=parts.len()).rev() {
            let candidate = parts[index..end].join("-");
            if !exists(&candidate) {
                continue;
            }
            if let Some(mut remainder) = segment(parts, end, exists) {
                let mut primitive_ids = Vec::with_capacity(remainder.len() + 1);
                primitive_ids.push(candidate);
                primitive_ids.append(&mut remainder);
                return Some(primitive_ids);
            }
        }
        None
    }

    let parts = composition.split('-').collect::<Vec<_>>();
    let Some(primitive_ids) = segment(&parts, 0, &exists) else {
        return Err(miette!(
            "primitive composition `{composition}` cannot be segmented into existing primitives"
        ));
    };

    if primitive_ids.len() < 2 {
        return Err(miette!(
            "activate_composed_primitive composition requires at least two existing primitive ids joined by '-'"
        ));
    }
    Ok(primitive_ids)
}

fn normalize_string_list(items: Vec<String>) -> Vec<String> {
    let mut seen = HashSet::new();
    let mut normalized = Vec::new();
    for item in items {
        let trimmed = item.trim();
        if trimmed.is_empty() {
            continue;
        }
        if seen.insert(trimmed.to_string()) {
            normalized.push(trimmed.to_string());
        }
    }
    normalized
}

fn extend_unique(target: &mut Vec<String>, additions: Vec<String>) {
    if additions.is_empty() {
        return;
    }
    let mut existing = target.iter().cloned().collect::<HashSet<_>>();
    for item in additions {
        if existing.insert(item.clone()) {
            target.push(item);
        }
    }
}

async fn workflow_run_records_file_path() -> PathBuf {
    daat_locus_paths()
        .await
        .state_dir()
        .join(WORKFLOWS_DIR_NAME)
        .join(PRIMITIVE_RUN_RECORDS_FILE_NAME)
}

fn workflow_run_records_io_lock() -> &'static tokio::sync::Mutex<()> {
    PRIMITIVE_RUN_RECORDS_IO_LOCK.get_or_init(|| tokio::sync::Mutex::new(()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::env;
    use tempfile::TempDir;

    #[tokio::test]
    async fn create_workflow_writes_primitive_markdown_and_can_reload() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary.clone()).await;
        let created = store
            .create_workflow(NewPrimitiveSpec {
                id: "repair-flaky-test-pipeline".to_string(),
                when_to_use: vec!["flaky test failure".to_string()],
                preconditions: vec!["failing logs available".to_string()],
                primitive_steps: vec!["collect evidence".to_string(), "verify fix".to_string()],
                done_criteria: vec!["result is stable".to_string()],
                recovery: vec!["return to previous stable state".to_string()],
            })
            .await
            .expect("create primitive spec");

        assert_eq!(created.id, "repair-flaky-test-pipeline");
        assert!(primary.join("repair-flaky-test-pipeline.md").exists());

        let reloaded = PrimitiveStore::open_scoped(primary.clone()).await;
        let loaded = reloaded
            .get("repair-flaky-test-pipeline")
            .expect("reloaded primitive spec");
        assert_eq!(loaded.primitive_steps.len(), 2);
    }

    #[tokio::test]
    async fn primitive_routing_catalog_includes_all_ids_and_ranks_relevant_primitives() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary).await;
        store
            .create_workflow(NewPrimitiveSpec {
                id: "zephyr-quartz-inspection".to_string(),
                when_to_use: vec!["inspect zephyr quartz artifacts".to_string()],
                preconditions: vec!["artifact path is known".to_string()],
                primitive_steps: vec!["inspect artifact metadata".to_string()],
                done_criteria: vec!["artifact findings are summarized".to_string()],
                recovery: vec![],
            })
            .await
            .expect("create matching primitive spec");
        store
            .create_workflow(NewPrimitiveSpec {
                id: "unrelated-ritual".to_string(),
                when_to_use: vec!["handle unrelated ritual tasks".to_string()],
                preconditions: vec![],
                primitive_steps: vec!["perform unrelated step".to_string()],
                done_criteria: vec!["unrelated result exists".to_string()],
                recovery: vec![],
            })
            .await
            .expect("create unrelated primitive spec");

        let catalog = store.primitive_routing_catalog("please inspect zephyr quartz", 8);
        let primitive_ids = catalog
            .primitive_ids
            .iter()
            .map(|summary| summary.id.as_str())
            .collect::<Vec<_>>();
        let relevant_ids = catalog
            .relevant_primitives
            .iter()
            .map(|summary| summary.id.as_str())
            .collect::<Vec<_>>();

        assert!(primitive_ids.contains(&"zephyr-quartz-inspection"));
        assert!(primitive_ids.contains(&"unrelated-ritual"));
        assert!(relevant_ids.contains(&"zephyr-quartz-inspection"));
        assert!(!relevant_ids.contains(&"unrelated-ritual"));
        assert_eq!(catalog.total_count, catalog.primitive_ids.len());
    }

    #[tokio::test]
    async fn compose_primitives_returns_existing_filename_sequence() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary).await;
        store
            .create_workflow(NewPrimitiveSpec {
                id: "alpha-project-scan".to_string(),
                when_to_use: vec!["inspect project".to_string()],
                preconditions: vec![],
                primitive_steps: vec!["inspect files".to_string()],
                done_criteria: vec!["findings exist".to_string()],
                recovery: vec![],
            })
            .await
            .expect("create inspect primitive");
        store
            .create_workflow(NewPrimitiveSpec {
                id: "beta-required-checks".to_string(),
                when_to_use: vec!["run checks".to_string()],
                preconditions: vec![],
                primitive_steps: vec!["run tests".to_string()],
                done_criteria: vec!["checks pass".to_string()],
                recovery: vec![],
            })
            .await
            .expect("create checks primitive");

        let composition = store
            .compose_primitives("alpha-project-scan-beta-required-checks")
            .expect("compose primitives");

        assert_eq!(
            composition.composition_id,
            "alpha-project-scan-beta-required-checks"
        );
        assert_eq!(
            composition.primitive_ids,
            vec!["alpha-project-scan", "beta-required-checks"]
        );
    }

    #[tokio::test]
    async fn compose_primitives_rejects_non_filename_input() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let store = PrimitiveStore::open_scoped(primary).await;

        let err = store
            .compose_primitives("inspect_local_project-run-required-checks")
            .expect_err("underscores are not legal primitive filename syntax");

        assert!(err.to_string().contains("use only lowercase a-z and '-'"));
    }

    #[test]
    fn primitive_summary_exposes_primitive_io_contract() {
        let summary = PrimitiveSpec {
            id: "modify-local-project".to_string(),
            when_to_use: vec!["local project needs edits".to_string()],
            preconditions: vec!["project path is known".to_string()],
            primitive_steps: vec![
                "inspect relevant code".to_string(),
                "apply targeted edits".to_string(),
                "avoid unrelated files".to_string(),
            ],
            done_criteria: vec!["requested change is implemented".to_string()],
            recovery: vec![],
        }
        .primitive_summary();

        assert_eq!(
            summary.capability_summary,
            "inspect relevant code; apply targeted edits; +1 more"
        );
        assert_eq!(summary.inputs_summary, "project path is known");
        assert_eq!(summary.outputs_summary, "requested change is implemented");
        assert_eq!(summary.when_to_use_summary, "local project needs edits");
    }

    #[tokio::test]
    async fn builtin_primitive_ids_cannot_be_overwritten() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary).await;

        let err = store
            .create_workflow(NewPrimitiveSpec {
                id: "author-workspace-app".to_string(),
                when_to_use: vec!["test".to_string()],
                preconditions: vec![],
                primitive_steps: vec!["step".to_string()],
                done_criteria: vec!["done".to_string()],
                recovery: vec![],
            })
            .await
            .expect_err("builtin primitive id should be reserved");

        assert!(err.to_string().contains("already exists"));
    }

    #[tokio::test]
    async fn builtin_primitives_are_read_only() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary).await;

        let err = store
            .apply_patch(PrimitiveSpecPatch {
                workflow_id: "author-workspace-app".to_string(),
                when_to_use_additions: vec!["extra".to_string()],
                precondition_additions: Vec::new(),
                workflow_step_additions: Vec::new(),
                done_criteria_additions: Vec::new(),
                recovery_additions: Vec::new(),
            })
            .await
            .expect_err("builtin primitive patch should be rejected");

        assert!(err.to_string().contains("read-only"));
    }

    #[tokio::test]
    async fn replace_workspace_primitive_spec_rewrites_complete_spec() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary.clone()).await;
        store
            .create_workflow(NewPrimitiveSpec {
                id: "search-todays-news".to_string(),
                when_to_use: vec!["search current news".to_string()],
                preconditions: vec!["network is available".to_string()],
                primitive_steps: vec![
                    "search aggregator".to_string(),
                    "repeat fallback searches".to_string(),
                ],
                done_criteria: vec!["sent news summary".to_string()],
                recovery: vec!["keep searching".to_string()],
            })
            .await
            .expect("create primitive");

        let updated = store
            .replace_workspace_workflow(
                "search-todays-news",
                PrimitiveSpec {
                    id: "search-todays-news".to_string(),
                    when_to_use: vec!["user asks for today's news".to_string()],
                    preconditions: vec!["date scope is known".to_string()],
                    primitive_steps: vec![
                        "choose approved news sources".to_string(),
                        "verify publication dates".to_string(),
                        "send concise summary".to_string(),
                    ],
                    done_criteria: vec!["summary cites sources".to_string()],
                    recovery: vec!["return a limited summary if sources are sparse".to_string()],
                },
            )
            .await
            .expect("replace primitive spec");

        assert_eq!(updated.primitive_steps.len(), 3);
        assert!(
            !updated
                .primitive_steps
                .iter()
                .any(|step| step == "repeat fallback searches")
        );

        let reloaded = PrimitiveStore::open_scoped(primary).await;
        let loaded = reloaded
            .get("search-todays-news")
            .expect("reloaded updated primitive spec");
        assert_eq!(
            loaded.primitive_steps,
            vec![
                "choose approved news sources",
                "verify publication dates",
                "send concise summary"
            ]
        );
    }

    #[tokio::test]
    async fn replace_rejects_builtin_primitive() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary).await;

        let err = store
            .replace_workspace_workflow(
                "author-workspace-app",
                PrimitiveSpec {
                    id: "author-workspace-app".to_string(),
                    when_to_use: vec!["test".to_string()],
                    preconditions: vec![],
                    primitive_steps: vec!["step".to_string()],
                    done_criteria: vec!["done".to_string()],
                    recovery: vec![],
                },
            )
            .await
            .expect_err("builtin primitive update should be rejected");

        assert!(err.to_string().contains("read-only"));
    }

    #[test]
    fn parse_workflow_file_accepts_crlf_line_endings() {
        let spec = parse_workflow_file(
            "---\r\nid: ignored-frontmatter\r\n---\r\n\r\n## When To Use\r\n- Windows checkout\r\n\r\n## Preconditions\r\n- A primitive spec file uses CRLF\r\n\r\n## Workflow\r\n1. Parse frontmatter\r\n\r\n## Done Criteria\r\n- Primitive spec is loaded\r\n\r\n## Recovery\r\n- Retry with normalized line endings\r\n",
            Some("crlf-workflow"),
        )
        .expect("parse CRLF workflow");

        assert_eq!(spec.id, "crlf-workflow");
        assert_eq!(spec.primitive_steps, vec!["Parse frontmatter"]);
        assert_eq!(spec.done_criteria, vec!["Primitive spec is loaded"]);
    }

    #[test]
    fn parse_workflow_file_uses_filename_id_over_frontmatter() {
        let spec = parse_workflow_file(
            "---\nid: arbitrary-content-id\nnotes: unrestricted\n---\n\n## When To Use\n- Any markdown content may differ from filename identity\n\n## Workflow\n- Load body\n\n## Done Criteria\n- Filename identity wins\n",
            Some("filename-identity"),
        )
        .expect("parse primitive spec with arbitrary frontmatter");

        assert_eq!(spec.id, "filename-identity");
        assert_eq!(spec.primitive_steps, vec!["Load body"]);
    }

    #[tokio::test]
    async fn merge_primitives_deletes_sources_and_updates_target() {
        let temp_dir = TempDir::new().expect("create primitive temp dir");
        let primary = temp_dir.path().join("workflows");
        let mut store = PrimitiveStore::open_scoped(primary.clone()).await;
        let target = store
            .create_workflow(NewPrimitiveSpec {
                id: "investigate-runtime-failure".to_string(),
                when_to_use: vec!["runtime failure".to_string()],
                preconditions: vec![],
                primitive_steps: vec!["collect logs".to_string()],
                done_criteria: vec!["cause is clear".to_string()],
                recovery: vec![],
            })
            .await
            .expect("create target");
        let source = store
            .create_workflow(NewPrimitiveSpec {
                id: "investigate-runtime-errors".to_string(),
                when_to_use: vec!["runtime error".to_string()],
                preconditions: vec![],
                primitive_steps: vec!["locate root cause".to_string()],
                done_criteria: vec!["fix direction is clear".to_string()],
                recovery: vec!["rollback".to_string()],
            })
            .await
            .expect("create source");

        let merged = store
            .merge_workflows(
                &target.id,
                std::slice::from_ref(&source.id),
                Some("duplicate".to_string()),
            )
            .await
            .expect("merge primitives");

        assert!(
            merged
                .primitive_steps
                .iter()
                .any(|item| item == "locate root cause")
        );
        assert!(store.get(&source.id).is_none());
        assert!(!primary.join(format!("{}.md", source.id)).exists());
    }

    #[tokio::test]
    async fn workflow_run_records_are_appended_and_deduped() {
        let temp_dir = TempDir::new().expect("workflow run record temp dir");
        let previous_home = env::var("DAAT_LOCUS_HOME").ok();
        unsafe {
            env::set_var("DAAT_LOCUS_HOME", temp_dir.path());
        }

        let record = PrimitiveRunRecord {
            run_id: "workflow-run:run-1".to_string(),
            workflow_id: "repair-flaky-test-pipeline".to_string(),
            started_at_ms: 1,
            ended_at_ms: 2,
            origin: "event:test".to_string(),
            outcome: PrimitiveRunOutcome::Completed,
            turn_count: 2,
            tool_action_count: 3,
            manual_fix_detected: false,
            rollback_detected: false,
            failure_types: vec!["tool_failure".to_string()],
            final_summary: "completed".to_string(),
        };
        let mut later_record = record.clone();
        later_record.run_id = "workflow-run:run-2".to_string();
        later_record.origin = "event:later".to_string();

        append_primitive_run_records(&[record.clone(), record.clone()])
            .await
            .expect("append workflow run records");
        let batch = load_primitive_run_batch()
            .await
            .expect("load workflow run records");
        let count = primitive_run_record_count()
            .await
            .expect("count workflow run records");
        append_primitive_run_records(&[later_record.clone()])
            .await
            .expect("append later workflow run record");
        compact_workflow_run_record_file(batch.next_offset)
            .await
            .expect("compact consumed workflow run records");
        let remaining_batch = load_primitive_run_batch()
            .await
            .expect("load remaining workflow run records");
        let remaining_count = primitive_run_record_count()
            .await
            .expect("count remaining workflow run records");

        match previous_home {
            Some(previous_home) => unsafe {
                env::set_var("DAAT_LOCUS_HOME", previous_home);
            },
            None => unsafe {
                env::remove_var("DAAT_LOCUS_HOME");
            },
        }

        assert_eq!(batch.records.len(), 1);
        assert_eq!(batch.unread_record_count, 1);
        assert!(batch.next_offset > 0);
        assert_eq!(count, 1);
        assert_eq!(batch.records[0].run_id, record.run_id);
        assert_eq!(batch.records[0].workflow_id, record.workflow_id);
        assert_eq!(remaining_count, 1);
        assert_eq!(remaining_batch.records.len(), 1);
        assert_eq!(remaining_batch.records[0].run_id, later_record.run_id);
    }
}
