//! Turn lifecycle: running a single interaction, executing tools, and
//! wiring the context pipeline + sub-agent harness around them.
//!
//! This file owns the "hot path" methods on `Agent`:
//!
//! - [`Agent::turn`] — the big one. Orchestrates system-prompt build,
//!   memory-context injection, the provider loop, tool dispatch, and
//!   the context pipeline (tool-result budget → microcompact →
//!   autocompact signal → session-memory extraction trigger).
//! - [`Agent::execute_tool_call`] / [`Agent::execute_tools`] — the
//!   per-call runners.
//! - [`Agent::build_parent_execution_context`] — snapshot helper for
//!   the parent-context task-local that sub-agents read.
//! - [`Agent::trim_history`], [`Agent::fetch_learned_context`],
//!   [`Agent::build_system_prompt`] — the small helpers `turn()` leans
//!   on every call.
//! - [`Agent::spawn_session_memory_extraction`] — the fire-and-forget
//!   background archivist fork.

use super::transcript;
use super::types::Agent;
use crate::core::event_bus::{publish_global, DomainEvent};
use crate::openhuman::agent::dispatcher::{ParsedToolCall, ToolExecutionResult};
use crate::openhuman::agent::harness;
use crate::openhuman::agent::hooks::{self, ToolCallRecord, TurnContext};
use crate::openhuman::agent::memory_loader::collect_recall_citations;
use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::agent::tool_policy::{
    ToolCallContext, ToolPolicyDecision, ToolPolicyRequest,
};
use crate::openhuman::agent_experience::{
    prepend_experience_block, render_experience_hits, AgentExperienceStore, ExperienceQuery,
};
use crate::openhuman::agent_tool_policy::render_tool_policy_boundary;
use crate::openhuman::context::prompt::{LearnedContextData, PromptContext, PromptTool};
use crate::openhuman::context::{ReductionOutcome, ARCHIVIST_EXTRACTION_PROMPT};
use crate::openhuman::inference::model_context::context_window_for_model;
use crate::openhuman::inference::provider::{
    ChatMessage, ChatRequest, ConversationMessage, ProviderDelta,
};
use crate::openhuman::memory::MemoryCategory;
use crate::openhuman::tools::traits::ToolCallOptions;
use crate::openhuman::tools::Tool;
use crate::openhuman::util::truncate_with_ellipsis;

use crate::openhuman::agent::harness::token_budget::{
    trim_chat_messages_to_budget, trim_conversation_history_to_budget,
};
use anyhow::Result;
use std::hash::{Hash, Hasher};
use std::sync::Arc;

