mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 18:19:05 +02:00
merge: preserve workspace callbacks through runtime selection extraction
Adopt upstream runtime-selection boundary and forward the task workspace checkpoint/phase callbacks without moving their run-log policy out of heartbeat. Runtime selection, project repository admission, and responsible-user invariants pass43 tests; server typecheck passes. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
0f8cea002a
46 files changed
+2399
-1103
No files matched your search
@@ -54,6 +54,19 @@ An old result cannot complete a newer operation.
|
||||
Deletion requires completed termination or a rejected hire.
|
||||
The existing restrictions for built-in agents and accounting still apply.
|
||||
|
||||
Messages accepted during setup remain queued until verification completes.
|
||||
After the worker commits `ready`, its driver immediately asks the scheduler to
|
||||
dispatch that agent's saved work. The scheduler rechecks company scope,
|
||||
invokability, budgets, concurrency, and task ownership through normal admission.
|
||||
If dispatch fails, readiness remains committed and the periodic queue sweep
|
||||
retries; setup is not repeated because of a scheduler failure.
|
||||
|
||||
Native OpenCode verification checks the selected model with the real hello
|
||||
probe. It validates the model identifier but does not first enumerate the full model
|
||||
catalog. Authentication failures, unavailable models, and probe timeouts still
|
||||
prevent readiness. Without an explicit local working directory, the native
|
||||
probe uses a temporary empty directory instead of the server's checkout.
|
||||
|
||||
## Commands and transactions
|
||||
|
||||
`requestHire` writes the agent record before preparation starts.
|
||||
|
||||
@@ -1351,6 +1351,16 @@ trace finalization immediately so a later error cannot lose cleanup state.
|
||||
Existing public failure helpers remain re-exported by `heartbeat.ts`.
|
||||
Keep completion policy changes separate from this extraction and dispatch.
|
||||
|
||||
Runtime selection is in `server/src/services/heartbeat/runtime-selection.ts`.
|
||||
`selectHeartbeatRuntime` constructs or restores native execution inputs, selects
|
||||
completion contracts and sessions, and persists native or legacy runtime choices.
|
||||
Native selection retains the row lock that serializes it with cancellation and
|
||||
controller lease renewal. The executor receives an explicit selection result;
|
||||
lifecycle and ownership callbacks update its teardown state immediately, including
|
||||
when preparation later throws. Public sandbox lifecycle and model fallback helpers
|
||||
remain re-exported by `heartbeat.ts`. Provider dispatch and final cleanup stay in
|
||||
the executor.
|
||||
|
||||
## Wake Context Delivery
|
||||
|
||||
Built-in adapters deliver wake context through the run prompt, including structured
|
||||
|
||||
@@ -72,7 +72,12 @@ runs and browser instances. Each hosted run records its invocation, parent run,
|
||||
event cursor and cost checkpoint. A leased reconciler polls minimal status,
|
||||
drains bounded event pages, discovers browser instances by conversation, and
|
||||
checks each instance's authoritative conversation association. Restarting the
|
||||
server resumes reconciliation. A lost paid-create response becomes an explicit
|
||||
server resumes reconciliation. The app schedules the next check from saved
|
||||
session deadlines and cleanup leases. Session writes wake it after commit.
|
||||
With no live sessions it performs no recurring browser database scan. Existing
|
||||
sessions still poll provider state (about three seconds while running, fifteen
|
||||
while idle), enforce expiry and access changes, and retry failed cleanup.
|
||||
A lost paid-create response becomes an explicit
|
||||
unknown outcome and is never submitted again automatically; the operator must
|
||||
inspect Browser Use before starting replacement work.
|
||||
|
||||
|
||||
@@ -438,10 +438,14 @@ Longer polling intervals alone do not complete an item.
|
||||
catch-up remain the recovery boundaries. Source:
|
||||
[email-channels.ts](../../server/src/services/email-channels.ts).
|
||||
|
||||
- [ ] **JOB-03 — Browser-use cleanup.** Current: every **3s** with due-session
|
||||
queries even when empty. Change: register actual session reconciliation/expiry
|
||||
deadlines. **Complete when:** no sessions means no queries, while active leases,
|
||||
expiry, and failed cleanup remain recoverable. Sources:
|
||||
- [x] **JOB-03 — Browser-use cleanup.** Replaced the empty **3s** scan with
|
||||
transaction-aware session signals and deadlines from existing `nextPollAt` and
|
||||
`leaseUntil` columns. Startup restores unfinished work; no live sessions means
|
||||
no recurring browser SQL or timer. Active and idle sessions retain provider
|
||||
polling, expiry checks, lease renewal, and cleanup retries. Credential-removal
|
||||
sweeps signal lease release so background cleanup resumes promptly. Standby
|
||||
suppresses SQL; idle drain defers dispatch with a memory-only wake. No new
|
||||
tables. Independent DB roots/direct SQL require restart recovery. Sources:
|
||||
[app.ts](../../server/src/app.ts),
|
||||
[browser-use.ts](../../server/src/services/browser-use.ts).
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { mkdtemp, rm, stat } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
@@ -72,6 +72,13 @@ vi.mock("@paperclipai/adapter-utils/execution-target", async () => {
|
||||
});
|
||||
|
||||
import { testEnvironment } from "./test.js";
|
||||
import { discoverOpenCodeModels, ensureOpenCodeModelConfiguredAndAvailable } from "./models.js";
|
||||
|
||||
vi.mock("./models.js", async () => ({
|
||||
...await vi.importActual<typeof import("./models.js")>("./models.js"),
|
||||
discoverOpenCodeModels: vi.fn().mockRejectedValue(new Error("catalog should not be needed")),
|
||||
ensureOpenCodeModelConfiguredAndAvailable: vi.fn().mockRejectedValue(new Error("catalog should not be needed")),
|
||||
}));
|
||||
|
||||
describe("opencode remote environment diagnostics", () => {
|
||||
const configHomes: string[] = [];
|
||||
@@ -89,6 +96,56 @@ describe("opencode remote environment diagnostics", () => {
|
||||
await Promise.all(configHomes.splice(0).map(dir => rm(dir, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
it.each(["passed", "auth_required", "model_unavailable", "timed_out"])(
|
||||
"validates a native model with the real probe path and no catalog scan (%s)", async outcome => {
|
||||
const successfulProbe = await runAdapterExecutionTargetProcess.getMockImplementation()!();
|
||||
runAdapterExecutionTargetProcess.mockResolvedValueOnce(outcome === "passed" ? successfulProbe : {
|
||||
...successfulProbe, exitCode: 1, stdout: "", timedOut: outcome === "timed_out",
|
||||
stderr: outcome === "auth_required" ? "invalid API key" : "ProviderModelNotFoundError",
|
||||
});
|
||||
const result = await testEnvironment({ companyId: "company-1", adapterType: "paperclip_runner",
|
||||
config: { model: "openrouter/deepseek/deepseek-v4-flash-0731", env: { XDG_CONFIG_HOME: configHome } } });
|
||||
expect(discoverOpenCodeModels).not.toHaveBeenCalled();
|
||||
expect(ensureOpenCodeModelConfiguredAndAvailable).not.toHaveBeenCalled();
|
||||
expect(runAdapterExecutionTargetProcess).toHaveBeenCalledExactlyOnceWith(
|
||||
expect.any(String), null, "opencode", ["run", "--format", "json", "--model", "openrouter/deepseek/deepseek-v4-flash-0731"],
|
||||
expect.objectContaining({ stdin: "Respond with hello." }),
|
||||
);
|
||||
expect(result.status).toBe(outcome === "passed" ? "pass" : "warn");
|
||||
expect(result.checks).toContainEqual(expect.objectContaining({ code: `opencode_hello_probe_${outcome}` }));
|
||||
const call = runAdapterExecutionTargetProcess.mock.calls[0] as unknown as [unknown, unknown, unknown, unknown, { cwd: string }];
|
||||
expect(path.basename(call[4].cwd)).toMatch(/^paperclip-opencode-native-probe-/);
|
||||
await expect(stat(call[4].cwd)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects a malformed native model before starting the probe", async () => {
|
||||
const result = await testEnvironment({ companyId: "company-1", adapterType: "paperclip_runner",
|
||||
config: { model: "missing-provider", env: { XDG_CONFIG_HOME: configHome } } });
|
||||
expect(result.status).toBe("fail");
|
||||
expect(runAdapterExecutionTargetProcess).not.toHaveBeenCalled();
|
||||
expect(discoverOpenCodeModels).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("preserves an explicitly configured native probe directory", async () => {
|
||||
await testEnvironment({ companyId: "company-1", adapterType: "paperclip_runner",
|
||||
config: { model: "provider/model", cwd: configHome, env: { XDG_CONFIG_HOME: configHome } } });
|
||||
const call = runAdapterExecutionTargetProcess.mock.calls[0] as unknown as [unknown, unknown, unknown, unknown, { cwd: string }];
|
||||
expect(call[4].cwd).toBe(configHome);
|
||||
expect((await stat(configHome)).isDirectory()).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps catalog validation for the legacy local adapter", async () => {
|
||||
vi.mocked(discoverOpenCodeModels).mockResolvedValueOnce([{ id: "provider/model", label: "Model" }]);
|
||||
vi.mocked(ensureOpenCodeModelConfiguredAndAvailable).mockResolvedValueOnce([{ id: "provider/model", label: "Model" }]);
|
||||
const result = await testEnvironment({ companyId: "company-1", adapterType: "opencode_local",
|
||||
config: { model: "provider/model", env: { XDG_CONFIG_HOME: configHome } } });
|
||||
expect(result.status).toBe("pass");
|
||||
expect(discoverOpenCodeModels).toHaveBeenCalledOnce();
|
||||
expect(ensureOpenCodeModelConfiguredAndAvailable).toHaveBeenCalledOnce();
|
||||
expect(runAdapterExecutionTargetProcess).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each([false, true])("stages remote runtime config assets for sandbox hello probes (managed=%s)", async (managed) => {
|
||||
const configHome = await mkdtemp(path.join(os.tmpdir(), "opencode-remote-test-config-"));
|
||||
configHomes.push(configHome);
|
||||
|
||||
@@ -25,7 +25,7 @@ import {
|
||||
prepareAdapterExecutionTargetRuntime,
|
||||
overrideAdapterExecutionTargetRemoteCwd,
|
||||
} from "@paperclipai/adapter-utils/execution-target";
|
||||
import { discoverOpenCodeModels, ensureOpenCodeModelConfiguredAndAvailable } from "./models.js";
|
||||
import { discoverOpenCodeModels, ensureOpenCodeModelConfiguredAndAvailable, requireOpenCodeModelId } from "./models.js";
|
||||
import { parseOpenCodeJsonl } from "./parse.js";
|
||||
import { SANDBOX_INSTALL_COMMAND } from "../index.js";
|
||||
import { prepareOpenCodeRuntimeConfig, prepareManagedOpenCodeRemoteHomes } from "./runtime-config.js";
|
||||
@@ -141,9 +141,16 @@ export async function testEnvironment(
|
||||
// still has the path available for cleanup in `finally` — otherwise the
|
||||
// `fs.mkdtemp` directory leaks on the early-throw path.
|
||||
let preparedRuntimeWorkspaceLocalDir: string | null = null;
|
||||
let nativeProbeWorkspaceLocalDir: string | null = null;
|
||||
try {
|
||||
let runtimeTarget: AdapterExecutionTarget | null = target ?? null;
|
||||
let runtimeCwd = cwd;
|
||||
// Native workspaces are assigned at execution time. A configuration probe
|
||||
// without an explicit workspace must not run in the server's checkout.
|
||||
if (!targetIsRemote && ctx.adapterType === "paperclip_runner" && !asString(config.cwd, "").trim()) {
|
||||
nativeProbeWorkspaceLocalDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-opencode-native-probe-"));
|
||||
runtimeCwd = nativeProbeWorkspaceLocalDir;
|
||||
}
|
||||
if (targetIsRemote) {
|
||||
preparedRuntimeWorkspaceLocalDir = await fs.mkdtemp(path.join(os.tmpdir(), `paperclip-opencode-envtest-${runId}-`));
|
||||
const preparedExecutionTargetRuntime = await prepareAdapterExecutionTargetRuntime({
|
||||
@@ -223,12 +230,30 @@ export async function testEnvironment(
|
||||
let modelValidationPassed = false;
|
||||
const configuredModel = asString(config.model, "").trim();
|
||||
|
||||
// The native runner already selects a concrete model. Its real hello probe
|
||||
// verifies model access and credentials; a full catalog scan adds another
|
||||
// cold CLI launch (and discovery retries) without verifying that request.
|
||||
const nativeModelProbe = ctx.adapterType === "paperclip_runner" && Boolean(configuredModel);
|
||||
|
||||
// Model discovery and validation use local child processes against
|
||||
// OpenCode's `models` subcommand and JSON config; these are not yet
|
||||
// wired through the execution target. When probing a remote env, skip
|
||||
// discovery/validation and rely on the remote hello probe to surface
|
||||
// model/auth issues directly.
|
||||
if (targetIsRemote && configuredModel) {
|
||||
if (nativeModelProbe) {
|
||||
try {
|
||||
requireOpenCodeModelId(configuredModel);
|
||||
modelValidationPassed = true;
|
||||
checks.push({
|
||||
code: "opencode_model_validation_via_probe",
|
||||
level: "info",
|
||||
message: "The configured native-runner model will be validated by the hello probe.",
|
||||
});
|
||||
} catch (error) {
|
||||
checks.push({ code: "opencode_model_invalid", level: "error",
|
||||
message: error instanceof Error ? error.message : "Configured model is invalid." });
|
||||
}
|
||||
} else if (targetIsRemote && configuredModel) {
|
||||
checks.push({
|
||||
code: "opencode_model_validation_skipped_remote",
|
||||
level: "info",
|
||||
@@ -305,7 +330,7 @@ export async function testEnvironment(
|
||||
const modelUnavailable = checks.some((check) => check.code === "opencode_hello_probe_model_unavailable");
|
||||
if (!configuredModel && !modelUnavailable) {
|
||||
// No model configured – skip model requirement if no model-related checks exist
|
||||
} else if (!targetIsRemote && configuredModel && canRunProbe) {
|
||||
} else if (!nativeModelProbe && !targetIsRemote && configuredModel && canRunProbe) {
|
||||
try {
|
||||
await ensureOpenCodeModelConfiguredAndAvailable({
|
||||
model: configuredModel,
|
||||
@@ -430,6 +455,9 @@ export async function testEnvironment(
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (nativeProbeWorkspaceLocalDir) {
|
||||
await fs.rm(nativeProbeWorkspaceLocalDir, { recursive: true, force: true }).catch(() => {});
|
||||
}
|
||||
await restoreWorkspace?.();
|
||||
if (!restoreWorkspace && preparedRuntimeWorkspaceLocalDir) {
|
||||
// Reached when `prepareAdapterExecutionTargetRuntime` threw before
|
||||
|
||||
@@ -238,6 +238,37 @@ provider-native operations. Full auto is resolved consistently for agent
|
||||
creation, adapter conversion, direct driver launches, and fresh/resumed turns.
|
||||
Explicitly stored restrictive modes still apply.
|
||||
|
||||
Native OpenCode forwards incremental text and reasoning parts as they arrive;
|
||||
the final full snapshot selects the final response without repeating streamed
|
||||
text. Delta parts must match an observed message and part identity. Identical
|
||||
chunks without event IDs remain distinct tokens. Each isolated launch disables
|
||||
automatic updates, remote model-catalog refreshes, and default plugins; the
|
||||
runner supplies the pinned executable, selected model, and assigned MCP tools.
|
||||
Provider inference and reasoning still contribute to time to first text. These
|
||||
startup settings do not disable model reasoning.
|
||||
|
||||
On macOS, the verified OpenCode executable keeps its private launch pathname
|
||||
until the provider process exits. Removing that pathname immediately after
|
||||
spawn can kill or stall the signed binary before its health check succeeds.
|
||||
Ownership, permissions, and file-identity checks remain enforced; cleanup also
|
||||
runs on failed launches, and retries create a fresh launch snapshot.
|
||||
|
||||
Call `session.startTurn({ message, reasoningMode: "disabled" })` to disable
|
||||
reasoning for one OpenCode/OpenRouter turn. `CapabilityLiveSession.sendMessage`
|
||||
accepts the same option in its second argument. The choice travels in the durable
|
||||
`turn.start` command and selects an OpenCode model variant for that prompt only.
|
||||
Omitting it (or passing `"default"`) on the next turn restores provider defaults,
|
||||
including on a warm or recovered session. It is not an environment variable or
|
||||
agent-wide setting. No automatic decision policy selects it. The model must
|
||||
support disabling reasoning; other runner providers reject an explicit selection
|
||||
before starting work. Native sessions expose support through the
|
||||
`perTurnReasoning` capability.
|
||||
|
||||
OpenCode's proxy sends agent instructions once per turn through the system prompt. The
|
||||
task envelope retains task-specific constraints and the completion contract,
|
||||
without repeating the system instructions. This applies to all users of the
|
||||
proxy; already-recorded conversation history is not rewritten.
|
||||
|
||||
The runner's authenticated bridge and controller still enforce company access,
|
||||
action claims, task modes, and governed approvals. Provider permission defaults
|
||||
do not change workspace isolation or grant credentials or connection access.
|
||||
|
||||
@@ -386,6 +386,13 @@ pub struct CodexSkillInput {
|
||||
pub path: String,
|
||||
}
|
||||
|
||||
/// Optional inputs for one turn; omitted fields preserve provider defaults.
|
||||
#[derive(Clone, Copy, Debug, Default)]
|
||||
pub struct CodexTurnOptions<'a> {
|
||||
pub skills: &'a [CodexSkillInput],
|
||||
pub reasoning_mode: Option<&'a str>,
|
||||
}
|
||||
|
||||
impl CodexSkillInput {
|
||||
pub fn validate(&self) -> Result<(), LocalRunnerError> {
|
||||
if self.input_type != "skill"
|
||||
@@ -1670,15 +1677,26 @@ impl CodexProvider {
|
||||
}
|
||||
|
||||
pub fn start_turn(&mut self, message: &str, cwd: &str) -> Result<Value, LocalRunnerError> {
|
||||
self.start_turn_with_skills(message, cwd, &[])
|
||||
self.start_turn_with_options(message, cwd, CodexTurnOptions::default())
|
||||
}
|
||||
|
||||
pub fn start_turn_with_skills(
|
||||
pub fn start_turn_with_options(
|
||||
&mut self,
|
||||
message: &str,
|
||||
cwd: &str,
|
||||
skills: &[CodexSkillInput],
|
||||
options: CodexTurnOptions<'_>,
|
||||
) -> Result<Value, LocalRunnerError> {
|
||||
let CodexTurnOptions {
|
||||
skills,
|
||||
reasoning_mode,
|
||||
} = options;
|
||||
if reasoning_mode.is_some_and(|mode| {
|
||||
self.config.provider != "opencode" || !matches!(mode, "default" | "disabled")
|
||||
}) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"turn.reasoningMode requires OpenCode and must be default or disabled",
|
||||
));
|
||||
}
|
||||
if skills.len() > 64 || (self.config.provider != "codex" && !skills.is_empty()) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"explicit skills require Codex and at most 64 selections",
|
||||
@@ -1727,6 +1745,9 @@ impl CodexProvider {
|
||||
let turn_params_object = turn_params
|
||||
.as_object_mut()
|
||||
.expect("Codex turn parameters are an object");
|
||||
if let Some(mode) = reasoning_mode {
|
||||
turn_params_object.insert("reasoningMode".to_owned(), json!(mode));
|
||||
}
|
||||
if self.permission_profile == "paperclip-runner-external-sandbox" {
|
||||
turn_params_object.insert(
|
||||
"sandboxPolicy".to_owned(),
|
||||
|
||||
@@ -13,7 +13,7 @@ use serde_json::{json, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::codex_provider::{
|
||||
CodexProvider, CodexProviderConfig, CodexProviderEvent, CodexSkillInput,
|
||||
CodexProvider, CodexProviderConfig, CodexProviderEvent, CodexSkillInput, CodexTurnOptions,
|
||||
ProviderStartupObservation, ProviderStartupStage, RejectedAcceptedTurn,
|
||||
MAX_SETTLED_PROVIDER_TURN_IDS,
|
||||
};
|
||||
@@ -2903,7 +2903,24 @@ impl CodexCommandExecutor {
|
||||
.validate()
|
||||
.map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
|
||||
}
|
||||
let reasoning_mode = match payload.get("reasoningMode") {
|
||||
None => None,
|
||||
Some(Value::String(mode)) if matches!(mode.as_str(), "default" | "disabled") => {
|
||||
Some(mode.as_str())
|
||||
}
|
||||
Some(_) => return Err(DurableRunnerError::invalid("invalid turn.reasoningMode")),
|
||||
};
|
||||
self.restore_provider_if_needed()?;
|
||||
if reasoning_mode.is_some()
|
||||
&& self
|
||||
.state
|
||||
.as_ref()
|
||||
.is_none_or(|state| state.config.provider != "opencode")
|
||||
{
|
||||
return Err(DurableRunnerError::invalid(
|
||||
"turn.reasoningMode requires OpenCode",
|
||||
));
|
||||
}
|
||||
if self
|
||||
.state
|
||||
.as_ref()
|
||||
@@ -2986,7 +3003,14 @@ impl CodexCommandExecutor {
|
||||
rejected_accepted_turn,
|
||||
) = {
|
||||
let provider = self.ensure_provider()?;
|
||||
let result = provider.start_turn_with_skills(text, &cwd, &skills);
|
||||
let result = provider.start_turn_with_options(
|
||||
text,
|
||||
&cwd,
|
||||
CodexTurnOptions {
|
||||
skills: &skills,
|
||||
reasoning_mode,
|
||||
},
|
||||
);
|
||||
(
|
||||
result,
|
||||
provider.completed_turn_authority().is_some(),
|
||||
@@ -6068,6 +6092,27 @@ mod tests {
|
||||
executor.state = Some(state);
|
||||
executor.restore_checked = true;
|
||||
|
||||
for mode in [
|
||||
json!(null),
|
||||
json!(false),
|
||||
json!("bad"),
|
||||
json!("disabled"),
|
||||
json!("default"),
|
||||
] {
|
||||
let error = executor
|
||||
.start_turn(&json!({"text": "must not reach the provider", "reasoningMode": mode}))
|
||||
.unwrap_err();
|
||||
assert!(error.to_string().contains("reasoningMode"));
|
||||
assert!(executor.provider.is_none());
|
||||
assert!(
|
||||
!executor
|
||||
.state
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.ambiguous_turn_start_pending
|
||||
);
|
||||
}
|
||||
|
||||
let error = executor
|
||||
.start_turn(&json!({"text": "must not reach the provider"}))
|
||||
.expect_err("the exact regular backlog limit must reject admission");
|
||||
|
||||
@@ -97,6 +97,34 @@ const driver: HarnessDriver = {
|
||||
};
|
||||
|
||||
describe("HarnessDriverBackend", () => {
|
||||
it.each(["default", "disabled"] as const)("rejects unsupported reasoning %s before starting provider work", async reasoningMode => {
|
||||
const startTurn = vi.fn(async () => ({ turnId: "turn-1" }));
|
||||
const session = Object.assign(new FakeHarnessSession(), { startTurn });
|
||||
const backend = new HarnessDriverBackend({ ...driver, openSession: async () => session });
|
||||
const opened = await backend.openSession({ identity: {
|
||||
runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1",
|
||||
} });
|
||||
expect((await opened.capabilities()).perTurnReasoning).toBe(false);
|
||||
const message = { role: "user" as const, text: "Hi" };
|
||||
await expect(opened.startTurn({ message, reasoningMode })).rejects.toThrow("Per-turn reasoning is not supported");
|
||||
expect(startTurn).not.toHaveBeenCalled();
|
||||
await expect(opened.startTurn({ message })).resolves.toEqual({ turnId: "turn-1" });
|
||||
expect(startTurn).toHaveBeenCalledExactlyOnceWith({ message });
|
||||
});
|
||||
|
||||
it("forwards reasoning when the provider explicitly supports it", async () => {
|
||||
const startTurn = vi.fn(async () => ({ turnId: "turn-1" }));
|
||||
const session = Object.assign(new FakeHarnessSession(), { startTurn, supportsTurnReasoning: () => true });
|
||||
const backend = new HarnessDriverBackend({ ...driver, openSession: async () => session });
|
||||
const opened = await backend.openSession({ identity: {
|
||||
runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1",
|
||||
} });
|
||||
expect((await opened.capabilities()).perTurnReasoning).toBe(true);
|
||||
const input = { message: { role: "user" as const, text: "Hi" }, reasoningMode: "disabled" as const };
|
||||
await opened.startTurn(input);
|
||||
expect(startTurn).toHaveBeenCalledExactlyOnceWith(input);
|
||||
});
|
||||
|
||||
it("refreshes negotiated controls after a warm handshake and separates follow-ups", async () => {
|
||||
let controls = { steering: false, queuedFollowUp: false };
|
||||
const steer = vi.fn(async () => {});
|
||||
|
||||
@@ -454,6 +454,7 @@ class HarnessNativeSession implements NativeSession {
|
||||
return {
|
||||
resume: true,
|
||||
typedEvents: true,
|
||||
perTurnReasoning: this.#session.supportsTurnReasoning?.() === true,
|
||||
steering: (negotiated?.steering ?? this.#steeringSupported) && this.#session.steer !== undefined,
|
||||
queuedFollowUp: negotiated?.queuedFollowUp === true && this.#session.steer !== undefined,
|
||||
interruption: this.#session.interrupt !== undefined,
|
||||
@@ -710,6 +711,9 @@ class HarnessNativeSession implements NativeSession {
|
||||
// The provider has no active turn to interrupt yet. Do not launch work
|
||||
// whose events cancellation would suppress and leave the owner waiting.
|
||||
if (this.#explicitlyCancelled) throw new Error("native_session_cancelled");
|
||||
if (input.reasoningMode !== undefined && this.#session.supportsTurnReasoning?.() !== true) {
|
||||
throw new Error("Per-turn reasoning is not supported by this provider");
|
||||
}
|
||||
this.#terminal = null;
|
||||
try {
|
||||
const started = await this.#session.startTurn(input);
|
||||
|
||||
@@ -190,11 +190,10 @@ describe("direct OpenCode HTTP instruction boundary", () => {
|
||||
const request = requests.at(-1)!;
|
||||
expect(request.providerID).toBe("openrouter"); expect(request.modelID).toBe("deepseek/deepseek-v4-flash-0731");
|
||||
const text = request.parts[0].text as string;
|
||||
expect(request.system).toBe(nativeSystemInstructions(value));
|
||||
if (schema === "v4" && phase === "start") {
|
||||
expect(request.system).toBe(nativeSystemInstructions(input));
|
||||
expect(JSON.parse(text).task.constraints).toEqual(nativeTaskConstraints(input));
|
||||
} else {
|
||||
expect(request.system).toBeUndefined();
|
||||
expect(JSON.parse(text)).toEqual(envelope);
|
||||
}
|
||||
directOpenCodeReceipts.push({ provider: "opencode", schema, phase, request: measure(request),
|
||||
|
||||
@@ -9,7 +9,7 @@ import {
|
||||
rmSync,
|
||||
writeFileSync,
|
||||
} from "node:fs";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { spawn, spawnSync } from "node:child_process";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import {
|
||||
@@ -28,7 +28,7 @@ describe("OpenCode runnerd proxy executable", () => {
|
||||
])).toEqual({ command: "/proc/self/fd/7", commandFd: 7 });
|
||||
});
|
||||
|
||||
it("executes the inherited artifact through the nested child fd mapping", () => {
|
||||
it("executes the inherited artifact through the nested child fd mapping", async () => {
|
||||
if (process.platform === "win32") return;
|
||||
if (process.platform === "darwin") {
|
||||
const root = mkdtempSync(join(tmpdir(), "paperclip-opencode-nested-"));
|
||||
@@ -45,22 +45,27 @@ describe("OpenCode runnerd proxy executable", () => {
|
||||
TRUSTED_OPENCODE_EXECUTABLE_ARG,
|
||||
command,
|
||||
]);
|
||||
binding.commandLifecycle?.beforeSpawn();
|
||||
const child = spawnSync(binding.command, [], { encoding: "utf8" });
|
||||
expect(child.error).toBeUndefined();
|
||||
expect(child.status).toBe(0);
|
||||
expect(child.stdout).toBe("verified-nested-spawn");
|
||||
binding.commandLifecycle?.afterSpawn();
|
||||
expect(() => lstatSync(command)).toThrow();
|
||||
expect(() => lstatSync(directory)).toThrow();
|
||||
binding.commandLifecycle?.beforeSpawn();
|
||||
const retriedChild = spawnSync(binding.command, [], { encoding: "utf8" });
|
||||
expect(retriedChild.error).toBeUndefined();
|
||||
expect(retriedChild.status).toBe(0);
|
||||
expect(retriedChild.stdout).toBe("verified-nested-spawn");
|
||||
binding.commandLifecycle?.afterSpawn();
|
||||
expect(() => lstatSync(command)).toThrow();
|
||||
expect(() => lstatSync(directory)).toThrow();
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
binding.commandLifecycle!.beforeSpawn();
|
||||
const child = spawn(binding.command, [], { stdio: ["ignore", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
child.stdout!.on("data", chunk => { stdout += chunk; });
|
||||
const exited = new Promise<number | null>((resolve, reject) => {
|
||||
child.once("error", reject);
|
||||
child.once("close", resolve);
|
||||
});
|
||||
binding.commandLifecycle!.afterSpawn();
|
||||
// Exercise the production async boundary, not spawnSync followed by
|
||||
// cleanup after the executable has already finished.
|
||||
expect(lstatSync(command).isFile()).toBe(true);
|
||||
expect(() => binding.commandLifecycle!.beforeSpawn()).toThrow("still in use");
|
||||
expect(await exited).toBe(0);
|
||||
expect(stdout).toBe("verified-nested-spawn");
|
||||
binding.commandLifecycle!.afterExit();
|
||||
binding.commandLifecycle!.afterExit();
|
||||
expect(() => lstatSync(command)).toThrow();
|
||||
expect(() => lstatSync(directory)).toThrow();
|
||||
}
|
||||
} finally {
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
@@ -131,6 +136,7 @@ describe("OpenCode runnerd proxy task envelope", () => {
|
||||
criterionIds: ["objective", "verification"],
|
||||
},
|
||||
})).toMatchObject({
|
||||
constraints: ["Use Paperclip MCP tools for semantic operations."],
|
||||
completionContract: {
|
||||
revision: "17",
|
||||
criteria: [
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
#!/usr/bin/env node
|
||||
import { parseOpenCodeReasoningMode } from "../drivers/opencode/reasoning-mode.js";
|
||||
import { createInterface } from "node:readline";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
@@ -365,6 +366,7 @@ async function handle(message: RpcMessage): Promise<void> {
|
||||
.join("\n");
|
||||
const turn = await session.startTurn({
|
||||
message: { role: "user", text: messageText },
|
||||
...(params.reasoningMode === undefined ? {} : { reasoningMode: parseOpenCodeReasoningMode(params.reasoningMode) }),
|
||||
});
|
||||
activeTurnId = turn.turnId;
|
||||
// OpenCode normally publishes session/turn startup over SSE, but a fast
|
||||
|
||||
@@ -39,6 +39,7 @@ export function trustedOpenCodeLaunchBinding(
|
||||
commandLifecycle?: {
|
||||
beforeSpawn(): void;
|
||||
afterSpawn(): void;
|
||||
afterExit(): void;
|
||||
};
|
||||
} {
|
||||
const command = args.length === 2 && args[0] === TRUSTED_OPENCODE_EXECUTABLE_ARG
|
||||
@@ -119,8 +120,7 @@ export function trustedOpenCodeLaunchBinding(
|
||||
let materializedIdentity: { dev: number; ino: number } | undefined;
|
||||
const materializeForSpawn = () => {
|
||||
if (materialized) {
|
||||
validateMacSnapshot(materializedIdentity);
|
||||
return;
|
||||
throw new Error("Verified OpenCode executable is still in use");
|
||||
}
|
||||
let destinationFd: number | undefined;
|
||||
try {
|
||||
@@ -184,10 +184,15 @@ export function trustedOpenCodeLaunchBinding(
|
||||
// attackers are outside the documented local-host trust boundary in
|
||||
// docs/durable-recovery.md. Keep the verified source on an unlinked
|
||||
// descriptor, rematerialize only at the syscall boundary (including
|
||||
// retries), then remove the executable pathname immediately.
|
||||
// retries). Keep its private pathname until the child exits: deleting
|
||||
// a signed executable during macOS startup can kill or stall the child.
|
||||
beforeSpawn: materializeForSpawn,
|
||||
afterSpawn() {
|
||||
validateMacSnapshot(materializedIdentity);
|
||||
},
|
||||
afterExit() {
|
||||
if (!materialized) return;
|
||||
validateMacSnapshot(materializedIdentity);
|
||||
unlinkSync(command);
|
||||
rmdirSync(dirname(command));
|
||||
materialized = false;
|
||||
|
||||
@@ -30,7 +30,8 @@ export function openCodeProxyTaskEnvelope(params: Record<string, unknown>) {
|
||||
}
|
||||
: {}),
|
||||
constraints: [
|
||||
text(params.baseInstructions, "Complete only the supplied task."),
|
||||
// baseInstructions already travels in the system prompt. Keep task
|
||||
// constraints here without duplicating the agent instruction bundle.
|
||||
"Use Paperclip MCP tools for semantic operations.",
|
||||
],
|
||||
});
|
||||
|
||||
@@ -498,6 +498,7 @@ export interface HarnessSessionRecoveryResult {
|
||||
}
|
||||
|
||||
export interface HarnessSession {
|
||||
supportsTurnReasoning?(): boolean;
|
||||
turnControlCapabilities?(): NativeTurnControlCapabilities | null;
|
||||
ids(): {
|
||||
driverSessionId: string;
|
||||
@@ -511,6 +512,8 @@ export interface HarnessSession {
|
||||
/** Set by orchestration only after successful provider-session recovery. */
|
||||
continuation?: true;
|
||||
requestedCollaborationMode?: "default" | "plan";
|
||||
/** OpenCode/OpenRouter only. Applies to this turn, never subsequent turns. */
|
||||
reasoningMode?: "default" | "disabled";
|
||||
}): Promise<{
|
||||
turnId: string;
|
||||
effectiveCollaborationMode?: "default" | "plan";
|
||||
|
||||
@@ -193,6 +193,8 @@ export interface NativeSession {
|
||||
/** Set by orchestration only after successful provider-session recovery. */
|
||||
continuation?: true;
|
||||
requestedCollaborationMode?: "default" | "plan";
|
||||
/** OpenCode/OpenRouter only. Applies to this turn, never subsequent turns. */
|
||||
reasoningMode?: "default" | "disabled";
|
||||
}): Promise<{
|
||||
turnId: string;
|
||||
effectiveCollaborationMode?: "default" | "plan";
|
||||
|
||||
@@ -41,6 +41,8 @@ export interface NativeSessionCapabilities {
|
||||
runtimeRequestHandoff?: boolean;
|
||||
goals?: boolean;
|
||||
threadLineage?: boolean;
|
||||
/** Can select reasoning for one turn without changing subsequent turns. */
|
||||
perTurnReasoning?: boolean;
|
||||
collaborationModes?: Array<"default" | "plan">;
|
||||
unsupported?: string[];
|
||||
}
|
||||
|
||||
@@ -35,6 +35,7 @@ export type CodexServerRequestHandler = (
|
||||
) => Promise<Record<string, unknown>>;
|
||||
|
||||
export interface CodexAppServerTransport {
|
||||
supportsTurnReasoning?(): boolean;
|
||||
/** Runner-owned live capability projection; absent on native Codex transports. */
|
||||
turnControlCapabilities?(): NativeTurnControlCapabilities | null;
|
||||
request(
|
||||
|
||||
+13
@@ -45,6 +45,19 @@ import {
|
||||
import { rehydrateRunnerdItemNotification } from "../../live/runnerd-codex-transport.js";
|
||||
|
||||
describe("Codex app-server Codex driver", () => {
|
||||
it.each(["default", "disabled"] as const)("rejects unsupported reasoning %s without sending a turn RPC", async reasoningMode => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({
|
||||
runId: "run-reasoning", normalizedSessionId: "normalized-reasoning", workingDirectory: WORKSPACE,
|
||||
});
|
||||
const message = { role: "user" as const, text: "Hi" };
|
||||
await expect(session.startTurn({ message, reasoningMode })).rejects.toThrow("Per-turn reasoning is not supported");
|
||||
expect(transport.calls.filter(call => call.method === "turn/start")).toHaveLength(0);
|
||||
await session.startTurn({ message });
|
||||
expect(transport.calls.filter(call => call.method === "turn/start")).toHaveLength(1);
|
||||
await session.close({ reason: "fixture complete" });
|
||||
});
|
||||
|
||||
it("retains one correlation-bound steering acknowledgement after the rehydrated runnerd echo", async () => {
|
||||
const transport = new FakeCodexTransport("thread-root");
|
||||
const session = await makeDriver([transport]).openSession({
|
||||
|
||||
@@ -82,6 +82,10 @@ export class CodexHarnessSession
|
||||
}
|
||||
}
|
||||
|
||||
supportsTurnReasoning(): boolean {
|
||||
return this.transport.supportsTurnReasoning?.() === true;
|
||||
}
|
||||
|
||||
turnControlCapabilities() {
|
||||
if (this.driverKind === "acpx_runtime") {
|
||||
return this.transport.turnControlCapabilities?.() ?? { steering: false, queuedFollowUp: false };
|
||||
@@ -157,11 +161,16 @@ export class CodexHarnessSession
|
||||
/** Set by orchestration only after successful provider-session recovery. */
|
||||
continuation?: true;
|
||||
requestedCollaborationMode?: "default" | "plan";
|
||||
/** OpenCode/OpenRouter only. Applies to this turn, never subsequent turns. */
|
||||
reasoningMode?: "default" | "disabled";
|
||||
}): Promise<{
|
||||
turnId: string;
|
||||
effectiveCollaborationMode: "default" | "plan";
|
||||
}> {
|
||||
this.assertProtocolIntegrity();
|
||||
if (input.reasoningMode !== undefined && !this.supportsTurnReasoning()) {
|
||||
throw new Error("Per-turn reasoning is not supported by this provider");
|
||||
}
|
||||
if (this.protocolFailed && this.protocolFailureCode) {
|
||||
throw new NativeProviderTerminalFailure(this.protocolFailureCode, false, this.protocolFailureMessage ?? undefined);
|
||||
}
|
||||
@@ -231,6 +240,7 @@ export class CodexHarnessSession
|
||||
try {
|
||||
response = await this.transport.request("turn/start", {
|
||||
threadId: this.opened.threadId,
|
||||
...(input.reasoningMode === undefined ? {} : { reasoningMode: input.reasoningMode }),
|
||||
...(this.reasoningEffort ? { effort: this.reasoningEffort } : {}),
|
||||
cwd: this.opened.context.workingDirectory,
|
||||
permissions:
|
||||
|
||||
@@ -906,6 +906,8 @@ describe("OpenCodeServerDriver", () => {
|
||||
expect(environment.keys).not.toContain("UNRELATED_SECRET");
|
||||
expect(environment.keys).not.toContain("PAPERCLIP_PROVIDER_TRACE_PATH");
|
||||
expect(environment.projectConfigDisabled).toBe("true");
|
||||
expect(environment.modelsFetchDisabled).toBe("true");
|
||||
expect(environment.defaultPluginsDisabled).toBe("true");
|
||||
expect(mcpEvidence.tools).toEqual(
|
||||
expect.arrayContaining(["paperclip_finish", "paperclip_block"]),
|
||||
);
|
||||
@@ -920,6 +922,46 @@ describe("OpenCodeServerDriver", () => {
|
||||
expect(diagnostics.join("\n")).toContain("[REDACTED]");
|
||||
});
|
||||
|
||||
it.each(["role-first", "role-last", "role-last-long"])("preserves incremental text and completion with identity checks (%s)", async (roleOrder) => {
|
||||
await chmod(fixture, 0o755);
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-stream-"));
|
||||
const workspace = await mkdtemp(join(tmpdir(), "paperclip-opencode-stream-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 },
|
||||
});
|
||||
const session = await driver.openSession({ runId: "stream", normalizedSessionId: "stream", workingDirectory: workspace });
|
||||
try {
|
||||
await session.startTurn({ message: { role: "user", text: `incremental-stream ${roleOrder}` } });
|
||||
const deltas: string[] = [];
|
||||
const reasoning: string[] = [];
|
||||
const events: PrpEvent[] = [];
|
||||
for await (const event of session.events()) {
|
||||
events.push(event);
|
||||
if (event.eventType === "item.delta" && event.payload.kind === "reasoning") reasoning.push(String(event.payload.text));
|
||||
if (event.eventType === "item.delta" && event.payload.kind === "agentMessage") {
|
||||
deltas.push(String(event.payload.text));
|
||||
// The fixture cannot finish its turn until a chunk reaches us.
|
||||
if (deltas.length === 1) {
|
||||
expect(await readFile(join(root, "stream", "data", "stream-released"), "utf8").catch(() => null)).toBeNull();
|
||||
}
|
||||
await writeFile(join(root, "stream", "data", "release-stream"), "ready");
|
||||
}
|
||||
if (TURN_TERMINAL_EVENT_TYPES.has(event.eventType)) break;
|
||||
}
|
||||
const expectedText = `Hi ${"👋".repeat(roleOrder === "role-last-long" ? 120 : 2)}`;
|
||||
expect(deltas).toEqual(roleOrder === "role-first" ? ["Hi ", "👋", "👋"] : [expectedText]);
|
||||
expect(reasoning).toEqual(["Thinking"]);
|
||||
expect(events.filter(event => event.eventType === "item.completed" && event.payload.kind === "agentMessage")
|
||||
.map(event => event.payload.text)).toEqual([expectedText]);
|
||||
} finally {
|
||||
await session.close({ reason: "stream test complete" });
|
||||
}
|
||||
});
|
||||
|
||||
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-"));
|
||||
@@ -1497,7 +1539,7 @@ describe("OpenCodeServerDriver", () => {
|
||||
await session.close({ reason: "test" });
|
||||
});
|
||||
|
||||
it("sends only the authoritative wake envelope when resuming an existing provider session", async () => {
|
||||
it("resends system instructions without repeating the task envelope on recovery", async () => {
|
||||
await chmod(fixture, 0o755);
|
||||
const root = await mkdtemp(
|
||||
join(tmpdir(), "paperclip-opencode-resume-context-"),
|
||||
@@ -1553,7 +1595,7 @@ describe("OpenCodeServerDriver", () => {
|
||||
],
|
||||
});
|
||||
expect(submittedPrompt).not.toHaveProperty("tools");
|
||||
expect(submittedPrompt).not.toHaveProperty("system");
|
||||
expect(submittedPrompt).toHaveProperty("system", "large original system context");
|
||||
await recovered!.session!.close({ reason: "recovery-test" });
|
||||
});
|
||||
|
||||
@@ -1942,9 +1984,11 @@ describe("OpenCodeServerDriver", () => {
|
||||
"*": permissionMode,
|
||||
external_directory: { "*": "deny", [`${workspace}/**`]: "allow" },
|
||||
});
|
||||
expect(config.autoupdate).toBe(false);
|
||||
expect(config.provider.openrouter.models).toHaveProperty(
|
||||
"deepseek/deepseek-v4-flash-0731",
|
||||
);
|
||||
expect(config.provider.openrouter.models["deepseek/deepseek-v4-flash-0731"].options).toBeUndefined();
|
||||
expect(config.permission.paperclip_finish).toBeUndefined();
|
||||
expect(config.permission["paperclip_*"]).toBe("allow");
|
||||
await session.close({ reason: "permission mode test complete" });
|
||||
@@ -2440,6 +2484,9 @@ describe("OpenCodeServerDriver", () => {
|
||||
afterSpawn: () => {
|
||||
commandLifecycle.push("after");
|
||||
},
|
||||
afterExit: () => {
|
||||
commandLifecycle.push("exit");
|
||||
},
|
||||
},
|
||||
fetch: async (input, init) => {
|
||||
if (
|
||||
@@ -2465,9 +2512,31 @@ describe("OpenCodeServerDriver", () => {
|
||||
expect(spawns.length).toBeGreaterThanOrEqual(2);
|
||||
expect(spawns.length).toBeLessThanOrEqual(3);
|
||||
expect(commandLifecycle).toEqual(
|
||||
Array.from({ length: spawns.length }, () => ["before", "after"]).flat(),
|
||||
[...Array.from({ length: spawns.length - 1 }, () => ["before", "after", "exit"]).flat(), "before", "after"],
|
||||
);
|
||||
await session.close({ reason: "test" });
|
||||
expect(commandLifecycle).toEqual(
|
||||
Array.from({ length: spawns.length }, () => ["before", "after", "exit"]).flat(),
|
||||
);
|
||||
});
|
||||
|
||||
it("releases the executable binding when spawning fails before health checks", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-spawn-error-"));
|
||||
roots.push(root);
|
||||
const lifecycle: string[] = [];
|
||||
const driver = new OpenCodeServerDriver({
|
||||
model: "openrouter/deepseek/deepseek-v4-flash-0731",
|
||||
runtimeDirectory: root,
|
||||
command: join(root, "missing-executable"),
|
||||
commandLifecycle: {
|
||||
beforeSpawn: () => { lifecycle.push("before"); },
|
||||
afterSpawn: () => { lifecycle.push("spawned"); },
|
||||
afterExit: () => { lifecycle.push("released"); },
|
||||
},
|
||||
});
|
||||
await expect(driver.openSession({ runId: "spawn-error", normalizedSessionId: "spawn-error",
|
||||
workingDirectory: root })).rejects.toThrow("ENOENT");
|
||||
expect(lifecycle).toEqual(["before", "released"]);
|
||||
});
|
||||
|
||||
it("reports startup exit details with stderr captured after spawn and redacted", async () => {
|
||||
|
||||
@@ -19,6 +19,8 @@ import { getCACertificates } from "node:tls";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path";
|
||||
|
||||
import { parseOpenCodeReasoningMode, type OpenCodeReasoningMode } from "./reasoning-mode.js";
|
||||
|
||||
import {
|
||||
CODEX_SKILLLESS_BASE_INSTRUCTIONS,
|
||||
createCodexTaskEnvelope,
|
||||
@@ -176,6 +178,7 @@ export interface OpenCodeServerDriverOptions {
|
||||
commandLifecycle?: {
|
||||
beforeSpawn(): void;
|
||||
afterSpawn(): void;
|
||||
afterExit?(): void;
|
||||
};
|
||||
runtimeDirectory: string;
|
||||
systemInstructions?: string;
|
||||
@@ -479,6 +482,7 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
Array<Record<string, unknown>>
|
||||
>();
|
||||
readonly #partText = new Map<string, string>();
|
||||
readonly #streamingParts = new Map<string, Record<string, unknown>>();
|
||||
readonly #completedTextPartIds = new Set<string>();
|
||||
readonly #completedReasoningPartIds = new Set<string>();
|
||||
readonly #completedTextParts: Array<{
|
||||
@@ -640,12 +644,22 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
.finally(() => this.#pumpEvents());
|
||||
}
|
||||
|
||||
supportsTurnReasoning(): boolean {
|
||||
return this.#model.startsWith("openrouter/");
|
||||
}
|
||||
|
||||
async startTurn(input: {
|
||||
message: NativeUserMessage;
|
||||
reasoningMode?: OpenCodeReasoningMode;
|
||||
}): Promise<{ turnId: string }> {
|
||||
if (this.#activeTurnId !== null)
|
||||
throw new Error("OpenCode session already has an active turn");
|
||||
const reasoningMode = parseOpenCodeReasoningMode(input.reasoningMode);
|
||||
if (input.reasoningMode !== undefined && !this.#model.startsWith("openrouter/")) {
|
||||
throw new Error("Per-turn OpenCode reasoning is supported only for OpenRouter models");
|
||||
}
|
||||
const turnId = `turn-${randomBytes(12).toString("hex")}`;
|
||||
this.#streamingParts.clear();
|
||||
this.#activeTurnId = turnId;
|
||||
this.#emit("turn.submitted", {
|
||||
envelopeSchema: this.#taskEnvelope.schema,
|
||||
@@ -655,12 +669,9 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
this.#emit("turn.started", { status: "inProgress" }, { turnId });
|
||||
const [providerID, ...modelParts] = this.#model.split("/");
|
||||
const modelID = modelParts.join("/");
|
||||
// A resumed OpenCode provider session already retains the original system
|
||||
// instructions and task envelope in its conversation. Repeating both on
|
||||
// every Paperclip continuation can overflow smaller context windows and
|
||||
// OpenCode then completes with `finish: unknown` and zero tokens. The
|
||||
// native model envelope still carries the authoritative wake delta,
|
||||
// interaction responses, completion contract, and current issue context.
|
||||
// Keep the task envelope only on the initial task-mode wake. OpenCode
|
||||
// rebuilds system instructions from the latest user message, so system
|
||||
// instructions must still accompany every prompt (including recovery).
|
||||
const prompt = this.#sendFullContext
|
||||
? JSON.stringify({
|
||||
task: this.#taskEnvelope,
|
||||
@@ -682,12 +693,15 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
// at this HTTP boundary.
|
||||
providerID,
|
||||
modelID,
|
||||
// Always select the empty default variant on an ordinary turn so a
|
||||
// previous disabled turn cannot change this turn's provider defaults.
|
||||
...(providerID === "openrouter"
|
||||
? { variant: reasoningMode === "disabled" ? "paperclip-no-reasoning" : "paperclip-default" }
|
||||
: {}),
|
||||
// Keep question enabled in the isolated config. OpenCode 1.18.32
|
||||
// turns a prompt's deprecated `tools` map into replacement session
|
||||
// permissions, so a sparse override here discards the session policy.
|
||||
...(this.#sendFullContext
|
||||
? { system: this.#systemInstructions }
|
||||
: {}),
|
||||
system: this.#systemInstructions,
|
||||
parts: [{ type: "text", text: prompt }],
|
||||
}),
|
||||
},
|
||||
@@ -1470,9 +1484,11 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
const event = record(value);
|
||||
const properties = record(event.properties);
|
||||
const type = text(event.type);
|
||||
const eventId = text(event.id, canonicalJson(value));
|
||||
if (this.#seenProviderEvents.has(eventId)) return;
|
||||
this.#seenProviderEvents.add(eventId);
|
||||
// Id-less deltas are additive: two identical chunks can be legitimate
|
||||
// adjacent tokens. Content deduplication would silently lose the second.
|
||||
const eventId = text(event.id) || (type === "message.part.delta" ? null : canonicalJson(value));
|
||||
if (eventId !== null && this.#seenProviderEvents.has(eventId)) return;
|
||||
if (eventId !== null) this.#seenProviderEvents.add(eventId);
|
||||
if (this.#seenProviderEvents.size > 10_000)
|
||||
this.#seenProviderEvents.delete(
|
||||
this.#seenProviderEvents.values().next().value!,
|
||||
@@ -1565,8 +1581,16 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (type === "message.part.updated") {
|
||||
const part = record(properties.part);
|
||||
if (type === "message.part.updated" || type === "message.part.delta") {
|
||||
let part = record(properties.part);
|
||||
if (type === "message.part.delta") {
|
||||
const previous = this.#streamingParts.get(text(properties.partID));
|
||||
// A delta has no role or part type. Only stream text for an observed
|
||||
// part with matching message identity; never infer assistant identity.
|
||||
if (!previous || properties.field !== "text" || typeof properties.delta !== "string"
|
||||
|| text(previous.messageID, text(previous.messageId)) !== text(properties.messageID)) return;
|
||||
part = { ...previous, text: text(previous.text) + properties.delta };
|
||||
}
|
||||
const messageId = text(part.messageID, text(part.messageId));
|
||||
if (!messageId) return;
|
||||
// Resolve the turn this message actually belongs to, not whichever
|
||||
@@ -1574,11 +1598,20 @@ class OpenCodeHarnessSession implements HarnessSession {
|
||||
// must stay attributed to the turn that created that message.
|
||||
const owningTurnId = this.#messageTurnIds.get(messageId) ?? turnId;
|
||||
if (!owningTurnId) return;
|
||||
if (owningTurnId === turnId && text(part.id) && ["text", "reasoning"].includes(text(part.type))) {
|
||||
this.#streamingParts.set(text(part.id), part);
|
||||
}
|
||||
const role = this.#messageRoles.get(messageId);
|
||||
if (role === "assistant") this.#emitAssistantPart(part, owningTurnId);
|
||||
else if (role === undefined) {
|
||||
const pending = this.#pendingMessageParts.get(messageId) ?? [];
|
||||
if (pending.length < 100) pending.push(part);
|
||||
// Text deltas are cumulative by this point. Keep the newest snapshot
|
||||
// (including completion metadata), not the first 100 token chunks.
|
||||
const previousIndex = ["text", "reasoning"].includes(text(part.type)) && text(part.id)
|
||||
? pending.findIndex(candidate => candidate.id === part.id && candidate.type === part.type)
|
||||
: -1;
|
||||
if (previousIndex >= 0) pending[previousIndex] = part;
|
||||
else if (pending.length < 100) pending.push(part);
|
||||
this.#pendingMessageParts.set(messageId, pending);
|
||||
}
|
||||
return;
|
||||
@@ -2233,12 +2266,23 @@ async function startRuntime(input: {
|
||||
input.trace?.addSensitiveValues([providerProxy.token]);
|
||||
}
|
||||
let child: ChildProcess | undefined;
|
||||
let launchPrepared = false;
|
||||
const releaseExecutable = () => {
|
||||
if (!launchPrepared) return;
|
||||
launchPrepared = false;
|
||||
try {
|
||||
input.options.commandLifecycle?.afterExit?.();
|
||||
} catch (error) {
|
||||
input.options.onDiagnostic?.(redact(`OpenCode executable cleanup failed: ${String(error)}`, sensitiveValues));
|
||||
}
|
||||
};
|
||||
try {
|
||||
const config = {
|
||||
$schema: "https://opencode.ai/config.json",
|
||||
model: input.options.model,
|
||||
small_model: input.options.model,
|
||||
share: "disabled",
|
||||
autoupdate: false,
|
||||
// The configured entry is already composed exactly once into the session
|
||||
// system prompt; siblings remain available through the read-only root.
|
||||
instructions: [],
|
||||
@@ -2257,7 +2301,15 @@ async function startRuntime(input: {
|
||||
},
|
||||
} : {}),
|
||||
models: {
|
||||
[providerModelId]: { name: providerModelId },
|
||||
[providerModelId]: {
|
||||
name: providerModelId,
|
||||
...(modelProvider === "openrouter"
|
||||
? { variants: {
|
||||
"paperclip-default": {},
|
||||
"paperclip-no-reasoning": { reasoning: { enabled: false } },
|
||||
} }
|
||||
: {}),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
@@ -2307,6 +2359,11 @@ async function startRuntime(input: {
|
||||
XDG_DATA_HOME: dataHome,
|
||||
XDG_CACHE_HOME: cacheHome,
|
||||
OPENCODE_DISABLE_PROJECT_CONFIG: "true",
|
||||
// The runner pins the executable, model and assigned MCP tools. A
|
||||
// fresh isolated cache must not fetch a catalog or install unrelated
|
||||
// default plugins before it can submit the first prompt.
|
||||
OPENCODE_DISABLE_MODELS_FETCH: "true",
|
||||
OPENCODE_DISABLE_DEFAULT_PLUGINS: "true",
|
||||
OPENCODE_SERVER_USERNAME: username,
|
||||
OPENCODE_SERVER_PASSWORD: password,
|
||||
...(providerProxy ? { NO_PROXY: [input.options.environment?.no_proxy ?? input.options.environment?.NO_PROXY, "127.0.0.1", "localhost"].filter(Boolean).join(",") } : {}),
|
||||
@@ -2319,6 +2376,7 @@ async function startRuntime(input: {
|
||||
stdio[input.options.commandFd] = input.options.commandFd;
|
||||
}
|
||||
input.options.commandLifecycle?.beforeSpawn();
|
||||
launchPrepared = true;
|
||||
child = spawn(
|
||||
input.options.command ?? resolvePinnedOpenCodeCommand(),
|
||||
["serve", "--hostname", "127.0.0.1", "--port", String(port)],
|
||||
@@ -2329,6 +2387,7 @@ async function startRuntime(input: {
|
||||
detached: globalThis.process.platform !== "win32" && isolateProcessGroup,
|
||||
},
|
||||
);
|
||||
child.once("exit", releaseExecutable);
|
||||
if (child.pid !== undefined) {
|
||||
try {
|
||||
input.options.commandLifecycle?.afterSpawn();
|
||||
@@ -2425,6 +2484,7 @@ async function startRuntime(input: {
|
||||
} catch {
|
||||
providerChild.kill("SIGKILL");
|
||||
}
|
||||
await waitForExit(providerChild, 2_000);
|
||||
}
|
||||
await rm(join(configHome, "opencode", "opencode.json"), {
|
||||
force: true,
|
||||
@@ -2441,6 +2501,8 @@ async function startRuntime(input: {
|
||||
await providerProxy?.close();
|
||||
await bridge.close().catch(() => {});
|
||||
child?.kill("SIGKILL");
|
||||
if (child?.pid) await waitForExit(child, 2_000);
|
||||
if (!child?.pid || child.exitCode !== null || child.signalCode !== null) releaseExecutable();
|
||||
await rm(join(configHome, "opencode", "opencode.json"), {
|
||||
force: true,
|
||||
}).catch(() => undefined);
|
||||
|
||||
@@ -30,6 +30,7 @@ describe("prepared OpenCode context transport", () => {
|
||||
runtimeDirectory: root,
|
||||
command: fixture,
|
||||
conversationMode: "prepared",
|
||||
systemInstructions: "Prepared agent instructions.",
|
||||
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
|
||||
fetch: async (input, init) => {
|
||||
if (String(input).endsWith("/prompt_async"))
|
||||
@@ -45,6 +46,7 @@ describe("prepared OpenCode context transport", () => {
|
||||
});
|
||||
await session.startTurn({
|
||||
message: { role: "user", text: "prepared OpenCode initial wake" },
|
||||
reasoningMode: "disabled",
|
||||
});
|
||||
for await (const event of session.events()) {
|
||||
if (event.eventType === "turn.completed") break;
|
||||
@@ -67,6 +69,8 @@ describe("prepared OpenCode context transport", () => {
|
||||
[{ type: "text", text: "completion-only OpenCode continuation" }],
|
||||
]);
|
||||
expect(JSON.stringify(submitted)).not.toContain('"task"');
|
||||
expect(submitted.map(body => body.variant)).toEqual(["paperclip-no-reasoning", "paperclip-default"]);
|
||||
expect(submitted.map(body => body.system)).toEqual(["Prepared agent instructions.", "Prepared agent instructions."]);
|
||||
await recovered!.session!.close({ reason: "prepared-test-recovery" });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,63 @@
|
||||
import { chmod, mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { OpenCodeServerDriver } from "./opencode-server-driver.js";
|
||||
import { parseOpenCodeReasoningMode } from "./reasoning-mode.js";
|
||||
import { openCodeProxyTaskEnvelope } from "../../cli/opencode-proxy-task-envelope.js";
|
||||
|
||||
describe("per-turn OpenCode reasoning", () => {
|
||||
it.each([undefined, "default"])("preserves provider defaults for %s", (value) => {
|
||||
expect(parseOpenCodeReasoningMode(value)).toBe("default");
|
||||
});
|
||||
it.each(["", "false", null, false, 0])("rejects malformed mode %s", (value) => {
|
||||
expect(() => parseOpenCodeReasoningMode(value)).toThrow("turn.reasoningMode");
|
||||
});
|
||||
it("selects reasoning independently on successive turns and sends instructions once", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-turn-reasoning-"));
|
||||
const fixture = resolve("test/fixtures/fake-opencode-server.mjs");
|
||||
await chmod(fixture, 0o755);
|
||||
const submitted: Array<{ variant?: string; system?: string; parts: Array<{ text: string }> }> = [];
|
||||
const instructions = "Unique shared agent instructions. Keep the requested scope.";
|
||||
const driver = new OpenCodeServerDriver({
|
||||
model: "openrouter/deepseek/deepseek-v4-flash-0731",
|
||||
runtimeDirectory: root,
|
||||
command: fixture,
|
||||
systemInstructions: instructions,
|
||||
taskEnvelope: openCodeProxyTaskEnvelope({ baseInstructions: instructions,
|
||||
completionContract: { revision: "17", criterionIds: ["objective"] } }),
|
||||
environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "fixture-key" },
|
||||
fetch: async (input, init) => {
|
||||
if (String(input).endsWith("/prompt_async")) submitted.push(JSON.parse(String(init?.body)));
|
||||
return fetch(input, init);
|
||||
},
|
||||
});
|
||||
let session: Awaited<ReturnType<OpenCodeServerDriver["openSession"]>> | undefined;
|
||||
try {
|
||||
session = await driver.openSession({ runId: "turn-reasoning", normalizedSessionId: "reasoning", workingDirectory: root });
|
||||
await expect(session.startTurn({ message: { role: "user", text: "invalid" }, reasoningMode: "bad" as never })).rejects.toThrow("turn.reasoningMode");
|
||||
expect(submitted).toHaveLength(0);
|
||||
for (const reasoningMode of [undefined, "disabled", undefined] as const) {
|
||||
await session.startTurn({ message: { role: "user", text: "Reply Hi." }, reasoningMode });
|
||||
for await (const event of session.events()) if (event.eventType === "turn.completed") break;
|
||||
}
|
||||
expect(submitted.map(body => body.variant)).toEqual(["paperclip-default", "paperclip-no-reasoning", "paperclip-default"]);
|
||||
expect(submitted[0]!.system).toBe(instructions);
|
||||
for (const body of submitted) {
|
||||
expect(body.system).toBe(instructions);
|
||||
expect(JSON.stringify(body).split(instructions)).toHaveLength(2);
|
||||
}
|
||||
const envelope = JSON.parse(submitted[0]!.parts[0].text);
|
||||
expect(envelope.task.completionContract.revision).toBe("17");
|
||||
expect(envelope.task.constraints).toEqual(["Use Paperclip MCP tools for semantic operations."]);
|
||||
const config = JSON.parse(await readFile(join(root, "reasoning/config/opencode/opencode.json"), "utf8"));
|
||||
expect(config.provider.openrouter.models["deepseek/deepseek-v4-flash-0731"]).toMatchObject({
|
||||
variants: { "paperclip-default": {}, "paperclip-no-reasoning": { reasoning: { enabled: false } } },
|
||||
});
|
||||
expect(config.provider.openrouter.models["deepseek/deepseek-v4-flash-0731"].options).toBeUndefined();
|
||||
} finally {
|
||||
await session?.close();
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,8 @@
|
||||
export type OpenCodeReasoningMode = "default" | "disabled";
|
||||
|
||||
/** Opt-in only. Omitting the setting preserves the provider's own defaults. */
|
||||
export function parseOpenCodeReasoningMode(value: unknown): OpenCodeReasoningMode {
|
||||
if (value === undefined || value === "default") return "default";
|
||||
if (value === "disabled") return "disabled";
|
||||
throw new Error("turn.reasoningMode must be default or disabled");
|
||||
}
|
||||
@@ -1514,7 +1514,7 @@ export class CapabilityLiveSession {
|
||||
async sendMessage(
|
||||
message: string,
|
||||
/** Launch-only diagnostics may opt out; qualification campaigns must not. */
|
||||
options: { allowMissingUsage?: boolean } = {},
|
||||
options: { allowMissingUsage?: boolean; reasoningMode?: "default" | "disabled" } = {},
|
||||
): Promise<CapabilityLiveTurnResult> {
|
||||
const value = message.trim();
|
||||
if (value.length === 0) throw new Error("Capability live messages cannot be empty");
|
||||
@@ -1617,6 +1617,7 @@ export class CapabilityLiveSession {
|
||||
permissions: CODEX_PERMISSION_PROFILE,
|
||||
runtimeWorkspaceRoots: [this.#config.workingDirectory],
|
||||
input: [userInput(value)],
|
||||
...(options.reasoningMode === undefined ? {} : { reasoningMode: options.reasoningMode }),
|
||||
});
|
||||
} catch (error) {
|
||||
if (this.#pendingTurnAdmission !== admission) {
|
||||
|
||||
@@ -7951,7 +7951,10 @@ it.each([true, false])("preserves prepared OpenCode cleanup errors (primary fail
|
||||
expect((failure as Error).message).not.toContain("fixture-secret");
|
||||
});
|
||||
|
||||
it("preserves prepared input and completion feedback through runnerd and the real OpenCode proxy boundary", async () => {
|
||||
it.each([
|
||||
{ model: "openrouter/deepseek/deepseek-v4-flash-0731", supportsReasoning: true },
|
||||
{ model: "anthropic/claude-sonnet-4", supportsReasoning: false },
|
||||
])("preserves prepared input and completion feedback through runnerd and the real OpenCode proxy boundary ($model)", async ({ model, supportsReasoning }) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-prepared-opencode-"));
|
||||
// GitHub-hosted Linux toolcache Node can be group-writable, unlike the AWS
|
||||
// fleet. Qualify an owned copy with strict permissions, never chmod the host
|
||||
@@ -8006,7 +8009,7 @@ it("preserves prepared input and completion feedback through runnerd and the rea
|
||||
const driver = new CodexAppServerDriver({
|
||||
taskEnvelope: task,
|
||||
conversationMode: "prepared",
|
||||
model: "openrouter/deepseek/deepseek-v4-flash-0731",
|
||||
model,
|
||||
transportFactory: () => bundle.transport,
|
||||
workingDirectoryAuthority: "remote_runner",
|
||||
environment: { PAPERCLIP_WORKSPACE_CWD: root },
|
||||
@@ -8025,7 +8028,15 @@ it("preserves prepared input and completion feedback through runnerd and the rea
|
||||
await withPreparedOpenCodeCleanup({
|
||||
run: async () => {
|
||||
session = await driver.openSession({ runId: "prepared-opencode", normalizedSessionId: "prepared-opencode", workingDirectory: root });
|
||||
await session.startTurn({ message: { role: "user", text: prepared } });
|
||||
expect(bundle.transport.supportsTurnReasoning?.()).toBe(supportsReasoning);
|
||||
const firstTurn = { message: { role: "user" as const, text: prepared } };
|
||||
if (!supportsReasoning) {
|
||||
await expect(session.startTurn({ ...firstTurn, reasoningMode: "disabled" }))
|
||||
.rejects.toThrow("Per-turn reasoning is not supported");
|
||||
await expect(bundle.transport.request("turn/start", { input: [{ type: "text", text: prepared }], reasoningMode: "disabled" }))
|
||||
.rejects.toThrow("requires an OpenCode/OpenRouter model");
|
||||
}
|
||||
await session.startTurn({ ...firstTurn, ...(supportsReasoning ? { reasoningMode: "disabled" as const } : {}) });
|
||||
const events: PrpEvent[] = [];
|
||||
for await (const event of session.events()) {
|
||||
events.push(event);
|
||||
@@ -8037,10 +8048,16 @@ it("preserves prepared input and completion feedback through runnerd and the rea
|
||||
expect(sessionRoots).toHaveLength(1);
|
||||
const requests = (await readFile(join(runtime, sessionRoots[0]!.name, "data/fake-prompt-requests.ndjson"), "utf8")).trim().split("\n").map((line) => JSON.parse(line));
|
||||
expect(requests.map((request) => request.parts)).toEqual([[{ type: "text", text: prepared }]]);
|
||||
expect(requests[0].variant).toBe(supportsReasoning ? "paperclip-no-reasoning" : undefined);
|
||||
const outcomes = JSON.parse(await readFile(join(runtime, sessionRoots[0]!.name, "data/fake-completion-feedback.json"), "utf8"));
|
||||
expect(outcomes).toHaveLength(2);
|
||||
expect(outcomes[0].result).toMatchObject({ isError: true, content: [{ text: expect.stringContaining("required document link") }] });
|
||||
expect(outcomes[1].result).toMatchObject({ content: [{ text: expect.stringContaining(feedback) }] });
|
||||
await session.startTurn({ message: { role: "user", text: "Reply Hi." } });
|
||||
for await (const event of session.events()) if (event.eventType === "turn.completed") break;
|
||||
const allRequests = (await readFile(join(runtime, sessionRoots[0]!.name, "data/fake-prompt-requests.ndjson"), "utf8")).trim().split("\n").map((line) => JSON.parse(line));
|
||||
expect(allRequests.map(request => request.variant)).toEqual(supportsReasoning
|
||||
? ["paperclip-no-reasoning", "paperclip-default"] : [undefined, undefined]);
|
||||
},
|
||||
closeSession: async () => { await session?.close(); },
|
||||
closeTransport: () => bundle.transport.close(),
|
||||
|
||||
@@ -3538,6 +3538,20 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
#providerIdentity: Record<string, unknown> | null = null;
|
||||
#turnControls: NativeTurnControlCapabilities = { steering: false, queuedFollowUp: false };
|
||||
|
||||
supportsTurnReasoning(): boolean {
|
||||
if (this.options.provider !== "opencode") return false;
|
||||
// Use the durable selected model on both fresh and recovered sessions.
|
||||
// Older states may retain the provider only in the preparation command.
|
||||
const state = this.#core?.store.state;
|
||||
const template = this.#runAttachTemplate ?? state?.runAttachTemplate
|
||||
?? state?.commands.findLast(command =>
|
||||
(command.type === "run.prepare" || command.type === "run.attach")
|
||||
&& record(command.payload).provider !== undefined,
|
||||
)?.payload;
|
||||
const model = record(record(template).provider).model;
|
||||
return typeof model === "string" && model.startsWith("openrouter/");
|
||||
}
|
||||
|
||||
turnControlCapabilities(): NativeTurnControlCapabilities | null {
|
||||
if (this.options.provider !== "acpx") return null;
|
||||
if (this.#closed || this.#failure) return { steering: false, queuedFollowUp: false };
|
||||
@@ -5883,6 +5897,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
.filter((item) => item.type !== "skill")
|
||||
.map((item) => (typeof item.text === "string" ? item.text : ""))
|
||||
.join("\n");
|
||||
const reasoningMode = params.reasoningMode;
|
||||
if (reasoningMode !== undefined && (
|
||||
!this.supportsTurnReasoning()
|
||||
|| (reasoningMode !== "default" && reasoningMode !== "disabled")
|
||||
)) throw new Error("turn.reasoningMode requires an OpenCode/OpenRouter model and must be default or disabled");
|
||||
const skills = resolveRunnerdCodexSkillInputs(
|
||||
input.filter((item) => item.type === "skill"),
|
||||
this.options.runtimeContext ?? null,
|
||||
@@ -5916,6 +5935,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
"turn.start",
|
||||
{
|
||||
text: message,
|
||||
...(reasoningMode === undefined ? {} : { reasoningMode }),
|
||||
...(skills.length ? { skills } : {}),
|
||||
turnId: pendingTurnId,
|
||||
},
|
||||
|
||||
@@ -99,6 +99,8 @@ await writeFile(
|
||||
home: process.env.HOME,
|
||||
configHome: process.env.XDG_CONFIG_HOME,
|
||||
projectConfigDisabled: process.env.OPENCODE_DISABLE_PROJECT_CONFIG,
|
||||
modelsFetchDisabled: process.env.OPENCODE_DISABLE_MODELS_FETCH,
|
||||
defaultPluginsDisabled: process.env.OPENCODE_DISABLE_DEFAULT_PLUGINS,
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -400,8 +402,7 @@ const server = createServer(async (request, response) => {
|
||||
const promptPayload = JSON.parse(Buffer.concat(chunks).toString("utf8"));
|
||||
await appendFile(join(process.env.XDG_DATA_HOME, "fake-prompt-requests.ndjson"), JSON.stringify(promptPayload) + "\n");
|
||||
if (
|
||||
promptPayload.providerID !== "openrouter" ||
|
||||
promptPayload.modelID !== "deepseek/deepseek-v4-flash-0731" ||
|
||||
`${promptPayload.providerID}/${promptPayload.modelID}` !== runtimeConfig.model ||
|
||||
"model" in promptPayload
|
||||
) {
|
||||
return json(response, 400, {
|
||||
@@ -427,11 +428,56 @@ const server = createServer(async (request, response) => {
|
||||
},
|
||||
},
|
||||
});
|
||||
emit({
|
||||
type: "message.part.delta",
|
||||
properties: {
|
||||
sessionID: session.id,
|
||||
messageID: "message-late-source",
|
||||
partID: "part-late-straggler",
|
||||
field: "text",
|
||||
delta: "late straggler text must not reach the next turn",
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
setTimeout(async () => {
|
||||
await callFirstPaperclipTool();
|
||||
const parsedPrompt = parsedPromptText(promptPayload);
|
||||
if (String(parsedPrompt.message ?? "").includes("incremental-stream")) {
|
||||
const messageID = `stream-message-${promptTurnSeq}`;
|
||||
const part = { id: `stream-part-${promptTurnSeq}`, messageID, sessionID: session.id, type: "text", text: "" };
|
||||
const reasoning = { ...part, id: `${part.id}-reasoning`, type: "reasoning" };
|
||||
const update = (value) => emit({ type: "message.part.updated", properties: { sessionID: session.id, part: value } });
|
||||
const delta = (value, text, extra = {}) => emit({ type: "message.part.delta", properties: { sessionID: session.id, messageID, partID: value.id, field: "text", delta: text, ...extra } });
|
||||
const announceRole = () => emit({ type: "message.updated", properties: { sessionID: session.id, info: { id: messageID, role: "assistant" } } });
|
||||
const roleFirst = String(parsedPrompt.message).includes("role-first");
|
||||
const longResponse = String(parsedPrompt.message).includes("role-last-long");
|
||||
if (roleFirst) announceRole();
|
||||
// Parts and deltas can precede the message's role announcement.
|
||||
update(part);
|
||||
update(reasoning);
|
||||
delta(reasoning, "Thinking");
|
||||
delta(part, "Hi ");
|
||||
const emojiCount = longResponse ? 120 : 2;
|
||||
for (let i = 0; i < emojiCount; i++) delta(part, "👋");
|
||||
delta(part, "wrong message", { messageID: "unrelated-message" });
|
||||
delta(part, "wrong session", { sessionID: "unrelated-session" });
|
||||
delta(part, "wrong field", { field: "output" });
|
||||
delta(part, "unknown part", { partID: "unknown-part" });
|
||||
const finalPart = { ...part, text: `Hi ${"👋".repeat(emojiCount)}`, time: { start: 1, end: 2 } };
|
||||
if (longResponse) update(finalPart);
|
||||
if (!roleFirst) announceRole();
|
||||
// The consumer must observe a delta before the completed snapshot
|
||||
// exists. This handshake makes the streaming regression deterministic.
|
||||
for (let i = 0; i < 500; i++) {
|
||||
if (await readFile(join(process.env.XDG_DATA_HOME, "release-stream"), "utf8").catch(() => "") === "ready") break;
|
||||
await new Promise(resolve => setTimeout(resolve, 10));
|
||||
}
|
||||
await writeFile(join(process.env.XDG_DATA_HOME, "stream-released"), "ready");
|
||||
update(finalPart);
|
||||
emit({ type: "session.idle", properties: { sessionID: session.id } });
|
||||
return;
|
||||
}
|
||||
if (String(parsedPrompt.message ?? "").includes("native-question")) {
|
||||
pendingQuestion = nativeQuestion();
|
||||
emit({
|
||||
|
||||
@@ -34,6 +34,8 @@ import {
|
||||
import { toolAccessService } from "../services/tool-access.js";
|
||||
import { createToolGatewayService } from "../services/tool-gateway.js";
|
||||
import { browserUseViewports } from "../services/browser-use-viewport.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { registerBrowserUseCleanup } from "../services/browser-use-work.js";
|
||||
import { browserUseService } from "../services/browser-use.js";
|
||||
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "../services/connector-runtime.js";
|
||||
import { registerAssignedMcpGateway } from "../services/native-runtime/assigned-mcp-tools.js";
|
||||
@@ -313,6 +315,47 @@ const actor = { actorType: "user" as const, actorId: "browser-reviewer" };
|
||||
},
|
||||
};
|
||||
}
|
||||
it("wakes the app worker from request-service commits and restores durable poll and lease deadlines", async () => {
|
||||
const f = await fixture();
|
||||
const background = browserUseService(db, f.request);
|
||||
const sweep = vi.spyOn(background, "sweep");
|
||||
const onError = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
try {
|
||||
await registerBrowserUseCleanup(coordinator, background, () => true).ready;
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
expect(await background.nextSweepAt()).toBeNull();
|
||||
const started = await f.service.execute(f.binding, f.grant, randomUUID(), "browser_start", { task: "Read example.com" }) as { sessionId: string };
|
||||
await vi.waitFor(() => expect(sweep.mock.calls.length).toBeGreaterThan(1));
|
||||
await coordinator.stop();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
|
||||
const poll = new Date(Date.now() + 60_000);
|
||||
const lease = new Date(Date.now() + 180_000);
|
||||
await db.update(browserUseSessions).set({ nextPollAt: poll, leaseUntil: lease })
|
||||
.where(eq(browserUseSessions.id, started.sessionId));
|
||||
expect(await background.nextSweepAt()).toBe(lease.getTime());
|
||||
const restarted = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
try {
|
||||
await registerBrowserUseCleanup(restarted, browserUseService(db, f.request), () => true).ready;
|
||||
expect(restarted.nextWakeAt()).toBe(lease.getTime());
|
||||
} finally { await restarted.stop(); }
|
||||
await db.update(browserUseSessions).set({ leaseUntil: new Date(0) })
|
||||
.where(eq(browserUseSessions.id, started.sessionId));
|
||||
expect(await background.nextSweepAt()).toBe(poll.getTime());
|
||||
await db.update(browserUseSessions).set({ leaseUntil: null })
|
||||
.where(eq(browserUseSessions.id, started.sessionId));
|
||||
expect(await background.nextSweepAt()).toBe(poll.getTime());
|
||||
const [session] = await db.select().from(browserUseSessions).where(eq(browserUseSessions.id, started.sessionId));
|
||||
await f.service.control(session, "end");
|
||||
expect(await background.nextSweepAt()).toBeLessThan(poll.getTime());
|
||||
await db.update(browserUseSessions).set({ status: "closed" }).where(eq(browserUseSessions.id, started.sessionId));
|
||||
expect(await background.nextSweepAt()).toBeNull();
|
||||
} finally {
|
||||
await coordinator.stop();
|
||||
await db.update(browserUseSessions).set({ status: "closed" }).where(eq(browserUseSessions.companyId, f.company.id));
|
||||
}
|
||||
});
|
||||
it("keeps Cloud instructions out of universal skills and unassigned runtime overlays", async () => {
|
||||
const skills = await listPaperclipSkillEntries(fileURLToPath(new URL("../", import.meta.url)), [fileURLToPath(new URL("../../../skills", import.meta.url))]);
|
||||
expect(skills.some(skill => skill.runtimeName === "paperclip")).toBe(true);
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { registerBrowserUseCleanup } from "../services/browser-use-work.js";
|
||||
|
||||
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
|
||||
afterEach(async () => {
|
||||
for (const coordinator of coordinators.splice(0)) await coordinator.stop();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
function setup() {
|
||||
vi.useFakeTimers();
|
||||
const execute = vi.fn(async () => [{ xid: "42", status: "committed" }]);
|
||||
const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>) {
|
||||
return callback({ execute });
|
||||
} });
|
||||
const enabled = vi.fn(() => true), primary = vi.fn(() => true), onError = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({
|
||||
owner, canRun: () => enabled() && primary(), canReconcile: primary, onError,
|
||||
});
|
||||
coordinators.push(coordinator);
|
||||
const service = { sweep: vi.fn(async () => {}), nextSweepAt: vi.fn(async (): Promise<number | null> => null) };
|
||||
const start = () => registerBrowserUseCleanup(coordinator, service, enabled);
|
||||
const commit = () => owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.browser));
|
||||
return { service, coordinator, start, commit, enabled, primary, onError };
|
||||
}
|
||||
describe("browser cleanup scheduling", () => {
|
||||
it("stays quiet after an empty startup, wakes on commit, and stops after the final session closes", async () => {
|
||||
const s = setup();
|
||||
await s.start().ready;
|
||||
await vi.advanceTimersByTimeAsync(3_600_000);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(1);
|
||||
expect(s.service.nextSweepAt).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
const due = Date.now() + 15_000;
|
||||
s.service.nextSweepAt.mockResolvedValue(due);
|
||||
await s.commit();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(2);
|
||||
expect(s.coordinator.nextWakeAt()).toBe(due);
|
||||
await vi.advanceTimersByTimeAsync(14_998);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(2);
|
||||
s.service.nextSweepAt.mockResolvedValue(null);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(3);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("restores a future lease/retry deadline at startup and lets a control commit preempt it", async () => {
|
||||
const s = setup();
|
||||
const due = Date.now() + 180_000;
|
||||
s.service.nextSweepAt.mockResolvedValue(due);
|
||||
await s.start().ready;
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(1);
|
||||
await s.commit();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(2);
|
||||
expect(s.coordinator.nextWakeAt()).toBe(due);
|
||||
});
|
||||
it("retries a failed scan and bounds already-due batches", async () => {
|
||||
const s = setup();
|
||||
s.service.sweep.mockRejectedValueOnce(new Error("database unavailable"));
|
||||
await s.start().ready;
|
||||
expect(s.onError).toHaveBeenCalledTimes(1);
|
||||
s.service.nextSweepAt.mockResolvedValue(Date.now() - 1000);
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(2);
|
||||
expect(s.coordinator.nextWakeAt()).toBe(Date.now() + 1000);
|
||||
s.service.nextSweepAt.mockResolvedValue(null);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("does no session SQL during standby or drain and recovers a first commit after drain", async () => {
|
||||
const s = setup();
|
||||
s.primary.mockReturnValue(false);
|
||||
await s.start().ready;
|
||||
await vi.advanceTimersByTimeAsync(6000);
|
||||
expect(s.service.sweep).not.toHaveBeenCalled();
|
||||
expect(s.service.nextSweepAt).not.toHaveBeenCalled();
|
||||
s.primary.mockReturnValue(true);
|
||||
s.enabled.mockReturnValue(false);
|
||||
await s.commit();
|
||||
await vi.advanceTimersByTimeAsync(6000);
|
||||
expect(s.service.sweep).not.toHaveBeenCalled();
|
||||
expect(s.service.nextSweepAt).not.toHaveBeenCalled();
|
||||
s.enabled.mockReturnValue(true);
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
expect(s.service.sweep).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("waits for in-flight cleanup on shutdown without rearming its deadline", async () => {
|
||||
const s = setup();
|
||||
let finish!: () => void;
|
||||
s.service.sweep.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
|
||||
const worker = s.start();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
let stopped = false;
|
||||
const stop = s.coordinator.stop().then(() => { stopped = true; });
|
||||
await Promise.resolve();
|
||||
expect(stopped).toBe(false);
|
||||
finish();
|
||||
await stop;
|
||||
await worker.ready;
|
||||
expect(s.service.nextSweepAt).not.toHaveBeenCalled();
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -25,6 +25,7 @@ import {
|
||||
import { heartbeatService } from "../services/heartbeat.ts";
|
||||
import { runningProcesses } from "../adapters/index.ts";
|
||||
import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js";
|
||||
import { createLifecycleDriver } from "../services/agent-lifecycle-driver.js";
|
||||
|
||||
const mockAdapterExecute = vi.hoisted(() =>
|
||||
vi.fn(async () => ({
|
||||
@@ -166,6 +167,27 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => {
|
||||
return { companyId, ownerUserId, agentId };
|
||||
}
|
||||
|
||||
it("dispatches saved work on readiness with company scoping and normal admission checks", async () => {
|
||||
const { companyId, agentId, ownerUserId } = await seedCompany();
|
||||
await db.update(agents).set({ status: "paused", lifecycleState: "verifying" }).where(eq(agents.id, agentId));
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "queued",
|
||||
invocationSource: "on_demand", contextSnapshot: { responsibleUserId: ownerUserId } }).returning();
|
||||
const driver = createLifecycleDriver(db, {} as never);
|
||||
const currentAgent = async () => (await db.select().from(agents).where(eq(agents.id, agentId)))[0];
|
||||
await driver.onReady!(await currentAgent());
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("queued");
|
||||
|
||||
await db.update(agents).set({ status: "idle", lifecycleState: "ready" }).where(eq(agents.id, agentId));
|
||||
await heartbeat.resumeQueuedRunsForAgent(randomUUID(), agentId);
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("queued");
|
||||
|
||||
await driver.onReady!(await currentAgent());
|
||||
expect(await waitForRun(db, run.id)).toMatchObject({ status: "succeeded", responsibleUserId: ownerUserId });
|
||||
expect(mockAdapterExecute).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("dispatches an interrupted queue under the clicking operator through the real startup path", async () => {
|
||||
const { companyId, agentId, ownerUserId } = await seedCompany();
|
||||
const operatorId = `operator-${randomUUID()}`, issueId = randomUUID(), commentId = randomUUID(), queueId = randomUUID();
|
||||
|
||||
+2
-7
@@ -9,6 +9,7 @@ import { customerSuccessRoutes } from "./routes/customer-success.js";
|
||||
import { cloudWarmStandbyMiddleware } from "./middleware/cloud-warm-standby.js";
|
||||
import type { CloudWarmStandby } from "./services/cloud-warm-standby.js";
|
||||
import { browserUseRoutes } from "./routes/browser-use.js";
|
||||
import { registerBrowserUseCleanup } from "./services/browser-use-work.js";
|
||||
import { browserUseService } from "./services/browser-use.js";
|
||||
import { slackToolRoutes } from "./routes/slack-tools.js";
|
||||
import { createPublicMcpOAuth, publicMcpConfig } from "./services/public-mcp/oauth.js";
|
||||
@@ -1292,12 +1293,7 @@ export async function createApp(
|
||||
);
|
||||
});
|
||||
};
|
||||
const browserUseTimer = setInterval(() => {
|
||||
if (isWarmStandby() || isIdleTaskDrainActive()) return;
|
||||
void browserUse.sweep().catch(() => logger.warn("Browser Use reconciliation failed; retrying."));
|
||||
}, 3000);
|
||||
browserUseTimer.unref?.();
|
||||
if (!isWarmStandby() && !isIdleTaskDrainActive()) void browserUse.sweep().catch(() => logger.warn("Browser Use startup reconciliation failed; retrying."));
|
||||
registerBrowserUseCleanup(deliveryWork, browserUse, () => !isIdleTaskDrainActive());
|
||||
let importTransferSweepTimer: ReturnType<typeof setInterval> | null =
|
||||
setInterval(
|
||||
sweepImportTransferSpools,
|
||||
@@ -1410,7 +1406,6 @@ export async function createApp(
|
||||
chatPublicationTimer = null;
|
||||
}
|
||||
await chatReconciliation.drain();
|
||||
clearInterval(browserUseTimer);
|
||||
if (importTransferSweepTimer) {
|
||||
clearInterval(importTransferSweepTimer);
|
||||
importTransferSweepTimer = null;
|
||||
|
||||
@@ -13,14 +13,58 @@ function fixture() {
|
||||
const store = { get: vi.fn(), claim: vi.fn().mockResolvedValue(agent), renew: vi.fn().mockResolvedValue(true),
|
||||
setRequiredPlugins: vi.fn().mockResolvedValue(true), completeHost: vi.fn().mockResolvedValue(true),
|
||||
completePlugin: vi.fn().mockResolvedValue(true), defer: vi.fn().mockResolvedValue(true),
|
||||
change: vi.fn().mockResolvedValue({ ...agent, lifecycleState: "ready" }), pending: vi.fn().mockResolvedValue([agent.id]) } satisfies LifecycleStore;
|
||||
change: vi.fn().mockResolvedValue({ ...agent, lifecycleState: "ready", lifecycleVersion: 4, lifecycleOperation: null }), pending: vi.fn().mockResolvedValue([agent.id]) } satisfies LifecycleStore;
|
||||
const driver = { requiredPluginIds: vi.fn().mockResolvedValue(["plugin"]), runHost: vi.fn().mockResolvedValue("complete"),
|
||||
runPlugin: vi.fn().mockResolvedValue("complete") } satisfies LifecycleDriver;
|
||||
runPlugin: vi.fn().mockResolvedValue("complete"), onReady: vi.fn().mockResolvedValue(undefined) } satisfies LifecycleDriver;
|
||||
const report = vi.fn();
|
||||
return { store, driver, report, worker: createLifecycleWorker(store, driver, report) };
|
||||
}
|
||||
afterEach(() => { vi.restoreAllMocks(); vi.useRealTimers(); });
|
||||
|
||||
describe("lifecycle readiness handoff", () => {
|
||||
it("notifies the scheduler only after host, plugins, and readiness have committed", async () => {
|
||||
const { store, driver, worker } = fixture();
|
||||
let commit!: (ready: LifecycleAgent) => void;
|
||||
store.change.mockReturnValue(new Promise(resolve => { commit = resolve; }));
|
||||
driver.onReady.mockImplementation(async ready => {
|
||||
expect(store.completeHost).toHaveBeenCalledOnce();
|
||||
expect(store.completePlugin).toHaveBeenCalledOnce();
|
||||
expect(store.change).toHaveBeenCalledOnce();
|
||||
expect(ready).toMatchObject({ lifecycleState: "ready", lifecycleVersion: 4, lifecycleOperation: null });
|
||||
});
|
||||
const work = worker.process(agent.id);
|
||||
await vi.waitFor(() => expect(store.change).toHaveBeenCalledOnce());
|
||||
expect(driver.onReady).not.toHaveBeenCalled();
|
||||
commit({ ...agent, lifecycleState: "ready", lifecycleVersion: 4, lifecycleOperation: null });
|
||||
await work;
|
||||
expect(driver.onReady).toHaveBeenCalledOnce();
|
||||
await worker.stop();
|
||||
});
|
||||
|
||||
it.each(["host", "plugin", "transition", "hold"])("does not dispatch when %s prevents readiness", async stage => {
|
||||
const { store, driver, worker } = fixture();
|
||||
if (stage === "host") driver.runHost.mockResolvedValue("pending");
|
||||
if (stage === "plugin") driver.runPlugin.mockResolvedValue("pending");
|
||||
if (stage === "transition") store.change.mockResolvedValue(null);
|
||||
if (stage === "hold") store.change.mockResolvedValue({ ...agent, lifecycleState: "paused" });
|
||||
await worker.process(agent.id);
|
||||
expect(driver.onReady).not.toHaveBeenCalled();
|
||||
await worker.stop();
|
||||
});
|
||||
|
||||
it("preserves committed readiness after a dispatch failure for the queue sweep to recover", async () => {
|
||||
const { store, driver, report, worker } = fixture();
|
||||
const error = new Error("queue unavailable");
|
||||
driver.onReady.mockRejectedValue(error);
|
||||
await worker.process(agent.id);
|
||||
expect(report).toHaveBeenCalledWith(expect.objectContaining({ stage: "ready", phase: "ready", version: 4 }), error);
|
||||
expect(store.defer).not.toHaveBeenCalled();
|
||||
expect(driver.runHost).toHaveBeenCalledOnce();
|
||||
expect(store.change).toHaveBeenCalledOnce();
|
||||
await worker.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("lifecycle failure diagnostics", () => {
|
||||
it.each(["host", "plugin", "claim", "renew", "scan", "defer"] as const)("reports %s failures with the available context", async stage => {
|
||||
const { store, driver, report, worker } = fixture();
|
||||
|
||||
@@ -27,11 +27,13 @@ export interface LifecycleDriver {
|
||||
requiredPluginIds(agent: LifecycleAgent): Promise<string[]>;
|
||||
runHost(agent: LifecycleAgent): Promise<"complete" | "pending">;
|
||||
runPlugin(agent: LifecycleAgent, pluginId: string): Promise<"complete" | "pending">;
|
||||
/** Notify the scheduler only after readiness has committed. */
|
||||
onReady?(agent: LifecycleAgent): Promise<void>;
|
||||
}
|
||||
|
||||
export interface LifecycleFailure {
|
||||
stage: "claim" | "renew" | "select_plugins" | "save_plugins" | "host" | "plugin"
|
||||
| "complete_host" | "complete_plugin" | "transition" | "defer" | "scan" | "policy_scan";
|
||||
| "complete_host" | "complete_plugin" | "transition" | "defer" | "scan" | "policy_scan" | "ready";
|
||||
agentId?: string;
|
||||
companyId?: string;
|
||||
phase?: AgentLifecycleState;
|
||||
|
||||
@@ -56,6 +56,14 @@ export function createLifecycleWorker(store: LifecycleStore, driver: LifecycleDr
|
||||
if (!await runStep("complete_plugin", () => store.completePlugin(agent, owner, pluginId), pluginId)) continue phase;
|
||||
}
|
||||
const changed = await runStep("transition", () => store.change(id, "complete", { version: agent.lifecycleVersion, owner }));
|
||||
if (changed?.lifecycleState === "ready" && !stopped && canRun()) {
|
||||
// Accepted messages may already be waiting for verification. Do not
|
||||
// leave them until the periodic queue sweep. Readiness is committed;
|
||||
// a dispatch failure must not replay setup or defer the old phase.
|
||||
await attempt({ ...context, stage: "ready", phase: changed.lifecycleState,
|
||||
version: changed.lifecycleVersion, operationId: changed.lifecycleOperation?.id },
|
||||
async () => { await driver.onReady?.(changed); }).catch(() => {});
|
||||
}
|
||||
if (changed && !isLifecycleWorkPending(changed.lifecycleState)) return;
|
||||
} catch {
|
||||
if (!await runStep("defer", () => store.defer(agent, owner, "The lifecycle step failed. Retry the operation.", new Date()))) continue;
|
||||
|
||||
@@ -10,6 +10,9 @@ export function createLifecycleDriver(db: Db, manager: PluginWorkerManager): Lif
|
||||
const { verify } = agentHarnessVerificationService(db, manager);
|
||||
const heartbeat = heartbeatService(db, { pluginWorkerManager: manager });
|
||||
return {
|
||||
async onReady(agent) {
|
||||
await heartbeat.resumeQueuedRunsForAgent(agent.companyId, agent.id);
|
||||
},
|
||||
async requiredPluginIds(agent) {
|
||||
if (agent.lifecycleRequiredPluginIds !== null) return agent.lifecycleRequiredPluginIds;
|
||||
const installed = await db.select().from(plugins).where(inArray(plugins.status, ["ready", "error", "installed", "upgrade_pending"]));
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
|
||||
/** The app owns this worker; request-scoped services only signal durable writes. */
|
||||
export function registerBrowserUseCleanup(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
service: { sweep(): Promise<void>; nextSweepAt(): Promise<number | null> },
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
let next: number | null = null;
|
||||
return coordinator.register(DELIVERY_QUEUES.browser, {
|
||||
retryMs: 3000,
|
||||
run: () => service.sweep(),
|
||||
hasPending: async () => {
|
||||
// During idle drain retain a memory-only wake, including startup or a
|
||||
// newly committed first session. Warm standby is gated by the coordinator.
|
||||
next = canRun() ? await service.nextSweepAt() : Date.now() + 3000;
|
||||
return false;
|
||||
},
|
||||
// Bound overdue batches without spinning on a busy database.
|
||||
nextRunAt: () => next === null ? null : Math.max(Date.now() + 1000, next),
|
||||
});
|
||||
}
|
||||
+125
-101
@@ -43,6 +43,7 @@ import { createCostEventInTransaction } from "./costs.js";
|
||||
import { createFinanceEventInTransaction } from "./finance.js";
|
||||
import { withAccountingTransaction } from "./accounting-transaction.js";
|
||||
import { logActivity } from "./activity-log.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { browserUseViewports } from "./browser-use-viewport.js";
|
||||
import { forbidden, notFound, conflict } from "../errors.js";
|
||||
|
||||
@@ -64,6 +65,7 @@ const argsSchema = z
|
||||
})
|
||||
.strict();
|
||||
const terminalSessions = ["closed", "failed"];
|
||||
const liveSessions = ["starting", "running", "idle", "stopping"];
|
||||
const requestMarker = (invocationId: string) => `\n\n[Paperclip request: ${invocationId}]`;
|
||||
|
||||
export function browserUseService(
|
||||
@@ -422,6 +424,7 @@ export function browserUseService(
|
||||
);
|
||||
const sid = args.sessionId;
|
||||
const s = await db.transaction(async (tx) => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser);
|
||||
let row: Session;
|
||||
if (sid) {
|
||||
await owned(binding, grant, sid);
|
||||
@@ -429,6 +432,7 @@ export function browserUseService(
|
||||
.update(sessions)
|
||||
.set({
|
||||
status: "starting",
|
||||
nextPollAt: new Date(),
|
||||
idleDeadline: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
@@ -486,6 +490,7 @@ export function browserUseService(
|
||||
...(maxCostUsd === undefined ? {} : { maxCostUsd }),
|
||||
});
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser);
|
||||
await tx
|
||||
.update(runs)
|
||||
.set({ providerRunId: result.id, status: result.status })
|
||||
@@ -507,6 +512,7 @@ export function browserUseService(
|
||||
const message = rejected ? error.message :
|
||||
"Browser run creation could not be confirmed. Paperclip is locating and stopping possible provider work. Do not start another run yet.";
|
||||
await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser);
|
||||
const [current] = await tx.select().from(sessions).where(eq(sessions.id, s.id)).for("update");
|
||||
// A restart must not leave a drained rejection attached to a starting
|
||||
// session: that combination has neither recovery work nor an idle timer.
|
||||
@@ -566,25 +572,28 @@ export function browserUseService(
|
||||
throw conflict("Only an idle browser can be kept open.");
|
||||
} else {
|
||||
if (terminalSessions.includes(s.status)) return;
|
||||
await db
|
||||
.update(sessions)
|
||||
.set({
|
||||
stopRequested: action,
|
||||
...(action === "end" ? { status: "stopping" } : {}),
|
||||
nextPollAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(sessions.id, s.id),
|
||||
action === "cancel"
|
||||
? or(
|
||||
isNull(sessions.stopRequested),
|
||||
ne(sessions.stopRequested, "end"),
|
||||
)
|
||||
: undefined,
|
||||
),
|
||||
);
|
||||
await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser);
|
||||
await tx
|
||||
.update(sessions)
|
||||
.set({
|
||||
stopRequested: action,
|
||||
...(action === "end" ? { status: "stopping" } : {}),
|
||||
nextPollAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(sessions.id, s.id),
|
||||
action === "cancel"
|
||||
? or(
|
||||
isNull(sessions.stopRequested),
|
||||
ne(sessions.stopRequested, "end"),
|
||||
)
|
||||
: undefined,
|
||||
),
|
||||
);
|
||||
});
|
||||
}
|
||||
await signal(
|
||||
s,
|
||||
@@ -1218,97 +1227,109 @@ export function browserUseService(
|
||||
);
|
||||
}
|
||||
}
|
||||
let sweeping = false;
|
||||
async function sweep(connectionId?: string) {
|
||||
if (sweeping) return;
|
||||
sweeping = true;
|
||||
try {
|
||||
const due = await db
|
||||
.select()
|
||||
.from(sessions)
|
||||
// Wait for a concurrent sweep rather than reporting completion while its
|
||||
// lease is still held. A filtered cleanup must still inspect its own scope.
|
||||
let sweeping: Promise<void> | null = null;
|
||||
async function sweep(connectionId?: string): Promise<void> {
|
||||
if (sweeping) {
|
||||
await sweeping;
|
||||
return sweep(connectionId);
|
||||
}
|
||||
sweeping = reconcileDue(connectionId);
|
||||
try { await sweeping; } finally { sweeping = null; }
|
||||
}
|
||||
async function reconcileDue(connectionId?: string) {
|
||||
const due = await db
|
||||
.select()
|
||||
.from(sessions)
|
||||
.where(
|
||||
and(
|
||||
connectionId ? eq(sessions.connectionId, connectionId) : undefined,
|
||||
inArray(sessions.status, liveSessions),
|
||||
lte(sessions.nextPollAt, new Date()),
|
||||
or(
|
||||
isNull(sessions.leaseUntil),
|
||||
lte(sessions.leaseUntil, new Date()),
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(30);
|
||||
for (const s of due) {
|
||||
const token = randomUUID();
|
||||
const lease = await db
|
||||
.update(sessions)
|
||||
.set({
|
||||
leaseToken: token,
|
||||
leaseUntil: new Date(Date.now() + 180_000),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
connectionId ? eq(sessions.connectionId, connectionId) : undefined,
|
||||
inArray(sessions.status, [
|
||||
"starting",
|
||||
"running",
|
||||
"idle",
|
||||
"stopping",
|
||||
]),
|
||||
lte(sessions.nextPollAt, new Date()),
|
||||
eq(sessions.id, s.id),
|
||||
or(
|
||||
isNull(sessions.leaseUntil),
|
||||
lte(sessions.leaseUntil, new Date()),
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(30);
|
||||
for (const s of due) {
|
||||
const token = randomUUID();
|
||||
const lease = await db
|
||||
.returning();
|
||||
if (!lease.length) continue;
|
||||
// Renew while paginating provider events or stopping multiple browsers.
|
||||
const renewal = setInterval(() => {
|
||||
void db
|
||||
.update(sessions)
|
||||
.set({ leaseUntil: new Date(Date.now() + 180_000) })
|
||||
.where(and(eq(sessions.id, s.id), eq(sessions.leaseToken, token)))
|
||||
.catch(() => {});
|
||||
}, 30_000);
|
||||
renewal.unref?.();
|
||||
let delay = s.status === "idle" ? 15000 : 3000;
|
||||
try {
|
||||
await reconcile(s);
|
||||
} catch (e) {
|
||||
delay = Math.max(
|
||||
10000,
|
||||
e instanceof BrowserUseError ? e.retryAfterMs : 0,
|
||||
);
|
||||
await db
|
||||
.update(sessions)
|
||||
.set({
|
||||
leaseToken: token,
|
||||
leaseUntil: new Date(Date.now() + 180_000),
|
||||
error:
|
||||
s.stopRequested === "end"
|
||||
? "Browser shutdown is not confirmed. Paperclip will retry."
|
||||
: "Browser synchronization failed. Paperclip will retry.",
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(sessions.id, s.id),
|
||||
or(
|
||||
isNull(sessions.leaseUntil),
|
||||
lte(sessions.leaseUntil, new Date()),
|
||||
),
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
if (!lease.length) continue;
|
||||
// Renew while paginating provider events or stopping multiple browsers.
|
||||
const renewal = setInterval(() => {
|
||||
void db
|
||||
.update(sessions)
|
||||
.set({ leaseUntil: new Date(Date.now() + 180_000) })
|
||||
.where(and(eq(sessions.id, s.id), eq(sessions.leaseToken, token)))
|
||||
.catch(() => {});
|
||||
}, 30_000);
|
||||
renewal.unref?.();
|
||||
let delay = s.status === "idle" ? 15000 : 3000;
|
||||
try {
|
||||
await reconcile(s);
|
||||
} catch (e) {
|
||||
delay = Math.max(
|
||||
10000,
|
||||
e instanceof BrowserUseError ? e.retryAfterMs : 0,
|
||||
);
|
||||
await db
|
||||
.update(sessions)
|
||||
.set({
|
||||
error:
|
||||
s.stopRequested === "end"
|
||||
? "Browser shutdown is not confirmed. Paperclip will retry."
|
||||
: "Browser synchronization failed. Paperclip will retry.",
|
||||
})
|
||||
.where(eq(sessions.id, s.id));
|
||||
} finally {
|
||||
clearInterval(renewal);
|
||||
await db
|
||||
.update(sessions)
|
||||
.set({
|
||||
leaseToken: null,
|
||||
leaseUntil: null,
|
||||
nextPollAt: new Date(
|
||||
Date.now() + delay + Math.floor(Math.random() * 500),
|
||||
),
|
||||
})
|
||||
.where(and(eq(sessions.id, s.id), eq(sessions.leaseToken, token)));
|
||||
.where(eq(sessions.id, s.id));
|
||||
} finally {
|
||||
clearInterval(renewal);
|
||||
const release = (tx: Pick<Db, "update">) => tx.update(sessions).set({
|
||||
leaseToken: null,
|
||||
leaseUntil: null,
|
||||
nextPollAt: new Date(Date.now() + delay + Math.floor(Math.random() * 500)),
|
||||
}).where(and(eq(sessions.id, s.id), eq(sessions.leaseToken, token)));
|
||||
if (connectionId) {
|
||||
// Credential-removal sweeps can race the app worker. Wake it when
|
||||
// their lease ends so it does not wait for the old lease expiry.
|
||||
await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser);
|
||||
await release(tx);
|
||||
});
|
||||
} else {
|
||||
await release(db);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
sweeping = false;
|
||||
}
|
||||
}
|
||||
async function nextSweepAt(): Promise<number | null> {
|
||||
const [row] = await db.select({
|
||||
// An overdue row owned by another sweep cannot run before its lease ends.
|
||||
at: sql<string | null>`min(greatest(${sessions.nextPollAt}, ${sessions.leaseUntil}))`,
|
||||
}).from(sessions).where(inArray(sessions.status, liveSessions));
|
||||
return row?.at == null ? null : new Date(row.at).getTime();
|
||||
}
|
||||
return {
|
||||
execute,
|
||||
sweep,
|
||||
nextSweepAt,
|
||||
list,
|
||||
viewer,
|
||||
resize,
|
||||
@@ -1322,16 +1343,19 @@ export function browserUseService(
|
||||
const scope = and(
|
||||
eq(sessions.companyId, companyId),
|
||||
eq(sessions.connectionId, connectionId),
|
||||
inArray(sessions.status, ["starting", "running", "idle", "stopping"]),
|
||||
inArray(sessions.status, liveSessions),
|
||||
);
|
||||
await db
|
||||
.update(sessions)
|
||||
.set({
|
||||
stopRequested: "end",
|
||||
nextPollAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(scope);
|
||||
await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser);
|
||||
await tx
|
||||
.update(sessions)
|
||||
.set({
|
||||
stopRequested: "end",
|
||||
nextPollAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(scope);
|
||||
});
|
||||
await sweep(connectionId);
|
||||
const [pending] = await db
|
||||
.select({ id: sessions.id })
|
||||
|
||||
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
browser: "browser-use",
|
||||
email: "email-channels",
|
||||
feedback: "feedback-exports",
|
||||
chatCompletion: "chat-completions",
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,233 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { agents, companies, completionContracts, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { CHAT_CONTROL_RECOVERY_ADMISSION_KEY } from "../chat-control-recovery-stop.js";
|
||||
import { buildNativeRuntimeContext } from "../native-runtime/runtime-context.js";
|
||||
import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js";
|
||||
import { prepareNativeWorkspaceSync } from "../native-runtime/native-workspace-sync.js";
|
||||
import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mode.js";
|
||||
import { canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js";
|
||||
import { selectHeartbeatRuntime, type HeartbeatRuntimeSelectionInput } from "./runtime-selection.js";
|
||||
|
||||
// Keep the real contract, session, and runtime-selection transactions. Replace only
|
||||
// filesystem materialization and sandbox synchronization at their external boundaries.
|
||||
vi.mock("../native-runtime/runtime-context.js", async (original) => ({
|
||||
...await original<typeof import("../native-runtime/runtime-context.js")>(),
|
||||
buildNativeRuntimeContext: vi.fn(async () => nativeRuntimeContextFixture()),
|
||||
}));
|
||||
vi.mock("../native-runtime/native-workspace-sync.js", async (original) => ({
|
||||
...await original<typeof import("../native-runtime/native-workspace-sync.js")>(),
|
||||
prepareNativeWorkspaceSync: vi.fn(async () => null),
|
||||
}));
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: Db;
|
||||
let instructionRoot: string;
|
||||
beforeAll(async () => {
|
||||
instructionRoot = await mkdtemp(join(tmpdir(), "paperclip-selection-instructions-"));
|
||||
await writeFile(join(instructionRoot, "AGENTS.md"), "Complete the assigned task.\n");
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-selection");
|
||||
db = createDb(database.connectionString);
|
||||
await db.execute(sql`set client_min_messages = warning`);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); if (instructionRoot) await rm(instructionRoot, { recursive: true, force: true }); });
|
||||
beforeEach(async () => {
|
||||
await db.execute(sql`truncate table companies restart identity cascade`);
|
||||
vi.clearAllMocks();
|
||||
vi.mocked(prepareNativeWorkspaceSync).mockResolvedValue(null);
|
||||
vi.mocked(buildNativeRuntimeContext).mockImplementation(async () => {
|
||||
const context = nativeRuntimeContextFixture();
|
||||
context.instructions.bundle.rootPath = instructionRoot;
|
||||
return { ...context, aggregateDigest: canonicalNativeRuntimeContextDigest(context) };
|
||||
});
|
||||
});
|
||||
|
||||
async function fixture(adapterType = "paperclip_runner") {
|
||||
const [company] = await db.insert(companies).values({ name: "Runtime selection", issuePrefix: "SEL" }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Selecting agent", status: "running", adapterType, adapterConfig: { provider: "codex", model: "gpt-5.6-luna" } }).returning();
|
||||
const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Complete this task", description: "Preserve the requested work", assigneeAgentId: agent.id }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, issueId: issue.id, scopeKind: "issue", status: "running" }).returning();
|
||||
const now = new Date();
|
||||
const input: HeartbeatRuntimeSelectionInput = {
|
||||
run, agent,
|
||||
nativeRuntimeResolution: resolveHeartbeatNativeRuntimeMode({ persisted: run, enabled: true, runtimeConfig: {}, adapterConfig: agent.adapterConfig, agent, issue, target: null, workspaceId: null }),
|
||||
task: { issueRef: issue, issueContext: null, context: { paperclipTaskMarkdown: "Complete this task" }, executionContinuation: null, safeWakeComments: [], safeWakeCommentContext: null, paperclipWakePayload: null, acceptedPlanContinuationWake: false },
|
||||
session: { persistedNativeExecutionInput: null, persistedRunnerProfile: {}, taskSessionCredentialCompatible: true, taskSessionDecodedParams: null, taskSessionForRun: null, isFailedChatRunRetry: false, getFreshSessionHandoff: vi.fn(async () => "Earlier context") },
|
||||
workspace: { persistedExecutionWorkspace: null, executionWorkspace: { cwd: "/tmp/paperclip-runtime-selection", repoUrl: null, repoRef: null, branchName: null }, executionTarget: null, isDotRun: false, projectRepositoryPaths: [],
|
||||
activeEnvironmentLease: { lease: { id: randomUUID(), companyId: company.id, environmentId: null, executionWorkspaceId: null, issueId: issue.id, heartbeatRunId: run.id, status: "active", leasePolicy: "ephemeral", provider: null, providerLeaseId: null, acquiredAt: now, lastUsedAt: now, expiresAt: null, releasedAt: null, failureReason: null, cleanupStatus: null, metadata: null, createdAt: now, updatedAt: now } },
|
||||
},
|
||||
config: { runtimeConfig: {}, runtimeSkillEntries: [], runScopedMentionedSkillKeys: [], instructionCopy: null, agentIdentity: undefined, issueAssigneeOverrides: null, resolvedConfig: {}, managedAiRuntime: undefined },
|
||||
recovery: {}, trace: { providerTraceRequested: false, providerTraceCapture: null },
|
||||
issuesSvc: { addComment: vi.fn() }, runtimeEnv: {}, onAdapterEvent: vi.fn(async () => {}),
|
||||
onNativeLifecycleSelected: vi.fn(), onProviderResourceDisposition: vi.fn(), onNativeOwnershipHeld: vi.fn(), stopControllerLease: vi.fn(),
|
||||
};
|
||||
return { input, issue, run, agent };
|
||||
}
|
||||
const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!);
|
||||
const coordinators = (id: string) => db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, id));
|
||||
const sandbox = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: "/workspace", reusableLeaseConfigured: false };
|
||||
|
||||
it("persists the native input, completion contract, and coordinator before transferring cleanup ownership", async () => {
|
||||
const { input, run } = await fixture();
|
||||
input.workspace.onCheckpoint = vi.fn(async () => {});
|
||||
input.workspace.onPhase = async (_name, work) => work();
|
||||
vi.mocked(prepareNativeWorkspaceSync).mockImplementationOnce(async args => {
|
||||
expect(await read(run.id)).toMatchObject({ runtimeMode: "native", nativeIssueId: input.task.issueRef!.id });
|
||||
expect(await coordinators(run.id)).toHaveLength(1);
|
||||
expect(input.stopControllerLease).toHaveBeenCalledOnce();
|
||||
expect(args).toMatchObject({ runId: run.id, workspaceId: run.id, workspaceLocalDir: input.workspace.executionWorkspace.cwd });
|
||||
expect(args.onCheckpoint).toBe(input.workspace.onCheckpoint);
|
||||
expect(args.onPhase).toBe(input.workspace.onPhase);
|
||||
return null;
|
||||
});
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
expect(selected.selected).toBe(true);
|
||||
if (!selected.selected) throw new Error("Expected selection");
|
||||
const saved = await read(run.id);
|
||||
expect(saved).toMatchObject({ runtimeMode: "native", driverKind: "codex_app_server", nativePhase: "observed", nativeSessionId: expect.any(String), runnerInstanceId: selected.nativeRunnerInstanceId,
|
||||
runnerProfileJson: { nativeExecutionInput: selected.nativeExecution, recoveryEventInventoryVersion: 1 } });
|
||||
expect(await db.select().from(completionContracts)).toHaveLength(1);
|
||||
expect(selected.getNativeFreshSessionHandoff).toBe(input.session.getFreshSessionHandoff);
|
||||
expect(input.onNativeOwnershipHeld).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("persists configured GitHub instruction skills without admitting provider-mentioned skills", async () => {
|
||||
const { input, issue, run } = await fixture();
|
||||
input.config.runScopedMentionedSkillKeys = ["company/test/configured"];
|
||||
input.task.context.paperclipWake = {
|
||||
reason: "External chat message received", externalChatProvider: "github", checkedOutByHarness: true,
|
||||
issue: { id: issue.id, workMode: "standard" },
|
||||
comments: [{ id: "review-comment", body: "Please use /provider-mentioned" }],
|
||||
commentIds: ["review-comment"], latestCommentId: "review-comment",
|
||||
commentWindow: { requestedCount: 1, includedCount: 1, missingCount: 0 }, fallbackFetchNeeded: false,
|
||||
};
|
||||
vi.mocked(buildNativeRuntimeContext).mockImplementationOnce(async () => {
|
||||
const context = nativeRuntimeContextFixture();
|
||||
context.instructions.bundle.rootPath = instructionRoot;
|
||||
context.skills = ["configured", "provider-mentioned"].map(name => ({
|
||||
key: `company/test/${name}`, runtimeName: name, versionId: null,
|
||||
bundle: { ...context.instructions.bundle },
|
||||
}));
|
||||
return { ...context, aggregateDigest: canonicalNativeRuntimeContextDigest(context) };
|
||||
});
|
||||
expect(await selectHeartbeatRuntime(db, input)).toMatchObject({
|
||||
selected: true, nativeExecution: { task: { description: "Configured GitHub instruction skills:\n/configured" } },
|
||||
});
|
||||
expect((await read(run.id)).runnerProfileJson).toMatchObject({
|
||||
nativeExecutionInput: { task: { description: "Configured GitHub instruction skills:\n/configured" } },
|
||||
});
|
||||
});
|
||||
|
||||
it("restores the immutable admitted input and saved checkpoint when newer context arrives", async () => {
|
||||
const { input, run } = await fixture();
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
if (!selected.selected || !selected.nativeExecution) throw new Error("Expected native selection");
|
||||
const saved = await read(run.id);
|
||||
const profile = { ...saved.runnerProfileJson, sessionCheckpoint: { retained: "checkpoint" } };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, run.id));
|
||||
input.run = await read(run.id);
|
||||
input.session.persistedRunnerProfile = profile;
|
||||
input.session.persistedNativeExecutionInput = selected.nativeExecution;
|
||||
input.task.context.paperclipTaskMarkdown = "A newer unrelated request";
|
||||
input.task.safeWakeComments = [{ id: randomUUID(), body: "New direction belongs to a separate wake" }];
|
||||
vi.mocked(buildNativeRuntimeContext).mockClear();
|
||||
const resumed = await selectHeartbeatRuntime(db, input);
|
||||
expect(resumed).toMatchObject({ selected: true, nativeExecution: selected.nativeExecution, nativeRunnerInstanceId: selected.nativeRunnerInstanceId });
|
||||
expect((await read(run.id)).runnerProfileJson).toMatchObject(profile);
|
||||
expect(await db.select().from(completionContracts)).toHaveLength(1);
|
||||
expect(await coordinators(run.id)).toHaveLength(1);
|
||||
expect(buildNativeRuntimeContext).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects an admitted input bound to another run without replacing the saved runtime", async () => {
|
||||
const { input, run } = await fixture();
|
||||
const selected = await selectHeartbeatRuntime(db, input);
|
||||
if (!selected.selected || !selected.nativeExecution) throw new Error("Expected native selection");
|
||||
const saved = await read(run.id);
|
||||
input.run = saved;
|
||||
input.session.persistedNativeExecutionInput = { ...selected.nativeExecution, binding: { ...selected.nativeExecution.binding, runId: randomUUID() } };
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("native_execution_input_persisted_binding_mismatch");
|
||||
expect((await read(run.id)).runnerProfileJson).toEqual(saved.runnerProfileJson);
|
||||
});
|
||||
|
||||
it.each(["cancelled", "startup_cancellation"])("does not create a native coordinator when %s wins preparation", async winner => {
|
||||
const { input, run } = await fixture();
|
||||
input.beforeNativeRuntimeSelection = async () => {
|
||||
await db.update(heartbeatRuns).set(winner === "cancelled" ? { status: "cancelled" } : { resultJson: { startupCancellation: { beforeNativeSelection: true } } }).where(eq(heartbeatRuns.id, run.id));
|
||||
};
|
||||
expect(await selectHeartbeatRuntime(db, input)).toEqual({ selected: false });
|
||||
expect(await read(run.id)).toMatchObject({ runtimeMode: "legacy", nativeSessionId: null, completionContractId: null });
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
expect(input.stopControllerLease).not.toHaveBeenCalled();
|
||||
expect(prepareNativeWorkspaceSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reports a lost controller lease before returning without native selection", async () => {
|
||||
const { input, run } = await fixture();
|
||||
await db.update(heartbeatRuns).set({ controllerBootId: randomUUID(), controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await selectHeartbeatRuntime(db, input)).toEqual({ selected: false });
|
||||
expect(input.onNativeOwnershipHeld).toHaveBeenCalledOnce();
|
||||
expect(input.stopControllerLease).not.toHaveBeenCalled();
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("rejects a competing persisted legacy selection atomically", async () => {
|
||||
const { input, run } = await fixture();
|
||||
input.beforeNativeRuntimeSelection = async () => {
|
||||
await db.update(heartbeatRuns).set({ runtimeModeResolvedAt: new Date(), runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, run.id));
|
||||
};
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("native_runtime_mode_conflict");
|
||||
expect((await read(run.id)).runtimeMode).toBe("legacy");
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
expect(input.stopControllerLease).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reports sandbox teardown policy even when a later preparation step throws", async () => {
|
||||
const { input, run } = await fixture();
|
||||
input.workspace.executionTarget = sandbox;
|
||||
input.beforeNativeRuntimeSelection = async () => { throw new Error("selection interrupted"); };
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("selection interrupted");
|
||||
expect(input.onNativeLifecycleSelected).toHaveBeenCalledWith({ provider: "codex", harness: "codex_app_server", lifecycleMode: "per_turn", sandboxResource: "destroy_after_turn" });
|
||||
expect(input.onProviderResourceDisposition).toHaveBeenCalledWith("destroy");
|
||||
expect(input.stopControllerLease).not.toHaveBeenCalled();
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("retains native admission and teardown state if workspace sync preparation fails", async () => {
|
||||
const { input, run } = await fixture();
|
||||
input.workspace.executionTarget = sandbox;
|
||||
vi.mocked(prepareNativeWorkspaceSync).mockRejectedValueOnce(new Error("sync preparation failed"));
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("sync preparation failed");
|
||||
expect(input.onProviderResourceDisposition).toHaveBeenCalledWith("destroy");
|
||||
expect(input.stopControllerLease).toHaveBeenCalledOnce();
|
||||
expect((await read(run.id)).runtimeMode).toBe("native");
|
||||
expect(await coordinators(run.id)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("rejects a warm native sandbox that cannot reuse its lease", async () => {
|
||||
const { input, run } = await fixture();
|
||||
input.workspace.executionTarget = { ...sandbox, runnerLifecyclePolicy: { mode: "warm", idleTimeoutMs: 300_000 } };
|
||||
await expect(selectHeartbeatRuntime(db, input)).rejects.toThrow("runner_warm_environment_requires_reusable_lease");
|
||||
expect(input.onNativeLifecycleSelected).not.toHaveBeenCalled();
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("preserves only server-owned admission and dispatch evidence when recording legacy selection", async () => {
|
||||
const { input, run } = await fixture("process");
|
||||
const admission = { version: 1, kind: "fixture" }, dispatch = { entered: true };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { [CHAT_CONTROL_RECOVERY_ADMISSION_KEY]: admission, adapterDispatch: dispatch, unrelated: "drop" } }).where(eq(heartbeatRuns.id, run.id));
|
||||
input.trace = { providerTraceRequested: true, providerTraceCapture: { metadata: { id: "trace-id" } } };
|
||||
expect(await selectHeartbeatRuntime(db, input)).toMatchObject({ selected: true, nativeExecution: null, nativeWorkspaceSync: null });
|
||||
expect((await read(run.id)).runnerProfileJson).toEqual({ [CHAT_CONTROL_RECOVERY_ADMISSION_KEY]: admission, adapterDispatch: dispatch, providerTrace: { mode: "raw", traceId: "trace-id", maxBytes: 64 * 1024 * 1024 } });
|
||||
expect(await coordinators(run.id)).toHaveLength(0);
|
||||
expect(input.stopControllerLease).not.toHaveBeenCalled();
|
||||
expect(buildNativeRuntimeContext).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
File diff suppressed because it is too large.
Load diff
@@ -461,7 +461,7 @@ describe("capability-gated connection tool refresh", () => {
|
||||
|
||||
it("wires exact-session recovery and guarded selected identity into heartbeat persistence", () => {
|
||||
const source = readFileSync(
|
||||
new URL("../heartbeat.ts", import.meta.url),
|
||||
new URL("../heartbeat/runtime-selection.ts", import.meta.url),
|
||||
"utf8",
|
||||
);
|
||||
expect(source).toContain("await findNativeSessionResumeRun(db,");
|
||||
|
||||
Reference in new issue
Block a user