proofborne 0.3.0

//! Local proof-carrying daemon (ADR 0007).
//!
//! The daemon is a thin loopback transport around the normal proof-gated
//! session path. It binds `127.0.0.1` only, speaks JSON-RPC 2.0 over
//! `Content-Length`-framed TCP, and serves requests sequentially.

use std::{
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use anyhow::{Context, Result, bail};
use proofborne_core::EventEnvelope;
use proofborne_runtime::{
    EventSink, PluginRegistry, PolicyPreset, RunOptions, RuntimePaths, SessionStore,
};
use serde_json::{Value, json};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::{TcpListener, TcpStream},
    time::timeout,
};

use crate::{load_contract, load_verifications, prepare_engine, resolve_workspace_argument};

const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
const SESSION_LIST_LIMIT: usize = 50;

/// Serves daemon requests on a bound loopback listener until shutdown.
pub async fn serve(workspace: PathBuf, listener: TcpListener) -> Result<()> {
    let store = SessionStore::open(
        RuntimePaths::for_workspace(&workspace)?
            .workspace_data
            .join("sessions.sqlite3"),
    )?;
    let store = Arc::new(store);
    let stop = Arc::new(AtomicBool::new(false));
    while !stop.load(Ordering::SeqCst) {
        let (stream, _) = listener.accept().await?;
        // Sequential v1 handling: a shutdown request sets `stop` and this loop
        // exits after the connection is served.
        let _ = handle_connection(stream, &workspace, &stop, store.clone()).await;
    }
    Ok(())
}

async fn handle_connection(
    mut stream: TcpStream,
    workspace: &Path,
    stop: &AtomicBool,
    store: Arc<SessionStore>,
) -> Result<()> {
    let request = match timeout(REQUEST_TIMEOUT, read_frame(&mut stream)).await {
        Ok(Ok(Some(request))) => request,
        Ok(Ok(None)) | Err(_) => return Ok(()),
        Ok(Err(error)) => return Err(error),
    };
    let id = request.get("id").cloned().unwrap_or(Value::Null);
    let method = request.get("method").and_then(Value::as_str).unwrap_or("");
    let params = request.get("params").cloned().unwrap_or(Value::Null);
    let (success, result) = match method {
        "daemon.status" => (true, status_result(workspace)),
        "daemon.sessions" => match list_sessions(&store) {
            Ok(result) => (true, result),
            Err(error) => (false, json!({"message": format!("{error:#}")})),
        },
        "daemon.session" => match session_detail(&store, &params) {
            Ok(result) => (true, result),
            Err(error) => (false, json!({"message": format!("{error:#}")})),
        },
        "task.run" => match run_task(workspace, &params).await {
            Ok(result) => (true, result),
            Err(error) => (false, json!({"message": format!("{error:#}")})),
        },
        "daemon.shutdown" => {
            stop.store(true, Ordering::SeqCst);
            (true, json!({"stopping": true}))
        }
        _ => (
            false,
            json!({"code": -32601, "message": format!("unsupported method {method}")}),
        ),
    };
    let response = if success {
        json!({"jsonrpc": "2.0", "id": id, "result": result})
    } else {
        json!({"jsonrpc": "2.0", "id": id, "error": result})
    };
    write_frame(&mut stream, &response).await?;
    Ok(())
}

fn status_result(workspace: &Path) -> Value {
    let plugins = PluginRegistry::discover(workspace);
    let mut status = json!({
        "version": env!("CARGO_PKG_VERSION"),
        "workspace": workspace.display().to_string(),
        "pid": std::process::id(),
    });
    match plugins {
        Ok(registry) => {
            status["pluginToolCount"] = json!(registry.tool_count());
        }
        Err(error) => {
            status["pluginError"] = json!(format!("{error}"));
        }
    }
    status
}

fn list_sessions(store: &SessionStore) -> Result<Value> {
    let sessions = store
        .list_sessions(SESSION_LIST_LIMIT)
        .context("list sessions")?;
    Ok(json!({ "sessions": sessions }))
}

fn session_detail(store: &SessionStore, params: &Value) -> Result<Value> {
    let id_text = params
        .get("id")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow::anyhow!("daemon.session requires an id"))?;
    let id = uuid::Uuid::parse_str(id_text).context("invalid session id")?;
    let record = store.load_session(id).context("load session")?;
    let events = store.load_events(id).context("load events")?;
    let final_event = events
        .iter()
        .rev()
        .find(|event| event.kind == "proof.finalized");
    Ok(json!({
        "id": record.id,
        "workspace": record.workspace,
        "provider": record.provider,
        "model": record.model,
        "status": record.status,
        "createdAt": record.created_at,
        "updatedAt": record.updated_at,
        "eventCount": events.len(),
        "finalEvent": final_event.map(|event| {
            json!({
                "outcome": event.payload.get("outcome"),
                "proofHash": event.payload.get("proofHash"),
            })
        }),
    }))
}