impl Agent {
    /// Executes a single interaction "turn" with the agent.
    ///
    /// This function is the primary driver of the agent's behavior. It manages the
    /// end-to-end lifecycle of a user request:
    ///
    /// 1. **Initialization**: Resumes from a session transcript if this is a new turn
    ///    to preserve KV-cache stability.
    /// 2. **Prompt Construction**: Builds the system prompt (only on the first turn)
    ///    incorporating learned context and tool instructions.
    /// 3. **Context Injection**: Enriches the user message with relevant memories
    ///    fetched via the [`MemoryLoader`].
    /// 4. **Execution Loop**: Enters a loop (up to `max_tool_iterations`) where it:
    ///    - Manages the context window (reduction/summarization).
    ///    - Calls the LLM provider.
    ///    - Parses and executes tool calls.
    ///    - Accumulates results into history.
    /// 5. **Synthesis**: Returns the final assistant response after all tools have
    ///    finished or the iteration budget is exhausted.
    /// 6. **Background Tasks**: Triggers episodic memory indexing and facts
    ///    extraction asynchronously.
    pub async fn turn(&mut self, user_message: &str) -> Result<String> {
        let turn_started = std::time::Instant::now();
        self.emit_progress(AgentProgress::TurnStarted).await;
        log::info!("[agent] turn started — awaiting user message processing");
        log::info!(
            "[agent_loop] turn start message_chars={} history_len={} max_tool_iterations={}",
            user_message.chars().count(),
            self.history.len(),
            self.config.max_tool_iterations
        );
        // ── Session transcript resume ─────────────────────────────────
        // On a fresh session (empty history), look for a previous
        // transcript to pre-populate the exact provider messages for
        // KV cache prefix reuse.
        if self.history.is_empty() && self.cached_transcript_messages.is_none() {
            self.try_load_session_transcript();
        }

        if self.history.is_empty() {
            // Learned context is only baked into the system prompt on the
            // very first turn — once the history is non-empty we reuse the
            // stored prompt verbatim to preserve the KV-cache prefix the
            // inference backend has already tokenised. Fetching it later
            // would just burn memory-store reads on data we throw away.
            if !self.connected_integrations_initialized {
                self.fetch_connected_integrations().await;
                // Sessions born without a cached Composio view still need
                // a one-shot delegation-surface reconcile before the system
                // prompt is frozen. The shared-Arc failure path returns
                // `false`, but on turn 1 the Arc should still be uniquely
                // owned; a `false` return here indicates a programmer error
                // and the warn-level log inside the helper already surfaces
                // it, so we keep the existing best-effort contract.
                let _ = self.refresh_delegation_tools();
            }
            let learned = self.fetch_learned_context().await;
            let rendered_prompt = self.build_system_prompt(learned)?;
            log::info!("[agent] system prompt built — initialising conversation history");
            log::info!(
                "[agent_loop] system prompt built chars={}",
                rendered_prompt.chars().count()
            );
            // User-file injection (PROFILE.md, MEMORY.md) puts
            // potentially-sensitive content (LinkedIn scrape output,
            // archivist-curated memories) into the system prompt. Avoid
            // leaking that to debug logs — log a length + content hash
            // instead. Narrow specialists (both flags off) keep the
            // full-body log so prompt-engineering iteration on
            // tools/safety sections stays easy.
            if self.omit_profile && self.omit_memory_md {
                log::debug!("[agent_loop] system prompt body:\n{}", rendered_prompt);
            } else {
                let mut hasher = std::collections::hash_map::DefaultHasher::new();
                rendered_prompt.hash(&mut hasher);
                log::debug!(
                    "[agent_loop] system prompt body redacted (contains PROFILE/MEMORY): chars={} hash={:016x}",
                    rendered_prompt.chars().count(),
                    hasher.finish()
                );
            }
            self.history
                .push(ConversationMessage::Chat(ChatMessage::system(
                    rendered_prompt,
                )));
            // Seed the per-turn mid-session refresh baseline with the
            // hash of whatever Composio actually returned just now.
            // Subsequent turns short-circuit unless this hash changes.
            self.last_seen_integrations_hash =
                crate::openhuman::composio::connected_set_hash(&self.connected_integrations);
        } else {
            // Deliberately do NOT rebuild the system prompt on subsequent
            // turns. The rendered prompt is the KV-cache prefix the inference
            // backend has already tokenised; replacing its bytes (even
            // cosmetically) forces the backend to re-prefill from scratch.
            //
            // Dynamic turn-to-turn context (memory recall, learned snippets)
            // rides on the user message via `memory_loader.load_context()`
            // — that's where the caller should inject anything that varies
            // between turns.
            //
            // *** Mid-session schema-only refresh ***
            //
            // The system prompt stays frozen, but the function-calling
            // schema (the `tools` field in the provider request) is sent
            // fresh on every API call — it's not part of the KV-cache
            // prefix. So we *can* react to Composio connect/disconnect
            // events mid-session by re-synthesising the `delegate_<toolkit>`
            // surface on `self.tools` / `self.tool_specs` and letting
            // the next provider call carry the new schema. KV cache stays
            // intact; the system prompt's `## Connected Integrations`
            // block goes mildly stale until the next session, but the
            // schema is the source of truth the model actually routes
            // against.
            //
            // The signal we react to is the process-wide
            // [`crate::openhuman::composio::INTEGRATIONS_CACHE`], kept
            // current by (a) the desktop UI's 5 s
            // `composio_list_connections` poll, (b) the post-OAuth
            // `ComposioConnectionCreatedSubscriber` invalidation, and
            // (c) the 60 s TTL fallback. We read it via the read-only
            // [`crate::openhuman::composio::cached_active_integrations`]
            // helper — never trigger a backend fetch ourselves, never
            // block on a writer.
            // Session agents built through `from_config_*` carry their
            // runtime `Config` snapshot directly, so this read avoids the
            // old `Config::load_or_init()` round-trip on every turn.
            //
            if let Some(cfg) = self.integration_runtime_config.as_ref() {
                if let Some(cache_view) =
                    crate::openhuman::composio::cached_active_integrations(cfg)
                {
                    let new_hash = crate::openhuman::composio::connected_set_hash(&cache_view);
                    if new_hash != self.last_seen_integrations_hash {
                        log::info!(
                            "[agent_loop] connection set changed mid-session (hash {:x} -> {:x}); refreshing tool schema (system prompt left intact for KV cache)",
                            self.last_seen_integrations_hash,
                            new_hash
                        );
                        // Snapshot the previous integration list so we
                        // can roll back if `refresh_delegation_tools`
                        // bails on a shared `Arc` — otherwise the
                        // agent's `connected_integrations` and its
                        // schema would disagree until the next
                        // event-driven refresh, and the
                        // `last_seen_integrations_hash` advance below
                        // would suppress retries.
                        let prev_integrations =
                            std::mem::replace(&mut self.connected_integrations, cache_view);
                        if self.refresh_delegation_tools() {
                            self.last_seen_integrations_hash = new_hash;
                            self.connected_integrations_initialized = true;
                        } else {
                            // Reconcile aborted (shared Arc) — restore
                            // the previous integration list so the
                            // next turn re-detects the same change and
                            // retries cleanly. We deliberately do NOT
                            // advance `last_seen_integrations_hash`.
                            self.connected_integrations = prev_integrations;
                        }
                    }
                }
            }
            // Cache empty/expired or config unavailable => no signal.
            // We leave the current tool surface alone and pick up any
            // real change on the next turn after the UI's 5 s poll has
            // repopulated [`INTEGRATIONS_CACHE`].

            log::trace!(
                "[agent_loop] system prompt reused (history_len={}) — KV cache prefix preserved",
                self.history.len()
            );
        }

        if self.auto_save {
            let _ = self
                .memory
                .store(
                    "",
                    "user_msg",
                    user_message,
                    MemoryCategory::Conversation,
                    None,
                )
                .await;
        }

        log::info!("[agent] loading memory context for user message");
        const MEMORY_CITATION_LIMIT: usize = 5;
        const MEMORY_CITATION_MIN_RELEVANCE: f64 = 0.4;
        match collect_recall_citations(
            self.memory.as_ref(),
            user_message,
            MEMORY_CITATION_LIMIT,
            MEMORY_CITATION_MIN_RELEVANCE,
        )
        .await
        {
            Ok(citations) => {
                log::debug!(
                    "[agent_loop] memory citations collected count={}",
                    citations.len()
                );
                self.last_turn_citations = citations;
            }
            Err(err) => {
                log::warn!("[agent_loop] memory citation collection failed: {err}");
                self.last_turn_citations.clear();
            }
        }
        let context = self
            .memory_loader
            .load_context(self.memory.as_ref(), user_message)
            .await
            .unwrap_or_default();

        // ── Memory-tree eager prefetch (#710 wiring) ──────────────────
        // The orchestrator session injects a cross-source digest on the
        // first turn AND every `tree_loader::REFRESH_INTERVAL` (30 min by
        // default) thereafter, so long-running conversations stay current
        // with newly-ingested memory. Each injection still rides on the
        // user message (NOT the system prompt) to keep the KV-cache prefix
        // stable. Failure is non-fatal — bare `context` is returned on any
        // error. The timestamp is bumped on every successful `load` (even
        // when the digest is empty) so an empty workspace doesn't get
        // re-queried every turn.
        //
        // Gate STM preemptive recall on the session turn index, independent
        // of tree-prefetch success/failure. (Previously keyed off
        // `last_tree_prefetch_at.is_none()`, which stays `None` when tree
        // prefetch fails — re-firing STM recall on every later turn.)
        let is_first_turn_for_stm = self.context.stats().session_memory_current_turn == 0;
        let now = std::time::Instant::now();
        let context = if crate::openhuman::agent::tree_loader::should_prefetch(
            self.last_tree_prefetch_at,
            now,
            crate::openhuman::agent::tree_loader::REFRESH_INTERVAL,
        ) {
            match crate::openhuman::config::rpc::load_config_with_timeout().await {
                Ok(cfg) => {
                    match crate::openhuman::agent::tree_loader::TreeContextLoader::load(&cfg).await
                    {
                        Ok(tree_ctx) => {
                            let was_first = self.last_tree_prefetch_at.is_none();
                            self.last_tree_prefetch_at = Some(now);
                            if !tree_ctx.is_empty() {
                                log::info!(
                                    "[memory_tree] tree context injected first_turn={} chars={}",
                                    was_first,
                                    tree_ctx.chars().count()
                                );
                                format!("{context}{tree_ctx}")
                            } else {
                                context
                            }
                        }
                        Err(e) => {
                            log::warn!("[memory_tree] tree_loader.load failed (non-fatal): {e}");
                            context
                        }
                    }
                }
                Err(e) => {
                    log::warn!(
                        "[memory_tree] tree_loader skipped — config load failed (non-fatal): {e}"
                    );
                    context
                }
            }
        } else {
            log::trace!("[memory_tree] tree_loader skipped — within refresh interval");
            context
        };

        // ── Phase 3 STM preemptive recall ────────────────────────────
        // On the very first turn only, assemble a bounded cross-thread
        // context block from the FTS5 episodic arm (keyword match) and the
        // segment-embedding arm (cosine similarity). The block rides on the
        // user message (NOT the system prompt) to keep the KV-cache prefix
        // stable, exactly like the tree-context injection above.
        //
        // Gate: `learning.stm_recall_enabled` must be true AND this must
        // be the first turn (STM is snapshot-frozen at session start).
        // Failure is non-fatal — bare `context` passes through untouched.
        let mut context = if is_first_turn_for_stm {
            // Load config to check the gate. Use a cached load (cheap).
            let stm_enabled = crate::openhuman::config::rpc::load_config_with_timeout()
                .await
                .map(|cfg| cfg.learning.stm_recall_enabled)
                .unwrap_or(true); // default: enabled

            if stm_enabled {
                if let Some(conn) = self.memory.sqlite_conn() {
                    use crate::openhuman::memory::stm_recall::recall::{stm_recall, StmRecallOpts};
                    let opts = StmRecallOpts {
                        exclude_session: &self.event_session_id,
                        query: if user_message.trim().is_empty() {
                            None
                        } else {
                            Some(user_message)
                        },
                        model_signature: None,
                    };
                    match stm_recall(&conn, &opts, None) {
                        Ok(block) if !block.is_empty() => {
                            let stm_md = block.render();
                            log::info!(
                                "[stm_recall] preemptive block injected: {} items, ~{} chars, fts5_candidates={}, dropped_dedup={}",
                                block.items.len(),
                                stm_md.chars().count(),
                                block.fts5_candidates,
                                block.dropped_dedup
                            );
                            format!("{stm_md}{context}")
                        }
                        Ok(_) => {
                            log::debug!(
                                "[stm_recall] preemptive recall: no cross-thread context found"
                            );
                            context
                        }
                        Err(e) => {
                            log::warn!("[stm_recall] preemptive recall failed (non-fatal): {e}");
                            context
                        }
                    }
                } else {
                    log::debug!("[stm_recall] preemptive recall skipped — no SQLite connection on memory backend");
                    context
                }
            } else {
                log::debug!("[stm_recall] preemptive recall skipped — stm_recall_enabled=false");
                context
            }
        } else {
            context
        };

        // ── Lane B: situational preferences (every turn) ─────────────────────
        // Recall topic-scoped preferences semantically relevant to THIS message
        // (model-aware embeddings, gated by vector similarity) and inject them
        // under a banner. Runs every turn — unlike the first-turn-gated tree/STM
        // blocks above — because the query changes per message; it rides the
        // per-turn context that's prepended to the user message (no KV-cache
        // cost). An unrelated message clears the similarity gate to nothing, so
        // no block is injected.
        {
            let situational =
                crate::openhuman::memory::preferences::recall_situational_preferences(
                    &self.memory,
                    user_message,
                )
                .await;
            if !situational.is_empty() {
                log::info!(
                    "[pref_recall] situational block injected: {} item(s)",
                    situational.len()
                );
                context.push_str("## Relevant preferences for this message\n\n");
                for pref in &situational {
                    context.push_str("- ");
                    context.push_str(pref.trim());
                    context.push('\n');
                }
                context.push('\n');
            } else {
                log::debug!("[pref_recall] no situational preference relevant to this message");
            }
        }

        let enriched = if context.is_empty() {
            log::info!("[agent] no memory context found — using raw user message");
            self.last_memory_context = None;
            user_message.to_string()
        } else {
            log::info!(
                "[agent] memory context loaded — enriching user message context_chars={}",
                context.chars().count()
            );
            self.last_memory_context = Some(context.clone());
            format!("{context}{user_message}")
        };

        let enriched = self
            .inject_agent_experience_context(user_message, enriched)
            .await;

        // ── SKILL.md body injection (#781) ───────────────────────────
        // Match installed SKILL.md skills against the user message and
        // prepend their bodies ahead of the memory-context block so the
        // LLM sees them at the top of the user turn. See the module
        // docs on [`crate::openhuman::skills::inject`] for the matching
        // heuristic and size cap rationale.
        let enriched = {
            use crate::openhuman::skills::inject;
            let matches = inject::match_skills(&self.skills, user_message);
            if matches.is_empty() {
                log::debug!(
                    "[skills:inject] no skill matches for user message (skill_catalog_len={})",
                    self.skills.len()
                );
                enriched
            } else {
                let injection = inject::render_injection(
                    &matches,
                    inject::DEFAULT_MAX_INJECTION_BYTES,
                    |skill| skill.read_body(),
                );
                let matched_count = injection.decisions.iter().filter(|d| d.matched).count();
                log::info!(
                    "[skills:inject] summary candidates={} matched={} injected_bytes={} truncated_any={}",
                    injection.decisions.len(),
                    matched_count,
                    injection.injected_bytes,
                    injection.truncated
                );
                if injection.rendered.is_empty() {
                    enriched
                } else {
                    format!("{}\n{}", injection.rendered, enriched)
                }
            }
        };

        self.history
            .push(ConversationMessage::Chat(ChatMessage::user(enriched)));

        // Pin the main agent to its configured model for the lifetime of
        // the session. Per-turn classification used to run here, but it
        // would flip `effective_model` mid-conversation (e.g. reasoning →
        // coding based on a single keyword). Every flip invalidates the
        // backend's KV cache namespace for this session, costing full
        // re-prefill on the very next turn. The main agent's job is to
        // decide *which sub-agent* to spawn — that routing lives in the
        // model prompt, not in the Rust-side classifier. Sub-agents pick
        // their own tier via `ModelSpec::Hint(...)` in their definition.
        let effective_model = self.model_name.clone();
        log::info!(
            "[agent_loop] model pinned model={} (per-turn classification disabled for KV cache stability)",
            effective_model
        );

        // Snapshot the parent's runtime once per turn so any
        // `spawn_subagent` invocation that fires inside this turn can
        // read it via the PARENT_CONTEXT task-local. We override the
        // model field with the post-classification effective model.
        let mut parent_context = self.build_parent_execution_context();
        parent_context.model_name = effective_model.clone();

        // Bump the session-memory turn counter. Used later by
        // `should_extract_session_memory` to decide whether to spawn a
        // background archivist fork at end-of-turn.
        self.context.tick_turn();

        // Collect tool call records across all iterations for post-turn hooks
        let mut all_tool_records: Vec<ToolCallRecord> = Vec::new();

        // Capture the last `Vec<ChatMessage>` sent to the provider so we
        // can persist it as a session transcript after the turn completes.
        let mut last_provider_messages: Option<Vec<ChatMessage>> = None;

        // Accumulate usage stats across iterations for the transcript.
        let mut cumulative_input_tokens: u64 = 0;
        let mut cumulative_output_tokens: u64 = 0;
        let mut cumulative_cached_input_tokens: u64 = 0;
        let mut cumulative_charged_usd: f64 = 0.0;

        // Per-turn usage from the final provider response, attached to the
        // last assistant message in the persisted transcript.
        let mut last_turn_usage: Option<transcript::TurnUsage> = None;

        let turn_body = async {
            for iteration in 0..self.config.max_tool_iterations {
                self.emit_progress(AgentProgress::IterationStarted {
                    iteration: (iteration + 1) as u32,
                    max_iterations: self.config.max_tool_iterations as u32,
                })
                .await;
                log::info!(
                    "[agent_loop] iteration start i={} history_len={}",
                    iteration + 1,
                    self.history.len()
                );

                if let Some(context_window) = context_window_for_model(&effective_model) {
                    let budget_outcome =
                        trim_conversation_history_to_budget(&mut self.history, context_window);
                    if budget_outcome.trimmed {
                        log::warn!(
                            "[agent_loop] pre-dispatch history trimmed model={} context_window={} original_tokens={} final_tokens={} messages_removed={}",
                            effective_model,
                            context_window,
                            budget_outcome.original_tokens,
                            budget_outcome.final_tokens,
                            budget_outcome.messages_removed
                        );
                    }
                }

                // Global context management: run the reduction chain
                // before every provider hit. Cheap when the guard is
                // healthy; executes the summarizer LLM call
                // internally when the pipeline asks for autocompaction
                // (summarization, microcompact, and the circuit
                // breaker all live inside [`ContextManager`]).
                let outcome = self.context.reduce_before_call(&mut self.history).await?;
                match &outcome {
                    ReductionOutcome::NoOp => {}
                    ReductionOutcome::Microcompacted {
                        envelopes_cleared,
                        entries_cleared,
                        bytes_freed,
                    } => {
                        log::info!(
                            "[agent_loop] context microcompact i={} envelopes={} entries={} bytes_freed={}",
                            iteration + 1,
                            envelopes_cleared,
                            entries_cleared,
                            bytes_freed
                        );
                    }
                    ReductionOutcome::Summarized(stats) => {
                        log::info!(
                            "[agent_loop] context autocompact summarized i={} messages_removed={} approx_tokens_freed={} summary_chars={}",
                            iteration + 1,
                            stats.messages_removed,
                            stats.approx_tokens_freed,
                            stats.summary_chars
                        );
                    }
                    ReductionOutcome::SummarizationFailed {
                        utilisation_pct,
                        reason,
                    } => {
                        log::warn!(
                            "[agent_loop] context summarizer failed i={} utilisation_pct={} reason={}",
                            iteration + 1,
                            utilisation_pct,
                            reason
                        );
                    }
                    ReductionOutcome::NotAttempted { utilisation_pct } => {
                        log::warn!(
                            "[agent_loop] context autocompact disabled in config i={} utilisation_pct={}",
                            iteration + 1,
                            utilisation_pct
                        );
                    }
                    ReductionOutcome::Exhausted {
                        utilisation_pct,
                        reason,
                    } => {
                        log::error!(
                            "[agent_loop] context exhausted i={} utilisation_pct={} reason={}",
                            iteration + 1,
                            utilisation_pct,
                            reason
                        );
                        return Err(anyhow::anyhow!(
                            "Context window exhausted ({utilisation_pct}% full): {reason}"
                        ));
                    }
                }

                // Use cached transcript messages on the first iteration of
                // a resumed session to provide a byte-identical prefix for
                // KV cache reuse. After `.take()` the cache is consumed;
                // subsequent iterations rebuild from history normally.
                let mut messages = if let Some(mut cached) = self.cached_transcript_messages.take()
                {
                    // Append only the delta (new user message) from the
                    // end of the current history.
                    let new_tail = self.tool_dispatcher.to_provider_messages(
                        &self.history[self.history.len().saturating_sub(1)..],
                    );
                    cached.extend(new_tail);
                    log::info!(
                        "[transcript] resumed from cached transcript prefix_len={} new_tail={}",
                        cached.len() - 1,
                        1
                    );
                    cached
                } else {
                    self.tool_dispatcher.to_provider_messages(&self.history)
                };
                if let Some(context_window) = context_window_for_model(&effective_model) {
                    let budget_outcome =
                        trim_chat_messages_to_budget(&mut messages, context_window);
                    if budget_outcome.trimmed {
                        log::warn!(
                            "[agent_loop] pre-dispatch provider messages trimmed model={} context_window={} original_tokens={} final_tokens={} messages_removed={}",
                            effective_model,
                            context_window,
                            budget_outcome.original_tokens,
                            budget_outcome.final_tokens,
                            budget_outcome.messages_removed
                        );
                    }
                }

                last_provider_messages = Some(messages.clone());

                log::info!(
                    "[agent] iteration {}/{} — sending request to provider model={}",
                    iteration + 1,
                    self.config.max_tool_iterations,
                    effective_model
                );
                log::info!(
                    "[agent_loop] provider request i={} messages={} send_tool_specs={}",
                    iteration + 1,
                    messages.len(),
                    self.tool_dispatcher.should_send_tool_specs()
                );
                let provider_started = std::time::Instant::now();
                // Only set up the streaming sink when someone is
                // listening for progress events. Without a listener the
                // channel buffer would fill up and back-pressure the
                // provider; skipping it also keeps the non-streaming
                // HTTP path alive for providers that don't implement
                // SSE.
                let iteration_for_stream = (iteration + 1) as u32;
                let (delta_tx_opt, delta_forwarder) = if self.on_progress.is_some() {
                    let (tx, mut rx) = tokio::sync::mpsc::channel::<ProviderDelta>(128);
                    let progress_tx = self.on_progress.clone();
                    let forwarder = tokio::spawn(async move {
                        while let Some(event) = rx.recv().await {
                            let Some(ref sink) = progress_tx else {
                                continue;
                            };
                            let mapped = match event {
                                ProviderDelta::TextDelta { delta } => AgentProgress::TextDelta {
                                    delta,
                                    iteration: iteration_for_stream,
                                },
                                ProviderDelta::ThinkingDelta { delta } => {
                                    AgentProgress::ThinkingDelta {
                                        delta,
                                        iteration: iteration_for_stream,
                                    }
                                }
                                ProviderDelta::ToolCallStart { call_id, tool_name } => {
                                    AgentProgress::ToolCallArgsDelta {
                                        call_id,
                                        tool_name,
                                        delta: String::new(),
                                        iteration: iteration_for_stream,
                                    }
                                }
                                ProviderDelta::ToolCallArgsDelta { call_id, delta } => {
                                    AgentProgress::ToolCallArgsDelta {
                                        call_id,
                                        tool_name: String::new(),
                                        delta,
                                        iteration: iteration_for_stream,
                                    }
                                }
                            };
                            // Await backpressure so streamed deltas arrive
                            // in order and aren't silently dropped when the
                            // downstream progress bridge is slow.
                            if sink.send(mapped).await.is_err() {
                                break;
                            }
                        }
                    });
                    (Some(tx), Some(forwarder))
                } else {
                    (None, None)
                };
                let response = match self
                    .provider
                    .chat(
                        ChatRequest {
                            messages: &messages,
                            tools: if self.tool_dispatcher.should_send_tool_specs() {
                                Some(self.visible_tool_specs.as_slice())
                            } else {
                                None
                            },
                            stream: delta_tx_opt.as_ref(),
                        },
                        &effective_model,
                        self.temperature,
                    )
                    .await
                {
                    Ok(resp) => {
                        log::info!(
                            "[agent_loop] provider response i={} elapsed_ms={} text_chars={} native_tool_calls={}",
                            iteration + 1,
                            provider_started.elapsed().as_millis(),
                            resp.text.as_ref().map_or(0, |t| t.chars().count()),
                            resp.tool_calls.len()
                        );
                        log::debug!("[agent_loop] provider response: {resp:?}");
                        // Feed the context manager (guard +
                        // session-memory token accounting). No-op when
                        // the provider doesn't return usage.
                        if let Some(ref usage) = resp.usage {
                            self.context.record_usage(usage);
                            cumulative_input_tokens += usage.input_tokens;
                            cumulative_output_tokens += usage.output_tokens;
                            cumulative_cached_input_tokens += usage.cached_input_tokens;
                            cumulative_charged_usd += usage.charged_amount_usd;
                            // Snapshot this turn's usage so the transcript
                            // writer can attribute it to the last assistant
                            // message.
                            last_turn_usage = Some(transcript::TurnUsage {
                                model: effective_model.clone(),
                                usage: transcript::MessageUsage {
                                    input: usage.input_tokens,
                                    output: usage.output_tokens,
                                    cached_input: usage.cached_input_tokens,
                                    cost_usd: usage.charged_amount_usd,
                                },
                                ts: chrono::Utc::now().to_rfc3339(),
                            });
                        } else {
                            // Missing usage on this iteration: clear any
                            // snapshot carried from a prior iteration so
                            // the transcript doesn't attribute stale
                            // numbers to the final assistant message.
                            last_turn_usage = None;
                        }
                        resp
                    }
                    Err(err) => {
                        drop(delta_tx_opt);
                        if let Some(handle) = delta_forwarder {
                            let _ = handle.await;
                        }
                        return Err(err);
                    }
                };
                drop(delta_tx_opt);
                if let Some(handle) = delta_forwarder {
                    let _ = handle.await;
                }

                let (text, calls) = self.tool_dispatcher.parse_response(&response);
                let calls = Self::with_fallback_tool_call_ids(calls, iteration);
                log::info!(
                    "[agent] provider responded — parsed tool_calls={} text_chars={}",
                    calls.len(),
                    text.chars().count()
                );
                log::info!(
                    "[agent_loop] parsed response i={} parsed_text_chars={} parsed_tool_calls={}",
                    iteration + 1,
                    text.chars().count(),
                    calls.len()
                );
                if calls.is_empty() {
                    let final_text = if text.is_empty() {
                        response.text.unwrap_or_default()
                    } else {
                        text
                    };
                    log::info!(
                        "[agent] no tool calls — returning final response after {} iteration(s)",
                        iteration + 1
                    );
                    log::info!(
                        "[agent_loop] final response i={} final_chars={}",
                        iteration + 1,
                        final_text.chars().count()
                    );

                    self.emit_progress(AgentProgress::TurnCompleted {
                        iterations: (iteration + 1) as u32,
                    })
                    .await;

                    self.history
                        .push(ConversationMessage::Chat(ChatMessage::assistant(
                            final_text.clone(),
                        )));
                    self.trim_history();

                    // Mirror the final assistant reply into the transcript
                    // snapshot so the JSONL persisted below captures the
                    // response (not just the prompt that was sent).
                    if let Some(ref mut msgs) = last_provider_messages {
                        msgs.push(ChatMessage::assistant(final_text.clone()));
                    }

                    // Persist the transcript **now** — right after the
                    // provider response lands — so a crash during hooks
                    // / memory-extraction / the outer epilogue can't
                    // lose the assistant's reply.
                    if let Some(ref messages) = last_provider_messages {
                        self.persist_session_transcript(
                            messages,
                            cumulative_input_tokens,
                            cumulative_output_tokens,
                            cumulative_cached_input_tokens,
                            cumulative_charged_usd,
                            last_turn_usage.as_ref(),
                        );
                    }

                    if self.auto_save {
                        let summary = truncate_with_ellipsis(&final_text, 100);
                        let _ = self
                            .memory
                            .store("", "assistant_resp", &summary, MemoryCategory::Daily, None)
                            .await;
                    }

                    // Session-memory tool-call accounting. The actual
                    // background extraction spawn happens *outside*
                    // `turn_body` so the spawned task can take an owned
                    // parent context without fighting the borrow
                    // checker against `self`. We capture the decision
                    // here and surface it via the manager's session
                    // state — the epilogue (below) reads
                    // `should_extract_session_memory()`.
                    self.context.record_tool_calls(all_tool_records.len());

                    // Fire post-turn hooks (non-blocking)
                    if !self.post_turn_hooks.is_empty() {
                        let ctx = TurnContext {
                            user_message: user_message.to_string(),
                            assistant_response: final_text.clone(),
                            tool_calls: all_tool_records,
                            turn_duration_ms: turn_started.elapsed().as_millis() as u64,
                            session_id: Some(self.event_session_id.clone())
                                .filter(|session_id| !session_id.trim().is_empty()),
                            agent_id: Some(self.agent_definition_id.clone())
                                .filter(|agent_id| !agent_id.trim().is_empty()),
                            entrypoint: Some(self.event_channel.clone())
                                .filter(|entrypoint| !entrypoint.trim().is_empty()),
                            iteration_count: iteration + 1,
                        };
                        hooks::fire_hooks(&self.post_turn_hooks, ctx);
                    }

                    return Ok(final_text);
                }

                if !text.is_empty() {
                    log::info!(
                        "[agent_loop] assistant pre-tool text i={} chars={}",
                        iteration + 1,
                        text.chars().count()
                    );
                    // Push the assistant text into history; rendering is
                    // the caller's responsibility (the CLI loop walks
                    // `agent.history()` after each turn, sub-agents and
                    // library consumers get whatever they need through
                    // the returned value / history accessors).
                    self.history
                        .push(ConversationMessage::Chat(ChatMessage::assistant(
                            text.clone(),
                        )));
                }
                let tool_names: Vec<&str> = calls.iter().map(|call| call.name.as_str()).collect();
                log::info!(
                    "[agent] dispatching {} tool(s): {:?}",
                    calls.len(),
                    tool_names
                );
                log::info!(
                    "[agent_loop] executing tools i={} names={:?}",
                    iteration + 1,
                    tool_names
                );
                let persisted_tool_calls =
                    Self::persisted_tool_calls_for_history(&response, &calls, iteration);
                log::info!(
                    "[agent_loop] persisting assistant tool calls i={} persisted_tool_calls={} parsed_tool_calls={}",
                    iteration + 1,
                    persisted_tool_calls.len(),
                    calls.len()
                );
                self.history.push(ConversationMessage::AssistantToolCalls {
                    text: if text.is_empty() {
                        None
                    } else {
                        Some(text.clone())
                    },
                    tool_calls: persisted_tool_calls,
                });

                // Persist the transcript **right after** the provider
                // response lands — before executing tools — so if the
                // session crashes mid-tool-call we still have the
                // assistant's response + tool-call intents on disk.
                // Rebuild `last_provider_messages` from the current
                // history so the snapshot includes whatever the
                // assistant just emitted (plain text + tool calls).
                last_provider_messages =
                    Some(self.tool_dispatcher.to_provider_messages(&self.history));
                if let Some(ref messages) = last_provider_messages {
                    self.persist_session_transcript(
                        messages,
                        cumulative_input_tokens,
                        cumulative_output_tokens,
                        cumulative_cached_input_tokens,
                        cumulative_charged_usd,
                        last_turn_usage.as_ref(),
                    );
                }

                let (results, records) = self.execute_tools(&calls, iteration).await;
                all_tool_records.extend(records);
                log::info!(
                    "[agent_loop] tool results complete i={} result_count={}",
                    iteration + 1,
                    results.len()
                );
                for r in &results {
                    log::info!(
                        "[agent] tool response name={} success={} output_chars={}",
                        r.name,
                        r.success,
                        r.output.chars().count(),
                    );
                    log::debug!(
                        "[agent] tool response body name={}: {}",
                        r.name,
                        truncate_with_ellipsis(&r.output, 300)
                    );
                }
                log::info!(
                    "[agent] all tools complete for iteration {} — looping back to provider",
                    iteration + 1
                );
                let formatted = self.tool_dispatcher.format_results(&results);
                self.history.push(formatted);
                self.trim_history();
                // Flush the transcript again now that tool results have
                // been appended — the pre-tool persist above only
                // captured the assistant's tool-call intents. A crash
                // or early-exit between iterations would otherwise lose
                // the tool output from the on-disk session record.
                let post_tool_messages = self.tool_dispatcher.to_provider_messages(&self.history);
                self.persist_session_transcript(
                    &post_tool_messages,
                    cumulative_input_tokens,
                    cumulative_output_tokens,
                    cumulative_cached_input_tokens,
                    cumulative_charged_usd,
                    last_turn_usage.as_ref(),
                );
                last_provider_messages = Some(post_tool_messages);
                log::info!(
                    "[agent_loop] iteration end i={} history_len={}",
                    iteration + 1,
                    self.history.len()
                );
            }

            log::warn!(
                "[agent] exceeded max tool iterations ({}) — aborting turn",
                self.config.max_tool_iterations
            );
            log::warn!(
                "[agent_loop] exceeded maximum tool iterations max={}",
                self.config.max_tool_iterations
            );
            // Return the typed `AgentError::MaxIterationsExceeded` variant
            // (boxed through `anyhow::Error`) so `Agent::run_single` can
            // downcast at the Sentry funnel and suppress emission — this is
            // a deterministic agent-state outcome, not a bug (OPENHUMAN-
            // TAURI-99 / -98). The `Display` text is preserved verbatim so
            // the user-visible chat-rendered "Error: Agent exceeded
            // maximum tool iterations" string is unchanged.
            Err(anyhow::Error::new(
                crate::openhuman::agent::error::AgentError::MaxIterationsExceeded {
                    max: self.config.max_tool_iterations,
                },
            ))
        }; // end of `turn_body` async block

        // Run the turn body inside the parent-execution-context scope so
        // that any `spawn_subagent` tool call fired during the loop can
        // read the parent's provider, tools, model, and workspace via
        // the PARENT_CONTEXT task-local.
        let result = harness::with_parent_context(parent_context, turn_body).await;

        // Session transcript persistence lives INSIDE the turn body —
        // one write per provider response, fired right after the
        // response lands (see the tool-call and terminal branches in
        // `turn_body`). A crash during tool execution no longer drops
        // the assistant's reply because it was already flushed to
        // disk before tool dispatch started. No outer-loop save is
        // needed here.

        // ── Session-memory extraction (stage 5) ───────────────────────
        //
        // If the pipeline's deltas have crossed all three thresholds
        // (token growth, tool calls, turn count), spawn a *background*
        // archivist sub-agent that will distil durable facts into the
        // workspace MEMORY.md file via the `update_memory_md` tool.
        //
        // The spawn is fire-and-forget: the main turn returns the
        // user-visible response immediately, and the archivist runs
        // asynchronously on the `agentic` tier. We optimistically mark
        // the extraction complete right away — if it actually fails,
        // we'll just retry on the next threshold window (a few turns
        // later), which is the right amount of retry behaviour for a
        // librarian task that's idempotent across reruns.
        if result.is_ok() && self.context.should_extract_session_memory() {
            self.spawn_session_memory_extraction().await;
            // Sibling pipeline (#1399): heuristic transcript ingestion
            // turns the just-written transcript into durable
            // conversational memory + reflections so a brand-new chat
            // can recover continuity. Background-only, never blocks the
            // user-facing turn return.
            self.spawn_transcript_ingestion();
        }

        result
    }

