harn-serve 0.10.166

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::convert::Infallible;
use std::ffi::OsString;
use std::net::SocketAddr;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path as AxumPath, Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Extension, Json, Router};
use futures::{stream, Stream, StreamExt};
use harn_vm::event_log::{AnyEventLog, EventLog, LogEvent, Topic};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;
use tokio::sync::{broadcast, mpsc, oneshot};
use uuid::Uuid;

use crate::adapters::acp::{
    run_acp_channel_server, AcpServerConfig, TaskAdmissionError, TaskPromptTarget,
};
use crate::auth::{AuthPolicy, AuthenticatedPrincipal};
use crate::permissions::{
    ActionClass, AuditFilter, DecisionScope, InMemoryPermissionStore, PermissionDecision,
    PermissionPolicy, PermissionRequest, PermissionStore, RememberRule, RememberSpec, RuleId,
};
use crate::tls::HttpTlsConfig;
use harn_vm::task_retry::{TaskRetryError, TaskRetryKey, TaskRetryPolicy};

mod artifacts;
mod authentication;
mod canonical_sessions;
mod events;
mod meta;
mod permissions;
mod session_control;
mod sessions;
mod task_admission;
mod task_store;
mod tasks;
mod workspaces;

const OPENAPI_YAML: &str = include_str!("../../../openapi.yaml");
pub const API_PROTOCOL_VERSION: &str = "agents-protocol-2026-10-11";
const ACTION_GRAPH_TOPIC: &str = "observability.action_graph";

#[derive(Clone, Debug)]
pub struct ApiHttpServeOptions {
    pub bind: SocketAddr,
    pub public_url: Option<String>,
    pub tls: HttpTlsConfig,
}

#[derive(Clone)]
pub struct ApiServerConfig {
    pub acp: AcpServerConfig,
    pub auth_policy: AuthPolicy,
    pub workspace_root: PathBuf,
    default_session_mode: String,
}

impl ApiServerConfig {
    pub fn for_pipeline(path: impl Into<String>) -> Self {
        let path = path.into();
        let root = canonical_sessions::workspace_root_for_pipeline(&path);
        Self {
            acp: AcpServerConfig::for_pipeline(path),
            auth_policy: AuthPolicy::allow_all(),
            workspace_root: root,
            default_session_mode: "ask".to_string(),
        }
    }

    pub fn with_auth_policy(mut self, auth_policy: AuthPolicy) -> Self {
        self.auth_policy = auth_policy;
        self
    }

    pub fn with_default_session_mode(mut self, mode_id: &str) -> Result<Self, String> {
        if !crate::adapters::acp::is_supported_session_mode(mode_id) {
            return Err(format!("unknown session mode: {mode_id}"));
        }
        self.default_session_mode = mode_id.to_string();
        Ok(self)
    }

    pub fn with_profile(mut self, profile: crate::adapters::acp::AcpProfileConfig) -> Self {
        self.acp.profile = profile;
        self
    }
}

#[derive(Clone)]
pub struct ApiServer {
    state: ApiState,
}

impl ApiServer {
    pub fn new(mut config: ApiServerConfig) -> Self {
        config.acp.auth_policy = AuthPolicy::allow_all();
        let provider_catalog = ProviderCatalogRuntime {
            llm_config_overrides: config.acp.llm_config_overrides.clone(),
            llm_capability_overrides: config.acp.llm_capability_overrides.clone(),
        };
        let (events_tx, _) = broadcast::channel(1024);
        let inner = Arc::new(Mutex::new(ApiStateInner::new(
            config.workspace_root.clone(),
        )));
        let (event_log, event_log_error) =
            match harn_vm::event_log::install_default_for_base_dir(&config.workspace_root) {
                Ok(log) => (Some(log), None),
                Err(error) => (None, Some(error.to_string())),
            };
        let canonical_sessions = canonical_sessions::open(&config.workspace_root);
        config.acp.task_admission = Some(Arc::new(task_admission::NativeTaskAdmission {
            store: canonical_sessions.clone(),
            inner: inner.clone(),
            events_tx: events_tx.clone(),
        }));
        let (client, response_rx) = AcpClient::start(config.acp);
        let state = ApiState {
            acp: client.clone(),
            inner,
            events_tx,
            auth_policy: config.auth_policy,
            permissions: Arc::new(InMemoryPermissionStore::default()),
            provider_catalog,
            default_session_mode: config.default_session_mode,
            event_log,
            event_log_error,
            canonical_sessions,
            session_control: Arc::new(tokio::sync::Mutex::new(())),
        };
        client.spawn_output_loop(response_rx, state.clone());
        Self { state }
    }

