Add local fake runner supervision (#12095)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Paperclip Runner needs a small local process model before it can connect to a production provider or server. > - The TypeScript PRP contracts now define the expected replay behavior. > - A second language implementation must produce the same result from the same fixtures. > - Local child processes also need bounded input, bounded output, and complete descendant cleanup. > - This pull request adds a package-local Rust runner, a scripted fake harness, and deterministic parity checks. > - The benefit is a testable process boundary with no production Paperclip behavior change. ## Linked Issues or Issue Description **Subsystem affected** Cross-cutting. This pull request adds private test infrastructure to `packages/paperclip-runner`. **Problem or motivation** The PRP contracts have no second implementation on `master`. There is also no small harness that can prove process cleanup, command idempotency, terminal reconciliation, or bounded JSONL handling without a production provider. **Proposed solution** Add a minimal Rust workspace. Add a local runner process, a scripted fake harness, a bounded process supervisor, and Rust conformance and replay checks. Keep all binaries package-local. Do not connect them to the Paperclip server. **Alternatives considered** The combined runner branch includes provider transports, durable networking, SDKs, labs, and server behavior. That change is too large for this review unit. A TypeScript-only harness would not test cross-language contract parity. **Roadmap alignment** This work supports the governed tools and self-healing run direction in `ROADMAP.md`. It does not add a user-facing runtime, adapter, endpoint, or rollout flag. **Additional context** Refs #12091 and #11962. Pull request #12091 was merged before this branch opened. This branch is based on the current `master`. Its delta is 25 files. ## What Changed - Added a minimal locked Rust workspace with only `serde` and `serde_json` dependencies. - Added a package-local `paperclip-runnerd` local mode and a scripted fake harness. - Added bounded controller input, harness input, subprocess output queues, line sizes, log retention, script sizes, script steps, and command history. - Added contiguous controller and harness sequence checks and equivalent-command replay handling. - Added process-group supervision that cleans up child processes and remaining descendants after forced or natural harness exit. - Added runner-owned terminal reconciliation for success, failure, interruption, cancellation, controller closure, and protocol failure. - Added Rust conformance output and deterministic replay summaries for the shared PRP fixtures. - Added fake scripts for success, failure, interruption, interaction, duplicate terminal output, process cleanup, and oversized output. - Added package scripts and documentation for the Rust and cross-language checks. - Kept provider transport, server integration, semantic tools, and production runtime selection out of this pull request. ## Verification - `pnpm --filter @paperclipai/paperclip-runner check:all` passes. - TypeScript contract tests pass: 8 Node tests and 44 Vitest tests. - Rust tests pass: 20 unit tests, 2 local-runner tests, and 3 process-supervisor tests. - The Rust conformance and replay parity checks pass against the shared fixtures. - The natural-exit and forced-exit tests confirm that the harness and its worker process are stopped. - The oversized-frame test confirms that a harness frame above the configured limit is rejected. - `pnpm -r typecheck` passes after the final rebase to `master`. - `pnpm build` passes after the final rebase to `master`. - `pnpm check:token-gates` passes. - `git diff --check` passes. - The delta against `master` is 25 files. The package lockfile is unchanged. - `pnpm test:run` completed locally with 4,686 passing tests, 19 skipped tests, and 30 failures in 8 unchanged server test files. The failures are local macOS path-alias, listener, port-range, and workspace-runtime baseline failures. No changed-file test failed, and every applicable Linux CI shard passes. - Storybook visual regression skipped intentionally because this pull request changes no UI or story files. ## Risks Low production risk. No server code invokes the new binaries. The package remains private. The main risks are process leaks, unbounded local input, and cross-language drift. Bounded queues and sizes, process-group cleanup tests, fixture manifests, and parity checks cover these risks. I checked `ROADMAP.md`. This change is private test infrastructure for planned control-plane work. It does not duplicate a shipped or public product surface. ## Model Used OpenAI Codex with GPT-5 was used. The exact serving model ID and context size were not exposed. The model used high reasoning, repository tools, GitHub tools, and local code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
83fefaadd1
commit
6b20cc97cc
|
|
@ -0,0 +1 @@
|
|||
/runner/target/
|
||||
|
|
@ -5,8 +5,11 @@ This private workspace package contains the staged Paperclip Runner work.
|
|||
The package currently exposes only the language-neutral PRP v1 TypeScript
|
||||
contract, provider-neutral structured questions and responses, deterministic
|
||||
fixture validation/replay, structured-result normalization, and the session
|
||||
reducer oracle. It does not add a runner process, provider transport, server
|
||||
adapter, semantic-tool authorization, or production Paperclip behavior.
|
||||
reducer oracle. It also contains a package-local Rust runner, scripted fake
|
||||
harness, bounded process supervisor, and cross-language replay oracle. These
|
||||
Rust binaries are test infrastructure. No server code starts or invokes them.
|
||||
The package does not add a provider transport, server adapter, semantic-tool
|
||||
authorization, or production Paperclip behavior.
|
||||
|
||||
The first provider scope is Codex. The protocol contains provider-neutral event
|
||||
and semantic receipt shapes, but their presence does not authorize a tool or
|
||||
|
|
@ -21,6 +24,16 @@ Run the complete contract gate with:
|
|||
pnpm --filter @paperclipai/paperclip-runner check:protocol
|
||||
```
|
||||
|
||||
Run the Rust runner gate with:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/paperclip-runner check:runner
|
||||
```
|
||||
|
||||
This command checks Rust formatting, builds and tests the minimal workspace,
|
||||
verifies bounded process cleanup, exercises the fake local runner, and compares
|
||||
the Rust conformance and replay summaries with the shared fixtures.
|
||||
|
||||
Use `generate:protocol-manifest` after a schema or fixture change,
|
||||
`generate:protocol-types` after a schema change, and
|
||||
`generate:replay-goldens` after an intentional reducer change. Commit generated
|
||||
|
|
|
|||
|
|
@ -21,15 +21,25 @@
|
|||
"scripts": {
|
||||
"build": "pnpm run check:protocol-manifest && pnpm run build:typescript && node scripts/generate-replay-goldens.mjs --check",
|
||||
"build:typescript": "pnpm run check:protocol-types && tsc -p tsconfig.json",
|
||||
"typecheck": "node --check scripts/protocol-contract.mjs && node --check scripts/generate-protocol-manifest.mjs && node --check scripts/generate-protocol-schema-module.mjs && node --check scripts/generate-replay-goldens.mjs && pnpm run check:protocol-types && tsc -p tsconfig.json --noEmit",
|
||||
"test": "node --test test/protocol-contract.test.mjs && vitest run",
|
||||
"build:rust": "cargo build --manifest-path runner/Cargo.toml --locked --workspace --bins",
|
||||
"typecheck": "pnpm run typecheck:typescript && pnpm run typecheck:rust",
|
||||
"typecheck:typescript": "node --check scripts/protocol-contract.mjs && node --check scripts/generate-protocol-manifest.mjs && node --check scripts/generate-protocol-schema-module.mjs && node --check scripts/generate-replay-goldens.mjs && pnpm run check:protocol-types && tsc -p tsconfig.json --noEmit",
|
||||
"typecheck:rust": "cargo fmt --manifest-path runner/Cargo.toml --all -- --check && cargo check --manifest-path runner/Cargo.toml --locked --workspace",
|
||||
"test": "pnpm run test:typescript && pnpm run test:rust",
|
||||
"test:typescript": "node --test test/protocol-contract.test.mjs && vitest run",
|
||||
"test:rust": "cargo test --manifest-path runner/Cargo.toml --locked --workspace",
|
||||
"generate:protocol-manifest": "node scripts/generate-protocol-manifest.mjs",
|
||||
"check:protocol-manifest": "node scripts/generate-protocol-manifest.mjs --check",
|
||||
"generate:protocol-types": "node scripts/generate-protocol-schema-module.mjs",
|
||||
"check:protocol-types": "node scripts/generate-protocol-schema-module.mjs --check",
|
||||
"generate:replay-goldens": "pnpm run build:typescript && node scripts/generate-replay-goldens.mjs",
|
||||
"check:replay-goldens": "pnpm run build:typescript && node scripts/generate-replay-goldens.mjs --check",
|
||||
"check:protocol": "pnpm run typecheck && pnpm run check:protocol-manifest && pnpm test && pnpm run check:replay-goldens"
|
||||
"check:protocol": "pnpm run typecheck:typescript && pnpm run check:protocol-manifest && pnpm run test:typescript && pnpm run check:replay-goldens",
|
||||
"check:conformance-parity": "cargo test --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core runs_the_mock_core_path_with_stable_output",
|
||||
"check:replay-parity": "cargo test --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core replay_fixture_parity",
|
||||
"check:runner": "pnpm run typecheck:rust && pnpm run test:rust && pnpm run check:conformance-parity && pnpm run check:replay-parity",
|
||||
"check:all": "pnpm run check:protocol && pnpm run check:runner",
|
||||
"trace:conformance:rust": "cargo run --quiet --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core --bin conformance-tracer"
|
||||
},
|
||||
"dependencies": {
|
||||
"ajv": "^8.20.0",
|
||||
|
|
|
|||
|
|
@ -35,3 +35,8 @@ The files in `fixtures/replay/golden/` are deterministic reducer oracles. Each
|
|||
accepted replay fixture has a complete session snapshot and a compact parity
|
||||
summary. `pnpm generate:replay-goldens` updates them after an intentional
|
||||
reducer change; package build and CI fail when they drift.
|
||||
|
||||
The files in `fixtures/local-runner/scripts/` drive the package-local fake
|
||||
harness. They cover successful, failed, interrupted, interactive, duplicate
|
||||
terminal, process-cleanup, and oversized-frame behavior without starting a
|
||||
production adapter.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,45 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "duplicate-terminal",
|
||||
"steps": [
|
||||
{
|
||||
"kind": "result",
|
||||
"payload": {
|
||||
"schema": "paperclip.run_result.v1",
|
||||
"reportedWorkDisposition": "done",
|
||||
"summary": "The duplicate terminal guard accepted one result.",
|
||||
"completionClaim": {
|
||||
"contractRevision": "local_runner-contract-v1",
|
||||
"objectiveSatisfied": true,
|
||||
"criteria": [],
|
||||
"remainingWork": []
|
||||
},
|
||||
"evidence": [],
|
||||
"verification": [],
|
||||
"attentionRequests": [],
|
||||
"artifacts": []
|
||||
}
|
||||
},
|
||||
{ "kind": "event", "event_type": "turn.completed", "payload": {} },
|
||||
{ "kind": "event", "event_type": "session.closed", "payload": {} },
|
||||
{
|
||||
"kind": "terminal",
|
||||
"payload": {
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "completed",
|
||||
"runTerminalState": "succeeded",
|
||||
"reportedWorkDisposition": "done"
|
||||
}
|
||||
},
|
||||
{
|
||||
"kind": "terminal",
|
||||
"payload": {
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "failed",
|
||||
"runTerminalState": "failed",
|
||||
"reportedWorkDisposition": "yielded"
|
||||
}
|
||||
},
|
||||
{ "kind": "exit", "code": 0 }
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "error",
|
||||
"steps": [
|
||||
{ "kind": "delay", "milliseconds": 5 },
|
||||
{ "kind": "log", "stream": "stderr", "text": "scripted harness failure" },
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.failed",
|
||||
"item_id": "item_local_runner_error",
|
||||
"payload": { "kind": "diagnostic", "message": "The fake harness failed as requested." }
|
||||
},
|
||||
{
|
||||
"kind": "result",
|
||||
"payload": {
|
||||
"schema": "paperclip.run_result.v1",
|
||||
"reportedWorkDisposition": "yielded",
|
||||
"summary": "The fake harness reported a scripted error.",
|
||||
"completionClaim": {
|
||||
"contractRevision": "local_runner-contract-v1",
|
||||
"objectiveSatisfied": false,
|
||||
"criteria": [],
|
||||
"remainingWork": [
|
||||
{ "description": "Inspect the scripted failure.", "blocksCompletion": true }
|
||||
]
|
||||
},
|
||||
"evidence": [],
|
||||
"verification": [
|
||||
{ "commandOrCheck": "scripted error scenario", "status": "failed" }
|
||||
],
|
||||
"attentionRequests": [],
|
||||
"artifacts": [],
|
||||
"continuation": {
|
||||
"kind": "retry",
|
||||
"summary": "Retry with a successful script.",
|
||||
"idempotencyKey": "local_runner-error-retry"
|
||||
}
|
||||
}
|
||||
},
|
||||
{ "kind": "event", "event_type": "turn.failed", "payload": {} },
|
||||
{ "kind": "event", "event_type": "session.failed", "payload": {} },
|
||||
{
|
||||
"kind": "terminal",
|
||||
"payload": {
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "failed",
|
||||
"runTerminalState": "failed",
|
||||
"reportedWorkDisposition": "yielded"
|
||||
}
|
||||
},
|
||||
{ "kind": "exit", "code": 7 }
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,90 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "happy-path",
|
||||
"steps": [
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.started",
|
||||
"item_id": "item_local_runner_command",
|
||||
"payload": { "kind": "command", "text": "pnpm test --filter fake" }
|
||||
},
|
||||
{ "kind": "delay", "milliseconds": 15 },
|
||||
{ "kind": "log", "stream": "stdout", "text": "prepare workspace" },
|
||||
{ "kind": "log", "stream": "stdout", "text": "load scripted driver" },
|
||||
{ "kind": "log", "stream": "stdout", "text": "run deterministic check" },
|
||||
{ "kind": "log", "stream": "stdout", "text": "write fixture output" },
|
||||
{ "kind": "log", "stream": "stdout", "text": "verification passed" },
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.delta",
|
||||
"item_id": "item_local_runner_command",
|
||||
"payload": { "text": "Fake driver output is live. " }
|
||||
},
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.completed",
|
||||
"item_id": "item_local_runner_command",
|
||||
"payload": { "kind": "command", "text": "Fake driver check passed." }
|
||||
},
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.started",
|
||||
"item_id": "item_local_runner_file",
|
||||
"payload": { "kind": "file_change", "text": "local-runner-output.json" }
|
||||
},
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.completed",
|
||||
"item_id": "item_local_runner_file",
|
||||
"payload": { "kind": "file_change", "text": "Wrote local-runner-output.json" }
|
||||
},
|
||||
{
|
||||
"kind": "result",
|
||||
"payload": {
|
||||
"schema": "paperclip.run_result.v1",
|
||||
"reportedWorkDisposition": "done",
|
||||
"summary": "The live fake-harness run completed successfully.",
|
||||
"completionClaim": {
|
||||
"contractRevision": "local_runner-contract-v1",
|
||||
"objectiveSatisfied": true,
|
||||
"criteria": [
|
||||
{
|
||||
"criterionId": "local-runner",
|
||||
"status": "satisfied",
|
||||
"evidenceRefs": ["event:item_local_runner_command"]
|
||||
}
|
||||
],
|
||||
"remainingWork": []
|
||||
},
|
||||
"evidence": [{ "kind": "event", "ref": "item_local_runner_command" }],
|
||||
"verification": [
|
||||
{ "commandOrCheck": "fake driver conformance", "status": "passed" }
|
||||
],
|
||||
"attentionRequests": [],
|
||||
"artifacts": [
|
||||
{ "kind": "fixture", "ref": "local-runner-output.json", "title": "Live trace" }
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "turn.completed",
|
||||
"payload": {}
|
||||
},
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "session.closed",
|
||||
"payload": { "reason": "script_complete" }
|
||||
},
|
||||
{
|
||||
"kind": "terminal",
|
||||
"payload": {
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "completed",
|
||||
"runTerminalState": "succeeded",
|
||||
"reportedWorkDisposition": "done"
|
||||
}
|
||||
},
|
||||
{ "kind": "exit", "code": 0 }
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,58 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "interrupted",
|
||||
"steps": [
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.started",
|
||||
"item_id": "item_local_runner_long_command",
|
||||
"payload": { "kind": "command", "text": "fake long-running command" }
|
||||
},
|
||||
{ "kind": "log", "stream": "stdout", "text": "waiting for interruption" },
|
||||
{ "kind": "await_interrupt" },
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.failed",
|
||||
"item_id": "item_local_runner_long_command",
|
||||
"payload": { "kind": "command", "message": "Interrupted by the operator." }
|
||||
},
|
||||
{
|
||||
"kind": "result",
|
||||
"payload": {
|
||||
"schema": "paperclip.run_result.v1",
|
||||
"reportedWorkDisposition": "yielded",
|
||||
"summary": "The scripted turn was interrupted.",
|
||||
"completionClaim": {
|
||||
"contractRevision": "local_runner-contract-v1",
|
||||
"objectiveSatisfied": false,
|
||||
"criteria": [],
|
||||
"remainingWork": [
|
||||
{ "description": "Start a replacement turn.", "blocksCompletion": true }
|
||||
]
|
||||
},
|
||||
"evidence": [],
|
||||
"verification": [
|
||||
{ "commandOrCheck": "turn interruption", "status": "passed" }
|
||||
],
|
||||
"attentionRequests": [],
|
||||
"artifacts": [],
|
||||
"continuation": {
|
||||
"kind": "same_agent",
|
||||
"summary": "A replacement turn can continue the session.",
|
||||
"idempotencyKey": "local_runner-interrupted-continuation"
|
||||
}
|
||||
}
|
||||
},
|
||||
{ "kind": "event", "event_type": "session.closed", "payload": {} },
|
||||
{
|
||||
"kind": "terminal",
|
||||
"payload": {
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "interrupted",
|
||||
"runTerminalState": "cancelled",
|
||||
"reportedWorkDisposition": "yielded"
|
||||
}
|
||||
},
|
||||
{ "kind": "exit", "code": 130 }
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "linger",
|
||||
"steps": [
|
||||
{ "kind": "spawn_worker" },
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.started",
|
||||
"item_id": "item_local_runner_linger",
|
||||
"payload": { "kind": "command", "text": "wait for supervisor cleanup" }
|
||||
},
|
||||
{ "kind": "await_interrupt" },
|
||||
{ "kind": "exit", "code": 130 }
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,12 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "oversized-line",
|
||||
"steps": [
|
||||
{
|
||||
"kind": "log",
|
||||
"stream": "stdout",
|
||||
"text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
|
||||
},
|
||||
{ "kind": "exit", "code": 0 }
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
{
|
||||
"schema": "paperclip.fake_harness.script.v1",
|
||||
"name": "permission-input",
|
||||
"steps": [
|
||||
{
|
||||
"kind": "request",
|
||||
"request": {
|
||||
"schema": "paperclip.runtime_request.v1",
|
||||
"requestKind": "runtime",
|
||||
"requestId": "request_local_runner_permission",
|
||||
"type": "permission",
|
||||
"status": "pending",
|
||||
"prompt": "Allow the fake driver to write its local fixture?",
|
||||
"choices": [
|
||||
{ "key": "allow", "label": "Allow" },
|
||||
{ "key": "deny", "label": "Deny" }
|
||||
]
|
||||
}
|
||||
},
|
||||
{ "kind": "await_resolution", "request_id": "request_local_runner_permission" },
|
||||
{
|
||||
"kind": "request",
|
||||
"request": {
|
||||
"schema": "paperclip.runtime_request.v1",
|
||||
"requestKind": "runtime",
|
||||
"requestId": "request_local_runner_input",
|
||||
"type": "input",
|
||||
"status": "pending",
|
||||
"prompt": "Name the local trace.",
|
||||
"choices": []
|
||||
}
|
||||
},
|
||||
{ "kind": "await_resolution", "request_id": "request_local_runner_input" },
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.started",
|
||||
"item_id": "item_local_runner_permission",
|
||||
"payload": { "kind": "assistant_message", "text": "" }
|
||||
},
|
||||
{ "kind": "delay", "milliseconds": 10 },
|
||||
{
|
||||
"kind": "event",
|
||||
"event_type": "item.completed",
|
||||
"item_id": "item_local_runner_permission",
|
||||
"payload": {
|
||||
"kind": "assistant_message",
|
||||
"text": "Permission and input were resolved through the local protocol."
|
||||
}
|
||||
},
|
||||
{
|
||||
"kind": "result",
|
||||
"payload": {
|
||||
"schema": "paperclip.run_result.v1",
|
||||
"reportedWorkDisposition": "done",
|
||||
"summary": "The permission and input flow completed.",
|
||||
"completionClaim": {
|
||||
"contractRevision": "local_runner-contract-v1",
|
||||
"objectiveSatisfied": true,
|
||||
"criteria": [],
|
||||
"remainingWork": []
|
||||
},
|
||||
"evidence": [],
|
||||
"verification": [
|
||||
{ "commandOrCheck": "permission and input round trip", "status": "passed" }
|
||||
],
|
||||
"attentionRequests": [],
|
||||
"artifacts": []
|
||||
}
|
||||
},
|
||||
{ "kind": "event", "event_type": "turn.completed", "payload": {} },
|
||||
{ "kind": "event", "event_type": "session.closed", "payload": {} },
|
||||
{
|
||||
"kind": "terminal",
|
||||
"payload": {
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "completed",
|
||||
"runTerminalState": "succeeded",
|
||||
"reportedWorkDisposition": "done"
|
||||
}
|
||||
},
|
||||
{ "kind": "exit", "code": 0 }
|
||||
]
|
||||
}
|
||||
|
|
@ -121,6 +121,48 @@
|
|||
"expectation": "accept",
|
||||
"compatibilityCase": "cross-language-input"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/duplicate-terminal.json",
|
||||
"sha256": "56928a21318ea4b390c65fb34f0d3114760619d79301f4d535391a8566b693ab",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/error.json",
|
||||
"sha256": "a4c74f8a85f8ab25adb13e83ae753fcac1a9cb289bad44d49be05df8f3170edb",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/happy-path.json",
|
||||
"sha256": "ccdeef3df1e2020b692da3bd361958643f2f3d5ec6e1e0b16cca9e0d9890da2f",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/interrupted.json",
|
||||
"sha256": "36dc94dc1a67d320a48921f2ed10f0e7ab760c518eea84d5f224e1e6ab8fa57c",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/linger.json",
|
||||
"sha256": "ac6fed60bea22b22c16501c03388828ccaae6e9ec44ceb2ec0e8b145ab590bb8",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/oversized-line.json",
|
||||
"sha256": "485b03653a2e41351f155b787a406f903e018b3c29b19c4ddf2dfc7c031552b7",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/local-runner/scripts/permission-input.json",
|
||||
"sha256": "9b64b78ed75006b9190c80fd88c5ab555ef458f7ad023e8ebfdac8de373e1f9c",
|
||||
"expectation": "accept",
|
||||
"compatibilityCase": "canonical"
|
||||
},
|
||||
{
|
||||
"path": "fixtures/questions/codex.json",
|
||||
"sha256": "ad673b6f9d70b74b53656e622d15fad1a66430f39d62b9ef71479d194a198bd6",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,107 @@
|
|||
# This file is automatically @generated by Cargo.
|
||||
# It is not intended for manual editing.
|
||||
version = 4
|
||||
|
||||
[[package]]
|
||||
name = "itoa"
|
||||
version = "1.0.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
|
||||
|
||||
[[package]]
|
||||
name = "memchr"
|
||||
version = "2.8.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
|
||||
|
||||
[[package]]
|
||||
name = "paperclip-runner-core"
|
||||
version = "0.0.0"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro2"
|
||||
version = "1.0.107"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
|
||||
dependencies = [
|
||||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quote"
|
||||
version = "1.0.47"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde"
|
||||
version = "1.0.229"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
|
||||
dependencies = [
|
||||
"serde_core",
|
||||
"serde_derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_core"
|
||||
version = "1.0.229"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
|
||||
dependencies = [
|
||||
"serde_derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_derive"
|
||||
version = "1.0.229"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_json"
|
||||
version = "1.0.151"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
|
||||
dependencies = [
|
||||
"itoa",
|
||||
"memchr",
|
||||
"serde",
|
||||
"serde_core",
|
||||
"zmij",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "3.0.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "unicode-ident"
|
||||
version = "1.0.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
|
||||
|
||||
[[package]]
|
||||
name = "zmij"
|
||||
version = "1.0.23"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
[workspace]
|
||||
members = ["crates/runner-core"]
|
||||
resolver = "2"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.0.0"
|
||||
edition = "2021"
|
||||
license = "MIT"
|
||||
publish = false
|
||||
|
||||
[workspace.dependencies]
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
[package]
|
||||
name = "paperclip-runner-core"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
publish.workspace = true
|
||||
|
||||
[dependencies]
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
|
||||
[[bin]]
|
||||
name = "conformance-tracer"
|
||||
path = "src/bin/conformance-tracer.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "paperclip-runnerd"
|
||||
path = "src/bin/paperclip-runnerd.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "fake-harness"
|
||||
path = "src/bin/fake-harness.rs"
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
use paperclip_runner_core::{run_conformance_tracer, CONFORMANCE_FIXTURE};
|
||||
|
||||
fn main() {
|
||||
match run_conformance_tracer(CONFORMANCE_FIXTURE) {
|
||||
Ok(output) => println!("{output}"),
|
||||
Err(error) => {
|
||||
eprintln!("Conformance tracer failed: {error}");
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
use std::path::PathBuf;
|
||||
use std::process::ExitCode;
|
||||
|
||||
use paperclip_runner_core::fake_harness::{load_fake_harness_script, run_fake_harness};
|
||||
use paperclip_runner_core::local_runner::LocalRunnerError;
|
||||
|
||||
fn value(args: &[String], name: &str) -> Result<String, LocalRunnerError> {
|
||||
let index = args
|
||||
.iter()
|
||||
.position(|argument| argument == name)
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("missing required argument {name}")))?;
|
||||
args.get(index + 1)
|
||||
.cloned()
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("missing value for {name}")))
|
||||
}
|
||||
|
||||
fn run() -> Result<i32, LocalRunnerError> {
|
||||
let args = std::env::args().skip(1).collect::<Vec<_>>();
|
||||
let script = load_fake_harness_script(&PathBuf::from(value(&args, "--script")?))?;
|
||||
let delay_override_ms = args
|
||||
.iter()
|
||||
.position(|argument| argument == "--delay-ms")
|
||||
.map(|index| {
|
||||
args.get(index + 1)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("missing value for --delay-ms"))?
|
||||
.parse::<u64>()
|
||||
.map_err(|error| LocalRunnerError::invalid(format!("invalid --delay-ms: {error}")))
|
||||
})
|
||||
.transpose()?;
|
||||
run_fake_harness(script, delay_override_ms)
|
||||
}
|
||||
|
||||
fn main() -> ExitCode {
|
||||
match run() {
|
||||
Ok(code) => ExitCode::from(u8::try_from(code.clamp(0, 255)).unwrap_or(1)),
|
||||
Err(error) => {
|
||||
eprintln!("fake-harness: {error}");
|
||||
ExitCode::FAILURE
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,65 @@
|
|||
use std::path::PathBuf;
|
||||
use std::process::ExitCode;
|
||||
use std::time::Duration;
|
||||
|
||||
use paperclip_runner_core::local_runner::{run_local_runner, LocalRunnerError, RunnerConfig};
|
||||
|
||||
fn value(args: &[String], name: &str) -> Result<String, LocalRunnerError> {
|
||||
let index = args
|
||||
.iter()
|
||||
.position(|argument| argument == name)
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("missing required argument {name}")))?;
|
||||
args.get(index + 1)
|
||||
.cloned()
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("missing value for {name}")))
|
||||
}
|
||||
|
||||
fn optional_u64(args: &[String], name: &str) -> Result<Option<u64>, LocalRunnerError> {
|
||||
let Some(index) = args.iter().position(|argument| argument == name) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let value = args
|
||||
.get(index + 1)
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("missing value for {name}")))?;
|
||||
value
|
||||
.parse::<u64>()
|
||||
.map(Some)
|
||||
.map_err(|error| LocalRunnerError::invalid(format!("invalid {name}: {error}")))
|
||||
}
|
||||
|
||||
fn usize_value(args: &[String], name: &str, default: usize) -> Result<usize, LocalRunnerError> {
|
||||
optional_u64(args, name)?.map_or(Ok(default), |value| {
|
||||
usize::try_from(value)
|
||||
.map_err(|error| LocalRunnerError::invalid(format!("invalid {name}: {error}")))
|
||||
})
|
||||
}
|
||||
|
||||
fn run() -> Result<(), LocalRunnerError> {
|
||||
let args = std::env::args().skip(1).collect::<Vec<_>>();
|
||||
run_local_runner(RunnerConfig {
|
||||
run_id: value(&args, "--run-id")?,
|
||||
normalized_session_id: value(&args, "--session-id")?,
|
||||
runner_instance_id: value(&args, "--runner-id")?,
|
||||
fake_harness_path: PathBuf::from(value(&args, "--fake-harness")?),
|
||||
script_path: PathBuf::from(value(&args, "--script")?),
|
||||
delay_override_ms: optional_u64(&args, "--delay-ms")?,
|
||||
log_max_lines: usize_value(&args, "--log-max-lines", 32)?,
|
||||
log_max_bytes: usize_value(&args, "--log-max-bytes", 16_384)?,
|
||||
command_history_limit: usize_value(&args, "--command-history-limit", 4096)?,
|
||||
controller_max_line_bytes: usize_value(&args, "--controller-max-line-bytes", 64 * 1024)?,
|
||||
harness_max_line_bytes: usize_value(&args, "--harness-max-line-bytes", 64 * 1024)?,
|
||||
shutdown_grace: Duration::from_millis(
|
||||
optional_u64(&args, "--shutdown-grace-ms")?.unwrap_or(100),
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
fn main() -> ExitCode {
|
||||
match run() {
|
||||
Ok(()) => ExitCode::SUCCESS,
|
||||
Err(error) => {
|
||||
eprintln!("paperclip-runnerd: {error}");
|
||||
ExitCode::FAILURE
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,327 @@
|
|||
use std::io::{Read, Write};
|
||||
use std::path::Path;
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::sync::mpsc::{self, Receiver};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde::Deserialize;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use crate::local_runner::{HarnessCommand, LocalRunnerError};
|
||||
use crate::process_supervisor::{read_bounded_line, BoundedLine};
|
||||
|
||||
const HARNESS_COMMAND_SCHEMA: &str = "paperclip.fake_harness.command.v1";
|
||||
const HARNESS_MESSAGE_SCHEMA: &str = "paperclip.fake_harness.message.v1";
|
||||
const HARNESS_COMMAND_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
const INTERACTIVE_REQUEST_TIMEOUT: Duration = Duration::from_secs(5 * 60);
|
||||
const HARNESS_COMMAND_QUEUE_CAPACITY: usize = 256;
|
||||
const HARNESS_MAX_COMMAND_BYTES: usize = 64 * 1024;
|
||||
const HARNESS_MAX_SCRIPT_BYTES: u64 = 1024 * 1024;
|
||||
const HARNESS_MAX_SCRIPT_STEPS: usize = 4096;
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct FakeHarnessScript {
|
||||
pub schema: String,
|
||||
pub name: String,
|
||||
pub steps: Vec<FakeHarnessStep>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
pub enum FakeHarnessStep {
|
||||
Delay {
|
||||
milliseconds: u64,
|
||||
},
|
||||
Log {
|
||||
stream: String,
|
||||
text: String,
|
||||
},
|
||||
Event {
|
||||
event_type: String,
|
||||
#[serde(default)]
|
||||
turn_id: Option<String>,
|
||||
#[serde(default)]
|
||||
item_id: Option<String>,
|
||||
#[serde(default)]
|
||||
payload: Value,
|
||||
},
|
||||
Request {
|
||||
request: Value,
|
||||
},
|
||||
AwaitResolution {
|
||||
request_id: String,
|
||||
},
|
||||
AwaitInterrupt,
|
||||
Result {
|
||||
payload: Value,
|
||||
},
|
||||
Terminal {
|
||||
payload: Value,
|
||||
},
|
||||
SpawnWorker,
|
||||
Exit {
|
||||
code: i32,
|
||||
},
|
||||
}
|
||||
|
||||
pub fn load_fake_harness_script(path: &Path) -> Result<FakeHarnessScript, LocalRunnerError> {
|
||||
let file = std::fs::File::open(path).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!(
|
||||
"failed to read fake harness script {}: {error}",
|
||||
path.display()
|
||||
))
|
||||
})?;
|
||||
let mut input = String::new();
|
||||
file.take(HARNESS_MAX_SCRIPT_BYTES + 1)
|
||||
.read_to_string(&mut input)
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!(
|
||||
"failed to read fake harness script {}: {error}",
|
||||
path.display()
|
||||
))
|
||||
})?;
|
||||
if input.len() as u64 > HARNESS_MAX_SCRIPT_BYTES {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"fake harness script cannot exceed {HARNESS_MAX_SCRIPT_BYTES} bytes"
|
||||
)));
|
||||
}
|
||||
let script: FakeHarnessScript = serde_json::from_str(&input).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("fake harness script must be valid JSON: {error}"))
|
||||
})?;
|
||||
if script.schema != "paperclip.fake_harness.script.v1" {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness script schema is unsupported",
|
||||
));
|
||||
}
|
||||
if script.name.trim().is_empty() || script.name.len() > 160 || script.steps.is_empty() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness script requires a name of at most 160 bytes and at least one step",
|
||||
));
|
||||
}
|
||||
if script.steps.len() > HARNESS_MAX_SCRIPT_STEPS {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"fake harness script cannot exceed {HARNESS_MAX_SCRIPT_STEPS} steps"
|
||||
)));
|
||||
}
|
||||
Ok(script)
|
||||
}
|
||||
|
||||
pub fn run_fake_harness(
|
||||
script: FakeHarnessScript,
|
||||
delay_override_ms: Option<u64>,
|
||||
) -> Result<i32, LocalRunnerError> {
|
||||
let (command_sender, command_receiver) =
|
||||
mpsc::sync_channel::<HarnessCommand>(HARNESS_COMMAND_QUEUE_CAPACITY);
|
||||
thread::spawn(move || {
|
||||
let stdin = std::io::stdin();
|
||||
let mut reader = stdin.lock();
|
||||
loop {
|
||||
match read_bounded_line(&mut reader, HARNESS_MAX_COMMAND_BYTES) {
|
||||
Ok(BoundedLine::Line(line)) if line.trim().is_empty() => {}
|
||||
Ok(BoundedLine::Line(line)) => {
|
||||
if let Ok(command) = serde_json::from_str::<HarnessCommand>(&line) {
|
||||
if command_sender.send(command).is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(BoundedLine::TooLong) => {}
|
||||
Ok(BoundedLine::Eof) | Err(_) => return,
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let mut output = FakeHarnessOutput::default();
|
||||
output.send(
|
||||
"ready",
|
||||
json!({ "version": "1.0.0", "script": script.name }),
|
||||
)?;
|
||||
receive_harness_command(&command_receiver, "session.open", None)?;
|
||||
output.send_event(
|
||||
"session.started",
|
||||
None,
|
||||
None,
|
||||
json!({ "driverSessionId": "driver_local_runner_fake" }),
|
||||
)?;
|
||||
output.send_event(
|
||||
"runtime.phase.changed",
|
||||
None,
|
||||
None,
|
||||
json!({ "phase": "executing" }),
|
||||
)?;
|
||||
let turn = receive_harness_command(&command_receiver, "turn.start", None)?;
|
||||
let turn_id = turn
|
||||
.payload
|
||||
.get("turnId")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("turn_local_runner")
|
||||
.to_owned();
|
||||
output.send_event("turn.started", Some(&turn_id), None, json!({}))?;
|
||||
|
||||
let mut workers = Vec::<Child>::new();
|
||||
for step in script.steps {
|
||||
match step {
|
||||
FakeHarnessStep::Delay { milliseconds } => {
|
||||
thread::sleep(Duration::from_millis(
|
||||
delay_override_ms.unwrap_or(milliseconds),
|
||||
));
|
||||
}
|
||||
FakeHarnessStep::Log { stream, text } => {
|
||||
output.send("log", json!({ "stream": stream, "text": text }))?;
|
||||
}
|
||||
FakeHarnessStep::Event {
|
||||
event_type,
|
||||
turn_id: step_turn_id,
|
||||
item_id,
|
||||
payload,
|
||||
} => {
|
||||
output.send_event(
|
||||
&event_type,
|
||||
step_turn_id.as_deref().or(Some(&turn_id)),
|
||||
item_id.as_deref(),
|
||||
payload,
|
||||
)?;
|
||||
}
|
||||
FakeHarnessStep::Request { request } => {
|
||||
output.send("request", json!({ "request": request }))?;
|
||||
}
|
||||
FakeHarnessStep::AwaitResolution { request_id } => {
|
||||
let command = receive_harness_command(
|
||||
&command_receiver,
|
||||
"request.resolve",
|
||||
Some(&request_id),
|
||||
)?;
|
||||
output.send_event(
|
||||
"runtime_request.resolved",
|
||||
Some(&turn_id),
|
||||
None,
|
||||
json!({
|
||||
"requestId": request_id,
|
||||
"response": command.payload.get("response").cloned().unwrap_or(Value::Null)
|
||||
}),
|
||||
)?;
|
||||
}
|
||||
FakeHarnessStep::AwaitInterrupt => {
|
||||
let command = receive_harness_command(&command_receiver, "turn.interrupt", None)?;
|
||||
output.send_event(
|
||||
"turn.interrupted",
|
||||
Some(&turn_id),
|
||||
None,
|
||||
json!({ "reason": command.payload.get("reason").cloned().unwrap_or_else(|| json!("operator")) }),
|
||||
)?;
|
||||
}
|
||||
FakeHarnessStep::Result { payload } => {
|
||||
output.send("result", json!({ "result": payload }))?;
|
||||
}
|
||||
FakeHarnessStep::Terminal { payload } => {
|
||||
output.send("terminal", json!({ "terminal": payload }))?;
|
||||
}
|
||||
FakeHarnessStep::SpawnWorker => {
|
||||
let worker = Command::new("sh")
|
||||
.args(["-c", "sleep 60"])
|
||||
.env_clear()
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.spawn()
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("failed to spawn fake worker: {error}"))
|
||||
})?;
|
||||
output.send(
|
||||
"diagnostic",
|
||||
json!({ "workerPid": worker.id(), "purpose": "cleanup_conformance" }),
|
||||
)?;
|
||||
workers.push(worker);
|
||||
}
|
||||
FakeHarnessStep::Exit { code } => {
|
||||
drop(workers);
|
||||
return Ok(code);
|
||||
}
|
||||
}
|
||||
}
|
||||
drop(workers);
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
fn receive_harness_command(
|
||||
receiver: &Receiver<HarnessCommand>,
|
||||
command_type: &str,
|
||||
request_id: Option<&str>,
|
||||
) -> Result<HarnessCommand, LocalRunnerError> {
|
||||
let timeout = if command_type == "request.resolve" {
|
||||
INTERACTIVE_REQUEST_TIMEOUT
|
||||
} else {
|
||||
HARNESS_COMMAND_TIMEOUT
|
||||
};
|
||||
let deadline = Instant::now() + timeout;
|
||||
loop {
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
if remaining.is_zero() {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"timed out waiting for {command_type}"
|
||||
)));
|
||||
}
|
||||
let command = receiver.recv_timeout(remaining).map_err(|_| {
|
||||
LocalRunnerError::invalid(format!("timed out waiting for {command_type}"))
|
||||
})?;
|
||||
if command.schema != HARNESS_COMMAND_SCHEMA {
|
||||
continue;
|
||||
}
|
||||
if command.command_type != command_type {
|
||||
continue;
|
||||
}
|
||||
if let Some(request_id) = request_id {
|
||||
if command.payload.get("requestId").and_then(Value::as_str) != Some(request_id) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return Ok(command);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FakeHarnessOutput {
|
||||
sequence: u64,
|
||||
}
|
||||
|
||||
impl FakeHarnessOutput {
|
||||
fn send(&mut self, message_type: &str, payload: Value) -> Result<(), LocalRunnerError> {
|
||||
self.sequence += 1;
|
||||
let message = json!({
|
||||
"schema": HARNESS_MESSAGE_SCHEMA,
|
||||
"messageSeq": self.sequence,
|
||||
"type": message_type,
|
||||
"payload": payload,
|
||||
});
|
||||
let stdout = std::io::stdout();
|
||||
let mut lock = stdout.lock();
|
||||
serde_json::to_writer(&mut lock, &message).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("harness message serialization failed: {error}"))
|
||||
})?;
|
||||
lock.write_all(b"\n")
|
||||
.and_then(|_| lock.flush())
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("harness message write failed: {error}"))
|
||||
})
|
||||
}
|
||||
|
||||
fn send_event(
|
||||
&mut self,
|
||||
event_type: &str,
|
||||
turn_id: Option<&str>,
|
||||
item_id: Option<&str>,
|
||||
payload: Value,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
self.send(
|
||||
"event",
|
||||
json!({
|
||||
"eventType": event_type,
|
||||
"turnId": turn_id,
|
||||
"itemId": item_id,
|
||||
"payload": payload,
|
||||
}),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,342 @@
|
|||
#![forbid(unsafe_code)]
|
||||
|
||||
pub mod fake_harness;
|
||||
pub mod local_runner;
|
||||
pub mod process_supervisor;
|
||||
pub mod replay;
|
||||
|
||||
use std::error::Error;
|
||||
use std::fmt::{self, Display, Formatter};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
pub const CONFORMANCE_FIXTURE_SCHEMA: &str = "paperclip.runner.conformance.fixture.v1";
|
||||
pub const CONFORMANCE_OUTPUT_SCHEMA: &str = "paperclip.runner.conformance.output.v1";
|
||||
pub const CONFORMANCE_FIXTURE: &str =
|
||||
include_str!("../../../../protocol/fixtures/conformance-minimal-run.json");
|
||||
pub const CONFORMANCE_EXPECTED_OUTPUT: &str =
|
||||
include_str!("../../../../protocol/fixtures/conformance-expected-output.json");
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct NativeRunIdentity {
|
||||
pub run_id: String,
|
||||
pub session_id: String,
|
||||
pub company_id: String,
|
||||
pub issue_id: String,
|
||||
pub agent_id: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct NativeRunEvent {
|
||||
pub event_id: String,
|
||||
pub run_id: String,
|
||||
pub sequence: u64,
|
||||
#[serde(rename = "type")]
|
||||
pub event_type: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct NativeRunResult {
|
||||
pub run_id: String,
|
||||
pub status: String,
|
||||
pub summary: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ConformanceFixture {
|
||||
pub schema_version: String,
|
||||
pub run: NativeRunIdentity,
|
||||
pub events: Vec<NativeRunEvent>,
|
||||
pub result: NativeRunResult,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ConformanceError(String);
|
||||
|
||||
impl ConformanceError {
|
||||
fn invalid(message: impl Into<String>) -> Self {
|
||||
Self(message.into())
|
||||
}
|
||||
}
|
||||
|
||||
impl Display for ConformanceError {
|
||||
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
|
||||
formatter.write_str(&self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for ConformanceError {}
|
||||
|
||||
fn validate_identifier(value: &str, path: &str) -> Result<(), ConformanceError> {
|
||||
let mut characters = value.chars();
|
||||
let starts_lowercase = characters
|
||||
.next()
|
||||
.is_some_and(|character| character.is_ascii_lowercase());
|
||||
let remainder_is_valid = characters.all(|character| {
|
||||
character.is_ascii_lowercase() || character.is_ascii_digit() || character == '_'
|
||||
});
|
||||
|
||||
if starts_lowercase && remainder_is_valid {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(ConformanceError::invalid(format!(
|
||||
"{path} must be a stable lowercase identifier"
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn validate_conformance_fixture(input: &str) -> Result<ConformanceFixture, ConformanceError> {
|
||||
let fixture: ConformanceFixture = serde_json::from_str(input).map_err(|error| {
|
||||
ConformanceError::invalid(format!("fixture must be valid JSON: {error}"))
|
||||
})?;
|
||||
|
||||
if fixture.schema_version != CONFORMANCE_FIXTURE_SCHEMA {
|
||||
return Err(ConformanceError::invalid(format!(
|
||||
"schemaVersion must be {CONFORMANCE_FIXTURE_SCHEMA}"
|
||||
)));
|
||||
}
|
||||
|
||||
for (path, value) in [
|
||||
("run.runId", fixture.run.run_id.as_str()),
|
||||
("run.sessionId", fixture.run.session_id.as_str()),
|
||||
("run.companyId", fixture.run.company_id.as_str()),
|
||||
("run.issueId", fixture.run.issue_id.as_str()),
|
||||
("run.agentId", fixture.run.agent_id.as_str()),
|
||||
] {
|
||||
validate_identifier(value, path)?;
|
||||
}
|
||||
|
||||
if fixture.events.len() < 2 {
|
||||
return Err(ConformanceError::invalid(
|
||||
"events must contain at least run.started and run.completed",
|
||||
));
|
||||
}
|
||||
|
||||
for (index, event) in fixture.events.iter().enumerate() {
|
||||
validate_identifier(&event.event_id, &format!("events[{index}].eventId"))?;
|
||||
if event.run_id != fixture.run.run_id {
|
||||
return Err(ConformanceError::invalid(format!(
|
||||
"events[{index}].runId must match run.runId"
|
||||
)));
|
||||
}
|
||||
let expected_sequence = index as u64 + 1;
|
||||
if event.sequence != expected_sequence {
|
||||
return Err(ConformanceError::invalid(format!(
|
||||
"events[{index}].sequence must be {expected_sequence}"
|
||||
)));
|
||||
}
|
||||
if !matches!(event.event_type.as_str(), "run.started" | "run.completed") {
|
||||
return Err(ConformanceError::invalid(format!(
|
||||
"events[{index}].type is unsupported"
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
if fixture
|
||||
.events
|
||||
.first()
|
||||
.map(|event| event.event_type.as_str())
|
||||
!= Some("run.started")
|
||||
{
|
||||
return Err(ConformanceError::invalid(
|
||||
"the first event must be run.started",
|
||||
));
|
||||
}
|
||||
if fixture.events.last().map(|event| event.event_type.as_str()) != Some("run.completed") {
|
||||
return Err(ConformanceError::invalid(
|
||||
"the final event must be run.completed",
|
||||
));
|
||||
}
|
||||
if fixture.result.run_id != fixture.run.run_id {
|
||||
return Err(ConformanceError::invalid(
|
||||
"result.runId must match run.runId",
|
||||
));
|
||||
}
|
||||
if fixture.result.status != "succeeded" {
|
||||
return Err(ConformanceError::invalid(
|
||||
"the Conformance fixture result must be succeeded",
|
||||
));
|
||||
}
|
||||
if fixture.result.summary.trim().is_empty() {
|
||||
return Err(ConformanceError::invalid(
|
||||
"result.summary must be a non-empty string",
|
||||
));
|
||||
}
|
||||
|
||||
Ok(fixture)
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct MockControlPlane {
|
||||
running: bool,
|
||||
opened_run_id: Option<String>,
|
||||
events: Vec<NativeRunEvent>,
|
||||
result: Option<NativeRunResult>,
|
||||
}
|
||||
|
||||
impl MockControlPlane {
|
||||
pub fn start(&mut self) -> Result<(), ConformanceError> {
|
||||
if self.running {
|
||||
return Err(ConformanceError::invalid(
|
||||
"mock control plane is already running",
|
||||
));
|
||||
}
|
||||
self.running = true;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn stop(&mut self) {
|
||||
self.running = false;
|
||||
}
|
||||
|
||||
pub fn open_run(&mut self, identity: &NativeRunIdentity) -> Result<(), ConformanceError> {
|
||||
self.require_running()?;
|
||||
if self.opened_run_id.is_some() {
|
||||
return Err(ConformanceError::invalid(
|
||||
"mock control plane accepts one Conformance run",
|
||||
));
|
||||
}
|
||||
self.opened_run_id = Some(identity.run_id.clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn append_event(&mut self, event: &NativeRunEvent) -> Result<u64, ConformanceError> {
|
||||
let run_id = self.require_run_id()?;
|
||||
if event.run_id != run_id {
|
||||
return Err(ConformanceError::invalid(
|
||||
"event runId does not match the opened run",
|
||||
));
|
||||
}
|
||||
let expected_sequence = self.events.len() as u64 + 1;
|
||||
if event.sequence != expected_sequence {
|
||||
return Err(ConformanceError::invalid(format!(
|
||||
"event sequence must be {expected_sequence}"
|
||||
)));
|
||||
}
|
||||
self.events.push(event.clone());
|
||||
Ok(event.sequence)
|
||||
}
|
||||
|
||||
pub fn complete_run(&mut self, result: &NativeRunResult) -> Result<(), ConformanceError> {
|
||||
let run_id = self.require_run_id()?;
|
||||
if result.run_id != run_id {
|
||||
return Err(ConformanceError::invalid(
|
||||
"result runId does not match the opened run",
|
||||
));
|
||||
}
|
||||
if self.events.last().map(|event| event.event_type.as_str()) != Some("run.completed") {
|
||||
return Err(ConformanceError::invalid(
|
||||
"run.completed must be ingested before the result",
|
||||
));
|
||||
}
|
||||
if self.result.is_some() {
|
||||
return Err(ConformanceError::invalid(
|
||||
"mock control plane accepts one terminal result",
|
||||
));
|
||||
}
|
||||
self.result = Some(result.clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn require_running(&self) -> Result<(), ConformanceError> {
|
||||
if self.running {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(ConformanceError::invalid(
|
||||
"mock control plane must be started first",
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn require_run_id(&self) -> Result<&str, ConformanceError> {
|
||||
self.require_running()?;
|
||||
self.opened_run_id
|
||||
.as_deref()
|
||||
.ok_or_else(|| ConformanceError::invalid("a run must be opened first"))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ConformanceTracerOutput<'a> {
|
||||
schema_version: &'static str,
|
||||
run_identity: ConformanceRunIdentityOutput<'a>,
|
||||
result: ConformanceResultOutput<'a>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ConformanceRunIdentityOutput<'a> {
|
||||
run_id: &'a str,
|
||||
session_id: &'a str,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ConformanceResultOutput<'a> {
|
||||
status: &'a str,
|
||||
summary: &'a str,
|
||||
}
|
||||
|
||||
pub fn run_conformance_tracer(input: &str) -> Result<String, ConformanceError> {
|
||||
let fixture = validate_conformance_fixture(input)?;
|
||||
let mut mock_core = MockControlPlane::default();
|
||||
mock_core.start()?;
|
||||
|
||||
let replay_result = (|| {
|
||||
mock_core.open_run(&fixture.run)?;
|
||||
for event in &fixture.events {
|
||||
mock_core.append_event(event)?;
|
||||
}
|
||||
mock_core.complete_run(&fixture.result)?;
|
||||
serde_json::to_string(&ConformanceTracerOutput {
|
||||
schema_version: CONFORMANCE_OUTPUT_SCHEMA,
|
||||
run_identity: ConformanceRunIdentityOutput {
|
||||
run_id: &fixture.run.run_id,
|
||||
session_id: &fixture.run.session_id,
|
||||
},
|
||||
result: ConformanceResultOutput {
|
||||
status: &fixture.result.status,
|
||||
summary: &fixture.result.summary,
|
||||
},
|
||||
})
|
||||
.map_err(|error| ConformanceError::invalid(format!("output serialization failed: {error}")))
|
||||
})();
|
||||
|
||||
mock_core.stop();
|
||||
replay_result
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn validates_the_shared_fixture() {
|
||||
let fixture = validate_conformance_fixture(CONFORMANCE_FIXTURE)
|
||||
.expect("shared fixture should validate");
|
||||
|
||||
assert_eq!(fixture.events.len(), 2);
|
||||
assert_eq!(fixture.events[1].sequence, 2);
|
||||
assert_eq!(fixture.result.status, "succeeded");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_a_sequence_gap() {
|
||||
let invalid = CONFORMANCE_FIXTURE.replacen(r#""sequence": 2"#, r#""sequence": 3"#, 1);
|
||||
|
||||
let error = validate_conformance_fixture(&invalid).expect_err("sequence gap must fail");
|
||||
assert_eq!(error.to_string(), "events[1].sequence must be 2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runs_the_mock_core_path_with_stable_output() {
|
||||
assert_eq!(
|
||||
run_conformance_tracer(CONFORMANCE_FIXTURE).expect("tracer should succeed"),
|
||||
CONFORMANCE_EXPECTED_OUTPUT.trim_end(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,906 @@
|
|||
use std::collections::HashMap;
|
||||
use std::error::Error;
|
||||
use std::fmt::{self, Display, Formatter};
|
||||
use std::io::Write;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::mpsc::{self, RecvTimeoutError};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use crate::process_supervisor::{
|
||||
read_bounded_line, BoundedLine, BoundedLogBuffer, ProcessExitFact, ProcessOutput,
|
||||
SupervisedProcess,
|
||||
};
|
||||
|
||||
const RUNNER_STREAM_SCHEMA: &str = "paperclip.runner.stream.v1";
|
||||
const RUNNER_COMMAND_SCHEMA: &str = "paperclip.prp.command.v1";
|
||||
const HARNESS_COMMAND_SCHEMA: &str = "paperclip.fake_harness.command.v1";
|
||||
const HARNESS_MESSAGE_SCHEMA: &str = "paperclip.fake_harness.message.v1";
|
||||
const CONTROLLER_COMMAND_QUEUE_CAPACITY: usize = 256;
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct LocalRunnerError(String);
|
||||
|
||||
impl LocalRunnerError {
|
||||
pub fn invalid(message: impl Into<String>) -> Self {
|
||||
Self(message.into())
|
||||
}
|
||||
}
|
||||
|
||||
impl Display for LocalRunnerError {
|
||||
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
|
||||
formatter.write_str(&self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for LocalRunnerError {}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct RunnerConfig {
|
||||
pub run_id: String,
|
||||
pub normalized_session_id: String,
|
||||
pub runner_instance_id: String,
|
||||
pub fake_harness_path: PathBuf,
|
||||
pub script_path: PathBuf,
|
||||
pub delay_override_ms: Option<u64>,
|
||||
pub log_max_lines: usize,
|
||||
pub log_max_bytes: usize,
|
||||
pub command_history_limit: usize,
|
||||
pub controller_max_line_bytes: usize,
|
||||
pub harness_max_line_bytes: usize,
|
||||
pub shutdown_grace: Duration,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RunnerCommand {
|
||||
pub schema: String,
|
||||
pub command_id: String,
|
||||
pub controller_seq: u64,
|
||||
#[serde(rename = "type")]
|
||||
pub command_type: String,
|
||||
pub issued_at: String,
|
||||
#[serde(default)]
|
||||
pub payload: Value,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HarnessCommand {
|
||||
pub schema: String,
|
||||
pub command_id: String,
|
||||
#[serde(rename = "type")]
|
||||
pub command_type: String,
|
||||
#[serde(default)]
|
||||
pub payload: Value,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct HarnessMessage {
|
||||
schema: String,
|
||||
message_seq: u64,
|
||||
#[serde(rename = "type")]
|
||||
message_type: String,
|
||||
#[serde(default)]
|
||||
payload: Value,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
enum RunnerStreamMessage<'a> {
|
||||
Event {
|
||||
event: &'a Value,
|
||||
},
|
||||
CommandReceipt {
|
||||
#[serde(rename = "commandId")]
|
||||
command_id: &'a str,
|
||||
#[serde(rename = "controllerSeq")]
|
||||
controller_seq: u64,
|
||||
status: &'a str,
|
||||
detail: &'a str,
|
||||
},
|
||||
Diagnostic {
|
||||
level: &'a str,
|
||||
message: &'a str,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
enum TerminalIntent {
|
||||
#[default]
|
||||
Natural,
|
||||
Interrupted,
|
||||
Cancelled,
|
||||
ControllerClosed,
|
||||
ProtocolViolation,
|
||||
}
|
||||
|
||||
fn enum_field<'a>(value: &'a Value, key: &str, allowed: &[&str]) -> Option<&'a str> {
|
||||
value
|
||||
.get(key)
|
||||
.and_then(Value::as_str)
|
||||
.filter(|candidate| allowed.contains(candidate))
|
||||
}
|
||||
|
||||
fn valid_terminal_proposal(value: &Value) -> bool {
|
||||
value.get("schema").and_then(Value::as_str) == Some("paperclip.prp.terminal.v1")
|
||||
&& enum_field(
|
||||
value,
|
||||
"turnTerminalState",
|
||||
&["completed", "failed", "interrupted", "cancelled"],
|
||||
)
|
||||
.is_some()
|
||||
&& enum_field(
|
||||
value,
|
||||
"runTerminalState",
|
||||
&["succeeded", "failed", "cancelled"],
|
||||
)
|
||||
.is_some()
|
||||
&& enum_field(
|
||||
value,
|
||||
"reportedWorkDisposition",
|
||||
&["done", "blocked", "needs_review", "yielded"],
|
||||
)
|
||||
.is_some()
|
||||
}
|
||||
|
||||
fn reconcile_terminal(
|
||||
proposal: Option<Value>,
|
||||
exit: &ProcessExitFact,
|
||||
semantic_result: Option<&Value>,
|
||||
intent: TerminalIntent,
|
||||
) -> Value {
|
||||
let (turn_terminal_state, run_terminal_state, reason) = match intent {
|
||||
TerminalIntent::Interrupted => ("interrupted", "cancelled", "turn_interrupted"),
|
||||
TerminalIntent::Cancelled => ("cancelled", "cancelled", "run_cancelled"),
|
||||
TerminalIntent::ControllerClosed => ("cancelled", "cancelled", "controller_closed"),
|
||||
TerminalIntent::ProtocolViolation => ("failed", "failed", "harness_protocol_violation"),
|
||||
TerminalIntent::Natural if exit.success && semantic_result.is_some() => {
|
||||
("completed", "succeeded", "successful_process_and_result")
|
||||
}
|
||||
TerminalIntent::Natural if !exit.success => ("failed", "failed", "harness_process_failed"),
|
||||
TerminalIntent::Natural => ("failed", "failed", "semantic_result_missing"),
|
||||
};
|
||||
|
||||
let proposal_valid = proposal.as_ref().is_some_and(valid_terminal_proposal);
|
||||
let proposal_contradicted = proposal.as_ref().is_some_and(|value| {
|
||||
proposal_valid
|
||||
&& (value.get("turnTerminalState").and_then(Value::as_str) != Some(turn_terminal_state)
|
||||
|| value.get("runTerminalState").and_then(Value::as_str)
|
||||
!= Some(run_terminal_state))
|
||||
});
|
||||
let proposed_disposition = proposal.as_ref().and_then(|value| {
|
||||
enum_field(
|
||||
value,
|
||||
"reportedWorkDisposition",
|
||||
&["done", "blocked", "needs_review", "yielded"],
|
||||
)
|
||||
.map(str::to_owned)
|
||||
});
|
||||
let result_disposition = semantic_result.and_then(|value| {
|
||||
enum_field(
|
||||
value,
|
||||
"reportedWorkDisposition",
|
||||
&["done", "blocked", "needs_review", "yielded"],
|
||||
)
|
||||
.map(str::to_owned)
|
||||
});
|
||||
let reported_work_disposition = if run_terminal_state == "succeeded" {
|
||||
result_disposition
|
||||
.or(proposed_disposition)
|
||||
.unwrap_or_else(|| "done".to_owned())
|
||||
} else {
|
||||
result_disposition
|
||||
.filter(|value| value != "done")
|
||||
.or_else(|| proposed_disposition.filter(|value| value != "done"))
|
||||
.unwrap_or_else(|| "yielded".to_owned())
|
||||
};
|
||||
let harness_terminal_proposal = proposal.unwrap_or(Value::Null);
|
||||
|
||||
json!({
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": turn_terminal_state,
|
||||
"runTerminalState": run_terminal_state,
|
||||
"reportedWorkDisposition": reported_work_disposition,
|
||||
"harnessTerminalProposal": harness_terminal_proposal,
|
||||
"reconciliation": {
|
||||
"authority": "runner",
|
||||
"reason": reason,
|
||||
"proposalValid": proposal_valid,
|
||||
"proposalContradicted": proposal_contradicted,
|
||||
"processExit": exit,
|
||||
"semanticResultObserved": semantic_result.is_some(),
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
struct RunnerState {
|
||||
config: RunnerConfig,
|
||||
source_seq: u64,
|
||||
next_controller_seq: u64,
|
||||
commands: HashMap<String, String>,
|
||||
next_harness_message_seq: u64,
|
||||
harness: Option<SupervisedProcess>,
|
||||
logs: BoundedLogBuffer,
|
||||
semantic_result: Option<Value>,
|
||||
pending_terminal: Option<Value>,
|
||||
terminal_intent: TerminalIntent,
|
||||
terminal_emitted: bool,
|
||||
stdout_closed: bool,
|
||||
}
|
||||
|
||||
impl RunnerState {
|
||||
fn new(config: RunnerConfig) -> Self {
|
||||
Self {
|
||||
logs: BoundedLogBuffer::new(config.log_max_lines, config.log_max_bytes),
|
||||
config,
|
||||
source_seq: 0,
|
||||
next_controller_seq: 1,
|
||||
commands: HashMap::new(),
|
||||
next_harness_message_seq: 1,
|
||||
harness: None,
|
||||
semantic_result: None,
|
||||
pending_terminal: None,
|
||||
terminal_intent: TerminalIntent::Natural,
|
||||
terminal_emitted: false,
|
||||
stdout_closed: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn write_stream(&self, message: &RunnerStreamMessage<'_>) -> Result<(), LocalRunnerError> {
|
||||
let stdout = std::io::stdout();
|
||||
let mut lock = stdout.lock();
|
||||
let envelope = json!({
|
||||
"schema": RUNNER_STREAM_SCHEMA,
|
||||
"message": message,
|
||||
});
|
||||
serde_json::to_writer(&mut lock, &envelope).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("stream serialization failed: {error}"))
|
||||
})?;
|
||||
lock.write_all(b"\n")
|
||||
.and_then(|_| lock.flush())
|
||||
.map_err(|error| LocalRunnerError::invalid(format!("stream write failed: {error}")))
|
||||
}
|
||||
|
||||
fn command_receipt(
|
||||
&self,
|
||||
command: &RunnerCommand,
|
||||
status: &str,
|
||||
detail: &str,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
self.write_stream(&RunnerStreamMessage::CommandReceipt {
|
||||
command_id: &command.command_id,
|
||||
controller_seq: command.controller_seq,
|
||||
status,
|
||||
detail,
|
||||
})
|
||||
}
|
||||
|
||||
fn diagnostic(&self, level: &str, message: &str) -> Result<(), LocalRunnerError> {
|
||||
self.write_stream(&RunnerStreamMessage::Diagnostic { level, message })
|
||||
}
|
||||
|
||||
fn emit_event(
|
||||
&mut self,
|
||||
event_type: &str,
|
||||
payload: Value,
|
||||
turn_id: Option<&str>,
|
||||
item_id: Option<&str>,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if self.terminal_emitted {
|
||||
return Ok(());
|
||||
}
|
||||
self.source_seq += 1;
|
||||
let mut event = json!({
|
||||
"schema": "paperclip.prp.event.v1",
|
||||
"sourceEventId": format!("event_{}_{:03}", self.config.run_id, self.source_seq),
|
||||
"sourceSeq": self.source_seq,
|
||||
"sourceInstanceId": self.config.runner_instance_id,
|
||||
"sourceKind": "runner",
|
||||
"runId": self.config.run_id,
|
||||
"normalizedSessionId": self.config.normalized_session_id,
|
||||
"eventType": event_type,
|
||||
"schemaVersion": 1,
|
||||
"priority": event_priority(event_type),
|
||||
"emittedAt": format!("2026-08-07T21:00:{:02}.{:03}Z", (self.source_seq / 1000) % 60, self.source_seq % 1000),
|
||||
"payload": payload,
|
||||
});
|
||||
if let Some(turn_id) = turn_id {
|
||||
event["turnId"] = Value::String(turn_id.to_owned());
|
||||
}
|
||||
if let Some(item_id) = item_id {
|
||||
event["itemId"] = Value::String(item_id.to_owned());
|
||||
}
|
||||
self.write_stream(&RunnerStreamMessage::Event { event: &event })?;
|
||||
if event_type == "run.terminal" {
|
||||
self.terminal_emitted = true;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn start_harness(&mut self) -> Result<(), LocalRunnerError> {
|
||||
self.emit_event(
|
||||
"runtime.phase.changed",
|
||||
json!({ "phase": "workspace_preparing" }),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
self.emit_event(
|
||||
"workspace.ready",
|
||||
json!({ "workingDirectory": "standalone-fixture" }),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
self.emit_event(
|
||||
"harness.starting",
|
||||
json!({ "driverKind": "fake", "transport": "stdio_jsonl" }),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
let mut args = vec![
|
||||
"--script".to_owned(),
|
||||
self.config.script_path.display().to_string(),
|
||||
];
|
||||
if let Some(milliseconds) = self.config.delay_override_ms {
|
||||
args.push("--delay-ms".to_owned());
|
||||
args.push(milliseconds.to_string());
|
||||
}
|
||||
let harness = SupervisedProcess::spawn(
|
||||
&self.config.fake_harness_path,
|
||||
&args,
|
||||
self.config.shutdown_grace,
|
||||
self.config.harness_max_line_bytes,
|
||||
)?;
|
||||
let ready_deadline = Instant::now() + Duration::from_secs(2);
|
||||
loop {
|
||||
let remaining = ready_deadline.saturating_duration_since(Instant::now());
|
||||
if remaining.is_zero() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness did not become ready",
|
||||
));
|
||||
}
|
||||
match harness.recv_timeout(remaining) {
|
||||
Ok(ProcessOutput::Stdout(line)) => {
|
||||
let message = parse_harness_message(&line)?;
|
||||
self.accept_harness_message_sequence(&message)?;
|
||||
if message.message_type == "ready" {
|
||||
self.emit_event(
|
||||
"harness.ready",
|
||||
json!({
|
||||
"driverKind": "fake",
|
||||
"driverVersion": message.payload.get("version").cloned().unwrap_or_else(|| json!("1.0.0")),
|
||||
"transport": "stdio_jsonl"
|
||||
}),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
break;
|
||||
}
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness emitted data before ready",
|
||||
));
|
||||
}
|
||||
Ok(ProcessOutput::Stderr(line)) => self.logs.push(format!("stderr: {line}")),
|
||||
Ok(ProcessOutput::StdoutError(message)) => {
|
||||
return Err(LocalRunnerError::invalid(message));
|
||||
}
|
||||
Ok(ProcessOutput::StdoutClosed) => {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness exited before ready",
|
||||
));
|
||||
}
|
||||
Ok(ProcessOutput::StderrClosed) => {}
|
||||
Err(RecvTimeoutError::Timeout) => {
|
||||
return Err(LocalRunnerError::invalid("fake harness ready timed out"));
|
||||
}
|
||||
Err(RecvTimeoutError::Disconnected) => {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness output channel closed",
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
self.harness = Some(harness);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn handle_command(&mut self, command: RunnerCommand) -> Result<(), LocalRunnerError> {
|
||||
if command.schema != RUNNER_COMMAND_SCHEMA {
|
||||
self.command_receipt(&command, "rejected", "unsupported command schema")?;
|
||||
return Ok(());
|
||||
}
|
||||
let canonical = canonical_json(&serde_json::to_value(&command).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("command canonicalization failed: {error}"))
|
||||
})?);
|
||||
if let Some(previous) = self.commands.get(&command.command_id) {
|
||||
if previous == &canonical {
|
||||
self.command_receipt(&command, "duplicate", "equivalent command already applied")?;
|
||||
} else {
|
||||
self.command_receipt(
|
||||
&command,
|
||||
"rejected",
|
||||
"commandId was reused with different data",
|
||||
)?;
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
if command.controller_seq != self.next_controller_seq {
|
||||
self.command_receipt(&command, "rejected", "controllerSeq is not contiguous")?;
|
||||
return Ok(());
|
||||
}
|
||||
if self.commands.len() >= self.config.command_history_limit.max(1) {
|
||||
self.command_receipt(&command, "rejected", "command history limit reached")?;
|
||||
return Ok(());
|
||||
}
|
||||
self.commands.insert(command.command_id.clone(), canonical);
|
||||
self.next_controller_seq += 1;
|
||||
|
||||
match command.command_type.as_str() {
|
||||
"run.prepare" => {
|
||||
if self.harness.is_some() {
|
||||
self.command_receipt(&command, "rejected", "run is already prepared")?;
|
||||
return Ok(());
|
||||
}
|
||||
self.start_harness()?;
|
||||
}
|
||||
"session.open" => {
|
||||
self.emit_event("session.starting", json!({}), None, None)?;
|
||||
self.send_harness_command(&command)?;
|
||||
}
|
||||
"turn.start" => {
|
||||
self.emit_event(
|
||||
"turn.submitted",
|
||||
json!({ "text": command.payload.get("text").cloned().unwrap_or(Value::Null) }),
|
||||
command.payload.get("turnId").and_then(Value::as_str),
|
||||
None,
|
||||
)?;
|
||||
self.send_harness_command(&command)?;
|
||||
}
|
||||
"request.resolve" | "session.close" => {
|
||||
self.send_harness_command(&command)?;
|
||||
}
|
||||
"turn.interrupt" | "turn.stop" => {
|
||||
self.terminal_intent = TerminalIntent::Interrupted;
|
||||
self.send_harness_command(&command)?;
|
||||
}
|
||||
"run.cancel" => {
|
||||
self.terminal_intent = TerminalIntent::Cancelled;
|
||||
self.send_harness_command(&command)?;
|
||||
}
|
||||
unsupported => {
|
||||
self.command_receipt(
|
||||
&command,
|
||||
"rejected",
|
||||
&format!("command {unsupported} is not implemented in Local runner"),
|
||||
)?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
self.command_receipt(&command, "accepted", "command applied")
|
||||
}
|
||||
|
||||
fn send_harness_command(&mut self, command: &RunnerCommand) -> Result<(), LocalRunnerError> {
|
||||
let harness = self
|
||||
.harness
|
||||
.as_mut()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("run.prepare must start the harness first"))?;
|
||||
harness.send(&HarnessCommand {
|
||||
schema: HARNESS_COMMAND_SCHEMA.to_owned(),
|
||||
command_id: command.command_id.clone(),
|
||||
command_type: command.command_type.clone(),
|
||||
payload: command.payload.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
fn drain_harness_output(&mut self) -> Result<(), LocalRunnerError> {
|
||||
loop {
|
||||
let output = match self.harness.as_ref().map(SupervisedProcess::try_recv) {
|
||||
Some(Ok(output)) => output,
|
||||
Some(Err(mpsc::TryRecvError::Empty)) | None => return Ok(()),
|
||||
Some(Err(mpsc::TryRecvError::Disconnected)) => {
|
||||
self.stdout_closed = true;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
match output {
|
||||
ProcessOutput::Stdout(line) => self.handle_harness_line(&line)?,
|
||||
ProcessOutput::Stderr(line) => self.logs.push(format!("stderr: {line}")),
|
||||
ProcessOutput::StdoutError(message) => {
|
||||
self.logs.push(format!("stdout: [{message}]"));
|
||||
self.terminal_intent = TerminalIntent::ProtocolViolation;
|
||||
self.terminate_harness()?;
|
||||
return Ok(());
|
||||
}
|
||||
ProcessOutput::StdoutClosed => self.stdout_closed = true,
|
||||
ProcessOutput::StderrClosed => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_harness_line(&mut self, line: &str) -> Result<(), LocalRunnerError> {
|
||||
let message = parse_harness_message(line)?;
|
||||
self.accept_harness_message_sequence(&message)?;
|
||||
match message.message_type.as_str() {
|
||||
"event" => {
|
||||
let event_type = message
|
||||
.payload
|
||||
.get("eventType")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("harness event requires eventType"))?
|
||||
.to_owned();
|
||||
let turn_id = message
|
||||
.payload
|
||||
.get("turnId")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_owned);
|
||||
let item_id = message
|
||||
.payload
|
||||
.get("itemId")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_owned);
|
||||
let payload = message
|
||||
.payload
|
||||
.get("payload")
|
||||
.cloned()
|
||||
.unwrap_or_else(|| json!({}));
|
||||
self.emit_event(&event_type, payload, turn_id.as_deref(), item_id.as_deref())?;
|
||||
}
|
||||
"request" => {
|
||||
let request =
|
||||
message.payload.get("request").cloned().ok_or_else(|| {
|
||||
LocalRunnerError::invalid("harness request requires request")
|
||||
})?;
|
||||
self.emit_event(
|
||||
"runtime_request.created",
|
||||
json!({ "request": request }),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
}
|
||||
"result" => {
|
||||
let result =
|
||||
message.payload.get("result").cloned().ok_or_else(|| {
|
||||
LocalRunnerError::invalid("harness result requires result")
|
||||
})?;
|
||||
if self.semantic_result.is_none() {
|
||||
self.semantic_result = Some(result.clone());
|
||||
self.emit_event("run.result.proposed", result, None, None)?;
|
||||
} else {
|
||||
self.emit_event(
|
||||
"harness.diagnostic",
|
||||
json!({
|
||||
"code": "duplicate_semantic_result_ignored",
|
||||
"message": "Duplicate semantic result ignored; the first result remains authoritative."
|
||||
}),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
"terminal" => {
|
||||
let terminal = message.payload.get("terminal").cloned().ok_or_else(|| {
|
||||
LocalRunnerError::invalid("harness terminal requires terminal")
|
||||
})?;
|
||||
if self.pending_terminal.is_none() {
|
||||
self.pending_terminal = Some(terminal);
|
||||
} else {
|
||||
self.emit_event(
|
||||
"harness.diagnostic",
|
||||
json!({
|
||||
"code": "duplicate_terminal_ignored",
|
||||
"message": "Duplicate terminal event ignored; the first terminal event remains authoritative."
|
||||
}),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
"log" => {
|
||||
let stream = message
|
||||
.payload
|
||||
.get("stream")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("stdout");
|
||||
let text = message
|
||||
.payload
|
||||
.get("text")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or_default();
|
||||
self.logs.push(format!("{stream}: {text}"));
|
||||
}
|
||||
"diagnostic" => {
|
||||
self.logs.push(format!("diagnostic: {}", message.payload));
|
||||
}
|
||||
"ready" => {
|
||||
self.diagnostic("warning", "duplicate harness ready message ignored")?;
|
||||
}
|
||||
unsupported => {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"unsupported fake harness message {unsupported} at sequence {}",
|
||||
message.message_seq
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn accept_harness_message_sequence(
|
||||
&mut self,
|
||||
message: &HarnessMessage,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if message.message_seq != self.next_harness_message_seq {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"fake harness message sequence must be {}; received {}",
|
||||
self.next_harness_message_seq, message.message_seq
|
||||
)));
|
||||
}
|
||||
self.next_harness_message_seq += 1;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn finish_harness(&mut self) -> Result<(), LocalRunnerError> {
|
||||
let exit = self
|
||||
.harness
|
||||
.as_mut()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("harness was not started"))?
|
||||
.wait()?;
|
||||
self.finalize_harness(exit)
|
||||
}
|
||||
|
||||
fn terminate_harness(&mut self) -> Result<(), LocalRunnerError> {
|
||||
let exit = self
|
||||
.harness
|
||||
.as_mut()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("harness was not started"))?
|
||||
.terminate_group()?;
|
||||
self.finalize_harness(exit)
|
||||
}
|
||||
|
||||
fn finalize_harness(&mut self, exit: ProcessExitFact) -> Result<(), LocalRunnerError> {
|
||||
self.emit_event(
|
||||
"harness.exited",
|
||||
json!({
|
||||
"processExit": &exit,
|
||||
"semanticResultObserved": self.semantic_result.is_some(),
|
||||
"logs": self.logs.snapshot(),
|
||||
}),
|
||||
None,
|
||||
None,
|
||||
)?;
|
||||
let terminal = reconcile_terminal(
|
||||
self.pending_terminal.take(),
|
||||
&exit,
|
||||
self.semantic_result.as_ref(),
|
||||
self.terminal_intent,
|
||||
);
|
||||
self.emit_event("run.terminal", terminal, None, None)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn run_local_runner(config: RunnerConfig) -> Result<(), LocalRunnerError> {
|
||||
let controller_max_line_bytes = config.controller_max_line_bytes.max(1);
|
||||
let (command_sender, command_receiver) = mpsc::sync_channel(CONTROLLER_COMMAND_QUEUE_CAPACITY);
|
||||
thread::spawn(move || {
|
||||
let stdin = std::io::stdin();
|
||||
let mut reader = stdin.lock();
|
||||
loop {
|
||||
let command = match read_bounded_line(&mut reader, controller_max_line_bytes) {
|
||||
Ok(BoundedLine::Line(line)) if line.trim().is_empty() => continue,
|
||||
Ok(BoundedLine::Line(line)) => serde_json::from_str::<RunnerCommand>(&line)
|
||||
.map_err(|error| format!("invalid command JSON: {error}")),
|
||||
Ok(BoundedLine::TooLong) => Err(format!(
|
||||
"controller command exceeded {controller_max_line_bytes} bytes"
|
||||
)),
|
||||
Ok(BoundedLine::Eof) => return,
|
||||
Err(error) => Err(format!("failed to read controller input: {error}")),
|
||||
};
|
||||
if command_sender.send(command).is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let mut state = RunnerState::new(config);
|
||||
loop {
|
||||
match command_receiver.recv_timeout(Duration::from_millis(2)) {
|
||||
Ok(Ok(command)) => state.handle_command(command)?,
|
||||
Ok(Err(error)) => state.diagnostic("error", &error)?,
|
||||
Err(RecvTimeoutError::Timeout) => {}
|
||||
Err(RecvTimeoutError::Disconnected) => {
|
||||
if state.harness.is_none() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"controller closed before run.prepare",
|
||||
));
|
||||
}
|
||||
state.terminal_intent = TerminalIntent::ControllerClosed;
|
||||
state.terminate_harness()?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
state.drain_harness_output()?;
|
||||
if state.terminal_emitted {
|
||||
return Ok(());
|
||||
}
|
||||
if state.stdout_closed {
|
||||
state.finish_harness()?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_harness_message(line: &str) -> Result<HarnessMessage, LocalRunnerError> {
|
||||
let message: HarnessMessage = serde_json::from_str(line).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("fake harness emitted invalid JSONL: {error}"))
|
||||
})?;
|
||||
if message.schema != HARNESS_MESSAGE_SCHEMA {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"fake harness message schema is unsupported",
|
||||
));
|
||||
}
|
||||
Ok(message)
|
||||
}
|
||||
|
||||
fn event_priority(event_type: &str) -> u8 {
|
||||
match event_type {
|
||||
"run.result.proposed"
|
||||
| "run.terminal"
|
||||
| "turn.completed"
|
||||
| "turn.failed"
|
||||
| "turn.interrupted"
|
||||
| "harness.exited" => 0,
|
||||
"item.delta" | "harness.diagnostic" | "runner.diagnostic" => 2,
|
||||
_ => 1,
|
||||
}
|
||||
}
|
||||
|
||||
fn canonical_json(value: &Value) -> String {
|
||||
match value {
|
||||
Value::Array(values) => format!(
|
||||
"[{}]",
|
||||
values
|
||||
.iter()
|
||||
.map(canonical_json)
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
),
|
||||
Value::Object(values) => {
|
||||
let mut keys = values.keys().collect::<Vec<_>>();
|
||||
keys.sort();
|
||||
format!(
|
||||
"{{{}}}",
|
||||
keys.into_iter()
|
||||
.map(|key| format!(
|
||||
"{}:{}",
|
||||
serde_json::to_string(key).expect("JSON object key should serialize"),
|
||||
canonical_json(&values[key])
|
||||
))
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
)
|
||||
}
|
||||
_ => value.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::BufReader;
|
||||
use std::path::Path;
|
||||
|
||||
use crate::fake_harness::load_fake_harness_script;
|
||||
use crate::process_supervisor::{read_bounded_line, BoundedLine};
|
||||
|
||||
#[test]
|
||||
fn bounded_logs_keep_only_the_tail() {
|
||||
let mut logs = BoundedLogBuffer::new(2, 12);
|
||||
logs.push("first");
|
||||
logs.push("second");
|
||||
logs.push("third");
|
||||
|
||||
assert_eq!(logs.snapshot().lines, vec!["second", "third"]);
|
||||
assert_eq!(logs.snapshot().dropped_lines, 1);
|
||||
assert!(logs.snapshot().retained_bytes <= 12);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bounded_line_reader_rejects_an_oversized_frame_and_recovers() {
|
||||
let mut source = vec![b'x'; 4_096];
|
||||
source.extend_from_slice(b"\nnext\n");
|
||||
let mut reader = BufReader::new(std::io::Cursor::new(source));
|
||||
|
||||
assert_eq!(
|
||||
read_bounded_line(&mut reader, 64).expect("oversized line should be classified"),
|
||||
BoundedLine::TooLong
|
||||
);
|
||||
assert_eq!(
|
||||
read_bounded_line(&mut reader, 64).expect("reader should continue after the frame"),
|
||||
BoundedLine::Line("next".to_owned())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nonzero_exit_overrides_a_success_terminal_proposal() {
|
||||
let terminal = reconcile_terminal(
|
||||
Some(json!({
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "completed",
|
||||
"runTerminalState": "succeeded",
|
||||
"reportedWorkDisposition": "done"
|
||||
})),
|
||||
&ProcessExitFact {
|
||||
exit_code: Some(7),
|
||||
success: false,
|
||||
signal: None,
|
||||
},
|
||||
Some(&json!({ "reportedWorkDisposition": "done" })),
|
||||
TerminalIntent::Natural,
|
||||
);
|
||||
|
||||
assert_eq!(terminal["turnTerminalState"], "failed");
|
||||
assert_eq!(terminal["runTerminalState"], "failed");
|
||||
assert_eq!(terminal["reportedWorkDisposition"], "yielded");
|
||||
assert_eq!(terminal["reconciliation"]["authority"], "runner");
|
||||
assert_eq!(terminal["reconciliation"]["proposalValid"], true);
|
||||
assert_eq!(terminal["reconciliation"]["proposalContradicted"], true);
|
||||
assert_eq!(
|
||||
terminal["harnessTerminalProposal"]["runTerminalState"],
|
||||
"succeeded"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn successful_process_and_result_override_a_failed_terminal_proposal() {
|
||||
let terminal = reconcile_terminal(
|
||||
Some(json!({
|
||||
"schema": "paperclip.prp.terminal.v1",
|
||||
"turnTerminalState": "failed",
|
||||
"runTerminalState": "failed",
|
||||
"reportedWorkDisposition": "yielded"
|
||||
})),
|
||||
&ProcessExitFact {
|
||||
exit_code: Some(0),
|
||||
success: true,
|
||||
signal: None,
|
||||
},
|
||||
Some(&json!({ "reportedWorkDisposition": "done" })),
|
||||
TerminalIntent::Natural,
|
||||
);
|
||||
|
||||
assert_eq!(terminal["turnTerminalState"], "completed");
|
||||
assert_eq!(terminal["runTerminalState"], "succeeded");
|
||||
assert_eq!(terminal["reportedWorkDisposition"], "done");
|
||||
assert_eq!(terminal["reconciliation"]["proposalContradicted"], true);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn canonical_json_sorts_object_keys_without_reordering_arrays() {
|
||||
assert_eq!(
|
||||
canonical_json(&json!({ "z": [2, 1], "a": { "b": true } })),
|
||||
r#"{"a":{"b":true},"z":[2,1]}"#
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn all_local_runner_scripts_load() {
|
||||
let scripts = Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../../../protocol/fixtures/local-runner/scripts");
|
||||
for name in [
|
||||
"happy-path",
|
||||
"permission-input",
|
||||
"interrupted",
|
||||
"error",
|
||||
"duplicate-terminal",
|
||||
"linger",
|
||||
"oversized-line",
|
||||
] {
|
||||
let script = load_fake_harness_script(&scripts.join(format!("{name}.json")))
|
||||
.expect("Local runner script should load");
|
||||
assert_eq!(script.name, name);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,412 @@
|
|||
use std::collections::VecDeque;
|
||||
use std::io::{self, BufRead, BufReader, Write};
|
||||
use std::path::Path;
|
||||
use std::process::{Child, ChildStdin, Command, ExitStatus, Stdio};
|
||||
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::process::CommandExt;
|
||||
|
||||
use serde::Serialize;
|
||||
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
|
||||
const PROCESS_OUTPUT_QUEUE_CAPACITY: usize = 256;
|
||||
|
||||
pub(crate) enum ProcessOutput {
|
||||
Stdout(String),
|
||||
Stderr(String),
|
||||
StdoutError(String),
|
||||
StdoutClosed,
|
||||
StderrClosed,
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub(crate) enum BoundedLine {
|
||||
Line(String),
|
||||
TooLong,
|
||||
Eof,
|
||||
}
|
||||
|
||||
pub(crate) fn read_bounded_line<R: BufRead>(
|
||||
reader: &mut R,
|
||||
max_bytes: usize,
|
||||
) -> io::Result<BoundedLine> {
|
||||
let max_bytes = max_bytes.max(1);
|
||||
let mut bytes = Vec::with_capacity(max_bytes.min(8 * 1024));
|
||||
let mut too_long = false;
|
||||
|
||||
loop {
|
||||
let buffer = reader.fill_buf()?;
|
||||
if buffer.is_empty() {
|
||||
if too_long {
|
||||
return Ok(BoundedLine::TooLong);
|
||||
}
|
||||
if bytes.is_empty() {
|
||||
return Ok(BoundedLine::Eof);
|
||||
}
|
||||
return String::from_utf8(bytes)
|
||||
.map(BoundedLine::Line)
|
||||
.map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error));
|
||||
}
|
||||
|
||||
let newline = buffer.iter().position(|byte| *byte == b'\n');
|
||||
let content_len = newline.unwrap_or(buffer.len());
|
||||
if !too_long {
|
||||
let remaining = max_bytes.saturating_sub(bytes.len());
|
||||
let copy_len = remaining.min(content_len);
|
||||
bytes.extend_from_slice(&buffer[..copy_len]);
|
||||
if content_len > remaining {
|
||||
too_long = true;
|
||||
bytes.clear();
|
||||
}
|
||||
}
|
||||
let consumed = newline.map_or(buffer.len(), |index| index + 1);
|
||||
reader.consume(consumed);
|
||||
|
||||
if newline.is_some() {
|
||||
if too_long {
|
||||
return Ok(BoundedLine::TooLong);
|
||||
}
|
||||
if bytes.last() == Some(&b'\r') {
|
||||
bytes.pop();
|
||||
}
|
||||
return String::from_utf8(bytes)
|
||||
.map(BoundedLine::Line)
|
||||
.map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn forward_bounded_output<R: io::Read + Send + 'static>(
|
||||
reader: R,
|
||||
sender: SyncSender<ProcessOutput>,
|
||||
stdout: bool,
|
||||
max_line_bytes: usize,
|
||||
) {
|
||||
thread::spawn(move || {
|
||||
let mut reader = BufReader::new(reader);
|
||||
loop {
|
||||
let output = match read_bounded_line(&mut reader, max_line_bytes) {
|
||||
Ok(BoundedLine::Line(line)) if stdout => ProcessOutput::Stdout(line),
|
||||
Ok(BoundedLine::Line(line)) => ProcessOutput::Stderr(line),
|
||||
Ok(BoundedLine::TooLong) if stdout => ProcessOutput::StdoutError(format!(
|
||||
"harness stdout frame exceeded {max_line_bytes} bytes"
|
||||
)),
|
||||
Ok(BoundedLine::TooLong) => ProcessOutput::Stderr(format!(
|
||||
"[harness stderr frame exceeded {max_line_bytes} bytes]"
|
||||
)),
|
||||
Ok(BoundedLine::Eof) => break,
|
||||
Err(error) if stdout => {
|
||||
ProcessOutput::StdoutError(format!("failed to read harness stdout: {error}"))
|
||||
}
|
||||
Err(error) => {
|
||||
ProcessOutput::Stderr(format!("[failed to read harness stderr: {error}]"))
|
||||
}
|
||||
};
|
||||
let terminal_output = matches!(&output, ProcessOutput::StdoutError(_));
|
||||
if sender.send(output).is_err() || terminal_output {
|
||||
return;
|
||||
}
|
||||
}
|
||||
let closed = if stdout {
|
||||
ProcessOutput::StdoutClosed
|
||||
} else {
|
||||
ProcessOutput::StderrClosed
|
||||
};
|
||||
let _ = sender.send(closed);
|
||||
});
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ProcessExitFact {
|
||||
pub exit_code: Option<i32>,
|
||||
pub success: bool,
|
||||
pub signal: Option<i32>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BoundedLogSnapshot {
|
||||
pub lines: Vec<String>,
|
||||
pub retained_bytes: usize,
|
||||
pub dropped_lines: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct BoundedLogBuffer {
|
||||
max_lines: usize,
|
||||
max_bytes: usize,
|
||||
retained_bytes: usize,
|
||||
dropped_lines: usize,
|
||||
lines: VecDeque<String>,
|
||||
}
|
||||
|
||||
impl BoundedLogBuffer {
|
||||
pub fn new(max_lines: usize, max_bytes: usize) -> Self {
|
||||
Self {
|
||||
max_lines: max_lines.max(1),
|
||||
max_bytes: max_bytes.max(1),
|
||||
retained_bytes: 0,
|
||||
dropped_lines: 0,
|
||||
lines: VecDeque::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn push(&mut self, line: impl Into<String>) {
|
||||
let mut line = line.into();
|
||||
if line.len() > self.max_bytes {
|
||||
let mut end = self.max_bytes;
|
||||
while !line.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
}
|
||||
line.truncate(end);
|
||||
}
|
||||
self.retained_bytes += line.len();
|
||||
self.lines.push_back(line);
|
||||
while self.lines.len() > self.max_lines || self.retained_bytes > self.max_bytes {
|
||||
if let Some(removed) = self.lines.pop_front() {
|
||||
self.retained_bytes = self.retained_bytes.saturating_sub(removed.len());
|
||||
self.dropped_lines += 1;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn snapshot(&self) -> BoundedLogSnapshot {
|
||||
BoundedLogSnapshot {
|
||||
lines: self.lines.iter().cloned().collect(),
|
||||
retained_bytes: self.retained_bytes,
|
||||
dropped_lines: self.dropped_lines,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct SupervisedProcess {
|
||||
child: Child,
|
||||
stdin: Option<ChildStdin>,
|
||||
output: Receiver<ProcessOutput>,
|
||||
process_group_id: u32,
|
||||
shutdown_grace: Duration,
|
||||
finished: bool,
|
||||
}
|
||||
|
||||
impl SupervisedProcess {
|
||||
pub fn spawn(
|
||||
program: &Path,
|
||||
args: &[String],
|
||||
shutdown_grace: Duration,
|
||||
max_line_bytes: usize,
|
||||
) -> Result<Self, LocalRunnerError> {
|
||||
let mut command = Command::new(program);
|
||||
command
|
||||
.args(args)
|
||||
.env_clear()
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped());
|
||||
for key in [
|
||||
"PATH",
|
||||
"PATHEXT",
|
||||
"SystemRoot",
|
||||
"WINDIR",
|
||||
"HOME",
|
||||
"USERPROFILE",
|
||||
"LANG",
|
||||
"LC_ALL",
|
||||
"TMPDIR",
|
||||
"TEMP",
|
||||
"TMP",
|
||||
"TZ",
|
||||
] {
|
||||
if let Some(value) = std::env::var_os(key) {
|
||||
command.env(key, value);
|
||||
}
|
||||
}
|
||||
#[cfg(unix)]
|
||||
command.process_group(0);
|
||||
|
||||
let mut child = command.spawn().map_err(|error| {
|
||||
LocalRunnerError::invalid(format!(
|
||||
"failed to start supervised process {}: {error}",
|
||||
program.display()
|
||||
))
|
||||
})?;
|
||||
let process_group_id = child.id();
|
||||
let stdin = child
|
||||
.stdin
|
||||
.take()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("supervised process stdin was not piped"))?;
|
||||
let stdout = child
|
||||
.stdout
|
||||
.take()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("supervised process stdout was not piped"))?;
|
||||
let stderr = child
|
||||
.stderr
|
||||
.take()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("supervised process stderr was not piped"))?;
|
||||
let (sender, output) = mpsc::sync_channel(PROCESS_OUTPUT_QUEUE_CAPACITY);
|
||||
forward_bounded_output(stdout, sender.clone(), true, max_line_bytes.max(1));
|
||||
forward_bounded_output(stderr, sender, false, max_line_bytes.max(1));
|
||||
|
||||
Ok(Self {
|
||||
child,
|
||||
stdin: Some(stdin),
|
||||
output,
|
||||
process_group_id,
|
||||
shutdown_grace,
|
||||
finished: false,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn id(&self) -> u32 {
|
||||
self.child.id()
|
||||
}
|
||||
|
||||
pub fn send<T: Serialize>(&mut self, value: &T) -> Result<(), LocalRunnerError> {
|
||||
let stdin = self
|
||||
.stdin
|
||||
.as_mut()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("supervised process stdin is closed"))?;
|
||||
serde_json::to_writer(&mut *stdin, value).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("command serialization failed: {error}"))
|
||||
})?;
|
||||
stdin
|
||||
.write_all(b"\n")
|
||||
.and_then(|_| stdin.flush())
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("failed to write process command: {error}"))
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn recv_timeout(
|
||||
&self,
|
||||
timeout: Duration,
|
||||
) -> Result<ProcessOutput, RecvTimeoutError> {
|
||||
self.output.recv_timeout(timeout)
|
||||
}
|
||||
|
||||
pub fn receive_stdout_line(
|
||||
&self,
|
||||
timeout: Duration,
|
||||
) -> Result<Option<String>, LocalRunnerError> {
|
||||
let deadline = Instant::now() + timeout;
|
||||
loop {
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
if remaining.is_zero() {
|
||||
return Ok(None);
|
||||
}
|
||||
match self.recv_timeout(remaining) {
|
||||
Ok(ProcessOutput::Stdout(line)) => return Ok(Some(line)),
|
||||
Ok(ProcessOutput::Stderr(_)) | Ok(ProcessOutput::StderrClosed) => {}
|
||||
Ok(ProcessOutput::StdoutError(message)) => {
|
||||
return Err(LocalRunnerError::invalid(message));
|
||||
}
|
||||
Ok(ProcessOutput::StdoutClosed) => return Ok(None),
|
||||
Err(RecvTimeoutError::Timeout) => return Ok(None),
|
||||
Err(RecvTimeoutError::Disconnected) => {
|
||||
return Err(LocalRunnerError::invalid("process output channel closed"));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn try_recv(&self) -> Result<ProcessOutput, mpsc::TryRecvError> {
|
||||
self.output.try_recv()
|
||||
}
|
||||
|
||||
pub fn try_wait(&mut self) -> Result<Option<ProcessExitFact>, LocalRunnerError> {
|
||||
self.child
|
||||
.try_wait()
|
||||
.map(|status| status.map(exit_fact))
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("failed to inspect process: {error}"))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn wait(&mut self) -> Result<ProcessExitFact, LocalRunnerError> {
|
||||
let status = self.child.wait().map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("failed to wait for process: {error}"))
|
||||
})?;
|
||||
// The group leader can exit while descendants remain alive. Reap the
|
||||
// leader first, then clear any remaining members of its private group.
|
||||
#[cfg(unix)]
|
||||
signal_process_group(self.process_group_id, "KILL");
|
||||
self.finished = true;
|
||||
Ok(exit_fact(status))
|
||||
}
|
||||
|
||||
pub fn terminate_group(&mut self) -> Result<ProcessExitFact, LocalRunnerError> {
|
||||
self.stdin.take();
|
||||
#[cfg(unix)]
|
||||
signal_process_group(self.process_group_id, "TERM");
|
||||
#[cfg(not(unix))]
|
||||
let _ = self.child.kill();
|
||||
|
||||
let deadline = Instant::now() + self.shutdown_grace;
|
||||
loop {
|
||||
if let Some(fact) = self.try_wait()? {
|
||||
#[cfg(unix)]
|
||||
signal_process_group(self.process_group_id, "KILL");
|
||||
self.finished = true;
|
||||
return Ok(fact);
|
||||
}
|
||||
if Instant::now() >= deadline {
|
||||
break;
|
||||
}
|
||||
thread::sleep(Duration::from_millis(5));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
signal_process_group(self.process_group_id, "KILL");
|
||||
#[cfg(not(unix))]
|
||||
let _ = self.child.kill();
|
||||
self.wait()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for SupervisedProcess {
|
||||
fn drop(&mut self) {
|
||||
if !self.finished {
|
||||
let _ = self.terminate_group();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn signal_process_group(process_group_id: u32, signal: &str) {
|
||||
let _ = Command::new("kill")
|
||||
.args([
|
||||
format!("-{signal}"),
|
||||
"--".to_owned(),
|
||||
format!("-{process_group_id}"),
|
||||
])
|
||||
.env_clear()
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.status();
|
||||
}
|
||||
|
||||
fn exit_fact(status: ExitStatus) -> ProcessExitFact {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::process::ExitStatusExt;
|
||||
ProcessExitFact {
|
||||
exit_code: status.code(),
|
||||
success: status.success(),
|
||||
signal: status.signal(),
|
||||
}
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
ProcessExitFact {
|
||||
exit_code: status.code(),
|
||||
success: status.success(),
|
||||
signal: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,408 @@
|
|||
use std::collections::HashMap;
|
||||
use std::error::Error;
|
||||
use std::fmt::{self, Display, Formatter};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
pub const PRP_PROTOCOL_VERSION: u64 = 1;
|
||||
const PRP_FIXTURE_VERSION: u64 = 1;
|
||||
const PRP_FIXTURE_SCHEMA: &str = "paperclip.prp.fixture.v1";
|
||||
const PRP_EVENT_SCHEMA: &str = "paperclip.prp.event.v1";
|
||||
const PRP_SEMANTIC_TOOL_SCHEMA: &str = "paperclip.prp.semantic_tool.v1";
|
||||
const PRP_STOP_REASON_SCHEMA: &str = "paperclip.prp.stop_reason.v1";
|
||||
const PRP_SEMANTIC_TOOLS_CAPABILITY_SCHEMA: &str = "paperclip.prp.semantic_tools.v1";
|
||||
const MAX_EXACT_JSON_INTEGER: u64 = 9_007_199_254_740_991;
|
||||
const MAX_RECORDED_MISSING_SEQUENCES: u64 = 256;
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ReplayFixture {
|
||||
schema: String,
|
||||
fixture_version: u64,
|
||||
protocol_version: u64,
|
||||
identity: ReplayIdentity,
|
||||
capabilities: Value,
|
||||
events: Vec<ReplayEvent>,
|
||||
result: Value,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ReplayIdentity {
|
||||
run_id: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ReplayEvent {
|
||||
schema: String,
|
||||
source_event_id: String,
|
||||
source_seq: u64,
|
||||
source_instance_id: String,
|
||||
source_kind: String,
|
||||
run_id: String,
|
||||
schema_version: u64,
|
||||
event_type: String,
|
||||
payload: Value,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ReplaySequenceGap {
|
||||
source_key: String,
|
||||
expected: u64,
|
||||
received: u64,
|
||||
missing_count: u64,
|
||||
missing: Vec<u64>,
|
||||
truncated: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ReplayParitySummary {
|
||||
run_id: String,
|
||||
integrity: String,
|
||||
timeline_count: usize,
|
||||
duplicate_event_ids: Vec<String>,
|
||||
gaps: Vec<ReplaySequenceGap>,
|
||||
turn_terminal_state: Option<String>,
|
||||
run_terminal_state: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ReplayError(String);
|
||||
|
||||
impl ReplayError {
|
||||
fn invalid(message: impl Into<String>) -> Self {
|
||||
Self(message.into())
|
||||
}
|
||||
}
|
||||
|
||||
impl Display for ReplayError {
|
||||
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
|
||||
formatter.write_str(&self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for ReplayError {}
|
||||
|
||||
pub fn reduce_replay_fixture(input: &str) -> Result<ReplayParitySummary, ReplayError> {
|
||||
let fixture: ReplayFixture = serde_json::from_str(input)
|
||||
.map_err(|error| ReplayError::invalid(format!("fixture must be valid JSON: {error}")))?;
|
||||
if fixture.protocol_version != PRP_PROTOCOL_VERSION {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required protocolVersion {}; expected {PRP_PROTOCOL_VERSION}",
|
||||
fixture.protocol_version
|
||||
)));
|
||||
}
|
||||
if fixture.fixture_version != PRP_FIXTURE_VERSION {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required fixtureVersion {}; expected {PRP_FIXTURE_VERSION}",
|
||||
fixture.fixture_version
|
||||
)));
|
||||
}
|
||||
if fixture.schema != PRP_FIXTURE_SCHEMA {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required fixture schema {}; expected {PRP_FIXTURE_SCHEMA}",
|
||||
fixture.schema
|
||||
)));
|
||||
}
|
||||
validate_optional_versioned_envelope(
|
||||
fixture.capabilities.get("semanticTools"),
|
||||
PRP_SEMANTIC_TOOLS_CAPABILITY_SCHEMA,
|
||||
"semanticTools",
|
||||
)?;
|
||||
|
||||
let mut delivery_counts = HashMap::<&str, usize>::new();
|
||||
for event in &fixture.events {
|
||||
*delivery_counts
|
||||
.entry(event.source_event_id.as_str())
|
||||
.or_default() += 1;
|
||||
}
|
||||
let mut duplicate_event_ids = delivery_counts
|
||||
.into_iter()
|
||||
.filter_map(|(event_id, count)| (count > 1).then(|| event_id.to_owned()))
|
||||
.collect::<Vec<_>>();
|
||||
duplicate_event_ids.sort();
|
||||
|
||||
let mut seen = HashMap::<&str, &ReplayEvent>::new();
|
||||
let mut cursors = HashMap::<String, u64>::new();
|
||||
let mut gaps = Vec::<ReplaySequenceGap>::new();
|
||||
let mut timeline_count = 0_usize;
|
||||
let mut turn_terminal_state = None;
|
||||
let mut run_terminal_state = None;
|
||||
let mut proposed_result_count = 0_usize;
|
||||
let mut terminal_count = 0_usize;
|
||||
|
||||
for event in &fixture.events {
|
||||
if event.source_seq > MAX_EXACT_JSON_INTEGER {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"sourceSeq {} exceeds the exact JSON integer range",
|
||||
event.source_seq
|
||||
)));
|
||||
}
|
||||
if event.schema != PRP_EVENT_SCHEMA {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required event schema {}; expected {PRP_EVENT_SCHEMA}",
|
||||
event.schema
|
||||
)));
|
||||
}
|
||||
if event.schema_version != 1 {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required event schemaVersion {}; expected 1",
|
||||
event.schema_version
|
||||
)));
|
||||
}
|
||||
validate_optional_versioned_envelope(
|
||||
event.payload.get("semantic_tool"),
|
||||
PRP_SEMANTIC_TOOL_SCHEMA,
|
||||
"semantic_tool",
|
||||
)?;
|
||||
validate_optional_versioned_envelope(
|
||||
event.payload.get("stopReason"),
|
||||
PRP_STOP_REASON_SCHEMA,
|
||||
"stopReason",
|
||||
)?;
|
||||
if event.run_id != fixture.identity.run_id {
|
||||
return Err(ReplayError::invalid(
|
||||
"event runId must match identity.runId",
|
||||
));
|
||||
}
|
||||
if let Some(previous) = seen.get(event.source_event_id.as_str()) {
|
||||
if *previous != event {
|
||||
return Err(ReplayError::invalid(
|
||||
"duplicate sourceEventId deliveries must be byte-equivalent",
|
||||
));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
seen.insert(event.source_event_id.as_str(), event);
|
||||
|
||||
let key = format!("{}:{}", event.source_kind, event.source_instance_id);
|
||||
let cursor = *cursors.get(&key).unwrap_or(&0);
|
||||
let expected = cursor + 1;
|
||||
if event.source_seq <= cursor {
|
||||
continue;
|
||||
}
|
||||
if event.source_seq > expected {
|
||||
let missing_count = event.source_seq - expected;
|
||||
let recorded_count = missing_count.min(MAX_RECORDED_MISSING_SEQUENCES);
|
||||
gaps.push(ReplaySequenceGap {
|
||||
source_key: key.clone(),
|
||||
expected,
|
||||
received: event.source_seq,
|
||||
missing_count,
|
||||
missing: (expected..expected + recorded_count).collect(),
|
||||
truncated: recorded_count < missing_count,
|
||||
});
|
||||
}
|
||||
cursors.insert(key, event.source_seq);
|
||||
timeline_count += 1;
|
||||
|
||||
match event.event_type.as_str() {
|
||||
"run.result.proposed" => {
|
||||
proposed_result_count += 1;
|
||||
if event.payload != fixture.result {
|
||||
return Err(ReplayError::invalid(
|
||||
"fixture result must match the run.result.proposed event payload",
|
||||
));
|
||||
}
|
||||
}
|
||||
"run.terminal" => {
|
||||
terminal_count += 1;
|
||||
turn_terminal_state = event
|
||||
.payload
|
||||
.get("turnTerminalState")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_owned);
|
||||
run_terminal_state = event
|
||||
.payload
|
||||
.get("runTerminalState")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_owned);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
if proposed_result_count != 1 {
|
||||
return Err(ReplayError::invalid(
|
||||
"scripted fixtures must contain exactly one unique run.result.proposed event",
|
||||
));
|
||||
}
|
||||
if terminal_count != 1 {
|
||||
return Err(ReplayError::invalid(
|
||||
"scripted fixtures must contain exactly one unique run.terminal event",
|
||||
));
|
||||
}
|
||||
|
||||
Ok(ReplayParitySummary {
|
||||
run_id: fixture.identity.run_id,
|
||||
integrity: if gaps.is_empty() {
|
||||
"complete".to_owned()
|
||||
} else {
|
||||
"gap_detected".to_owned()
|
||||
},
|
||||
timeline_count,
|
||||
duplicate_event_ids,
|
||||
gaps,
|
||||
turn_terminal_state,
|
||||
run_terminal_state,
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_optional_versioned_envelope(
|
||||
envelope: Option<&Value>,
|
||||
expected_schema: &str,
|
||||
field: &str,
|
||||
) -> Result<(), ReplayError> {
|
||||
let Some(envelope) = envelope else {
|
||||
return Ok(());
|
||||
};
|
||||
let schema = envelope
|
||||
.get("schema")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("missing");
|
||||
if schema != expected_schema {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required {field} schema {schema}; expected {expected_schema}"
|
||||
)));
|
||||
}
|
||||
let version = envelope
|
||||
.get("schemaVersion")
|
||||
.and_then(Value::as_u64)
|
||||
.unwrap_or(0);
|
||||
if version != 1 {
|
||||
return Err(ReplayError::invalid(format!(
|
||||
"unsupported required {field} schemaVersion {version}; expected 1"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
reduce_replay_fixture, ReplayParitySummary, MAX_EXACT_JSON_INTEGER,
|
||||
MAX_RECORDED_MISSING_SEQUENCES,
|
||||
};
|
||||
use serde_json::Value;
|
||||
|
||||
fn assert_fixture_parity(fixture: &str, expected: &str) {
|
||||
let actual = reduce_replay_fixture(fixture).expect("fixture should reduce");
|
||||
let expected: ReplayParitySummary =
|
||||
serde_json::from_str(expected).expect("golden summary should parse");
|
||||
assert_eq!(actual, expected);
|
||||
}
|
||||
|
||||
macro_rules! parity_case {
|
||||
($name:ident, $fixture:literal, $golden:literal) => {
|
||||
#[test]
|
||||
fn $name() {
|
||||
assert_fixture_parity(
|
||||
include_str!(concat!(
|
||||
"../../../../protocol/fixtures/replay/",
|
||||
$fixture,
|
||||
".json"
|
||||
)),
|
||||
include_str!(concat!(
|
||||
"../../../../protocol/fixtures/replay/golden/",
|
||||
$golden,
|
||||
".summary.json"
|
||||
)),
|
||||
);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
parity_case!(replay_fixture_parity_happy_path, "happy-path", "happy-path");
|
||||
parity_case!(replay_fixture_parity_failed_run, "failed-run", "failed-run");
|
||||
parity_case!(
|
||||
replay_fixture_parity_interrupted_run,
|
||||
"interrupted-run",
|
||||
"interrupted-run"
|
||||
);
|
||||
parity_case!(
|
||||
replay_fixture_parity_duplicate_event,
|
||||
"duplicate-event",
|
||||
"duplicate-event"
|
||||
);
|
||||
parity_case!(replay_fixture_parity_source_gap, "source-gap", "source-gap");
|
||||
parity_case!(
|
||||
replay_fixture_parity_unknown_optional_fields,
|
||||
"unknown-optional-fields",
|
||||
"unknown-optional-fields"
|
||||
);
|
||||
|
||||
#[test]
|
||||
fn replay_fixture_parity_rejects_unsupported_required_version() {
|
||||
let error = reduce_replay_fixture(include_str!(
|
||||
"../../../../protocol/fixtures/replay/unsupported-required-version.json"
|
||||
))
|
||||
.expect_err("PRP v2 fixture must fail closed");
|
||||
assert!(error
|
||||
.to_string()
|
||||
.contains("unsupported required protocolVersion 2"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_fixture_parity_rejects_unsupported_fixture_version() {
|
||||
let fixture = include_str!("../../../../protocol/fixtures/replay/happy-path.json")
|
||||
.replace("\"fixtureVersion\": 1", "\"fixtureVersion\": 2");
|
||||
let error =
|
||||
reduce_replay_fixture(&fixture).expect_err("fixture version 2 must fail closed");
|
||||
assert!(error
|
||||
.to_string()
|
||||
.contains("unsupported required fixtureVersion 2"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_fixture_parity_bounds_large_sequence_gap_details() {
|
||||
let mut fixture: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/fixtures/replay/source-gap.json"
|
||||
))
|
||||
.expect("fixture should parse");
|
||||
fixture["events"][2]["sourceSeq"] = Value::from(MAX_EXACT_JSON_INTEGER - 2);
|
||||
fixture["events"][3]["sourceSeq"] = Value::from(MAX_EXACT_JSON_INTEGER - 1);
|
||||
fixture["events"][4]["sourceSeq"] = Value::from(MAX_EXACT_JSON_INTEGER);
|
||||
|
||||
let summary = reduce_replay_fixture(&fixture.to_string()).expect("fixture should reduce");
|
||||
let gap = summary.gaps.first().expect("large gap should be recorded");
|
||||
assert_eq!(gap.expected, 3);
|
||||
assert_eq!(gap.received, MAX_EXACT_JSON_INTEGER - 2);
|
||||
assert_eq!(gap.missing_count, MAX_EXACT_JSON_INTEGER - 5);
|
||||
assert_eq!(gap.missing.len() as u64, MAX_RECORDED_MISSING_SEQUENCES);
|
||||
assert_eq!(gap.missing.last(), Some(&258));
|
||||
assert!(gap.truncated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_fixture_parity_rejects_inexact_json_sequence_values() {
|
||||
let mut fixture: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/fixtures/replay/happy-path.json"
|
||||
))
|
||||
.expect("fixture should parse");
|
||||
fixture["events"][0]["sourceSeq"] = Value::from(MAX_EXACT_JSON_INTEGER + 1);
|
||||
|
||||
let error = reduce_replay_fixture(&fixture.to_string())
|
||||
.expect_err("an inexact JSON sequence must fail closed");
|
||||
assert!(error
|
||||
.to_string()
|
||||
.contains("exceeds the exact JSON integer range"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_fixture_parity_rejects_a_mismatched_declared_result() {
|
||||
let mut fixture: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/fixtures/replay/happy-path.json"
|
||||
))
|
||||
.expect("fixture should parse");
|
||||
fixture["result"]["summary"] = Value::String("Contradictory result".to_owned());
|
||||
let error = reduce_replay_fixture(&fixture.to_string())
|
||||
.expect_err("mismatched result must fail closed");
|
||||
assert!(error
|
||||
.to_string()
|
||||
.contains("fixture result must match the run.result.proposed event payload"));
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,126 @@
|
|||
use std::io::{Read, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Command, Stdio};
|
||||
|
||||
use serde_json::{json, Value};
|
||||
|
||||
fn run_fixture(script_name: &str, commands: &[Value]) -> (std::process::ExitStatus, Vec<Value>) {
|
||||
let runnerd = PathBuf::from(env!("CARGO_BIN_EXE_paperclip-runnerd"));
|
||||
let fake_harness = PathBuf::from(env!("CARGO_BIN_EXE_fake-harness"));
|
||||
let script = Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../../../protocol/fixtures/local-runner/scripts")
|
||||
.join(format!("{script_name}.json"));
|
||||
let mut child = Command::new(runnerd)
|
||||
.args([
|
||||
"--run-id",
|
||||
"run_local_test",
|
||||
"--session-id",
|
||||
"session_local_test",
|
||||
"--runner-id",
|
||||
"runner_local_test",
|
||||
"--fake-harness",
|
||||
])
|
||||
.arg(fake_harness)
|
||||
.args(["--script"])
|
||||
.arg(script)
|
||||
.args(["--delay-ms", "0", "--shutdown-grace-ms", "100"])
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::inherit())
|
||||
.spawn()
|
||||
.expect("local runner should start");
|
||||
let mut stdin = child.stdin.take().expect("runner stdin should be piped");
|
||||
for command in commands {
|
||||
serde_json::to_writer(&mut stdin, command).expect("command should serialize");
|
||||
stdin.write_all(b"\n").expect("command should write");
|
||||
}
|
||||
stdin.flush().expect("commands should flush");
|
||||
|
||||
let mut output = String::new();
|
||||
child
|
||||
.stdout
|
||||
.take()
|
||||
.expect("runner stdout should be piped")
|
||||
.read_to_string(&mut output)
|
||||
.expect("runner output should be readable");
|
||||
let status = child.wait().expect("local runner should exit");
|
||||
let messages = output
|
||||
.lines()
|
||||
.map(|line| serde_json::from_str(line).expect("runner output should be JSONL"))
|
||||
.collect();
|
||||
(status, messages)
|
||||
}
|
||||
|
||||
fn command(id: &str, sequence: u64, command_type: &str, payload: Value) -> Value {
|
||||
json!({
|
||||
"schema": "paperclip.prp.command.v1",
|
||||
"commandId": id,
|
||||
"controllerSeq": sequence,
|
||||
"type": command_type,
|
||||
"issuedAt": "2026-08-07T21:00:00.000Z",
|
||||
"payload": payload,
|
||||
})
|
||||
}
|
||||
|
||||
fn event_type(message: &Value) -> Option<&str> {
|
||||
(message["message"]["kind"] == "event")
|
||||
.then(|| message["message"]["event"]["eventType"].as_str())
|
||||
.flatten()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn happy_path_emits_one_result_and_one_terminal() {
|
||||
let commands = [
|
||||
command("command_prepare", 1, "run.prepare", json!({})),
|
||||
command("command_session", 2, "session.open", json!({})),
|
||||
command(
|
||||
"command_turn",
|
||||
3,
|
||||
"turn.start",
|
||||
json!({ "turnId": "turn_local_test", "text": "Run the fixture." }),
|
||||
),
|
||||
];
|
||||
let (status, messages) = run_fixture("happy-path", &commands);
|
||||
|
||||
assert!(status.success());
|
||||
assert_eq!(
|
||||
messages
|
||||
.iter()
|
||||
.filter(|message| event_type(message) == Some("run.result.proposed"))
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
let terminals = messages
|
||||
.iter()
|
||||
.filter(|message| event_type(message) == Some("run.terminal"))
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(terminals.len(), 1);
|
||||
assert_eq!(
|
||||
terminals[0]["message"]["event"]["payload"]["runTerminalState"],
|
||||
"succeeded"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn equivalent_command_redelivery_is_idempotent() {
|
||||
let prepare = command("command_prepare", 1, "run.prepare", json!({}));
|
||||
let commands = [
|
||||
prepare.clone(),
|
||||
prepare,
|
||||
command("command_session", 2, "session.open", json!({})),
|
||||
command(
|
||||
"command_turn",
|
||||
3,
|
||||
"turn.start",
|
||||
json!({ "turnId": "turn_local_test", "text": "Run the fixture." }),
|
||||
),
|
||||
];
|
||||
let (status, messages) = run_fixture("happy-path", &commands);
|
||||
|
||||
assert!(status.success());
|
||||
assert!(messages.iter().any(|message| {
|
||||
message["message"]["kind"] == "command_receipt"
|
||||
&& message["message"]["commandId"] == "command_prepare"
|
||||
&& message["message"]["status"] == "duplicate"
|
||||
}));
|
||||
}
|
||||
|
|
@ -0,0 +1,176 @@
|
|||
#![cfg(unix)]
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::process::Stdio;
|
||||
use std::time::Duration;
|
||||
|
||||
use paperclip_runner_core::local_runner::HarnessCommand;
|
||||
use paperclip_runner_core::process_supervisor::SupervisedProcess;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
fn process_exists(pid: u64) -> bool {
|
||||
Command::new("kill")
|
||||
.args(["-0", &pid.to_string()])
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.status()
|
||||
.is_ok_and(|status| status.success())
|
||||
}
|
||||
|
||||
fn spawn_linger_process() -> (SupervisedProcess, u32, u64) {
|
||||
let harness = PathBuf::from(env!("CARGO_BIN_EXE_fake-harness"));
|
||||
let script = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../../../protocol/fixtures/local-runner/scripts/linger.json");
|
||||
let mut process = SupervisedProcess::spawn(
|
||||
&harness,
|
||||
&[
|
||||
"--script".to_owned(),
|
||||
script.display().to_string(),
|
||||
"--delay-ms".to_owned(),
|
||||
"1".to_owned(),
|
||||
],
|
||||
Duration::from_millis(50),
|
||||
64 * 1024,
|
||||
)
|
||||
.expect("fake harness should start");
|
||||
let harness_pid = process.id();
|
||||
process
|
||||
.receive_stdout_line(Duration::from_secs(1))
|
||||
.expect("ready line should be readable")
|
||||
.expect("ready line should exist");
|
||||
process
|
||||
.send(&HarnessCommand {
|
||||
schema: "paperclip.fake_harness.command.v1".to_owned(),
|
||||
command_id: "open".to_owned(),
|
||||
command_type: "session.open".to_owned(),
|
||||
payload: json!({}),
|
||||
})
|
||||
.expect("session.open should send");
|
||||
process
|
||||
.receive_stdout_line(Duration::from_secs(1))
|
||||
.expect("session line should be readable")
|
||||
.expect("session line should exist");
|
||||
process
|
||||
.send(&HarnessCommand {
|
||||
schema: "paperclip.fake_harness.command.v1".to_owned(),
|
||||
command_id: "turn".to_owned(),
|
||||
command_type: "turn.start".to_owned(),
|
||||
payload: json!({ "turnId": "turn_cleanup" }),
|
||||
})
|
||||
.expect("turn.start should send");
|
||||
|
||||
let mut worker_pid = None;
|
||||
for _ in 0..5 {
|
||||
let line = process
|
||||
.receive_stdout_line(Duration::from_secs(1))
|
||||
.expect("harness output should be readable")
|
||||
.expect("harness output should continue");
|
||||
let message: Value = serde_json::from_str(&line).expect("harness output should be JSON");
|
||||
if message["type"] == "diagnostic" {
|
||||
worker_pid = message["payload"]["workerPid"].as_u64();
|
||||
break;
|
||||
}
|
||||
}
|
||||
let worker_pid = worker_pid.expect("linger script should report its worker pid");
|
||||
assert!(process_exists(u64::from(harness_pid)));
|
||||
assert!(process_exists(worker_pid));
|
||||
(process, harness_pid, worker_pid)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn forced_process_group_cleanup_stops_harness_and_worker() {
|
||||
let (mut process, harness_pid, worker_pid) = spawn_linger_process();
|
||||
|
||||
process
|
||||
.terminate_group()
|
||||
.expect("process group cleanup should finish");
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
assert!(!process_exists(u64::from(harness_pid)));
|
||||
assert!(!process_exists(worker_pid));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn natural_harness_exit_also_cleans_up_workers() {
|
||||
let (mut process, harness_pid, worker_pid) = spawn_linger_process();
|
||||
process
|
||||
.send(&HarnessCommand {
|
||||
schema: "paperclip.fake_harness.command.v1".to_owned(),
|
||||
command_id: "interrupt".to_owned(),
|
||||
command_type: "turn.interrupt".to_owned(),
|
||||
payload: json!({ "reason": "cleanup_test" }),
|
||||
})
|
||||
.expect("turn.interrupt should send");
|
||||
process
|
||||
.wait()
|
||||
.expect("harness should exit after interruption");
|
||||
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
assert!(!process_exists(u64::from(harness_pid)));
|
||||
assert!(!process_exists(worker_pid));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn oversized_harness_stdout_frame_is_rejected() {
|
||||
let harness = PathBuf::from(env!("CARGO_BIN_EXE_fake-harness"));
|
||||
let script = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
|
||||
.join("../../../protocol/fixtures/local-runner/scripts/oversized-line.json");
|
||||
let mut process = SupervisedProcess::spawn(
|
||||
&harness,
|
||||
&[
|
||||
"--script".to_owned(),
|
||||
script.display().to_string(),
|
||||
"--delay-ms".to_owned(),
|
||||
"1".to_owned(),
|
||||
],
|
||||
Duration::from_millis(50),
|
||||
512,
|
||||
)
|
||||
.expect("fake harness should start");
|
||||
process
|
||||
.receive_stdout_line(Duration::from_secs(1))
|
||||
.expect("ready frame should fit")
|
||||
.expect("ready frame should exist");
|
||||
process
|
||||
.send(&HarnessCommand {
|
||||
schema: "paperclip.fake_harness.command.v1".to_owned(),
|
||||
command_id: "open".to_owned(),
|
||||
command_type: "session.open".to_owned(),
|
||||
payload: json!({}),
|
||||
})
|
||||
.expect("session.open should send");
|
||||
process
|
||||
.receive_stdout_line(Duration::from_secs(1))
|
||||
.expect("session frame should fit")
|
||||
.expect("session frame should exist");
|
||||
process
|
||||
.send(&HarnessCommand {
|
||||
schema: "paperclip.fake_harness.command.v1".to_owned(),
|
||||
command_id: "turn".to_owned(),
|
||||
command_type: "turn.start".to_owned(),
|
||||
payload: json!({ "turnId": "turn_oversized" }),
|
||||
})
|
||||
.expect("turn.start should send");
|
||||
process
|
||||
.receive_stdout_line(Duration::from_secs(1))
|
||||
.expect("turn frame should fit")
|
||||
.expect("turn frame should exist");
|
||||
|
||||
let mut oversized_error = None;
|
||||
for _ in 0..3 {
|
||||
match process.receive_stdout_line(Duration::from_secs(1)) {
|
||||
Ok(Some(_)) => {}
|
||||
Ok(None) => break,
|
||||
Err(error) => {
|
||||
oversized_error = Some(error);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
let error = oversized_error.expect("oversized harness frame must be rejected");
|
||||
assert!(error.to_string().contains("exceeded 512 bytes"));
|
||||
|
||||
process
|
||||
.terminate_group()
|
||||
.expect("oversized-frame harness should be cleaned up");
|
||||
}
|
||||
Loading…
Reference in New Issue