    async fn inject_agent_experience_context(
        &self,
        user_message: &str,
        enriched: String,
    ) -> String {
        const MAX_EXPERIENCE_HITS: usize = 3;
        const MAX_EXPERIENCE_BLOCK_BYTES: usize = 2048;

        if !self.learning_enabled {
            return enriched;
        }

        let tools = self
            .visible_tool_specs
            .iter()
            .map(|spec| spec.name.clone())
            .collect();
        let store = AgentExperienceStore::new(self.memory.clone());
        let query = ExperienceQuery {
            query: user_message.to_string(),
            tools,
            tags: Vec::new(),
            agent_id: Some(self.agent_definition_id.clone()).filter(|id| !id.trim().is_empty()),
            entrypoint: Some(self.event_channel.clone())
                .filter(|entrypoint| !entrypoint.trim().is_empty()),
            max_hits: MAX_EXPERIENCE_HITS,
        };

        match store.retrieve(query).await {
            Ok(hits) => {
                let matched_hits: Vec<_> = hits
                    .into_iter()
                    .filter(|hit| !hit.match_reasons.is_empty())
                    .collect();
                let block = render_experience_hits(&matched_hits, MAX_EXPERIENCE_BLOCK_BYTES);
                if block.is_empty() {
                    return enriched;
                }
                log::debug!(
                    "[agent-experience] injected {} experience hit(s) bytes={}",
                    matched_hits.len(),
                    block.len()
                );
                prepend_experience_block(&enriched, &block)
            }
            Err(err) => {
                log::warn!("[agent-experience] retrieval failed (non-fatal): {err}");
                enriched
            }
        }
    }