    pub async fn run_http(self: Arc<Self>, options: ApiHttpServeOptions) -> Result<(), String> {
        let router = api_router(self.state.clone());
        let router =
            crate::apply_transport_layers(router, &crate::TransportConfig::default_enabled());
        let router = crate::tls::apply_security_headers(router, &options.tls);
        let listener = crate::tls::bind_listener(options.bind)?;
        let local_addr = listener
            .local_addr()
            .map_err(|error| format!("failed to read local addr: {error}"))?;
        let advertised = options
            .public_url
            .clone()
            .unwrap_or_else(|| format!("{}://{local_addr}", options.tls.advertised_scheme()));
        eprintln!("[harn] Agents API server ready on {advertised}");
        eprintln!("[harn] OpenAPI document: {advertised}/openapi.json");
        crate::tls::serve_router_from_tcp(listener, router, &options.tls)
            .await
            .map_err(|error| format!("Agents API server failed: {error}"))
    }
}

#[derive(Clone)]
struct ApiState {
    acp: AcpClient,
    inner: Arc<Mutex<ApiStateInner>>,
    events_tx: broadcast::Sender<ApiEvent>,
    auth_policy: AuthPolicy,
    /// The permission primitive shared by REST, ACP suspend/respond, and
    /// eventual `harness.permissions.*` host calls.
    permissions: Arc<InMemoryPermissionStore>,
    provider_catalog: ProviderCatalogRuntime,
    default_session_mode: String,
    event_log: Option<Arc<AnyEventLog>>,
    event_log_error: Option<String>,
    canonical_sessions: Option<crate::sessions::SharedSessionStore>,
    /// Serialize restoration and configuration acknowledgements, never execution.
    session_control: Arc<tokio::sync::Mutex<()>>,
}

#[derive(Clone, Default)]
struct ProviderCatalogRuntime {
    llm_config_overrides: Option<harn_vm::llm_config::ProvidersConfig>,
    llm_capability_overrides: Option<harn_vm::llm::capabilities::CapabilitiesFile>,
}

impl ProviderCatalogRuntime {
    fn artifact(&self) -> harn_vm::provider_catalog::ProviderCatalogArtifact {
        harn_vm::provider_catalog::artifact_with_overrides(
            self.llm_config_overrides.as_ref(),
            self.llm_capability_overrides.as_ref(),
        )
    }
}

struct ApiStateInner {
    sessions: BTreeMap<String, Value>,
    messages: HashMap<String, Vec<Value>>,
    artifacts: BTreeMap<String, Value>,
    permissions: BTreeMap<String, PendingPermission>,
    workspaces: BTreeMap<String, Value>,
    events: Vec<ApiEvent>,
    root_workspace_id: String,
    root_workspace_path: PathBuf,
}

impl ApiStateInner {
    fn new(workspace_root: PathBuf) -> Self {
        let now = now_rfc3339();
        let root_workspace_id = "local".to_string();
        let mut workspaces = BTreeMap::new();
        workspaces.insert(
            root_workspace_id.clone(),
            json!({
                "id": root_workspace_id,
                "object": "workspace",
                "created_at": now,
                "updated_at": now,
                "metadata": {},
                "name": "Local workspace",
                "root": workspace_root.to_string_lossy(),
                "default_branch_id": null,
                "host": "local",
                "repository": null,
                "tenant_id": null,
                "capabilities": [
                    "sessions",
                    "tasks",
                    "events",
                    "artifacts",
                    "permissions",
                    "workflow_trigger_runs",
                    "workspace.files.read"
                ],
                "connectors": [],
                "quota_id": null
            }),
        );
        Self {
            sessions: BTreeMap::new(),
            messages: HashMap::new(),
            artifacts: BTreeMap::new(),
            permissions: BTreeMap::new(),
            workspaces,
            events: Vec::new(),
            root_workspace_id,
            root_workspace_path: workspace_root,
        }
    }
}

#[derive(Clone)]
struct AcpClient {
    request_tx: mpsc::UnboundedSender<Value>,
    pending: Arc<tokio::sync::Mutex<HashMap<u64, oneshot::Sender<Value>>>>,
    next_id: Arc<AtomicU64>,
}

struct AcpCallError {
    code: Option<i64>,
    message: String,
}

impl AcpClient {
    fn start(config: AcpServerConfig) -> (Self, mpsc::UnboundedReceiver<String>) {
        let (request_tx, request_rx) = mpsc::unbounded_channel();
        let (response_tx, response_rx) = mpsc::unbounded_channel();
        harn_parser::runtime_stack::builder()
            .name("harn-acp-client".to_owned())
            .spawn(move || {
                let runtime = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .expect("start ACP runtime");
                runtime.block_on(run_acp_channel_server(config, request_rx, response_tx));
            })
            .unwrap_or_else(|error| panic!("spawn harn-acp-client VM thread: {error}"));
        (
            Self {
                request_tx,
                pending: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
                next_id: Arc::new(AtomicU64::new(1)),
            },
            response_rx,
        )
    }

