JCode

    4  impl Agent {
    5      /// Run a single turn with the given user message
    6      pub async fn run_once(&mut self, user_message: &str) -> Result<()> {
    7          self.add_message(
    8              Role::User,
    9              vec![ContentBlock::Text {
   10                  text: user_message.to_string(),
   11                  cache_control: None,
   12              }],
   13          );
   14          self.session.save()?;
   15          if trace_enabled() {
   16              eprintln!("[trace] session_id {}", self.session.id);
   17          }
   18          let _ = self.run_turn(true).await?;
   19          Ok(())
   20      }
   21  
   22      pub async fn run_once_capture(&mut self, user_message: &str) -> Result<String> {
   23          self.add_message(
   24              Role::User,
   25              vec![ContentBlock::Text {
   26                  text: user_message.to_string(),
   27                  cache_control: None,
   28              }],
   29          );
   30          self.session.save()?;
   31          if trace_enabled() {
   32              eprintln!("[trace] session_id {}", self.session.id);
   33          }
   34          self.run_turn(false).await
   35      }

   17      pub(super) async fn run_turn(&mut self, print_output: bool) -> Result<String> {
   18          self.set_log_context();
   19          crate::session_metrics::record_turn(&self.session.id);
   20          // Mark this session as actively streaming for presence UIs (e.g. the
   21          // macOS menu bar indicator). Cleared automatically on every exit path.
   22          let _streaming_guard = crate::session::StreamingGuard::new(self.session.id.clone());
   23          // Register this turn's cancel signal so session-level cancels reach
   24          // this in-flight turn even through stale control handles (issue #428).
   25          let _turn_cancel_guard = crate::turn_cancel_registry::register_active_turn(
   26              &self.session.id,
   27              self.graceful_shutdown.clone(),
   28          );
   29          let mut final_text = String::new();
   30          let trace = trace_enabled();
   31          let mut context_limit_retries = 0u32;
   32          let mut incomplete_continuations = 0u32;
   33          let mut empty_post_tool_continuations = 0u32;
   34  
   35          loop {
   36              let repaired = self.repair_missing_tool_outputs();
   37              if repaired > 0 {
   38                  logging::warn(&format!(
   39                      "Recovered {} missing tool output(s) before API call",
   40                      repaired
      … 18 lines omitted; exact range 17–68 …
   59              }
   60  
   61              let tools = self.tool_definitions().await;
   62              let messages: std::sync::Arc<[Message]> = messages.into();
   63              // Non-blocking memory: uses pending result from last turn, spawns check for next turn
   64              let memory_pending =
   65                  self.build_memory_prompt_nonblocking_shared(std::sync::Arc::clone(&messages), None);
   66              // Use split prompt for better caching - static content cached, dynamic not
   67              let split_prompt = self.build_system_prompt_split(None);
   68              self.log_prompt_prefix_accounting(&split_prompt, &tools);

  455                      StreamEvent::RetryRollback { attempt, max } => {
  456                          // Transient transport fault mid-stream; the provider is
  457                          // replaying the request. Discard this attempt's partial
  458                          // output so the replay doesn't duplicate it in history.
  459                          logging::warn(&format!(
  460                              "Mid-stream retry rollback (attempt {}/{}): discarding partial output ({} text chars, {} tool calls)",
  461                              attempt,
  462                              max,
  463                              text_content.len(),
  464                              tool_calls.len(),
  465                          ));
  466                          if print_output && !text_content.is_empty() {
  467                              // Already-printed text can't be unprinted on a plain
  468                              // stdout stream; mark the discontinuity instead.
  469                              println!("\n[connection interrupted, retrying response from the top]");
  470                              io::stdout().flush()?;
  471                          }
  472                          text_content.clear();
  473                          tool_calls.clear();
  474                          current_tool = None;
  475                          current_tool_input.clear();
  476                          sdk_tool_results.clear();
  477                          generated_image_contexts.clear();
  478                          reasoning_content.clear();
  479                          reasoning_signature.clear();
  480                          openai_reasoning_items.clear();
  481                          openai_native_compaction = None;
  482                          saw_message_end = false;
  483                          stop_reason = None;
  484                      }

    5      /// Run turns until no more tool calls
    6      /// Maximum number of context-limit compaction retries before giving up.
    7      pub(super) const MAX_CONTEXT_LIMIT_RETRIES: u32 = 5;
    8      pub(super) const MAX_INCOMPLETE_CONTINUATION_ATTEMPTS: u32 = 3;
    9      /// Retries allowed when the provider returns an empty response right after
   10      /// tool results. This is a transient provider hiccup, not a signal that the
   11      /// task is finished, so a single retry is too few: one empty response
   12      /// observed once in 43 turns silently ended a 20-hour benchmark run with the
   13      /// task half-done. The counter is per turn-loop, so a genuinely finished
   14      /// agent still exits promptly.
   15      pub(crate) const MAX_EMPTY_POST_TOOL_CONTINUATION_ATTEMPTS: u32 = 5;

   76  pub trait Provider: Send + Sync {
   77      /// Send messages and get a streaming response.
   78      /// resume_session_id: Optional session ID to resume a previous conversation (provider-specific).
   79      async fn complete(
   80          &self,
   81          messages: &[Message],
   82          tools: &[ToolDefinition],
   83          system: &str,
   84          resume_session_id: Option<&str>,
   85      ) -> Result<EventStream>;
   86  
   87      /// Send messages with split system prompt for better caching.
   88      async fn complete_split(
   89          &self,
   90          messages: &[Message],
   91          tools: &[ToolDefinition],
   92          system_static: &str,
   93          system_dynamic: &str,
   94          resume_session_id: Option<&str>,
   95      ) -> Result<EventStream> {
   96          let dynamic_messages = messages_with_dynamic_system_context(messages, system_dynamic);
   97          self.complete(&dynamic_messages, tools, system_static, resume_session_id)
   98              .await
   99      }
      … 17 lines omitted; exact range 76–126 …
  117      fn display_name(&self) -> String {
  118          self.name().to_string()
  119      }
  120  
  121      /// Get the model identifier being used.
  122      fn model(&self) -> String {
  123          "unknown".to_string()
  124      }
  125  
  126      /// Human-readable description of the auth method the active provider will

  328  /// MultiProvider wraps multiple providers and allows seamless model switching
  329  pub struct MultiProvider {
  330      /// Claude Code CLI provider
  331      claude: RwLock<Option<Arc<dyn Provider>>>,
  332      /// Direct Anthropic API provider (no Python dependency)
  333      anthropic: RwLock<Option<Arc<dyn Provider>>>,
  334      openai: RwLock<Option<Arc<dyn Provider>>>,
  335      /// GitHub Copilot API provider (direct API, hot-swappable after login).
  336      /// Held as `dyn Provider`: the concrete runtime lives downstream in
  337      /// `jcode-provider-copilot-runtime` and is instantiated through
  338      /// `external::instantiate_external_provider`.
  339      copilot_api: RwLock<Option<Arc<dyn Provider>>>,
  340      /// Antigravity provider (direct HTTPS, hot-swappable after login). Held as
  341      /// `dyn Provider`: the concrete runtime lives downstream in
  342      /// `jcode-provider-antigravity-runtime` and is instantiated through
  343      /// `external::instantiate_external_provider`.
  344      antigravity: RwLock<Option<Arc<dyn Provider>>>,
  345      /// Gemini provider (hot-swappable after login). Held as `dyn Provider`:
  346      /// the concrete runtime lives downstream in `jcode-provider-gemini-runtime`
  347      /// and is instantiated through `external::instantiate_external_provider`.
  348      gemini: RwLock<Option<Arc<dyn Provider>>>,
  349      /// Cursor provider (native/direct API, hot-swappable after login). Held as
  350      /// `dyn Provider`: the concrete runtime lives downstream in
  351      /// `jcode-provider-cursor-runtime` and is instantiated through
      … 13 lines omitted; exact range 328–374 …
  365      active_openai_compatible_profile: RwLock<Option<String>>,
  366      active: RwLock<ActiveProvider>,
  367      /// Use Claude CLI instead of direct API (legacy mode)
  368      use_claude_cli: bool,
  369      /// Notifications generated during provider/account auto-selection.
  370      /// The TUI should drain and display these on session start.
  371      startup_notices: RwLock<Vec<String>>,
  372      /// CLI/environment selection to use when creating fresh sessions. This is
  373      /// only an initial preference and never restricts later model switches.
  374      initial_provider: Option<ActiveProvider>,

  485                      StreamEvent::MessageEnd {
  486                          stop_reason: reason,
  487                      } => {
  488                          saw_message_end = true;
  489                          if reason.is_some() {
  490                              stop_reason = reason;
  491                          }
  492                          // Don't break yet - wait for SessionId which comes after MessageEnd
  493                          // (but stream close will also end the loop for providers without SessionId)
  494                      }
  495                      StreamEvent::SessionId(sid) => {
  496                          if trace {
  497                              eprintln!("[trace] session_id {}", sid);
  498                          }
  499                          self.provider_session_id = Some(sid.clone());
  500                          self.session.provider_session_id = Some(sid);
  501                          // We've received session_id, can exit the loop now
  502                          if saw_message_end {
  503                              break;
  504                          }
  505                      }
  506                      StreamEvent::UpstreamProvider { provider } => {
  507                          // Log upstream provider for local trace output
  508                          if trace {
      … 29 lines omitted; exact range 485–547 …
  538                              let tokens_str = pre_tokens
  539                                  .map(|t| format!(" ({} tokens)", t))
  540                                  .unwrap_or_default();
  541                              crate::terminal_println!(
  542                                  "📦 Context compacted ({}){}",
  543                                  trigger,
  544                                  tokens_str
  545                              );
  546                          }
  547                      }

  451  /// Build system prompt split into static (cacheable) and dynamic parts
  452  /// This improves cache hit rate by keeping frequently-changing content separate
  453  pub fn build_system_prompt_split(
  454      skill_prompt: Option<&str>,
  455      available_skills: &[SkillInfo],
  456      is_selfdev: bool,
  457      memory_prompt: Option<&str>,
  458      working_dir: Option<&Path>,
  459  ) -> (SplitSystemPrompt, ContextInfo) {
  460      build_system_prompt_split_with_capabilities(
  461          skill_prompt,
  462          available_skills,
  463          is_selfdev,
  464          memory_prompt,
  465          working_dir,
  466          PromptCapabilities::current(),
  467      )
  468  }
  469  
  470  pub fn build_system_prompt_split_with_capabilities(
  471      skill_prompt: Option<&str>,
  472      available_skills: &[SkillInfo],
  473      is_selfdev: bool,
  474      memory_prompt: Option<&str>,
      … 73 lines omitted; exact range 451–557 …
  548      info.total_chars = static_part.len() + dynamic_part.len();
  549  
  550      (
  551          SplitSystemPrompt {
  552              static_part,
  553              dynamic_part,
  554          },
  555          info,
  556      )
  557  }

  128  /// Manages background compaction of conversation context.
  129  ///
  130  /// Does NOT own message data. The caller owns the messages and passes
  131  /// references into methods that need them. After compaction, the manager
  132  /// records `compacted_count` — the number of leading messages that have
  133  /// been summarized and should be skipped when building API payloads.
  134  pub struct CompactionManager {
  135      /// Number of leading messages that have been compacted into the summary.
  136      /// When building API messages, skip the first `compacted_count` messages.
  137      compacted_count: usize,
  138  
  139      /// Active summary (if we've compacted before)
  140      active_summary: Option<Summary>,
  141  
  142      /// Rolling char estimate for the active (non-compacted) message suffix.
  143      ///
  144      /// In the common append-only case this is maintained incrementally, so token
  145      /// estimation does not need to rescan the entire active history every time.
  146      /// Bundled with its own dirty flag so the value and staleness can never
  147      /// drift apart (see [`ActiveCharEstimate`]).
  148      active_chars: ActiveCharEstimate,
  149  
  150      /// Background compaction task handle
  151      pending_task: Option<JoinHandle<Result<CompactionResult>>>,
      … 44 lines omitted; exact range 128–205 …
  196      embedding_history: VecDeque<Vec<f32>>,
  197  
  198      /// Local cache for semantic compaction embeddings keyed by truncated-text hash.
  199      /// Stores both successful embeddings and failed lookups (`None`) so repeated
  200      /// semantic scans do not redo the same work.
  201      semantic_embed_cache: HashMap<u64, (Option<Vec<f32>>, u64)>,
  202  
  203      /// Monotonic recency counter for the semantic embedding cache LRU.
  204      semantic_embed_cache_counter: u64,
  205  }

  456      // ── Anti-signal guard (shared by proactive + semantic) ──────────────────
  457  
  458      /// Returns `true` when any anti-signal fires and we should NOT compact
  459      /// proactively right now.
  460      ///
  461      /// Anti-signals are universal guards applied before the mode-specific
  462      /// trigger logic. They prevent wasted work and respect user intent.
  463      fn anti_signals_block(&self, all_messages: &[Message]) -> bool {
  464          let cfg = &self.compaction_config;
  465  
  466          // 1. Already compacting — never double-trigger.
  467          if self.pending_task.is_some() {
  468              return true;
  469          }
  470  
  471          // 2. Context below the proactive floor — too early regardless of trend.
  472          let usage = self.context_usage_with(all_messages);
  473          if usage < cfg.proactive_floor {
  474              return true;
  475          }
  476  
  477          // 3. Not enough token history to project from.
  478          if self.token_history.len() < cfg.min_samples {
  479              return true;
      … 54 lines omitted; exact range 456–543 …
  534              let delta = ((snapshots[i] as f64) - (snapshots[i - 1] as f64)).max(0.0);
  535              ewma_delta = alpha * delta + (1.0 - alpha) * ewma_delta;
  536          }
  537          let Some(current) = snapshots.last().copied().map(|value| value as f64) else {
  538              return false;
  539          };
  540          let projected = current + ewma_delta * cfg.lookahead_turns as f64;
  541  
  542          crate::logging::info(&format!(
  543              "[compaction/proactive] current={:.0} ewma_delta={:.1}/turn projected@{}turns={:.0} threshold={:.0}",

   90      fn is_context_limit_error(error: &str) -> bool {
   91          let lower = error.to_lowercase();
   92          lower.contains("context length")
   93              || lower.contains("context window")
   94              || lower.contains("maximum context")
   95              || lower.contains("max context")
   96              || lower.contains("token limit")
   97              || lower.contains("too many tokens")
   98              || lower.contains("prompt is too long")
   99              || lower.contains("input is too long")
  100              || lower.contains("request too large")
  101              || lower.contains("length limit")
  102              || lower.contains("maximum tokens")
  103              || (lower.contains("exceeded") && lower.contains("tokens"))
  104      }
  105  
  106      /// Best-effort emergency recovery after a context-limit error.
  107      ///
  108      /// Performs a synchronous hard compaction and resets provider session state,
  109      /// allowing the caller to retry the same turn immediately.
  110      pub(super) fn try_auto_compact_after_context_limit(&mut self, error: &str) -> bool {
  111          if crate::provider::openai_request::is_openai_encrypted_content_too_large_error(error)
  112              && self.try_recover_oversized_openai_native_compaction()
  113          {
      … 59 lines omitted; exact range 90–182 …
  173                  "context_limit_auto_compaction",
  174              )
  175              .with_session_id(self.session.id.clone())
  176              .with_detail(format!(
  177                  "dropped_messages={dropped},usage_pct={usage_pct:.1}"
  178              ))
  179              .force_attribution(),
  180          );
  181  
  182          true

    1  //! Persistent Memory Agent
    2  //!
    3  //! A dedicated Haiku-powered agent for memory management that runs alongside
    4  //! the main agent. It has access to memory-specific tools only (no code execution).
    5  //!
    6  //! Architecture:
    7  //! - Receives context updates from main agent via channel
    8  //! - Uses embeddings for fast similarity search
    9  //! - Uses Haiku LLM to decide what's relevant and dig deeper
   10  //! - Surfaces relevant memories to main agent via PENDING_MEMORY
   11  
   12  use anyhow::Result;
   13  use chrono::Utc;
   14  use std::collections::{HashMap, HashSet};
   15  use std::sync::Arc;
   16  use std::sync::Mutex;
   17  use std::sync::atomic::{AtomicU64, Ordering};
   18  use std::time::Instant;
   19  use tokio::sync::mpsc;
   20  
   21  use crate::embedding;
   22  use crate::memory::{self, MemoryEntry, MemoryManager};
   23  use crate::memory_graph::{ClusterEntry, EdgeKind, MemoryGraph};
   24  use crate::memory_types::{MemoryEventKind, MemoryState, StepResult, StepStatus};
      … 11 lines omitted; exact range 1–45 …
   36  }
   37  
   38  /// Channel capacity for context updates
   39  const CONTEXT_CHANNEL_CAPACITY: usize = 16;
   40  
   41  /// Similarity threshold for topic change detection (lower = more different)
   42  const TOPIC_CHANGE_THRESHOLD: f32 = 0.3;
   43  
   44  /// Maximum memories to surface per turn
   45  const MAX_MEMORIES_PER_TURN: usize = 5;

  159  trait MemoryEntryEmbeddingExt {
  160      fn ensure_embedding(&mut self) -> bool;
  161  }
  162  
  163  impl MemoryEntryEmbeddingExt for MemoryEntry {
  164      /// Generate and set embedding if not already present.
  165      /// Returns true if embedding was generated, false if already exists or failed.
  166      fn ensure_embedding(&mut self) -> bool {
  167          if self.embedding.is_some() {
  168              return false;
  169          }
  170  
  171          match crate::embedding_backend::embed_passage_active(&self.content) {
  172              Ok((embedding, model_id)) => {
  173                  // Tag with the ACTIVE backend's model id so dense search only
  174                  // compares vectors from the same model/vector space. Untagged
  175                  // legacy memories are treated as local MiniLM via
  176                  // effective_embedding_model().
  177                  self.set_embedding(Some(embedding), Some(model_id));
  178                  true
  179              }
  180              Err(err) => {
  181                  crate::logging::info(&format!("Failed to generate embedding: {err}"));
  182                  false
  183              }
  184          }
  185      }

    9  pub const TOOL_INTENT_DESCRIPTION: &str = concat!(
   10      "Short natural-language label explaining why this tool call is being made. ",
   11      "Used for compact UI display only. Required on every call; do not use this instead of required tool parameters."
   12  );
   13  
   14  pub fn intent_schema_property() -> Value {
   15      serde_json::json!({
   16          "type": "string",
   17          "description": TOOL_INTENT_DESCRIPTION,
   18      })
   19  }
   20  
   21  /// Ensure a tool parameter schema declares the shared `intent` property and
   22  /// marks it required. Applied centrally when converting tools to provider
   23  /// definitions so every tool (including MCP proxies) asks the model for an
   24  /// intent without each tool wiring it manually.
   25  pub fn ensure_intent_in_schema(mut schema: Value) -> Value {
   26      let Some(object) = schema.as_object_mut() else {
   27          return schema;
   28      };
   29      // Only touch object-shaped parameter schemas.
   30      let is_object_schema = object
   31          .get("type")
   32          .and_then(|t| t.as_str())
      … 23 lines omitted; exact range 9–65 …
   56          _ => {
   57              object.insert(
   58                  "required".to_string(),
   59                  Value::Array(vec![Value::String("intent".to_string())]),
   60              );
   61          }
   62      }
   63  
   64      schema
   65  }

  878              // Execute tools and add results
  879              let mut tool_results_dirty = false;
  880              for tc in tool_calls {
  881                  let message_id = assistant_message_id
  882                      .clone()
  883                      .unwrap_or_else(|| self.session.id.clone());
  884  
  885                  if let Some(error_msg) = tc.validation_error() {
  886                      logging::warn(&error_msg);
  887                      Bus::global().publish(BusEvent::ToolUpdated(ToolEvent {
  888                          session_id: self.session.id.clone(),
  889                          message_id: message_id.clone(),
  890                          tool_call_id: tc.id.clone(),
  891                          tool_name: tc.name.clone(),
  892                          status: ToolStatus::Error,
  893                          intent: tc.intent.clone(),
  894                          title: None,
  895                      }));
  896                      if print_output {
  897                          println!("\n  → {}", error_msg);
  898                      }
  899                      self.add_message(
  900                          Role::User,
  901                          vec![ContentBlock::ToolResult {
      … 6 lines omitted; exact range 878–917 …
  908                      continue;
  909                  }
  910  
  911                  self.validate_tool_allowed(&tc.name)?;
  912  
  913                  let is_native_tool = JCODE_NATIVE_TOOLS.contains(&tc.name.as_str());
  914  
  915                  // Check if SDK already executed this tool
  916                  if let Some((sdk_content, sdk_is_error)) = sdk_tool_results.remove(&tc.id) {
  917                      // For native tools, ignore SDK errors and execute locally

  627          // Context overflow guard: check if this output would push us over the limit
  628          output = self.guard_context_overflow(name, output).await;
  629  
  630          let mut fields = Self::tool_lifecycle_fields("done", name, resolved_name, &input, &ctx);
  631          fields.push(("elapsed_ms".to_string(), latency_ms.to_string()));
  632          fields.push(("output_bytes".to_string(), output.output.len().to_string()));
  633          fields.push((
  634              "output_chars".to_string(),
  635              output.output.chars().count().to_string(),
  636          ));
  637          fields.push(("image_count".to_string(), output.images.len().to_string()));
  638          crate::logging::event_info("TOOL_LIFECYCLE", fields);
  639  
  640          Ok(output)
  641      }
  642  
  643      /// Check if a tool output would overflow the context window and truncate if needed.
  644      /// Returns the (possibly truncated) output.
  645      async fn guard_context_overflow(&self, tool_name: &str, output: ToolOutput) -> ToolOutput {
  646          let compaction = self.compaction.read().await;
  647          let budget = compaction.token_budget();
  648          if budget == 0 {
  649              return output;
  650          }
      … 18 lines omitted; exact range 627–678 …
  669          let remaining = if current_tokens < threshold_tokens {
  670              threshold_tokens - current_tokens
  671          } else {
  672              // Already over threshold — allow a small amount for the error message
  673              budget / 50 // ~2% of budget for the truncation notice
  674          };
  675          let max_tokens = remaining.min(single_max_tokens);
  676  
  677          // Convert token limit back to approximate character limit
  678          let max_chars = max_tokens * 4;

  543      /// Execute a tool by name
  544      pub async fn execute(&self, name: &str, input: Value, ctx: ToolContext) -> Result<ToolOutput> {
  545          let tools = self.tools.read().await;
  546          let resolved_name = Self::resolve_tool_name(name);
  547          if let Some(policy) = session_tool_policy(&ctx.session_id) {
  548              if let Some(allowed) = policy.allowed_tools.as_ref()
  549                  && !allowed.contains(resolved_name)
  550              {
  551                  return Err(anyhow::anyhow!("Tool '{}' is not allowed", resolved_name));
  552              }
  553              if policy.disabled_tools.contains(resolved_name) {
  554                  return Err(anyhow::anyhow!("Tool '{}' is disabled", resolved_name));
  555              }
  556          }
  557          let tool = match tools.get(resolved_name) {
  558              Some(tool) => tool.clone(),
  559              None => {
  560                  // List available tools so the model can recover instead of
  561                  // spiraling through hallucinated names like "ToolSearch" (#104).
  562                  let mut available: Vec<&str> = tools.keys().map(|k| k.as_str()).collect();
  563                  available.sort_unstable();
  564                  let suggestions = Self::closest_tool_names(name, &available);
  565                  let mut msg = format!("Unknown tool: {name}.");
  566                  if !suggestions.is_empty() {
      … 25 lines omitted; exact range 543–601 …
  592              if let crate::hooks::GateDecision::Block { reason } = decision {
  593                  let mut fields =
  594                      Self::tool_lifecycle_fields("blocked", name, resolved_name, &input, &ctx);
  595                  fields.push(("block_reason".to_string(), reason.clone()));
  596                  crate::logging::event_warn("TOOL_LIFECYCLE", fields);
  597                  return Err(anyhow::anyhow!(
  598                      "Tool call blocked by pre_tool hook: {reason}"
  599                  ));
  600              }
  601          }

    1  //! The destructive-command gate for the `bash` tool (issue #604).
    2  //!
    3  //! Kept in its own file so the policy seam is easy to find and review: this is
    4  //! the only thing standing between a model's `rm -rf` and the user's data.
    5  
    6  /// Apply the deterministic destructive-command gate, returning refusal text
    7  /// when the command must not run as-issued.
    8  ///
    9  /// Stage 1 is a pure blast-radius assessment; stage 2 turns a `Confirm` verdict
   10  /// into a reflection prompt that a blind retry cannot satisfy. Catastrophic
   11  /// targets (`/`, `$HOME`, credential stores, device nodes) are denied outright.
   12  /// See issue #604.
   13  pub(super) fn destructive_command_refusal(
   14      command: &str,
   15      justification: Option<&str>,
   16      working_dir: Option<std::path::PathBuf>,
   17  ) -> Option<String> {
   18      let risk_ctx = jcode_command_risk::RiskContext::from_env(working_dir);
   19      let assessment = jcode_command_risk::assess(command, &risk_ctx);
   20      if assessment.level.runs_immediately() {
   21          return None;
   22      }
   23  
   24      let justification = jcode_command_risk::Justification {
      … 5 lines omitted; exact range 1–39 …
   30              crate::logging::warn(&format!("[bash] denied destructive command: {command}"));
   31              Some(reason)
   32          }
   33          jcode_command_risk::GateOutcome::Reflect { prompt } => {
   34              crate::logging::info(&format!(
   35                  "[bash] destructive command held for justification: {command}"
   36              ));
   37              Some(prompt)
   38          }
   39      }

  724          // Foreground execution with stdin detection
  725          self.execute_foreground(&params, &ctx).await
  726      }
  727  }
  728  
  729  impl BashTool {
  730      async fn execute_foreground(
  731          &self,
  732          params: &BashInput,
  733          ctx: &ToolContext,
  734      ) -> Result<ToolOutput> {
  735          #[cfg(unix)]
  736          if self.supports_reload_persistence(ctx) {
  737              return self
  738                  .execute_reload_persistable_foreground(params, ctx)
  739                  .await;
  740          }
  741  
  742          let timeout_ms = params.timeout.unwrap_or(DEFAULT_TIMEOUT_MS).min(600000);
  743          let timeout_duration = Duration::from_millis(timeout_ms);
  744  
  745          let has_stdin_channel = ctx.stdin_request_tx.is_some();
  746  
  747          let mut command = build_shell_command(&params.command);
      … 3 lines omitted; exact range 724–760 …
  751              .stderr(Stdio::piped());
  752  
  753          if has_stdin_channel {
  754              command.stdin(Stdio::piped());
  755          }
  756  
  757          if let Some(ref dir) = ctx.working_dir {
  758              command.current_dir(dir);
  759          }
  760          let mut child = command.spawn()?;

    1  //! MCP Manager - manages MCP server connections for a single session.
    2  //!
    3  //! In daemon mode with a shared pool, servers marked `shared: true` (the default)
    4  //! are managed by the pool and reused across sessions. Servers marked `shared: false`
    5  //! (e.g., Playwright with browser state) are spawned per-session.
    6  
    7  use super::client::{McpClient, McpHandle};
    8  use super::pool::SharedMcpPool;
    9  use super::protocol::{McpConfig, McpServerConfig, McpToolDef, ToolCallResult};
   10  use anyhow::{Context, Result};
   11  use serde::Serialize;
   12  use std::collections::HashMap;
   13  use std::sync::Arc;
   14  use tokio::sync::RwLock;
   15  
   16  /// Bound on how long a tool call will wait for a not-yet-connected MCP server
   17  /// to come up before failing with a clean tool error. Keeps a slow/hanging
   18  /// server from blocking a single tool call forever (and never blocks spawn).
   19  const CONNECT_ON_CALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
   20  
   21  /// Meter a completed tool call for partner-discovery provenance. No-op for
   22  /// servers without discovery provenance (the overwhelmingly common case) and
   23  /// whenever `sponsors.enabled` is false. Counts only; never content.
   24  fn meter_provenance_call(server: &str, result: &Result<ToolCallResult>) {
      … 25 lines omitted; exact range 1–59 …
   50      /// Handles from the shared pool (shared servers)
   51      pool_handles: RwLock<HashMap<String, McpHandle>>,
   52      /// Per-session owned clients (non-shared / stateful servers)
   53      owned_clients: RwLock<HashMap<String, McpClient>>,
   54      config: McpConfig,
   55      session_id: String,
   56      /// Project directory used to resolve project-local MCP config. `None`
   57      /// loads only global config and never consults the process working directory.
   58      project_dir: Option<std::path::PathBuf>,
   59  }

  335      pub(super) async fn tool_definitions(&mut self) -> Vec<ToolDefinition> {
  336          if self.session.is_canary {
  337              self.registry.register_selfdev_tools().await;
  338          }
  339  
  340          // Return locked tools if available (prevents cache invalidation from
  341          // tools arriving asynchronously after the first API request).
  342          //
  343          // Exception: MCP servers connect on a background task and register
  344          // `mcp__*` tools seconds after the session starts — typically *after*
  345          // the first turn has already locked the snapshot. We deliberately do
  346          // NOT block the first turn on MCP connection: servers can be slow or
  347          // hang, and we want the user to be able to talk to the agent the moment
  348          // the session spawns. The price is that the first locked snapshot is
  349          // missing MCP tools, and the only other unlock path fires when the model
  350          // calls the `mcp` management tool — which it cannot do without first
  351          // seeing MCP tools (#206).
  352          //
  353          // So, exactly once per locked snapshot, if MCP tools have since appeared
  354          // in the registry, we rebuild. This is a single intentional provider
  355          // prompt-cache miss (the turn MCP tools first appear). The
  356          // `mcp_late_register_resolved` flag makes this a one-shot check so we do
  357          // not rescan the registry on every subsequent turn.
  358          if let Some(ref locked) = self.locked_tools {
      … 25 lines omitted; exact range 335–393 …
  384          let tools = self.build_filtered_tool_definitions().await;
  385  
  386          // Lock the tool list to prevent cache invalidation when more tools
  387          // arrive asynchronously mid-session.
  388          logging::info(&format!(
  389              "Locking tool list at {} tools for cache stability",
  390              tools.len()
  391          ));
  392          self.locked_tools = Some(tools.clone());
  393          tools

  374  pub fn build_system_prompt_full_with_capabilities(
  375      skill_prompt: Option<&str>,
  376      available_skills: &[SkillInfo],
  377      is_selfdev: bool,
  378      memory_prompt: Option<&str>,
  379      working_dir: Option<&Path>,
  380      capabilities: PromptCapabilities,
  381  ) -> (String, ContextInfo) {
  382      let mut parts = base_system_prompt_parts(capabilities);
  383      let mut info = ContextInfo {
  384          system_prompt_chars: parts.join("\n\n").len(),
  385          ..Default::default()
  386      };
  387  
  388      // Add self-dev guidance only in active self-dev sessions. Normal sessions
  389      // learn about the on-ramp from the mode-aware `selfdev` tool schema.
  390      if is_selfdev {
  391          let selfdev_prompt = build_selfdev_prompt_for_working_dir(working_dir);
  392          info.selfdev_chars = selfdev_prompt.len();
  393          parts.push(selfdev_prompt);
  394      }
  395  
  396      // Add AGENTS.md instructions with tracking (from working_dir or cwd)
  397      let (md_content, md_info) = load_agents_md_files_from_dir(working_dir);
      … 41 lines omitted; exact range 374–448 …
  439  
  440      // Add active skill prompt
  441      if let Some(skill) = skill_prompt {
  442          parts.push(format!("# Active Skill\n\n{}", skill));
  443      }
  444  
  445      let prompt = parts.join("\n\n");
  446      info.total_chars = prompt.len();
  447  
  448      (prompt, info)

   11  /// A skill definition from SKILL.md
   12  #[derive(Debug, Clone)]
   13  pub struct Skill {
   14      pub name: String,
   15      pub description: String,
   16      pub allowed_tools: Option<Vec<String>>,
   17      pub content: String,
   18      pub path: PathBuf,
   19      search_text: String,
   20  }
   21  
   22  #[derive(Debug, Deserialize)]
   23  struct SkillFrontmatter {
   24      name: String,
   25      description: String,
   26      #[serde(rename = "allowed-tools")]
   27      allowed_tools: Option<String>,
   28  }
   29  
   30  /// Registry of available skills
   31  #[derive(Debug, Default, Clone)]
   32  pub struct SkillRegistry {
   33      skills: HashMap<String, Skill>,
   34  }
      … 16 lines omitted; exact range 11–60 …
   51  impl SkillRegistry {
   52      /// Process-wide shared mutable registry used by both `skill_manage` and
   53      /// direct slash invocation paths. Keeping a single registry prevents slash
   54      /// commands from seeing a stale startup-only skill snapshot after reloads.
   55      ///
   56      /// Holds GLOBAL skills only (plugins, `~/.jcode/skills/`,
   57      /// `~/.agents/skills/`). Project-local skills are a per-session overlay
   58      /// composed at read time from the session's workspace root (issue #457);
   59      /// they must never enter this shared registry, and the daemon's startup
   60      /// cwd must never influence its contents.

 1528  pub(super) async fn run_swarm_task(
 1529      agent: Arc<Mutex<Agent>>,
 1530      description: &str,
 1531      subagent_type: &str,
 1532      prompt: &str,
 1533  ) -> Result<String> {
 1534      let started = Instant::now();
 1535      let (provider, registry, session_id, working_dir, coordinator_model, provider_key, route) = {
 1536          let agent = agent.lock().await;
 1537          (
 1538              agent.provider_fork(),
 1539              agent.registry(),
 1540              agent.session_id().to_string(),
 1541              agent.working_dir().map(PathBuf::from),
 1542              agent.provider_model(),
 1543              agent.session_provider_key(),
 1544              agent.session_route_api_method(),
 1545          )
 1546      };
 1547      let parent_session_id = session_id.clone();
 1548      let mut session = Session::create(
 1549          Some(session_id),
 1550          Some(format!("{} (@{} swarm)", description, subagent_type)),
 1551      );
      … 52 lines omitted; exact range 1528–1613 …
 1604                      ("child_session_id", child_session_id),
 1605                      ("subagent_type", subagent_type.to_string()),
 1606                      ("error", error.to_string()),
 1607                      ("elapsed_ms", started.elapsed().as_millis().to_string()),
 1608                  ],
 1609              );
 1610              Err(error)
 1611          }
 1612      }
 1613  }

   75  /// Reasoning-effort sentinel for the **deep task graph** mode: strongest model
   76  /// reasoning AND the comprehensive DAG-first swarm workflow (decompose into a
   77  /// validated task graph, critique/verify gates, typed artifact handoffs). Sits
   78  /// one rung above [`SWARM_EFFORT`] on the effort ladder: `... xhigh`, `swarm`
   79  /// (light fan-out), `swarm-deep` (deep task graph). Providers translate this to
   80  /// their strongest real effort, while the UI/session keep the literal marker so
   81  /// the agent knows to inject [`SWARM_DEEP_EFFORT_DIRECTIVE`].
   82  pub const SWARM_DEEP_EFFORT: &str = "swarm-deep";
   83  
   84  /// System-prompt directive injected when the active reasoning effort is
   85  /// [`SWARM_EFFORT`]. Instructs the agent to lean on the swarm tooling.
   86  pub const SWARM_EFFORT_DIRECTIVE: &str = "# Swarm Effort\n\nYou are running at the maximum reasoning effort with swarm orchestration enabled. For any non-trivial task, decompose the work and use the `swarm` tool to spawn and coordinate parallel agents (spawn workers with concrete prompts, assign tasks, and collect their reports) instead of doing everything yourself in one thread. Prefer parallelizing independent subtasks across swarm members, and use a coordinator/plan when the work has multiple stages. Only skip the swarm for trivial, single-step requests.";
   87  
   88  /// System-prompt directive injected when the active reasoning effort is
   89  /// [`SWARM_DEEP_EFFORT`]. Instructs the agent to run the comprehensive DAG-first
   90  /// task-graph workflow.
   91  pub const SWARM_DEEP_EFFORT_DIRECTIVE: &str = "# Deep Task Graph\n\nYou are running at maximum reasoning effort with the deep task-graph swarm workflow. Treat the task DAG as the primary object, not ad hoc agent chat. Workflow:\n\n1. Seed a graph with `swarm task_graph` using `mode: \"deep\"`: lay out nodes (kind explore|implement|verify|fix|synthesize) and `depends_on` edges instead of answering directly. (At this effort the server already defaults the plan to deep, but pass `mode: \"deep\"` explicitly anyway.) The engine auto-inserts a plan-wide root gate over your seed: the plan cannot finish until a final adversarial audit passes, and that audit can inject new top-level work.\n2. For any node that is too big, `swarm expand_node` to decompose it into a child sub-DAG (you become its planner/integrator). In deep mode a critique/verify gate is auto-inserted before a composite node can close. The graph is EXPECTED to outgrow its seed, often by several times: growth (expansions and gate-injected gaps) is the system working, not scope creep. plan_status reports seeded-vs-grown counts.\n3. Finish each node with `swarm complete_node` and a typed artifact: `findings`, `evidence` (file:line / commit refs), `validation`, `open_questions`, a required `confidence` (low|medium|high; report low honestly, it routes follow-up work to shore up that scope), and an honest `what_i_did_not_check`. Downstream nodes are hydrated with these artifacts automatically. There is no other way to close a deep node: a turn ending without expand_node/complete_node re-queues the node to a fresh worker and fails it on repeat.\n4. When a critique/verify gate finds gaps or failures, use `swarm inject_gap` to add new nodes; the parent cannot close until they drain. A passing gate artifact must account for EVERY node it audited by id (the server rejects rubber stamps), and cannot pass over a low-confidence sibling without addressing it explicitly, so treat low-confidence siblings as priority probe targets.\n5. Use `swarm run_plan` to drive the graph to completion. It returns immediately and drives the plan as a background task (progress card + wake on completion), so keep working or answer the user while it runs; check `swarm plan_status` or `bg` for progress. Deep mode fans out wide automatically (many workers run in parallel, bounded only by the swarm member cap), so prefer decomposing into MANY independent sibling nodes rather than a few serial ones: keep the ready set wide so run_plan can dispatch lots of agents at once. Only add `depends_on` edges for real data dependencies.\n\nComprehensiveness is structural: prefer decomposition + gates over a single thorough answer, so it is very unlikely any nook or cranny is missed.";

  213  /// Durable, persistable portion of a swarm member.
  214  #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
  215  pub struct SwarmMemberRecord {
  216      pub session_id: String,
  217      pub working_dir: Option<PathBuf>,
  218      pub swarm_id: Option<String>,
  219      pub swarm_enabled: bool,
  220      pub status: SwarmLifecycleStatus,
  221      pub detail: Option<String>,
  222      /// Stable label of the task/role this member was spawned or assigned for.
  223      #[serde(default, skip_serializing_if = "Option::is_none")]
  224      pub task_label: Option<String>,
  225      pub friendly_name: Option<String>,
  226      pub report_back_to_session_id: Option<String>,
  227      #[serde(default, skip_serializing_if = "Option::is_none")]
  228      pub latest_completion_report: Option<String>,
  229      pub role: SwarmRole,
  230      pub is_headless: bool,
  231  }
  232  
  233  /// Bidirectional index for swarm channel subscriptions.
  234  #[derive(Clone, Debug, Default, PartialEq, Eq)]
  235  pub struct ChannelIndex {
  236      pub by_swarm_channel: HashMap<String, HashMap<String, HashSet<String>>>,
      … 7 lines omitted; exact range 213–253 …
  244              .or_default()
  245              .entry(channel.to_string())
  246              .or_default()
  247              .insert(session_id.to_string());
  248          self.by_session
  249              .entry(session_id.to_string())
  250              .or_default()
  251              .entry(swarm_id.to_string())
  252              .or_default()
  253              .insert(channel.to_string());

  307      pub fn save(&mut self) -> Result<()> {
  308          self.updated_at = Utc::now();
  309          let path = session_path(&self.id)?;
  310          let journal_path = session_journal_path_from_snapshot(&path);
  311          let start = std::time::Instant::now();
  312          let snapshot_bytes_before = file_len_or_zero(&path);
  313          let journal_bytes_before = file_len_or_zero(&journal_path);
  314          let current_meta = self.journal_meta();
  315          let metadata_needs_snapshot = self
  316              .persist_state
  317              .last_meta
  318              .as_ref()
  319              .is_some_and(|prev| metadata_requires_snapshot(prev, &current_meta));
  320          let vectors_need_snapshot = !self.persist_state.snapshot_exists
  321              || self.persist_state.messages_mode == PersistVectorMode::Full
  322              || self.persist_state.env_snapshots_mode == PersistVectorMode::Full
  323              || self.persist_state.memory_injections_mode == PersistVectorMode::Full
  324              || self.persist_state.replay_events_mode == PersistVectorMode::Full
  325              || self.messages.len() < self.persist_state.messages_len
  326              || self.env_snapshots.len() < self.persist_state.env_snapshots_len
  327              || self.memory_injections.len() < self.persist_state.memory_injections_len
  328              || self.replay_events.len() < self.persist_state.replay_events_len;
  329  
  330          let delta_messages = self
      … 55 lines omitted; exact range 307–395 …
  386                  Ok(()) => {
  387                      self.reset_persist_state(true);
  388                      let journal_stat_start = Instant::now();
  389                      let journal_bytes_after = file_len_or_zero(&journal_path);
  390                      let journal_stat_ms = journal_stat_start.elapsed().as_millis();
  391                      if journal_bytes_after > MAX_SESSION_JOURNAL_BYTES {
  392                          let checkpoint_start = Instant::now();
  393                          let result = self.checkpoint_snapshot(&path, &journal_path);
  394                          let checkpoint_ms = checkpoint_start.elapsed().as_millis();
  395                          let journal_bytes_after = file_len_or_zero(&journal_path);

   26  /// Attempt to recover complete entries from a journal line that failed the
   27  /// strict one-entry-per-line parse.
   28  ///
   29  /// If a writer died mid-append (torn line without a trailing newline), the
   30  /// next successful append starts writing on the same line, producing
   31  /// `<torn json><complete entry json>\n` or `<entry json><entry json>\n`.
   32  /// Serialized entries always begin with `{"meta":` (struct field order), so
   33  /// scan for candidate starts and stream-parse consecutive complete entries
   34  /// from the first position that yields any.
   35  fn salvage_glued_journal_entries(line: &str, mut apply: impl FnMut(SessionJournalEntry)) -> usize {
   36      const ENTRY_START: &str = "{\"meta\":";
   37      let mut salvaged = 0usize;
   38      let mut search_from = 0usize;
   39      while let Some(rel) = line
   40          .get(search_from..)
   41          .and_then(|rest| rest.find(ENTRY_START))
   42      {
   43          let candidate_start = search_from + rel;
   44          let mut stream = serde_json::Deserializer::from_str(&line[candidate_start..])
   45              .into_iter::<SessionJournalEntry>();
   46          let mut parsed = Vec::new();
   47          for item in &mut stream {
   48              match item {
   49                  Ok(entry) => parsed.push(entry),
      … 69 lines omitted; exact range 26–128 …
  119                  ("phase", "journal_replay_corruption".to_string()),
  120                  ("path", journal_path.display().to_string()),
  121                  ("entries_replayed", stats.entries.to_string()),
  122                  ("lines_skipped", stats.skipped_lines.to_string()),
  123                  ("entries_salvaged", stats.salvaged_entries.to_string()),
  124              ],
  125          );
  126      }
  127  
  128      Ok(stats)

  603          crate::logging::event_info(
  604              "TOOL_LIFECYCLE",
  605              Self::tool_lifecycle_fields("start", name, resolved_name, &input, &ctx),
  606          );
  607  
  608          let started_at = std::time::Instant::now();
  609          let result = tool.execute(input.clone(), ctx.clone()).await;
  610          let latency_ms = started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64;
  611  
  612          crate::telemetry::record_tool_execution(resolved_name, &input, result.is_ok(), latency_ms);
  613          Self::fire_post_tool_hook(resolved_name, &ctx, &result, latency_ms);
  614  
  615          let mut output = match result {
  616              Ok(output) => output,
  617              Err(error) => {
  618                  let mut fields =
  619                      Self::tool_lifecycle_fields("error", name, resolved_name, &input, &ctx);
  620                  fields.push(("elapsed_ms".to_string(), latency_ms.to_string()));
  621                  fields.push(("error".to_string(), crate::util::format_error_chain(&error)));
  622                  crate::logging::event_warn("TOOL_LIFECYCLE", fields);
  623                  return Err(error);
  624              }
  625          };
  626  
      … 2 lines omitted; exact range 603–638 …
  629  
  630          let mut fields = Self::tool_lifecycle_fields("done", name, resolved_name, &input, &ctx);
  631          fields.push(("elapsed_ms".to_string(), latency_ms.to_string()));
  632          fields.push(("output_bytes".to_string(), output.output.len().to_string()));
  633          fields.push((
  634              "output_chars".to_string(),
  635              output.output.chars().count().to_string(),
  636          ));
  637          fields.push(("image_count".to_string(), output.images.len().to_string()));
  638          crate::logging::event_info("TOOL_LIFECYCLE", fields);

  388  async fn test_guard_at_95_triggers_hard_compact() {
  389      let mut manager = CompactionManager::new().with_budget(1_000);
  390      let mut messages = Vec::new();
  391      for i in 0..20 {
  392          messages.push(make_text_message(
  393              Role::User,
  394              &format!("message {} with padding {}", i, "x".repeat(50)),
  395          ));
  396          manager.notify_message_added();
  397      }
  398      // 96% usage — above critical threshold
  399      manager.update_observed_input_tokens(960);
  400  
  401      let provider: Arc<dyn Provider> = Arc::new(MockSummaryProvider);
  402      let action = manager.ensure_context_fits(&messages, provider);
  403      assert!(
  404          matches!(action, CompactionAction::HardCompacted(_)),
  405          "SHOULD hard-compact at 96%"
  406      );
  407      assert!(
  408          manager.compacted_count > 0,
  409          "compacted_count should increase after hard compact"
  410      );
  411      assert!(
      … 30 lines omitted; exact range 388–451 …
  442      );
  443      // First message should be the emergency summary
  444      match &api_messages[0].content[0] {
  445          ContentBlock::Text { text, .. } => {
  446              assert!(text.contains("Previous Conversation Summary"));
  447              assert!(text.contains("Emergency compaction"));
  448          }
  449          _ => panic!("expected text summary block"),
  450      }
  451  }