car-ffi-common 0.32.1

//! JSON wrapper for `car-workflow::WorkflowEngine`.
//!
//! Lets JS / Python / WS callers run a multi-stage workflow definition
//! end-to-end. The workflow JSON shape comes from
//! [`car_workflow::Workflow`] (serde-derived); the result is
//! [`car_workflow::WorkflowResult`] serialized to JSON.
//!
//! Like `car_ffi_common::multi`, this takes an `AgentRunner` from the
//! caller — the FFI bindings supply their own (NAPI's
//! `StoredAgentRunner`, PyO3's `PyAgentRunner`) so the workflow's
//! agent stages dispatch through the registered JS/Python callback.

use car_multi::{AgentRunner, SharedInfra};
use car_workflow::{CheckpointStore, PausedWorkflow, Workflow, WorkflowEngine};
use std::collections::HashMap;
use std::sync::Arc;

/// List resumable workflow checkpoints under `runs_dir` (EPIC H / H1). Returns
/// a JSON array of `CheckpointSummary` (`{run_id, paused_stage_id, prompt,
/// created_at}`) — the embedded-mode counterpart of the daemon's
/// `workflow.list_paused`, so a caller that persists its own checkpoints can
/// rediscover resumable runs after a restart. In-flight and corrupt
/// checkpoints are omitted.
pub fn list_paused_workflows(runs_dir: &str) -> Result<String, String> {
    let store = CheckpointStore::open(runs_dir).map_err(|e| e.to_string())?;
    let summaries = store.list_summaries().map_err(|e| e.to_string())?;
    serde_json::to_string(&summaries).map_err(|e| e.to_string())
}

/// Run a Workflow JSON definition to completion. Returns
/// [`WorkflowResult`](car_workflow::WorkflowResult) as JSON.
///
/// `initial_state`, when given, is seeded into workflow state before the run
/// starts — the inter-workflow chaining hook: hand one workflow's
/// `final_state` (or any caller-supplied map) to the next so its edge
/// conditions and stages can read upstream results. `None` is exactly the
/// prior behavior (state starts empty, save the pinned `goal`). The reserved
/// `goal` drift anchor is always re-derived from the workflow definition and
/// cannot be injected via `initial_state`.
///
/// If the workflow hits a human-in-the-loop approval gate, the result has
/// `status == "paused"` and a `paused` checkpoint object. Hand that checkpoint
/// back to [`resume_workflow`] (with the human's response) to continue. The
/// caller owns checkpoint persistence — this function performs no I/O so the
/// engine stays pure; the server layer adds durable, exactly-once storage.
pub async fn run_workflow(
    workflow_json: &str,
    initial_state: Option<HashMap<String, serde_json::Value>>,
    runner: Arc<dyn AgentRunner>,
) -> Result<String, String> {
    let workflow: Workflow =
        serde_json::from_str(workflow_json).map_err(|e| format!("invalid workflow JSON: {}", e))?;
    let infra = SharedInfra::new();
    let engine = WorkflowEngine::new(runner, infra);
    let result = engine
        .run_with_state(&workflow, initial_state.unwrap_or_default())
        .await
        .map_err(|e| format!("workflow error: {}", e))?;
    serde_json::to_string(&result).map_err(|e| e.to_string())
}

