lean-ctx-server 3.12.3

// SPDX-License-Identifier: Apache-2.0
//! Post-dispatch side-effect stages for `LeanCtxServer::call_tool_guarded`
//! (issue #144).
//!
//! These blocks run *after* a tool produced its (already post-processed) output
//! and perform pure side effects — they never mutate the text returned to the
//! model. They are tightly coupled to `&self` (locks, peers, spawned tasks), so
//! they live as thin `LeanCtxServer` methods rather than free functions. Moving
//! them out of the guarded path keeps that function a readable orchestrator;
//! behaviour, ordering and await points are identical to the inlined versions.

#[allow(clippy::wildcard_imports)]
use super::*;

impl LeanCtxServer {
    /// Record the tool receipt, infer/apply intent, persist the session when due,
    /// trigger auto-consolidation, and attribute token cost. All work is either
    /// synchronous bookkeeping under the session lock or fire-and-forget blocking
    /// tasks; nothing here feeds back into the tool output.
    pub(super) async fn record_receipt_and_cost(
        &self,
        name: &str,
        args: Option<&serde_json::Map<String, serde_json::Value>>,
        action: Option<&str>,
        result_text: &str,
        output_token_count: usize,
    ) {
        let input = helpers::canonical_args_string(args);
        let input_md5 = helpers::hash_fast(&input);
        let output_md5 = helpers::hash_fast(result_text);
        // A server-wide last-task snapshot is not an execution identity: another
        // concurrent call may have replaced it while this call was running.
        let envelope = lean_ctx::core::task_spine::TaskSpine::current();
        let agent_id = if let Some(task) = envelope.as_ref() {
            Some(task.agent_id.as_str().to_owned())
        } else {
            self.agent_id.read().await.clone()
        };
        let client_name = self.client_name.read().await.clone();
        let task_id = envelope.as_ref().map(|envelope| envelope.task_id.as_str());
        let session_save_due = {
            let empty_args = serde_json::Map::new();
            let args_map = args.unwrap_or(&empty_args);
            let mut session = self.session.write().await;
            session.record_tool_receipt_with_task(
                name,
                action,
                &input_md5,
                &output_md5,
                agent_id.as_deref(),
                Some(&client_name),
                task_id,
            );

            if let Some(intent) = lean_ctx::core::intent_protocol::infer_from_tool_call(
                name,
                action,
                args_map,
                session.project_root.as_deref(),
            ) {
                session.record_intent(intent);
            }
            session.should_save()
        };

        if session_save_due {
            let ir_clone = self.context_ir.clone();
            let session = self.session.instance();
            tokio::spawn(async move {
                lean_ctx::core::session::SessionState::save_shared_logged(session).await;
                tokio::task::spawn_blocking(move || {
                    if let Some(ir) = ir_clone
                        && let Ok(ir_guard) = ir.try_read()
                    {
                        ir_guard.save();
                    }
                    // Flush the persistent stub index on the same batch cadence so
                    // recent full deliveries survive a restart as cheap stubs (#955).
                    lean_ctx::core::read_stub_index::persist();
                });
            });
        }

        if self.autonomy.is_enabled() {
            let (calls, project_root) = {
                let session = self.session.read().await;
                (session.stats.total_tool_calls, session.project_root.clone())
            };

            if let Some(root) = project_root
                && lean_ctx::core::autonomy::should_auto_consolidate(&self.autonomy, calls)
            {
                let root_clone = root.clone();
                // Imports this server's own session, never the project's
                // latest saved one (I07/S03).
                let own = self.session.instance();
                tokio::task::spawn_blocking(move || {
                    lean_ctx::core::session::caller::scope(Some(own), || {
                        lean_ctx::core::policy::runtime::in_project(root_clone.clone(), || {
                            lean_ctx::core::consolidation_engine::consolidate_latest(
                                &root_clone,
                                lean_ctx::core::consolidation_engine::ConsolidationBudgets::default(
                                ),
                            )
                        })
                    })
                });
            }
        }

        let agent_key = agent_id.unwrap_or_else(|| "unknown".to_string());
        let input_token_count = lean_ctx::core::tokens::count_tokens(&input) as u64;
        let output_token_count_u64 = output_token_count as u64;
        let name_owned = name.to_string();
        tokio::task::spawn_blocking(move || {
            let mut store = lean_ctx::core::a2a::cost_attribution::CostStore::load();
            store.record_tool_call(
                &agent_key,
                &client_name,
                &name_owned.clone(),
                input_token_count,
                output_token_count_u64,
                0,
            );
            if let Err(e) = store.save() {
                tracing::warn!("lean-ctx: failed to persist cost attribution: {e}");
            }

            // R29: Dispatch-level evidence for kernel health + adaptive signals.
            lean_ctx::core::context_kernel::evidence_wiring::record_from_tool_dispatch(
                &name_owned.clone(),
                input_token_count as usize,
                output_token_count_u64 as usize,
                0, // savings tracked separately per-tool
            );

            // R30: Response evidence — output token tracking.
            lean_ctx::core::context_kernel::response_evidence::record_response(
                &name_owned.clone(),
                output_token_count_u64 as usize,
                false,
            );
        });
    }

