diff --git a/docs/vrs/05-harness-state/requirements.md b/docs/vrs/05-harness-state/requirements.md index 75cb6b59..9b4dfc20 100644 --- a/docs/vrs/05-harness-state/requirements.md +++ b/docs/vrs/05-harness-state/requirements.md @@ -117,7 +117,7 @@ authorizes or changes delivery. nothing there awaits a human (matching the projection's `Held { Review }` → `active` / `blockedOn: none` row) — while `WaitingOnApproval` and `WaitingOnUserInput` report `active` with - `blockedOn: human`. `Compaction` reports `active`, and + `blockedOn: human`. `Compaction` and `UnknownProtocol` report `active`, and `NotLoaded`/`SystemError`/`AwaitingStatus` withhold rather than write. `Held` — a delivery predicate — never appears in the published vocabulary. - **OHS-R06 Heartbeat only on evidence:** A writer re-stamps the record on the @@ -195,8 +195,9 @@ The measurements are #268's, taken 2026-08-16/17 on one host and carried with their original caveats: 1298 presence files all legacy one-line records, 4 transitions per turn 0.1–0.4 ms apart, Claude hook timelines (blocked entry in 2 of 9 captures, exit in 1), silent Claude death under SIGTERM/SIGKILL, and -the Codex `activeFlags` schema present on all supported codex-cli versions -(#268's first comment). The shipped code evidence is in-repo: the Codex state +the Codex `activeFlags` schema present in the measured codex-cli versions +(#268's first comment). The startup gate now checks that generated schema +directly. The shipped code evidence is in-repo: the Codex state machine and its hold reasons, the unfiltered agent-dir watch beside the presence refresh that writes into it, and `src/harness_state.rs`, which implements the envelope this file ratifies. diff --git a/docs/vrs/05-harness-state/spec.md b/docs/vrs/05-harness-state/spec.md index 110fe4cb..53e4fd41 100644 --- a/docs/vrs/05-harness-state/spec.md +++ b/docs/vrs/05-harness-state/spec.md @@ -183,11 +183,21 @@ comes first. The cross-check is a narrowing of the ungraceful-death window (provably dead sessions: pidfile present, process gone), not its closure — OHS-T04/OHS-R07 say exactly this, and no death tombstone is attempted: the kill that removes the registry entry leaves nothing behind to prove death -with, and fabricating evidence is the one thing this design never does. A Codex -binary whose delivery-critical schema projection does not match an admitted -fingerprint produces no Codex observed state at all: provider launch is refused -before the control channel starts. Fingerprint admission and live behavioral -evidence remain separate. +with, and fabricating evidence is the one thing this design never does. + +Each controlled Codex startup generates the installed app-server schema. st2 +checks the required methods, response fields, blocking flags, and data shapes +that native delivery uses. A compatible patch or minor release starts without +a source change. A missing required element or a changed critical shape stops +before the app-server starts. The wrapper sends one idempotent rejection report +to the agent's declared supervisor. The version string is diagnostic data, not +an admission proxy. + +Additive item kinds and server-request methods do not stop startup. A listed +element is reviewed and safe to ignore. An unlisted element creates an +`UnknownProtocol` delivery hold. The next safe thread status releases the hold. +This fail-closed runtime rule keeps the agent present and prevents a silent +delivery into a new hold that st2 does not understand. ## Codex producer (OHS-R05) @@ -205,6 +215,7 @@ complement of steerable, a delivery predicate (decision 0001's boundary). | `Held { ConflictingTurn }` | `active` | `none` | `none` | `conflictingTurn` — two turns believed live is maximally active | | `Held { Review }` | `active` | `none` | `none` | `review` — review's enter and exit are model-emitted items inside a running turn; nothing awaits a human | | `Held { Compaction }` | `active` | `none` | `none` | `compaction` | +| `Held { UnknownProtocol }` | `active` | `none` | `none` | `unknownProtocol` — an additive protocol element blocks delivery until a safe thread status arrives | | `Held { WaitingOnApproval }` | `active` | `human` | `permission` | `waitingOnApproval` | | `Held { WaitingOnUserInput }` | `active` | `human` | `question` | `waitingOnUserInput` | | `Held { NotLoaded }` | *withhold* | — | — | thread not loaded proves nothing about work | diff --git a/docs/vrs/spec.md b/docs/vrs/spec.md index 0f9cddba..d1380090 100644 --- a/docs/vrs/spec.md +++ b/docs/vrs/spec.md @@ -1065,9 +1065,9 @@ policy, compat field, or API version constant — but it does publish its TypeScript declarations, and those govern this coupling. `checks.pi-extension-types` type-checks the shipped extension against a pinned pi release at build time, which is what makes the asset's otherwise-erased -`import type` load-bearing. This follows the repo's existing rule that st2 pins -where skew fails silently (`pty`, `codex-cli`) and not where it fails loudly -(`claude`); the extension's one silent surface was its idle proof, and using +`import type` load-bearing. This follows the repo's existing rule that st2 validates +where skew fails silently (`pty`, the Codex app-server schema) and not where it fails +loudly (`claude`); the extension's one silent surface was its idle proof, and using that proof without calling it is now a build error. A pi agent never enters the DING path. `deliver` and `ding` are mutually diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs index d992c93e..0fef4425 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -33,40 +33,60 @@ use tungstenite::{Message as WebSocketMessage, WebSocket}; use crate::{ding, harness_context, harness_state, message, run, status}; -/// A Codex binary is admitted by the exact delivery-critical schema projection st2 consumes, not -/// by its release number. Behavioral evidence remains a separate gate: a matching schema does not -/// claim that a model turn, resume, or durable receipt was exercised for an arbitrary build. -/// -/// This fingerprint is the canonical projection generated by codex-cli 0.151.0. It covers every -/// outbound method arm st2 emits, the inbound notification arms that drive delivery state, their -/// recursively referenced payload definitions, and delivery-critical response definitions. -const ADMITTED_CODEX_PROTOCOL_FINGERPRINTS: &[&str] = - &["40d024cc88b67c9d325edd944c87a460dc2fc1094a1b74bba7ccd2347aefe9b6"]; -const CODEX_PROTOCOL_SCHEMA: &str = "codex_app_server_protocol.v2.schemas.json"; -const CLIENT_NOTIFICATION_SCHEMA: &str = "ClientNotification.json"; -const CRITICAL_CLIENT_REQUESTS: &[(&str, &str)] = &[ - ("initialize", "InitializeParams"), - ("hooks/list", "HooksListParams"), - ("thread/loaded/list", "ThreadLoadedListParams"), - ("thread/resume", "ThreadResumeParams"), - ("turn/start", "TurnStartParams"), - ("turn/steer", "TurnSteerParams"), +const REQUIRED_CODEX_CLIENT_REQUESTS: &[&str] = &[ + "hooks/list", + "initialize", + "thread/loaded/list", + "thread/resume", + "turn/start", + "turn/steer", +]; +const REQUIRED_CODEX_CLIENT_NOTIFICATIONS: &[&str] = &["initialized"]; +const REQUIRED_CODEX_SERVER_NOTIFICATIONS: &[&str] = &[ + "item/completed", + "item/started", + "thread/started", + "thread/status/changed", + "turn/completed", + "turn/started", ]; -const CRITICAL_SERVER_NOTIFICATIONS: &[(&str, &str)] = &[ - ("thread/started", "ThreadStartedNotification"), - ("thread/status/changed", "ThreadStatusChangedNotification"), - ("turn/started", "TurnStartedNotification"), - ("turn/completed", "TurnCompletedNotification"), - ("item/started", "ItemStartedNotification"), - ("item/completed", "ItemCompletedNotification"), - ("thread/compacted", "ContextCompactedNotification"), +// The control observer does not answer server requests. A listed request is reviewed and safe to +// ignore. An unlisted request creates a delivery hold until the thread reports a safe status. +const CLASSIFIED_CODEX_SERVER_REQUESTS: &[&str] = &[ + "account/chatgptAuthTokens/refresh", + "applyPatchApproval", + "attestation/generate", + "currentTime/read", + "execCommandApproval", + "item/commandExecution/requestApproval", + "item/fileChange/requestApproval", + "item/permissions/requestApproval", + "item/tool/call", + "item/tool/requestUserInput", + "mcpServer/elicitation/request", ]; -const CRITICAL_RESPONSE_DEFINITIONS: &[&str] = &[ - "HooksListResponse", - "ThreadLoadedListResponse", - "ThreadResumeResponse", - "TurnStartResponse", - "TurnSteerResponse", +// A listed item is reviewed and safe to ignore unless `observe` handles it explicitly. An unlisted +// item creates a delivery hold until the thread reports a safe status. +const CLASSIFIED_CODEX_THREAD_ITEMS: &[&str] = &[ + "agentMessage", + "collabAgentToolCall", + "commandExecution", + "contextCompaction", + "dynamicToolCall", + "enteredReviewMode", + "exitedReviewMode", + "fileChange", + "functionCallOutput", + "hookPrompt", + "imageGeneration", + "imageView", + "mcpToolCall", + "plan", + "reasoning", + "sleep", + "subAgentActivity", + "userMessage", + "webSearch", ]; const RUNTIME_SCHEMA: &str = "st2.codex-runtime.v1"; const BINDING_SCHEMA: &str = "st2.codex-thread-binding.v1"; @@ -219,6 +239,7 @@ pub enum CodexHoldReason { ConflictingTurn, Review, Compaction, + UnknownProtocol, NotLoaded, SystemError, UnknownStatus, @@ -275,6 +296,9 @@ impl CodexObservedState { CodexHoldReason::Compaction => { Some(observation(Activity::Active, BlockedOn::None).with_reason("compaction")) } + CodexHoldReason::UnknownProtocol => Some( + observation(Activity::Active, BlockedOn::None).with_reason("unknownProtocol"), + ), // Codex positively reported active; st2 merely cannot name a steerable turn. CodexHoldReason::ActiveWithoutTurn => Some( observation(Activity::Active, BlockedOn::None).with_reason("activeWithoutTurn"), @@ -309,6 +333,7 @@ struct CodexDeliveryConfig { inbox: PathBuf, identity: String, this_host: String, + supervisor: Option, } impl CodexDeliveryConfig { @@ -321,14 +346,57 @@ impl CodexDeliveryConfig { catalog_root.display() ) })?; + let supervisor = crate::discover(catalog_root) + .specs + .into_iter() + .find(|spec| spec.path.parent() == Some(agent_dir.as_path())) + .and_then(|spec| spec.supervisor); Ok(Self { catalog_root: catalog_root.to_path_buf(), inbox: message::inbox_dir(&agent_dir), agent_dir, identity: identity.to_string(), this_host, + supervisor, }) } + + fn report_protocol_rejection(&self, codex: &str, error: &anyhow::Error) { + let Some(supervisor) = self.supervisor.as_deref() else { + eprintln!( + "st2 codex: agent '{}' has no supervisor for a protocol rejection report", + self.identity + ); + return; + }; + let subject = format!("Codex protocol rejected: {}", self.identity); + let body = format!( + "st2 rejected the installed Codex app-server protocol for agent '{}'. Native delivery did not start. Codex executable: '{}'. Error: {error:#}", + self.identity, codex + ); + let mut key_hash = Sha256::new(); + key_hash.update(b"st2.codex-protocol-rejection.v1"); + key_hash.update(body.as_bytes()); + let idempotency_key = format!("st2.codex-protocol-rejection.v1:{:x}", key_hash.finalize()); + let tags = ["codex-protocol".to_string(), "launch-rejected".to_string()]; + if let Err(report_error) = message::send_to_resolved_inbox( + &self.catalog_root, + supervisor, + &self.this_host, + &self.identity, + Some(&subject), + None, + &tags, + &body, + Some(&idempotency_key), + None, + ) { + eprintln!( + "st2 codex: failed to report agent '{}' protocol rejection to supervisor '{}': {report_error:#}", + self.identity, supervisor + ); + } + } } #[derive(Debug, Clone, PartialEq, Eq)] @@ -1266,19 +1334,20 @@ impl CodexControlState { return Ok(false); } let item_type = required_string(message, "/params/item/type", method)?; - // Codex 0.151 `ThreadItem` has nineteen variants; only these three carry - // steerability. Every other item reports work inside a turn that the turn and - // thread status already model, so it is ignored on purpose — a later protocol - // item that gates or releases input has to be added here explicitly, because - // silently dropping one is how `exitedReviewMode` stayed unmatched. Review is - // also the only hold the protocol ends with a typed item of its own: + // The admitted `ThreadItem` schema has only three variants that change + // steerability. Every other classified item reports work inside a turn that the + // turn and thread status already model, so it is ignored on purpose. A later + // protocol item that gates or releases input must be added here explicitly. + // Silently dropping one is how `exitedReviewMode` stayed unmatched. Review is + // also the only hold that the protocol ends with a typed item of its own: // `contextCompaction` has no exit item, so both of its lifecycle edges keep // holding until the thread proves otherwise. let (reason, released) = match item_type { "enteredReviewMode" => (CodexHoldReason::Review, false), "exitedReviewMode" => (CodexHoldReason::Review, true), "contextCompaction" => (CodexHoldReason::Compaction, false), - _ => return Ok(false), + _ if CLASSIFIED_CODEX_THREAD_ITEMS.contains(&item_type) => return Ok(false), + _ => (CodexHoldReason::UnknownProtocol, false), }; let turn_id = required_string(message, "/params/turnId", method)?; if released { @@ -1287,6 +1356,11 @@ impl CodexControlState { self.observe_non_steerable(turn_id, reason); } } + _ if message.get("id").is_some() + && !CLASSIFIED_CODEX_SERVER_REQUESTS.contains(&method) => + { + self.observe_unknown_protocol(); + } _ => return Ok(false), } Ok(self.observed != before) @@ -1332,6 +1406,7 @@ impl CodexControlState { reason: CodexHoldReason::Review | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol | CodexHoldReason::ConflictingTurn, .. }, @@ -1368,7 +1443,10 @@ impl CodexControlState { self.observed.clone() } CodexObservedState::Held { - reason: reason @ (CodexHoldReason::Review | CodexHoldReason::Compaction), + reason: + reason @ (CodexHoldReason::Review + | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol), .. } => CodexObservedState::Held { reason: *reason, @@ -1408,6 +1486,7 @@ impl CodexControlState { reason: CodexHoldReason::Review | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol | CodexHoldReason::ConflictingTurn | CodexHoldReason::WaitingOnApproval | CodexHoldReason::WaitingOnUserInput @@ -1445,14 +1524,18 @@ impl CodexControlState { } if current_reason == &reason && matches!( reason, - CodexHoldReason::Review | CodexHoldReason::Compaction + CodexHoldReason::Review + | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol ) => { self.observed.clone() } _ if matches!( reason, - CodexHoldReason::Review | CodexHoldReason::Compaction + CodexHoldReason::Review + | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol ) => { CodexObservedState::Held { @@ -1487,17 +1570,32 @@ impl CodexControlState { turn_id: turn_id.to_string(), }; } + + fn observe_unknown_protocol(&mut self) { + if matches!(self.observed, CodexObservedState::TerminalError { .. }) { + return; + } + let turn_id = match &self.observed { + CodexObservedState::Active { turn_id } + | CodexObservedState::Held { + turn_id: Some(turn_id), + .. + } => Some(turn_id.clone()), + _ => None, + }; + self.observed = CodexObservedState::Held { + reason: CodexHoldReason::UnknownProtocol, + turn_id, + }; + } } /// Read the delivery-relevant part of `ThreadStatus.activeFlags`: the first flag that says this /// thread is blocked on a human rather than on the model. /// -/// `activeFlags` is a required property of the `active` arm of `ThreadStatus` on both supported -/// codex-cli versions, and appears on no other arm, so a missing or malformed array reads as no -/// flag instead of failing the frame - a schema surprise must not kill the control watcher. -/// Unknown future flag values degrade the same way, to a plain `active` status; admitting a new -/// codex-cli version already requires a delivery-critical schema comparison, which is where a new -/// flag value has to be classified. +/// The startup gate requires `activeFlags` on the `active` arm of `ThreadStatus`. A missing or +/// malformed runtime array reads as no flag instead of killing the control watcher. The startup +/// gate rejects an unclassified flag before launch. fn human_blocking_flag(status: Option<&Value>) -> Option { status? .get("activeFlags")? @@ -1530,8 +1628,11 @@ pub fn run_controlled( !codex_argv.is_empty(), "Codex controlled launch argv is empty" ); - ensure_supported_protocol(&codex_argv[0])?; let delivery = CodexDeliveryConfig::resolve(catalog_root, &identity)?; + if let Err(error) = ensure_supported_protocol(&codex_argv[0]) { + delivery.report_protocol_rejection(&codex_argv[0], &error); + return Err(error); + } let state_dir = state_dir(catalog_root, &identity); secure_dir(&state_dir)?; @@ -1589,44 +1690,7 @@ fn run_controlled_owned( .parent() .context("Codex app-server socket has no parent")?; secure_dir(socket_dir)?; - match fs::symlink_metadata(&socket_path) { - Ok(metadata) => { - anyhow::ensure!( - metadata.file_type().is_socket(), - "Codex app-server path already exists and is not a socket: {}", - socket_path.display() - ); - match UnixStream::connect(&socket_path) { - Ok(_) => anyhow::bail!( - "Codex app-server socket {} is already live; refusing a second control owner", - socket_path.display() - ), - Err(error) - if matches!( - error.kind(), - std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound - ) => - { - fs::remove_file(&socket_path).with_context(|| { - format!("removing stale Codex socket {}", socket_path.display()) - })?; - } - Err(error) => { - return Err(error).with_context(|| { - format!( - "checking existing Codex socket {} before launch", - socket_path.display() - ) - }); - } - } - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => { - return Err(error) - .with_context(|| format!("checking Codex socket path {}", socket_path.display())); - } - } + prepare_socket_for_launch(&socket_path)?; // Publish a new incarnation only after this process holds the owner lock and has proved that no // older daemon is live. A rejected second owner must not invalidate the first owner's binding. @@ -1667,13 +1731,13 @@ fn run_controlled_owned( .stdin(Stdio::null()) .stdout(log.try_clone()?) .stderr(log); - let mut server = spawn_process_group(&mut server_command) + let mut server = spawn_process_group(&mut server_command, Some(&socket_path)) .with_context(|| format!("starting {} app-server", codex_argv[0]))?; let result = diagnostics .record("appServerStarted", json!({ "pid": server.id() })) .and_then(|_| { run_connected( - &mut server, + server.child_mut(), &socket_path, &runtime, &codex_argv, @@ -1682,11 +1746,52 @@ fn run_controlled_owned( diagnostics, ) }); - terminate_process_group(&mut server); - let _ = fs::remove_file(&socket_path); + server.terminate(); result } +fn prepare_socket_for_launch(socket_path: &Path) -> Result<()> { + match fs::symlink_metadata(socket_path) { + Ok(metadata) => { + anyhow::ensure!( + metadata.file_type().is_socket(), + "Codex app-server path already exists and is not a socket: {}", + socket_path.display() + ); + match UnixStream::connect(socket_path) { + Ok(_) => anyhow::bail!( + "Codex app-server socket {} is already live; refusing a second control owner", + socket_path.display() + ), + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound + ) => + { + fs::remove_file(socket_path).with_context(|| { + format!("removing stale Codex socket {}", socket_path.display()) + })?; + } + Err(error) => { + return Err(error).with_context(|| { + format!( + "checking existing Codex socket {} before launch", + socket_path.display() + ) + }); + } + } + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error) + .with_context(|| format!("checking Codex socket path {}", socket_path.display())); + } + } + Ok(()) +} + fn run_connected( server: &mut Child, socket_path: &Path, @@ -2041,12 +2146,13 @@ fn preflight_hook_trust( .stdin(Stdio::null()) .stdout(log.try_clone()?) .stderr(log.try_clone()?); - let mut server = spawn_process_group(&mut server_command) + let mut server = spawn_process_group(&mut server_command, Some(socket_path)) .with_context(|| format!("starting {codex} hook-trust preflight app-server"))?; let result = diagnostics .record("hookTrustPreflightStarted", json!({ "pid": server.id() })) .and_then(|_| { - let Some(control) = connect_control(&mut server, socket_path, STARTUP_TIMEOUT)? else { + let Some(control) = connect_control(server.child_mut(), socket_path, STARTUP_TIMEOUT)? + else { // Stop requested mid-preflight: skip the projection — the launch proceeds to the // connect stage, whose own stop check exits gracefully before the TUI starts. return Ok(None); @@ -2056,8 +2162,7 @@ fn preflight_hook_trust( }; query_hook_trust_projection(&mut websocket, cwd) }); - terminate_process_group(&mut server); - let _ = fs::remove_file(socket_path); + server.terminate(); let projection = result?; diagnostics.record( "hookTrustPreflightComplete", @@ -2951,8 +3056,20 @@ fn completed_tui(status: ExitStatus) -> Result<()> { Ok(()) } +struct CodexProtocolSchemas { + protocol: Value, + client_requests: Value, + client_notifications: Value, + server_requests: Value, + server_notifications: Value, +} + fn ensure_supported_protocol(codex: &str) -> Result<()> { - let schema_dir = tempfile::tempdir().context("creating Codex schema admission directory")?; + let version = codex_version(codex)?; + let generated = tempfile::Builder::new() + .prefix("st2-codex-protocol-") + .tempdir() + .context("creating a temporary Codex protocol schema directory")?; let output = Command::new(codex) .args([ "app-server", @@ -2960,147 +3077,553 @@ fn ensure_supported_protocol(codex: &str) -> Result<()> { "--experimental", "--out", ]) - .arg(schema_dir.path()) + .arg(generated.path()) .output() - .with_context(|| format!("generating Codex app-server schemas with {codex}"))?; + .with_context(|| format!("generating the Codex app-server schema from {codex}"))?; anyhow::ensure!( output.status.success(), "{codex} app-server schema generation failed: {}", String::from_utf8_lossy(&output.stderr).trim() ); - let fingerprint = delivery_critical_schema_fingerprint(schema_dir.path())?; - anyhow::ensure!( - ADMITTED_CODEX_PROTOCOL_FINGERPRINTS.contains(&fingerprint.as_str()), - "unsupported Codex app-server delivery schema fingerprint '{fingerprint}'" - ); - Ok(()) -} - -fn delivery_critical_schema_fingerprint(schema_dir: &Path) -> Result { - let aggregate = read_schema_json(&schema_dir.join(CODEX_PROTOCOL_SCHEMA))?; - let client_notification = read_schema_json(&schema_dir.join(CLIENT_NOTIFICATION_SCHEMA))?; - let projection = delivery_critical_schema_projection(&aggregate, &client_notification)?; - let canonical = - serde_json::to_vec(&projection).context("serializing Codex schema projection")?; - Ok(format!("{:x}", Sha256::digest(canonical))) + let read_schema = |name: &str| -> Result { + let path = generated.path().join(name); + let bytes = + fs::read(&path).with_context(|| format!("reading generated Codex schema {name}"))?; + serde_json::from_slice(&bytes) + .with_context(|| format!("parsing generated Codex schema {name}")) + }; + let schemas = CodexProtocolSchemas { + protocol: read_schema("codex_app_server_protocol.v2.schemas.json")?, + client_requests: read_schema("ClientRequest.json")?, + client_notifications: read_schema("ClientNotification.json")?, + server_requests: read_schema("ServerRequest.json")?, + server_notifications: read_schema("ServerNotification.json")?, + }; + verify_codex_protocol_schemas(&schemas) + .with_context(|| format!("Codex app-server schema from {version} is incompatible")) } -fn read_schema_json(path: &Path) -> Result { - serde_json::from_slice( - &fs::read(path).with_context(|| format!("reading Codex schema {}", path.display()))?, - ) - .with_context(|| format!("parsing Codex schema {}", path.display())) +fn codex_version(codex: &str) -> Result { + let output = Command::new(codex) + .arg("--version") + .output() + .with_context(|| format!("reading Codex version from {codex}"))?; + anyhow::ensure!( + output.status.success(), + "{codex} --version failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + let actual = String::from_utf8(output.stdout) + .context("Codex version output is not UTF-8")? + .trim() + .to_string(); + anyhow::ensure!(!actual.is_empty(), "{codex} --version printed nothing"); + Ok(actual) } -fn delivery_critical_schema_projection( - aggregate: &Value, - client_notification: &Value, -) -> Result { - let definitions = aggregate +fn verify_codex_protocol_schemas(schemas: &CodexProtocolSchemas) -> Result<()> { + let definitions = schemas + .protocol .get("definitions") .and_then(Value::as_object) - .context("Codex aggregate schema has no definitions object")?; - let mut refs = BTreeSet::new(); - let mut unions = BTreeMap::::new(); - for (union, required) in [ - ("ClientRequest", CRITICAL_CLIENT_REQUESTS), - ("ServerNotification", CRITICAL_SERVER_NOTIFICATIONS), + .context("aggregate schema has no definitions object")?; + + require_methods( + &schemas.client_requests, + REQUIRED_CODEX_CLIENT_REQUESTS, + "client request", + )?; + require_methods( + &schemas.client_notifications, + REQUIRED_CODEX_CLIENT_NOTIFICATIONS, + "client notification", + )?; + require_methods( + &schemas.server_notifications, + REQUIRED_CODEX_SERVER_NOTIFICATIONS, + "server notification", + )?; + schema_methods(&schemas.server_requests, "server request")?; + + let status_variants = schema_variants(definitions, "ThreadStatus", "type")?; + for status in ["notLoaded", "idle", "systemError", "active"] { + anyhow::ensure!( + status_variants.contains_key(status), + "ThreadStatus has no '{status}' variant" + ); + } + let active = status_variants + .get("active") + .context("ThreadStatus has no active variant")?; + let active_flags = + required_property(definitions, active, "activeFlags", "ThreadStatus.active")?; + let active_flag = require_array(definitions, active_flags, "ThreadStatus.activeFlags")?; + anyhow::ensure!( + active_flag == schema_definition(definitions, "ThreadActiveFlag")?, + "ThreadStatus.activeFlags does not contain ThreadActiveFlag" + ); + let actual_active_flags = schema_enum(definitions, "ThreadActiveFlag")?; + anyhow::ensure!( + actual_active_flags == string_set(&["waitingOnApproval", "waitingOnUserInput"]), + "ThreadActiveFlag changed: {}", + actual_active_flags + .into_iter() + .collect::>() + .join(", ") + ); + + let item_variants = schema_variants(definitions, "ThreadItem", "type")?; + for item in [ + "contextCompaction", + "enteredReviewMode", + "exitedReviewMode", + "userMessage", ] { - let schema = definitions - .get(union) - .with_context(|| format!("Codex aggregate schema has no {union} definition"))?; - let mut arms = BTreeMap::::new(); - for (method, payload) in required { - let arm = required_method_arm(schema, method, Some(payload))?; - collect_definition_refs(&arm, &mut refs); - arms.insert((*method).to_owned(), arm); - } - unions.insert(union.to_owned(), serde_json::to_value(arms)?); + anyhow::ensure!( + item_variants.contains_key(item), + "ThreadItem has no '{item}' variant" + ); } + let user_message = item_variants + .get("userMessage") + .context("ThreadItem has no userMessage variant")?; + require_property_type( + definitions, + user_message, + "clientId", + "string", + false, + "ThreadItem.userMessage", + )?; - let initialized = required_method_arm(client_notification, "initialized", None)?; - for definition in CRITICAL_RESPONSE_DEFINITIONS { - refs.insert((*definition).to_owned()); + let user_input_variants = schema_variants(definitions, "UserInput", "type")?; + let text_input = user_input_variants + .get("text") + .context("UserInput has no text variant")?; + require_property_type( + definitions, + text_input, + "text", + "string", + true, + "UserInput.text", + )?; + let text_elements = property(definitions, text_input, "text_elements", "UserInput.text")?; + require_array(definitions, text_elements, "UserInput.text.text_elements")?; + + for (definition, path) in [ + ("ClientInfo", &["name"][..]), + ("ClientInfo", &["version"][..]), + ("Thread", &["id"][..]), + ("Turn", &["id"][..]), + ("ThreadStatusChangedNotification", &["threadId"][..]), + ("TurnStartedNotification", &["threadId"][..]), + ("TurnStartedNotification", &["turn", "id"][..]), + ("TurnCompletedNotification", &["threadId"][..]), + ("TurnCompletedNotification", &["turn", "id"][..]), + ("ItemStartedNotification", &["threadId"][..]), + ("ItemStartedNotification", &["turnId"][..]), + ("ItemCompletedNotification", &["threadId"][..]), + ("ItemCompletedNotification", &["turnId"][..]), + ("ThreadStartedNotification", &["thread", "id"][..]), + ("ThreadResumeParams", &["threadId"][..]), + ("ThreadResumeResponse", &["thread", "id"][..]), + ("TurnStartParams", &["threadId"][..]), + ("TurnStartResponse", &["turn", "id"][..]), + ("TurnSteerParams", &["threadId"][..]), + ("TurnSteerParams", &["expectedTurnId"][..]), + ("TurnSteerResponse", &["turnId"][..]), + ] { + let schema = required_schema_path(definitions, definition, path)?; + require_type( + definitions, + schema, + "string", + &format!("{definition}.{}", path.join(".")), + )?; } - let mut projected_definitions = BTreeMap::::new(); - while let Some(name) = refs - .iter() - .find(|name| !projected_definitions.contains_key(*name)) - .cloned() - { - let definition = definitions - .get(&name) - .with_context(|| format!("Codex schema reference '#/definitions/{name}' is missing"))? - .clone(); - collect_definition_refs(&definition, &mut refs); - projected_definitions.insert(name, definition); - } - - Ok(json!({ - "arms": unions, - "clientNotification": initialized, - "definitions": projected_definitions, - })) + + require_property_type( + definitions, + schema_definition(definitions, "ClientInfo")?, + "title", + "string", + false, + "ClientInfo", + )?; + require_property_type( + definitions, + schema_definition(definitions, "InitializeCapabilities")?, + "experimentalApi", + "boolean", + false, + "InitializeCapabilities", + )?; + required_schema_path(definitions, "InitializeParams", &["clientInfo"])?; + + let thread_status = required_schema_path(definitions, "Thread", &["status"])?; + anyhow::ensure!( + thread_status == schema_definition(definitions, "ThreadStatus")?, + "Thread.status does not use ThreadStatus" + ); + let resume_status = + required_schema_path(definitions, "ThreadResumeResponse", &["thread", "status"])?; + anyhow::ensure!( + resume_status == schema_definition(definitions, "ThreadStatus")?, + "ThreadResumeResponse.thread.status does not use ThreadStatus" + ); + let started_status = required_schema_path( + definitions, + "ThreadStartedNotification", + &["thread", "status"], + )?; + anyhow::ensure!( + started_status == schema_definition(definitions, "ThreadStatus")?, + "ThreadStartedNotification.thread.status does not use ThreadStatus" + ); + let changed_status = + required_schema_path(definitions, "ThreadStatusChangedNotification", &["status"])?; + anyhow::ensure!( + changed_status == schema_definition(definitions, "ThreadStatus")?, + "ThreadStatusChangedNotification.status does not use ThreadStatus" + ); + + let turns = required_schema_path(definitions, "Thread", &["turns"])?; + let turn = require_array(definitions, turns, "Thread.turns")?; + anyhow::ensure!( + turn == schema_definition(definitions, "Turn")?, + "Thread.turns does not contain Turn" + ); + let items = required_schema_path(definitions, "Turn", &["items"])?; + let item = require_array(definitions, items, "Turn.items")?; + anyhow::ensure!( + item == schema_definition(definitions, "ThreadItem")?, + "Turn.items does not contain ThreadItem" + ); + for notification in ["ItemStartedNotification", "ItemCompletedNotification"] { + let item = required_schema_path(definitions, notification, &["item"])?; + anyhow::ensure!( + item == schema_definition(definitions, "ThreadItem")?, + "{notification}.item does not use ThreadItem" + ); + } + + for params in ["TurnStartParams", "TurnSteerParams"] { + let input = required_schema_path(definitions, params, &["input"])?; + let input_item = require_array(definitions, input, &format!("{params}.input"))?; + anyhow::ensure!( + input_item == schema_definition(definitions, "UserInput")?, + "{params}.input does not contain UserInput" + ); + require_property_type( + definitions, + schema_definition(definitions, params)?, + "clientUserMessageId", + "string", + false, + params, + )?; + } + let loaded = required_schema_path(definitions, "ThreadLoadedListResponse", &["data"])?; + let loaded_item = require_array(definitions, loaded, "ThreadLoadedListResponse.data")?; + require_type( + definitions, + loaded_item, + "string", + "ThreadLoadedListResponse.data item", + )?; + let hook_cwds = property( + definitions, + schema_definition(definitions, "HooksListParams")?, + "cwds", + "HooksListParams", + )?; + let hook_cwd = require_array(definitions, hook_cwds, "HooksListParams.cwds")?; + require_type(definitions, hook_cwd, "string", "HooksListParams.cwds item")?; + verify_hook_schema(definitions)?; + Ok(()) } -fn required_method_arm( - union: &Value, - method: &str, - expected_payload: Option<&str>, -) -> Result { - let arms = union - .get("oneOf") - .and_then(Value::as_array) - .context("Codex protocol union has no oneOf arms")?; - let matches = arms - .iter() - .filter(|arm| { - arm.pointer("/properties/method/enum") - .and_then(Value::as_array) - .is_some_and(|values| { - values.len() == 1 && values[0].as_str() == Some(method) - }) - }) - .collect::>(); +fn verify_hook_schema(definitions: &serde_json::Map) -> Result<()> { + let data = required_schema_path(definitions, "HooksListResponse", &["data"])?; + let entry = require_array(definitions, data, "HooksListResponse.data")?; + anyhow::ensure!( + entry == schema_definition(definitions, "HooksListEntry")?, + "HooksListResponse.data does not contain HooksListEntry" + ); + let hooks = required_schema_path(definitions, "HooksListEntry", &["hooks"])?; + let hook = require_array(definitions, hooks, "HooksListEntry.hooks")?; anyhow::ensure!( - matches.len() == 1, - "Codex protocol method '{method}' must have exactly one schema arm, found {}", - matches.len() + hook == schema_definition(definitions, "HookMetadata")?, + "HooksListEntry.hooks does not contain HookMetadata" ); - let arm = matches[0]; - if let Some(payload) = expected_payload { - let expected = format!("#/definitions/{payload}"); - let actual = arm - .pointer("/properties/params/$ref") - .and_then(Value::as_str); + for (property, expected_type) in [ + ("currentHash", "string"), + ("isManaged", "boolean"), + ("key", "string"), + ] { + require_property_type( + definitions, + schema_definition(definitions, "HookMetadata")?, + property, + expected_type, + true, + "HookMetadata", + )?; + } + let trust_status = required_schema_path(definitions, "HookMetadata", &["trustStatus"])?; + anyhow::ensure!( + trust_status == schema_definition(definitions, "HookTrustStatus")?, + "HookMetadata.trustStatus does not use HookTrustStatus" + ); + let statuses = schema_enum(definitions, "HookTrustStatus")?; + anyhow::ensure!( + statuses == string_set(&["managed", "modified", "trusted", "untrusted"]), + "HookTrustStatus changed: {}", + statuses.into_iter().collect::>().join(", ") + ); + Ok(()) +} + +fn string_set(values: &[&str]) -> BTreeSet { + values.iter().map(|value| (*value).to_string()).collect() +} + +fn schema_methods(schema: &Value, label: &str) -> Result> { + let arms = schema + .get("oneOf") + .and_then(Value::as_array) + .with_context(|| format!("{label} schema has no oneOf array"))?; + let mut methods = BTreeSet::new(); + for arm in arms { + let required = arm + .get("required") + .and_then(Value::as_array) + .with_context(|| format!("{label} arm has no required array"))?; anyhow::ensure!( - actual == Some(expected.as_str()), - "Codex protocol method '{method}' must reference payload '{expected}', found {}", - actual.unwrap_or("") + required + .iter() + .any(|value| value.as_str() == Some("method")), + "{label} arm does not require method" + ); + let values = arm + .pointer("/properties/method/enum") + .and_then(Value::as_array) + .with_context(|| format!("{label} arm has no method enum"))?; + anyhow::ensure!(values.len() == 1, "{label} arm method enum is not exact"); + let method = values[0] + .as_str() + .with_context(|| format!("{label} arm method is not a string"))?; + anyhow::ensure!( + methods.insert(method.to_string()), + "{label} method '{method}' is duplicated" ); } - Ok(arm.clone()) + Ok(methods) } -fn collect_definition_refs(value: &Value, refs: &mut BTreeSet) { - match value { - Value::Object(object) => { - if let Some(reference) = object.get("$ref").and_then(Value::as_str) - && let Some(name) = reference.strip_prefix("#/definitions/") - { - refs.insert(name.to_owned()); - } - for child in object.values() { - collect_definition_refs(child, refs); - } +fn require_methods(schema: &Value, required: &[&str], label: &str) -> Result<()> { + let methods = schema_methods(schema, label)?; + let missing = string_set(required) + .difference(&methods) + .cloned() + .collect::>(); + anyhow::ensure!( + missing.is_empty(), + "missing {label} methods: {}", + missing.join(", ") + ); + Ok(()) +} + +fn schema_definition<'a>( + definitions: &'a serde_json::Map, + name: &str, +) -> Result<&'a Value> { + definitions + .get(name) + .with_context(|| format!("aggregate schema has no {name} definition")) +} + +fn resolve_schema<'a>( + definitions: &'a serde_json::Map, + mut schema: &'a Value, +) -> Result<&'a Value> { + for _ in 0..16 { + if let Some(reference) = schema.get("$ref").and_then(Value::as_str) { + let name = reference + .strip_prefix("#/definitions/") + .with_context(|| format!("unsupported schema reference '{reference}'"))?; + schema = schema_definition(definitions, name)?; + continue; } - Value::Array(values) => { - for child in values { - collect_definition_refs(child, refs); - } + if let Some(all_of) = schema.get("allOf").and_then(Value::as_array) { + anyhow::ensure!(all_of.len() == 1, "schema allOf is not a single reference"); + schema = &all_of[0]; + continue; } - _ => {} + return Ok(schema); + } + anyhow::bail!("schema reference depth exceeds 16") +} + +fn schema_variants<'a>( + definitions: &'a serde_json::Map, + definition: &str, + discriminator: &str, +) -> Result> { + let schema = schema_definition(definitions, definition)?; + let variants = schema + .get("oneOf") + .and_then(Value::as_array) + .with_context(|| format!("{definition} has no oneOf variants"))?; + let mut found = BTreeMap::new(); + for variant in variants { + let variant = resolve_schema(definitions, variant)?; + let required = variant + .get("required") + .and_then(Value::as_array) + .with_context(|| format!("{definition} variant has no required array"))?; + anyhow::ensure!( + required + .iter() + .any(|value| value.as_str() == Some(discriminator)), + "{definition} variant does not require {discriminator}" + ); + let values = variant + .pointer(&format!("/properties/{discriminator}/enum")) + .and_then(Value::as_array) + .with_context(|| format!("{definition} variant has no {discriminator} enum"))?; + anyhow::ensure!( + values.len() == 1, + "{definition} variant discriminator is not exact" + ); + let value = values[0] + .as_str() + .with_context(|| format!("{definition} discriminator is not a string"))?; + anyhow::ensure!( + found.insert(value.to_string(), variant).is_none(), + "{definition} discriminator '{value}' is duplicated" + ); } + Ok(found) +} + +fn schema_enum( + definitions: &serde_json::Map, + definition: &str, +) -> Result> { + let values = schema_definition(definitions, definition)? + .get("enum") + .and_then(Value::as_array) + .with_context(|| format!("{definition} has no enum"))?; + values + .iter() + .map(|value| { + value + .as_str() + .map(str::to_string) + .with_context(|| format!("{definition} has a non-string enum value")) + }) + .collect() +} + +fn property<'a>( + definitions: &'a serde_json::Map, + schema: &'a Value, + name: &str, + label: &str, +) -> Result<&'a Value> { + let schema = resolve_schema(definitions, schema)?; + let property = schema + .get("properties") + .and_then(Value::as_object) + .and_then(|properties| properties.get(name)) + .with_context(|| format!("{label} has no {name} property"))?; + resolve_schema(definitions, property) +} + +fn required_property<'a>( + definitions: &'a serde_json::Map, + schema: &'a Value, + name: &str, + label: &str, +) -> Result<&'a Value> { + let schema = resolve_schema(definitions, schema)?; + let required = schema + .get("required") + .and_then(Value::as_array) + .with_context(|| format!("{label} has no required array"))?; + anyhow::ensure!( + required.iter().any(|value| value.as_str() == Some(name)), + "{label} does not require {name}" + ); + property(definitions, schema, name, label) +} + +fn required_schema_path<'a>( + definitions: &'a serde_json::Map, + definition: &str, + path: &[&str], +) -> Result<&'a Value> { + let mut schema = schema_definition(definitions, definition)?; + let mut label = definition.to_string(); + for component in path { + schema = required_property(definitions, schema, component, &label)?; + label.push('.'); + label.push_str(component); + } + Ok(schema) +} + +fn require_property_type( + definitions: &serde_json::Map, + schema: &Value, + property_name: &str, + expected_type: &str, + required: bool, + label: &str, +) -> Result<()> { + let property = if required { + required_property(definitions, schema, property_name, label)? + } else { + property(definitions, schema, property_name, label)? + }; + require_type( + definitions, + property, + expected_type, + &format!("{label}.{property_name}"), + ) +} + +fn require_type( + definitions: &serde_json::Map, + schema: &Value, + expected: &str, + label: &str, +) -> Result<()> { + let schema = resolve_schema(definitions, schema)?; + let matches = match schema.get("type") { + Some(Value::String(actual)) => actual == expected, + Some(Value::Array(actual)) => actual.iter().any(|value| value.as_str() == Some(expected)), + _ => false, + }; + anyhow::ensure!(matches, "{label} does not accept {expected}"); + Ok(()) +} + +fn require_array<'a>( + definitions: &'a serde_json::Map, + schema: &'a Value, + label: &str, +) -> Result<&'a Value> { + let schema = resolve_schema(definitions, schema)?; + require_type(definitions, schema, "array", label)?; + let items = schema + .get("items") + .with_context(|| format!("{label} has no item schema"))?; + resolve_schema(definitions, items) } pub fn state_dir(catalog_root: &Path, identity: &str) -> PathBuf { @@ -3411,28 +3934,125 @@ fn poll_json_message(websocket: &mut WebSocket) -> Result std::io::Result { +/// One app-server process group and the write end of its wrapper-liveness channel. +/// +/// A watchdog in the dedicated process group owns the read end. The watchdog kills only that +/// group if this wrapper disappears without running Rust cleanup. Its membership also prevents +/// the operating system from reusing the group ID before cleanup. +struct OwnedProcessGroup { + child: Child, + watchdog: Child, + owner_write: Option, + socket_path: Option, + active: bool, +} + +impl OwnedProcessGroup { + fn id(&self) -> u32 { + self.child.id() + } + + fn child_mut(&mut self) -> &mut Child { + &mut self.child + } + + fn terminate(&mut self) { + if !self.active { + return; + } + self.active = false; + let process_group = self.watchdog.id() as i32; + unsafe { + libc::kill(-process_group, libc::SIGKILL); + } + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = self.watchdog.kill(); + let _ = self.watchdog.wait(); + if let Some(socket_path) = self.socket_path.as_deref() { + let _ = fs::remove_file(socket_path); + } + self.owner_write.take(); + } +} + +impl Drop for OwnedProcessGroup { + fn drop(&mut self) { + self.terminate(); + } +} + +fn set_close_on_exec(fd: libc::c_int) -> std::io::Result<()> { + let mut flags = unsafe { libc::fcntl(fd, libc::F_GETFD) }; + if flags == -1 { + return Err(std::io::Error::last_os_error()); + } + flags |= libc::FD_CLOEXEC; + if unsafe { libc::fcntl(fd, libc::F_SETFD, flags) } == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) +} + +/// Spawn a provider launcher in an isolated, wrapper-owned process group. +/// +/// Explicit cleanup covers normal returns and Rust errors. The in-group watchdog covers wrapper +/// crashes, SIGKILL, and supervisor teardown. The watchdog holds the group ID until cleanup, so a +/// stale PID can never identify a process group that belongs to another live owner. A crash can +/// leave one dead socket file; the next launch proves that it has no listener and removes it. +fn spawn_process_group( + command: &mut Command, + socket_path: Option<&Path>, +) -> std::io::Result { + let (watchdog_read, owner_write) = UnixStream::pair()?; + set_close_on_exec(owner_write.as_raw_fd())?; + let owner_write_fd = owner_write.as_raw_fd(); + let mut watchdog_command = Command::new("/bin/sh"); + watchdog_command + .arg("-c") + .arg("IFS= read -r ignored; kill -KILL 0") + .arg("st2-codex-watchdog") + .stdin(Stdio::from(std::os::fd::OwnedFd::from(watchdog_read))) + .stdout(Stdio::null()) + .stderr(Stdio::null()); unsafe { - command.pre_exec(|| { - if libc::setsid() == -1 { - Err(std::io::Error::last_os_error()) - } else { - Ok(()) + watchdog_command.pre_exec(|| { + if libc::setpgid(0, 0) == -1 { + return Err(std::io::Error::last_os_error()); } + Ok(()) }); } - command.spawn() -} - -fn terminate_process_group(child: &mut Child) { - let process_group = child.id() as i32; + let mut watchdog = watchdog_command.spawn()?; + let watchdog_process_group = watchdog.id() as i32; unsafe { - libc::kill(-process_group, libc::SIGKILL); + command.pre_exec(move || { + if libc::setpgid(0, watchdog_process_group) == -1 { + return Err(std::io::Error::last_os_error()); + } + libc::close(owner_write_fd); + Ok(()) + }); } - let _ = child.kill(); - let _ = child.wait(); + let child = match command.spawn() { + Ok(child) => child, + Err(error) => { + drop(owner_write); + unsafe { + libc::kill(-watchdog_process_group, libc::SIGKILL); + } + let _ = watchdog.kill(); + let _ = watchdog.wait(); + return Err(error); + } + }; + Ok(OwnedProcessGroup { + child, + watchdog, + owner_write: Some(owner_write), + socket_path: socket_path.map(Path::to_path_buf), + active: true, + }) } fn terminate_child(child: &mut Child) { @@ -3485,104 +4105,454 @@ mod tests { result.unwrap().is_none(), "a stop while the server sits silent mid-handshake must return the graceful None" ); - assert!( - started.elapsed() < STARTUP_TIMEOUT, - "the stop must unblock the handshake well before the startup timeout" + assert!( + started.elapsed() < STARTUP_TIMEOUT, + "the stop must unblock the handshake well before the startup timeout" + ); + } + use std::os::unix::fs::PermissionsExt; + use std::os::unix::net::UnixListener; + + #[cfg(target_os = "linux")] + fn linux_process_state(pid: i32) -> Option { + std::fs::read_to_string(format!("/proc/{pid}/stat")) + .ok()? + .rsplit_once(") ")? + .1 + .chars() + .next() + } + + fn process_can_retain_cleanup_resources(pid: i32) -> bool { + #[cfg(target_os = "linux")] + if linux_process_state(pid) == Some('Z') { + return false; + } + crate::host_lock::process_alive(pid) + } + + fn object_schema(required: &[&str], properties: &[(&str, Value)]) -> Value { + json!({ + "type": "object", + "required": required, + "properties": properties + .iter() + .map(|(name, schema)| ((*name).to_string(), schema.clone())) + .collect::>() + }) + } + + fn reference(name: &str) -> Value { + json!({ "$ref": format!("#/definitions/{name}") }) + } + + fn array_of(items: Value) -> Value { + json!({ "type": "array", "items": items }) + } + + fn tagged_variant(name: &str, required: &[&str], properties: &[(&str, Value)]) -> Value { + let mut all_required = vec!["type"]; + all_required.extend(required); + let mut all_properties = vec![("type", json!({ "type": "string", "enum": [name] }))]; + all_properties.extend(properties.iter().cloned()); + object_schema(&all_required, &all_properties) + } + + fn method_schema(methods: &[&str]) -> Value { + json!({ + "oneOf": methods + .iter() + .map(|method| object_schema( + &["method"], + &[("method", json!({ "type": "string", "enum": [method] }))], + )) + .collect::>() + }) + } + + fn compatible_protocol_schemas() -> CodexProtocolSchemas { + let mut definitions = serde_json::Map::new(); + definitions.insert( + "ThreadActiveFlag".into(), + json!({ + "type": "string", + "enum": ["waitingOnApproval", "waitingOnUserInput"] + }), + ); + definitions.insert( + "ThreadStatus".into(), + json!({ + "oneOf": [ + tagged_variant("notLoaded", &[], &[]), + tagged_variant("idle", &[], &[]), + tagged_variant("systemError", &[], &[]), + tagged_variant( + "active", + &["activeFlags"], + &[("activeFlags", array_of(reference("ThreadActiveFlag")))], + ) + ] + }), + ); + definitions.insert( + "ThreadItem".into(), + json!({ + "oneOf": [ + tagged_variant("contextCompaction", &[], &[]), + tagged_variant("enteredReviewMode", &[], &[]), + tagged_variant("exitedReviewMode", &[], &[]), + tagged_variant( + "userMessage", + &[], + &[("clientId", json!({ "type": ["string", "null"] }))], + ) + ] + }), + ); + definitions.insert("TextElement".into(), object_schema(&[], &[])); + definitions.insert( + "UserInput".into(), + json!({ + "oneOf": [tagged_variant( + "text", + &["text"], + &[ + ("text", json!({ "type": "string" })), + ("text_elements", array_of(reference("TextElement"))), + ], + )] + }), + ); + definitions.insert( + "ClientInfo".into(), + object_schema( + &["name", "version"], + &[ + ("name", json!({ "type": "string" })), + ("title", json!({ "type": ["string", "null"] })), + ("version", json!({ "type": "string" })), + ], + ), + ); + definitions.insert( + "InitializeCapabilities".into(), + object_schema(&[], &[("experimentalApi", json!({ "type": "boolean" }))]), + ); + definitions.insert( + "InitializeParams".into(), + object_schema( + &["clientInfo"], + &[ + ("clientInfo", reference("ClientInfo")), + ("capabilities", reference("InitializeCapabilities")), + ], + ), + ); + definitions.insert( + "Thread".into(), + object_schema( + &["id", "status", "turns"], + &[ + ("id", json!({ "type": "string" })), + ("status", reference("ThreadStatus")), + ("turns", array_of(reference("Turn"))), + ], + ), + ); + definitions.insert( + "Turn".into(), + object_schema( + &["id", "items"], + &[ + ("id", json!({ "type": "string" })), + ("items", array_of(reference("ThreadItem"))), + ], + ), + ); + for notification in ["TurnStartedNotification", "TurnCompletedNotification"] { + definitions.insert( + notification.into(), + object_schema( + &["threadId", "turn"], + &[ + ("threadId", json!({ "type": "string" })), + ("turn", reference("Turn")), + ], + ), + ); + } + for notification in ["ItemStartedNotification", "ItemCompletedNotification"] { + definitions.insert( + notification.into(), + object_schema( + &["threadId", "turnId", "item"], + &[ + ("threadId", json!({ "type": "string" })), + ("turnId", json!({ "type": "string" })), + ("item", reference("ThreadItem")), + ], + ), + ); + } + definitions.insert( + "ThreadStartedNotification".into(), + object_schema(&["thread"], &[("thread", reference("Thread"))]), + ); + definitions.insert( + "ThreadStatusChangedNotification".into(), + object_schema( + &["threadId", "status"], + &[ + ("threadId", json!({ "type": "string" })), + ("status", reference("ThreadStatus")), + ], + ), + ); + definitions.insert( + "ThreadResumeParams".into(), + object_schema(&["threadId"], &[("threadId", json!({ "type": "string" }))]), + ); + definitions.insert( + "ThreadResumeResponse".into(), + object_schema(&["thread"], &[("thread", reference("Thread"))]), + ); + definitions.insert( + "TurnStartParams".into(), + object_schema( + &["threadId", "input"], + &[ + ("threadId", json!({ "type": "string" })), + ("input", array_of(reference("UserInput"))), + ("clientUserMessageId", json!({ "type": ["string", "null"] })), + ], + ), + ); + definitions.insert( + "TurnSteerParams".into(), + object_schema( + &["threadId", "expectedTurnId", "input"], + &[ + ("threadId", json!({ "type": "string" })), + ("expectedTurnId", json!({ "type": "string" })), + ("input", array_of(reference("UserInput"))), + ("clientUserMessageId", json!({ "type": ["string", "null"] })), + ], + ), + ); + definitions.insert( + "TurnStartResponse".into(), + object_schema(&["turn"], &[("turn", reference("Turn"))]), + ); + definitions.insert( + "TurnSteerResponse".into(), + object_schema(&["turnId"], &[("turnId", json!({ "type": "string" }))]), + ); + definitions.insert( + "ThreadLoadedListResponse".into(), + object_schema( + &["data"], + &[("data", array_of(json!({ "type": "string" })))], + ), + ); + definitions.insert( + "HooksListParams".into(), + object_schema(&[], &[("cwds", array_of(json!({ "type": "string" })))]), + ); + definitions.insert( + "HooksListResponse".into(), + object_schema( + &["data"], + &[("data", array_of(reference("HooksListEntry")))], + ), ); - } - use std::os::unix::fs::PermissionsExt; - use std::os::unix::net::UnixListener; - - #[cfg(target_os = "linux")] - fn linux_process_state(pid: i32) -> Option { - std::fs::read_to_string(format!("/proc/{pid}/stat")) - .ok()? - .rsplit_once(") ")? - .1 - .chars() - .next() - } - - fn process_can_retain_cleanup_resources(pid: i32) -> bool { - #[cfg(target_os = "linux")] - if linux_process_state(pid) == Some('Z') { - return false; + definitions.insert( + "HooksListEntry".into(), + object_schema( + &["hooks"], + &[("hooks", array_of(reference("HookMetadata")))], + ), + ); + definitions.insert( + "HookMetadata".into(), + object_schema( + &["currentHash", "isManaged", "key", "trustStatus"], + &[ + ("currentHash", json!({ "type": "string" })), + ("isManaged", json!({ "type": "boolean" })), + ("key", json!({ "type": "string" })), + ("trustStatus", reference("HookTrustStatus")), + ], + ), + ); + definitions.insert( + "HookTrustStatus".into(), + json!({ + "type": "string", + "enum": ["managed", "modified", "trusted", "untrusted"] + }), + ); + CodexProtocolSchemas { + protocol: json!({ "definitions": definitions }), + client_requests: method_schema(REQUIRED_CODEX_CLIENT_REQUESTS), + client_notifications: method_schema(REQUIRED_CODEX_CLIENT_NOTIFICATIONS), + server_requests: method_schema(&["currentTime/read"]), + server_notifications: method_schema(REQUIRED_CODEX_SERVER_NOTIFICATIONS), } - crate::host_lock::process_alive(pid) } - fn protocol_schema_fixture() -> (Value, Value) { - let arm = |method: &str, payload: &str| { - json!({ - "type": "object", - "properties": { - "method": { "enum": [method] }, - "params": { "$ref": format!("#/definitions/{payload}") } - }, - "required": ["method", "params"] - }) - }; - let mut definitions = serde_json::Map::new(); - for (union, required) in [ - ("ClientRequest", CRITICAL_CLIENT_REQUESTS), - ("ServerNotification", CRITICAL_SERVER_NOTIFICATIONS), + fn write_fake_codex( + root: &Path, + name: &str, + version: &str, + schemas: &CodexProtocolSchemas, + ) -> PathBuf { + let fixture = root.join(format!("{name}-schemas")); + fs::create_dir(&fixture).unwrap(); + for (filename, schema) in [ + ( + "codex_app_server_protocol.v2.schemas.json", + &schemas.protocol, + ), + ("ClientRequest.json", &schemas.client_requests), + ("ClientNotification.json", &schemas.client_notifications), + ("ServerRequest.json", &schemas.server_requests), + ("ServerNotification.json", &schemas.server_notifications), ] { - definitions.insert( - union.to_owned(), - json!({ - "oneOf": required - .iter() - .map(|(method, payload)| arm(method, payload)) - .collect::>() - }), - ); - for (_, payload) in required { - definitions.insert((*payload).to_owned(), json!({ "type": "object" })); - } - } - for response in CRITICAL_RESPONSE_DEFINITIONS { - definitions.insert((*response).to_owned(), json!({ "type": "object" })); + fs::write(fixture.join(filename), serde_json::to_vec(schema).unwrap()).unwrap(); } - ( - json!({ "definitions": definitions }), - json!({ - "oneOf": [{ - "type": "object", - "properties": { "method": { "enum": ["initialized"] } }, - "required": ["method"] - }] - }), + let path = root.join(name); + fs::write( + &path, + format!( + "#!/bin/sh\nif [ \"$1\" = \"--version\" ]; then printf '%s\\n' '{version}'; exit 0; fi\nout=\nwhile [ \"$#\" -gt 0 ]; do if [ \"$1\" = \"--out\" ]; then out=$2; break; fi; shift; done\n[ -n \"$out\" ] || exit 2\ncp '{fixture}/'*.json \"$out/\"\n", + fixture = fixture.display() + ), ) + .unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o755)).unwrap(); + path } #[test] - fn protocol_admission_proves_each_method_payload_linkage() { - let (mut aggregate, client_notification) = protocol_schema_fixture(); - delivery_critical_schema_projection(&aggregate, &client_notification).unwrap(); - let start = aggregate - .pointer_mut("/definitions/ClientRequest/oneOf") - .and_then(Value::as_array_mut) + fn protocol_schema_gate_accepts_a_compatible_release_and_rejects_shape_drift() { + let tmp = tempfile::tempdir().unwrap(); + let compatible = compatible_protocol_schemas(); + let patch = write_fake_codex( + tmp.path(), + "codex-compatible-patch", + "codex-cli 0.150.0", + &compatible, + ); + ensure_supported_protocol(patch.to_str().unwrap()).unwrap(); + + let mut incompatible = compatible_protocol_schemas(); + incompatible + .protocol + .pointer_mut("/definitions/ThreadActiveFlag/enum") .unwrap() - .iter_mut() - .find(|arm| { - arm.pointer("/properties/method/enum/0") - .and_then(Value::as_str) - == Some("turn/start") - }) - .unwrap(); - start["properties"]["params"]["$ref"] = - Value::String("#/definitions/TurnSteerParams".to_owned()); + .as_array_mut() + .unwrap() + .push(Value::String("waitingOnFutureInput".into())); + let incompatible = write_fake_codex( + tmp.path(), + "codex-incompatible-schema", + "codex-cli 0.150.1", + &incompatible, + ); + let error = ensure_supported_protocol(incompatible.to_str().unwrap()).unwrap_err(); + assert!(format!("{error:#}").contains("ThreadActiveFlag changed")); + } - let error = - delivery_critical_schema_projection(&aggregate, &client_notification).unwrap_err(); - assert!(error.to_string().contains("turn/start")); - assert!(error.to_string().contains("TurnStartParams")); + #[test] + fn protocol_schema_gate_accepts_additive_items_and_server_requests() { + let mut schemas = compatible_protocol_schemas(); + schemas + .server_requests + .get_mut("oneOf") + .unwrap() + .as_array_mut() + .unwrap() + .push( + method_schema(&["future/request"]) + .get_mut("oneOf") + .unwrap() + .as_array_mut() + .unwrap() + .remove(0), + ); + schemas + .protocol + .pointer_mut("/definitions/ThreadItem/oneOf") + .unwrap() + .as_array_mut() + .unwrap() + .push(tagged_variant("futureItem", &[], &[])); + + verify_codex_protocol_schemas(&schemas).unwrap(); } #[test] - #[ignore = "requires an installed Codex binary whose schema has completed admission review"] - fn installed_codex_protocol_fingerprint_is_admitted() { - ensure_supported_protocol("codex").unwrap(); + fn protocol_rejection_reaches_the_declared_supervisor_once() { + let tmp = tempfile::tempdir().unwrap(); + let worker = tmp.path().join("agents/h/worker/agent.kdl"); + let supervisor = tmp.path().join("agents/h/cos/agent.kdl"); + fs::create_dir_all(worker.parent().unwrap()).unwrap(); + fs::create_dir_all(supervisor.parent().unwrap()).unwrap(); + fs::write( + &worker, + r#"agent "worker" { + host "h" + supervisor "h.cos" + command "true" +} +"#, + ) + .unwrap(); + fs::write( + &supervisor, + r#"agent "cos" { + host "h" + command "true" +} +"#, + ) + .unwrap(); + let mut incompatible = compatible_protocol_schemas(); + incompatible + .protocol + .pointer_mut("/definitions/ThreadActiveFlag/enum") + .unwrap() + .as_array_mut() + .unwrap() + .push(Value::String("waitingOnFutureInput".into())); + let codex = write_fake_codex( + tmp.path(), + "codex-rejected", + "codex-cli 0.150.1", + &incompatible, + ); + let argv = vec![codex.display().to_string()]; + + for _ in 0..2 { + let error = run_controlled( + tmp.path(), + "h.worker".into(), + "h.worker".into(), + argv.clone(), + ) + .unwrap_err(); + assert!(format!("{error:#}").contains("ThreadActiveFlag changed")); + } + + let inbox = message::list_inbox(&message::inbox_dir(supervisor.parent().unwrap())).unwrap(); + assert_eq!(inbox.len(), 1, "the rejection report was not idempotent"); + assert_eq!(inbox[0].from.as_deref(), Some("h.worker")); + assert_eq!( + inbox[0].subject.as_deref(), + Some("Codex protocol rejected: h.worker") + ); + assert!(inbox[0].body.contains("Native delivery did not start")); + assert!(inbox[0].body.contains("ThreadActiveFlag changed")); } #[test] @@ -4004,6 +4974,7 @@ mod tests { agent_dir, identity: "h.worker".into(), this_host: "h".into(), + supervisor: None, } } @@ -6135,6 +7106,107 @@ mod tests { } } + #[test] + fn an_unclassified_item_holds_until_the_next_idle_status() { + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + let mut state = CodexControlState::new(&runtime, "thread-main".into()); + state + .observe(&json!({ + "method": "turn/started", + "params": { "threadId": "thread-main", "turn": { "id": "turn-1" } } + })) + .unwrap(); + + assert!( + state + .observe(&json!({ + "method": "item/completed", + "params": { + "threadId": "thread-main", + "turnId": "turn-1", + "item": { "type": "futureBlockingItem", "id": "item-1" } + } + })) + .unwrap() + ); + assert!(matches!( + state.observed(), + CodexObservedState::Held { + reason: CodexHoldReason::UnknownProtocol, + turn_id: Some(turn_id), + } if turn_id == "turn-1" + )); + + assert!( + state + .observe(&json!({ + "method": "thread/status/changed", + "params": { + "threadId": "thread-main", + "status": { "type": "idle" } + } + })) + .unwrap() + ); + assert_eq!(state.observed(), &CodexObservedState::Idle); + } + + #[test] + fn an_unclassified_server_request_holds_until_the_next_idle_status() { + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + let mut state = CodexControlState::new(&runtime, "thread-main".into()); + state + .observe(&json!({ + "method": "turn/started", + "params": { "threadId": "thread-main", "turn": { "id": "turn-1" } } + })) + .unwrap(); + + assert!( + !state + .observe(&json!({ + "id": 1, + "method": "item/commandExecution/requestApproval", + "params": {} + })) + .unwrap() + ); + assert!(matches!( + state.observed(), + CodexObservedState::Active { .. } + )); + + assert!( + state + .observe(&json!({ + "id": 2, + "method": "future/request", + "params": {} + })) + .unwrap() + ); + assert_eq!( + state.observed(), + &CodexObservedState::Held { + reason: CodexHoldReason::UnknownProtocol, + turn_id: Some("turn-1".into()), + } + ); + + assert!( + state + .observe(&json!({ + "method": "thread/status/changed", + "params": { + "threadId": "thread-main", + "status": { "type": "idle" } + } + })) + .unwrap() + ); + assert_eq!(state.observed(), &CodexObservedState::Idle); + } + #[test] fn an_errored_turn_completes_into_the_named_error_not_a_conflicting_turn() { // Replays the captured terminal-error ordering (#264): a usage limit emits @@ -6521,7 +7593,16 @@ mod tests { .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()); - let mut launcher = spawn_process_group(&mut command).unwrap(); + let mut launcher = spawn_process_group(&mut command, None).unwrap(); + let mut foreign_command = Command::new("/bin/sh"); + foreign_command + .arg("-c") + .arg("sleep 60") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let mut foreign_owner = spawn_process_group(&mut foreign_command, None).unwrap(); + let foreign_pid = foreign_owner.id() as i32; let deadline = Instant::now() + Duration::from_secs(1); // The shell's `>` redirection creates an empty pidfile before `printf` // writes, so wait for parsable content, not mere file existence. @@ -6540,7 +7621,12 @@ mod tests { "the native descendant was not alive before cleanup" ); - terminate_process_group(&mut launcher); + launcher.terminate(); + assert!( + process_can_retain_cleanup_resources(foreign_pid), + "cleanup killed a different live owner" + ); + foreign_owner.terminate(); let deadline = Instant::now() + Duration::from_secs(1); while process_can_retain_cleanup_resources(descendant) && Instant::now() < deadline { std::thread::sleep(Duration::from_millis(10)); @@ -6557,6 +7643,204 @@ mod tests { ); } + #[test] + fn dropping_a_process_group_owner_reaps_the_group_and_socket() { + let temporary = tempfile::tempdir().unwrap(); + let socket_path = temporary.path().join("app-server.sock"); + let _listener = UnixListener::bind(&socket_path).unwrap(); + let mut command = Command::new("/bin/sh"); + command + .arg("-c") + .arg("sleep 60") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let launcher = spawn_process_group(&mut command, Some(&socket_path)).unwrap(); + let launcher_pid = launcher.id() as i32; + assert!(process_can_retain_cleanup_resources(launcher_pid)); + + drop(launcher); + let deadline = Instant::now() + Duration::from_secs(1); + while (process_can_retain_cleanup_resources(launcher_pid) || socket_path.exists()) + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + !process_can_retain_cleanup_resources(launcher_pid), + "the app-server survived owner cleanup" + ); + assert!( + !socket_path.exists(), + "the app-server socket survived owner cleanup" + ); + } + + #[test] + fn a_live_socket_refuses_a_second_control_owner() { + let temporary = tempfile::tempdir().unwrap(); + let socket_path = temporary.path().join("app-server.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + + let error = prepare_socket_for_launch(&socket_path).unwrap_err(); + + assert!( + error + .to_string() + .contains("refusing a second control owner") + ); + assert!(socket_path.exists(), "the live owner socket was removed"); + assert!( + UnixStream::connect(&socket_path).is_ok(), + "the first owner stopped accepting connections" + ); + drop(listener); + } + + #[test] + fn a_dead_socket_is_removed_before_launch() { + let temporary = tempfile::tempdir().unwrap(); + let socket_path = temporary.path().join("app-server.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + drop(listener); + assert!(socket_path.exists()); + + prepare_socket_for_launch(&socket_path).unwrap(); + + assert!(!socket_path.exists(), "the dead socket was not removed"); + } + + #[test] + fn a_killed_wrapper_reaps_its_app_server_and_the_next_launch_recovers_its_socket() { + const TEST_NAME: &str = "codex_app_server::tests::a_killed_wrapper_reaps_its_app_server_and_the_next_launch_recovers_its_socket"; + const ROLE: &str = "ST2_CODEX_ORPHAN_TEST_ROLE"; + const SOCKET_PATH: &str = "ST2_CODEX_ORPHAN_TEST_SOCKET"; + const PID_PATH: &str = "ST2_CODEX_ORPHAN_TEST_PID"; + const READY_PATH: &str = "ST2_CODEX_ORPHAN_TEST_READY"; + + match std::env::var(ROLE).as_deref() { + Ok("server") => { + let socket_path = PathBuf::from(std::env::var_os(SOCKET_PATH).unwrap()); + let ready_path = PathBuf::from(std::env::var_os(READY_PATH).unwrap()); + let _listener = UnixListener::bind(socket_path).unwrap(); + fs::write(ready_path, b"ready").unwrap(); + loop { + std::thread::sleep(Duration::from_secs(60)); + } + } + Ok("wrapper") => { + let pid_path = PathBuf::from(std::env::var_os(PID_PATH).unwrap()); + let socket_path = PathBuf::from(std::env::var_os(SOCKET_PATH).unwrap()); + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .arg("--exact") + .arg(TEST_NAME) + .arg("--nocapture") + .env(ROLE, "server") + .env(SOCKET_PATH, std::env::var_os(SOCKET_PATH).unwrap()) + .env(READY_PATH, std::env::var_os(READY_PATH).unwrap()) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let server = spawn_process_group(&mut command, Some(&socket_path)).unwrap(); + fs::write(pid_path, server.id().to_string()).unwrap(); + loop { + std::thread::sleep(Duration::from_secs(60)); + } + } + Ok(role) => panic!("unknown orphan test role {role}"), + Err(_) => {} + } + + let temporary = tempfile::tempdir().unwrap(); + let socket_path = temporary.path().join("app-server.sock"); + let pid_path = temporary.path().join("app-server.pid"); + let ready_path = temporary.path().join("app-server.ready"); + let mut wrapper = Command::new(std::env::current_exe().unwrap()) + .arg("--exact") + .arg(TEST_NAME) + .arg("--nocapture") + .env(ROLE, "wrapper") + .env(SOCKET_PATH, &socket_path) + .env(PID_PATH, &pid_path) + .env(READY_PATH, &ready_path) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + while (!pid_path.is_file() || !ready_path.is_file()) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + if !pid_path.is_file() || !ready_path.is_file() { + let _ = wrapper.kill(); + let _ = wrapper.wait(); + panic!("the wrapper did not start its app-server"); + } + let server_pid = fs::read_to_string(&pid_path) + .expect("the wrapper did not report its app-server PID") + .parse::() + .unwrap(); + assert!( + process_can_retain_cleanup_resources(server_pid), + "the app-server was not alive before the wrapper died" + ); + assert!( + fs::symlink_metadata(&socket_path) + .unwrap() + .file_type() + .is_socket(), + "the app-server did not bind its socket" + ); + + unsafe { + libc::kill(wrapper.id() as i32, libc::SIGKILL); + } + let _ = wrapper.wait(); + let deadline = Instant::now() + Duration::from_secs(2); + while process_can_retain_cleanup_resources(server_pid) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + let server_survived = process_can_retain_cleanup_resources(server_pid); + if server_survived { + unsafe { + libc::kill(server_pid, libc::SIGKILL); + } + } + assert!(!server_survived, "the app-server survived its wrapper"); + assert!( + socket_path.exists(), + "the app-server did not leave the expected recoverable socket" + ); + let refusal_deadline = Instant::now() + Duration::from_secs(2); + let refusal = loop { + match UnixStream::connect(&socket_path) { + Ok(stream) if Instant::now() < refusal_deadline => { + drop(stream); + std::thread::sleep(Duration::from_millis(10)); + } + Ok(_) => panic!("the residual socket still had a live listener"), + Err(error) => break error, + } + }; + assert_eq!(refusal.kind(), std::io::ErrorKind::ConnectionRefused); + + prepare_socket_for_launch(&socket_path) + .expect("the next launch did not recover the residual socket"); + assert!( + !socket_path.exists(), + "the next launch did not remove the residual socket" + ); + let replacement = UnixListener::bind(&socket_path) + .expect("the next app-server could not bind the recovered socket"); + assert!( + UnixStream::connect(&socket_path).is_ok(), + "the replacement app-server socket did not accept a connection" + ); + drop(replacement); + } + #[test] fn app_server_configuration_extraction_fails_closed_at_ambiguous_boundaries() { let missing =