    fn spawn_output_loop(&self, mut response_rx: mpsc::UnboundedReceiver<String>, state: ApiState) {
        let client = self.clone();
        tokio::spawn(async move {
            while let Some(line) = response_rx.recv().await {
                let Ok(message) = serde_json::from_str::<Value>(&line) else {
                    continue;
                };
                if message.get("method").is_none() && message.get("id").is_some() {
                    client.resolve_response(message).await;
                    continue;
                }
                if message.get("method").is_some() && message.get("id").is_some() {
                    state.handle_acp_request(message).await;
                    continue;
                }
                state.handle_acp_notification(message);
            }
        });
    }

    async fn call(&self, method: &str, params: Value) -> Result<Value, String> {
        self.call_structured(method, params)
            .await
            .map_err(|error| error.message)
    }

    async fn call_structured(&self, method: &str, params: Value) -> Result<Value, AcpCallError> {
        let id = self.next_id.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.pending.lock().await.insert(id, tx);
        let request = json!({
            "jsonrpc": "2.0",
            "id": id,
            "method": method,
            "params": params,
        });
        if self.request_tx.send(request).is_err() {
            self.pending.lock().await.remove(&id);
            return Err(AcpCallError {
                code: None,
                message: "ACP runtime is not running".into(),
            });
        }
        let response = rx.await.map_err(|_| AcpCallError {
            code: None,
            message: "ACP runtime stopped before responding".into(),
        })?;
        if let Some(error) = response.get("error") {
            return Err(AcpCallError {
                code: error.get("code").and_then(Value::as_i64),
                message: error
                    .get("message")
                    .and_then(Value::as_str)
                    .map(str::to_string)
                    .unwrap_or_else(|| error.to_string()),
            });
        }
        Ok(response.get("result").cloned().unwrap_or(Value::Null))
    }

    async fn resolve_response(&self, response: Value) {
        let Some(id) = response.get("id").and_then(Value::as_u64) else {
            return;
        };
        let sender = self.pending.lock().await.remove(&id);
        if let Some(sender) = sender {
            let _ = sender.send(response);
        }
    }

    fn send_raw(&self, message: Value) {
        let _ = self.try_send_raw(message);
    }

    fn try_send_raw(&self, message: Value) -> Result<(), String> {
        self.request_tx
            .send(message)
            .map_err(|_| "ACP runtime is not running".to_owned())
    }
}

#[derive(Clone, Serialize)]
struct ApiEvent {
    id: String,
    object: &'static str,
    created_at: String,
    updated_at: String,
    metadata: Value,
    resource: ResourcePointer,
    sequence: u64,
    event: String,
    session_id: Option<String>,
    task_id: Option<String>,
    payload: Value,
}

#[derive(Clone, Serialize)]
struct ResourcePointer {
    object: String,
    id: String,
}

impl ApiEvent {
    fn new(
        id: String,
        created_at: String,
        event: String,
        session_id: Option<String>,
        task_id: Option<String>,
        payload: Value,
        fallback_resource: ResourcePointer,
    ) -> Self {
        let resource = payload
            .get("object")
            .and_then(Value::as_str)
            .filter(|object| !object.is_empty())
            .zip(
                payload
                    .get("id")
                    .and_then(Value::as_str)
                    .filter(|id| !id.is_empty()),
            )
            .map(|(object, id)| ResourcePointer {
                object: object.into(),
                id: id.into(),
            })
            .or_else(|| {
                task_id.as_ref().map(|id| ResourcePointer {
                    object: "task".into(),
                    id: id.clone(),
                })
            })
            .or_else(|| {
                session_id.as_ref().map(|id| ResourcePointer {
                    object: "session".into(),
                    id: id.clone(),
                })
            })
            .unwrap_or(fallback_resource);
        Self {
            sequence: event_seq(&id),
            id,
            object: "event",
            updated_at: created_at.clone(),
            created_at,
            metadata: json!({}),
            resource,
            event,
            session_id,
            task_id,
            payload,
        }
    }
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum EventPersistence {
    TaskControl,
    LiveProjection,
    SessionMessage,
}

#[derive(Clone)]
struct PendingPermission {
    public: Value,
    rpc_id: Option<u64>,
    hitl: bool,
    response_in_flight: bool,
}

impl ApiState {
    async fn handle_acp_request(&self, message: Value) {
        let id = message.get("id").cloned().unwrap_or(Value::Null);
        match message.get("method").and_then(Value::as_str) {
            Some("host/capabilities") => {
                self.acp.send_raw(json!({
                    "jsonrpc": "2.0",
                    "id": id,
                    "result": {}
                }));
            }
            Some("session/request_permission") => {
                self.register_permission_request(message);
            }
            Some(method) => {
                self.acp.send_raw(json!({
                    "jsonrpc": "2.0",
                    "id": id,
                    "error": {
                        "code": -32601,
                        "message": format!("Unsupported local API host request: {method}")
                    }
                }));
            }
            None => {}
        }
    }

