mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-19 20:44:16 +02:00
fix: verify response waits before provider dispatch
Retain the final admission receipt when preparation has started but the provider callback has not run. Conflicting execution evidence still requires recovery. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
6d997f889c
commit
d4e6196fb9
3 files changed
+26
-6
No files matched your search
@@ -191,19 +191,31 @@ describePostgres("already available connection requests", () => {
|
||||
details: expect.objectContaining({ previousConnectionId: f.connection.id, connectionId: replacement.id }) })]));
|
||||
});
|
||||
|
||||
it("rearms an unstarted wait without manufacturing a workspace repair hold", async () => {
|
||||
it.each(["queued", "running"] as const)("rearms an unstarted %s wait without manufacturing a workspace repair hold", async (status) => {
|
||||
const f = await seed(); const run = await retry(f);
|
||||
await satisfiedConnectionIntentService(db).sweepPending();
|
||||
const key = `connection-intent:${f.intent.id}:expired`;
|
||||
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", status: "claimed", idempotencyKey: key, runId: run.id }).returning();
|
||||
await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() });
|
||||
await db.update(heartbeatRuns).set({ status: "queued", wakeupRequestId: wake!.id, resultJson: null,
|
||||
await db.update(heartbeatRuns).set({ status, startedAt: status === "running" ? new Date() : null,
|
||||
runtimeModeResolvedAt: status === "running" ? new Date() : null,
|
||||
controllerBootId: status === "running" ? randomUUID() : null,
|
||||
executionStage: status === "running" ? "preparing" : null, wakeupRequestId: wake!.id, resultJson: null,
|
||||
contextSnapshot: { issueId: f.issueId, interactionId: f.intent.id, connectionIntentResolution: "existing_connection",
|
||||
interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", wakeReason: "issue_commented", source: "connection_intent.resolved" } }).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.update(connectionGrants).set({ status: "revoked" });
|
||||
expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id,
|
||||
expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" });
|
||||
const dispatch = createPostgresRunDispatchAdapter(db);
|
||||
if (status === "queued") {
|
||||
expect(await dispatch.cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id,
|
||||
expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" });
|
||||
} else {
|
||||
const invoke = vi.fn(async () => null);
|
||||
expect(await dispatch.dispatchResolvedInteractionIfCurrent({ companyId: f.companyId, runId: run.id,
|
||||
expectedStatus: "running", now: new Date(), dispatch: invoke })).toMatchObject({ dispatched: false,
|
||||
cancellation: { outcome: "cancelled", errorCode: "issue_waiting_for_response" } });
|
||||
expect(invoke).not.toHaveBeenCalled();
|
||||
}
|
||||
const cancelled = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0];
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, cancelled)).toBe(false);
|
||||
expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).toBeNull();
|
||||
|
||||
@@ -882,6 +882,7 @@ export function createPostgresRunDispatchAdapter(
|
||||
decision: Extract<ReturnType<typeof decideQueuedRunStaleness>, { stale: true }>,
|
||||
expectedStatus: "queued" | "running",
|
||||
now: Date,
|
||||
providerDispatchNotStarted = false,
|
||||
): Promise<CancelStaleQueuedRunOutcome> {
|
||||
const [row] = await tx
|
||||
.update(heartbeatRuns)
|
||||
@@ -893,6 +894,9 @@ export function createPostgresRunDispatchAdapter(
|
||||
resultJson: {
|
||||
...parseObject(run.resultJson),
|
||||
stopReason: decision.errorCode,
|
||||
...(providerDispatchNotStarted && decision.errorCode === "issue_waiting_for_response"
|
||||
? { preDispatchResponseWait: { version: 1, providerWorkStarted: false } }
|
||||
: {}),
|
||||
...(decision.errorCode === "execution_reconciliation_required"
|
||||
? { executionWait: decision.details }
|
||||
: {}),
|
||||
@@ -1146,6 +1150,7 @@ export function createPostgresRunDispatchAdapter(
|
||||
decision,
|
||||
input.expectedStatus,
|
||||
input.now,
|
||||
true, // This locked gate has not handed off to the provider callback.
|
||||
);
|
||||
return { dispatched: false as const, cancellation };
|
||||
}
|
||||
|
||||
@@ -14,6 +14,9 @@ type DispatchFields = "startedAt" | "runtimeModeResolvedAt" | "processPid" | "pr
|
||||
export function isPreDispatchReviewWait(
|
||||
run: Pick<Run, "runtimeMode" | "status" | "errorCode" | "resultJson"> & Partial<Pick<Run, DispatchFields>>,
|
||||
): boolean {
|
||||
const finalResponseWait = run.errorCode === "issue_waiting_for_response" &&
|
||||
(run.resultJson?.preDispatchResponseWait as { version?: number; providerWorkStarted?: boolean } | undefined)?.version === 1 &&
|
||||
(run.resultJson?.preDispatchResponseWait as { providerWorkStarted?: boolean } | undefined)?.providerWorkStarted === false;
|
||||
return run.runtimeMode === "legacy" && run.status === "cancelled" &&
|
||||
(run.errorCode === "issue_continuation_waiting_on_review" ||
|
||||
(run.errorCode === "issue_waiting_for_response" &&
|
||||
@@ -26,10 +29,10 @@ export function isPreDispatchReviewWait(
|
||||
run.resultJson?.stopReason === run.errorCode &&
|
||||
run.resultJson?.timeoutSource === "stale_queued_run_gate" &&
|
||||
run.resultJson?.workspaceRestoreFailure !== "restore_unsafe_archive" &&
|
||||
run.startedAt === null && run.runtimeModeResolvedAt === null &&
|
||||
(finalResponseWait || (run.startedAt === null && run.runtimeModeResolvedAt === null)) &&
|
||||
run.processPid === null && run.processGroupId === null && run.processStartedAt === null &&
|
||||
run.nativeIssueId === null && run.nativeSessionId === null && run.sessionIdAfter === null &&
|
||||
run.controllerBootId === null && run.controllerLeaseExpiresAt === null && run.executionStage === null;
|
||||
(finalResponseWait || (run.controllerBootId === null && run.controllerLeaseExpiresAt === null && run.executionStage === null));
|
||||
}
|
||||
|
||||
/** The row receipt is a candidate, not permission to discard conflicting
|
||||
|
||||
Reference in new issue
Block a user