    // ─────────────────────────────────────────────────────────────────
    // Per-call tool execution
    // ─────────────────────────────────────────────────────────────────

    /// Executes a single tool call and returns the result and execution record.
    ///
    /// This method:
    /// 1. Emits telemetry events for the start of execution.
    /// 2. Handles the special `spawn_subagent` tool with `fork` context.
    /// 3. Validates tool visibility and availability.
    /// 4. Dispatches to the underlying tool implementation.
    /// 5. Applies per-result byte budgets to prevent context window bloat.
    /// 6. Sanitizes and records the outcome for post-turn hooks.
    pub(super) async fn execute_tool_call(
        &self,
        call: &ParsedToolCall,
        iteration: usize,
    ) -> (ToolExecutionResult, ToolCallRecord) {
        let started = std::time::Instant::now();
        publish_global(DomainEvent::ToolExecutionStarted {
            tool_name: call.name.clone(),
            session_id: self.event_session_id().to_string(),
        });
        // Synthesise a fallback id for prompt-guided (non-native) tool
        // calls so downstream consumers always have a stable key to
        // reconcile tool_call / tool_args_delta / tool_result rows by.
        // A random uuid guarantees uniqueness even when the same tool
        // name appears multiple times in the same iteration's parsed
        // calls.
        let call_id = call.tool_call_id.clone().unwrap_or_else(|| {
            format!(
                "turn-{iteration}-{}-{}",
                call.name,
                uuid::Uuid::new_v4().simple()
            )
        });
        self.emit_progress(AgentProgress::ToolCallStarted {
            call_id: call_id.clone(),
            tool_name: call.name.clone(),
            arguments: call.arguments.clone(),
            iteration: (iteration + 1) as u32,
        })
        .await;
        log::info!("[agent] executing tool: {}", call.name);
        log::info!("[agent_loop] tool start name={}", call.name);

        let (raw_result, success) = if !self.visible_tool_names.is_empty()
            && !self.visible_tool_names.contains(&call.name)
        {
            log::warn!(
                "[agent] blocked tool call '{}' — not in visible tool set",
                call.name
            );
            (
                format!("Tool '{}' is not available to this agent", call.name),
                false,
            )
        } else if let Some(tool) = self.tools.iter().find(|t| t.name() == call.name) {
            let session_decision = self.tool_policy_session.decision_for(&call.name);
            if session_decision.is_denied() {
                let required = session_decision
                    .required_permission
                    .map(|permission| permission.to_string())
                    .unwrap_or_else(|| "unknown".to_string());
                (
                    format!(
                        "Tool '{}' blocked by tool policy: requires {}, channel '{}' allows {}",
                        call.name,
                        required,
                        self.event_channel,
                        session_decision.allowed_permission
                    ),
                    false,
                )
            } else {
                let context = ToolCallContext::session(
                    self.event_session_id(),
                    self.event_channel(),
                    self.agent_definition_id.to_string(),
                    call_id.clone(),
                    (iteration + 1) as u32,
                );
                let policy_request =
                    ToolPolicyRequest::new(call.name.clone(), call.arguments.clone(), context);
                if let ToolPolicyDecision::Deny { reason } =
                    self.tool_policy.check(&policy_request).await
                {
                    tracing::debug!(
                        tool = call.name.as_str(),
                        policy = self.tool_policy.name(),
                        reason = %reason,
                        "[agent_loop] tool denied by policy"
                    );
                    (
                        format!(
                            "Tool '{}' denied by policy '{}': {reason}",
                            call.name,
                            self.tool_policy.name()
                        ),
                        false,
                    )
                } else {
                    // Per-call options: ask the tool for markdown output when the
                    // context manager is configured to prefer it. Tools that
                    // implement `execute_with_options` will populate
                    // `markdown_formatted`; others fall through to the default
                    // implementation which forwards to `execute`.
                    let prefer_markdown = self.context.prefer_markdown_tool_output();
                    let options = ToolCallOptions { prefer_markdown };
                    let outcome = tool
                        .execute_with_options(call.arguments.clone(), options)
                        .await;
                    match outcome {
                        Ok(r) => {
                            if !r.is_error {
                                let mut output = r.output_for_llm(prefer_markdown);
                                if prefer_markdown && r.markdown_formatted.is_some() {
                                    log::debug!(
                                        "[agent_loop] tool={} returned markdown payload bytes={}",
                                        call.name,
                                        output.len()
                                    );
                                }
                                // Issue #574 — if a payload summarizer is wired
                                // in (orchestrator session only) and the output
                                // exceeds the configured threshold, hand it to
                                // the summarizer sub-agent before it enters
                                // history. On any failure or below-threshold
                                // payload, leave `output` untouched and let the
                                // existing tool_result_budget_bytes truncation
                                // pipeline handle it downstream.
                                if let Some(ps) = self.payload_summarizer.as_ref() {
                                    log::debug!(
                                    "[agent_loop] payload_summarizer intercepting tool={} bytes={}",
                                    call.name,
                                    output.len()
                                );
                                    match ps.maybe_summarize(&call.name, None, &output).await {
                                        Ok(Some(payload)) => {
                                            log::info!(
                                            "[agent_loop] payload_summarizer compressed tool={} {}->{} bytes",
                                            call.name,
                                            payload.original_bytes,
                                            payload.summary_bytes
                                        );
                                            output = payload.summary;
                                        }
                                        Ok(None) => {
                                            log::debug!(
                                            "[agent_loop] payload_summarizer pass-through tool={} bytes={}",
                                            call.name,
                                            output.len()
                                        );
                                        }
                                        Err(e) => {
                                            log::warn!(
                                            "[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)",
                                            call.name,
                                            e
                                        );
                                        }
                                    }
                                }
                                (output, true)
                            } else {
                                (
                                    format!("Error: {}", r.output_for_llm(prefer_markdown)),
                                    false,
                                )
                            }
                        }
                        Err(e) => (format!("Error executing {}: {e}", call.name), false),
                    }
                }
            }
        } else {
            (format!("Unknown tool: {}", call.name), false)
        };

        // Context pipeline stage 1: apply the per-result byte budget
        // *inline* before the result enters history. This is the only
        // cache-safe reduction stage — the truncated body has never
        // been sent to the backend so it creates no cache invalidation.
        // Source the budget from the context manager so it tracks the
        // resolved `context.tool_result_budget_bytes` (including any
        // env/config overrides) rather than the deprecated
        // `agent.tool_result_budget_bytes` field.
        let budget_bytes = self.context.tool_result_budget_bytes();
        let (result, budget_outcome) =
            crate::openhuman::context::apply_tool_result_budget(raw_result, budget_bytes);
        if budget_outcome.truncated {
            log::info!(
                "[agent_loop] tool_result_budget applied name={} original_bytes={} final_bytes={} dropped_bytes={}",
                call.name,
                budget_outcome.original_bytes,
                budget_outcome.final_bytes,
                budget_outcome.original_bytes - budget_outcome.final_bytes
            );
        }

        let elapsed_ms = started.elapsed().as_millis() as u64;
        publish_global(DomainEvent::ToolExecutionCompleted {
            tool_name: call.name.clone(),
            session_id: self.event_session_id().to_string(),
            success,
            elapsed_ms,
        });
        self.emit_progress(AgentProgress::ToolCallCompleted {
            call_id: call_id.clone(),
            tool_name: call.name.clone(),
            success,
            output_chars: result.chars().count(),
            elapsed_ms,
            iteration: (iteration + 1) as u32,
        })
        .await;
        log::info!(
            "[agent] tool completed: {} success={} elapsed_ms={}",
            call.name,
            success,
            elapsed_ms
        );
        log::debug!(
            "[agent] tool output for {}: {}",
            call.name,
            truncate_with_ellipsis(&result, 500)
        );
        log::info!(
            "[agent_loop] tool finish name={} elapsed_ms={} output_chars={} success={}",
            call.name,
            elapsed_ms,
            result.chars().count(),
            success
        );

        let output_summary = hooks::sanitize_tool_output(&result, &call.name, success);

        let record = ToolCallRecord {
            name: call.name.clone(),
            arguments: call.arguments.clone(),
            success,
            output_summary,
            duration_ms: elapsed_ms,
        };

        let exec_result = ToolExecutionResult {
            name: call.name.clone(),
            output: result,
            success,
            tool_call_id: call.tool_call_id.clone(),
        };

        (exec_result, record)
    }