    fn handle_acp_notification(&self, message: Value) {
        let Some(method) = message
            .get("method")
            .and_then(Value::as_str)
            .map(str::to_string)
        else {
            return;
        };
        match method.as_str() {
            "session/update" => self.register_session_update(message),
            _ => {
                let params = message.get("params").cloned().unwrap_or_else(|| json!({}));
                let Ok(task_id) = self.frame_task_id(&params) else {
                    tracing::warn!("Refused invalid task-owned ACP notification");
                    return;
                };
                let session_id = params
                    .get("sessionId")
                    .and_then(Value::as_str)
                    .map(str::to_owned);
                let _ = self.append_event_with_persistence(
                    session_id,
                    task_id,
                    &method,
                    message,
                    EventPersistence::LiveProjection,
                );
            }
        }
    }

    fn register_permission_request(&self, message: Value) {
        let params = message.get("params").cloned().unwrap_or_else(|| json!({}));
        let session_id = params
            .get("sessionId")
            .and_then(Value::as_str)
            .map(str::to_string);
        let request_id = params
            .pointer("/toolCall/toolCallId")
            .or_else(|| params.pointer("/toolCall/_meta/harn/approvalRequest/id"))
            .and_then(Value::as_str)
            .map(str::to_string)
            .unwrap_or_else(|| format!("permission_{}", Uuid::now_v7()));
        let task_id = match self.frame_task_id(&params) {
            Ok(task_id) => task_id,
            Err(error) => {
                self.acp
                    .send_raw(json!({"jsonrpc":"2.0", "id":message["id"],
                    "error":{"code":error.rpc_code(), "message":error.message()}}));
                return;
            }
        };
        let now = now_rfc3339();
        let public = json!({
            "id": request_id,
            "object": "permission_request",
            "created_at": now,
            "updated_at": now,
            "metadata": {},
            "session_id": session_id,
            "task_id": task_id,
            "status": "pending",
            "source": "acp",
            "action": params.pointer("/toolCall/_meta/harn/toolName")
                .or_else(|| params.pointer("/toolCall/title"))
                .cloned()
                .unwrap_or(Value::Null),
            "request": params,
            "response": null
        });
        let rpc_id = message.get("id").and_then(Value::as_u64);
        let pending = PendingPermission {
            public: public.clone(),
            rpc_id,
            hitl: false,
            response_in_flight: false,
        };
        {
            let mut inner = self.inner.lock().expect("api state poisoned");
            inner.permissions.insert(
                public["id"].as_str().expect("permission id").to_string(),
                pending,
            );
        }
        if let Some(task_id) = public.get("task_id").and_then(Value::as_str) {
            if self
                .change_task(
                    task_id,
                    harn_session_store::task::TaskStatus::AuthRequired,
                    None,
                )
                .is_err()
            {
                eprintln!("[harn] could not commit task permission state");
                return;
            }
        }
        let event_session_id = public["session_id"].as_str().map(str::to_string);
        let event_task_id = public["task_id"].as_str().map(str::to_string);
        self.append_event_from_resource(
            event_session_id,
            event_task_id,
            "permission.requested",
            public,
        );
    }

    fn register_session_update(&self, message: Value) {
        let params = message.get("params").cloned().unwrap_or_else(|| json!({}));
        let session_id = params
            .get("sessionId")
            .and_then(Value::as_str)
            .map(str::to_string);
        let Ok(task_id) = self.frame_task_id(&params) else {
            tracing::warn!("Refused invalid task-owned ACP session update");
            return;
        };
        if params
            .pointer("/update/sessionUpdate")
            .and_then(Value::as_str)
            == Some("hitl_request")
        {
            self.register_hitl_request(session_id.clone(), task_id.clone(), &params);
        }
        if params
            .pointer("/update/sessionUpdate")
            .and_then(Value::as_str)
            == Some("artifact")
        {
            artifacts::register_harn_session_artifact(
                self,
                session_id.clone(),
                task_id.clone(),
                &params,
            );
        }
        let _ = self.append_event_with_persistence(
            session_id,
            task_id,
            "session.update",
            params,
            EventPersistence::LiveProjection,
        );
    }

