harn-serve 0.10.129

use super::*;

impl AcpServer {
    pub fn new(config: AcpServerConfig) -> Self {
        Self::new_with_output(config, AcpOutput::stdout())
    }

    /// Create an ACP server that writes responses and notifications to a
    /// caller-provided output sink.
    ///
    /// Prefer [`crate::EmbeddedAgent`] or [`run_acp_channel_server_with_handle`]
    /// unless the host already owns the compatible current-thread runtime and
    /// wants to drive incoming JSON-RPC messages directly.
    pub fn new_with_output(config: AcpServerConfig, output: AcpOutput) -> Self {
        let notifier_output = output.clone();
        let known_sessions = super::session_watch::KnownSessions::default();
        let notifier_sessions = known_sessions.clone();
        let llm_config_overrides = config.llm_config_overrides.clone();
        let runtime_provider_endpoint_overrides = config
            .runtime_configurator
            .runtime_provider_endpoint_overrides();
        let llm_capability_overrides = config.llm_capability_overrides.clone();
        let concurrent_controls = ConcurrentSessionControls::new(
            config.auth_policy.methods.is_empty() || config.authenticated_principal.is_some(),
            serde_json::json!({
                "authMethods": config.auth_policy.acp_auth_methods(),
            }),
        );

        Self {
            descriptor: AdapterDescriptor {
                id: "acp".to_string(),
                caller_shape: "agent-session".to_string(),
                supports_streaming: true,
                supports_cancel: true,
            },
            pipeline: config.pipeline,
            auth_policy: config.auth_policy,
            authenticated_principal: config.authenticated_principal,
            runtime_configurator: config.runtime_configurator,
            sessions: HashMap::new(),
            concurrent_controls,
            timeline_subscriptions: HashMap::new(),
            next_id: AtomicU64::new(1),
            pending: Arc::new(Mutex::new(HashMap::new())),
            session_cancellations: Arc::new(std::sync::Mutex::new(HashMap::new())),
            output,
            compile_cache: None,
            vm_baseline_cache: None,
            profile: config.profile,
            llm_config_overrides,
            runtime_provider_endpoint_overrides,
            llm_capability_overrides,
            default_budget: config.budget,
            sandbox: config.sandbox,
            active_bulk_auth: std::sync::Mutex::new(None),
            known_sessions,
            _session_change_subscription: harn_vm::subscribe_session_changes(Arc::new(
                super::session_watch::SessionInfoNotifier::new(notifier_output, notifier_sessions),
            )),
        }
    }

    /// Record a session this client now has, so the session-change notifier
    /// forwards its metadata changes. Paired with every `sessions.insert`; a
    /// session missing here is simply never pushed, never mis-pushed.
    pub(super) fn track_known_session(&self, session_id: &str) {
        let mut known = self
            .known_sessions
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        known.insert(session_id.to_string());
    }

    /// Dispatch an ACP request with this server's LLM routing context.
    ///
    /// Provider and capability overlays are per-request ambient state, not
    /// constructor-time thread state: two embedded servers may share a Tokio
    /// worker and suspend independently while a provider request is in flight.
    pub async fn handle_incoming_message(&mut self, msg: serde_json::Value) {
        let provider_overrides = self.llm_config_overrides.clone();
        let runtime_provider_endpoint_overrides = self.runtime_provider_endpoint_overrides.clone();
        let capability_overrides = self.llm_capability_overrides.clone();
        // Erase the large method-router state machine before it enters the
        // generic ambient wrapper. This intentionally pays one heap allocation
        // per message and virtual dispatch per poll so the wrapper's generated
        // code stays independent of the router state and drop glue.
        let dispatch: std::pin::Pin<Box<dyn std::future::Future<Output = ()> + '_>> =
            Box::pin(self.handle_incoming_message_scoped(msg));
        harn_vm::orchestration::scope_llm_runtime_overrides_with_provider_endpoints(
            provider_overrides,
            capability_overrides,
            runtime_provider_endpoint_overrides,
            dispatch,
        )
        .await;
    }

    /// Compile `source` for `target_pipeline` (or the default entry point
    /// when `target_pipeline` is None), reusing the cached chunk when the
    /// file at `source_path` has the same mtime as the last cache fill and
    /// the target hasn't changed.
    ///
    /// Returns `(chunk, hit)` so the caller can keep its existing compile-
    /// time telemetry meaningful (hits report ~0 ms).
    ///
    /// Inline-mode prompts pass `source_path: None` and never hit cache —
    /// the source is freshly generated per turn so there's nothing to reuse.
    pub(super) fn compile_pipeline_cached(
        &mut self,
        source: &str,
        source_path: Option<&Path>,
        target_pipeline: Option<&str>,
    ) -> Result<(harn_vm::Chunk, bool), String> {
        let target_owned = target_pipeline.map(|s| s.to_string());
        let cache_key = source_path.and_then(|path| {
            std::fs::metadata(path)
                .and_then(|m| m.modified())
                .ok()
                .map(|mtime| (path.to_path_buf(), mtime))
        });
        if let Some((ref path, mtime)) = cache_key {
            if let Some(entry) = self.compile_cache.as_ref() {
                if entry.path == *path
                    && entry.mtime == mtime
                    && entry.target_pipeline == target_owned
                    && entry.source == source
                {
                    return Ok((entry.chunk.clone(), true));
                }
            }
        }
        let chunk = match target_pipeline {
            Some(name) => harn_vm::compile_source_named(source, name),
            None => harn_vm::compile_source(source),
        }
        .map_err(|e| format!("Compilation error: {e}"))?;
        if let Some((path, mtime)) = cache_key {
            self.compile_cache = Some(CompileCacheEntry {
                path,
                mtime,
                target_pipeline: target_owned,
                source: source.to_string(),
                chunk: chunk.clone(),
            });
        }
        Ok((chunk, false))
    }