    /// Executes multiple tool calls in sequence.
    ///
    /// Collects results and execution records for all requested tools in a single batch.
    pub(super) async fn execute_tools(
        &self,
        calls: &[ParsedToolCall],
        iteration: usize,
    ) -> (Vec<ToolExecutionResult>, Vec<ToolCallRecord>) {
        let mut results = Vec::with_capacity(calls.len());
        let mut records = Vec::with_capacity(calls.len());
        for call in calls {
            let (exec_result, record) = self.execute_tool_call(call, iteration).await;
            results.push(exec_result);
            records.push(record);
        }
        (results, records)
    }

    // ─────────────────────────────────────────────────────────────────
    // Sub-agent context snapshots
    // ─────────────────────────────────────────────────────────────────

    /// Snapshot the parent's runtime so spawned sub-agents can read
    /// it via the [`harness::PARENT_CONTEXT`] task-local.
    pub(super) fn build_parent_execution_context(&self) -> harness::ParentExecutionContext {
        harness::ParentExecutionContext {
            provider: Arc::clone(&self.provider),
            all_tools: Arc::clone(&self.tools),
            all_tool_specs: Arc::clone(&self.tool_specs),
            model_name: self.model_name.clone(),
            temperature: self.temperature,
            workspace_dir: self.workspace_dir.clone(),
            memory: Arc::clone(&self.memory),
            agent_config: self.config.clone(),
            skills: Arc::new(self.skills.clone()),
            memory_context: Arc::new(self.last_memory_context.clone()),
            session_id: self.event_session_id().to_string(),
            channel: self.event_channel().to_string(),
            connected_integrations: self.connected_integrations.clone(),
            tool_call_format: self.tool_dispatcher.tool_call_format(),
            session_key: self.session_key.clone(),
            session_parent_prefix: self.session_parent_prefix.clone(),
            on_progress: self.on_progress.clone(),
        }
    }

