From e6c6738ebbd1629aecdc99e800287c8c5c4b7487 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sat, 29 Aug 2026 20:48:27 +0200 Subject: [PATCH 1/4] fix(codex): admit compatible app-server schemas --- docs/vrs/05-harness-state/requirements.md | 5 +- docs/vrs/05-harness-state/spec.md | 13 +- docs/vrs/spec.md | 6 +- src/codex_app_server.rs | 1163 +++++++++++++++++++-- 4 files changed, 1115 insertions(+), 72 deletions(-) diff --git a/docs/vrs/05-harness-state/requirements.md b/docs/vrs/05-harness-state/requirements.md index 75cb6b59..5f8845b9 100644 --- a/docs/vrs/05-harness-state/requirements.md +++ b/docs/vrs/05-harness-state/requirements.md @@ -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 6585906c..1d04540c 100644 --- a/docs/vrs/05-harness-state/spec.md +++ b/docs/vrs/05-harness-state/spec.md @@ -183,10 +183,15 @@ 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. And -hosts running codex-cli at or above 0.148 produce no Codex observed state at -all: `SUPPORTED_CODEX_CLI_VERSIONS` refuses the launch, correctly, until the -pin moves (#267). +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 methods, response fields, blocking flags, and classified item and +server-request kinds that native delivery uses. A compatible patch or minor +release starts without a source change. A delivery-critical schema change +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. ## Codex producer (OHS-R05) diff --git a/docs/vrs/spec.md b/docs/vrs/spec.md index 2f609fbb..0a6fedac 100644 --- a/docs/vrs/spec.md +++ b/docs/vrs/spec.md @@ -1010,9 +1010,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 2222db5e..4fecdf0d 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -9,7 +9,7 @@ //! inbox head and submits typed input only when that state proves an idle or one exact regular //! active turn. -use std::collections::{BTreeMap, VecDeque}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::fs::{self, File, OpenOptions}; use std::io::{Read as _, Write}; use std::net::Shutdown; @@ -33,20 +33,60 @@ use tungstenite::{Message as WebSocketMessage, WebSocket}; use crate::{ding, harness_context, harness_state, message, run, status}; -/// Every admitted version has a delivery-critical schema comparison and live remote-TUI evidence. -/// A later version stays rejected until both checks are repeated; semantic-version proximity is -/// not compatibility evidence for this experimental provider surface. -/// -/// `codex-cli 0.147.0` is admitted on a completed schema comparison against 0.146.0 and on live -/// evidence that reached a submitted `turn/start` against the real binary. Its live evidence stops -/// short of a completed model turn: the account was over its usage limit when the check ran. The -/// `turn/start` response body, `turn/started`, `turn/completed`, the typed `item/completed` -/// receipt, `turn/steer`, and the `thread/resume` subscription path are therefore unproven on this -/// version. See #267. -pub const SUPPORTED_CODEX_CLI_VERSIONS: &[&str] = &[ - "codex-cli 0.145.0", - "codex-cli 0.146.0", - "codex-cli 0.147.0", +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", +]; +// The control observer does not answer server requests. A new request class must be reviewed before +// st2 can prove that the interactive client, and not this observer, owns the response. +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", +]; +// Unknown item kinds can add a delivery hold that the thread status does not expose. Each item kind +// must therefore be classified before startup, even when st2 does not otherwise read its fields. +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"; @@ -286,6 +326,7 @@ struct CodexDeliveryConfig { inbox: PathBuf, identity: String, this_host: String, + supervisor: Option, } impl CodexDeliveryConfig { @@ -298,14 +339,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)] @@ -1244,12 +1328,12 @@ impl CodexControlState { return Ok(false); } let item_type = required_string(message, "/params/item/type", method)?; - // Codex 0.146/0.147 `ThreadItem` has eighteen 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 { @@ -1469,12 +1553,9 @@ impl CodexControlState { /// 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")? @@ -1507,8 +1588,11 @@ pub fn run_controlled( !codex_argv.is_empty(), "Codex controlled launch argv is empty" ); - ensure_supported_version(&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)?; @@ -2924,7 +3008,54 @@ fn completed_tui(status: ExitStatus) -> Result<()> { Ok(()) } -fn ensure_supported_version(codex: &str) -> Result<()> { +struct CodexProtocolSchemas { + protocol: Value, + client_requests: Value, + client_notifications: Value, + server_requests: Value, + server_notifications: Value, +} + +fn ensure_supported_protocol(codex: &str) -> Result<()> { + 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", + "generate-json-schema", + "--experimental", + "--out", + ]) + .arg(generated.path()) + .output() + .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 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 codex_version(codex: &str) -> Result { let output = Command::new(codex) .arg("--version") .output() @@ -2938,14 +3069,536 @@ fn ensure_supported_version(codex: &str) -> Result<()> { .context("Codex version output is not UTF-8")? .trim() .to_string(); + anyhow::ensure!(!actual.is_empty(), "{codex} --version printed nothing"); + Ok(actual) +} + +fn verify_codex_protocol_schemas(schemas: &CodexProtocolSchemas) -> Result<()> { + let definitions = schemas + .protocol + .get("definitions") + .and_then(Value::as_object) + .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", + )?; + let server_requests = schema_methods(&schemas.server_requests, "server request")?; + let classified_server_requests = string_set(CLASSIFIED_CODEX_SERVER_REQUESTS); + let unknown_server_requests = server_requests + .difference(&classified_server_requests) + .cloned() + .collect::>(); + anyhow::ensure!( + unknown_server_requests.is_empty(), + "unclassified server request methods: {}", + unknown_server_requests.join(", ") + ); + + 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")?; + let classified_items = string_set(CLASSIFIED_CODEX_THREAD_ITEMS); + let unknown_items = item_variants + .keys() + .filter(|item| !classified_items.contains(*item)) + .cloned() + .collect::>(); + anyhow::ensure!( + unknown_items.is_empty(), + "unclassified ThreadItem variants: {}", + unknown_items.join(", ") + ); + for item in [ + "contextCompaction", + "enteredReviewMode", + "exitedReviewMode", + "userMessage", + ] { + 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 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(".")), + )?; + } + + 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 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!( + hook == schema_definition(definitions, "HookMetadata")?, + "HooksListEntry.hooks does not contain HookMetadata" + ); + 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!( + 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(methods) +} + +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; + } + 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!( - SUPPORTED_CODEX_CLI_VERSIONS.contains(&actual.as_str()), - "unsupported Codex app-server protocol version '{actual}' (expected one of: {})", - SUPPORTED_CODEX_CLI_VERSIONS.join(", ") + 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 { let base = std::env::var_os("XDG_STATE_HOME") .map(PathBuf::from) @@ -3354,42 +4007,425 @@ mod tests { 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")))], + ), + ); + 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), + } + } + + 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), + ] { + fs::write(fixture.join(filename), serde_json::to_vec(schema).unwrap()).unwrap(); + } + 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_version_gate_accepts_only_the_exact_allowlist() { + fn protocol_schema_gate_accepts_a_compatible_release_and_rejects_shape_drift() { let tmp = tempfile::tempdir().unwrap(); - let write_version = |name: &str, version: &str| { - let path = tmp.path().join(name); - fs::write(&path, format!("#!/bin/sh\nprintf '%s\\n' '{version}'\n")).unwrap(); - fs::set_permissions(&path, fs::Permissions::from_mode(0o755)).unwrap(); - path - }; - for (index, version) in SUPPORTED_CODEX_CLI_VERSIONS.iter().enumerate() { - ensure_supported_version( - write_version(&format!("codex-admitted-{index}"), version) - .to_str() - .unwrap(), - ) - .unwrap(); - } - // An unadmitted release stays rejected until both policy checks are repeated for it, and - // an alpha cannot carry live evidence at all. - for (index, unadmitted) in ["codex-cli 0.144.0", "codex-cli 0.148.0-alpha.21"] - .into_iter() - .enumerate() - { - let error = ensure_supported_version( - write_version(&format!("codex-unadmitted-{index}"), unadmitted) - .to_str() - .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() + .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")); + } + + #[test] + fn protocol_schema_gate_rejects_an_unclassified_server_request() { + 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), + ); + let error = verify_codex_protocol_schemas(&schemas).unwrap_err(); + assert!( + error + .to_string() + .contains("unclassified server request methods: future/request") + ); + } + + #[test] + 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!(error.to_string().contains(unadmitted)); - assert!( - error - .to_string() - .contains(&SUPPORTED_CODEX_CLI_VERSIONS.join(", ")) - ); + 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] @@ -3790,6 +4826,7 @@ mod tests { agent_dir, identity: "h.worker".into(), this_host: "h".into(), + supervisor: None, } } From 680f73d37268cbb2d8cefb1d4acb164b49dafb9e Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sat, 29 Aug 2026 21:44:25 +0200 Subject: [PATCH 2/4] fix(codex): hold additive protocol elements --- docs/vrs/05-harness-state/requirements.md | 2 +- docs/vrs/05-harness-state/spec.md | 19 ++- src/codex_app_server.rs | 194 +++++++++++++++++----- 3 files changed, 171 insertions(+), 44 deletions(-) diff --git a/docs/vrs/05-harness-state/requirements.md b/docs/vrs/05-harness-state/requirements.md index 5f8845b9..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 diff --git a/docs/vrs/05-harness-state/spec.md b/docs/vrs/05-harness-state/spec.md index 1d04540c..1f68347f 100644 --- a/docs/vrs/05-harness-state/spec.md +++ b/docs/vrs/05-harness-state/spec.md @@ -186,12 +186,18 @@ kill that removes the registry entry leaves nothing behind to prove death 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 methods, response fields, blocking flags, and classified item and -server-request kinds that native delivery uses. A compatible patch or minor -release starts without a source change. A delivery-critical schema change -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. +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) @@ -209,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/src/codex_app_server.rs b/src/codex_app_server.rs index 4fecdf0d..03a4f146 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -50,8 +50,8 @@ const REQUIRED_CODEX_SERVER_NOTIFICATIONS: &[&str] = &[ "turn/completed", "turn/started", ]; -// The control observer does not answer server requests. A new request class must be reviewed before -// st2 can prove that the interactive client, and not this observer, owns the response. +// 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", @@ -65,8 +65,8 @@ const CLASSIFIED_CODEX_SERVER_REQUESTS: &[&str] = &[ "item/tool/requestUserInput", "mcpServer/elicitation/request", ]; -// Unknown item kinds can add a delivery hold that the thread status does not expose. Each item kind -// must therefore be classified before startup, even when st2 does not otherwise read its fields. +// 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", @@ -239,6 +239,7 @@ pub enum CodexHoldReason { ConflictingTurn, Review, Compaction, + UnknownProtocol, NotLoaded, SystemError, WaitingOnApproval, @@ -294,6 +295,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"), @@ -1340,7 +1344,8 @@ impl CodexControlState { "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 { @@ -1349,6 +1354,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) @@ -1394,6 +1404,7 @@ impl CodexControlState { reason: CodexHoldReason::Review | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol | CodexHoldReason::ConflictingTurn, .. }, @@ -1430,7 +1441,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, @@ -1470,6 +1484,7 @@ impl CodexControlState { reason: CodexHoldReason::Review | CodexHoldReason::Compaction + | CodexHoldReason::UnknownProtocol | CodexHoldReason::ConflictingTurn | CodexHoldReason::WaitingOnApproval | CodexHoldReason::WaitingOnUserInput @@ -1506,14 +1521,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 { @@ -1548,6 +1567,24 @@ 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 @@ -3095,17 +3132,7 @@ fn verify_codex_protocol_schemas(schemas: &CodexProtocolSchemas) -> Result<()> { REQUIRED_CODEX_SERVER_NOTIFICATIONS, "server notification", )?; - let server_requests = schema_methods(&schemas.server_requests, "server request")?; - let classified_server_requests = string_set(CLASSIFIED_CODEX_SERVER_REQUESTS); - let unknown_server_requests = server_requests - .difference(&classified_server_requests) - .cloned() - .collect::>(); - anyhow::ensure!( - unknown_server_requests.is_empty(), - "unclassified server request methods: {}", - unknown_server_requests.join(", ") - ); + schema_methods(&schemas.server_requests, "server request")?; let status_variants = schema_variants(definitions, "ThreadStatus", "type")?; for status in ["notLoaded", "idle", "systemError", "active"] { @@ -3135,17 +3162,6 @@ fn verify_codex_protocol_schemas(schemas: &CodexProtocolSchemas) -> Result<()> { ); let item_variants = schema_variants(definitions, "ThreadItem", "type")?; - let classified_items = string_set(CLASSIFIED_CODEX_THREAD_ITEMS); - let unknown_items = item_variants - .keys() - .filter(|item| !classified_items.contains(*item)) - .cloned() - .collect::>(); - anyhow::ensure!( - unknown_items.is_empty(), - "unclassified ThreadItem variants: {}", - unknown_items.join(", ") - ); for item in [ "contextCompaction", "enteredReviewMode", @@ -4340,7 +4356,7 @@ mod tests { } #[test] - fn protocol_schema_gate_rejects_an_unclassified_server_request() { + fn protocol_schema_gate_accepts_additive_items_and_server_requests() { let mut schemas = compatible_protocol_schemas(); schemas .server_requests @@ -4356,12 +4372,15 @@ mod tests { .unwrap() .remove(0), ); - let error = verify_codex_protocol_schemas(&schemas).unwrap_err(); - assert!( - error - .to_string() - .contains("unclassified server request methods: future/request") - ); + schemas + .protocol + .pointer_mut("/definitions/ThreadItem/oneOf") + .unwrap() + .as_array_mut() + .unwrap() + .push(tagged_variant("futureItem", &[], &[])); + + verify_codex_protocol_schemas(&schemas).unwrap(); } #[test] @@ -6958,6 +6977,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 From d521e87d60a881f18bb3416a016a9e92941346af Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sat, 29 Aug 2026 20:25:54 +0200 Subject: [PATCH 3/4] fix(codex): reap app-server groups after wrapper death --- src/codex_app_server.rs | 420 ++++++++++++++++++++++++++++++++++------ 1 file changed, 356 insertions(+), 64 deletions(-) diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs index 03a4f146..e78973ec 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -1687,44 +1687,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. @@ -1765,13 +1728,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, @@ -1780,11 +1743,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, @@ -2139,12 +2143,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); @@ -2154,8 +2159,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", @@ -3923,28 +3927,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. +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; if [ -n \"$1\" ]; then /bin/rm -f -- \"$1\"; fi; kill -KILL 0") + .arg("st2-codex-watchdog") + .arg(socket_path.unwrap_or_else(|| Path::new(""))) + .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) { @@ -7464,7 +7565,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. @@ -7484,7 +7594,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)); @@ -7501,6 +7616,183 @@ 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 an_app_server_group_dies_when_its_wrapper_is_killed() { + const TEST_NAME: &str = + "codex_app_server::tests::an_app_server_group_dies_when_its_wrapper_is_killed"; + 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) || socket_path.exists()) + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(10)); + } + let server_survived = process_can_retain_cleanup_resources(server_pid); + let socket_survived = socket_path.exists(); + if server_survived { + unsafe { + libc::kill(server_pid, libc::SIGKILL); + } + } + let _ = fs::remove_file(&socket_path); + assert!(!server_survived, "the app-server survived its wrapper"); + assert!( + !socket_survived, + "the app-server socket survived its wrapper" + ); + } + #[test] fn app_server_configuration_extraction_fails_closed_at_ambiguous_boundaries() { let missing = From 0c3a0dd9502f3ea384daaff81c19468df1316e33 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sat, 29 Aug 2026 23:04:20 +0200 Subject: [PATCH 4/4] fix(codex): recover crash sockets on relaunch --- src/codex_app_server.rs | 40 +++++++++++++++++++++++++++------------- 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs index e78973ec..fcd59d2d 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -3991,7 +3991,8 @@ fn set_close_on_exec(fd: libc::c_int) -> std::io::Result<()> { /// /// 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. +/// 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>, @@ -4002,9 +4003,8 @@ fn spawn_process_group( let mut watchdog_command = Command::new("/bin/sh"); watchdog_command .arg("-c") - .arg("IFS= read -r ignored; if [ -n \"$1\" ]; then /bin/rm -f -- \"$1\"; fi; kill -KILL 0") + .arg("IFS= read -r ignored; kill -KILL 0") .arg("st2-codex-watchdog") - .arg(socket_path.unwrap_or_else(|| Path::new(""))) .stdin(Stdio::from(std::os::fd::OwnedFd::from(watchdog_read))) .stdout(Stdio::null()) .stderr(Stdio::null()); @@ -7684,9 +7684,8 @@ mod tests { } #[test] - fn an_app_server_group_dies_when_its_wrapper_is_killed() { - const TEST_NAME: &str = - "codex_app_server::tests::an_app_server_group_dies_when_its_wrapper_is_killed"; + 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"; @@ -7773,24 +7772,39 @@ mod tests { } let _ = wrapper.wait(); let deadline = Instant::now() + Duration::from_secs(2); - while (process_can_retain_cleanup_resources(server_pid) || socket_path.exists()) - && Instant::now() < deadline - { + 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); - let socket_survived = socket_path.exists(); if server_survived { unsafe { libc::kill(server_pid, libc::SIGKILL); } } - let _ = fs::remove_file(&socket_path); assert!(!server_survived, "the app-server survived its wrapper"); assert!( - !socket_survived, - "the app-server socket survived its wrapper" + socket_path.exists(), + "the app-server did not leave the expected recoverable socket" + ); + assert_eq!( + UnixStream::connect(&socket_path).unwrap_err().kind(), + std::io::ErrorKind::ConnectionRefused, + "the residual socket still had a live listener" + ); + + 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]