feat(runner): add phase 4b protocol server

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-08 04:44:00 +00:00
1 parent 5870a1a432
commit 62d384b383
27 files changed
+8060 -45

No files matched your search

+9 -1
View File
@@ -13,7 +13,10 @@ the Rust outbound WebSocket client, private durable state, replay and restart
recovery, and safe CLI/browser diagnostics. Phase 4 adds a direct, skillless
Codex app-server driver, semantic completion tools, canonical event tracing,
and a real-model example against the mock core. These phases do not import or
start Paperclip's server, UI, CLI, or production database.
start Paperclip's server, UI, CLI, or production database. The Phase 4b lower
layer adds browser-resolved provider requests, goal capability detection,
subagent lineage, control-race semantics, and a package-local demo server that
keeps Codex authentication server-side.
## Phase 0–4 quick start
@@ -55,6 +58,7 @@ pnpm --filter @paperclipai/paperclip-runner replay:phase1
pnpm --filter @paperclipai/paperclip-runner trace:phase2 -- --scenario happy-path
pnpm --filter @paperclipai/paperclip-runner trace:phase3 -- --fault lost-ack
pnpm --filter @paperclipai/paperclip-runner trace:phase4
pnpm --filter @paperclipai/paperclip-runner demo:phase4b -- --host 127.0.0.1 --port 4174
pnpm --filter @paperclipai/paperclip-runner browser:dev --host 127.0.0.1 --port 4179
```
@@ -80,6 +84,8 @@ pnpm --filter @paperclipai/paperclip-runner browser:dev --host 127.0.0.1 --port
| `record:phase3` | Regenerate the complete Phase 3 fault matrix and exact per-fault traces. |
| `trace:phase4` | Run the mock core with a real, local skillless Codex app-server session. |
| `record:phase4` | Run the safe Codex task and record its validated, normalized trace. |
| `demo:phase4b` | Start the package-local HTTP/SSE server with server-only Codex authentication. |
| `record:phase4b` | Run a safe real Codex task through the demo server and record reconnect/replay evidence. |
| `browser:dev` | Start the standalone live/replay/recovery browser devtool. |
| `test:browser` | Exercise static replay and live scenarios, then capture temporary screenshots under ignored test output. |
| `verify` | Run the complete deterministic Phase 0 through Phase 4 acceptance sequence. |
@@ -97,6 +103,8 @@ pnpm --filter @paperclipai/paperclip-runner browser:dev --host 127.0.0.1 --port
- [Phase 3 durable transport reference](docs/phase-03-durable-transport.md)
- [Phase 4 skillless Codex tutorial](docs/tutorials/phase-04-skillless-codex.md)
- [Phase 4 skillless Codex driver reference](docs/phase-04-skillless-codex-driver.md)
- [Phase 4b protocol/server tutorial](docs/tutorials/phase-04b-protocol-server.md)
- [Phase 4b protocol/server reference](docs/phase-04b-protocol-server.md)
- [PRP compatibility/versioning policy](docs/protocol-compatibility.md)
- [Cumulative end-to-end tutorial](docs/tutorials/end-to-end.md)
- [Journal guide](docs/journal.md)
+2
View File
@@ -10,6 +10,7 @@ Each tutorial starts with an explanation of what the phase is and what the phase
- [Phase 2: local runner and fake harness](tutorials/phase-02-local-runner.md) — run a supervised Rust process, exercise scripted live scenarios, resolve requests, interrupt a turn, and confirm live/replay parity.
- [Phase 3: break recovery on purpose](tutorials/phase-03-break-recovery.md) — lose an ACK, drop the socket, restart the runner side, replay durable events, and inspect recovery diagnostics.
- [Phase 4: run the skillless Codex driver](tutorials/phase-04-skillless-codex.md) — run a safe real-model task, inspect the exact context boundary, steer or interrupt it, and confirm one replay-safe result.
- [Phase 4b: run the protocol demo server](tutorials/phase-04b-protocol-server.md) — exercise the server-only Codex boundary, canonical replay, typed controls, and reconnect with `curl`.
- [Cumulative end-to-end tutorial](tutorials/end-to-end.md) — the shortest complete workflow available at the current phase.
## Reference
@@ -19,6 +20,7 @@ Each tutorial starts with an explanation of what the phase is and what the phase
- [Phase 2 local protocol and supervision](phase-02-local-protocol.md)
- [Phase 3 durable transport and recovery](phase-03-durable-transport.md)
- [Phase 4 skillless Codex driver](phase-04-skillless-codex-driver.md)
- [Phase 4b protocol and demo server](phase-04b-protocol-server.md)
- [Engineering journal guide](journal.md)
- [Dated shadcn/ui and AI Elements compatibility note](research/2026-08-07-ui-library-compatibility.md)
- [Phase 4b live-console interaction map](design/phase-4b-interaction-map.md)
@@ -0,0 +1,115 @@
# Phase 4b Protocol and Demo Server
## Scope
This is the lowest Phase 4b layer. It extends the package-local Codex driver,
typed harness contract, deterministic fixtures, and mock-core demo server. It
does not import or change Paperclip server, UI, database, or production routes.
The browser boundary is HTTP plus server-sent events. The browser sends typed
actions to the demo server. Only the server starts `codex app-server`, reads
the existing Codex login, owns the working directory, and resumes provider
threads. No provider or Paperclip credential is serialized to browser state.
## Deterministic fixture
`protocol/fixtures/phase-04b/driver-conformance.json` records the exact Codex
0.132.0 wire shapes used by conformance tests:
- command, file, permission, user-input, and MCP elicitation requests;
- accept, accept-for-session, decline, cancel, and submitted answers;
- goal get, set, pause, resume, and clear calls;
- subagent parent/child thread source metadata;
- same-turn and stale-turn controls;
- interrupt-before-start and terminal-race outcomes;
- stable reconnect identities and source cursor; and
- hostile diagnostic strings that must be redacted.
The fixture loader rejects missing cases, duplicate request identities,
incomplete goal operations, and incomplete lineage or recovery identity.
## Browser-resolved requests
The driver no longer automatically declines supported runtime requests. It
emits `runtime_request.created`, keeps the JSON-RPC response pending, and waits
for `resolveRuntimeRequest`.
| Upstream request | Canonical kind | Allowed response shape |
| --- | --- | --- |
| `item/commandExecution/requestApproval` | `command_approval` | accept, session accept, decline, cancel |
| `item/fileChange/requestApproval` | `file_approval` | accept, session accept, decline, cancel |
| `item/permissions/requestApproval` | `permission_approval` | empty safe permission grant for turn or session |
| `item/tool/requestUserInput` | `user_input` | typed answer map, decline, cancel |
| `mcpServer/elicitation/request` | `elicitation` | typed content, decline, cancel |
The permission response cannot widen the skillless filesystem or network
boundary. A stale turn, duplicate response, provider cleanup, terminal turn,
protocol failure, close, or reconnect settles the pending request exactly once.
Only the resolution action enters the canonical event; submitted secrets or
answers do not enter the acknowledgement payload.
## Steering and interruption
Successful same-turn steering emits a completed
`steering_acknowledgement` item. A different or child identity is rejected
with `HarnessStaleTurnError` and `stale_turn_rejected`; direct child steering
is never emulated.
An interrupt received while `turn/start` is waiting for a provider turn ID is
queued and sent as soon as the ID arrives. After a terminal fact wins the race,
further steering or interruption returns `already_terminal`. A successful
interrupt request emits an `interrupt_acknowledgement` item without replacing
the session.
## Goals and lineage
The driver probes `thread/goal/get` after create or resume. Success enables
get, set, pause, resume, and clear. A method-not-found response disables goals
and advertises the precise unsupported capability. The installed Codex 0.132.0
binary generated goal bindings but did not enable the method at runtime in the
recorded real session, so the evidence correctly shows `goals: false`.
Root and child threads use `thread.id`, `thread.sessionId`, `forkedFromId`, and
`source.subAgent.thread_spawn` to build lineage. Child start, status, and close
notifications become canonical `thread_lineage` items. Unrelated thread
notifications are ignored instead of being attached to the active session.
## Demo-server API
Start it with:
```sh
pnpm --filter @paperclipai/paperclip-runner demo:phase4b -- \
--host 127.0.0.1 --port 4174
```
Important routes:
- `GET /api/phase4b/health`
- `POST /api/phase4b/sessions`
- `GET /api/phase4b/sessions/:id`
- `GET /api/phase4b/sessions/:id/events?after=:cursor`
- `GET /api/phase4b/sessions/:id/stream?after=:cursor`
- `POST /api/phase4b/sessions/:id/steer`
- `POST /api/phase4b/sessions/:id/interrupt`
- `POST /api/phase4b/sessions/:id/requests/:requestId/resolve`
- `POST /api/phase4b/sessions/:id/goal/:operation`
- `POST /api/phase4b/sessions/:id/reconnect`
The server chooses the working directory. A create body cannot override it.
Every JSON and event response passes a second bounded redaction layer. Replay
uses the same validated PRP events and reducer as the live view.
## Evidence
Run deterministic conformance first, then the real Codex boundary:
```sh
pnpm --filter @paperclipai/paperclip-runner exec vitest run \
src/drivers/codex/codex-app-server-driver.test.ts \
src/mock-core/phase4b-demo-server.test.ts
pnpm --filter @paperclipai/paperclip-runner record:phase4b
```
See the [verification record](../knowledge/evidence/2026-08-08-phase-04b-protocol-server-verification.md)
and [real server trace](../knowledge/evidence/phase-04b-real-codex-server.json).
@@ -3,7 +3,7 @@
## What this tutorial is
This tutorial combines each implemented Native Runner phase into one procedure.
It currently includes Phase 0 through Phase 4.
It currently includes Phase 0 through the Phase 4b protocol/server layer.
## What this tutorial proves
@@ -49,6 +49,11 @@ semantic completion tools, and the same reducer/replay proof used by fixtures.
stable session identities and exactly one result and terminal event.
21. Open the [Phase 4 journal entry](../../knowledge/journal/2026-08-08-phase-04.md)
and its linked real-session trace and verification evidence.
22. Follow [Phase 4b: Run the Protocol Demo Server](phase-04b-protocol-server.md).
23. Confirm requests stay pending for a typed browser decision, stale steering
is rejected, and reconnect keeps the same run and session identities.
24. Open the [Phase 4b protocol/server journal](../../knowledge/journal/2026-08-08-phase-04b-protocol-server.md)
and its linked deterministic and real-Codex evidence.
The one-command form after installation is:
@@ -103,6 +108,18 @@ pnpm --filter @paperclipai/paperclip-runner verify:rootless
idempotent, and a changed duplicate is rejected;
- unsupported capabilities degrade through explicit redacted diagnostics;
- the real trace and its replay reduce to the same final snapshot.
- supported provider requests wait for one typed browser resolution and clean
up exactly once;
- same-turn steering is acknowledged while stale and direct-child steering are
rejected;
- pre-start interrupts queue until the provider turn ID exists and terminal
races return `already_terminal`;
- goal controls are capability probed and disabled precisely when unavailable;
- parent/child activity derives from provider thread identities;
- the demo server fixes the workspace and keeps Codex authentication out of
browser JSON, events, and diagnostics;
- refresh/reconnect replays canonical events and resumes the exact persisted
provider thread.
Phase 5 remains uncreated until the Phase 4 review and human checkpoint are
complete.
Phase 5 remains gated on the complete Phase 4b browser, review, QA, live-service,
and human-checkpoint evidence.
@@ -0,0 +1,112 @@
# Phase 4b: Run the Protocol Demo Server
## What this is
This tutorial runs the Phase 4b server layer. It has no browser UI yet. You can
use `curl` to act like the browser.
The server starts Codex. Codex login data stays on the server. The JSON replies
must not contain a Paperclip key, an OpenAI key, or a bearer value.
## Step 1: Run deterministic checks
From the repository root, run:
```sh
pnpm --filter @paperclipai/paperclip-runner exec vitest run \
src/drivers/codex/codex-app-server-driver.test.ts \
src/mock-core/phase4b-demo-server.test.ts
```
Expected result: both files pass. The tests cover requests, goals, lineage,
steering, interrupt races, reconnect, replay, and redaction.
## Step 2: Start the server
Use an empty directory outside your Codex home:
```sh
phase4b_workspace="$(mktemp -d)"
pnpm --filter @paperclipai/paperclip-runner demo:phase4b -- \
--host 127.0.0.1 \
--port 4174 \
--working-directory "$phase4b_workspace"
```
Keep this terminal open. The first output line is JSON. It names the host,
port, and server-owned working directory. It also says
`"credentialsExposed":false`.
## Step 3: Check the boundary
In a second terminal, run:
```sh
curl -s http://127.0.0.1:4174/api/phase4b/health | jq
```
Expected facts:
- `status` is `ok`;
- `boundary` is `package-local-mock-core`;
- `providerAuthentication` is `server-side`; and
- `credentialsExposed` is `false`.
## Step 4: Start one safe session
```sh
curl -s -X POST http://127.0.0.1:4174/api/phase4b/sessions \
-H 'content-type: application/json' \
--data '{"objective":"Create demo.txt with exactly: phase4b demo","message":"Complete and verify the exact file task."}' \
| tee "$phase4b_workspace/session.json" | jq
```
Copy `sessionId` and `activeTurnId` from the reply. The server ignores any
browser-supplied working-directory or credential field.
## Step 5: Read live or replayed events
```sh
session_id="$(jq -r .sessionId "$phase4b_workspace/session.json")"
curl -s "http://127.0.0.1:4174/api/phase4b/sessions/$session_id/events?after=0" | jq
```
The reply contains canonical PRP events and a cursor. A later request can pass
that cursor as `after`. The same events go through the same reducer during live
view and replay.
For a live stream, run:
```sh
curl -N "http://127.0.0.1:4174/api/phase4b/sessions/$session_id/stream?after=0"
```
## Step 6: Reconnect the same session
```sh
curl -s -X POST \
"http://127.0.0.1:4174/api/phase4b/sessions/$session_id/reconnect" \
-H 'content-type: application/json' --data '{}' | jq
```
Confirm that `runId`, `normalizedSessionId`, `driverSessionId`, and
`providerSessionId` did not change. The new events include `session.resumed`.
## Step 7: Record real evidence
Stop the manual server. Then run:
```sh
pnpm --filter @paperclipai/paperclip-runner record:phase4b
```
This starts an ephemeral loopback server, runs one safe real Codex task,
checks goal capability, steers when the turn is still active, reconnects,
checks the exact file, and writes the redacted evidence JSON.
Read:
- `packages/paperclip-runner/knowledge/evidence/phase-04b-real-codex-server.json`
- `packages/paperclip-runner/knowledge/evidence/2026-08-08-phase-04b-protocol-server-verification.md`
The deterministic fixture is the conformance authority. A real model can vary.
@@ -0,0 +1,100 @@
---
type: Verification Evidence
title: Phase 4b protocol and demo-server verification
description: Deterministic and real-Codex evidence for browser-resolved requests, goals, lineage, controls, reconnect, replay, and credential isolation.
tags: [native-runner, phase-4b, codex, protocol, demo-server, evidence]
status: stable
generated: { by: openai/gpt-5, at: 2026-08-08T04:43:00Z }
---
# Scope
This record covers the lowest Phase 4b driver, protocol fixture, and
package-local demo server. It changes nothing outside
`packages/paperclip-runner/` and exposes no Paperclip or provider credential to
the browser boundary.
# Deterministic evidence
```sh
pnpm --filter @paperclipai/paperclip-runner typecheck:typescript
pnpm --filter @paperclipai/paperclip-runner exec vitest run \
src/drivers/codex/codex-app-server-driver.test.ts \
src/mock-core/phase4b-demo-server.test.ts
```
Result: TypeScript typecheck passed. The two focused files passed 37 tests.
The demo-server test needed permission to bind an ephemeral loopback port; its
first in-sandbox run failed with `listen EPERM`, then the same test passed with
loopback access.
The deterministic fixture covers all five upstream request kinds, five goal
operations, root/child lineage, same-turn steering, stale rejection,
interrupt-before-start, terminal races, recovery identities, and redaction.
# Package acceptance
```sh
pnpm --filter @paperclipai/paperclip-runner verify
pnpm --filter @paperclipai/paperclip-runner verify:rootless
```
The direct acceptance command passed its build, all typechecks, 107 TypeScript
tests, 39 Rust tests, script tests, and token gate before Chromium failed to
launch because this host does not install `libatk-1.0.so.0`. The repository's
supported rootless wrapper downloaded the browser libraries into the run-owned
scratch directory and then passed the complete acceptance path: 9 Playwright
tests, standalone/import and documentation checks, Phase 0 and Phase 1 parity,
and the Phase 0 through Phase 3 traces.
# Real Codex evidence
```sh
pnpm --filter @paperclipai/paperclip-runner record:phase4b
```
The first attempt correctly failed because the run-owned scratch directory was
outside the injected `PAPERCLIP_WORKSPACE_CWD`. The recorder now binds the
driver's assigned workspace root to that exact scratch directory. The second
attempt reached Codex but found that the generated goal methods were not
advertised at runtime. The recorder now records this as a precise disabled
capability instead of invoking a fake control. The final attempt passed.
Recorded result:
- Codex CLI/app-server: `0.132.0`;
- model/provider: recorded in the canonical session context;
- 190 canonical events before reconnect;
- 5 canonical events after the replay cursor, including `session.resumed`;
- exact output: `phase4b.txt` contains `phase4b real codex evidence`;
- exactly one semantic result and one turn terminal;
- continuous source sequence;
- stable run, normalized session, driver thread, and provider session IDs;
- explicit `goals: false` with `capability not advertised`;
- server-side provider authentication and no credential value in evidence.
See the [real server trace](phase-04b-real-codex-server.json).
# Failures fixed
- Replaced automatic request decline with one pending typed browser resolution.
- Added cleanup for provider resolution, terminal, protocol failure, close, and
reconnect so a request cannot receive two responses.
- Added visible same-turn steering acknowledgements and exact stale-turn errors.
- Queued pre-start interrupts until the provider turn ID exists and return
`already_terminal` when completion wins.
- Added capability-probed goal methods rather than version-only assumptions.
- Added parent/child lineage derived from upstream identity instead of inferred
UI nesting.
- Added a second bounded browser redaction pass; narrowed its sensitive-key
matcher after a test found that `tokenBudget` was incorrectly redacted.
- Redacted nested Paperclip home paths from Codex sandbox metadata after the
final evidence scan found a server-side writable-root path.
- Made the demo server use only its configured workspace after the boundary
test supplied a hostile browser path.
# Remaining ownership
The Phase 4b browser issue owns the React UI, request cards, controls, protocol
inspector, screenshots, accessibility checks, and persistent live-service
handoff. This lowest layer is ready for that consumer.
@@ -8,3 +8,4 @@
* [Phase 2 verification](2026-08-07-phase-02-verification.md) - Rust supervisor, fake-driver scenarios, cleanup, live/replay parity, browser flows, screenshots, and complete package verification.
* [Phase 3 verification](2026-08-07-phase-03-verification.md) - Lost-ACK replay, restart recovery, command deduplication, storage, and redacted diagnostics.
* [Phase 4 verification](2026-08-08-phase-04-verification.md) - Direct Codex conformance, exact skillless context, semantic completion, recovery, and real-session replay.
* [Phase 4b protocol/server verification](2026-08-08-phase-04b-protocol-server-verification.md) - Browser-resolved requests, goals, lineage, controls, reconnect, redaction, and real Codex server evidence.
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,72 @@
---
type: Engineering Journal Entry
title: Phase 4b protocol and demo-server layer
description: Decisions, failures, fixes, and evidence for browser-resolved Codex requests and the server-only credential boundary.
tags: [native-runner, phase-4b, codex, protocol, demo-server]
status: stable
generated: { by: openai/gpt-5, at: 2026-08-08T04:35:00Z }
entry_kind: phase
phase: "4b"
---
# Context
Phase 4 proved one skillless real Codex run. Phase 4b needs a browser, but the
browser must not own Codex authentication or invent a second event model. This
entry records the lower driver and server contract built before the UI.
# Decisions
1. Store the installed Codex 0.132.0 wire shapes in one deterministic fixture.
2. Keep provider server requests pending until one typed browser resolution.
3. Map browser actions to narrow response unions. Never accept an arbitrary raw
provider response from the browser.
4. Keep permission acceptance empty so it cannot widen the skillless sandbox.
5. Emit visible canonical acknowledgement items for steer and interrupt.
6. Reject stale turns and direct child steering. Never emulate either action.
7. Queue interruption only while `turn/start` lacks its provider ID. After a
terminal, return `already_terminal`.
8. Probe goal support with the side-effect-free get method. Generated bindings
alone do not prove runtime capability.
9. Derive subagent lineage from upstream thread source and parent identity.
10. Fix the demo server's workspace at construction. Browser bodies cannot
choose filesystem roots.
11. Replay only validated canonical events through the existing reducer.
12. Apply browser redaction after driver redaction as defense in depth.
# Evidence
See the [verification record](../evidence/2026-08-08-phase-04b-protocol-server-verification.md),
[real Codex server trace](../evidence/phase-04b-real-codex-server.json),
[reference](../../docs/phase-04b-protocol-server.md), and
[tutorial](../../docs/tutorials/phase-04b-protocol-server.md).
# Failures
- The first HTTP test could not bind loopback inside the default sandbox. It
passed unchanged with explicit loopback permission.
- The initial public redactor treated every key containing `token` as a secret
and hid `tokenBudget`. The matcher now targets credential key names, not
usage fields.
- The first real recorder workspace conflicted with the heartbeat's assigned
workspace root. The recorder now supplies the exact scratch root to the
driver rather than weakening containment.
- Codex generated goal types but returned method unavailable at runtime. The
driver now disables the controls and records the diagnostic.
# Known gaps
- The demo server is an in-memory tracer boundary, not production persistence.
- Pending provider requests cannot survive loss of the provider process. The
canonical pending fact remains replayable, and recovery records cancellation.
- The lower layer does not render UI or provide screenshots. The next Phase 4b
browser task consumes this API.
- The real model did not exercise every request kind or a subagent. The
deterministic fixture remains the conformance authority for those paths.
# Follow-up questions
- Should Phase 5 freeze HTTP polling plus SSE, or expose a transport-neutral
client interface first?
- Which Codex feature advertisement should replace the goal method probe when
the app-server publishes a stable capability list?
@@ -9,3 +9,4 @@
* [Phase 3 durable outbound transport and recovery](2026-08-07-phase-03.md) - WebSocket authentication, durable ACK replay, command deduplication, and recovery diagnostics.
* [Phase 4 skillless Codex app-server driver](2026-08-08-phase-04.md) - Direct app-server mapping, context isolation, semantic completion, real-model failures, and fixes.
* [Phase 4b UX gate — interaction map and component decisions](2026-08-08-phase-04b-ux.md) - Approved live-console interaction contract, adapt/reject component record, and keyboard/accessibility acceptance criteria.
* [Phase 4b protocol and demo-server layer](2026-08-08-phase-04b-protocol-server.md) - Browser-resolved provider requests, control races, goals, lineage, replay, and server-only credentials.
@@ -18,3 +18,6 @@
shadcn/AI Elements adapt/reject decision record with a zero-new-runtime-
dependency rule, keyboard/accessibility acceptance criteria, four token
additions, and annotated desktop/mobile mockups.
* **Phase 4b protocol/server layer**: Added deterministic Codex request, goal,
lineage, control-race, reconnect, replay, and redaction fixtures; a typed
server-only demo boundary; and a passing real Codex server trace.
+2
View File
@@ -46,6 +46,8 @@
"record:phase2": "pnpm run trace:phase2 -- --quiet --output knowledge/evidence/phase-02-happy-path.json",
"trace:phase3": "pnpm run build:typescript && pnpm run build:phase2-binaries && node dist/cli/phase3-recovery.js",
"trace:phase4": "pnpm run build:typescript && node dist/cli/phase4-codex.js",
"demo:phase4b": "pnpm run build:typescript && node scripts/phase4b-demo-server.mjs",
"record:phase4b": "pnpm run build:typescript && node scripts/record-phase4b-evidence.mjs",
"record:phase4": "pnpm run build:typescript && node scripts/record-phase4-evidence.mjs",
"diagnose:phase3": "pnpm run trace:phase3",
"record:phase3": "pnpm run build:typescript && pnpm run build:phase2-binaries && node scripts/record-phase3-evidence.mjs",
@@ -0,0 +1,111 @@
{
"schema": "paperclip.runner.phase4b.conformance.v1",
"codexVersion": "0.132.0",
"runtimeRequests": [
{
"id": "command-accept-session",
"method": "item/commandExecution/requestApproval",
"requestKind": "command_approval",
"resolution": { "action": "accept_for_session" },
"expectedResponse": { "decision": "acceptForSession" }
},
{
"id": "file-reject",
"method": "item/fileChange/requestApproval",
"requestKind": "file_approval",
"resolution": { "action": "decline" },
"expectedResponse": { "decision": "decline" }
},
{
"id": "permission-accept-turn",
"method": "item/permissions/requestApproval",
"requestKind": "permission_approval",
"resolution": { "action": "accept" },
"expectedResponse": { "permissions": {}, "scope": "turn" }
},
{
"id": "user-input-submit",
"method": "item/tool/requestUserInput",
"requestKind": "user_input",
"resolution": {
"action": "submit",
"answers": { "preferred_name": { "answers": ["Ada"] } }
},
"expectedResponse": {
"answers": { "preferred_name": { "answers": ["Ada"] } }
}
},
{
"id": "elicitation-cancel",
"method": "mcpServer/elicitation/request",
"requestKind": "elicitation",
"resolution": { "action": "cancel" },
"expectedResponse": { "action": "cancel", "content": null, "_meta": null }
}
],
"goals": [
{
"action": "get",
"method": "thread/goal/get",
"params": {}
},
{
"action": "set",
"method": "thread/goal/set",
"params": { "objective": "Ship the Phase 4b tracer", "status": "active", "tokenBudget": 4096 }
},
{
"action": "pause",
"method": "thread/goal/set",
"params": { "status": "paused" }
},
{
"action": "resume",
"method": "thread/goal/set",
"params": { "status": "active" }
},
{
"action": "clear",
"method": "thread/goal/clear",
"params": {}
}
],
"lineage": {
"rootThreadId": "thread-root",
"childThread": {
"id": "thread-child",
"sessionId": "provider-session-1",
"source": {
"subAgent": {
"thread_spawn": {
"parent_thread_id": "thread-root",
"depth": 1,
"agent_path": ["researcher"],
"agent_nickname": "Scout",
"agent_role": "researcher"
}
}
},
"agentNickname": "Scout",
"agentRole": "researcher"
}
},
"controls": {
"sameTurnSteer": { "turnId": "turn-1", "expected": "acknowledged" },
"staleTurnSteer": { "turnId": "turn-stale", "expected": "stale_turn" },
"interruptBeforeStart": { "expected": "queued" },
"interruptAfterTerminal": { "expected": "already_terminal" }
},
"reconnect": {
"runId": "run-phase4b",
"normalizedSessionId": "normalized-phase4b",
"driverSessionId": "thread-root",
"providerSessionId": "provider-session-1",
"lastSourceSequence": 17
},
"redactionMarkers": [
"Bearer browser-secret",
"OPENAI_API_KEY=provider-secret",
"https://user:password@example.test/path?token=query-secret"
]
}
@@ -0,0 +1,52 @@
import { mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import { resolve } from "node:path";
import {
CodexAppServerDriver,
Phase4bDemoServer,
} from "../dist/index.js";
function option(name) {
const index = process.argv.indexOf(name);
return index < 0 ? undefined : process.argv[index + 1];
}
const host = option("--host") ?? "127.0.0.1";
const parsedPort = Number.parseInt(option("--port") ?? "4174", 10);
if (!Number.isInteger(parsedPort) || parsedPort < 0 || parsedPort > 65535) {
throw new Error("--port must be an integer from 0 to 65535");
}
const requestedDirectory = option("--working-directory");
const workingDirectory = requestedDirectory === undefined
? await mkdtemp(resolve(process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? tmpdir(), "phase4b-demo-"))
: resolve(requestedDirectory);
const server = new Phase4bDemoServer({
host,
port: parsedPort,
workingDirectory,
driverFactory: (taskEnvelope) => new CodexAppServerDriver({
taskEnvelope,
onDiagnostic: (message) => process.stderr.write(`[phase4b] ${message}\n`),
}),
});
const address = await server.start();
process.stdout.write(`${JSON.stringify({
schema: "paperclip.runner.phase4b.server.v1",
...address,
workingDirectory,
providerAuthentication: "server-side",
credentialsExposed: false,
})}\n`);
let closing = false;
async function close() {
if (closing) return;
closing = true;
await server.close();
}
process.once("SIGINT", () => void close().finally(() => process.exit(0)));
process.once("SIGTERM", () => void close().finally(() => process.exit(0)));
@@ -0,0 +1,184 @@
import { mkdtemp, readFile, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { dirname, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import {
CodexAppServerDriver,
Phase4bDemoServer,
} from "../dist/index.js";
const packageRoot = resolve(dirname(fileURLToPath(import.meta.url)), "..");
const workspace = await mkdtemp(resolve(
process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? tmpdir(),
"phase4b-real-codex-",
));
const output = resolve(
packageRoot,
process.argv[2] ?? "knowledge/evidence/phase-04b-real-codex-server.json",
);
const server = new Phase4bDemoServer({
host: "127.0.0.1",
port: 0,
workingDirectory: workspace,
driverFactory: (taskEnvelope) => new CodexAppServerDriver({
taskEnvelope,
environment: { ...process.env, PAPERCLIP_WORKSPACE_CWD: workspace },
}),
});
const address = await server.start();
const commandLog = [];
async function request(path, init) {
commandLog.push({ method: init?.method ?? "GET", path });
const response = await fetch(`${address.url}${path}`, init);
const body = await response.json();
if (!response.ok) throw new Error(`${path} returned ${response.status}: ${JSON.stringify(body)}`);
return body;
}
function post(path, body) {
return request(path, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify(body),
});
}
function normalize(value) {
if (typeof value === "string") return value.replaceAll(workspace, "<phase4b-workspace>");
if (Array.isArray(value)) return value.map(normalize);
if (typeof value !== "object" || value === null) return value;
return Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, normalize(entry)]));
}
try {
const health = await request("/api/phase4b/health");
const created = await post("/api/phase4b/sessions", {
objective: "Create phase4b.txt with exactly: phase4b real codex evidence",
message: "Complete the exact file task, verify it, and return the semantic result.",
});
const sessionId = created.sessionId;
const turnId = created.activeTurnId;
let goalSet;
let goalPause;
let goalResume;
if (created.capabilities?.goals === true) {
goalSet = await post(`/api/phase4b/sessions/${sessionId}/goal/set`, {
objective: "Complete the Phase 4b real Codex evidence turn",
tokenBudget: 8192,
});
goalPause = await post(`/api/phase4b/sessions/${sessionId}/goal/pause`, {});
goalResume = await post(`/api/phase4b/sessions/${sessionId}/goal/resume`, {});
} else {
const unsupported = { supported: false, diagnostic: "capability not advertised" };
goalSet = unsupported;
goalPause = unsupported;
goalResume = unsupported;
}
let steer;
try {
steer = await post(`/api/phase4b/sessions/${sessionId}/steer`, {
turnId,
text: "Keep the file content exact and include the verification in the semantic result.",
});
} catch (error) {
steer = { race: "already_terminal", diagnostic: String(error) };
}
const deadline = Date.now() + 120_000;
let replay;
while (Date.now() < deadline) {
replay = await request(`/api/phase4b/sessions/${sessionId}/events?after=0`);
if (replay.events.some((event) => [
"turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled",
].includes(event.eventType))) break;
await new Promise((resolve) => setTimeout(resolve, 250));
}
if (!replay?.events.some((event) => [
"turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled",
].includes(event.eventType))) {
throw new Error("real Phase 4b turn did not terminalize before the timeout");
}
const beforeReconnect = await request(`/api/phase4b/sessions/${sessionId}`);
const reconnected = await post(`/api/phase4b/sessions/${sessionId}/reconnect`, {});
const afterReconnect = await request(
`/api/phase4b/sessions/${sessionId}/events?after=${replay.cursor}`,
);
const fileText = await readFile(resolve(workspace, "phase4b.txt"), "utf8");
const serialized = JSON.stringify({
health,
created,
goalSet,
goalPause,
goalResume,
steer,
replay,
beforeReconnect,
reconnected,
afterReconnect,
});
const secretValues = [process.env.PAPERCLIP_API_KEY, process.env.OPENAI_API_KEY]
.filter((value) => typeof value === "string" && value.length > 0);
const events = replay.events;
const sessionContext = events.find((event) => event.eventType === "session.started")
?.payload?.context;
const codexVersion = typeof sessionContext?.codexVersion === "string"
? sessionContext.codexVersion
: "not reported";
const evidence = normalize({
schema: "paperclip.runner.phase4b.real-evidence.v1",
recordedAt: new Date().toISOString(),
codexVersion,
server: {
host: address.host,
port: address.port,
providerAuthentication: health.providerAuthentication,
credentialsExposed: health.credentialsExposed,
},
commandLog,
identity: {
sessionId,
runId: created.runId,
normalizedSessionId: created.normalizedSessionId,
driverSession: created.driverSession,
reconnectedDriverSession: reconnected.driverSession,
},
capabilities: reconnected.capabilities,
goals: {
set: goalSet.goal ?? goalSet,
paused: goalPause.goal ?? goalPause,
resumed: goalResume.goal ?? goalResume,
},
steering: steer,
file: { path: "phase4b.txt", text: fileText },
events,
reconnectEvents: afterReconnect.events,
assertions: {
exactFile: fileText === "phase4b real codex evidence",
semanticResult: events.filter((event) => event.eventType === "run.result.proposed").length === 1,
oneTurnTerminal: events.filter((event) => [
"turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled",
].includes(event.eventType)).length === 1,
continuousSourceSequence: events.every((event, index) => event.sourceSeq === index + 1),
stableRunIdentity: reconnected.runId === created.runId &&
reconnected.normalizedSessionId === created.normalizedSessionId,
stableProviderIdentity: JSON.stringify(reconnected.driverSession) === JSON.stringify(created.driverSession),
replayAfterReconnect: afterReconnect.events.some((event) => event.eventType === "session.resumed"),
goalCapabilityExplicit: typeof reconnected.capabilities?.goals === "boolean",
credentialsAbsent: health.credentialsExposed === false &&
secretValues.every((secret) => !serialized.includes(secret)),
hostPathsAbsent: !serialized.includes("/srv/paperclip/home") &&
!serialized.includes(".paperclip/instances/"),
},
});
if (!Object.values(evidence.assertions).every(Boolean)) {
throw new Error(`Phase 4b evidence assertion failed: ${JSON.stringify(evidence.assertions)}`);
}
await writeFile(output, `${JSON.stringify(evidence, null, 2)}\n`);
process.stdout.write(`${JSON.stringify({ output, assertions: evidence.assertions })}\n`);
} finally {
await server.close();
}
@@ -358,6 +358,19 @@ Phase 5 conditionally: do not create or start Phase 5 until the complete Phase
tutorial, screenshots, OKF journal entry, review gates, QA evidence, live
service handoff, and human checkpoint are complete.
**Lowest-layer implementation evidence (2026-08-08):** the driver/protocol and
demo-server slice is implemented under `packages/paperclip-runner/`. It adds a
Codex 0.132.0 deterministic conformance fixture, typed browser resolution for
command/file/permission/user-input/elicitation requests, same-turn steering
acknowledgement and stale rejection, queued pre-start interruption and terminal
race semantics, capability-probed goals, upstream-derived parent/child lineage,
exact resume/replay identity, bounded redaction, and an HTTP/SSE demo server
with a fixed server-owned workspace and server-only Codex authentication. The
real server evidence records one exact file result, one semantic result, one
terminal, continuous source sequence, stable reconnect identity, and explicit
runtime goal unavailability. Browser rendering, screenshots, reviews, QA, live
service handoff, and the human checkpoint remain later Phase 4b gates.
## Phase 5 — Browser reference console and reusable SDK surface
**Purpose:** generalize the Phase 4b browser proof into a stable reference
@@ -6657,6 +6657,17 @@ requests, goal operations, and parent/child lineage are explicit Phase 4b
contracts and must receive deterministic conformance fixtures before UI work
claims them.
The implemented lowest-layer contract keeps supported provider server requests
pending behind a typed resolution union, emits canonical acknowledgement or
cleanup facts exactly once, probes goal support through `thread/goal/get`, and
derives lineage from upstream thread identity/source fields. The package-local
HTTP/SSE demo server fixes its workspace at construction, resumes the exact
persisted driver/provider thread, replays only validated PRP events, and applies
a second bounded redaction pass before browser delivery. The real Codex 0.132.0
recording found generated goal bindings but no runtime-advertised goal method;
the capability is therefore disabled with an exact diagnostic instead of
presenting fabricated controls.
Use source-adapted shadcn/ui and Vercel AI Elements primitives where they fit,
after UX review. Keep the package's own protocol and reducer authoritative;
third-party chat message or tool-part types must not become the runner protocol.
@@ -37,6 +37,79 @@ export class HarnessReconciliationError extends Error {
}
}
export class HarnessOperationAlreadyTerminalError extends Error {
readonly code = "already_terminal" as const;
constructor(operation: string) {
super(`${operation} lost a race with the committed turn terminal`);
this.name = "HarnessOperationAlreadyTerminalError";
}
}
export class HarnessStaleTurnError extends Error {
readonly code = "stale_turn" as const;
constructor(turnId: string) {
super(`turn ${turnId} is not the active turn`);
this.name = "HarnessStaleTurnError";
}
}
export type HarnessRuntimeRequestKind =
| "command_approval"
| "file_approval"
| "permission_approval"
| "user_input"
| "elicitation";
export interface HarnessRuntimeRequest {
requestId: string;
requestKind: HarnessRuntimeRequestKind;
method: string;
turnId: string;
itemId: string;
status: "pending";
prompt: string;
details: Record<string, unknown>;
}
export type HarnessRuntimeRequestResolution =
| { action: "accept" | "accept_for_session" | "decline" | "cancel" }
| {
action: "submit";
answers: Record<string, { answers: string[] }>;
}
| {
action: "submit";
content: Record<string, unknown>;
};
export interface HarnessThreadGoal {
threadId: string;
objective: string;
status: "active" | "paused" | "blocked" | "usageLimited" | "budgetLimited" | "complete";
tokenBudget: number | null;
tokensUsed: number;
timeUsedSeconds: number;
createdAt: number;
updatedAt: number;
}
export type HarnessGoalOperation =
| { action: "get" }
| { action: "set"; objective: string; tokenBudget?: number | null }
| { action: "pause" | "resume" | "clear" };
export interface HarnessThreadLineageEntry {
threadId: string;
providerSessionId: string | null;
parentThreadId: string | null;
depth: number;
nickname: string | null;
role: string | null;
status: string;
}
export interface PersistedHarnessSemanticResult {
result: PrpStructuredRunResult;
fingerprint: string;
@@ -58,6 +131,9 @@ export interface PersistedHarnessSession {
activeTurnId?: string | null;
semanticResult?: PersistedHarnessSemanticResult | null;
terminalTurns?: PersistedHarnessTurnTerminal[];
pendingRuntimeRequests?: HarnessRuntimeRequest[];
goal?: HarnessThreadGoal | null;
lineage?: HarnessThreadLineageEntry[];
lastSourceSequence?: number;
}
@@ -76,7 +152,15 @@ export interface HarnessSession {
events(): AsyncIterable<PrpEvent>;
startTurn(input: { message: NativeUserMessage }): Promise<{ turnId: string }>;
steer?(input: { turnId: string; message: NativeUserMessage }): Promise<void>;
interrupt?(input: { turnId: string; reason?: string }): Promise<void>;
interrupt?(input: { turnId?: string; reason?: string }): Promise<void>;
pendingRuntimeRequests?(): HarnessRuntimeRequest[];
resolveRuntimeRequest?(input: {
requestId: string;
turnId: string;
resolution: HarnessRuntimeRequestResolution;
}): Promise<void>;
goal?(input: HarnessGoalOperation): Promise<HarnessThreadGoal | null>;
lineage?(): HarnessThreadLineageEntry[];
read?(): Promise<Record<string, unknown>>;
reconcile?(): Promise<Record<string, unknown>>;
usage?(): Promise<Record<string, unknown> | null>;
@@ -37,7 +37,7 @@ export interface NativeSession {
events(input?: { afterCursor?: string | null }): AsyncIterable<NativeRunEvent>;
startTurn(input: { message: NativeUserMessage }): Promise<{ turnId: string }>;
steer?(input: { turnId: string; message: NativeUserMessage }): Promise<void>;
interrupt?(input: { turnId: string; reason?: string }): Promise<void>;
interrupt?(input: { turnId?: string; reason?: string }): Promise<void>;
cancel?(input: { reason: string }): Promise<void>;
result(): Promise<NativeRunResult | null>;
snapshot(): Promise<PersistedNativeSession>;
@@ -50,6 +50,11 @@ export interface Phase4ModelContextSnapshot {
environmentKeys: string[];
dynamicToolNames: string[];
modelInputKinds: ["text"];
phase4b?: {
runtimeRequestResolution: boolean;
goals: boolean;
threadLineage: boolean;
};
envelope: Phase4TaskEnvelope;
}
@@ -33,6 +33,9 @@ export interface NativeSessionCapabilities {
reconciliation?: boolean;
usage?: boolean;
dynamicTools?: boolean;
runtimeRequestResolution?: boolean;
goals?: boolean;
threadLineage?: boolean;
unsupported?: string[];
}
@@ -1,4 +1,5 @@
import { describe, expect, it } from "vitest";
import { fileURLToPath } from "node:url";
import {
PHASE4_BLOCK_RESULT_OUTPUT_SCHEMA,
@@ -8,7 +9,10 @@ import {
} from "../../contracts/phase4.js";
import {
HarnessCapabilityUnavailableError,
HarnessOperationAlreadyTerminalError,
HarnessReconciliationError,
HarnessStaleTurnError,
type HarnessRuntimeRequestResolution,
} from "../../contracts/harness-driver.js";
import {
applyPrpEvent,
@@ -20,6 +24,7 @@ import {
type PrpEvent,
type PrpStructuredRunResult,
} from "../../protocol/phase1-contract.js";
import { loadPhase4bConformanceFixture } from "../../protocol/phase4b-fixture.js";
import {
CodexAppServerDriver,
createIsolatedCodexAppServerArgs,
@@ -71,6 +76,8 @@ class FakeCodexTransport implements CodexAppServerTransport {
handler: CodexServerRequestHandler = async () => ({});
rejectMethods = new Map<string, Error>();
readResponse: Record<string, unknown> | null = null;
turnStartResponse: Promise<Record<string, unknown>> | null = null;
goalState: Record<string, unknown> | null = null;
constructor(
readonly threadId = "thread-1",
@@ -102,7 +109,29 @@ class FakeCodexTransport implements CodexAppServerTransport {
instructionSources: [],
};
}
if (method === "turn/start") return { turn: { id: "turn-1", status: "inProgress", items: [] } };
if (method === "turn/start") {
return this.turnStartResponse ?? { turn: { id: "turn-1", status: "inProgress", items: [] } };
}
if (method === "thread/goal/get") return { goal: this.goalState };
if (method === "thread/goal/set") {
this.goalState = {
threadId: this.threadId,
objective: typeof params.objective === "string"
? params.objective
: String(this.goalState?.objective ?? "Ship the Phase 4b tracer"),
status: params.status ?? this.goalState?.status ?? "active",
tokenBudget: params.tokenBudget ?? this.goalState?.tokenBudget ?? null,
tokensUsed: 0,
timeUsedSeconds: 0,
createdAt: 1,
updatedAt: 2,
};
return { goal: this.goalState };
}
if (method === "thread/goal/clear") {
this.goalState = null;
return {};
}
if (method === "thread/read") {
return this.readResponse ?? { thread: { id: this.threadId, sessionId: this.providerSessionId, cwd: "/workspace", turns: [{ id: "turn-1", status: "inProgress", items: [] }] } };
}
@@ -139,6 +168,11 @@ const envelope = createPhase4TaskEnvelope({
criteria: [{ id: "file", requirement: "hello.txt contains hello" }],
});
const phase4bFixturePath = fileURLToPath(new URL(
"../../../protocol/fixtures/phase-04b/driver-conformance.json",
import.meta.url,
));
const result: PrpStructuredRunResult = {
schema: "paperclip.run_result.v1",
reportedWorkDisposition: "done",
@@ -246,11 +280,19 @@ describe("Codex app-server Phase 4 driver", () => {
transport.push("item/started", { threadId: "thread-1", turnId: turn.turnId, item: { id: "cmd-1", type: "commandExecution", command: "printf hello" } });
transport.push("item/commandExecution/outputDelta", { threadId: "thread-1", turnId: turn.turnId, itemId: "cmd-1", delta: "hello" });
transport.push("item/completed", { threadId: "thread-1", turnId: turn.turnId, item: { id: "file-1", type: "fileChange", changes: [{ path: "hello.txt" }] } });
expect(await transport.invoke({
const requestResolution = transport.invoke({
id: "request-1",
method: "item/tool/requestUserInput",
params: { threadId: "thread-1", turnId: turn.turnId, itemId: "question-1" },
})).toEqual({ answers: {} });
});
await new Promise((resolve) => setTimeout(resolve, 0));
expect(session.pendingRuntimeRequests?.()).toHaveLength(1);
await session.resolveRuntimeRequest?.({
requestId: "request-1",
turnId: turn.turnId,
resolution: { action: "cancel" },
});
expect(await requestResolution).toEqual({ answers: {} });
transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: turn.turnId, tokenUsage: { total: { inputTokens: 10, outputTokens: 4 }, modelContextWindow: 128000 } });
transport.push("item/completed", { threadId: "thread-1", turnId: turn.turnId, item: { id: "answer-1", type: "agentMessage", text: JSON.stringify(result) } });
transport.push("turn/completed", { threadId: "thread-1", turn: { id: turn.turnId, status: "completed", items: [] } });
@@ -383,11 +425,224 @@ describe("Codex app-server Phase 4 driver", () => {
await session.steer?.({ turnId, message: { role: "user", text: "Use a shorter answer." } });
await session.interrupt?.({ turnId, reason: "operator requested" });
expect(transport.calls.map((call) => call.method)).toEqual([
"initialize", "thread/start", "turn/start", "turn/steer", "turn/interrupt",
"initialize", "thread/start", "thread/goal/get", "turn/start", "turn/steer", "turn/interrupt",
]);
expect(session.ids()).toEqual({ driverSessionId: "thread-1", providerSessionId: "provider-session-1", displayId: "thread-1" });
});
it("loads the deterministic Phase 4b wire fixture", async () => {
const fixture = await loadPhase4bConformanceFixture(phase4bFixturePath);
expect(fixture.runtimeRequests.map(({ requestKind }) => requestKind)).toEqual([
"command_approval",
"file_approval",
"permission_approval",
"user_input",
"elicitation",
]);
expect(fixture.goals.map(({ action }) => action)).toEqual([
"get", "set", "pause", "resume", "clear",
]);
});
it("holds browser-resolved upstream requests and returns the exact fixture responses", async () => {
const fixture = await loadPhase4bConformanceFixture(phase4bFixturePath);
for (const scenario of fixture.runtimeRequests) {
const transport = new FakeCodexTransport();
const session = await makeDriver([transport]).openSession({
runId: `run-${scenario.id}`,
normalizedSessionId: `normalized-${scenario.id}`,
workingDirectory: "/workspace",
});
const { turnId } = await session.startTurn({ message: { role: "user", text: "Exercise request." } });
const response = transport.invoke({
id: scenario.id,
method: scenario.method,
params: {
threadId: "thread-1",
turnId,
itemId: `item-${scenario.id}`,
reason: `fixture ${scenario.id}`,
},
});
await new Promise((resolve) => setTimeout(resolve, 0));
expect(session.pendingRuntimeRequests?.()).toEqual([
expect.objectContaining({
requestId: scenario.id,
requestKind: scenario.requestKind,
method: scenario.method,
turnId,
}),
]);
await expect(session.resolveRuntimeRequest?.({
requestId: scenario.id,
turnId,
resolution: { action: "forged" } as unknown as HarnessRuntimeRequestResolution,
})).rejects.toThrow("unsupported runtime request resolution");
expect(session.pendingRuntimeRequests?.()).toHaveLength(1);
await session.resolveRuntimeRequest?.({
requestId: scenario.id,
turnId,
resolution: scenario.resolution as HarnessRuntimeRequestResolution,
});
expect(await response).toEqual(scenario.expectedResponse);
expect(session.pendingRuntimeRequests?.()).toEqual([]);
await expect(session.resolveRuntimeRequest?.({
requestId: scenario.id,
turnId,
resolution: { action: "cancel" },
})).rejects.toBeInstanceOf(HarnessCapabilityUnavailableError);
await session.close({ reason: "fixture complete" });
}
});
it("acknowledges same-turn steering and rejects stale or child steering", async () => {
const fixture = await loadPhase4bConformanceFixture(phase4bFixturePath);
const transport = new FakeCodexTransport("thread-root");
const session = await makeDriver([transport]).openSession({
runId: "run-phase4b-controls",
normalizedSessionId: "normalized-phase4b-controls",
workingDirectory: "/workspace",
});
const { turnId } = await session.startTurn({ message: { role: "user", text: "Start." } });
await session.steer?.({ turnId, message: { role: "user", text: "Stay concise." } });
await expect(session.steer?.({
turnId: fixture.controls.staleTurnSteer.turnId!,
message: { role: "user", text: "Stale." },
})).rejects.toBeInstanceOf(HarnessStaleTurnError);
transport.push("thread/started", { thread: fixture.lineage.childThread });
await new Promise((resolve) => setTimeout(resolve, 0));
expect(session.lineage?.()).toEqual(expect.arrayContaining([
expect.objectContaining({ threadId: "thread-root", parentThreadId: null, depth: 0 }),
expect.objectContaining({
threadId: "thread-child",
parentThreadId: "thread-root",
depth: 1,
nickname: "Scout",
role: "researcher",
}),
]));
await expect(session.steer?.({
turnId: "thread-child",
message: { role: "user", text: "Do not emulate child steering." },
})).rejects.toBeInstanceOf(HarnessStaleTurnError);
const events = session.events()[Symbol.asyncIterator]();
const observed: PrpEvent[] = [];
for (let index = 0; index < 9; index += 1) {
const next = await events.next();
if (next.done) break;
observed.push(next.value);
}
expect(observed).toEqual(expect.arrayContaining([
expect.objectContaining({
eventType: "item.completed",
payload: expect.objectContaining({ kind: "steering_acknowledgement", status: "acknowledged" }),
}),
expect.objectContaining({
eventType: "harness.diagnostic",
payload: expect.objectContaining({ code: "stale_turn_rejected" }),
}),
expect.objectContaining({
eventType: "item.started",
payload: expect.objectContaining({ kind: "thread_lineage" }),
}),
]));
await session.close({ reason: "fixture complete" });
});
it("queues an interrupt before turn identity and reports a terminal race precisely", async () => {
const transport = new FakeCodexTransport();
let releaseStart!: (value: Record<string, unknown>) => void;
transport.turnStartResponse = new Promise((resolve) => { releaseStart = resolve; });
const session = await makeDriver([transport]).openSession({
runId: "run-interrupt-races",
normalizedSessionId: "normalized-interrupt-races",
workingDirectory: "/workspace",
});
const starting = session.startTurn({ message: { role: "user", text: "Start slowly." } });
await new Promise((resolve) => setTimeout(resolve, 0));
await session.interrupt?.({ reason: "operator" });
releaseStart({ turn: { id: "turn-1", status: "inProgress", items: [] } });
await expect(starting).resolves.toEqual({ turnId: "turn-1" });
expect(transport.calls.map(({ method }) => method)).toEqual(expect.arrayContaining([
"turn/start", "turn/interrupt",
]));
transport.push("turn/completed", {
threadId: "thread-1",
turn: { id: "turn-1", status: "interrupted", items: [] },
});
await new Promise((resolve) => setTimeout(resolve, 0));
await expect(session.interrupt?.({ turnId: "turn-1" }))
.rejects.toBeInstanceOf(HarnessOperationAlreadyTerminalError);
await session.close({ reason: "fixture complete" });
});
it("maps all goal operations and advertises an exact unsupported state", async () => {
const fixture = await loadPhase4bConformanceFixture(phase4bFixturePath);
const transport = new FakeCodexTransport();
const session = await makeDriver([transport]).openSession({
runId: "run-goals",
normalizedSessionId: "normalized-goals",
workingDirectory: "/workspace",
});
await session.goal?.({ action: "get" });
await session.goal?.({ action: "set", objective: "Ship the Phase 4b tracer", tokenBudget: 4096 });
await session.goal?.({ action: "pause" });
await session.goal?.({ action: "resume" });
await session.goal?.({ action: "clear" });
const goalCalls = transport.calls.filter(({ method }) => method.startsWith("thread/goal/"));
expect(goalCalls.slice(1).map(({ method, params }) => ({
method,
params: Object.fromEntries(Object.entries(params).filter(([key]) => key !== "threadId")),
}))).toEqual(fixture.goals.map(({ method, params }) => ({ method, params })));
const unsupportedTransport = new FakeCodexTransport();
unsupportedTransport.rejectMethods.set("thread/goal/get", new Error("method not found"));
const unsupportedDriver = makeDriver([unsupportedTransport]);
const unsupported = await unsupportedDriver.openSession({
runId: "run-goals-unsupported",
normalizedSessionId: "normalized-goals-unsupported",
workingDirectory: "/workspace",
});
expect((await unsupportedDriver.descriptor()).capabilities).toMatchObject({ goals: false });
await expect(unsupported.goal?.({ action: "get" }))
.rejects.toBeInstanceOf(HarnessCapabilityUnavailableError);
await session.close({ reason: "fixture complete" });
await unsupported.close({ reason: "fixture complete" });
});
it("redacts browser-visible request details and diagnostics from the fixture markers", async () => {
const fixture = await loadPhase4bConformanceFixture(phase4bFixturePath);
const transport = new FakeCodexTransport();
const session = await makeDriver([transport]).openSession({
runId: "run-redaction",
normalizedSessionId: "normalized-redaction",
workingDirectory: "/workspace",
});
const { turnId } = await session.startTurn({ message: { role: "user", text: "Request input." } });
const pending = transport.invoke({
id: "redacted-request",
method: "item/tool/requestUserInput",
params: {
threadId: "thread-1",
turnId,
itemId: "redacted-item",
reason: fixture.redactionMarkers.join(" "),
authorization: "Bearer browser-secret",
},
});
await new Promise((resolve) => setTimeout(resolve, 0));
const serialized = JSON.stringify(session.pendingRuntimeRequests?.());
for (const marker of fixture.redactionMarkers) expect(serialized).not.toContain(marker);
expect(serialized).toContain("[REDACTED]");
await session.resolveRuntimeRequest?.({
requestId: "redacted-request",
turnId,
resolution: { action: "cancel" },
});
await pending;
await session.close({ reason: "fixture complete" });
});
it("makes duplicate semantic completion idempotent and rejects changed payloads", async () => {
const transport = new FakeCodexTransport();
const session = await makeDriver([transport]).openSession({ runId: "run-result", normalizedSessionId: "normalized-result", workingDirectory: "/workspace" });
@@ -759,7 +1014,9 @@ describe("Codex app-server Phase 4 driver", () => {
expect(recovery).toMatchObject({ recovered: true });
expect(recovery?.session?.ids()).toEqual(original.ids());
expect(await recovery?.session?.reconcile?.()).toMatchObject({ thread: { id: "thread-1" } });
expect(second.calls.map((call) => call.method)).toEqual(["initialize", "thread/read", "thread/resume", "thread/read"]);
expect(second.calls.map((call) => call.method)).toEqual([
"initialize", "thread/read", "thread/resume", "thread/goal/get", "thread/read",
]);
expect((await recovery?.session?.snapshot())?.activeTurnId).toBe("turn-1");
});
@@ -1076,7 +1333,7 @@ describe("Codex app-server Phase 4 driver", () => {
await expect(session.steer?.({ turnId, message: { role: "user", text: "Steer." } })).rejects.toBeInstanceOf(HarnessCapabilityUnavailableError);
const iterator = session.events()[Symbol.asyncIterator]();
const events: PrpEvent[] = [];
for (let index = 0; index < 5; index += 1) {
for (let index = 0; index < 6; index += 1) {
const next = await iterator.next();
if (next.value) events.push(next.value);
}
@@ -5,6 +5,12 @@ import type {
HarnessDriverDescriptor,
HarnessSession,
HarnessSessionRecoveryResult,
HarnessGoalOperation,
HarnessRuntimeRequest,
HarnessRuntimeRequestKind,
HarnessRuntimeRequestResolution,
HarnessThreadGoal,
HarnessThreadLineageEntry,
OpenHarnessSessionInput,
PersistedHarnessSession,
PersistedHarnessSemanticResult,
@@ -12,7 +18,9 @@ import type {
} from "../../contracts/harness-driver.js";
import {
HarnessCapabilityUnavailableError,
HarnessOperationAlreadyTerminalError,
HarnessReconciliationError,
HarnessStaleTurnError,
} from "../../contracts/harness-driver.js";
import {
PHASE4_BLOCK_RESULT_OUTPUT_SCHEMA,
@@ -91,6 +99,9 @@ export interface CodexAppServerDriverOptions {
usage: boolean;
reconciliation: boolean;
dynamicTools: boolean;
runtimeRequestResolution: boolean;
goals: boolean;
threadLineage: boolean;
}>;
}
@@ -109,6 +120,12 @@ interface OpenedCodexThread {
threadId: string;
providerSessionId: string | null;
context: Phase4ModelContextSnapshot;
lineage: HarnessThreadLineageEntry;
}
interface PendingRuntimeRequest {
request: HarnessRuntimeRequest;
settle: (response: Record<string, unknown>) => void;
}
function record(value: unknown): Record<string, unknown> {
@@ -281,6 +298,9 @@ export class CodexAppServerDriver implements HarnessDriver {
usage: true,
reconciliation: true,
dynamicTools: true,
runtimeRequestResolution: true,
goals: true,
threadLineage: true,
...options.capabilities,
};
if (!this.#caps.read) this.#caps.reconciliation = false;
@@ -305,6 +325,9 @@ export class CodexAppServerDriver implements HarnessDriver {
reconciliation: this.#caps.reconciliation,
usage: this.#caps.usage,
dynamicTools: this.#caps.dynamicTools,
runtimeRequestResolution: this.#caps.runtimeRequestResolution,
goals: this.#caps.goals,
threadLineage: this.#caps.threadLineage,
unsupported,
},
};
@@ -324,11 +347,14 @@ export class CodexAppServerDriver implements HarnessDriver {
persistExtendedHistory: false,
});
const opened = this.#openedThread(response, initialize, workingDirectory);
const goal = await this.#discoverGoal(transport, opened.threadId);
if (opened.context.phase4b) opened.context.phase4b.goals = this.#caps.goals;
return this.#session({
transport,
runId: input.runId,
normalizedSessionId: input.normalizedSessionId,
opened,
goal,
resumed: false,
sourceSequence: 0,
});
@@ -384,6 +410,8 @@ export class CodexAppServerDriver implements HarnessDriver {
await transport.close();
return { recovered: false, reason: "provider resumed a different provider session" };
}
const goal = await this.#discoverGoal(transport, opened.threadId);
if (opened.context.phase4b) opened.context.phase4b.goals = this.#caps.goals;
return {
recovered: true,
session: this.#session({
@@ -391,10 +419,13 @@ export class CodexAppServerDriver implements HarnessDriver {
runId: snapshot.runId,
normalizedSessionId: snapshot.normalizedSessionId,
opened,
goal,
resumed: true,
activeTurnId: snapshot.activeTurnId ?? null,
semanticResult: snapshot.semanticResult ?? null,
terminalTurns: snapshot.terminalTurns ?? [],
stalePendingRuntimeRequests: snapshot.pendingRuntimeRequests ?? [],
lineage: snapshot.lineage,
sourceSequence: snapshot.lastSourceSequence ?? 0,
}),
};
@@ -424,6 +455,23 @@ export class CodexAppServerDriver implements HarnessDriver {
return initialized;
}
async #discoverGoal(
transport: CodexAppServerTransport,
threadId: string,
): Promise<HarnessThreadGoal | null | undefined> {
if (!this.#caps.goals) return undefined;
try {
const response = await transport.request("thread/goal/get", { threadId });
return parseThreadGoal(response.goal);
} catch (error) {
this.#caps.goals = false;
this.#options.onDiagnostic?.(
redactCodexDiagnostic(`thread goals unavailable: ${String(error)}`),
);
return undefined;
}
}
#openedThread(
response: Record<string, unknown>,
initialize: Record<string, unknown>,
@@ -488,8 +536,14 @@ export class CodexAppServerDriver implements HarnessDriver {
).sort(),
dynamicToolNames: this.#caps.dynamicTools ? [...PHASE4_SEMANTIC_TOOL_NAMES] : [],
modelInputKinds: ["text"],
phase4b: {
runtimeRequestResolution: this.#caps.runtimeRequestResolution,
goals: this.#caps.goals,
threadLineage: this.#caps.threadLineage,
},
envelope: structuredClone(this.#options.taskEnvelope),
},
lineage: lineageFromThread(thread),
};
}
@@ -498,10 +552,13 @@ export class CodexAppServerDriver implements HarnessDriver {
runId: string;
normalizedSessionId: string;
opened: OpenedCodexThread;
goal?: HarnessThreadGoal | null;
resumed: boolean;
activeTurnId?: string | null;
semanticResult?: PersistedHarnessSemanticResult | null;
terminalTurns?: PersistedHarnessTurnTerminal[];
stalePendingRuntimeRequests?: HarnessRuntimeRequest[];
lineage?: HarnessThreadLineageEntry[];
sourceSequence: number;
}): CodexHarnessSession {
return new CodexHarnessSession({
@@ -538,6 +595,12 @@ class CodexHarnessSession implements HarnessSession {
#terminal = false;
#turnStarted = false;
readonly #terminalTurns = new Map<string, string>();
readonly #pendingRuntimeRequests = new Map<string, PendingRuntimeRequest>();
readonly #lineage = new Map<string, HarnessThreadLineageEntry>();
#goal: HarnessThreadGoal | null = null;
#interruptQueued = false;
#steerSequence = 0;
#interruptSequence = 0;
constructor(input: {
transport: CodexAppServerTransport;
@@ -549,6 +612,9 @@ class CodexHarnessSession implements HarnessSession {
activeTurnId?: string | null;
semanticResult?: PersistedHarnessSemanticResult | null;
terminalTurns?: PersistedHarnessTurnTerminal[];
stalePendingRuntimeRequests?: HarnessRuntimeRequest[];
lineage?: HarnessThreadLineageEntry[];
goal?: HarnessThreadGoal | null;
sourceSequence: number;
now: () => Date;
runnerInstanceId: string;
@@ -564,6 +630,13 @@ class CodexHarnessSession implements HarnessSession {
this.#now = input.now;
this.#runnerInstanceId = input.runnerInstanceId;
this.#capabilities = input.capabilities;
this.#goal = input.goal === undefined ? null : structuredClone(input.goal);
for (const entry of input.lineage ?? [input.opened.lineage]) {
this.#lineage.set(entry.threadId, structuredClone(entry));
}
if (!this.#lineage.has(input.opened.lineage.threadId)) {
this.#lineage.set(input.opened.lineage.threadId, structuredClone(input.opened.lineage));
}
if (input.semanticResult) {
const validation = validatePrpStructuredRunResult(input.semanticResult.result);
if (!validation.ok || canonicalJson(validation.result) !== input.semanticResult.fingerprint) {
@@ -587,6 +660,17 @@ class CodexHarnessSession implements HarnessSession {
providerSessionId: input.opened.providerSessionId,
context: input.opened.context,
});
for (const stale of input.stalePendingRuntimeRequests ?? []) {
this.#emit("runtime_request.cancelled", {
requestId: stale.requestId,
reason: "transport_recovered",
}, { turnId: stale.turnId, itemId: stale.itemId });
}
this.#emit("item.completed", {
kind: "thread_lineage",
text: `Root thread ${input.opened.threadId}`,
lineage: input.opened.lineage,
}, { itemId: `thread:${input.opened.threadId}` });
this.#emit("item.completed", {
kind: "model",
text: `${input.opened.context.model} (${input.opened.context.modelProvider})`,
@@ -649,36 +733,159 @@ class CodexHarnessSession implements HarnessSession {
}
this.#activeTurnId ??= turnId;
this.#emit("turn.accepted", { turnId }, { turnId });
if (this.#interruptQueued) {
this.#interruptQueued = false;
await this.#sendInterrupt(turnId, "queued_before_start");
}
return { turnId };
}
async steer(input: { turnId: string; message: NativeUserMessage }): Promise<void> {
this.#requireCapability("steering");
this.#requireActiveTurn(input.turnId);
this.#requireActiveTurn(input.turnId, "steering");
try {
await this.#transport.request("turn/steer", {
threadId: this.#opened.threadId,
input: [userInput(input.message)],
expectedTurnId: input.turnId,
});
if (this.#activeTurnId !== input.turnId) {
throw new HarnessOperationAlreadyTerminalError("steering");
}
this.#emit("item.completed", {
kind: "steering_acknowledgement",
text: "Steering acknowledged for the active turn.",
status: "acknowledged",
}, {
turnId: input.turnId,
itemId: `${input.turnId}:steer:${++this.#steerSequence}`,
});
} catch (error) {
if (error instanceof HarnessOperationAlreadyTerminalError) throw error;
throw this.#unsupported("steering", error);
}
}
async interrupt(input: { turnId: string; reason?: string }): Promise<void> {
async interrupt(input: { turnId?: string; reason?: string }): Promise<void> {
this.#requireCapability("interruption");
this.#requireActiveTurn(input.turnId);
if (this.#turnStartPending && this.#activeTurnId === null) {
this.#interruptQueued = true;
this.#emit("item.completed", {
kind: "interrupt_acknowledgement",
text: "Interrupt queued until the provider assigns the turn identity.",
status: "queued",
}, { itemId: `interrupt:queued:${++this.#interruptSequence}` });
return;
}
if (this.#terminal || this.#activeTurnId === null) {
throw new HarnessOperationAlreadyTerminalError("interruption");
}
const turnId = input.turnId ?? this.#activeTurnId;
this.#requireActiveTurn(turnId, "interruption");
await this.#sendInterrupt(turnId, input.reason);
}
async #sendInterrupt(turnId: string, reason?: string): Promise<void> {
try {
await this.#transport.request("turn/interrupt", {
threadId: this.#opened.threadId,
turnId: input.turnId,
turnId,
});
if (this.#activeTurnId !== turnId) {
throw new HarnessOperationAlreadyTerminalError("interruption");
}
this.#emit("item.completed", {
kind: "interrupt_acknowledgement",
text: "Interrupt accepted for the active turn.",
status: "acknowledged",
reason: boundedText(reason),
}, {
turnId,
itemId: `${turnId}:interrupt:${++this.#interruptSequence}`,
});
} catch (error) {
if (error instanceof HarnessOperationAlreadyTerminalError) throw error;
throw this.#unsupported("interruption", error);
}
}
pendingRuntimeRequests(): HarnessRuntimeRequest[] {
return [...this.#pendingRuntimeRequests.values()].map(({ request }) =>
structuredClone(request));
}
async resolveRuntimeRequest(input: {
requestId: string;
turnId: string;
resolution: HarnessRuntimeRequestResolution;
}): Promise<void> {
this.#requireCapability("runtimeRequestResolution");
const pending = this.#pendingRuntimeRequests.get(input.requestId);
if (pending === undefined) {
throw new HarnessCapabilityUnavailableError(
"runtime request resolution",
`request ${input.requestId} is no longer pending`,
);
}
if (pending.request.turnId !== input.turnId || this.#activeTurnId !== input.turnId) {
throw new HarnessStaleTurnError(input.turnId);
}
const response = runtimeRequestResponse(pending.request, input.resolution);
this.#pendingRuntimeRequests.delete(input.requestId);
this.#emit("runtime_request.resolved", {
requestId: input.requestId,
resolution: input.resolution.action,
}, { turnId: input.turnId, itemId: pending.request.itemId });
pending.settle(response);
}
async goal(input: HarnessGoalOperation): Promise<HarnessThreadGoal | null> {
this.#requireCapability("goals");
let method: string;
let params: Record<string, unknown> = { threadId: this.#opened.threadId };
if (input.action === "get") {
method = "thread/goal/get";
} else if (input.action === "clear") {
method = "thread/goal/clear";
} else {
method = "thread/goal/set";
if (input.action === "set") {
params = {
...params,
objective: input.objective,
status: "active",
...(input.tokenBudget !== undefined ? { tokenBudget: input.tokenBudget } : {}),
};
} else {
params = { ...params, status: input.action === "pause" ? "paused" : "active" };
}
}
try {
const response = await this.#transport.request(method, params);
const goal = input.action === "clear" ? null : parseThreadGoal(response.goal);
if (!["get", "clear"].includes(input.action) && goal === null) {
throw new Error(`${method} omitted its goal`);
}
this.#goal = goal;
const kind = input.action === "clear" ? "goal_cleared" : "goal";
this.#emit("item.completed", {
kind,
text: goal === null
? input.action === "clear" ? "Thread goal cleared." : "No thread goal is configured."
: `Thread goal ${goal.status}: ${goal.objective}`,
action: input.action,
goal,
}, { itemId: `${this.#opened.threadId}:goal:${this.#sourceSequence + 1}` });
return goal === null ? null : structuredClone(goal);
} catch (error) {
throw this.#unsupported(`goal ${input.action}`, error);
}
}
lineage(): HarnessThreadLineageEntry[] {
return [...this.#lineage.values()].map((entry) => structuredClone(entry));
}
async read(): Promise<Record<string, unknown>> {
this.#requireCapability("read");
try {
@@ -787,11 +994,15 @@ class CodexHarnessSession implements HarnessSession {
turnId,
fingerprint,
})),
pendingRuntimeRequests: this.pendingRuntimeRequests(),
goal: this.#goal === null ? null : structuredClone(this.#goal),
lineage: this.lineage(),
lastSourceSequence: this.#sourceSequence,
};
}
async close(): Promise<void> {
this.#cancelPendingRequests("session_closed");
this.#events.close();
await this.#transport.close();
}
@@ -828,6 +1039,84 @@ class CodexHarnessSession implements HarnessSession {
});
return;
}
if (notification.method === "thread/started") {
if (!this.#capabilities.threadLineage) return;
const thread = record(params.thread);
const lineage = lineageFromThread(thread);
if (
lineage.threadId.length === 0 ||
lineage.threadId === this.#opened.threadId ||
lineage.parentThreadId === null ||
!this.#lineage.has(lineage.parentThreadId)
) {
return;
}
this.#lineage.set(lineage.threadId, lineage);
this.#emit("item.started", {
kind: "thread_lineage",
text: `${lineage.nickname ?? lineage.role ?? "Child agent"} started.`,
lineage,
}, { itemId: `thread:${lineage.threadId}` });
return;
}
if (notification.method === "thread/status/changed") {
const lineage = this.#lineage.get(threadId);
if (lineage === undefined || threadId === this.#opened.threadId) return;
lineage.status = threadStatus(params.status);
this.#emit("item.delta", {
kind: "thread_lineage",
text: `${lineage.nickname ?? lineage.role ?? "Child agent"}: ${lineage.status}`,
lineage,
}, { itemId: `thread:${lineage.threadId}` });
return;
}
if (notification.method === "thread/closed") {
const lineage = this.#lineage.get(threadId);
if (lineage === undefined || threadId === this.#opened.threadId) return;
lineage.status = "closed";
this.#emit("item.completed", {
kind: "thread_lineage",
text: `${lineage.nickname ?? lineage.role ?? "Child agent"} closed.`,
lineage,
}, { itemId: `thread:${lineage.threadId}` });
return;
}
if (notification.method === "thread/goal/updated") {
if (threadId !== this.#opened.threadId) return;
const goal = parseThreadGoal(params.goal);
if (goal === null) return;
this.#goal = goal;
this.#emit("item.completed", {
kind: "goal",
text: `Thread goal ${goal.status}: ${goal.objective}`,
action: "notification",
goal,
}, { turnId: turnId || undefined, itemId: `${threadId}:goal:update:${this.#sourceSequence + 1}` });
return;
}
if (notification.method === "thread/goal/cleared") {
if (threadId !== this.#opened.threadId) return;
this.#goal = null;
this.#emit("item.completed", {
kind: "goal_cleared",
text: "Thread goal cleared.",
action: "notification",
goal: null,
}, { itemId: `${threadId}:goal:clear:${this.#sourceSequence + 1}` });
return;
}
if (notification.method === "serverRequest/resolved") {
const requestId = String(params.requestId ?? "");
const pending = this.#pendingRuntimeRequests.get(requestId);
if (pending === undefined || threadId !== this.#opened.threadId) return;
this.#pendingRuntimeRequests.delete(requestId);
this.#emit("runtime_request.cancelled", {
requestId,
reason: "provider_resolved",
}, { turnId: pending.request.turnId, itemId: pending.request.itemId });
pending.settle(safeRequestResponse(pending.request.method, "cancel"));
return;
}
if (threadId !== this.#opened.threadId) {
this.#failProtocol(
"thread_binding_mismatch",
@@ -962,7 +1251,8 @@ class CodexHarnessSession implements HarnessSession {
return { success: true, contentItems: [{ type: "inputText", text: "Semantic completion accepted." }] };
}
if (!isTurnScopedServerRequest(request.method)) return safeRequestResponse(request.method);
const requestKind = runtimeRequestKind(request.method);
if (requestKind === null) return safeRequestResponse(request.method);
const requestTurnId = text(request.params.turnId);
if (
text(request.params.threadId) !== this.#opened.threadId ||
@@ -974,30 +1264,30 @@ class CodexHarnessSession implements HarnessSession {
this.#failProtocol("runtime_request_binding_mismatch", "Runtime request did not name the active thread and turn.");
return safeRequestResponse(request.method);
}
const requestKind = request.method.includes("requestApproval") ? "approval" : "user_input";
const requestId = String(request.id);
this.#emit("runtime_request.created", {
request: {
requestId,
requestKind,
type: request.method,
status: "pending",
prompt: "Codex requires an external runtime decision.",
details: redactCodexValue(boundedCodexValue(request.params)),
},
}, {
turnId: requestTurnId,
itemId: text(request.params.itemId, requestId),
});
const response = safeRequestResponse(request.method);
this.#emit("runtime_request.resolved", {
if (this.#pendingRuntimeRequests.has(requestId)) {
this.#failProtocol("runtime_request_duplicate", "Provider reused a pending runtime request identity.");
return safeRequestResponse(request.method);
}
const runtimeRequest: HarnessRuntimeRequest = {
requestId,
resolution: "declined_by_skillless_driver",
}, {
requestKind,
method: request.method,
turnId: requestTurnId,
itemId: text(request.params.itemId, requestId),
status: "pending",
prompt: runtimeRequestPrompt(requestKind, request.params),
details: record(redactCodexValue(boundedCodexValue(request.params))),
};
this.#emit("runtime_request.created", {
request: { ...runtimeRequest, type: runtimeRequest.method },
}, {
turnId: requestTurnId,
itemId: runtimeRequest.itemId,
});
return new Promise<Record<string, unknown>>((settle) => {
this.#pendingRuntimeRequests.set(requestId, { request: runtimeRequest, settle });
});
return response;
}
#captureResultFromItem(item: Record<string, unknown>, turnId: string): boolean {
@@ -1106,6 +1396,7 @@ class CodexHarnessSession implements HarnessSession {
: status === "cancelled"
? "turn.cancelled"
: "turn.completed";
this.#cancelPendingRequests("turn_terminal");
this.#emit(eventType, boundedPayload({
status,
error: turn.error ?? null,
@@ -1173,12 +1464,30 @@ class CodexHarnessSession implements HarnessSession {
return item === null ? null : tryParseResult(item.text);
}
#requireActiveTurn(turnId: string): void {
#requireActiveTurn(turnId: string, operation: string): void {
if (this.#activeTurnId !== turnId) {
throw this.#unsupported("active turn operation", "turn is no longer active");
this.#emit("harness.diagnostic", {
code: "stale_turn_rejected",
operation,
turnId,
activeTurnId: this.#activeTurnId,
message: `Rejected ${operation} for a stale turn identity.`,
});
throw new HarnessStaleTurnError(turnId);
}
}
#cancelPendingRequests(reason: string): void {
for (const [requestId, pending] of this.#pendingRuntimeRequests) {
this.#emit("runtime_request.cancelled", { requestId, reason }, {
turnId: pending.request.turnId,
itemId: pending.request.itemId,
});
pending.settle(safeRequestResponse(pending.request.method, "cancel"));
}
this.#pendingRuntimeRequests.clear();
}
#notificationNamesActiveTurn(turnId: string, kind: string): boolean {
if (
this.#protocolFailed ||
@@ -1219,6 +1528,7 @@ class CodexHarnessSession implements HarnessSession {
this.#protocolFailed = true;
this.#protocolFailureCode = code;
this.#protocolFailureMessage = redactCodexDiagnostic(message);
this.#cancelPendingRequests("protocol_failed");
this.#emit("session.failed", {
code,
message: redactCodexDiagnostic(message),
@@ -1377,14 +1687,148 @@ function rejectedToolCall(message: string): Record<string, unknown> {
return { success: false, contentItems: [{ type: "inputText", text: message }] };
}
function isTurnScopedServerRequest(method: string): boolean {
return method.startsWith("item/") || method.startsWith("tool/") || method === "mcpServer/elicitation/request";
function runtimeRequestKind(method: string): HarnessRuntimeRequestKind | null {
if (method === "item/commandExecution/requestApproval" || method === "execCommandApproval") {
return "command_approval";
}
if (method === "item/fileChange/requestApproval" || method === "applyPatchApproval") {
return "file_approval";
}
if (method === "item/permissions/requestApproval") return "permission_approval";
if (method === "item/tool/requestUserInput" || method === "tool/requestUserInput") {
return "user_input";
}
if (method === "mcpServer/elicitation/request") return "elicitation";
return null;
}
function runtimeRequestPrompt(
kind: HarnessRuntimeRequestKind,
params: Record<string, unknown>,
): string {
const reason = text(params.reason, text(params.message));
if (reason.length > 0) return boundedText(redactCodexDiagnostic(reason));
const labels: Record<HarnessRuntimeRequestKind, string> = {
command_approval: "Codex requests approval to run a command.",
file_approval: "Codex requests approval to change files.",
permission_approval: "Codex requests additional runtime permissions.",
user_input: "Codex requests user input.",
elicitation: "A tool requests structured user input.",
};
return labels[kind];
}
function runtimeRequestResponse(
request: HarnessRuntimeRequest,
resolution: HarnessRuntimeRequestResolution,
): Record<string, unknown> {
const action = (resolution as { action?: unknown }).action;
if (!["accept", "accept_for_session", "decline", "cancel", "submit"].includes(String(action))) {
throw new Error(`unsupported runtime request resolution ${String(action)}`);
}
if (request.requestKind === "command_approval" || request.requestKind === "file_approval") {
if (resolution.action === "submit") {
throw new Error(`${request.requestKind} does not accept submitted form data`);
}
const decisions = {
accept: "accept",
accept_for_session: "acceptForSession",
decline: "decline",
cancel: "cancel",
} as const;
return { decision: decisions[resolution.action] };
}
if (request.requestKind === "permission_approval") {
if (resolution.action === "submit") {
throw new Error("permission approval does not accept submitted form data");
}
return {
permissions: {},
scope: resolution.action === "accept_for_session" ? "session" : "turn",
};
}
if (request.requestKind === "user_input") {
if (resolution.action === "submit" && "answers" in resolution) {
return { answers: structuredClone(resolution.answers) };
}
if (resolution.action === "accept" || resolution.action === "accept_for_session") {
throw new Error("user input requires submit, decline, or cancel");
}
return { answers: {} };
}
if (resolution.action === "submit" && "content" in resolution) {
return { action: "accept", content: structuredClone(resolution.content), _meta: null };
}
if (resolution.action === "accept_for_session") {
throw new Error("elicitation does not support session acceptance");
}
return {
action: resolution.action === "submit" ? "accept" : resolution.action,
content: null,
_meta: null,
};
}
function parseThreadGoal(value: unknown): HarnessThreadGoal | null {
const goal = record(value);
const threadId = text(goal.threadId);
const objective = text(goal.objective);
const status = text(goal.status);
if (
threadId.length === 0 ||
objective.length === 0 ||
!["active", "paused", "blocked", "usageLimited", "budgetLimited", "complete"].includes(status)
) {
return null;
}
return {
threadId,
objective,
status: status as HarnessThreadGoal["status"],
tokenBudget: typeof goal.tokenBudget === "number" ? goal.tokenBudget : null,
tokensUsed: typeof goal.tokensUsed === "number" ? goal.tokensUsed : 0,
timeUsedSeconds: typeof goal.timeUsedSeconds === "number" ? goal.timeUsedSeconds : 0,
createdAt: typeof goal.createdAt === "number" ? goal.createdAt : 0,
updatedAt: typeof goal.updatedAt === "number" ? goal.updatedAt : 0,
};
}
function threadStatus(value: unknown): string {
if (typeof value === "string") return value;
return text(record(value).type, "unknown");
}
function lineageFromThread(value: unknown): HarnessThreadLineageEntry {
const thread = record(value);
const source = record(thread.source);
const subAgent = source.subAgent ?? source.subagent;
const subAgentRecord = record(subAgent);
const spawn = record(subAgentRecord.thread_spawn ?? subAgentRecord.threadSpawn);
const parentThreadId = text(
spawn.parent_thread_id ?? spawn.parentThreadId,
text(thread.forkedFromId),
) || null;
return {
threadId: text(thread.id),
providerSessionId: text(thread.sessionId) || null,
parentThreadId,
depth: typeof spawn.depth === "number" ? spawn.depth : parentThreadId === null ? 0 : 1,
nickname: text(thread.agentNickname, text(spawn.agent_nickname ?? spawn.agentNickname)) || null,
role: text(thread.agentRole, text(spawn.agent_role ?? spawn.agentRole)) || null,
status: threadStatus(thread.status),
};
}
function isBoundCodexNotification(method: string): boolean {
return (
method === "turn/started" ||
method === "turn/completed" ||
method === "thread/started" ||
method === "thread/status/changed" ||
method === "thread/closed" ||
method === "thread/goal/updated" ||
method === "thread/goal/cleared" ||
method === "serverRequest/resolved" ||
method === "thread/tokenUsage/updated" ||
method === "error" ||
method === "warning" ||
@@ -1395,18 +1839,18 @@ function isBoundCodexNotification(method: string): boolean {
);
}
function safeRequestResponse(method: string): Record<string, unknown> {
function safeRequestResponse(method: string, action: "decline" | "cancel" = "decline"): Record<string, unknown> {
if (method === "item/permissions/requestApproval") {
return { permissions: {} };
return { permissions: {}, scope: "turn" };
}
if (method === "mcpServer/elicitation/request") {
return { action: "decline", content: null };
return { action, content: null, _meta: null };
}
if (method === "item/tool/requestUserInput" || method === "tool/requestUserInput") {
return { answers: {} };
}
if (method.includes("requestApproval")) {
return { decision: "decline" };
if (method.includes("requestApproval") || method === "execCommandApproval" || method === "applyPatchApproval") {
return { decision: action };
}
return {};
}
+2
View File
@@ -9,10 +9,12 @@ export * from "./drivers/codex/app-server-transport.js";
export * from "./drivers/codex/codex-app-server-driver.js";
export * from "./mock-core/mock-control-plane-adapter.js";
export * from "./mock-core/phase4-codex-runner.js";
export * from "./mock-core/phase4b-demo-server.js";
export * from "./mock-core/phase2-local-runner.js";
export * from "./mock-core/phase3-recovery.js";
export * from "./protocol/phase0-fixture.js";
export * from "./protocol/phase1-contract.js";
export * from "./protocol/phase4b-fixture.js";
export * from "./protocol/phase1-loader.js";
export * from "./reducer/session-reducer.js";
export * from "./tracer/phase0-runner.js";
@@ -0,0 +1,355 @@
import { afterEach, describe, expect, it } from "vitest";
import type {
HarnessDriver,
HarnessDriverDescriptor,
HarnessGoalOperation,
HarnessRuntimeRequest,
HarnessRuntimeRequestResolution,
HarnessSession,
HarnessSessionRecoveryResult,
OpenHarnessSessionInput,
PersistedHarnessSession,
} from "../contracts/harness-driver.js";
import type { PrpEvent } from "../protocol/phase1-contract.js";
import { Phase4bDemoServer } from "./phase4b-demo-server.js";
class Queue<T> implements AsyncIterable<T> {
#values: T[] = [];
#waiters: Array<(value: IteratorResult<T>) => void> = [];
#closed = false;
push(value: T): void {
const waiter = this.#waiters.shift();
if (waiter) waiter({ value, done: false });
else this.#values.push(value);
}
close(): void {
this.#closed = true;
for (const waiter of this.#waiters.splice(0)) waiter({ value: undefined, done: true });
}
[Symbol.asyncIterator](): AsyncIterator<T> {
return {
next: async () => {
const value = this.#values.shift();
if (value !== undefined) return { value, done: false };
if (this.#closed) return { value: undefined, done: true };
return new Promise((resolve) => this.#waiters.push(resolve));
},
};
}
}
class StubSession implements HarnessSession {
readonly queue = new Queue<PrpEvent>();
readonly runId: string;
readonly normalizedSessionId: string;
sourceSequence: number;
activeTurnId: string | null;
pending: HarnessRuntimeRequest[];
currentGoal: PersistedHarnessSession["goal"];
constructor(input: {
runId: string;
normalizedSessionId: string;
sourceSequence?: number;
activeTurnId?: string | null;
pending?: HarnessRuntimeRequest[];
goal?: PersistedHarnessSession["goal"];
resumed?: boolean;
}) {
this.runId = input.runId;
this.normalizedSessionId = input.normalizedSessionId;
this.sourceSequence = input.sourceSequence ?? 0;
this.activeTurnId = input.activeTurnId ?? null;
this.pending = structuredClone(input.pending ?? []);
this.currentGoal = input.goal ?? null;
this.emit(input.resumed ? "session.resumed" : "session.started", {
driverSessionId: "thread-demo",
providerSessionId: "provider-demo",
diagnostic: "Bearer server-only-secret",
sandbox: {
writableRoots: [
"/srv/paperclip/home/.paperclip/instances/company/codex-home/memories",
],
},
});
}
ids() {
return {
driverSessionId: "thread-demo",
providerSessionId: "provider-demo",
displayId: "thread-demo",
};
}
events(): AsyncIterable<PrpEvent> {
return this.queue;
}
async startTurn(): Promise<{ turnId: string }> {
this.activeTurnId = "turn-demo";
this.emit("turn.submitted", {});
this.emit("turn.accepted", { turnId: "turn-demo" }, { turnId: "turn-demo" });
this.emit("turn.started", { status: "inProgress" }, { turnId: "turn-demo" });
const request: HarnessRuntimeRequest = {
requestId: "request-demo",
requestKind: "command_approval",
method: "item/commandExecution/requestApproval",
turnId: "turn-demo",
itemId: "command-demo",
status: "pending",
prompt: "Approve the deterministic command.",
details: { authorization: "Bearer browser-must-not-see" },
};
this.pending = [request];
this.emit("runtime_request.created", {
request: { ...request, type: request.method },
}, { turnId: request.turnId, itemId: request.itemId });
return { turnId: "turn-demo" };
}
async steer(input: { turnId: string }): Promise<void> {
this.emit("item.completed", {
kind: "steering_acknowledgement",
status: "acknowledged",
text: "Steering acknowledged.",
}, { turnId: input.turnId, itemId: "steer-demo" });
}
async interrupt(input: { turnId?: string }): Promise<void> {
this.emit("item.completed", {
kind: "interrupt_acknowledgement",
status: "acknowledged",
text: "Interrupt acknowledged.",
}, { turnId: input.turnId, itemId: "interrupt-demo" });
}
pendingRuntimeRequests(): HarnessRuntimeRequest[] {
return structuredClone(this.pending);
}
async resolveRuntimeRequest(input: {
requestId: string;
turnId: string;
resolution: HarnessRuntimeRequestResolution;
}): Promise<void> {
this.pending = this.pending.filter(({ requestId }) => requestId !== input.requestId);
this.emit("runtime_request.resolved", {
requestId: input.requestId,
resolution: input.resolution.action,
}, { turnId: input.turnId, itemId: "command-demo" });
}
async goal(input: HarnessGoalOperation) {
if (input.action === "clear") this.currentGoal = null;
else if (input.action === "set") {
this.currentGoal = {
threadId: "thread-demo",
objective: input.objective,
status: "active",
tokenBudget: input.tokenBudget ?? null,
tokensUsed: 0,
timeUsedSeconds: 0,
createdAt: 1,
updatedAt: 2,
};
}
return this.currentGoal ?? null;
}
lineage() {
return [{
threadId: "thread-demo",
providerSessionId: "provider-demo",
parentThreadId: null,
depth: 0,
nickname: null,
role: null,
status: "active",
}];
}
async snapshot(): Promise<PersistedHarnessSession> {
return {
driverKind: "stub",
driverSessionId: "thread-demo",
providerSessionId: "provider-demo",
runId: this.runId,
normalizedSessionId: this.normalizedSessionId,
activeTurnId: this.activeTurnId,
pendingRuntimeRequests: this.pendingRuntimeRequests(),
goal: this.currentGoal ?? null,
lineage: this.lineage(),
lastSourceSequence: this.sourceSequence,
};
}
async close(): Promise<void> {
this.queue.close();
}
private emit(
eventType: PrpEvent["eventType"],
payload: Record<string, unknown>,
refs: { turnId?: string; itemId?: string } = {},
): void {
const sourceSeq = ++this.sourceSequence;
this.queue.push({
schema: "paperclip.prp.event.v1",
sourceEventId: `stub:${this.runId}:${sourceSeq}`,
sourceSeq,
sourceInstanceId: "stub",
sourceKind: "runner",
runId: this.runId,
normalizedSessionId: this.normalizedSessionId,
...refs,
eventType,
schemaVersion: 1,
priority: 1,
emittedAt: "2026-08-08T12:00:00.000Z",
payload,
});
}
}
class StubDriver implements HarnessDriver {
openInputs: OpenHarnessSessionInput[] = [];
async descriptor(): Promise<HarnessDriverDescriptor> {
return {
kind: "stub",
displayName: "Stub",
version: "1",
capabilities: {
resume: true,
typedEvents: true,
steering: true,
interruption: true,
structuredResult: true,
runtimeRequestResolution: true,
goals: true,
threadLineage: true,
},
};
}
async openSession(input: OpenHarnessSessionInput): Promise<HarnessSession> {
this.openInputs.push(structuredClone(input));
return new StubSession(input);
}
async recoverSession(snapshot: PersistedHarnessSession): Promise<HarnessSessionRecoveryResult> {
return {
recovered: true,
session: new StubSession({
runId: snapshot.runId!,
normalizedSessionId: snapshot.normalizedSessionId!,
sourceSequence: snapshot.lastSourceSequence,
activeTurnId: snapshot.activeTurnId,
pending: snapshot.pendingRuntimeRequests,
goal: snapshot.goal,
resumed: true,
}),
};
}
}
const servers: Phase4bDemoServer[] = [];
afterEach(async () => {
await Promise.all(servers.splice(0).map((server) => server.close()));
});
describe("Phase 4b package-local demo server", () => {
it("keeps credentials and workspace selection server-side across replay and reconnect", async () => {
const driver = new StubDriver();
const server = new Phase4bDemoServer({
workingDirectory: "/safe/server-owned-workspace",
driverFactory: () => driver,
});
servers.push(server);
const { url } = await server.start();
const health = await fetch(`${url}/api/phase4b/health`).then((response) => response.json());
expect(health).toEqual(expect.objectContaining({
status: "ok",
providerAuthentication: "server-side",
credentialsExposed: false,
}));
const created = await fetch(`${url}/api/phase4b/sessions`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
objective: "Exercise the server boundary.",
workingDirectory: "/browser/chosen-path",
authorization: "Bearer browser-secret",
}),
}).then((response) => response.json()) as Record<string, unknown>;
expect(driver.openInputs[0]?.workingDirectory).toBe("/safe/server-owned-workspace");
expect(created).toMatchObject({
providerAuthentication: "server-side",
credentialsExposed: false,
activeTurnId: "turn-demo",
pendingRequests: [{ requestId: "request-demo" }],
});
expect(JSON.stringify(created)).not.toContain("browser-secret");
expect(JSON.stringify(created)).not.toContain("browser-must-not-see");
expect(JSON.stringify(created)).not.toContain("server-only-secret");
expect(JSON.stringify(created)).not.toContain("/srv/paperclip/home");
const sessionId = String(created.sessionId);
const replay = await fetch(`${url}/api/phase4b/sessions/${sessionId}/events?after=0`)
.then((response) => response.json()) as { events: PrpEvent[]; cursor: number; replay: boolean };
expect(replay.replay).toBe(true);
expect(replay.events.length).toBeGreaterThanOrEqual(5);
expect(replay.events.every((event, index) => event.sourceSeq === index + 1)).toBe(true);
expect(JSON.stringify(replay)).not.toContain("/srv/paperclip/home");
expect(JSON.stringify(replay)).toContain("<server-path>");
const resolved = await fetch(
`${url}/api/phase4b/sessions/${sessionId}/requests/request-demo/resolve`,
{
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
turnId: "turn-demo",
resolution: { action: "accept_for_session" },
}),
},
).then((response) => response.json()) as Record<string, unknown>;
expect(resolved.pendingRequests).toEqual([]);
const withGoal = await fetch(`${url}/api/phase4b/sessions/${sessionId}/goal/set`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ objective: "Keep the goal durable.", tokenBudget: 1024 }),
}).then((response) => response.json()) as Record<string, unknown>;
expect(withGoal.goal).toMatchObject({
objective: "Keep the goal durable.",
status: "active",
tokenBudget: 1024,
});
const reconnected = await fetch(`${url}/api/phase4b/sessions/${sessionId}/reconnect`, {
method: "POST",
headers: { "content-type": "application/json" },
body: "{}",
}).then((response) => response.json()) as Record<string, unknown>;
expect(reconnected).toMatchObject({
sessionId,
runId: created.runId,
normalizedSessionId: created.normalizedSessionId,
driverSession: created.driverSession,
goal: withGoal.goal,
});
const afterReconnect = await fetch(
`${url}/api/phase4b/sessions/${sessionId}/events?after=${replay.cursor}`,
).then((response) => response.json()) as { events: PrpEvent[] };
expect(afterReconnect.events.some(({ eventType }) => eventType === "session.resumed")).toBe(true);
});
});
@@ -0,0 +1,412 @@
import { randomUUID } from "node:crypto";
import {
createServer,
type IncomingMessage,
type Server,
type ServerResponse,
} from "node:http";
import type {
HarnessDriver,
HarnessGoalOperation,
HarnessRuntimeRequestResolution,
HarnessSession,
} from "../contracts/harness-driver.js";
import {
createPhase4TaskEnvelope,
type Phase4TaskEnvelope,
} from "../contracts/phase4.js";
import type { NativeSessionCapabilities } from "../contracts/types.js";
import {
validatePrpEvent,
type PrpCapabilities,
type PrpEvent,
} from "../protocol/phase1-contract.js";
import {
applyPrpEvent,
createSessionSnapshotFromMetadata,
} from "../reducer/session-reducer.js";
import { redactCodexDiagnostic } from "../drivers/codex/app-server-transport.js";
const MAX_BROWSER_BODY_BYTES = 64 * 1024;
const MAX_BROWSER_EVENTS = 4096;
export interface Phase4bDemoServerOptions {
workingDirectory: string;
driverFactory: (envelope: Phase4TaskEnvelope) => HarnessDriver;
host?: string;
port?: number;
}
interface DemoEntry {
id: string;
runId: string;
normalizedSessionId: string;
envelope: Phase4TaskEnvelope;
driver: HarnessDriver;
session: HarnessSession;
capabilities: NativeSessionCapabilities;
events: PrpEvent[];
subscribers: Set<ServerResponse>;
consumeTask: Promise<void>;
}
interface BrowserCreateBody {
objective?: string;
message?: string;
}
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
}
function browserSafe(value: unknown, depth = 0): unknown {
if (depth > 10) return "[TRUNCATED]";
if (typeof value === "string") return redactBrowserString(value);
if (Array.isArray(value)) return value.slice(0, 256).map((entry) => browserSafe(entry, depth + 1));
if (typeof value !== "object" || value === null) return value;
return Object.fromEntries(
Object.entries(value as Record<string, unknown>).slice(0, 256).map(([key, entry]) => [
key,
isSensitiveBrowserKey(key)
? "[REDACTED]"
: browserSafe(entry, depth + 1),
]),
);
}
function redactBrowserString(value: string): string {
return redactCodexDiagnostic(value).replace(
/\/(?:srv\/paperclip\/home|home\/[^/\s]+)\/\.paperclip\/[^\s"'<>),\]}]*/g,
"<server-path>",
);
}
function isSensitiveBrowserKey(key: string): boolean {
return /^(?:api[_-]?key|token|accessToken|refreshToken|secret|password|authorization|cookie)$/i.test(key) ||
/(?:^|_)(?:api[_-]?key|token|secret|password|authorization|cookie)$/i.test(key);
}
function json(response: ServerResponse, status: number, body: unknown): void {
const serialized = JSON.stringify(browserSafe(body));
response.writeHead(status, {
"content-type": "application/json; charset=utf-8",
"cache-control": "no-store",
"content-length": Buffer.byteLength(serialized),
});
response.end(serialized);
}
async function readJson(request: IncomingMessage): Promise<Record<string, unknown>> {
const chunks: Buffer[] = [];
let size = 0;
for await (const chunk of request) {
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
size += buffer.length;
if (size > MAX_BROWSER_BODY_BYTES) throw new Error("browser request body is too large");
chunks.push(buffer);
}
if (chunks.length === 0) return {};
const value = JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown;
if (typeof value !== "object" || value === null || Array.isArray(value)) {
throw new Error("browser request body must be a JSON object");
}
return value as Record<string, unknown>;
}
function prpCapabilities(capabilities: NativeSessionCapabilities): PrpCapabilities {
return {
schema: "paperclip.prp.capabilities.v1",
sessionReusePolicy: "reuse_per_issue",
driver: { kind: "phase4b-demo", version: "1" },
steer: capabilities.steering,
interrupt: capabilities.interruption,
resume: capabilities.resume,
runtimeRequests: capabilities.runtimeRequestResolution ?? false,
structuredResult: capabilities.structuredResult,
typedEvents: capabilities.typedEvents,
goals: capabilities.goals ?? false,
threadLineage: capabilities.threadLineage ?? false,
...(capabilities.unsupported?.length ? { unsupported: capabilities.unsupported } : {}),
};
}
function pathParts(request: IncomingMessage): string[] {
const url = new URL(request.url ?? "/", "http://phase4b.invalid");
return url.pathname.split("/").filter(Boolean);
}
function errorStatus(error: unknown): number {
const code = record(error).code;
if (code === "stale_turn" || code === "already_terminal") return 409;
return 400;
}
export class Phase4bDemoServer {
readonly #options: Required<Pick<Phase4bDemoServerOptions, "host" | "port">> &
Omit<Phase4bDemoServerOptions, "host" | "port">;
readonly #entries = new Map<string, DemoEntry>();
#server: Server | null = null;
constructor(options: Phase4bDemoServerOptions) {
this.#options = {
...options,
host: options.host ?? "127.0.0.1",
port: options.port ?? 0,
};
}
async start(): Promise<{ host: string; port: number; url: string }> {
if (this.#server !== null) throw new Error("Phase 4b demo server is already started");
this.#server = createServer((request, response) => {
void this.#handle(request, response).catch((error) => {
if (!response.headersSent) {
json(response, errorStatus(error), {
error: record(error).code ?? "invalid_request",
message: redactCodexDiagnostic(String(error)),
});
} else {
response.end();
}
});
});
await new Promise<void>((resolve, reject) => {
this.#server!.once("error", reject);
this.#server!.listen(this.#options.port, this.#options.host, () => resolve());
});
const address = this.#server.address();
if (address === null || typeof address === "string") {
throw new Error("Phase 4b demo server did not bind a TCP address");
}
return {
host: this.#options.host,
port: address.port,
url: `http://${this.#options.host}:${address.port}`,
};
}
async close(): Promise<void> {
const entries = [...this.#entries.values()];
this.#entries.clear();
await Promise.all(entries.map(async (entry) => {
for (const subscriber of entry.subscribers) subscriber.end();
await entry.session.close({ reason: "demo_server_closed" });
await entry.consumeTask.catch(() => undefined);
}));
if (this.#server === null) return;
const server = this.#server;
this.#server = null;
await new Promise<void>((resolve, reject) => server.close((error) => {
if (error) reject(error);
else resolve();
}));
}
async #handle(request: IncomingMessage, response: ServerResponse): Promise<void> {
const parts = pathParts(request);
if (request.method === "GET" && parts.join("/") === "api/phase4b/health") {
json(response, 200, {
status: "ok",
boundary: "package-local-mock-core",
providerAuthentication: "server-side",
credentialsExposed: false,
});
return;
}
if (request.method === "POST" && parts.join("/") === "api/phase4b/sessions") {
const body = await readJson(request) as BrowserCreateBody;
const objective = typeof body.objective === "string" && body.objective.trim().length > 0
? body.objective.trim()
: "Complete the Phase 4b demo task safely.";
const message = typeof body.message === "string" ? body.message : objective;
const entry = await this.#create(objective, message);
json(response, 201, await this.#publicState(entry));
return;
}
if (parts.length < 4 || parts[0] !== "api" || parts[1] !== "phase4b" || parts[2] !== "sessions") {
json(response, 404, { error: "not_found" });
return;
}
const entry = this.#entries.get(parts[3]!);
if (entry === undefined) {
json(response, 404, { error: "session_not_found" });
return;
}
const action = parts[4];
if (request.method === "GET" && action === undefined) {
json(response, 200, await this.#publicState(entry));
return;
}
if (request.method === "GET" && action === "events") {
const url = new URL(request.url ?? "/", "http://phase4b.invalid");
const after = Math.max(0, Number.parseInt(url.searchParams.get("after") ?? "0", 10) || 0);
json(response, 200, {
events: entry.events.slice(after),
cursor: entry.events.length,
replay: after === 0,
});
return;
}
if (request.method === "GET" && action === "stream") {
const url = new URL(request.url ?? "/", "http://phase4b.invalid");
const after = Math.max(0, Number.parseInt(url.searchParams.get("after") ?? "0", 10) || 0);
response.writeHead(200, {
"content-type": "text/event-stream; charset=utf-8",
"cache-control": "no-store",
connection: "keep-alive",
});
for (const event of entry.events.slice(after)) {
response.write(`data: ${JSON.stringify(browserSafe(event))}\n\n`);
}
entry.subscribers.add(response);
request.once("close", () => entry.subscribers.delete(response));
return;
}
if (request.method !== "POST") {
json(response, 405, { error: "method_not_allowed" });
return;
}
const body = await readJson(request);
if (action === "steer") {
await entry.session.steer?.({
turnId: String(body.turnId ?? ""),
message: { role: "user", text: String(body.text ?? "") },
});
} else if (action === "interrupt") {
await entry.session.interrupt?.({
...(typeof body.turnId === "string" ? { turnId: body.turnId } : {}),
...(typeof body.reason === "string" ? { reason: body.reason } : {}),
});
} else if (action === "requests" && parts[5] && parts[6] === "resolve") {
await entry.session.resolveRuntimeRequest?.({
requestId: parts[5],
turnId: String(body.turnId ?? ""),
resolution: record(body.resolution) as HarnessRuntimeRequestResolution,
});
} else if (action === "goal" && parts[5]) {
await entry.session.goal?.(goalOperation(parts[5], body));
} else if (action === "reconnect") {
await this.#reconnect(entry);
} else {
json(response, 404, { error: "action_not_found" });
return;
}
json(response, 200, await this.#publicState(entry));
}
async #create(objective: string, message: string): Promise<DemoEntry> {
const envelope = createPhase4TaskEnvelope({
objective,
contractRevision: "phase4b-demo-v1",
criteria: [{ id: "objective", requirement: objective }],
});
const driver = this.#options.driverFactory(envelope);
const id = randomUUID();
const runId = `phase4b-run-${id}`;
const normalizedSessionId = `phase4b-session-${id}`;
const session = await driver.openSession({
runId,
normalizedSessionId,
workingDirectory: this.#options.workingDirectory,
});
const descriptor = await driver.descriptor();
const entry: DemoEntry = {
id,
runId,
normalizedSessionId,
envelope,
driver,
session,
capabilities: descriptor.capabilities,
events: [],
subscribers: new Set(),
consumeTask: Promise.resolve(),
};
this.#entries.set(id, entry);
entry.consumeTask = this.#consume(entry, session);
await session.startTurn({ message: { role: "user", text: message } });
await new Promise<void>((resolve) => setImmediate(resolve));
return entry;
}
async #consume(entry: DemoEntry, session: HarnessSession): Promise<void> {
for await (const rawEvent of session.events()) {
const event = browserSafe(rawEvent) as PrpEvent;
const validation = validatePrpEvent(event);
if (!validation.ok) continue;
if (entry.events.length >= MAX_BROWSER_EVENTS) entry.events.shift();
entry.events.push(event);
const serialized = `data: ${JSON.stringify(event)}\n\n`;
for (const subscriber of entry.subscribers) subscriber.write(serialized);
}
}
async #reconnect(entry: DemoEntry): Promise<void> {
const snapshot = await entry.session.snapshot();
await entry.session.close({ reason: "browser_reconnect" });
await entry.consumeTask.catch(() => undefined);
const recovery = await entry.driver.recoverSession?.(snapshot);
if (!recovery?.recovered || recovery.session === undefined) {
throw new Error(`session resume failed: ${recovery?.reason ?? "driver cannot resume"}`);
}
entry.session = recovery.session;
entry.consumeTask = this.#consume(entry, recovery.session);
}
async #publicState(entry: DemoEntry): Promise<Record<string, unknown>> {
const driverSession = entry.session.ids();
const metadata = {
fixtureName: "phase4b-demo",
identity: {
schema: "paperclip.prp.identity.v1" as const,
companyId: "package-local-demo",
issueId: "phase4b-demo",
runId: entry.runId,
environmentLeaseId: "package-local",
runnerInstanceId: "phase4b-demo-server",
normalizedSessionId: entry.normalizedSessionId,
driverSessionId: driverSession.driverSessionId,
...(driverSession.providerSessionId
? { providerSessionId: driverSession.providerSessionId }
: {}),
},
capabilities: prpCapabilities(entry.capabilities),
};
const snapshot = entry.events.reduce(
applyPrpEvent,
createSessionSnapshotFromMetadata(metadata),
);
const harnessSnapshot = await entry.session.snapshot();
return {
sessionId: entry.id,
runId: entry.runId,
normalizedSessionId: entry.normalizedSessionId,
providerAuthentication: "server-side",
credentialsExposed: false,
capabilities: entry.capabilities,
driverSession,
activeTurnId: harnessSnapshot.activeTurnId ?? snapshot.activeTurnId,
pendingRequests: entry.session.pendingRuntimeRequests?.() ?? [],
goal: harnessSnapshot.goal ?? null,
lineage: entry.session.lineage?.() ?? [],
cursor: entry.events.length,
snapshot,
};
}
}
function goalOperation(action: string, body: Record<string, unknown>): HarnessGoalOperation {
if (action === "get" || action === "pause" || action === "resume" || action === "clear") {
return { action };
}
if (action === "set" && typeof body.objective === "string" && body.objective.trim().length > 0) {
return {
action,
objective: body.objective.trim(),
...(typeof body.tokenBudget === "number" ? { tokenBudget: body.tokenBudget } : {}),
};
}
throw new Error(`unsupported goal operation ${action}`);
}
@@ -0,0 +1,107 @@
import { readFile } from "node:fs/promises";
export const PHASE4B_CONFORMANCE_SCHEMA =
"paperclip.runner.phase4b.conformance.v1" as const;
export interface Phase4bRuntimeRequestFixture {
id: string;
method: string;
requestKind: string;
resolution: Record<string, unknown> & { action: string };
expectedResponse: Record<string, unknown>;
}
export interface Phase4bGoalFixture {
action: "get" | "set" | "pause" | "resume" | "clear";
method: string;
params: Record<string, unknown>;
}
export interface Phase4bConformanceFixture {
schema: typeof PHASE4B_CONFORMANCE_SCHEMA;
codexVersion: string;
runtimeRequests: Phase4bRuntimeRequestFixture[];
goals: Phase4bGoalFixture[];
lineage: {
rootThreadId: string;
childThread: Record<string, unknown> & { id: string; sessionId: string };
};
controls: Record<string, { turnId?: string; expected: string }>;
reconnect: {
runId: string;
normalizedSessionId: string;
driverSessionId: string;
providerSessionId: string;
lastSourceSequence: number;
};
redactionMarkers: string[];
}
function record(value: unknown): Record<string, unknown> | null {
return typeof value === "object" && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: null;
}
function nonEmpty(value: unknown): value is string {
return typeof value === "string" && value.length > 0;
}
/** Load the deterministic Phase 4b wire fixture without trusting its shape. */
export async function loadPhase4bConformanceFixture(
path: string,
): Promise<Phase4bConformanceFixture> {
const value = JSON.parse(await readFile(path, "utf8")) as unknown;
const fixture = record(value);
if (fixture?.schema !== PHASE4B_CONFORMANCE_SCHEMA) {
throw new Error("Phase 4b fixture has an unsupported schema");
}
if (!nonEmpty(fixture.codexVersion)) {
throw new Error("Phase 4b fixture must name the observed Codex version");
}
if (!Array.isArray(fixture.runtimeRequests) || fixture.runtimeRequests.length === 0) {
throw new Error("Phase 4b fixture must include runtime request cases");
}
if (!Array.isArray(fixture.goals) || fixture.goals.length !== 5) {
throw new Error("Phase 4b fixture must include all five goal operations");
}
const requestIds = new Set<string>();
for (const candidate of fixture.runtimeRequests) {
const request = record(candidate);
const resolution = record(request?.resolution);
if (
!nonEmpty(request?.id) ||
requestIds.has(request.id) ||
!nonEmpty(request.method) ||
!nonEmpty(request.requestKind) ||
!nonEmpty(resolution?.action) ||
record(request.expectedResponse) === null
) {
throw new Error("Phase 4b fixture contains an invalid runtime request case");
}
requestIds.add(request.id);
}
const actions = new Set(fixture.goals.map((candidate) => record(candidate)?.action));
if (!["get", "set", "pause", "resume", "clear"].every((action) => actions.has(action))) {
throw new Error("Phase 4b fixture goal operation set is incomplete");
}
const lineage = record(fixture.lineage);
const child = record(lineage?.childThread);
const reconnect = record(fixture.reconnect);
if (
!nonEmpty(lineage?.rootThreadId) ||
!nonEmpty(child?.id) ||
!nonEmpty(child.sessionId) ||
!nonEmpty(reconnect?.runId) ||
!nonEmpty(reconnect.normalizedSessionId) ||
!nonEmpty(reconnect.driverSessionId) ||
!nonEmpty(reconnect.providerSessionId) ||
!Number.isInteger(reconnect.lastSourceSequence)
) {
throw new Error("Phase 4b fixture identity or lineage is incomplete");
}
if (!Array.isArray(fixture.redactionMarkers) || !fixture.redactionMarkers.every(nonEmpty)) {
throw new Error("Phase 4b fixture redaction markers are invalid");
}
return value as Phase4bConformanceFixture;
}