/// Run a sequence of Workflow JSON definitions as a chain: each next
/// workflow's initial state is the previous [`WorkflowResult`]'s
/// `final_state`, merged over the caller's `initial_state` (the previous
/// result wins on key collisions). The chain stops at the first
/// non-`completed` result.
///
/// `workflows_json` is a JSON array of Workflow objects. Every workflow is
/// **statically pre-validated** (via [`car_workflow::verify_workflow`])
/// before *any* of them executes: a chain must never perform external side
/// effects and then die on a manifest defect in a later workflow, so
/// structural garbage anywhere in the array rejects the whole chain up front
/// with an `Err`.
///
/// Returns JSON `{ results: [WorkflowResult, ...], status, paused_at_index?,
/// error?, failed_at_index? }` where `results` holds every workflow that ran
/// (in order), `status` is `"completed"` when the whole chain finished or
/// the stopping result's status otherwise, and `paused_at_index` names the
/// workflow index whose run parked at an approval gate (its `paused`
/// checkpoint is inside that result).
///
/// A **mid-chain runtime error** (cycle limit reached, stage not found, and
/// other engine errors that static verification cannot rule out) does NOT
/// discard the chain: the call still returns `Ok` with `status: "failed"`,
/// the `results` accumulated so far (including any delivery evidence from
/// completed workflows), a top-level `error` message, and `failed_at_index`
/// naming the workflow whose engine run produced no result. A workflow that
/// *ran* and failed appears as the last entry of `results` instead (no
/// top-level `error`). `Err` is reserved for call-level problems (bad JSON,
/// empty array, failed pre-validation) where nothing has executed.
///
/// All workflows in the chain share one `SharedInfra` (state store + event
/// log), so a later workflow can observe an earlier one's `state_write`s
/// beyond the explicit `final_state` seeding. Checkpoint persistence stays
/// caller-owned, exactly like [`run_workflow`]; the daemon layer persists a
/// paused intermediate durably so `workflow.resume` can pick it up.
pub async fn chain_workflows(
    workflows_json: &str,
    initial_state: Option<HashMap<String, serde_json::Value>>,
    runner: Arc<dyn AgentRunner>,
) -> Result<String, String> {
    let workflows: Vec<Workflow> = serde_json::from_str(workflows_json)
        .map_err(|e| format!("invalid workflows JSON (expected an array): {}", e))?;
    if workflows.is_empty() {
        return Err("workflow chain requires at least one workflow".to_string());
    }

    // Statically pre-validate EVERY workflow before executing any. Rejecting
    // manifest defects up front means a chain never lands external side
    // effects and then aborts on structural garbage downstream.
    for (i, workflow) in workflows.iter().enumerate() {
        let report = car_workflow::verify_workflow(workflow);
        if !report.valid {
            let errors: Vec<String> = report
                .issues
                .iter()
                .filter(|issue| issue.severity == "error")
                .map(|issue| issue.message.clone())
                .collect();
            return Err(format!(
                "workflow at index {i} ('{}') failed static verification: {}",
                workflow.id,
                errors.join("; ")
            ));
        }
    }

    let base = initial_state.unwrap_or_default();
    let mut carry = base.clone();
    let mut results: Vec<car_workflow::WorkflowResult> = Vec::with_capacity(workflows.len());
    let mut status = car_workflow::WorkflowStatus::Completed;
    let mut paused_at_index: Option<usize> = None;
    let mut chain_error: Option<(usize, String)> = None;

    let infra = SharedInfra::new();
    let engine = WorkflowEngine::new(runner, infra);
    for (i, workflow) in workflows.iter().enumerate() {
        let result = match engine.run_with_state(workflow, carry.clone()).await {
            Ok(r) => r,
            Err(e) => {
                // A runtime engine error produced no WorkflowResult for this
                // workflow — but the completed workflows before it already
                // performed external effects (deliveries!). Preserve their
                // results and report the failure in-band; a retry decision
                // needs this evidence to avoid double-delivery.
                status = car_workflow::WorkflowStatus::Failed;
                chain_error = Some((i, format!("workflow '{}': {}", workflow.id, e)));
                break;
            }
        };
        let done = result.status == car_workflow::WorkflowStatus::Completed;
        if !done {
            status = result.status.clone();
            if result.is_paused() {
                paused_at_index = Some(i);
            }
            results.push(result);
            break;
        }
        // Next workflow's seed: the caller's base merged with this result's
        // final_state, final_state winning.
        carry = base.clone();
        for (k, v) in &result.final_state {
            carry.insert(k.clone(), v.clone());
        }
        results.push(result);
    }

    let mut out = serde_json::json!({
        "results": results,
        "status": status,
    });
    if let Some(idx) = paused_at_index {
        out["paused_at_index"] = serde_json::json!(idx);
    }
    if let Some((idx, error)) = chain_error {
        out["failed_at_index"] = serde_json::json!(idx);
        out["error"] = serde_json::json!(error);
    }
    serde_json::to_string(&out).map_err(|e| e.to_string())
}

