mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
fix: continue native chats after verified worker-crash cleanup
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
8326e33ada
commit
b41549f9fc
7 files changed
+209
-16
No files matched your search
@@ -1,3 +1,4 @@
|
||||
import * as nativeExecutor from "./native-runtime/native-session-executor.js";
|
||||
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
@@ -374,6 +375,32 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
return result;
|
||||
});
|
||||
|
||||
it.each(["verified", "unproven", "changed", "dry_run", "retry", "duplicate"])(
|
||||
"requires exact local cleanup for a new turn after worker loss (%s)", async mode => {
|
||||
const f = await seed();
|
||||
await db.update(heartbeatRuns).set({ errorCode: "native_session_cleanup_quarantined" }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
const retire = vi.fn(() => mode !== "changed");
|
||||
const verify = vi.spyOn(nativeExecutor, "verifyStoppedNativeSessionForContinuation").mockResolvedValue(
|
||||
mode === "unproven" ? null : { evidence: { runId: f.sourceRunId, schema: "paperclip.stopped_native_conversation.v1" }, retire });
|
||||
try {
|
||||
const result = mode === "retry"
|
||||
? await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db,
|
||||
reason: "retry_failed_run", failedRunId: f.sourceRunId }))
|
||||
: await admit(f, mode === "dry_run");
|
||||
expect(Boolean(result)).toBe(["verified", "dry_run", "duplicate"].includes(mode));
|
||||
expect(retire).toHaveBeenCalledTimes(["verified", "changed", "duplicate"].includes(mode) ? 1 : 0);
|
||||
const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(Boolean(actions[0].evidence.explicitUserContinuation)).toBe(["verified", "duplicate"].includes(mode));
|
||||
const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceRunId),
|
||||
eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified")));
|
||||
expect(events).toHaveLength(["verified", "duplicate"].includes(mode) ? 1 : 0);
|
||||
if (mode === "duplicate") {
|
||||
expect(await admit(f)).toBeNull();
|
||||
expect(retire).toHaveBeenCalledTimes(1);
|
||||
}
|
||||
} finally { verify.mockRestore(); }
|
||||
});
|
||||
|
||||
it.each(["suspended", "ready", "wrong_run", "wrong_thread", "active_provider", "pending_tool", "pending_output", "missing_state", "new_launch"])(
|
||||
"recovers a historical run without process metadata only from exact suspended state (%s)", async kind => {
|
||||
const f = await seed();
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import { readQueuedInteractionResponse } from "./queued-interaction-response.js";
|
||||
import { isCancelledNativeStartup } from "./cancelled-native-startup.js";
|
||||
import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession } from "./native-local-process-stop.js";
|
||||
@@ -146,6 +147,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start.");
|
||||
|
||||
const sources: Run[] = [];
|
||||
const stoppedSessions: Array<{ evidence: Record<string, unknown>; retire: () => boolean }> = [];
|
||||
const cancelledStartupIds = new Set<string>();
|
||||
for (const action of actions) {
|
||||
const runId = action.evidence.runId ?? action.evidence.sourceRunId;
|
||||
@@ -235,6 +237,14 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
if (run.processGroupId && !processStopped(-run.processGroupId)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically.");
|
||||
}
|
||||
}
|
||||
if (!remote && run.runtimeMode === "native" && run.errorCode === "native_session_cleanup_quarantined") {
|
||||
// This is new user input, never permission to retry the interrupted turn.
|
||||
if (retry) return blocked("cleanup_quarantined", "Send a new message after the previous provider has stopped.");
|
||||
const { verifyStoppedNativeSessionForContinuation } = await import("./native-runtime/native-session-executor.js");
|
||||
const stopped = await verifyStoppedNativeSessionForContinuation(db, run);
|
||||
if (!stopped) return blocked("local_cleanup", "Waiting for the previous provider and its tools to stop. Your message is saved.");
|
||||
stoppedSessions.push(stopped);
|
||||
}
|
||||
sources.push(run);
|
||||
}
|
||||
const nativeSources = sources.filter(run => run.runtimeMode === "native");
|
||||
@@ -253,6 +263,15 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
context: { previousRunId: previous.id, wakeCommentId: commentId },
|
||||
summary: null, exposeLowTrustRaw: false });
|
||||
if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
|
||||
for (const stopped of stoppedSessions) {
|
||||
if (!stopped.retire()) return blocked("local_cleanup", "The previous provider cleanup changed. Your message is saved.");
|
||||
await appendHeartbeatRunEvent(db, {
|
||||
companyId, runId: String(stopped.evidence.runId), agentId,
|
||||
eventType: "native.stopped_conversation_verified", stream: "system", level: "info",
|
||||
message: "The old runner and provider stopped. New user input will use a fresh session; prior action outcomes remain recorded.",
|
||||
payload: stopped.evidence,
|
||||
});
|
||||
}
|
||||
const authorization = { actorId, commentId, ...(response ? { interactionId: response.source.interactionId } : {}), ...(retry ? { failedRunId: input.failedRunId } : {}),
|
||||
...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}),
|
||||
...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId,
|
||||
|
||||
@@ -29,6 +29,7 @@ import {
|
||||
createPrpSemanticToolResultEnvelope,
|
||||
validatePrpStructuredRunResult,
|
||||
validatePrpEvent,
|
||||
parseNativeExecutionInput,
|
||||
type NativeExecutionInputV1,
|
||||
type PrpEvent,
|
||||
} from "@paperclipai/paperclip-runner";
|
||||
@@ -253,6 +254,7 @@ import {
|
||||
assertRemoteRunnerBuildMetadata,
|
||||
nativeSessionFailureDisposition,
|
||||
nativeFailedRunRetryStateIsSafe,
|
||||
verifyStoppedNativeSessionForContinuation,
|
||||
nativePreProviderRetryAfterCleanupStateIsSafe,
|
||||
reconcileRetainedNativeSessionCleanup,
|
||||
retainedNativeCleanupJournalMatches,
|
||||
@@ -3853,6 +3855,75 @@ describe("retained native cleanup activation", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("stopped native conversation physical cleanup", () => {
|
||||
it.each(["codex", "acpx"].flatMap(provider => [
|
||||
"stopped", "provider_alive", "worker_alive", "missing_receipt", "wrong_receipt", "new_launch",
|
||||
"foreign_run", "foreign_company", "foreign_runner", "remote", "unreleased", "changed_state", "changed_pid", "symlink",
|
||||
].map(mode => ({ provider, mode }))))("$provider $mode", async ({ provider, mode }) => {
|
||||
const base = await mkdtemp(join(tmpdir(), "native-conversation-cleanup-"));
|
||||
const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = base;
|
||||
const input = parseNativeExecutionInput({ ...execution,
|
||||
provider: provider === "codex" ? execution.provider : { kind: "acpx", agent: "claude", model: "claude-sonnet-5", permissionPolicy: "interactive",
|
||||
profile: { driverKind: "acpx_runtime", protocolVersion: 1, acpxVersion: "0.13.1", agent: "claude", agentProfileVersion: 1,
|
||||
agentServerPackage: "@zed-industries/claude-agent-acp", agentServerVersion: "1", agentRuntimePackage: null,
|
||||
agentRuntimeVersion: null, commandDigest: "fixture" } },
|
||||
session: { ...execution.session, driverKind: provider === "codex" ? "codex_app_server" : "acpx_runtime" },
|
||||
});
|
||||
const canonical = (value: any): string => value && typeof value === "object" && !Array.isArray(value)
|
||||
? `{${Object.entries(value).filter(([, v]) => v !== undefined).sort(([a], [b]) => a.localeCompare(b)).map(([k,v]) => `${JSON.stringify(k)}:${canonical(v)}`).join(",")}}`
|
||||
: JSON.stringify(value);
|
||||
const hash = (value: unknown) => createHash("sha256").update(canonical(value)).digest("hex");
|
||||
const providerScope = input.provider.kind === "acpx" ? { kind: "acpx", agent: input.provider.agent, profile: input.provider.profile } : { kind: "codex" };
|
||||
const root = join(base, hash({ schema: "paperclip.native-session-scope.v2", companyId: input.binding.companyId,
|
||||
agentId: input.binding.agentId, workspace: { kind: "managed", executionWorkspaceId: input.binding.executionWorkspaceId },
|
||||
provider: { driverKind: input.session.driverKind, identity: providerScope }, normalizedSessionId: input.session.normalizedSessionId }));
|
||||
const identity = { runId: input.binding.runId, normalizedSessionId: input.session.normalizedSessionId,
|
||||
runnerInstanceId: "runner-crashed", environmentLeaseId: "lease", turnId: "turn", itemId: "item" };
|
||||
const event = { schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceKind: "runner", sourceSeq: 1,
|
||||
sourceEventId: "provider-identity", sourceInstanceId: identity.runnerInstanceId, runId: identity.runId,
|
||||
normalizedSessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId,
|
||||
priority: 0, emittedAt: new Date().toISOString(), eventType: "session.started",
|
||||
payload: { providerSessionId: "provider-session", processId: mode === "provider_alive" ? process.pid : 99_999_998 } };
|
||||
const receipt = { sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:1`,
|
||||
sourcePayloadSha256: mode === "wrong_receipt" ? "wrong" : hash(event), payload: { prpEvent: event } };
|
||||
const stop = { eventType: mode === "new_launch" ? "native.process_start_requested" : "native.local_process_stopped",
|
||||
payload: { processPid: mode === "worker_alive" ? process.pid : 99_999_999, processGroupId: mode === "worker_alive" ? process.pid : 99_999_999 } };
|
||||
let queryIndex = 0;
|
||||
const db = { select() { const rows = [
|
||||
[{ provider: mode === "remote" ? "daytona" : "local", releasedAt: mode === "unreleased" ? null : new Date() }],
|
||||
[stop], mode === "missing_receipt" ? [] : [receipt],
|
||||
][queryIndex++] ?? [];
|
||||
const q: any = { from: () => q, where: () => q, orderBy: () => q, limit: () => q,
|
||||
then: Promise.resolve(rows).then.bind(Promise.resolve(rows)) }; return q;
|
||||
} } as unknown as Db;
|
||||
const run = { id: input.binding.runId, companyId: mode === "foreign_company" ? "foreign" : input.binding.companyId,
|
||||
agentId: input.binding.agentId, nativeIssueId: input.binding.issueId, runtimeMode: "native", status: "failed", finishedAt: new Date(),
|
||||
nativeSessionId: input.session.normalizedSessionId, runnerInstanceId: mode === "foreign_runner" ? "foreign" : identity.runnerInstanceId,
|
||||
runnerProfileJson: { nativeExecutionInput: input, nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("local") } } as typeof heartbeatRuns.$inferSelect;
|
||||
try {
|
||||
await mkdir(join(root, "runner"), { recursive: true });
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
const runnerPath = join(root, "runner/runner-state.json");
|
||||
await writeFile(join(root, "control-plane/control-plane-state.json"), JSON.stringify({ ...durableControlPlaneState(identity), committedEvents: [{ envelope: { payload: event } }] }));
|
||||
await writeFile(runnerPath, JSON.stringify(durableRunnerState({ ...identity, ...(mode === "foreign_run" ? { runId: "foreign" } : {}) }, "ready")));
|
||||
await writeFile(join(root, `runner/${provider}-provider-state.json`), JSON.stringify({ lifecycle: "turn_active" }));
|
||||
if (mode === "symlink") { await rename(runnerPath, join(base, "external")); await symlink(join(base, "external"), runnerPath); }
|
||||
const proof = await verifyStoppedNativeSessionForContinuation(db, run);
|
||||
if (["stopped", "changed_state", "changed_pid"].includes(mode)) {
|
||||
expect(proof).not.toBeNull();
|
||||
if (mode === "changed_state") await writeFile(runnerPath, "{}");
|
||||
const kill = mode === "changed_pid" ? vi.spyOn(process, "kill").mockReturnValue(true) : null;
|
||||
try { expect(proof!.retire()).toBe(mode === "stopped"); } finally { kill?.mockRestore(); }
|
||||
expect(await readFile(join(root, `runner/${provider}-provider-state.json`), "utf8")).toBe('{"lifecycle":"turn_active"}');
|
||||
} else expect(proof).toBeNull();
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
|
||||
await rm(base, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("explicit failed native retry physical evidence", () => {
|
||||
it.each([
|
||||
"suspended",
|
||||
|
||||
@@ -1756,7 +1756,7 @@ const CLEANUP_CANONICAL_FILES = [
|
||||
] as const;
|
||||
const CLEANUP_ACTIVATION_FILE = "cleanup-activation.json";
|
||||
|
||||
function cleanupStateSnapshot(root: string) {
|
||||
function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state.json") {
|
||||
if (
|
||||
![root, resolve(root, "runner"), resolve(root, "control-plane")].every(
|
||||
isSafeNativeStateDirectory,
|
||||
@@ -1764,7 +1764,8 @@ function cleanupStateSnapshot(root: string) {
|
||||
) {
|
||||
throw new Error("native_cleanup_maintenance_unproven");
|
||||
}
|
||||
const bytes = CLEANUP_CANONICAL_FILES.map((file) =>
|
||||
const files = [...CLEANUP_CANONICAL_FILES.slice(0, 2), `runner/${providerFile}`];
|
||||
const bytes = files.map((file) =>
|
||||
readBoundedNativeFile(
|
||||
resolve(root, file),
|
||||
NATIVE_RUNNER_STATE_MAX_BYTES,
|
||||
@@ -2362,6 +2363,79 @@ export function rebaseRetainedNativeCleanupProviderHome(
|
||||
}
|
||||
}
|
||||
|
||||
/** Physical cleanup for an explicitly authorized NEW conversation turn. Unlike
|
||||
* automatic replacement, this does not certify or replay the interrupted actions.
|
||||
* The caller holds the issue/controller/run locks and preserves their history.
|
||||
* Retain the old durable root permanently; only the exact in-memory cleanup
|
||||
* owner is retired, and the successor must use a fresh normalized session.
|
||||
*/
|
||||
export async function verifyStoppedNativeSessionForContinuation(
|
||||
db: Db,
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
): Promise<{ evidence: Record<string, unknown>; retire: () => boolean } | null> {
|
||||
try {
|
||||
if (run.runtimeMode !== "native" || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status) ||
|
||||
!run.finishedAt || !run.nativeIssueId || !run.nativeSessionId || !run.runnerInstanceId) return null;
|
||||
const execution = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput);
|
||||
if (!(["codex", "acpx"] as string[]).includes(execution.provider.kind) ||
|
||||
execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId ||
|
||||
execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId ||
|
||||
nativeSessionKey(execution) !== run.nativeSessionId ||
|
||||
record(run.runnerProfileJson).nativeToolContractFingerprint !== nativeToolContractFingerprintForTarget("local")) return null;
|
||||
const scope = nativeSessionScopeKey(execution);
|
||||
const idle = () => !activeNativeSessions.has(run.id) && !executingRunnerdSessionScopes.has(scope) &&
|
||||
!initializingSessionToolAuthorities.has(scope) && !warmNativeSessions.has(scope);
|
||||
if (!idle()) return null;
|
||||
const leases = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id)));
|
||||
if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt || lease.cleanupStatus === "failed")) return null;
|
||||
const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id);
|
||||
if (!stopped) return null;
|
||||
const root = scopedRunnerdStateRoot(execution);
|
||||
const providerFile = runnerProviderStateFilename(execution);
|
||||
const snapshot = cleanupStateSnapshot(root, providerFile);
|
||||
const identity = record(snapshot.control.identity);
|
||||
if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId ||
|
||||
!["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key =>
|
||||
typeof identity[key] === "string" && identity[key] && snapshot.runner[key] === identity[key]) ||
|
||||
!Array.isArray(snapshot.control.committedEvents)) return null;
|
||||
const providerEvents = snapshot.control.committedEvents.map(entry => record(record(record(entry).envelope).payload))
|
||||
.filter(event => ["session.started", "session.resumed"].includes(String(event.eventType)));
|
||||
if (!providerEvents.length) return null;
|
||||
const receipts = await db.select().from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.sourceInstanceId, run.runnerInstanceId),
|
||||
inArray(heartbeatRunEvents.eventType, ["session.started", "session.resumed"])));
|
||||
const providerPids = new Set<number>();
|
||||
for (const event of providerEvents) {
|
||||
const provider = record(event.payload);
|
||||
if (!validatePrpEvent(event).ok || event.sourceKind !== "runner" || event.sourceInstanceId !== run.runnerInstanceId ||
|
||||
event.runId !== run.id || event.normalizedSessionId !== run.nativeSessionId ||
|
||||
!cleanupProcessAbsent(provider.processId) || provider.processId === stopped.processPid) return null;
|
||||
const receipt = receipts.find(row => row.sourceEventId === `${run.runnerInstanceId}:${run.id}:${event.sourceSeq}`);
|
||||
const durable = record(record(receipt?.payload).prpEvent);
|
||||
const durableProvider = record(durable.payload);
|
||||
if (!receipt || !validatePrpEvent(durable).ok || durable.sourceInstanceId !== event.sourceInstanceId ||
|
||||
durable.runId !== run.id || durable.normalizedSessionId !== run.nativeSessionId ||
|
||||
receipt.sourcePayloadSha256 !== nativeSha256(durable) ||
|
||||
(durableProvider.processId !== undefined && durableProvider.processId !== provider.processId) ||
|
||||
(durableProvider.driverSessionId ?? durableProvider.providerSessionId) !== provider.providerSessionId) return null;
|
||||
providerPids.add(provider.processId);
|
||||
}
|
||||
if (receipts.length !== providerEvents.length) return null;
|
||||
const unchanged = () => idle() && cleanupProcessAbsent(stopped.processPid) &&
|
||||
[...providerPids].every(cleanupProcessAbsent) && cleanupStateSnapshot(root, providerFile).fingerprint === snapshot.fingerprint;
|
||||
return {
|
||||
evidence: { schema: "paperclip.stopped_native_conversation.v1", runId: run.id,
|
||||
nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId,
|
||||
processPid: stopped.processPid, providerPids: [...providerPids], stateFingerprint: snapshot.fingerprint },
|
||||
retire: () => unchanged() && completeTerminatedLocalNativeSessionCleanup({
|
||||
companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId!,
|
||||
}),
|
||||
};
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
/** Prove that a crashed local Codex turn ended without external effects. No provider
|
||||
* is launched and no retained state is rewritten. The successor must use a fresh
|
||||
* normalized session; the old directory remains available for investigation. */
|
||||
|
||||
@@ -975,7 +975,7 @@ export const runnerSuites: readonly RunnerSuiteFixture[] = [
|
||||
profiles: runnerProfiles.filter(profile => ["runner-codex", "runner-acpx-claude"].includes(profile.id))
|
||||
.map(profile => productionStoryProfile(defaultPermissionProfile(profile))),
|
||||
environments: [localEnvironment], tasks: chatQualificationTasks, expectedMatrixSize: 6,
|
||||
definitionMetadata: { version: 6, permissions: "production-defaults", instructions: "production", crashBoundary: "verified-native-worker-pid-at-file-wait", recovery: "user-visible-retry", answerGrading: "exact-grounded-propositions-plus-separate-semantic-review", scheduling: "explicit-only" },
|
||||
definitionMetadata: { version: 7, permissions: "production-defaults", instructions: "production", crashBoundary: "verified-native-worker-pid-at-file-wait", recovery: "new-user-message-after-verified-cleanup", answerGrading: "exact-grounded-propositions-plus-separate-semantic-review", scheduling: "explicit-only" },
|
||||
},
|
||||
...(process.env.PAPERCLIP_RUNNER_E2E_CONNECTION_REVIEWS === "1" ? [connectionReviewSuite] : []),
|
||||
{
|
||||
|
||||
@@ -5,12 +5,12 @@ import { assertActiveHandoff, assertAnswerFacts, assertCrashRecovered, assertWor
|
||||
import { runnerMatrix } from "./catalog.js";
|
||||
import { buildRunnerE2EProcessEnvironment } from "./harness-env.js";
|
||||
|
||||
const run = { id: "old", agentId: "original", companyId: "company", runtimeMode: "native", status: "cancelled", errorCode: "issue_reassigned", finishedAt: "2026-09-21T00:01:00Z", contextSnapshot: { issueId: "task" } };
|
||||
const run = { id: "old", nativeSessionId: "old-session", agentId: "original", companyId: "company", runtimeMode: "native", status: "cancelled", errorCode: "issue_reassigned", finishedAt: "2026-09-21T00:01:00Z", contextSnapshot: { issueId: "task" } };
|
||||
const successor = { ...run, id: "next", agentId: "successor", status: "succeeded", startedAt: "2026-09-21T00:01:01Z" };
|
||||
const before = { id: "task", assigneeAgentId: "original", description: "Preserve scope", projectId: null };
|
||||
const plan = { body: "Friday REFERENCE", latestRevisionId: "revision" };
|
||||
const handoff = { before, after: { ...before, status: "done", assigneeAgentId: "successor" }, oldRun: run, boundary: { ...run, status: "running" }, runs: [run, successor], successorId: "successor", planBefore: plan, planAfter: plan, draft: { id: "draft", latestRevisionId: "v1", body: "REFERENCE" }, draftAfter: { id: "draft", body: "REFERENCE" }, draftRevisions: [{ id: "v1", body: "REFERENCE" }], output: { body: "REFERENCE", createdByAgentId: "successor" }, reference: "REFERENCE", audit: [{ action: "issue.reassigned", details: { source: "paperclip_runner_protocol" } }], taskIds: ["task"] };
|
||||
const recovery = { boundary: { ...run, status: "running" }, failed: { ...run, status: "failed" }, runs: [{ ...run, status: "failed" }, { ...successor, agentId: "original" }], issueId: "task", prompt: "Read my brief", comments: [{ body: "Read my brief" }, { body: "REFERENCE MARKER", authorAgentId: "original", createdByRunId: "next" }], reference: "REFERENCE", marker: "MARKER", planBefore: { body: "plan MARKER", latestRevisionId: "v1" }, planAfter: { body: "plan MARKER", latestRevisionId: "v1" } };
|
||||
const recovery = { boundary: { ...run, status: "running" }, failed: { ...run, status: "failed" }, runs: [{ ...run, status: "failed" }, { ...successor, agentId: "original", nativeSessionId: "fresh-session", contextSnapshot: { issueId: "task", previousRunId: "old", forceFreshSession: true } }], issueId: "task", prompt: "Read my brief", comments: [{ body: "Read my brief" }, { body: "REFERENCE MARKER", authorAgentId: "original", createdByRunId: "next" }], reference: "REFERENCE", marker: "MARKER", planBefore: { body: "plan MARKER", latestRevisionId: "v1" }, planAfter: { body: "plan MARKER", latestRevisionId: "v1" } };
|
||||
|
||||
describe("remaining native chat qualification", () => {
|
||||
it("calibrates the pidfd helper against reuse and wrong-identity faults", () => {
|
||||
@@ -44,6 +44,8 @@ describe("remaining native chat qualification", () => {
|
||||
{ failed: { ...recovery.failed, id: "unrelated" } },
|
||||
{ runs: [recovery.runs[0]!, { ...successor, status: "failed" }] },
|
||||
{ runs: [recovery.runs[0]!, { ...successor, contextSnapshot: { issueId: "new-chat" } }] },
|
||||
{ runs: [recovery.runs[0]!, { ...recovery.runs[1]!, nativeSessionId: "old-session" }] },
|
||||
{ runs: [recovery.runs[0]!, { ...recovery.runs[1]!, contextSnapshot: { issueId: "task", previousRunId: "unrelated", forceFreshSession: true } }] },
|
||||
{ comments: [...recovery.comments, recovery.comments[0]!] },
|
||||
{ comments: [recovery.comments[0]!, { ...recovery.comments[1], createdByRunId: "old" }] },
|
||||
{ comments: [recovery.comments[0]!, { ...recovery.comments[1], body: "MARKER" }] },
|
||||
|
||||
@@ -63,6 +63,9 @@ export function assertCrashRecovered(e: {
|
||||
expect(e.runs.find(r => r.id === e.failed.id)).toMatchObject({ status: "failed", contextSnapshot: { issueId: e.issueId } });
|
||||
const retry = e.runs.find(r => r.id !== e.failed.id)!;
|
||||
expect(retry).toMatchObject({ agentId: e.failed.agentId, status: "succeeded", runtimeMode: "native", contextSnapshot: { issueId: e.issueId } });
|
||||
expect(retry.contextSnapshot).toMatchObject({ previousRunId: e.failed.id, forceFreshSession: true });
|
||||
expect((retry as Row).nativeSessionId).not.toBe(e.failed.nativeSessionId);
|
||||
expect((retry as Row).nativeSessionId).toBeTruthy();
|
||||
expect(e.comments.filter(c => !c.authorAgentId && c.body === e.prompt)).toHaveLength(1);
|
||||
const replies = e.comments.filter(c => c.authorAgentId && c.body.includes(e.marker));
|
||||
expect(replies).toHaveLength(1);
|
||||
@@ -145,7 +148,7 @@ export async function runWorkerCrash(context: Context) {
|
||||
const processIdentity = JSON.parse(execFileSync("python3", [faultHelper, "inspect", String(boundary.processPid), boundary.id], { encoding: "utf8" }));
|
||||
const command = execFileSync("ps", ["-p", String(boundary.processPid), "-o", "command="], { encoding: "utf8" });
|
||||
assertWorkerIdentity(boundary, command, input.execution.environment.id);
|
||||
await input.evidence("chat-worker-fault.json", { boundary, planBefore, processIdentity, fault: "SIGKILL through verified Linux pidfd", recovery: "visible Retry button" });
|
||||
await input.evidence("chat-worker-fault.json", { boundary, planBefore, processIdentity, fault: "SIGKILL through verified Linux pidfd", recovery: "new user message after verified provider cleanup" });
|
||||
const fault = JSON.parse(execFileSync("python3", [faultHelper, "kill", String(boundary.processPid), boundary.id, processIdentity.startTicks], { encoding: "utf8" }));
|
||||
expect(fault.signalled).toBe(true);
|
||||
await input.evidence("chat-worker-fault-delivered.json", fault);
|
||||
@@ -177,17 +180,14 @@ export async function runWorkerCrash(context: Context) {
|
||||
await input.evidence("chat-worker-quarantine.json", { failed, retryStatus: refused.status(), savedPlanPreserved: true,
|
||||
usableRecovery: false, classification: "product recovery boundary; no provider retry was admitted" });
|
||||
await input.capture("worker-quarantined", "Worker recovery requires reconciliation", "worker-quarantined.png");
|
||||
throw new Error("worker_crash_recovery_unqualified: native_session_cleanup_quarantined requires explicit reconciliation; generic Retry is correctly unavailable");
|
||||
// Reconcile the fixture's known outcomes through ordinary user input.
|
||||
// The saved plan is verified above; the only interrupted command reads a
|
||||
// local brief. The new provider must discover its reference from the file.
|
||||
await sendChatMessage(input.page, `The prior worker failed. The plan is already saved and must remain unchanged. The interrupted command only waited for my local brief, which is now available. Continue in a fresh conversation turn: read the brief using node ${wait.scriptPath}, then reply with its reference and ${marker}. Do not create tasks or projects.`);
|
||||
}
|
||||
else {
|
||||
throw new Error(`worker_crash_unexpected_recovery_boundary: ${failed.errorCode}`);
|
||||
}
|
||||
const retry = input.page.getByRole("button", { name: "Retry", exact: true });
|
||||
await expect(retry).toHaveCount(1, { timeout: 60_000 });
|
||||
const [retryResponse] = await Promise.all([
|
||||
input.page.waitForResponse(response => response.request().method() === "POST" &&
|
||||
new URL(response.url()).pathname === `/api/agents/${input.fixtures.agent.id}/wakeup`),
|
||||
retry.click(),
|
||||
]);
|
||||
await input.evidence("chat-worker-retry-response.json", { status: retryResponse.status(), body: await retryResponse.json() });
|
||||
expect(retryResponse.ok(), "The visible Retry must admit an attempt before waiting for its result").toBe(true);
|
||||
await context.idle(2);
|
||||
const e = { boundary, failed, runs: await context.allRuns(), issueId: context.issue().id,
|
||||
prompt, comments: await context.comments(), reference, marker, planBefore,
|
||||
|
||||
Reference in new issue
Block a user