    // ─────────────────────────────────────────────────────────────────
    // History & prompt helpers
    // ─────────────────────────────────────────────────────────────────

    /// Emit a lifecycle progress event. Uses `send().await` so control
    /// events (turn/iteration boundaries, tool_call_started/completed,
    /// turn_completed) survive downstream backpressure from the
    /// higher-frequency streamed deltas that share the same `on_progress`
    /// channel — dropping one of these would desync the web-channel
    /// progress bridge (e.g. a tool row stuck in `running` forever).
    /// A closed sink is logged and ignored; no progress subscriber is
    /// equivalent to success.
    async fn emit_progress(&self, event: AgentProgress) {
        if let Some(ref tx) = self.on_progress {
            if let Err(e) = tx.send(event).await {
                log::warn!("[agent] progress sink closed while emitting lifecycle event: {e}");
            }
        }
    }

    /// Truncates the conversation history to the configured maximum message count.
    ///
    /// System messages are always preserved. Older non-system messages are
    /// dropped first.
    pub(super) fn trim_history(&mut self) {
        let max = self.config.max_history_messages;
        if self.history.len() <= max {
            return;
        }

        let mut system_messages = Vec::new();
        let mut other_messages = Vec::new();

        for msg in self.history.drain(..) {
            match &msg {
                ConversationMessage::Chat(chat) if chat.role == "system" => {
                    system_messages.push(msg);
                }
                _ => other_messages.push(msg),
            }
        }

        if other_messages.len() > max {
            let drop_count = other_messages.len() - max;
            other_messages.drain(0..drop_count);
        }

        self.history = system_messages;
        self.history.extend(other_messages);
    }

    /// Bound a resumed transcript prefix to the agent history window.
    ///
    /// Resume paths may load a long prior transcript directly into
    /// `cached_transcript_messages` (provider-ready `ChatMessage`s), which
    /// bypasses `self.history`-based trimming/reduction. Keep at most
    /// `max_history_messages` entries while preserving the leading system
    /// message when present.
    pub(super) fn bound_cached_transcript_messages(
        &self,
        messages: Vec<ChatMessage>,
    ) -> Vec<ChatMessage> {
        let max = self.config.max_history_messages.max(1);
        if messages.len() <= max {
            return messages;
        }

        if matches!(messages.first(), Some(msg) if msg.role == "system") {
            let keep_tail = max.saturating_sub(1);
            let start = messages.len().saturating_sub(keep_tail);
            let mut bounded = Vec::with_capacity(max);
            bounded.push(messages[0].clone());
            if keep_tail > 0 {
                bounded.extend(messages[start..].iter().cloned());
            }
            bounded
        } else {
            let start = messages.len().saturating_sub(max);
            messages[start..].to_vec()
        }
    }

    /// Pre-fetches learned context data from memory (observations, patterns, user profile).
    ///
    /// This is an async, non-blocking operation that populates the context
    /// for the system prompt.
    ///
    /// # Explicit-preferences narrow path
    ///
    /// When `learning_enabled` is `false` but `explicit_preferences_enabled`
    /// is `true`, only the `user_profile` namespace (pinned preferences from
    /// the `remember_preference` tool) is fetched and returned.  All other
    /// inference-derived data (observations, patterns, reflections, tree
    /// summaries) remains empty — the inference stack is not touched.
    pub(super) async fn fetch_learned_context(&self) -> LearnedContextData {
        // Fast path: neither the full learning subsystem nor the explicit
        // preferences path is active — skip all memory reads.
        if !self.learning_enabled && !self.explicit_preferences_enabled {
            tracing::debug!(
                "[learning] fetch_learned_context: both learning_enabled and \
                 explicit_preferences_enabled are false — returning empty context"
            );
            return LearnedContextData::default();
        }

        // Narrow explicit-preferences path (Lane A): inject the latest-N general
        // (always-on) preferences written via `save_preference`. Topic-scoped
        // (situational) prefs are NOT injected here — they ride the user message
        // via per-turn recall (Lane B). The legacy `user_profile` pinned namespace
        // is no longer read here; explicit prefs now live in `user_pref_general`.
        if !self.learning_enabled && self.explicit_preferences_enabled {
            let general = crate::openhuman::memory::preferences::load_general_preferences(
                &self.memory,
                crate::openhuman::memory::preferences::STANDING_PREFS_LIMIT,
            )
            .await;
            tracing::debug!(
                "[learning] fetch_learned_context: explicit_preferences_enabled — loaded {} general preference(s) for the system prompt",
                general.len()
            );
            return LearnedContextData {
                user_profile: general,
                ..LearnedContextData::default()
            };
        }

        // Full learning path: fetch all inference-derived data.
        tracing::debug!(
            "[learning] fetch_learned_context: learning_enabled=true — fetching full context"
        );

        let obs_entries = self
            .memory
            .list(
                Some("learning_observations"),
                Some(&MemoryCategory::Custom("learning_observations".into())),
                None,
            )
            .await
            .unwrap_or_default();

        let pat_entries = self
            .memory
            .list(
                Some("learning_patterns"),
                Some(&MemoryCategory::Custom("learning_patterns".into())),
                None,
            )
            .await
            .unwrap_or_default();

        // Standing preferences come from the explicit two-lane store (Lane A),
        // not the inferred `user_profile` facets — those are demoted: no longer
        // injected as ground truth. A high-confidence inferred facet should be
        // *proposed* to the user (and pinned via `save_preference` on
        // confirmation), not silently treated as a standing preference.
        let general = crate::openhuman::memory::preferences::load_general_preferences(
            &self.memory,
            crate::openhuman::memory::preferences::STANDING_PREFS_LIMIT,
        )
        .await;

        // Explicit user reflections — privileged memory class. Pulled
        // separately from observations/patterns so the prompt assembly
        // can render them ahead of generic tree summaries.
        let reflection_entries = self
            .memory
            .list(
                Some(crate::openhuman::learning::reflection::REFLECTIONS_NAMESPACE),
                Some(&MemoryCategory::Custom(
                    crate::openhuman::learning::reflection::REFLECTIONS_NAMESPACE.into(),
                )),
                None,
            )
            .await
            .unwrap_or_default();

        // Pull every namespace's root-level summary from the tree
        // summarizer. This is the densest user memory we can hand the
        // orchestrator: each root holds up to 20 000 tokens of distilled
        // long-term context. Done synchronously here because the calls
        // are filesystem reads, not provider/network round-trips, and
        // happen exactly once per session (only on the first turn).
        //
        // Per-namespace + total caps come from the user-facing memory
        // window preset on `AgentConfig` so changing the slider in the
        // UI takes effect on the very next session-start.
        let limits = self.config.resolved_memory_limits();
        let tree_root_summaries = collect_tree_root_summaries(
            &self.workspace_dir,
            limits.per_namespace_max_chars,
            limits.total_tree_max_chars,
        );

        LearnedContextData {
            observations: obs_entries
                .iter()
                .rev()
                .take(5)
                .map(|e| sanitize_learned_entry(&e.content))
                .collect(),
            patterns: pat_entries
                .iter()
                .take(3)
                .map(|e| sanitize_learned_entry(&e.content))
                .collect(),
            user_profile: general,
            // Cap reflections at 10 to keep the privileged section
            // bounded — the issue requires reflections improve context
            // rather than flood it. Newest first.
            reflections: reflection_entries
                .iter()
                .rev()
                .take(10)
                .map(|e| sanitize_learned_entry(&e.content))
                .collect(),
            tree_root_summaries,
        }
    }

    /// Fetches the user's active Composio connections and populates
    /// `self.connected_integrations` so the system prompt can surface them.
    ///
    /// Delegates to the shared [`crate::openhuman::composio::fetch_connected_integrations`]
    /// which is the single source of truth for integration discovery.
    ///
    /// **No session-scoped Composio client is cached on the agent any
    /// more (#1710 Wave 2)**. Every downstream caller that needs to
    /// dispatch a Composio action now resolves a fresh client via
    /// [`crate::openhuman::composio::client::create_composio_client`]
    /// at call time so the live `composio.mode` toggle is honoured
    /// without rebuilding the session — see `ComposioActionTool`,
    /// `ProviderContext::execute`, the 5 migrated agent tools in
    /// `composio/tools.rs`, and the spawn-time per-action tool build
    /// path in `subagent_runner/ops.rs`.
    pub async fn fetch_connected_integrations(&mut self) {
        let config = match self.integration_runtime_config.clone() {
            Some(config) => config,
            None => match crate::openhuman::config::Config::load_or_init().await {
                Ok(config) => config,
                Err(e) => {
                    log::debug!(
                        "[agent] skipping connected integrations fetch: config load failed: {e}"
                    );
                    return;
                }
            },
        };
        self.connected_integrations =
            crate::openhuman::composio::fetch_connected_integrations(&config).await;
        self.connected_integrations_initialized = true;
    }

