RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(runner): bridge Codex dynamic tools (#12382) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The native runner keeps provider execution separate from Paperclip
tool authority.
> - The Rust authorization bridge can now validate a run-scoped semantic
tool catalog.
> - Codex still needs a bounded protocol adapter for that validated
catalog.
> - The adapter must advertise the same catalog after provider recovery.
> - The adapter must reject unadvertised calls and mismatched results.
> - This pull request adds that Codex-only transport boundary.
> - The benefit is a fail-closed provider bridge that remains inactive
until durable wiring supplies tools.

## Linked Issues or Issue Description

**Subsystem affected**

`packages/paperclip-runner` and its Rust Codex app-server provider.

**Problem or motivation**

The native runner has a validated semantic tool catalog, but the Rust
Codex provider cannot project that catalog or correlate Codex tool calls
with Paperclip results. A resumed Codex thread also needs the same
run-scoped catalog.

**Proposed solution**

Add an explicit Codex start path that accepts validated tools. Send
those tools through `dynamicTools` for both thread start and thread
resume. Validate thread, turn, call, operation, result identity, and
size before data crosses the provider boundary.

**Roadmap alignment**

This work supports the shipped governed MCP Tool Gateway and
self-healing run milestones. It does not add a user-facing adapter or
enable the experimental runner.

## What Changed

- Add a Codex dynamic-tool projection for explicit authorized tool sets.
- Advertise the same tool set on new and resumed provider threads.
- Correlate bounded Codex tool calls and Paperclip semantic results.
- Reject calls outside the active thread or turn, unadvertised tools,
reused request identities, mismatched results, and oversized values.
- Keep the current durable backend on the zero-tool path until the next
wiring change.
- Extend the fake Codex app server and integration tests for success,
recovery, denial, and correlation behavior.

## Verification

- `cargo test --manifest-path
packages/paperclip-runner/runner/Cargo.toml --workspace`
- `cargo clippy --manifest-path
packages/paperclip-runner/runner/Cargo.toml --workspace --all-targets --
-D warnings -A clippy::manual_is_multiple_of -A
clippy::filter_map_bool_then`
- `pnpm -r typecheck`
- `pnpm build`
- The PR changes 4 files relative to `runner-rust-semantic-tool-bridge`.

## Risks

Low activation risk. The existing `CodexProvider::start` path still
supplies an empty tool set. The durable backend treats a tool event as
an error until a later PR attaches the durable authorization bridge. The
new transport rejects unknown tools and invalid provider bindings.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

OpenAI Codex with GPT-5 and repository tool use.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
This commit is contained in:
Dotta 2026-08-30 07:56:17 -05:00 committed by GitHub
parent 3ba0e7f64f
commit cf6db7b523
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 2675 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

309
packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs
View File

@ -2,6 +2,8 @@ use std::fs::{self, OpenOptions};
use std::io::{self, BufRead, Write};
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::thread;
use std::time::Duration;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
@ -47,6 +49,19 @@ fn log_call(path: Option<&Path>, method: &str) -> io::Result<()> {
writeln!(file, "{method}")
}
fn has_task_context_tool(message: &Value) -> bool {
message
.pointer("/params/dynamicTools")
.and_then(Value::as_array)
.is_some_and(|tools| {
tools.iter().any(|tool| {
tool.get("name").and_then(Value::as_str) == Some("get_task_context")
&& tool.get("description").and_then(Value::as_str) == Some("Read task context.")
&& tool.pointer("/inputSchema/type").and_then(Value::as_str) == Some("object")
})
})
}
fn finish_turn(state_path: &Path, state: &mut FakeState, status: &str) -> io::Result<()> {
let turn_id = state
.active_turn_id
@ -79,18 +94,110 @@ fn finish_turn(state_path: &Path, state: &mut FakeState, status: &str) -> io::Re
save_state(state_path, state)
}
fn emit_ambiguous_turn_evidence(
state_path: &Path,
state: &mut FakeState,
emit_turn_started: bool,
conflicting_identity: bool,
) -> io::Result<()> {
let turn_id = state
.active_turn_id
.clone()
.unwrap_or_else(|| "provider-turn-2".to_owned());
if emit_turn_started {
send(json!({
"method": "turn/started",
"params": {"turn": {"id": turn_id}}
}))?;
}
if conflicting_identity {
send(json!({
"method": "turn/completed",
"params": {"turn": {"id": "provider-turn-conflict", "status": "completed"}}
}))
} else {
finish_turn(state_path, state, "completed")
}
}
fn run() -> Result<(), Box<dyn std::error::Error>> {
let args = std::env::args().skip(1).collect::<Vec<_>>();
let state_path =
PathBuf::from(argument(&args, "--state-file").ok_or("--state-file is required")?);
let call_log = argument(&args, "--call-log").map(PathBuf::from);
let emit_question = args.iter().any(|value| value == "--emit-question");
let emit_tool_call = args.iter().any(|value| value == "--emit-tool-call");
let replay_completed_tool_call = args
.iter()
.any(|value| value == "--replay-completed-tool-call");
let complete_after_tool_call = args
.iter()
.any(|value| value == "--complete-after-tool-call");
let exit_after_tool_call_completion = args
.iter()
.any(|value| value == "--exit-after-tool-call-completion");
let require_dynamic_tool = args.iter().any(|value| value == "--require-dynamic-tool");
let hold_turn = args.iter().any(|value| value == "--hold-turn");
let exit_after_turn_start = args.iter().any(|value| value == "--exit-after-turn-start");
let exit_after_turn_completion = args
.iter()
.any(|value| value == "--exit-after-turn-completion");
let emit_post_completion_warning = args
.iter()
.any(|value| value == "--emit-post-completion-warning");
let fail_after_turn_completion = args
.iter()
.any(|value| value == "--fail-after-turn-completion");
let fail_after_second_turn_start = args
.iter()
.any(|value| value == "--fail-after-second-turn-start");
let reject_second_turn_start = args
.iter()
.any(|value| value == "--reject-second-turn-start");
let emit_turn_before_rejected_second_start = args
.iter()
.any(|value| value == "--emit-turn-before-rejected-second-start");
let malformed_error_second_turn_start = args
.iter()
.any(|value| value == "--malformed-error-second-turn-start");
let missing_id_second_turn_start = args
.iter()
.any(|value| value == "--missing-id-second-turn-start");
let fail_after_accepting_second_turn_before_response = args
.iter()
.any(|value| value == "--fail-after-accepting-second-turn-before-response");
let exit_after_accepting_second_turn_before_response = args
.iter()
.any(|value| value == "--exit-after-accepting-second-turn-before-response");
let complete_ambiguous_second_turn = args
.iter()
.any(|value| value == "--complete-ambiguous-second-turn");
let retain_ambiguous_second_turn_active = args
.iter()
.any(|value| value == "--retain-ambiguous-second-turn-active");
let hold_ambiguous_second_turn_after_item = args
.iter()
.any(|value| value == "--hold-ambiguous-second-turn-after-item");
let complete_ambiguous_second_turn_before_response = args
.iter()
.any(|value| value == "--complete-ambiguous-second-turn-before-response");
let conflicting_ambiguous_second_turn = args
.iter()
.any(|value| value == "--conflicting-ambiguous-second-turn");
let omit_ambiguous_turn_started = args
.iter()
.any(|value| value == "--omit-ambiguous-turn-started");
let fail_after_thread_read = args.iter().any(|value| value == "--fail-after-thread-read");
let exit_after_thread_read = args.iter().any(|value| value == "--exit-after-thread-read");
let fail_after_turn_completion_delay_ms =
argument(&args, "--fail-after-turn-completion-delay-ms")
.map(|value| value.parse::<u64>())
.transpose()?;
let pre_response_notification = args
.iter()
.any(|value| value == "--notification-before-response");
let mut state = load_state(&state_path);
let mut turn_start_count = 0_u64;
for line in io::stdin().lock().lines() {
let message: Value = serde_json::from_str(&line?)?;
@ -99,6 +206,42 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
finish_turn(&state_path, &mut state, "completed")?;
continue;
}
if message.get("method").is_none() && message.get("id") == Some(&json!("tool-request-1")) {
if message.pointer("/result/success") == Some(&json!(false)) {
log_call(call_log.as_deref(), "tool-response:failure")?;
if state.active_turn_id.is_some() {
finish_turn(&state_path, &mut state, "failed")?;
}
continue;
}
if message.pointer("/result/success") != Some(&json!(true)) {
return Err("semantic tool response omitted success".into());
}
let text = message
.pointer("/result/contentItems/0/text")
.and_then(Value::as_str)
.ok_or("semantic tool response omitted content text")?;
let result: Value = serde_json::from_str(text)?;
if result != json!({"ok": true, "task": {"id": "task-1"}}) {
return Err("semantic tool response changed the operation result".into());
}
if replay_completed_tool_call {
send(json!({
"id": "tool-request-replay",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": state.active_turn_id,
"callId": "semantic-call-1",
"tool": "get_task_context",
"arguments": {}
}
}))?;
continue;
}
finish_turn(&state_path, &mut state, "completed")?;
continue;
}
let Some(method) = message.get("method").and_then(Value::as_str) else {
continue;
};
@ -111,6 +254,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
}))?,
"initialized" => {}
"thread/start" => {
if require_dynamic_tool && !has_task_context_tool(&message) {
return Err("thread/start omitted the authorized dynamic tool".into());
}
state.thread_id = "codex-thread-1".to_owned();
state.active_turn_id = None;
save_state(&state_path, &state)?;
@ -125,10 +271,15 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
"result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}}
}))?;
}
"thread/resume" => send(json!({
"id": id,
"result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}}
}))?,
"thread/resume" => {
if require_dynamic_tool && !has_task_context_tool(&message) {
return Err("thread/resume omitted the authorized dynamic tool".into());
}
send(json!({
"id": id,
"result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}}
}))?;
}
"thread/read" => {
let turns = state
.active_turn_id
@ -139,10 +290,113 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
"id": id,
"result": {"thread": {"id": state.thread_id, "turns": turns}}
}))?;
if fail_after_thread_read {
return Err("configured failure after thread read".into());
} else if exit_after_thread_read {
return Ok(());
}
}
"turn/start" => {
state.active_turn_id = Some("provider-turn-1".to_owned());
turn_start_count += 1;
if reject_second_turn_start && turn_start_count == 2 {
send(json!({
"method": "warning",
"params": {"message": "buffered before replacement rejection"}
}))?;
if emit_turn_before_rejected_second_start {
send(json!({
"method": "turn/started",
"params": {"turn": {"id": "provider-turn-contradiction"}}
}))?;
}
send(json!({
"id": id,
"error": {"code": -32000, "message": "replacement turn rejected"}
}))?;
return Err("configured failure after second turn rejection".into());
}
let emits_ambiguous_turn_evidence = turn_start_count == 2
&& (complete_ambiguous_second_turn
|| complete_ambiguous_second_turn_before_response
|| conflicting_ambiguous_second_turn);
let provider_turn_id = if emits_ambiguous_turn_evidence
|| (turn_start_count == 2
&& (retain_ambiguous_second_turn_active
|| hold_ambiguous_second_turn_after_item))
{
"provider-turn-2"
} else {
"provider-turn-1"
};
state.active_turn_id = Some(provider_turn_id.to_owned());
save_state(&state_path, &state)?;
if complete_ambiguous_second_turn_before_response && turn_start_count == 2 {
emit_ambiguous_turn_evidence(
&state_path,
&mut state,
!omit_ambiguous_turn_started,
false,
)?;
}
if fail_after_accepting_second_turn_before_response && turn_start_count == 2 {
if emits_ambiguous_turn_evidence
&& !complete_ambiguous_second_turn_before_response
{
emit_ambiguous_turn_evidence(
&state_path,
&mut state,
!omit_ambiguous_turn_started,
conflicting_ambiguous_second_turn,
)?;
}
return Err("configured failure after accepting second turn".into());
}
if exit_after_accepting_second_turn_before_response && turn_start_count == 2 {
return Ok(());
}
if malformed_error_second_turn_start && turn_start_count == 2 {
send(json!({"id": id, "error": {}}))?;
if emits_ambiguous_turn_evidence
&& !complete_ambiguous_second_turn_before_response
{
emit_ambiguous_turn_evidence(
&state_path,
&mut state,
!omit_ambiguous_turn_started,
conflicting_ambiguous_second_turn,
)?;
}
return Err("configured failure after malformed turn error".into());
}
if missing_id_second_turn_start && turn_start_count == 2 {
send(json!({
"id": id,
"result": {"turn": {"status": "inProgress"}}
}))?;
if hold_ambiguous_second_turn_after_item {
send(json!({
"method": "item/completed",
"params": {"item": {
"id": "replacement-message-before-terminal",
"type": "agentMessage",
"status": "completed",
"text": "Replacement output before terminal authority."
}}
}))?;
continue;
}
if emits_ambiguous_turn_evidence
&& !complete_ambiguous_second_turn_before_response
{
emit_ambiguous_turn_evidence(
&state_path,
&mut state,
!omit_ambiguous_turn_started,
conflicting_ambiguous_second_turn,
)?;
}
return Err("configured failure after missing turn identity".into());
}
send(json!({
"id": id,
"result": {"turn": {"id": "provider-turn-1", "status": "inProgress"}}
@ -151,8 +405,28 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
"method": "turn/started",
"params": {"turn": {"id": "provider-turn-1"}}
}))?;
if exit_after_turn_start {
if fail_after_second_turn_start && turn_start_count == 2 {
return Err("configured failure after second turn start".into());
} else if exit_after_turn_start {
return Ok(());
} else if emit_tool_call {
send(json!({
"id": "tool-request-1",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"callId": "semantic-call-1",
"tool": "get_task_context",
"arguments": {}
}
}))?;
if complete_after_tool_call {
finish_turn(&state_path, &mut state, "completed")?;
if exit_after_tool_call_completion {
return Ok(());
}
}
} else if emit_question {
send(json!({
"id": "runtime-request-1",
@ -176,6 +450,29 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
}))?;
} else if !hold_turn {
finish_turn(&state_path, &mut state, "completed")?;
if emit_post_completion_warning {
send(json!({
"method": "warning",
"params": {"message": "provider remained live after terminal"}
}))?;
}
if fail_after_turn_completion {
if let Some(delay_ms) = fail_after_turn_completion_delay_ms {
thread::sleep(Duration::from_millis(delay_ms));
// Make the post-terminal liveness observation
// deterministic even when parallel tests delay the
// controller's next poll until after this process
// exits.
send(json!({
"method": "warning",
"params": {"message": "provider remained live after terminal"}
}))?;
}
return Err("configured failure after turn completion".into());
}
if exit_after_turn_completion {
return Ok(());
}
}
}
"turn/steer" => send(json!({"id": id, "result": {"accepted": true}}))?,

