Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-10-26 18:31:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(runner): resume native Codex tool calls exactly once

Browse Source 
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-26 19:39:46 +00:00
1 parent 381b763846
commit eb191dddc9
21 files changed
+42646 -91
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/db/src/migrations/0228_polite_black_knight.sql
+18
View File
@@ -0,0 +1,18 @@
CREATE TABLE "native_semantic_receipts" (
"id" uuid PRIMARY KEY NOT NULL,
"company_id" uuid NOT NULL,
"issue_id" uuid NOT NULL,
"run_id" uuid NOT NULL,
"operation_id" text NOT NULL,
"scope_digest" text NOT NULL,
"input_digest" text NOT NULL,
"status" text DEFAULT 'claimed' NOT NULL,
"outcome" jsonb,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"updated_at" timestamp with time zone DEFAULT now() NOT NULL
);
--> statement-breakpoint
ALTER TABLE "native_semantic_receipts" ADD CONSTRAINT "native_semantic_receipts_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "native_semantic_receipts" ADD CONSTRAINT "native_semantic_receipts_issue_company_fk" FOREIGN KEY ("company_id","issue_id") REFERENCES "public"."issues"("company_id","id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "native_semantic_receipts" ADD CONSTRAINT "native_semantic_receipts_run_owner_fk" FOREIGN KEY ("company_id","issue_id","run_id") REFERENCES "public"."heartbeat_runs"("company_id","native_issue_id","id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
CREATE UNIQUE INDEX "native_semantic_receipts_run_scope_uq" ON "native_semantic_receipts" USING btree ("run_id","scope_digest");
packages/db/src/migrations/meta/0228_snapshot.json
Generated
+41419
View File
File diff suppressed because it is too large. Load diff
packages/db/src/migrations/meta/_journal.json
Generated
+7
View File
@@ -1583,6 +1583,13 @@
"when": 1787668790923,
"tag": "0227_modern_pandemic",
"breakpoints": true
},
{
"idx": 228,
"version": "7",
"when": 1787773030696,
"tag": "0228_polite_black_knight",
"breakpoints": true
}
]
}
packages/db/src/schema/index.ts
+1
View File
@@ -114,6 +114,7 @@ export { heartbeatRunEvents } from "./heartbeat_run_events.js";
export { completionContracts } from "./completion_contracts.js";
export { nativeRunResults } from "./native_run_results.js";
export { nativeRunFinalizations } from "./native_run_finalizations.js";
export { nativeSemanticReceipts } from "./native_semantic_receipts.js";
export { workAssessments } from "./work_assessments.js";
export { statusDecisions } from "./status_decisions.js";
export { statusDecisionEffects } from "./status_decision_effects.js";
packages/db/src/schema/native_semantic_receipts.ts
+50
View File
@@ -0,0 +1,50 @@
import {
foreignKey,
jsonb,
pgTable,
text,
timestamp,
uniqueIndex,
uuid,
} from "drizzle-orm/pg-core";
import { companies } from "./companies.js";
import { heartbeatRuns } from "./heartbeat_runs.js";
import { issues } from "./issues.js";
export const nativeSemanticReceipts = pgTable(
"native_semantic_receipts",
{
id: uuid("id").primaryKey(),
companyId: uuid("company_id").notNull().references(() => companies.id),
issueId: uuid("issue_id").notNull(),
runId: uuid("run_id").notNull(),
operationId: text("operation_id").notNull(),
scopeDigest: text("scope_digest").notNull(),
inputDigest: text("input_digest").notNull(),
status: text("status").notNull().default("claimed"),
outcome: jsonb("outcome").$type<Record<string, unknown>>(),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
},
(table) => ({
issueCompanyFk: foreignKey({
columns: [table.companyId, table.issueId],
foreignColumns: [issues.companyId, issues.id],
name: "native_semantic_receipts_issue_company_fk",
}),
runOwnerFk: foreignKey({
columns: [table.companyId, table.issueId, table.runId],
foreignColumns: [
heartbeatRuns.companyId,
heartbeatRuns.nativeIssueId,
heartbeatRuns.id,
],
name: "native_semantic_receipts_run_owner_fk",
}),
runScopeUq: uniqueIndex("native_semantic_receipts_run_scope_uq").on(
table.runId,
table.scopeDigest,
),
}),
);
packages/paperclip-runner/protocol/manifest.json
+1 -1
View File
@@ -30,7 +30,7 @@
{
"path": "schemas/event.schema.json",
"id": "https://paperclip.dev/schemas/prp/v1/event.schema.json",
"sha256": "d69ac4acfd9cd265f9dcb6a6580e949a67a50347093d0a7c12c2000a2da6d482"
"sha256": "abc19ded0556ac785d54a1bfb8bf3e6a16dce0985ad232a8489f26966ccd35c3"
},
{
"path": "schemas/fixture.schema.json",
packages/paperclip-runner/protocol/schemas/event.schema.json
+1 -1
View File
@@ -43,7 +43,7 @@
"session.starting", "session.started", "session.resuming", "session.resumed", "session.reconciled", "session.updated", "session.closed", "session.failed",
"turn.submitted", "turn.accepted", "turn.started", "turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled",
"item.started", "item.delta", "item.completed", "item.failed", "usage.reported",
"semantic_tool.input", "semantic_tool.result",
"semantic_tool.input", "semantic_tool.result", "semantic_tool.reconciled",
"mcp_app.discovered", "mcp_app.resource.resolved", "mcp_app.initializing", "mcp_app.ready", "mcp_app.tool_input", "mcp_app.tool_result", "mcp_app.action.requested", "mcp_app.action.resolved", "mcp_app.host_context.changed", "mcp_app.failed", "mcp_app.teardown",
"runtime_request.created", "runtime_request.resolved", "runtime_request.expired", "runtime_request.cancelled",
"interaction.request.proposed", "interaction.request.materialized", "interaction.request.rejected", "interaction.response.progressed", "interaction.response.resolved", "interaction.response.delivered",
packages/paperclip-runner/runner/CODEX_PROVIDER.md
+17 -6
View File
@@ -14,6 +14,9 @@ execution and finalization path.
- The server also binds the immutable completion-contract revision and criterion
identifiers. A completed Codex turn emits one `run.result.proposed` followed
by one `run.terminal`; server finalization accepts only that bound pair.
- `run.prepare` also binds the coordinator's run-scoped semantic-tool
projection. Runnerd supplies the same immutable inventory to both
`thread/start` and `thread/resume`.
- `session.open` initializes Codex and starts a thread. A recovered runner
resumes the recorded thread and reads it before accepting another turn.
- `turn.start` requires bounded non-empty `payload.text`. `turn.steer`,
@@ -37,9 +40,15 @@ later turn cannot diverge from Codex's native state.
PRP journals every command before the provider effect. Exact command replay
returns the durable result without invoking Codex again. A crash in the effect
window remains indeterminate and is not retried. Codex JSON-RPC notifications
received before a synchronous response are buffered rather than lost. Reusing
a pending structured-input request ID with different content fails closed.
window remains indeterminate and is not retried. Semantic tool calls use a
narrower native recovery contract: runnerd persists the Codex `callId`, tool,
and input before emitting `semantic_tool.input`. If Codex reissues that pending
request after `thread/resume`, runnerd emits `semantic_tool.reconciled` instead
of a second input event and delivers the control plane's durable result to the
same call ID. Reuse with different content fails closed. Codex JSON-RPC
notifications received before a synchronous response are buffered rather than
lost. Reusing a pending structured-input request ID with different content
also fails closed.
Normalized events remain in the provider sidecar until the durable PRP outbox
has committed and acknowledged them. The server persists the native binding
before process launch and reuses it after restart, including when the rollout
@@ -51,6 +60,8 @@ Codex-native envelopes do not cross PRP. The provider backend emits bounded,
redacted session, turn, item, plan, usage, tool-execution, notice, and structured
input events. Unknown notifications are ignored.
Codex starts with an empty dynamic-tool inventory. Catalog presence is not
authorization. The coordinator projects only the already landed, same-task
read bindings for the exact company and native run.
Catalog presence is not authorization. The coordinator projects only bindings
authorized for the exact company, task, actor, and native run. Read operations
are run-bound. `report_progress` additionally uses a durable semantic receipt
and a deterministic comment identity, so an interrupted or repeated dispatch
can produce only one task comment.
packages/paperclip-runner/runner/DURABLE_TRANSPORT.md
+12 -4
View File
@@ -53,6 +53,13 @@ the process dies inside the effect window, recovery records an indeterminate
result and refuses to execute that command again. Recent results are bounded;
commands older than the compacted controller cursor fail closed.
Pending Codex semantic requests are provider events, not replayed control
commands. Their native call identity is persisted in the provider sidecar. A
restarted runner resumes the same Codex thread, recognizes the reissued call,
and records reconciliation without producing another semantic input. The
server's deterministic result command and durable semantic receipt then finish
the original request and make the application effect idempotent.
State written before complete-command fingerprints existed is migrated by
compacting its legacy command journal through the last recorded controller
sequence. The runner can recover, but it rejects redelivery of those older
@@ -69,7 +76,8 @@ Durable mode is selected only when `paperclip-runnerd` receives
`--connect-url`. Its executor accepts a Codex app-server descriptor and bound
completion contract through `run.prepare`, owns the provider process group,
resumes the persisted Codex thread after runner restart, and translates
provider notifications to PRP events. The server supplies the bootstrap ticket
only through the child environment, stores the process identity for bounded
cancellation, and waits for the durable result and terminal pair. The existing
local fake-runner mode remains unchanged.
provider notifications to PRP events. The server supplies each one-use
bootstrap ticket only through the child environment, restarts an unexpectedly
exited runner up to a fixed bound against the same state directory, stores each
process identity for bounded cancellation, and waits for the durable result
and terminal pair. The existing local fake-runner mode remains unchanged.
packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs
+55 -4
View File
@@ -11,6 +11,8 @@ use serde_json::{json, Value};
struct FakeState {
thread_id: String,
active_turn_id: Option<String>,
#[serde(default)]
pending_semantic_call: bool,
}
fn argument(args: &[String], name: &str) -> Option<String> {
@@ -34,6 +36,7 @@ fn load_state(path: &Path) -> FakeState {
.unwrap_or_else(|| FakeState {
thread_id: "codex-thread-1".to_owned(),
active_turn_id: None,
pending_semantic_call: false,
})
}
@@ -79,12 +82,31 @@ fn finish_turn(state_path: &Path, state: &mut FakeState, status: &str) -> io::Re
save_state(state_path, state)
}
fn send_semantic_call(state: &FakeState) -> io::Result<()> {
send(json!({
"id": "semantic-rpc-1",
"method": "item/tool/call",
"params": {
"threadId": state.thread_id,
"turnId": "provider-turn-1",
"itemId": "semantic-item-1",
"callId": "semantic-call-1",
"tool": "report_progress",
"arguments": {
"idempotencyKey": "native-resume-proof-1",
"body": "Native resume completed one semantic effect."
},
}
}))
}
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_semantic_tool = args.iter().any(|value| value == "--emit-semantic-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 pre_response_notification = args
@@ -94,6 +116,13 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
for line in io::stdin().lock().lines() {
let message: Value = serde_json::from_str(&line?)?;
if message.get("method").is_none() && message.get("id") == Some(&json!("semantic-rpc-1")) {
log_call(call_log.as_deref(), "semantic_tool/result")?;
state.pending_semantic_call = false;
save_state(&state_path, &state)?;
finish_turn(&state_path, &mut state, "completed")?;
continue;
}
if message.get("method").is_none() && message.get("id") == Some(&json!("runtime-request-1"))
{
finish_turn(&state_path, &mut state, "completed")?;
@@ -111,8 +140,21 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
}))?,
"initialized" => {}
"thread/start" => {
if emit_semantic_tool
&& message
.pointer("/params/dynamicTools")
.and_then(Value::as_array)
.is_none_or(|tools| {
!tools.iter().any(|tool| {
tool.get("name").and_then(Value::as_str) == Some("report_progress")
})
})
{
return Err("semantic test requires the projected dynamic tool".into());
}
state.thread_id = "codex-thread-1".to_owned();
state.active_turn_id = None;
state.pending_semantic_call = false;
save_state(&state_path, &state)?;
if pre_response_notification {
send(json!({
@@ -125,10 +167,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" => {
send(json!({
"id": id,
"result": {"thread": {"id": state.thread_id, "sessionId": "codex-account-session"}}
}))?;
if state.pending_semantic_call {
send_semantic_call(&state)?;
}
}
"thread/read" => {
let turns = state
.active_turn_id
@@ -174,6 +221,10 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
}]
}
}))?;
} else if emit_semantic_tool {
state.pending_semantic_call = true;
save_state(&state_path, &state)?;
send_semantic_call(&state)?;
} else if !hold_turn {
finish_turn(&state_path, &mut state, "completed")?;
}
packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs
+141 -5
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;
@@ -39,6 +39,34 @@ pub struct CodexProviderConfig {
pub approval_policy: String,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct CodexDynamicTool {
pub name: String,
pub description: String,
pub input_schema: Value,
}
impl CodexDynamicTool {
pub fn validate(&self) -> Result<(), LocalRunnerError> {
if self.name.is_empty()
|| self.name.len() > 160
|| self
.name
.chars()
.any(|character| !(character.is_ascii_alphanumeric() || "_-".contains(character)))
|| self.description.is_empty()
|| self.description.len() > 4_000
|| !self.input_schema.is_object()
{
return Err(LocalRunnerError::invalid(
"Codex dynamic tool definition is malformed or oversized",
));
}
Ok(())
}
}
impl CodexProviderConfig {
pub fn validate(&self) -> Result<(), LocalRunnerError> {
if self.provider != "codex" || self.driver != "codex_app_server" {
@@ -109,6 +137,11 @@ pub enum CodexProviderEvent {
request_id: String,
question_set: Value,
},
ToolCall {
call_id: String,
operation_id: String,
input: Value,
},
Exited {
exit_code: Option<i32>,
success: bool,
@@ -124,6 +157,13 @@ struct PendingRuntimeRequest {
option_labels: QuestionOptionLabels,
}
#[derive(Clone, Debug, PartialEq)]
struct PendingToolCall {
rpc_id: Value,
operation_id: String,
input: Value,
}
pub struct CodexProvider {
process: SupervisedProcess,
next_request_id: u64,
@@ -132,6 +172,8 @@ pub struct CodexProvider {
active_provider_turn_id: Option<String>,
pending_messages: VecDeque<Value>,
pending_runtime_requests: BTreeMap<String, PendingRuntimeRequest>,
pending_tool_calls: BTreeMap<String, PendingToolCall>,
authorized_tool_names: BTreeSet<String>,
expected_shutdown: bool,
}
@@ -139,8 +181,17 @@ impl CodexProvider {
pub fn start(
config: &CodexProviderConfig,
resume_thread_id: Option<&str>,
dynamic_tools: &[CodexDynamicTool],
) -> Result<Self, LocalRunnerError> {
config.validate()?;
if dynamic_tools.len() > 128 {
return Err(LocalRunnerError::invalid(
"Codex dynamic tool inventory exceeds the 128 tool limit",
));
}
for tool in dynamic_tools {
tool.validate()?;
}
let mut provider = Self {
process: SupervisedProcess::spawn(
&config.command,
@@ -154,6 +205,8 @@ impl CodexProvider {
active_provider_turn_id: None,
pending_messages: VecDeque::new(),
pending_runtime_requests: BTreeMap::new(),
pending_tool_calls: BTreeMap::new(),
authorized_tool_names: dynamic_tools.iter().map(|tool| tool.name.clone()).collect(),
expected_shutdown: false,
};
let initialized = provider.request(
@@ -183,14 +236,12 @@ impl CodexProvider {
let params_object = params
.as_object_mut()
.expect("Codex thread parameters are an object");
params_object.insert("dynamicTools".to_owned(), json!(dynamic_tools));
params_object.insert("experimentalRawEvents".to_owned(), json!(false));
let method = if let Some(thread_id) = resume_thread_id {
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"
};
let opened = provider.request(method, params)?;
@@ -326,6 +377,30 @@ impl CodexProvider {
Ok(())
}
pub fn deliver_tool_result(
&mut self,
call_id: &str,
result: &Value,
is_error: bool,
) -> Result<(), LocalRunnerError> {
let pending = self
.pending_tool_calls
.get(call_id)
.cloned()
.ok_or_else(|| {
LocalRunnerError::invalid("semantic result has no pending Codex tool call")
})?;
self.process.send(&json!({
"id": pending.rpc_id,
"result": {
"success": !is_error,
"contentItems": [{"type": "inputText", "text": result.to_string()}],
},
}))?;
self.pending_tool_calls.remove(call_id);
Ok(())
}
pub fn poll(&mut self) -> Result<Option<CodexProviderEvent>, LocalRunnerError> {
let message = if let Some(message) = self.pending_messages.pop_front() {
message
@@ -347,6 +422,66 @@ impl CodexProvider {
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 = params
.get("callId")
.and_then(Value::as_str)
.filter(|value| !value.is_empty() && value.len() <= 160)
.ok_or_else(|| LocalRunnerError::invalid("Codex tool call id is invalid"))?
.to_owned();
let operation_id = params
.get("tool")
.and_then(Value::as_str)
.filter(|value| !value.is_empty() && value.len() <= 160)
.ok_or_else(|| LocalRunnerError::invalid("Codex tool name is invalid"))?
.to_owned();
if !self.authorized_tool_names.contains(&operation_id) {
return Err(LocalRunnerError::invalid(
"Codex requested a tool outside its authorized inventory",
));
}
let input = params
.get("arguments")
.filter(|value| value.is_object())
.cloned()
.ok_or_else(|| {
LocalRunnerError::invalid("Codex tool arguments must be an object")
})?;
let pending = PendingToolCall {
rpc_id,
operation_id: operation_id.clone(),
input: input.clone(),
};
if let Some(existing) = self.pending_tool_calls.get(&call_id) {
if existing.operation_id != pending.operation_id
|| existing.input != pending.input
{
return Err(LocalRunnerError::invalid(
"Codex reused a tool call id with different input",
));
}
}
self.pending_tool_calls.insert(call_id.clone(), pending);
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()) {
@@ -420,6 +555,7 @@ impl CodexProvider {
pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> {
self.expected_shutdown = true;
self.cancel_pending_runtime_requests()?;
self.pending_tool_calls.clear();
self.process.terminate_group().map(|_| ())
}
packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs
+78 -1
View File
@@ -3,6 +3,7 @@ use std::time::Instant;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use super::state::{
Command, CommandDisposition, DurableState, DurableStateStore, EventPriority,
@@ -306,7 +307,10 @@ fn poll_executor_events<E: CommandExecutor>(
if events.is_empty() {
return Ok(());
}
for event in events {
for mut event in events {
if event.event_type == "semantic_tool.input" {
event.payload = semantic_tool_input_payload(state, event.payload)?;
}
// Commit and acknowledge one event at a time. If a later event is
// oversized or the outbox is full, the accepted prefix is already
// durable and the unacknowledged suffix remains with the executor.
@@ -332,6 +336,79 @@ fn poll_executor_events<E: CommandExecutor>(
Ok(())
}
fn semantic_tool_input_payload(
state: &DurableState,
payload: Value,
) -> Result<Value, DurableRunnerError> {
let call_id = payload
.get("callId")
.and_then(Value::as_str)
.ok_or_else(|| DurableRunnerError::invalid("semantic tool input omitted callId"))?;
let operation_id = payload
.get("operationId")
.and_then(Value::as_str)
.ok_or_else(|| DurableRunnerError::invalid("semantic tool input omitted operationId"))?;
let input = payload
.get("input")
.cloned()
.ok_or_else(|| DurableRunnerError::invalid("semantic tool input omitted input"))?;
let digest = format!(
"sha256:{:x}",
Sha256::digest(canonical_json(&input).as_bytes())
);
Ok(json!({
"semantic_tool": {
"schema": "paperclip.prp.semantic_tool.v1",
"schemaVersion": 1,
"phase": "input",
"operationId": operation_id,
"callId": call_id,
"correlation": {
"runId": state.run_id,
"normalizedSessionId": state.normalized_session_id,
"turnId": state.turn_id,
"itemId": state.item_id,
},
"idempotencyKey": input.get("idempotencyKey").and_then(Value::as_str),
"input": input,
"content": {
"digest": digest,
"redactionDisposition": "digest_only",
"references": [],
},
},
}))
}
fn canonical_json(value: &Value) -> String {
match value {
Value::Array(values) => format!(
"[{}]",
values
.iter()
.map(canonical_json)
.collect::<Vec<_>>()
.join(",")
),
Value::Object(values) => {
let mut keys = values.keys().collect::<Vec<_>>();
keys.sort();
format!(
"{{{}}}",
keys.into_iter()
.map(|key| format!(
"{}:{}",
serde_json::to_string(key).expect("JSON object key should serialize"),
canonical_json(&values[key])
))
.collect::<Vec<_>>()
.join(",")
)
}
_ => value.to_string(),
}
}
fn process_command<E: CommandExecutor>(
state: &mut DurableState,
store: &DurableStateStore,
packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
+245 -11
View File
@@ -1,4 +1,4 @@
use std::collections::{HashSet, VecDeque};
use std::collections::{BTreeMap, HashSet, VecDeque};
use std::fs;
use std::io::{Read, Write};
use std::path::PathBuf;
@@ -11,7 +11,9 @@ use std::os::unix::fs::PermissionsExt;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use crate::codex_provider::{CodexProvider, CodexProviderConfig, CodexProviderEvent};
use crate::codex_provider::{
CodexDynamicTool, CodexProvider, CodexProviderConfig, CodexProviderEvent,
};
use crate::durable::{
create_private_temporary_file, open_private_regular_file, verify_private_directory, Command,
CommandExecution, CommandExecutor, DurableRunnerError, EventPriority, PolledEvent,
@@ -30,6 +32,51 @@ struct CompletionContractBinding {
criterion_ids: Vec<String>,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "camelCase")]
struct PendingSemanticCall {
operation_id: String,
input: Value,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "camelCase")]
struct CompletedSemanticCall {
operation_id: String,
input: Value,
result: Value,
is_error: bool,
}
fn semantic_tools(payload: &Value) -> Result<Vec<CodexDynamicTool>, DurableRunnerError> {
let tools = payload
.get("semanticTools")
.cloned()
.unwrap_or_else(|| json!([]));
let tools: Vec<CodexDynamicTool> = serde_json::from_value(tools).map_err(|error| {
DurableRunnerError::invalid(format!("run.prepare semanticTools is invalid: {error}"))
})?;
if tools.len() > 128 {
return Err(DurableRunnerError::invalid(
"run.prepare semanticTools exceeds the 128 tool limit",
));
}
for tool in &tools {
tool.validate()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
}
let unique = tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<HashSet<_>>();
if unique.len() != tools.len() {
return Err(DurableRunnerError::invalid(
"run.prepare semanticTools contains duplicate names",
));
}
Ok(tools)
}
fn initial_provider_event_seq() -> u64 {
1
}
@@ -156,6 +203,8 @@ struct CodexProviderState {
lifecycle: String,
config: CodexProviderConfig,
#[serde(default)]
semantic_tools: Vec<CodexDynamicTool>,
#[serde(default)]
completion_contract: Option<CompletionContractBinding>,
#[serde(default)]
thread_id: Option<String>,
@@ -166,6 +215,10 @@ struct CodexProviderState {
#[serde(default)]
last_agent_message: Option<String>,
#[serde(default)]
pending_semantic_calls: BTreeMap<String, PendingSemanticCall>,
#[serde(default)]
completed_semantic_calls: BTreeMap<String, CompletedSemanticCall>,
#[serde(default)]
pending_events: VecDeque<PolledEvent>,
#[serde(default = "initial_provider_event_seq")]
next_provider_event_seq: u64,
@@ -174,6 +227,7 @@ struct CodexProviderState {
impl CodexProviderState {
fn new(
config: CodexProviderConfig,
semantic_tools: Vec<CodexDynamicTool>,
completion_contract: Option<CompletionContractBinding>,
) -> Self {
let thread_id = config.provider_session_id.clone();
@@ -181,11 +235,14 @@ impl CodexProviderState {
schema: PROVIDER_STATE_SCHEMA.to_owned(),
lifecycle: "prepared".to_owned(),
config,
semantic_tools,
completion_contract,
thread_id,
provider_session_id: None,
active_provider_turn_id: None,
last_agent_message: None,
pending_semantic_calls: BTreeMap::new(),
completed_semantic_calls: BTreeMap::new(),
pending_events: VecDeque::new(),
next_provider_event_seq: initial_provider_event_seq(),
}
@@ -195,6 +252,10 @@ impl CodexProviderState {
self.config
.validate()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
for tool in &self.semantic_tools {
tool.validate()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
}
let mut pending_event_ids = HashSet::new();
if self.schema != PROVIDER_STATE_SCHEMA
|| !matches!(
@@ -228,6 +289,24 @@ impl CodexProviderState {
.last_agent_message
.as_ref()
.is_some_and(|value| value.is_empty() || value.len() > 1_000_000)
|| self.semantic_tools.len() > 128
|| self.pending_semantic_calls.len() > 128
|| self.completed_semantic_calls.len() > 128
|| self.pending_semantic_calls.iter().any(|(call_id, call)| {
call_id.is_empty()
|| call_id.len() > 160
|| call.operation_id.is_empty()
|| call.operation_id.len() > 160
|| !call.input.is_object()
|| self.completed_semantic_calls.contains_key(call_id)
})
|| self.completed_semantic_calls.iter().any(|(call_id, call)| {
call_id.is_empty()
|| call_id.len() > 160
|| call.operation_id.is_empty()
|| call.operation_id.len() > 160
|| !call.input.is_object()
})
|| (self.thread_id.is_none()
&& (self.provider_session_id.is_some()
|| self.active_provider_turn_id.is_some()
@@ -354,9 +433,10 @@ 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| {
DurableRunnerError::invalid(format!("failed to resume Codex provider: {error}"))
})?;
let provider = CodexProvider::start(&state.config, Some(&thread_id), &state.semantic_tools)
.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);
self.provider = Some(provider);
if provider_had_exited || recovered_active_turn_id != previous_active_turn_id {
@@ -457,8 +537,12 @@ impl CodexCommandExecutor {
.validate()
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
let completion_contract = completion_contract(payload)?;
let semantic_tools = semantic_tools(payload)?;
if let Some(state) = &self.state {
if state.config != config || state.completion_contract != completion_contract {
if state.config != config
|| state.semantic_tools != semantic_tools
|| state.completion_contract != completion_contract
{
return Err(DurableRunnerError::invalid(
"Codex provider or completion contract changed across the durable run",
));
@@ -469,7 +553,11 @@ impl CodexCommandExecutor {
));
}
} else {
self.state = Some(CodexProviderState::new(config, completion_contract));
self.state = Some(CodexProviderState::new(
config,
semantic_tools,
completion_contract,
));
self.save_state()?;
}
Ok(CommandExecution::result(json!({
@@ -490,10 +578,14 @@ 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 provider = CodexProvider::start(
&state.config,
state.thread_id.as_deref(),
&state.semantic_tools,
)
.map_err(|error| {
DurableRunnerError::invalid(format!("failed to start Codex provider: {error}"))
})?;
self.provider = Some(provider);
}
self.provider
@@ -675,6 +767,75 @@ impl CodexCommandExecutor {
})
}
fn resolve_semantic_tool(
&mut self,
payload: &Value,
) -> Result<CommandExecution, DurableRunnerError> {
let call_id = payload
.get("callId")
.and_then(Value::as_str)
.filter(|value| !value.is_empty() && value.len() <= 160)
.ok_or_else(|| DurableRunnerError::invalid("semantic_tool.result requires callId"))?;
let result = payload
.get("result")
.cloned()
.ok_or_else(|| DurableRunnerError::invalid("semantic_tool.result requires result"))?;
let is_error = payload
.get("isError")
.and_then(Value::as_bool)
.unwrap_or(false);
if let Some(completed) = self
.state
.as_ref()
.and_then(|state| state.completed_semantic_calls.get(call_id))
{
if completed.result != result || completed.is_error != is_error {
return Err(DurableRunnerError::invalid(
"semantic_tool.result conflicts with the completed call",
));
}
return Ok(CommandExecution::result(json!({
"status": "duplicate",
"callId": call_id,
})));
}
let pending = self
.state
.as_ref()
.and_then(|state| state.pending_semantic_calls.get(call_id))
.cloned()
.ok_or_else(|| {
DurableRunnerError::invalid("semantic_tool.result has no pending call")
})?;
let state = self
.state
.as_mut()
.expect("Codex state exists while resolving semantic tool");
state.pending_semantic_calls.remove(call_id);
state.completed_semantic_calls.insert(
call_id.to_owned(),
CompletedSemanticCall {
operation_id: pending.operation_id,
input: pending.input,
result: result.clone(),
is_error,
},
);
self.save_state()?;
// Persist the completed result before crossing the provider boundary.
// If delivery is interrupted, Codex reissues the same native call on
// thread resume and poll_provider redelivers this stored result.
self.ensure_provider()?
.deliver_tool_result(call_id, &result, is_error)
.map_err(|error| {
DurableRunnerError::invalid(format!("Codex tool result delivery failed: {error}"))
})?;
Ok(CommandExecution::result(json!({
"status": "delivered",
"callId": call_id,
})))
}
fn close_session(&mut self) -> Result<CommandExecution, DurableRunnerError> {
if let Some(provider) = self.provider.as_mut() {
provider.shutdown().map_err(|error| {
@@ -811,6 +972,74 @@ impl CodexCommandExecutor {
})?;
self.save_state()?;
}
CodexProviderEvent::ToolCall {
call_id,
operation_id,
input,
} => {
let state = self
.state
.as_mut()
.expect("Codex state remains available while polling");
if let Some(completed) = state.completed_semantic_calls.get(&call_id).cloned() {
if completed.operation_id != operation_id || completed.input != input {
return Err(DurableRunnerError::invalid(
"Codex reused a completed tool call id with different input",
));
}
self.provider
.as_mut()
.expect("provider remains available while reconciling a tool call")
.deliver_tool_result(&call_id, &completed.result, completed.is_error)
.map_err(|error| {
DurableRunnerError::invalid(format!(
"Codex reconciled tool result delivery failed: {error}"
))
})?;
state.push_event(NormalizedProviderEvent {
event_type: "semantic_tool.reconciled".to_owned(),
priority: EventPriority::P0,
payload: json!({
"callId": call_id,
"operationId": operation_id,
"outcome": "completed_result_redelivered",
}),
})?;
} else if let Some(pending) = state.pending_semantic_calls.get(&call_id) {
if pending.operation_id != operation_id || pending.input != input {
return Err(DurableRunnerError::invalid(
"Codex reused a pending tool call id with different input",
));
}
state.push_event(NormalizedProviderEvent {
event_type: "semantic_tool.reconciled".to_owned(),
priority: EventPriority::P0,
payload: json!({
"callId": call_id,
"operationId": operation_id,
"outcome": "pending_call_resumed",
}),
})?;
} else {
state.pending_semantic_calls.insert(
call_id.clone(),
PendingSemanticCall {
operation_id: operation_id.clone(),
input: input.clone(),
},
);
state.push_event(NormalizedProviderEvent {
event_type: "semantic_tool.input".to_owned(),
priority: EventPriority::P0,
payload: json!({
"callId": call_id,
"operationId": operation_id,
"input": input,
}),
})?;
}
self.save_state()?;
}
CodexProviderEvent::Exited { exit_code, success } => {
self.provider = None;
if let Some(state) = self.state.as_mut() {
@@ -859,6 +1088,7 @@ impl CommandExecutor for CodexCommandExecutor {
self.interrupt_turn(&command.command_type)
}
"request.resolve" => self.resolve_request(&command.payload),
"semantic_tool.result" => self.resolve_semantic_tool(&command.payload),
"session.snapshot" => self.snapshot(),
"session.close" | "session.destroy" => self.close_session(),
"runner.drain" | "runner.suspend" | "runner.shutdown" => {
@@ -937,11 +1167,14 @@ mod tests {
instructions: String::new(),
approval_policy: "never".to_owned(),
},
semantic_tools: Vec::new(),
completion_contract: None,
thread_id: Some("thread-1".to_owned()),
provider_session_id: None,
active_provider_turn_id: None,
last_agent_message: None,
pending_semantic_calls: BTreeMap::new(),
completed_semantic_calls: BTreeMap::new(),
pending_events: VecDeque::new(),
next_provider_event_seq: initial_provider_event_seq(),
};
@@ -966,6 +1199,7 @@ mod tests {
instructions: String::new(),
approval_policy: "never".to_owned(),
},
Vec::new(),
Some(CompletionContractBinding {
revision: "1".to_owned(),
criterion_ids: vec!["objective".to_owned()],
packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
+122 -2
View File
@@ -3,7 +3,7 @@ use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use paperclip_runner_core::codex_provider::{
CodexProvider, CodexProviderConfig, CodexProviderEvent,
CodexDynamicTool, CodexProvider, CodexProviderConfig, CodexProviderEvent,
};
use paperclip_runner_core::durable::{Command, CommandExecutor, DurableRunnerError, PolledEvent};
use paperclip_runner_core::provider_backend::CodexCommandExecutor;
@@ -51,6 +51,22 @@ fn provider_config(directory: &Path, switches: &[&str]) -> CodexProviderConfig {
}
}
fn semantic_tool() -> CodexDynamicTool {
CodexDynamicTool {
name: "report_progress".to_owned(),
description: "Append one durable progress update.".to_owned(),
input_schema: json!({
"type": "object",
"properties": {
"idempotencyKey": {"type": "string"},
"body": {"type": "string"}
},
"required": ["idempotencyKey", "body"],
"additionalProperties": false,
}),
}
}
fn command(id: &str, sequence: u64, command_type: &str, payload: Value) -> Command {
Command {
schema: "paperclip.prp.command.v1".to_owned(),
@@ -84,7 +100,7 @@ fn poll_and_ack(
fn codex_transport_buffers_notifications_while_waiting_for_responses() {
let directory = temporary_directory("buffering");
let config = provider_config(&directory, &["--notification-before-response"]);
let mut provider = CodexProvider::start(&config, None).expect("start fake Codex provider");
let mut provider = CodexProvider::start(&config, None, &[]).expect("start fake Codex provider");
let event = provider
.poll()
.expect("poll buffered notification")
@@ -180,6 +196,110 @@ fn durable_backend_resumes_the_active_thread_without_restarting_the_turn() {
fs::remove_dir_all(directory).expect("remove Codex integration-test directory");
}
#[test]
fn durable_backend_resumes_one_pending_native_tool_call_with_the_same_identity() {
let directory = temporary_directory("semantic-resume");
let config = provider_config(&directory, &["--emit-semantic-tool"]);
let tools = vec![semantic_tool()];
let mut first = CodexCommandExecutor::new(&directory);
first
.execute(&command(
"prepare",
1,
"run.prepare",
json!({"provider": config, "semanticTools": tools}),
))
.expect("prepare Codex provider with its authorized tool inventory");
first
.execute(&command("open", 2, "session.open", json!({})))
.expect("open Codex session");
first
.execute(&command(
"turn",
3,
"turn.start",
json!({"text": "Read the active task once."}),
))
.expect("start provider turn");
let mut first_input = None;
for _ in 0..32 {
let events = first.poll_events().expect("poll semantic tool input");
if let Some(event) = events
.iter()
.find(|event| event.event_type == "semantic_tool.input")
{
first_input = Some(event.clone());
first
.acknowledge_events(events.len())
.expect("acknowledge semantic tool input");
break;
}
first
.acknowledge_events(events.len())
.expect("acknowledge events before semantic input");
}
let first_input = first_input.expect("observe semantic tool input");
assert_eq!(first_input.payload["callId"], "semantic-call-1");
assert_eq!(first_input.payload["operationId"], "report_progress");
assert_eq!(call_count(&directory, "turn/start"), 1);
first.shutdown().expect("stop first provider process");
drop(first);
let mut recovered = CodexCommandExecutor::new(&directory);
let mut reconciled = None;
for _ in 0..32 {
let events = recovered
.poll_events()
.expect("poll reconciled semantic call");
if let Some(event) = events
.iter()
.find(|event| event.event_type == "semantic_tool.reconciled")
{
reconciled = Some(event.clone());
recovered
.acknowledge_events(events.len())
.expect("acknowledge reconciliation");
break;
}
recovered
.acknowledge_events(events.len())
.expect("acknowledge events before reconciliation");
}
let reconciled = reconciled.expect("observe native tool-call reconciliation");
assert_eq!(reconciled.payload["callId"], first_input.payload["callId"]);
assert_eq!(reconciled.payload["outcome"], "pending_call_resumed");
assert_eq!(call_count(&directory, "thread/resume"), 1);
assert_eq!(call_count(&directory, "turn/start"), 1);
recovered
.execute(&command(
"semantic-result",
4,
"semantic_tool.result",
json!({
"callId": "semantic-call-1",
"result": {"ok": true, "value": {"task": "NCV-1"}},
"isError": false,
}),
))
.expect("deliver semantic tool result to resumed Codex request");
let mut completed = false;
for _ in 0..32 {
let events = poll_and_ack(&mut recovered).expect("poll completed resumed turn");
completed |= events
.iter()
.any(|event| event.event_type == "turn.completed");
if completed {
break;
}
}
assert!(completed);
assert_eq!(call_count(&directory, "semantic_tool/result"), 1);
recovered.shutdown().expect("stop recovered provider");
fs::remove_dir_all(directory).expect("remove Codex integration-test directory");
}
#[test]
fn provider_exit_preserves_and_reconciles_the_active_turn() {
let directory = temporary_directory("exit-active-turn");
packages/paperclip-runner/src/protocol/generated/schema-bundle.ts
+1
View File
@@ -3032,6 +3032,7 @@ export const eventSchema = {
"usage.reported",
"semantic_tool.input",
"semantic_tool.result",
"semantic_tool.reconciled",
"mcp_app.discovered",
"mcp_app.resource.resolved",
"mcp_app.initializing",
server/src/services/native-runtime/native-codex-runner.integration.test.ts
+57 -6
View File
@@ -6,7 +6,7 @@ import { createServer, type Server } from "node:http";
import { tmpdir } from "node:os";
import { resolve } from "node:path";
import { eq } from "drizzle-orm";
import { and, eq } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
@@ -15,6 +15,7 @@ import {
createDb,
heartbeatRunEvents,
heartbeatRuns,
issueComments,
issues,
nativeRunFinalizations,
nativeRunResults,
@@ -174,6 +175,11 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
environmentLeaseId: "lease-native-codex-e2e",
});
const logs: string[] = [];
const runnerPids: number[] = [];
let resolveRestart!: () => void;
const restarted = new Promise<void>((resolve) => {
resolveRestart = resolve;
});
const execute = executeNativeCodexRunner({
db,
companyId,
@@ -201,13 +207,25 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
resolve(runtimeRoot, "fake-codex-state.json"),
"--call-log",
resolve(runtimeRoot, "fake-codex-calls.log"),
"--emit-semantic-tool",
],
providerVersion: "fake-codex-v1",
},
onLog: async (_stream, chunk) => {
logs.push(chunk);
},
onSpawn: async () => undefined,
onSemanticToolInputCommitted: async ({ callId, operationId }) => {
expect(callId).toBe("semantic-call-1");
expect(operationId).toBe("report_progress");
const firstPid = runnerPids[0];
if (!firstPid) throw new Error("First Runner D process was not recorded");
process.kill(process.platform === "win32" ? firstPid : -firstPid, "SIGKILL");
await restarted;
},
onSpawn: async ({ pid }) => {
runnerPids.push(pid);
if (runnerPids.length === 2) resolveRestart();
},
});
const result = await execute.catch((error) => {
throw new Error(
@@ -240,6 +258,8 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
},
});
expect(logs.join("\n")).not.toContain("PAPERCLIP_RUNNER_BOOTSTRAP_TICKET");
expect(runnerPids).toHaveLength(2);
expect(new Set(runnerPids).size).toBe(2);
const [persistedResult] = await db
.select()
@@ -251,15 +271,45 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
.from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, runId));
expect(finalization).toMatchObject({ phase: "workspace_finalizing" });
const eventTypes = await db
.select({ eventType: heartbeatRunEvents.eventType })
const nativeEvents = await db
.select({
eventType: heartbeatRunEvents.eventType,
payload: heartbeatRunEvents.payload,
})
.from(heartbeatRunEvents)
.where(eq(heartbeatRunEvents.runId, runId));
expect(eventTypes.map((event) => event.eventType)).toEqual(expect.arrayContaining([
expect(nativeEvents.map((event) => event.eventType)).toEqual(expect.arrayContaining([
"turn.completed",
"run.result.proposed",
"run.terminal",
"semantic_tool.input",
"semantic_tool.reconciled",
]));
const semanticInputs = nativeEvents.filter(
(event) => event.eventType === "semantic_tool.input",
);
expect(semanticInputs).toHaveLength(1);
const semanticEvent = semanticInputs[0]?.payload?.prpEvent as
| Record<string, unknown>
| undefined;
const semanticPayload = semanticEvent?.payload as
| Record<string, unknown>
| undefined;
expect(
(semanticPayload?.semantic_tool as Record<string, unknown>)?.callId,
).toBe("semantic-call-1");
const progressEffects = await db
.select({ body: issueComments.body })
.from(issueComments)
.where(
and(
eq(issueComments.issueId, issueId),
eq(issueComments.createdByRunId, runId),
),
);
expect(progressEffects).toEqual([
{ body: "Native resume completed one semantic effect." },
]);
const resumedRunId = randomUUID();
const [resumedRun] = await db.insert(heartbeatRuns).values({
@@ -331,6 +381,7 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
"utf8",
);
expect(providerCalls.match(/^thread\/start$/gm)).toHaveLength(1);
expect(providerCalls.match(/^thread\/resume$/gm)).toHaveLength(1);
expect(providerCalls.match(/^thread\/resume$/gm)).toHaveLength(2);
expect(providerCalls.match(/^semantic_tool\/result$/gm)).toHaveLength(1);
}, 60_000);
});
server/src/services/native-runtime/native-codex-runner.ts
+85 -45
View File
@@ -12,6 +12,7 @@ import { runnerPrpCoordinator } from "./runner-prp-coordinator.js";
const moduleDirectory = dirname(fileURLToPath(import.meta.url));
const RUNNER_VERSION = "paperclip-runner-v1";
const MAX_RUNNER_RESTARTS = 3;
function executableName(): string {
return process.platform === "win32" ? "paperclip-runnerd.exe" : "paperclip-runnerd";
@@ -157,6 +158,11 @@ export async function executeNativeCodexRunner(input: {
args: string[];
providerVersion?: string;
};
/** Internal restart-test seam. Production semantic dispatch does not pause. */
onSemanticToolInputCommitted?: (input: {
readonly callId: string;
readonly operationId: string;
}) => Promise<void>;
onLog: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
onSpawn: (meta: {
pid: number;
@@ -177,6 +183,7 @@ export async function executeNativeCodexRunner(input: {
const prepared = await runnerPrpCoordinator(input.db, {
stateRoot: resolve(runtimeRoot, "control-plane"),
onSemanticToolInputCommitted: input.onSemanticToolInputCommitted,
}).prepare({
companyId: input.companyId,
issueId: input.issueId,
@@ -206,60 +213,91 @@ export async function executeNativeCodexRunner(input: {
instructions: "",
approvalPolicy: "never",
},
semanticTools: prepared.semanticTools,
completionContract: input.completionContract,
}, `prepare_${input.runId}`);
prepared.queueCommand("session.open", {}, `open_${input.runId}`);
prepared.queueCommand("turn.start", { text: input.prompt }, `turn_${input.runId}`);
const child = spawn(binary, buildNativeRunnerArguments({
connectUrl: prepared.connectUrl,
stateDirectory: runnerStateDirectory,
runnerInstanceId: input.runnerInstanceId,
environmentLeaseId: input.environmentLeaseId,
runId: input.runId,
normalizedSessionId: input.normalizedSessionId,
turnId: input.turnId,
itemId: input.itemId,
runnerDigest,
maxRuntimeMs: input.timeoutMs,
}), {
cwd: input.cwd,
detached: process.platform !== "win32",
env: {
...process.env,
...input.environment,
PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: prepared.bootstrapTicket,
},
stdio: ["ignore", "pipe", "pipe"],
});
const exit = waitForExit(child);
child.stdout?.on("data", (chunk: Buffer) => {
void input.onLog("stdout", chunk.toString("utf8"));
});
child.stderr?.on("data", (chunk: Buffer) => {
void input.onLog("stderr", chunk.toString("utf8"));
});
let activeChild: ChildProcess | null = null;
let activeExit: Promise<unknown> | null = null;
try {
if (!child.pid) throw new Error("paperclip_runner_process_not_started");
await input.onSpawn({
pid: child.pid,
processGroupId: process.platform === "win32" ? null : child.pid,
startedAt: new Date().toISOString(),
});
const completed = await Promise.race([
prepared.waitForTerminal(input.timeoutMs),
exit.then(async ({ code, signal }) => {
const recovered = await prepared.waitForTerminal(2_000).catch(() => null);
if (recovered) return recovered;
const terminal = prepared.waitForTerminal(input.timeoutMs);
let bootstrapTicket = prepared.bootstrapTicket;
let restartCount = 0;
let completed: Awaited<typeof terminal> | null = null;
while (completed === null) {
const child = spawn(binary, buildNativeRunnerArguments({
connectUrl: prepared.connectUrl,
stateDirectory: runnerStateDirectory,
runnerInstanceId: input.runnerInstanceId,
environmentLeaseId: input.environmentLeaseId,
runId: input.runId,
normalizedSessionId: input.normalizedSessionId,
turnId: input.turnId,
itemId: input.itemId,
runnerDigest,
maxRuntimeMs: input.timeoutMs,
}), {
cwd: input.cwd,
detached: process.platform !== "win32",
env: {
...process.env,
...input.environment,
PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: bootstrapTicket,
},
stdio: ["ignore", "pipe", "pipe"],
});
const exit = waitForExit(child);
activeChild = child;
activeExit = exit;
child.stdout?.on("data", (chunk: Buffer) => {
void input.onLog("stdout", chunk.toString("utf8"));
});
child.stderr?.on("data", (chunk: Buffer) => {
void input.onLog("stderr", chunk.toString("utf8"));
});
if (!child.pid) throw new Error("paperclip_runner_process_not_started");
await input.onSpawn({
pid: child.pid,
processGroupId: process.platform === "win32" ? null : child.pid,
startedAt: new Date().toISOString(),
});
const outcome = await Promise.race([
terminal.then((value) => ({ kind: "terminal" as const, value })),
exit.then((value) => ({ kind: "exit" as const, value })),
]);
if (outcome.kind === "terminal") {
completed = outcome.value;
break;
}
activeChild = null;
activeExit = null;
const afterExit = await Promise.race([
terminal.then((value) => ({ kind: "terminal" as const, value })),
new Promise<{ kind: "restart" }>((resolveRestart) => {
const timer = setTimeout(() => resolveRestart({ kind: "restart" }), 250);
timer.unref();
}),
]);
if (afterExit.kind === "terminal") {
completed = afterExit.value;
break;
}
if (restartCount >= MAX_RUNNER_RESTARTS) {
throw new Error(
`paperclip_runner_process_exited: code=${code ?? "null"} signal=${signal ?? "null"}`,
`paperclip_runner_process_exited: code=${outcome.value.code ?? "null"} signal=${outcome.value.signal ?? "null"}`,
);
}),
]);
}
restartCount += 1;
bootstrapTicket = prepared.issueBootstrapTicket();
}
if (completed === null) throw new Error("paperclip_runner_terminal_missing");
prepared.queueCommand("session.close", {}, `close_${input.runId}`);
prepared.queueCommand("runner.shutdown", {}, `shutdown_${input.runId}`);
await stopChild(child, exit, true);
if (activeChild && activeExit) {
await stopChild(activeChild, activeExit, true);
}
const succeeded = completed.terminal.runTerminalState === "succeeded";
return {
@@ -285,7 +323,9 @@ export async function executeNativeCodexRunner(input: {
summary: completed.result.summary,
};
} finally {
await stopChild(child, exit).catch(() => undefined);
if (activeChild && activeExit) {
await stopChild(activeChild, activeExit).catch(() => undefined);
}
await prepared.release();
}
}
server/src/services/native-runtime/runner-prp-coordinator.test.ts
+62
View File
@@ -14,9 +14,11 @@ import {
createDb,
heartbeatRunEvents,
heartbeatRuns,
issueComments,
issues,
nativeRunFinalizations,
nativeRunResults,
nativeSemanticReceipts,
} from "@paperclipai/db";
import type {
PrpEvent,
@@ -118,7 +120,9 @@ describeEmbeddedPostgres("hidden runner PRP coordinator", () => {
runnerPrpWebSocketInternals.resetForTests();
await db.delete(nativeRunFinalizations);
await db.delete(nativeRunResults);
await db.delete(nativeSemanticReceipts);
await db.delete(heartbeatRunEvents);
await db.delete(issueComments);
await db.update(issues).set({ executionRunId: null });
await db.delete(heartbeatRuns);
await db.delete(completionContracts);
@@ -270,6 +274,7 @@ describeEmbeddedPostgres("hidden runner PRP coordinator", () => {
"list_documents",
"read_document",
"list_document_revisions",
"report_progress",
]);
expect(
runnerPrpWebSocketInternals.activeRegistration({
@@ -338,6 +343,63 @@ describeEmbeddedPostgres("hidden runner PRP coordinator", () => {
});
});
it("applies one durable progress effect for idempotent semantic retries", async () => {
const seed = await seedNativeRun();
const authority = new PaperclipRunnerSemanticAuthority(db, {
companyId: seed.companyId,
issueId: seed.issueId,
runId: seed.runId,
agentId: seed.agentId,
});
const call = {
callId: "progress-call-1",
operationId: "report_progress",
correlation: {
runId: seed.runId,
normalizedSessionId: seed.sessionId,
turnId: "turn-1",
itemId: "item-1",
},
input: {
idempotencyKey: "progress-once",
body: "The native run resumed and completed its pending call.",
},
};
const applied = await authority.dispatch(call);
expect(applied).toMatchObject({
ok: true,
operationId: "report_progress",
duplicate: false,
value: { disposition: "applied" },
});
const duplicate = await authority.dispatch({
...call,
callId: "progress-call-2",
});
expect(duplicate).toMatchObject({
ok: true,
operationId: "report_progress",
duplicate: true,
});
const comments = await db
.select({ body: issueComments.body })
.from(issueComments)
.where(eq(issueComments.createdByRunId, seed.runId));
expect(comments).toEqual([
{ body: "The native run resumed and completed its pending call." },
]);
await expect(
authority.dispatch({
...call,
callId: "progress-call-3",
input: { ...call.input, body: "Conflicting retry." },
}),
).resolves.toMatchObject({
ok: false,
error: { code: "idempotency_conflict" },
});
});
it("persists events and results idempotently and leases finalization", async () => {
const seed = await seedNativeRun();
const nativeStore = store(seed);
server/src/services/native-runtime/runner-prp-coordinator.ts
+14
View File
@@ -40,6 +40,7 @@ export interface PreparedRunnerPrpSession {
readonly connectUrl: string;
/** One-use secret. Pass it only through the runner's protected bootstrap channel. */
readonly bootstrapTicket: string;
issueBootstrapTicket(): string;
readonly semanticTools: readonly PaperclipSemanticToolDefinition[];
queueCommand(
type: string,
@@ -123,6 +124,11 @@ export function runnerPrpCoordinator(
db: Db,
options: {
readonly stateRoot: string;
/** Internal restart-test seam. Production dispatch has no pause here. */
readonly onSemanticToolInputCommitted?: (input: {
readonly callId: string;
readonly operationId: string;
}) => Promise<void>;
},
) {
const stateRoot = resolve(options.stateRoot);
@@ -264,6 +270,10 @@ export function runnerPrpCoordinator(
}
},
onSemanticToolInput: async (call) => {
await options.onSemanticToolInputCommitted?.({
callId: call.callId,
operationId: call.operationId,
});
const result = await semanticAuthority.dispatch({
callId: call.callId,
operationId: call.operationId,
@@ -291,6 +301,10 @@ export function runnerPrpCoordinator(
return {
connectUrl: registration.connectUrl,
bootstrapTicket,
issueBootstrapTicket: () => {
if (released) throw new Error("runner_prp_session_released");
return authority.issueBootstrapTicket(bootstrapTtlMs);
},
semanticTools,
queueCommand: (type, payload = {}, commandId) => {
if (released) throw new Error("runner_prp_session_released");
server/src/services/native-runtime/runner-semantic-authority.ts
+256 -5
View File
@@ -1,3 +1,5 @@
import { createHash } from "node:crypto";
import { and, desc, eq, isNull } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@@ -9,14 +11,18 @@ import {
issueComments,
issueDocuments,
issues,
nativeSemanticReceipts,
} from "@paperclipai/db";
import {
PaperclipSemanticDispatcher,
digestPaperclipSemanticContent,
type PaperclipJsonValue,
type PaperclipSemanticActionBinding,
type PaperclipSemanticActionId,
type PaperclipSemanticAuthorizationRecord,
type PaperclipSemanticIdempotencyStore,
type PaperclipSemanticRunContext,
type PaperclipSemanticStoredOutcome,
type PaperclipSemanticToolCall,
type PaperclipSemanticToolDefinition,
type PaperclipSemanticToolResult,
@@ -37,6 +43,10 @@ const READ_OPERATION_IDS = [
"list_document_revisions",
] as const satisfies readonly PaperclipSemanticActionId[];
const WRITE_OPERATION_IDS = [
"report_progress",
] as const satisfies readonly PaperclipSemanticActionId[];
type BoundContext = {
readonly run: typeof heartbeatRuns.$inferSelect;
readonly issue: typeof issues.$inferSelect;
@@ -56,6 +66,183 @@ function requiredString(value: unknown): string {
return value;
}
function requiredBody(value: unknown): string {
if (typeof value !== "string" || value.length === 0 || value.length > 20_000) {
throw new Error("paperclip_runner_semantic_input_invalid");
}
return value;
}
function stableUuid(value: string): string {
const digest = createHash("sha256").update(value).digest("hex");
return [
digest.slice(0, 8),
digest.slice(8, 12),
`5${digest.slice(13, 16)}`,
`8${digest.slice(17, 20)}`,
digest.slice(20, 32),
].join("-");
}
function semanticScope(input: {
companyId: string;
runId: string;
operationId: PaperclipSemanticActionId;
idempotencyKey: string;
}): string {
return digestPaperclipSemanticContent([
input.companyId,
input.runId,
input.operationId,
input.idempotencyKey,
]);
}
function claimToken(scope: string): string {
return stableUuid(`paperclip-runner-semantic:${scope}`);
}
class RunnerSemanticIdempotencyStore implements PaperclipSemanticIdempotencyStore {
readonly #db: Db;
readonly #binding: PaperclipRunnerSemanticBinding;
constructor(db: Db, binding: PaperclipRunnerSemanticBinding) {
this.#db = db;
this.#binding = binding;
}
async claim(input: {
readonly scope: string;
readonly operationId: PaperclipSemanticActionId;
readonly inputDigest: string;
}) {
const token = claimToken(input.scope);
const inserted = await this.#db
.insert(nativeSemanticReceipts)
.values({
id: token,
companyId: this.#binding.companyId,
issueId: this.#binding.issueId,
runId: this.#binding.runId,
operationId: input.operationId,
scopeDigest: input.scope,
inputDigest: input.inputDigest,
})
.onConflictDoNothing()
.returning({ id: nativeSemanticReceipts.id });
if (inserted.length > 0) return { kind: "claimed" as const, token };
const [existing] = await this.#db
.select({
operationId: nativeSemanticReceipts.operationId,
scopeDigest: nativeSemanticReceipts.scopeDigest,
inputDigest: nativeSemanticReceipts.inputDigest,
status: nativeSemanticReceipts.status,
outcome: nativeSemanticReceipts.outcome,
updatedAt: nativeSemanticReceipts.updatedAt,
})
.from(nativeSemanticReceipts)
.where(
and(
eq(nativeSemanticReceipts.id, token),
eq(nativeSemanticReceipts.companyId, this.#binding.companyId),
eq(nativeSemanticReceipts.issueId, this.#binding.issueId),
eq(nativeSemanticReceipts.runId, this.#binding.runId),
),
)
.limit(1);
if (!existing) throw new Error("paperclip_runner_semantic_receipt_missing");
if (
existing.operationId !== input.operationId ||
existing.scopeDigest !== input.scope ||
existing.inputDigest !== input.inputDigest
) {
return { kind: "conflict" as const };
}
if (existing.status === "claimed") {
if (Date.now() - existing.updatedAt.getTime() < 60_000) {
return { kind: "in_progress" as const };
}
const reclaimed = await this.#db
.update(nativeSemanticReceipts)
.set({ updatedAt: new Date() })
.where(
and(
eq(nativeSemanticReceipts.id, token),
eq(nativeSemanticReceipts.status, "claimed"),
eq(nativeSemanticReceipts.updatedAt, existing.updatedAt),
),
)
.returning({ id: nativeSemanticReceipts.id });
return reclaimed.length === 1
? { kind: "claimed" as const, token }
: { kind: "in_progress" as const };
}
if (existing.status === "completed" && existing.outcome) {
return {
kind: "duplicate" as const,
outcome: existing.outcome as unknown as PaperclipSemanticStoredOutcome,
};
}
return { kind: "conflict" as const };
}
async complete(token: string, outcome: PaperclipSemanticStoredOutcome): Promise<void> {
const updated = await this.#db
.update(nativeSemanticReceipts)
.set({
status: "completed",
outcome: outcome as unknown as Record<string, unknown>,
updatedAt: new Date(),
})
.where(
and(
eq(nativeSemanticReceipts.id, token),
eq(nativeSemanticReceipts.companyId, this.#binding.companyId),
eq(nativeSemanticReceipts.issueId, this.#binding.issueId),
eq(nativeSemanticReceipts.runId, this.#binding.runId),
eq(nativeSemanticReceipts.status, "claimed"),
),
)
.returning({ id: nativeSemanticReceipts.id });
if (updated.length !== 1) throw new Error("paperclip_runner_semantic_receipt_missing");
}
async recover(token: string, outcome: PaperclipSemanticStoredOutcome): Promise<void> {
const [existing] = await this.#db
.select({
status: nativeSemanticReceipts.status,
outcome: nativeSemanticReceipts.outcome,
})
.from(nativeSemanticReceipts)
.where(eq(nativeSemanticReceipts.id, token))
.limit(1);
if (existing?.status === "completed" && existing.outcome) {
const stored = existing.outcome as unknown as PaperclipSemanticStoredOutcome;
if (JSON.stringify(stored) === JSON.stringify(outcome)) return;
throw new Error("paperclip_runner_semantic_receipt_conflict");
}
await this.complete(token, outcome);
}
async release(token: string): Promise<void> {
const released = await this.#db
.delete(nativeSemanticReceipts)
.where(
and(
eq(nativeSemanticReceipts.id, token),
eq(nativeSemanticReceipts.companyId, this.#binding.companyId),
eq(nativeSemanticReceipts.issueId, this.#binding.issueId),
eq(nativeSemanticReceipts.runId, this.#binding.runId),
eq(nativeSemanticReceipts.status, "claimed"),
),
)
.returning({ id: nativeSemanticReceipts.id });
if (released.length !== 1) {
throw new Error("paperclip_runner_semantic_receipt_release_failed");
}
}
}
function jsonValue(value: unknown): PaperclipJsonValue {
return JSON.parse(JSON.stringify(value)) as PaperclipJsonValue;
}
@@ -68,8 +255,8 @@ function activeAgentStatus(status: string): "active" | "inactive" {
/**
* Run-scoped semantic authority for the hidden native coordinator.
* This first server slice binds only same-task read operations. A catalog
* entry remains undiscoverable until a later PR adds its guarded binding.
* Catalog entries remain undiscoverable until this authority supplies a
* run-scoped binding. Mutations additionally require a durable receipt store.
*/
export class PaperclipRunnerSemanticAuthority {
readonly #db: Db;
@@ -81,9 +268,11 @@ export class PaperclipRunnerSemanticAuthority {
this.#binding = structuredClone(binding);
this.#dispatcher = new PaperclipSemanticDispatcher({
contextProvider: (runId) => this.#context(runId),
bindings: READ_OPERATION_IDS.map((operationId) =>
this.#readBinding(operationId),
),
bindings: [
...READ_OPERATION_IDS.map((operationId) => this.#readBinding(operationId)),
...WRITE_OPERATION_IDS.map((operationId) => this.#writeBinding(operationId)),
],
idempotencyStore: new RunnerSemanticIdempotencyStore(db, binding),
});
}
@@ -276,6 +465,68 @@ export class PaperclipRunnerSemanticAuthority {
};
}
#writeBinding(
operationId: (typeof WRITE_OPERATION_IDS)[number],
): PaperclipSemanticActionBinding {
return {
operationId,
execute: async (invocation) => {
const context = await this.#loadBoundContext();
this.#assertActiveContext(context, true);
const idempotencyKey = requiredString(invocation.input.idempotencyKey);
const body = requiredBody(invocation.input.body);
const scope = semanticScope({
companyId: this.#binding.companyId,
runId: this.#binding.runId,
operationId,
idempotencyKey,
});
const commentId = claimToken(scope);
await this.#db
.insert(issueComments)
.values({
id: commentId,
companyId: this.#binding.companyId,
issueId: this.#binding.issueId,
authorAgentId: this.#binding.agentId,
authorType: "agent",
createdByRunId: this.#binding.runId,
body,
})
.onConflictDoNothing();
const [comment] = await this.#db
.select({
id: issueComments.id,
body: issueComments.body,
authorAgentId: issueComments.authorAgentId,
createdByRunId: issueComments.createdByRunId,
})
.from(issueComments)
.where(eq(issueComments.id, commentId))
.limit(1);
if (
!comment ||
comment.body !== body ||
comment.authorAgentId !== this.#binding.agentId ||
comment.createdByRunId !== this.#binding.runId
) {
throw new Error("paperclip_runner_semantic_effect_conflict");
}
return {
value: jsonValue({
commandId: `runner-semantic:${commentId}`,
disposition: "applied",
stateRevision: context.issue.statusVersion,
entityRefs: [commentId],
scheduledWakeIds: [],
}),
stateRevision: context.issue.statusVersion,
references: [{ kind: "task", id: context.issue.id }],
};
},
};
}
async #context(requestedRunId: string): Promise<PaperclipSemanticRunContext> {
if (requestedRunId !== this.#binding.runId) {
throw new Error("paperclip_runner_semantic_run_mismatch");
server/src/vendor/paperclip-runner/index.ts
Vendored
+4
View File
@@ -14,7 +14,9 @@ export type {
PaperclipSemanticActionBinding,
PaperclipSemanticActionId,
PaperclipSemanticAuthorizationRecord,
PaperclipSemanticIdempotencyStore,
PaperclipSemanticRunContext,
PaperclipSemanticStoredOutcome,
PaperclipSemanticToolCall,
PaperclipSemanticToolDefinition,
PaperclipSemanticToolResult,
@@ -35,6 +37,8 @@ const runner = await import(sourceUrl.href) as RunnerModule;
export const DurablePrpControlPlane = runner.DurablePrpControlPlane;
export const PaperclipSemanticDispatcher = runner.PaperclipSemanticDispatcher;
export const digestPaperclipSemanticContent =
runner.digestPaperclipSemanticContent;
export const validatePrpEvent = runner.validatePrpEvent;
export const validatePrpStructuredRunResult =
runner.validatePrpStructuredRunResult;