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

feat(runner): durably reconcile Codex tools (#12384) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The Paperclip Runner gives an agent a durable execution boundary.
> - The Codex transport can now advertise a run-scoped semantic tool
catalog.
> - The durable backend did not yet persist tool calls or correlate
their results.
> - A restart could therefore lose the boundary between a provider call
and a Paperclip action.
> - This pull request binds authorized calls, durable events, results,
replay, and cancellation.
> - The benefit is safe semantic tool recovery without duplicate
Paperclip actions.

## Linked Issues or Issue Description

Refs #12382

**What existing behavior does this improve?**

This improves the durable Codex provider backend in
`@paperclipai/paperclip-runner`.

**Current behavior**

The Codex transport can project authorized dynamic tools. The durable
backend rejects their calls because it cannot persist and recover their
identities.

**Proposed behavior**

The durable backend records each authorized call before it emits the
semantic input event. It records each result before it sends the result
to Codex. It reconciles exact provider replays without another Paperclip
action.

**Reason and benefit**

This order prevents duplicate semantic actions after a process restart.
It also keeps unauthorized, changed, oversized, and late calls closed.

**Breaking changes**

None. A run without an authorized tool catalog still starts Codex with
no dynamic tools.

## What Changed

- Persist the authorized tool catalog with the Codex provider state.
- Emit correlated and redacted semantic input, reconciliation, and
result events.
- Reconcile exact pending and completed calls after a provider restart.
- Reject catalog drift, changed replay input, malformed results, and
unauthorized operations.
- Complete pending tool calls with a durable failure when a turn stops.
- Bound retained tool values and validate recovered state before
provider startup.
- Bind production runner events to the active run, session, turn, and
item identities.

## Verification

- `cargo fmt --all -- --check`
- `cargo test --workspace`
- `pnpm -r typecheck`
- `pnpm build`
- Confirmed that the PR changes 9 files against
`runner-codex-dynamic-tools`.
- Confirmed that dependency installation did not change
`pnpm-lock.yaml`.

## Risks

The main risk is a mismatch between recovered provider state and the
controller tool catalog. Recovery validates the complete catalog and its
digest before Codex starts. The backend persists a call before it emits
work and persists a result before it returns the result to Codex.

This PR does not enable the server adapter or change any direct adapter
path.

## 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 11:31:36 -05:00 committed by GitHub
parent 9e4be0e60c
commit 6c85fa060e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 7645 additions and 590 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -120,6 +120,66 @@ fn emit_ambiguous_turn_evidence(
}
}
fn emit_ambiguous_turn_item() -> io::Result<()> {
send(json!({
"method": "item/completed",
"params": {"item": {
"id": "replacement-message-before-terminal",
"type": "agentMessage",
"status": "completed",
"text": "Replacement output before terminal authority."
}}
}))
}
fn send_question(state: &FakeState) -> io::Result<()> {
let turn_id = state.active_turn_id.as_deref().unwrap_or("provider-turn-1");
send(json!({
"id": "runtime-request-1",
"method": "item/tool/requestUserInput",
"params": {
"threadId": state.thread_id,
"turnId": turn_id,
"itemId": "question-item-1",
"isBlocking": true,
"title": "Deployment input",
"questions": [{
"id": "environment",
"header": "Environment",
"question": "Where should we deploy?",
"options": [
{"label": "Staging", "description": "Deploy safely."},
{"label": "Production", "description": "Deploy directly."}
]
}]
}
}))
}
fn send_runtime_request_flood(state: &FakeState, interrupt_count: u64) -> io::Result<()> {
let turn_id = state.active_turn_id.as_deref().unwrap_or("provider-turn-1");
for index in 0..160_u64 {
send(json!({
"id": format!("runtime-flood-{interrupt_count}-{index}"),
"method": "item/tool/requestUserInput",
"params": {
"threadId": state.thread_id,
"turnId": turn_id,
"itemId": format!("question-item-{interrupt_count}-{index}"),
"isBlocking": true,
"title": "Bounded cleanup input",
"questions": [{
"id": "environment",
"header": "Environment",
"question": "Where should we deploy?",
"options": [{"label": "Staging", "description": "Deploy safely."}],
}],
},
}))?;
}
Ok(())
}
fn run() -> Result<(), Box<dyn std::error::Error>> {
let args = std::env::args().skip(1).collect::<Vec<_>>();
let state_path =
@ -136,6 +196,19 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
let exit_after_tool_call_completion = args
.iter()
.any(|value| value == "--exit-after-tool-call-completion");
let emit_tool_call_on_resume = args
.iter()
.any(|value| value == "--emit-tool-call-on-resume");
let resume_unowned_turn_when_marked = args
.iter()
.any(|value| value == "--resume-unowned-turn-when-marked");
let replay_completed_tool_call_count = argument(&args, "--replay-completed-tool-call-count")
.map(|value| value.parse::<u64>())
.transpose()?
.unwrap_or_default();
let finish_turn_with_pending_tool = args
.iter()
.any(|value| value == "--finish-turn-with-pending-tool");
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");
@ -163,6 +236,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
let missing_id_second_turn_start = args
.iter()
.any(|value| value == "--missing-id-second-turn-start");
let missing_id_live_turn_start = args
.iter()
.any(|value| value == "--missing-id-live-turn-start");
let fail_after_accepting_second_turn_before_response = args
.iter()
.any(|value| value == "--fail-after-accepting-second-turn-before-response");
@ -184,32 +260,93 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
let conflicting_ambiguous_second_turn = args
.iter()
.any(|value| value == "--conflicting-ambiguous-second-turn");
let ambiguous_older_reused_turn = args
.iter()
.any(|value| value == "--ambiguous-older-reused-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 fail_first_interrupt = args.iter().any(|value| value == "--fail-first-interrupt");
let accept_interrupt_without_terminal_once = args
.iter()
.any(|value| value == "--accept-interrupt-without-terminal-once");
let accept_interrupt_without_terminal = args
.iter()
.any(|value| value == "--accept-interrupt-without-terminal");
let flood_runtime_requests_on_interrupt = args
.iter()
.any(|value| value == "--flood-runtime-requests-on-interrupt");
let interrupt_terminal_delay_ms = argument(&args, "--interrupt-terminal-delay-ms")
.map(|value| value.parse::<u64>())
.transpose()?;
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 delayed_tool_after_failed_turn = args
.iter()
.any(|value| value == "--delayed-tool-after-failed-turn");
let delayed_tool_after_next_turn_start = args
.iter()
.any(|value| value == "--delayed-tool-after-next-turn-start");
let delayed_tool_after_third_turn_start = args
.iter()
.any(|value| value == "--delayed-tool-after-third-turn-start");
let delayed_tool_after_second_turn_completion = args
.iter()
.any(|value| value == "--delayed-tool-after-second-turn-completion");
let tool_after_reused_turn_start = args
.iter()
.any(|value| value == "--tool-after-reused-turn-start");
let tool_after_older_reused_turn_start = args
.iter()
.any(|value| value == "--tool-after-older-reused-turn-start");
let question_before_failed_turn = args
.iter()
.any(|value| value == "--question-before-failed-turn");
let reuse_question_id = args.iter().any(|value| value == "--reuse-question-id");
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;
let mut interrupt_count = 0_u64;
let mut delayed_interrupt_terminal_scheduled = false;
let mut answered_questions = 0u8;
let mut replayed_completed_tool_calls = 0_u64;
for line in io::stdin().lock().lines() {
let message: Value = serde_json::from_str(&line?)?;
if message.get("method").is_none()
&& message
.get("id")
.and_then(Value::as_str)
.is_some_and(|id| id.starts_with("runtime-flood-"))
{
let outcome = if message.get("error").is_some() {
"runtime-response:rejected"
} else {
"runtime-response:cancelled"
};
log_call(call_log.as_deref(), outcome)?;
continue;
}
if message.get("method").is_none() && message.get("id") == Some(&json!("runtime-request-1"))
{
if reuse_question_id && answered_questions == 0 {
answered_questions = 1;
send_question(&state)?;
continue;
}
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() {
if state.active_turn_id.is_some() && !hold_turn {
finish_turn(&state_path, &mut state, "failed")?;
}
continue;
@ -225,7 +362,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
if result != json!({"ok": true, "task": {"id": "task-1"}}) {
return Err("semantic tool response changed the operation result".into());
}
if replay_completed_tool_call {
log_call(call_log.as_deref(), &format!("tool-response:{text}"))?;
if replay_completed_tool_call && replayed_completed_tool_calls == 0 {
replayed_completed_tool_calls += 1;
send(json!({
"id": "tool-request-replay",
"method": "item/tool/call",
@ -237,9 +376,22 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
"arguments": {}
}
}))?;
continue;
} else if replayed_completed_tool_calls < replay_completed_tool_call_count {
replayed_completed_tool_calls += 1;
send(json!({
"id": "tool-request-1",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": state.active_turn_id,
"callId": "semantic-call-1",
"tool": "get_task_context",
"arguments": {}
}
}))?;
} else if !hold_turn {
finish_turn(&state_path, &mut state, "completed")?;
}
finish_turn(&state_path, &mut state, "completed")?;
continue;
}
let Some(method) = message.get("method").and_then(Value::as_str) else {
@ -275,10 +427,31 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
if require_dynamic_tool && !has_task_context_tool(&message) {
return Err("thread/resume omitted the authorized dynamic tool".into());
}
let unowned_turn_marker = state_path.with_file_name("resume-unowned-turn");
if resume_unowned_turn_when_marked && unowned_turn_marker.exists() {
state.active_turn_id = Some("provider-turn-unowned".to_owned());
save_state(&state_path, &state)?;
fs::remove_file(unowned_turn_marker)?;
}
send(json!({
"id": id,
"result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}}
}))?;
if emit_tool_call_on_resume {
if let Some(turn_id) = state.active_turn_id.as_deref() {
send(json!({
"id": "tool-request-1",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": turn_id,
"callId": "semantic-call-1",
"tool": "get_task_context",
"arguments": {}
}
}))?;
}
}
}
"thread/read" => {
let turns = state
@ -319,17 +492,31 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
&& (complete_ambiguous_second_turn
|| complete_ambiguous_second_turn_before_response
|| conflicting_ambiguous_second_turn);
let provider_turn_id = if emits_ambiguous_turn_evidence
let provider_turn_id = if turn_start_count == 3 && ambiguous_older_reused_turn {
"provider-turn-1".to_owned()
} else if emits_ambiguous_turn_evidence
|| (turn_start_count == 2
&& (retain_ambiguous_second_turn_active
|| hold_ambiguous_second_turn_after_item))
{
"provider-turn-2"
"provider-turn-2".to_owned()
} else if tool_after_reused_turn_start
|| (tool_after_older_reused_turn_start && turn_start_count == 3)
{
"provider-turn-1".to_owned()
} else {
"provider-turn-1"
format!("provider-turn-{turn_start_count}")
};
state.active_turn_id = Some(provider_turn_id.to_owned());
state.active_turn_id = Some(provider_turn_id.clone());
save_state(&state_path, &state)?;
if ambiguous_older_reused_turn && turn_start_count == 3 {
send(json!({
"method": "turn/started",
"params": {"turn": {"id": provider_turn_id}}
}))?;
send(json!({"id": id, "error": {}}))?;
continue;
}
if complete_ambiguous_second_turn_before_response && turn_start_count == 2 {
emit_ambiguous_turn_evidence(
&state_path,
@ -355,6 +542,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
return Ok(());
}
if malformed_error_second_turn_start && turn_start_count == 2 {
if hold_ambiguous_second_turn_after_item {
emit_ambiguous_turn_item()?;
}
send(json!({"id": id, "error": {}}))?;
if emits_ambiguous_turn_evidence
&& !complete_ambiguous_second_turn_before_response
@ -374,15 +564,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
"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."
}}
}))?;
emit_ambiguous_turn_item()?;
continue;
}
if emits_ambiguous_turn_evidence
@ -397,16 +579,103 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
}
return Err("configured failure after missing turn identity".into());
}
if missing_id_live_turn_start {
send(json!({
"id": id,
"result": {"turn": {"status": "inProgress"}}
}))?;
continue;
}
send(json!({
"id": id,
"result": {"turn": {"id": "provider-turn-1", "status": "inProgress"}}
"result": {"turn": {"id": provider_turn_id, "status": "inProgress"}}
}))?;
send(json!({
"method": "turn/started",
"params": {"turn": {"id": "provider-turn-1"}}
"params": {"turn": {"id": provider_turn_id}}
}))?;
if fail_after_second_turn_start && turn_start_count == 2 {
return Err("configured failure after second turn start".into());
} else if question_before_failed_turn {
send_question(&state)?;
send(json!({
"method": "turn/completed",
"params": {
"threadId": state.thread_id,
"turn": {"id": provider_turn_id, "status": "failed"}
}
}))?;
state.active_turn_id = None;
save_state(&state_path, &state)?;
} else if delayed_tool_after_failed_turn {
send(json!({
"method": "turn/failed",
"params": {
"threadId": state.thread_id,
"turn": {"id": provider_turn_id, "status": "failed"}
}
}))?;
state.active_turn_id = None;
save_state(&state_path, &state)?;
send(json!({
"id": "tool-request-delayed",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": provider_turn_id,
"callId": "semantic-call-delayed",
"tool": "get_task_context",
"arguments": {}
}
}))?;
} else if delayed_tool_after_next_turn_start && turn_start_count == 2 {
send(json!({
"id": "tool-request-delayed",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"callId": "semantic-call-delayed",
"tool": "get_task_context",
"arguments": {}
}
}))?;
} else if delayed_tool_after_third_turn_start && turn_start_count == 3 {
send(json!({
"id": "tool-request-two-turns-delayed",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"callId": "semantic-call-two-turns-delayed",
"tool": "get_task_context",
"arguments": {}
}
}))?;
} else if tool_after_reused_turn_start && turn_start_count == 2 {
send(json!({
"id": "tool-request-reused-turn",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"callId": "semantic-call-reused-turn",
"tool": "get_task_context",
"arguments": {}
}
}))?;
} else if tool_after_older_reused_turn_start && turn_start_count == 3 {
send(json!({
"id": "tool-request-older-reused-turn",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"callId": "semantic-call-older-reused-turn",
"tool": "get_task_context",
"arguments": {}
}
}))?;
} else if exit_after_turn_start {
return Ok(());
} else if emit_tool_call {
@ -415,41 +684,35 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"turnId": provider_turn_id,
"callId": "semantic-call-1",
"tool": "get_task_context",
"arguments": {}
}
}))?;
if complete_after_tool_call {
if complete_after_tool_call || finish_turn_with_pending_tool {
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",
"method": "item/tool/requestUserInput",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"itemId": "question-item-1",
"isBlocking": true,
"title": "Deployment input",
"questions": [{
"id": "environment",
"header": "Environment",
"question": "Where should we deploy?",
"options": [
{"label": "Staging", "description": "Deploy safely."},
{"label": "Production", "description": "Deploy directly."}
]
}]
}
}))?;
send_question(&state)?;
} else if !hold_turn {
finish_turn(&state_path, &mut state, "completed")?;
if delayed_tool_after_second_turn_completion && turn_start_count == 2 {
send(json!({
"id": "tool-request-idle-two-turns-delayed",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"callId": "semantic-call-idle-two-turns-delayed",
"tool": "get_task_context",
"arguments": {}
}
}))?;
}
if emit_post_completion_warning {
send(json!({
"method": "warning",
@ -477,8 +740,44 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
}
"turn/steer" => send(json!({"id": id, "result": {"accepted": true}}))?,
"turn/interrupt" => {
send(json!({"id": id, "result": {"accepted": true}}))?;
finish_turn(&state_path, &mut state, "interrupted")?;
interrupt_count += 1;
if fail_first_interrupt && interrupt_count == 1 {
send(json!({
"id": id,
"error": {"code": -32001, "message": "configured interrupt failure"}
}))?;
} else {
send(json!({"id": id, "result": {"accepted": true}}))?;
if flood_runtime_requests_on_interrupt {
send_runtime_request_flood(&state, interrupt_count)?;
}
if !accept_interrupt_without_terminal
&& !(accept_interrupt_without_terminal_once
&& interrupt_count == if fail_first_interrupt { 2 } else { 1 })
{
if let Some(delay_ms) = interrupt_terminal_delay_ms {
if !delayed_interrupt_terminal_scheduled {
delayed_interrupt_terminal_scheduled = true;
let delayed_state_path = state_path.clone();
let mut delayed_state = state.clone();
thread::spawn(move || {
thread::sleep(Duration::from_millis(delay_ms));
if let Err(error) = finish_turn(
&delayed_state_path,
&mut delayed_state,
"interrupted",
) {
eprintln!(
"failed to emit delayed interrupt terminal: {error}"
);
}
});
}
} else {
finish_turn(&state_path, &mut state, "interrupted")?;
}
}
}
}
_ if id.is_some() => send(json!({
"id": id,

42
packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs
View File

@ -50,28 +50,26 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> {
optional_u64(args, name).map(|value| Duration::from_millis(value.unwrap_or(default)))
};
let state_dir = PathBuf::from(value(args, "--state-dir")?);
run_durable_runner(
DurableRunnerConfig {
connect_url: value(args, "--connect-url")?,
state_dir: state_dir.clone(),
runner_instance_id: value(args, "--runner-id")?,
environment_lease_id: value(args, "--environment-lease-id")?,
run_id: value(args, "--run-id")?,
normalized_session_id: value(args, "--session-id")?,
turn_id: value(args, "--turn-id")?,
item_id: value(args, "--item-id")?,
runner_version: value(args, "--runner-version")?,
runner_digest: value(args, "--runner-digest")?,
max_outbox_bytes: usize_value(args, "--max-outbox-bytes", 16 * 1024 * 1024)?,
p0_reserve_bytes: usize_value(args, "--p0-reserve-bytes", 1024 * 1024)?,
max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?,
reconnect_delay: duration("--reconnect-delay-ms", 250)?,
max_runtime: duration("--max-runtime-ms", 60 * 60 * 1000)?,
},
ticket,
CodexCommandExecutor::new(state_dir),
)
.map_err(|error| LocalRunnerError::invalid(error.to_string()))
let config = DurableRunnerConfig {
connect_url: value(args, "--connect-url")?,
state_dir: state_dir.clone(),
runner_instance_id: value(args, "--runner-id")?,
environment_lease_id: value(args, "--environment-lease-id")?,
run_id: value(args, "--run-id")?,
normalized_session_id: value(args, "--session-id")?,
turn_id: value(args, "--turn-id")?,
item_id: value(args, "--item-id")?,
runner_version: value(args, "--runner-version")?,
runner_digest: value(args, "--runner-digest")?,
max_outbox_bytes: usize_value(args, "--max-outbox-bytes", 16 * 1024 * 1024)?,
p0_reserve_bytes: usize_value(args, "--p0-reserve-bytes", 1024 * 1024)?,
max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?,
reconnect_delay: duration("--reconnect-delay-ms", 250)?,
max_runtime: duration("--max-runtime-ms", 60 * 60 * 1000)?,
};
let executor = CodexCommandExecutor::with_runner_config(state_dir, &config);
run_durable_runner(config, ticket, executor)
.map_err(|error| LocalRunnerError::invalid(error.to_string()))
}
fn run() -> Result<(), LocalRunnerError> {

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

File diff suppressed because it is too large Load Diff

4
packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs
View File

@ -9,12 +9,14 @@ use std::time::Duration;
pub use runner::{run_durable_runner, CommandExecution, CommandExecutor, PolledEvent};
pub(crate) use state::{
create_private_temporary_file, open_private_regular_file, redact_text, verify_private_directory,
create_private_temporary_file, open_private_regular_file, redact_text, sanitize_value,
verify_private_directory,
};
pub use state::{
Command, CommandDisposition, DurableState, DurableStateStore, EventPriority,
StoredCommandResult, StoredOutboxEvent,
};
pub(crate) use transport::current_unix_ms;
pub const PROTOCOL: &str = "paperclip.runner";
pub const PROTOCOL_VERSION: u64 = 1;

37
packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs
View File

@ -1078,7 +1078,23 @@ fn sensitive_key(key: &str) -> bool {
.any(|needle| normalized.contains(needle))
}
fn sanitize_value(value: &Value) -> Value {
fn protocol_authorization_boundary(key: &str, value: &Value) -> bool {
key.eq_ignore_ascii_case("authorizationBoundary")
&& value.as_str().is_some_and(|boundary| {
matches!(
boundary,
"company"
| "actor"
| "active_task"
| "grant"
| "governed_action"
| "lock"
| "revision"
)
})
}
pub(crate) fn sanitize_value(value: &Value) -> Value {
match value {
Value::Object(object) => Value::Object(
object
@ -1086,7 +1102,9 @@ fn sanitize_value(value: &Value) -> Value {
.map(|(key, value)| {
(
key.clone(),
if sensitive_key(key) {
if protocol_authorization_boundary(key, value) {
value.clone()
} else if sensitive_key(key) {
Value::String("[REDACTED]".to_owned())
} else {
sanitize_value(value)
@ -1323,6 +1341,21 @@ mod tests {
);
}
#[test]
fn protocol_authorization_boundary_is_not_redacted_as_a_credential() {
let sanitized = sanitize_value(&json!({
"authorizationBoundary": "active_task",
"nested": {"authorizationBoundary": "Bearer secret-value"},
"authorization": "Bearer secret-value",
}));
assert_eq!(sanitized["authorizationBoundary"], json!("active_task"));
assert_eq!(sanitized["authorization"], json!("[REDACTED]"));
assert_eq!(
sanitized["nested"]["authorizationBoundary"],
json!("[REDACTED]")
);
}
#[test]
fn outbox_reserves_capacity_for_p0_and_bounds_frames() {
let mut bounds_config = config(PathBuf::from("unused"));

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

File diff suppressed because it is too large Load Diff

1557
packages/paperclip-runner/runner/crates/runner-core/src/provider_bridge.rs
View File

File diff suppressed because it is too large Load Diff

56
packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs
View File

@ -12,6 +12,29 @@ pub struct NormalizedProviderEvent {
pub payload: Value,
}
pub(crate) fn normalized_codex_terminal_event_type(
method: &str,
params: &Value,
) -> Option<&'static str> {
let status = match method {
"turn/failed" => "failed",
"turn/cancelled" => "cancelled",
"turn/interrupted" => "interrupted",
"turn/completed" => string(
params
.pointer("/turn/status")
.or_else(|| params.get("status")),
),
_ => return None,
};
Some(match status {
"failed" | "error" => "turn.failed",
"cancelled" | "canceled" => "turn.cancelled",
"interrupted" | "aborted" => "turn.interrupted",
_ => "turn.completed",
})
}
fn bounded_text(value: &str, max_chars: usize) -> String {
redact_text(value).chars().take(max_chars).collect()
}
@ -118,18 +141,19 @@ pub fn normalize_codex_notification(method: &str, params: &Value) -> Vec<Normali
"providerTurnId": params.pointer("/turn/id").or_else(|| params.get("turnId")).and_then(Value::as_str),
}),
),
"turn/completed" => {
let status = string(
params
.pointer("/turn/status")
.or_else(|| params.get("status")),
);
let event_type = match status {
"failed" | "error" => "turn.failed",
"cancelled" | "canceled" => "turn.cancelled",
"interrupted" | "aborted" => "turn.interrupted",
_ => "turn.completed",
"turn/completed" | "turn/failed" | "turn/cancelled" | "turn/interrupted" => {
let status = match method {
"turn/failed" => "failed",
"turn/cancelled" => "cancelled",
"turn/interrupted" => "interrupted",
_ => string(
params
.pointer("/turn/status")
.or_else(|| params.get("status")),
),
};
let event_type = normalized_codex_terminal_event_type(method, params)
.expect("matched Codex terminal method has a normalized terminal type");
push(
&mut events,
event_type,
@ -366,6 +390,16 @@ mod tests {
);
assert_eq!(terminal[0].event_type, "turn.failed");
assert_eq!(terminal[0].priority, EventPriority::P0);
for (method, expected) in [
("turn/failed", "turn.failed"),
("turn/cancelled", "turn.cancelled"),
("turn/interrupted", "turn.interrupted"),
] {
let terminal =
normalize_codex_notification(method, &json!({"turnId": "provider-turn"}));
assert_eq!(terminal[0].event_type, expected);
assert_eq!(terminal[0].priority, EventPriority::P0);
}
let usage = normalize_codex_notification(
"thread/tokenUsage/updated",

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

File diff suppressed because it is too large Load Diff

540
packages/paperclip-runner/runner/crates/runner-core/tests/provider_bridge.rs
View File

@ -73,6 +73,13 @@ fn rejects_unknown_tools_and_conflicting_duplicate_results() {
};
bridge.apply_result(result.clone()).unwrap();
bridge.apply_result(result).unwrap();
assert!(bridge
.replay_result("call-1", "get_task_context", &json!({}))
.unwrap()
.is_some());
assert!(bridge
.replay_result("call-1", "get_task_context", &json!({"changed": true}))
.is_err());
assert!(bridge
.apply_result(ToolResult {
call_id: "call-1".to_owned(),
@ -108,6 +115,29 @@ fn catalog_digest_matches_the_typescript_canonical_json_contract() {
);
}
#[test]
fn catalog_digest_normalizes_json_numbers_like_javascript() {
let operation = AuthorizedTool {
operation_id: "get_task_context".into(),
version: 1,
description: "Read the active task context.".into(),
input_schema: json!({
"type": "object",
"properties": {
"limit": { "type": "number", "default": 1.0 },
"epsilon": { "type": "number", "default": 1e-6 },
},
}),
response_schema: json!({ "type": "object" }),
};
let digest = authorized_tool_catalog_digest(&[operation]).unwrap();
assert_eq!(
digest,
"sha256:1c93693d9b5b48b46c83cd1c11d1ea329774f1b9b0ae741197cb2b8e992c4b8d"
);
}
#[test]
fn validates_the_operation_value_inside_a_semantic_dispatch_envelope() {
let mut set = tools("sha256:catalog-a");
@ -120,7 +150,7 @@ fn validates_the_operation_value_inside_a_semantic_dispatch_envelope() {
let mut bridge = ProviderToolBridge::default();
set.catalog_digest = digest('a');
set.catalog_digest = authorized_tool_catalog_digest(&set.operations).unwrap();
bridge.prepare(set).unwrap();
bridge.prepare(set.clone()).unwrap();
bridge
.begin_call("call-1".into(), "get_task_context".into(), json!({}))
.unwrap();
@ -139,6 +169,25 @@ fn validates_the_operation_value_inside_a_semantic_dispatch_envelope() {
})
.unwrap();
assert_eq!(bridge.pending_calls().count(), 0);
let mut second = ProviderToolBridge::default();
second.prepare(set).unwrap();
second
.begin_call("call-2".into(), "get_task_context".into(), json!({}))
.unwrap();
second
.apply_result(ToolResult {
call_id: "call-2".into(),
operation_id: "get_task_context".into(),
result: json!({
"ok": true,
"operationId": "get_task_context",
"callId": "call-2",
"value": { "value": "accepted" }
}),
is_error: false,
})
.unwrap();
}
#[test]
@ -159,6 +208,24 @@ fn rejects_noncanonical_digests_and_oversized_contract_values() {
json!({ "value": "x".repeat(1024 * 1024) }),
)
.is_err());
let retained = json!({"value": "x".repeat(700 * 1024)});
for index in 0..5 {
bridge
.begin_call(
format!("call-{index}"),
"get_task_context".into(),
retained.clone(),
)
.unwrap();
}
assert!(bridge
.begin_call(
"call-over-aggregate-limit".into(),
"get_task_context".into(),
retained,
)
.is_err());
}
#[test]
@ -203,12 +270,214 @@ fn recovery_preserves_completed_call_replay_identities() {
let encoded = serde_json::to_string(&bridge).unwrap();
let mut recovered: ProviderToolBridge = serde_json::from_str(&encoded).unwrap();
recovered.validate_recovered().unwrap();
recovered.attach_existing_run().unwrap();
assert!(recovered
.begin_call("call-1".into(), "get_task_context".into(), json!({}))
.is_err());
}
#[test]
fn recovered_bridge_rejects_tampered_authorization_state() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
let mut encoded = serde_json::to_value(&bridge).unwrap();
encoded["authorized"]["get_task_context"]["description"] = json!("Tampered");
let recovered: ProviderToolBridge = serde_json::from_value(encoded).unwrap();
assert!(recovered.validate_recovered().is_err());
}
#[test]
fn cancellation_completes_pending_calls_and_rejects_late_results() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
bridge
.begin_call("call-1".into(), "get_task_context".into(), json!({}))
.unwrap();
let cancelled = bridge
.cancel_pending_calls("provider_turn_stopped")
.unwrap();
assert_eq!(cancelled.len(), 1);
assert!(cancelled[0].is_error);
assert_eq!(bridge.pending_calls().count(), 0);
assert!(bridge
.apply_result(ToolResult {
call_id: "call-1".into(),
operation_id: "get_task_context".into(),
result: json!({"ok": true}),
is_error: false,
})
.is_err());
}
#[test]
fn turn_settlement_releases_value_capacity_without_reusing_call_ids() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
bridge
.begin_call("call-1".into(), "get_task_context".into(), json!({}))
.unwrap();
bridge
.apply_result(ToolResult {
call_id: "call-1".into(),
operation_id: "get_task_context".into(),
result: json!({"ok": true}),
is_error: false,
})
.unwrap();
assert!(bridge
.settle_turn("provider_turn_terminated")
.unwrap()
.is_empty());
assert!(bridge
.begin_call("call-1".into(), "get_task_context".into(), json!({}))
.is_err());
bridge
.begin_call("call-2".into(), "get_task_context".into(), json!({}))
.expect("a new turn can use a fresh call id after releasing exact values");
}
#[test]
fn completed_receipts_are_exact_until_the_controlled_turn_limit() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
for index in 0..4_096 {
let call_id = format!("call-{index}");
bridge
.begin_call(call_id.clone(), "get_task_context".into(), json!({}))
.expect("a completed call must release concurrent capacity");
bridge
.apply_result(ToolResult {
call_id,
operation_id: "get_task_context".into(),
result: json!({"ok": true}),
is_error: false,
})
.unwrap();
}
let error = bridge
.begin_call(
"call-after-limit".into(),
"get_task_context".into(),
json!({}),
)
.expect_err("the bounded exact receipt ledger must stop the active turn");
assert!(error.is_active_turn_receipt_limit());
assert!(bridge
.replay_result("call-0", "get_task_context", &json!({}))
.unwrap()
.is_some());
assert!(bridge
.replay_result("call-4095", "get_task_context", &json!({}))
.unwrap()
.is_some());
let recovered: ProviderToolBridge =
serde_json::from_str(&serde_json::to_string(&bridge).unwrap()).unwrap();
recovered.validate_recovered().unwrap();
}
#[test]
fn turn_settlement_cannot_be_blocked_by_completed_value_pressure() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
let large = json!({"value": "x".repeat(700 * 1024)});
for index in 0..5 {
bridge
.begin_call(
format!("call-{index}"),
"get_task_context".into(),
large.clone(),
)
.unwrap();
}
for index in 0..4 {
bridge
.apply_result(ToolResult {
call_id: format!("call-{index}"),
operation_id: "get_task_context".into(),
result: large.clone(),
is_error: false,
})
.unwrap();
}
let settled = bridge.settle_turn("provider_turn_terminated").unwrap();
assert_eq!(settled.len(), 1);
assert!(settled[0].is_error);
assert!(bridge
.begin_call("call-0".into(), "get_task_context".into(), json!({}))
.is_err());
bridge
.begin_call(
"call-after-settlement".into(),
"get_task_context".into(),
json!({}),
)
.expect("settlement releases prior turn value retention");
}
#[test]
fn recovered_turn_preserves_exact_results_at_the_value_boundary() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
let large = json!({"value": "x".repeat(700 * 1024)});
for index in 0..5 {
bridge
.begin_call(
format!("call-{index}"),
"get_task_context".into(),
large.clone(),
)
.unwrap();
}
for index in 0..5 {
bridge
.apply_result(ToolResult {
call_id: format!("call-{index}"),
operation_id: "get_task_context".into(),
result: large.clone(),
is_error: false,
})
.unwrap();
}
let error = bridge
.begin_call("call-next".into(), "get_task_context".into(), large.clone())
.expect_err("exact replay values must not be discarded for later work");
assert!(error.is_active_turn_receipt_limit());
assert_eq!(
bridge
.replay_result("call-0", "get_task_context", &large)
.unwrap()
.unwrap()
.result,
large,
);
bridge
.apply_result(ToolResult {
call_id: "call-0".into(),
operation_id: "get_task_context".into(),
result: large.clone(),
is_error: false,
})
.expect("a matching exact result receipt remains idempotent");
assert!(bridge
.apply_result(ToolResult {
call_id: "call-0".into(),
operation_id: "get_task_context".into(),
result: json!({"value": "changed"}),
is_error: false,
})
.is_err());
let recovered: ProviderToolBridge =
serde_json::from_str(&serde_json::to_string(&bridge).unwrap()).unwrap();
recovered.validate_recovered().unwrap();
}
#[test]
fn recovery_preserves_pending_calls_for_the_existing_run() {
let mut bridge = ProviderToolBridge::default();
@ -252,7 +521,7 @@ fn recovery_rejects_nonempty_state_without_a_catalog_digest() {
let error = recovered
.attach_existing_run()
.expect_err("nonempty recovered state must remain bound to a catalog digest");
assert!(error.to_string().contains("omit the catalog digest"));
assert!(error.to_string().contains("omitted its catalog identity"));
}
#[test]
@ -269,8 +538,8 @@ fn recovery_rejects_tampered_authorization_catalog_bindings() {
let mut recovered: ProviderToolBridge = serde_json::from_value(changed_contract).unwrap();
let error = recovered
.attach_existing_run()
.expect_err("recovery must recompute the catalog digest");
assert!(error.to_string().contains("catalog digest"));
.expect_err("recovery must reconstruct the authorized catalog projection");
assert!(error.to_string().contains("changed its authorized catalog"));
let mut changed_map_key = encoded;
let authorized = changed_map_key["authorized"].as_object_mut().unwrap();
@ -280,7 +549,7 @@ fn recovery_rejects_tampered_authorization_catalog_bindings() {
let error = recovered
.attach_existing_run()
.expect_err("recovery must bind map keys to declared operation identities");
assert!(error.to_string().contains("identities are inconsistent"));
assert!(error.to_string().contains("changed its authorized catalog"));
}
#[test]
@ -326,18 +595,21 @@ fn recovery_rejects_tampered_retained_result_contracts() {
let completed = serde_json::to_value(&bridge).unwrap();
let mut unauthorized = completed.clone();
unauthorized["completed"]["call-1"]["operationId"] = json!("delete_company");
let mut recovered: ProviderToolBridge = serde_json::from_value(unauthorized).unwrap();
assert!(recovered.attach_existing_run().is_err());
unauthorized["completed"]["call-1"]["result"]["operationId"] = json!("delete_company");
let error = serde_json::from_value::<ProviderToolBridge>(unauthorized)
.expect_err("durable decoding must reject mismatched call and result identities");
assert!(error
.to_string()
.contains("retained provider tool receipt identity is inconsistent"));
let mut invalid_output = completed;
invalid_output["completed"]["call-1"]["result"] = json!(["not", "an", "object"]);
invalid_output["completed"]["call-1"]["result"]["result"] = json!(["not", "an", "object"]);
let mut recovered: ProviderToolBridge = serde_json::from_value(invalid_output).unwrap();
assert!(recovered.attach_existing_run().is_err());
bridge.settle_turn().unwrap();
bridge.settle_turn("provider_turn_terminated").unwrap();
let mut invalid_settled_output = serde_json::to_value(&bridge).unwrap();
invalid_settled_output["settledResults"]["call-1"]["result"] = json!("invalid");
invalid_settled_output["settledResults"]["call-1"]["result"]["result"] = json!("invalid");
let mut recovered: ProviderToolBridge = serde_json::from_value(invalid_settled_output).unwrap();
assert!(recovered.attach_existing_run().is_err());
}
@ -389,7 +661,7 @@ fn settles_completed_receipts_before_the_next_turn() {
assert!(bridge
.begin_call("call-next".into(), "get_task_context".into(), json!({}))
.is_err());
bridge.settle_turn().unwrap();
bridge.settle_turn("provider_turn_terminated").unwrap();
assert!(bridge
.begin_call("call-next".into(), "get_task_context".into(), json!({}))
.is_ok());
@ -410,7 +682,7 @@ fn settlement_preserves_call_ids_for_the_durable_run() {
is_error: false,
})
.unwrap();
bridge.settle_turn().unwrap();
bridge.settle_turn("provider_turn_terminated").unwrap();
let replay = ToolResult {
call_id: "call-1".into(),
@ -456,7 +728,7 @@ fn settlement_preserves_call_ids_for_the_durable_run() {
}
#[test]
fn reserves_identity_capacity_before_accepting_a_call() {
fn exact_identity_overflow_saturates_the_durable_run() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
@ -472,27 +744,147 @@ fn reserves_identity_capacity_before_accepting_a_call() {
bridge
.begin_call("last-call".into(), "get_task_context".into(), json!({}))
.unwrap();
bridge
.apply_result(ToolResult {
call_id: "last-call".into(),
operation_id: "get_task_context".into(),
result: json!({"ok": true}),
is_error: false,
})
.unwrap();
bridge.settle_turn().unwrap();
assert!(bridge
let overflow = bridge
.begin_call("overflow".into(), "get_task_context".into(), json!({}))
.expect_err("the pending call reserves the final exact identity slot");
assert!(overflow.is_active_turn_receipt_limit());
let encoded = serde_json::to_string(&bridge).unwrap();
let mut recovered: ProviderToolBridge = serde_json::from_str(&encoded).unwrap();
recovered.attach_existing_run().unwrap();
assert!(recovered.durable_run_receipt_limit_reached());
recovered
.settle_turn("provider_turn_terminated")
.expect("the controlled turn stop retains replay protection");
assert!(recovered.durable_run_receipt_limit_reached());
assert!(recovered.has_completed_call("settled-0"));
assert!(recovered.has_completed_call("last-call"));
let stopped_turn_receipt = recovered
.replay_result("last-call", "get_task_context", &json!({}))
.unwrap()
.expect("the call admitted before exhaustion retains an exact terminal receipt");
assert!(stopped_turn_receipt.is_error);
assert_eq!(
stopped_turn_receipt.result["error"]["code"],
"provider_turn_terminated"
);
assert!(recovered
.begin_call("settled-0".into(), "get_task_context".into(), json!({}))
.is_err());
assert!(recovered
.begin_call("overflow".into(), "get_task_context".into(), json!({}))
.is_err());
assert!(bridge.settle_turn().is_ok());
recovered.prepare_turn().unwrap();
assert!(recovered
.replay_result("last-call", "get_task_context", &json!({}))
.unwrap()
.is_none());
assert!(recovered.has_completed_call("last-call"));
assert!(recovered.has_completed_call("settled-0"));
assert!(recovered.durable_run_receipt_limit_reached());
let saturation = recovered
.begin_call("next-call".into(), "get_task_context".into(), json!({}))
.expect_err("a turn boundary must not reopen the saturated durable run");
assert!(saturation.is_active_turn_receipt_limit());
let encoded = serde_json::to_string(&recovered).unwrap();
let mut recovered: ProviderToolBridge = serde_json::from_str(&encoded).unwrap();
recovered.attach_existing_run().unwrap();
assert!(recovered.has_completed_call("settled-0"));
assert!(recovered.has_completed_call("last-call"));
assert!(recovered.durable_run_receipt_limit_reached());
recovered.attach_run(tools("computed")).unwrap();
assert!(!recovered.durable_run_receipt_limit_reached());
recovered
.begin_call("overflow".into(), "get_task_context".into(), json!({}))
.expect("a new durable run receives a fresh tool-call identity ledger");
}
#[test]
fn settled_result_byte_exhaustion_recovers_after_turn_cleanup() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
let large_result = json!({"value": "x".repeat(750 * 1024)});
for index in 0..10 {
let call_id = format!("large-settled-{index}");
bridge
.begin_call(call_id.clone(), "get_task_context".into(), json!({}))
.unwrap();
bridge
.apply_result(ToolResult {
call_id,
operation_id: "get_task_context".into(),
result: large_result.clone(),
is_error: false,
})
.unwrap();
}
bridge.settle_turn("provider_turn_terminated").unwrap();
let error = bridge
.begin_call(
"over-byte-limit".into(),
"get_task_context".into(),
json!({}),
)
.expect_err("settled byte exhaustion must stop the durable run");
assert!(error.is_active_turn_receipt_limit());
let encoded = serde_json::to_string(&bridge).unwrap();
let mut recovered: ProviderToolBridge = serde_json::from_str(&encoded).unwrap();
recovered.attach_existing_run().unwrap();
assert!(recovered.durable_run_receipt_limit_reached());
assert_eq!(
recovered
.replay_result("large-settled-0", "get_task_context", &json!({}))
.unwrap()
.unwrap()
.result,
large_result
);
assert!(recovered
.begin_call(
"over-byte-limit".into(),
"get_task_context".into(),
json!({})
)
.is_err());
recovered.prepare_turn().unwrap();
assert!(recovered
.replay_result("large-settled-0", "get_task_context", &json!({}))
.unwrap()
.is_none());
assert!(recovered.has_completed_call("large-settled-0"));
assert!(!recovered.durable_run_receipt_limit_reached());
recovered
.begin_call(
"after-turn-boundary".into(),
"get_task_context".into(),
json!({}),
)
.expect("releasing bulky results clears transient byte pressure");
recovered
.settle_turn("provider_turn_terminated")
.expect("the admitted next-turn call remains settleable");
recovered.attach_run(tools("computed")).unwrap();
recovered
.begin_call(
"over-byte-limit".into(),
"get_task_context".into(),
json!({}),
)
.expect("a new durable run resets the settled result budget");
}
#[test]
fn reserves_settled_result_bytes_before_accepting_a_call() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
let large_result = json!({"value": "x".repeat(900 * 1024)});
let large_result = json!({"value": "x".repeat(700 * 1024)});
let mut completed = 0;
for index in 0..20 {
@ -523,7 +915,7 @@ fn reserves_settled_result_bytes_before_accepting_a_call() {
)
.is_err());
bridge
.settle_turn()
.settle_turn("provider_turn_terminated")
.expect("settlement cannot strand results whose bytes were reserved at admission");
assert_eq!(
bridge
@ -534,7 +926,7 @@ fn reserves_settled_result_bytes_before_accepting_a_call() {
is_error: false,
})
.unwrap(),
json!({"value": "x".repeat(900 * 1024)})
json!({"value": "x".repeat(700 * 1024)})
);
}
@ -544,15 +936,22 @@ fn recovery_rejects_an_oversized_settled_result_envelope() {
bridge.prepare(tools("computed")).unwrap();
let mut encoded = serde_json::to_value(&bridge).unwrap();
let settled = encoded["settledResults"].as_object_mut().unwrap();
for index in 0..10 {
for index in 0..12 {
let call_id = format!("recovered-large-{index}");
settled.insert(
call_id.clone(),
json!({
"callId": call_id,
"operationId": "get_task_context",
"result": {"value": "x".repeat(900 * 1024)},
"isError": false
"call": {
"callId": call_id,
"operationId": "get_task_context",
"input": {}
},
"result": {
"callId": call_id,
"operationId": "get_task_context",
"result": {"value": "x".repeat(700 * 1024)},
"isError": false
}
}),
);
}
@ -568,18 +967,28 @@ fn recovery_rejects_state_without_room_for_a_pending_result() {
bridge.prepare(tools("computed")).unwrap();
let mut encoded = serde_json::to_value(&bridge).unwrap();
let settled = encoded["settledResults"].as_object_mut().unwrap();
for index in 0..8 {
let mut settled_ids = Vec::new();
for index in 0..11 {
let call_id = format!("recovered-large-{index}");
settled_ids.push(json!(call_id));
settled.insert(
call_id.clone(),
json!({
"callId": call_id,
"operationId": "get_task_context",
"result": {"value": "x".repeat(900 * 1024)},
"isError": false
"call": {
"callId": call_id,
"operationId": "get_task_context",
"input": {}
},
"result": {
"callId": call_id,
"operationId": "get_task_context",
"result": {"value": "x".repeat(700 * 1024)},
"isError": false
}
}),
);
}
encoded["settledCallIds"] = serde_json::Value::Array(settled_ids);
encoded["pending"]["pending-call"] = json!({
"callId": "pending-call",
"operationId": "get_task_context",
@ -594,13 +1003,62 @@ fn recovery_rejects_state_without_room_for_a_pending_result() {
}
#[test]
fn refuses_to_settle_receipts_while_calls_are_pending() {
fn settles_pending_receipts_with_explicit_terminal_results() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
bridge
.begin_call("call-1".into(), "get_task_context".into(), json!({}))
.unwrap();
assert!(bridge.settle_turn().is_err());
assert_eq!(bridge.pending_calls().count(), 1);
let settled = bridge.settle_turn("provider_turn_terminated").unwrap();
assert_eq!(settled.len(), 1);
assert_eq!(settled[0].call_id, "call-1");
assert!(settled[0].is_error);
assert_eq!(bridge.pending_calls().count(), 0);
}
#[test]
fn full_identity_ledger_fails_closed_across_turn_and_recovery() {
let mut bridge = ProviderToolBridge::default();
bridge.prepare(tools("computed")).unwrap();
let mut encoded = serde_json::to_value(&bridge).unwrap();
encoded["settledCallIds"] = serde_json::Value::Array(
(0..65_536)
.map(|index| json!(format!("settled-{index:05}")))
.collect(),
);
let mut recovered: ProviderToolBridge = serde_json::from_value(encoded).unwrap();
recovered.validate_recovered().unwrap();
let saturation = recovered
.begin_call("current-call".into(), "get_task_context".into(), json!({}))
.expect_err("a full exact ledger must stop fresh work");
assert!(saturation.is_active_turn_receipt_limit());
assert!(recovered.durable_run_receipt_limit_reached());
assert!(recovered.has_completed_call("settled-65535"));
assert!(recovered.has_completed_call("settled-00000"));
assert!(!recovered.has_completed_call("current-call"));
assert!(recovered
.begin_call("settled-00000".into(), "get_task_context".into(), json!({}))
.is_err());
recovered.prepare_turn().unwrap();
let fresh = recovered
.begin_call("fresh-call".into(), "get_task_context".into(), json!({}))
.expect_err("a turn boundary must preserve durable-run saturation");
assert!(fresh.is_active_turn_receipt_limit());
let round_trip = serde_json::to_value(&recovered).unwrap();
assert_eq!(
round_trip["settledCallIds"].as_array().unwrap().len(),
65_536
);
let mut recovered_again: ProviderToolBridge = serde_json::from_value(round_trip).unwrap();
recovered_again.attach_existing_run().unwrap();
assert!(recovered_again.durable_run_receipt_limit_reached());
assert!(recovered_again.has_completed_call("settled-00000"));
recovered_again.attach_run(tools("computed")).unwrap();
assert!(!recovered_again.durable_run_receipt_limit_reached());
recovered_again
.begin_call("current-call".into(), "get_task_context".into(), json!({}))
.expect("only a new durable run resets replay authority");
}