765
packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs
View File

@ -1,4 +1,4 @@
use std::collections::{BTreeMap, VecDeque};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::path::{Path, PathBuf};
use std::time::Duration;
@ -8,10 +8,14 @@ use serde_json::{json, Value};
use crate::durable::redact_text;
use crate::local_runner::LocalRunnerError;
use crate::process_supervisor::SupervisedProcess;
use crate::provider_bridge::{AuthorizedTool, ToolResult};
pub const CODEX_APP_SERVER_MAX_FRAME_BYTES: usize = 4 * 1024 * 1024;
const MAX_BUFFERED_MESSAGES: usize = 1_024;
const MAX_INSTRUCTIONS_BYTES: usize = 1024 * 1024;
const MAX_PENDING_TOOL_REQUESTS: usize = 4_096;
const MAX_PENDING_TOOL_REQUEST_BYTES: usize = 16 * 1024 * 1024;
const MAX_COMPLETED_TOOL_CALL_IDS: usize = 4_096;
type QuestionOptionLabels = BTreeMap<String, BTreeMap<String, String>>;
type QuestionSetMapping = (String, Value, QuestionOptionLabels);
@ -101,6 +105,11 @@ impl CodexProviderConfig {
#[derive(Clone, Debug, PartialEq)]
pub enum CodexProviderEvent {
ToolCall {
call_id: String,
operation_id: String,
input: Value,
},
Notification {
method: String,
params: Value,
@ -112,9 +121,41 @@ pub enum CodexProviderEvent {
Exited {
exit_code: Option<i32>,
success: bool,
completed_turn_authoritative: bool,
completed_turn_observed_by_process: bool,
completion_reconciles_exit: bool,
process_generation: u64,
completed_turn_process_generation: Option<u64>,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct CompletedTurnAuthority {
process_generation: u64,
provider_turn_id: String,
}
enum ProviderRequestError {
Rejected(LocalRunnerError),
Ambiguous(LocalRunnerError),
}
impl ProviderRequestError {
fn into_inner(self) -> LocalRunnerError {
match self {
Self::Rejected(error) | Self::Ambiguous(error) => error,
}
}
}
#[derive(Clone, Debug, PartialEq)]
struct PendingToolRequest {
rpc_id: Value,
operation_id: String,
input: Value,
retained_bytes: usize,
}
#[derive(Clone, Debug, PartialEq)]
struct PendingRuntimeRequest {
rpc_id: Value,
@ -124,23 +165,79 @@ struct PendingRuntimeRequest {
option_labels: QuestionOptionLabels,
}
struct BufferedProviderMessage {
value: Value,
}
enum AmbiguousTurnMessage {
Ready,
Deferred,
ReconciledWithStart,
ReconciledNeedsStart { provider_turn_id: String },
}
pub struct CodexProvider {
process: SupervisedProcess,
next_request_id: u64,
thread_id: String,
provider_session_id: Option<String>,
active_provider_turn_id: Option<String>,
pending_messages: VecDeque<Value>,
pending_messages: VecDeque<BufferedProviderMessage>,
deferred_ambiguous_messages: VecDeque<BufferedProviderMessage>,
authorized_tool_ids: BTreeSet<String>,
pending_tool_requests: BTreeMap<String, PendingToolRequest>,
completed_tool_call_ids: BTreeSet<String>,
pending_tool_request_bytes: usize,
pending_runtime_requests: BTreeMap<String, PendingRuntimeRequest>,
expected_shutdown: bool,
process_generation: u64,
completed_turn_authority: Option<CompletedTurnAuthority>,
completion_reconciliation_pending: bool,
ambiguous_turn_start_pending: bool,
}
impl CodexProvider {
pub fn start(
config: &CodexProviderConfig,
resume_thread_id: Option<&str>,
) -> Result<Self, LocalRunnerError> {
Self::start_with_tools_for_generation(config, std::iter::empty(), resume_thread_id, 1)
}
pub fn start_with_tools(
config: &CodexProviderConfig,
authorized_tools: impl IntoIterator<Item = AuthorizedTool>,
resume_thread_id: Option<&str>,
) -> Result<Self, LocalRunnerError> {
Self::start_with_tools_for_generation(config, authorized_tools, resume_thread_id, 1)
}
pub(crate) fn start_for_generation(
config: &CodexProviderConfig,
resume_thread_id: Option<&str>,
process_generation: u64,
) -> Result<Self, LocalRunnerError> {
Self::start_with_tools_for_generation(
config,
std::iter::empty(),
resume_thread_id,
process_generation,
)
}
fn start_with_tools_for_generation(
config: &CodexProviderConfig,
authorized_tools: impl IntoIterator<Item = AuthorizedTool>,
resume_thread_id: Option<&str>,
process_generation: u64,
) -> Result<Self, LocalRunnerError> {
config.validate()?;
if process_generation == 0 {
return Err(LocalRunnerError::invalid(
"Codex process generation must be positive",
));
}
let (dynamic_tools, authorized_tool_ids) = codex_dynamic_tools(authorized_tools)?;
let mut provider = Self {
process: SupervisedProcess::spawn(
&config.command,
@ -153,8 +250,17 @@ impl CodexProvider {
provider_session_id: None,
active_provider_turn_id: None,
pending_messages: VecDeque::new(),
deferred_ambiguous_messages: VecDeque::new(),
authorized_tool_ids,
pending_tool_requests: BTreeMap::new(),
completed_tool_call_ids: BTreeSet::new(),
pending_tool_request_bytes: 0,
pending_runtime_requests: BTreeMap::new(),
expected_shutdown: false,
process_generation,
completed_turn_authority: None,
completion_reconciliation_pending: false,
ambiguous_turn_start_pending: false,
};
let initialized = provider.request(
"initialize",
@ -179,6 +285,7 @@ impl CodexProvider {
"permissions": "paperclip-runner-workspace-only",
"runtimeWorkspaceRoots": [config.cwd],
"baseInstructions": config.instructions,
"dynamicTools": dynamic_tools,
});
let params_object = params
.as_object_mut()
@ -187,9 +294,6 @@ impl CodexProvider {
params_object.insert("threadId".to_owned(), json!(thread_id));
"thread/resume"
} else {
// This PR does not grant any semantic tools. A later catalog and
// authorization layer can project a run-scoped inventory here.
params_object.insert("dynamicTools".to_owned(), json!([]));
params_object.insert("experimentalRawEvents".to_owned(), json!(false));
"thread/start"
};
@ -215,7 +319,9 @@ impl CodexProvider {
if resume_thread_id.is_some() {
let snapshot = provider.read_thread()?;
provider.active_provider_turn_id = latest_active_turn_id(&snapshot);
provider.active_provider_turn_id = latest_active_turn_id(&snapshot)
.map(|turn_id| bounded_identifier(Some(&turn_id), "Codex turn id"))
.transpose()?;
}
Ok(provider)
}
@ -236,18 +342,68 @@ impl CodexProvider {
self.active_provider_turn_id.as_deref()
}
pub(crate) fn ambiguous_turn_start_pending(&self) -> bool {
self.ambiguous_turn_start_pending
}
pub(crate) fn restore_completed_turn_authority(
&mut self,
authoritative: bool,
process_generation: Option<u64>,
provider_turn_id: Option<&str>,
) {
self.completed_turn_authority = authoritative.then(|| CompletedTurnAuthority {
// Legacy state did not record the generation. Generation zero is
// deliberately older than every supervised process generation.
process_generation: process_generation.unwrap_or(0),
provider_turn_id: provider_turn_id
.unwrap_or("durable-completed-turn")
.to_owned(),
});
// Resuming a completed durable thread and reading its provider state
// is recovery, not new turn work. Keep the prior terminal authoritative
// until start_turn explicitly revokes it.
self.expected_shutdown = authoritative;
// Completion authority is durable across provider generations. Output,
// probes, and process restarts are session-liveness observations; none
// of them supersedes a completed result. Only accepting a replacement
// turn identity revokes this authority.
self.completion_reconciliation_pending = false;
}
pub(crate) fn completed_turn_authority(&self) -> Option<(u64, &str)> {
self.completed_turn_authority.as_ref().map(|authority| {
(
authority.process_generation,
authority.provider_turn_id.as_str(),
)
})
}
pub fn start_turn(&mut self, message: &str, cwd: &str) -> Result<Value, LocalRunnerError> {
if self.active_provider_turn_id.is_some() {
return Err(LocalRunnerError::invalid(
"Codex already has an active provider turn",
));
}
if self.ambiguous_turn_start_pending {
return Err(LocalRunnerError::invalid(
"Codex has an unresolved ambiguous provider turn start",
));
}
if message.is_empty() || message.len() > MAX_INSTRUCTIONS_BYTES {
return Err(LocalRunnerError::invalid(
"Codex turn text is empty or exceeds the 1 MiB limit",
));
}
let result = self.request(
// Preserve the prior durable result until a replacement turn identity
// is accepted. A rejected, ambiguous, or transport-failed attempt does
// not prove that replacement work superseded the completed turn.
let prior_reconciliation_pending = self.completion_reconciliation_pending;
self.completion_reconciliation_pending = false;
let prior_buffered_message_count = self.pending_messages.len();
self.ambiguous_turn_start_pending = true;
let result = match self.request_classified(
"turn/start",
json!({
"threadId": self.thread_id,
@ -256,18 +412,92 @@ impl CodexProvider {
"runtimeWorkspaceRoots": [cwd],
"input": [{"type": "text", "text": message, "text_elements": []}],
}),
) {
Ok(result) => result,
Err(ProviderRequestError::Rejected(error)) => {
// A definite rejection proves no replacement work began.
// Only diagnostics without provider-work identity belong to
// that rejected request. Contradictory turn/item evidence or a
// server request leaves the start ambiguous until a validated
// replacement identity is observed.
let definite_rejection = self
.pending_messages
.iter()
.skip(prior_buffered_message_count)
.all(|buffered| is_unbound_rejected_turn_diagnostic(&buffered.value));
if definite_rejection {
self.ambiguous_turn_start_pending = false;
self.completion_reconciliation_pending = prior_reconciliation_pending;
}
return Err(error);
}
Err(ProviderRequestError::Ambiguous(error)) => return Err(error),
};
let provider_turn_id = bounded_identifier(
result
.pointer("/turn/id")
.or_else(|| result.get("turnId"))
.and_then(Value::as_str),
"Codex turn id",
)?;
let provider_turn_id = result
.pointer("/turn/id")
.or_else(|| result.get("turnId"))
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.ok_or_else(|| LocalRunnerError::invalid("Codex turn/start omitted turn.id"))?
.to_owned();
self.active_provider_turn_id = Some(provider_turn_id);
// Only a validated provider turn identity proves that replacement
// work exists and supersedes the prior completed result.
self.accept_replacement_turn(provider_turn_id);
Ok(result)
}
fn classify_ambiguous_turn_message(
&mut self,
message: &Value,
) -> Result<AmbiguousTurnMessage, LocalRunnerError> {
if !self.ambiguous_turn_start_pending {
return Ok(AmbiguousTurnMessage::Ready);
}
let Some(method) = message.get("method").and_then(Value::as_str) else {
return Ok(AmbiguousTurnMessage::Deferred);
};
let params = message.get("params").cloned().unwrap_or(Value::Null);
let provider_turn_id = notification_turn_id(&params);
let identity_required = matches!(method, "turn/started" | "turn/completed");
if provider_turn_id.is_none() && !identity_required {
return Ok(AmbiguousTurnMessage::Deferred);
}
validate_notification_binding(&self.thread_id, None, &params)?;
let provider_turn_id =
bounded_identifier(provider_turn_id, "Codex turn id").map_err(|_| {
LocalRunnerError::invalid(format!(
"Codex {method} cannot resolve an ambiguous turn start without a valid turn id"
))
})?;
if self
.completed_turn_authority
.as_ref()
.is_some_and(|authority| authority.provider_turn_id == provider_turn_id)
{
return Err(LocalRunnerError::invalid(format!(
"Codex {method} notification reused the previously completed turn id while resolving an ambiguous turn start"
)));
}
self.accept_replacement_turn(provider_turn_id.clone());
if method == "turn/started" && message.get("id").is_none() {
Ok(AmbiguousTurnMessage::ReconciledWithStart)
} else {
Ok(AmbiguousTurnMessage::ReconciledNeedsStart { provider_turn_id })
}
}
fn accept_replacement_turn(&mut self, provider_turn_id: String) {
self.ambiguous_turn_start_pending = false;
self.expected_shutdown = false;
self.completed_turn_authority = None;
self.completion_reconciliation_pending = false;
self.completed_tool_call_ids.clear();
self.active_provider_turn_id = Some(provider_turn_id);
}
pub fn steer_turn(&mut self, message: &str) -> Result<Value, LocalRunnerError> {
let turn_id = self
.active_provider_turn_id
@ -293,7 +523,7 @@ impl CodexProvider {
.active_provider_turn_id
.clone()
.ok_or_else(|| LocalRunnerError::invalid("Codex has no active provider turn"))?;
self.cancel_pending_runtime_requests()?;
self.cancel_pending_requests()?;
self.request(
"turn/interrupt",
json!({"threadId": self.thread_id, "turnId": turn_id}),
@ -301,6 +531,11 @@ impl CodexProvider {
}
pub fn read_thread(&mut self) -> Result<Value, LocalRunnerError> {
// Probing provider state does not supersede an authoritative terminal.
// A later replacement turn must still establish its own identity.
// It does prove the provider remained live after that terminal, so a
// subsequent nonzero exit is a separate idle-session failure.
self.completion_reconciliation_pending = false;
self.request(
"thread/read",
json!({"threadId": self.thread_id, "includeTurns": true}),
@ -327,14 +562,48 @@ impl CodexProvider {
}
pub fn poll(&mut self) -> Result<Option<CodexProviderEvent>, LocalRunnerError> {
let message = if let Some(message) = self.pending_messages.pop_front() {
message
let buffered = self.pending_messages.pop_front();
let message = if let Some(buffered) = buffered {
buffered.value
} else {
let Some(line) = self.process.receive_stdout_line(Duration::from_millis(1))? else {
return if let Some(exit) = self.process.try_wait()? {
let exit = self.process.try_wait()?;
return if let Some(exit) = exit {
let completed_turn_authoritative = self.completed_turn_authority.is_some()
&& self.active_provider_turn_id.is_none();
let completed_turn_observed_by_process = self
.completed_turn_authority
.as_ref()
.is_some_and(|authority| {
authority.process_generation == self.process_generation
});
// A durable terminal remains the run outcome, but it only
// reconciles the process generation that produced it. A
// later recovered provider can fail independently while
// leaving the already-recorded turn result intact.
let completion_reconciles_exit = completed_turn_authoritative
&& completed_turn_observed_by_process
&& self.completion_reconciliation_pending;
Ok(Some(CodexProviderEvent::Exited {
exit_code: exit.exit_code,
success: exit.success && self.expected_shutdown,
// A clean idle exit after a terminal is healthy. A
// nonzero exit still makes the provider unavailable,
// but the durable terminal reconciles it instead of
// allowing the session to fail retroactively. Fresh
// turn work explicitly revokes the prior authority.
// An unresolved start may already have created fresh
// work, so even a clean exit must fail that session.
success: exit.success
&& !self.ambiguous_turn_start_pending
&& (self.expected_shutdown || completed_turn_authoritative),
completed_turn_authoritative,
completed_turn_observed_by_process,
completion_reconciles_exit,
process_generation: self.process_generation,
completed_turn_process_generation: self
.completed_turn_authority
.as_ref()
.map(|authority| authority.process_generation),
}))
} else {
Ok(None)
@ -343,10 +612,157 @@ impl CodexProvider {
parse_provider_message(&line)?
};
if self.completed_turn_authority.is_some()
&& self.active_provider_turn_id.is_none()
&& message.get("method").and_then(Value::as_str) != Some("turn/completed")
{
// Output after the terminal proves the provider entered an idle
// liveness phase. Keep the result authoritative, but do not let it
// hide a later process failure.
self.completion_reconciliation_pending = false;
}
match self.classify_ambiguous_turn_message(&message)? {
AmbiguousTurnMessage::Ready => {}
AmbiguousTurnMessage::Deferred => {
if self.deferred_ambiguous_messages.len() >= MAX_BUFFERED_MESSAGES {
return Err(LocalRunnerError::invalid(
"Codex emitted too many messages before resolving an ambiguous turn start",
));
}
self.deferred_ambiguous_messages
.push_back(BufferedProviderMessage { value: message });
return Ok(None);
}
AmbiguousTurnMessage::ReconciledWithStart => {
let mut replay = std::mem::take(&mut self.deferred_ambiguous_messages);
replay.append(&mut self.pending_messages);
self.pending_messages = replay;
}
AmbiguousTurnMessage::ReconciledNeedsStart { provider_turn_id } => {
let mut replay = std::mem::take(&mut self.deferred_ambiguous_messages);
replay.push_back(BufferedProviderMessage { value: message });
replay.append(&mut self.pending_messages);
self.pending_messages = replay;
return Ok(Some(CodexProviderEvent::Notification {
method: "turn/started".to_owned(),
params: json!({
"threadId": self.thread_id,
"turn": {"id": provider_turn_id, "status": "inProgress"},
"reconciled": true,
}),
}));
}
}
if let (Some(rpc_id), Some(method)) = (
message.get("id").cloned(),
message.get("method").and_then(Value::as_str),
) {
if method == "item/tool/call" {
let params = message.get("params").cloned().unwrap_or(Value::Null);
if params.get("threadId").and_then(Value::as_str) != Some(self.thread_id.as_str()) {
return Err(LocalRunnerError::invalid(
"Codex tool call named another thread",
));
}
let active_turn_id = self.active_provider_turn_id.as_deref().ok_or_else(|| {
LocalRunnerError::invalid("Codex tool call arrived outside an active turn")
})?;
if params.get("turnId").and_then(Value::as_str) != Some(active_turn_id) {
return Err(LocalRunnerError::invalid(
"Codex tool call named another turn",
));
}
let call_id = bounded_identifier(
params.get("callId").and_then(Value::as_str),
"Codex tool callId",
)?;
let operation_id = bounded_identifier(
params.get("tool").and_then(Value::as_str),
"Codex tool name",
)?;
if !self.authorized_tool_ids.contains(&operation_id) {
self.process.send(&json!({
"id": rpc_id,
"result": codex_tool_failure("Paperclip did not authorize this tool for the run"),
}))?;
return Err(LocalRunnerError::invalid(format!(
"Codex requested unauthorized tool {}",
bounded_method(&operation_id)
)));
}
let input = params.get("arguments").cloned().unwrap_or(Value::Null);
let input_bytes = serde_json::to_vec(&input)
.map_err(|error| {
LocalRunnerError::invalid(format!(
"Codex tool arguments are not serializable: {error}"
))
})?
.len();
let rpc_id_bytes = serde_json::to_vec(&rpc_id)
.map_err(|error| {
LocalRunnerError::invalid(format!(
"Codex JSON-RPC id is not serializable: {error}"
))
})?
.len();
let retained_bytes = pending_tool_request_size([
input_bytes,
rpc_id_bytes,
call_id.len(),
operation_id.len(),
])?;
let pending = PendingToolRequest {
rpc_id: rpc_id.clone(),
operation_id: operation_id.clone(),
input: input.clone(),
retained_bytes,
};
if self.completed_tool_call_ids.contains(&call_id) {
return Err(LocalRunnerError::invalid(
"Codex reused a completed tool call id",
));
}
if let Some(existing) = self.pending_tool_requests.get(&call_id) {
if existing != &pending {
return Err(LocalRunnerError::invalid(
"Codex reused a tool call id with different input",
));
}
return Ok(None);
}
if self
.pending_tool_requests
.values()
.any(|existing| existing.rpc_id == rpc_id)
{
return Err(LocalRunnerError::invalid(
"Codex reused a pending JSON-RPC id for another tool call",
));
}
if self.pending_tool_requests.len() >= MAX_PENDING_TOOL_REQUESTS {
return Err(LocalRunnerError::invalid(
"Codex emitted too many pending tool calls",
));
}
if self.completed_tool_call_ids.len() >= MAX_COMPLETED_TOOL_CALL_IDS {
return Err(LocalRunnerError::invalid(
"Codex emitted too many completed tool calls in one turn",
));
}
let retained_request_bytes = retain_pending_tool_request_bytes(
self.pending_tool_request_bytes,
retained_bytes,
)?;
self.pending_tool_requests.insert(call_id.clone(), pending);
self.pending_tool_request_bytes = retained_request_bytes;
return Ok(Some(CodexProviderEvent::ToolCall {
call_id,
operation_id,
input,
}));
}
if method == "item/tool/requestUserInput" {
let params = message.get("params").cloned().unwrap_or(Value::Null);
if params.get("threadId").and_then(Value::as_str) != Some(self.thread_id.as_str()) {
@ -407,7 +823,24 @@ impl CodexProvider {
&params,
)?;
if method == "turn/completed" {
let provider_turn_id = self.active_provider_turn_id.clone().ok_or_else(|| {
LocalRunnerError::invalid(
"Codex completion arrived outside an active provider turn",
)
})?;
self.active_provider_turn_id = None;
self.expected_shutdown = true;
self.completed_turn_authority = Some(CompletedTurnAuthority {
process_generation: self.process_generation,
provider_turn_id,
});
self.completion_reconciliation_pending = true;
// The provider terminal is authoritative once received. Clear
// local request ownership and attempt courtesy responses, but
// a provider that already closed stdin must not turn the
// completed turn back into a transport failure.
let _ = self.cancel_pending_requests();
self.completed_tool_call_ids.clear();
}
return Ok(Some(CodexProviderEvent::Notification {
method: method.to_owned(),
@ -417,60 +850,248 @@ impl CodexProvider {
Ok(None)
}
pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> {
self.expected_shutdown = true;
self.cancel_pending_runtime_requests()?;
self.process.terminate_group().map(|_| ())
}
fn cancel_pending_runtime_requests(&mut self) -> Result<(), LocalRunnerError> {
let pending = std::mem::take(&mut self.pending_runtime_requests);
for request in pending.into_values() {
self.process.send(&json!({
"id": request.rpc_id,
"result": {"answers": {}},
}))?;
pub fn deliver_tool_result(&mut self, result: &ToolResult) -> Result<(), LocalRunnerError> {
let pending = self
.pending_tool_requests
.get(&result.call_id)
.cloned()
.ok_or_else(|| {
LocalRunnerError::invalid("Codex tool result has no pending JSON-RPC request")
})?;
if pending.operation_id != result.operation_id {
return Err(LocalRunnerError::invalid(
"Codex tool result operation does not match its call",
));
}
let result_bytes = serde_json::to_vec(&result.result).map_err(|error| {
LocalRunnerError::invalid(format!("Codex tool result is not serializable: {error}"))
})?;
if result_bytes.len() > 1024 * 1024 {
return Err(LocalRunnerError::invalid(
"Codex tool result exceeds the 1 MiB limit",
));
}
let text = String::from_utf8(result_bytes)
.expect("serde_json always serializes JSON values as valid UTF-8");
self.process.send(&json!({
"id": pending.rpc_id,
"result": {
"success": !result.is_error,
"contentItems": [{"type": "inputText", "text": text}],
},
}))?;
if let Some(completed) = self.pending_tool_requests.remove(&result.call_id) {
self.pending_tool_request_bytes = self
.pending_tool_request_bytes
.saturating_sub(completed.retained_bytes);
self.completed_tool_call_ids.insert(result.call_id.clone());
}
Ok(())
}
pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> {
self.expected_shutdown = true;
self.cancel_pending_requests()?;
self.process.terminate_group().map(|_| ())
}
fn cancel_pending_requests(&mut self) -> Result<(), LocalRunnerError> {
let pending_runtime = std::mem::take(&mut self.pending_runtime_requests);
let pending = std::mem::take(&mut self.pending_tool_requests);
self.pending_tool_request_bytes = 0;
let mut first_error = None;
for request in pending_runtime.into_values() {
if let Err(error) = self.process.send(&json!({
"id": request.rpc_id,
"result": {"answers": {}},
})) {
first_error.get_or_insert(error);
}
}
for request in pending.into_values() {
if let Err(error) = self.process.send(&json!({
"id": request.rpc_id,
"result": codex_tool_failure("Paperclip stopped the active provider turn"),
})) {
first_error.get_or_insert(error);
}
}
match first_error {
Some(error) => Err(error),
None => Ok(()),
}
}
fn request(&mut self, method: &str, params: Value) -> Result<Value, LocalRunnerError> {
self.request_classified(method, params)
.map_err(ProviderRequestError::into_inner)
}
fn request_classified(
&mut self,
method: &str,
params: Value,
) -> Result<Value, ProviderRequestError> {
let request_id = self.next_request_id;
self.next_request_id = self
.next_request_id
.checked_add(1)
.ok_or_else(|| LocalRunnerError::invalid("Codex request id exhausted"))?;
self.next_request_id = self.next_request_id.checked_add(1).ok_or_else(|| {
ProviderRequestError::Rejected(LocalRunnerError::invalid("Codex request id exhausted"))
})?;
self.process
.send(&json!({"id": request_id, "method": method, "params": params}))?;
.send(&json!({"id": request_id, "method": method, "params": params}))
.map_err(ProviderRequestError::Ambiguous)?;
loop {
let line = self
.process
.receive_stdout_line(Duration::from_secs(30))?
.receive_stdout_line(Duration::from_secs(30))
.map_err(ProviderRequestError::Ambiguous)?
.ok_or_else(|| {
LocalRunnerError::invalid(format!("Codex {method} response timed out"))
ProviderRequestError::Ambiguous(LocalRunnerError::invalid(format!(
"Codex {method} response timed out"
)))
})?;
let message = parse_provider_message(&line)?;
let message = parse_provider_message(&line).map_err(ProviderRequestError::Ambiguous)?;
if message.get("id").and_then(Value::as_u64) == Some(request_id)
&& message.get("method").is_none()
{
if let Some(error) = message.get("error") {
return Err(LocalRunnerError::invalid(format!(
"Codex {method} failed: {}",
redact_text(&error.to_string())
let well_formed_rejection = error.get("code").and_then(Value::as_i64).is_some()
&& error.get("message").and_then(Value::as_str).is_some();
if !well_formed_rejection || message.get("result").is_some() {
return Err(ProviderRequestError::Ambiguous(LocalRunnerError::invalid(
format!("Codex {method} returned an invalid error response"),
)));
}
return Err(ProviderRequestError::Rejected(LocalRunnerError::invalid(
format!("Codex {method} failed: {}", redact_text(&error.to_string())),
)));
}
return Ok(message.get("result").cloned().unwrap_or(Value::Null));
}
if self.pending_messages.len() >= MAX_BUFFERED_MESSAGES {
return Err(LocalRunnerError::invalid(
return Err(ProviderRequestError::Ambiguous(LocalRunnerError::invalid(
"Codex emitted too many messages before a request response",
));
)));
}
self.pending_messages.push_back(message);
self.pending_messages
.push_back(BufferedProviderMessage { value: message });
}
}
}
fn pending_tool_request_size(
parts: impl IntoIterator<Item = usize>,
) -> Result<usize, LocalRunnerError> {
parts.into_iter().try_fold(0usize, |total, part| {
total
.checked_add(part)
.ok_or_else(|| LocalRunnerError::invalid("Codex pending tool request size overflowed"))
})
}
fn retain_pending_tool_request_bytes(
current: usize,
incoming: usize,
) -> Result<usize, LocalRunnerError> {
current
.checked_add(incoming)
.filter(|total| *total <= MAX_PENDING_TOOL_REQUEST_BYTES)
.ok_or_else(|| {
LocalRunnerError::invalid(
"Codex pending tool requests exceed the 16 MiB aggregate limit",
)
})
}
fn codex_dynamic_tools(
authorized_tools: impl IntoIterator<Item = AuthorizedTool>,
) -> Result<(Vec<Value>, BTreeSet<String>), LocalRunnerError> {
let mut dynamic_tools = Vec::new();
let mut operation_ids = BTreeSet::new();
for tool in authorized_tools {
if dynamic_tools.len() >= 256 {
return Err(LocalRunnerError::invalid(
"Codex authorized tool set exceeds the operation limit",
));
}
let operation_id = bounded_identifier(Some(&tool.operation_id), "Codex tool name")?;
let mut characters = operation_id.chars();
let valid_first = characters
.next()
.is_some_and(|character| character.is_ascii_alphanumeric());
let valid_rest = characters.all(|character| {
character.is_ascii_alphanumeric() || matches!(character, '_' | '-' | '.' | ':')
});
if !valid_first || !valid_rest {
return Err(LocalRunnerError::invalid(
"Codex tool name is not a valid operation id",
));
}
if tool.version != 1
|| tool.description.trim().is_empty()
|| tool.description.len() > 16 * 1024
|| tool.description.contains('\0')
|| !tool.input_schema.is_object()
|| !tool.response_schema.is_object()
{
return Err(LocalRunnerError::invalid(format!(
"Codex tool {} has an incomplete provider contract",
bounded_method(&operation_id)
)));
}
let input_schema_bytes = serde_json::to_vec(&tool.input_schema).map_err(|error| {
LocalRunnerError::invalid(format!("Codex tool input schema is invalid: {error}"))
})?;
if input_schema_bytes.len() > 1024 * 1024 {
return Err(LocalRunnerError::invalid(
"Codex tool input schema exceeds the 1 MiB limit",
));
}
jsonschema::validator_for(&tool.input_schema).map_err(|_| {
LocalRunnerError::invalid(format!(
"Codex tool {} has an invalid input JSON Schema",
bounded_method(&operation_id)
))
})?;
if !operation_ids.insert(operation_id.clone()) {
return Err(LocalRunnerError::invalid(
"Codex authorized tool names must be unique",
));
}
dynamic_tools.push(json!({
"name": operation_id,
"description": tool.description,
"inputSchema": tool.input_schema,
}));
}
if serde_json::to_vec(&dynamic_tools)
.map_err(|error| {
LocalRunnerError::invalid(format!("Codex dynamic tool set is invalid: {error}"))
})?
.len()
> 4 * 1024 * 1024
{
return Err(LocalRunnerError::invalid(
"Codex dynamic tool set exceeds the 4 MiB limit",
));
}
Ok((dynamic_tools, operation_ids))
}
fn bounded_identifier(value: Option<&str>, label: &str) -> Result<String, LocalRunnerError> {
let value = value.ok_or_else(|| LocalRunnerError::invalid(format!("{label} is required")))?;
if value.is_empty() || value.len() > 160 || value.chars().any(char::is_control) {
return Err(LocalRunnerError::invalid(format!("{label} is invalid")));
}
Ok(value.to_owned())
}
fn codex_tool_failure(message: &str) -> Value {
json!({
"success": false,
"contentItems": [{"type": "inputText", "text": message}],
})
}
fn parse_provider_message(line: &str) -> Result<Value, LocalRunnerError> {
let value: Value = serde_json::from_str(line).map_err(|error| {
LocalRunnerError::invalid(format!("Codex emitted invalid JSON-RPC: {error}"))
@ -483,6 +1104,28 @@ fn parse_provider_message(line: &str) -> Result<Value, LocalRunnerError> {
Ok(value)
}
fn is_unbound_rejected_turn_diagnostic(message: &Value) -> bool {
message.get("id").is_none()
&& message.get("method").and_then(Value::as_str) == Some("warning")
&& message
.get("params")
.is_none_or(|params| !contains_provider_work_binding(params))
}
fn contains_provider_work_binding(value: &Value) -> bool {
match value {
Value::Array(values) => values.iter().any(contains_provider_work_binding),
Value::Object(fields) => fields.iter().any(|(key, child)| {
(matches!(key.as_str(), "threadId" | "turnId" | "itemId" | "requestId")
&& !child.is_null())
|| (matches!(key.as_str(), "thread" | "turn" | "item" | "request")
&& child.get("id").is_some_and(|id| !id.is_null()))
|| contains_provider_work_binding(child)
}),
_ => false,
}
}
fn validate_notification_binding(
thread_id: &str,
active_turn_id: Option<&str>,
@ -498,11 +1141,7 @@ fn validate_notification_binding(
"Codex notification named another thread",
));
}
let notification_turn_id = params
.get("turnId")
.or_else(|| params.pointer("/turn/id"))
.and_then(Value::as_str)
.filter(|value| !value.is_empty());
let notification_turn_id = notification_turn_id(params);
if let Some(active_turn_id) = active_turn_id {
if notification_turn_id.is_some_and(|value| value != active_turn_id) {
return Err(LocalRunnerError::invalid(
@ -513,6 +1152,14 @@ fn validate_notification_binding(
Ok(())
}
fn notification_turn_id(params: &Value) -> Option<&str> {
params
.get("turnId")
.or_else(|| params.pointer("/turn/id"))
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
}
fn latest_active_turn_id(snapshot: &Value) -> Option<String> {
snapshot
.pointer("/thread/turns")
@ -855,4 +1502,18 @@ mod tests {
)
.is_ok());
}
#[test]
fn bounds_all_retained_pending_tool_request_data_in_aggregate() {
let request_bytes = pending_tool_request_size([1, 2, 3, 4]).unwrap();
assert_eq!(request_bytes, 10);
assert_eq!(
retain_pending_tool_request_bytes(MAX_PENDING_TOOL_REQUEST_BYTES - 10, request_bytes)
.unwrap(),
MAX_PENDING_TOOL_REQUEST_BYTES
);
assert!(retain_pending_tool_request_bytes(MAX_PENDING_TOOL_REQUEST_BYTES, 1).is_err());
assert!(retain_pending_tool_request_bytes(usize::MAX, 1).is_err());
assert!(pending_tool_request_size([usize::MAX, 1]).is_err());
}
}

8
packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs
View File

@ -308,9 +308,11 @@ impl SupervisedProcess {
}
Ok(ProcessOutput::StdoutClosed) => return Ok(None),
Err(RecvTimeoutError::Timeout) => return Ok(None),
Err(RecvTimeoutError::Disconnected) => {
return Err(LocalRunnerError::invalid("process output channel closed"));
}
// The reader threads end when the child closes its output.
// Let the caller reconcile that closure with the authoritative
// process exit status instead of turning the channel teardown
// into a transport failure.
Err(RecvTimeoutError::Disconnected) => return Ok(None),
}
}
}

306
packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
View File

@ -164,6 +164,15 @@ struct CodexProviderState {
#[serde(default)]
active_provider_turn_id: Option<String>,
#[serde(default)]
ambiguous_turn_start_pending: bool,
#[serde(default)]
completed_turn_authoritative: bool,
#[serde(default)]
provider_process_generation: u64,
#[serde(default)]
completed_turn_process_generation: Option<u64>,
#[serde(default)]
completed_provider_turn_id: Option<String>,
last_agent_message: Option<String>,
#[serde(default)]
pending_events: VecDeque<PolledEvent>,
@ -185,6 +194,11 @@ impl CodexProviderState {
thread_id,
provider_session_id: None,
active_provider_turn_id: None,
ambiguous_turn_start_pending: false,
completed_turn_authoritative: false,
provider_process_generation: 0,
completed_turn_process_generation: None,
completed_provider_turn_id: None,
last_agent_message: None,
pending_events: VecDeque::new(),
next_provider_event_seq: initial_provider_event_seq(),
@ -213,6 +227,10 @@ impl CodexProviderState {
.active_provider_turn_id
.as_ref()
.is_some_and(|value| value.is_empty() || value.len() > 240)
|| self
.completed_provider_turn_id
.as_ref()
.is_some_and(|value| value.is_empty() || value.len() > 240)
|| self.completion_contract.as_ref().is_some_and(|contract| {
contract.revision.is_empty()
|| contract.revision.len() > 120
@ -233,6 +251,20 @@ impl CodexProviderState {
|| self.active_provider_turn_id.is_some()
|| matches!(self.lifecycle.as_str(), "session_open" | "turn_active")))
|| (self.lifecycle == "turn_active" && self.active_provider_turn_id.is_none())
|| (self.ambiguous_turn_start_pending
&& (self.thread_id.is_none()
|| self.active_provider_turn_id.is_some()
|| matches!(
self.lifecycle.as_str(),
"prepared" | "turn_active" | "closed"
)))
|| (self.completed_turn_authoritative && self.active_provider_turn_id.is_some())
|| (!self.completed_turn_authoritative
&& (self.completed_turn_process_generation.is_some()
|| self.completed_provider_turn_id.is_some()))
|| self
.completed_turn_process_generation
.is_some_and(|generation| generation > self.provider_process_generation)
|| (matches!(
self.lifecycle.as_str(),
"prepared" | "session_open" | "closed"
@ -279,6 +311,25 @@ impl CodexProviderState {
}
Ok(())
}
fn reconcile_active_provider_turn(&mut self, active_provider_turn_id: Option<String>) {
self.active_provider_turn_id = active_provider_turn_id;
if self.active_provider_turn_id.is_some() {
// A newly discovered turn supersedes completion authority from the
// prior turn. Persisting both would make the recovered state
// invalid and could misclassify a later provider exit.
self.completed_turn_authoritative = false;
self.completed_turn_process_generation = None;
self.completed_provider_turn_id = None;
self.ambiguous_turn_start_pending = false;
self.last_agent_message = None;
}
self.lifecycle = if self.active_provider_turn_id.is_some() {
"turn_active".to_owned()
} else {
"session_open".to_owned()
};
}
}
pub struct CodexCommandExecutor {
@ -354,22 +405,56 @@ impl CodexCommandExecutor {
DurableRunnerError::invalid("recoverable Codex state omitted its thread id")
})?;
let previous_active_turn_id = state.active_provider_turn_id.clone();
let provider = CodexProvider::start(&state.config, Some(&thread_id)).map_err(|error| {
let process_generation = state
.provider_process_generation
.checked_add(1)
.ok_or_else(|| DurableRunnerError::invalid("Codex process generation exhausted"))?;
let completed_turn_authoritative = state.completed_turn_authoritative;
let completed_turn_process_generation = state.completed_turn_process_generation;
let completed_provider_turn_id = state.completed_provider_turn_id.clone();
let ambiguous_turn_start_pending = state.ambiguous_turn_start_pending;
let mut provider = CodexProvider::start_for_generation(
&state.config,
Some(&thread_id),
process_generation,
)
.map_err(|error| {
DurableRunnerError::invalid(format!("failed to resume Codex provider: {error}"))
})?;
let recovered_active_turn_id = provider.active_provider_turn_id().map(str::to_owned);
if ambiguous_turn_start_pending {
let recovered_turn_id = recovered_active_turn_id.as_deref().ok_or_else(|| {
DurableRunnerError::invalid(
"cannot safely recover an ambiguous Codex turn start without an active replacement turn",
)
})?;
if completed_provider_turn_id.as_deref() == Some(recovered_turn_id) {
return Err(DurableRunnerError::invalid(
"ambiguous Codex turn recovery reused the previously completed turn identity",
));
}
}
provider.restore_completed_turn_authority(
completed_turn_authoritative
&& recovered_active_turn_id.is_none()
&& !ambiguous_turn_start_pending,
completed_turn_process_generation,
completed_provider_turn_id.as_deref(),
);
self.provider = Some(provider);
if provider_had_exited || recovered_active_turn_id != previous_active_turn_id {
self.state
.as_mut()
.expect("Codex state remains available during recovery")
.provider_process_generation = process_generation;
if provider_had_exited
|| ambiguous_turn_start_pending
|| recovered_active_turn_id != previous_active_turn_id
{
let state = self
.state
.as_mut()
.expect("Codex state remains available during recovery");
state.active_provider_turn_id = recovered_active_turn_id.clone();
state.lifecycle = if recovered_active_turn_id.is_some() {
"turn_active".to_owned()
} else {
"session_open".to_owned()
};
state.reconcile_active_provider_turn(recovered_active_turn_id.clone());
state.push_event(NormalizedProviderEvent {
event_type: "session.reconciled".to_owned(),
priority: EventPriority::P0,
@ -380,8 +465,8 @@ impl CodexCommandExecutor {
"activeProviderTurnId": recovered_active_turn_id,
}),
})?;
self.save_state()?;
}
self.save_state()?;
Ok(())
}
@ -490,11 +575,29 @@ impl CodexCommandExecutor {
"Codex provider session is closed",
));
}
let provider = CodexProvider::start(&state.config, state.thread_id.as_deref())
.map_err(|error| {
DurableRunnerError::invalid(format!("failed to start Codex provider: {error}"))
})?;
let process_generation = state
.provider_process_generation
.checked_add(1)
.ok_or_else(|| DurableRunnerError::invalid("Codex process generation exhausted"))?;
let mut provider = CodexProvider::start_for_generation(
&state.config,
state.thread_id.as_deref(),
process_generation,
)
.map_err(|error| {
DurableRunnerError::invalid(format!("failed to start Codex provider: {error}"))
})?;
provider.restore_completed_turn_authority(
state.completed_turn_authoritative && provider.active_provider_turn_id().is_none(),
state.completed_turn_process_generation,
state.completed_provider_turn_id.as_deref(),
);
self.provider = Some(provider);
self.state
.as_mut()
.expect("Codex state remains available after provider start")
.provider_process_generation = process_generation;
self.save_state()?;
}
self.provider
.as_mut()
@ -586,11 +689,46 @@ impl CodexCommandExecutor {
.config
.cwd
.clone();
let (provider_turn_id, thread_id) = {
self.ensure_provider()?;
{
let state = self
.state
.as_mut()
.expect("Codex state remains available before turn/start dispatch");
state.ambiguous_turn_start_pending = true;
}
self.save_state()?;
let (start_result, completion_authority_retained, ambiguous_turn_start_pending) = {
let provider = self.ensure_provider()?;
provider.start_turn(text, &cwd).map_err(|error| {
DurableRunnerError::invalid(format!("Codex turn/start failed: {error}"))
})?;
let result = provider.start_turn(text, &cwd);
(
result,
provider.completed_turn_authority().is_some(),
provider.ambiguous_turn_start_pending(),
)
};
if let Err(error) = start_result {
let state = self
.state
.as_mut()
.expect("Codex state remains available after turn/start failure");
state.ambiguous_turn_start_pending = ambiguous_turn_start_pending;
if !completion_authority_retained && !ambiguous_turn_start_pending {
state.completed_turn_authoritative = false;
state.completed_turn_process_generation = None;
state.completed_provider_turn_id = None;
state.last_agent_message = None;
}
self.save_state()?;
return Err(DurableRunnerError::invalid(format!(
"Codex turn/start failed: {error}"
)));
}
let (provider_turn_id, thread_id) = {
let provider = self
.provider
.as_ref()
.expect("Codex provider remains available after turn/start acceptance");
(
provider
.active_provider_turn_id()
@ -606,6 +744,10 @@ impl CodexCommandExecutor {
.as_mut()
.expect("Codex state exists after turn start");
state.active_provider_turn_id = Some(provider_turn_id.clone());
state.ambiguous_turn_start_pending = false;
state.completed_turn_authoritative = false;
state.completed_turn_process_generation = None;
state.completed_provider_turn_id = None;
state.last_agent_message = None;
state.lifecycle = "turn_active".to_owned();
self.save_state()?;
@ -687,6 +829,7 @@ impl CodexCommandExecutor {
.as_mut()
.ok_or_else(|| DurableRunnerError::invalid("Codex provider is not prepared"))?;
state.active_provider_turn_id = None;
state.ambiguous_turn_start_pending = false;
state.lifecycle = "closed".to_owned();
let thread_id = state.thread_id.clone();
self.save_state()?;
@ -717,6 +860,11 @@ impl CodexCommandExecutor {
fn poll_provider(&mut self) -> Result<(), DurableRunnerError> {
self.restore()?;
// `restore_checked` records that the durable file was loaded even when
// provider recovery failed. Retry the provider reconciliation here so
// an ambiguous-start failure cannot degrade into an empty successful
// poll on the same executor.
self.restore_provider_if_needed()?;
if self
.state
.as_ref()
@ -738,7 +886,32 @@ impl CodexCommandExecutor {
})?;
let Some(event) = event else { break };
match event {
CodexProviderEvent::ToolCall {
call_id,
operation_id,
..
} => {
return Err(DurableRunnerError::invalid(format!(
"Codex emitted semantic tool call {call_id} for {operation_id} before the durable tool bridge was attached"
)));
}
CodexProviderEvent::Notification { method, params } => {
let active_provider_turn_id = if method == "turn/started" {
self.provider
.as_ref()
.and_then(CodexProvider::active_provider_turn_id)
.map(str::to_owned)
} else {
None
};
let completed_turn_authority = if method == "turn/completed" {
self.provider
.as_ref()
.and_then(CodexProvider::completed_turn_authority)
.map(|(generation, turn_id)| (generation, turn_id.to_owned()))
} else {
None
};
let normalized = normalize_codex_notification(&method, &params);
let terminal_event_type = normalized
.iter()
@ -767,8 +940,26 @@ impl CodexCommandExecutor {
.map(|text| text.chars().take(1_000_000).collect());
}
}
if method == "turn/started" {
let provider_turn_id = active_provider_turn_id.ok_or_else(|| {
DurableRunnerError::invalid(
"Codex turn start notification omitted active turn authority",
)
})?;
state.reconcile_active_provider_turn(Some(provider_turn_id));
}
if method == "turn/completed" {
let (process_generation, provider_turn_id) = completed_turn_authority
.ok_or_else(|| {
DurableRunnerError::invalid(
"Codex completion omitted process and turn authority",
)
})?;
state.active_provider_turn_id = None;
state.completed_turn_authoritative = true;
state.completed_turn_process_generation = Some(process_generation);
state.completed_provider_turn_id = Some(provider_turn_id);
state.ambiguous_turn_start_pending = false;
state.lifecycle = "session_open".to_owned();
}
state.extend_events(normalized)?;
@ -811,18 +1002,49 @@ impl CodexCommandExecutor {
})?;
self.save_state()?;
}
CodexProviderEvent::Exited { exit_code, success } => {
CodexProviderEvent::Exited {
exit_code,
success,
completed_turn_authoritative,
completed_turn_observed_by_process,
completion_reconciles_exit,
process_generation,
completed_turn_process_generation,
} => {
self.provider = None;
if let Some(state) = self.state.as_mut() {
if !success {
let state = self
.state
.as_mut()
.expect("Codex state remains available while polling");
// The durable terminal remains the run outcome. Use the
// provider's generation correlation only to decide
// whether this separate session exit belongs to that
// completion or is a later idle-provider failure.
state.lifecycle = "provider_exited".to_owned();
state.push_event(NormalizedProviderEvent {
event_type: "session.failed".to_owned(),
// A completed turn remains authoritative, while the
// reusable provider session independently becomes
// unavailable. Avoid emitting session.failed for
// already successful work, but never leave the
// durable lifecycle open after a nonzero exit.
event_type: if completion_reconciles_exit {
"session.reconciled"
} else {
"session.failed"
}
.to_owned(),
priority: EventPriority::P0,
payload: json!({
"provider": "codex",
"code": "provider_exited",
"exitCode": exit_code,
"expected": success,
"previousTurnCompleted": completed_turn_authoritative,
"completedByExitedProcess": completed_turn_observed_by_process,
"processGeneration": process_generation,
"completedTurnProcessGeneration": completed_turn_process_generation,
"activeProviderTurnId": Value::Null,
}),
})?;
}
@ -941,6 +1163,11 @@ mod tests {
thread_id: Some("thread-1".to_owned()),
provider_session_id: None,
active_provider_turn_id: None,
ambiguous_turn_start_pending: false,
completed_turn_authoritative: false,
provider_process_generation: 0,
completed_turn_process_generation: None,
completed_provider_turn_id: None,
last_agent_message: None,
pending_events: VecDeque::new(),
next_provider_event_seq: initial_provider_event_seq(),
@ -948,6 +1175,45 @@ mod tests {
assert!(state.validate().is_err());
}
#[test]
fn recovered_active_turn_revokes_prior_completion_authority() {
let mut state = CodexProviderState::new(
CodexProviderConfig {
provider: "codex".to_owned(),
driver: "codex_app_server".to_owned(),
provider_version: "test".to_owned(),
command: PathBuf::from("codex"),
args: vec!["app-server".to_owned()],
cwd: std::env::current_dir()
.unwrap()
.to_string_lossy()
.into_owned(),
model: None,
provider_session_id: None,
instructions: String::new(),
approval_policy: "never".to_owned(),
},
None,
);
state.thread_id = Some("thread-1".to_owned());
state.lifecycle = "session_open".to_owned();
state.completed_turn_authoritative = true;
state.provider_process_generation = 1;
state.completed_turn_process_generation = Some(1);
state.completed_provider_turn_id = Some("turn-1".to_owned());
state.last_agent_message = Some("old turn output".to_owned());
state.reconcile_active_provider_turn(Some("turn-2".to_owned()));
assert_eq!(state.lifecycle, "turn_active");
assert_eq!(state.active_provider_turn_id.as_deref(), Some("turn-2"));
assert!(!state.completed_turn_authoritative);
assert!(state.completed_turn_process_generation.is_none());
assert!(state.completed_provider_turn_id.is_none());
assert!(state.last_agent_message.is_none());
assert!(state.validate().is_ok());
}
#[test]
fn emits_a_structured_result_before_the_terminal_event() {
let mut state = CodexProviderState::new(

1368
packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
View File

File diff suppressed because it is too large Load Diff