    fn register_hitl_request(
        &self,
        session_id: Option<String>,
        task_id: Option<String>,
        params: &Value,
    ) {
        let update = params.get("update").cloned().unwrap_or_else(|| json!({}));
        let harn = update
            .pointer("/_meta/harn")
            .cloned()
            .unwrap_or_else(|| json!({}));
        let request_id = harn
            .get("requestId")
            .and_then(Value::as_str)
            .map(str::to_string)
            .unwrap_or_else(|| format!("hitl_{}", Uuid::now_v7()));
        let now = now_rfc3339();
        let public = json!({
            "id": request_id,
            "object": "permission_request",
            "created_at": now,
            "updated_at": now,
            "metadata": {},
            "session_id": session_id,
            "task_id": task_id,
            "status": "pending",
            "source": "hitl",
            "action": harn.get("kind").cloned().unwrap_or(Value::Null),
            "request": harn.get("payload").cloned().unwrap_or(Value::Null),
            "response": null
        });
        let pending = PendingPermission {
            public: public.clone(),
            rpc_id: None,
            hitl: true,
            response_in_flight: false,
        };
        {
            let mut inner = self.inner.lock().expect("api state poisoned");
            inner.permissions.insert(request_id, pending);
        }
        if let Some(task_id) = public.get("task_id").and_then(Value::as_str) {
            let waiting = if harn.get("kind").and_then(Value::as_str) == Some("question") {
                harn_session_store::task::TaskStatus::InputRequired
            } else {
                harn_session_store::task::TaskStatus::AuthRequired
            };
            if self.change_task(task_id, waiting, None).is_err() {
                eprintln!("[harn] could not commit task permission state");
                return;
            }
        }
        self.append_event_from_resource(
            public["session_id"].as_str().map(str::to_string),
            public["task_id"].as_str().map(str::to_string),
            "permission.requested",
            public,
        );
    }

    fn frame_task_id(&self, params: &Value) -> Result<Option<String>, TaskAdmissionError> {
        let Some(target) = TaskPromptTarget::from_params(params)? else {
            return Ok(None);
        };
        let session_id = params
            .get("sessionId")
            .and_then(Value::as_str)
            .ok_or(TaskAdmissionError::InvalidTaskContext)?;
        let task = self
            .task_store()
            .map_err(|_| TaskAdmissionError::StorageUnavailable)?
            .get_task(&target.task_id)
            .map_err(|error| match error {
                harn_session_store::StoreError::NotFound(_) => {
                    TaskAdmissionError::InvalidTaskContext
                }
                _ => TaskAdmissionError::StorageUnavailable,
            })?;
        if task.session_id != session_id {
            return Err(TaskAdmissionError::InvalidTaskContext);
        }
        Ok(Some(target.task_id))
    }

    fn append_event(
        &self,
        session_id: Option<String>,
        task_id: Option<String>,
        event: &str,
        payload: Value,
    ) {
        self.append_event_from_resource(session_id, task_id, event, payload);
    }

    fn append_event_from_resource(
        &self,
        session_id: Option<String>,
        task_id: Option<String>,
        event: &str,
        payload: Value,
    ) {
        let _ = self.append_event_with_persistence(
            session_id,
            task_id,
            event,
            payload,
            EventPersistence::TaskControl,
        );
    }

    fn append_event_with_persistence(
        &self,
        session_id: Option<String>,
        task_id: Option<String>,
        event: &str,
        payload: Value,
        persistence: EventPersistence,
    ) -> Option<ApiEvent> {
        if let Some(task_id) = task_id
            .as_deref()
            .filter(|_| persistence == EventPersistence::TaskControl)
        {
            return match self.task_store().and_then(|store| {
                store.append_task_event(
                    task_id,
                    harn_session_store::task::NewTaskControlEvent {
                        event: event.into(),
                        payload,
                    },
                )
            }) {
                Ok(event) => Some(self.publish_task_event(event)),
                Err(_) => {
                    eprintln!("[harn] could not commit task control event");
                    None
                }
            };
        }
        let id = match self
            .task_store()
            .and_then(|store| store.next_control_event_id())
        {
            Ok(id) => id,
            Err(_) => {
                eprintln!("[harn] could not allocate control event identity");
                return None;
            }
        };
        let mut inner = self.inner.lock().expect("api state poisoned");
        let api_event = ApiEvent::new(
            id.clone(),
            now_rfc3339(),
            event.to_string(),
            session_id,
            task_id,
            payload,
            ResourcePointer {
                object: "workspace".into(),
                id: inner.root_workspace_id.clone(),
            },
        );
        if let Some(session_id) = api_event.session_id.as_deref() {
            if persistence == EventPersistence::SessionMessage {
                inner
                    .messages
                    .entry(session_id.into())
                    .or_default()
                    .push(api_event.payload.clone());
            }
            if let Some(session) = inner.sessions.get_mut(session_id) {
                session["last_event_id"] = json!(id);
                session["updated_at"] = json!(api_event.created_at.clone());
            }
        }
        inner.events.push(api_event.clone());
        drop(inner);
        let _ = self.events_tx.send(api_event.clone());
        Some(api_event)
    }

