mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
fix(chat): preserve deferred Retry source batches (#17883)
## Thinking Path > - Paperclip manages agent work through tasks and governed execution. > - Connected conversations retain the source and authority for each request. > - Task Retry pins the full saved GitHub input batch before it waits for cleanup. > - Resuming that wait could append an independent board queue after authorization. > - Dispatch then rejected the changed batch before the provider started. > - This pull request keeps the authorized batch intact and restores certified undelivered input from historical failed attempts. > - Each board message retains a separate turn and its own admission checks. ## Linked Issues or Issue Description **What happened?** A deferred GitHub task Retry could consume a separate board message while promoting its saved receipt. The new run then failed with `chat_failed_run_retry_not_authorized`. A second Retry could also fail because the historical run retained the mixed batch. **Expected behavior** Retry must preserve the source batch it authorized. Separate board input must keep its own execution path. Historical input may be restored only when durable evidence proves that the rejected attempt never started a provider. **Steps to reproduce** 1. Retain ten authorized GitHub requests behind a failed native turn. 2. Start task Retry while the predecessor still lacks its exact termination receipt. 3. Save an independent board message. 4. Complete cleanup and resume the same Retry receipt. 5. Observe the mixed batch rejected during dispatch. Retry that failed attempt to reproduce the second refusal. Refs #17725 for the existing saved-input Retry contract. This change repairs the later queue merge and its retained failed state. Related #17878 covers snapshot continuation and later turns in recovered directories; its workspace repair can be reviewed independently. ## What Changed - Keep deferred chat Retry promotion scoped to its exact receipt. Ordinary board Retry coalescing retains its existing behavior. - Let a new explicit task Retry restore a historical mixed batch only after strict never-started, source, ownership, allocation, board-comment and prior-delivery checks. - Preserve failed-run history and original budget-skipped receipts. Record restored queue ownership in activity. - Add production queue and HTTP-route regressions. Update the execution contract and verification maps. ## Verification - Before the repair, the deferred ten-input batch failed before identity initialization. The same regression now reaches the real initializer with the exact GitHub batch. The independent board message then reaches its own initializer in a separate turn. - Before historical recovery, ordinary task Retry refused the persisted mixed failed batch. The repair passes that case without changing the failed run. HTTP duplicate requests retain one Retry owner. An additional compatibility probe restored the old queue candidate predicate and removed all historical mixed-state fixture writes: the real producer created the exact batch and the complete historical HTTP recovery case passed (1/1). The fixed-source historical fixture now uses the production serializers. - The affected Retry group passes 86 cases on commit `325c00158df5569f5b42731cda4ecc4584909d99`, including already-coalesced external receipts. Refusals cover edited sources, active controllers, unresolved allocations, changed receipts, foreign external receipt ownership and prior delivery of either board or GitHub input, including immutable budget-skipped receipts. - Full workspace `pnpm -r typecheck` and `pnpm build` pass with these source changes. Map, module-boundary and diff checks pass. - Full local `pnpm test:run` was started. The host cannot create additional PostgreSQL semaphores; database-dependent suites fail or skip at startup. The affected database group runs against a separate migrated PostgreSQL 17.10 test database with Node 24.22.0 on macOS arm64. - The regression deliberately stops after the real identity initializer to avoid provider allocation. It proves HTTP, queue, source authorization, identity and separate continuation behavior. It does not claim a live model result, GitHub publication, deployment or recovery of the original hosted task. - Current head `325c00158df5569f5b42731cda4ecc4584909d99`: all checks are terminal with no failures or pending checks; required `ci / verify` and `ci / e2e` pass. Conditional Storybook checks are skipped. Greptile and Paperclip each reviewed this exact commit at 5/5. Both earlier review threads are resolved. Paperclip retains one optional robustness suggestion for batch ordering; the verified historical producer order remains a strict recovery contract. ## Risks Restoring input could repeat work if the prior turn had executed. The existing strict bootstrap proof rejects provider output, native state, process or controller evidence, and allocations. Restoration also checks current external authority, unchanged board revisions and other run history. An uncertain executor remains held. No migration, provider fallback or relaxation of dispatch validation is included. ## Model Used OpenAI Codex, based on GPT-6, with reasoning, repository inspection, code editing and test execution. The exact runtime model ID and context-window size are not exposed in this session. A separately shared regression supplied independent reproduction evidence. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass (affected database regressions; full local suite has the host limitation described above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3be54534c2
commit
585b4e6af8
7 files changed
+279
-8
No files matched your search
@@ -1582,6 +1582,14 @@ delta when attaching a new run. Other stale-event and authority checks remain.
|
||||
|
||||
### Explicit user continuation after execution failure
|
||||
|
||||
A deferred chat Retry retains its authorized source batch when it resumes.
|
||||
Separate board queues do not join that batch. For a historical mixed batch
|
||||
refused before provider startup, a new task Retry can restore the undelivered
|
||||
receipts only after certifying the exact bootstrap failure, source authority,
|
||||
unchanged board comments and absence of execution, allocation or prior delivery.
|
||||
The restoration is audited. Historical runs and budget-skipped receipts remain
|
||||
unchanged; each input still passes its ordinary admission and dispatch checks.
|
||||
|
||||
When a user's comment also assigns the task back to its agent, the assignment
|
||||
wake retains that comment's continuation intent. Admission validates the saved
|
||||
comment, its author, task, timing and delivery receipt through the same checks as
|
||||
|
||||
@@ -50,6 +50,14 @@ current access, task eligibility, pause and budget gates must still pass before
|
||||
one successor starts. Restarts reuse the receipt; they do not create another
|
||||
Retry generation or another external publication.
|
||||
|
||||
Resuming a chat Retry preserves its exact authorized input batch. Independent
|
||||
board messages keep their own queue receipts and enter a separate turn. If an
|
||||
older queue merged those messages and dispatch refused the batch before any
|
||||
provider work, a new deliberate Retry can restore the undelivered receipts.
|
||||
Restoration requires the exact bootstrap failure, no execution or allocation,
|
||||
current chat authority, unchanged board input and no prior delivery. It records
|
||||
an audit event and retains the failed run and original skipped receipts.
|
||||
|
||||
Several retained recovery holds can describe the same failed execution. A
|
||||
historical hold without a system recovery record uses the current exact Retry's
|
||||
workspace strategy for that run. It does not borrow another run's strategy or
|
||||
|
||||
@@ -1019,3 +1019,10 @@ comment, session generation, and admitted instruction snapshot. A separate retry
|
||||
receipt preserves the original skipped receipt and older failed native result.
|
||||
Current access and normal native stop/cleanup guards still govern execution;
|
||||
board wake metadata alone cannot manufacture chat authority or replay old work.
|
||||
|
||||
Deferred chat Retry excludes independent board queues from its pinned batch.
|
||||
An explicit board Retry can restore a historical mixed batch only with exact
|
||||
never-started evidence, current source authority, unchanged board comments and
|
||||
no prior delivery or allocation. The server records the restoration in activity;
|
||||
failed runs and original skipped receipts remain history. This grants no new
|
||||
agent retry capability and does not weaken execution-time source validation.
|
||||
@@ -272,6 +272,13 @@ After successful admission, reload and verify that the prior skipped receipt doe
|
||||
not strand a later real provider reply or tree resume. Provider fixtures support
|
||||
this behavior; live fixed-source runtime and destination proof remain separate.
|
||||
|
||||
Also defer Retry behind cleanup, queue an independent board message, then
|
||||
complete cleanup. Verify one run receives the exact GitHub batch and the board
|
||||
message enters its own turn. For an older mixed batch rejected before provider
|
||||
startup, ordinary task Retry must recover only certified undelivered receipts.
|
||||
Changed input, active ownership, allocations and prior delivery must keep the
|
||||
refusal effective. Check the restoration audit and unchanged failed-run history.
|
||||
|
||||
### `delegated-configuration`
|
||||
|
||||
Endpoint activity hides private-task events while retaining a next-page cursor across private-only windows, so older readable history stays reachable. Actual-JWT fixtures verify this separately from live provider delivery.
|
||||
|
||||
@@ -61898,9 +61898,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
});
|
||||
|
||||
it.each([
|
||||
...(["real_budget_ingress", "legacy_budget_ingress", "legacy_prior_response", "legacy_execution", "legacy_execution_log", "legacy_execution_lease", "legacy_execution_profile", "legacy_execution_event", "legacy_execution_controller", "legacy_execution_intent", "legacy_no_prior", "legacy_prior_late", "legacy_prior_unprocessed", "legacy_prior_delivery_unprocessed", "legacy_prior_numeric_app", "legacy_response_conflict", "legacy_prior_wrong_app", "legacy_prior_wrong_agent", "legacy_prior_unlinked", "legacy_prior_retargeted", "legacy_explicit_null", "legacy_explicit_empty", "legacy_explicit_malformed", "legacy_explicit_mismatch", "legacy_current_app_changed", "source_app_changed", "source_destination_changed", "accepted", "multiple_saved", "real_queue", "committed_succeeded", "committed_consumed", "committed_failure", "committed_failed_publication", "committed_unknown_publication", "committed_pending_publication", "committed_mixed_publication", "committed_consumed_publication", "committed_succeeded_publication", "committed_delayed_cleanup", "committed_unknown_cleanup", "committed_live_controller", "unreconciled", "source_edited", "source_claimed", "malformed_previous", "endpoint_paused", "principal_revoked", "still_budget_blocked"] as const)
|
||||
...(["real_budget_ingress", "legacy_budget_ingress", "legacy_prior_response", "legacy_execution", "legacy_execution_log", "legacy_execution_lease", "legacy_execution_profile", "legacy_execution_event", "legacy_execution_controller", "legacy_execution_intent", "legacy_no_prior", "legacy_prior_late", "legacy_prior_unprocessed", "legacy_prior_delivery_unprocessed", "legacy_prior_numeric_app", "legacy_response_conflict", "legacy_prior_wrong_app", "legacy_prior_wrong_agent", "legacy_prior_unlinked", "legacy_prior_retargeted", "legacy_explicit_null", "legacy_explicit_empty", "legacy_explicit_malformed", "legacy_explicit_mismatch", "legacy_current_app_changed", "source_app_changed", "source_destination_changed", "accepted", "multiple_saved", "real_queue", "committed_succeeded", "committed_consumed", "committed_failure", "committed_failed_publication", "committed_unknown_publication", "committed_pending_publication", "committed_mixed_publication", "committed_consumed_publication", "committed_succeeded_publication", "committed_delayed_cleanup", "committed_board", "committed_board_existing", "committed_board_existing_coalesced", "committed_board_existing_source_edited", "committed_board_existing_board_edited", "committed_board_existing_controller", "committed_board_existing_allocation", "committed_board_existing_receipt", "committed_board_existing_delivered", "committed_board_existing_source_owner", "committed_board_existing_source_delivered", "committed_unknown_cleanup", "committed_live_controller", "unreconciled", "source_edited", "source_claimed", "malformed_previous", "endpoint_paused", "principal_revoked", "still_budget_blocked"] as const)
|
||||
.map(mode => ({ provider: "github" as const, mode })),
|
||||
])("retries the newer saved $provider request instead of the failed request after a budget hold: $mode", async ({ provider, mode }) => {
|
||||
const queuedBoard = mode === "committed_board" || mode.startsWith("committed_board_existing");
|
||||
const legacyIngress = mode.startsWith("legacy_");
|
||||
const realBudgetIngress = mode === "real_budget_ingress" || legacyIngress;
|
||||
const executionRevalidation = mode.startsWith("legacy_execution");
|
||||
@@ -61983,6 +61984,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
endpointId: context.endpoint.id, provider, thread: context.thread.thread,
|
||||
message: makeMessage({ id: "second-saved-after-budget-hold", text: "@maya include the follow-up",
|
||||
userId: "U-SAFE-PROGRESS", mentioned: true }), trigger: "mention" });
|
||||
if (queuedBoard) for (let index = 1; index < 10; index++) await deliverMessage({
|
||||
callbacks: context.runtime.configurations.get(context.endpoint.id)!.callbacks,
|
||||
endpointId: context.endpoint.id, provider, thread: context.thread.thread,
|
||||
message: makeMessage({ id: `saved-followup-${index}`, text: `@maya review follow-up ${index}`,
|
||||
userId: "U-SAFE-PROGRESS", mentioned: true }), trigger: "mention" });
|
||||
const savedInputs = await db.select().from(chatActions).where(and(
|
||||
eq(chatActions.companyId, companyId), eq(chatActions.conversationId, context.conversation.id),
|
||||
eq(chatActions.kind, "inbound_wakeup"), sql`${chatActions.id} <> ${context.action.id}`))
|
||||
@@ -62008,7 +62014,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
expect(await unadmitted()).toHaveLength(1);
|
||||
// The budget was cleared. Retry must reconsider this independent saved
|
||||
// request through chat admission, not replay the older failed turn.
|
||||
const held = mode === "unreconciled" || mode === "committed_delayed_cleanup" || mode === "committed_unknown_cleanup" || mode === "committed_live_controller";
|
||||
const held = mode === "unreconciled" || mode === "committed_delayed_cleanup" || queuedBoard || mode === "committed_unknown_cleanup" || mode === "committed_live_controller";
|
||||
if (mode === "unreconciled" || mode.startsWith("committed_") || executionRevalidation) {
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: context.issue.id,
|
||||
nativePhase: "terminal_failure", runnerProfileJson: {} }).where(eq(heartbeatRuns.id, context.runId));
|
||||
@@ -62035,7 +62041,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
provider: "daytona", providerLeaseId: "owned-saved-input-sandbox" };
|
||||
await db.insert(environmentLeases).values({ ...lease, issueId: context.issue.id,
|
||||
status: "expired", cleanupStatus: "success", releasedAt: new Date(), leasePolicy: "ephemeral",
|
||||
metadata: mode === "committed_unknown_cleanup" || mode === "committed_delayed_cleanup" ? {} : {
|
||||
metadata: mode === "committed_unknown_cleanup" || mode === "committed_delayed_cleanup" || queuedBoard ? {} : {
|
||||
remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) } });
|
||||
}
|
||||
}
|
||||
@@ -62309,18 +62315,169 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
expect((await db.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, context.issue.id)))))
|
||||
.toEqual(expect.arrayContaining([expect.objectContaining({ status: "active" })]));
|
||||
if (mode === "committed_delayed_cleanup") {
|
||||
if (mode === "committed_delayed_cleanup" || queuedBoard) {
|
||||
const [boardComment] = queuedBoard ? await db.insert(issueComments).values({
|
||||
companyId, issueId: context.issue.id, authorType: "user", authorUserId: "owner-user",
|
||||
body: "Keep this independent continuation for its own authorized turn.",
|
||||
}).returning() : [];
|
||||
const [boardWake] = boardComment ? await db.insert(agentWakeupRequests).values({ companyId,
|
||||
agentId: context.fixture.assignedAgentId, source: "automation", triggerDetail: "system",
|
||||
reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "user", requestedByActorId: "owner-user",
|
||||
payload: { issueId: context.issue.id, commentId: boardComment.id, wakeCommentIds: [boardComment.id],
|
||||
_paperclipWakeContext: { issueId: context.issue.id, taskKey: context.issue.id,
|
||||
source: "issue.comment.reopen", wakeReason: "issue_commented",
|
||||
wakeCommentId: boardComment.id, wakeCommentIds: [boardComment.id] } },
|
||||
}).returning() : [];
|
||||
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, context.runId));
|
||||
await db.update(environmentLeases).set({ metadata: { ...lease.metadata,
|
||||
remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) },
|
||||
}).where(eq(environmentLeases.id, lease.id));
|
||||
await reconcileTaskChatRetryReceipts(db);
|
||||
await context.service.processFailedChatRunRetry(receipt.chatActionId!);
|
||||
const [resumed] = await readAgentWakeupRequestsById(receipt.chatActionId!);
|
||||
let [resumed] = await readAgentWakeupRequestsById(receipt.chatActionId!);
|
||||
expect(resumed).toMatchObject({ id: admitted.id, status: "queued" });
|
||||
expect(resumed.runId).toBeTruthy();
|
||||
expect((await readHeartbeatRunsById(resumed.runId!))[0]).toMatchObject({ status: "queued",
|
||||
contextSnapshot: { wakeCommentId: saved.payload.commentId } });
|
||||
if (boardWake) {
|
||||
const heartbeat = heartbeatService(db);
|
||||
if (mode.startsWith("committed_board_existing")) {
|
||||
// Persist the historical producer's merged batch, then let real
|
||||
// dispatch reject it before any identity or provider starts.
|
||||
const [queued] = await readHeartbeatRunsById(resumed.runId!);
|
||||
if (mode === "committed_board_existing_coalesced") {
|
||||
await db.update(chatActions).set({ status: "processed" }).where(inArray(chatActions.id, savedInputs.map(source => source.id)));
|
||||
await db.update(agentWakeupRequests).set({ status: "coalesced", runId: queued.id,
|
||||
payload: sql`${agentWakeupRequests.payload} || ${JSON.stringify({ coalescedIntoWakeupRequestId: resumed.id })}::jsonb`,
|
||||
}).where(inArray(agentWakeupRequests.id, savedInputs.map(source => source.id)));
|
||||
}
|
||||
const mixedIds = [boardComment!.id, ...expectedComments];
|
||||
// Replaying this test with the queue fix reverted uses the old
|
||||
// producer's actual persisted merge. With the fixed producer,
|
||||
// retain historical state using its shared serializers.
|
||||
if (!(queued.contextSnapshot?.wakeCommentIds as string[])?.includes(boardComment!.id)) {
|
||||
const { withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload } =
|
||||
await import("../services/issue-queued-comment-queue.js");
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot:
|
||||
withQueuedCommentIdsInRunContext(queued.contextSnapshot, mixedIds, [boardWake.id]) })
|
||||
.where(eq(heartbeatRuns.id, queued.id));
|
||||
await db.update(agentWakeupRequests).set({ payload:
|
||||
withQueuedCommentIdsInWakePayload(resumed.payload, mixedIds) })
|
||||
.where(eq(agentWakeupRequests.id, resumed.id));
|
||||
await db.update(agentWakeupRequests).set({ status: "coalesced", runId: queued.id,
|
||||
finishedAt: new Date() }).where(eq(agentWakeupRequests.id, boardWake.id));
|
||||
}
|
||||
expect((await readHeartbeatRunsById(queued.id))[0].contextSnapshot).toMatchObject({
|
||||
wakeCommentId: expectedComments.at(-1), wakeCommentIds: mixedIds,
|
||||
queuedCommentSourceWakeIds: [boardWake.id],
|
||||
});
|
||||
expect((await readAgentWakeupRequestsById(boardWake.id))[0]).toMatchObject({
|
||||
status: "coalesced", runId: queued.id, payload: { issueId: context.issue.id },
|
||||
});
|
||||
expect((await readAgentWakeupRequestsById(boardWake.id))[0].payload)
|
||||
.not.toHaveProperty("coalescedIntoWakeupRequestId");
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"), isNull(heartbeatRuns.wakeupRequestId)));
|
||||
await db.update(agents).set({ status: "idle" }).where(eq(agents.id, context.fixture.assignedAgentId));
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [failed] = await readHeartbeatRunsById(queued.id);
|
||||
expect(failed).toMatchObject({ status: "failed", errorCode: "chat_failed_run_retry_not_authorized",
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } });
|
||||
await db.insert(heartbeatRuns).values({ companyId, agentId: context.fixture.assignedAgentId,
|
||||
status: "running", startedAt: new Date() });
|
||||
await db.update(agents).set({ status: "running" }).where(eq(agents.id, context.fixture.assignedAgentId));
|
||||
const nextInput = { ...request, observedRunId: failed.id, idempotencyKey: randomUUID() };
|
||||
if (mode.endsWith("source_edited")) await db.update(issueComments).set({ body: "Changed external source" })
|
||||
.where(eq(issueComments.id, String(saved.payload.commentId)));
|
||||
if (mode.endsWith("board_edited")) await db.update(issueComments).set({ body: "Changed board source",
|
||||
updatedAt: new Date(boardComment!.createdAt.getTime() + 1) })
|
||||
.where(eq(issueComments.id, boardComment!.id));
|
||||
if (mode.endsWith("controller")) await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(heartbeatRuns.id, failed.id));
|
||||
if (mode.endsWith("allocation")) await db.insert(environmentLeases).values({ companyId,
|
||||
heartbeatRunId: failed.id, issueId: context.issue.id, provider: "daytona", providerLeaseId: "unknown-executor",
|
||||
status: "active", leasePolicy: "ephemeral" });
|
||||
if (mode.endsWith("receipt")) await db.update(agentWakeupRequests).set({ status: "claimed" })
|
||||
.where(eq(agentWakeupRequests.id, boardWake.id));
|
||||
if (mode.endsWith("source_owner")) await db.update(agentWakeupRequests).set({
|
||||
status: "coalesced", runId: context.runId,
|
||||
payload: sql`${agentWakeupRequests.payload} || ${JSON.stringify({ coalescedIntoWakeupRequestId: resumed.id })}::jsonb`,
|
||||
}).where(eq(agentWakeupRequests.id, saved.id));
|
||||
if (mode.endsWith("delivered")) await db.insert(heartbeatRuns).values({ companyId,
|
||||
agentId: context.fixture.assignedAgentId, status: "succeeded", finishedAt: new Date(),
|
||||
createdAt: new Date(failed.createdAt.getTime() - 1),
|
||||
contextSnapshot: { issueId: context.issue.id, wakeCommentIds:
|
||||
[mode.endsWith("source_delivered") ? saved.payload.commentId : boardComment!.id] },
|
||||
});
|
||||
const inputIds = [boardWake.id, ...savedInputs.map(source => source.id)];
|
||||
const before = await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)).orderBy(asc(agentWakeupRequests.id));
|
||||
if (mode !== "committed_board_existing" && mode !== "committed_board_existing_coalesced") {
|
||||
await expect(requestTaskRetry(nextInput)).rejects.toMatchObject({
|
||||
details: { code: "chat_failed_run_retry_not_authorized" },
|
||||
});
|
||||
expect(await db.select().from(agentWakeupRequests).where(inArray(agentWakeupRequests.id, inputIds)).orderBy(asc(agentWakeupRequests.id)))
|
||||
.toEqual(before);
|
||||
return;
|
||||
}
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
app.use((req, _res, next) => { req.actor = boardActor(companyId); next(); });
|
||||
app.use("/api", issueRoutes(db, createStorageService().storage, { chatRunRetries: context.service }));
|
||||
app.use(errorHandler);
|
||||
const httpRequest = (await import("supertest")).default;
|
||||
const retryHttp = () => httpRequest(app).post(`/api/issues/${context.issue.id}/retry`).send({
|
||||
idempotencyKey: nextInput.idempotencyKey, observedRunId: nextInput.observedRunId,
|
||||
}).expect(202);
|
||||
const next = (await retryHttp()).body;
|
||||
expect(next).toMatchObject({ accepted: true });
|
||||
expect((await retryHttp()).body.requestId).toBe(next.requestId);
|
||||
expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId),
|
||||
eq(activityLog.runId, failed.id), eq(activityLog.action, "issue.retry_inputs_restored")))).toHaveLength(1);
|
||||
expect(await context.service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" });
|
||||
[resumed] = await readAgentWakeupRequestsById(next.chatActionId!);
|
||||
expect(resumed.runId).toBeTruthy();
|
||||
if (mode === "committed_board_existing_coalesced") {
|
||||
const restored = await db.select().from(agentWakeupRequests)
|
||||
.where(inArray(agentWakeupRequests.id, savedInputs.map(source => source.id)));
|
||||
expect(restored).toHaveLength(10);
|
||||
expect(restored.every(wake => wake.status === "coalesced" && wake.runId === resumed.runId)).toBe(true);
|
||||
}
|
||||
expect((await readHeartbeatRunsById(failed.id))[0].contextSnapshot).toEqual(failed.contextSnapshot);
|
||||
}
|
||||
const realInitialize = runIdentityModule.initializeRunIdentity;
|
||||
const initialized: Array<{ runId: string; messageId: string | null }> = [];
|
||||
expect((await readAgentWakeupRequestsById(boardWake.id))[0])
|
||||
.toMatchObject({ status: "deferred_issue_execution", runId: null });
|
||||
const probe = vi.spyOn(runIdentityModule, "initializeRunIdentity").mockImplementation(async (database, input) => {
|
||||
const identity = await realInitialize(database, input);
|
||||
if (input.companyId !== companyId) return identity;
|
||||
initialized.push({ runId: input.runId, messageId: identity.messageId });
|
||||
throw new Error("certified_batch_identity_probe_complete");
|
||||
});
|
||||
try {
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"), isNull(heartbeatRuns.wakeupRequestId)));
|
||||
await db.update(agents).set({ status: "idle" }).where(eq(agents.id, context.fixture.assignedAgentId));
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [successor] = await readHeartbeatRunsById(resumed.runId!);
|
||||
expect(successor).toMatchObject({ error: "certified_batch_identity_probe_complete",
|
||||
contextSnapshot: { wakeCommentIds: expectedComments } });
|
||||
expect(successor.contextSnapshot?.queuedCommentSourceWakeIds).toBeUndefined();
|
||||
const [boardReceipt] = await readAgentWakeupRequestsById(boardWake.id);
|
||||
expect(boardReceipt.runId).toBeTruthy();
|
||||
expect(boardReceipt.runId).not.toBe(resumed.runId);
|
||||
expect(initialized).toEqual([
|
||||
{ runId: resumed.runId, messageId: saved.payload.commentId },
|
||||
{ runId: boardReceipt.runId, messageId: boardComment!.id },
|
||||
]);
|
||||
expect((await readHeartbeatRunsById(boardReceipt.runId!))[0]).toMatchObject({
|
||||
contextSnapshot: { wakeCommentIds: [boardComment!.id] },
|
||||
});
|
||||
} finally { probe.mockRestore(); await heartbeat.drainActiveRunExecutions(); }
|
||||
}
|
||||
await context.service.processFailedChatRunRetry(receipt.chatActionId!);
|
||||
await reconcileTaskChatRetryReceipts(db);
|
||||
expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run =>
|
||||
@@ -62354,7 +62511,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
}
|
||||
}
|
||||
expect((await readAgentWakeupRequestsById(saved.id))[0]).toMatchObject(mode === "source_claimed"
|
||||
? { status: "queued", runId: context.runId } : { status: "skipped", runId: null });
|
||||
? { status: "queued", runId: context.runId } : mode === "committed_board_existing_coalesced"
|
||||
? { status: "coalesced", runId: expect.any(String) } : { status: "skipped", runId: null });
|
||||
expect((await readHeartbeatRunsById(context.runId))[0]).toMatchObject({ status: "failed" });
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(previousPublications);
|
||||
} finally { await context.service.shutdown(); }
|
||||
|
||||
@@ -7,6 +7,7 @@ import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-no
|
||||
import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js";
|
||||
import { readChatAuthorizationRun } from "./authorization-run-read.js";
|
||||
import { selectTaskRetryInput, claimTaskRetryInput, settleTaskRetryInput, releaseUnadmittedTaskRetryInput, taskRetryInputSchema, type TaskRetryInput } from "./task-retry-input.js";
|
||||
import { queuedCommentIdsFromWakePayload } from "../issue-queued-comment-queue.js";
|
||||
import type { TaskRetryWakeAuthority } from "./github/task-retry-authorization.js";
|
||||
|
||||
import {
|
||||
@@ -76,6 +77,7 @@ import {
|
||||
|
||||
import {
|
||||
logActivity,
|
||||
persistActivity,
|
||||
} from "../activity-log.js";
|
||||
import {
|
||||
type IssueAssignmentWakeupDeps,
|
||||
@@ -1361,6 +1363,82 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
|
||||
}
|
||||
});
|
||||
|
||||
/** A historical queue merge can consume independent board input before chat
|
||||
* dispatch rejects the changed batch. A new board Retry may restore those
|
||||
* never-delivered receipts, after certifying both their original authorities.
|
||||
* The failed run and its canonical Retry remain unchanged history. */
|
||||
async function restoreUnstartedRetryInput(tx: DbOrTransaction, input: {
|
||||
companyId: string; issueId: string; agentId: string; failedRunId: string; initiatedByUserId: string;
|
||||
}) {
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId),
|
||||
eq(heartbeatRuns.id, input.failedRunId),
|
||||
)).for("share", { noWait: true });
|
||||
const queueIds = run?.contextSnapshot?.queuedCommentSourceWakeIds;
|
||||
if (run?.errorCode !== "chat_failed_run_retry_not_authorized" || !Array.isArray(queueIds) || !queueIds.length) return;
|
||||
if (queueIds.length > 50 || queueIds.some(id => typeof id !== "string" || !isUuidLike(id)) ||
|
||||
new Set(queueIds).size !== queueIds.length || !await isPreProviderChatFailure(tx, run)) throw failedChatRetryDenied();
|
||||
const source = await authorizeFailedChatRunRetry(tx, { ...input, retryActionId: run.wakeupRequestId!, runId: run.id });
|
||||
const sourceIds = source.taskRetryInput?.sourceActionIds;
|
||||
if (source.provider !== "github" || !sourceIds?.length || !source.intent.payload.taskRetryIntentId)
|
||||
throw failedChatRetryDenied();
|
||||
const queues = await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, input.agentId),
|
||||
inArray(agentWakeupRequests.id, queueIds as string[]),
|
||||
)).for("update");
|
||||
const byId = new Map(queues.map(queue => [queue.id, queue]));
|
||||
if (queues.length !== queueIds.length) throw failedChatRetryDenied();
|
||||
const boardIds = (queueIds as string[]).flatMap(id => {
|
||||
const queue = byId.get(id)!;
|
||||
const ids = queuedCommentIdsFromWakePayload(queue.payload);
|
||||
if (queue.status !== "coalesced" || queue.runId !== run.id || queue.payload?.issueId !== input.issueId ||
|
||||
queue.requestedByActorType !== "user" || !queue.requestedByActorId ||
|
||||
!["issue_commented", "issue_reopened_via_comment"].includes(queue.reason ?? "") ||
|
||||
queue.idempotencyKey?.startsWith("chat-inbound:") || queue.payload?.queuedCommentInterrupt ||
|
||||
queue.payload?.coalescedIntoWakeupRequestId || !ids.length || ids.some(id => !isUuidLike(id)))
|
||||
throw failedChatRetryDenied();
|
||||
return ids;
|
||||
});
|
||||
if (new Set(boardIds).size !== boardIds.length || boardIds.some(id => source.commentIds.includes(id)) ||
|
||||
JSON.stringify(run.contextSnapshot?.wakeCommentIds) !== JSON.stringify([...boardIds, ...source.commentIds]) ||
|
||||
run.contextSnapshot?.wakeCommentId !== source.commentIds.at(-1)) throw failedChatRetryDenied();
|
||||
const comments = await tx.select().from(issueComments).where(and(
|
||||
eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId),
|
||||
inArray(issueComments.id, boardIds),
|
||||
)).for("share", { noWait: true });
|
||||
const commentsById = new Map(comments.map(comment => [comment.id, comment]));
|
||||
for (const queue of queues) for (const id of queuedCommentIdsFromWakePayload(queue.payload)) {
|
||||
const comment = commentsById.get(id);
|
||||
if (!comment || comment.authorType !== "user" || comment.authorUserId !== queue.requestedByActorId ||
|
||||
comment.createdByRunId || comment.deletedAt || !comment.body.trim() ||
|
||||
comment.updatedAt.getTime() !== comment.createdAt.getTime()) throw failedChatRetryDenied();
|
||||
}
|
||||
const [delivered] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), ne(heartbeatRuns.id, run.id),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${input.issueId}`,
|
||||
sql`not (${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.startedAt} is null)`,
|
||||
// Budget-skipped source receipts remain immutable even after a later
|
||||
// Retry delivers them. Check run history for both input authorities.
|
||||
or(...[...boardIds, ...source.commentIds].map(id => or(sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`,
|
||||
sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`))),
|
||||
)).limit(1);
|
||||
if (delivered) throw failedChatRetryDenied();
|
||||
// Source authorization above checked every external receipt and its owner.
|
||||
// Reset only this certified unstarted batch under the task's admission lock.
|
||||
const skippedIds = source.taskRetryInput?.skippedActionIds ?? [];
|
||||
await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", runId: null, finishedAt: null,
|
||||
payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(),
|
||||
}).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, input.agentId),
|
||||
inArray(agentWakeupRequests.id, [...sourceIds, ...queueIds as string[]]),
|
||||
ne(agentWakeupRequests.status, "skipped")));
|
||||
await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "user", actorId: input.initiatedByUserId,
|
||||
action: "issue.retry_inputs_restored", entityType: "issue", entityId: input.issueId, runId: run.id,
|
||||
details: { sourceRetryActionId: source.intent.id, externalRequestIds: sourceIds, boardRequestIds: queueIds } });
|
||||
// The old skipped receipts stay unchanged. The certified failed attempt
|
||||
// proves this batch was never delivered even though it predates that run.
|
||||
return { sourceActionIds: sourceIds, skippedActionIds: skippedIds } satisfies TaskRetryInput;
|
||||
}
|
||||
|
||||
async function prepareFailedChatRunRetry(
|
||||
tx: DbOrTransaction,
|
||||
input: {
|
||||
@@ -1372,7 +1450,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
|
||||
retryIntentId?: string;
|
||||
},
|
||||
): Promise<{ actionId: string; issueId: string }> {
|
||||
const taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(tx, input) : undefined;
|
||||
let taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(tx, input) : undefined;
|
||||
if (input.retryIntentId && taskRetryInput && !taskRetryInput.sourceActionIds.length)
|
||||
taskRetryInput = await restoreUnstartedRetryInput(tx, input) ?? taskRetryInput;
|
||||
const source = await failedChatRetrySource(tx, { ...input, taskRetryIntentId: input.retryIntentId, taskRetryInput });
|
||||
const providerActionId = input.retryIntentId
|
||||
? `task-retry:${input.retryIntentId}` : `failed-run-retry:${source.failedRunId}`;
|
||||
|
||||
@@ -4247,7 +4247,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
inArray(agentWakeupRequests.agentId, handoffSource ? [agentId, handoffSource.agentId] : [agentId]),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`,
|
||||
canCoalesceComments || opts.taskRetryRequestId ? undefined : eq(agentWakeupRequests.id, executionWaitRequestId!),
|
||||
// Chat Retry already pins its complete authorized source
|
||||
// batch. Keep separate board input for its own admission.
|
||||
canCoalesceComments || (opts.taskRetryRequestId && !failedChatRetry)
|
||||
? undefined : eq(agentWakeupRequests.id, executionWaitRequestId!),
|
||||
),
|
||||
)
|
||||
.orderBy(asc(agentWakeupRequests.requestedAt))
|
||||
|
||||
Reference in new issue
Block a user