/// Resume a workflow parked at an approval gate.
///
/// `paused_json` is the `paused` checkpoint object returned by [`run_workflow`]
/// (or a prior `resume_workflow`); `input_json` is a JSON object of the human's
/// response fields. Returns the next [`WorkflowResult`](car_workflow::WorkflowResult)
/// as JSON — which may itself be `paused` again if the run hits another gate.
pub async fn resume_workflow(
    paused_json: &str,
    input_json: &str,
    runner: Arc<dyn AgentRunner>,
) -> Result<String, String> {
    let paused: PausedWorkflow = serde_json::from_str(paused_json)
        .map_err(|e| format!("invalid paused checkpoint JSON: {}", e))?;
    let input: HashMap<String, serde_json::Value> = serde_json::from_str(input_json)
        .map_err(|e| format!("invalid approval input JSON: {}", e))?;
    let infra = SharedInfra::new();
    let engine = WorkflowEngine::new(runner, infra);
    let result = engine
        .resume(paused, input)
        .await
        .map_err(|e| format!("workflow resume error: {}", e))?;
    serde_json::to_string(&result).map_err(|e| e.to_string())
}

/// Build the external-item automation recipe from an
/// [`AutomationSpec`](car_workflow::AutomationSpec) JSON into a runnable
/// [`Workflow`](car_workflow::Workflow) JSON.
///
/// Lowers the poll → dedup → per-item agent → deliver loop to a plain workflow:
/// poll a source, drop items already handled in a prior run (persistent
/// content-hash dedup), fan an ephemeral agent over each new item, then
/// optionally deliver the results. Hand the returned JSON to [`run_workflow`]
/// (typically on a schedule). Stateless — performs no I/O.
pub fn build_automation_workflow(spec_json: &str) -> Result<String, String> {
    let spec: car_workflow::AutomationSpec = serde_json::from_str(spec_json)
        .map_err(|e| format!("invalid automation spec JSON: {}", e))?;
    serde_json::to_string(&spec.build()).map_err(|e| e.to_string())
}