    fn history(&self, filter: &EventFilter) -> harn_session_store::StoreResult<Vec<ApiEvent>> {
        let task_events = self
            .ready_task_store()?
            .task_events(filter.task_id.as_deref(), filter.session_id.as_deref())?;
        let inner = self.inner.lock().expect("api state poisoned");
        let mut events: Vec<_> = inner
            .events
            .iter()
            .filter(|event| filter.matches(event))
            .cloned()
            .chain(
                task_events
                    .into_iter()
                    .map(ApiEvent::from)
                    .filter(|event| filter.matches(event)),
            )
            .collect();
        events.sort_by(|a, b| a.id.cmp(&b.id));
        Ok(events)
    }
}

#[derive(Clone, Default, Deserialize)]
struct ListQuery {
    limit: Option<usize>,
    cursor: Option<String>,
    after: Option<String>,
    workspace_id: Option<String>,
    session_id: Option<String>,
    task_id: Option<String>,
    path: Option<String>,
}

#[derive(Clone, Default)]
struct EventFilter {
    session_id: Option<String>,
    task_id: Option<String>,
    after: Option<String>,
}

impl EventFilter {
    fn from_query(query: &ListQuery, session_id: Option<String>, task_id: Option<String>) -> Self {
        Self {
            session_id: session_id.or_else(|| query.session_id.clone()),
            task_id: task_id.or_else(|| query.task_id.clone()),
            after: query.after.clone().or_else(|| query.cursor.clone()),
        }
    }