    /// Re-synthesise `delegate_*` tools for the orchestrator's `subagents`
    /// declaration using the live `connected_integrations` slice, and
    /// reconcile the resulting set into `self.tools` / `self.tool_specs` /
    /// `self.visible_tool_specs` / `self.visible_tool_names`.
    ///
    /// **Reconciliation strategy** — full rebuild of the synthesised
    /// subset:
    ///
    ///   1. Drop every tool whose name was in [`Self::synthesized_tool_names`]
    ///      from the previous synthesis. Direct tools (`query_memory`,
    ///      `cron_add`, …) are untouched because their names are not in
    ///      that set.
    ///   2. Append the freshly collected synthesis output verbatim.
    ///   3. Replace `synthesized_tool_names` with the new set so the
    ///      next refresh has a clean mask to undo.
    ///
    /// This is safer than appending-only or strict-diff reconcile:
    ///
    ///   * Stale tools after a revoke can never leak — anything from the
    ///     previous synthesis is unconditionally dropped, the new set is
    ///     authoritative.
    ///   * Direct tools can never be accidentally removed — only names
    ///     in `synthesized_tool_names` are touched.
    ///   * Duplicate registration is impossible — retain+extend
    ///     guarantees every final entry is either a non-synthesised
    ///     direct tool or a member of the fresh `synthed` set.
    ///
    /// **When to call**: on turn 1 only when the session was built
    /// without a prewarmed Composio cache snapshot, and on any
    /// subsequent turn where the connection set has changed since the
    /// last reconcile (detected via
    /// [`Self::last_seen_integrations_hash`] vs.
    /// [`crate::openhuman::composio::cached_active_integrations`]).
    ///
    /// **Atomicity**: when `Arc::get_mut` fails (a sub-agent or other
    /// caller has already captured a clone of the tool list), we restore
    /// the previous `synthesized_tool_names` and bail. The next refresh
    /// attempt will re-apply the full transition cleanly rather than
    /// resuming from a partial state. This should never happen on a turn
    /// boundary in production — sub-agents always drop their snapshots
    /// before the parent's next turn — but it's defended against anyway.
    ///
    /// **Return value** — `true` when the agent's tool surface is now
    /// consistent with `self.connected_integrations` (either because a
    /// successful reconcile applied, or because no reconcile was needed).
    /// `false` only when the Arc was shared and the reconcile was
    /// aborted; callers should treat this as "retry next turn" and
    /// **not** advance any signal they use to gate future refreshes
    /// (e.g. `last_seen_integrations_hash`) — otherwise a one-shot
    /// shared-Arc collision could suppress further reconciliation until
    /// another integration event happened to bump the hash again.
    pub fn refresh_delegation_tools(&mut self) -> bool {
        use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
        use crate::openhuman::tools::orchestrator_tools::collect_orchestrator_tools;

        let Some(reg) = AgentDefinitionRegistry::global() else {
            // No registry — there's nothing we can do until the
            // registry is initialised. The agent's surface stays at
            // whatever the builder produced; callers can safely treat
            // this as "no reconcile needed right now".
            return true;
        };
        let Some(def) = reg.get(&self.agent_definition_id) else {
            log::debug!(
                "[agent] refresh_delegation_tools: definition '{}' not in registry — skipping",
                self.agent_definition_id
            );
            return true;
        };
        if def.subagents.is_empty() {
            return true;
        }

        let synthed = collect_orchestrator_tools(def, reg, &self.connected_integrations);
        let synthed_names: std::collections::HashSet<String> =
            synthed.iter().map(|t| t.name().to_string()).collect();
        let synthed_specs: Vec<crate::openhuman::tools::ToolSpec> =
            synthed.iter().map(|t| t.spec()).collect();

        // Skip the Arc mutation entirely when neither the previous nor
        // the next synthesis produced any names — saves an
        // `Arc::get_mut` probe on agents that don't declare `subagents`.
        if self.synthesized_tool_names.is_empty() && synthed_names.is_empty() {
            return true;
        }

        // Mask used to drop the previous synthesis. We `take` it now so
        // the restore-on-failure path below can put it back if the Arc
        // turns out to be shared.
        let old_synth = std::mem::take(&mut self.synthesized_tool_names);

        match (
            Arc::get_mut(&mut self.tools),
            Arc::get_mut(&mut self.tool_specs),
        ) {
            (Some(tools_vec), Some(specs_vec)) => {
                tools_vec.retain(|t| !old_synth.contains(t.name()));
                specs_vec.retain(|s| !old_synth.contains(&s.name));
                tools_vec.extend(synthed);
                specs_vec.extend(synthed_specs);
            }
            _ => {
                log::warn!(
                    "[agent] refresh_delegation_tools: tools/tool_specs Arc is shared — \
                     cannot reconcile delegation surface (would have produced {} synthesised tool(s)). \
                     Restoring previous synthesized_tool_names so the next refresh retries cleanly.",
                    synthed_names.len()
                );
                self.synthesized_tool_names = old_synth;
                return false;
            }
        }

        // `visible_tool_names` carries an explicit allowlist for
        // [`ToolScope::Named`] agents. Drop the previously-synthesised
        // names and add the new ones so the visible set tracks the
        // tool list. Wildcard-scope agents keep this empty ("no
        // filter") and never need touching.
        if !self.visible_tool_names.is_empty() {
            for name in &old_synth {
                self.visible_tool_names.remove(name);
            }
            for name in &synthed_names {
                self.visible_tool_names.insert(name.clone());
            }
        }

        // Rebuild the visible-spec cache from the new tool_specs so the
        // next provider call carries the reconciled schema. Dedup
        // afterward so a delegate synthesised here (e.g.
        // `delegate_name = "research"`) doesn't collide with a
        // same-named skill tool on the wire — Anthropic 400s on dup
        // tool names where OpenHuman's backend silently accepts.
        self.rebuild_tool_policy_session();

        // Compute add/remove deltas for the log line — useful when
        // diagnosing a Composio connect/revoke that should have rebuilt
        // the surface but didn't. Materialise to owned `Vec<String>`
        // so we can move `synthed_names` into `self.synthesized_tool_names`
        // below without the log-statement reborrow blocking the move.
        let added: Vec<String> = synthed_names
            .iter()
            .filter(|n| !old_synth.contains(n.as_str()))
            .cloned()
            .collect();
        let removed: Vec<String> = old_synth
            .iter()
            .filter(|n| !synthed_names.contains(n.as_str()))
            .cloned()
            .collect();

        self.synthesized_tool_names = synthed_names;

        log::info!(
            "[agent] refresh_delegation_tools: reconciled delegation surface for agent '{}' (display='{}'); now {} synthesised tool(s); added={:?} removed={:?}",
            self.agent_definition_id,
            self.agent_definition_name,
            self.synthesized_tool_names.len(),
            added,
            removed
        );
        true
    }

    /// Builds the system prompt for the current turn, including tool
    /// instructions and learned context.
    pub fn build_system_prompt(&self, learned: LearnedContextData) -> Result<String> {
        let tools_slice: &[Box<dyn Tool>] = self.tools.as_slice();
        let instructions = self
            .tool_dispatcher
            .prompt_instructions_for_specs(self.visible_tool_specs.as_slice())
            .unwrap_or_else(|| self.tool_dispatcher.prompt_instructions(tools_slice));
        // Adapt the owned Box<dyn Tool> slice into the shared PromptTool
        // shape that every prompt-building call-site uses. Temporary vec
        // borrows from `tools_slice` and lives for the duration of the
        // prompt build.
        let prompt_tools = PromptTool::from_tools(tools_slice);
        let prompt_visible_tool_names = self.tool_policy_session.visible_tool_names_for_prompt();
        let ctx = PromptContext {
            workspace_dir: &self.workspace_dir,
            model_name: &self.model_name,
            agent_id: &self.agent_definition_name,
            tools: &prompt_tools,
            skills: &self.skills,
            dispatcher_instructions: &instructions,
            learned,
            visible_tool_names: &prompt_visible_tool_names,
            tool_call_format: self.tool_dispatcher.tool_call_format(),
            connected_integrations: &self.connected_integrations,
            connected_identities_md: crate::openhuman::agent::prompts::render_connected_identities(
            ),
            include_profile: !self.omit_profile,
            include_memory_md: !self.omit_memory_md,
            curated_snapshot: None,
            user_identity: crate::openhuman::app_state::peek_cached_current_user_identity(),
        };
        // Route through the global context manager so every
        // prompt-building call-site — main agent, sub-agent runner,
        // channel runtimes — shares one builder configuration.
        let mut prompt = self.context.build_system_prompt(&ctx)?;
        if let Some(boundary) = render_tool_policy_boundary(&self.tool_policy_session, 2048) {
            prompt = format!("{boundary}\n\n{prompt}");
        }
        Ok(prompt)
    }

    // ─────────────────────────────────────────────────────────────────
    // Session transcript helpers
    // ─────────────────────────────────────────────────────────────────

    /// Try to load a previous session transcript for KV cache resume.
    ///
    /// Best-effort: failures are logged and silently ignored.
    pub(super) fn try_load_session_transcript(&mut self) {
        match transcript::find_latest_transcript(&self.workspace_dir, &self.agent_definition_name) {
            Some(path) => {
                log::info!(
                    "[transcript] found previous transcript path={}",
                    path.display()
                );
                match transcript::read_transcript(&path) {
                    Ok(session) => {
                        if session.messages.is_empty() {
                            log::debug!(
                                "[transcript] previous transcript is empty — skipping resume"
                            );
                            return;
                        }
                        let loaded_count = session.messages.len();
                        log::info!("[transcript] loaded {} messages for resume", loaded_count);
                        let bounded = self.bound_cached_transcript_messages(session.messages);
                        if bounded.len() < loaded_count {
                            log::warn!(
                                "[transcript] resume prefix trimmed from {} to {} messages (max_history_messages={})",
                                loaded_count,
                                bounded.len(),
                                self.config.max_history_messages
                            );
                        }
                        self.cached_transcript_messages = Some(bounded);
                    }
                    Err(err) => {
                        log::warn!(
                            "[transcript] failed to parse previous transcript {}: {err}",
                            path.display()
                        );
                    }
                }
            }
            None => {
                log::debug!(
                    "[transcript] no previous transcript found for agent={}",
                    self.agent_definition_name
                );
            }
        }
    }

