fix(runner): keep streaming after task completion tools (#13108)
Keep receiving provider events after paperclip_finish, drain pending event persistence, and select the final assistant answer after the provider turn ends. Preserve cancellation, failure, and governed-wait behavior. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
5cb4f061dd
commit
ca96e1eb0a
|
|
@ -785,6 +785,22 @@ Examples:
|
|||
|
||||
Auto-recovery preserves the existing owner. It does not choose a replacement agent.
|
||||
|
||||
### Completion tools and final answers
|
||||
|
||||
A completion tool such as `paperclip_finish` reports task disposition; it does not
|
||||
end the provider turn. Paperclip continues persisting and displaying provider
|
||||
events until an authoritative turn terminal arrives. The completion report starts
|
||||
no interruption timer. Existing execution timeouts, cancellation, governed waits,
|
||||
and active-goal rules still apply. A later failed or cancelled terminal remains
|
||||
failed or cancelled even when the agent already reported completed work.
|
||||
|
||||
The final assistant message is the visible task response. Response selection runs
|
||||
after preceding event persistence completes; the completion summary cannot replace
|
||||
an available final answer. Existing fallback and explicit-comment precedence still
|
||||
apply. Stream closure without a turn terminal is not proof of success. Event
|
||||
replay uses the existing source receipts and never repeats provider work merely
|
||||
to recover recorded output.
|
||||
|
||||
### Provider continuity and bounded finalization
|
||||
|
||||
A permanently unusable established provider session may be replaced only with evidence that its predecessor is stopped and fenced, completed results and workspace state are preserved, required task history is available, and pending effects have been reconciled. A provider-native shell command or external write without a reliable outcome receipt is unknown. Unknown effects, integrity failures, and unverified process ownership never authorize speculative replay. Once automatic recovery is ruled out, Paperclip selects a conservative default: preserve recorded work, stop the affected task, and retain a durable no-replay hold. Unknown action outcomes remain unknown. No reconciliation form or user diagnosis is required.
|
||||
|
|
|
|||
|
|
@ -136,7 +136,7 @@ const CODEX_COLLABORATION_RUNTIME_INSTRUCTIONS = `## Codex-style collaboration
|
|||
- Before the first tool call in a turn, send a brief commentary update describing the immediate work you are starting.
|
||||
- During tool-driven work, send concise commentary updates at meaningful transitions so the user can follow progress without opening raw logs.
|
||||
- Reserve \`report_progress\` for meaningful durable milestones on longer work. Do not call it merely to create a completion comment on a short run; Paperclip materializes the final assistant response as the durable completion comment.
|
||||
- Invoke the semantic completion tool exactly once before the final assistant response. After it succeeds, send one self-contained final response with the outcome and verification, then do not call another tool.`;
|
||||
- Invoke the semantic completion tool exactly once before the final assistant response. After it succeeds, send one self-contained final response with the outcome and verification, then do not call another tool. The completion tool records task disposition; Paperclip keeps receiving your answer until the provider turn ends.`;
|
||||
|
||||
export function withCodexCollaborationRuntimeInstructions(
|
||||
instructions: string,
|
||||
|
|
|
|||
|
|
@ -529,8 +529,8 @@ describe("executeNativeSession recovery", () => {
|
|||
workingNow: true,
|
||||
});
|
||||
if (!lateGoal) yield runnerEvent(2, "run.result.proposed", result);
|
||||
// A newly observed goal must revoke the ordinary semantic-result
|
||||
// timeout, including when the proposal arrived first.
|
||||
// A newly observed goal owns the lifetime even when the completion
|
||||
// proposal arrived first.
|
||||
await new Promise((resolve) => setTimeout(resolve, 25));
|
||||
yield runnerEvent(3, "session.goal.updated", {
|
||||
goal: {
|
||||
|
|
@ -607,7 +607,6 @@ describe("executeNativeSession recovery", () => {
|
|||
controlPlane: port,
|
||||
runnerInstanceId: "runner-agent-goal",
|
||||
controlPlaneInstanceId: "control-agent-goal",
|
||||
semanticResultTerminalGraceMs: 5,
|
||||
});
|
||||
|
||||
expect(startTurn).toHaveBeenCalledOnce();
|
||||
|
|
@ -4154,7 +4153,7 @@ describe("executeNativeSession recovery", () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it("retains a reusable session after its remote semantic-result cancellation settles", async () => {
|
||||
it("uses the execution timeout when a completion report has no provider terminal", async () => {
|
||||
const lifecycle: string[] = [];
|
||||
let releaseProvider = () => {};
|
||||
const providerReleased = new Promise<void>((resolve) => {
|
||||
|
|
@ -4261,25 +4260,23 @@ describe("executeNativeSession recovery", () => {
|
|||
controlPlane: port,
|
||||
runnerInstanceId: "runner-recovery",
|
||||
controlPlaneInstanceId: "control-recovery",
|
||||
semanticResultTerminalGraceMs: 0,
|
||||
timeoutMs: 20,
|
||||
keepSessionOpen: true,
|
||||
onSession: (current) => retainedSessions.push(current),
|
||||
}),
|
||||
).resolves.toMatchObject({ result, terminal });
|
||||
).rejects.toThrow("native session timed out after 20ms");
|
||||
|
||||
expect(cancel).toHaveBeenCalledWith({
|
||||
reason: "Paperclip accepted the durable semantic result.",
|
||||
reason: "Native session event consumption failed.",
|
||||
signal: expect.any(AbortSignal),
|
||||
});
|
||||
expect(providerResult).not.toHaveBeenCalled();
|
||||
expect(events.map((event) => event.eventType)).toEqual([
|
||||
"run.result.proposed",
|
||||
"run.result.accepted",
|
||||
"run.terminal",
|
||||
]);
|
||||
expect(lifecycle).toEqual(["cancelled"]);
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(retainedSessions).toEqual([session]);
|
||||
expect(lifecycle).toContain("cancelled");
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
expect(retainedSessions.at(-1)).toBeNull();
|
||||
});
|
||||
|
||||
it("retains a reusable session while a semantic terminal releases its remote subscription", async () => {
|
||||
|
|
@ -4374,112 +4371,139 @@ describe("executeNativeSession recovery", () => {
|
|||
expect(retainedSessions).toEqual([session]);
|
||||
});
|
||||
|
||||
it("retains provider output emitted after a durable semantic result", async () => {
|
||||
const cancel = vi.fn(() => ({ cleanup: Promise.resolve() }));
|
||||
const close = vi.fn(async () => undefined);
|
||||
const events: PrpEvent[] = [];
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
async capabilities() {
|
||||
return {
|
||||
resume: true,
|
||||
typedEvents: true,
|
||||
steering: false,
|
||||
interruption: true,
|
||||
structuredResult: true,
|
||||
};
|
||||
},
|
||||
async *events() {
|
||||
yield runnerEvent(1, "run.result.proposed", result);
|
||||
yield runnerEvent(2, "item.completed", {
|
||||
item: { type: "assistant_message", text: "Final response." },
|
||||
});
|
||||
yield runnerEvent(3, "turn.completed");
|
||||
},
|
||||
async startTurn() {
|
||||
return { turnId: "turn-recovery" };
|
||||
},
|
||||
cancel,
|
||||
async result() {
|
||||
return null;
|
||||
},
|
||||
async snapshot() {
|
||||
return {
|
||||
backendKind: "mock",
|
||||
sessionId: identity.sessionId,
|
||||
identity,
|
||||
providerSessionId: "provider-recovery",
|
||||
cursor: "3",
|
||||
activeTurnId: null,
|
||||
pendingRuntimeRequests: [],
|
||||
lineage: [],
|
||||
};
|
||||
},
|
||||
close,
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
async descriptor() {
|
||||
return {
|
||||
kind: "mock",
|
||||
name: "semantic-result-final-response-backend",
|
||||
version: "1",
|
||||
capabilities: await session.capabilities(),
|
||||
};
|
||||
},
|
||||
async openSession() {
|
||||
return session;
|
||||
},
|
||||
};
|
||||
const port: ControlPlanePort = {
|
||||
async openRun() {},
|
||||
async checkpointSession() {},
|
||||
async appendEvent(event) {
|
||||
events.push(structuredClone(event as PrpEvent));
|
||||
const sourceEvents = events.filter(
|
||||
(candidate) => candidate.sourceInstanceId === event.sourceInstanceId,
|
||||
);
|
||||
return {
|
||||
cursor: events.length,
|
||||
highestContiguousSourceSeq: highestContiguous(sourceEvents),
|
||||
disposition: "committed",
|
||||
};
|
||||
},
|
||||
async replayEvents(replay) {
|
||||
const sourceEvents = events.filter(
|
||||
(event) => event.sourceInstanceId === replay.sourceInstanceId,
|
||||
);
|
||||
return {
|
||||
events: structuredClone(
|
||||
sourceEvents.filter(
|
||||
(event) => event.sourceSeq > replay.afterSourceSeq,
|
||||
it.each([
|
||||
["turn.completed", "succeeded"],
|
||||
["turn.failed", "failed"],
|
||||
["turn.cancelled", "cancelled"],
|
||||
["turn.interrupted", "cancelled"],
|
||||
["stream.closed", null],
|
||||
] as const)("persists a delayed final answer before settling %s", async (eventType, runTerminalState) => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
let releasePersistence!: () => void;
|
||||
const persistence = new Promise<void>((resolve) => { releasePersistence = resolve; });
|
||||
let persistingAnswer = false;
|
||||
const cancel = vi.fn(() => ({ cleanup: Promise.resolve() }));
|
||||
const close = vi.fn(async () => undefined);
|
||||
const events: PrpEvent[] = [];
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
async capabilities() {
|
||||
return {
|
||||
resume: true,
|
||||
typedEvents: true,
|
||||
steering: false,
|
||||
interruption: true,
|
||||
structuredResult: true,
|
||||
};
|
||||
},
|
||||
async *events() {
|
||||
yield runnerEvent(1, "run.result.proposed", result);
|
||||
await new Promise((resolve) => setTimeout(resolve, 6_000));
|
||||
yield runnerEvent(2, "item.completed", {
|
||||
kind: "agentMessage", channel: "final", text: "Final response.",
|
||||
});
|
||||
if (eventType !== "stream.closed") yield runnerEvent(3, eventType);
|
||||
},
|
||||
async startTurn() {
|
||||
return { turnId: "turn-recovery" };
|
||||
},
|
||||
cancel,
|
||||
async result() {
|
||||
return null;
|
||||
},
|
||||
async snapshot() {
|
||||
return {
|
||||
backendKind: "mock",
|
||||
sessionId: identity.sessionId,
|
||||
identity,
|
||||
providerSessionId: "provider-recovery",
|
||||
cursor: "3",
|
||||
activeTurnId: null,
|
||||
pendingRuntimeRequests: [],
|
||||
lineage: [],
|
||||
};
|
||||
},
|
||||
close,
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
async descriptor() {
|
||||
return {
|
||||
kind: "mock",
|
||||
name: "semantic-result-final-response-backend",
|
||||
version: "1",
|
||||
capabilities: await session.capabilities(),
|
||||
};
|
||||
},
|
||||
async openSession() {
|
||||
return session;
|
||||
},
|
||||
};
|
||||
const port: ControlPlanePort = {
|
||||
async openRun() {},
|
||||
async checkpointSession() {},
|
||||
async appendEvent(event) {
|
||||
if (event.eventType === "item.completed") {
|
||||
persistingAnswer = true;
|
||||
await persistence;
|
||||
}
|
||||
events.push(structuredClone(event as PrpEvent));
|
||||
const sourceEvents = events.filter(
|
||||
(candidate) => candidate.sourceInstanceId === event.sourceInstanceId,
|
||||
);
|
||||
return {
|
||||
cursor: events.length,
|
||||
highestContiguousSourceSeq: highestContiguous(sourceEvents),
|
||||
disposition: "committed",
|
||||
};
|
||||
},
|
||||
async replayEvents(replay) {
|
||||
const sourceEvents = events.filter(
|
||||
(event) => event.sourceInstanceId === replay.sourceInstanceId,
|
||||
);
|
||||
return {
|
||||
events: structuredClone(
|
||||
sourceEvents.filter(
|
||||
(event) => event.sourceSeq > replay.afterSourceSeq,
|
||||
),
|
||||
),
|
||||
),
|
||||
highestContiguousSourceSeq: highestContiguous(sourceEvents),
|
||||
};
|
||||
},
|
||||
async completeRun() {},
|
||||
};
|
||||
highestContiguousSourceSeq: highestContiguous(sourceEvents),
|
||||
};
|
||||
},
|
||||
async completeRun() {},
|
||||
};
|
||||
|
||||
await expect(
|
||||
executeNativeSession({
|
||||
input,
|
||||
backend,
|
||||
controlPlane: port,
|
||||
runnerInstanceId: "runner-recovery",
|
||||
controlPlaneInstanceId: "control-recovery",
|
||||
semanticResultTerminalGraceMs: 50,
|
||||
}),
|
||||
).resolves.toMatchObject({ result, terminal });
|
||||
let completed = false;
|
||||
const execution = executeNativeSession({
|
||||
input, backend, controlPlane: port,
|
||||
runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery",
|
||||
}).then((value) => { completed = true; return value; });
|
||||
await vi.waitFor(() => expect(events).toHaveLength(1));
|
||||
await vi.advanceTimersByTimeAsync(6_001);
|
||||
expect(persistingAnswer).toBe(true);
|
||||
expect(completed).toBe(false);
|
||||
expect(events.map((event) => event.eventType)).toEqual(["run.result.proposed"]);
|
||||
expect(cancel).not.toHaveBeenCalled();
|
||||
releasePersistence();
|
||||
if (eventType === "stream.closed") {
|
||||
await expect(execution).rejects.toThrow("native event stream closed before a turn terminal fact");
|
||||
expect(events.map((event) => event.eventType)).toEqual(["run.result.proposed", "item.completed"]);
|
||||
return;
|
||||
}
|
||||
await expect(execution).resolves.toMatchObject({ result, terminal: { runTerminalState } });
|
||||
|
||||
expect(cancel).not.toHaveBeenCalled();
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
expect(events.map((event) => event.eventType)).toEqual([
|
||||
"run.result.proposed",
|
||||
"item.completed",
|
||||
"turn.completed",
|
||||
"run.result.accepted",
|
||||
"run.terminal",
|
||||
]);
|
||||
expect(cancel).not.toHaveBeenCalled();
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
expect(events.map((event) => event.eventType)).toEqual([
|
||||
"run.result.proposed",
|
||||
"item.completed",
|
||||
eventType,
|
||||
"run.result.accepted",
|
||||
"run.terminal",
|
||||
]);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects a mismatched checkpoint before it mutates control-plane state", async () => {
|
||||
|
|
|
|||
|
|
@ -37,10 +37,9 @@ import {
|
|||
} from "./live/runnerd-codex-transport.js";
|
||||
|
||||
export const DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS = 120_000;
|
||||
export const DEFAULT_NATIVE_SEMANTIC_RESULT_TERMINAL_GRACE_MS = 5_000;
|
||||
const OPTIONAL_SESSION_CANCELLATION_GRACE_MS = 100;
|
||||
const FAILED_OPERATION_SETTLEMENT_GRACE_MS = 100;
|
||||
// Retaining a remote provider requires its semantic-result interruption to be
|
||||
// Retaining a remote provider requires terminal subscription teardown to be
|
||||
// acknowledged before the session is handed to another run. Daytona command
|
||||
// round trips routinely exceed the generic failed-operation grace, but remain
|
||||
// bounded by the transport. Other iterator and handoff cleanup keeps the short
|
||||
|
|
@ -146,8 +145,6 @@ export interface ExecuteNativeSessionOptions {
|
|||
checkpointTimeoutMs?: number;
|
||||
/** Internal test seam; production uses the fixed 120-second platform policy. */
|
||||
runtimeInputLiveWindowMs?: number;
|
||||
/** Internal test seam; production gives the provider five seconds to end after a result. */
|
||||
semanticResultTerminalGraceMs?: number;
|
||||
onSession?: (session: NativeSession | null) => void;
|
||||
/** Observes why a retained session was removed from warm reuse. */
|
||||
onSessionQuarantined?: (reason: string) => Promise<void> | void;
|
||||
|
|
@ -794,7 +791,6 @@ async function consumeTurn(
|
|||
input: NativeExecutionInput,
|
||||
timeoutMs: number,
|
||||
runtimeInputLiveWindowMs: number,
|
||||
semanticResultTerminalGraceMs: number,
|
||||
reusableSessionCancellationGraceMs: number,
|
||||
closeFailedSession: () => Promise<void>,
|
||||
quarantineSession: (reason: string) => void,
|
||||
|
|
@ -807,12 +803,9 @@ async function consumeTurn(
|
|||
initialGoal?: HarnessThreadGoal | null,
|
||||
) {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
let semanticResultTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
const semanticResultGraceExpired = Symbol("semantic_result_grace_expired");
|
||||
const appendAbort = new AbortController();
|
||||
const governedCleanupOperations = new Set<Promise<unknown>>();
|
||||
let governedCancellationCommitted = false;
|
||||
let semanticCancellationCommitted = false;
|
||||
let semanticResultObserved = false;
|
||||
let deferredGovernedCleanupSettlement: Promise<unknown> | null = null;
|
||||
let deferredSessionCancellationSettlement: Promise<unknown> | null = null;
|
||||
|
|
@ -859,11 +852,6 @@ async function consumeTurn(
|
|||
let goalControlObserved = false;
|
||||
let latestSessionGoal: HarnessThreadGoal | null = null;
|
||||
let resultSource: "semantic_result" | "governed_wait" | null = null;
|
||||
let semanticResultEvent: PrpEvent | null = null;
|
||||
let semanticResultDeadline: Promise<
|
||||
typeof semanticResultGraceExpired
|
||||
> | null = null;
|
||||
let pendingNext: ReturnType<typeof eventIterator.next> | null = null;
|
||||
let providerFailure: NativeProviderTerminalFailure | null = null;
|
||||
const settleDurableResult = (
|
||||
event: PrpEvent,
|
||||
|
|
@ -878,7 +866,6 @@ async function consumeTurn(
|
|||
signal: appendAbort.signal,
|
||||
});
|
||||
governedCancellationCommitted = true;
|
||||
semanticCancellationCommitted = event.eventType === "run.result.proposed";
|
||||
const cleanup = cancellation.cleanup;
|
||||
governedCleanupOperations.add(cleanup);
|
||||
void cleanup
|
||||
|
|
@ -892,37 +879,11 @@ async function consumeTurn(
|
|||
};
|
||||
};
|
||||
while (true) {
|
||||
pendingNext ??= eventIterator.next().catch(error => { throw providerFailure ?? error; });
|
||||
const next =
|
||||
semanticResultDeadline === null
|
||||
? await pendingNext
|
||||
: await Promise.race([pendingNext, semanticResultDeadline]);
|
||||
if (next === semanticResultGraceExpired) {
|
||||
void pendingNext.catch(() => undefined);
|
||||
if (semanticResultEvent === null || governedResult === null) {
|
||||
throw new Error("native_semantic_result_grace_lost_result");
|
||||
}
|
||||
return settleDurableResult(
|
||||
semanticResultEvent,
|
||||
governedResult,
|
||||
"Paperclip accepted the durable semantic result.",
|
||||
);
|
||||
}
|
||||
pendingNext = null;
|
||||
const next = await eventIterator.next().catch((error) => {
|
||||
throw providerFailure ?? error;
|
||||
});
|
||||
if (stopConsumer) throw new Error("native event consumer stopped");
|
||||
if (next.done) {
|
||||
if (
|
||||
resultSource === "semantic_result" &&
|
||||
semanticResultEvent !== null &&
|
||||
governedResult !== null
|
||||
) {
|
||||
return {
|
||||
event: semanticResultEvent,
|
||||
eventCount,
|
||||
highestContiguousSourceSeq,
|
||||
governedResult,
|
||||
};
|
||||
}
|
||||
if (providerFailure) throw providerFailure;
|
||||
throw new Error(
|
||||
"native event stream closed before a turn terminal fact",
|
||||
|
|
@ -969,10 +930,8 @@ async function consumeTurn(
|
|||
sessionGoalObserved = true;
|
||||
latestSessionGoal = eventGoal;
|
||||
// A harness-created goal may arrive after a semantic result proposal.
|
||||
// The goal owns the durable lifetime; revoke the old grace deadline.
|
||||
// The goal owns the durable lifetime instead of the completion proposal.
|
||||
if (resultSource === "semantic_result") {
|
||||
if (semanticResultTimer !== null) clearTimeout(semanticResultTimer);
|
||||
semanticResultDeadline = null;
|
||||
governedResult = null;
|
||||
resultSource = null;
|
||||
}
|
||||
|
|
@ -1068,17 +1027,7 @@ async function consumeTurn(
|
|||
}
|
||||
governedResult = validation.result;
|
||||
resultSource = "semantic_result";
|
||||
semanticResultEvent = event;
|
||||
semanticResultObserved = true;
|
||||
if (session.cancel !== undefined) {
|
||||
semanticResultDeadline = new Promise((resolve) => {
|
||||
semanticResultTimer = setTimeout(
|
||||
() => resolve(semanticResultGraceExpired),
|
||||
semanticResultTerminalGraceMs,
|
||||
);
|
||||
semanticResultTimer.unref?.();
|
||||
});
|
||||
}
|
||||
}
|
||||
if (governedResult === null && resolveGovernedWait) {
|
||||
if (appendAbort.signal.aborted) {
|
||||
|
|
@ -1095,9 +1044,8 @@ async function consumeTurn(
|
|||
}
|
||||
if (governedResult !== null && !isTurnTerminal(event) && !sessionGoalObserved) {
|
||||
if (resultSource === "semantic_result") {
|
||||
// Give the provider a short grace to publish its final assistant
|
||||
// message and terminal after the semantic tool returns. If no
|
||||
// terminal arrives, the deadline above finalizes the durable result.
|
||||
// A completion tool reports task disposition, not provider termination.
|
||||
// Persist the final answer and authoritative terminal before finalizing.
|
||||
continue;
|
||||
}
|
||||
return settleDurableResult(
|
||||
|
|
@ -1161,6 +1109,7 @@ async function consumeTurn(
|
|||
}
|
||||
}
|
||||
if (isTurnTerminal(event)) {
|
||||
if (providerFailure) throw providerFailure;
|
||||
return {
|
||||
event,
|
||||
eventCount,
|
||||
|
|
@ -1281,14 +1230,13 @@ async function consumeTurn(
|
|||
// Iterator and provider cleanup own no control-plane mutation authority.
|
||||
// A reusable session with a semantic result needs a longer bounded
|
||||
// window for the remote event subscription to release. This applies
|
||||
// both when Paperclip forced an interrupt and when the provider emitted
|
||||
// its own terminal immediately afterward: the latter still crosses the
|
||||
// remote PRP acknowledgement boundary and routinely takes longer than
|
||||
// after the provider terminal, which still crosses the remote PRP
|
||||
// acknowledgement boundary and routinely takes longer than
|
||||
// the generic local cleanup grace. Governed waits and unrelated stalled
|
||||
// cleanup retain the short fail-closed boundary.
|
||||
const teardownSettled = await settlesWithin(
|
||||
passiveTeardownSettlement,
|
||||
semanticCancellationCommitted || semanticResultObserved
|
||||
semanticResultObserved
|
||||
? reusableSessionCancellationGraceMs
|
||||
: FAILED_OPERATION_SETTLEMENT_GRACE_MS,
|
||||
);
|
||||
|
|
@ -1296,7 +1244,6 @@ async function consumeTurn(
|
|||
quarantineSession("provider_event_teardown_timed_out");
|
||||
}
|
||||
if (timer !== undefined) clearTimeout(timer);
|
||||
if (semanticResultTimer !== undefined) clearTimeout(semanticResultTimer);
|
||||
removeExternalAbort();
|
||||
}
|
||||
}
|
||||
|
|
@ -2257,8 +2204,6 @@ export async function executeNativeSession(
|
|||
options.timeoutMs ?? 900_000,
|
||||
options.runtimeInputLiveWindowMs ??
|
||||
DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS,
|
||||
options.semanticResultTerminalGraceMs ??
|
||||
DEFAULT_NATIVE_SEMANTIC_RESULT_TERMINAL_GRACE_MS,
|
||||
options.keepSessionOpen
|
||||
? REUSABLE_SESSION_CANCELLATION_SETTLEMENT_GRACE_MS
|
||||
: FAILED_OPERATION_SETTLEMENT_GRACE_MS,
|
||||
|
|
@ -2381,13 +2326,17 @@ export async function executeNativeSession(
|
|||
? await session.result()
|
||||
: {
|
||||
result: consumed.governedResult,
|
||||
terminal: {
|
||||
schema: "paperclip.prp.terminal.v1",
|
||||
turnTerminalState: "completed",
|
||||
runTerminalState: "succeeded",
|
||||
reportedWorkDisposition:
|
||||
consumed.governedResult.reportedWorkDisposition,
|
||||
},
|
||||
terminal: isTurnTerminal(terminalEvent)
|
||||
? terminalFromEvent(
|
||||
terminalEvent,
|
||||
consumed.governedResult.reportedWorkDisposition,
|
||||
)
|
||||
: {
|
||||
schema: "paperclip.prp.terminal.v1",
|
||||
turnTerminalState: "completed",
|
||||
runTerminalState: "succeeded",
|
||||
reportedWorkDisposition: consumed.governedResult.reportedWorkDisposition,
|
||||
},
|
||||
turnId: terminalEvent.turnId ?? null,
|
||||
};
|
||||
signal.throwIfAborted();
|
||||
|
|
|
|||
|
|
@ -3313,13 +3313,11 @@ describe("native warm session supervision", () => {
|
|||
.mockReset()
|
||||
.mockImplementationOnce(async (options) => {
|
||||
expect(options.existingSession).toBeUndefined();
|
||||
expect(options.semanticResultTerminalGraceMs).toBe(30_000);
|
||||
options.onSession?.(sharedSession);
|
||||
return result;
|
||||
})
|
||||
.mockImplementationOnce(async (options) => {
|
||||
expect(options.existingSession).toBe(sharedSession);
|
||||
expect(options.semanticResultTerminalGraceMs).toBe(30_000);
|
||||
return result;
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -195,12 +195,6 @@ const TERMINAL_HEARTBEAT_RUN_STATUSES = new Set([
|
|||
const NATIVE_SESSION_EXECUTION_LEASE_TTL_MS = 20 * 60_000;
|
||||
const NATIVE_SESSION_EXECUTION_LEASE_RENEW_INTERVAL_MS = 5 * 60_000;
|
||||
const NATIVE_SESSION_CANCELLATION_CLEANUP_GRACE_MS = 2_000;
|
||||
// A reusable provider must publish its terminal suffix before the next run can
|
||||
// rotate PRP authority. Remote Codex can take more than the ordinary five-second
|
||||
// result grace to flush its final answer over Daytona, so retain the bounded
|
||||
// turn long enough to reach a naturally quiescent, reusable state. This adds no
|
||||
// delay when the provider terminates normally.
|
||||
const NATIVE_WARM_SEMANTIC_RESULT_TERMINAL_GRACE_MS = 30_000;
|
||||
const NATIVE_RUNTIME_REQUEST_RESOLUTION_CACHE_MAX = 256;
|
||||
type NativeRuntimeRequestResolution = {
|
||||
runId: string;
|
||||
|
|
@ -4898,10 +4892,6 @@ async function executePaperclipNativeSessionWithinScope(
|
|||
keepSessionOpen: warmSessionId !== null,
|
||||
sessionGoalControl: input.sessionGoalControl,
|
||||
resumeSessionGoalHeartbeat: input.resumeSessionGoalHeartbeat,
|
||||
semanticResultTerminalGraceMs:
|
||||
warmSessionId === null
|
||||
? undefined
|
||||
: NATIVE_WARM_SEMANTIC_RESULT_TERMINAL_GRACE_MS,
|
||||
// Every durable runner must finish its bounded suspension before
|
||||
// the next run verifies and rotates the saved authority.
|
||||
requireSessionCloseBeforeReturn: runnerdBackend !== null,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { asc, eq } from "drizzle-orm";
|
||||
import { readCompletedAssistantMessageCandidate, resolveHeartbeatRunResponse, selectHeartbeatRunFinalAgentMessage } from "../heartbeat-run-summary.js";
|
||||
import {
|
||||
activityLog,
|
||||
agentWakeupRequests,
|
||||
|
|
@ -495,7 +496,7 @@ describe("PaperclipControlPlanePort conformance", () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it("completes one selected Paperclip task through the public package session contract", async () => {
|
||||
it("persists a delayed final answer and replay before resolving the selected task response", async () => {
|
||||
const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity;
|
||||
const sessionId = taskSessionId;
|
||||
const evidenceRef = `work_product:${taskWorkProductId}`;
|
||||
|
|
@ -519,9 +520,13 @@ describe("PaperclipControlPlanePort conformance", () => {
|
|||
emittedAt: `2026-08-09T02:59:0${sourceSeq}.000Z`,
|
||||
payload,
|
||||
});
|
||||
const finalText = "The launch has two stages. Source: https://example.invalid/launch/STREAM-42";
|
||||
const finalAnswer = event(2, "item.completed", { kind: "agentMessage", channel: "final", text: finalText });
|
||||
const events = [
|
||||
event(1, "run.result.proposed", taskResult),
|
||||
event(2, "turn.completed", {}),
|
||||
finalAnswer,
|
||||
finalAnswer, // Exact replay must not create another stored answer.
|
||||
event(3, "turn.completed", {}),
|
||||
];
|
||||
const backend: NativeSessionBackend = {
|
||||
async descriptor() {
|
||||
|
|
@ -537,9 +542,13 @@ describe("PaperclipControlPlanePort conformance", () => {
|
|||
return {
|
||||
identity: () => input.identity,
|
||||
async capabilities() { return { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true }; },
|
||||
async *events() { yield* events; },
|
||||
async *events() {
|
||||
yield events[0]!;
|
||||
await new Promise((resolve) => setTimeout(resolve, 6_000));
|
||||
yield* events.slice(1);
|
||||
},
|
||||
async startTurn() { return { turnId: "turn-phase6-paperclip-task" }; },
|
||||
cancel() { return { cleanup: Promise.resolve() }; },
|
||||
cancel() { throw new Error("A completion report must not interrupt the provider"); },
|
||||
async result() { return { result: taskResult, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId: "turn-phase6-paperclip-task" }; },
|
||||
async snapshot() { return { backendKind: "mock", sessionId, identity: input.identity, providerSessionId: "provider-phase6-paperclip-task" }; },
|
||||
async close() {},
|
||||
|
|
@ -606,6 +615,23 @@ describe("PaperclipControlPlanePort conformance", () => {
|
|||
controlPlaneInstanceId: "phase6-control-plane",
|
||||
});
|
||||
expect(completed.terminal.runTerminalState).toBe("succeeded");
|
||||
const stored = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, taskRunId)).orderBy(asc(heartbeatRunEvents.seq));
|
||||
const answers = stored.filter((row) => row.eventType === "item.completed");
|
||||
expect(answers).toHaveLength(1);
|
||||
const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({
|
||||
candidates: answers.flatMap((row) => {
|
||||
const candidate = readCompletedAssistantMessageCandidate({ seq: row.seq, prpEvent: row.payload?.prpEvent });
|
||||
return candidate ? [candidate] : [];
|
||||
}),
|
||||
});
|
||||
const response = resolveHeartbeatRunResponse({
|
||||
resultJson: { nativeResult: taskResult }, finalAgentMessage,
|
||||
});
|
||||
expect(resolveHeartbeatRunResponse({ resultJson: { nativeResult: taskResult } }).text).toBe(taskResult.summary);
|
||||
expect(response.text).toBe(finalText);
|
||||
expect(response.decision.chosenSource).toBe("final_agent_message");
|
||||
expect(stored.findIndex((row) => row.eventType === "run.result.accepted"))
|
||||
.toBeGreaterThan(stored.findIndex((row) => row.eventType === "item.completed"));
|
||||
await finalizeNativeRun({ db, runId: taskRunId, workspaceFinalizeStatus: "succeeded" });
|
||||
await expect(port.completeRun({
|
||||
result: taskResult,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,2 @@
|
|||
#!/bin/sh
|
||||
exec node "$(dirname "$0")/../in-feed-codex.mjs" --completion-stream-fixture "$@"
|
||||
|
|
@ -55,6 +55,23 @@ async function finish(text, evidenceRef) {
|
|||
evidence: [{ ref: evidenceRef }], verification: [{ commandOrCheck: 'Fixture outcome', status: 'passed' }], attentionRequests: [], artifacts: [] });
|
||||
}
|
||||
async function execute() {
|
||||
if (process.argv.includes('--completion-stream-fixture')) {
|
||||
// Report completion before a deliberately slow final answer. The real
|
||||
// runner and control plane must keep listening after the tool succeeds.
|
||||
await finish('The fixture work is complete.', 'fixture:STREAM-42');
|
||||
// Exceed both former completion deadlines: cold (5s) and reusable (30s).
|
||||
await new Promise((resolve) => setTimeout(resolve, 31_000));
|
||||
const itemId = `answer-${turnId}`;
|
||||
const text = 'STREAM-42: The launch has two stages. Source: https://example.invalid/launch/STREAM-42';
|
||||
send({ method: 'item/started', params: { threadId, turnId, item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text: '' } } });
|
||||
send({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta: text.slice(0, 40) } });
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
send({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta: text.slice(40) } });
|
||||
send({ method: 'item/completed', params: { threadId, turnId, item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text } } });
|
||||
send({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } });
|
||||
return;
|
||||
}
|
||||
|
||||
if (recoveryFixture) {
|
||||
send({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId: `progress-${turnId}`, delta: 'Checking the current request and available connections.' } });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,103 @@
|
|||
import { expect, test } from '@playwright/test';
|
||||
import { execFileSync, spawn, type ChildProcess } from 'node:child_process';
|
||||
import { writeFile } from 'node:fs/promises';
|
||||
import { resolve } from 'node:path';
|
||||
|
||||
async function stopTestDrive(child: ChildProcess | undefined): Promise<void> {
|
||||
if (!child?.pid) return;
|
||||
const signal = (name: NodeJS.Signals) => {
|
||||
try { process.kill(-child.pid!, name); }
|
||||
catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ESRCH') throw error;
|
||||
}
|
||||
};
|
||||
const waitForExit = (timeoutMs: number) => {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve(true);
|
||||
return new Promise<boolean>((done) => {
|
||||
const exited = () => { clearTimeout(timer); done(true); };
|
||||
const timer = setTimeout(() => { child.removeListener('exit', exited); done(false); }, timeoutMs);
|
||||
child.once('exit', exited);
|
||||
});
|
||||
};
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
const stopped = waitForExit(15_000);
|
||||
signal('SIGTERM');
|
||||
if (await stopped) return;
|
||||
const killed = waitForExit(5_000);
|
||||
signal('SIGKILL');
|
||||
if (!(await killed)) throw new Error('test-drive process did not exit after SIGKILL');
|
||||
}
|
||||
|
||||
test('a completion tool does not cut off a delayed final answer', async ({ page }, info) => {
|
||||
await page.setViewportSize({ width: 1440, height: 1000 });
|
||||
const root = resolve(import.meta.dirname, '../../..');
|
||||
let child: ChildProcess | undefined;
|
||||
let logs = '';
|
||||
try {
|
||||
const env = { ...process.env, STREAM_FIXTURE_KEY: 'not-a-real-model-key', NODE_ENV: 'test',
|
||||
PATH: `${root}/tests/e2e/fixtures/completion-stream-bin:${process.env.PATH}` };
|
||||
delete env.DATABASE_URL;
|
||||
delete env.DATABASE_MIGRATION_URL;
|
||||
child = spawn(process.execPath, ['cli/node_modules/tsx/dist/cli.mjs', 'cli/src/index.ts', 'test-drive',
|
||||
'--harness', 'codex', '--api-key-env', 'STREAM_FIXTURE_KEY', '--company-name', 'Stream completion', '--no-browser'],
|
||||
{ cwd: root, env, detached: true, stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
child.stdout!.on('data', (chunk) => { logs += chunk.toString(); });
|
||||
child.stderr!.on('data', (chunk) => { logs += chunk.toString(); });
|
||||
await expect.poll(() => logs.match(/Paperclip is ready at (http:\/\/127\.0\.0\.1:\d+)/)?.[1], { timeout: 100_000 }).toBeTruthy();
|
||||
const base = logs.match(/Paperclip is ready at (http:\/\/127\.0\.0\.1:\d+)/)![1]!;
|
||||
const api = async (path: string) => {
|
||||
const response = await page.request.get(`${base}/api${path}`);
|
||||
expect(response.ok(), await response.text()).toBeTruthy();
|
||||
return response.json();
|
||||
};
|
||||
const health = await api('/health');
|
||||
expect(health).toMatchObject({ status: 'ok', deploymentMode: 'local_trusted', bootstrapStatus: 'ready',
|
||||
serverInfo: { git: { branchName: execFileSync('git', ['branch', '--show-current'], { cwd: root, encoding: 'utf8' }).trim() } } });
|
||||
const [company] = await api('/companies');
|
||||
const [agent] = await api(`/companies/${company.id}/agents`);
|
||||
expect(await api(`/companies/${company.id}/issues`)).toEqual([]);
|
||||
expect(await api(`/companies/${company.id}/heartbeat-runs`)).toEqual([]);
|
||||
const prefix = `/${company.issuePrefix}`;
|
||||
await page.goto(base + prefix + '/dashboard');
|
||||
await page.goto(base + prefix + '/company/settings/instance/experimental');
|
||||
if (!(await api('/instance/settings/experimental')).enableNativeRunner) {
|
||||
await page.getByRole('switch', { name: 'Toggle Paperclip Runner experimental setting' }).click();
|
||||
}
|
||||
await expect.poll(async () => (await api('/instance/settings/experimental')).enableNativeRunner).toBe(true);
|
||||
await page.goto(base + prefix + `/agents/${agent.id}/configuration`);
|
||||
await page.getByRole('button', { name: 'Codex', exact: true }).click();
|
||||
await page.getByRole('button', { name: /Paperclip Runner/ }).click();
|
||||
await page.getByRole('button', { name: 'Save changes', exact: true }).click();
|
||||
await expect.poll(async () => (await api(`/agents/${agent.id}`)).adapterType).toBe('paperclip_runner');
|
||||
await page.getByRole('link', { name: 'Tasks', exact: true }).click();
|
||||
await page.getByRole('button', { name: 'New Task', exact: true }).last().click();
|
||||
await page.getByPlaceholder('Task title').fill('Summarize the launch decisions and include the source link');
|
||||
await page.getByRole('button', { name: 'Assignee', exact: true }).click();
|
||||
await page.getByRole('button', { name: 'CEO', exact: true }).click();
|
||||
await page.getByRole('button', { name: 'Create Task', exact: true }).click();
|
||||
await expect.poll(async () => (await api(`/companies/${company.id}/issues`)).length).toBe(1);
|
||||
const [issue] = await api(`/companies/${company.id}/issues`);
|
||||
await page.goto(`${base}${prefix}/issues/${issue.identifier}`);
|
||||
await expect.poll(async () => (await api(`/companies/${company.id}/heartbeat-runs`))[0]?.status, { timeout: 60_000 }).toBe('succeeded');
|
||||
await expect(page.getByText('STREAM-42: The launch has two stages.', { exact: false }).first()).toBeVisible();
|
||||
await expect(page.getByRole('link', { name: 'https://example.invalid/launch/STREAM-42', exact: true }).first()).toBeVisible();
|
||||
await expect(page.getByRole('textbox', { name: 'editable markdown', exact: true })).toBeEditable();
|
||||
await page.reload();
|
||||
await expect(page.getByText('STREAM-42: The launch has two stages.', { exact: false }).first()).toBeVisible();
|
||||
const runs = await api(`/companies/${company.id}/heartbeat-runs`);
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]).toMatchObject({ status: 'succeeded' });
|
||||
const run = await api(`/heartbeat-runs/${runs[0].id}`);
|
||||
expect(run.runtimeMode).toBe('native');
|
||||
expect(run.resultJson.presentationDecision.chosenSource).toBe('final_agent_message');
|
||||
const comments = await api(`/issues/${issue.id}/comments`);
|
||||
expect(comments.filter((comment: { body: string }) => comment.body.includes('STREAM-42'))).toHaveLength(1);
|
||||
await page.screenshot({ path: info.outputPath('delayed-final-answer.png'), fullPage: true });
|
||||
await writeFile(info.outputPath('evidence.json'), JSON.stringify({ base, health, companyId: company.id, issueId: issue.id,
|
||||
runId: run.id, dependency: 'deterministic Codex app-server fixture', delayMs: 31000,
|
||||
dataDir: logs.match(/Data directory: ([^\n\r]+)/)?.[1], presentation: run.resultJson.presentationDecision }, null, 2));
|
||||
} finally {
|
||||
try { await stopTestDrive(child); }
|
||||
finally { await writeFile(info.outputPath('test-drive.log'), logs); }
|
||||
}
|
||||
});
|
||||
Loading…
Reference in New Issue