async fn run_task(workspace: &Path, params: &Value) -> Result<Value> {
    let contract_param = params
        .get("contract")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow::anyhow!("task.run requires a contract path"))?;
    let verification_param = params.get("verification").and_then(Value::as_str);
    let policy = params.get("policy").and_then(Value::as_str);
    let trust_workspace = params
        .get("trustWorkspace")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let profile = params.get("profile").and_then(Value::as_str);

    let contract_path = contained_path(workspace, Path::new(contract_param))?;
    let verification_path = verification_param
        .map(|value| contained_path(workspace, Path::new(value)))
        .transpose()?;
    let verifications = verification_path
        .as_deref()
        .map(load_verifications)
        .transpose()?
        .unwrap_or_default();
    let contract = load_contract(None, Some(&contract_path), false, &verifications)?;
    let policy_preset = policy.map(parse_policy_preset).transpose()?;
    let prepared = prepare_engine(workspace, profile, policy_preset).await?;

    let mut options = RunOptions::new(workspace.to_path_buf(), &prepared.model, contract);
    options.policy = prepared.policy;
    options.routing = prepared.routing.clone();
    if trust_workspace {
        if options.policy.preset != PolicyPreset::Workspace {
            bail!("trustWorkspace requires policy = workspace");
        }
        options.policy.trusted_workspace = true;
    }
    options.verification = verifications;
    options.redactor = prepared.redactor;

    let sink = CollectingSink::default();
    let result = prepared
        .engine
        .run_chain(prepared.provider_chain, options, &sink)
        .await?;
    let proof_hash = sink
        .events()
        .iter()
        .rev()
        .find(|event| event.kind == "proof.finalized")
        .and_then(|event| {
            event
                .payload
                .get("proofHash")
                .and_then(Value::as_str)
                .map(ToOwned::to_owned)
        });
    Ok(json!({
        "sessionId": result.session_id,
        "outcome": format!("{:?}", result.outcome),
        "proofHash": proof_hash,
    }))
}

fn contained_path(workspace: &Path, path: &Path) -> Result<PathBuf> {
    let resolved = resolve_workspace_argument(workspace, path);
    let canonical_workspace = workspace
        .canonicalize()
        .with_context(|| format!("workspace is not canonicalizable: {}", workspace.display()))?;
    let canonical = resolved
        .canonicalize()
        .with_context(|| format!("daemon path is not canonicalizable: {}", resolved.display()))?;
    if canonical.starts_with(&canonical_workspace) {
        Ok(canonical)
    } else {
        bail!("daemon path escapes the workspace: {}", resolved.display())
    }
}

fn parse_policy_preset(value: &str) -> Result<PolicyPreset> {
    value
        .parse()
        .with_context(|| format!("invalid policy preset {value:?}"))
}

/// Sends one JSON-RPC request and returns the result (or an error).
pub async fn request(addr: &str, method: &str, params: Value) -> Result<Value> {
    request_with_timeout(addr, method, params, REQUEST_TIMEOUT).await
}

/// Sends one JSON-RPC request with an explicit response timeout (needed for
/// long-running proof-gated `task.run` calls from clients).
pub async fn request_with_timeout(
    addr: &str,
    method: &str,
    params: Value,
    response_timeout: Duration,
) -> Result<Value> {
    let mut stream = TcpStream::connect(addr).await?;
    let request = json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params});
    write_frame(&mut stream, &request).await?;
    let response = timeout(response_timeout, read_frame(&mut stream))
        .await
        .context("daemon request timed out")?
        .context("daemon closed the connection")?
        .ok_or_else(|| anyhow::anyhow!("daemon closed the connection"))?;
    if let Some(error) = response.get("error") {
        bail!(
            "daemon error: {}",
            error
                .get("message")
                .and_then(Value::as_str)
                .unwrap_or("unknown")
        );
    }
    Ok(response.get("result").cloned().unwrap_or(Value::Null))
}

async fn read_frame(stream: &mut TcpStream) -> Result<Option<Value>> {
    let mut content_length = None;
    let mut header = Vec::new();
    loop {
        let mut byte = [0u8; 1];
        let read = stream.read(&mut byte).await?;
        if read == 0 {
            return Ok(None);
        }
        header.push(byte[0]);
        if header.ends_with(b"\r\n\r\n") || header.ends_with(b"\n\n") {
            break;
        }
        if header.len() > 64 * 1024 {
            bail!("daemon header exceeds the boundary");
        }
    }
    let header_text = String::from_utf8_lossy(&header);
    for line in header_text.lines() {
        if let Some((name, value)) = line.split_once(':')
            && name.eq_ignore_ascii_case("content-length")
        {
            content_length = Some(value.trim().parse::<usize>()?);
        }
    }
    let length = content_length.ok_or_else(|| anyhow::anyhow!("frame omitted Content-Length"))?;
    if length > MAX_FRAME_BYTES {
        bail!("daemon frame exceeds the 16 MiB boundary");
    }
    let mut body = vec![0u8; length];
    stream.read_exact(&mut body).await?;
    Ok(Some(serde_json::from_slice(&body)?))
}

