feat(runner): add qualified OpenCode runtime (#12588)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Paperclip Runner provides a durable execution boundary for supported
providers.
> - The current production runtime supports Codex but cannot execute
OpenCode sessions.
> - OpenCode needs a qualified transport, strict input mapping, and
normalized events.
> - This pull request adds the OpenCode runtime as one isolated provider
unit.
> - The benefit is a reviewable provider expansion that does not weaken
the existing Codex path.

## Linked Issues or Issue Description

**Subsystem affected**

Cross-cutting: packages/paperclip-runner and the Codex-local adapter
configuration contract.

**Problem or motivation**

Paperclip Runner has provider-neutral contracts, but the production
backend factory cannot start a qualified OpenCode session. This blocks
OpenCode from using the durable runner path.

**Proposed solution**

Add the qualified OpenCode app-server proxy, driver, MCP bridge,
backend, fixtures, and factory wiring. Keep existing Codex behavior
unchanged.

**Alternatives considered**

Keeping OpenCode only on the direct adapter path would avoid this
runtime work, but it would not provide durable runner recovery or
normalized provider events.

**Roadmap alignment**

ROADMAP.md does not list a conflicting provider-runtime project. This
change extends the existing Paperclip Runner architecture.

## What Changed

- Added the qualified OpenCode app-server proxy and input queue.
- Added collaboration-mode and provider-event normalization.
- Added the OpenCode MCP bridge and native session backend.
- Added strict fixtures and focused unit coverage.
- Added only the package exports and adapter configuration required by
this runtime.
- Kept deferred SDK, lab, eval, and public package surfaces out of this
change.

## Verification

- GitHub Actions is the authoritative verification environment for this
PR.
- Run the package type checks and focused OpenCode tests in CI.
- Run repository typecheck, test, build, security, and policy gates
through the stack-aware workflow.
- Local tests were not run because this checkout is resource
constrained.

## Risks

- OpenCode protocol changes could affect event normalization or
recovery.
- The driver fails closed on malformed input and unsupported runtime
behavior.
- Existing Codex selection remains unchanged unless the stored provider
is OpenCode.

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

## Model Used

OpenAI Codex, GPT-5.6, with repository tools, code execution, and
parallel agent review.

## 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 linked existing issues or described the issue in-PR
following the relevant issue template
- [x] I have not referenced internal or instance-local Paperclip issues
or links
- [x] My branch name describes the change and contains no internal
Paperclip ticket id
- [ ] I have run tests locally and they pass — GitHub Actions is
authoritative for this resource-constrained checkout
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

## Stack

- Position: 1 of 4
- Base: master
- Next: additional qualified provider runtimes
This commit is contained in:
Dotta authored and GitHub committed 2026-08-31 10:43:34 -05:00
1 parent 39eafad47d
commit 2e5a24e177
29 files changed
+4783 -30

No files matched your search

@@ -106,4 +106,44 @@ describe("buildPaperclipRunnerConfig", () => {
expect(config).not.toHaveProperty(unsupportedKey);
}
});
it("builds a bounded OpenCode runner profile from schema values", () => {
const config = buildPaperclipRunnerConfig(makeValues({
adapterType: "paperclip_runner",
model: "",
codexEngine: "acp",
dangerouslyBypassSandbox: true,
adapterSchemaValues: {
provider: "opencode",
opencodePermissionMode: "ask",
lifecycleMode: "warm",
idleTimeoutMs: 45_000,
},
}));
expect(config).toMatchObject({
provider: "opencode",
model: "openrouter/deepseek/deepseek-v4-flash-0731",
opencodePermissionMode: "ask",
lifecycleMode: "warm",
idleTimeoutMs: 45_000,
});
expect(config).not.toHaveProperty("engine");
expect(config).not.toHaveProperty("dangerouslyBypassApprovalsAndSandbox");
});
it("falls back to safe OpenCode defaults for invalid schema values", () => {
expect(buildPaperclipRunnerConfig(makeValues({
adapterSchemaValues: {
provider: "opencode",
opencodePermissionMode: "unrestricted",
lifecycleMode: "forever",
idleTimeoutMs: -1,
},
}))).toMatchObject({
provider: "opencode",
opencodePermissionMode: "allow",
lifecycleMode: "per_turn",
});
});
});
@@ -61,7 +61,15 @@ export function buildCodexLocalConfig(v: CreateConfigValues): Record<string, unk
return ac;
}
/** Build the Codex-only profile accepted by the experimental Rust runner. */
function paperclipRunnerProvider(value: unknown): "codex" | "opencode" {
return value === "opencode" ? "opencode" : "codex";
}
function openCodePermissionMode(value: unknown): "allow" | "ask" | "deny" {
return value === "ask" || value === "deny" ? value : "allow";
}
/** Build a provider profile accepted by the experimental Rust runner. */
export function buildPaperclipRunnerConfig(v: CreateConfigValues): Record<string, unknown> {
const config = buildCodexLocalConfig(v);
for (const unsupportedKey of [
@@ -82,5 +90,27 @@ export function buildPaperclipRunnerConfig(v: CreateConfigValues): Record<string
]) {
delete config[unsupportedKey];
}
return { ...config, provider: "codex" };
const schemaValues = v.adapterSchemaValues ?? {};
const provider = paperclipRunnerProvider(schemaValues.provider);
if (provider === "codex") {
return { ...config, provider };
}
const lifecycleMode = schemaValues.lifecycleMode === "warm" ? "warm" : "per_turn";
const configuredIdleTimeoutMs = schemaValues.idleTimeoutMs;
const idleTimeoutMs = typeof configuredIdleTimeoutMs === "number"
&& Number.isSafeInteger(configuredIdleTimeoutMs)
&& configuredIdleTimeoutMs > 0
? configuredIdleTimeoutMs
: 300_000;
return {
...config,
provider,
model: v.model || "openrouter/deepseek/deepseek-v4-flash-0731",
opencodePermissionMode: openCodePermissionMode(
schemaValues.opencodePermissionMode,
),
lifecycleMode,
...(lifecycleMode === "warm" ? { idleTimeoutMs } : {}),
};
}
+13 -3
View File
@@ -20,7 +20,10 @@ The package also publishes the canonical semantic action declarations and
their input and output schemas. Its package-local dispatcher projects only
bound, run-authorized actions and emits redacted semantic receipts.
The first and only installed provider is Codex. Dynamic semantic tools remain
The production-selected provider remains Codex. The package also contains the
qualified OpenCode 1.18.17 driver, authenticated loopback MCP bridge, and
runnerd-compatible proxy, but this slice does not authorize the server to
select OpenCode for a fresh native run. Dynamic semantic tools remain
undiscoverable unless the hidden server coordinator projects one of the five
same-task read bindings for an already persisted native Codex run. Catalog
membership alone does not grant authority. The server can now create and start
@@ -36,8 +39,8 @@ The package has two initial public surfaces:
- `@paperclipai/paperclip-runner/testing` adds Node-only fixture loading and a
provider-neutral semantic conformance kit for deterministic test adapters.
No SDK, browser, React, eval, live-console, lab, or provider-experiment entry
point is exported. The package remains private in this wave. The server route
No SDK, browser, React, eval, live-console, or lab entry point is exported. The
package remains private in this wave. The server route
at `/api/runner/v1/connect/:runId` has no authority until the hidden coordinator
registers an exact existing run binding. Fresh native starts are rejected
unless the instance `enableNativeRunner` flag is enabled. Existing direct
@@ -49,6 +52,13 @@ an installed server does not depend on a separate system Rust installation or
a manually copied binary. `pnpm-lock.yaml` remains under the repository's
existing lockfile process.
The `paperclip-runner-opencode-proxy` binary adapts the pinned OpenCode server
protocol to the normalized harness contract. It admits only a bounded provider
environment, uses an authenticated loopback MCP endpoint, validates dynamic
tool inputs, keeps terminal completion tools private to the runner boundary,
and preserves exact provider session identity for recovery. The backend factory
requires an explicit runtime directory before constructing this provider.
The package also builds `paperclip-runner-acpx-sidecar`. This bounded v2
stdin/stdout bridge admits the qualified Codex ACPX profile only. It validates
the exact model, session identity, tool catalog, structured input, and terminal
+3 -1
View File
@@ -8,6 +8,7 @@
},
"type": "module",
"bin": {
"paperclip-runner-opencode-proxy": "./dist/cli/opencode-app-server-proxy.js",
"paperclip-runner-acpx-sidecar": "./dist/cli/acpx-runtime-sidecar.js"
},
"exports": {
@@ -58,7 +59,8 @@
"@agentclientprotocol/codex-acp": "1.6.2",
"acpx": "0.13.1",
"ajv": "^8.20.0",
"json-schema-to-ts": "^3.1.1"
"json-schema-to-ts": "^3.1.1",
"opencode-ai": "1.18.17"
},
"devDependencies": {
"@types/node": "^24.0.0",
@@ -0,0 +1,43 @@
{
"schema": "paperclip.question_adapter_fixture.v1",
"adapter": "opencode",
"nativeRequest": {
"type": "question.asked",
"properties": {
"id": "request-1",
"sessionID": "session-1",
"title": "Deployment input",
"questions": [{
"question": "Where should we deploy?",
"header": "Environment",
"required": true,
"options": [
{ "label": "Staging", "description": "Deploy to staging first." },
{ "label": "Production", "description": "Deploy directly to production." }
],
"multiple": false
}]
}
},
"canonicalQuestionSet": {
"schema": "paperclip.question_set.v1",
"title": "Deployment input",
"submitLabel": "Submit answers",
"questions": [{
"id": "environment",
"header": "Environment",
"prompt": "Where should we deploy?",
"required": true,
"answerMode": "single_select",
"options": [
{ "id": "option-1", "label": "Staging", "description": "Deploy to staging first." },
{ "id": "option-2", "label": "Production", "description": "Deploy directly to production." }
]
}]
},
"canonicalResponse": {
"schema": "paperclip.question_response.v1",
"answers": { "environment": { "selectedOptionIds": ["option-1"] } }
},
"nativeResponse": { "answers": [["Staging"]] }
}
@@ -187,6 +187,12 @@
"expectation": "accept",
"compatibilityCase": "codex-structured-input"
},
{
"path": "fixtures/questions/opencode.json",
"sha256": "7e5f753d04321f78a38a4362b2a4c7aeacbb4736d4db72a699f23c245884da72",
"expectation": "accept",
"compatibilityCase": "opencode-structured-input"
},
{
"path": "fixtures/replay/duplicate-event.json",
"sha256": "779c146966d2825ffcde88bfff8b1fb238c162d40c578980ad3bd1a7def405d7",
@@ -93,6 +93,23 @@ function acpxExecution(agent: "codex" | "pi" = "codex"): NativeExecutionInput {
};
}
function opencodeExecution(): NativeExecutionInput {
return {
...execution(),
session: {
normalizedSessionId: "session",
driverKind: "opencode_server",
protocolVersion: 1,
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
},
provider: {
kind: "opencode",
model: "openrouter/model",
permissionMode: "allow",
},
};
}
describe("native backend factory", () => {
it("constructs the Codex backend without starting its transport", async () => {
const backend = createNativeSessionBackend(execution(), {
@@ -111,17 +128,27 @@ describe("native backend factory", () => {
});
});
it("fails closed when a deferred provider reaches the factory", () => {
it("requires an explicit runtime root for OpenCode", () => {
expect(() =>
createNativeSessionBackend(
execution({
kind: "opencode",
model: "openrouter/model",
}),
),
).toThrow(
"Native backend for opencode is not included in the Codex-first runner",
);
createNativeSessionBackend(opencodeExecution()),
).toThrow("OpenCode native backend requires an instance runtime directory");
});
it("constructs the OpenCode backend without starting its process", async () => {
const backend = createNativeSessionBackend(opencodeExecution(), {
opencodeRuntimeDirectory: "/runtime",
});
await expect(backend.descriptor()).resolves.toMatchObject({
kind: "runner",
name: "opencode_server",
version: "1.18.17",
capabilities: {
resume: true,
interruption: true,
dynamicTools: true,
},
});
});
it("constructs the qualified Codex ACPX backend without starting ACPX", async () => {
@@ -12,6 +12,7 @@ import {
createCodexAcpxNativeSessionBackend,
type CodexAcpxNativeSessionBackendOptions,
} from "./codex-acpx-native-backend.js";
import { createOpenCodeNativeSessionBackend } from "./opencode-native-backend.js";
export interface NativeBackendFactoryOptions extends Omit<
CodexNativeSessionBackendOptions,
@@ -24,6 +25,9 @@ export interface NativeBackendFactoryOptions extends Omit<
acpxEnvironment?: NodeJS.ProcessEnv;
acpxManagedCodexCredentialSourcePath?: string;
acpxDynamicToolHandler?: CodexAcpxNativeSessionBackendOptions["dynamicToolHandler"];
opencodeRuntimeDirectory?: string;
opencodeEnvironment?: NodeJS.ProcessEnv;
opencodeCommand?: string;
}
/**
@@ -35,6 +39,22 @@ export function createNativeSessionBackend(
input: NativeExecutionInput,
options: NativeBackendFactoryOptions = {},
): NativeSessionBackend {
if (input.provider.kind === "opencode") {
if (!options.opencodeRuntimeDirectory?.trim()) {
throw new Error(
"OpenCode native backend requires an instance runtime directory",
);
}
return createOpenCodeNativeSessionBackend(input, {
runtimeDirectory: options.opencodeRuntimeDirectory,
environment: options.opencodeEnvironment,
command: options.opencodeCommand,
runnerInstanceId: options.runnerInstanceId,
onSpawn: options.onSpawn,
dynamicTools: options.dynamicTools,
dynamicToolHandler: options.dynamicToolHandler,
});
}
if (input.provider.kind === "acpx") {
if (input.provider.agent !== "codex") {
throw new Error(
@@ -0,0 +1,32 @@
import { createCodexTaskEnvelope } from "../contracts/codex.js";
import type { NativeExecutionInput } from "../contracts/native-execution.js";
import type { NativeSessionBackend } from "../contracts/native-session-backend.js";
import { OpenCodeServerDriver, type OpenCodeServerDriverOptions } from "../drivers/opencode/opencode-server-driver.js";
import { HarnessDriverBackend } from "./harness-driver-backend.js";
import { nativeSystemInstructions, nativeTaskConstraints } from "./runtime-context.js";
export function createOpenCodeNativeSessionBackend(
input: NativeExecutionInput,
options: Omit<OpenCodeServerDriverOptions, "model" | "taskEnvelope">,
): NativeSessionBackend {
if (input.provider.kind !== "opencode" || !input.provider.model) {
throw new Error("OpenCode native backend requires a persisted OpenCode provider/model selection");
}
return new HarnessDriverBackend(new OpenCodeServerDriver({
...options,
systemInstructions: nativeSystemInstructions(input),
runtimeContext: "runtimeContext" in input ? input.runtimeContext : null,
model: input.provider.model,
permissionMode: input.provider.permissionMode ?? "allow",
taskEnvelope: createCodexTaskEnvelope({
objective: input.completionContract.contract.objective,
contractRevision: input.completionContract.contract.revision,
criteria: input.completionContract.contract.criteria,
constraints: [
"Work only inside the supplied working directory.",
...nativeTaskConstraints(input),
"Return one semantic completion result through paperclip_finish or paperclip_block.",
],
}),
}));
}
@@ -0,0 +1,23 @@
import { describe, expect, it } from "vitest";
import { openCodeProxyTaskEnvelope } from "./opencode-proxy-task-envelope.js";
describe("OpenCode runnerd proxy task envelope", () => {
it("uses the durable completion-contract binding instead of a demo revision", () => {
expect(openCodeProxyTaskEnvelope({
baseInstructions: "Complete only this task.",
completionContract: {
revision: "17",
criterionIds: ["objective", "verification"],
},
})).toMatchObject({
completionContract: {
revision: "17",
criteria: [
{ id: "objective" },
{ id: "verification" },
],
},
});
});
});
@@ -0,0 +1,340 @@
#!/usr/bin/env node
import { createInterface } from "node:readline";
import { readFileSync } from "node:fs";
import { createRequire } from "node:module";
import { resolve } from "node:path";
import type { HarnessRuntimeRequestResolution, HarnessSession, PersistedHarnessSession } from "../contracts/harness-driver.js";
import { OpenCodeServerDriver } from "../drivers/opencode/opencode-server-driver.js";
import { parseNativeRuntimeContext } from "../contracts/runtime-context.js";
import { openCodeProxyTaskEnvelope } from "./opencode-proxy-task-envelope.js";
import {
shouldAnnounceOpenCodeProxyTurn,
shouldForwardOpenCodeProxyItem,
} from "./opencode-proxy-events.js";
import { enqueueOpenCodeProxyInput } from "./opencode-proxy-input.js";
import {
assertOpenCodeProxyCollaborationMode,
openCodeProxyCollaborationModes,
} from "./opencode-proxy-collaboration-mode.js";
type RpcMessage = { id?: string | number; method?: string; params?: unknown; result?: unknown; error?: unknown };
const input = createInterface({ input: process.stdin, crlfDelay: Infinity });
const pending = new Map<string, { resolve(value: unknown): void; reject(error: Error): void }>();
let nextServerRequestId = 1;
let driver: OpenCodeServerDriver | null = null;
let session: HarnessSession | null = null;
let eventPump: Promise<void> | null = null;
let cwd = "";
let activeModel = "";
let activeTurnId: string | null = null;
const announcedTurnIds = new Set<string>();
function openCodeCommand(): string {
const configured = process.env.PAPERCLIP_OPENCODE_COMMAND?.trim();
if (configured && configured !== "opencode") return configured;
try {
return createRequire(import.meta.url).resolve("opencode-ai/bin/opencode.exe");
} catch {
return configured || "opencode";
}
}
function send(value: unknown): void {
process.stdout.write(`${JSON.stringify(value)}\n`);
}
function requestController(method: string, params: unknown): Promise<unknown> {
const id = `opencode-${nextServerRequestId++}`;
send({ id, method, params });
return new Promise((resolveValue, reject) => pending.set(id, { resolve: resolveValue, reject }));
}
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
}
function text(value: unknown, fallback = ""): string {
return typeof value === "string" ? value : fallback;
}
function runtimeDirectory(): string {
const configured = process.env.PAPERCLIP_OPENCODE_RUNTIME_DIR?.trim();
if (!configured) throw new Error("PAPERCLIP_OPENCODE_RUNTIME_DIR is required");
return resolve(configured);
}
async function open(params: Record<string, unknown>, resume: boolean): Promise<Record<string, unknown>> {
if (session) return threadResponse(session.ids().providerSessionId ?? session.ids().driverSessionId);
cwd = resolve(text(params.cwd, process.cwd()));
const model = text(params.model);
if (!model.includes("/")) throw new Error("OpenCode proxy requires model in provider/model form");
activeModel = model;
const dynamicTools = Array.isArray(params.dynamicTools) ? params.dynamicTools.map(record) : [];
const runtimeContextPath = process.env.PAPERCLIP_NATIVE_RUNTIME_CONTEXT_PATH?.trim();
const runtimeContext = runtimeContextPath
? parseNativeRuntimeContext(JSON.parse(readFileSync(runtimeContextPath, "utf8")))
: null;
driver = new OpenCodeServerDriver({
model,
command: openCodeCommand(),
runtimeDirectory: runtimeDirectory(),
environment: process.env,
runnerInstanceId: process.env.PAPERCLIP_RUNNER_INSTANCE_ID ?? "paperclip-runnerd-opencode",
taskEnvelope: openCodeProxyTaskEnvelope(params),
systemInstructions: text(params.baseInstructions, "Complete only the supplied task."),
runtimeContext,
dynamicTools,
dynamicToolHandler: async (call) => requestController("item/tool/call", {
threadId: session?.ids().driverSessionId ?? "opening",
turnId: call.turnId,
callId: call.callId,
tool: call.tool,
arguments: call.arguments,
}),
onDiagnostic: (message) => process.stderr.write(`[opencode] ${message}\n`),
// runnerd gives this proxy its own process group. Keep `opencode serve` in
// that same group so runnerd's TERM/KILL fallback cannot orphan it.
isolateProcessGroup: false,
});
if (resume) {
const threadId = text(params.threadId);
const snapshot: PersistedHarnessSession = {
driverKind: "opencode_server",
driverSessionId: threadId,
providerSessionId: threadId,
runId: process.env.PAPERCLIP_RUN_ID ?? "runnerd-run",
normalizedSessionId: process.env.PAPERCLIP_NORMALIZED_SESSION_ID ?? threadId,
activeTurnId: null,
lastSourceSequence: 0,
};
const recovered = await driver.recoverSession(snapshot);
if (!recovered.recovered || !recovered.session) throw new Error(recovered.reason ?? "OpenCode recovery failed");
session = recovered.session;
} else {
session = await driver.openSession({
runId: process.env.PAPERCLIP_RUN_ID ?? "runnerd-run",
normalizedSessionId: process.env.PAPERCLIP_NORMALIZED_SESSION_ID ?? `runnerd-${Date.now()}`,
workingDirectory: cwd,
});
}
eventPump = pumpEvents(session);
void eventPump.catch((error) => failProxy(error));
return threadResponse(session.ids().providerSessionId ?? session.ids().driverSessionId);
}
function threadResponse(id: string): Record<string, unknown> {
return { thread: { id, sessionId: id, cwd }, model: activeModel };
}
function announceTurnStarted(opened: HarnessSession, turnId: string): void {
if (!shouldAnnounceOpenCodeProxyTurn(announcedTurnIds, turnId)) return;
send({
method: "turn/started",
params: {
threadId: opened.ids().driverSessionId,
turnId,
turn: { id: turnId, status: "inProgress" },
},
});
}
async function pumpEvents(opened: HarnessSession): Promise<void> {
for await (const event of opened.events()) {
const payload = record(event.payload);
if (event.eventType === "turn.started") {
if (typeof event.turnId === "string") announceTurnStarted(opened, event.turnId);
} else if (event.eventType === "item.started" || event.eventType === "item.delta" || event.eventType === "item.completed") {
// The inner OpenCode driver publishes session-scoped model metadata as
// an item before any turn exists. The outer Codex protocol facade emits
// its own model item, so forwarding this unbound duplicate would quite
// correctly trip the facade's strict turn-binding validator.
if (!shouldForwardOpenCodeProxyItem({ turnId: event.turnId, kind: payload.kind })) continue;
const assistantText = payload.kind === "text" && typeof payload.text === "string"
? payload.text
: null;
const method = event.eventType === "item.delta" && assistantText !== null
? "item/agentMessage/delta"
: event.eventType.replace(".", "/");
const providerItem = record(payload.item);
const item = assistantText === null
? payload.item ?? { id: event.itemId, type: payload.kind, text: payload.text }
: { ...providerItem, id: event.itemId, type: "agentMessage", text: assistantText };
send({
method,
params: {
threadId: opened.ids().driverSessionId,
turnId: event.turnId,
itemId: event.itemId,
...payload,
item,
...(method === "item/agentMessage/delta" ? { delta: assistantText } : {}),
},
});
if (payload.kind === "usage") {
const usage = record(payload.usage);
const cache = record(usage.cache);
send({ method: "thread/tokenUsage/updated", params: {
threadId: opened.ids().driverSessionId,
turnId: event.turnId,
tokenUsage: { total: {
inputTokens: usage.inputTokens ?? usage.input ?? 0,
outputTokens: usage.outputTokens ?? usage.output ?? 0,
cachedInputTokens: usage.cachedInputTokens ?? cache.read ?? 0,
reasoningTokens: usage.reasoningTokens ?? usage.reasoning ?? 0,
costUsd: usage.costUsd ?? usage.cost ?? null,
} },
} });
}
} else if (event.eventType === "run.result.proposed") {
send({
method: "paperclip/runResult",
params: {
threadId: opened.ids().driverSessionId,
turnId: event.turnId,
itemId: event.itemId ?? "semantic-result",
result: payload,
},
});
} else if (event.eventType === "runtime_request.created") {
const request = record(payload.request);
const requestId = text(request.requestId);
const turnId = text(request.turnId, text(event.turnId, activeTurnId ?? ""));
if (!requestId || !turnId || !opened.resolveRuntimeRequest) {
throw new Error("OpenCode runtime request is not resolvable");
}
// Keep consuming OpenCode SSE while the controller waits for the user.
// If the underlying provider disappears, the stream can then fail the
// proxy and runnerd will expire the still-pending canonical request.
void (async () => {
const controllerResponse = record(await requestController("paperclip/runtimeRequest", {
request,
}));
await opened.resolveRuntimeRequest!({
requestId,
turnId,
resolution: record(controllerResponse.resolution) as HarnessRuntimeRequestResolution,
});
})().catch((error) => failProxy(error));
} else if (["turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled"].includes(event.eventType)) {
const status = event.eventType.slice("turn.".length);
send({ method: "turn/completed", params: { threadId: opened.ids().driverSessionId, turnId: event.turnId, turn: { id: event.turnId, status } } });
activeTurnId = null;
} else if (event.eventType === "harness.diagnostic") {
send({ method: "warning", params: { threadId: opened.ids().driverSessionId, message: payload.message ?? payload.code } });
}
}
}
async function handle(message: RpcMessage): Promise<void> {
if (message.method === undefined && message.id !== undefined) {
const waiter = pending.get(String(message.id));
if (!waiter) return;
pending.delete(String(message.id));
if (message.error !== undefined) waiter.reject(new Error(JSON.stringify(message.error)));
else {
const contentItems = record(message.result).contentItems;
waiter.resolve(Array.isArray(contentItems) ? contentItems[0] ?? message.result : message.result);
}
return;
}
if (!message.method || message.id === undefined) return;
const params = record(message.params);
let result: unknown;
switch (message.method) {
case "initialize":
result = { user: { sessionId: "opencode" }, serverInfo: { name: "opencode", version: "1.18.17" } };
break;
case "thread/start": result = await open(params, false); break;
case "thread/resume": result = await open(params, true); break;
case "collaborationMode/list":
result = openCodeProxyCollaborationModes(activeModel);
break;
case "turn/start": {
if (!session) throw new Error("OpenCode thread is not open");
assertOpenCodeProxyCollaborationMode(params);
const inputItems = Array.isArray(params.input) ? params.input.map(record) : [];
const messageText = inputItems.map((entry) => text(entry.text)).filter(Boolean).join("\n");
const turn = await session.startTurn({ message: { role: "user", text: messageText } });
activeTurnId = turn.turnId;
// OpenCode normally publishes session/turn startup over SSE, but a fast
// completion can make the synchronous prompt response the only place the
// authoritative turn id is observed. Emit the normalized notification
// after this request's JSON-RPC response when SSE did not already announce
// it. runnerd buffers notifications received while awaiting a response,
// so replaying an earlier SSE announcement would violate strict turn
// binding at the outer driver.
queueMicrotask(() => {
announceTurnStarted(session!, turn.turnId);
});
result = { turn: { id: turn.turnId, status: "inProgress" } };
break;
}
case "turn/interrupt":
if (!session?.interrupt) throw new Error("OpenCode interruption is unavailable");
await session.interrupt({ turnId: text(params.turnId, activeTurnId ?? "") });
result = true;
break;
case "thread/read":
if (!session) throw new Error("OpenCode thread is not open");
result = { thread: { id: session.ids().driverSessionId, cwd, turns: activeTurnId ? [{ id: activeTurnId, status: "inProgress" }] : [] }, transcript: await session.read?.() };
break;
default: throw new Error(`Unsupported OpenCode proxy method ${message.method}`);
}
send({ id: message.id, result });
}
let pendingInput = Promise.resolve();
let bootstrapFailure: Error | null = null;
input.on("line", (line) => {
if (!line.trim()) return;
let message: RpcMessage;
try { message = JSON.parse(line) as RpcMessage; }
catch (error) { process.stderr.write(`Invalid JSON-RPC input: ${String(error)}\n`); return; }
pendingInput = enqueueOpenCodeProxyInput(
pendingInput,
async () => {
if (bootstrapFailure) {
throw new Error(
`OpenCode provider bootstrap failed before ${message.method ?? "the dependent command"}: ${bootstrapFailure.message}`,
);
}
await handle(message);
},
(error) => {
const normalized = error instanceof Error ? error : new Error(String(error));
if (
message.method === "initialize"
|| message.method === "thread/start"
|| message.method === "thread/resume"
) {
bootstrapFailure ??= normalized;
}
send({
id: message.id,
error: { code: -32000, message: normalized.message },
});
},
);
});
let shutdownPromise: Promise<void> | null = null;
function failProxy(error: unknown): void {
process.stderr.write(`[opencode] fatal proxy error: ${error instanceof Error ? error.message : String(error)}\n`);
void shutdown(1);
}
function shutdown(exitCode = 0): Promise<void> {
shutdownPromise ??= (async () => {
for (const waiter of pending.values()) waiter.reject(new Error("OpenCode proxy is shutting down"));
pending.clear();
await session?.close({ reason: "proxy_shutdown", force: true }).catch(() => {});
await eventPump?.catch(() => {});
})().finally(() => process.exit(exitCode));
return shutdownPromise;
}
input.on("close", () => { void pendingInput.then(() => shutdown()); });
process.on("SIGTERM", () => { void shutdown(); });
process.on("SIGINT", () => { void shutdown(); });
@@ -0,0 +1,44 @@
import { describe, expect, it } from "vitest";
import {
assertOpenCodeProxyCollaborationMode,
openCodeProxyCollaborationModes,
} from "./opencode-proxy-collaboration-mode.js";
describe("OpenCode runnerd proxy collaboration mode", () => {
it("advertises proxy-owned planning against the resolved model", () => {
expect(
openCodeProxyCollaborationModes(
"openrouter/deepseek/deepseek-v4-flash-0731",
),
).toEqual({
data: [
{
name: "Plan",
mode: "plan",
model: "openrouter/deepseek/deepseek-v4-flash-0731",
reasoning_effort: null,
},
],
});
});
it("accepts the negotiated plan payload on turn/start", () => {
expect(() =>
assertOpenCodeProxyCollaborationMode({
collaborationMode: {
mode: "plan",
settings: { model: "openrouter/example" },
},
}),
).not.toThrow();
});
it("rejects collaboration modes the proxy does not implement", () => {
expect(() =>
assertOpenCodeProxyCollaborationMode({
collaborationMode: { mode: "pair-programming" },
}),
).toThrow("Unsupported OpenCode proxy collaboration mode pair-programming");
});
});
@@ -0,0 +1,41 @@
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
}
/**
* OpenCode does not expose Codex app-server collaboration presets itself, but
* the Paperclip proxy implements planning semantics through its task envelope,
* read-only permission profile, and Paperclip runtime tools. Advertise that
* proxy-owned capability so runnerd can negotiate planning mode without
* pretending OpenCode supplied a native Codex preset.
*/
export function openCodeProxyCollaborationModes(model: string) {
const resolvedModel = model.trim();
if (!resolvedModel) {
throw new Error(
"OpenCode collaboration modes are unavailable before thread/start resolves a model",
);
}
return {
data: [
{
name: "Plan",
mode: "plan",
model: resolvedModel,
reasoning_effort: null,
},
],
};
}
export function assertOpenCodeProxyCollaborationMode(params: unknown): void {
const collaborationMode = record(record(params).collaborationMode);
if (Object.keys(collaborationMode).length === 0) return;
if (collaborationMode.mode !== "plan") {
throw new Error(
`Unsupported OpenCode proxy collaboration mode ${String(collaborationMode.mode)}`,
);
}
}
@@ -0,0 +1,30 @@
import { describe, expect, it } from "vitest";
import {
shouldAnnounceOpenCodeProxyTurn,
shouldForwardOpenCodeProxyItem,
} from "./opencode-proxy-events.js";
describe("OpenCode runnerd proxy event boundary", () => {
it("drops only the duplicate session-scoped model item", () => {
expect(
shouldForwardOpenCodeProxyItem({ kind: "model" }),
).toBe(false);
expect(
shouldForwardOpenCodeProxyItem({
turnId: "turn-1",
kind: "model",
}),
).toBe(true);
expect(
shouldForwardOpenCodeProxyItem({ kind: "agentMessage" }),
).toBe(true);
});
it("announces one outer turn when SSE wins the response race", () => {
const announced = new Set<string>();
expect(shouldAnnounceOpenCodeProxyTurn(announced, "turn-1")).toBe(true);
expect(shouldAnnounceOpenCodeProxyTurn(announced, "turn-1")).toBe(false);
expect([...announced]).toEqual(["turn-1"]);
});
});
@@ -0,0 +1,15 @@
export function shouldForwardOpenCodeProxyItem(input: {
turnId?: string;
kind?: unknown;
}): boolean {
return !(input.turnId === undefined && input.kind === "model");
}
export function shouldAnnounceOpenCodeProxyTurn(
announcedTurnIds: Set<string>,
turnId: string,
): boolean {
if (announcedTurnIds.has(turnId)) return false;
announcedTurnIds.add(turnId);
return true;
}
@@ -0,0 +1,59 @@
import { describe, expect, it } from "vitest";
import { enqueueOpenCodeProxyInput } from "./opencode-proxy-input.js";
describe("OpenCode proxy input sequencing", () => {
it("drains requests in order before EOF shutdown", async () => {
let releaseFirst!: () => void;
const firstBlocked = new Promise<void>((resolve) => {
releaseFirst = resolve;
});
const events: string[] = [];
let pending = Promise.resolve();
pending = enqueueOpenCodeProxyInput(
pending,
async () => {
events.push("initialize:start");
await firstBlocked;
events.push("initialize:complete");
},
() => events.push("initialize:error"),
);
pending = enqueueOpenCodeProxyInput(
pending,
async () => {
events.push("thread:start");
},
() => events.push("thread:error"),
);
const shutdown = pending.then(() => {
events.push("shutdown");
});
await Promise.resolve();
expect(events).toEqual(["initialize:start"]);
releaseFirst();
await shutdown;
expect(events).toEqual([
"initialize:start",
"initialize:complete",
"thread:start",
"shutdown",
]);
});
it("reports a failed queued request without rejecting the drain", async () => {
const events: string[] = [];
let pending = Promise.resolve();
pending = enqueueOpenCodeProxyInput(
pending,
async () => {
throw new Error("initialize failed");
},
(error) => events.push((error as Error).message),
);
await pending;
expect(events).toEqual(["initialize failed"]);
});
});
@@ -0,0 +1 @@
export { enqueueSerialInput as enqueueOpenCodeProxyInput } from "./serial-input-queue.js";
@@ -0,0 +1,38 @@
import { createCodexTaskEnvelope } from "../contracts/codex.js";
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
function text(value: unknown, fallback = ""): string {
return typeof value === "string" ? value : fallback;
}
export function openCodeProxyTaskEnvelope(params: Record<string, unknown>) {
const completionContract = record(params.completionContract);
const revision = text(completionContract.revision).trim();
const criterionIds = Array.isArray(completionContract.criterionIds)
? completionContract.criterionIds
.filter((value): value is string => typeof value === "string" && value.trim().length > 0)
.map((value) => value.trim())
: [];
return createCodexTaskEnvelope({
objective: "Complete the provider turn supplied by Paperclip Runner.",
...(revision && criterionIds.length > 0
? {
contractRevision: revision,
criteria: criterionIds.map((id) => ({
id,
requirement: `Satisfy the Paperclip completion criterion ${id}.`,
})),
}
: {}),
constraints: [
text(params.baseInstructions, "Complete only the supplied task."),
"Work only inside the supplied working directory.",
"Use Paperclip MCP tools for semantic operations.",
],
});
}
@@ -17,5 +17,10 @@ export function enqueueSerialInput(
// Diagnostics must not poison the queue or skip later input frames.
}
};
return pending.catch(reportError).then(operation).catch(reportError);
return pending
.then(operation, (error) => {
reportError(error);
return operation();
})
.catch(reportError);
}
@@ -0,0 +1,62 @@
import { chmod, mkdtemp, readFile, stat, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import { createProviderTraceFileSink } from "./provider-trace-file-sink.js";
const roots: string[] = [];
afterEach(async () => {
const { rm } = await import("node:fs/promises");
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
});
describe("ProviderTraceFileSink", () => {
it("hardens an existing file before it appends provider frames", async () => {
const root = await mkdtemp(join(tmpdir(), "paperclip-provider-trace-mode-"));
roots.push(root);
const tracePath = join(root, "provider.ndjson");
await writeFile(tracePath, "", { mode: 0o644 });
await chmod(tracePath, 0o644);
const sink = await createProviderTraceFileSink({
path: tracePath,
provider: "opencode",
channel: "test",
});
expect(statMode(await stat(tracePath))).toBe(0o600);
await sink?.finish();
});
it("redacts registered credentials before raw frames reach disk", async () => {
const root = await mkdtemp(join(tmpdir(), "paperclip-provider-trace-secret-"));
roots.push(root);
const tracePath = join(root, "provider.ndjson");
const sink = await createProviderTraceFileSink({
path: tracePath,
provider: "opencode",
channel: "test",
});
sink?.addSensitiveValues(["provider-secret", "bridge-secret"]);
sink?.frame({
direction: "provider_stderr",
raw: "credential=provider-secret authorization=bridge-secret",
transport: "process_stderr",
});
await sink?.finish();
const trace = await readFile(tracePath, "utf8");
expect(trace).not.toContain("provider-secret");
expect(trace).not.toContain("bridge-secret");
const frame = JSON.parse(trace.trim().split("\n")[0]!) as Record<string, unknown>;
expect(Buffer.from(String(frame.rawBase64), "base64").toString("utf8"))
.toBe("credential=[REDACTED] authorization=[REDACTED]");
});
});
function statMode(value: Awaited<ReturnType<typeof stat>>): number {
return value.mode & 0o777;
}
@@ -0,0 +1,248 @@
import { createHash } from "node:crypto";
import { open, type FileHandle } from "node:fs/promises";
const DEFAULT_MAX_BYTES = 64 * 1024 * 1024;
const MAX_PENDING_RECORDS = 1_024;
type TraceRecord = Record<string, unknown>;
type ProviderTraceDirection =
| "client_to_provider"
| "provider_to_client"
| "provider_stderr";
type ProviderTraceDisposition =
| "mapped"
| "generic"
| "ignored"
| "rejected"
| "operator_only";
type ProviderTraceFieldMapping = {
inputPath?: string;
outputPath?: string;
action: "copied" | "renamed" | "normalized" | "derived" | "dropped" | "redacted";
reason?: string;
};
/**
* A bounded, fire-and-forget file spool for native providers implemented in
* TypeScript. Provider execution never awaits writes on its hot path; a slow
* or failed sidecar becomes an incomplete trace instead of run backpressure.
*/
export class ProviderTraceFileSink {
readonly #file: FileHandle;
readonly #provider: string;
readonly #channel: string;
readonly #maxBytes: number;
readonly #sensitiveValues = new Set<string>();
readonly #queue: string[] = [];
#nextFrameId: number;
#nextDebugSequence: number;
#capturedBytes: number;
#draining: Promise<void> | null = null;
#closed = false;
#truncated = false;
#incompleteReason: string | null = null;
constructor(input: {
file: FileHandle;
provider: string;
channel: string;
maxBytes?: number;
nextFrameId?: number;
nextDebugSequence?: number;
capturedBytes?: number;
}) {
this.#file = input.file;
this.#provider = input.provider;
this.#channel = input.channel;
this.#maxBytes = input.maxBytes ?? DEFAULT_MAX_BYTES;
this.#nextFrameId = input.nextFrameId ?? 1;
this.#nextDebugSequence = input.nextDebugSequence ?? 1;
this.#capturedBytes = input.capturedBytes ?? 0;
}
addSensitiveValues(values: readonly (string | undefined)[]): void {
for (const value of values) {
if (value) this.#sensitiveValues.add(value);
}
}
frame(input: {
direction: ProviderTraceDirection;
raw: Uint8Array | string;
transport: string;
nativeMethod?: string;
}): number | null {
if (this.#closed) return null;
const raw = this.#redact(
typeof input.raw === "string" ? Buffer.from(input.raw) : Buffer.from(input.raw),
);
if (this.#capturedBytes + raw.byteLength > this.#maxBytes) {
this.#truncated = true;
return null;
}
const frameId = this.#nextFrameId++;
this.#capturedBytes += raw.byteLength;
this.#append({
kind: "frame",
schema: "paperclip.provider_trace_frame.v1",
frameId,
timestamp: new Date().toISOString(),
direction: input.direction,
transport: input.transport,
provider: this.#provider,
...(input.nativeMethod ? { nativeMethod: input.nativeMethod } : {}),
byteLength: raw.byteLength,
digest: `sha256:${createHash("sha256").update(raw).digest("hex")}`,
rawBase64: raw.toString("base64"),
});
return frameId;
}
interpretation(input: {
frameId: number;
stage: string;
ruleId: string;
disposition: ProviderTraceDisposition;
emittedEventIds?: string[];
droppedFields?: string[];
fieldMappings?: ProviderTraceFieldMapping[];
reason: string;
}): void {
if (this.#closed) return;
this.#append({
kind: "interpretation",
schema: "paperclip.provider_trace_interpretation.v1",
frameId: input.frameId,
stage: input.stage,
ruleId: input.ruleId,
disposition: input.disposition,
emittedEventIds: input.emittedEventIds ?? [],
droppedFields: input.droppedFields ?? [],
fieldMappings: input.fieldMappings ?? [],
reason: input.reason,
});
}
async finish(input: { reason?: string | null } = {}): Promise<void> {
if (this.#closed) {
await this.#draining;
return;
}
const reason = input.reason ?? this.#incompleteReason;
const status = reason
? "incomplete"
: this.#truncated
? "truncated"
: "complete";
const acknowledgedDebugSequence = this.#nextDebugSequence - 1;
this.#append({
kind: "trace_status",
status,
acknowledgedDebugSequence,
reason: reason ?? (this.#truncated ? "provider_trace_max_bytes_exceeded" : null),
});
this.#closed = true;
await this.#draining;
await this.#file.close().catch(() => undefined);
}
#append(value: TraceRecord): void {
const record = {
...value,
debugChannel: this.#channel,
debugSequence: this.#nextDebugSequence++,
};
if (this.#queue.length >= MAX_PENDING_RECORDS) {
this.#incompleteReason ??= "trace_sidecar_spool_full";
return;
}
this.#queue.push(`${JSON.stringify(record)}\n`);
if (this.#draining === null) this.#scheduleDrain();
}
#scheduleDrain(): void {
this.#draining = this.#drain().finally(() => {
this.#draining = null;
if (this.#queue.length > 0) this.#scheduleDrain();
});
}
async #drain(): Promise<void> {
while (this.#queue.length > 0) {
const line = this.#queue.shift()!;
try {
await this.#file.appendFile(line, { encoding: "utf8" });
} catch {
this.#incompleteReason ??= "trace_sidecar_write_failed";
this.#queue.length = 0;
return;
}
}
}
#redact(raw: Buffer): Buffer {
if (this.#sensitiveValues.size === 0) return raw;
let value = raw.toString("utf8");
for (const sensitive of [...this.#sensitiveValues].sort(
(left, right) => right.length - left.length,
)) {
value = value.split(sensitive).join("[REDACTED]");
}
return Buffer.from(value, "utf8");
}
}
export async function createProviderTraceFileSink(input: {
path?: string;
provider: string;
channel: string;
maxBytes?: string | number;
}): Promise<ProviderTraceFileSink | null> {
if (!input.path) return null;
const maxBytes = typeof input.maxBytes === "number"
? input.maxBytes
: Number.parseInt(input.maxBytes ?? "", 10);
// Keep one descriptor for the sink lifetime. This prevents a path swap
// between permission hardening, replay inspection, and later appends.
const file = await open(input.path, "a+", 0o600);
let existing: string;
try {
// Opening an existing file preserves its prior mode. Tighten it before
// reading or appending so no provider frame has a readable exposure window.
await file.chmod(0o600);
existing = await file.readFile({ encoding: "utf8" });
} catch (error) {
await file.close().catch(() => undefined);
throw error;
}
let nextFrameId = 1;
let nextDebugSequence = 1;
let capturedBytes = 0;
for (const line of existing.split("\n")) {
if (!line.trim()) continue;
try {
const entry = JSON.parse(line) as Record<string, unknown>;
if (entry.kind === "frame") {
nextFrameId = Math.max(nextFrameId, Number(entry.frameId) + 1 || 1);
capturedBytes += Number(entry.byteLength) || 0;
}
if (entry.debugChannel === input.channel) {
nextDebugSequence = Math.max(
nextDebugSequence,
Number(entry.debugSequence) + 1 || 1,
);
}
} catch {
// The server-side integrity pass will classify a malformed prior record.
}
}
return new ProviderTraceFileSink({
file,
provider: input.provider,
channel: input.channel,
maxBytes: Number.isSafeInteger(maxBytes) && maxBytes > 0 ? maxBytes : DEFAULT_MAX_BYTES,
nextFrameId,
nextDebugSequence,
capturedBytes,
});
}
@@ -0,0 +1,182 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { canonicalOpenCodeMcpToolName, startOpenCodeMcpBridge, type OpenCodeMcpBridge } from "./mcp-bridge.js";
const bridges: OpenCodeMcpBridge[] = [];
afterEach(async () => {
await Promise.all(bridges.splice(0).map((bridge) => bridge.close()));
});
async function rpc(
bridge: OpenCodeMcpBridge,
body: Record<string, unknown>,
secret = bridge.secret,
) {
return fetch(bridge.url, {
method: "POST",
headers: {
Authorization: `Bearer ${secret}`,
"Content-Type": "application/json",
Accept: "application/json, text/event-stream",
},
body: JSON.stringify({ jsonrpc: "2.0", ...body }),
});
}
describe("OpenCode MCP bridge", () => {
it("binds to loopback, requires authentication, and exposes only its closed catalog", async () => {
const bridge = await startOpenCodeMcpBridge({
secret: "session-secret",
tools: [{ name: "documents.read", description: "Read", inputSchema: { type: "object" } }],
handler: async () => ({ ok: true }),
});
bridges.push(bridge);
expect(new URL(bridge.url).hostname).toBe("127.0.0.1");
expect((await rpc(bridge, { id: 1, method: "tools/list" }, "wrong")).status).toBe(401);
const listed = await rpc(bridge, { id: 2, method: "tools/list" });
expect(await listed.json()).toMatchObject({
result: { tools: [
{ name: "documents.read" },
{ name: "paperclip_finish" },
{ name: "paperclip_block" },
] },
});
});
it("admits runner-private extension operations without advertising them to the model", async () => {
const handler = vi.fn(async ({ tool }) => ({ decision: tool === "__paperclip_permission" ? "accept" : "decline" }));
const bridge = await startOpenCodeMcpBridge({
tools: [{ name: "documents.read", inputSchema: { type: "object" } }],
privateTools: [{ name: "__paperclip_permission", inputSchema: { type: "object" } }],
handler,
});
bridges.push(bridge);
const listed = await (await rpc(bridge, { id: 1, method: "tools/list" })).json() as { result: { tools: Array<{ name: string }> } };
expect(listed.result.tools.map((tool) => tool.name)).not.toContain("__paperclip_permission");
expect(await (await rpc(bridge, {
id: "permission-1",
method: "tools/call",
params: { name: "__paperclip_permission", arguments: {} },
})).json()).toMatchObject({ result: { content: [{ text: "{\"decision\":\"accept\"}" }] } });
expect(handler).toHaveBeenCalledWith(expect.objectContaining({ tool: "__paperclip_permission" }));
});
it("maps prefixed names and replays identical duplicate calls exactly once", async () => {
const handler = vi.fn(async () => ({ value: 7 }));
const bridge = await startOpenCodeMcpBridge({
tools: [{ name: "documents.read", inputSchema: { type: "object" } }],
handler,
});
bridges.push(bridge);
const request = { id: "call-1", method: "tools/call", params: { name: "paperclip_documents.read", arguments: { id: "doc" } } };
expect(await (await rpc(bridge, request)).json()).toMatchObject({ result: { content: [{ text: "{\"value\":7}" }] } });
expect(await (await rpc(bridge, request)).json()).toMatchObject({ result: { content: [{ text: "{\"value\":7}" }] } });
expect(handler).toHaveBeenCalledTimes(1);
expect(handler).toHaveBeenCalledWith(expect.objectContaining({ tool: "documents.read", callId: "call-1", arguments: { id: "doc" } }));
});
it("fails conflicting duplicate identities and handler errors closed", async () => {
const bridge = await startOpenCodeMcpBridge({
tools: [{ name: "documents.read", inputSchema: { type: "object" } }],
handler: async () => { throw new Error("denied"); },
});
bridges.push(bridge);
const failed = await rpc(bridge, { id: "call-2", method: "tools/call", params: { name: "documents.read", arguments: { id: "a" } } });
expect(await failed.json()).toMatchObject({ result: { isError: true, content: [{ text: "denied" }] } });
const conflict = await rpc(bridge, { id: "call-2", method: "tools/call", params: { name: "documents.read", arguments: { id: "b" } } });
expect(await conflict.json()).toMatchObject({ result: { isError: true, content: [{ text: "Duplicate call identity conflict." }] } });
});
it("normalizes all supported OpenCode MCP prefixes", () => {
expect(canonicalOpenCodeMcpToolName("paperclip_documents.read")).toBe("documents.read");
expect(canonicalOpenCodeMcpToolName("paperclip__documents.read")).toBe("documents.read");
expect(canonicalOpenCodeMcpToolName("paperclip.documents.read")).toBe("documents.read");
expect(canonicalOpenCodeMcpToolName("paperclip_paperclip_finish")).toBe("paperclip_finish");
});
it("validates inputs before dispatch", async () => {
const handler = vi.fn(async () => ({ ok: true }));
const bridge = await startOpenCodeMcpBridge({
tools: [{
name: "documents.read",
inputSchema: { type: "object", required: ["id"], properties: { id: { type: "string" } }, additionalProperties: false },
}],
handler,
});
bridges.push(bridge);
const response = await rpc(bridge, {
id: "invalid",
method: "tools/call",
params: { name: "documents.read", arguments: { id: 9 } },
});
expect(await response.json()).toMatchObject({ result: { isError: true } });
expect(handler).not.toHaveBeenCalled();
});
it("times out and cancels in-flight calls", async () => {
const observed = vi.fn();
const bridge = await startOpenCodeMcpBridge({
tools: [{ name: "documents.read", inputSchema: { type: "object" } }],
timeoutMs: 20,
handler: ({ signal }) => new Promise((_resolve, reject) => {
signal.addEventListener("abort", () => { observed(); reject(new Error("aborted")); }, { once: true });
}),
});
bridges.push(bridge);
const response = await rpc(bridge, {
id: "slow-call",
method: "tools/call",
params: { name: "documents.read", arguments: {} },
});
expect(await response.json()).toMatchObject({ result: { isError: true, content: [{ text: "Paperclip tool call timed out" }] } });
expect(observed).toHaveBeenCalledOnce();
});
it("allows human-gated private calls to outlive ordinary tool timeouts", async () => {
const bridge = await startOpenCodeMcpBridge({
tools: [{ name: "documents.read", inputSchema: { type: "object" } }],
privateTools: [{ name: "__paperclip_runtime_input", inputSchema: { type: "object" } }],
timeoutMs: 20,
privateToolTimeoutMs: 200,
handler: async ({ tool }) => {
await new Promise((resolve) => setTimeout(resolve, 50));
return { tool, resolved: true };
},
});
bridges.push(bridge);
const response = await rpc(bridge, {
id: "human-input",
method: "tools/call",
params: { name: "__paperclip_runtime_input", arguments: {} },
});
expect(await response.json()).toMatchObject({
result: { content: [{ text: expect.stringContaining("\"resolved\":true") }] },
});
});
it("honors MCP cancellation notifications", async () => {
const started = vi.fn();
const bridge = await startOpenCodeMcpBridge({
tools: [{ name: "documents.read", inputSchema: { type: "object" } }],
handler: ({ signal }) => new Promise((_resolve, reject) => {
started();
signal.addEventListener("abort", () => reject(new Error("handler aborted")), { once: true });
}),
});
bridges.push(bridge);
const pending = rpc(bridge, {
id: "cancel-me",
method: "tools/call",
params: { name: "documents.read", arguments: {} },
});
await vi.waitFor(() => expect(started).toHaveBeenCalledOnce());
expect((await rpc(bridge, {
method: "notifications/cancelled",
params: { requestId: "cancel-me", reason: "turn stopped" },
})).status).toBe(202);
expect(await (await pending).json()).toMatchObject({
result: { isError: true, content: [{ text: "Paperclip tool call cancelled" }] },
});
});
});
@@ -0,0 +1,356 @@
import { randomBytes, timingSafeEqual } from "node:crypto";
import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
import { Ajv2020, type ValidateFunction } from "ajv/dist/2020.js";
import {
PRP_BLOCK_RESULT_PROVIDER_INPUT_SCHEMA,
PRP_BLOCK_TOOL_NAME,
PRP_COMPLETION_RESULT_PROVIDER_INPUT_SCHEMA,
PRP_COMPLETION_TOOL_NAME,
} from "../../contracts/completion-result.js";
export interface OpenCodeMcpToolDefinition {
name: string;
description?: string;
inputSchema: Record<string, unknown>;
}
export interface OpenCodeMcpCall {
tool: string;
callId: string;
arguments: unknown;
signal: AbortSignal;
}
export interface OpenCodeMcpBridgeOptions {
tools?: readonly Readonly<Record<string, unknown>>[];
/** Runner-owned operations callable only by a trusted harness extension and never advertised to the model. */
privateTools?: readonly Readonly<Record<string, unknown>>[];
handler: (call: OpenCodeMcpCall) => Promise<unknown>;
timeoutMs?: number;
/** Human-gated private operations may remain pending much longer than model tool calls. */
privateToolTimeoutMs?: number;
maxBodyBytes?: number;
secret?: string;
}
export interface OpenCodeMcpBridge {
url: string;
secret: string;
close(): Promise<void>;
}
const DEFAULT_TIMEOUT_MS = 30_000;
const DEFAULT_MAX_BODY_BYTES = 1_048_576;
export function canonicalOpenCodeMcpToolName(name: string): string {
const trimmed = name.trim();
if (trimmed === PRP_COMPLETION_TOOL_NAME || trimmed === PRP_BLOCK_TOOL_NAME) return trimmed;
for (const prefix of ["paperclip__", "paperclip_", "paperclip."]) {
if (trimmed.startsWith(prefix)) return trimmed.slice(prefix.length);
}
return trimmed;
}
export async function startOpenCodeMcpBridge(
options: OpenCodeMcpBridgeOptions,
): Promise<OpenCodeMcpBridge> {
const secret = options.secret ?? randomBytes(32).toString("base64url");
const tools = normalizeTools(options.tools ?? []);
const privateTools = normalizeTools(options.privateTools ?? [], false);
const admittedTools = [...tools, ...privateTools.filter((candidate) => !tools.some((tool) => tool.name === candidate.name))];
const validators = compileValidators(admittedTools);
const calls = new Map<string, { fingerprint: string; promise: Promise<unknown> }>();
const controllers = new Map<string, AbortController>();
const server = createServer((request, response) => {
void handleRequest(request, response, {
secret,
tools,
admittedTools,
validators,
calls,
controllers,
handler: options.handler,
timeoutMs: options.timeoutMs ?? DEFAULT_TIMEOUT_MS,
privateToolTimeoutMs: options.privateToolTimeoutMs ?? options.timeoutMs ?? DEFAULT_TIMEOUT_MS,
privateToolNames: new Set(privateTools.map((tool) => tool.name)),
maxBodyBytes: options.maxBodyBytes ?? DEFAULT_MAX_BODY_BYTES,
});
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => {
server.off("error", reject);
resolve();
});
});
const address = server.address();
if (address === null || typeof address === "string") {
await closeServer(server);
throw new Error("OpenCode MCP bridge failed to bind a loopback port");
}
return {
url: `http://127.0.0.1:${address.port}/mcp`,
secret,
close: async () => {
for (const controller of controllers.values()) controller.abort();
await closeServer(server);
},
};
}
function normalizeTools(
input: readonly Readonly<Record<string, unknown>>[],
includeTerminalTools = true,
): OpenCodeMcpToolDefinition[] {
const tools: OpenCodeMcpToolDefinition[] = [];
const seen = new Set<string>();
for (const raw of input) {
const name = typeof raw.name === "string" ? canonicalOpenCodeMcpToolName(raw.name) : "";
if (!name || seen.has(name)) continue;
const inputSchema = isRecord(raw.inputSchema) ? structuredClone(raw.inputSchema) : { type: "object" };
tools.push({
name,
...(typeof raw.description === "string" ? { description: raw.description } : {}),
inputSchema,
});
seen.add(name);
}
if (includeTerminalTools) for (const terminal of [
{ name: PRP_COMPLETION_TOOL_NAME, description: "Return the semantic completion result.", inputSchema: PRP_COMPLETION_RESULT_PROVIDER_INPUT_SCHEMA },
{ name: PRP_BLOCK_TOOL_NAME, description: "Return the semantic blocked result.", inputSchema: PRP_BLOCK_RESULT_PROVIDER_INPUT_SCHEMA },
]) {
if (!seen.has(terminal.name)) tools.push(terminal);
}
return tools;
}
function compileValidators(tools: readonly OpenCodeMcpToolDefinition[]): Map<string, ValidateFunction> {
const ajv = new Ajv2020({ allErrors: true, strict: false });
return new Map(tools.map((tool) => [tool.name, ajv.compile(tool.inputSchema)]));
}
async function handleRequest(
request: IncomingMessage,
response: ServerResponse,
context: {
secret: string;
tools: OpenCodeMcpToolDefinition[];
admittedTools: OpenCodeMcpToolDefinition[];
validators: Map<string, ValidateFunction>;
calls: Map<string, { fingerprint: string; promise: Promise<unknown> }>;
controllers: Map<string, AbortController>;
handler: OpenCodeMcpBridgeOptions["handler"];
timeoutMs: number;
privateToolTimeoutMs: number;
privateToolNames: ReadonlySet<string>;
maxBodyBytes: number;
},
): Promise<void> {
setSecurityHeaders(response);
if (!authorized(request.headers.authorization, context.secret)) {
response.statusCode = 401;
response.end();
return;
}
if (request.method === "GET") {
response.statusCode = 405;
response.setHeader("Allow", "POST, DELETE");
response.end();
return;
}
if (request.method === "DELETE") {
response.statusCode = 204;
response.end();
return;
}
if (request.method !== "POST" || request.url !== "/mcp") {
response.statusCode = 404;
response.end();
return;
}
let message: Record<string, unknown>;
try {
message = parseMessage(await readBody(request, context.maxBodyBytes));
} catch (error) {
writeRpc(response, null, undefined, rpcError(-32700, safeError(error)));
return;
}
const id = message.id ?? null;
const method = typeof message.method === "string" ? message.method : "";
if (method === "notifications/cancelled") {
const params = isRecord(message.params) ? message.params : {};
const requestId = params.requestId;
if (typeof requestId === "string" || typeof requestId === "number") {
context.controllers.get(String(requestId))?.abort();
}
response.statusCode = 202;
response.end();
return;
}
if (method === "notifications/initialized") {
response.statusCode = 202;
response.end();
return;
}
if (method === "initialize") {
response.setHeader("Mcp-Session-Id", randomBytes(16).toString("hex"));
writeRpc(response, id, {
protocolVersion: "2025-03-26",
capabilities: { tools: { listChanged: false } },
serverInfo: { name: "paperclip-runner", version: "1" },
});
return;
}
if (method === "ping") {
writeRpc(response, id, {});
return;
}
if (method === "tools/list") {
writeRpc(response, id, { tools: context.tools });
return;
}
if (method !== "tools/call") {
writeRpc(response, id, undefined, rpcError(-32601, "Method not found"));
return;
}
const params = isRecord(message.params) ? message.params : {};
const rawName = typeof params.name === "string" ? params.name : "";
const tool = canonicalOpenCodeMcpToolName(rawName);
if (!context.admittedTools.some((entry) => entry.name === tool)) {
writeRpc(response, id, { isError: true, content: [{ type: "text", text: "Unsupported tool." }] });
return;
}
const callId = typeof id === "string" || typeof id === "number" ? String(id) : randomBytes(12).toString("hex");
const args = params.arguments ?? {};
const validator = context.validators.get(tool);
if (validator === undefined || !validator(args)) {
const detail = validator?.errors?.map((error) => `${error.instancePath || "/"} ${error.message ?? "is invalid"}`).join("; ");
writeRpc(response, id, {
isError: true,
content: [{ type: "text", text: `Invalid tool input${detail ? `: ${detail}` : "."}` }],
});
return;
}
const fingerprint = canonicalJson({ tool, args });
const existing = context.calls.get(callId);
if (existing && existing.fingerprint !== fingerprint) {
writeRpc(response, id, { isError: true, content: [{ type: "text", text: "Duplicate call identity conflict." }] });
return;
}
const controller = existing === undefined ? new AbortController() : undefined;
const execution = existing?.promise ?? withCancellationAndTimeout(
context.handler({ tool, callId, arguments: structuredClone(args), signal: controller!.signal }),
controller!,
context.privateToolNames.has(tool) ? context.privateToolTimeoutMs : context.timeoutMs,
);
if (!existing) {
if (context.calls.size >= 2_048) context.calls.delete(context.calls.keys().next().value!);
context.calls.set(callId, { fingerprint, promise: execution });
context.controllers.set(callId, controller!);
void execution.finally(() => context.controllers.delete(callId)).catch(() => undefined);
}
try {
const result = await execution;
writeRpc(response, id, { content: [{ type: "text", text: safeJson(result) }] });
} catch (error) {
writeRpc(response, id, { isError: true, content: [{ type: "text", text: safeError(error) }] });
}
}
function authorized(value: string | undefined, secret: string): boolean {
if (!value?.startsWith("Bearer ")) return false;
const supplied = Buffer.from(value.slice(7));
const expected = Buffer.from(secret);
return supplied.length === expected.length && timingSafeEqual(supplied, expected);
}
function readBody(request: IncomingMessage, maxBytes: number): Promise<string> {
return new Promise((resolve, reject) => {
const chunks: Buffer[] = [];
let size = 0;
request.on("data", (chunk: Buffer) => {
size += chunk.length;
if (size > maxBytes) {
reject(new Error("MCP request exceeded the retained payload limit"));
request.destroy();
return;
}
chunks.push(chunk);
});
request.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
request.on("error", reject);
});
}
function parseMessage(body: string): Record<string, unknown> {
const parsed: unknown = JSON.parse(body);
if (!isRecord(parsed) || parsed.jsonrpc !== "2.0") throw new Error("Invalid JSON-RPC request");
return parsed;
}
function writeRpc(response: ServerResponse, id: unknown, result?: unknown, error?: unknown): void {
response.statusCode = 200;
response.setHeader("Content-Type", "application/json");
response.end(JSON.stringify({ jsonrpc: "2.0", id, ...(error === undefined ? { result } : { error }) }));
}
function rpcError(code: number, message: string) {
return { code, message };
}
function setSecurityHeaders(response: ServerResponse): void {
response.setHeader("Cache-Control", "no-store");
response.setHeader("X-Content-Type-Options", "nosniff");
}
function closeServer(server: Server): Promise<void> {
return new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
function withCancellationAndTimeout<T>(
promise: Promise<T>,
controller: AbortController,
timeoutMs: number,
): Promise<T> {
return new Promise((resolve, reject) => {
let settled = false;
const finish = (callback: () => void) => {
if (settled) return;
settled = true;
clearTimeout(timer);
controller.signal.removeEventListener("abort", onAbort);
callback();
};
const onAbort = () => finish(() => reject(new Error("Paperclip tool call cancelled")));
const timer = setTimeout(() => {
finish(() => reject(new Error("Paperclip tool call timed out")));
controller.abort();
}, timeoutMs);
timer.unref?.();
controller.signal.addEventListener("abort", onAbort, { once: true });
promise.then(
(value) => finish(() => resolve(value)),
(error) => finish(() => reject(error)),
);
});
}
function safeError(error: unknown): string {
return (error instanceof Error ? error.message : String(error)).slice(0, 2_000);
}
function safeJson(value: unknown): string {
const serialized = JSON.stringify(value);
return (serialized ?? "null").slice(0, 64 * 1024);
}
function canonicalJson(value: unknown): string {
if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`;
if (isRecord(value)) return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${canonicalJson(value[key])}`).join(",")}}`;
return JSON.stringify(value) ?? "undefined";
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
@@ -0,0 +1,920 @@
import { chmod, lstat, mkdir, mkdtemp, readFile, readdir, rm, stat, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import {
NATIVE_RUNTIME_ASSET_SCHEMA,
PAPERCLIP_EXECUTION_PROMPT,
PAPERCLIP_EXECUTION_PROMPT_REVISION,
canonicalNativeRuntimeContextDigest,
nativeRuntimePromptDigest,
type NativeRuntimeContextSnapshot,
} from "../../contracts/runtime-context.js";
import { OpenCodeServerDriver, openCodeServerDriverInternals } from "./opencode-server-driver.js";
const roots: string[] = [];
const fixture = resolve("test/fixtures/fake-opencode-server.mjs");
function runtimeContext(skillRoot: string, instructionRoot: string): NativeRuntimeContextSnapshot {
const digest = "0".repeat(64);
const value = {
prompt: {
revision: PAPERCLIP_EXECUTION_PROMPT_REVISION,
text: PAPERCLIP_EXECUTION_PROMPT,
digest: nativeRuntimePromptDigest(),
},
instructions: {
entryPath: "AGENTS.md",
bundle: {
schema: NATIVE_RUNTIME_ASSET_SCHEMA,
digest,
manifestDigest: digest,
rootPath: instructionRoot,
fileCount: 2,
totalBytes: 2,
},
},
skills: [{
key: "company/assigned",
runtimeName: "assigned",
versionId: "version-1",
bundle: {
schema: NATIVE_RUNTIME_ASSET_SCHEMA,
digest,
manifestDigest: digest,
rootPath: skillRoot,
fileCount: 2,
totalBytes: 2,
},
}],
mcp: { assignmentSetId: "assigned", digest, bindingId: "binding" },
} satisfies Omit<NativeRuntimeContextSnapshot, "aggregateDigest">;
return { ...value, aggregateDigest: canonicalNativeRuntimeContextDigest(value) };
}
async function makeWritable(root: string): Promise<void> {
const info = await lstat(root).catch(() => null);
if (!info) return;
if (info.isDirectory()) {
await chmod(root, 0o700);
for (const entry of await readdir(root)) await makeWritable(join(root, entry));
} else {
await chmod(root, 0o600);
}
}
afterEach(async () => {
await Promise.all(roots.splice(0).map(async (root) => {
await makeWritable(root);
await rm(root, { recursive: true, force: true });
}));
});
describe("OpenCodeServerDriver", () => {
it("advertises within-turn plans as unsupported", async () => {
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: "/tmp/paperclip-opencode-capabilities",
command: fixture,
});
const descriptor = await driver.descriptor();
expect(descriptor.capabilities.typedEventFamilies
.find((family) => family.family === "plan"))
.toMatchObject({ availability: "unsupported" });
});
it("starts an authenticated isolated server, creates a session, streams usage, aborts, and cleans up", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const tracePath = join(root, "provider-trace.ndjson");
const spawns: Array<{ pid: number; processGroupId: number | null }> = [];
const diagnostics: string[] = [];
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: {
PATH: process.env.PATH,
OPENROUTER_API_KEY: "test-openrouter-key",
PAPERCLIP_API_KEY: "must-not-leak",
UNRELATED_SECRET: "must-not-leak",
PAPERCLIP_PROVIDER_TRACE_PATH: tracePath,
PAPERCLIP_PROVIDER_TRACE_MAX_BYTES: String(64 * 1024 * 1024),
},
onSpawn: async (meta) => { spawns.push(meta); },
onDiagnostic: (message) => diagnostics.push(message),
});
const session = await driver.openSession({ runId: "run-1", normalizedSessionId: "normalized/1", workingDirectory: workspace });
expect(session.ids()).toMatchObject({ providerSessionId: "ses_fake_1" });
expect(spawns).toHaveLength(1);
const turn = await session.startTurn({ message: { role: "user", text: "finish" } });
const events = [];
for await (const event of session.events()) events.push(event);
expect(events.map((event) => event.eventType)).toContain("turn.completed");
expect(events).toContainEqual(expect.objectContaining({ eventType: "run.result.proposed" }));
expect(events.filter((event) => event.eventType === "item.delta")).toHaveLength(1);
expect(events.find((event) => event.eventType === "item.delta")?.payload.text).toBe("done [guide](guide.md)");
expect(events.find(
(event) => event.eventType === "item.completed" && event.payload.kind === "agentMessage",
)?.payload).toMatchObject({
channel: "final",
providerPhase: "final_answer",
text: "done [guide](guide.md)",
});
expect(events.filter(
(event) => event.eventType === "item.completed" && event.payload.kind === "usage",
)).toHaveLength(1);
expect(events.find((event) => event.eventType === "workspace.change.updated")?.payload)
.toMatchObject({ schema: "paperclip.workspace.diff.v1", source: "harness_reported", totals: { files: 1 } });
expect(events.find((event) => event.eventType === "workspace.diff.recorded")?.payload)
.toMatchObject({ schema: "paperclip.workspace.diff.v1", complete: true });
expect(events.find((event) => event.eventType === "workspace.file.referenced")?.payload)
.toMatchObject({ schema: "paperclip.workspace.file_reference.v1", path: "guide.md" });
expect(JSON.stringify(events)).not.toContain("submitted prompt must not become assistant text");
expect(await session.usage()).toMatchObject({
input: 3,
output: 2,
costUsd: 0.001,
provider: "openrouter",
driverVersion: "1.18.17",
});
await session.interrupt?.({ turnId: turn.turnId });
const snapshot = await session.snapshot();
expect(snapshot.driverKind).toBe("opencode_server");
const sessionRoot = join(root, "normalized_1");
expect((await stat(sessionRoot)).mode & 0o777).toBe(0o700);
expect((await stat(join(sessionRoot, "config", "opencode", "opencode.json"))).mode & 0o777).toBe(0o600);
const config = await readFile(join(sessionRoot, "config", "opencode", "opencode.json"), "utf8");
await session.close({ reason: "test" });
const recovered = await driver.recoverSession?.(snapshot);
expect(recovered).toMatchObject({ recovered: true });
await expect(recovered?.session?.read?.()).resolves.toMatchObject({ sessionId: "ses_fake_1", messages: [] });
await recovered?.session?.close({ reason: "recovery-test" });
const traceEntries = (await readFile(tracePath, "utf8"))
.trim()
.split("\n")
.map((line) => JSON.parse(line) as Record<string, unknown>);
const traceFrames = traceEntries.filter((entry) => entry.kind === "frame");
expect(traceFrames.map((entry) => entry.direction)).toEqual(
expect.arrayContaining(["client_to_provider", "provider_to_client", "provider_stderr"]),
);
expect(traceFrames.some((entry) => entry.nativeMethod === "SSE /event")).toBe(true);
expect(traceEntries.some(
(entry) => entry.stage === "typescript_opencode_driver_normalization"
&& Array.isArray(entry.emittedEventIds)
&& entry.emittedEventIds.length > 0,
)).toBe(true);
expect(traceEntries.at(-1)).toMatchObject({
kind: "trace_status",
debugChannel: "typescript_opencode_native",
status: "complete",
});
const environment = JSON.parse(await readFile(join(sessionRoot, "data", "fake-environment.json"), "utf8"));
const mcpEvidence = JSON.parse(await readFile(join(sessionRoot, "data", "fake-mcp-evidence.json"), "utf8"));
expect(environment.keys).toContain("OPENROUTER_API_KEY");
expect(environment.keys).not.toContain("PAPERCLIP_API_KEY");
expect(environment.keys).not.toContain("UNRELATED_SECRET");
expect(environment.keys).not.toContain("PAPERCLIP_PROVIDER_TRACE_PATH");
expect(environment.projectConfigDisabled).toBe("true");
expect(mcpEvidence.tools).toEqual(expect.arrayContaining(["paperclip_finish", "paperclip_block"]));
expect(config).toContain("openrouter/deepseek/deepseek-v4-flash-0731");
expect(config).toContain('"*": "allow"');
expect(config).toContain('"external_directory": "deny"');
expect(events.some((event) => event.eventType === "runtime_request.created")).toBe(false);
expect(config).not.toContain("test-openrouter-key");
expect(diagnostics.join("\n")).not.toContain("test-openrouter-key");
expect(diagnostics.join("\n")).toContain("[REDACTED]");
});
it("maps OpenCode's normal abort error to cancellation without a false provider failure notice", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-abort-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-abort-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-aborted",
normalizedSessionId: "aborted",
workingDirectory: workspace,
});
await session.startTurn({ message: { role: "user", text: "session-aborted" } });
const events = [];
for await (const event of session.events()) events.push(event);
expect(events.map((event) => event.eventType)).toContain("turn.cancelled");
expect(events.map((event) => event.eventType)).not.toContain("turn.failed");
expect(events.map((event) => event.eventType)).not.toContain("provider.notice.recorded");
await session.close({ reason: "test" });
});
it("keeps OpenCode in an outer supervisor process group when requested", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
let processGroupId: number | null | undefined;
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
isolateProcessGroup: false,
onSpawn: async (meta) => { processGroupId = meta.processGroupId; },
});
const session = await driver.openSession({ runId: "run-supervised", normalizedSessionId: "supervised", workingDirectory: workspace });
expect(processGroupId).toBeNull();
await session.close({ reason: "test" });
});
it("uses the native system channel, selected skill tree, instruction root, and assigned MCP gateway", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-context-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-context-workspace-"));
const skillRoot = join(root, "skill-source");
const instructionRoot = join(root, "instruction-source");
roots.push(root, workspace);
await Promise.all([
mkdir(join(skillRoot, "references"), { recursive: true }),
mkdir(instructionRoot, { recursive: true }),
]);
await Promise.all([
writeFile(join(skillRoot, "SKILL.md"), "# Assigned\nRead references/support.md\n"),
writeFile(join(skillRoot, "references", "support.md"), "skill support\n"),
writeFile(join(instructionRoot, "AGENTS.md"), "Read sibling.md\n"),
writeFile(join(instructionRoot, "sibling.md"), "instruction sibling\n"),
]);
const systemInstructions = `${PAPERCLIP_EXECUTION_PROMPT}\n\nRead sibling.md\n\nRead-only instruction sibling root: ${instructionRoot}`;
let submittedPrompt: Record<string, unknown> | null = null;
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
systemInstructions,
runtimeContext: runtimeContext(skillRoot, instructionRoot),
environment: {
PATH: process.env.PATH,
OPENROUTER_API_KEY: "fixture-key",
PAPERCLIP_NATIVE_MCP_NAME: "paperclip-assigned",
PAPERCLIP_NATIVE_MCP_URL: "https://paperclip.example/mcp",
PAPERCLIP_NATIVE_MCP_TOKEN: "x".repeat(40),
},
fetch: async (input, init) => {
if (String(input).endsWith("/prompt_async")) submittedPrompt = JSON.parse(String(init?.body));
return fetch(input, init);
},
});
const session = await driver.openSession({
runId: "run-context",
normalizedSessionId: "context",
workingDirectory: workspace,
});
await session.startTurn({ message: { role: "user", text: "finish" } });
for await (const event of session.events()) {
if (event.eventType === "turn.completed") break;
}
expect(submittedPrompt).toMatchObject({
system: systemInstructions,
tools: { question: true },
});
expect(JSON.stringify(submittedPrompt?.parts ?? null)).not.toContain(PAPERCLIP_EXECUTION_PROMPT);
const sessionRoot = join(root, "context");
const isolatedHomes = (await readdir(sessionRoot)).filter((entry) => entry.startsWith("home-"));
expect(isolatedHomes).toHaveLength(1);
await expect(readFile(join(sessionRoot, isolatedHomes[0]!, ".claude", "skills", "assigned", "references", "support.md"), "utf8"))
.resolves.toBe("skill support\n");
const config = JSON.parse(await readFile(join(sessionRoot, "config", "opencode", "opencode.json"), "utf8"));
expect(config).toMatchObject({
instructions: [],
plugin: [],
tools: { question: true },
permission: { external_directory: { "*": "deny", [`${instructionRoot}/**`]: "allow" } },
mcp: {
paperclip: { enabled: true },
"paperclip-assigned": { url: "https://paperclip.example/mcp", enabled: true },
},
});
await expect(readFile(join(instructionRoot, "sibling.md"), "utf8")).resolves.toBe("instruction sibling\n");
await session.close({ reason: "test" });
});
it("sends only the authoritative wake envelope when resuming an existing provider session", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-resume-context-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-resume-workspace-"));
roots.push(root, workspace);
let submittedPrompt: Record<string, unknown> | null = null;
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
systemInstructions: "large original system context",
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
fetch: async (input, init) => {
if (String(input).endsWith("/prompt_async")) submittedPrompt = JSON.parse(String(init?.body));
return fetch(input, init);
},
});
const seed = await driver.openSession({
runId: "run-seed",
normalizedSessionId: "resume-context",
workingDirectory: workspace,
});
const snapshot = await seed.snapshot();
await seed.close({ reason: "seed complete" });
const recovered = await driver.recoverSession?.(snapshot);
expect(recovered).toMatchObject({ recovered: true });
await recovered!.session!.startTurn({
message: { role: "user", text: "{\"interactionResponses\":[{\"answer\":\"friendly\"}]}" },
});
for await (const event of recovered!.session!.events()) {
if (event.eventType === "turn.completed") break;
}
expect(submittedPrompt).toMatchObject({
providerID: "openrouter",
modelID: "deepseek/deepseek-v4-flash-0731",
tools: { question: true },
parts: [{ type: "text", text: "{\"interactionResponses\":[{\"answer\":\"friendly\"}]}" }],
});
expect(submittedPrompt).not.toHaveProperty("system");
await recovered!.session!.close({ reason: "recovery-test" });
});
it("normalizes native question.asked events and replies with ordered OpenCode answer arrays", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-native-question",
normalizedSessionId: "native-question",
workingDirectory: workspace,
});
const { turnId } = await session.startTurn({ message: { role: "user", text: "native-question" } });
const iterator = session.events()[Symbol.asyncIterator]();
let requestEvent: Awaited<ReturnType<typeof iterator.next>>["value"] | null = null;
for (let count = 0; count < 30; count += 1) {
const event = await iterator.next();
if (event.done) break;
if (event.value.eventType === "runtime_request.created") {
requestEvent = event.value;
break;
}
}
expect(requestEvent?.payload).toMatchObject({
request: {
schema: "paperclip.runtime_request.v2",
requestKind: "runtime",
type: "input",
input: {
schema: "paperclip.question_set.v1",
questions: [
{ id: "environment", answerMode: "single_select", required: false, customAnswer: { enabled: true } },
{ id: "regions", answerMode: "multi_select", required: false },
],
},
},
});
await new Promise((resolveDelay) => setTimeout(resolveDelay, 20));
expect(session.pendingRuntimeRequests?.()).toHaveLength(1);
await session.resolveRuntimeRequest?.({
requestId: "question-native-1",
turnId,
resolution: {
action: "submit",
response: {
schema: "paperclip.question_response.v1",
answers: {
environment: { customText: "Canary" },
regions: { selectedOptionIds: ["option-1", "option-2"] },
},
},
},
});
const reply = JSON.parse(await readFile(join(root, "native-question", "data", "fake-question-reply.json"), "utf8"));
expect(reply).toEqual({
url: expect.stringContaining(`directory=${encodeURIComponent(workspace)}`),
body: { answers: [["Canary"], ["US", "EU"]] },
});
const terminalQuestionEvents = (await session.transcript?.())?.events.filter((event) =>
event.payload.requestId === "question-native-1"
&& ["runtime_request.resolved", "runtime_request.cancelled"].includes(event.eventType)
) ?? [];
expect(terminalQuestionEvents).toHaveLength(1);
expect(terminalQuestionEvents[0]?.payload).toMatchObject({
action: "submit",
response: {
schema: "paperclip.question_response.v1",
answers: {
environment: { customText: "Canary" },
regions: { selectedOptionIds: ["option-1", "option-2"] },
},
},
});
await session.close({ reason: "test" });
});
it("hands a native question to the durable interaction path without touching permissions", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-handoff-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-handoff-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-native-question-handoff",
normalizedSessionId: "native-question-handoff",
workingDirectory: workspace,
});
const { turnId } = await session.startTurn({ message: { role: "user", text: "native-question" } });
const iterator = session.events()[Symbol.asyncIterator]();
let created: Awaited<ReturnType<typeof iterator.next>>["value"] | null = null;
for (let count = 0; count < 30; count += 1) {
const next = await iterator.next();
if (next.done) break;
if (next.value.eventType === "runtime_request.created") {
created = next.value;
break;
}
}
expect(created).toMatchObject({
payload: { request: { requestId: "question-native-1", type: "input" } },
});
const firstHandoff = session.handoffRuntimeRequest?.({
requestId: "question-native-1",
turnId,
reason: "durable_handoff",
signal: new AbortController().signal,
});
expect(firstHandoff?.result).toBe("handed_off");
await firstHandoff?.cleanup;
const repeatedHandoff = session.handoffRuntimeRequest?.({
requestId: "question-native-1",
turnId,
reason: "durable_handoff",
signal: new AbortController().signal,
});
expect(repeatedHandoff?.result).toBe("already_settled");
await repeatedHandoff?.cleanup;
let expired: Awaited<ReturnType<typeof iterator.next>>["value"] | null = null;
for (let count = 0; count < 20; count += 1) {
const next = await iterator.next();
if (next.done) break;
if (next.value.eventType === "runtime_request.expired") {
expired = next.value;
break;
}
}
expect(expired).toMatchObject({
payload: {
requestId: "question-native-1",
reason: "durable_handoff",
replayAllowed: false,
requestType: "input",
request: {
schema: "paperclip.runtime_request.v2",
requestId: "question-native-1",
type: "input",
},
},
});
expect(session.pendingRuntimeRequests?.()).toEqual([]);
const terminal = (await session.transcript?.())?.events.filter((event) =>
event.payload.requestId === "question-native-1"
&& ["runtime_request.resolved", "runtime_request.cancelled", "runtime_request.expired"].includes(event.eventType)
);
expect(terminal).toHaveLength(1);
await session.close({ reason: "handoff test complete" });
});
it("recovers a missed pending question from the list endpoint and rejects it", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-recovery-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-recovery-workspace-"));
roots.push(root, workspace);
const normalizedSessionId = "question-recovery";
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const seed = await driver.openSession({
runId: "run-question-recovery",
normalizedSessionId,
workingDirectory: workspace,
});
const snapshot = await seed.snapshot();
await seed.close({ reason: "simulate reconnect" });
const dataRoot = join(root, normalizedSessionId, "data");
await mkdir(dataRoot, { recursive: true });
await writeFile(join(dataRoot, "fake-pending-question.json"), JSON.stringify({
id: "question-native-1",
sessionID: "ses_fake_1",
questions: [{
id: "environment",
header: "Environment",
question: "Where should we deploy?",
options: [{ label: "Staging" }, { label: "Production" }],
custom: true,
}],
}));
const recovered = await driver.recoverSession?.({
...snapshot,
activeTurnId: "turn-recovered-question",
});
expect(recovered).toMatchObject({ recovered: true });
const session = recovered!.session!;
const iterator = session.events()[Symbol.asyncIterator]();
let created: Awaited<ReturnType<typeof iterator.next>>["value"] | null = null;
for (let count = 0; count < 20; count += 1) {
const next = await iterator.next();
if (next.done) break;
if (next.value.eventType === "runtime_request.created") {
created = next.value;
break;
}
}
expect(created).toMatchObject({
payload: { request: { requestId: "question-native-1", type: "input" } },
});
await session.resolveRuntimeRequest?.({
requestId: "question-native-1",
turnId: "turn-recovered-question",
resolution: { action: "decline" },
});
expect(session.pendingRuntimeRequests?.()).toEqual([]);
await session.close({ reason: "recovery test complete" });
});
it.each([
{ style: "legacy", action: "accept" as const, reply: "once" },
{ style: "v2", action: "accept_for_session" as const, reply: "always" },
{ style: "v2", action: "decline" as const, reply: "reject" },
])("normalizes $style permission events and maps $action to $reply", async ({ style, action, reply }) => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), `paperclip-opencode-permission-${style}-`));
const workspace = await mkdtemp(join(tmpdir(), `paperclip-opencode-permission-${style}-workspace-`));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
permissionMode: "ask",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const normalizedSessionId = `native-permission-${style}-${action}`;
const session = await driver.openSession({
runId: `run-${normalizedSessionId}`,
normalizedSessionId,
workingDirectory: workspace,
});
const { turnId } = await session.startTurn({
message: { role: "user", text: `native-permission-${style}` },
});
const iterator = session.events()[Symbol.asyncIterator]();
let created: Awaited<ReturnType<typeof iterator.next>>["value"] | null = null;
for (let count = 0; count < 30; count += 1) {
const next = await iterator.next();
if (next.done) break;
if (next.value.eventType === "runtime_request.created") {
created = next.value;
break;
}
}
expect(created).toMatchObject({
payload: {
request: {
requestId: "permission-native-1",
type: "permission",
actions: ["accept", "accept_for_session", "decline", "cancel"],
},
},
});
await session.resolveRuntimeRequest?.({
requestId: "permission-native-1",
turnId,
resolution: { action },
});
const persisted = JSON.parse(
await readFile(join(root, normalizedSessionId, "data", "fake-permission-reply.json"), "utf8"),
);
expect(persisted).toEqual({
url: expect.stringContaining(`directory=${encodeURIComponent(workspace)}`),
body: { reply },
});
expect(session.pendingRuntimeRequests?.()).toEqual([]);
await session.close({ reason: "permission test complete" });
});
it.each(["allow", "ask", "deny"] as const)("pins the %s permission mode while retaining external-directory denial", async (permissionMode) => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), `paperclip-opencode-mode-${permissionMode}-`));
const workspace = await mkdtemp(join(tmpdir(), `paperclip-opencode-mode-${permissionMode}-workspace-`));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
permissionMode,
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({ runId: `run-${permissionMode}`, normalizedSessionId: permissionMode, workingDirectory: workspace });
const config = JSON.parse(await readFile(join(root, permissionMode, "config", "opencode", "opencode.json"), "utf8"));
expect(config.permission).toMatchObject({ "*": permissionMode, external_directory: "deny" });
expect(config.permission.paperclip_finish).toBeUndefined();
expect(config.permission["paperclip_*"]).toBe("allow");
await session.close({ reason: "permission mode test complete" });
});
it("clears a stale active turn that already has a persisted terminal fingerprint", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-stale-turn-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-stale-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const seed = await driver.openSession({
runId: "run-stale-turn",
normalizedSessionId: "stale-turn",
workingDirectory: workspace,
});
const snapshot = await seed.snapshot();
await seed.close({ reason: "seed complete" });
const recoveryDriver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const recovery = await recoveryDriver.recoverSession?.({
...snapshot,
activeTurnId: "turn-already-terminal",
terminalTurns: [{ turnId: "turn-already-terminal", fingerprint: "terminal-fingerprint" }],
});
expect(recovery).toMatchObject({ recovered: true });
await expect(recovery!.session!.snapshot()).resolves.toMatchObject({ activeTurnId: null });
await recovery!.session!.close({ reason: "recovery-test" });
});
it("validates provider/model form", async () => {
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
roots.push(root);
const driver = new OpenCodeServerDriver({ model: "bad", runtimeDirectory: root });
await expect(driver.validateConfig?.({ model: "bad" })).resolves.toMatchObject({ ok: false });
await expect(driver.validateConfig?.({ model: "openrouter/model" })).resolves.toMatchObject({ ok: true });
});
it("normalizes a structured block result", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({ runId: "run-block", normalizedSessionId: "block", workingDirectory: workspace });
await session.startTurn({ message: { role: "user", text: "block-result" } });
const events = [];
for await (const event of session.events()) events.push(event);
expect(events.find((event) => event.eventType === "run.result.proposed")?.payload)
.toMatchObject({ reportedWorkDisposition: "blocked", blocker: { reasonCode: "test_dependency" } });
await session.close({ reason: "test" });
});
it("designates the substantive pre-finish response instead of a trailing acknowledgement", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-final-selection",
normalizedSessionId: "final-selection",
workingDirectory: workspace,
});
await session.startTurn({ message: { role: "user", text: "text-before-finish" } });
const events = [];
for await (const event of session.events()) events.push(event);
const finalMessages = events.filter(
(event) => event.eventType === "item.completed"
&& event.payload.kind === "agentMessage"
&& event.payload.channel === "final",
);
expect(finalMessages).toHaveLength(1);
expect(finalMessages[0]?.payload.text).toBe("Substantive answer before finish.");
expect(events.filter((event) => event.eventType === "item.delta").map((event) => event.payload.text))
.toEqual(expect.arrayContaining(["Substantive answer before finish.", "Done."]));
await session.close({ reason: "test" });
});
it("does not promote opening commentary to the final response when work follows it", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-commentary-only",
normalizedSessionId: "commentary-only",
workingDirectory: workspace,
});
await session.startTurn({ message: { role: "user", text: "commentary-only-before-work" } });
const events = [];
for await (const event of session.events()) events.push(event);
expect(events.filter(
(event) => event.eventType === "item.completed"
&& event.payload.kind === "agentMessage"
&& event.payload.channel === "final",
)).toHaveLength(0);
expect(events.filter((event) => event.eventType === "item.delta").map((event) => event.payload.text))
.toContain("I will inspect the workspace first.");
expect(events).toContainEqual(expect.objectContaining({ eventType: "run.result.proposed" }));
await session.close({ reason: "test" });
});
it("correlates the final response to the native message containing the semantic tool", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-correlated-final-selection",
normalizedSessionId: "correlated-final-selection",
workingDirectory: workspace,
});
await session.startTurn({ message: { role: "user", text: "correlated-final-message" } });
const events = [];
for await (const event of session.events()) events.push(event);
const finalMessages = events.filter(
(event) => event.eventType === "item.completed"
&& event.payload.kind === "agentMessage"
&& event.payload.channel === "final",
);
expect(finalMessages).toHaveLength(1);
expect(finalMessages[0]?.payload.text).toBe("Correlated substantive final answer.");
expect(events.filter((event) => event.eventType === "item.delta").map((event) => event.payload.text))
.toEqual(expect.arrayContaining([
"I will write the result now.",
"Correlated substantive final answer.",
"Done.",
]));
await session.close({ reason: "test" });
});
it("prefers a substantive post-tool answer over schema commentary in the tool message", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
});
const session = await driver.openSession({
runId: "run-post-tool-final-selection",
normalizedSessionId: "post-tool-final-selection",
workingDirectory: workspace,
});
await session.startTurn({ message: { role: "user", text: "final-after-tool-commentary" } });
const events = [];
for await (const event of session.events()) events.push(event);
const finalMessages = events.filter(
(event) => event.eventType === "item.completed"
&& event.payload.kind === "agentMessage"
&& event.payload.channel === "final",
);
expect(finalMessages).toHaveLength(1);
expect(finalMessages[0]?.payload.text)
.toBe("This is the complete substantive answer emitted after the accepted completion tool call.");
await session.close({ reason: "test" });
});
it("rejects malformed and oversized SSE frames", async () => {
const stream = (value: string) => new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(new TextEncoder().encode(value));
controller.close();
},
});
const consume = async (value: string) => {
for await (const _event of openCodeServerDriverInternals.parseSse(stream(value))) {
// Consume the complete stream so parse failures are observed.
}
};
await expect(consume("data: {bad json}\n\n")).rejects.toThrow();
await expect(consume(`data: ${"x".repeat(1_048_577)}`)).rejects.toThrow("payload limit");
});
it("restarts OpenCode when session startup fails transiently", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
let failSessionCreate = true;
const spawns: number[] = [];
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: fixture,
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
fetch: async (input, init) => {
if (failSessionCreate && String(input).endsWith("/session") && init?.method === "POST") {
failSessionCreate = false;
throw new Error("transient session initialization failure");
}
return fetch(input, init);
},
onSpawn: async ({ pid }) => { spawns.push(pid); },
});
const session = await driver.openSession({ runId: "run-retry", normalizedSessionId: "retry", workingDirectory: workspace });
expect(session.ids().providerSessionId).toBe("ses_fake_1");
expect(spawns).toHaveLength(2);
await session.close({ reason: "test" });
});
it("reports startup exit details with stderr captured after spawn and redacted", async () => {
await chmod(fixture, 0o755);
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-driver-"));
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-workspace-"));
roots.push(root, workspace);
const exitingFixture = join(root, "exit-before-health.mjs");
await writeFile(
exitingFixture,
"#!/usr/bin/env node\nprocess.stderr.write(`credential=${process.env.OPENROUTER_API_KEY}\\nauthorization=super-secret-opencode-token\\n`);\nprocess.exit(17);\n",
{ mode: 0o755 },
);
const driver = new OpenCodeServerDriver({
model: "openrouter/deepseek/deepseek-v4-flash-0731",
runtimeDirectory: root,
command: exitingFixture,
environment: {
PATH: process.env.PATH,
OPENROUTER_API_KEY: "fixture-key",
},
});
const error = await driver.openSession({
runId: "run-startup-exit",
normalizedSessionId: "startup-exit",
workingDirectory: workspace,
}).then(
() => "provider unexpectedly started",
(cause: unknown) => String(cause),
);
expect(error).toContain("provider_process_exited");
expect(error).toContain("provider=opencode");
expect(error).toContain("stage=health");
expect(error).toContain("[REDACTED]");
expect(error).not.toContain("fixture-key");
expect(error).not.toContain("super-secret-opencode-token");
});
});
File diff suppressed because it is too large. Load diff
+5
View File
@@ -11,10 +11,15 @@ export * from "./contracts/question-set.js";
export * from "./contracts/runtime-context.js";
export * from "./contracts/types.js";
export * from "./backends/harness-driver-backend.js";
export { createOpenCodeNativeSessionBackend } from "./backends/opencode-native-backend.js";
export {
createNativeSessionBackend,
type NativeBackendFactoryOptions,
} from "./backends/native-backend-factory.js";
export {
OpenCodeServerDriver,
type OpenCodeServerDriverOptions,
} from "./drivers/opencode/opencode-server-driver.js";
export * from "./native-session-runtime.js";
export {
DurablePrpControlPlane,
@@ -37,9 +37,11 @@ test("the runner pins only the Codex ACPX production dependencies", () => {
);
});
test("the package exposes only the reviewed Codex ACPX sidecar binary", () => {
test("the package exposes only the reviewed provider transport binaries", () => {
assert.deepEqual(runnerPackage.bin, {
"paperclip-runner-acpx-sidecar": "./dist/cli/acpx-runtime-sidecar.js",
"paperclip-runner-opencode-proxy":
"./dist/cli/opencode-app-server-proxy.js",
});
});
@@ -0,0 +1,307 @@
#!/usr/bin/env node
import { createServer } from "node:http";
import { mkdir, readFile, writeFile } from "node:fs/promises";
import { join } from "node:path";
const args = process.argv.slice(2);
if (process.env.OPENROUTER_API_KEY) process.stderr.write(`fixture credential=${process.env.OPENROUTER_API_KEY}\n`);
const port = Number(args[args.indexOf("--port") + 1]);
const username = process.env.OPENCODE_SERVER_USERNAME ?? "opencode";
const password = process.env.OPENCODE_SERVER_PASSWORD ?? "";
const expectedAuth = `Basic ${Buffer.from(`${username}:${password}`).toString("base64")}`;
const session = { id: "ses_fake_1", title: "Paperclip fake" };
const clients = new Set();
let eventConnections = 0;
const runtimeConfig = JSON.parse(await readFile(join(process.env.XDG_CONFIG_HOME, "opencode", "opencode.json"), "utf8"));
const mcp = runtimeConfig.mcp?.paperclip;
let mcpRequestId = 1;
const mcpEvidence = { tools: [], calls: [] };
function nativeQuestion() {
return {
id: "question-native-1",
sessionID: session.id,
questions: [
{
id: "environment",
header: "Environment",
question: "Where should we deploy?",
options: [
{ label: "Staging", description: "Deploy to staging first." },
{ label: "Production", description: "Deploy directly to production." }
],
custom: true
},
{
id: "regions",
header: "Regions",
question: "Which regions?",
options: [{ label: "US" }, { label: "EU" }],
multiple: true
}
]
};
}
function nativePermission(style) {
return style === "legacy"
? {
id: "permission-native-1",
type: "tool",
pattern: "echo OK",
sessionID: session.id,
messageID: "message-permission",
callID: "call-permission",
title: "Run validation command",
metadata: {},
time: { created: Date.now() },
}
: {
id: "permission-native-1",
sessionID: session.id,
permission: "bash",
patterns: ["echo OK"],
always: ["echo *"],
metadata: {},
tool: { messageID: "message-permission", callID: "call-permission" },
};
}
let pendingQuestion = await readFile(join(process.env.XDG_DATA_HOME, "fake-pending-question.json"), "utf8")
.then((value) => JSON.parse(value))
.catch(() => null);
let pendingPermission = await readFile(join(process.env.XDG_DATA_HOME, "fake-pending-permission.json"), "utf8")
.then((value) => JSON.parse(value))
.catch(() => null);
let permissionStyle = "v2";
await mkdir(process.env.XDG_DATA_HOME, { recursive: true });
await writeFile(join(process.env.XDG_DATA_HOME, "fake-environment.json"), JSON.stringify({
keys: Object.keys(process.env).sort(),
home: process.env.HOME,
configHome: process.env.XDG_CONFIG_HOME,
projectConfigDisabled: process.env.OPENCODE_DISABLE_PROJECT_CONFIG,
}));
function json(response, status, value) {
response.statusCode = status;
response.setHeader("Content-Type", "application/json");
response.end(JSON.stringify(value));
}
function emit(value) {
const frame = `data: ${JSON.stringify(value)}\n\n`;
for (const response of clients) response.write(frame);
}
async function mcpRequest(method, params) {
const response = await fetch(mcp.url, {
method: "POST",
headers: { ...mcp.headers, "Content-Type": "application/json", Accept: "application/json, text/event-stream" },
body: JSON.stringify({ jsonrpc: "2.0", id: mcpRequestId++, method, ...(params ? { params } : {}) }),
});
return response.json();
}
async function callFirstPaperclipTool() {
if (!mcp?.url) return;
await mcpRequest("initialize", { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "fake-opencode", version: "1" } });
const listed = await mcpRequest("tools/list");
mcpEvidence.tools = (listed.result?.tools ?? []).map((entry) => entry.name);
const tool = listed.result?.tools?.find((entry) =>
!["paperclip_finish", "paperclip_block"].includes(entry.name)
&& (entry.inputSchema?.required?.length ?? 0) === 0
);
if (tool) mcpEvidence.calls.push(await mcpRequest("tools/call", { name: `paperclip_${tool.name}`, arguments: {} }));
await writeFile(join(process.env.XDG_DATA_HOME, "fake-mcp-evidence.json"), JSON.stringify(mcpEvidence));
}
async function callTerminalTool(promptBody) {
const prompt = JSON.parse(promptBody.parts?.[0]?.text ?? "{}");
const blocked = String(prompt.message ?? "").includes("block-result");
const result = {
schema: "paperclip.run_result.v1",
reportedWorkDisposition: blocked ? "blocked" : "done",
summary: blocked ? "Fake provider is blocked." : "Fake provider completed the task.",
completionClaim: {
contractRevision: prompt.task?.completionContract?.revision ?? "codex-demo-v1",
objectiveSatisfied: !blocked,
criteria: (prompt.task?.completionContract?.criteria ?? []).map((criterion) => ({
criterionId: criterion.id,
status: blocked ? "unknown" : "satisfied",
evidenceRefs: [],
})),
remainingWork: blocked ? [{ description: "Waiting on a test dependency.", blocksCompletion: true }] : [],
},
evidence: [],
verification: [],
attentionRequests: blocked ? [{ kind: "blocker", summary: "Waiting on a test dependency." }] : [],
artifacts: [],
...(blocked ? { blocker: {
reasonCode: "test_dependency",
owner: { kind: "external", name: "fixture" },
unblockAction: "Release the fixture dependency.",
scope: "current_track",
} } : {}),
};
return mcpRequest("tools/call", {
name: blocked ? "paperclip_block" : "paperclip_finish",
arguments: result,
});
}
const server = createServer(async (request, response) => {
if (request.headers.authorization !== expectedAuth) return json(response, 401, { error: "unauthorized" });
if (request.url === "/global/health") return json(response, 200, { healthy: true, version: "1.18.17" });
if (request.url === "/event") {
eventConnections += 1;
response.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache" });
response.write(`data: ${JSON.stringify({ type: "server.connected", properties: {} })}\n\n`);
if (eventConnections === 1) {
response.end();
return;
}
clients.add(response);
request.on("close", () => clients.delete(response));
return;
}
if (request.method === "POST" && request.url === "/session") return json(response, 200, session);
if (request.method === "GET" && request.url === `/session/${session.id}`) return json(response, 200, session);
if (request.method === "GET" && request.url === `/session/${session.id}/message`) return json(response, 200, []);
if (request.method === "GET" && request.url === "/session/status") return json(response, 200, { [session.id]: { type: "idle" } });
if (request.method === "GET" && request.url?.startsWith("/question?")) return json(response, 200, pendingQuestion ? [pendingQuestion] : []);
if (request.method === "GET" && request.url?.startsWith("/permission?")) return json(response, 200, pendingPermission ? [pendingPermission] : []);
if (request.method === "POST" && request.url?.startsWith("/question/question-native-1/reply?")) {
const chunks = [];
request.on("data", (chunk) => chunks.push(chunk));
request.on("end", async () => {
const body = JSON.parse(Buffer.concat(chunks).toString("utf8"));
await writeFile(join(process.env.XDG_DATA_HOME, "fake-question-reply.json"), JSON.stringify({ url: request.url, body }));
const repliedQuestion = pendingQuestion;
pendingQuestion = null;
emit({ type: "question.replied", id: "event-question-replied", properties: repliedQuestion });
setTimeout(() => {
json(response, 200, true);
emit({ type: "session.idle", id: "event-question-idle", properties: { sessionID: session.id } });
}, 10);
});
return;
}
if (request.method === "POST" && request.url?.startsWith("/question/question-native-1/reject?")) {
const rejectedQuestion = pendingQuestion;
pendingQuestion = null;
emit({ type: "question.rejected", id: "event-question-rejected", properties: rejectedQuestion });
setTimeout(() => {
json(response, 200, true);
emit({ type: "session.idle", id: "event-question-rejected-idle", properties: { sessionID: session.id } });
}, 10);
return;
}
if (request.method === "POST" && request.url?.startsWith("/permission/permission-native-1/reply?")) {
const chunks = [];
request.on("data", (chunk) => chunks.push(chunk));
request.on("end", async () => {
const body = JSON.parse(Buffer.concat(chunks).toString("utf8"));
await writeFile(join(process.env.XDG_DATA_HOME, "fake-permission-reply.json"), JSON.stringify({ url: request.url, body }));
pendingPermission = null;
emit({
type: permissionStyle === "legacy" ? "permission.replied" : "permission.v2.replied",
id: "event-permission-replied",
properties: permissionStyle === "legacy"
? { sessionID: session.id, permissionID: "permission-native-1", response: body.reply }
: { sessionID: session.id, requestID: "permission-native-1", reply: body.reply },
});
setTimeout(() => {
json(response, 200, true);
emit({ type: "session.idle", id: "event-permission-idle", properties: { sessionID: session.id } });
}, 10);
});
return;
}
if (request.method === "POST" && request.url === `/session/${session.id}/abort`) return json(response, 200, true);
if (request.method === "POST" && request.url === `/session/${session.id}/prompt_async`) {
const chunks = [];
request.on("data", (chunk) => chunks.push(chunk));
request.on("end", () => {
const promptPayload = JSON.parse(Buffer.concat(chunks).toString("utf8"));
if (promptPayload.providerID !== "openrouter" || promptPayload.modelID !== "deepseek/deepseek-v4-flash-0731" || "model" in promptPayload) {
return json(response, 400, { error: "OpenCode 1.18 prompt model fields must be top-level" });
}
json(response, 204, null);
setTimeout(async () => {
await callFirstPaperclipTool();
const parsedPrompt = JSON.parse(promptPayload.parts?.[0]?.text ?? "{}");
if (String(parsedPrompt.message ?? "").includes("native-question")) {
pendingQuestion = nativeQuestion();
emit({ type: "question.asked", id: "event-question-native-1", properties: pendingQuestion });
emit({ type: "question.asked", id: "event-question-native-1", properties: pendingQuestion });
return;
}
if (String(parsedPrompt.message ?? "").includes("native-permission")) {
permissionStyle = String(parsedPrompt.message ?? "").includes("legacy") ? "legacy" : "v2";
pendingPermission = nativePermission(permissionStyle);
emit({
type: permissionStyle === "legacy" ? "permission.updated" : "permission.v2.asked",
id: `event-permission-${permissionStyle}`,
properties: pendingPermission,
});
return;
}
if (String(parsedPrompt.message ?? "").includes("session-aborted")) {
emit({
type: "session.error",
id: "event-session-aborted",
properties: {
sessionID: session.id,
error: { name: "MessageAbortedError", data: { message: "Aborted" } },
},
});
return;
}
const textBeforeFinish = String(parsedPrompt.message ?? "").includes("text-before-finish");
const correlatedFinal = String(parsedPrompt.message ?? "").includes("correlated-final-message");
const finalAfterToolCommentary = String(parsedPrompt.message ?? "").includes("final-after-tool-commentary");
const commentaryOnlyBeforeWork = String(parsedPrompt.message ?? "").includes("commentary-only-before-work");
emit({ type: "message.updated", id: "event-user", properties: { sessionID: session.id, info: { id: "message-user", sessionID: session.id, role: "user" } } });
emit({ type: "message.part.updated", id: "event-user-part", properties: { sessionID: session.id, part: { id: "part-user", messageID: "message-user", type: "text", text: "submitted prompt must not become assistant text" } } });
emit({ type: "message.updated", id: "event-assistant", properties: { sessionID: session.id, info: { id: "message-assistant", sessionID: session.id, role: "assistant" } } });
emit({ type: "message.part.updated", id: "event-patch", properties: { sessionID: session.id, part: { id: "part-patch", messageID: "message-assistant", type: "patch", files: ["src/index.ts"] } } });
if (commentaryOnlyBeforeWork) {
emit({ type: "message.part.updated", id: "event-opening-commentary", properties: { sessionID: session.id, part: { id: "part-opening-commentary", messageID: "message-assistant", type: "text", text: "I will inspect the workspace first.", time: { start: 1, end: 2 } } } });
emit({ type: "message.part.updated", id: "event-work-tool", properties: { sessionID: session.id, part: { id: "part-work-tool", messageID: "message-assistant", type: "tool", tool: "bash", state: { status: "completed", input: { command: "pwd" }, output: "/workspace" } } } });
await callTerminalTool(promptPayload);
} else if (finalAfterToolCommentary) {
emit({ type: "message.part.updated", id: "event-tool-commentary", properties: { sessionID: session.id, part: { id: "part-tool-commentary", messageID: "message-assistant", type: "text", text: "The schema might need another value; I will try again.", time: { start: 1, end: 2 } } } });
await callTerminalTool(promptPayload);
emit({ type: "message.part.updated", id: "event-tool-part", properties: { sessionID: session.id, part: { id: "part-tool", messageID: "message-assistant", type: "tool", tool: "paperclip_paperclip_finish", state: { status: "completed", output: "accepted" } } } });
emit({ type: "message.updated", id: "event-post-tool-final-message", properties: { sessionID: session.id, info: { id: "message-post-tool-final", sessionID: session.id, role: "assistant" } } });
emit({ type: "message.part.updated", id: "event-post-tool-final", properties: { sessionID: session.id, part: { id: "part-post-tool-final", messageID: "message-post-tool-final", type: "text", text: "This is the complete substantive answer emitted after the accepted completion tool call.", time: { start: 3, end: 4 } } } });
} else if (correlatedFinal) {
emit({ type: "message.part.updated", id: "event-commentary", properties: { sessionID: session.id, part: { id: "part-commentary", messageID: "message-assistant", type: "text", text: "I will write the result now.", time: { start: 1, end: 2 } } } });
await callTerminalTool(promptPayload);
emit({ type: "message.updated", id: "event-final-message", properties: { sessionID: session.id, info: { id: "message-final", sessionID: session.id, role: "assistant" } } });
emit({ type: "message.part.updated", id: "event-finish-part", properties: { sessionID: session.id, part: { id: "part-finish", messageID: "message-final", type: "tool", tool: "paperclip_paperclip_finish", state: { status: "completed", output: "accepted" } } } });
emit({ type: "message.part.updated", id: "event-correlated-answer", properties: { sessionID: session.id, part: { id: "part-correlated-answer", messageID: "message-final", type: "text", text: "Correlated substantive final answer.", time: { start: 3, end: 4 } } } });
emit({ type: "message.part.updated", id: "event-duplicate-finish", properties: { sessionID: session.id, part: { id: "part-duplicate-finish", messageID: "message-final", type: "tool", tool: "unknown", callID: "functions.paperclip_paperclip_finish:24", state: { status: "error", error: "Tool execution aborted", metadata: { interrupted: true } } } } });
emit({ type: "message.updated", id: "event-trailing-message", properties: { sessionID: session.id, info: { id: "message-trailing", sessionID: session.id, role: "assistant" } } });
emit({ type: "message.part.updated", id: "event-trailing-ack", properties: { sessionID: session.id, part: { id: "part-trailing-ack", messageID: "message-trailing", type: "text", text: "Done.", time: { start: 5, end: 6 } } } });
} else if (textBeforeFinish) {
emit({ type: "message.part.updated", id: "event-answer", properties: { sessionID: session.id, part: { id: "part-answer", messageID: "message-assistant", type: "text", text: "Substantive answer before finish.", time: { start: 1, end: 2 } } } });
await callTerminalTool(promptPayload);
emit({ type: "message.part.updated", id: "event-ack", properties: { sessionID: session.id, part: { id: "part-ack", messageID: "message-assistant", type: "text", text: "Done.", time: { start: 3, end: 4 } } } });
} else {
await callTerminalTool(promptPayload);
emit({ type: "message.part.updated", id: "event-1", properties: { sessionID: session.id, part: { id: "part-1", messageID: "message-assistant", type: "text", text: "done [guide](guide.md)", time: { start: 1, end: 2 } } } });
emit({ type: "message.part.updated", id: "event-1", properties: { sessionID: session.id, part: { id: "part-1", messageID: "message-assistant", type: "text", text: "done" } } });
}
emit({ type: "message.updated", id: "event-2", properties: { info: { id: "message-assistant", sessionID: session.id, role: "assistant", tokens: { input: 3, output: 2 }, cost: 0.001 } } });
emit({ type: "session.idle", id: "event-3", properties: { sessionID: session.id } });
}, 100);
});
return;
}
json(response, 404, { error: "not found" });
});
server.listen(port, "127.0.0.1");
for (const signal of ["SIGTERM", "SIGINT"]) process.on(signal, () => server.close(() => process.exit(0)));
@@ -488,12 +488,17 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => {
await heartbeat.resumeQueuedRuns();
await waitForCondition(async () => {
const run = await db
.select({ status: heartbeatRuns.status })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
return run?.status === "cancelled";
const [run, wakeup] = await Promise.all([
db.select({ status: heartbeatRuns.status })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null),
db.select({ status: agentWakeupRequests.status })
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, wakeupRequestId))
.then((rows) => rows[0] ?? null),
]);
return run?.status === "cancelled" && wakeup?.status === "skipped";
});
const [run, wakeup, issue] = await Promise.all([
@@ -580,12 +585,17 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => {
await heartbeat.resumeQueuedRuns();
await waitForCondition(async () => {
const run = await db
.select({ status: heartbeatRuns.status })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
return run?.status === "cancelled";
const [run, wakeup] = await Promise.all([
db.select({ status: heartbeatRuns.status })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null),
db.select({ status: agentWakeupRequests.status })
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, wakeupRequestId))
.then((rows) => rows[0] ?? null),
]);
return run?.status === "cancelled" && wakeup?.status === "skipped";
});
const [run, wakeup, issue] = await Promise.all([