    pub(super) async fn prepare_vm_baseline_cached(
        &mut self,
        source: &str,
        source_path: Option<&Path>,
        target_pipeline: Option<&str>,
        cwd: &Path,
        project_root: &Path,
        mode_id: &str,
    ) -> Result<(Option<harn_vm::VmBaseline>, Option<bool>, u64), String> {
        let Some(source_path) = source_path else {
            return Ok((None, None, 0));
        };

        let prepare_started = Instant::now();
        let target_owned = target_pipeline.map(str::to_string);
        let cache_key = std::fs::metadata(source_path)
            .and_then(|m| m.modified())
            .ok()
            .map(|mtime| (source_path.to_path_buf(), mtime));
        let project_root = Some(project_root.to_path_buf());

        if let Some((ref path, mtime)) = cache_key {
            if let Some(entry) = self.vm_baseline_cache.as_ref() {
                if entry.path == *path
                    && entry.mtime == mtime
                    && entry.target_pipeline == target_owned
                    && entry.source == source
                    && entry.cwd == cwd
                    && entry.project_root == project_root
                    && entry.mode_id == mode_id
                {
                    return Ok((
                        Some(entry.baseline.clone()),
                        Some(true),
                        prepare_started.elapsed().as_millis() as u64,
                    ));
                }
            }
        }

        let baseline = execute::prepare_vm_baseline(
            source,
            source_path,
            cwd,
            project_root.as_deref(),
            self.runtime_configurator.clone(),
        )
        .await?;
        if let Some((path, mtime)) = cache_key {
            self.vm_baseline_cache = Some(VmBaselineCacheEntry {
                path,
                mtime,
                target_pipeline: target_owned,
                source: source.to_string(),
                cwd: cwd.to_path_buf(),
                project_root,
                mode_id: mode_id.to_string(),
                baseline: baseline.clone(),
            });
        } else {
            self.vm_baseline_cache = None;
        }

        Ok((
            Some(baseline),
            Some(false),
            prepare_started.elapsed().as_millis() as u64,
        ))
    }

    /// Write a complete JSON-RPC message to the current transport.
    pub(super) fn write_line(&self, line: &str) {
        self.output.write_line(line);
    }

    /// Send a JSON-RPC success response.
    pub(super) fn send_response(&self, id: &serde_json::Value, result: serde_json::Value) {
        let response = harn_vm::jsonrpc::response(id.clone(), result);
        if let Ok(line) = serde_json::to_string(&response) {
            self.write_line(&line);
        }
    }

    /// Send a JSON-RPC error response.
    pub(super) fn send_error(&self, id: &serde_json::Value, code: i64, message: &str) {
        let response = harn_vm::jsonrpc::error_response(id.clone(), code, message);
        if let Ok(line) = serde_json::to_string(&response) {
            self.write_line(&line);
        }
    }

    pub(super) fn send_error_with_data(
        &self,
        id: &serde_json::Value,
        code: i64,
        message: &str,
        data: serde_json::Value,
    ) {
        let response = harn_vm::jsonrpc::error_response_with_data(id.clone(), code, message, data);
        if let Ok(line) = serde_json::to_string(&response) {
            self.write_line(&line);
        }
    }

    pub(super) fn send_session_open_error(
        &self,
        id: &serde_json::Value,
        error: &harn_vm::agent_sessions::SessionOpenError,
    ) {
        match error {
            harn_vm::agent_sessions::SessionOpenError::CapacityExhausted {
                limit,
                active,
                protected,
            } => self.send_error_with_data(
                id,
                -32000,
                &error.to_string(),
                serde_json::json!({
                    "code": "agent_session_capacity_exhausted",
                    "limit": limit,
                    "active": active,
                    "protected": protected,
                }),
            ),
            harn_vm::agent_sessions::SessionOpenError::LineageRejected { session_id, reason } => {
                self.send_error_with_data(
                    id,
                    -32000,
                    &error.to_string(),
                    serde_json::json!({
                        "code": "agent_session_lineage_rejected",
                        "session_id": session_id,
                        "reason": reason,
                    }),
                );
            }
        }
    }

