OpenAI Codex
从源码学会它怎么工作

  153  pub(crate) async fn run_turn(
  154      sess: Arc<Session>,
  155      turn_context: Arc<TurnContext>,
  156      turn_extension_data: Arc<codex_extension_api::ExtensionData>,
  157      input: Vec<TurnInput>,
  158      prewarmed_client_session: Option<ModelClientSession>,
  159      cancellation_token: CancellationToken,
  160  ) -> CodexResult<Option<String>> {
  161      let mut client_session =
  162          prewarmed_client_session.unwrap_or_else(|| sess.services.model_client.new_session());
  163      // TODO(ccunningham): Pre-turn compaction runs before context updates and the
  164      // new user message are recorded. Estimate pending incoming items (context
  165      // diffs/full reinjection + user input) and trigger compaction preemptively
  166      // when they would push the thread over the compaction threshold.
  167      if let Err(err) = run_pre_sampling_compact(
  168          &sess,
  169          &turn_context,
  170          &mut client_session,
  171          &cancellation_token,
  172      )
  173      .await
  174      {
  175          if matches!(err.details(), CodexErrorDetails::TurnAborted) {
  176              run_hooks_and_record_inputs(&sess, &turn_context, &input).await;
  177              return Err(err);
      … 86 lines omitted; exact range 153–274 …
  264          if run_hooks_and_record_inputs(&sess, &turn_context, &pending_input).await {
  265              break;
  266          }
  267  
  268          let window_id = sess.current_window_id().await;
  269          super::rollout_budget::maybe_record_reminder(
  270              sess.as_ref(),
  271              turn_context.as_ref(),
  272              &window_id,
  273          )
  274          .await;

 2034      tool_runtime: ToolCallRuntime,
 2035      sess: Arc<Session>,
 2036      turn_context: Arc<TurnContext>,
 2037      turn_store: Arc<codex_extension_api::ExtensionData>,
 2038      client_session: &mut ModelClientSession,
 2039      responses_metadata: &CodexResponsesMetadata,
 2040      turn_diff_tracker: SharedTurnDiffTracker,
 2041      prompt: &Prompt,
 2042      cancellation_token: CancellationToken,
 2043  ) -> CodexResult<SamplingRequestResult> {
 2044      feedback_tags!(
 2045          model = turn_context.model_info.slug.clone(),
 2046          approval_policy = turn_context.approval_policy.value(),
 2047          sandbox_policy = &turn_context.sandbox_policy(),
 2048          effort = turn_context.reasoning_effort,
 2049          auth_mode = sess.services.auth_manager.auth_mode(),
 2050          features = sess.features.enabled_features(),
 2051      );
 2052      let inference_trace = sess.services.rollout_thread_trace.inference_trace_context(
 2053          turn_context.sub_id.as_str(),
 2054          turn_context.model_info.slug.as_str(),
 2055          turn_context.provider.info().name.as_str(),
 2056      );
 2057      let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling();
 2058      let uses_sequential_cutoff_reasoning_summaries = turn_context
      … 99 lines omitted; exact range 2034–2168 …
 2158                      let item_id = previous.id();
 2159                      flush_assistant_text_segments_for_item(
 2160                          &sess,
 2161                          &turn_context,
 2162                          plan_mode_state.as_mut(),
 2163                          &mut assistant_message_stream_parsers,
 2164                          &item_id,
 2165                      )
 2166                      .await;
 2167                  }
 2168                  if let Some(state) = plan_mode_state.as_mut()

 1176  async fn run_sampling_request(
 1177      sess: Arc<Session>,
 1178      step_context: Arc<StepContext>,
 1179      turn_store: Arc<codex_extension_api::ExtensionData>,
 1180      turn_diff_tracker: SharedTurnDiffTracker,
 1181      client_session: &mut ModelClientSession,
 1182      responses_metadata: &CodexResponsesMetadata,
 1183      input: Vec<ResponseItem>,
 1184      cancellation_token: CancellationToken,
 1185  ) -> CodexResult<(SamplingRequestResult, Vec<ResponseItem>)> {
 1186      let turn_context = Arc::clone(&step_context.turn);
 1187      let router = Arc::clone(&step_context.tool_router);
 1188  
 1189      let base_instructions = sess.get_base_instructions().await;
 1190  
 1191      let tool_runtime = ToolCallRuntime::new(
 1192          Arc::clone(&router),
 1193          Arc::clone(&sess),
 1194          Arc::clone(&step_context),
 1195          Arc::clone(&turn_diff_tracker),
 1196      );
 1197      let _code_mode_worker = sess.services.code_mode_service.start_turn_worker(
 1198          &sess,
 1199          Arc::clone(&step_context),
 1200          Arc::clone(&router),
      … 62 lines omitted; exact range 1176–1273 …
 1263              max_retries,
 1264              err,
 1265              client_session,
 1266              &sess,
 1267              &turn_context,
 1268              ResponsesStreamRequest::Sampling,
 1269          )
 1270          .await?;
 1271          turn_context.turn_timing_state.record_sampling_retry();
 1272      }
 1273  }

   54  /// Wire protocol that the provider speaks.
   55  #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, JsonSchema)]
   56  #[serde(rename_all = "lowercase")]
   57  pub enum WireApi {
   58      /// The Responses API exposed by OpenAI at `/v1/responses`.
   59      #[default]
   60      Responses,
   61  }
   62  
   63  impl fmt::Display for WireApi {
   64      fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
   65          let value = match self {
   66              Self::Responses => "responses",
   67          };
   68          f.write_str(value)
   69      }
   70  }
   71  
   72  impl<'de> Deserialize<'de> for WireApi {
   73      fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
   74      where
   75          D: serde::Deserializer<'de>,
   76      {
   77          let value = String::deserialize(deserializer)?;
   78          match value.as_str() {
   79              "responses" => Ok(Self::Responses),
   80              "chat" => Err(serde::de::Error::custom(CHAT_WIRE_API_REMOVED_ERROR)),
   81              _ => Err(serde::de::Error::unknown_variant(&value, &["responses"])),
   82          }
   83      }
   84  }

    1  //! Session- and turn-scoped helpers for talking to model provider APIs.
    2  //!
    3  //! `ModelClient` is intended to live for the lifetime of a Codex session and holds the stable
    4  //! configuration and state needed to talk to a provider (auth, provider selection, conversation id,
    5  //! and transport fallback state).
    6  //!
    7  //! Per-turn settings (model selection, reasoning controls, telemetry context, and turn metadata)
    8  //! are passed explicitly to streaming and unary methods so that the turn lifetime is visible at the
    9  //! call site.
   10  //!
   11  //! A [`ModelClientSession`] is created per turn and is used to stream one or more Responses API
   12  //! requests during that turn. It caches a Responses WebSocket connection (opened lazily) and stores
   13  //! per-turn state such as the `x-codex-turn-state` token used for sticky routing.
   14  //!
   15  //! WebSocket prewarm is a v2-only `response.create` with `generate=false`; it waits for completion
   16  //! so the next request can reuse the same connection and `previous_response_id`.
   17  //!
   18  //! Turn execution performs prewarm as a best-effort step before the first stream request so the
   19  //! subsequent request can reuse the same connection.
   20  //!
   21  //! ## Retry-Budget Tradeoff
   22  //!
   23  //! WebSocket prewarm is treated as the first websocket connection attempt for a turn. If it
   24  //! fails, normal stream retry/fallback logic handles recovery on the same turn.

   48  /// Typed runtime contract for locally executed tools.
   49  ///
   50  /// Implementers provide the shared `ToolExecutor` behavior plus optional
   51  /// core-owned metadata for hooks, telemetry, tool search, and argument diffs.
   52  pub(crate) trait CoreToolRuntime: ToolExecutor<ToolInvocation> {
   53      fn matches_kind(&self, payload: &ToolPayload) -> bool {
   54          matches!(
   55              payload,
   56              ToolPayload::Function { .. } | ToolPayload::ToolSearch { .. }
   57          )
   58      }
   59  
   60      /// Whether cancellation should let the handler finish teardown before the
   61      /// host returns an aborted tool response.
   62      fn waits_for_runtime_cancellation(&self) -> bool {
   63          false
   64      }
   65  
   66      fn telemetry_tags<'a>(
   67          &'a self,
   68          _invocation: &'a ToolInvocation,
   69      ) -> BoxFuture<'a, ToolTelemetryTags> {
   70          Box::pin(async { Vec::new() })
   71      }
   72  
      … 66 lines omitted; exact range 48–149 …
  139          Ok(ToolInvocation {
  140              payload: ToolPayload::Function { arguments },
  141              ..invocation
  142          })
  143      }
  144  
  145      /// Creates an optional consumer for streamed tool argument diffs.
  146      fn create_diff_consumer(&self) -> Option<Box<dyn ToolArgumentDiffConsumer>> {
  147          None
  148      }
  149  }

   74      #[instrument(level = "trace", skip_all)]
   75      pub(crate) fn handle_tool_call(
   76          self,
   77          call: ToolCall,
   78          cancellation_token: CancellationToken,
   79      ) -> impl std::future::Future<Output = Result<ResponseInputItem, CodexErr>> {
   80          let error_call = call.clone();
   81          let future =
   82              self.handle_tool_call_with_source(call, ToolCallSource::Direct, cancellation_token);
   83          async move {
   84              match future.await {
   85                  Ok(response) => Ok(response.into_response()),
   86                  Err(FunctionCallError::Fatal(message)) => Err(CodexErr::Fatal(message)),
   87                  Err(other) => Ok(Self::failure_response(error_call, other)),
   88              }
   89          }
   90          .in_current_span()
   91      }
   92  
   93      #[instrument(level = "trace", skip_all)]
   94      pub(crate) fn handle_tool_call_with_source(
   95          self,
   96          call: ToolCall,
   97          source: ToolCallSource,
   98          cancellation_token: CancellationToken,
      … 36 lines omitted; exact range 74–145 …
  135                  } else {
  136                      Either::Right(lock.write().await)
  137                  };
  138                  // Admission through the parallel-execution gate marks the end
  139                  // of dispatch waiting and the start of handler execution.
  140                  if let Some(execution_started_at) = execution_started_at {
  141                      let _ = execution_started_at.set(Instant::now());
  142                  }
  143  
  144                  router
  145                      .dispatch_tool_call_with_terminal_outcome(

  637  async fn shell_family_registers_visible_unified_exec_and_hidden_legacy_shell() {
  638      let plan = probe(|turn| {
  639          set_features(turn, &[Feature::ShellTool, Feature::UnifiedExec]);
  640          set_feature(turn, Feature::ShellZshFork, /*enabled*/ false);
  641          turn.model_info.shell_type = ConfigShellToolType::ShellCommand;
  642      })
  643      .await;
  644  
  645      plan.assert_visible_contains(&["exec_command", "write_stdin"]);
  646      plan.assert_visible_lacks(&["shell_command"]);
  647      plan.assert_registered_contains(&["exec_command", "write_stdin", "shell_command"]);
  648      assert_eq!(plan.exposure("shell_command"), ToolExposure::Hidden);
  649      assert!(has_parameter(plan.visible_spec("exec_command"), "shell"));
  650  }
  651  
  652  #[tokio::test]
  653  async fn shell_zsh_fork_stays_standalone_until_unified_exec_composition_is_enabled() {
  654      let standalone = probe(|turn| {
  655          set_features(turn, &[Feature::ShellTool, Feature::UnifiedExec]);
  656          set_feature(turn, Feature::ShellZshFork, /*enabled*/ true);
  657          set_feature(turn, Feature::UnifiedExecZshFork, /*enabled*/ false);
  658          turn.model_info.shell_type = ConfigShellToolType::ShellCommand;
  659      })
  660      .await;
  661  
      … 36 lines omitted; exact range 637–708 …
  698      let plan = probe(|turn| {
  699          set_features(
  700              turn,
  701              &[
  702                  Feature::ShellTool,
  703                  Feature::UnifiedExec,
  704                  Feature::ShellZshFork,
  705                  Feature::UnifiedExecZshFork,
  706              ],
  707          );
  708          turn.unified_exec_shell_mode =

   38  /// Transcript of thread history
   39  #[derive(Debug, Clone, Default)]
   40  pub(crate) struct ContextManager {
   41      /// The oldest items are at the beginning of the vector. Snapshots share the vector until a
   42      /// caller needs to mutate it, avoiding deep copies for read-only history consumers.
   43      items: Arc<Vec<ResponseItem>>,
   44      /// Bumped whenever history is rewritten, such as compaction or rollback.
   45      history_version: u64,
   46      token_info: Option<TokenUsageInfo>,
   47      /// Reference context snapshot used for diffing and producing model-visible
   48      /// settings update items.
   49      ///
   50      /// This is the baseline for the next regular model turn, and may already
   51      /// match the current turn after context updates are persisted.
   52      ///
   53      /// When this is `None`, settings diffing treats the next turn as having no
   54      /// baseline and emits a full reinjection of context state. Rollback may
   55      /// also clear this when it trims a mixed initial-context developer bundle
   56      /// whose non-diff fragments no longer exist in the surviving history.
   57      reference_context_item: Option<TurnContextItem>,
   58      /// World state most recently appended to model-visible history.
   59      world_state_baseline: Option<WorldStateSnapshot>,
   60  }

  189      pub(crate) fn remove_first_item(&mut self) {
  190          if !self.items.is_empty() {
  191              // Remove the oldest item (front of the list). Items are ordered from
  192              // oldest → newest, so index 0 is the first entry recorded.
  193              let items = Arc::make_mut(&mut self.items);
  194              let removed = items.remove(0);
  195              // If the removed item participates in a call/output pair, also remove
  196              // its corresponding counterpart to keep the invariants intact without
  197              // running a full normalization pass.
  198              normalize::remove_corresponding_for(items, &removed);
  199              self.world_state_baseline = None;
  200          }
  201      }
  202  
  203      pub(crate) fn replace(&mut self, items: Vec<ResponseItem>) {
  204          self.items = Arc::new(items);
  205          self.history_version = self.history_version.saturating_add(1);
  206          self.world_state_baseline = None;
  207      }
  208  
  209      /// Drop the last `num_turns` instruction turns from this history.
  210      ///
  211      /// Instruction turns are history messages that should behave like a new prompt boundary:
  212      /// ordinary user messages and structured assistant inter-agent instructions.
  213      ///
      … 24 lines omitted; exact range 189–248 …
  238          let mut cut_idx = if n_from_end >= user_positions.len() {
  239              first_instruction_turn_idx
  240          } else {
  241              user_positions[user_positions.len() - n_from_end]
  242          };
  243  
  244          cut_idx =
  245              self.trim_pre_turn_context_updates(&snapshot, first_instruction_turn_idx, cut_idx);
  246  
  247          self.replace(snapshot[..cut_idx].to_vec());
  248      }

  240  async fn run_compact_task_inner_impl(
  241      sess: Arc<Session>,
  242      turn_context: Arc<TurnContext>,
  243      input: Vec<UserInput>,
  244      initial_context_injection: InitialContextInjection,
  245      compaction_metadata: CompactionTurnMetadata,
  246  ) -> CodexResult<String> {
  247      let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new());
  248      sess.emit_turn_item_started(&turn_context, &compaction_item)
  249          .await;
  250      let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input);
  251  
  252      let mut history = sess.clone_history().await;
  253      history.record_items(
  254          &[initial_input_for_turn.into()],
  255          turn_context.model_info.truncation_policy.into(),
  256      );
  257  
  258      let max_retries = turn_context.provider.info().stream_max_retries();
  259      let mut retries = 0;
  260      let mut client_session = sess.services.model_client.new_session();
  261      // Reuse one client session so turn-scoped state (sticky routing, websocket incremental
  262      // request tracking)
  263      // survives retries within this compact turn.
  264      let window_id = sess.current_window_id().await;
      … 43 lines omitted; exact range 240–318 …
  308              }
  309              Err(e) if matches!(e.details(), CodexErrorDetails::ContextWindowExceeded) => {
  310                  if turn_input_len > 1 {
  311                      // Trim from the beginning to preserve cache (prefix-based) and keep recent messages intact.
  312                      error!(
  313                          "Context window exceeded while compacting; removing oldest history item. Error: {e}"
  314                      );
  315                      history.remove_first_item();
  316                      retries = 0;
  317                      continue;
  318                  }

  890  /// Determines the conditions under which the user is consulted to approve
  891  /// running the command proposed by Codex.
  892  #[derive(
  893      Debug,
  894      Clone,
  895      Copy,
  896      Default,
  897      PartialEq,
  898      Eq,
  899      Hash,
  900      Serialize,
  901      Deserialize,
  902      Display,
  903      JsonSchema,
  904      TS,
  905  )]
  906  #[serde(rename_all = "kebab-case")]
  907  #[strum(serialize_all = "kebab-case")]
  908  pub enum AskForApproval {
  909      /// Under this policy, only "known safe" commands—as determined by
  910      /// `is_safe_command()`—that **only read files** are auto‑approved.
  911      /// Everything else will ask the user to approve.
  912      #[serde(rename = "untrusted")]
  913      #[strum(serialize = "untrusted")]
  914      UnlessTrusted,
      … 7 lines omitted; exact range 890–932 …
  922      ///
  923      /// When a field is `true`, commands in that category are allowed. When it
  924      /// is `false`, those requests are automatically rejected instead of shown
  925      /// to the user.
  926      #[strum(serialize = "granular")]
  927      Granular(GranularApprovalConfig),
  928  
  929      /// Never ask the user to approve commands. Failures are immediately returned
  930      /// to the model, and never escalated to the user for approval.
  931      Never,
  932  }

 2295      pub async fn request_command_approval(
 2296          &self,
 2297          turn_context: &TurnContext,
 2298          call_id: String,
 2299          approval_id: Option<String>,
 2300          environment_id: Option<String>,
 2301          command: Vec<String>,
 2302          cwd: AbsolutePathBuf,
 2303          reason: Option<String>,
 2304          network_approval_context: Option<NetworkApprovalContext>,
 2305          proposed_execpolicy_amendment: Option<ExecPolicyAmendment>,
 2306          additional_permissions: Option<AdditionalPermissionProfile>,
 2307          available_decisions: Option<Vec<ReviewDecision>>,
 2308          plugin_attribution_override: Option<PluginCommandAttribution>,
 2309      ) -> ReviewDecision {
 2310          let _elicitation = self.services.elicitations.register();
 2311          //  command-level approvals use `call_id`.
 2312          // `approval_id` is only present for subcommand callbacks (execve intercept)
 2313          let effective_approval_id = approval_id.clone().unwrap_or_else(|| call_id.clone());
 2314          // Add the tx_approve callback to the map before sending the request.
 2315          let (tx_approve, rx_approve) = oneshot::channel();
 2316          let prev_entry = {
 2317              let mut active = self.active_turn.lock().await;
 2318              match active.as_mut() {
 2319                  Some(at) => {
      … 46 lines omitted; exact range 2295–2376 …
 2366              cwd,
 2367              reason,
 2368              network_approval_context,
 2369              proposed_execpolicy_amendment,
 2370              proposed_network_policy_amendments,
 2371              additional_permissions,
 2372              available_decisions: Some(available_decisions),
 2373              parsed_cmd,
 2374          });
 2375          self.send_event(turn_context, event).await;
 2376          rx_approve.await.unwrap_or(ReviewDecision::Abort)

   34  #[derive(Clone, Copy, Debug, PartialEq, Eq)]
   35  pub enum SandboxType {
   36      None,
   37      MacosSeatbelt,
   38      LinuxSeccomp,
   39      WindowsRestrictedToken,
   40  }
   41  
   42  impl SandboxType {
   43      pub fn as_metric_tag(self) -> &'static str {
   44          match self {
   45              SandboxType::None => "none",
   46              SandboxType::MacosSeatbelt => "seatbelt",
   47              SandboxType::LinuxSeccomp => "seccomp",
   48              SandboxType::WindowsRestrictedToken => "windows_sandbox",
   49          }
   50      }
   51  }
   52  
   53  #[derive(Clone, Copy, Debug, PartialEq, Eq)]
   54  pub enum SandboxablePreference {
   55      Auto,
   56      Require,
   57      Forbid,
   58  }
      … 4 lines omitted; exact range 34–73 …
   63      } else if cfg!(target_os = "linux") {
   64          Some(SandboxType::LinuxSeccomp)
   65      } else if cfg!(target_os = "windows") {
   66          if windows_sandbox_enabled {
   67              Some(SandboxType::WindowsRestrictedToken)
   68          } else {
   69              None
   70          }
   71      } else {
   72          None
   73      }

  203  fn extensible_builtin_parent_profile(profile_name: &str) -> Option<PermissionProfileToml> {
  204      let file_system = match profile_name {
  205          BUILT_IN_READ_ONLY_PROFILE => FileSystemSandboxPolicy::read_only(),
  206          BUILT_IN_WORKSPACE_PROFILE => FileSystemSandboxPolicy::workspace_write(
  207              &[],
  208              /*exclude_tmpdir_env_var*/ false,
  209              /*exclude_slash_tmp*/ false,
  210          ),
  211          _ => return None,
  212      };
  213      Some(permission_profile_toml_from_file_system_policy(file_system))

   66  pub(crate) struct McpServerConnection {
   67      identity: Option<McpServerConnectionIdentity>,
   68      client: AsyncManagedClient,
   69  }
   70  
   71  impl McpServerConnection {
   72      async fn reusable_client(
   73          &self,
   74          desired: &McpServerConnectionIdentity,
   75      ) -> Option<ManagedClient> {
   76          let current = self.identity.as_ref()?;
   77          if !current.has_same_connection_config(desired) {
   78              return None;
   79          }
   80          if !self.client.startup_complete.load(Ordering::Acquire) {
   81              return None;
   82          }
   83          let client = self.client.client().await.ok()?;
   84          if client.client.is_closed().await {
   85              return None;
   86          }
   87          let Ok(desired_credentials) = desired.oauth_credentials() else {
   88              return Some(client);
   89          };
   90          let reusable = match client.client.managed_oauth_credentials().await {
      … 16 lines omitted; exact range 66–117 …
  107      fn cancel_startup(&self) {
  108          if !self.client.startup_complete.load(Ordering::Acquire) {
  109              self.client.cancel_token.cancel();
  110          }
  111      }
  112  }
  113  
  114  impl Drop for McpServerConnection {
  115      fn drop(&mut self) {
  116          self.client.cancel_token.cancel();
  117      }

   34  /// Returns whether a tool may be included in model-facing tool declarations.
   35  ///
   36  /// Tools without visibility metadata remain visible. Tools with visibility
   37  /// metadata are hidden unless they explicitly include `model`.
   38  ///
   39  /// <https://github.com/modelcontextprotocol/ext-apps/blob/main/specification/2026-01-26/apps.mdx#resource-discovery>
   40  pub fn tool_is_model_visible(tool: &ToolInfo) -> bool {
   41      let Some(visibility) = tool
   42          .tool
   43          .meta
   44          .as_deref()
   45          .and_then(|meta| meta.get(MCP_UI_META_KEY))
   46          .and_then(serde_json::Value::as_object)
   47          .and_then(|ui| ui.get(MCP_UI_VISIBILITY_META_KEY))
   48          .and_then(serde_json::Value::as_array)
   49      else {
   50          return true;
   51      };
   52      visibility
   53          .iter()
   54          .any(|target| target.as_str() == Some(MCP_UI_MODEL_VISIBILITY))
   55  }

    1  //! AGENTS.md discovery and user instruction assembly.
    2  //!
    3  //! Project-level documentation is primarily stored in files named `AGENTS.md`.
    4  //! Additional fallback filenames can be configured via `project_doc_fallback_filenames`.
    5  //! We include the concatenation of all files found along the path from the
    6  //! project root to the current working directory as follows:
    7  //!
    8  //! 1.  Determine the project root by walking upwards from the current working
    9  //!     directory until a configured `project_root_markers` entry is found.
   10  //!     When `project_root_markers` is unset, the default marker list is used
   11  //!     (`.git`). If no marker is found, only the current working directory is
   12  //!     considered. An empty marker list disables parent traversal.
   13  //! 2.  Collect every `AGENTS.md` found from the project root down to the
   14  //!     current working directory (inclusive) and concatenate their contents in
   15  //!     that order.
   16  //! 3.  We do **not** walk past the project root.

 3336      pub(crate) async fn build_initial_context_with_world_state(
 3337          &self,
 3338          turn_context: &TurnContext,
 3339          world_state: &WorldState,
 3340      ) -> Vec<ResponseItem> {
 3341          let mut developer_sections = Vec::<String>::with_capacity(8);
 3342          let mut contextual_user_sections = Vec::<String>::with_capacity(2);
 3343          let mut separate_developer_sections = Vec::<String>::new();
 3344          let (session_source, auto_compact_window_ids) = {
 3345              let state = self.state.lock().await;
 3346              (
 3347                  state.session_configuration.session_source.clone(),
 3348                  state.auto_compact_window_ids(),
 3349              )
 3350          };
 3351          let separate_guardian_developer_message =
 3352              crate::guardian::is_guardian_reviewer_source(&session_source);
 3353          // Keep the guardian policy prompt out of the aggregated developer bundle so it
 3354          // stays isolated as its own top-level developer message for guardian subagents.
 3355          if !separate_guardian_developer_message
 3356              && let Some(developer_instructions) = turn_context.developer_instructions.as_deref()
 3357              && !developer_instructions.is_empty()
 3358          {
 3359              developer_sections.push(developer_instructions.to_string());
 3360          }
      … 26 lines omitted; exact range 3336–3397 …
 3387                          msg: EventMsg::Warning(WarningEvent {
 3388                              message: warning_message,
 3389                          }),
 3390                      })
 3391                      .await;
 3392                  }
 3393                  if !host_catalog_in_world_state {
 3394                      developer_sections.push(skills_instructions.render());
 3395                  }
 3396              }
 3397          }

  103  pub(crate) async fn run_pending_session_start_hooks(
  104      sess: &Arc<Session>,
  105      turn_context: &Arc<TurnContext>,
  106  ) -> bool {
  107      while let Some(session_start_source) = sess.take_pending_session_start_source().await {
  108          // Pending session-start hooks are reused to dispatch thread-spawn subagent
  109          // starts. Other subagent sessions are internal/system work and do not run
  110          // start hooks.
  111          let target = match &turn_context.session_source {
  112              SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_role, .. })
  113                  if matches!(
  114                      session_start_source,
  115                      codex_hooks::SessionStartSource::Startup
  116                  ) =>
  117              {
  118                  let context = subagent_hook_context(sess, agent_role);
  119                  StartHookTarget::SubagentStart {
  120                      turn_id: turn_context.sub_id.clone(),
  121                      agent_id: context.agent_id,
  122                      agent_type: context.agent_type,
  123                  }
  124              }
  125              SessionSource::SubAgent(_) => return false,
  126              _ => StartHookTarget::SessionStart {
  127                  source: session_start_source,
      … 82 lines omitted; exact range 103–220 …
  210      {
  211          PreToolUseHookResult::Blocked(format!(
  212              "Command blocked by PreToolUse hook: {reason}. Command: {command}"
  213          ))
  214      } else {
  215          PreToolUseHookResult::Blocked(format!(
  216              "Tool call blocked by PreToolUse hook: {reason}. Tool: {}",
  217              tool_name.name()
  218          ))
  219      }
  220  }

   70      pub(crate) fork_parent_spawn_call_id: Option<String>,
   71      pub(crate) fork_mode: Option<SpawnAgentForkMode>,
   72      pub(crate) parent_thread_id: Option<ThreadId>,
   73      pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
   74  }
   75  
   76  #[derive(Clone, Debug)]
   77  pub(crate) struct LiveAgent {
   78      pub(crate) thread_id: ThreadId,
   79      pub(crate) metadata: AgentMetadata,
   80      pub(crate) status: AgentStatus,
   81  }
   82  
   83  #[derive(Clone, Debug, Serialize, PartialEq, Eq)]
   84  pub(crate) struct ListedAgent {
   85      pub(crate) agent_name: String,
   86      pub(crate) agent_status: AgentStatus,
   87  }
   88  
   89  /// Control-plane handle for multi-agent operations.
   90  /// `AgentControl` is held by each session (via `SessionServices`). It provides capability to
   91  /// spawn new agents and the inter-agent communication layer.
   92  /// An `AgentControl` instance is intended to be created at most once per root thread/session
   93  /// tree. That same `AgentControl` is then shared with every sub-agent spawned from that root,
   94  /// which keeps the registry scoped to that root thread rather than the entire `ThreadManager`.
      … 6 lines omitted; exact range 70–111 …
  101      /// This is `Weak` to avoid reference cycles and shadow persistence of the form
  102      /// `ThreadManagerState -> CodexThread -> Session -> SessionServices -> ThreadManagerState`.
  103      manager: Weak<ThreadManagerState>,
  104      state: Arc<AgentRegistry>,
  105      v2_residency: Arc<V2Residency>,
  106      agent_execution_limiter: Arc<AgentExecutionLimiter>,
  107      /// Session-scoped state shared by the root thread and every cloned sub-agent control handle.
  108      rollout_budget: Arc<RolloutBudget>,
  109  }
  110  
  111  impl AgentControl {

   39  async fn handle_spawn_agent(
   40      invocation: ToolInvocation,
   41  ) -> Result<SpawnAgentResult, FunctionCallError> {
   42      let ToolInvocation {
   43          session,
   44          turn,
   45          payload,
   46          call_id,
   47          ..
   48      } = invocation;
   49      let arguments = function_arguments(payload)?;
   50      let args: SpawnAgentArgs = parse_arguments(&arguments)?;
   51      let fork_mode = args.fork_mode()?;
   52      let role_name = args
   53          .agent_type
   54          .as_deref()
   55          .map(str::trim)
   56          .filter(|role| !role.is_empty());
   57  
   58      let message = message_content(args.message)?;
   59      let session_source = turn.session_source.clone();
   60      let child_depth = next_thread_spawn_depth(&session_source);
   61      let mut config =
   62          build_agent_spawn_config(&session.get_base_instructions().await, turn.as_ref())?;
   63      if let Some(service_tier) = args.service_tier.as_ref() {
      … 91 lines omitted; exact range 39–165 …
  155  
  156      let hide_agent_metadata = turn.config.multi_agent_v2.hide_spawn_agent_metadata;
  157      if hide_agent_metadata {
  158          Ok(SpawnAgentResult::HiddenMetadata { task_name })
  159      } else {
  160          Ok(SpawnAgentResult::WithNickname {
  161              task_name,
  162              nickname,
  163          })
  164      }
  165  }

 1547      pub(crate) fn effective_agent_max_threads(
 1548          &self,
 1549          multi_agent_version: MultiAgentVersion,
 1550      ) -> Option<usize> {
 1551          match multi_agent_version {
 1552              MultiAgentVersion::V2 => Some(
 1553                  self.multi_agent_v2
 1554                      .max_concurrent_threads_per_session
 1555                      .saturating_sub(1),
 1556              ),
 1557              MultiAgentVersion::Disabled | MultiAgentVersion::V1 => {
 1558                  self.agent_max_threads.or(DEFAULT_AGENT_MAX_THREADS)
 1559              }
 1560          }

   93  pub enum RolloutRecorderParams {
   94      Create {
   95          session_id: SessionId,
   96          conversation_id: ThreadId,
   97          forked_from_id: Option<ThreadId>,
   98          parent_thread_id: Option<ThreadId>,
   99          source: Box<SessionSource>,
  100          thread_source: Option<ThreadSource>,
  101          originator: String,
  102          base_instructions: BaseInstructions,
  103          dynamic_tools: Vec<DynamicToolSpec>,
  104          selected_capability_roots: Vec<SelectedCapabilityRoot>,
  105          multi_agent_version: Option<MultiAgentVersion>,
  106          history_mode: ThreadHistoryMode,
  107          history_base: Option<HistoryPosition>,
  108          subagent_history_start_ordinal: Option<u64>,
  109          initial_window_id: Option<String>,
  110      },
  111      Resume {
  112          path: PathBuf,
  113      },
  114  }
  115  
  116  enum RolloutCmd {
  117      AddItems(Vec<RolloutItem>),
      … 43 lines omitted; exact range 93–171 …
  161      }
  162  
  163      /// Return the terminal writer-task failure, if the task exited with an error.
  164      fn terminal_failure(&self) -> Option<IoError> {
  165          let guard = self
  166              .terminal_failure
  167              .lock()
  168              .unwrap_or_else(std::sync::PoisonError::into_inner);
  169          guard.as_ref().map(|err| clone_io_error(err.as_ref()))
  170      }
  171  }

    1  //! SQLite-backed state for rollout metadata.
    2  //!
    3  //! This crate is intentionally small and focused: it extracts rollout metadata
    4  //! from JSONL rollouts and mirrors it into a local SQLite database. Backfill
    5  //! orchestration and rollout scanning live in `codex-core`.
    6  
    7  const _: () = assert!(
    8      libsqlite3_sys::SQLITE_VERSION_NUMBER >= 3_051_003,
    9      "bundled SQLite must include the WAL-reset corruption fix",
   10  );

   74  use codex_otel::SessionTelemetry;
   75  use codex_otel::current_span_w3c_trace_context;
   76  use codex_protocol::auth::AuthMode;
   77  
   78  use codex_protocol::ThreadId;
   79  use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
   80  use codex_protocol::config_types::Verbosity as VerbosityConfig;
   81  use codex_protocol::models::ContentItem;
   82  use codex_protocol::models::ResponseItem;
   83  use codex_protocol::openai_models::ModelInfo;
   84  use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
   85  use codex_protocol::protocol::InternalSessionSource;
   86  use codex_protocol::protocol::SessionSource;
   87  use codex_protocol::protocol::W3cTraceContext;
   88  use codex_rollout_trace::CompactionTraceContext;
   89  use codex_rollout_trace::InferenceTraceAttempt;
   90  use codex_rollout_trace::InferenceTraceContext;
   91  use codex_tools::create_tools_json_for_responses_api;

    1                                   Apache License
    2                             Version 2.0, January 2004
    3                          http://www.apache.org/licenses/
    4  
    5  TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
    6  
    7  1.  Definitions.
    8  
    9      "License" shall mean the terms and conditions for use, reproduction,
   10      and distribution as defined by Sections 1 through 9 of this document.
   11  
   12      "Licensor" shall mean the copyright owner or entity authorized by
   13      the copyright owner that is granting the License.
   14  
   15      "Legal Entity" shall mean the union of the acting entity and all
   16      other entities that control, are controlled by, or are under common
   17      control with that entity. For the purposes of this definition,
   18      "control" means (i) the power, direct or indirect, to cause the
   19      direction or management of such entity, whether by contract or
   20      otherwise, or (ii) ownership of fifty percent (50%) or more of the
   21      outstanding shares, or (iii) beneficial ownership of such entity.
   22  
   23      "You" (or "Your") shall mean an individual or Legal Entity
   24      exercising permissions granted by this License.
   25  
   26      "Source" form shall mean the preferred form for making modifications,
   27      including but not limited to software source code, documentation
   28      source, and configuration files.