feat(agents): add watchdog and forked-agent support

Add watchdog scheduling and forked-agent spawn/resume support across the agent control plane, schemas, rollout metadata, and TUI surfaces.

Rolls up follow-up fixes for watchdog lifecycle behavior and guidance: helper cleanup/one-shot enforcement, wait semantics for watchdog check-in agents, and clarified tool descriptions so models align with runtime behavior.
This commit is contained in:
Friel
2026-02-24 14:05:06 -08:00
parent fefdc03b25
commit 993d3874e1
41 changed files with 5727 additions and 313 deletions
@@ -1286,6 +1286,14 @@
],
"type": "object"
},
"CollabAgentSpawnMode": {
"enum": [
"spawn",
"fork",
"watchdog"
],
"type": "string"
},
"CommandExecutionApprovalDecision": {
"oneOf": [
{
@@ -3844,6 +3852,15 @@
],
"description": "Thread ID of the sender."
},
"spawn_mode": {
"allOf": [
{
"$ref": "#/definitions/CollabAgentSpawnMode"
}
],
"default": "spawn",
"description": "Spawn mode used for this agent."
},
"status": {
"allOf": [
{
@@ -2,6 +2,7 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { AgentStatus } from "./AgentStatus";
import type { CollabAgentSpawnMode } from "./CollabAgentSpawnMode";
import type { ThreadId } from "./ThreadId";
export type CollabAgentSpawnEndEvent = {
@@ -29,7 +30,11 @@ new_agent_role?: string | null,
* Initial prompt sent to the agent. Can be empty to prevent CoT leaking at the
* beginning.
*/
prompt: string,
prompt: string,
/**
* Spawn mode used for this agent.
*/
spawn_mode: CollabAgentSpawnMode,
/**
* Last known status of the new agent reported to the sender agent.
*/
@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type CollabAgentSpawnMode = "spawn" | "fork" | "watchdog";
@@ -166,6 +166,7 @@ impl ThreadHistoryBuilder {
RolloutItem::Compacted(payload) => self.handle_compacted(payload),
RolloutItem::TurnContext(_)
| RolloutItem::SessionMeta(_)
| RolloutItem::ForkReference(_)
| RolloutItem::ResponseItem(_) => {}
}
}
+35
View File
@@ -6,6 +6,13 @@
"description": "A path that is guaranteed to be absolute and normalized (though it is not guaranteed to be canonicalized or exist on the filesystem).\n\nIMPORTANT: When deserializing an `AbsolutePathBuf`, a base path must be set using [AbsolutePathBufGuard::new]. If no base path is set, the deserialization will fail unless the path being deserialized is already absolute.",
"type": "string"
},
"AgentRoleSpawnMode": {
"enum": [
"spawn",
"fork"
],
"type": "string"
},
"AgentRoleToml": {
"additionalProperties": false,
"properties": {
@@ -20,6 +27,18 @@
"description": {
"description": "Human-facing role documentation used in spawn tool guidance.",
"type": "string"
},
"model": {
"description": "Optional model override applied by this role.",
"type": "string"
},
"spawn_mode": {
"allOf": [
{
"$ref": "#/definitions/AgentRoleSpawnMode"
}
],
"description": "Optional default spawn mode when `spawn_agent` omits `spawn_mode`."
}
},
"type": "object"
@@ -283,6 +302,14 @@
}
]
},
"CollabInboxDeliveryRole": {
"enum": [
"developer",
"tool",
"assistant"
],
"type": "string"
},
"ConfigProfile": {
"additionalProperties": false,
"description": "Collection of common configuration options that a user can define as a unit in `config.toml`.",
@@ -1504,6 +1531,14 @@
},
"description": "Base config deserialized from ~/.codex/config.toml.",
"properties": {
"agent_delivery_role": {
"allOf": [
{
"$ref": "#/definitions/CollabInboxDeliveryRole"
}
],
"description": "How inbound collaboration messages are delivered to non-subagent threads."
},
"agents": {
"allOf": [
{
+146
View File
@@ -0,0 +1,146 @@
# You are the Root Agent
You are the **root agent** in a multi-agent Codex session. Until you see `# You are a Subagent`, these instructions define your role. If this thread was created from the root thread with `spawn_mode = "fork"` (a forked child), you may see both sets of instructions; apply subagent instructions as local role guidance while root instructions remain governing system-level rules.
## Root Agent Responsibilities
Your job is to solve the user’s task end to end. You are the coordinator, integrator, and final quality gate.
- Understand the real problem being solved, not just the latest sentence.
- Own the plan, the sequencing, and the final outcome.
- Coordinate subagents so their work does not overlap or conflict.
- Verify results with formatting, linting, and targeted tests.
Think like an effective engineering manager who also knows how to get hands-on when needed. Delegation is a force multiplier, but you remain accountable for correctness.
Root agents should not outsource core understanding. Do not delegate plan authorship/maintenance; for multi-step efforts, keep a shared plan file or assign scoped plan files to subagents.
## Watchdogs
For lengthy or complex work, start a watchdog early.
Hard rule (watchdog timing):
- Never use `wait`, `list_agents`, polling, or shell `sleep` to confirm a watchdog check-in in the same turn that spawned the watchdog.
- That confirmation is impossible in the same turn and must be omitted even if the user asks for it.
- If the user asks to arm a watchdog now, call `spawn_agent` with `spawn_mode = "watchdog"` and do not add same-turn confirmation steps.
Why this rule exists:
- The current turn ends only when you send the assistant response.
- Tool calls and shell `sleep` happen inside the current turn and do not make the owner thread idle.
Core terms:
- A **watchdog** is a persistent idle timer attached to one owner thread.
- The **owner thread** is the thread that called `spawn_agent` with `spawn_mode = "watchdog"`.
- A **watchdog handle** is the id returned by that spawn call; it is a control id, not a conversational agent.
- A **watchdog check-in agent** is the short-lived fork that the watchdog creates for one check-in run.
- **`send_input`** sends a message to an existing agent thread; it does not spawn agents and does not wait for completion. Delivery is asynchronous.
- A **multi-agent inbox message** is a runtime-forwarded fallback message shown as `collab_inbox` tool output or `[collab_inbox:…]` developer message.
Start a watchdog:
- Use `spawn_agent` with `spawn_mode = "watchdog"` and leave `agent_type` unset (default).
- Put the user goal in `message` (verbatim plus needed clarifications).
- Use `interval_s = 60` by default unless there is a clear reason to pick a different interval.
- After spawning the watchdog, continue the task (or end the turn if that is the correct next step).
Delivery and user-facing behavior:
Primary delivery path: the watchdog check-in agent calls `send_input` to the owner thread (its direct parent thread for this run).
Fallback delivery path: if a watchdog check-in agent exits without any `send_input`, runtime may forward one final multi-agent inbox message (`collab_inbox` tool output or `[collab_inbox:…]` developer message). This fallback is best-effort and not guaranteed.
- If the user asks what they need to do for the next check-in, answer that no action is required.
- Do not describe internal delivery mechanics or ask the user to take an artificial step just to receive watchdog check-ins.
Operational notes:
- Do not call `send_input` on watchdog handles.
- If every id passed to `wait` is a watchdog handle, `wait` returns an immediate correction; this does not mean a new check-in happened.
- The tool returns a watchdog handle ID. When you no longer need the watchdog, stop it by calling `close_agent` on that handle ID.
Treat watchdog guidance as high-priority execution feedback. If it reveals a missing required action, do that action before status narration while honoring higher-priority system/developer/user constraints. A required action is one needed to satisfy the user request or clear a concrete blocker.
Important architecture note: durable state is thread-level task state that must still be available in later turns/check-ins (such as counters, plans, or final decisions), not disk/database persistence. Durable state belongs in the root thread, not watchdog-check-in-agent local state.
## Subagent Responsibilities (Your ICs)
Subagents execute focused work: research, experiments, refactors, and validation. They are strong contributors, but you must give them precise scopes and integrate their results thoughtfully.
Subagents can become confused if the world changes while they are idle. Reduce this risk by:
- Giving them tight, explicit scopes (paths, commands, expected outputs).
- Providing updates when you change course.
- Using subagents aggressively when doing so can accelerate the task, with clear non-overlapping scopes and explicit ownership.
## Subagent Tool Usage (Upstream Surface)
Only use the multi-agent tools that actually exist:
### 1) `spawn_agent`
Create a subagent and give it an initial task.
Parameters:
- `message` (required): the task description.
- `agent_type` (optional): the role to assign (`default`, `explorer`, `fast-worker`, or `worker`).
- `spawn_mode` (optional): one of `spawn`, `fork`, or `watchdog`.
- `interval_s` (optional): watchdog interval in seconds when `spawn_mode = "watchdog"`.
Guidance:
- When `spawn_mode` is omitted, the default is `fork` unless the selected role overrides it.
- Use `agent_type = "explorer"` for specific codebase questions; it defaults to context-free `spawn`.
- Use `agent_type = "fast-worker"` for tightly constrained execution work that can run from a self-contained prompt; it also defaults to context-free `spawn`.
- Use `agent_type = "worker"` for broader implementation work that should inherit current-thread context; it defaults to `fork`.
- Choose `fork` vs `spawn` by context requirements first (not by task shape).
- Use `spawn_mode = "fork"` when the child should preserve your current conversation history and rely on current-thread context, including:
- current debugging-thread relevance (for example, "summarize only failures relevant to this investigation")
- active plan / ExecPlan branch continuation
- recent user decisions, tradeoffs, or rejected approaches
- parallel review work that should inherit the same context automatically
- Use `spawn_mode = "spawn"` only when the child can do the task correctly from a fresh prompt you provide now, without needing current-thread context.
- For `spawn`, make the task, inputs, and expected output explicit (especially for independent, output-heavy work where you want the child to distill results and keep the root thread context clean).
- Needle-in-a-haystack searches are strong `spawn` candidates when the child can search from a precise prompt without current-thread context.
- Do not choose `spawn` solely because work is output-heavy or command-heavy if it still depends on current-thread context.
- Use `spawn_mode = "watchdog"` for long-running work that needs periodic oversight.
- When using `spawn_mode = "watchdog"`, keep `agent_type` at the default.
### 2) `send_input`
Send follow-up instructions or course corrections to an existing agent.
Guidance:
- Use `interrupt = true` sparingly. Prefer to let agents complete coherent chunks of work.
- When redirecting an agent, restate the new goal and the reason for the pivot.
- Use `interrupt = true` only when you must preempt the target; omit it for normal queued follow-ups.
- Subagents can call `send_input` without an `id` (or with `id = "parent"` / `id = "root"`). In this runtime those forms resolve to the immediate parent thread.
- Treat explicit `send_input` deliveries as the primary path and multi-agent inbox messages (`collab_inbox` tool calls or `[collab_inbox:…]` messages) as fallback inbound agent messages.
### 3) `wait`
Wait for one or more agents to complete or report status.
Guidance:
- You do not need to wait after every spawn. Do useful parallel work, then wait when you need results.
- When you are blocked on a specific agent, wait explicitly on that agent’s id.
- If `wait` includes watchdog handles, it reports their current status but does not block on them.
- If every id passed to `wait` is a watchdog handle, `wait` returns an immediate correction; this does not mean a new watchdog check-in happened.
- Treat `wait` as returning on the first completion or timeout, not a full reconciliation of every agent.
- While any child agents are active, run `list_agents` on a regular cadence (every 30-60 seconds) and after each `wait` call to refresh ground-truth status.
- Keep an explicit set of outstanding agent ids. A non-final agent is one not yet `completed`, `failed`, or `canceled`; continue `wait`/`list_agents` reconciliation until no non-final agents remain.
### 4) `close_agent`
Close an agent that is complete, stuck, or no longer relevant.
Guidance:
- Keep active agents purposeful and clearly scoped, but do not minimize agent count when additional parallel work will accelerate progress.
- Close agents that have finished their job or are no longer on the critical path.
## Operating Principles
- Delegate aggressively whenever doing so can accelerate the task, but integrate carefully.
- Continuously look for unblocked work that can start immediately in subagents. Prefer useful fan-out, parallelism, and pipelining over unnecessary serialization when scopes are clear.
- Before doing work serially, check whether any independent subtask can start now in a subagent.
- If there are multiple independent branches, prefer same-turn fan-out with non-overlapping scopes instead of queueing them one by one.
- Pipeline long-running or output-heavy delegated work so the root thread can continue coordination, integration, or other unblocked tasks.
- Prefer clear, explicit instructions over cleverness.
- Prefer execution over hypothetical narration. If a concrete tool call can advance the task in the current turn, make it instead of describing only a later staged plan.
- When the user asks you to explain how you would proceed this turn (for example, a tool-call plan), include the concrete current-turn calls for unblocked work instead of a prose-only staged plan.
- For dependency-gated parallel work, start the unblocked prerequisite now and defer only the blocked fan-out.
- When you receive subagent output, verify it before relying on it.
- Do not reference tools outside the upstream multi-agent surface.
@@ -0,0 +1 @@
model_reasoning_effort = "medium"
File diff suppressed because it is too large Load Diff
+8
View File
@@ -52,6 +52,14 @@ pub(crate) fn exceeds_thread_spawn_depth_limit(depth: i32, max_depth: i32) -> bo
}
impl Guards {
pub(crate) fn tracked_thread_ids(&self) -> Vec<ThreadId> {
let active_agents = self
.active_agents
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
active_agents.threads_set.iter().copied().collect()
}
pub(crate) fn reserve_spawn_slot(
self: &Arc<Self>,
max_threads: Option<usize>,
+5
View File
@@ -2,10 +2,15 @@ pub(crate) mod control;
mod guards;
pub(crate) mod role;
pub(crate) mod status;
mod watchdog;
pub(crate) use codex_protocol::protocol::AgentStatus;
pub(crate) use control::AgentControl;
pub(crate) use control::WatchdogParentCompactionResult;
pub(crate) use guards::exceeds_thread_spawn_depth_limit;
pub(crate) use guards::max_thread_spawn_depth;
pub(crate) use guards::next_thread_spawn_depth;
pub(crate) use status::agent_status_from_event;
pub(crate) use watchdog::DEFAULT_WATCHDOG_INTERVAL_S;
pub(crate) use watchdog::RemovedWatchdog;
pub(crate) use watchdog::WatchdogRegistration;
+166 -49
View File
@@ -1,4 +1,5 @@
use crate::config::AgentRoleConfig;
use crate::config::AgentRoleSpawnMode;
use crate::config::Config;
use crate::config::ConfigOverrides;
use crate::config::deserialize_config_toml_with_base;
@@ -16,50 +17,80 @@ use toml::Value as TomlValue;
pub const DEFAULT_ROLE_NAME: &str = "default";
const AGENT_TYPE_UNAVAILABLE_ERROR: &str = "agent type is currently not available";
fn resolve_role_config(
user_defined_roles: &BTreeMap<String, AgentRoleConfig>,
role_name: Option<&str>,
) -> Result<(AgentRoleConfig, bool), String> {
let role_name = role_name.unwrap_or(DEFAULT_ROLE_NAME);
user_defined_roles
.get(role_name)
.cloned()
.map(|role| (role, false))
.or_else(|| {
built_in::configs()
.get(role_name)
.cloned()
.map(|role| (role, true))
})
.ok_or_else(|| format!("unknown agent_type '{role_name}'"))
}
pub(crate) fn default_spawn_mode_for_role(
user_defined_roles: &BTreeMap<String, AgentRoleConfig>,
role_name: Option<&str>,
) -> AgentRoleSpawnMode {
resolve_role_config(user_defined_roles, role_name)
.ok()
.and_then(|(role, _)| role.spawn_mode)
.unwrap_or_default()
}
/// Applies a role config layer to a mutable config and preserves unspecified keys.
pub(crate) async fn apply_role_to_config(
config: &mut Config,
role_name: Option<&str>,
) -> Result<(), String> {
let role_name = role_name.unwrap_or(DEFAULT_ROLE_NAME);
let (config_file, is_built_in) = config
.agent_roles
.get(role_name)
.map(|role| (&role.config_file, false))
.or_else(|| {
built_in::configs()
.get(role_name)
.map(|role| (&role.config_file, true))
})
.ok_or_else(|| format!("unknown agent_type '{role_name}'"))?;
let Some(config_file) = config_file.as_ref() else {
return Ok(());
};
let (role, is_built_in) = resolve_role_config(&config.agent_roles, role_name)?;
let mut role_layer_toml = if let Some(config_file) = role.config_file.as_ref() {
let (role_config_contents, role_config_base) = if is_built_in {
(
built_in::config_file_contents(config_file)
.map(str::to_owned)
.ok_or_else(|| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?,
config.codex_home.as_path(),
)
} else {
(
tokio::fs::read_to_string(config_file)
.await
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?,
config_file
.parent()
.ok_or_else(|| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?,
)
};
let (role_config_contents, role_config_base) = if is_built_in {
(
built_in::config_file_contents(config_file)
.map(str::to_owned)
.ok_or_else(|| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?,
config.codex_home.as_path(),
)
let role_config_toml: TomlValue = toml::from_str(&role_config_contents)
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?;
deserialize_config_toml_with_base(role_config_toml.clone(), role_config_base)
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?;
resolve_relative_paths_in_config_toml(role_config_toml, role_config_base)
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?
} else {
(
tokio::fs::read_to_string(config_file)
.await
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?,
config_file
.parent()
.ok_or_else(|| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?,
)
TomlValue::Table(Default::default())
};
let role_config_toml: TomlValue = toml::from_str(&role_config_contents)
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?;
deserialize_config_toml_with_base(role_config_toml.clone(), role_config_base)
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?;
let role_layer_toml = resolve_relative_paths_in_config_toml(role_config_toml, role_config_base)
.map_err(|_| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?;
if let Some(model) = role.model {
let table = role_layer_toml
.as_table_mut()
.ok_or_else(|| AGENT_TYPE_UNAVAILABLE_ERROR.to_string())?;
table.insert("model".to_string(), TomlValue::String(model));
}
if role_layer_toml
.as_table()
.is_some_and(toml::map::Map::is_empty)
{
return Ok(());
}
let mut layers: Vec<ConfigLayerEntry> = config
.config_layer_stack
@@ -136,10 +167,23 @@ Available roles:
}
fn format_role(name: &str, declaration: &AgentRoleConfig) -> String {
if let Some(description) = &declaration.description {
format!("{name}: {{\n{description}\n}}")
} else {
let has_inline_metadata = declaration.model.is_some() || declaration.spawn_mode.is_some();
if declaration.description.is_none() && !has_inline_metadata {
format!("{name}: no description")
} else {
let mut body = Vec::new();
if let Some(description) = &declaration.description {
body.push(description.clone());
}
let default_spawn_mode = match declaration.spawn_mode.unwrap_or_default() {
AgentRoleSpawnMode::Spawn => "spawn",
AgentRoleSpawnMode::Fork => "fork",
};
body.push(format!("Default spawn mode: {default_spawn_mode}"));
if let Some(model) = &declaration.model {
body.push(format!("Model override: {model}"));
}
format!("{name}: {{\n{}\n}}", body.join("\n"))
}
}
}
@@ -155,7 +199,7 @@ mod built_in {
DEFAULT_ROLE_NAME.to_string(),
AgentRoleConfig {
description: Some("Default agent.".to_string()),
config_file: None,
..Default::default()
}
),
(
@@ -169,9 +213,28 @@ Rules:
- Trust explorer results without verification.
- Run explorers in parallel when useful.
- Reuse existing explorers for related questions."#.to_string()),
model: Some("gpt-5.3-codex-spark".to_string()),
spawn_mode: Some(AgentRoleSpawnMode::Spawn),
config_file: Some("explorer.toml".to_string().parse().unwrap_or_default()),
}
),
(
"fast-worker".to_string(),
AgentRoleConfig {
description: Some(r#"Use `fast-worker` for tightly constrained problems.
Typical tasks:
- Make a small, localized code change
- Execute a narrowly scoped command sequence
- Handle an isolated fix from a self-contained prompt
Rules:
- Keep scope tight and avoid broad repo exploration.
- Treat the prompt as self-contained and do not assume shared context.
- Prefer direct execution over extended analysis."#.to_string()),
model: Some("gpt-5.3-codex-spark".to_string()),
spawn_mode: Some(AgentRoleSpawnMode::Spawn),
..Default::default()
}
),
(
"worker".to_string(),
AgentRoleConfig {
@@ -183,7 +246,7 @@ Typical tasks:
Rules:
- Explicitly assign **ownership** of the task (files / responsibility).
- Always tell workers they are **not alone in the codebase**, and they should ignore edits made by others without touching them."#.to_string()),
config_file: None,
..Default::default()
}
),
(
@@ -196,7 +259,9 @@ This includes, but not only:
* explicit ask to wait for something
When YOU wait for the `awaiter` agent to be done, use the largest possible timeout."#.to_string()),
spawn_mode: Some(AgentRoleSpawnMode::Spawn),
config_file: Some("awaiter.toml".to_string().parse().unwrap_or_default()),
..Default::default()
}
)
])
@@ -282,7 +347,6 @@ mod tests {
}
#[tokio::test]
#[ignore = "No role requiring it for now"]
async fn apply_explorer_role_sets_model_and_adds_session_flags_layer() {
let (_home, mut config) = test_config_with_cli_overrides(Vec::new()).await;
let before_layers = session_flags_layer_count(&config);
@@ -291,11 +355,24 @@ mod tests {
.await
.expect("explorer role should apply");
assert_eq!(config.model.as_deref(), Some("gpt-5.1-codex-mini"));
assert_eq!(config.model.as_deref(), Some("gpt-5.3-codex-spark"));
assert_eq!(config.model_reasoning_effort, Some(ReasoningEffort::Medium));
assert_eq!(session_flags_layer_count(&config), before_layers + 1);
}
#[tokio::test]
async fn apply_fast_worker_role_sets_model_and_adds_session_flags_layer() {
let (_home, mut config) = test_config_with_cli_overrides(Vec::new()).await;
let before_layers = session_flags_layer_count(&config);
apply_role_to_config(&mut config, Some("fast-worker"))
.await
.expect("fast-worker role should apply");
assert_eq!(config.model.as_deref(), Some("gpt-5.3-codex-spark"));
assert_eq!(session_flags_layer_count(&config), before_layers + 1);
}
#[tokio::test]
async fn apply_role_returns_unavailable_for_missing_user_role_file() {
let (_home, mut config) = test_config_with_cli_overrides(Vec::new()).await;
@@ -304,6 +381,7 @@ mod tests {
AgentRoleConfig {
description: None,
config_file: Some(PathBuf::from("/path/does/not/exist.toml")),
..Default::default()
},
);
@@ -323,6 +401,7 @@ mod tests {
AgentRoleConfig {
description: None,
config_file: Some(role_path),
..Default::default()
},
);
@@ -351,6 +430,7 @@ mod tests {
AgentRoleConfig {
description: None,
config_file: Some(role_path),
..Default::default()
},
);
@@ -390,6 +470,7 @@ writable_roots = ["./sandbox-root"]
AgentRoleConfig {
description: None,
config_file: Some(role_path),
..Default::default()
},
);
@@ -443,6 +524,7 @@ writable_roots = ["./sandbox-root"]
AgentRoleConfig {
description: None,
config_file: Some(role_path),
..Default::default()
},
);
@@ -454,6 +536,38 @@ writable_roots = ["./sandbox-root"]
assert_eq!(session_flags_layer_count(&config), before_layers + 1);
}
#[test]
fn default_spawn_mode_for_role_defaults_to_fork_and_honors_role_overrides() {
let user_defined_roles = BTreeMap::from([(
"researcher".to_string(),
AgentRoleConfig {
spawn_mode: Some(AgentRoleSpawnMode::Spawn),
..Default::default()
},
)]);
assert_eq!(
default_spawn_mode_for_role(&user_defined_roles, None),
AgentRoleSpawnMode::Fork
);
assert_eq!(
default_spawn_mode_for_role(&user_defined_roles, Some("worker")),
AgentRoleSpawnMode::Fork
);
assert_eq!(
default_spawn_mode_for_role(&user_defined_roles, Some("explorer")),
AgentRoleSpawnMode::Spawn
);
assert_eq!(
default_spawn_mode_for_role(&user_defined_roles, Some("fast-worker")),
AgentRoleSpawnMode::Spawn
);
assert_eq!(
default_spawn_mode_for_role(&user_defined_roles, Some("researcher")),
AgentRoleSpawnMode::Spawn
);
}
#[test]
fn spawn_tool_spec_build_deduplicates_user_defined_built_in_roles() {
let user_defined_roles = BTreeMap::from([
@@ -461,7 +575,7 @@ writable_roots = ["./sandbox-root"]
"explorer".to_string(),
AgentRoleConfig {
description: Some("user override".to_string()),
config_file: None,
..Default::default()
},
),
("researcher".to_string(), AgentRoleConfig::default()),
@@ -470,8 +584,9 @@ writable_roots = ["./sandbox-root"]
let spec = spawn_tool_spec::build(&user_defined_roles);
assert!(spec.contains("researcher: no description"));
assert!(spec.contains("explorer: {\nuser override\n}"));
assert!(spec.contains("default: {\nDefault agent.\n}"));
assert!(spec.contains("explorer: {\nuser override\nDefault spawn mode: fork\n}"));
assert!(spec.contains("default: {\nDefault agent.\nDefault spawn mode: fork\n}"));
assert!(spec.contains("fast-worker: {"));
assert!(!spec.contains("Explorers are fast and authoritative."));
}
@@ -481,14 +596,16 @@ writable_roots = ["./sandbox-root"]
"aaa".to_string(),
AgentRoleConfig {
description: Some("first".to_string()),
config_file: None,
..Default::default()
},
)]);
let spec = spawn_tool_spec::build(&user_defined_roles);
let user_index = spec.find("aaa: {\nfirst\n}").expect("find user role");
let user_index = spec
.find("aaa: {\nfirst\nDefault spawn mode: fork\n}")
.expect("find user role");
let built_in_index = spec
.find("default: {\nDefault agent.\n}")
.find("default: {\nDefault agent.\nDefault spawn mode: fork\n}")
.expect("find built-in role");
assert!(user_index < built_in_index);
+566
View File
@@ -0,0 +1,566 @@
use super::control::AgentControl;
use super::guards::Guards;
use super::guards::exceeds_thread_spawn_depth_limit;
use super::status::is_final;
use crate::codex::load_watchdog_prompt;
use crate::config::Config;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
use crate::thread_manager::ThreadManagerState;
use codex_protocol::ThreadId;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::Weak;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use tokio::sync::Mutex;
use tokio::time::Duration;
use tokio::time::Instant;
use tracing::info;
use tracing::warn;
pub(crate) const DEFAULT_WATCHDOG_INTERVAL_S: i64 = 60;
// Tick frequently enough to respect short intervals (e.g., 1s demos).
const WATCHDOG_TICK_SECONDS: i64 = 1;
#[derive(Clone)]
pub(crate) struct WatchdogRegistration {
pub(crate) owner_thread_id: ThreadId,
pub(crate) target_thread_id: ThreadId,
pub(crate) child_depth: i32,
pub(crate) interval_s: i64,
pub(crate) prompt: String,
pub(crate) config: Config,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct RemovedWatchdog {
pub(crate) target_thread_id: ThreadId,
pub(crate) active_helper_id: Option<ThreadId>,
}
struct WatchdogEntry {
registration: WatchdogRegistration,
interval: Duration,
last_trigger: Instant,
active_helper_id: Option<ThreadId>,
owner_idle_since: Option<Instant>,
owner_was_running: bool,
force_due_once: bool,
generation: i64,
}
pub(crate) struct WatchdogManager {
manager: Weak<ThreadManagerState>,
guards: Arc<Guards>,
registrations: Mutex<HashMap<ThreadId, WatchdogEntry>>,
started: AtomicBool,
next_generation: AtomicI64,
}
impl WatchdogManager {
pub(crate) fn new(manager: Weak<ThreadManagerState>, guards: Arc<Guards>) -> Arc<Self> {
Arc::new(Self {
manager,
guards,
registrations: Mutex::new(HashMap::new()),
started: AtomicBool::new(false),
next_generation: AtomicI64::new(1),
})
}
pub(crate) fn start(self: &Arc<Self>) {
if self
.started
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
return;
}
let manager = Arc::clone(self);
tokio::spawn(async move {
manager.run_loop().await;
});
}
pub(crate) async fn register(
self: &Arc<Self>,
registration: WatchdogRegistration,
) -> CodexResult<Vec<RemovedWatchdog>> {
if exceeds_thread_spawn_depth_limit(
registration.child_depth,
registration.config.agent_max_depth,
) {
let max_depth = registration.config.agent_max_depth;
return Err(CodexErr::UnsupportedOperation(format!(
"agent depth limit reached: max depth is {max_depth}"
)));
}
let interval = interval_duration(registration.interval_s)?;
let generation = self.next_generation.fetch_add(1, Ordering::AcqRel);
let now = Instant::now();
let entry = WatchdogEntry {
registration,
interval,
last_trigger: now,
active_helper_id: None,
owner_idle_since: Some(now),
owner_was_running: false,
force_due_once: false,
generation,
};
let mut registrations = self.registrations.lock().await;
let superseded_targets: Vec<ThreadId> = registrations
.iter()
.filter_map(|(target_thread_id, existing_entry)| {
(existing_entry.registration.owner_thread_id == entry.registration.owner_thread_id
&& *target_thread_id != entry.registration.target_thread_id)
.then_some(*target_thread_id)
})
.collect();
let mut superseded = Vec::new();
for superseded_target in superseded_targets {
if let Some(removed) = registrations.remove(&superseded_target) {
superseded.push(RemovedWatchdog {
target_thread_id: superseded_target,
active_helper_id: removed.active_helper_id,
});
}
}
registrations.insert(entry.registration.target_thread_id, entry);
Ok(superseded)
}
async fn run_loop(self: Arc<Self>) {
let tick = tick_duration();
loop {
self.run_once().await;
if self.manager.upgrade().is_none() {
break;
}
tokio::time::sleep(tick).await;
}
}
pub(crate) async fn run_once(self: &Arc<Self>) {
let Some(manager_state) = self.manager.upgrade() else {
self.registrations.lock().await.clear();
return;
};
let snapshots: Vec<(ThreadId, i64)> = {
let registrations = self.registrations.lock().await;
registrations
.iter()
.map(|(target_id, entry)| (*target_id, entry.generation))
.collect()
};
let now = Instant::now();
for (target_id, generation) in snapshots {
self.evaluate(&manager_state, target_id, generation, now)
.await;
}
}
async fn evaluate(
self: &Arc<Self>,
manager_state: &Arc<ThreadManagerState>,
target_thread_id: ThreadId,
generation: i64,
now: Instant,
) {
let Some(snapshot) = self.snapshot(target_thread_id, generation).await else {
return;
};
// The owner thread is what we actually monitor for idle time and context.
let owner_thread = manager_state.get_thread(snapshot.owner_thread_id).await;
let owner_status = match owner_thread.as_ref() {
Ok(thread) => thread.agent_status().await,
Err(_) => AgentStatus::NotFound,
};
if is_watchdog_terminated(&owner_status) {
self.remove_if_generation(target_thread_id, generation)
.await;
return;
}
let force_due = self
.take_force_due_if_generation(target_thread_id, generation)
.await;
let owner_has_active_turn = match owner_thread {
Ok(thread) => thread.has_active_turn().await,
Err(_) => false,
};
let owner_running = (is_running(&owner_status) || owner_has_active_turn) && !force_due;
let owner_idle_since = self
.update_owner_idle_state_if_generation(
target_thread_id,
generation,
owner_running,
now,
force_due,
)
.await;
if owner_running {
return;
}
let owner_idle_since = owner_idle_since.or(snapshot.owner_idle_since);
let Some(owner_idle_since) = owner_idle_since else {
return;
};
if now.duration_since(owner_idle_since) < snapshot.interval {
return;
}
let control_for_spawn = AgentControl::from_parts(
self.manager.clone(),
Arc::clone(&self.guards),
Arc::clone(self),
);
if let Some(helper_id) = snapshot.active_helper_id {
let helper_status = get_status(manager_state, helper_id).await;
if !is_final(&helper_status) {
return;
}
let helper_sent_input = manager_state
.get_thread(helper_id)
.await
.map(|thread| thread.last_completed_turn_used_collab_send_input())
.unwrap_or(false);
if let AgentStatus::Completed(Some(message)) = &helper_status
&& !message.trim().is_empty()
&& !helper_sent_input
{
if let Err(err) = control_for_spawn
.send_collab_message(snapshot.owner_thread_id, helper_id, message.clone())
.await
{
warn!(
helper_id = %helper_id,
owner_thread_id = %snapshot.owner_thread_id,
"watchdog helper forward failed: {err}"
);
} else {
info!(
helper_id = %helper_id,
owner_thread_id = %snapshot.owner_thread_id,
"watchdog forwarded helper completion to owner"
);
}
}
if let Err(err) = control_for_spawn.shutdown_agent(helper_id).await {
warn!(
helper_id = %helper_id,
owner_thread_id = %snapshot.owner_thread_id,
"watchdog helper cleanup failed: {err}"
);
}
self.update_after_spawn(target_thread_id, generation, now, None)
.await;
return;
}
if now.duration_since(snapshot.last_trigger) < snapshot.interval {
return;
}
let session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: snapshot.owner_thread_id,
depth: snapshot.child_depth,
agent_nickname: None,
agent_role: None,
});
let mut helper_config = snapshot.config.clone();
// Watchdog helpers are short-lived check-ins. Keep their state in-memory and
// avoid writing full forked rollout history to disk for each helper run.
helper_config.ephemeral = true;
let helper_prompt =
watchdog_helper_prompt(&helper_config, snapshot.owner_thread_id, &snapshot.prompt)
.await;
let spawn_result = control_for_spawn
.fork_agent(
helper_config,
vec![UserInput::Text {
text: helper_prompt,
text_elements: Vec::new(),
}],
snapshot.owner_thread_id,
// Keep full history for helper forks to maximize model-side caching efficiency.
usize::MAX,
session_source,
)
.await;
match spawn_result {
Ok(helper_id) => {
info!("watchdog spawned helper {helper_id} for target {target_thread_id}");
self.update_after_spawn(target_thread_id, generation, now, Some(helper_id))
.await;
}
Err(err) => {
warn!("watchdog spawn failed for target {target_thread_id}: {err}");
self.update_after_spawn(target_thread_id, generation, now, None)
.await;
}
}
}
async fn snapshot(
&self,
target_thread_id: ThreadId,
generation: i64,
) -> Option<WatchdogSnapshot> {
let registrations = self.registrations.lock().await;
let entry = registrations.get(&target_thread_id)?;
if entry.generation != generation {
return None;
}
Some(WatchdogSnapshot {
owner_thread_id: entry.registration.owner_thread_id,
child_depth: entry.registration.child_depth,
prompt: entry.registration.prompt.clone(),
config: entry.registration.config.clone(),
interval: entry.interval,
last_trigger: entry.last_trigger,
active_helper_id: entry.active_helper_id,
owner_idle_since: entry.owner_idle_since,
})
}
async fn update_owner_idle_state_if_generation(
&self,
target_thread_id: ThreadId,
generation: i64,
owner_running: bool,
now: Instant,
force_due: bool,
) -> Option<Instant> {
let mut registrations = self.registrations.lock().await;
let entry = registrations.get_mut(&target_thread_id)?;
if entry.generation != generation {
return None;
}
if force_due {
return entry.owner_idle_since;
}
if owner_running {
entry.owner_idle_since = None;
entry.owner_was_running = true;
return None;
}
if entry.owner_was_running || entry.owner_idle_since.is_none() {
entry.owner_idle_since = Some(now);
}
entry.owner_was_running = false;
entry.owner_idle_since
}
async fn take_force_due_if_generation(
&self,
target_thread_id: ThreadId,
generation: i64,
) -> bool {
let mut registrations = self.registrations.lock().await;
let Some(entry) = registrations.get_mut(&target_thread_id) else {
return false;
};
if entry.generation != generation || !entry.force_due_once {
return false;
}
entry.force_due_once = false;
true
}
async fn update_after_spawn(
&self,
target_thread_id: ThreadId,
generation: i64,
now: Instant,
active_helper_id: Option<ThreadId>,
) {
let mut registrations = self.registrations.lock().await;
let Some(entry) = registrations.get_mut(&target_thread_id) else {
return;
};
if entry.generation != generation {
return;
}
entry.last_trigger = now;
entry.active_helper_id = active_helper_id;
}
async fn remove_if_generation(&self, target_thread_id: ThreadId, generation: i64) {
let mut registrations = self.registrations.lock().await;
let Some(entry) = registrations.get(&target_thread_id) else {
return;
};
if entry.generation == generation {
registrations.remove(&target_thread_id);
}
}
pub(crate) async fn unregister(&self, target_thread_id: ThreadId) -> Option<RemovedWatchdog> {
let mut registrations = self.registrations.lock().await;
registrations
.remove(&target_thread_id)
.map(|removed| RemovedWatchdog {
target_thread_id,
active_helper_id: removed.active_helper_id,
})
}
pub(crate) async fn owner_for_active_helper(
&self,
helper_thread_id: ThreadId,
) -> Option<ThreadId> {
let registrations = self.registrations.lock().await;
registrations.values().find_map(|entry| {
(entry.active_helper_id == Some(helper_thread_id))
.then_some(entry.registration.owner_thread_id)
})
}
pub(crate) async fn registered_targets(&self, candidate_ids: &[ThreadId]) -> HashSet<ThreadId> {
let registrations = self.registrations.lock().await;
candidate_ids
.iter()
.copied()
.filter(|candidate_id| registrations.contains_key(candidate_id))
.collect()
}
pub(crate) async fn take_for_owner(&self, owner_thread_id: ThreadId) -> Vec<RemovedWatchdog> {
let mut registrations = self.registrations.lock().await;
let removed_targets: Vec<ThreadId> = registrations
.iter()
.filter_map(|(target_thread_id, entry)| {
(entry.registration.owner_thread_id == owner_thread_id).then_some(*target_thread_id)
})
.collect();
let mut removed = Vec::new();
for removed_target in removed_targets {
if let Some(entry) = registrations.remove(&removed_target) {
removed.push(RemovedWatchdog {
target_thread_id: removed_target,
active_helper_id: entry.active_helper_id,
});
}
}
removed
}
#[cfg(test)]
#[allow(dead_code)]
pub(crate) async fn set_active_helper_for_tests(
&self,
target_thread_id: ThreadId,
helper_thread_id: ThreadId,
) {
let mut registrations = self.registrations.lock().await;
let Some(entry) = registrations.get_mut(&target_thread_id) else {
return;
};
let due_at = Instant::now() - entry.interval;
entry.last_trigger = due_at;
entry.owner_idle_since = Some(due_at);
entry.owner_was_running = false;
entry.active_helper_id = Some(helper_thread_id);
}
#[cfg(test)]
#[allow(dead_code)]
pub(crate) async fn force_due_for_tests(&self, target_thread_id: ThreadId) {
let mut registrations = self.registrations.lock().await;
let Some(entry) = registrations.get_mut(&target_thread_id) else {
return;
};
let due_at = Instant::now() - entry.interval;
entry.last_trigger = due_at;
entry.owner_idle_since = Some(due_at);
entry.owner_was_running = false;
entry.force_due_once = true;
}
}
#[derive(Clone)]
struct WatchdogSnapshot {
owner_thread_id: ThreadId,
child_depth: i32,
prompt: String,
config: Config,
interval: Duration,
last_trigger: Instant,
active_helper_id: Option<ThreadId>,
owner_idle_since: Option<Instant>,
}
async fn get_status(
manager_state: &Arc<ThreadManagerState>,
thread_id: ThreadId,
) -> codex_protocol::protocol::AgentStatus {
let Ok(thread) = manager_state.get_thread(thread_id).await else {
return codex_protocol::protocol::AgentStatus::NotFound;
};
thread.agent_status().await
}
fn is_running(status: &codex_protocol::protocol::AgentStatus) -> bool {
matches!(
status,
codex_protocol::protocol::AgentStatus::PendingInit
| codex_protocol::protocol::AgentStatus::Running
)
}
fn is_watchdog_terminated(status: &codex_protocol::protocol::AgentStatus) -> bool {
matches!(
status,
codex_protocol::protocol::AgentStatus::Shutdown
| codex_protocol::protocol::AgentStatus::NotFound
)
}
fn interval_duration(interval_s: i64) -> CodexResult<Duration> {
if interval_s <= 0 {
return Err(CodexErr::UnsupportedOperation(
"interval_s must be greater than zero".to_string(),
));
}
let seconds = u64::try_from(interval_s).map_err(|_| {
CodexErr::UnsupportedOperation(format!("interval_s out of range: {interval_s}"))
})?;
Ok(Duration::from_secs(seconds))
}
fn tick_duration() -> Duration {
let seconds = u64::try_from(WATCHDOG_TICK_SECONDS).unwrap_or(5);
Duration::from_secs(seconds)
}
async fn watchdog_helper_prompt(
config: &Config,
target_thread_id: ThreadId,
prompt: &str,
) -> String {
let watchdog_prompt = load_watchdog_prompt(&config.codex_home).await;
if prompt.trim().is_empty() {
format!("{watchdog_prompt}\n\nTarget agent id: {target_thread_id}")
} else {
format!("{watchdog_prompt}\n\nTarget agent id: {target_thread_id}\n\n{prompt}")
}
}
+449 -38
View File
@@ -4,7 +4,9 @@ use std::fmt::Debug;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use crate::AuthManager;
use crate::CodexAuth;
@@ -103,6 +105,7 @@ use rmcp::model::ReadResourceResult;
use rmcp::model::RequestId;
use serde_json;
use serde_json::Value;
use tokio::fs;
use tokio::sync::Mutex;
use tokio::sync::RwLock;
use tokio::sync::oneshot;
@@ -219,6 +222,7 @@ use crate::rollout::RolloutRecorderParams;
use crate::rollout::map_session_init_error;
use crate::rollout::metadata;
use crate::rollout::policy::EventPersistenceMode;
use crate::rollout::truncation;
use crate::shell;
use crate::shell_snapshot::ShellSnapshot;
use crate::skills::SkillError;
@@ -276,6 +280,37 @@ use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_readiness::Readiness;
use codex_utils_readiness::ReadinessFlag;
const ROOT_AGENT_PROMPT_FALLBACK: &str = include_str!("../root_agent_prompt.md");
const SUBAGENT_PROMPT_FALLBACK: &str = include_str!("../subagent_prompt.md");
const WATCHDOG_PROMPT_FALLBACK: &str = include_str!("../watchdog_agent_prompt.md");
async fn load_agent_prompt_fallback(
codex_home: &Path,
fallback: &str,
override_filename: &str,
) -> String {
let override_path = codex_home.join(override_filename);
if let Ok(contents) = fs::read_to_string(&override_path).await
&& !contents.trim().is_empty()
{
return contents;
}
fallback.to_string()
}
async fn load_root_agent_prompt(codex_home: &Path) -> String {
load_agent_prompt_fallback(codex_home, ROOT_AGENT_PROMPT_FALLBACK, "AGENTS.root.md").await
}
async fn load_subagent_prompt(codex_home: &Path) -> String {
load_agent_prompt_fallback(codex_home, SUBAGENT_PROMPT_FALLBACK, "AGENTS.subagent.md").await
}
pub(crate) async fn load_watchdog_prompt(codex_home: &Path) -> String {
load_agent_prompt_fallback(codex_home, WATCHDOG_PROMPT_FALLBACK, "AGENTS.watchdog.md").await
}
/// The high-level interface to the Codex system.
/// It operates as a queue pair where you send submissions and receive events.
pub struct Codex {
@@ -355,6 +390,21 @@ impl Codex {
)
.await;
let role_prompt = if config.features.enabled(Feature::Collab) {
if let SessionSource::SubAgent(_) = session_source {
Some(load_subagent_prompt(&config.codex_home).await)
} else {
Some(load_root_agent_prompt(&config.codex_home).await)
}
} else {
None
};
let developer_instructions = match (role_prompt, config.developer_instructions.clone()) {
(Some(prompt), Some(existing)) => Some(format!("{prompt}\n\n{existing}")),
(Some(prompt), None) => Some(prompt),
(None, existing) => existing,
};
// Resolve base instructions for the session. Priority order:
// 1. config.base_instructions override
// 2. conversation history => session_meta.base_instructions
@@ -409,7 +459,7 @@ impl Codex {
provider: config.model_provider.clone(),
collaboration_mode,
model_reasoning_summary: config.model_reasoning_summary,
developer_instructions: config.developer_instructions.clone(),
developer_instructions,
user_instructions,
personality: config.personality,
base_instructions,
@@ -524,6 +574,14 @@ impl Codex {
pub(crate) fn enabled(&self, feature: Feature) -> bool {
self.session.enabled(feature)
}
pub(crate) async fn has_active_turn(&self) -> bool {
self.session.has_active_turn().await
}
pub(crate) fn last_completed_turn_used_collab_send_input(&self) -> bool {
self.session.last_completed_turn_used_collab_send_input()
}
}
/// Context for an initialized model agent
@@ -543,6 +601,10 @@ pub(crate) struct Session {
pub(crate) services: SessionServices,
js_repl: Arc<JsReplHandle>,
next_internal_sub_id: AtomicU64,
/// Tracks whether the current turn delivered a collab inbox message via send_input.
turn_used_collab_send_input: AtomicBool,
/// Snapshots whether the last completed turn used collab send_input.
last_completed_turn_used_collab_send_input: AtomicBool,
}
#[derive(Clone, Debug)]
@@ -793,6 +855,7 @@ impl SessionConfiguration {
reasoning_effort: self.collaboration_mode.reasoning_effort(),
personality: self.personality,
session_source: self.session_source.clone(),
collab_inbox_delivery_role: self.original_config_do_not_use.collab_inbox_delivery_role,
}
}
@@ -1382,6 +1445,8 @@ impl Session {
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
turn_used_collab_send_input: AtomicBool::new(false),
last_completed_turn_used_collab_send_input: AtomicBool::new(false),
});
if let Some(network_policy_decider_session) = network_policy_decider_session {
let mut guard = network_policy_decider_session.write().await;
@@ -1499,6 +1564,43 @@ impl Session {
self.services.state_db.clone()
}
pub(crate) async fn has_active_turn(&self) -> bool {
self.active_turn.lock().await.is_some()
}
pub(crate) async fn parent_thread_id(&self) -> Option<ThreadId> {
let state = self.state.lock().await;
match &state.session_configuration.session_source {
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id, ..
}) => Some(*parent_thread_id),
_ => None,
}
}
pub(crate) fn mark_turn_used_collab_send_input(&self) {
self.turn_used_collab_send_input
.store(true, Ordering::Release);
}
pub(crate) fn reset_turn_collab_send_input_flag(&self) {
self.turn_used_collab_send_input
.store(false, Ordering::Release);
}
pub(crate) fn snapshot_collab_send_input_on_turn_complete(&self) {
let used_collab_send_input = self
.turn_used_collab_send_input
.swap(false, Ordering::AcqRel);
self.last_completed_turn_used_collab_send_input
.store(used_collab_send_input, Ordering::Release);
}
pub(crate) fn last_completed_turn_used_collab_send_input(&self) -> bool {
self.last_completed_turn_used_collab_send_input
.load(Ordering::Acquire)
}
/// Ensure all rollout writes are durably flushed.
pub(crate) async fn flush_rollout(&self) {
let recorder = {
@@ -1532,6 +1634,7 @@ impl Session {
}
pub(crate) async fn route_realtime_text_input(self: &Arc<Self>, text: String) {
let mut previous_context = None;
handlers::user_input_or_turn(
self,
self.next_internal_sub_id(),
@@ -1542,6 +1645,7 @@ impl Session {
}],
final_output_json_schema: None,
},
&mut previous_context,
)
.await;
}
@@ -1701,6 +1805,15 @@ impl Session {
let previous_model =
previous_regular_turn_context_item.map(|ctx| ctx.model.clone());
self.set_previous_model(previous_model).await;
let mut rollout_items = rollout_items;
let fork_reference = rollout_items.iter().find_map(|item| match item {
RolloutItem::ForkReference(_) => Some(item.clone()),
_ => None,
});
// During the initial fork startup we already have inherited rollout items in
// memory; keep fork references for on-disk resume, but do not recurse into them
// during this startup reconstruction.
rollout_items.retain(|item| !matches!(item, RolloutItem::ForkReference(_)));
// Always add response items to conversation history
let reconstructed_history = self
@@ -1721,8 +1834,12 @@ impl Session {
self.set_mcp_tool_selection(selected_tools).await;
}
// If persisting, persist all rollout items as-is (recorder filters)
if !rollout_items.is_empty() {
// For forks, persist only a compact fork reference marker instead of copying all
// inherited rollout items into the child rollout file.
if let Some(reference) = fork_reference {
self.persist_rollout_items(&[reference]).await;
} else if !rollout_items.is_empty() {
// Fallback for older forks that do not include a fork reference marker.
self.persist_rollout_items(&rollout_items).await;
}
@@ -2847,34 +2964,89 @@ impl Session {
rollout_items: &[RolloutItem],
) -> Vec<ResponseItem> {
let mut history = ContextManager::new();
for item in rollout_items {
match item {
RolloutItem::ResponseItem(response_item) => {
history.record_items(
std::iter::once(response_item),
turn_context.truncation_policy,
);
}
RolloutItem::Compacted(compacted) => {
if let Some(replacement) = &compacted.replacement_history {
history.replace(replacement.clone());
} else {
let user_messages = collect_user_messages(history.raw_items());
let rebuilt = compact::build_compacted_history(
self.build_initial_context(turn_context).await,
&user_messages,
&compacted.message,
);
history.replace(rebuilt);
self.apply_rollout_items_to_history(&mut history, turn_context, rollout_items, 0)
.await;
history.raw_items().to_vec()
}
async fn apply_rollout_items_to_history(
&self,
history: &mut ContextManager,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
initial_depth: usize,
) {
const MAX_FORK_REFERENCE_DEPTH: usize = 8;
let mut stack: Vec<(Vec<RolloutItem>, usize, usize)> =
vec![(rollout_items.to_vec(), 0, initial_depth)];
while let Some((items, mut idx, depth)) = stack.pop() {
while idx < items.len() {
match &items[idx] {
RolloutItem::ForkReference(reference) => {
if depth >= MAX_FORK_REFERENCE_DEPTH {
warn!(
"skipping fork reference recursion at depth {} for {:?}",
depth, reference.rollout_path
);
idx += 1;
continue;
}
let parent_history =
match RolloutRecorder::get_rollout_history(&reference.rollout_path)
.await
{
Ok(history) => history,
Err(err) => {
warn!(
"failed to load fork reference rollout {:?}: {err}",
reference.rollout_path
);
idx += 1;
continue;
}
};
let parent_items = parent_history.get_rollout_items();
let parent_items =
truncation::truncate_rollout_before_nth_user_message_from_start(
&parent_items,
reference.nth_user_message,
);
// Process referenced parent items before continuing with the current list.
stack.push((items, idx + 1, depth));
stack.push((parent_items, 0, depth + 1));
break;
}
RolloutItem::ResponseItem(response_item) => {
history.record_items(
std::iter::once(response_item),
turn_context.truncation_policy,
);
}
RolloutItem::Compacted(compacted) => {
if let Some(replacement) = &compacted.replacement_history {
history.replace(replacement.clone());
} else {
let user_messages = collect_user_messages(history.raw_items());
let rebuilt = compact::build_compacted_history(
self.build_initial_context(turn_context).await,
&user_messages,
&compacted.message,
);
history.replace(rebuilt);
}
}
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => {
history.drop_last_n_user_turns(rollback.num_turns);
}
_ => {}
}
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => {
history.drop_last_n_user_turns(rollback.num_turns);
}
_ => {}
idx += 1;
}
}
history.raw_items().to_vec()
}
/// Append ResponseItems to the in-memory conversation history only.
@@ -3608,6 +3780,10 @@ impl Session {
}
async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiver<Submission>) {
// Seed with context in case the first op changes settings or injects response items before a
// normal user turn starts.
let mut previous_context = Some(sess.new_default_turn().await.to_turn_context_item());
// To break out of this loop, send Op::Shutdown.
while let Ok(sub) = rx_sub.recv().await {
debug!(?sub, "Submission");
@@ -3679,7 +3855,17 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
.await;
}
Op::UserInput { .. } | Op::UserTurn { .. } => {
handlers::user_input_or_turn(&sess, sub.id.clone(), sub.op).await;
handlers::user_input_or_turn(&sess, sub.id.clone(), sub.op, &mut previous_context)
.await;
}
Op::InjectResponseItems { items } => {
handlers::inject_response_items(
&sess,
sub.id.clone(),
items,
&mut previous_context,
)
.await;
}
Op::ExecApproval {
id: approval_id,
@@ -3759,7 +3945,13 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
handlers::set_thread_name(&sess, sub.id.clone(), name).await;
}
Op::RunUserShellCommand { command } => {
handlers::run_user_shell_command(&sess, sub.id.clone(), command).await;
handlers::run_user_shell_command(
&sess,
sub.id.clone(),
command,
&mut previous_context,
)
.await;
}
Op::ResolveElicitation {
server_name,
@@ -3787,7 +3979,6 @@ mod handlers {
use crate::codex::Session;
use crate::codex::SessionSettingsUpdate;
use crate::codex::SteerInputError;
use crate::codex::spawn_review_thread;
use crate::config::Config;
@@ -3821,16 +4012,21 @@ mod handlers {
use codex_protocol::protocol::ThreadNameUpdatedEvent;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::WarningEvent;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_protocol::skill_approval::SkillApprovalResponse;
use crate::context_manager::is_user_turn_boundary;
use crate::parse_turn_item;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::items::TurnItem;
use codex_protocol::mcp::RequestId as ProtocolRequestId;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::user_input::UserInput;
use codex_rmcp_client::ElicitationAction;
use codex_rmcp_client::ElicitationResponse;
@@ -3864,7 +4060,12 @@ mod handlers {
}
}
pub async fn user_input_or_turn(sess: &Arc<Session>, sub_id: String, op: Op) {
pub async fn user_input_or_turn(
sess: &Arc<Session>,
sub_id: String,
op: Op,
previous_context: &mut Option<TurnContextItem>,
) {
let (items, updates) = match op {
Op::UserTurn {
cwd,
@@ -3925,15 +4126,112 @@ mod handlers {
// Attempt to inject input into current task.
if let Err(SteerInputError::NoActiveTurn(items)) = sess.steer_input(items, None).await {
let previous_model = sess.previous_model().await;
let update_items = sess.build_settings_update_items(
previous_context.as_ref(),
previous_model.as_deref(),
&current_context,
);
if !update_items.is_empty() {
sess.record_conversation_items(&current_context, &update_items)
.await;
}
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
let regular_task = sess.take_startup_regular_task().await.unwrap_or_default();
sess.spawn_task(Arc::clone(&current_context), items, regular_task)
.await;
*previous_context = Some(current_context.to_turn_context_item());
}
}
pub async fn run_user_shell_command(sess: &Arc<Session>, sub_id: String, command: String) {
pub async fn inject_response_items(
sess: &Arc<Session>,
sub_id: String,
items: Vec<ResponseInputItem>,
previous_context: &mut Option<TurnContextItem>,
) {
const MAX_TURN_RESTART_ATTEMPTS: usize = 3;
let mut pending_items = items;
let mut attempts = 0usize;
loop {
match sess.inject_response_items(pending_items).await {
Ok(()) => return,
Err(items_without_active_turn) => {
pending_items = items_without_active_turn;
}
}
if attempts >= MAX_TURN_RESTART_ATTEMPTS {
warn!(
attempts,
remaining_items = pending_items.len(),
"dropping response items after repeated turn restart failures"
);
return;
}
attempts += 1;
let mut turn_input =
pop_leading_user_message_input(&mut pending_items).unwrap_or_default();
if turn_input.is_empty() {
turn_input.push(UserInput::Text {
text: String::new(),
text_elements: Vec::new(),
});
}
let turn_sub_id = if attempts == 1 {
sub_id.clone()
} else {
format!("{sub_id}-retry-{attempts}")
};
let current_context = sess.new_default_turn_with_sub_id(turn_sub_id).await;
current_context.otel_manager.user_prompt(&turn_input);
let previous_model = sess.previous_model().await;
let update_items = sess.build_settings_update_items(
previous_context.as_ref(),
previous_model.as_deref(),
&current_context,
);
if !update_items.is_empty() {
sess.record_conversation_items(&current_context, &update_items)
.await;
}
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
let regular_task = sess.take_startup_regular_task().await.unwrap_or_default();
sess.spawn_task(Arc::clone(&current_context), turn_input, regular_task)
.await;
*previous_context = Some(current_context.to_turn_context_item());
if pending_items.is_empty() {
return;
}
}
}
fn pop_leading_user_message_input(
items: &mut Vec<ResponseInputItem>,
) -> Option<Vec<UserInput>> {
let first_item = items.first().cloned()?;
let response_item: ResponseItem = first_item.into();
let TurnItem::UserMessage(user_message) = parse_turn_item(&response_item)? else {
return None;
};
let _ = items.remove(0);
Some(user_message.content)
}
pub async fn run_user_shell_command(
sess: &Arc<Session>,
sub_id: String,
command: String,
previous_context: &mut Option<TurnContextItem>,
) {
if let Some((turn_context, cancellation_token)) =
sess.active_turn_context_and_cancellation_token().await
{
@@ -3958,6 +4256,7 @@ mod handlers {
UserShellCommandTask::new(command),
)
.await;
*previous_context = Some(turn_context.to_turn_context_item());
}
pub async fn resolve_elicitation(
@@ -4609,12 +4908,9 @@ async fn spawn_review_thread(
per_turn_config.model = Some(model.clone());
per_turn_config.features = review_features.clone();
if let Err(err) = per_turn_config.web_search_mode.set(review_web_search_mode) {
let fallback_value = per_turn_config.web_search_mode.value();
tracing::warn!(
error = %err,
?review_web_search_mode,
?fallback_value,
"review web_search_mode is disallowed by requirements; keeping constrained value"
"failed to force review web_search_mode=disabled; keeping constrained value"
);
}
@@ -8266,6 +8562,8 @@ mod tests {
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
turn_used_collab_send_input: AtomicBool::new(false),
last_completed_turn_used_collab_send_input: AtomicBool::new(false),
};
(session, turn_context)
@@ -8421,6 +8719,8 @@ mod tests {
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
turn_used_collab_send_input: AtomicBool::new(false),
last_completed_turn_used_collab_send_input: AtomicBool::new(false),
});
(session, turn_context, rx_event)
@@ -8641,9 +8941,15 @@ mod tests {
let mut state = session.state.lock().await;
state.set_reference_context_item(None);
}
let mut previous_context = None;
handlers::run_user_shell_command(&session, "sub-id".to_string(), "echo shell".to_string())
.await;
handlers::run_user_shell_command(
&session,
"sub-id".to_string(),
"echo shell".to_string(),
&mut previous_context,
)
.await;
let deadline = StdDuration::from_secs(5);
let start = std::time::Instant::now();
@@ -8804,6 +9110,111 @@ mod tests {
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn inject_response_items_handler_adds_pending_input_for_active_turn() {
let (sess, tc, _rx) = make_session_and_context_with_rx().await;
let input = vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}];
sess.spawn_task(
Arc::clone(&tc),
input,
NeverEndingTask {
kind: TaskKind::Regular,
listen_to_cancellation_token: false,
},
)
.await;
let mut previous_context = None;
let injected_item = ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: "collab inbox test".to_string(),
}],
};
handlers::inject_response_items(
&sess,
"inject-active-turn".to_string(),
vec![injected_item.clone()],
&mut previous_context,
)
.await;
let pending = sess.get_pending_input().await;
assert_eq!(pending, vec![injected_item]);
sess.abort_all_tasks(TurnAbortReason::Replaced).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn submission_loop_handles_inject_response_items_op() {
let (sess, tc, _rx) = make_session_and_context_with_rx().await;
let input = vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}];
sess.spawn_task(
Arc::clone(&tc),
input,
NeverEndingTask {
kind: TaskKind::Regular,
listen_to_cancellation_token: false,
},
)
.await;
let config = sess.get_config().await;
let (tx_sub, rx_sub) = async_channel::bounded(4);
let session = Arc::clone(&sess);
let loop_handle = tokio::spawn(async move {
submission_loop(session, config, rx_sub).await;
});
let injected_item = ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: "submission-loop collab inbox".to_string(),
}],
};
tx_sub
.send(Submission {
id: "inject-op".to_string(),
op: Op::InjectResponseItems {
items: vec![injected_item.clone()],
},
})
.await
.expect("submit inject op");
let pending = tokio::time::timeout(Duration::from_secs(2), async {
loop {
let pending = sess.get_pending_input().await;
if !pending.is_empty() {
break pending;
}
tokio::task::yield_now().await;
}
})
.await
.expect("inject op should be routed to pending input");
assert_eq!(pending, vec![injected_item]);
tx_sub
.send(Submission {
id: "shutdown-op".to_string(),
op: Op::Shutdown,
})
.await
.expect("submit shutdown op");
loop_handle
.await
.expect("submission loop task should exit cleanly");
sess.abort_all_tasks(TurnAbortReason::Replaced).await;
}
#[tokio::test]
async fn steer_input_requires_active_turn() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
+14
View File
@@ -1,6 +1,7 @@
use crate::agent::AgentStatus;
use crate::codex::Codex;
use crate::codex::SteerInputError;
use crate::config::types::CollabInboxDeliveryRole;
use crate::error::Result as CodexResult;
use crate::features::Feature;
use crate::file_watcher::WatchRegistration;
@@ -32,6 +33,7 @@ pub struct ThreadConfigSnapshot {
pub reasoning_effort: Option<ReasoningEffort>,
pub personality: Option<Personality>,
pub session_source: SessionSource,
pub collab_inbox_delivery_role: CollabInboxDeliveryRole,
}
pub struct CodexThread {
@@ -115,6 +117,14 @@ impl CodexThread {
.await;
}
pub(crate) async fn has_active_turn(&self) -> bool {
self.codex.has_active_turn().await
}
pub(crate) fn last_completed_turn_used_collab_send_input(&self) -> bool {
self.codex.last_completed_turn_used_collab_send_input()
}
pub fn rollout_path(&self) -> Option<PathBuf> {
self.rollout_path.clone()
}
@@ -123,6 +133,10 @@ impl CodexThread {
self.codex.state_db()
}
pub(crate) async fn flush_rollout(&self) {
self.codex.session.flush_rollout().await;
}
pub async fn config_snapshot(&self) -> ThreadConfigSnapshot {
self.codex.thread_config_snapshot().await
}
+62 -2
View File
@@ -2,6 +2,7 @@ use crate::auth::AuthCredentialsStoreMode;
use crate::config::edit::ConfigEdit;
use crate::config::edit::ConfigEditsBuilder;
use crate::config::types::AppsConfigToml;
use crate::config::types::CollabInboxDeliveryRole;
use crate::config::types::DEFAULT_OTEL_ENVIRONMENT;
use crate::config::types::History;
use crate::config::types::McpServerConfig;
@@ -371,6 +372,9 @@ pub struct Config {
/// Memories subsystem settings.
pub memories: MemoriesConfig,
/// How inbound collaboration messages are delivered to non-subagent threads.
pub collab_inbox_delivery_role: CollabInboxDeliveryRole,
/// Directory containing all Codex state (defaults to `~/.codex` but can be
/// overridden by the `CODEX_HOME` environment variable).
pub codex_home: PathBuf,
@@ -1204,6 +1208,9 @@ pub struct ConfigToml {
/// Agent-related settings (thread limits, etc.).
pub agents: Option<AgentsToml>,
/// How inbound collaboration messages are delivered to non-subagent threads.
pub agent_delivery_role: Option<CollabInboxDeliveryRole>,
/// Memories subsystem settings.
pub memories: Option<MemoriesToml>,
@@ -1357,10 +1364,22 @@ pub struct AgentsToml {
pub roles: BTreeMap<String, AgentRoleToml>,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, Default, PartialEq, Eq, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum AgentRoleSpawnMode {
Spawn,
#[default]
Fork,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AgentRoleConfig {
/// Human-facing role documentation used in spawn tool guidance.
pub description: Option<String>,
/// Optional model override applied by this role.
pub model: Option<String>,
/// Optional default spawn mode when `spawn_agent` omits `spawn_mode`.
pub spawn_mode: Option<AgentRoleSpawnMode>,
/// Path to a role-specific config layer.
pub config_file: Option<PathBuf>,
}
@@ -1371,6 +1390,12 @@ pub struct AgentRoleToml {
/// Human-facing role documentation used in spawn tool guidance.
pub description: Option<String>,
/// Optional model override applied by this role.
pub model: Option<String>,
/// Optional default spawn mode when `spawn_agent` omits `spawn_mode`.
pub spawn_mode: Option<AgentRoleSpawnMode>,
/// Path to a role-specific config layer.
/// Relative paths are resolved relative to the `config.toml` that defines them.
pub config_file: Option<AbsolutePathBuf>,
@@ -1817,6 +1842,7 @@ impl Config {
.as_ref()
.and_then(|agents| agents.max_threads)
.or(DEFAULT_AGENT_MAX_THREADS);
let collab_inbox_delivery_role = cfg.agent_delivery_role.unwrap_or_default();
if agent_max_threads == Some(0) {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
@@ -1849,6 +1875,8 @@ impl Config {
name.clone(),
AgentRoleConfig {
description: role.description.clone(),
model: role.model.clone(),
spawn_mode: role.spawn_mode,
config_file,
},
))
@@ -2143,6 +2171,7 @@ impl Config {
memories: cfg.memories.unwrap_or_default().into(),
agent_max_spawn_depth,
agent_job_max_runtime_seconds,
collab_inbox_delivery_role,
codex_home,
sqlite_home,
log_dir,
@@ -2531,6 +2560,16 @@ phase_2_model = "gpt-5"
phase_2_model: Some("gpt-5".to_string()),
}
);
let agent_delivery_role = r#"
agent_delivery_role = "assistant"
"#;
let agent_delivery_role_cfg = toml::from_str::<ConfigToml>(agent_delivery_role)
.expect("TOML deserialization should succeed");
assert_eq!(
Some(CollabInboxDeliveryRole::Assistant),
agent_delivery_role_cfg.agent_delivery_role
);
}
#[test]
@@ -4486,6 +4525,7 @@ model = "gpt-5.1-codex"
AgentRoleToml {
description: Some("Research role".to_string()),
config_file: Some(AbsolutePathBuf::from_absolute_path(missing_path)?),
..Default::default()
},
)]),
}),
@@ -4521,6 +4561,8 @@ model = "gpt-5.1-codex"
codex_home.path().join(CONFIG_TOML_FILE),
r#"[agents.researcher]
description = "Research role"
model = "gpt-5.3-codex-spark"
spawn_mode = "spawn"
config_file = "./agents/researcher.toml"
"#,
)
@@ -4538,6 +4580,20 @@ config_file = "./agents/researcher.toml"
.and_then(|role| role.config_file.as_ref()),
Some(&role_config_path)
);
assert_eq!(
config
.agent_roles
.get("researcher")
.and_then(|role| role.model.as_deref()),
Some("gpt-5.3-codex-spark")
);
assert_eq!(
config
.agent_roles
.get("researcher")
.and_then(|role| role.spawn_mode),
Some(AgentRoleSpawnMode::Spawn)
);
Ok(())
}
@@ -4756,6 +4812,7 @@ model_verbosity = "high"
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
collab_inbox_delivery_role: CollabInboxDeliveryRole::Tool,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -4882,6 +4939,7 @@ model_verbosity = "high"
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
collab_inbox_delivery_role: CollabInboxDeliveryRole::Tool,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -5006,6 +5064,7 @@ model_verbosity = "high"
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
collab_inbox_delivery_role: CollabInboxDeliveryRole::Tool,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -5116,6 +5175,7 @@ model_verbosity = "high"
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
collab_inbox_delivery_role: CollabInboxDeliveryRole::Tool,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
@@ -5207,8 +5267,8 @@ model_verbosity = "high"
]),
mcp_servers: None,
rules: None,
enforce_residency: None,
network: None,
enforce_residency: None,
};
let requirement_source = crate::config_loader::RequirementSource::Unknown;
let requirement_source_for_error = requirement_source.clone();
@@ -5811,8 +5871,8 @@ mcp_oauth_callback_url = "https://example.com/callback"
allowed_web_search_modes: None,
mcp_servers: None,
rules: None,
enforce_residency: None,
network: None,
enforce_residency: None,
};
let config = ConfigBuilder::default()
+10
View File
@@ -41,6 +41,16 @@ pub struct WindowsToml {
pub sandbox: Option<WindowsSandboxModeToml>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
#[derive(Default)]
pub enum CollabInboxDeliveryRole {
Developer,
#[default]
Tool,
Assistant,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum McpServerDisabledReason {
Unknown,
+67 -1
View File
@@ -1061,6 +1061,9 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result<HeadTai
RolloutItem::Compacted(_) => {
// Not included in `head`; skip.
}
RolloutItem::ForkReference(_) => {
// Not included in `head`; skip.
}
RolloutItem::EventMsg(ev) => {
if let EventMsg::UserMessage(user) = ev {
summary.saw_user_event = true;
@@ -1112,6 +1115,7 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result<Vec<serde_json::Va
head.push(value);
}
}
RolloutItem::ForkReference(_) => {}
RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => {}
@@ -1240,6 +1244,63 @@ async fn find_thread_path_by_id_str_in_subdir(
Ok(found)
}
async fn try_unarchive_thread_path_by_id_str(
codex_home: &Path,
id_str: &str,
) -> io::Result<Option<PathBuf>> {
let Some(archived_path) =
find_thread_path_by_id_str_in_subdir(codex_home, ARCHIVED_SESSIONS_SUBDIR, id_str).await?
else {
return Ok(None);
};
let Some(file_name) = archived_path.file_name().map(OsStr::to_owned) else {
tracing::error!(
"archived rollout path for thread {id_str} missing file name: {}",
archived_path.display()
);
return Ok(None);
};
let Some((year, month, day)) = rollout_date_parts(&file_name) else {
tracing::error!(
"archived rollout path for thread {id_str} missing filename timestamp: {}",
archived_path.display()
);
return Ok(None);
};
let restored_dir = codex_home
.join(SESSIONS_SUBDIR)
.join(year)
.join(month)
.join(day);
tokio::fs::create_dir_all(&restored_dir).await?;
let restored_path = restored_dir.join(&file_name);
match tokio::fs::rename(&archived_path, &restored_path).await {
Ok(()) => {}
Err(err) => {
if tokio::fs::try_exists(&restored_path).await.unwrap_or(false) {
tracing::debug!(
"archived rollout for thread {id_str} already restored concurrently to {}",
restored_path.display()
);
} else {
return Err(err);
}
}
}
if let Some(state_db_ctx) = state_db::open_if_present(codex_home, "").await
&& let Ok(thread_id) = ThreadId::from_string(id_str)
{
let _ = state_db_ctx
.mark_unarchived(thread_id, restored_path.as_path())
.await;
}
Ok(Some(restored_path))
}
/// Locate a recorded thread rollout file by its UUID string using the existing
/// paginated listing implementation. Returns `Ok(Some(path))` if found, `Ok(None)` if not present
/// or the id is invalid.
@@ -1247,7 +1308,12 @@ pub async fn find_thread_path_by_id_str(
codex_home: &Path,
id_str: &str,
) -> io::Result<Option<PathBuf>> {
find_thread_path_by_id_str_in_subdir(codex_home, SESSIONS_SUBDIR, id_str).await
if let Some(active_path) =
find_thread_path_by_id_str_in_subdir(codex_home, SESSIONS_SUBDIR, id_str).await?
{
return Ok(Some(active_path));
}
try_unarchive_thread_path_by_id_str(codex_home, id_str).await
}
/// Locate an archived thread rollout file by its UUID string.
+1
View File
@@ -67,6 +67,7 @@ pub(crate) fn builder_from_items(
) -> Option<ThreadMetadataBuilder> {
if let Some(session_meta) = items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line),
RolloutItem::ForkReference(_) => None,
RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
+4 -3
View File
@@ -17,9 +17,10 @@ pub(crate) fn is_persisted_response_item(item: &RolloutItem, mode: EventPersiste
RolloutItem::ResponseItem(item) => should_persist_response_item(item),
RolloutItem::EventMsg(ev) => should_persist_event_msg(ev, mode),
// Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns).
RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {
true
}
RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::SessionMeta(_)
| RolloutItem::ForkReference(_) => true,
}
}
+3
View File
@@ -552,6 +552,9 @@ impl RolloutRecorder {
RolloutItem::ResponseItem(item) => {
items.push(RolloutItem::ResponseItem(item));
}
RolloutItem::ForkReference(item) => {
items.push(RolloutItem::ForkReference(item));
}
RolloutItem::Compacted(item) => {
items.push(RolloutItem::Compacted(item));
}
+10
View File
@@ -220,6 +220,16 @@ pub(crate) fn last_assistant_message_from_item(
pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Option<ResponseItem> {
match input {
ResponseInputItem::FunctionCall {
name,
arguments,
call_id,
} => Some(ResponseItem::FunctionCall {
id: None,
name: name.clone(),
arguments: arguments.clone(),
call_id: call_id.clone(),
}),
ResponseInputItem::FunctionCallOutput { call_id, output } => {
Some(ResponseItem::FunctionCallOutput {
call_id: call_id.clone(),
+2 -1
View File
@@ -193,7 +193,7 @@ impl Session {
turn_context
.turn_metadata_state
.cancel_git_enrichment_task();
self.snapshot_collab_send_input_on_turn_complete();
let mut active = self.active_turn.lock().await;
let mut pending_input = Vec::<ResponseInputItem>::new();
let mut should_clear_active_turn = false;
@@ -224,6 +224,7 @@ impl Session {
}
async fn register_new_active_task(&self, task: RunningTask) {
self.reset_turn_collab_send_input_flag();
let mut active = self.active_turn.lock().await;
let mut turn = ActiveTurn::default();
turn.add_task(task);
+68 -7
View File
@@ -22,6 +22,7 @@ use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationModeMask;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::protocol::ForkReferenceItem;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::McpServerRefreshConfig;
use codex_protocol::protocol::Op;
@@ -361,16 +362,36 @@ impl ThreadManager {
path: PathBuf,
persist_extended_history: bool,
) -> CodexResult<NewThread> {
let history = RolloutRecorder::get_rollout_history(&path).await?;
let history = truncate_before_nth_user_message(history, nth_user_message);
self.state
.spawn_thread(
.fork_thread_with_source(
nth_user_message,
config,
history,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
Vec::new(),
persist_extended_history,
path,
self.state.session_source.clone(),
)
.await
}
/// Fork an existing thread while explicitly controlling the session source of the
/// forked thread.
pub async fn fork_thread_with_source(
&self,
nth_user_message: usize,
config: Config,
path: PathBuf,
session_source: SessionSource,
persist_extended_history: bool,
) -> CodexResult<NewThread> {
self.state
.fork_thread_with_source(
nth_user_message,
config,
self.agent_control(),
persist_extended_history,
path,
session_source,
)
.await
}
@@ -399,6 +420,15 @@ impl ThreadManagerState {
.ok_or_else(|| CodexErr::ThreadNotFound(thread_id))
}
/// Return a snapshot of all currently managed threads.
pub(crate) async fn list_threads(&self) -> Vec<(ThreadId, Arc<CodexThread>)> {
let threads = self.threads.read().await;
threads
.iter()
.map(|(thread_id, thread)| (*thread_id, Arc::clone(thread)))
.collect()
}
/// Send an operation to a thread by ID.
pub(crate) async fn send_op(&self, thread_id: ThreadId, op: Op) -> CodexResult<String> {
let thread = self.get_thread(thread_id).await?;
@@ -463,7 +493,6 @@ impl ThreadManagerState {
)
.await
}
/// Spawn a new thread with optional history and register it with the manager.
pub(crate) async fn spawn_thread(
&self,
@@ -552,6 +581,38 @@ impl ThreadManagerState {
pub(crate) fn notify_thread_created(&self, thread_id: ThreadId) {
let _ = self.thread_created_tx.send(thread_id);
}
pub(crate) async fn fork_thread_with_source(
&self,
nth_user_message: usize,
config: Config,
agent_control: AgentControl,
persist_extended_history: bool,
path: PathBuf,
session_source: SessionSource,
) -> CodexResult<NewThread> {
let history = RolloutRecorder::get_rollout_history(&path).await?;
let mut history = truncate_before_nth_user_message(history, nth_user_message);
if let InitialHistory::Forked(items) = &mut history {
items.insert(
0,
RolloutItem::ForkReference(ForkReferenceItem {
rollout_path: path.clone(),
nth_user_message,
}),
);
}
self.spawn_thread_with_source(
config,
history,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
Vec::new(),
persist_extended_history,
)
.await
}
}
/// Return a prefix of `items` obtained by cutting strictly before the nth user message
File diff suppressed because it is too large Load Diff
+151 -52
View File
@@ -552,7 +552,7 @@ fn create_view_image_tool() -> ToolSpec {
})
}
fn create_collab_input_items_schema() -> JsonSchema {
fn create_multi_agent_input_items_schema() -> JsonSchema {
let properties = BTreeMap::from([
(
"type".to_string(),
@@ -605,31 +605,47 @@ fn create_collab_input_items_schema() -> JsonSchema {
}
fn create_spawn_agent_tool(config: &ToolsConfig) -> ToolSpec {
let properties = BTreeMap::from([
(
"message".to_string(),
JsonSchema::String {
description: Some(
"Initial plain-text task for the new agent. Use either message or items."
.to_string(),
),
},
),
("items".to_string(), create_collab_input_items_schema()),
(
"agent_type".to_string(),
JsonSchema::String {
description: Some(crate::agent::role::spawn_tool_spec::build(
&config.agent_roles,
)),
},
),
]);
let mut properties = BTreeMap::new();
properties.insert(
"message".to_string(),
JsonSchema::String {
description: Some(
"Initial plain-text task for the new agent. Use either message or items."
.to_string(),
),
},
);
properties.insert("items".to_string(), create_multi_agent_input_items_schema());
properties.insert(
"agent_type".to_string(),
JsonSchema::String {
description: Some(crate::agent::role::spawn_tool_spec::build(
&config.agent_roles,
)),
},
);
properties.insert(
"spawn_mode".to_string(),
JsonSchema::String {
description: Some(
"Spawn behavior: fork (default), spawn (fresh context), or watchdog (idle-time check-ins). Roles may override the omitted-mode default. Watchdog mode returns a virtual handle, not a conversational worker. Watchdog check-ins are asynchronous and cannot occur until after the current turn ends and the owner thread is idle."
.to_string(),
),
},
);
properties.insert(
"interval_s".to_string(),
JsonSchema::Number {
description: Some(
"Watchdog interval in seconds; used only when spawn_mode = watchdog.".to_string(),
),
},
);
ToolSpec::Function(ResponsesApiTool {
name: "spawn_agent".to_string(),
description:
"Spawn a sub-agent for a well-scoped task. Returns the agent id (and user-facing nickname when available) to use to communicate with this agent."
"Spawn a sub-agent for a well-scoped task. Returns the agent id (and user-facing nickname when available) to use to communicate with this agent. Watchdog mode returns a control handle (not a conversational agent); watchdog check-ins are asynchronous and can only arrive after the current turn ends and the owner thread becomes idle. Do not wait/sleep/poll to force a watchdog check-in."
.to_string(),
strict: false,
parameters: JsonSchema::Object {
@@ -766,33 +782,35 @@ fn create_report_agent_job_result_tool() -> ToolSpec {
}
fn create_send_input_tool() -> ToolSpec {
let properties = BTreeMap::from([
(
"id".to_string(),
JsonSchema::String {
description: Some("Agent id to message (from spawn_agent).".to_string()),
},
),
(
"message".to_string(),
JsonSchema::String {
description: Some(
"Legacy plain-text message to send to the agent. Use either message or items."
.to_string(),
),
},
),
("items".to_string(), create_collab_input_items_schema()),
(
"interrupt".to_string(),
JsonSchema::Boolean {
description: Some(
"When true, stop the agent's current task and handle this immediately. When false (default), queue this message."
.to_string(),
),
},
),
]);
let mut properties = BTreeMap::new();
properties.insert(
"id".to_string(),
JsonSchema::String {
description: Some(
"Agent id to message (from spawn_agent). Optional: omit (or use \"parent\"/\"root\") to message the parent thread when available."
.to_string(),
),
},
);
properties.insert(
"message".to_string(),
JsonSchema::String {
description: Some(
"Legacy plain-text message to send to the agent. Use either message or items."
.to_string(),
),
},
);
properties.insert("items".to_string(), create_multi_agent_input_items_schema());
properties.insert(
"interrupt".to_string(),
JsonSchema::Boolean {
description: Some(
"When true, stop the agent's current task and handle this immediately. When false (default), queue this message."
.to_string(),
),
},
);
ToolSpec::Function(ResponsesApiTool {
name: "send_input".to_string(),
@@ -802,7 +820,7 @@ fn create_send_input_tool() -> ToolSpec {
strict: false,
parameters: JsonSchema::Object {
properties,
required: Some(vec!["id".to_string()]),
required: None,
additional_properties: Some(false.into()),
},
})
@@ -831,6 +849,83 @@ fn create_resume_agent_tool() -> ToolSpec {
})
}
fn create_compact_parent_context_tool() -> ToolSpec {
let mut properties = BTreeMap::new();
properties.insert(
"reason".to_string(),
JsonSchema::String {
description: Some(
"Optional short reason describing why the parent appears stuck (for example, repeated non-progress loop)."
.to_string(),
),
},
);
properties.insert(
"evidence".to_string(),
JsonSchema::String {
description: Some(
"Optional concrete evidence of non-progress (for example, repeated identical replies with no tool/file actions)."
.to_string(),
),
},
);
ToolSpec::Function(ResponsesApiTool {
name: "compact_parent_context".to_string(),
description: "Watchdog-only: request context compaction for the watchdog's parent thread when it is idle and appears stuck in a non-progress loop."
.to_string(),
strict: false,
parameters: JsonSchema::Object {
properties,
required: None,
additional_properties: Some(false.into()),
},
})
}
fn create_list_agents_tool() -> ToolSpec {
let mut properties = BTreeMap::new();
properties.insert(
"id".to_string(),
JsonSchema::String {
description: Some(
"Identifier of the parent agent whose spawned agents to list. Defaults to the current agent."
.to_string(),
),
},
);
properties.insert(
"recursive".to_string(),
JsonSchema::Boolean {
description: Some(
"When true (default), include all descendants recursively. When false, include only direct children."
.to_string(),
),
},
);
properties.insert(
"all".to_string(),
JsonSchema::Boolean {
description: Some(
"When true, ignore id/recursive and return all tracked open agents that currently count toward this session's agent limit."
.to_string(),
),
},
);
ToolSpec::Function(ResponsesApiTool {
name: "list_agents".to_string(),
description: "List agents spawned by an agent, optionally recursively. This is a status view, not a way to trigger watchdog check-ins; polling `list_agents` during the same turn will not make a watchdog fire."
.to_string(),
strict: false,
parameters: JsonSchema::Object {
properties,
required: None,
additional_properties: Some(false.into()),
},
})
}
fn create_wait_tool() -> ToolSpec {
let mut properties = BTreeMap::new();
properties.insert(
@@ -838,7 +933,7 @@ fn create_wait_tool() -> ToolSpec {
JsonSchema::Array {
items: Box::new(JsonSchema::String { description: None }),
description: Some(
"Agent ids to wait on. Pass multiple ids to wait for whichever finishes first."
"Agent ids to wait on. Pass multiple ids to wait for whichever finishes first. Watchdog handle ids cannot be waited on for new check-ins. If all ids are watchdog handles, wait returns an immediate correction instead of blocking. If mixed with normal agent ids, wait still waits on normal agents and includes current watchdog handle statuses. Do not use `list_agents` or shell `sleep` to poll for same-turn watchdog check-ins."
.to_string(),
),
},
@@ -854,7 +949,7 @@ fn create_wait_tool() -> ToolSpec {
ToolSpec::Function(ResponsesApiTool {
name: "wait".to_string(),
description: "Wait for agents to reach a final status. Completed statuses may include the agent's final message. Returns empty status when timed out. Once the agent reaches his final status, a notification message will be received containing the same completed status."
description: "Wait for agents to reach a final status. Completed statuses may include the agent's final message. Returns empty status when timed out. Watchdog handles are status-only in this tool: wait cannot block for watchdog check-ins. If only watchdog handles are provided, wait returns an immediate correction; if mixed with normal agents, wait includes current watchdog handle statuses while waiting on normal agents. Sleeping or polling cannot make a watchdog check-in happen while the current turn is active."
.to_string(),
strict: false,
parameters: JsonSchema::Object {
@@ -1803,11 +1898,15 @@ pub(crate) fn build_specs(
builder.push_spec(create_spawn_agent_tool(config));
builder.push_spec(create_send_input_tool());
builder.push_spec(create_resume_agent_tool());
builder.push_spec(create_compact_parent_context_tool());
builder.push_spec(create_list_agents_tool());
builder.push_spec(create_wait_tool());
builder.push_spec(create_close_agent_tool());
builder.register_handler("spawn_agent", multi_agent_handler.clone());
builder.register_handler("send_input", multi_agent_handler.clone());
builder.register_handler("resume_agent", multi_agent_handler.clone());
builder.register_handler("compact_parent_context", multi_agent_handler.clone());
builder.register_handler("list_agents", multi_agent_handler.clone());
builder.register_handler("wait", multi_agent_handler.clone());
builder.register_handler("close_agent", multi_agent_handler);
}
+39
View File
@@ -0,0 +1,39 @@
# You are a Subagent
You are a **subagent** in a multi-agent Codex session. Your role is no longer root. Your goal is the task given by the parent/root agent.
Term definitions in this file:
- **parent thread**: the thread that spawned this subagent.
- **root thread**: the top-level user-facing thread.
- **`send_input`**: send a message to an existing agent thread; it does not spawn agents. Delivery is asynchronous.
- **durable state**: thread-level task state needed across later turns/check-ins (not disk/database persistence).
- In this runtime, when a subagent calls `send_input`, `id = "parent"` and `id = "root"` both route to the immediate parent thread.
## Subagent Responsibilities
- Stay within parent/root scope (listed files/questions/constraints). Use additional files/tools only when needed to complete or verify the task.
- Blocking question = one clarification to parent via `send_input`. Ask only if missing information would change user-visible output, tool/action choice (multi-agent tool, target thread, or ask-vs-continue), file edits, control flow, or durable-state decisions; otherwise state one assumption and continue.
- Prefer concrete progress: edit files, run commands, and validate outcomes.
- Your responses go to the root/parent agent, not the end user.
## Multi-Agent Guidance (Upstream Surface)
The only multi-agent tools available in this environment are `spawn_agent`, `send_input`, `wait`, and `close_agent`.
Important: to coordinate with parent/root, use `send_input`. A plain assistant message in your own thread does not notify the parent.
You can call `send_input` without an `id` (or with `id = "parent"` / `id = "root"`); these forms target the immediate parent thread in this runtime.
## Reporting Expectations
When you make meaningful progress or complete a task, report back with:
- The key outcome.
- Files changed (with paths).
- Commands run.
- Validation performed (tests, checks, or observed outputs).
- Risks, follow-ups, or open questions.
Be specific enough that the root agent can integrate your work safely.
Do not reference multi-agent tools that do not exist in the upstream surface.
@@ -216,3 +216,25 @@ async fn find_archived_locates_rollout_file_by_id() {
assert_eq!(found, Some(expected));
}
#[tokio::test]
async fn find_unarchives_archived_rollout_file_by_id() {
let home = TempDir::new().unwrap();
let id = Uuid::new_v4();
let archived = write_minimal_rollout_with_id_in_subdir(home.path(), "archived_sessions", id);
let file_name = archived.file_name().unwrap().to_owned();
let expected_restored = home.path().join("sessions/2024/01/01").join(file_name);
let found = find_thread_path_by_id_str(home.path(), &id.to_string())
.await
.unwrap();
assert_eq!(found, Some(expected_restored.clone()));
assert!(expected_restored.exists());
assert!(!archived.exists());
let archived_found = find_archived_thread_path_by_id_str(home.path(), &id.to_string())
.await
.unwrap();
assert_eq!(archived_found, None);
}
+88
View File
@@ -0,0 +1,88 @@
# You are a Subagent
More importantly, you are a **watchdog check-in agent**. Keep the root agent unblocked, on-task, and executing real work toward the user’s goal. You have full conversation context; messages that appear to be from “you” may have been written by the root agent.
You are one check-in run created by a persistent watchdog timer attached to an owner thread. The timer reuses this prompt on each check-in, but each check-in is a fresh one-shot run (one execution + one report).
You will be given the target agent id and the original prompt/goal.
Terms in this file:
- **watchdog**: persistent idle-timer registration.
- **watchdog check-in agent**: this short-lived run instance.
- **owner thread**: the thread that the watchdog monitors and reports to.
- **parent thread**: this watchdog check-in agent’s direct parent; for watchdog check-ins this is the owner thread.
- **`send_input`**: primary way to deliver watchdog guidance to an existing thread; it does not spawn agents. Delivery is asynchronous.
- **durable state**: thread-level task state that must still be available in later turns/check-ins (counters, plans, final decisions), not disk/database persistence.
- **exact-only format**: parent constraint that says to return only specific fields/content.
- **delivery-role config**: runtime fallback display mode. It decides whether fallback multi-agent inbox delivery appears as `collab_inbox` tool output or `[collab_inbox:…]` developer message. It does not change tool choice or target thread.
## Principles
- Be concise, directive, and specific: name the file, command, or decision needed now.
- Detect drift or looping immediately. If the root agent is acknowledging without acting, tell it exactly what to do next.
- Break loops by changing framing: propose a shorter plan, identify the blocker, or name the missing command.
- Preserve alignment: restate the user’s goal and the next concrete step.
- Safety and correctness: call out missing tests, skipped checks, or unclear acceptance criteria.
- Output precedence is: system/developer/policy rules first, then parent-task output constraints. If the parent requires exact-only format (for example "only"), return exactly the requested fields/content unless higher-priority rules require extra content.
- If exact-only format is not required, include all requested fields/content and you may add 1-2 short non-conflicting guidance sentences.
## Operating Procedure (Every Time You Run)
1. Re-evaluate the user’s latest request and the current status. Independently verify status when needed by reading files, running commands, and checking plan files against recent changes.
2. Identify the single highest-impact next action (or a very short ordered list).
3. Direct the root agent to execute it now (include paths and commands).
4. If blocked, propose one or two crisp unblockers.
5. If the goal appears complete, say so and direct the root agent to close unneeded agents.
Tone: direct, actionable, minimally polite. Optimize for progress over narration.
## Detect Looping and Reward Hacking
The root agent may slip into patterns that look like progress but are not. Interrupt those patterns.
Watch for:
- Tests that always pass (tautologies, `assert!(true)`, mocks that cannot fail).
- Marking items complete with only stub implementations.
- "Fixes" that comment out failing tests or code without addressing root causes.
- Claiming success without running required format/lint/tests.
- Ignoring explicit user requirements in favor of quicker but incomplete shortcuts.
When you detect these, prescribe the corrective action explicitly.
## Multi-Agent Tools (Upstream Surface)
Use only the multi-agent tools that exist here:
- `spawn_agent` (prefer `spawn_mode = "fork"` when shared context matters).
- `send_input`.
- `compact_parent_context` (watchdog-only recovery tool; see below).
- `wait`.
- `close_agent`.
There is no cancel tool. Use `close_agent` to stop agents that are done or no longer needed.
When recommending watchdogs to the root agent, keep `agent_type` at the default.
Important: send watchdog check-in output with `send_input` to the owner/parent thread. A plain assistant message in your own watchdog check-in thread is not a reliable delivery path to the owner.
Each watchdog check-in runs in a fresh one-shot watchdog check-in agent with no guaranteed continuity across check-ins. Do not keep durable state in watchdog-check-in-agent local memory/files; treat local state as run-local only. Ask the parent to track durable state, and use `send_input` (without `id`, or `id = "parent"`/`"root"`) to report results.
`send_input` is the primary path for watchdog delivery to parent/owner. If a watchdog check-in agent finishes without `send_input`, runtime may forward one final multi-agent inbox message (for example `collab_inbox` tool output or `[collab_inbox:…]` developer message), depending on delivery-role config. This fallback is best-effort.
For token protocols (for example `ping N` / `pong N`), treat those as literal text counters, not shell commands. Do not call command-execution tools unless the prompt explicitly asks you to execute commands.
## Parent Recovery via Context Compaction
`compact_parent_context` asks the system to abbreviate/compact redundant parent-thread context so the parent can recover from loops.
Use it only as a last resort:
- The parent has been repeatedly non-responsive across multiple watchdog check-ins.
- The parent is taking no meaningful actions (no concrete commands/edits/tests) and making no progress.
- You already sent at least one direct corrective instruction with `send_input`, and it was ignored.
Do not call `compact_parent_context` for routine nudges or normal delays. Prefer precise `send_input` guidance first.
## Style
Be explicit when precision matters. Your job is to drive real progress toward the user’s goal.
@@ -1,3 +1,30 @@
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::AgentReasoningEvent;
use codex_core::protocol::AgentStatus;
use codex_core::protocol::AskForApproval;
use codex_core::protocol::CollabAgentSpawnBeginEvent;
use codex_core::protocol::CollabAgentSpawnEndEvent;
use codex_core::protocol::CollabAgentSpawnMode;
use codex_core::protocol::CollabWaitingEndEvent;
use codex_core::protocol::ErrorEvent;
use codex_core::protocol::Event;
use codex_core::protocol::EventMsg;
use codex_core::protocol::ExecCommandBeginEvent;
use codex_core::protocol::ExecCommandEndEvent;
use codex_core::protocol::ExecCommandSource;
use codex_core::protocol::ExecCommandStatus as CoreExecCommandStatus;
use codex_core::protocol::FileChange;
use codex_core::protocol::McpInvocation;
use codex_core::protocol::McpToolCallBeginEvent;
use codex_core::protocol::McpToolCallEndEvent;
use codex_core::protocol::PatchApplyBeginEvent;
use codex_core::protocol::PatchApplyEndEvent;
use codex_core::protocol::PatchApplyStatus as CorePatchApplyStatus;
use codex_core::protocol::SandboxPolicy;
use codex_core::protocol::SessionConfiguredEvent;
use codex_core::protocol::WarningEvent;
use codex_core::protocol::WebSearchBeginEvent;
use codex_core::protocol::WebSearchEndEvent;
use codex_exec::event_processor_with_jsonl_output::EventProcessorWithJsonOutput;
use codex_exec::exec_events::AgentMessageItem;
use codex_exec::exec_events::CollabAgentState;
@@ -575,6 +602,7 @@ fn collab_spawn_begin_and_end_emit_item_events() {
new_agent_nickname: None,
new_agent_role: None,
prompt: prompt.clone(),
spawn_mode: CollabAgentSpawnMode::Spawn,
status: AgentStatus::Running,
}),
);
+15
View File
@@ -73,6 +73,11 @@ pub enum ResponseInputItem {
role: String,
content: Vec<ContentItem>,
},
FunctionCall {
name: String,
arguments: String,
call_id: String,
},
FunctionCallOutput {
call_id: String,
output: FunctionCallOutputPayload,
@@ -664,6 +669,16 @@ impl From<ResponseInputItem> for ResponseItem {
end_turn: None,
phase: None,
},
ResponseInputItem::FunctionCall {
name,
arguments,
call_id,
} => Self::FunctionCall {
id: None,
name,
arguments,
call_id,
},
ResponseInputItem::FunctionCallOutput { call_id, output } => {
Self::FunctionCallOutput { call_id, output }
}
+50
View File
@@ -33,6 +33,7 @@ use crate::message_history::HistoryEntry;
use crate::models::BaseInstructions;
use crate::models::ContentItem;
use crate::models::MessagePhase;
use crate::models::ResponseInputItem;
use crate::models::ResponseItem;
use crate::models::WebSearchAction;
use crate::num_format::format_with_separators;
@@ -72,6 +73,27 @@ pub const ENVIRONMENT_CONTEXT_CLOSE_TAG: &str = "</environment_context>";
pub const COLLABORATION_MODE_OPEN_TAG: &str = "<collaboration_mode>";
pub const COLLABORATION_MODE_CLOSE_TAG: &str = "</collaboration_mode>";
pub const USER_MESSAGE_BEGIN: &str = "## My request for Codex:";
pub const COLLAB_INBOX_KIND: &str = "collab_inbox";
pub const COLLAB_INBOX_MESSAGE_PREFIX: &str = "[collab_inbox:";
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema)]
pub struct CollabInboxPayload {
pub injected: bool,
pub kind: String,
pub sender_thread_id: ThreadId,
pub message: String,
}
impl CollabInboxPayload {
pub fn new(sender_thread_id: ThreadId, message: String) -> Self {
Self {
injected: true,
kind: COLLAB_INBOX_KIND.to_string(),
sender_thread_id,
message,
}
}
}
/// Submission Queue Entry - requests from user
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)]
@@ -161,6 +183,14 @@ pub enum Op {
final_output_json_schema: Option<Value>,
},
/// Inject non-user input items (tool output or assistant/developer messages).
///
/// Used for agent-to-agent inbox delivery without re-encoding as a user message.
InjectResponseItems {
/// Response input items to inject and optionally run.
items: Vec<ResponseInputItem>,
},
/// Similar to [`Op::UserInput`], but contains additional context required
/// for a turn of a [`crate::codex_thread::CodexThread`].
UserTurn {
@@ -2071,10 +2101,17 @@ pub struct SessionMetaLine {
pub git: Option<GitInfo>,
}
#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS)]
pub struct ForkReferenceItem {
pub rollout_path: PathBuf,
pub nth_user_message: usize,
}
#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS)]
#[serde(tag = "type", content = "payload", rename_all = "snake_case")]
pub enum RolloutItem {
SessionMeta(SessionMetaLine),
ForkReference(ForkReferenceItem),
ResponseItem(ResponseItem),
Compacted(CompactedItem),
TurnContext(TurnContextItem),
@@ -2855,6 +2892,16 @@ pub struct CollabAgentSpawnBeginEvent {
pub prompt: String,
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS, Default)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum CollabAgentSpawnMode {
#[default]
Spawn,
Fork,
Watchdog,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct CollabAgentRef {
/// Thread ID of the receiver/new agent.
@@ -2898,6 +2945,9 @@ pub struct CollabAgentSpawnEndEvent {
/// Initial prompt sent to the agent. Can be empty to prevent CoT leaking at the
/// beginning.
pub prompt: String,
/// Spawn mode used for this agent.
#[serde(default)]
pub spawn_mode: CollabAgentSpawnMode,
/// Last known status of the new agent reported to the sender agent.
pub status: AgentStatus,
}
+1
View File
@@ -22,6 +22,7 @@ pub fn apply_rollout_item(
RolloutItem::TurnContext(turn_ctx) => apply_turn_context(metadata, turn_ctx),
RolloutItem::EventMsg(event) => apply_event_msg(metadata, event),
RolloutItem::ResponseItem(item) => apply_response_item(metadata, item),
RolloutItem::ForkReference(_) => {}
RolloutItem::Compacted(_) => {}
}
if metadata.model_provider.is_empty() {
+1
View File
@@ -1555,6 +1555,7 @@ fn extract_dynamic_tools(items: &[RolloutItem]) -> Option<Option<Vec<DynamicTool
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.dynamic_tools.clone()),
RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::ForkReference(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => None,
})
+935 -13
View File
File diff suppressed because it is too large Load Diff
+16
View File
@@ -9,6 +9,7 @@
//! quits without reaching into the app loop or coupling to shutdown/exit sequencing.
use std::path::PathBuf;
use std::sync::Arc;
use codex_chatgpt::connectors::AppInfo;
use codex_file_search::FileMatch;
@@ -21,6 +22,7 @@ use codex_utils_approval_presets::ApprovalPreset;
use crate::bottom_pane::ApprovalRequest;
use crate::bottom_pane::StatusLineItem;
use crate::history_cell::HistoryCell;
use crate::history_cell::SubagentStatusCell;
use codex_core::features::Feature;
use codex_protocol::config_types::CollaborationModeMask;
@@ -46,6 +48,15 @@ pub(crate) struct ConnectorsSnapshot {
#[derive(Debug)]
pub(crate) enum AppEvent {
CodexEvent(Event),
/// Forward an event for a specific thread id.
///
/// Used by background listeners for non-active threads so the app can keep
/// subagent status/panel state in sync even when those threads are not the
/// focused chat thread.
CodexThreadEvent {
thread_id: ThreadId,
event: Event,
},
/// Open the agent picker for switching active threads.
OpenAgentPicker,
/// Switch the active thread to the selected agent.
@@ -139,6 +150,11 @@ pub(crate) enum AppEvent {
StartCommitAnimation,
StopCommitAnimation,
CommitTick,
StartSubagentAnimation,
StopSubagentAnimation,
SubagentTick,
UpdateSubagentPanel(Arc<SubagentStatusCell>),
ClearSubagentPanel,
/// Update the current reasoning effort in the running app and widget.
UpdateReasoningEffort(Option<ReasoningEffort>),
+113 -38
View File
@@ -80,7 +80,9 @@ use codex_protocol::config_types::Settings;
#[cfg(target_os = "windows")]
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::items::AgentMessageItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::ResponseItem;
use codex_protocol::models::local_image_label_text;
use codex_protocol::parse_command::ParsedCommand;
use codex_protocol::protocol::AgentMessageDeltaEvent;
@@ -91,7 +93,10 @@ use codex_protocol::protocol::AgentReasoningRawContentDeltaEvent;
use codex_protocol::protocol::AgentReasoningRawContentEvent;
use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
use codex_protocol::protocol::BackgroundEventEvent;
use codex_protocol::protocol::COLLAB_INBOX_KIND;
use codex_protocol::protocol::COLLAB_INBOX_MESSAGE_PREFIX;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::CollabInboxPayload;
use codex_protocol::protocol::CreditsSnapshot;
use codex_protocol::protocol::DeprecationNoticeEvent;
use codex_protocol::protocol::ErrorEvent;
@@ -114,6 +119,7 @@ use codex_protocol::protocol::McpToolCallEndEvent;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::PatchApplyBeginEvent;
use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::RawResponseItemEvent;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::ReviewTarget;
use codex_protocol::protocol::SkillMetadata as ProtocolSkillMetadata;
@@ -230,6 +236,7 @@ use crate::history_cell::AgentMessageCell;
use crate::history_cell::HistoryCell;
use crate::history_cell::McpToolCallCell;
use crate::history_cell::PlainHistoryCell;
use crate::history_cell::SubagentStatusCell;
use crate::history_cell::WebSearchCell;
use crate::key_hint;
use crate::key_hint::KeyBinding;
@@ -245,6 +252,7 @@ use crate::slash_command::SlashCommand;
use crate::status::RateLimitSnapshotDisplay;
use crate::status_indicator_widget::STATUS_DETAILS_DEFAULT_MAX_LINES;
use crate::status_indicator_widget::StatusDetailsCapitalization;
use crate::text_formatting::extract_first_bold;
use crate::text_formatting::truncate_text;
use crate::tui::FrameRequester;
mod interrupts;
@@ -347,6 +355,37 @@ fn is_unified_exec_source(source: ExecCommandSource) -> bool {
)
}
fn collab_inbox_message_from_item(item: &ResponseItem) -> Option<(Option<String>, String)> {
match item {
ResponseItem::FunctionCallOutput { output, .. } => {
let text = output.body.to_text()?;
let payload: CollabInboxPayload = serde_json::from_str(&text).ok()?;
if !payload.injected || payload.kind != COLLAB_INBOX_KIND {
return None;
}
Some((Some(payload.sender_thread_id.to_string()), payload.message))
}
ResponseItem::Message { content, .. } => {
let text = content.iter().find_map(|item| match item {
ContentItem::InputText { text } | ContentItem::OutputText { text } => {
Some(text.as_str())
}
_ => None,
})?;
let rest = text.strip_prefix(COLLAB_INBOX_MESSAGE_PREFIX)?;
let (sender, message) = rest.split_once(']')?;
let message = message.trim_start().to_string();
let sender = sender.trim().to_string();
if sender.is_empty() {
Some((None, message))
} else {
Some((Some(sender), message))
}
}
_ => None,
}
}
fn is_standard_tool_call(parsed_cmd: &[ParsedCommand]) -> bool {
!parsed_cmd.is_empty()
&& parsed_cmd
@@ -1273,6 +1312,23 @@ impl ChatWidget {
self.request_redraw();
}
fn on_collab_event(&mut self, cell: PlainHistoryCell) {
self.flush_answer_stream_with_separator();
self.add_to_history(cell);
self.request_redraw();
}
fn on_raw_response_item(&mut self, event: RawResponseItemEvent) {
if let Some((sender, message)) = collab_inbox_message_from_item(&event.item) {
let hint = sender.map(|sender| format!("from {sender}"));
self.add_to_history(history_cell::new_info_event(
format!("Agent message: {message}"),
hint,
));
self.request_redraw();
}
}
fn on_agent_message_delta(&mut self, delta: String) {
self.handle_streaming_delta(delta);
}
@@ -2189,12 +2245,6 @@ impl ChatWidget {
self.had_work_activity = true;
}
fn on_collab_event(&mut self, cell: PlainHistoryCell) {
self.flush_answer_stream_with_separator();
self.add_to_history(cell);
self.request_redraw();
}
fn on_get_history_entry_response(
&mut self,
event: codex_protocol::protocol::GetHistoryEntryResponseEvent,
@@ -2293,6 +2343,53 @@ impl ChatWidget {
self.run_commit_tick();
}
pub(crate) fn on_subagent_panel_updated(&mut self, panel: Arc<SubagentStatusCell>) {
let state_handle = panel.state_handle();
if let Some(active) = self.active_cell.as_mut()
&& let Some(existing) = active.as_any_mut().downcast_mut::<SubagentStatusCell>()
{
if existing.matches_state(&state_handle) {
self.bump_active_cell_revision();
self.request_redraw();
return;
}
*existing = panel.as_ref().clone();
self.bump_active_cell_revision();
self.request_redraw();
return;
}
if self.active_cell.is_none() {
self.active_cell = Some(Box::new(panel.as_ref().clone()));
self.bump_active_cell_revision();
self.request_redraw();
}
}
pub(crate) fn clear_subagent_panel(&mut self) {
if self
.active_cell
.as_ref()
.is_some_and(|cell| cell.as_any().is::<SubagentStatusCell>())
{
self.active_cell = None;
self.bump_active_cell_revision();
self.request_redraw();
}
}
pub(crate) fn on_subagent_tick(&mut self) {
if self
.active_cell
.as_ref()
.is_some_and(|cell| cell.as_any().is::<SubagentStatusCell>())
{
self.bump_active_cell_revision();
self.request_redraw();
}
}
/// Runs a regular periodic commit tick.
fn run_commit_tick(&mut self) {
self.run_commit_tick_with_scope(CommitTickScope::AnyMode);
@@ -3895,6 +3992,14 @@ impl ChatWidget {
fn flush_active_cell(&mut self) {
if let Some(active) = self.active_cell.take() {
// Subagent status is a transient panel, not transcript history. If we
// flush it into history every time another cell is inserted, the
// transcript gets spammed with repeated identical "Subagents ..." blocks.
// Keep the panel mounted so later transcript cells do not make it disappear.
if active.as_any().is::<SubagentStatusCell>() {
self.active_cell = Some(active);
return;
}
self.needs_final_message_separator = true;
self.app_event_tx.send(AppEvent::InsertHistoryCell(active));
}
@@ -4408,8 +4513,8 @@ impl ChatWidget {
});
}
}
EventMsg::RawResponseItem(_)
| EventMsg::ItemStarted(_)
EventMsg::RawResponseItem(ev) => self.on_raw_response_item(ev),
EventMsg::ItemStarted(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::ReasoningContentDelta(_)
| EventMsg::ReasoningRawContentDelta(_)
@@ -7775,36 +7880,6 @@ const PLACEHOLDERS: [&str; 8] = [
"Use /skills to list available skills",
];
// Extract the first bold (Markdown) element in the form **...** from `s`.
// Returns the inner text if found; otherwise `None`.
fn extract_first_bold(s: &str) -> Option<String> {
let bytes = s.as_bytes();
let mut i = 0usize;
while i + 1 < bytes.len() {
if bytes[i] == b'*' && bytes[i + 1] == b'*' {
let start = i + 2;
let mut j = start;
while j + 1 < bytes.len() {
if bytes[j] == b'*' && bytes[j + 1] == b'*' {
// Found closing **
let inner = &s[start..j];
let trimmed = inner.trim();
if !trimmed.is_empty() {
return Some(trimmed.to_string());
} else {
return None;
}
}
j += 1;
}
// No closing; stop searching (wait for more deltas)
return None;
}
i += 1;
}
None
}
async fn fetch_rate_limits(base_url: String, auth: CodexAuth) -> Vec<RateLimitSnapshot> {
match BackendClient::from_auth(base_url, &auth) {
Ok(client) => match client.get_rate_limits_many().await {
+45
View File
@@ -11,6 +11,9 @@ use crate::app_event_sender::AppEventSender;
use crate::bottom_pane::FeedbackAudience;
use crate::bottom_pane::LocalImageAttachment;
use crate::bottom_pane::MentionBinding;
use crate::history_cell::SubagentPanelAgent;
use crate::history_cell::SubagentPanelState;
use crate::history_cell::SubagentStatusCell;
use crate::history_cell::UserHistoryCell;
use crate::test_backend::VT100Backend;
use crate::tui::FrameRequester;
@@ -50,6 +53,7 @@ use codex_protocol::protocol::AgentMessageDeltaEvent;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::AgentReasoningDeltaEvent;
use codex_protocol::protocol::AgentReasoningEvent;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
use codex_protocol::protocol::BackgroundEventEvent;
use codex_protocol::protocol::CodexErrorInfo;
@@ -103,6 +107,9 @@ use serial_test::serial;
use std::collections::BTreeMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
use std::time::Instant;
use tempfile::NamedTempFile;
use tempfile::tempdir;
use tokio::sync::mpsc::error::TryRecvError;
@@ -1824,6 +1831,44 @@ fn lines_to_single_string(lines: &[ratatui::text::Line<'static>]) -> String {
s
}
#[tokio::test]
async fn subagent_panel_is_not_flushed_into_transcript_history() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
let state = Arc::new(StdMutex::new(SubagentPanelState {
started_at: Instant::now(),
total_agents: 1,
running_count: 0,
running_agents: vec![SubagentPanelAgent {
ordinal: 1,
name: "user-request-derisk-implement".to_string(),
status: AgentStatus::PendingInit,
is_watchdog: true,
preview: "watchdog idle".to_string(),
latest_update_at: Instant::now(),
}],
}));
chat.on_subagent_panel_updated(Arc::new(SubagentStatusCell::new(state, true)));
chat.add_to_history(history_cell::new_error_event("follow-up cell".to_string()));
let inserted = drain_insert_history(&mut rx);
assert_eq!(
inserted.len(),
1,
"subagent panel should remain transient and not be inserted into transcript history"
);
let rendered = lines_to_single_string(&inserted[0]);
assert!(rendered.contains("follow-up cell"));
assert!(!rendered.contains("Subagents"));
assert!(
chat.active_cell
.as_ref()
.is_some_and(|cell| cell.as_any().is::<SubagentStatusCell>()),
"subagent panel should stay mounted after other history cells are inserted"
);
}
fn make_token_info(total_tokens: i64, context_window: i64) -> TokenUsageInfo {
fn usage(total_tokens: i64) -> TokenUsage {
TokenUsage {
+315
View File
@@ -25,6 +25,8 @@ use crate::render::line_utils::line_to_static;
use crate::render::line_utils::prefix_lines;
use crate::render::line_utils::push_owned_lines;
use crate::render::renderable::Renderable;
use crate::shimmer::shimmer_spans;
use crate::status_indicator_widget::fmt_elapsed_compact;
use crate::style::proposed_plan_style;
use crate::style::user_message_style;
use crate::text_formatting::format_and_truncate_tool_result;
@@ -50,6 +52,7 @@ use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::plan_tool::PlanItemArg;
use codex_protocol::plan_tool::StepStatus;
use codex_protocol::plan_tool::UpdatePlanArgs;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::McpAuthStatus;
use codex_protocol::protocol::McpInvocation;
@@ -73,6 +76,8 @@ use std::collections::HashMap;
use std::io::Cursor;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use std::time::Instant;
use tracing::error;
@@ -477,6 +482,225 @@ impl HistoryCell for PlainHistoryCell {
}
}
#[derive(Clone, Debug)]
pub(crate) struct SubagentPanelAgent {
pub(crate) ordinal: i32,
pub(crate) name: String,
pub(crate) status: AgentStatus,
pub(crate) is_watchdog: bool,
pub(crate) preview: String,
pub(crate) latest_update_at: Instant,
}
#[derive(Clone, Debug)]
pub(crate) struct SubagentPanelState {
pub(crate) started_at: Instant,
pub(crate) total_agents: i32,
pub(crate) running_count: i32,
pub(crate) running_agents: Vec<SubagentPanelAgent>,
}
impl SubagentPanelState {
pub(crate) fn running_count(&self) -> i32 {
self.running_count
}
pub(crate) fn has_animating_agents(&self, now: Instant) -> bool {
self.running_agents
.iter()
.any(|agent| should_shimmer(agent, now))
}
}
#[derive(Clone, Debug)]
pub(crate) struct SubagentStatusCell {
state: Arc<Mutex<SubagentPanelState>>,
animations_enabled: bool,
}
impl SubagentStatusCell {
pub(crate) fn new(
state: Arc<Mutex<SubagentPanelState>>,
animations_enabled: bool,
) -> SubagentStatusCell {
SubagentStatusCell {
state,
animations_enabled,
}
}
pub(crate) fn state_handle(&self) -> Arc<Mutex<SubagentPanelState>> {
Arc::clone(&self.state)
}
pub(crate) fn matches_state(&self, other: &Arc<Mutex<SubagentPanelState>>) -> bool {
Arc::ptr_eq(&self.state, other)
}
}
impl HistoryCell for SubagentStatusCell {
fn display_lines(&self, width: u16) -> Vec<Line<'static>> {
let state = {
let guard = self.state.lock().expect("subagent panel state lock");
guard.clone()
};
if state.running_agents.is_empty() {
return Vec::new();
}
let elapsed = fmt_elapsed_compact(state.started_at.elapsed().as_secs());
let running_count = state.running_count();
let total_agents = state.total_agents.max(running_count);
let count_label = subagent_count_label(total_agents, running_count);
let header_suffix = format!("({elapsed} • {count_label} • esc to interrupt)");
let mut lines = Vec::new();
lines.push(Line::from(vec![
"• ".dim(),
"Subagents".bold(),
" ".into(),
header_suffix.dim(),
]));
let mut running_agents = state.running_agents;
running_agents.sort_by(|left, right| left.ordinal.cmp(&right.ordinal));
let preview_budget = running_preview_budget(width);
let now = Instant::now();
lines.extend(running_agents.into_iter().map(|agent| {
let preview = truncate_text(agent.preview.trim(), preview_budget);
let mut spans: Vec<Span<'static>> =
vec!["• ".dim(), format!("[#{}] ", agent.ordinal).dim()];
if agent.is_watchdog {
spans.push("[watchdog] ".magenta().dim());
}
spans.push(Span::from(agent.name.clone()));
spans.push(" ".into());
spans.push(status_span_for_panel(&agent));
spans.push(" — ".dim());
if self.animations_enabled && should_shimmer(&agent, now) {
spans.extend(shimmer_spans(&preview));
} else {
spans.push(Span::from(preview));
}
Line::from(spans)
}));
lines
}
fn transcript_animation_tick(&self) -> Option<u64> {
if !self.animations_enabled {
return None;
}
let guard = self.state.lock().expect("subagent panel state lock");
let now = Instant::now();
if !guard.has_animating_agents(now) {
return None;
}
Some((now.duration_since(guard.started_at).as_millis() / 100) as u64)
}
}
pub(crate) fn new_subagent_spawned_cell(name: &str, prompt_preview: &str) -> PlainHistoryCell {
let mut lines = Vec::new();
lines.push(Line::from(vec![
"• ".dim(),
"Spawned subagent ".into(),
Span::from(name.to_string()).bold(),
]));
let preview = truncate_text(prompt_preview.trim(), 240);
if !preview.is_empty() {
lines.push(Line::from(vec![
" └ ".dim(),
Span::from(format!("\"{preview}\"")).dim(),
]));
}
PlainHistoryCell::new(lines)
}
pub(crate) fn new_subagent_update_cell(
name: &str,
status: &AgentStatus,
summary: &str,
) -> PlainHistoryCell {
let mut spans: Vec<Span<'static>> = vec![
"• ".dim(),
"Subagent update: ".into(),
Span::from(name.to_string()).bold(),
" ".into(),
status_label_span(status),
];
let summary = truncate_text(summary.trim(), 240);
if !summary.is_empty() {
spans.push(" — ".dim());
spans.push(Span::from(summary));
}
PlainHistoryCell::new(vec![Line::from(spans)])
}
fn running_preview_budget(width: u16) -> usize {
let width = width as usize;
width.saturating_sub(24).clamp(60, 160)
}
fn is_running_status(status: &AgentStatus) -> bool {
matches!(status, AgentStatus::PendingInit | AgentStatus::Running)
}
fn status_span_for_panel(agent: &SubagentPanelAgent) -> Span<'static> {
match &agent.status {
AgentStatus::PendingInit if agent.is_watchdog => "idle".dim(),
AgentStatus::PendingInit | AgentStatus::Running => "running".cyan().bold(),
AgentStatus::Completed(_) => "completed".green(),
AgentStatus::Errored(_) => "errored".red(),
AgentStatus::Shutdown => "shutdown".dim(),
AgentStatus::NotFound => "not found".red(),
}
}
const SUBAGENT_SHIMMER_WINDOW: Duration = Duration::from_secs(1);
fn should_shimmer(agent: &SubagentPanelAgent, now: Instant) -> bool {
if agent.is_watchdog && matches!(agent.status, AgentStatus::PendingInit) {
return false;
}
is_running_status(&agent.status)
&& now.saturating_duration_since(agent.latest_update_at) <= SUBAGENT_SHIMMER_WINDOW
}
fn status_label_span(status: &AgentStatus) -> Span<'static> {
match status {
AgentStatus::PendingInit | AgentStatus::Running => "running".cyan().bold(),
AgentStatus::Completed(_) => "completed".green(),
AgentStatus::Errored(_) => "errored".red(),
AgentStatus::Shutdown => "shutdown".dim(),
AgentStatus::NotFound => "not found".red(),
}
}
fn subagent_count_label(total: i32, running: i32) -> String {
if total <= 0 || running <= 0 {
return "no subagents running".to_string();
}
let total_label = subagent_pluralize(total, "subagent");
if running >= total {
return format!("{total_label} running");
}
format!("{total_label}, {running} running")
}
fn subagent_pluralize(count: i32, singular: &str) -> String {
if count == 1 {
format!("1 {singular}")
} else {
format!("{count} {singular}s")
}
}
#[cfg_attr(debug_assertions, allow(dead_code))]
#[derive(Debug)]
pub(crate) struct UpdateAvailableHistoryCell {
@@ -2404,6 +2628,9 @@ mod tests {
use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Instant;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::mcp::Tool;
@@ -2436,6 +2663,94 @@ mod tests {
render_lines(&cell.transcript_lines(u16::MAX))
}
#[test]
fn subagent_panel_renders_watchdog_handle_as_idle() {
let state = Arc::new(Mutex::new(SubagentPanelState {
started_at: Instant::now(),
total_agents: 1,
running_count: 0,
running_agents: vec![SubagentPanelAgent {
ordinal: 1,
name: "watchdog-agent".to_string(),
status: AgentStatus::PendingInit,
is_watchdog: true,
preview: "monitor parent progress".to_string(),
latest_update_at: Instant::now(),
}],
}));
let cell = SubagentStatusCell::new(state, true);
let lines = render_lines(&cell.display_lines(120));
assert!(lines[0].contains("no subagents running"));
assert!(lines[1].contains("[watchdog] watchdog-agent idle"));
}
#[test]
fn subagent_panel_animation_tick_ignores_idle_watchdogs() {
let state = Arc::new(Mutex::new(SubagentPanelState {
started_at: Instant::now(),
total_agents: 1,
running_count: 0,
running_agents: vec![SubagentPanelAgent {
ordinal: 1,
name: "watchdog-agent".to_string(),
status: AgentStatus::PendingInit,
is_watchdog: true,
preview: "monitor parent progress".to_string(),
latest_update_at: Instant::now(),
}],
}));
let cell = SubagentStatusCell::new(state, true);
assert_eq!(cell.transcript_animation_tick(), None);
}
#[test]
fn subagent_panel_animation_tick_runs_for_recent_running_updates() {
let state = Arc::new(Mutex::new(SubagentPanelState {
started_at: Instant::now(),
total_agents: 1,
running_count: 1,
running_agents: vec![SubagentPanelAgent {
ordinal: 1,
name: "worker-agent".to_string(),
status: AgentStatus::Running,
is_watchdog: false,
preview: "working".to_string(),
latest_update_at: Instant::now(),
}],
}));
let cell = SubagentStatusCell::new(state, true);
assert!(
cell.transcript_animation_tick().is_some(),
"recent running updates should animate"
);
}
#[test]
fn subagent_panel_animation_tick_stops_when_updates_are_stale() {
let stale_update = Instant::now()
.checked_sub(Duration::from_secs(2))
.unwrap_or_else(Instant::now);
let state = Arc::new(Mutex::new(SubagentPanelState {
started_at: Instant::now(),
total_agents: 1,
running_count: 1,
running_agents: vec![SubagentPanelAgent {
ordinal: 1,
name: "worker-agent".to_string(),
status: AgentStatus::Running,
is_watchdog: false,
preview: "working".to_string(),
latest_update_at: stale_update,
}],
}));
let cell = SubagentStatusCell::new(state, true);
assert_eq!(cell.transcript_animation_tick(), None);
}
fn image_block(data: &str) -> serde_json::Value {
serde_json::to_value(Content::image(data.to_string(), "image/png"))
.expect("image content should serialize")
+10 -9
View File
@@ -18,9 +18,9 @@ use ratatui::text::Span;
use std::collections::HashMap;
use std::collections::HashSet;
const COLLAB_PROMPT_PREVIEW_GRAPHEMES: usize = 160;
const COLLAB_AGENT_ERROR_PREVIEW_GRAPHEMES: usize = 160;
const COLLAB_AGENT_RESPONSE_PREVIEW_GRAPHEMES: usize = 240;
const MULTI_AGENT_PROMPT_PREVIEW_GRAPHEMES: usize = 160;
const MULTI_AGENT_ERROR_PREVIEW_GRAPHEMES: usize = 160;
const MULTI_AGENT_RESPONSE_PREVIEW_GRAPHEMES: usize = 240;
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct AgentPickerThreadEntry {
@@ -83,6 +83,7 @@ pub(crate) fn spawn_end(ev: CollabAgentSpawnEndEvent) -> PlainHistoryCell {
new_agent_role,
prompt,
status: _,
..
} = ev;
let title = match new_thread_id {
@@ -178,7 +179,6 @@ pub(crate) fn close_end(ev: CollabCloseEndEvent) -> PlainHistoryCell {
receiver_agent_role,
status: _,
} = ev;
collab_event(
title_with_agent(
"Closed",
@@ -200,7 +200,6 @@ pub(crate) fn resume_begin(ev: CollabResumeBeginEvent) -> PlainHistoryCell {
receiver_agent_nickname,
receiver_agent_role,
} = ev;
collab_event(
title_with_agent(
"Resuming",
@@ -223,7 +222,6 @@ pub(crate) fn resume_end(ev: CollabResumeEndEvent) -> PlainHistoryCell {
receiver_agent_role,
status,
} = ev;
collab_event(
title_with_agent(
"Resumed",
@@ -305,7 +303,7 @@ fn prompt_line(prompt: &str) -> Option<Line<'static>> {
} else {
Some(Line::from(Span::from(truncate_text(
trimmed,
COLLAB_PROMPT_PREVIEW_GRAPHEMES,
MULTI_AGENT_PROMPT_PREVIEW_GRAPHEMES,
))))
}
}
@@ -416,7 +414,7 @@ fn status_summary_spans(status: &AgentStatus) -> Vec<Span<'static>> {
if let Some(message) = message.as_ref() {
let message_preview = truncate_text(
&message.split_whitespace().collect::<Vec<_>>().join(" "),
COLLAB_AGENT_RESPONSE_PREVIEW_GRAPHEMES,
MULTI_AGENT_RESPONSE_PREVIEW_GRAPHEMES,
);
if !message_preview.is_empty() {
spans.push(Span::from(" - ").dim());
@@ -429,7 +427,7 @@ fn status_summary_spans(status: &AgentStatus) -> Vec<Span<'static>> {
let mut spans = vec![Span::from("Error").red()];
let error_preview = truncate_text(
&error.split_whitespace().collect::<Vec<_>>().join(" "),
COLLAB_AGENT_ERROR_PREVIEW_GRAPHEMES,
MULTI_AGENT_ERROR_PREVIEW_GRAPHEMES,
);
if !error_preview.is_empty() {
spans.push(Span::from(" - ").dim());
@@ -446,6 +444,7 @@ fn status_summary_spans(status: &AgentStatus) -> Vec<Span<'static>> {
mod tests {
use super::*;
use crate::history_cell::HistoryCell;
use codex_protocol::protocol::CollabAgentSpawnMode;
use insta::assert_snapshot;
use pretty_assertions::assert_eq;
use ratatui::style::Color;
@@ -467,6 +466,7 @@ mod tests {
new_agent_nickname: Some("Robie".to_string()),
new_agent_role: Some("explorer".to_string()),
prompt: "Compute 11! and reply with just the integer result.".to_string(),
spawn_mode: CollabAgentSpawnMode::Spawn,
status: AgentStatus::PendingInit,
});
@@ -547,6 +547,7 @@ mod tests {
new_agent_nickname: Some("Robie".to_string()),
new_agent_role: Some("explorer".to_string()),
prompt: String::new(),
spawn_mode: CollabAgentSpawnMode::Spawn,
status: AgentStatus::PendingInit,
});
+47
View File
@@ -114,6 +114,33 @@ pub(crate) fn truncate_text(text: &str, max_graphemes: usize) -> String {
}
}
/// Extract the first Markdown bold chunk (`**...**`) from `text`.
///
/// Returns trimmed inner text when both delimiters are present and non-empty.
/// If the opening delimiter is present without a closing delimiter, returns `None`
/// so callers can wait for additional streamed text.
pub(crate) fn extract_first_bold(text: &str) -> Option<String> {
let bytes = text.as_bytes();
let mut i = 0usize;
while i + 1 < bytes.len() {
if bytes[i] == b'*' && bytes[i + 1] == b'*' {
let start = i + 2;
let mut j = start;
while j + 1 < bytes.len() {
if bytes[j] == b'*' && bytes[j + 1] == b'*' {
let inner = &text[start..j];
let trimmed = inner.trim();
return (!trimmed.is_empty()).then_some(trimmed.to_string());
}
j += 1;
}
return None;
}
i += 1;
}
None
}
/// Truncate a path-like string to the given display width, keeping leading and trailing segments
/// where possible and inserting a single Unicode ellipsis between them. If an individual segment
/// cannot fit, it is front-truncated with an ellipsis.
@@ -440,6 +467,26 @@ mod tests {
assert_eq!(truncated.len(), 10); // 7 'a's + 3 dots
}
#[test]
fn test_extract_first_bold_returns_first_chunk() {
let text = "hello **Working on parser** and **another**";
assert_eq!(
extract_first_bold(text),
Some("Working on parser".to_string())
);
}
#[test]
fn test_extract_first_bold_ignores_empty_inner_text() {
assert_eq!(extract_first_bold("start **** end"), None);
assert_eq!(extract_first_bold("start ** ** end"), None);
}
#[test]
fn test_extract_first_bold_requires_closing_delimiter() {
assert_eq!(extract_first_bold("start **incomplete"), None);
}
#[test]
fn test_format_json_compact_simple_object() {
let json = r#"{ "name": "John", "age": 30 }"#;
+17
View File
@@ -24,6 +24,23 @@ Codex can run a notification hook when the agent finishes a turn. See the config
- https://developers.openai.com/codex/config-reference
## Agent Inbox Delivery
Inbound messages from other agents (for example, when a watchdog uses `send_input`) can be injected
as different roles in the root thread. Configure this with `agent_delivery_role`:
```toml
agent_delivery_role = "tool" # tool | developer | assistant
```
`tool` (default) injects a synthetic multi-agent inbox `collab_inbox` tool-call + tool-output pair
so inbound agent messages are explicit tool activity in the transcript. `developer` injects a
developer message with an explicit `[collab_inbox:...]` prefix so it is not mistaken for user
input. `assistant` injects the same prefix using the assistant role.
This setting applies to non-subagent threads (for example, the root thread). Messages sent to
subagents via `send_input` are still delivered as user input.
## JSON Schema
The generated JSON Schema for `config.toml` lives at `codex-rs/core/config.schema.json`.