test: preserve task binding before native selection and require explicit resumption

This commit is contained in:
Dotta committed 2026-10-21 12:29:09 -05:00
1 parent 5b7dfc43e3
commit 6036cee2f4
8 files changed
+50 -16

No files matched your search

@@ -132,9 +132,9 @@ const support = await getEmbeddedPostgresTestSupport();
expect(comment.authorUserId).toBe("second-user"); expect(comment.authorUserId).toBe("second-user");
} else expect(receipt.status).toBe("deferred_issue_execution"); } else expect(receipt.status).toBe("deferred_issue_execution");
}); });
it.each(["ready", "unacknowledged", "pause", "recovery", "controller", "process_running", "identity_missing", "remote_pending", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"])("delivers a saved native message after run-only Stop exactly once (%s)", async gate => { it.each(["ready", "unacknowledged", "pause", "resolved_recovery", "controller", "process_running", "identity_missing", "remote_pending", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"])("retains pre-Stop input and delivers once only after a later authorized message (%s)", async gate => {
const f = await seed(); const f = await seed();
if (gate !== "recovery") await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); if (gate !== "resolved_recovery") await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issueId)); await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issueId));
await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureDetail: null, leaseOwner: gate === "controller" ? "still-cleaning" : null }) await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureDetail: null, leaseOwner: gate === "controller" ? "still-cleaning" : null })
.where(eq(nativeRunFinalizations.runId, f.sourceRunId)); .where(eq(nativeRunFinalizations.runId, f.sourceRunId));
@@ -189,12 +189,27 @@ const support = await getEmbeddedPostgresTestSupport();
const heartbeat = heartbeatService(db); const heartbeat = heartbeatService(db);
await heartbeat.resumeRemoteStopComments(source); await heartbeat.resumeRemoteStopComments(source);
await heartbeat.resumeRemoteStopComments(source); await heartbeat.resumeRemoteStopComments(source);
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(0);
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)))[0])
.toMatchObject({ status: "deferred_issue_execution", runId: null });
const resumeId = randomUUID();
await db.insert(issueComments).values({ id: resumeId, companyId: f.companyId, issueId: f.issueId,
authorType: "user", authorUserId: "board", body: "Resume with my earlier direction.", createdAt: new Date("2026-09-11T11:00:00Z") });
queuedIds = [...queuedIds, resumeId];
expectedIds = [...expectedIds, resumeId];
await db.update(agentWakeupRequests).set({ payload: {
issueId: f.issueId, commentId: resumeId, _paperclipWakeContext: { issueId: f.issueId, wakeReason: "issue_commented",
wakeCommentId: resumeId, wakeCommentIds: queuedIds },
} }).where(eq(agentWakeupRequests.id, queueId));
await heartbeat.resumeRemoteStopComments(source);
await heartbeat.resumeRemoteStopComments(source);
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
expect(successors).toHaveLength(["ready", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"].includes(gate) ? 1 : 0); expect(successors).toHaveLength(["ready", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors", "resolved_recovery"].includes(gate) ? 1 : 0);
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId));
if (["ready", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"].includes(gate)) { if (["ready", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors", "resolved_recovery"].includes(gate)) {
expect(wake).toMatchObject({ status: "coalesced", runId: successors[0].id, requestedByActorId: "board" }); expect(wake).toMatchObject({ status: "coalesced", runId: successors[0].id, requestedByActorId: "board" });
expect(successors[0].contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds }); expect(successors[0].contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds });
if (gate === "resolved_recovery") expect(successors[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
} else expect(wake.status).toBe("deferred_issue_execution"); } else expect(wake.status).toBe("deferred_issue_execution");
}); });
@@ -520,8 +520,6 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations).where(and( leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),
)) : []; )) : [];
// Stop ends one response. Saved user input can enter ordinary admission
// once its executor settles; it does not need a manufactured crash incident.
// The ordinary path still enforces task holds, ownership, and native session // The ordinary path still enforces task holds, ownership, and native session
// cleanup before starting a provider. // cleanup before starting a provider.
const stoppedNativeContinuation = currentRun && isAcknowledgedNativeStop(currentRun) && const stoppedNativeContinuation = currentRun && isAcknowledgedNativeStop(currentRun) &&
+5 -2
View File
@@ -118,8 +118,11 @@ pnpm test:e2e:runner -- --list --suite followup-stop-resume
pnpm test:e2e:runner -- --id followup-stop-resume.runner-codex.local.followup-stop-resume --max-automatic-retries 0 pnpm test:e2e:runner -- --id followup-stop-resume.runner-codex.local.followup-stop-resume --max-automatic-retries 0
``` ```
The checkpoint poll treats invalid, duplicate, corrupt, or unauthorized evidence Before native selection, the observer verifies the existing task and context
as a permanent attempt failure. An artifact that is not registered yet may wait. bindings while the native task ID is still unset. Every present task ID must
agree. A checkpoint still requires the selected native run and its owned provider
turn. The checkpoint poll treats invalid, duplicate, corrupt, or unauthorized
evidence as a permanent attempt failure. An artifact that is not registered yet may wait.
The final result opens its actual attachment chip and downloads through the file The final result opens its actual attachment chip and downloads through the file
tab. Both inner and outer failure captures use the same closed lifecycle fields. tab. Both inner and outer failure captures use the same closed lifecycle fields.
They do not put prompt, provider text, configuration, or logs into api-state.json. They do not put prompt, provider text, configuration, or logs into api-state.json.
@@ -5,9 +5,9 @@ import { captureFollowupStopFailureApiState, followupStopApiState } from "./foll
it("uses the same closed selection for outer failure capture without reading conversations or logs", async () => { it("uses the same closed selection for outer failure capture without reading conversations or logs", async () => {
const issue = { id: "task", companyId: "company", status: "in_progress", executionRunId: "run", executionWorkspaceId: "workspace", const issue = { id: "task", companyId: "company", status: "in_progress", executionRunId: "run", executionWorkspaceId: "workspace",
description: "PRIVATE PROMPT", title: "PRIVATE TITLE" }; description: "PRIVATE PROMPT", title: "PRIVATE TITLE" };
const run = { id: "run", companyId: "company", agentId: "agent", nativeIssueId: "task", status: "cancelled", nativePhase: "terminal_failure", const run = { id: "run", companyId: "company", agentId: "agent", issueId: "task", nativeIssueId: "task", status: "cancelled", nativePhase: "terminal_failure",
runtimeMode: "native", startedAt: "2026-10-21T12:00:00Z", finishedAt: "2026-10-21T12:00:01Z", runtimeMode: "native", startedAt: "2026-10-21T12:00:00Z", finishedAt: "2026-10-21T12:00:01Z",
contextSnapshot: { prompt: "PRIVATE PROMPT" }, resultJson: { summary: "PRIVATE PROVIDER TEXT" }, runnerProfileJson: { token: "PRIVATE CONFIG" } }; contextSnapshot: { issueId: "task", prompt: "PRIVATE PROMPT" }, resultJson: { summary: "PRIVATE PROVIDER TEXT" }, runnerProfileJson: { token: "PRIVATE CONFIG" } };
const get = vi.fn(async (route: string) => { const get = vi.fn(async (route: string) => {
if (route === "/api/issues/task") return issue; if (route === "/api/issues/task") return issue;
if (route === "/api/companies/company/heartbeat-runs?agentId=agent&limit=20") return [run]; if (route === "/api/companies/company/heartbeat-runs?agentId=agent&limit=20") return [run];
@@ -23,7 +23,7 @@ it("uses the same closed selection for outer failure capture without reading con
expect(JSON.stringify(evidence.mock.calls)).not.toContain("PRIVATE"); expect(JSON.stringify(evidence.mock.calls)).not.toContain("PRIVATE");
expect(evidence.mock.calls[0]![1]).toEqual({ capturePhase: "failure", expect(evidence.mock.calls[0]![1]).toEqual({ capturePhase: "failure",
issue: { id: "task", companyId: "company", status: "in_progress", executionRunId: "run", executionWorkspaceId: "workspace" }, issue: { id: "task", companyId: "company", status: "in_progress", executionRunId: "run", executionWorkspaceId: "workspace" },
runs: [{ id: "run", companyId: "company", agentId: "agent", nativeIssueId: "task", status: "cancelled", nativePhase: "terminal_failure", runs: [{ id: "run", companyId: "company", agentId: "agent", issueId: "task", nativeIssueId: "task", contextIssueId: "task", status: "cancelled", nativePhase: "terminal_failure",
runtimeMode: "native", startedAt: "2026-10-21T12:00:00Z", finishedAt: "2026-10-21T12:00:01Z" }] }); runtimeMode: "native", startedAt: "2026-10-21T12:00:00Z", finishedAt: "2026-10-21T12:00:01Z" }] });
}); });
+1 -1
View File
@@ -5,7 +5,7 @@ export function followupStopApiState(phase: string, issue: Row | null | undefine
return { capturePhase: phase, return { capturePhase: phase,
issue: issue ? { id: issue.id, companyId: issue.companyId, status: issue.status, executionRunId: issue.executionRunId, issue: issue ? { id: issue.id, companyId: issue.companyId, status: issue.status, executionRunId: issue.executionRunId,
executionWorkspaceId: issue.executionWorkspaceId } : null, executionWorkspaceId: issue.executionWorkspaceId } : null,
runs: runs.map(run => ({ id: run.id, companyId: run.companyId, agentId: run.agentId, nativeIssueId: run.nativeIssueId, runs: runs.map(run => ({ id: run.id, companyId: run.companyId, agentId: run.agentId, issueId: run.issueId, nativeIssueId: run.nativeIssueId, contextIssueId: run.contextSnapshot?.issueId,
status: run.status, nativePhase: run.nativePhase, runtimeMode: run.runtimeMode, startedAt: run.startedAt, finishedAt: run.finishedAt })) }; status: run.status, nativePhase: run.nativePhase, runtimeMode: run.runtimeMode, startedAt: run.startedAt, finishedAt: run.finishedAt })) };
} }
@@ -11,7 +11,7 @@ import { createTaskThroughUi, submitTaskReply } from "./user-actions.js";
import { waitForRunDelivery } from "./run-delivery.js"; import { waitForRunDelivery } from "./run-delivery.js";
import { followupStopApiState } from "./followup-stop-evidence.js"; import { followupStopApiState } from "./followup-stop-evidence.js";
import { downloadTaskTextArtifact } from "./task-text-artifact-download.js"; import { downloadTaskTextArtifact } from "./task-text-artifact-download.js";
import { isActiveStopCheckpointRun, exactSyntheticJson, followupStopFiles, gradeFollowupStopResume, STOP_QUIET_WINDOW_MS, type FollowupStopObservation } from "./followup-stop-resume.js"; import { stopRunBelongsToTask, isActiveStopCheckpointRun, exactSyntheticJson, followupStopFiles, gradeFollowupStopResume, STOP_QUIET_WINDOW_MS, type FollowupStopObservation } from "./followup-stop-resume.js";
import type { LiveFixtureValues } from "./live-fixtures.js"; import type { LiveFixtureValues } from "./live-fixtures.js";
import type { MatrixExecution } from "./types.js"; import type { MatrixExecution } from "./types.js";
@@ -41,7 +41,7 @@ export async function runFollowupStopResumeFlow(input: {
api.get<Row[]>(`/api/companies/${fixtures.company.id}/heartbeat-runs?agentId=${fixtures.agent.id}&limit=20`)]); api.get<Row[]>(`/api/companies/${fixtures.company.id}/heartbeat-runs?agentId=${fixtures.agent.id}&limit=20`)]);
issue = current; issue = current;
runs = await Promise.all(listed.map(run => api.get<Row>(`/api/heartbeat-runs/${run.id}`))); runs = await Promise.all(listed.map(run => api.get<Row>(`/api/heartbeat-runs/${run.id}`)));
if (runs.some(run => run.companyId !== fixtures.company.id || run.agentId !== fixtures.agent.id || run.nativeIssueId !== issue!.id)) if (runs.some(run => !stopRunBelongsToTask(run, observation.binding)))
throw new Error("Stop/resume admitted work outside its original task"); throw new Error("Stop/resume admitted work outside its original task");
input.observe(issue, runs, gradeFollowupStopResume(observation)); input.observe(issue, runs, gradeFollowupStopResume(observation));
} }
@@ -84,7 +84,7 @@ export async function runFollowupStopResumeFlow(input: {
load: async () => { load: async () => {
await refresh(); await refresh();
const run = runs[0]; const run = runs[0];
if (runs.length !== 1 || !run || run.status !== "running") return false; if (runs.length !== 1 || !run || run.status !== "running" || run.runtimeMode !== "native" || run.nativeIssueId !== issue!.id) return false;
const events = prpEvents(await runEvents(run.id), run.id); const events = prpEvents(await runEvents(run.id), run.id);
if (!isActiveStopCheckpointRun(run, events)) return false; if (!isActiveStopCheckpointRun(run, events)) return false;
const checkpoint = await readRegisteredArtifacts(api, issue!.id, run.id, [files.checkpoint]) const checkpoint = await readRegisteredArtifacts(api, issue!.id, run.id, [files.checkpoint])
+14 -1
View File
@@ -1,6 +1,6 @@
import { describe, expect, it } from "vitest"; import { describe, expect, it } from "vitest";
import { runnerMatrix } from "./catalog.js"; import { runnerMatrix } from "./catalog.js";
import { isActiveStopCheckpointRun, exactSyntheticJson, gradeFollowupStopResume, type FollowupStopObservation } from "./followup-stop-resume.js"; import { stopRunBelongsToTask, isActiveStopCheckpointRun, exactSyntheticJson, gradeFollowupStopResume, type FollowupStopObservation } from "./followup-stop-resume.js";
function fixture(): FollowupStopObservation { function fixture(): FollowupStopObservation {
return { binding: { companyId: "company", issueId: "task", agentId: "agent" }, return { binding: { companyId: "company", issueId: "task", agentId: "agent" },
@@ -78,3 +78,16 @@ it("observes the actual active provider through its durable turn rather than a f
...["run.result.accepted", "run.terminal", "turn.completed", "turn.failed", "turn.cancelled"].map(eventType => [started, { runId: "run", eventType }])]) ...["run.result.accepted", "run.terminal", "turn.completed", "turn.failed", "turn.cancelled"].map(eventType => [started, { runId: "run", eventType }])])
expect(isActiveStopCheckpointRun(run, events)).toBe(false); expect(isActiveStopCheckpointRun(run, events)).toBe(false);
}); });
it("keeps the recorded task binding during preparation without treating an absent native binding as foreign work", () => {
const binding = { companyId: "company", agentId: "agent", issueId: "task" };
const queued = { ...binding, nativeIssueId: null, contextSnapshot: { issueId: "task" }, runtimeMode: "legacy", runtimeModeResolvedAt: null };
expect(stopRunBelongsToTask(queued, binding)).toBe(true);
const selected = { ...queued, nativeIssueId: "task", runtimeMode: "native" };
expect(stopRunBelongsToTask(selected, binding)).toBe(true);
for (const run of [{ ...queued, companyId: "foreign" }, { ...queued, agentId: "foreign" },
{ ...queued, issueId: "foreign" }, { ...queued, nativeIssueId: "foreign" },
{ ...queued, contextSnapshot: { issueId: "foreign" } }, { ...queued, nativeIssueId: "" },
{ ...queued, issueId: undefined, contextSnapshot: null }]) expect(stopRunBelongsToTask(run, binding)).toBe(false);
});
+5
View File
@@ -25,6 +25,11 @@ export const followupStopTask: RunnerTaskFixture = {
].join("\n"); ].join("\n");
}, buildMatchers: () => [], }, buildMatchers: () => [],
}; };
export function stopRunBelongsToTask(run: { companyId?: unknown; agentId?: unknown; issueId?: unknown; nativeIssueId?: unknown;
contextSnapshot?: { issueId?: unknown } | null }, binding: { companyId: string; agentId: string; issueId: string }) {
const ids = [run.issueId, run.nativeIssueId, run.contextSnapshot?.issueId].filter(id => id !== null && id !== undefined);
return run.companyId === binding.companyId && run.agentId === binding.agentId && ids.length > 0 && ids.every(id => id === binding.issueId);
}
export function isActiveStopCheckpointRun(run: { id?: string; status?: string } | undefined, export function isActiveStopCheckpointRun(run: { id?: string; status?: string } | undefined,
events: readonly { runId?: string; eventType?: string; sourceKind?: string }[]) { events: readonly { runId?: string; eventType?: string; sourceKind?: string }[]) {
if (!run?.id || run.status !== "running") return false; if (!run?.id || run.status !== "running") return false;