    /// Persist the exact provider messages as a session transcript.
    ///
    /// Writes JSONL as source of truth and re-renders the companion `.md`
    /// for human readability. Best-effort: failures are logged and silently
    /// ignored. The JSONL conversation store remains the authoritative
    /// persistence layer; session transcripts are an optimization for KV
    /// cache stability.
    ///
    /// `turn_usage` — when `Some`, attributes per-message token/cost figures
    /// to the last assistant message in the written transcript.
    pub(super) fn persist_session_transcript(
        &mut self,
        messages: &[ChatMessage],
        input_tokens: u64,
        output_tokens: u64,
        cached_input_tokens: u64,
        charged_amount_usd: f64,
        turn_usage: Option<&transcript::TurnUsage>,
    ) {
        // Resolve the transcript path on first write. The stem is
        // `{parent_prefix}__{session_key}` for sub-agents (producing a
        // flat hierarchical filename) or just `{session_key}` for a
        // root session. Prefix chaining is already done by the
        // sub-agent runner when it populates `session_parent_prefix`.
        if self.session_transcript_path.is_none() {
            let stem = match &self.session_parent_prefix {
                Some(prefix) => format!("{}__{}", prefix, self.session_key),
                None => self.session_key.clone(),
            };
            match transcript::resolve_keyed_transcript_path(&self.workspace_dir, &stem) {
                Ok(path) => {
                    log::info!(
                        "[transcript] new session transcript path={}",
                        path.display()
                    );
                    self.session_transcript_path = Some(path);
                }
                Err(err) => {
                    log::warn!("[transcript] failed to resolve transcript path: {err}");
                    return;
                }
            }
        }

        let path = self.session_transcript_path.as_ref().unwrap();
        let now = chrono::Utc::now().to_rfc3339();

        let meta = transcript::TranscriptMeta {
            agent_name: self.agent_definition_name.clone(),
            dispatcher: if self.tool_dispatcher.should_send_tool_specs() {
                "native".into()
            } else {
                "xml".into()
            },
            created: now.clone(),
            updated: now,
            turn_count: self.context.stats().session_memory_current_turn as usize,
            input_tokens,
            output_tokens,
            cached_input_tokens,
            charged_amount_usd,
            thread_id: crate::openhuman::inference::provider::thread_context::current_thread_id(),
        };

        if let Err(err) = transcript::write_transcript(path, messages, &meta, turn_usage) {
            log::warn!(
                "[transcript] failed to write transcript {}: {err}",
                path.display()
            );
        }
    }

    // ─────────────────────────────────────────────────────────────────
    // Session-memory extraction (stage 5 of the context pipeline)
    // ─────────────────────────────────────────────────────────────────

    /// Spawn a background archivist sub-agent to extract durable facts
    /// from the recent conversation into `MEMORY.md`. Fire-and-forget.
    ///
    /// Gated by [`context_pipeline::SessionMemoryState::should_extract`]
    /// — see its docs for the threshold invariants. Safe to call from
    /// inside `turn()` after the turn body has settled.
    pub(super) async fn spawn_session_memory_extraction(&mut self) {
        // ── Flush the trailing open segment before the session winds down ──
        //
        // The ArchivistHook manages per-turn segment lifecycle but cannot
        // force-close the *last* open segment because there is no explicit
        // "session end" event in the turn loop. `spawn_session_memory_extraction`
        // is the closest available signal: it fires when the context manager
        // decides the session has accumulated enough material to archive.
        //
        // GUARANTEE: the flush is *awaited* here (not fire-and-forget) so
        // the trailing segment always receives its recap + embedding + tree
        // ingest before the function returns, even during runtime wind-down.
        // This honours the doc-comment guarantee on `flush_open_segment` in
        // `archivist.rs`. No deadlock risk: no mutex guard is held across
        // this await point.
        if let Some(ref archivist) = self.archivist_hook {
            let session_id = self.event_session_id.clone();
            log::debug!(
                "[archivist] awaiting flush_open_segment for session={session_id} at session wind-down"
            );
            archivist.flush_open_segment(&session_id).await;
        }

        let Some(registry) = harness::AgentDefinitionRegistry::global() else {
            log::debug!("[session_memory] registry not initialised — skipping extraction spawn");
            return;
        };
        let Some(definition) = registry.get("archivist").cloned() else {
            log::debug!(
                "[session_memory] archivist definition not found — skipping extraction spawn"
            );
            return;
        };

        // Build a dedicated ParentExecutionContext for the background
        // task. The in-progress turn's context has already been
        // consumed by the `with_parent_context` scope above, so this is
        // a fresh snapshot.
        let parent_ctx = self.build_parent_execution_context();
        let extraction_prompt = ARCHIVIST_EXTRACTION_PROMPT.to_string();

        // Flip the extraction state to "in-progress" so future
        // should_extract checks return false until the archivist
        // finishes. We then hand a shared handle to the spawned task
        // so it can mark the extraction complete (resets deltas) on
        // success, or failed (keeps deltas intact for retry) on error.
        // This replaces the old optimistic `mark_complete` that
        // silently dropped the retry window when extractions failed.
        let stats_snapshot = self.context.stats();
        self.context.mark_session_memory_started();
        let sm_handle = self.context.session_memory_handle();

        log::info!(
            "[session_memory] spawning background archivist extraction (turn={}, tokens={})",
            stats_snapshot.session_memory_current_turn,
            stats_snapshot.session_memory_total_tokens
        );

        tokio::spawn(async move {
            let options = harness::SubagentRunOptions::default();
            let fut = harness::run_subagent(&definition, &extraction_prompt, options);
            let result = harness::with_parent_context(parent_ctx, fut).await;
            match result {
                Ok(outcome) => {
                    tracing::info!(
                        agent_id = %outcome.agent_id,
                        task_id = %outcome.task_id,
                        iterations = outcome.iterations,
                        output_chars = outcome.output.chars().count(),
                        "[session_memory] archivist extraction completed"
                    );
                    if let Ok(mut sm) = sm_handle.lock() {
                        sm.mark_extraction_complete();
                    }
                }
                Err(err) => {
                    tracing::warn!(
                        error = %err,
                        "[session_memory] archivist extraction failed — will retry after next threshold crossing"
                    );
                    // Leave the deltas intact so the next threshold
                    // crossing schedules another attempt. Clearing
                    // `extraction_in_progress` lets the retry
                    // actually fire.
                    if let Ok(mut sm) = sm_handle.lock() {
                        sm.mark_extraction_failed();
                    }
                }
            }
        });
    }

    /// Spawn a background task that ingests the current session
    /// transcript into the conversational-memory store.
    ///
    /// Issue #1399: complements `spawn_session_memory_extraction`. The
    /// archivist path writes dense bullets into `MEMORY.md`; this path
    /// extracts importance-tagged, provenance-bearing memories via the
    /// heuristic [`crate::openhuman::learning::transcript_ingest`]
    /// pipeline. The two are deliberately independent so the prompt
    /// retrieval layer can pull from `conversation_memory` without
    /// needing the archivist's extraction to have fired this session.
    ///
    /// Fire-and-forget: failures are logged, never propagated.
    pub(super) fn spawn_transcript_ingestion(&self) {
        let Some(path) = self.session_transcript_path.clone() else {
            log::debug!("[transcript_ingest] no session transcript path yet — skipping spawn");
            return;
        };
        let memory = std::sync::Arc::clone(&self.memory);

        tokio::spawn(async move {
            match crate::openhuman::learning::transcript_ingest::ingest_transcript_path(
                memory.as_ref(),
                &path,
            )
            .await
            {
                Ok(report) => tracing::info!(
                    transcript = %path.display(),
                    extracted = report.extracted,
                    stored = report.stored,
                    deduped = report.deduped,
                    reflections_stored = report.reflections_stored,
                    "[transcript_ingest] background ingest complete"
                ),
                Err(err) => tracing::warn!(
                    transcript = %path.display(),
                    error = %err,
                    "[transcript_ingest] background ingest failed — will retry next threshold window"
                ),
            }
        });
    }
}

/// Wrapper around
/// [`crate::openhuman::tree_summarizer::store::collect_root_summaries_with_caps`]
/// that takes user-resolved per-namespace and total caps. The actual
/// limits are derived from the active
/// [`crate::openhuman::config::schema::agent::MemoryContextWindow`]
/// preset by [`crate::openhuman::config::schema::agent::AgentConfig::resolved_memory_limits`].
fn collect_tree_root_summaries(
    workspace_dir: &std::path::Path,
    per_namespace_cap: usize,
    total_cap: usize,
) -> Vec<(String, String)> {
    crate::openhuman::tree_summarizer::store::collect_root_summaries_with_caps(
        workspace_dir,
        per_namespace_cap,
        total_cap,
    )
}

/// Sanitize a learned memory entry before injecting into the system prompt.
/// Strips raw data, limits length, and removes potential secrets.
fn sanitize_learned_entry(content: &str) -> String {
    let trimmed = content.trim();
    if trimmed.is_empty() {
        return String::new();
    }
    // Truncate to a safe length
    let max_len = 200;
    let sanitized: String = trimmed.chars().take(max_len).collect();
    // Strip anything that looks like a secret/token
    if sanitized.contains("Bearer ")
        || sanitized.contains("sk-")
        || sanitized.contains("ghp_")
        || sanitized.contains("-----BEGIN")
    {
        return "[redacted: potential secret]".to_string();
    }
    sanitized
}

#[cfg(test)]
#[path = "turn_tests.rs"]
mod tests;