    pub(super) fn emit_control_outcome(
        &self,
        session_id: &str,
        method: &str,
        outcome: &str,
        status: &str,
        actor: serde_json::Value,
        target: serde_json::Value,
        reason: Option<&str>,
    ) {
        harn_vm::agent_events::emit_event(&harn_vm::agent_events::AgentEvent::ControlOutcome {
            session_id: session_id.to_string(),
            control_id: control_id(),
            method: method.to_string(),
            outcome: outcome.to_string(),
            status: status.to_string(),
            actor,
            target,
            reason: reason.map(str::to_string),
            metadata: serde_json::Value::Null,
        });
    }

    /// Record an accepted control word and publish its outcome. See
    /// [`record_and_emit_control`], which owns both halves.
    pub(super) fn record_and_emit_control(
        &self,
        session_id: &str,
        control: harn_session_store::ControlEvent,
    ) {
        record_and_emit_control(session_id, control);
    }

    /// Send a JSON-RPC notification (no id, no response expected).
    pub(super) fn send_notification(&self, method: &str, params: serde_json::Value) {
        let notification = harn_vm::jsonrpc::notification(method, params);
        if let Ok(line) = serde_json::to_string(&notification) {
            self.write_line(&line);
        }
    }

    /// Emit a terminal prompt failure as a single typed JSON-RPC error.
    ///
    /// A terminal failure is never assistant content: this path emits exactly
    /// one JSON-RPC error carrying the typed [`AcpPromptErrorData`] and never a
    /// `session/update` `agent_message_chunk`. Compile/setup/runtime string
    /// errors carry the `generic_throw` class and no route facts.
    pub(super) fn send_prompt_error(&self, id: &serde_json::Value, message: &str) {
        self.send_prompt_failure(
            id,
            message,
            harn_vm::llm::AgentTerminalClass::GenericThrow,
            super::types::AcpPromptFailureFacts::default(),
        );
    }

    /// Emit a terminal prompt failure with the typed class and structured
    /// machine facts projected from the thrown error dict.
    pub(super) fn send_prompt_failure(
        &self,
        id: &serde_json::Value,
        message: &str,
        terminal_class: harn_vm::llm::AgentTerminalClass,
        facts: super::types::AcpPromptFailureFacts,
    ) {
        let data = super::types::AcpPromptErrorData::with_facts(terminal_class, facts);
        self.send_error_with_data(
            id,
            -32000,
            message,
            serde_json::to_value(data).expect("ACP prompt error data must serialize"),
        );
        eprintln!("{message}");
    }

    pub(super) fn send_prompt_protocol_error(&self, id: &serde_json::Value, message: &str) {
        let data = super::types::AcpPromptErrorData::new(
            harn_vm::llm::AgentTerminalClass::AgentLoopProtocolFailure,
        );
        self.send_error_with_data(
            id,
            -32602,
            message,
            serde_json::to_value(data).expect("ACP prompt error data must serialize"),
        );
        eprintln!("{message}");
    }

    /// Generate a unique session ID.
    pub(super) fn next_session_id(&mut self) -> String {
        uuid::Uuid::new_v4().to_string()
    }

    pub(super) fn register_session_cancellation(
        &mut self,
        session_id: &str,
    ) -> SessionCancellation {
        let cancellation = SessionCancellation::default();
        self.session_cancellations
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(session_id.to_string(), cancellation.clone());
        cancellation
    }
}

/// Record an accepted control word and publish its outcome.
///
/// One function owns both halves so a stored row and the live
/// `control_outcome` notification always carry the same `control_id`,
/// and so no acceptance site can emit the wire event while forgetting
/// the durable one. Every ACP path that accepts a stop or an injection
/// routes through here: the dispatch handlers, and the transport-level
/// preemption paths that answer a control while a prompt is in flight.
/// Those preemption paths are the mid-turn case — the one the record
/// exists for — so leaving them out would have made the record present
/// exactly when nobody needed it.
///
/// The journal write can miss: a control aimed at a session this VM
/// thread does not own has no event stream. That outcome rides on the
/// notification's `metadata.recorded` rather than being swallowed,
/// because "no control row in the store" must not read the same as "no
/// control happened".
pub(super) fn record_and_emit_control(session_id: &str, control: harn_session_store::ControlEvent) {
    let outcome = harn_vm::agent_sessions::record_control_event(session_id, &control);
    harn_vm::agent_events::emit_event(&harn_vm::agent_events::AgentEvent::ControlOutcome {
        session_id: session_id.to_string(),
        control_id: control.control_id.clone(),
        method: control.method.clone(),
        outcome: "accepted".to_string(),
        status: control.status.clone(),
        actor: control.actor.clone(),
        target: serde_json::json!({
            "sessionId": session_id,
            "messageId": control.message_id,
        }),
        reason: None,
        metadata: serde_json::json!({
            "action": control.action.as_str(),
            "requestedMode": control.requested_mode,
            "deliveryMode": control.delivery_mode,
            "recorded": outcome.as_str(),
        }),
    });
}