async fn write_frame(stream: &mut TcpStream, message: &Value) -> Result<()> {
    let bytes = serde_json::to_vec(message)?;
    if bytes.len() > MAX_FRAME_BYTES {
        bail!("daemon response exceeds the 16 MiB boundary");
    }
    stream
        .write_all(format!("Content-Length: {}\r\n\r\n", bytes.len()).as_bytes())
        .await?;
    stream.write_all(&bytes).await?;
    stream.flush().await?;
    Ok(())
}

#[derive(Debug, Default)]
struct CollectingSink(std::sync::Mutex<Vec<EventEnvelope>>);

impl CollectingSink {
    fn events(&self) -> Vec<EventEnvelope> {
        self.0.lock().unwrap().clone()
    }
}

impl EventSink for CollectingSink {
    fn emit(&self, event: &EventEnvelope) -> Result<(), String> {
        self.0.lock().unwrap().push(event.clone());
        Ok(())
    }
}

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

    fn workspace_with_contract() -> (tempfile::TempDir, PathBuf, PathBuf) {
        let workspace = tempfile::tempdir().unwrap();
        let config_dir = workspace.path().join(".proofborne");
        std::fs::create_dir_all(&config_dir).unwrap();
        std::fs::write(
            config_dir.join("config.toml"),
            "default_profile = \"mock\"\n\n[providers.mock]\nprovider = \"mock\"\nmodel = \"deterministic-v1\"\n",
        )
        .unwrap();
        let contract = workspace.path().join("contract.json");
        let verification = workspace.path().join("verification.json");
        std::fs::write(
            &contract,
            serde_json::to_string_pretty(&json!({
                "schemaVersion": "proofborne.v1",
                "id": "019fbf00-0000-7000-8000-000000000001",
                "goal": "daemon test task",
                "claimScope": "task",
                "criteria": [
                    {
                        "id": "runtime_completed",
                        "description": "runtime completed",
                        "required": true,
                        "minimumEvidence": 1,
                        "evidenceRequirement": {
                            "allowedKinds": ["runtime"],
                            "allowedProducers": ["proofborne.runtime"],
                            "minimumObservations": 1,
                            "freshness": "final_workspace_state",
                            "requireArtifacts": false,
                            "minimumIndependentProducers": 1,
                            "minimumAssurance": "observed"
                        },
                        "state": "pending",
                        "evidenceIds": []
                    },
                    {
                        "id": "tests_pass",
                        "description": "verification passes",
                        "required": true,
                        "minimumEvidence": 1,
                        "evidenceRequirement": {
                            "allowedKinds": ["process"],
                            "allowedProducers": ["verify.exec"],
                            "minimumObservations": 1,
                            "freshness": "final_workspace_state",
                            "requireArtifacts": false,
                            "minimumIndependentProducers": 1,
                            "minimumAssurance": "observed"
                        },
                        "state": "pending",
                        "evidenceIds": []
                    }
                ],
                "createdAt": "2026-08-02T00:00:00Z",
                "confirmed": true
            }))
            .unwrap(),
        )
        .unwrap();
        std::fs::write(
            &verification,
            serde_json::to_string(&json!([{
                "criterionId": "tests_pass",
                "program": std::env::current_exe().unwrap().to_string_lossy(),
                "args": ["--help"],
                "timeoutSeconds": 30
            }]))
            .unwrap(),
        )
        .unwrap();
        (workspace, contract, verification)
    }

    #[tokio::test]
    async fn daemon_status_task_run_and_shutdown() {
        let (workspace, contract, verification) = workspace_with_contract();
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap().to_string();
        let server = tokio::spawn(serve(workspace.path().to_path_buf(), listener));

        let status = request(&addr, "daemon.status", json!({})).await.unwrap();
        assert_eq!(status["version"], env!("CARGO_PKG_VERSION"));
        let canonical = workspace.path().canonicalize().unwrap();
        let reported = PathBuf::from(status["workspace"].as_str().unwrap());
        assert_eq!(reported.canonicalize().unwrap(), canonical);
        assert!(status["pid"].as_i64().is_some());

        let result = request(
            &addr,
            "task.run",
            json!({
                "contract": contract.to_string_lossy(),
                "verification": verification.to_string_lossy(),
                "policy": "workspace",
                "trustWorkspace": true
            }),
        )
        .await
        .unwrap();
        assert_eq!(result["outcome"], "Verified");
        assert!(result["sessionId"].as_str().is_some());
        assert!(result["proofHash"].as_str().is_some());

        let shutdown = request(&addr, "daemon.shutdown", json!({})).await.unwrap();
        assert_eq!(shutdown["stopping"], true);
        server.await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn daemon_rejects_paths_escaping_the_workspace() {
        let (workspace, _, _) = workspace_with_contract();
        let outside = tempfile::NamedTempFile::new().unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap().to_string();
        let server = tokio::spawn(serve(workspace.path().to_path_buf(), listener));

        let response = request(
            &addr,
            "task.run",
            json!({"contract": outside.path().to_string_lossy()}),
        )
        .await;
        assert!(response.is_err());

        request(&addr, "daemon.shutdown", json!({})).await.unwrap();
        server.await.unwrap().unwrap();
    }
}