    fn matches(&self, event: &ApiEvent) -> bool {
        if self.after.as_ref().is_some_and(|after| event.id <= *after) {
            return false;
        }
        if let Some(session_id) = self.session_id.as_deref() {
            if event.session_id.as_deref() != Some(session_id) {
                return false;
            }
        }
        if let Some(task_id) = self.task_id.as_deref() {
            if event.task_id.as_deref() != Some(task_id) {
                return false;
            }
        }
        true
    }
}

fn api_router(state: ApiState) -> Router {
    let canonical_sessions = state.canonical_sessions.clone();
    let public = Router::new()
        .route("/health", get(meta::health))
        .route("/version", get(meta::version))
        .route("/openapi.json", get(meta::openapi_json))
        .route("/v1", get(meta::api_root))
        .route_layer(axum::middleware::from_fn_with_state(
            TaskRetryPolicy::Unsupported,
            retry_policy_guard,
        ));
    let router = Router::new()
        .route("/v1/runtime", get(meta::runtime))
        .route("/v1/capabilities", get(meta::capabilities))
        .route("/v1/provider-catalog", get(meta::provider_catalog))
        .route("/v1/tools", get(meta::list_tools))
        .route("/v1/tools/{tool_id}", get(meta::get_tool))
        .route(
            "/v1/workspaces",
            get(workspaces::list_workspaces).post(workspaces::create_workspace),
        )
        .route(
            "/v1/workspaces/{workspace_id}",
            get(workspaces::get_workspace).patch(workspaces::update_workspace),
        )
        .route(
            "/v1/workspaces/{workspace_id}/files",
            get(workspaces::read_workspace_file).put(workspaces::write_workspace_file),
        )
        .route(
            "/v1/sessions",
            get(sessions::list_sessions).post(sessions::create_session),
        )
        .route(
            "/v1/sessions/{session_id}",
            get(sessions::get_session).patch(sessions::update_session),
        )
        .route(
            "/v1/sessions/{session_id}/view",
            get(sessions::get_session_view),
        )
        .route(
            "/v1/sessions/{session_id}/close",
            post(sessions::close_session),
        )
        .route(
            "/v1/sessions/{session_id}/live-clients",
            get(sessions::list_session_live_clients),
        )
        .route(
            "/v1/sessions/{session_id}/attach",
            post(sessions::attach_session_client),
        )
        .route(
            "/v1/sessions/{session_id}/takeover",
            post(sessions::takeover_session_client),
        )
        .route(
            "/v1/sessions/{session_id}/detach",
            post(sessions::detach_session_client),
        )
        .route(
            "/v1/sessions/{session_id}/heartbeat",
            post(sessions::heartbeat_session_client),
        )
        .route(
            "/v1/sessions/{session_id}/fork",
            post(sessions::fork_session),
        )
        .route(
            "/v1/sessions/{session_id}/truncate",
            post(sessions::truncate_session),
        )
        .route(
            "/v1/sessions/{session_id}/messages",
            get(sessions::list_session_messages).post(sessions::append_session_message),
        )
        .route(
            "/v1/sessions/{session_id}/tasks",
            get(sessions::list_session_tasks),
        )
        .route("/v1/tasks", get(tasks::list_tasks))
        .route("/v1/tasks/{task_id}", get(tasks::get_task))
        .route("/v1/tasks/{task_id}/cancel", post(tasks::cancel_task))
        .route(
            "/v1/tasks/{task_id}/messages",
            post(tasks::append_task_message),
        )
        .route(
            "/v1/artifacts",
            get(artifacts::list_artifacts).post(artifacts::register_artifact),
        )
        .route("/v1/artifacts/{artifact_id}", get(artifacts::get_artifact))
        .route(
            "/v1/artifacts/{artifact_id}/content",
            get(artifacts::download_artifact_content),
        )
        .route("/v1/events", get(events::list_events))
        .route(
            "/v1/workflow-trigger-runs",
            get(events::list_workflow_trigger_runs),
        )
        .route("/v1/events/stream", get(events::stream_events))
        .route(
            "/v1/sessions/{session_id}/events",
            get(events::list_session_events),
        )
        .route(
            "/v1/sessions/{session_id}/events/stream",
            get(events::stream_session_events),
        )
        .route("/v1/tasks/{task_id}/events", get(events::list_task_events))
        .route(
            "/v1/tasks/{task_id}/events/stream",
            get(events::stream_task_events),
        )
        .route(
            "/v1/tasks/{task_id}/stream",
            get(events::stream_task_events),
        )
        .route(
            "/v1/permission-requests",
            get(permissions::list_permission_requests),
        )
        .route(
            "/v1/tasks/{task_id}/permission-requests",
            get(permissions::list_task_permission_requests),
        )
        .route(
            "/v1/permission-requests/{request_id}/respond",
            post(permissions::respond_permission_request),
        )
        .route(
            "/v1/permissions/policy",
            get(permissions::get_permissions_policy).put(permissions::put_permissions_policy),
        );
    canonical_sessions::mount(router, canonical_sessions)
        .route(
            "/v1/permissions/rules",
            get(permissions::list_permission_rules).post(permissions::create_permission_rule),
        )
        .route(
            "/v1/permissions/rules/{rule_id}",
            axum::routing::delete(permissions::revoke_permission_rule),
        )
        .route(
            "/v1/permissions/history",
            get(permissions::get_permission_history),
        )
        .route("/v1/permissions/check", post(permissions::check_permission))
        // Apply the default to every existing route, including nested stores.
        // The two supported methods are then merged with their own policy.
        .route_layer(axum::middleware::from_fn_with_state(
            TaskRetryPolicy::Unsupported,
            retry_policy_guard,
        ))
        .route("/v1/tasks", task_create_route(post(tasks::submit_task)))
        .route(
            "/v1/sessions/{session_id}/tasks",
            task_create_route(post(sessions::submit_session_task)),
        )
        .route_layer(axum::middleware::from_fn_with_state(
            state.auth_policy.clone(),
            authentication::shared_workspace_authority,
        ))
        .route_layer(axum::middleware::from_fn(protocol_version_guard))
        .merge(public)
        .layer(DefaultBodyLimit::max(crate::DEFAULT_HTTP_BODY_LIMIT_BYTES))
        .with_state(state)
}

async fn protocol_version_guard(
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> Response {
    let mut versions = request
        .headers()
        .get_all("harn-agents-protocol-version")
        .iter();
    if versions.next().and_then(|version| version.to_str().ok()) != Some(API_PROTOCOL_VERSION)
        || versions.next().is_some()
    {
        return (
            StatusCode::UPGRADE_REQUIRED,
            Json(json!({"error": {
                "code": "unsupported_protocol_version",
                "message": "provide the current Harn Agents Protocol version",
                "type": "request_error",
                "details": {"supported_versions": [API_PROTOCOL_VERSION]},
            }})),
        )
            .into_response();
    }
    next.run(request).await
}

fn task_create_route(
    route: axum::routing::MethodRouter<ApiState>,
) -> axum::routing::MethodRouter<ApiState> {
    route.layer(axum::middleware::from_fn_with_state(
        TaskRetryPolicy::TaskCreate,
        retry_policy_guard,
    ))
}

async fn retry_policy_guard(
    State(policy): State<TaskRetryPolicy>,
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> Response {
    if request.headers().contains_key("idempotency-key")
        && policy.supported_method() != Some(request.method().as_str())
    {
        return api_error(
            StatusCode::BAD_REQUEST,
            "idempotency_not_supported",
            "this operation does not support an idempotency key",
        );
    }
    next.run(request).await
}

enum TaskRetryHeaderError {
    MultipleKeys,
    Key(TaskRetryError),
}

impl IntoResponse for TaskRetryHeaderError {
    fn into_response(self) -> Response {
        let message = match self {
            Self::MultipleKeys => "provide one idempotency key".to_owned(),
            Self::Key(error) => error.to_string(),
        };
        api_error(StatusCode::BAD_REQUEST, "invalid_idempotency_key", &message)
    }
}

fn task_retry_key(headers: &HeaderMap) -> Result<Option<TaskRetryKey>, TaskRetryHeaderError> {
    let mut keys = headers.get_all("idempotency-key").iter();
    let Some(raw) = keys.next() else {
        return Ok(None);
    };
    if keys.next().is_some() {
        return Err(TaskRetryHeaderError::MultipleKeys);
    }
    let raw = raw
        .to_str()
        .map_err(|_| TaskRetryHeaderError::Key(TaskRetryError::InvalidKey))?;
    TaskRetryKey::parse(raw)
        .map(Some)
        .map_err(TaskRetryHeaderError::Key)
}

/// Render a scope mismatch as the REST API's standard `forbidden` error
/// payload, mirroring how a cloud gateway reports the same case.
/// Uses the same `kind`/`required_scopes`/`granted_scopes`/`missing_scopes`
/// fields as the JSON-RPC adapters; the only difference is the REST
/// adapter inlines them under a single `error` envelope object instead
/// of nesting under `error.data`.
fn forbidden_api_error(required: &BTreeSet<String>, granted: &BTreeSet<String>) -> Response {
    let mut body = crate::forbidden_data_payload(required, granted);
    if let Some(map) = body.as_object_mut() {
        map.insert(
            "message".to_string(),
            json!(crate::forbidden_message(required, granted)),
        );
    }
    (StatusCode::FORBIDDEN, Json(json!({ "error": body }))).into_response()
}

/// 400 response for a request body that failed to parse as JSON.
fn invalid_json_response() -> Response {
    api_error(
        StatusCode::BAD_REQUEST,
        "invalid_json",
        "request body must be JSON",
    )
}

fn parse_json_body(body: &Bytes) -> Result<Value, serde_json::Error> {
    if body.is_empty() {
        return Ok(json!({}));
    }
    serde_json::from_slice(body)
}

fn list_response<T: Serialize>(data: Vec<T>) -> Value {
    json!({
        "object": "list",
        "data": data,
        "has_more": false,
        "next_cursor": null
    })
}

fn limit_values(mut values: Vec<Value>, limit: Option<usize>) -> Vec<Value> {
    if let Some(limit) = limit {
        values.truncate(limit);
    }
    values
}

fn api_error(status: StatusCode, code: &str, message: &str) -> Response {
    (
        status,
        Json(json!({
            "error": {
                "code": code,
                "message": message
            }
        })),
    )
        .into_response()
}

fn now_rfc3339() -> String {
    // No `Clock` reaches these handlers; threading one through the axum state
    // would be invasive. See `harn_clock::system_now_rfc3339`.
    harn_clock::system_now_rfc3339()
}

fn event_seq(id: &str) -> u64 {
    id.strip_prefix("event_")
        .and_then(|value| u64::from_str_radix(value, 16).ok())
        .unwrap_or(0)
}

fn merge_mutable_fields(target: &mut Value, input: &Value, fields: &[&str]) {
    for field in fields {
        if let Some(value) = input.get(*field) {
            target[*field] = value.clone();
        }
    }
}

fn workspace_root_locked(inner: &ApiStateInner, workspace_id: &str) -> Option<PathBuf> {
    if workspace_id == inner.root_workspace_id {
        return Some(inner.root_workspace_path.clone());
    }
    inner
        .workspaces
        .get(workspace_id)
        .and_then(|workspace| workspace.get("root").and_then(Value::as_str))
        .map(PathBuf::from)
}

fn message_resource(session_id: &str, task_id: Option<&str>, input: Value) -> Value {
    message_resource_with_id(
        session_id,
        task_id,
        input,
        format!("message_{}", Uuid::now_v7()),
    )
}

fn message_resource_with_id(
    session_id: &str,
    task_id: Option<&str>,
    input: Value,
    message_id: String,
) -> Value {
    let now = now_rfc3339();
    let message = normalize_message_input(input);
    json!({
        "id": message_id,
        "object": "message",
        "created_at": now,
        "updated_at": now,
        "metadata": message.get("metadata").cloned().unwrap_or_else(|| json!({})),
        "session_id": session_id,
        "task_id": task_id,
        "role": message.get("role").and_then(Value::as_str).unwrap_or("user"),
        "parts": message.get("parts").cloned().unwrap_or_else(|| json!([]))
    })
}

fn normalize_message_input(input: Value) -> Value {
    if input.get("role").is_some() && input.get("parts").is_some() {
        return input;
    }
    if let Some(text) = input
        .as_str()
        .or_else(|| input.get("text").and_then(Value::as_str))
    {
        return json!({
            "role": input.get("role").and_then(Value::as_str).unwrap_or("user"),
            "parts": [{
                "type": "text",
                "text": text,
                "visibility": "public"
            }],
            "metadata": input.get("metadata").cloned().unwrap_or_else(|| json!({}))
        });
    }
    let role = input
        .get("role")
        .and_then(Value::as_str)
        .unwrap_or("user")
        .to_string();
    json!({
        "role": role,
        "parts": [{
            "type": "json",
            "value": input,
            "visibility": "public"
        }],
        "metadata": {}
    })
}

fn prompt_text(input: &Value) -> Option<String> {
    if let Some(text) = input
        .as_str()
        .or_else(|| input.get("text").and_then(Value::as_str))
    {
        return Some(text.to_string());
    }
    if let Some(message) = input.get("message") {
        return prompt_text(message);
    }
    if let Some(value) = input.get("input") {
        return prompt_text(value);
    }
    let parts = input.get("parts")?.as_array()?;
    let text = parts
        .iter()
        .filter_map(|part| part.get("text").and_then(Value::as_str))
        .collect::<Vec<_>>()
        .join("\n");
    (!text.trim().is_empty()).then_some(text)
}

#[cfg(test)]
mod tests;