diff --git a/README.md b/README.md index f962865..0e2f377 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,7 @@ Consumers ignore previews after the first final chunk and preserve each thought' Final content is acknowledged through the shared event publisher before the Run terminal, keeping large thoughts outside the terminal batch limit. Each successful Run exports an immutable, verified native bundle under `.state/native-checkpoints/` and includes its descriptor in `run.completed`. +OpenAI cancellation before the first successful native checkpoint starts a fresh thread for the next input; cancellation after a successful checkpoint resumes the established thread. In mosoo, the host backs up that bundle before atomically committing the checkpoint descriptor and canonical completion receipt. The next Run waits for terminal acknowledgement, backend settlement, required cleanup, and input-command acknowledgement. A cleanup failure blocks handoff while preserving the committed outcome. diff --git a/src/runtimes/acp/acp-agent-process.ts b/src/runtimes/acp/acp-agent-process.ts index 4b8af86..da9f74b 100644 --- a/src/runtimes/acp/acp-agent-process.ts +++ b/src/runtimes/acp/acp-agent-process.ts @@ -4,10 +4,10 @@ import { once } from "node:events"; import { mkdir } from "node:fs/promises"; import type { Readable, Writable } from "node:stream"; +import type { AgentDriverContext } from "../../core/agent-driver-backend"; import { summarizePath } from "../../observability/driver-debug"; import type { DriverStartInput } from "../../protocol/start"; import { raceWithAbort, settlePromiseWithTimeout } from "../../utils/async"; -import type { AgentDriverContext } from "../../core/agent-driver-backend"; import { bindSpawnedProcess, createProcessTreeEnvironment, @@ -50,6 +50,7 @@ export async function startAcpAgentProcess( supervisionOptions: AcpAgentProcessSupervisionOptions = {}, ): Promise<{ readonly process: AcpAgentProcess; + readonly releaseStartupAbort: () => void; readonly ready: Promise; }> { const command = supervisionOptions.command ?? readFallbackCommand(); @@ -89,14 +90,22 @@ export async function startAcpAgentProcess( watchdog: null, }; agentProcessSupervision.set(agentProcess, supervision); + let startupAbortReleased = false; const onAbort = () => signalAcpAgentProcess(agentProcess, "SIGKILL"); + const releaseStartupAbort = () => { + if (startupAbortReleased) { + return; + } + startupAbortReleased = true; + signal.removeEventListener("abort", onAbort); + }; signal.addEventListener("abort", onAbort, { once: true }); const exited = Promise.withResolvers(); const onExit = () => { agentProcess.off("exit", onExit); agentProcess.off("close", onExit); signalLinuxProcessMarker(supervision.marker, "SIGKILL"); - signal.removeEventListener("abort", onAbort); + releaseStartupAbort(); exited.resolve(); }; agentProcess.once("exit", onExit); @@ -136,7 +145,7 @@ export async function startAcpAgentProcess( ); void ready.catch(() => {}); - return { process: agentProcess, ready }; + return { process: agentProcess, ready, releaseStartupAbort }; } async function waitForAcpAgentProcessReady( diff --git a/src/runtimes/acp/acp-client-request-handler.ts b/src/runtimes/acp/acp-client-request-handler.ts index 14225f0..f1fedd9 100644 --- a/src/runtimes/acp/acp-client-request-handler.ts +++ b/src/runtimes/acp/acp-client-request-handler.ts @@ -19,8 +19,8 @@ import type { WriteTextFileResponse, } from "@agentclientprotocol/sdk"; -import type { DriverEventInput } from "../../protocol/events"; import type { AgentDriverContext } from "../../core/agent-driver-backend"; +import type { DriverEventInput } from "../../protocol/events"; import { AcpTurnStateLimitError, type AcpAssistantTranscriptState, @@ -41,6 +41,7 @@ interface AcpClientRequestHandlerOptions { nativeSessionId(): string | null; onUpdateFailure(error: Error): void; push(context: AgentDriverContext, reason: string, events: DriverEventInput[]): Promise; + pushSession: AcpClientRequestHandlerOptions["push"]; readonly turnEvents: AcpAssistantTranscriptState; } @@ -63,6 +64,7 @@ export class AcpClientRequestHandler { readonly #pathScope: AcpPathScope; readonly #pendingPermissions = new Set>(); readonly #push: AcpClientRequestHandlerOptions["push"]; + readonly #pushSession: AcpClientRequestHandlerOptions["push"]; #permissionIngressClosed = false; #stopping = false; #transcriptTail: Promise = Promise.resolve(); @@ -77,6 +79,7 @@ export class AcpClientRequestHandler { this.#nativeSessionId = options.nativeSessionId; this.#onUpdateFailure = options.onUpdateFailure; this.#push = options.push; + this.#pushSession = options.pushSession; this.#turnEvents = options.turnEvents; this.#pathScope = new AcpPathScope({ allowedRoots: options.allowedRoots, @@ -456,7 +459,8 @@ export class AcpClientRequestHandler { return; } - await this.#push(context, "driver.acp.session.update", events); + const push = isTurnScopedSessionUpdate(params) ? this.#push : this.#pushSession; + await push(context, "driver.acp.session.update", events); }); } diff --git a/src/runtimes/acp/acp-driver-backend.ts b/src/runtimes/acp/acp-driver-backend.ts index ffc6ee5..ab851cb 100644 --- a/src/runtimes/acp/acp-driver-backend.ts +++ b/src/runtimes/acp/acp-driver-backend.ts @@ -1,3 +1,6 @@ +import { dirname, join } from "node:path"; +import { Readable, Writable } from "node:stream"; + import { client as createAcpClient, methods as acpMethods, @@ -10,9 +13,9 @@ import type { ClientContext, InitializeResponse, } from "@agentclientprotocol/sdk"; -import { Readable, Writable } from "node:stream"; -import { dirname, join } from "node:path"; +import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend"; +import { AGENT_DRIVER_VERSION } from "../../core/version"; import { summarizePath, summarizePathCollection } from "../../observability/driver-debug"; import type { DriverEventInput } from "../../protocol/events"; import type { RunId } from "../../protocol/id"; @@ -20,8 +23,6 @@ import type { DriverRuntime } from "../../protocol/runtime"; import type { DriverStartInput } from "../../protocol/start"; import type { RuntimeCommandInput } from "../../runtime-command"; import { raceWithAbort, settlePromiseWithTimeout } from "../../utils/async"; -import { AGENT_DRIVER_VERSION } from "../../core/version"; -import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend"; import { DriverEventPublisher } from "../driver-event-publisher"; import { buildRuntimeBootstrapText, @@ -33,7 +34,6 @@ import { import { startAcpAgentProcess, stopAcpAgentProcess } from "./acp-agent-process"; import type { AcpAgentProcess } from "./acp-agent-process"; import { AcpClientRequestHandler } from "./acp-client-request-handler"; -import { limitAcpInput } from "./acp-input-limit"; import { ACP_PROTOCOL_VERSION, appendOpenCodeInstruction, @@ -49,11 +49,11 @@ import { supportsSessionLoad, supportsSessionResume, } from "./acp-configuration"; +import { limitAcpInput } from "./acp-input-limit"; import { toAuthEvent, toInitializeEvents, toSessionReadyEvents } from "./acp-session-events"; import { setupAcpSession } from "./acp-session-setup"; import { withAcpStartupStage } from "./acp-startup"; import { AcpTurnController } from "./acp-turn-controller"; - import { exportOpenCodeCheckpoint, openCodeDataPath, @@ -130,6 +130,8 @@ export class AcpDriverBackend implements AgentDriverBackend { nativeSessionId: () => this.#nativeSessionId, onUpdateFailure: (error) => this.#connection?.close(error), push: async (context, reason, events) => this.#push(context, reason, events), + pushSession: async (context, reason, events) => + this.#eventPublisher.pushSession(context, reason, events), turnEvents: this.#turnController.events, }); } @@ -462,6 +464,7 @@ export class AcpDriverBackend implements AgentDriverBackend { } connection.signal.throwIfAborted(); connectionReady = true; + startedProcess.releaseStartupAbort(); return setup; } catch (error) { @@ -613,10 +616,25 @@ export class AcpDriverBackend implements AgentDriverBackend { await this.#joinStop(); return; } - if (resumeSignal.aborted) { + if (!providerPromptAdmitted) { return; } - if (!providerPromptAdmitted) { + + if (resumeSignal.aborted) { + const transport = this.#connection; + const agentProcess = this.#agentProcess; + if (transport !== null) { + this.#connection = null; + transport.close(new Error("ACP cancelled turn stopping provider process.")); + } + if (agentProcess !== null) { + await this.#stopOwnedAgent( + context, + agentProcess, + "turn.cancelled.stop", + Date.now() + ACP_RECYCLE_BUDGET_MS, + ); + } return; } diff --git a/src/runtimes/openai/app-server-driver-backend.ts b/src/runtimes/openai/app-server-driver-backend.ts index c186874..9780fbd 100644 --- a/src/runtimes/openai/app-server-driver-backend.ts +++ b/src/runtimes/openai/app-server-driver-backend.ts @@ -1,3 +1,4 @@ +import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend"; import { isDriverFullAccess } from "../../core/driver-permission-policy"; import { pushLosslessEvents } from "../../core/driver-runtime-io"; import { @@ -10,12 +11,13 @@ import { toDurationMs, } from "../../core/driver-runtime-timing"; import { summarizePath, summarizeRuntimeCommandInput } from "../../observability/driver-debug"; +import type { RuntimeRunCompletedPayload } from "../../protocol/events"; import { driverIdTimeMs, type RunId } from "../../protocol/id"; +import type { NativeCheckpoint } from "../../protocol/native-checkpoint"; import type { DriverRuntime } from "../../protocol/runtime"; import type { DriverStartInput } from "../../protocol/start"; import type { RuntimeCommandInput } from "../../runtime-command"; import { raceWithAbort, settlePromiseWithTimeout } from "../../utils/async"; -import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend"; import { DriverEventPublisher } from "../driver-event-publisher"; import { createRuntimeSourceEventId } from "../runtime-public-id"; import { @@ -23,11 +25,10 @@ import { computeRuntimeBootstrapDigest, writeSkillBootstrapArtifacts, } from "../skill-bootstrap"; -import { OpenAiAppServerClient } from "./app-server-client"; import { openAiAgentTasksClosedEvent } from "./app-server-agent-task-events"; +import { OpenAiAppServerClient } from "./app-server-client"; import { MOSOO_OPENAI_RUNTIME_SANDBOX_MODE } from "./app-server-env"; import { OpenAiAppServerEventBridge } from "./app-server-event-bridge"; -import { createOpenAiNativeCheckpoint, restoreOpenAiNativeCheckpoint } from "./native-checkpoint"; import { toOpenAiProtocolError } from "./app-server-event-mapping"; import type { ApprovalPolicy, @@ -36,6 +37,7 @@ import type { ThreadStartResponse, TurnStatus, } from "./app-server-protocol"; +import { createOpenAiNativeCheckpoint, restoreOpenAiNativeCheckpoint } from "./native-checkpoint"; /** * App-server approval policy derived from the driver permission policy. @@ -122,6 +124,7 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend { #pendingTurnStartServerRequests: Promise | null = null; #pendingTurnStartUpdates: Promise | null = null; #pendingCancellationSettlement: Promise | null = null; + #nativeCheckpoint: NativeCheckpoint | null; #restartThreadId: string | null = null; #stopping = false; #threadId: string | null = null; @@ -154,13 +157,24 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend { push: async (context, reason, events) => this.#eventPublisher.push(context, reason, events), pushSession: async (context, reason, events) => this.#eventPublisher.pushSession(context, reason, events), - pushTerminal: async (context, reason, closures, terminal, cancellationSignal) => - this.#eventPublisher.pushTerminal(context, reason, closures, terminal, cancellationSignal), + pushTerminal: async (context, reason, closures, terminal, cancellationSignal) => { + await this.#eventPublisher.pushTerminal( + context, + reason, + closures, + terminal, + cancellationSignal, + ); + if (terminal.kind === "run.completed") { + this.#nativeCheckpoint = (terminal.payload as RuntimeRunCompletedPayload).checkpoint; + } + }, requireThreadId: () => this.#requireThreadId(), }); constructor(payload: DriverStartInput) { this.#payload = payload; + this.#nativeCheckpoint = payload.execution.session.nativeCheckpoint; } async start(context: AgentDriverContext, signal: AbortSignal): Promise { @@ -183,6 +197,7 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend { restoreOpenAiNativeCheckpoint(this.#payload, signal), ); const nativeResumeThreadId = readResumeThreadId(this.#payload); + this.#nativeCheckpoint = this.#payload.execution.session.nativeCheckpoint; // Resume emits historical usage before its response; bind the verified native identity first. this.#threadId = nativeResumeThreadId; const client = this.#createClient(context); @@ -388,13 +403,43 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend { const client = this.#createClient(context); const startupCancellation = new AbortController(); - const resumeThreadId = this.#restartThreadId ?? this.#threadId; + const resumeThreadId = + this.#nativeCheckpoint !== null ? (this.#restartThreadId ?? this.#threadId) : null; + const checkpoint = this.#nativeCheckpoint; + if (resumeThreadId !== null) { + if (checkpoint === null) { + throw new Error("OpenAI native checkpoint is unavailable for thread resume."); + } + if ( + checkpoint.nativeRef.runtimeId !== this.runtime || + checkpoint.nativeRef.kind !== "openai_thread_id" || + checkpoint.nativeRef.value !== resumeThreadId + ) { + throw new Error("OpenAI native checkpoint does not match the resumed thread."); + } + } this.#threadId = resumeThreadId; this.#client = client; this.#clientStartupCancellation = startupCancellation; this.#clientStopRequested = false; try { + if (checkpoint !== null && resumeThreadId !== null) { + await restoreOpenAiNativeCheckpoint( + { + ...this.#payload, + execution: { + ...this.#payload.execution, + session: { + ...this.#payload.execution.session, + nativeCheckpoint: checkpoint, + nativeResumeRef: checkpoint.nativeRef, + }, + }, + }, + startupCancellation.signal, + ); + } await client.start(startupCancellation.signal); const threadResult = await this.#startThread( client, @@ -1015,7 +1060,7 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend { ): Promise { const isCurrentClient = this.#client === client; if (isCurrentClient) { - this.#restartThreadId = threadId; + this.#restartThreadId = this.#nativeCheckpoint !== null ? threadId : null; this.#clientStopRequested = true; } this.#clientStartupCancellation?.abort(new DriverTurnCancelledError(reason)); diff --git a/src/runtimes/openai/app-server-item-events.ts b/src/runtimes/openai/app-server-item-events.ts index 6adbedc..4eb6723 100644 --- a/src/runtimes/openai/app-server-item-events.ts +++ b/src/runtimes/openai/app-server-item-events.ts @@ -56,6 +56,10 @@ function toOpenAiMessageSnapshotEvents(input: { readonly sourcePrefix: string; readonly text: string; }): DriverEventInput[] { + // Native tool turns can close an assistant item without text. Its lifecycle + // still closes, but there is no text snapshot to publish. + if (input.text.length === 0) return []; + const createAddedEvent = (content: string): DriverEventInput => ({ delivery: "lossless", kind: "message.added", diff --git a/tests/acp-agent-process.test.ts b/tests/acp-agent-process.test.ts index 9659b07..c9d274a 100644 --- a/tests/acp-agent-process.test.ts +++ b/tests/acp-agent-process.test.ts @@ -6,6 +6,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { createAgentDriverContext } from "../src/core/agent-driver-backend"; import { createDisabledLogger } from "../src/observability"; import { createDriverStartInputFromBootPayload } from "../src/protocol/start"; import { @@ -15,7 +16,6 @@ import { } from "../src/runtimes/acp/acp-agent-process"; import { buildChildEnv } from "../src/runtimes/acp/acp-configuration"; import { spawnLinuxProcessTreeWatchdog } from "../src/runtimes/child-process"; -import { createAgentDriverContext } from "../src/core/agent-driver-backend"; import { driverBootPayload, driverStartInput } from "./driver-boot-payload-fixture"; function createHarness() { @@ -183,6 +183,40 @@ describe("ACP agent process lifecycle", () => { } }); + test("stops owning the startup abort after the process is handed off", async () => { + const harness = createHarness(); + const root = await mkdtemp(join(tmpdir(), "driver-acp-startup-handoff-")); + const payload = createPayload(root); + const controller = new AbortController(); + let child: AcpAgentProcess | undefined; + + try { + const started = await startAcpAgentProcess( + harness.context, + payload, + buildChildEnv(payload), + controller.signal, + { + args: ["-e", "setInterval(() => {}, 1000)"], + command: process.execPath, + }, + ); + child = started.process; + await started.ready; + started.releaseStartupAbort(); + controller.abort(new Error("startup deadline elapsed")); + await Bun.sleep(50); + + expect(child.exitCode).toBeNull(); + expect(child.signalCode).toBeNull(); + } finally { + if (child !== undefined && child.exitCode === null && child.signalCode === null) { + await stopAcpAgentProcess(harness.context, child, "test.cleanup").catch(() => {}); + } + await rm(root, { force: true, recursive: true }); + } + }); + test.each(["SIGTERM", "SIGKILL"] as const)("observes exit after %s", async (exitSignal) => { const harness = createHarness(); const process = await startTestAgentProcess( diff --git a/tests/acp-client-request-handler.feature b/tests/acp-client-request-handler.feature new file mode 100644 index 0000000..91e4163 --- /dev/null +++ b/tests/acp-client-request-handler.feature @@ -0,0 +1,8 @@ +Feature: ACP session metadata delivery + + Scenario: Deliver session metadata after a cancelled run terminal + Given a cancelled run terminal is still being acknowledged + When the provider sends a session metadata update after cancellation + Then the metadata waits for the run terminal + And the metadata is delivered as a session event + And the terminal settlement remains the only run terminal diff --git a/tests/acp-client-request-handler.test.ts b/tests/acp-client-request-handler.test.ts index 4a5da6c..00ff229 100644 --- a/tests/acp-client-request-handler.test.ts +++ b/tests/acp-client-request-handler.test.ts @@ -8,18 +8,19 @@ import { DriverPermissionBroker, PermissionEventDeliveryError, } from "../src/core/driver-permission-broker"; -import type { DriverPermissionRequest } from "../src/host-ports"; import { DriverEventRejectedError, type DriverRuntimeEventPort, } from "../src/core/driver-runtime-io"; -import type { DriverEventInput } from "../src/protocol/events"; +import type { DriverPermissionRequest } from "../src/host-ports"; import { createDisabledLogger } from "../src/observability"; -import { AcpClientRequestHandler } from "../src/runtimes/acp/acp-client-request-handler"; +import type { DriverEventInput } from "../src/protocol/events"; +import type { RunId } from "../src/protocol/id"; import { AcpAssistantTranscriptState } from "../src/runtimes/acp/acp-assistant-transcript-state"; +import { AcpClientRequestHandler } from "../src/runtimes/acp/acp-client-request-handler"; import { DriverEventPublisher } from "../src/runtimes/driver-event-publisher"; import { beginAcpTranscript } from "./acp-test-helpers"; -import { driverStartInput } from "./driver-boot-payload-fixture"; +import { driverStartInput, DRIVER_TEST_IDS } from "./driver-boot-payload-fixture"; const BASE_HANDLER_OPTIONS = { allowedRoots: [], @@ -28,6 +29,7 @@ const BASE_HANDLER_OPTIONS = { isCancelling: () => false, nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, + pushSession: async () => {}, } satisfies Omit[0], "push" | "turnEvents">; describe("ACP client request handler", () => { @@ -296,6 +298,9 @@ describe("ACP client request handler", () => { push: async (_context, reason) => { pushedReasons.push(reason); }, + pushSession: async (_context, reason) => { + pushedReasons.push(reason); + }, turnEvents: new AcpAssistantTranscriptState(), }); const notification = { @@ -319,6 +324,89 @@ describe("ACP client request handler", () => { expect(pushedReasons).toEqual(["driver.acp.session.update"]); }); + test("delivers metadata through the session lane after a cancelled terminal", async () => { + const terminalEntered = Promise.withResolvers(); + const releaseTerminal = Promise.withResolvers(); + const metadataPushEntered = Promise.withResolvers(); + const runId: RunId = DRIVER_TEST_IDS.runId; + let activeRunId: RunId | null = runId; + let metadataLane: "run" | "session" | null = null; + let sequence = 0; + const delivered: DriverEventInput[] = []; + const context = createAgentDriverContext({ + eventSink: { + currentRunId: () => activeRunId, + pushEvents: async ({ events }) => { + delivered.push(...events); + if (events.some((event) => event.kind === "run.cancelled")) { + terminalEntered.resolve(); + await releaseTerminal.promise; + } + return { + accepted: events.map((event) => ({ + eventId: event.sourceEventId!, + seq: ++sequence, + type: event.kind, + })), + }; + }, + }, + logger: createDisabledLogger(), + payload: driverStartInput, + permission: { request: async () => "reject_once" }, + }); + const publisher = new DriverEventPublisher("acp-fallback", () => "native-session-1"); + const handler = new AcpClientRequestHandler({ + ...BASE_HANDLER_OPTIONS, + push: async (pushContext, reason, events) => { + metadataLane = "run"; + metadataPushEntered.resolve(); + await publisher.push(pushContext, reason, events); + }, + pushSession: async (pushContext, reason, events) => { + metadataLane = "session"; + metadataPushEntered.resolve(); + await publisher.pushSession(pushContext, reason, events); + }, + turnEvents: beginAcpTranscript(), + }); + const terminal = publisher.pushTerminal(context, "cancelled", [], { + kind: "run.cancelled", + payload: { runId, status: "cancelled" }, + runId, + }); + + await terminalEntered.promise; + const metadata = handler.enqueueUpdate(context, { + sessionId: "native-session-1", + update: { + availableCommands: [{ description: "Resume", name: "resume" }], + sessionUpdate: "available_commands_update", + }, + }); + + let metadataSettled = false; + void metadata.then( + () => { + metadataSettled = true; + }, + () => { + metadataSettled = true; + }, + ); + await metadataPushEntered.promise; + expect(metadataLane).toBe("session"); + expect(metadataSettled).toBe(false); + expect(delivered.map((event) => event.kind)).toEqual(["run.cancelled"]); + releaseTerminal.resolve(); + await Promise.all([terminal, metadata]); + + expect(delivered.map((event) => [event.kind, event.runId ?? null])).toEqual([ + ["run.cancelled", runId], + ["session.commands.updated", null], + ]); + }); + test("passes the runtime environment to terminal child processes", async () => { const handler = new AcpClientRequestHandler({ ...BASE_HANDLER_OPTIONS, @@ -361,6 +449,9 @@ describe("ACP client request handler", () => { await gate.promise; } }, + pushSession: async (_context, reason) => { + pushedReasons.push(reason); + }, turnEvents, }); const update = (text: string) => diff --git a/tests/acp-driver-backend.feature b/tests/acp-driver-backend.feature new file mode 100644 index 0000000..f521531 --- /dev/null +++ b/tests/acp-driver-backend.feature @@ -0,0 +1,20 @@ +Feature: ACP cancelled provider tree recycling + + Scenario: Observe a detached child before recycling a cancelled provider + Given a provider prompt is blocked and the cancellation fixture will spawn a detached child + When ACP cancellation is requested while the cancellation event publication is held + And the fixture publishes the detached child's PID + And the cancellation event publication is released + Then the cancelled provider tree is recycled before terminal delivery + And the detached child is no longer running + And the ACP session resumes for the next turn + + Scenario: Stop an active provider tree before its cancellation terminal + Given a blocked Linux provider prompt has a detached child running + When ACP session stop is requested + Then the detached child is no longer running when run.cancelled is delivered + + Scenario: Keep a resumed provider alive after the recycle deadline + Given a cancelled provider is successfully resumed + When the next turn completes after the recycle deadline + Then the resumed provider remains alive until explicit cleanup diff --git a/tests/acp-driver-backend.test.ts b/tests/acp-driver-backend.test.ts index dbd74d6..8e8fcd0 100644 --- a/tests/acp-driver-backend.test.ts +++ b/tests/acp-driver-backend.test.ts @@ -344,6 +344,14 @@ const handle = (message) => { } if (message.params.prompt[0]?.text === "hang") { pendingPromptId = message.id; + if (latePidPath && process.env.TEST_SPAWN_EARLY_CHILD === "1") { + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { + detached: true, + stdio: "ignore", + }); + appendFileSync(latePidPath, String(child.pid) + "\n"); + child.unref(); + } send({ jsonrpc: "2.0", method: "session/update", @@ -657,6 +665,7 @@ async function createHarness( onEvents?(events: readonly DriverEventInput[]): void; readonly permission?: AgentDriverPermissionPort["request"]; readonly spawnLateChild?: boolean; + readonly spawnEarlyChild?: boolean; readonly updateBeforeSessionResponse?: boolean; readonly writeOnClose?: boolean; } = {}, @@ -694,6 +703,7 @@ async function createHarness( TEST_HANG_CLOSE: options.hangClose ? "1" : "0", TEST_METADATA_ON_RESUME: options.metadataOnResume ? "1" : "0", TEST_SPAWN_LATE_CHILD: options.spawnLateChild ? "1" : "0", + TEST_SPAWN_EARLY_CHILD: options.spawnEarlyChild ? "1" : "0", TEST_UPDATE_BEFORE_SESSION_RESPONSE: options.updateBeforeSessionResponse ? "1" : "0", ...(options.authenticate ? { MOSOO_ACP_AUTH_METHOD_ID: "test-auth" } : {}), }, @@ -752,6 +762,7 @@ async function createHarness( toRuntimeEventInput( { createId: () => createDriverId() as EventId, + ...(eventRunId === null ? { draftRunIdPolicy: "ignore" as const } : {}), driverInstanceId: DRIVER_TEST_IDS.driverInstanceId, occurredAt: "2026-10-10T00:00:00.000Z", ...(eventRunId === null ? {} : { runId: eventRunId }), @@ -1749,14 +1760,12 @@ describe("ACP driver backend lifecycle", () => { blockResume: true, spawnLateChild: true, }); + const firstRunId = createDriverId() as RunId; + const secondRunId = createDriverId() as RunId; try { const eventOffset = harness.events.length; - const input = harness.backend.handleInput( - harness.context, - { text: "hang" }, - DRIVER_TEST_IDS.runId as RunId, - ); + const input = harness.backend.handleInput(harness.context, { text: "hang" }, firstRunId); void input.catch(() => {}); await waitForAcpTestCondition( async () => (await harness.methods()).includes("session/prompt"), @@ -1770,6 +1779,10 @@ describe("ACP driver backend lifecycle", () => { async () => (await harness.methods()).includes("session/cancel"), "ACP session/cancel request", ); + await waitForAcpTestCondition( + async () => (await readFile(harness.latePidPath, "utf8").catch(() => "")).trim() !== "", + "ACP late child PID publication", + ); gate.release(); await expect(cancellation).resolves.toBeUndefined(); await waitForAcpTestCondition( @@ -1810,12 +1823,14 @@ describe("ACP driver backend lifecycle", () => { expect(harness.lifecycleFailures).toEqual([]); await expect( - harness.backend.handleInput( - harness.context, - { text: "after recycle" }, - DRIVER_TEST_IDS.secondRunId as RunId, - ), + harness.backend.handleInput(harness.context, { text: "after recycle" }, secondRunId), ).resolves.toBeUndefined(); + const latestAgentPid = Number.parseInt( + (await readFile(harness.agentPidPath, "utf8")).trim().split("\n").at(-1) ?? "0", + 10, + ); + await Bun.sleep(4_600); + expect(isRunning(latestAgentPid)).toBe(true); } finally { await writeFile(harness.resumeGatePath, "cleanup").catch(() => {}); await harness.destroy(); @@ -1899,6 +1914,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -1951,6 +1967,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -2009,6 +2026,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -2068,6 +2086,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -2134,6 +2153,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -2207,6 +2227,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -2287,6 +2308,7 @@ describe("ACP driver backend lifecycle", () => { nativeSessionId: () => "native-session-1", onUpdateFailure: () => {}, push: async () => {}, + pushSession: async () => {}, turnEvents: turn.events, }); const connection = { @@ -3016,70 +3038,93 @@ describe("ACP driver backend lifecycle", () => { } }, 7_000); - test("stops an active session without resuming its cancelled provider", async () => { - const harness = await createHarness(); - const runtimeState = new DriverRuntimeStateMachine("ready"); - const socket = new FakeDriverRuntimeIo([ - { - commandId: "acp-stop-input", - input: { text: "hang" }, - kind: "input.start", - requestId: "acp-stop-request", - runId: DRIVER_TEST_IDS.runId, - }, - { - commandId: "acp-stop-session", - kind: "session.stop", - reason: "test.stop", - }, - ]); - const nextCommand = socket.nextCommand.bind(socket); - socket.nextCommand = async (signal) => { - const command = await nextCommand(signal); - if (command?.kind === "session.stop") { - await waitForAcpTestCondition( - async () => (await harness.methods()).includes("session/prompt"), - "ACP session/prompt request", - ); - } - return command; - }; - const { dispatcher, logger } = createDispatcher({ - backend: harness.backend, - isShuttingDown: () => runtimeState.isShuttingDown(), - runtimeState, - shutdown: async (_runtimeIo, reason) => - harness.backend.stop(harness.context, reason, new AbortController().signal), - }); + test.skipIf(process.platform !== "linux")( + "stops an active session without resuming its cancelled provider", + async () => { + let detachedChildLiveAtCancellation: boolean | null = null; + const harness = await createHarness({ spawnEarlyChild: true }); + const runtimeState = new DriverRuntimeStateMachine("ready"); + const socket = new FakeDriverRuntimeIo([ + { + commandId: "acp-stop-input", + input: { text: "hang" }, + kind: "input.start", + requestId: "acp-stop-request", + runId: DRIVER_TEST_IDS.runId, + }, + { + commandId: "acp-stop-session", + kind: "session.stop", + reason: "test.stop", + }, + ]); + const nextCommand = socket.nextCommand.bind(socket); + socket.nextCommand = async (signal) => { + const command = await nextCommand(signal); + if (command?.kind === "session.stop") { + await waitForAcpTestCondition( + async () => (await harness.methods()).includes("session/prompt"), + "ACP session/prompt request", + ); + await waitForAcpTestCondition( + async () => (await readFile(harness.latePidPath, "utf8").catch(() => "")).trim() !== "", + "ACP early child PID publication before session stop", + ); + const earlyPid = Number.parseInt(readFileSync(harness.latePidPath, "utf8").trim(), 10); + expect(isRunning(earlyPid)).toBe(true); + } + return command; + }; + const pushEvents = socket.pushEvents.bind(socket); + socket.pushEvents = async (input) => { + if (input.events.some(({ kind }) => kind === "run.cancelled")) { + await waitForAcpTestCondition( + async () => (await readFile(harness.latePidPath, "utf8").catch(() => "")).trim() !== "", + "ACP late child PID publication before cancellation terminal", + ); + const latePid = Number.parseInt(readFileSync(harness.latePidPath, "utf8").trim(), 10); + detachedChildLiveAtCancellation = isRunning(latePid); + } + return pushEvents(input); + }; + const { dispatcher, logger } = createDispatcher({ + backend: harness.backend, + isShuttingDown: () => runtimeState.isShuttingDown(), + runtimeState, + shutdown: async (_runtimeIo, reason) => + harness.backend.stop(harness.context, reason, new AbortController().signal), + }); - try { - await dispatcher.run(socket, logger); + try { + await dispatcher.run(socket, logger); - const events = socket.pushedEvents.flatMap(({ events }) => events); - expect( - events - .filter( - ({ kind }) => - kind === "run.cancelled" || kind === "run.completed" || kind === "run.failed", - ) - .map(({ kind }) => kind), - ).toEqual(["run.cancelled"]); - expect(events.filter(({ kind }) => kind === "run.cancel.requested")).toHaveLength(1); - expect(await harness.methods()).not.toContain("session/resume"); - expect(socket.updates).toContainEqual({ - commandId: "acp-stop-input", - status: "cancelled", - }); - expect(socket.updates).toContainEqual({ - commandId: "acp-stop-session", - status: "completed", - }); - expect(socket.completedRunReasons).toEqual(["completed"]); - expect(runtimeState.status()).toBe("stopped"); - } finally { - await harness.destroy(); - } - }); + const events = socket.pushedEvents.flatMap(({ events }) => events); + expect( + events + .filter( + ({ kind }) => + kind === "run.cancelled" || kind === "run.completed" || kind === "run.failed", + ) + .map(({ kind }) => kind), + ).toEqual(["run.cancelled"]); + expect(events.filter(({ kind }) => kind === "run.cancel.requested")).toHaveLength(1); + expect(await harness.methods()).not.toContain("session/resume"); + expect(detachedChildLiveAtCancellation).toBe(false); + expect(socket.updates).toContainEqual({ + commandId: "acp-stop-input", + status: "cancelled", + }); + expect(socket.updates).toContainEqual({ + commandId: "acp-stop-session", + status: "completed", + }); + expect(socket.completedRunReasons).toEqual(["completed"]); + expect(runtimeState.status()).toBe("stopped"); + } finally { + await harness.destroy(); + } + }, + ); test("drains burst updates sent immediately before the prompt response", async () => { const harness = await createHarness(); diff --git a/tests/driver-artifact-live-message-terminal.feature b/tests/driver-artifact-live-message-terminal.feature new file mode 100644 index 0000000..e8cd4e1 --- /dev/null +++ b/tests/driver-artifact-live-message-terminal.feature @@ -0,0 +1,16 @@ +Feature: Message and thought lifecycle terminal pairing + + Scenario: Accept every canonical message terminal outcome + Given a message.started event has one message ID + When the lifecycle assertion validates its terminal event + Then message.completed, message.cancelled, and message.failed each close exactly one message + + Scenario: Accept every canonical thought terminal outcome + Given a thought.started event has one thought ID + When the lifecycle assertion validates its terminal event + Then thought.completed and thought.cancelled each close exactly one thought + + Scenario: Reject non-terminal message events + Given a message event has a kind outside the message terminal contract + When the lifecycle assertion validates its terminal event + Then the event is rejected diff --git a/tests/driver-artifact-live-message-terminal.test.ts b/tests/driver-artifact-live-message-terminal.test.ts new file mode 100644 index 0000000..757e01f --- /dev/null +++ b/tests/driver-artifact-live-message-terminal.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, test } from "bun:test"; + +import { + CANONICAL_MESSAGE_TERMINAL_KINDS, + CANONICAL_THOUGHT_TERMINAL_KINDS, + isCanonicalMessageTerminalKind, + isCanonicalThoughtTerminalKind, +} from "./driver-artifact-live-message-terminal"; + +describe("packed driver lifecycle terminal predicates", () => { + test("accepts every canonical message terminal outcome", () => { + expect(CANONICAL_MESSAGE_TERMINAL_KINDS).toEqual([ + "message.cancelled", + "message.completed", + "message.failed", + ]); + + for (const kind of CANONICAL_MESSAGE_TERMINAL_KINDS) { + expect(isCanonicalMessageTerminalKind(kind)).toBe(true); + } + }); + + test.each([ + "message.started", + "message.delta", + "message.added", + "item.completed", + "run.failed", + "message.unknown", + ])('rejects "%s"', (kind) => { + expect(isCanonicalMessageTerminalKind(kind)).toBe(false); + }); + + test("accepts every canonical thought terminal outcome", () => { + expect(CANONICAL_THOUGHT_TERMINAL_KINDS).toEqual(["thought.cancelled", "thought.completed"]); + + for (const kind of CANONICAL_THOUGHT_TERMINAL_KINDS) { + expect(isCanonicalThoughtTerminalKind(kind)).toBe(true); + } + }); + + test.each(["thought.started", "thought.delta", "message.completed", "run.cancelled"])( + 'rejects "%s" as a thought terminal', + (kind) => { + expect(isCanonicalThoughtTerminalKind(kind)).toBe(false); + }, + ); +}); diff --git a/tests/driver-artifact-live-message-terminal.ts b/tests/driver-artifact-live-message-terminal.ts new file mode 100644 index 0000000..972ac9b --- /dev/null +++ b/tests/driver-artifact-live-message-terminal.ts @@ -0,0 +1,29 @@ +export const CANONICAL_MESSAGE_TERMINAL_KINDS = [ + "message.cancelled", + "message.completed", + "message.failed", +] as const; + +export type CanonicalMessageTerminalKind = (typeof CANONICAL_MESSAGE_TERMINAL_KINDS)[number]; + +export const CANONICAL_THOUGHT_TERMINAL_KINDS = ["thought.cancelled", "thought.completed"] as const; + +export type CanonicalThoughtTerminalKind = (typeof CANONICAL_THOUGHT_TERMINAL_KINDS)[number]; + +export function isCanonicalMessageTerminalKind( + value: unknown, +): value is CanonicalMessageTerminalKind { + return ( + typeof value === "string" && + (CANONICAL_MESSAGE_TERMINAL_KINDS as readonly string[]).includes(value) + ); +} + +export function isCanonicalThoughtTerminalKind( + value: unknown, +): value is CanonicalThoughtTerminalKind { + return ( + typeof value === "string" && + (CANONICAL_THOUGHT_TERMINAL_KINDS as readonly string[]).includes(value) + ); +} diff --git a/tests/driver-artifact-live-permission.feature b/tests/driver-artifact-live-permission.feature new file mode 100644 index 0000000..7755918 --- /dev/null +++ b/tests/driver-artifact-live-permission.feature @@ -0,0 +1,8 @@ +Feature: Supervised permission terminal outcomes + + Scenario: Preserve provider-specific terminal semantics after rejecting a permission + Given a supervised permission request is resolved with reject_once + When the provider handles the rejected tool request + Then OpenAI interrupted turns produce run.cancelled and a cancelled input command + And other live runtimes produce run.completed and a completed input command + And permission lifecycle and rejected file assertions remain valid diff --git a/tests/driver-artifact-live-runtime-notice.feature b/tests/driver-artifact-live-runtime-notice.feature new file mode 100644 index 0000000..de36d24 --- /dev/null +++ b/tests/driver-artifact-live-runtime-notice.feature @@ -0,0 +1,26 @@ +Feature: OpenAI runtime notices in artifact lifecycle checks + + Scenario: Allow a known driver runtime notice before run.started + Given a message.added event has driver actor and origin, agent role, and a known runtime notice subtype + When the packed driver lifecycle assertion checks projection ordering + Then the notice may precede run.started + + Scenario: Keep ordinary agent output subject to start ordering + Given a message.added event has driver actor and origin with the agent role and ordinary output content + When the streamed-message pairing assertion checks the output + Then the output must have message.started and message.completed events + + Scenario: Keep a known driver runtime notice standalone + Given a message.added event has driver actor and origin, agent role, and a known runtime notice subtype + When the streamed-message pairing assertion checks the notice + Then the notice does not require message.started or message.completed events + + Scenario: Ignore a known runtime notice when checking resume history + Given a known driver runtime notice precedes a resumed run.started event + When the historical projection assertion checks events before run.started + Then the notice is allowed while historical output remains rejected + + Scenario: Reject runtime notices with untrusted metadata + Given a message.added event has a foreign actor or origin, or an unknown subtype + When the packed driver lifecycle assertion classifies the event + Then it is not treated as a known runtime notice diff --git a/tests/driver-artifact-live-runtime-notice.test.ts b/tests/driver-artifact-live-runtime-notice.test.ts new file mode 100644 index 0000000..c328d2e --- /dev/null +++ b/tests/driver-artifact-live-runtime-notice.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, test } from "bun:test"; + +import { + isOpenAiRuntimeNotice, + OPENAI_RUNTIME_NOTICE_SUBTYPES, +} from "./driver-artifact-live-runtime-notice"; + +interface TestEvent { + readonly actor: string; + readonly kind: string; + readonly origin: string; + readonly payload: Record; +} + +function runtimeNotice(overrides: Partial = {}): TestEvent { + return { + actor: "driver", + kind: "message.added", + origin: "driver", + payload: { + content: "A runtime notice.", + messageId: "message-1", + role: "agent", + subtype: "warning", + }, + ...overrides, + }; +} + +describe("OpenAI runtime notice predicate", () => { + test.each(OPENAI_RUNTIME_NOTICE_SUBTYPES)("recognizes the %s subtype", (subtype) => { + expect( + isOpenAiRuntimeNotice(runtimeNotice({ payload: { ...runtimeNotice().payload, subtype } })), + ).toBe(true); + }); + + test("keeps ordinary agent output in lifecycle pairing", () => { + expect( + isOpenAiRuntimeNotice( + runtimeNotice({ + payload: { + content: "The answer.", + messageId: "message-1", + role: "agent", + }, + }), + ), + ).toBe(false); + }); + + test("allows known runtime notices to remain standalone", () => { + expect(isOpenAiRuntimeNotice(runtimeNotice())).toBe(true); + }); + + test.each([ + ["actor", { actor: "agent" }], + ["origin", { origin: "runtime" }], + ] as const)("rejects a foreign %s", (_label, override) => { + expect(isOpenAiRuntimeNotice(runtimeNotice(override))).toBe(false); + }); + + test("rejects an unknown subtype", () => { + expect( + isOpenAiRuntimeNotice( + runtimeNotice({ payload: { ...runtimeNotice().payload, subtype: "warning_extra" } }), + ), + ).toBe(false); + }); + + test("requires the agent role and message.added kind", () => { + expect( + isOpenAiRuntimeNotice( + runtimeNotice({ + kind: "message.delta", + payload: { ...runtimeNotice().payload, role: "user" }, + }), + ), + ).toBe(false); + }); +}); diff --git a/tests/driver-artifact-live-runtime-notice.ts b/tests/driver-artifact-live-runtime-notice.ts new file mode 100644 index 0000000..ce55c3d --- /dev/null +++ b/tests/driver-artifact-live-runtime-notice.ts @@ -0,0 +1,35 @@ +export const OPENAI_RUNTIME_NOTICE_SUBTYPES = [ + "warning", + "guardian_warning", + "model_provider_auth_recovery_started", + "model_provider_auth_recovery_completed", + "strict_review_required", + "windows_world_writable_warning", +] as const; + +interface RuntimeNoticeEvent { + readonly actor?: string | undefined; + readonly kind: string; + readonly origin?: string | undefined; + readonly payload: unknown; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +export function isOpenAiRuntimeNotice(event: RuntimeNoticeEvent): boolean { + if (event.kind !== "message.added" || event.actor !== "driver" || event.origin !== "driver") { + return false; + } + + if (!isRecord(event.payload) || event.payload["role"] !== "agent") { + return false; + } + + const subtype = event.payload["subtype"]; + return ( + typeof subtype === "string" && + (OPENAI_RUNTIME_NOTICE_SUBTYPES as readonly string[]).includes(subtype) + ); +} diff --git a/tests/driver-artifact-live-tool-status.feature b/tests/driver-artifact-live-tool-status.feature new file mode 100644 index 0000000..00da33f --- /dev/null +++ b/tests/driver-artifact-live-tool-status.feature @@ -0,0 +1,23 @@ +Feature: Tool call status validation in artifact lifecycle checks + + Scenario: Accept every canonical tool call status + Given a packed driver emits a tool.call.updated event + When the lifecycle assertion validates its status + Then cancelled, completed, failed, and running are accepted + + Scenario: Reject statuses outside the tool call contract + Given a packed driver emits a tool.call.updated event + When the lifecycle assertion validates an unknown status + Then the status is rejected + + Scenario: Allow authoritative terminal patches without reopening the tool + Given a tool emits running, failed, and authoritative failed updates + When the lifecycle assertion summarizes the updates + Then the tool has a running update before terminal updates + And every terminal update has the same status + And no running update follows a terminal update + + Scenario: Reject a tool that reopens after terminal status + Given a tool emits running, completed, and running updates + When the lifecycle assertion summarizes the updates + Then a running update follows a terminal update diff --git a/tests/driver-artifact-live-tool-status.test.ts b/tests/driver-artifact-live-tool-status.test.ts new file mode 100644 index 0000000..6917465 --- /dev/null +++ b/tests/driver-artifact-live-tool-status.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, test } from "bun:test"; + +import { + CANONICAL_TOOL_CALL_STATUSES, + inspectToolCallUpdateLifecycle, + isCanonicalToolCallStatus, +} from "./driver-artifact-live-tool-status"; + +describe("packed driver tool call status predicate", () => { + test("accepts every canonical protocol status", () => { + expect(CANONICAL_TOOL_CALL_STATUSES).toEqual(["cancelled", "completed", "failed", "running"]); + + for (const status of CANONICAL_TOOL_CALL_STATUSES) { + expect(isCanonicalToolCallStatus(status)).toBe(true); + } + }); + + test.each(["", "queued", "pending", "unknown"])('rejects "%s"', (status) => { + expect(isCanonicalToolCallStatus(status)).toBe(false); + }); + + test("allows authoritative terminal patches without reopening the tool", () => { + expect(inspectToolCallUpdateLifecycle(["running", "failed", "failed"])).toEqual({ + firstTerminalIndex: 1, + hasRunningUpdate: true, + runningAfterTerminal: false, + terminalStatuses: ["failed", "failed"], + }); + }); + + test("detects a running update after terminal status", () => { + expect(inspectToolCallUpdateLifecycle(["running", "completed", "running"])).toEqual({ + firstTerminalIndex: 1, + hasRunningUpdate: true, + runningAfterTerminal: true, + terminalStatuses: ["completed"], + }); + }); +}); diff --git a/tests/driver-artifact-live-tool-status.ts b/tests/driver-artifact-live-tool-status.ts new file mode 100644 index 0000000..d28cde2 --- /dev/null +++ b/tests/driver-artifact-live-tool-status.ts @@ -0,0 +1,49 @@ +import type { RuntimeToolCallUpdatedPayload } from "../src/protocol/events/runtime-event-types"; + +export const CANONICAL_TOOL_CALL_STATUSES = [ + "cancelled", + "completed", + "failed", + "running", +] as const satisfies readonly RuntimeToolCallUpdatedPayload["status"][]; + +export type CanonicalToolCallStatus = (typeof CANONICAL_TOOL_CALL_STATUSES)[number]; + +export interface ToolCallUpdateLifecycle { + readonly firstTerminalIndex: number | null; + readonly hasRunningUpdate: boolean; + readonly runningAfterTerminal: boolean; + readonly terminalStatuses: readonly Exclude[]; +} + +export function isCanonicalToolCallStatus( + value: unknown, +): value is RuntimeToolCallUpdatedPayload["status"] { + return ( + typeof value === "string" && (CANONICAL_TOOL_CALL_STATUSES as readonly string[]).includes(value) + ); +} + +export function inspectToolCallUpdateLifecycle( + statuses: readonly unknown[], +): ToolCallUpdateLifecycle { + const firstTerminalIndex = statuses.findIndex((status) => status !== "running"); + const terminalStatuses = statuses.flatMap>( + (status) => { + if (status === "cancelled" || status === "completed" || status === "failed") { + return [status]; + } + return []; + }, + ); + + return { + firstTerminalIndex: firstTerminalIndex === -1 ? null : firstTerminalIndex, + hasRunningUpdate: statuses.includes("running"), + runningAfterTerminal: + firstTerminalIndex === -1 + ? false + : statuses.slice(firstTerminalIndex + 1).some((status) => status === "running"), + terminalStatuses, + }; +} diff --git a/tests/driver-artifact-live-usage.feature b/tests/driver-artifact-live-usage.feature new file mode 100644 index 0000000..5bc18e7 --- /dev/null +++ b/tests/driver-artifact-live-usage.feature @@ -0,0 +1,11 @@ +Feature: ACP live usage metadata validation + + Scenario: Accept native OpenCode step totals + Given an ACP fallback run emits usage from OpenCode native step totals + When the lifecycle assertion validates the usage metadata + Then the native_step_totals source and anthropic_bucketed contract are accepted + + Scenario: Reject stale or unknown ACP usage metadata + Given an ACP fallback run emits usage metadata + When the lifecycle assertion validates a non-native source or contract + Then the usage metadata is rejected diff --git a/tests/driver-artifact-live-usage.test.ts b/tests/driver-artifact-live-usage.test.ts new file mode 100644 index 0000000..43de4bd --- /dev/null +++ b/tests/driver-artifact-live-usage.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, test } from "bun:test"; + +import { isCanonicalAcpFallbackUsageMetadata } from "./driver-artifact-live-usage"; + +describe("ACP live usage metadata predicate", () => { + test("accepts native OpenCode step totals with the ACP contract", () => { + expect(isCanonicalAcpFallbackUsageMetadata("native_step_totals", "anthropic_bucketed")).toBe( + true, + ); + }); + + test.each([ + ["prompt_response", "anthropic_bucketed"], + ["session_update", "anthropic_bucketed"], + ["native_step_totals", "openai_runtime_total_with_cached_breakdown"], + ["unknown", "anthropic_bucketed"], + ] as const)("rejects %s with %s", (source, usageContract) => { + expect(isCanonicalAcpFallbackUsageMetadata(source, usageContract)).toBe(false); + }); +}); diff --git a/tests/driver-artifact-live-usage.ts b/tests/driver-artifact-live-usage.ts new file mode 100644 index 0000000..ae5722b --- /dev/null +++ b/tests/driver-artifact-live-usage.ts @@ -0,0 +1,17 @@ +import { ACP_USAGE_CONTRACT } from "../src/runtimes/acp/acp-types"; + +export const ACP_FALLBACK_USAGE_METADATA = { + sources: ["native_step_totals"], + usageContract: ACP_USAGE_CONTRACT, +} as const; + +export function isCanonicalAcpFallbackUsageMetadata( + source: unknown, + usageContract: unknown, +): boolean { + return ( + typeof source === "string" && + (ACP_FALLBACK_USAGE_METADATA.sources as readonly string[]).includes(source) && + usageContract === ACP_FALLBACK_USAGE_METADATA.usageContract + ); +} diff --git a/tests/driver-artifact-live-workspace.feature b/tests/driver-artifact-live-workspace.feature new file mode 100644 index 0000000..28feab0 --- /dev/null +++ b/tests/driver-artifact-live-workspace.feature @@ -0,0 +1,7 @@ +Feature: Live workspace CRUD preserves the requested file bytes + + Scenario: Edit a Unicode path while preserving its trailing newline + Given the workspace contains a file named "edit target-δΈ­.txt" with "before" and a trailing newline + When the live prompt requests replacing "before" with "after" and preserving the trailing newline + Then the file contains exactly "after" and a trailing newline + And the requested read, create, and delete operations also complete diff --git a/tests/driver-artifact-live.test.ts b/tests/driver-artifact-live.test.ts index c4145fc..cbc1353 100644 --- a/tests/driver-artifact-live.test.ts +++ b/tests/driver-artifact-live.test.ts @@ -9,6 +9,20 @@ import { isAbsolute, join, resolve } from "node:path"; import { DRIVER_PROTOCOL_VERSION, type DriverModelConnection } from "../src/protocol/boot"; import { parseRunId } from "../src/protocol/id"; import { parseNativeCheckpoint, type NativeCheckpoint } from "../src/protocol/native-checkpoint"; +import { + CANONICAL_MESSAGE_TERMINAL_KINDS, + CANONICAL_THOUGHT_TERMINAL_KINDS, + isCanonicalMessageTerminalKind, + isCanonicalThoughtTerminalKind, + type CanonicalMessageTerminalKind, + type CanonicalThoughtTerminalKind, +} from "./driver-artifact-live-message-terminal"; +import { isOpenAiRuntimeNotice } from "./driver-artifact-live-runtime-notice"; +import { + inspectToolCallUpdateLifecycle, + isCanonicalToolCallStatus, +} from "./driver-artifact-live-tool-status"; +import { ACP_FALLBACK_USAGE_METADATA } from "./driver-artifact-live-usage"; import { DriverArtifactTestController, expectedDriverCapabilities, @@ -135,10 +149,7 @@ interface StartBoundaryObservation { } const EXPECTED_USAGE_METADATA = { - "acp-fallback": { - sources: ["prompt_response", "session_update"], - usageContract: "anthropic_bucketed", - }, + "acp-fallback": ACP_FALLBACK_USAGE_METADATA, "claude-agent-sdk": { sources: ["session_update"], usageContract: "anthropic_bucketed", @@ -464,22 +475,37 @@ function isPromptInputProjection(event: DriverArtifactTestEvent): boolean { ); } +type PairedLifecycleTerminalKind = + | "item.completed" + | CanonicalMessageTerminalKind + | CanonicalThoughtTerminalKind; + +function isPairedLifecycleTerminal( + event: DriverArtifactTestEvent, + terminalKinds: readonly PairedLifecycleTerminalKind[], +): boolean { + return terminalKinds.includes(event.kind as PairedLifecycleTerminalKind); +} + function expectPairedLifecycle( events: readonly DriverArtifactTestEvent[], startedKind: "item.started" | "message.started" | "thought.started", - completedKind: "item.completed" | "message.completed" | "thought.completed", + terminalKinds: readonly PairedLifecycleTerminalKind[], idField: "itemId" | "messageId" | "thoughtId", runTerminalIndex: number, ): void { for (const event of events.filter( - (candidate) => candidate.kind === startedKind || candidate.kind === completedKind, + (candidate) => + candidate.kind === startedKind || isPairedLifecycleTerminal(candidate, terminalKinds), )) { expect(payloadString(event, idField)).not.toBeNull(); } const ids = new Set( events - .filter((event) => event.kind === startedKind || event.kind === completedKind) + .filter( + (event) => event.kind === startedKind || isPairedLifecycleTerminal(event, terminalKinds), + ) .map((event) => payloadString(event, idField)) .filter((id): id is string => id !== null), ); @@ -488,14 +514,17 @@ function expectPairedLifecycle( const started = events .map((event, index) => ({ event, index })) .filter(({ event }) => event.kind === startedKind && payloadString(event, idField) === id); - const completed = events + const terminals = events .map((event, index) => ({ event, index })) - .filter(({ event }) => event.kind === completedKind && payloadString(event, idField) === id); + .filter( + ({ event }) => + isPairedLifecycleTerminal(event, terminalKinds) && payloadString(event, idField) === id, + ); expect(started).toHaveLength(1); - expect(completed).toHaveLength(1); - expect(started[0]!.index).toBeLessThan(completed[0]!.index); - expect(completed[0]!.index).toBeLessThan(runTerminalIndex); + expect(terminals).toHaveLength(1); + expect(started[0]!.index).toBeLessThan(terminals[0]!.index); + expect(terminals[0]!.index).toBeLessThan(runTerminalIndex); } } @@ -530,7 +559,7 @@ function expectSingleRunLifecycle( for (const [index, event] of runEvents.entries()) { if (isRunProjection(event)) { - if (!isPromptInputProjection(event)) { + if (!isPromptInputProjection(event) && !isOpenAiRuntimeNotice(event)) { expect(startedIndex).toBeLessThan(index); } expect(index).toBeLessThan(terminalIndex); @@ -540,15 +569,15 @@ function expectSingleRunLifecycle( expectPairedLifecycle( runEvents, "message.started", - "message.completed", + CANONICAL_MESSAGE_TERMINAL_KINDS, "messageId", terminalIndex, ); - expectPairedLifecycle(runEvents, "item.started", "item.completed", "itemId", terminalIndex); + expectPairedLifecycle(runEvents, "item.started", ["item.completed"], "itemId", terminalIndex); expectPairedLifecycle( runEvents, "thought.started", - "thought.completed", + CANONICAL_THOUGHT_TERMINAL_KINDS, "thoughtId", terminalIndex, ); @@ -557,6 +586,9 @@ function expectSingleRunLifecycle( if (event.kind !== "message.delta" && event.kind !== "message.added") { continue; } + if (isOpenAiRuntimeNotice(event)) { + continue; + } if (event.kind === "message.added" && payloadRecord(event)["role"] !== "agent") { continue; } @@ -571,15 +603,15 @@ function expectSingleRunLifecycle( (candidate) => candidate.kind === "message.started" && payloadString(candidate, "messageId") === messageId, ); - const messageCompletedIndex = runEvents.findIndex( + const messageTerminalIndex = runEvents.findIndex( (candidate) => - candidate.kind === "message.completed" && + isCanonicalMessageTerminalKind(candidate.kind) && payloadString(candidate, "messageId") === messageId, ); expect(messageStartedIndex).toBeGreaterThanOrEqual(0); - expect(messageCompletedIndex).toBeGreaterThanOrEqual(0); + expect(messageTerminalIndex).toBeGreaterThanOrEqual(0); expect(messageStartedIndex).toBeLessThan(index); - expect(index).toBeLessThan(messageCompletedIndex); + expect(index).toBeLessThan(messageTerminalIndex); } for (const [index, event] of runEvents.entries()) { @@ -596,21 +628,21 @@ function expectSingleRunLifecycle( (candidate) => candidate.kind === "thought.started" && payloadString(candidate, "thoughtId") === thoughtId, ); - const thoughtCompletedIndex = runEvents.findIndex( + const thoughtTerminalIndex = runEvents.findIndex( (candidate) => - candidate.kind === "thought.completed" && + isCanonicalThoughtTerminalKind(candidate.kind) && payloadString(candidate, "thoughtId") === thoughtId, ); expect(thoughtStartedIndex).toBeGreaterThanOrEqual(0); - expect(thoughtCompletedIndex).toBeGreaterThanOrEqual(0); + expect(thoughtTerminalIndex).toBeGreaterThanOrEqual(0); expect(thoughtStartedIndex).toBeLessThan(index); - expect(index).toBeLessThan(thoughtCompletedIndex); + expect(index).toBeLessThan(thoughtTerminalIndex); } const toolEvents = runEvents.filter((event) => event.kind === "tool.call.updated"); for (const event of toolEvents) { expect(payloadString(event, "toolCallId")).not.toBeNull(); - expect(["completed", "failed", "running"]).toContain(String(payloadRecord(event)["status"])); + expect(isCanonicalToolCallStatus(payloadRecord(event)["status"])).toBe(true); } for (const toolCallId of new Set( @@ -627,15 +659,15 @@ function expectSingleRunLifecycle( const terminalUpdates = updates.filter( ({ event }) => payloadRecord(event)["status"] !== "running", ); + const lifecycle = inspectToolCallUpdateLifecycle( + updates.map(({ event }) => payloadRecord(event)["status"]), + ); - expect(updates.some(({ event }) => payloadRecord(event)["status"] === "running")).toBe(true); - expect(terminalUpdates).toHaveLength(1); - expect( - updates.every( - ({ event, index }) => - payloadRecord(event)["status"] !== "running" || index < terminalUpdates[0]!.index, - ), - ).toBe(true); + expect(lifecycle.hasRunningUpdate).toBe(true); + expect(lifecycle.terminalStatuses.length).toBeGreaterThan(0); + expect(new Set(lifecycle.terminalStatuses).size).toBe(1); + expect(lifecycle.runningAfterTerminal).toBe(false); + expect(terminalUpdates.length).toBe(lifecycle.terminalStatuses.length); } const usageEvents = runEvents.filter((candidate) => candidate.kind === "usage.updated"); @@ -733,7 +765,12 @@ function expectNoHistoricalRunProjections( expect( controller.events .slice(0, runStartedIndex) - .filter((event) => isRunProjection(event) && !isPromptInputProjection(event)), + .filter( + (event) => + isRunProjection(event) && + !isPromptInputProjection(event) && + !isOpenAiRuntimeNotice(event), + ), ).toEqual([]); } @@ -1074,7 +1111,7 @@ async function testWorkspace(runtimeCase: LiveRuntimeCase): Promise { await withController(runtimeCase, paths, async (controller) => { const events = await runTurn( controller, - `Use available file and shell tools to perform every step in order: read ${JSON.stringify(sourceName)}; create ${JSON.stringify(outputName)} containing exactly the token you read; replace before with after in ${JSON.stringify(editName)}; delete ${JSON.stringify(deleteName)}. Reply with exactly workspace-done.`, + `Use available file and shell tools to perform every step in order: read ${JSON.stringify(sourceName)}; create ${JSON.stringify(outputName)} containing exactly the token you read; replace before with after in ${JSON.stringify(editName)} and preserve its trailing newline; delete ${JSON.stringify(deleteName)}. Reply with exactly workspace-done.`, ); expect((await readFile(join(paths.workspacePath, outputName), "utf8")).trim()).toBe(token); @@ -2332,6 +2369,10 @@ async function testSupervisedPermission(runtimeCase: LiveRuntimeCase): Promise event.runId === rejected.runId && event.kind === "run.completed", + (event) => event.runId === rejected.runId && event.kind === rejectedTerminalKind, rejected.eventIndex, LIVE_TURN_TIMEOUT_MS, - "rejected run completion", + "rejected run terminal", ), ]); expect(rejectUpdate.status).toBe("completed"); - expect(rejectedUpdate.status).toBe("completed"); + expect(rejectedUpdate.status).toBe(rejectedInputCommandStatus); expect(payloadRecord(rejectedResolution)).toMatchObject({ outcome: "reject_once", reason: "rejected", requestId: rejectedRequestId, }); - expect(rejectedEvent.kind).toBe("run.completed"); + expect(rejectedEvent.kind).toBe(rejectedTerminalKind); expectPermissionLifecycle( controller.eventsSince(rejected.eventIndex), rejected.runId, @@ -2367,7 +2408,7 @@ async function testSupervisedPermission(runtimeCase: LiveRuntimeCase): Promise {}, 1000); ] as const)("%s the 1024-message queue limit", async (_label, messageCount, rejected) => { const firstNotification = Promise.withResolvers(); const notificationGate = Promise.withResolvers(); + const notificationsHandled = Promise.withResolvers(); let handled = 0; const harness = await createClientHarness( (directory) => ` @@ -2034,16 +2035,25 @@ setInterval(() => {}, 1000); firstNotification.resolve(); await notificationGate.promise; } + if (handled === messageCount) { + notificationsHandled.resolve(); + } }, ); + const burstFlushed = Promise.withResolvers(); + const watcher = watch(harness.directory, (_event, filename) => { + if (filename === "burst-flushed") burstFlushed.resolve(); + }); + watcher.once("error", burstFlushed.reject); try { await harness.client.start(); await firstNotification.promise; + const burstFlushedPath = join(harness.directory, "burst-flushed"); expect(handled).toBe(1); expect(harness.protocolErrors).toEqual([]); if (!rejected) { - expect(await Bun.file(join(harness.directory, "burst-flushed")).exists()).toBe(false); + expect(existsSync(burstFlushedPath)).toBe(false); } notificationGate.resolve(); @@ -2052,7 +2062,15 @@ setInterval(() => {}, 1000); await expect(drain).rejects.toThrow("message queue limit exceeded"); } else { await drain; - expect(await Bun.file(join(harness.directory, "burst-flushed")).exists()).toBe(true); + // Draining admitted work does not certify the peer's entire burst. + // Observe both its write callback and the final delivered notification. + const completed = await settlePromiseWithTimeout( + Promise.all([burstFlushed.promise, notificationsHandled.promise]), + { label: "notification burst flush and delivery", timeoutMs: 1_000 }, + ); + expect(completed.status).toBe("completed"); + // Read fresh filesystem state after observing the write callback. + expect(existsSync(burstFlushedPath)).toBe(true); } expect(handled).toBe(Math.min(messageCount, 1_024)); if (rejected) { @@ -2062,6 +2080,7 @@ setInterval(() => {}, 1000); expect(harness.protocolErrors).toEqual([]); } } finally { + watcher.close(); notificationGate.resolve(); await harness.client.stop(); } diff --git a/tests/openai-app-server-event-bridge-terminal.test.ts b/tests/openai-app-server-event-bridge-terminal.test.ts index d46aae4..71d5b24 100644 --- a/tests/openai-app-server-event-bridge-terminal.test.ts +++ b/tests/openai-app-server-event-bridge-terminal.test.ts @@ -143,6 +143,87 @@ function createPublisherHarness( } describe("OpenAi app-server event bridge", () => { + test.each([ + "item completion", + "turn snapshot", + "filtered item completion", + "omitted turn items", + "replayed item completion", + "streamed item completion", + ])( + "accepts an empty final assistant message from %s without publishing empty text or reusing progress", + async (source) => { + const { bridge, context, delivered } = createPublisherHarness(); + const trackedTurn = bridge.trackTurn("turn-1", DRIVER_TEST_IDS.runId); + void trackedTurn.catch(() => {}); + // Given a valid progress message followed by an authoritative empty final. + const progress = { + id: "message-progress", + phase: "commentary", + text: "PROGRESS", + type: "agentMessage", + }; + const final = { + id: "message-final", + phase: "final", + text: source === "filtered item completion" ? "\uE200cite\uE202turn7search12\uE201" : "", + type: "agentMessage", + }; + await bridge.handleNotification(context, "item/completed", { + item: progress, + threadId: "thread-1", + turnId: "turn-1", + }); + if (source === "streamed item completion") { + await bridge.handleNotification(context, "item/agentMessage/delta", { + delta: "TRANSIENT", + itemId: final.id, + threadId: "thread-1", + turnId: "turn-1", + }); + } + + // When the provider delivers it through either native completion path. + if (source !== "turn snapshot") { + await bridge.handleNotification(context, "item/completed", { + item: final, + threadId: "thread-1", + turnId: "turn-1", + }); + } + if (source === "replayed item completion") { + await bridge.handleNotification(context, "item/completed", { + item: final, + threadId: "thread-1", + turnId: "turn-1", + }); + } + await bridge.handleNotification(context, "turn/completed", { + threadId: "thread-1", + turn: { + id: "turn-1", + items: source === "omitted turn items" ? [] : [progress, final], + itemsView: source === "omitted turn items" ? "notLoaded" : "full", + status: "completed", + }, + }); + + // Then strict event validation succeeds, lifecycles close, and the + // completed run does not misidentify the earlier progress as its final. + await expect(trackedTurn).resolves.toBeUndefined(); + expect( + delivered + .filter((event) => event.kind === "message.added") + .map((event) => readEventPayloadString(event, "content")), + ).toEqual(["PROGRESS"]); + expect(delivered.filter((event) => event.kind === "message.started")).toHaveLength(2); + expect(delivered.filter((event) => event.kind === "message.completed")).toHaveLength(2); + const terminals = delivered.filter((event) => event.kind === "run.completed"); + expect(terminals).toHaveLength(1); + expect(terminals[0]!.payload).not.toHaveProperty("finalMessageId"); + }, + ); + test("preserves the first reasoning delta when its start delivery fails", async () => { const { bridge, context, delivered } = createPublisherHarness({ failEventOnce: "thought.started", diff --git a/tests/openai-app-server-startup.feature b/tests/openai-app-server-startup.feature new file mode 100644 index 0000000..4d0691f --- /dev/null +++ b/tests/openai-app-server-startup.feature @@ -0,0 +1,23 @@ +Feature: OpenAI cancellation native-thread recovery + + Scenario: Start a fresh thread after cancelling before native history exists + Given the OpenAI app-server has started a thread with no successful persisted turn + When that first turn is cancelled after its start response is acknowledged + And no native turn has completed + And the next input is dispatched + Then the replacement client starts a fresh thread + And it does not resume the cancelled thread + + Scenario: Resume an established native thread after cancellation + Given the OpenAI app-server has a native checkpoint for the thread + When a turn on that thread is cancelled before its start response completes + And the next input is dispatched + Then the replacement client resumes the existing thread + + Scenario: Preserve same-process history across a completed turn and cancellation + Given a turn completes and writes a native checkpoint in the current process + When the next turn is cancelled + And the app-server's SQLite projection is stale + And another input is dispatched + Then the latest verified checkpoint is restored before the replacement client starts + And the replacement client resumes the established native thread diff --git a/tests/openai-app-server-startup.test.ts b/tests/openai-app-server-startup.test.ts index 9289196..a8cb643 100644 --- a/tests/openai-app-server-startup.test.ts +++ b/tests/openai-app-server-startup.test.ts @@ -51,9 +51,11 @@ interface CancellationHarnessOptions { readonly holdCancellationRequest?: boolean; readonly holdRunCancellation?: boolean; readonly holdMessageStart?: boolean; + readonly holdSecondTurnCompletion?: boolean; readonly holdTurnStartAfterItems?: boolean; readonly holdToolCompletion?: boolean; readonly nativeResume?: boolean; + readonly requireCheckpointRestoreOnResume?: boolean; readonly restartResumeError?: string; readonly threadId?: string; readonly turnId?: string; @@ -285,6 +287,9 @@ const persistCompletedTurn = (turnId) => { type: "event_msg", payload: { type: "task_complete", turn_id: turnId, error: null }, }) + "\\n"); + if (${JSON.stringify(cancellationOptions.requireCheckpointRestoreOnResume ?? false)}) { + writeFileSync(join(process.env.CODEX_SQLITE_HOME, "state_1.sqlite"), "stale projection"); + } }; process.stdin.setEncoding("utf8"); process.stdin.on("data", (chunk) => { @@ -320,11 +325,15 @@ process.stdin.on("data", (chunk) => { ((launchNumber > 1 || turnNumber > 1) && !${JSON.stringify(cancellationOptions.emitToolCompletionOnTurnStart ?? false)}); const holdTurnStart = + request.method === "turn/start" && (${JSON.stringify(holdFirstTurnStartResponse)} || ${JSON.stringify(cancellationOptions.holdTurnStartAfterItems ?? false)}) && launchNumber === 1 && - request.method === "turn/start" && turnNumber === 1; + const holdSecondTurnCompletion = + ${JSON.stringify(cancellationOptions.holdSecondTurnCompletion ?? false)} && + launchNumber === 1 && + turnNumber === 2; const response = request.method === "initialized" ? null @@ -336,9 +345,19 @@ process.stdin.on("data", (chunk) => { ${JSON.stringify(cancellationOptions.failInitialThreadStart ?? false)} ? { id: request.id, error: { code: -32600, message: "initial thread start failed" } } : request.method === "thread/resume" - ? resumeError === null + ? resumeError === null && + !(${JSON.stringify(cancellationOptions.requireCheckpointRestoreOnResume ?? false)} && + existsSync(join(process.env.CODEX_SQLITE_HOME, "state_1.sqlite"))) ? { id: request.id, result: threadStartResult } - : { id: request.id, error: { code: -32600, message: resumeError } } + : { + id: request.id, + error: { + code: -32600, + message: + resumeError ?? + "no rollout found for thread id " + ${JSON.stringify(cancellationOptions.threadId ?? "fresh-thread")}, + }, + } : request.method === "thread/start" ? { id: request.id, result: threadStartResult } : request.method === "turn/start" && turnStartErrorMessage !== null @@ -428,9 +447,11 @@ process.stdin.on("data", (chunk) => { () => sendToolLifecycle( turnId, - launchNumber > 1 || - turnNumber > 1 || - !${JSON.stringify(cancellationOptions.holdTurnStartAfterItems ?? false)}, + !holdSecondTurnCompletion && + (launchNumber > 1 || + turnNumber > 1 || + !${JSON.stringify(cancellationOptions.holdTurnStartAfterItems ?? false)}), + !holdSecondTurnCompletion, ), 5, ); @@ -1136,12 +1157,7 @@ describe("OpenAI app-server startup", () => { false, "full_access", true, - operation === "cancel" - ? { - restartResumeError: - "failed to read thread: thread-store internal error: failed to read thread /tmp/rollout.jsonl: rollout at /tmp/rollout.jsonl is empty", - } - : {}, + {}, ); const reason = `test.${operation}`; @@ -1205,10 +1221,7 @@ describe("OpenAI app-server startup", () => { ), { label: "turn after registration-window cancellation", timeoutMs: 1_000 }, ), - ).resolves.toMatchObject({ - status: "failed", - error: { message: expect.stringContaining("rollout at /tmp/rollout.jsonl is empty") }, - }); + ).resolves.toMatchObject({ status: "completed" }); expect( harness.events .filter( @@ -1217,12 +1230,13 @@ describe("OpenAI app-server startup", () => { ["run.cancelled", "run.completed", "run.failed"].includes(event.kind), ) .map((event) => event.kind), - ).toEqual(["run.failed"]); + ).toEqual(["run.completed"]); const requests = (await readFile(harness.requestLog, "utf8")) .trim() .split("\n") .map((line) => JSON.parse(line) as { method: string }); - expect(requests.filter((request) => request.method === "thread/start")).toHaveLength(1); + expect(requests.filter((request) => request.method === "thread/start")).toHaveLength(2); + expect(requests.some((request) => request.method === "thread/resume")).toBe(false); expect(requests.some((request) => request.method === "thread/inject_items")).toBe(false); } } finally { @@ -1579,7 +1593,8 @@ describe("OpenAI app-server startup", () => { requests .filter((request) => request.method === "thread/resume") .map((request) => request.params?.threadId), - ).toEqual(["fresh-thread"]); + ).toEqual([]); + expect(requests.filter((request) => request.method === "thread/start")).toHaveLength(2); expect(requests.filter((request) => request.method === "turn/start")).toHaveLength(2); } finally { await harness.releaseTurnStartResponse(); @@ -1587,6 +1602,211 @@ describe("OpenAI app-server startup", () => { } }); + test("dispatcher cancellation resumes a thread with persisted native history", async () => { + const harness = await createCancellationHarness({ + holdTurnStartAfterItems: true, + nativeResume: true, + }); + const kernel = new AgentDriverKernelCore({ + backendFactory: (payload) => new OpenAiAppServerDriverBackend(payload), + hostPorts: { + skill: { + materialize: async () => [], + }, + }, + logger: harness.logger, + }); + + try { + await kernel.start(harness.payload); + const firstInput = kernel.dispatch({ + commandId: "pending-turn-start-with-history", + input: { text: "first" }, + kind: "input.start", + requestId: "pending-turn-with-history-request", + runId: DRIVER_TEST_IDS.runId, + }); + void firstInput.catch(() => {}); + await expect( + settlePromiseWithTimeout( + (async () => { + while (!(await Bun.file(harness.turnStartHeldMarker).exists())) { + await Bun.sleep(1); + } + })(), + { label: "held turn start with native history", timeoutMs: 1_000 }, + ), + ).resolves.toMatchObject({ status: "completed" }); + + await expect( + settlePromiseWithTimeout(Promise.all([firstInput, kernel.cancel("test.cancel")]), { + label: "native-history turn/start cancellation", + timeoutMs: ACTIVE_TURN_CANCEL_GRACE_MS, + }), + ).resolves.toMatchObject({ status: "completed" }); + + await expect( + settlePromiseWithTimeout( + kernel.dispatch({ + commandId: "input-after-native-history-cancellation", + input: { text: "second" }, + kind: "input.start", + requestId: "request-after-native-history-cancellation", + runId: DRIVER_TEST_IDS.secondRunId, + }), + { label: "input after native-history cancellation", timeoutMs: 1_000 }, + ), + ).resolves.toMatchObject({ status: "completed" }); + + const requests = (await readFile(harness.requestLog, "utf8")) + .trim() + .split("\n") + .map( + (line) => + JSON.parse(line) as { + method: string; + params?: { threadId?: string }; + }, + ); + expect( + requests + .filter((request) => request.method === "thread/resume") + .map((request) => request.params?.threadId), + ).toEqual(["fresh-thread", "fresh-thread"]); + expect(requests.filter((request) => request.method === "thread/start")).toHaveLength(0); + } finally { + await harness.releaseTurnStartResponse(); + await kernel.stop("test complete").catch(() => {}); + } + }); + + test("cancellation after turn start ACK starts a fresh thread without native history", async () => { + const harness = await createCancellationHarness({}); + + try { + await harness.backend.start(harness.context, new AbortController().signal); + const firstInput = harness.backend.handleInput( + harness.context, + { text: "first" }, + DRIVER_TEST_IDS.runId, + ); + void firstInput.catch(() => {}); + await expect( + settlePromiseWithTimeout( + (async () => { + while (!harness.events.some((event) => event.kind === "run.started")) { + await Bun.sleep(1); + } + })(), + { label: "OpenAI turn after start ACK", timeoutMs: 250 }, + ), + ).resolves.toMatchObject({ status: "completed" }); + + await expect( + harness.backend.cancelActiveTurn(harness.context, "test.cancel"), + ).resolves.toBeUndefined(); + await expect(firstInput).rejects.toThrow("test.cancel"); + await expect( + harness.backend.handleInput( + harness.context, + { text: "second" }, + DRIVER_TEST_IDS.secondRunId, + ), + ).resolves.toBeUndefined(); + + const requests = (await readFile(harness.requestLog, "utf8")) + .trim() + .split("\n") + .map( + (line) => + JSON.parse(line) as { + method: string; + params?: { threadId?: string }; + }, + ); + expect( + requests + .filter((request) => request.method === "thread/resume") + .map((request) => request.params?.threadId), + ).toEqual([]); + expect(requests.filter((request) => request.method === "thread/start")).toHaveLength(2); + } finally { + await harness.backend.stop(harness.context, "test complete", new AbortController().signal); + } + }); + + test("cancellation after a completed turn resumes the established native thread", async () => { + const harness = await createCancellationHarness({ + emitToolCompletionOnTurnStart: true, + holdSecondTurnCompletion: true, + requireCheckpointRestoreOnResume: true, + }); + + try { + await harness.backend.start(harness.context, new AbortController().signal); + await harness.backend.handleInput(harness.context, { text: "first" }, DRIVER_TEST_IDS.runId); + expect( + harness.events.some( + (event) => event.runId === DRIVER_TEST_IDS.runId && event.kind === "run.completed", + ), + ).toBe(true); + + const cancelledInput = harness.backend.handleInput( + harness.context, + { text: "cancelled" }, + DRIVER_TEST_IDS.secondRunId, + ); + void cancelledInput.catch(() => {}); + await expect( + settlePromiseWithTimeout( + (async () => { + while ( + !harness.events.some( + (event) => + event.runId === DRIVER_TEST_IDS.secondRunId && event.kind === "run.started", + ) + ) { + await Bun.sleep(1); + } + })(), + { label: "cancelled turn after completed native history", timeoutMs: 1_000 }, + ), + ).resolves.toMatchObject({ status: "completed" }); + await expect( + harness.backend.cancelActiveTurn(harness.context, "test.cancel"), + ).resolves.toBeUndefined(); + await expect(cancelledInput).rejects.toThrow("test.cancel"); + + await expect( + harness.backend.handleInput(harness.context, { text: "third" }, DRIVER_TEST_IDS.thirdRunId), + ).resolves.toBeUndefined(); + + const requests = (await readFile(harness.requestLog, "utf8")) + .trim() + .split("\n") + .map( + (line) => + JSON.parse(line) as { + method: string; + params?: { threadId?: string }; + }, + ); + expect(requests.filter((request) => request.method === "thread/start")).toHaveLength(1); + expect( + requests + .filter((request) => request.method === "thread/resume") + .map((request) => request.params?.threadId), + ).toEqual(["fresh-thread"]); + expect( + harness.events.some( + (event) => event.runId === DRIVER_TEST_IDS.thirdRunId && event.kind === "run.completed", + ), + ).toBe(true); + } finally { + await harness.backend.stop(harness.context, "test complete", new AbortController().signal); + } + }); + test("dispatcher cancellation replaces an unselected completed terminal", async () => { const harness = await createCancellationHarness({ terminalNotificationBeforeTurnStartResponse: true, @@ -2280,6 +2500,7 @@ describe("OpenAI app-server startup", () => { test("retains a failed restart client until shutdown retries its cleanup", async () => { const harness = await createCancellationHarness({ + nativeResume: true, restartResumeError: "restart resume failed", }); const nativeKill = process.kill; diff --git a/tests/pi-driver-live-turn.test.ts b/tests/pi-driver-live-turn.test.ts new file mode 100644 index 0000000..a80456b --- /dev/null +++ b/tests/pi-driver-live-turn.test.ts @@ -0,0 +1,24 @@ +import { expect, test } from "bun:test"; + +import { parseRuntimeCommand } from "../src/runtime-command"; +import { createPiLiveTurn } from "./pi-driver-live-turn"; + +test("keeps event metadata outside the input.start command", () => { + const turn = createPiLiveTurn({ + commandId: "command-1", + eventIndex: 7, + requestId: "request-1", + runId: "run-1", + text: "hello", + }); + + expect(() => parseRuntimeCommand(turn.command)).not.toThrow(); + expect(Object.keys(turn.command).toSorted()).toEqual([ + "commandId", + "input", + "kind", + "requestId", + "runId", + ]); + expect(turn.eventIndex).toBe(7); +}); diff --git a/tests/pi-driver-live-turn.ts b/tests/pi-driver-live-turn.ts new file mode 100644 index 0000000..5530b5f --- /dev/null +++ b/tests/pi-driver-live-turn.ts @@ -0,0 +1,33 @@ +import type { InputStartCommand } from "../src/runtime-command"; + +export interface PiLiveTurnInput { + readonly commandId: string; + readonly eventIndex: number; + readonly requestId: string; + readonly runId: string; + readonly text: string; +} + +export interface PiLiveTurn { + readonly command: InputStartCommand; + readonly commandId: string; + readonly eventIndex: number; + readonly requestId: string; + readonly runId: string; +} + +export function createPiLiveTurn(input: PiLiveTurnInput): PiLiveTurn { + return { + command: { + commandId: input.commandId, + input: { text: input.text }, + kind: "input.start", + requestId: input.requestId, + runId: input.runId, + }, + commandId: input.commandId, + eventIndex: input.eventIndex, + requestId: input.requestId, + runId: input.runId, + }; +} diff --git a/tests/pi-driver-live-upstream.test.ts b/tests/pi-driver-live-upstream.test.ts new file mode 100644 index 0000000..ccf9214 --- /dev/null +++ b/tests/pi-driver-live-upstream.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, test } from "bun:test"; + +import { + DEFAULT_PI_LIVE_UPSTREAM_BASE_URL, + PI_LIVE_UPSTREAM_ENV, + piLiveChatCompletionsUrl, +} from "./pi-driver-live-upstream"; + +describe("Pi live upstream fixture", () => { + test("appends chat completions to the configured base URL", () => { + expect( + piLiveChatCompletionsUrl({ + [PI_LIVE_UPSTREAM_ENV]: "http://configured.example/v1/", + }), + ).toBe("http://configured.example/v1/chat/completions"); + }); + + test("uses the default OpenRouter base URL when unset", () => { + expect(piLiveChatCompletionsUrl({})).toBe( + `${DEFAULT_PI_LIVE_UPSTREAM_BASE_URL}/chat/completions`, + ); + }); +}); diff --git a/tests/pi-driver-live-upstream.ts b/tests/pi-driver-live-upstream.ts new file mode 100644 index 0000000..5b7572e --- /dev/null +++ b/tests/pi-driver-live-upstream.ts @@ -0,0 +1,17 @@ +export const PI_LIVE_UPSTREAM_ENV = "AGENT_DRIVER_LIVE_OPENROUTER_BASE_URL"; +export const DEFAULT_PI_LIVE_UPSTREAM_BASE_URL = "https://openrouter.ai/api/v1"; + +export function piLiveChatCompletionsUrl( + environment: Readonly> = process.env, +): string { + const configured = environment[PI_LIVE_UPSTREAM_ENV]?.trim(); + const baseUrl = + configured && configured.length > 0 + ? configured.replace(/\/$/, "") + : DEFAULT_PI_LIVE_UPSTREAM_BASE_URL; + const url = new URL(baseUrl); + if (url.protocol !== "http:" && url.protocol !== "https:") { + throw new Error(`${PI_LIVE_UPSTREAM_ENV} must use HTTP or HTTPS.`); + } + return `${baseUrl}/chat/completions`; +} diff --git a/tests/pi-driver-live.feature b/tests/pi-driver-live.feature new file mode 100644 index 0000000..6d1817b --- /dev/null +++ b/tests/pi-driver-live.feature @@ -0,0 +1,17 @@ +Feature: Pi live upstream endpoint fixture + + Scenario: Forward Pi live requests through a configured OpenAI-compatible endpoint + Given the Pi live fixture has an OpenAI-compatible base URL ending in /v1 + When the fixture builds the chat completions request URL + Then it appends /chat/completions to the configured base URL + + Scenario: Preserve the default Pi live upstream endpoint + Given the Pi live fixture has no configured base URL + When the fixture builds the chat completions request URL + Then it uses the default OpenRouter base URL followed by /chat/completions + + Scenario: Keep event metadata out of the input command + Given a Pi live turn tracks an event index for event assertions + When the fixture enqueues its input.start command + Then the command contains only the production input.start fields + And the event index remains fixture metadata diff --git a/tests/pi-driver-live.test.ts b/tests/pi-driver-live.test.ts index 80bf7cf..5f1bdab 100644 --- a/tests/pi-driver-live.test.ts +++ b/tests/pi-driver-live.test.ts @@ -16,8 +16,10 @@ import { import type { DriverArtifactTestEvent } from "./driver-artifact-test-controller"; import { driverBootPayload } from "./driver-boot-payload-fixture"; import { messageText } from "./driver-event-test-helpers"; +import { createPiLiveTurn } from "./pi-driver-live-turn"; +import { piLiveChatCompletionsUrl } from "./pi-driver-live-upstream"; -// Explicitly opt in: these tests make billable requests through OpenRouter. +// Explicitly opt in: these tests make billable requests through the configured upstream. // The local forwarder certifies the Driver/Pi wire, not hosted Worker/D1 behavior. const enabled = process.env["AGENT_DRIVER_PI_LIVE"] === "1"; const liveTest = enabled ? test : test.skip; @@ -49,6 +51,7 @@ async function fixture(policy: "full_access" | "supervised" = "full_access") { const mcpProof = `server-proof-${randomUUID()}`; const mcpMarker = `marker-${randomUUID()}`; const requests: { model: unknown; status: number; messages: unknown }[] = []; + const upstreamUrl = piLiveChatCompletionsUrl(); let mcpCalls = 0; let unauthorized = 0; const proxy = Bun.serve({ @@ -72,7 +75,7 @@ async function fixture(policy: "full_access" | "supervised" = "full_access") { const observed = { model: parsed.model, status: 0, messages: parsed.messages }; requests.push(observed); console.log(`Pi live: upstream request ${requests.length}`); - const upstream = await fetch("https://openrouter.ai/api/v1/chat/completions", { + const upstream = await fetch(upstreamUrl, { method: "POST", headers: { Authorization: `Bearer ${key}`, "Content-Type": "application/json" }, body, @@ -240,13 +243,14 @@ async function fixture(policy: "full_access" | "supervised" = "full_access") { } function begin(controller: DriverArtifactTestController, text: string) { - const turn = { + const turn = createPiLiveTurn({ commandId: createDriverId(), requestId: createDriverId(), runId: createDriverId(), eventIndex: controller.events.length, - }; - controller.enqueue({ ...turn, kind: "input.start", input: { text } }); + text, + }); + controller.enqueue(turn.command); return turn; }