    /// Context OS: persist the shared session snapshot and publish the matching
    /// bus events (primary `ToolCallRecorded` plus any secondary kind). No-op
    /// outside shared session mode. Fire-and-forget on a blocking task.
    pub(super) async fn persist_shared_context_os(
        &self,
        name: &str,
        action: Option<&str>,
        args: Option<&serde_json::Map<String, serde_json::Value>>,
    ) {
        if self.session_mode != crate::tools::SessionMode::Shared {
            return;
        }
        let ws = self.workspace_id.clone();
        let ch = self.channel_id.clone();
        let rt = self.context_os.clone();
        let agent = self.agent_id.read().await.clone();
        let tool = name.to_string();
        let tool_action = action.map(str::to_string);
        let tool_path = helpers::get_str(args, "path");
        let tool_category = helpers::get_str(args, "category");
        let tool_key = helpers::get_str(args, "key");
        let session_snapshot = self.session.read().await.clone();
        let session_task = session_snapshot.task.clone();
        tokio::task::spawn_blocking(move || {
            let Some(rt) = rt else {
                return;
            };
            let Some(root) = session_snapshot.project_root.as_deref() else {
                return;
            };
            rt.shared_sessions
                .persist_best_effort(root, &ws, &ch, &session_snapshot);
            rt.metrics.record_session_persisted();

            let mut base_payload = serde_json::json!({
                "tool": tool,
                "action": tool_action,
            });
            if let Some(ref p) = tool_path {
                base_payload["path"] = serde_json::Value::String(p.clone());
            }
            if let Some(ref c) = tool_category {
                base_payload["category"] = serde_json::Value::String(c.clone());
            }
            if let Some(ref k) = tool_key {
                base_payload["key"] = serde_json::Value::String(k.clone());
            }
            if let Some(ref t) = session_task {
                base_payload["reasoning"] = serde_json::Value::String(t.description.clone());
            }

            if rt
                .bus
                .append(
                    &ws,
                    &ch,
                    &lean_ctx::core::context_os::ContextEventKindV1::ToolCallRecorded,
                    agent.as_deref(),
                    base_payload.clone(),
                )
                .is_some()
            {
                rt.metrics.record_event_appended();
                rt.metrics.record_event_broadcast();
            }

            if let Some(secondary) =
                lean_ctx::core::context_os::secondary_event_kind(&tool, tool_action.as_deref())
                && rt
                    .bus
                    .append(&ws, &ch, &secondary, agent.as_deref(), base_payload)
                    .is_some()
            {
                rt.metrics.record_event_appended();
                rt.metrics.record_event_broadcast();
            }
        });
    }
}