/// Static analysis: validate a workflow definition without running it.
/// Returns the verification report as JSON.
///
/// Manually serialize the report because car_workflow::WorkflowVerifyResult
/// isn't Serde-derived (and shouldn't be — its Debug-only fields are
/// for human review, not API contract).
pub fn verify_workflow(workflow_json: &str) -> Result<String, String> {
    let workflow: Workflow =
        serde_json::from_str(workflow_json).map_err(|e| format!("invalid workflow JSON: {}", e))?;
    let report = car_workflow::verify_workflow(&workflow);
    let json = serde_json::json!({
        "valid": report.valid,
        "has_cycles": report.has_cycles,
        "reachable_stages": report.reachable_stages,
        "unreachable_stages": report.unreachable_stages,
        "issues": report.issues.iter().map(|i| format!("{:?}", i)).collect::<Vec<_>>(),
        // Advisory semantic findings (dangling edge-condition keys, unproduced
        // state dependencies). Non-blocking; do not affect `valid`.
        "semantic": car_workflow::semantic_issues(&workflow),
    });
    Ok(json.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::{json, Value};

    /// Chain tests use proposal-only workflows, so the runner is never invoked.
    struct NoopRunner;

    #[async_trait::async_trait]
    impl AgentRunner for NoopRunner {
        async fn run(
            &self,
            _spec: &car_multi::AgentSpec,
            _task: &str,
            _runtime: &car_engine::Runtime,
            _mailbox: &car_multi::Mailbox,
        ) -> Result<car_multi::AgentOutput, car_multi::MultiError> {
            Err(car_multi::MultiError::NoOutput)
        }
    }

    fn runner() -> Arc<dyn AgentRunner> {
        Arc::new(NoopRunner)
    }

    /// A single-stage workflow whose proposal writes `key` = `value` via a
    /// `state_write` action.
    fn writer_workflow(id: &str, key: &str, value: Value) -> Value {
        json!({
            "id": id,
            "name": id,
            "start": "write",
            "stages": [{
                "id": "write",
                "name": "write",
                "step": {
                    "type": "proposal",
                    "proposal": {
                        "id": format!("p-{id}"),
                        "source": "test",
                        "actions": [{
                            "id": "w1",
                            "type": "state_write",
                            "parameters": { "key": key, "value": value }
                        }],
                        "timestamp": "2026-01-01T00:00:00Z",
                        "context": {}
                    }
                }
            }],
            "edges": []
        })
    }

    /// A workflow that branches on `handoff == "from-a"`: check -> yes on the
    /// conditional edge, otherwise the unconditional fallback to no.
    fn branching_workflow() -> Value {
        let noop_stage = |id: &str| {
            json!({
                "id": id,
                "name": id,
                "step": {
                    "type": "proposal",
                    "proposal": {
                        "id": format!("p-{id}"),
                        "source": "test",
                        "actions": [],
                        "timestamp": "2026-01-01T00:00:00Z",
                        "context": {}
                    }
                }
            })
        };
        json!({
            "id": "wf-b",
            "name": "WF B",
            "start": "check",
            "stages": [noop_stage("check"), noop_stage("yes"), noop_stage("no")],
            "edges": [
                {
                    "from": "check",
                    "to": "yes",
                    "conditions": [{
                        "key": "handoff",
                        "operator": "eq",
                        "value": "from-a",
                        "description": ""
                    }]
                },
                { "from": "check", "to": "no" }
            ]
        })
    }

    fn ran_stages(result: &Value) -> Vec<String> {
        result["stages"]
            .as_array()
            .unwrap()
            .iter()
            .map(|s| s["stage_id"].as_str().unwrap().to_string())
            .collect()
    }

    #[tokio::test]
    async fn run_workflow_seeds_initial_state() {
        let wf = branching_workflow();
        let mut seed = HashMap::new();
        seed.insert("handoff".to_string(), json!("from-a"));

        let out = run_workflow(&wf.to_string(), Some(seed), runner())
            .await
            .unwrap();
        let result: Value = serde_json::from_str(&out).unwrap();
        assert_eq!(result["status"], "completed");
        assert!(ran_stages(&result).contains(&"yes".to_string()));

        // None keeps today's behavior: the fallback branch runs.
        let out = run_workflow(&wf.to_string(), None, runner()).await.unwrap();
        let result: Value = serde_json::from_str(&out).unwrap();
        assert!(ran_stages(&result).contains(&"no".to_string()));
    }

    #[tokio::test]
    async fn chain_threads_final_state_between_workflows() {
        // A writes handoff="from-a"; B's conditional edge reads it.
        let workflows = json!([
            writer_workflow("wf-a", "handoff", json!("from-a")),
            branching_workflow(),
        ]);
        let out = chain_workflows(&workflows.to_string(), None, runner())
            .await
            .unwrap();
        let chained: Value = serde_json::from_str(&out).unwrap();

        assert_eq!(chained["status"], "completed");
        assert!(chained.get("paused_at_index").is_none());
        let results = chained["results"].as_array().unwrap();
        assert_eq!(results.len(), 2);
        assert!(
            ran_stages(&results[1]).contains(&"yes".to_string()),
            "A's final_state routed B down the yes branch"
        );
    }

    #[tokio::test]
    async fn chain_stops_at_paused_intermediate() {
        let gate = json!({
            "id": "wf-gate",
            "name": "gated",
            "start": "gate",
            "stages": [{
                "id": "gate",
                "name": "gate",
                "step": { "type": "approval", "prompt": "ok?", "output_key": "approval" }
            }],
            "edges": []
        });
        // The gate pauses, so the second workflow never runs.
        let workflows = json!([gate, writer_workflow("wf-a", "x", json!(1))]);
        let out = chain_workflows(&workflows.to_string(), None, runner())
            .await
            .unwrap();
        let chained: Value = serde_json::from_str(&out).unwrap();

        assert_eq!(chained["status"], "paused");
        assert_eq!(chained["paused_at_index"], 0);
        let results = chained["results"].as_array().unwrap();
        assert_eq!(results.len(), 1, "chain stops at the paused workflow");
        assert!(results[0]["paused"].is_object(), "checkpoint is carried");
    }

    /// A workflow whose sole stage is a deliver fan-out with one succeeding
    /// (state_write) sink — completed delivery evidence for chain tests.
    fn deliver_workflow(id: &str) -> Value {
        json!({
            "id": id,
            "name": id,
            "start": "deliver",
            "stages": [{
                "id": "deliver",
                "name": "deliver",
                "step": {
                    "type": "deliver",
                    "sinks": [{
                        "id": "sink-ok",
                        "source": "test",
                        "actions": [{
                            "id": "d1",
                            "type": "state_write",
                            "parameters": { "key": "notify", "value": "sent" }
                        }],
                        "timestamp": "2026-01-01T00:00:00Z",
                        "context": {}
                    }]
                }
            }],
            "edges": []
        })
    }

    /// A workflow that passes static verification (a cycle is only a warning)
    /// but dies at runtime on the loop guard: a <-> b with max_iterations 2.
    fn cycle_limited_workflow() -> Value {
        let noop_stage = |id: &str| {
            json!({
                "id": id,
                "name": id,
                "step": {
                    "type": "proposal",
                    "proposal": {
                        "id": format!("p-{id}"),
                        "source": "test",
                        "actions": [],
                        "timestamp": "2026-01-01T00:00:00Z",
                        "context": {}
                    }
                }
            })
        };
        json!({
            "id": "wf-cycle",
            "name": "cyclic",
            "start": "a",
            "stages": [noop_stage("a"), noop_stage("b")],
            "edges": [
                { "from": "a", "to": "b" },
                { "from": "b", "to": "a" }
            ],
            "max_iterations": 2
        })
    }

    #[tokio::test]
    async fn chain_preserves_results_on_mid_chain_runtime_error() {
        // Workflow A completes, delivering through its sink (an external
        // side effect); workflow B hits the runtime cycle limit — an engine
        // Err that produces no WorkflowResult. The chain must NOT discard
        // A's delivery evidence: it returns a normal result with the
        // results-so-far, status "failed", and failed_at_index.
        let workflows = json!([deliver_workflow("wf-a"), cycle_limited_workflow()]);
        let out = chain_workflows(&workflows.to_string(), None, runner())
            .await
            .expect("mid-chain runtime error must not become a call-level Err");
        let chained: Value = serde_json::from_str(&out).unwrap();

        assert_eq!(chained["status"], "failed");
        assert_eq!(chained["failed_at_index"], 1);
        let err = chained["error"].as_str().unwrap();
        assert!(err.contains("wf-cycle"), "error names the workflow: {err}");
        assert!(
            err.to_lowercase().contains("cycle") || err.contains("iteration"),
            "error carries the engine cause: {err}"
        );

        // A's result — including its per-sink delivery evidence — survives.
        let results = chained["results"].as_array().unwrap();
        assert_eq!(results.len(), 1, "only the workflows that ran are recorded");
        assert_eq!(results[0]["status"], "completed");
        let output = &results[0]["stages"][0]["output"];
        assert_eq!(output["type"], "deliver");
        assert_eq!(output["results"][0]["sink_id"], "sink-ok");
        assert_eq!(output["results"][0]["ok"], true);
    }

    #[tokio::test]
    async fn chain_pre_validates_every_workflow_before_running_any() {
        // Workflow B is structural garbage (start stage doesn't exist).
        // The chain must reject up front — before A performs any side
        // effects — rather than delivering A and then dying on B.
        let broken = json!({
            "id": "wf-broken",
            "name": "broken",
            "start": "missing",
            "stages": [],
            "edges": []
        });
        let workflows = json!([deliver_workflow("wf-a"), broken]);
        let err = chain_workflows(&workflows.to_string(), None, runner())
            .await
            .expect_err("structurally invalid workflow must reject the chain up front");
        assert!(err.contains("index 1"), "err: {err}");
        assert!(err.contains("static verification"), "err: {err}");
        assert!(err.contains("wf-broken"), "err: {err}");
    }
}