fix(chat): admit task Retry after GitHub reconnect (#17733)

## Thinking Path

> - Paperclip manages agents and their tasks.
> - Task Retry must preserve saved input and enforce current access.
> - A GitHub reconnect invalidates old runtime callbacks.
> - Retry reused that old callback authority and rejected valid new
intent.
> - Retry also replayed an older failed request when a follow-up was
already saved.
> - This change admits a deliberate Retry under current authority and
transfers saved input once.

## Linked Issues or Issue Description

Refs #16965, #17053 and #17317. These repairs preserve work and contain
old executors. This change fixes the subsequent chat admission boundary.
A search found no overlapping open reconnect/admission fix.

**What happened?**

Ordinary task Retry returned `chat_failed_run_retry_not_authorized`
after the same GitHub App reconnected. The endpoint was healthy, but the
saved request still carried its old runtime generation. A newer queued
follow-up also made Retry reject the older failed request instead of
continuing the follow-up. No successor was admitted.

HTTP qualification found two related defects. Concurrent clicks could
omit the canonical chat action from a response. Signed-in board dispatch
also passed an incomplete task resource to the permission resolver,
which rejected a permitted mutation.

**Expected behavior**

A deliberate Retry checks current authority for the same App, sender,
repository, conversation and task. It continues saved input once. A
later reconnect, changed App, revoked access or changed source prevents
dispatch. Old runtime callbacks remain invalid.

**Steps to reproduce**

1. Admit a GitHub request and record a failed native run.
2. Save a follow-up while execution is blocked.
3. Reconnect the same GitHub App.
4. Click ordinary task Retry.
5. Before this change, admission returns 409. Without reconnect, the
newer input alone also causes 409.

**Paperclip version or commit**

Baseline source: `41c42895fa862c586ff841f8c50200d93ffc83a4`.

**Deployment mode**

Source checkout on macOS. Regressions use real PostgreSQL transactions
and the assembled chat, HTTP route and heartbeat admission services.
Provider boundaries are simulated. No hosted recovery is claimed.

## What Changed

- Record the admitting App identity for new GitHub deliveries.
Historical deliveries can use their original working-comment binding.
Reauthorize deliberate Retry under a pinned current generation and
credential-reference fingerprint.
- Select the complete saved-input queue from the failed run's actual
conversation before replaying the older failed batch. Keep exact-run
Retry bounded while ordinary task Retry supports larger queues. Transfer
its receipts with the dispatch transaction. Recover the same ownership
after restart or a lost response. Release unadmitted input after a
definitive rejection.
- Preserve canonical chat action IDs across concurrent clicks. Supply
current task status and assignment fields to board mutation checks.
- Keep exact-run Retry, stale webhook, containment, source integrity,
publication, pause and budget checks. Update recovery contracts and
verification recipes.

## Verification

- Red: all three new PostgreSQL regressions failed on unchanged
production code. Reconnect failed at chat authorization. Newer input
failed at the obsolete-source guard. The combined case also failed.
- Green: 61 selected chat integration cases passed. These include 27 new
cases and 34 existing cases for containment, retained-source recovery,
exact batches and source revocation.
- A further red/green PostgreSQL regression reproduces the 51-message
overflow identified during review. All 51 messages reach one successor,
including a later Retry of that successor. A separate 52-message
consumed-batch regression also passes.
- A competing conversation retains its own deferred message. Distinct
automatic policy contexts are checked independently; revoking only the
later policy blocks the whole unadmitted batch. Batch validation still
rejects a later message with a changed App binding or deleted source. A
real PostgreSQL query capture reduced the 51-message preparation from
1,103 calls to fewer than 180.
- All 16 durable task Retry tests passed.
- The real HTTP route returns 202 and admits one queued successor for
concurrent deliveries. Fault injection covers rollback during input
handoff, response loss after dispatch commit, and permission loss when
resuming an existing durable wait. A wait only adopts input after
current admission succeeds; a one-shot settlement marker keeps status
projection read-only. A new deliberate Retry works after another
reconnect or a failed successor.
- Current-App, repository, sender, source edit/delete, reassignment,
endpoint pause and board-permission controls passed. Historical
deliveries remain unchanged.
- Final-head CI: 55 checks passed, three expected skips. The unchanged
minute-bucket rate-limit assertion passed locally against PostgreSQL and
on its one CI rerun. Greptile reviewed the final commit at 5/5 with no
unresolved threads.
- Workspace typecheck and full build passed. Map alignment, module
boundaries, UI token gates and `git diff --check` passed.
- Hosted task recovery and a live provider turn remain unverified. The
local tests hold agent capacity to inspect admission without starting a
model or publishing a GitHub response.

## Risks

- A deliberate GitHub Retry can admit old saved input under current
connection authority. It requires the same App binding and current
access. It pins that admission; it does not rewrite the historical
delivery or make stale callbacks valid.
- The durable chat queue has an optional admission callback. It
transfers input only when the transaction contains a valid queued
execution or saved wait. Other chat consumers have no callback and
retain their behavior.
- No migration, Cloud dependency, deployment, configuration change or
workspace cleanup is included.

## Model Used

OpenAI GPT-6 through Codex, with reasoning, source inspection, code
execution and GitHub tools. The exact serving variant and context-window
size are not exposed in this session.

## 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
- [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:
DottaandPaperclip authored and GitHub committed 2026-10-26 10:26:45 -05:00
1 parent ad861a7f94
commit 338bf75761
14 files changed
+808 -122

No files matched your search

+20 -2
View File
@@ -15,8 +15,26 @@ new generation; a repeated HTTP request cannot.
The existing exact-run retry endpoint retains its source-bound contract.
Connected conversations use their existing inbound authorization owner. A board
Retry changes the workspace strategy, not the external principal, source batch,
conversation access or publication authority.
Retry preserves external principal, conversation access and publication checks.
A GitHub Retry selects already-saved, undelivered input before replaying the
failed run's older request. The selected receipts transfer to the retry owner
under the dispatch transaction. They link to its one successor, including after
a restart or a lost dispatch response. Rejected, unadmitted retries release the
saved input for another deliberate Retry.
A deliberate GitHub Retry can admit saved input after the same App reconnects.
It verifies the App binding in the accepted delivery or its historical working
comment receipt, then checks current sender, repository, conversation and task
access. The retry records the current runtime generation and credential-reference
fingerprint. Another reconnect invalidates that admission. Historical deliveries,
stale webhook callbacks and the exact-run retry contract retain their fences.
Task Retry keeps the complete saved-input batch, including batches larger than
the exact-run endpoint's limit. Source rows, receipts and invalidations are read
in batches. Within one transaction, messages with the same principal and automatic
policy context share a current-access check; every message still proves its App,
destination, comment link and immutable source. Different automatic policy contexts
receive separate access checks. Input in another conversation remains queued.
A connected-conversation Retry can save its exact request while the matching
system recovery generation verifies the old executor. Saving that wait grants
+7
View File
@@ -4,6 +4,13 @@ Product map: [Recovery](../feature-map/recovery.md).
Feature ID: `recovery`.
Ordinary board Retry can reauthorize saved GitHub input for the same App after
reconnect. It pins current runtime authority and preserves sender, repository,
conversation and task checks. Pending input takes precedence over the older
failed request and transfers once to the retry owner. This grants no new agent
Retry API or stale webhook authority. Local PostgreSQL and HTTP-route regressions
cover these transitions; hosted execution and delivery remain separate acceptance.
Source revision: `3bbc2104b65caf9f7763938a01260a5673e12f7a`.
Review depth and executed proof are separate. The operation records enumerate the named behavior and access boundaries. Their evidence states describe the reviewed sources; run results and environment limits belong in the PR or verification artifacts. A listed API or source test does not establish a live authorized journey. Follow the [map contract](../README.md).
+8
View File
@@ -89,6 +89,14 @@ The connected-chat database regression in
exercises admission and queue continuation. It does not establish hosted
provider recovery or successful delivery.
For GitHub, save a follow-up behind a failed run, then reconnect the same App.
Click ordinary Retry and verify that one successor receives the saved follow-up
instead of replaying the older request. Reload and repeat the same request key;
there must be one successor and no duplicate publication. A second reconnect
after admission, changed App, revoked sender or changed repository must prevent
dispatch. An unadmitted rejection keeps the follow-up available for another
deliberate Retry. Historical deliveries must remain unchanged.
### `post-stop-message`
After a stop, send a new instruction on the same task or inspect one already
@@ -131,6 +131,7 @@ import { deliverReconciledExecutions } from "../services/execution-recovery-reso
import {
unadmittedChatWakeupCondition,
authorizeFailedChatRunRetryWake,
createDurableChatWakeupRequest,
} from "../services/durable-chat-wakeup.js";
import {
createChatWebhookDiagnostics,
@@ -60661,6 +60662,358 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
return { ...context, issue, action, receipt, runId };
}
async function currentGitHubTaskRetryFixture(withPending = false, pendingCount = 1) {
const f = await failedChatRetryFixture("github", "42", true);
const companyId = f.fixture.companyId, issueId = f.issue.id;
await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(f.action.deliveryId!);
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId,
nativePhase: "terminal_failure", runnerProfileJson: {} }).where(eq(heartbeatRuns.id, f.runId));
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: f.runId, phase: "terminal_failure" });
await db.insert(heartbeatRunEvents).values({ companyId, agentId: f.fixture.assignedAgentId, runId: f.runId,
seq: 1, eventType: "native.local_process_stopped", payload: { processPid: 999999999, processGroupId: null } });
let pending: typeof chatActions.$inferSelect | undefined;
if (withPending) {
for (let index = 0; index < pendingCount; index++) {
await deliverMessage({ callbacks: f.runtime.configurations.get(f.endpoint.id)!.callbacks,
endpointId: f.endpoint.id, provider: "github", thread: f.thread.thread,
message: makeMessage({ id: String(990092 + index), text: `@maya continue saved follow-up ${index}`, userId: "42", mentioned: true }),
trigger: "mention" });
const [{ source }] = await db.select({ source: chatActions }).from(chatActions)
.innerJoin(chatDeliveries, eq(chatDeliveries.id, chatActions.deliveryId)).where(and(
eq(chatActions.conversationId, f.conversation.id), eq(chatActions.kind, "inbound_wakeup"),
eq(chatDeliveries.providerEventId, `${f.thread.thread.id}:${990092 + index}`)));
pending ??= source;
await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution", runId: null })
.where(eq(agentWakeupRequests.id, source.id));
await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(source.deliveryId!);
}
}
const [originalDelivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, f.action.deliveryId!));
await db.update(agents).set({ adapterType: "paperclip_runner", status: "running",
runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } })
.where(eq(agents.id, f.fixture.assignedAgentId));
await db.insert(heartbeatRuns).values({ companyId, agentId: f.fixture.assignedAgentId,
status: "running", startedAt: new Date() });
f.wakeup.mockImplementation(heartbeatService(db).wakeup);
return { ...f, pending, originalDelivery };
}
it.each(["reconnected", "newer_input", "reconnected_newer_input"] as const)(
"admits current GitHub task Retry without replaying obsolete input: %s", async kind => {
const f = await currentGitHubTaskRetryFixture(kind.includes("newer_input"));
const { pending, originalDelivery } = f;
try {
const companyId = f.fixture.companyId, issueId = f.issue.id;
if (kind.includes("reconnected")) await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user");
const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant",
observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry };
const request = await requestTaskRetry(input);
expect(request.accepted).toBe(true);
expect((await requestTaskRetry(input)).requestId).toBe(request.requestId);
expect(await f.service.processFailedChatRunRetry(request.chatActionId!)).toMatchObject({ status: "queued" });
const [wake] = await readAgentWakeupRequestsById(request.chatActionId!);
expect(wake.runId).toBeTruthy();
const expectedComment = pending?.payload.commentId ?? f.action.payload.commentId;
const [run] = await readHeartbeatRunsById(wake.runId!);
expect(run).toMatchObject({ status: "queued", retryOfRunId: f.runId,
contextSnapshot: { wakeCommentIds: [expectedComment], taskRetryRequestId: request.requestId } });
if (pending) expect((await readAgentWakeupRequestsById(pending.id))[0])
.toMatchObject({ status: "coalesced", runId: wake.runId });
await f.service.processFailedChatRunRetry(request.chatActionId!);
await heartbeatService(db).resumeQueuedRuns();
expect((await readHeartbeatRunsByCompanyId(companyId)).filter(r => r.contextSnapshot?.taskRetryRequestId === request.requestId))
.toHaveLength(1);
expect((await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, originalDelivery.id)))[0].normalizedEvent)
.toEqual(originalDelivery.normalizedEvent);
} finally { await f.service.shutdown(); }
},
);
it.each(["legacy_receipt", "concurrent_clicks", "lost_dispatch_reply", "restart", "retry_failed_successor", "http", "handoff_rollback", "board_permission_wait", "peer_conversation", "backlog_51", "consumed_backlog_52"] as const)(
"preserves one current GitHub Retry across %s", async mode => {
const f = await currentGitHubTaskRetryFixture(true, mode.includes("backlog") ? 51 : mode === "peer_conversation" ? 2 : 1);
try {
const companyId = f.fixture.companyId, issueId = f.issue.id;
if (mode === "consumed_backlog_52") {
const sources = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id),
eq(chatActions.kind, "inbound_wakeup"))).orderBy(asc(chatActions.createdAt), asc(chatActions.id));
expect(sources).toHaveLength(52);
const commentIds = sources.map(source => source.payload.commentId);
await db.update(heartbeatRuns).set({ finishedAt: new Date(), contextSnapshot: {
...(await readHeartbeatRunsById(f.runId))[0].contextSnapshot,
wakeCommentIds: commentIds, wakeCommentId: commentIds.at(-1),
} }).where(eq(heartbeatRuns.id, f.runId));
await db.update(agentWakeupRequests).set({ runId: f.runId, status: "coalesced",
payload: sql`${agentWakeupRequests.payload} || ${JSON.stringify({ coalescedIntoWakeupRequestId: f.action.id })}::jsonb`,
}).where(and(inArray(agentWakeupRequests.id, sources.map(source => source.id)),
sql`${agentWakeupRequests.id} <> ${f.action.id}::uuid`));
}
let peerInput: typeof chatActions.$inferSelect | undefined;
if (mode === "peer_conversation") {
const originId = randomUUID();
await db.insert(chatConversations).values({ ...f.conversation, id: originId, sessionGeneration: 2 });
await db.update(chatActions).set({ conversationId: originId,
payload: sql`jsonb_set(${chatActions.payload}, '{sessionGeneration}', '2'::jsonb)`,
}).where(inArray(chatActions.id, [f.action.id, f.pending!.id]));
await db.update(chatDeliveries).set({ conversationId: originId })
.where(inArray(chatDeliveries.id, [f.action.deliveryId!, f.pending!.deliveryId!]));
await db.update(chatMessageLinks).set({ conversationId: originId })
.where(inArray(chatMessageLinks.deliveryId, [f.action.deliveryId!, f.pending!.deliveryId!]));
[peerInput] = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id),
eq(chatActions.kind, "inbound_wakeup")));
expect(peerInput).toBeDefined();
}
if (mode === "legacy_receipt") {
const [originalEndpoint] = await readChatEndpointsById(f.endpoint.id);
await db.insert(chatActions).values({ companyId, endpointId: f.endpoint.id,
conversationId: f.conversation.id, deliveryId: f.pending!.deliveryId,
kind: "github_response_comment", providerActionId: `github_response:${f.pending!.deliveryId}`,
payload: { botExternalId: originalEndpoint.botExternalId, agentId: f.fixture.assignedAgentId },
status: "processed" }).onConflictDoNothing();
await db.update(chatDeliveries).set({
normalizedEvent: sql`jsonb_set(${chatDeliveries.normalizedEvent}, '{githubAuthority}',
(${chatDeliveries.normalizedEvent}->'githubAuthority') - 'botExternalId')`,
}).where(eq(chatDeliveries.id, f.pending!.deliveryId!));
}
await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user");
const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant",
observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry };
if (mode === "board_permission_wait") {
input.actorId = `retry-operator-${companyId}`;
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: input.actorId,
status: "active", membershipRole: "operator" });
}
const queryEvents: Record<string, unknown>[] = [];
const capture = mode === "backlog_51" ? createDatabaseResponseCapture({ start: Date.now() - 1,
end: Date.now() + 59_000 }, event => queryEvents.push(event)) : undefined;
if (capture) installDatabaseResponseCapture(db, { chatActions, chatDeliveries, chatMessageLinks, issueComments,
agentWakeupRequests, chatEndpoints, chatConversations }, capture);
let one: Awaited<ReturnType<typeof requestTaskRetry>>, two: typeof one;
if (mode === "http") {
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: f.service }));
app.use(errorHandler);
const send = () => request(app).post(`/api/issues/${issueId}/retry`).send({
idempotencyKey: input.idempotencyKey, observedRunId: input.observedRunId,
}).expect(202);
const replies = await Promise.all([send(), send()]);
[one, two] = replies.map(response => response.body);
} else [one, two] = await Promise.all([requestTaskRetry(input), requestTaskRetry({ ...input,
idempotencyKey: mode === "concurrent_clicks" ? randomUUID() : input.idempotencyKey })]);
if (capture) {
capture.flush();
expect(Number(queryEvents.find(event => event.event === "paperclip.db_response_capture.summary.v1")?.startedQueries))
.toBeLessThan(180);
}
expect(two.requestId).toBe(one.requestId);
expect(one).toMatchObject({ accepted: true, chatActionId: expect.any(String) });
let service = f.service;
if (mode === "restart") {
await service.shutdown();
service = createService(f.runtime, f.providerFetch, { wakeup: heartbeatService(db).wakeup }).service;
}
if (mode === "lost_dispatch_reply") f.wakeup.mockImplementationOnce(async (...args) => {
await heartbeatService(db).wakeup(...args);
throw new Error("Simulated response loss after dispatch commit");
});
if (mode === "handoff_rollback") f.wakeup.mockImplementationOnce(async (agentId, options) => {
const original = options!.durableChatRequest!;
return heartbeatService(db).wakeup(agentId, { ...options, durableChatRequest: createDurableChatWakeupRequest({
...original, onAdmitted: async (tx, receipt) => {
await original.onAdmitted!(tx, receipt);
throw new Error("Simulated interruption before admission commit");
},
}) });
});
if (mode === "board_permission_wait") await db.update(companyMemberships).set({ membershipRole: "viewer" })
.where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, input.actorId)));
await Promise.all([service.processFailedChatRunRetry(one.chatActionId!),
service.processFailedChatRunRetry(one.chatActionId!)]);
if (mode === "handoff_rollback" || mode === "board_permission_wait") {
expect(await readAgentWakeupRequestsById(one.chatActionId!)).toHaveLength(0);
expect((await readAgentWakeupRequestsById(f.pending!.id))[0])
.toMatchObject({ status: "deferred_issue_execution", runId: null });
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === one.requestId)).toHaveLength(0);
}
if (mode === "board_permission_wait") await db.update(companyMemberships).set({ membershipRole: "operator" })
.where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, input.actorId)));
await vi.waitFor(async () => {
await service.processFailedChatRunRetry(one.chatActionId!);
expect((await readAgentWakeupRequestsById(one.chatActionId!))[0]?.runId,
JSON.stringify((await db.select().from(chatActions).where(eq(chatActions.id, one.chatActionId!)))[0]?.result)).toBeTruthy();
}, { timeout: 10_000, interval: 100 });
const [parent] = await readAgentWakeupRequestsById(one.chatActionId!);
expect(parent.runId).toBeTruthy();
expect((await readAgentWakeupRequestsById(f.pending!.id))[0]).toMatchObject({
status: "coalesced", runId: mode === "consumed_backlog_52" ? f.runId : parent.runId,
});
if (peerInput) expect((await readAgentWakeupRequestsById(peerInput.id))[0])
.toMatchObject({ status: "deferred_issue_execution", runId: null });
if (mode === "backlog_51") {
const saved = await db.select({ id: chatActions.id, payload: chatActions.payload,
status: agentWakeupRequests.status, runId: agentWakeupRequests.runId }).from(chatActions)
.innerJoin(agentWakeupRequests, eq(agentWakeupRequests.id, chatActions.id)).where(and(
eq(chatActions.conversationId, f.conversation.id), eq(chatActions.kind, "inbound_wakeup"),
sql`${chatActions.id} <> ${f.action.id}::uuid`,
)).orderBy(asc(chatActions.createdAt), asc(chatActions.id));
expect(saved).toHaveLength(51);
expect(saved.every(source => source.status === "coalesced" && source.runId === parent.runId)).toBe(true);
expect((await readHeartbeatRunsById(parent.runId!))[0].contextSnapshot?.wakeCommentIds)
.toEqual(saved.map(source => source.payload.commentId));
}
await service.processFailedChatRunRetry(one.chatActionId!);
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === one.requestId)).toHaveLength(1);
expect(await db.select().from(chatActions).where(and(eq(chatActions.companyId, companyId),
eq(chatActions.kind, "failed_run_retry")))).toHaveLength(1);
if (mode === "retry_failed_successor" || mode === "backlog_51") {
await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(),
errorCode: "adapter_failed", runtimeMode: "native", nativeIssueId: issueId, nativePhase: "terminal_failure",
}).where(eq(heartbeatRuns.id, parent.runId!));
await db.update(agentWakeupRequests).set({ status: "failed" }).where(eq(agentWakeupRequests.id, one.chatActionId!));
await db.update(issues).set({ executionRunId: null, status: "blocked" }).where(eq(issues.id, issueId));
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: parent.runId!, phase: "terminal_failure" });
await db.insert(heartbeatRunEvents).values({ companyId, agentId: f.fixture.assignedAgentId, runId: parent.runId!,
seq: 1, eventType: "native.local_process_stopped", payload: { processPid: 999999998, processGroupId: null } });
await service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user");
const next = await requestTaskRetry({ ...input, observedRunId: parent.runId!, idempotencyKey: randomUUID(),
prepareChatRetry: service.prepareFailedChatRunRetry });
expect(next.requestId).not.toBe(one.requestId);
expect(await service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" });
const [second] = await readAgentWakeupRequestsById(next.chatActionId!);
expect((await readHeartbeatRunsById(second.runId!))[0]).toMatchObject({
retryOfRunId: parent.runId, contextSnapshot: {
wakeCommentIds: (await readHeartbeatRunsById(parent.runId!))[0].contextSnapshot!.wakeCommentIds,
},
});
}
} finally { await f.service.shutdown(); }
},
);
it("leaves saved GitHub input pending when a persisted Retry wait loses board authority", async () => {
const f = await currentGitHubTaskRetryFixture(true);
try {
const companyId = f.fixture.companyId, issueId = f.issue.id, actorId = `retry-operator-${companyId}`;
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: actorId,
status: "active", membershipRole: "operator" });
const intent = await requestTaskRetry({ db, companyId, issueId, actorId, actorSource: "cloud_tenant",
observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry });
let savedOptions: Parameters<typeof f.wakeup>[1];
f.wakeup.mockImplementationOnce(async (_agentId, options) => {
savedOptions = options;
const request = options!.durableChatRequest!;
// Persist the pre-handoff wait as a server interrupted before queue adoption would.
await db.insert(agentWakeupRequests).values({ id: request.id, companyId, agentId: request.agentId,
source: options!.source, triggerDetail: options!.triggerDetail, reason: options!.reason,
requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId,
requestedAt: request.requestedAt, idempotencyKey: request.idempotencyKey,
status: "deferred_issue_execution", payload: { ...options!.payload,
taskRetryRequestId: intent.requestId, _paperclipWakeContext: options!.contextSnapshot } });
return null;
});
expect(await f.service.processFailedChatRunRetry(intent.chatActionId!)).toMatchObject({ status: "deferred" });
await db.update(companyMemberships).set({ membershipRole: "viewer" }).where(and(
eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, actorId)));
expect(await heartbeatService(db).wakeup(f.fixture.assignedAgentId, savedOptions!)).toBeNull();
expect((await readAgentWakeupRequestsById(f.pending!.id))[0])
.toMatchObject({ status: "deferred_issue_execution", runId: null });
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === intent.requestId)).toHaveLength(0);
await db.update(companyMemberships).set({ membershipRole: "operator" }).where(and(
eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, actorId)));
await heartbeatService(db).wakeup(f.fixture.assignedAgentId, savedOptions!);
const [parent] = await readAgentWakeupRequestsById(intent.chatActionId!);
expect(parent.runId).toBeTruthy();
expect((await readAgentWakeupRequestsById(f.pending!.id))[0]).toMatchObject({ status: "coalesced", runId: parent.runId });
} finally { await f.service.shutdown(); }
});
it.each(["changed_bot", "disabled_repository", "retargeted_repository", "revoked_sender", "paused_endpoint",
"edited_source", "deleted_source", "reassigned_task", "second_reconnect", "later_message_wrong_bot", "later_message_deleted", "later_message_revoked_policy"] as const)(
"keeps current GitHub task Retry authority gated: %s", async gate => {
const f = await currentGitHubTaskRetryFixture(true, gate.startsWith("later_message") ? 2 : 1);
try {
const companyId = f.fixture.companyId, issueId = f.issue.id;
await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user");
const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant",
observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry };
if (gate === "later_message_revoked_policy") {
const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.id, f.conversation.resourceId!));
await db.update(chatEndpointResources).set({ availability: "available",
metadata: { ...resource.metadata, providerRepositoryId: "418" },
}).where(eq(chatEndpointResources.id, resource.id));
await db.insert(chatGitHubConfigurations).values({ companyId, endpointId: f.endpoint.id,
updatedByUserId: "owner-user", configuration: { version: 1, toolsEnabled: true, responsibleUserId: "owner-user",
memberAccess: "all_linked", people: [], repositories: {},
defaults: { ...defaultGitHubReviewPolicy(), invocation: "linked_authors", events: ["opened", "synchronize"] },
} });
const sources = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id),
eq(chatActions.kind, "inbound_wakeup"), sql`${chatActions.id} <> ${f.action.id}::uuid`))
.orderBy(asc(chatActions.createdAt), asc(chatActions.id));
for (const [index, source] of sources.entries()) {
const context = { event: index === 0 ? "opened" : "synchronize", deliveryId: source.deliveryId,
repositoryId: "418", repository: resource.providerResourceId,
pullNumber: 7, title: "Saved review request", body: "", baseSha: "a".repeat(40), headSha: "b".repeat(40),
baseBranch: "master", author: { id: "42", login: "octocat", isBot: false },
sender: { id: "42", login: "octocat" }, draft: false, labels: [] };
await db.update(chatDeliveries).set({
normalizedEvent: sql`${chatDeliveries.normalizedEvent} || ${JSON.stringify({ githubAutomatic: { context } })}::jsonb`,
}).where(eq(chatDeliveries.id, source.deliveryId!));
}
}
const accepted = await requestTaskRetry(input);
if (gate.startsWith("later_message")) {
const [later] = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id),
eq(chatActions.kind, "inbound_wakeup"), sql`${chatActions.id} not in (${f.action.id}::uuid, ${f.pending!.id}::uuid)`));
if (gate === "later_message_wrong_bot") await db.update(chatDeliveries).set({
normalizedEvent: sql`jsonb_set(${chatDeliveries.normalizedEvent}, '{githubAuthority,botExternalId}', ${JSON.stringify(randomUUID())}::jsonb)`,
}).where(eq(chatDeliveries.id, later.deliveryId!));
else if (gate === "later_message_deleted") await db.update(issueComments).set({ deletedAt: new Date() })
.where(eq(issueComments.id, String(later.payload.commentId)));
else await db.update(chatGitHubConfigurations).set({ revision: sql`${chatGitHubConfigurations.revision} + 1`,
configuration: sql`jsonb_set(${chatGitHubConfigurations.configuration}, '{defaults,events}', '["opened"]'::jsonb)`,
}).where(eq(chatGitHubConfigurations.endpointId, f.endpoint.id));
}
if (gate === "changed_bot") await db.update(chatEndpoints).set({ botExternalId: randomUUID() })
.where(eq(chatEndpoints.id, f.endpoint.id));
if (gate === "disabled_repository") await db.update(chatEndpointResources).set({ enabled: false })
.where(eq(chatEndpointResources.id, f.conversation.resourceId!));
if (gate === "retargeted_repository") await db.update(chatEndpointResources).set({ providerResourceId: "another/repository" })
.where(eq(chatEndpointResources.id, f.conversation.resourceId!));
if (gate === "revoked_sender") await db.update(chatIdentityLinks).set({ status: "revoked" })
.where(and(eq(chatIdentityLinks.endpointId, f.endpoint.id), eq(chatIdentityLinks.principalId, f.pending!.principalId!)));
if (gate === "paused_endpoint") await db.update(chatEndpoints).set({ status: "paused" })
.where(eq(chatEndpoints.id, f.endpoint.id));
if (gate === "edited_source") await db.update(issueComments).set({ body: "Changed after Retry", updatedAt: new Date() })
.where(eq(issueComments.id, String(f.pending!.payload.commentId)));
if (gate === "deleted_source") await db.update(issueComments).set({ deletedAt: new Date() })
.where(eq(issueComments.id, String(f.pending!.payload.commentId)));
if (gate === "reassigned_task") await db.update(issues).set({ assigneeAgentId: f.fixture.replacementAgentId })
.where(eq(issues.id, issueId));
if (gate === "second_reconnect") await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user");
expect(await f.service.processFailedChatRunRetry(accepted.chatActionId!)).toMatchObject({ status: "failed" });
expect((await readAgentWakeupRequestsById(f.pending!.id))[0]).toMatchObject({ status: "deferred_issue_execution", runId: null });
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === accepted.requestId)).toHaveLength(0);
const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, accepted.chatActionId!));
expect(settled.result?.taskRetryInputReconciled).toBe(true);
const [intentReceipt] = await readAgentWakeupRequestsById(accepted.requestId!);
await f.service.processFailedChatRunRetry(accepted.chatActionId!);
expect((await readAgentWakeupRequestsById(accepted.requestId!))[0].updatedAt).toEqual(intentReceipt.updatedAt);
if (gate === "second_reconnect") {
const next = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() });
expect(next.requestId).not.toBe(accepted.requestId);
expect(await f.service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" });
}
} finally { await f.service.shutdown(); }
},
);
it.each([
"settled",
"distinct_account",
+3 -15
View File
@@ -1,3 +1,4 @@
import { credentialFingerprint } from "./chat-channels/credential-fingerprint.js";
import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js";
import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js";
import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js";
@@ -312,20 +313,6 @@ function runtimeGeneration(setup: ChatEndpointSetupState): number {
return Number.isSafeInteger(value) && value >= 0 ? value : 0;
}
function credentialFingerprint(refs: ToolCredentialSecretRef[]): string {
const stable = refs
.map((ref) => ({
configPath: ref.configPath,
secretId: ref.secretId,
versionSelector: ref.versionSelector ?? "latest",
}))
.sort((left, right) =>
`${left.configPath}:${left.secretId}:${left.versionSelector}`.localeCompare(
`${right.configPath}:${right.secretId}:${right.versionSelector}`,
),
);
return createHash("sha256").update(JSON.stringify(stable)).digest("hex");
}
type ConversationRow = typeof chatConversations.$inferSelect;
type DeliveryRow = typeof chatDeliveries.$inferSelect;
@@ -847,7 +834,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
const { lockTelegramCallbackNoticeAuthority } = createTelegramCallbackNoticeAuthority({ nonDirectDestinationAllowed, lockCurrentPrincipalAuthorization });
const { stageProviderEffect, processProviderEffect, processPendingProviderEffects } = createProviderEffects(db, { endpointRecord, runtimeCallbackEndpoint, withCredentialMutationLease, runtimeFor, invalidateRuntime, lockCurrentPrincipalAuthorization, redactError, MAX_ERROR_TEXT, nonDirectDestinationAllowed, lockTelegramCallbackNoticeAuthority });
const { publicationFiles } = createOutboundAttachments(db, { get storage() { return options.storage; } });
const { authorizeInboundWakeup, authorizeInboundWorkStart, authorizeFastResponse } = createInboundWakeAuthority(db, { runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization });
const { authorizeInboundWakeup, authorizeGitHubTaskRetryBatch, authorizeInboundWorkStart, authorizeFastResponse } = createInboundWakeAuthority(db, { runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization });
const { ingestAttachments, attachmentOmissionDetail } = createInboundAttachments(db, { authorizeInboundWakeup, lifecycleRuntimeFence, sanitizeFilename, issuesSvc, isExternalActionAuthorizationChange, redactError, get storage() { return options.storage; } });
const { stageInboundWakeup, acceptInboundWakeup, settleRejectedInboundWakeups, processInboundWakeup } = createInboundWakeups(db, { authorizeInboundWakeup, authorizeInboundWorkStart, attachmentOmissionDetail, isExternalActionAuthorizationChange, redactError, fetchImpl, get heartbeat() { return options.heartbeat; } });
const { settleTelegramAttachmentOnlyFailure } = createTelegramAttachmentFailure(db, {
@@ -1203,6 +1190,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
pendingFailedRetries,
} = createChatFailedRunRetry(db, {
authorizeInboundWakeup,
authorizeGitHubTaskRetryBatch,
isExternalActionAuthorizationChange,
inboundWakeScope,
lifecycleRuntimeFence,
@@ -0,0 +1,10 @@
import { createHash } from "node:crypto";
import type { ToolCredentialSecretRef } from "@paperclipai/shared";
export function credentialFingerprint(refs: ToolCredentialSecretRef[]): string {
const stable = refs.map(ref => ({ configPath: ref.configPath, secretId: ref.secretId,
versionSelector: ref.versionSelector ?? "latest" })).sort((left, right) =>
`${left.configPath}:${left.secretId}:${left.versionSelector}`.localeCompare(
`${right.configPath}:${right.secretId}:${right.versionSelector}`));
return createHash("sha256").update(JSON.stringify(stable)).digest("hex");
}
@@ -3,6 +3,8 @@ import { readTaskChatRetryIntent, taskChatRetryCanAwaitRecovery, taskChatRetryEx
import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-notifications.js";
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 type { TaskRetryWakeAuthority } from "./github/task-retry-authorization.js";
import {
createHash,
@@ -93,14 +95,16 @@ type RetryRuntimeFence = { credentialFingerprint: string; generation: number };
export type ChatFailedRetrySource = Awaited<ReturnType<ReturnType<typeof createChatFailedRunRetry>["failedChatRetrySource"]>>;
type ChatFailedRunRetryOptions = {
authorizeGitHubTaskRetryBatch: ReturnType<typeof import("./inbound/wakeup-authorization.js").createInboundWakeAuthority>["authorizeGitHubTaskRetryBatch"];
authorizeInboundWakeup: (
database: DbOrTransaction,
action: typeof chatActions.$inferSelect,
notice?: { nonblocking: true; terminal: boolean },
notice?: { nonblocking: true; terminal: boolean; taskRetry?: TaskRetryWakeAuthority },
) => Promise<{
conversation: Pick<typeof chatConversations.$inferSelect, "sessionGeneration">;
endpoint: Pick<typeof chatEndpoints.$inferSelect, "provider">;
delivery: DeliveryRow;
retryRuntimeFence?: RetryRuntimeFence;
}>;
isExternalActionAuthorizationChange: (error: unknown) => boolean;
inboundWakeScope: (action: typeof chatActions.$inferSelect) => string;
@@ -112,6 +116,7 @@ type ChatFailedRunRetryOptions = {
export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOptions) {
const {
authorizeInboundWakeup,
authorizeGitHubTaskRetryBatch,
isExternalActionAuthorizationChange,
inboundWakeScope,
lifecycleRuntimeFence,
@@ -131,14 +136,14 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
}
}
function retryCommentIds(context: Record<string, unknown>): string[] {
function retryCommentIds(context: Record<string, unknown>, taskRetryInput?: TaskRetryInput): string[] {
const ids =
context.wakeCommentIds ??
(context.wakeCommentId ? [context.wakeCommentId] : []);
if (
!Array.isArray(ids) ||
ids.length === 0 ||
ids.length > 50 ||
(!taskRetryInput && ids.length > 50) ||
ids.some((id) => typeof id !== "string" || !isUuidLike(id)) ||
new Set(ids).size !== ids.length ||
(context.wakeCommentId !== undefined &&
@@ -168,6 +173,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
failedRunId: string;
sourceScopeSha256: string;
}>;
taskRetryInput?: TaskRetryInput;
sourceScopeSha256: string;
};
function hashFailedRetrySource(
@@ -651,6 +657,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
runId?: string;
publication?: boolean;
taskRetryIntentId?: string;
taskRetryInput?: TaskRetryInput;
taskRetryReauthorization?: TaskRetryWakeAuthority;
/** Internal selector for already committed output, never retry admission. */
committedResponse?: {
resultId: string;
@@ -734,7 +742,22 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
(context.taskKey !== issue.identifier && context.taskKey !== issue.id)
)
throw failedChatRetryDenied();
const commentIds = retryCommentIds(context);
const selection = input.taskRetryInput && taskRetryInputSchema.safeParse(input.taskRetryInput);
if (selection && !selection.success) throw failedChatRetryDenied();
const selectedInput = selection?.success ? selection.data : undefined;
if (selectedInput && (!input.taskRetryIntentId || provider !== "github")) throw failedChatRetryDenied();
const reauthorization = input.taskRetryReauthorization ?? (selectedInput ? {
intentId: input.taskRetryIntentId!, sourceRunId: failedRun.id, runtimeFence: selectedInput.runtimeFence,
} : undefined);
const pendingSources = selectedInput?.sourceActionIds.length ? await tx.select().from(chatActions).where(and(
eq(chatActions.companyId, input.companyId), eq(chatActions.kind, "inbound_wakeup"),
sql`${chatActions.payload}->>'issueId' = ${issue.id}`,
inArray(chatActions.id, selectedInput.sourceActionIds),
)) : null;
if (pendingSources && pendingSources.length !== selectedInput!.sourceActionIds.length) throw failedChatRetryDenied();
const pendingById = new Map(pendingSources?.map(source => [source.id, source]));
const commentIds = pendingSources ? selectedInput!.sourceActionIds.map(id =>
String(pendingById.get(id)!.payload.commentId)) : retryCommentIds(context, selectedInput);
const [accepted] = await tx
.select()
.from(nativeRunResults)
@@ -868,7 +891,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
answerDelivery
)
throw failedChatRetryDenied();
if (context.chatFailedRunRetry) {
if (context.chatFailedRunRetry && !pendingSources) {
const hint = context.chatFailedRunRetry as Record<string, unknown>;
const [priorIntent] = await tx
.select()
@@ -914,6 +937,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
committedResponse: undefined,
taskRetryIntentId: typeof priorIntent.payload.taskRetryIntentId === "string"
? priorIntent.payload.taskRetryIntentId : undefined,
taskRetryInput: priorIntent.payload.taskRetryInput as TaskRetryInput | undefined,
retryActionId: priorIntent.id,
taskRetryReauthorization: reauthorization,
},
ancestors,
);
@@ -945,6 +971,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
return hashFailedRetrySource({
...parentScope,
failedRunId: failedRun.id,
...(selectedInput ? { taskRetryInput: { ...selectedInput, runtimeFence: reauthorization?.runtimeFence } } : {}),
retryAncestors: [
...parent.retryAncestors,
{
@@ -958,20 +985,21 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
// Include every receipt that fed this run, including a deferred owner's
// coalesced children. A mutable latest-comment hint cannot drop a sibling.
const admittedBatch =
await tx.execute(sql`select source.id, source.payload->>'commentId' as comment_id
pendingSources ? [] : await tx.execute(sql`select source.id, source.payload->>'commentId' as comment_id
from agent_wakeup_requests receipt join chat_actions source on source.id = receipt.id
where receipt.company_id = ${input.companyId}::uuid and receipt.agent_id = ${input.agentId}::uuid
and (receipt.run_id = ${failedRun.id}::uuid or receipt.id = ${failedRun.wakeupRequestId}::uuid
or receipt.payload->>'coalescedIntoWakeupRequestId' = ${failedRun.wakeupRequestId})
and source.company_id = ${input.companyId}::uuid and source.kind = 'inbound_wakeup' limit 51`);
and source.company_id = ${input.companyId}::uuid and source.kind = 'inbound_wakeup'
${selectedInput ? sql`` : sql`limit 51`}`);
if (
admittedBatch.length !== commentIds.length ||
!pendingSources && (admittedBatch.length !== commentIds.length ||
admittedBatch.some(
(entry) => !commentIds.includes(String(entry.comment_id)),
)
))
)
throw failedChatRetryDenied();
const sources = await tx
const sources = pendingSources ?? await tx
.select()
.from(chatActions)
.where(
@@ -986,8 +1014,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
),
);
if (sources.length !== commentIds.length) throw failedChatRetryDenied();
const sourcesByCommentId = new Map(sources.map(source => [source.payload.commentId, source]));
const ordered = commentIds.map(
(id) => sources.find((action) => action.payload.commentId === id)!,
(id) => sourcesByCommentId.get(id)!,
);
const first = ordered[0]!;
if (
@@ -1011,39 +1040,53 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
);
if (!identity[0]?.locked)
throw new NativeChatReviewPresentationContentionError();
const actionIds = ordered.map(action => action.id);
const currentActions = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, input.companyId),
inArray(chatActions.id, actionIds))).orderBy(asc(chatActions.id)).for("update", { noWait: true });
const deliveries = await tx.select().from(chatDeliveries).where(and(eq(chatDeliveries.companyId, input.companyId),
inArray(chatDeliveries.id, ordered.map(action => action.deliveryId!))))
.orderBy(asc(chatDeliveries.id)).for("update", { noWait: true });
const comments = await tx.select().from(issueComments).where(and(eq(issueComments.companyId, input.companyId),
eq(issueComments.issueId, issue.id), inArray(issueComments.id, commentIds)))
.orderBy(asc(issueComments.id)).for("share", { noWait: true });
const receipts = await tx.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, input.companyId),
inArray(agentWakeupRequests.id, [...actionIds, failedRun.wakeupRequestId!, ...(input.retryActionId ? [input.retryActionId] : [])])))
.orderBy(asc(agentWakeupRequests.id)).for("share", { noWait: true });
const actionsById = new Map(currentActions.map(action => [action.id, action]));
const deliveriesById = new Map(deliveries.map(delivery => [delivery.id, delivery]));
const commentsById = new Map(comments.map(comment => [comment.id, comment]));
const receiptsById = new Map(receipts.map(receipt => [receipt.id, receipt]));
const lifecycleRows = await tx.select({
target: sql<string>`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId'`,
context: sql<Record<string, unknown>>`${chatDeliveries.normalizedEvent}->'runtimeContext'`,
}).from(chatDeliveries).where(and(eq(chatDeliveries.companyId, input.companyId),
eq(chatDeliveries.endpointId, first.endpointId),
or(isNull(chatDeliveries.conversationId), eq(chatDeliveries.conversationId, first.conversationId)),
ne(chatDeliveries.state, "filtered"), inArray(chatDeliveries.eventKind, ["message_updated", "message_deleted", "message_restored"]),
inArray(sql<string>`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId'`, deliveries.map(delivery => delivery.providerEventId))));
const lifecycleKey = (target: string, context: Record<string, unknown> | null) =>
JSON.stringify([target, Object.entries(context ?? {}).sort(([left], [right]) => left.localeCompare(right))]);
const invalidated = new Set(lifecycleRows.map(row => lifecycleKey(row.target, row.context)));
const batch = reauthorization ? await authorizeGitHubTaskRetryBatch(tx, ordered, {
nonblocking: true, terminal: input.publication === true, taskRetry: reauthorization,
}).catch(error => { if (isExternalActionAuthorizationChange(error)) throw failedChatRetryDenied(); throw error; }) : null;
const canonicalSources: Array<Record<string, unknown>> = [];
let destination: Awaited<ReturnType<typeof authorizeInboundWakeup>> | null =
null;
for (const action of ordered) {
destination = await authorizeInboundWakeup(tx, action, {
destination = batch?.get(action.id) ?? await authorizeInboundWakeup(tx, action, {
nonblocking: true,
terminal: input.publication === true,
...(reauthorization ? { taskRetry: reauthorization } : {}),
}).catch((error) => {
if (isExternalActionAuthorizationChange(error))
throw failedChatRetryDenied();
throw error;
});
const [delivery] = await tx
.select()
.from(chatDeliveries)
.where(eq(chatDeliveries.id, action.deliveryId!))
.for("update", { noWait: true });
const [current] = await tx
.select()
.from(chatActions)
.where(eq(chatActions.id, action.id))
.for("update", { noWait: true });
const [comment] = await tx
.select()
.from(issueComments)
.where(
and(
eq(issueComments.id, String(action.payload.commentId)),
eq(issueComments.companyId, input.companyId),
eq(issueComments.issueId, issue.id),
),
)
.for("share", { noWait: true });
if (reauthorization && !reauthorization.runtimeFence) reauthorization.runtimeFence = destination.retryRuntimeFence;
const delivery = deliveriesById.get(action.deliveryId!);
const current = actionsById.get(action.id);
const comment = commentsById.get(String(action.payload.commentId));
if (
!current ||
current.status !== "processed" ||
@@ -1053,43 +1096,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
!comment ||
comment.deletedAt ||
comment.updatedAt.getTime() !== comment.createdAt.getTime() ||
comment.updatedAt >
(!pendingSources && comment.updatedAt >
(input.committedResponse
? accepted!.createdAt
: failedRun.finishedAt!) ||
: failedRun.finishedAt!)) ||
destination.endpoint.provider !== provider ||
JSON.stringify(lifecycleRuntimeFence(delivery)) !==
JSON.stringify(lifecycleRuntimeFence(destination.delivery))
)
throw failedChatRetryDenied();
const [lifecycle] = await tx
.select({ id: chatDeliveries.id })
.from(chatDeliveries)
.where(
and(
eq(chatDeliveries.companyId, input.companyId),
eq(chatDeliveries.endpointId, first.endpointId),
or(
isNull(chatDeliveries.conversationId),
eq(chatDeliveries.conversationId, first.conversationId),
),
ne(chatDeliveries.state, "filtered"),
sql`${chatDeliveries.normalizedEvent}->'runtimeContext' = ${JSON.stringify(lifecycleRuntimeFence(delivery))}::jsonb`,
inArray(chatDeliveries.eventKind, [
"message_updated",
"message_deleted",
"message_restored",
]),
sql`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId' = ${delivery.providerEventId}`,
),
)
.limit(1);
if (lifecycle) throw failedChatRetryDenied();
const [receipt] = await tx
.select()
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, action.id))
.for("share", { noWait: true });
if (invalidated.has(lifecycleKey(delivery.providerEventId, lifecycleRuntimeFence(delivery)))) throw failedChatRetryDenied();
const receipt = receiptsById.get(action.id);
if (!receipt) throw failedChatRetryDenied();
assertFailedRetryReceipt(
createDurableChatWakeupRequest({
@@ -1107,6 +1124,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
}),
receipt,
);
if (pendingSources) {
const direct = receipt.status === "deferred_issue_execution" && !receipt.runId
&& receipt.payload?.coalescedIntoWakeupRequestId === undefined;
const owned = input.retryActionId && receipt.status === "coalesced"
&& receipt.payload?.coalescedIntoWakeupRequestId === input.retryActionId;
if (!direct && !owned) throw failedChatRetryDenied();
if (owned && receipt.runId) {
const owner = receiptsById.get(input.retryActionId!);
if (owner?.runId !== receipt.runId) throw failedChatRetryDenied();
}
}
let owner = receipt;
const seen = new Set<string>();
for (
@@ -1126,25 +1154,16 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
)
throw failedChatRetryDenied();
seen.add(ownerId);
const [next] = await tx
.select()
.from(agentWakeupRequests)
.where(
and(
eq(agentWakeupRequests.id, ownerId),
eq(agentWakeupRequests.companyId, input.companyId),
),
)
.for("share", { noWait: true });
const next = receiptsById.get(ownerId);
if (!next) throw failedChatRetryDenied();
owner = next;
}
if (
owner.id !== failedRun.wakeupRequestId ||
(!pendingSources && (owner.id !== failedRun.wakeupRequestId ||
owner.runId !== failedRun.id ||
owner.agentId !== input.agentId ||
owner.requestedByActorType !== receipt.requestedByActorType ||
owner.requestedByActorId !== receipt.requestedByActorId ||
owner.requestedByActorId !== receipt.requestedByActorId)) ||
["skipped", "cancelled"].includes(receipt.status)
)
throw failedChatRetryDenied();
@@ -1158,7 +1177,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
bodySha256: createHash("sha256").update(comment.body).digest("hex"),
omissionReasons: action.payload.attachmentOmissionReasons ?? {},
receiptId: receipt.id,
ownerId: owner.id,
ownerId: pendingSources ? action.id : owner.id,
});
}
if (!destination) throw failedChatRetryDenied();
@@ -1204,6 +1223,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
| "system",
requestedByActorId: String(first.payload.requestedByActorId),
retryAncestors: [],
...(selectedInput ? { taskRetryInput: { ...selectedInput,
runtimeFence: selectedInput.runtimeFence ?? destination.retryRuntimeFence } } : {}),
};
return hashFailedRetrySource(scope);
}
@@ -1240,7 +1261,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
retryIntentId?: string;
},
): Promise<{ actionId: string; issueId: string }> {
const source = await failedChatRetrySource(tx, { ...input, taskRetryIntentId: input.retryIntentId });
const taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(tx, input) : undefined;
const source = await failedChatRetrySource(tx, { ...input, taskRetryIntentId: input.retryIntentId, taskRetryInput });
const providerActionId = input.retryIntentId
? `task-retry:${input.retryIntentId}` : `failed-run-retry:${source.failedRunId}`;
const [existing] = await tx
@@ -1351,6 +1373,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
failedRunId: snapshot.payload.failedRunId,
taskRetryIntentId: typeof snapshot.payload.taskRetryIntentId === "string"
? snapshot.payload.taskRetryIntentId : undefined,
taskRetryInput: snapshot.payload.taskRetryInput as TaskRetryInput | undefined,
});
if (typeof snapshot.payload.taskRetryIntentId === "string" && !await readTaskChatRetryIntent(tx as unknown as Db, {
companyId: input.companyId, issueId: input.issueId, agentId: input.agentId,
@@ -1444,7 +1467,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
context.source !== `chat:${source.provider}` ||
context.taskKey !== source.taskKey ||
context.retryOfRunId !== source.failedRunId ||
JSON.stringify(retryCommentIds(context)) !==
JSON.stringify(retryCommentIds(context, source.taskRetryInput)) !==
JSON.stringify(source.commentIds) ||
hint?.version !== 1 ||
hint.actionId !== source.intent.id ||
@@ -1518,8 +1541,19 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
};
}
async function reconcileFailedTaskRetry(action: typeof chatActions.$inferSelect) {
if (!["failed", "cancelled"].includes(action.status) || !action.payload.taskRetryIntentId ||
action.result?.taskRetryInputReconciled === true) return action;
await releaseUnadmittedTaskRetryInput(db, { companyId: action.companyId, retryActionId: action.id });
const [settled] = await db.update(chatActions).set({
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || '{"taskRetryInputReconciled":true}'::jsonb`,
}).where(and(eq(chatActions.companyId, action.companyId), eq(chatActions.id, action.id),
inArray(chatActions.status, ["failed", "cancelled"]))).returning();
return settled ?? action;
}
async function claimFailedChatRunRetry(actionId: string) {
const [candidate] = await db
const [snapshot] = await db
.select()
.from(chatActions)
.where(
@@ -1528,7 +1562,10 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
eq(chatActions.kind, "failed_run_retry"),
),
);
if (!candidate) throw notFound("Chat retry request not found");
if (!snapshot) throw notFound("Chat retry request not found");
// Reconcile once at settlement; a crash before the marker is safely
// retried here. Status projection itself performs no writes or row locks.
const candidate = await reconcileFailedTaskRetry(snapshot);
const now = new Date();
if (
isShuttingDown() ||
@@ -1571,7 +1608,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
)
.returning();
});
return failedChatRetryReceipt(current ?? claimed);
return failedChatRetryReceipt(await reconcileFailedTaskRetry(current ?? claimed));
};
try {
const scope = {
@@ -1644,6 +1681,12 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
true, // The durable wake callback validates admission; later phases recheck containment.
);
},
onAdmitted: async (tx, receipt) => {
if (!source.taskRetryInput) return;
await claimTaskRetryInput(tx, { companyId: claimed.companyId,
retryActionId: claimed.id, sourceActionIds: source.taskRetryInput.sourceActionIds });
await settleTaskRetryInput(tx, { companyId: claimed.companyId, retryActionId: claimed.id, runId: receipt.runId });
},
}),
});
}
@@ -1652,6 +1695,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, claimed.id));
if (!receipt) throw new Error("chat_retry_receipt_not_committed");
await settleTaskRetryInput(db, { companyId: claimed.companyId, retryActionId: claimed.id, runId: receipt.runId });
return settle(
["skipped", "failed", "cancelled"].includes(receipt.status)
? "failed"
@@ -37,7 +37,7 @@ export async function retainGitHubDeliveryAuthority(
linkedUserId: string | null, taskUserId: string | null,
): Promise<void> {
const guest = !linkedUserId;
await taskTx.update(chatDeliveries).set({ normalizedEvent: sql`${chatDeliveries.normalizedEvent} || ${JSON.stringify({ githubAuthority: { guest, responsibleUserId: taskUserId, sponsorUserId: guest ? taskEndpoint.sponsorUserId : null } })}::jsonb` }).where(eq(chatDeliveries.id, deliveryId));
await taskTx.update(chatDeliveries).set({ normalizedEvent: sql`${chatDeliveries.normalizedEvent} || ${JSON.stringify({ githubAuthority: { guest, botExternalId: taskEndpoint.botExternalId, responsibleUserId: taskUserId, sponsorUserId: guest ? taskEndpoint.sponsorUserId : null } })}::jsonb` }).where(eq(chatDeliveries.id, deliveryId));
}
export async function retainGitHubReviewReceipt(taskTx: DbOrTransaction, endpoint: EndpointRow, issueId: string, deliveryId: string, githubAutomatic: AutomaticRequest): Promise<void> {
@@ -0,0 +1,126 @@
import { and, eq, inArray } from "drizzle-orm";
import { chatActions, chatConversations, chatDeliveries, chatEndpointResources, chatMessageLinks, issueComments, toolConnections } from "@paperclipai/db";
import { forbidden } from "../../../errors.js";
import type { DbOrTransaction, EndpointRow, LifecycleRuntimeFence } from "../runtime-types.js";
import { credentialFingerprint } from "../credential-fingerprint.js";
import { readTaskChatRetryIntent } from "../task-retry-authority.js";
export type TaskRetryWakeAuthority = {
intentId: string;
sourceRunId: string;
runtimeFence?: LifecycleRuntimeFence;
};
export function githubTaskRetryAppMatches(endpoint: EndpointRow, delivery: typeof chatDeliveries.$inferSelect,
response: { payload: Record<string, unknown> } | undefined) {
const original = delivery.normalizedEvent.githubAuthority as { botExternalId?: unknown } | undefined;
return original?.botExternalId !== undefined ? original.botExternalId === endpoint.botExternalId
: response?.payload.botExternalId === endpoint.botExternalId && response.payload.agentId === endpoint.assignedAgentId;
}
export function githubTaskRetryDestinationMatches(delivery: typeof chatDeliveries.$inferSelect,
conversation: typeof chatConversations.$inferSelect, resource: typeof chatEndpointResources.$inferSelect | null) {
const originalResource = delivery.normalizedEvent.resource as { providerResourceId?: unknown } | undefined;
const originalThread = delivery.normalizedEvent.conversation as { externalThreadId?: unknown } | undefined;
const automatic = (delivery.normalizedEvent.githubAutomatic ?? delivery.normalizedEvent.githubIssue) as
{ context?: { repositoryId?: unknown } } | undefined;
return resource && originalResource?.providerResourceId === resource.providerResourceId
&& originalThread?.externalThreadId === conversation.externalThreadId
&& (automatic?.context?.repositoryId === undefined
|| automatic.context.repositoryId === resource.metadata.providerRepositoryId);
}
export async function githubTaskRetryRuntimeFence(tx: DbOrTransaction, endpoint: EndpointRow,
action: typeof chatActions.$inferSelect, delivery: typeof chatDeliveries.$inferSelect,
authority: TaskRetryWakeAuthority): Promise<LifecycleRuntimeFence> {
const deny = () => forbidden("This accepted chat message is no longer authorized to start work",
{ code: "chat_action_authorization_changed" });
const retry = await readTaskChatRetryIntent(tx as Parameters<typeof readTaskChatRetryIntent>[0], {
companyId: action.companyId, issueId: String(action.payload.issueId),
agentId: String(action.payload.agentId), sourceRunId: authority.sourceRunId, intentId: authority.intentId,
});
if (!retry || endpoint.provider !== "github" || !endpoint.botExternalId || action.kind !== "inbound_wakeup"
|| action.status !== "processed" || delivery.state !== "processed") throw deny();
const [response] = await tx.select({ payload: chatActions.payload }).from(chatActions).where(and(
eq(chatActions.companyId, action.companyId), eq(chatActions.endpointId, endpoint.id),
eq(chatActions.conversationId, action.conversationId!), eq(chatActions.deliveryId, delivery.id),
eq(chatActions.kind, "github_response_comment"),
));
if (!githubTaskRetryAppMatches(endpoint, delivery, response)) throw deny();
const [connection] = await tx.select({ refs: toolConnections.credentialSecretRefs,
enabled: toolConnections.enabled, status: toolConnections.status }).from(toolConnections).where(and(
eq(toolConnections.companyId, endpoint.companyId), eq(toolConnections.id, endpoint.connectionId),
));
if (!connection?.enabled || connection.status !== "active") throw deny();
const generation = Number((endpoint.setup as { runtimeGeneration?: unknown }).runtimeGeneration ?? 0);
if (!Number.isSafeInteger(generation) || generation < 0) throw deny();
return authority.runtimeFence ?? { generation, credentialFingerprint: credentialFingerprint(connection.refs ?? []) };
}
type GitHubRetryBinding = {
endpoint: EndpointRow;
conversation: typeof chatConversations.$inferSelect;
resource: typeof chatEndpointResources.$inferSelect | null;
delivery: typeof chatDeliveries.$inferSelect;
retryRuntimeFence?: LifecycleRuntimeFence;
};
export function createGitHubTaskRetryBatchAuthority<T extends GitHubRetryBinding>(authorize: (
tx: DbOrTransaction, action: typeof chatActions.$inferSelect,
notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority },
) => Promise<T>) {
async function authorizeGitHubTaskRetryBatch(tx: DbOrTransaction, actions: Array<typeof chatActions.$inferSelect>,
notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority }) {
const deny = () => forbidden("This accepted chat message is no longer authorized to start work",
{ code: "chat_action_authorization_changed" });
const first = actions[0];
if (!first) throw deny();
const commonScope = (action: typeof chatActions.$inferSelect) => JSON.stringify([
action.companyId, action.endpointId, action.conversationId, action.principalId, action.kind, action.status,
...["version", "issueId", "agentId", "sessionGeneration", "requestedByActorType", "requestedByActorId"]
.map(key => action.payload[key]),
]);
if (actions.some(action => commonScope(action) !== commonScope(first))) throw deny();
const deliveries = await tx.select().from(chatDeliveries).where(and(eq(chatDeliveries.companyId, first.companyId),
inArray(chatDeliveries.id, actions.map(action => action.deliveryId!))));
const byDelivery = new Map(deliveries.map(delivery => [delivery.id, delivery]));
const legacy = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, first.companyId),
eq(chatActions.endpointId, first.endpointId), eq(chatActions.conversationId, first.conversationId!),
eq(chatActions.kind, "github_response_comment"), inArray(chatActions.deliveryId, deliveries.map(delivery => delivery.id))));
const legacyByDelivery = new Map(legacy.map(action => [action.deliveryId, action]));
const links = await tx.select({ deliveryId: chatMessageLinks.deliveryId, commentId: chatMessageLinks.commentId,
authorUserId: issueComments.authorUserId, deletedAt: issueComments.deletedAt }).from(chatMessageLinks)
.innerJoin(issueComments, and(eq(issueComments.id, chatMessageLinks.commentId),
eq(issueComments.companyId, first.companyId), eq(issueComments.issueId, String(first.payload.issueId))))
.where(and(eq(chatMessageLinks.companyId, first.companyId), eq(chatMessageLinks.endpointId, first.endpointId),
eq(chatMessageLinks.conversationId, first.conversationId!), eq(chatMessageLinks.direction, "inbound"),
inArray(chatMessageLinks.deliveryId, deliveries.map(delivery => delivery.id))));
const expectedUserId = first.payload.requestedByActorType === "user" ? first.payload.requestedByActorId : null;
const linked = new Set(links.filter(link => (!link.deletedAt || notice.terminal) && link.authorUserId === expectedUserId)
.map(link => `${link.deliveryId}:${link.commentId}`));
const groups = new Map<string, T>();
const authorized = new Map<string, T>();
for (const action of actions) {
const delivery = byDelivery.get(action.deliveryId!);
if (!delivery || !linked.has(`${delivery.id}:${action.payload.commentId}`)) throw deny();
// Reuse only the same live grant within this transaction. Automatic review
// policies with different contexts still receive their own full check.
const grant = JSON.stringify(delivery.normalizedEvent.githubAutomatic ?? delivery.normalizedEvent.githubIssue ?? null);
let current = groups.get(grant);
if (!current) {
current = await authorize(tx, action, notice);
if (current.endpoint.provider !== "github") throw deny();
notice.taskRetry.runtimeFence ??= current.retryRuntimeFence;
groups.set(grant, current);
}
if (delivery.state !== "processed" || delivery.endpointId !== current.endpoint.id ||
delivery.conversationId !== current.conversation.id || delivery.principalId !== action.principalId ||
!githubTaskRetryAppMatches(current.endpoint, delivery, legacyByDelivery.get(delivery.id)) ||
!githubTaskRetryDestinationMatches(delivery, current.conversation, current.resource)) throw deny();
authorized.set(action.id, { ...current, delivery });
}
return authorized;
}
return authorizeGitHubTaskRetryBatch;
}
@@ -11,6 +11,7 @@ import { agentWakeupRequests, agents, chatActions, chatConversations, chatDelive
import { forbidden } from "../../../errors.js";
import { resolveGitHubWakeAuthority, isGitHubAutomaticDelivery } from "../github/wake-authority.js";
import { assertSlackWakeMention } from "../slack/wake-authority.js";
import { githubTaskRetryRuntimeFence, githubTaskRetryDestinationMatches, createGitHubTaskRetryBatchAuthority, type TaskRetryWakeAuthority } from "../github/task-retry-authorization.js";
import type { createChatInboundAuthority } from "./authorization.js";
import type { createChatMessageLifecycle } from "./lifecycle.js";
type WakeAuthorityOptions = {
@@ -41,7 +42,7 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions
async function authorizeInboundWakeup(
tx: DbOrTransaction,
action: typeof chatActions.$inferSelect,
notice?: { nonblocking: true; terminal: boolean },
notice?: { nonblocking: true; terminal: boolean; taskRetry?: TaskRetryWakeAuthority },
) {
const payload = action.payload;
const deny = () =>
@@ -106,6 +107,8 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions
.limit(1)
.then((rows) => rows[0] ?? null);
const fence = delivery ? lifecycleRuntimeFence(delivery) : null;
const currentFence = endpoint && delivery && notice?.taskRetry
? await githubTaskRetryRuntimeFence(tx, endpoint, action, delivery, notice.taskRetry) : fence;
if (
!endpoint ||
endpoint.assignedAgentId !== payload.agentId ||
@@ -114,8 +117,8 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions
delivery.endpointId !== endpoint.id ||
delivery.conversationId !== action.conversationId ||
delivery.principalId !== action.principalId ||
!fence ||
!(await runtimeCallbackEndpoint(tx as unknown as DbTransaction, endpoint.id, fence, [
!currentFence ||
!(await runtimeCallbackEndpoint(tx as unknown as DbTransaction, endpoint.id, currentFence, [
"verifying",
"active",
]))
@@ -160,6 +163,7 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions
endpoint,
action.principalId,
);
if (notice?.taskRetry && !githubTaskRetryDestinationMatches(delivery, conversation, resource)) throw deny();
if (endpoint.provider === "github") {
const automatic = await resolveGitHubWakeAuthority(tx, endpoint, delivery);
if (automatic.kind === "denied") throw deny();
@@ -214,9 +218,12 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions
source.authorUserId !== expectedUserId
)
throw deny();
return { issue, delivery, endpoint, conversation };
return { issue, delivery, endpoint, conversation, resource,
...(notice?.taskRetry && currentFence ? { retryRuntimeFence: currentFence } : {}) };
}
const authorizeGitHubTaskRetryBatch = createGitHubTaskRetryBatchAuthority(authorizeInboundWakeup);
async function authorizeInboundWorkStart(
tx: DbOrTransaction,
action: typeof chatActions.$inferSelect,
@@ -241,5 +248,5 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions
attachments: normalized.message?.attachments?.map(a => `${a.name ?? "Attachment"} (${a.mimeType ?? "unknown type"})`)
};
}
return { authorizeInboundWakeup, authorizeInboundWorkStart, authorizeFastResponse };
return { authorizeInboundWakeup, authorizeGitHubTaskRetryBatch, authorizeInboundWorkStart, authorizeFastResponse };
}
@@ -0,0 +1,95 @@
import { z } from "zod";
import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
import { agentWakeupRequests, chatActions, chatConversations, chatEndpoints, heartbeatRuns, issues, type Db } from "@paperclipai/db";
import type { DbOrTransaction } from "./runtime-types.js";
export const taskRetryInputSchema = z.object({
// This is a server-selected snapshot of the durable queue, not client input.
// Truncating it strands older requests or makes the newer-input guard reject
// every Retry. Dispatch already carries the full queue with bounded inline text.
sourceActionIds: z.array(z.string().uuid()).refine(ids => new Set(ids).size === ids.length),
runtimeFence: z.object({ generation: z.number().int().nonnegative(),
credentialFingerprint: z.string().regex(/^[a-f0-9]{64}$/) }).optional(),
});
export type TaskRetryInput = z.infer<typeof taskRetryInputSchema>;
export async function selectTaskRetryInput(tx: DbOrTransaction, input: {
companyId: string; issueId: string; agentId: string; failedRunId: string;
}) {
const [conversation] = await tx.select({ id: chatConversations.id }).from(chatConversations)
.innerJoin(chatEndpoints, eq(chatEndpoints.id, chatConversations.endpointId))
.innerJoin(chatActions, and(eq(chatActions.companyId, chatConversations.companyId),
eq(chatActions.conversationId, chatConversations.id)))
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, chatConversations.companyId),
eq(heartbeatRuns.wakeupRequestId, chatActions.id))).where(and(
eq(chatConversations.companyId, input.companyId), eq(chatConversations.issueId, input.issueId),
eq(heartbeatRuns.id, input.failedRunId), eq(heartbeatRuns.agentId, input.agentId),
eq(chatEndpoints.provider, "github"), eq(chatEndpoints.assignedAgentId, input.agentId),
eq(chatEndpoints.externalExecutionPolicy, "restricted"),
)).limit(1);
if (!conversation) return undefined;
const pending = await tx.select({ id: chatActions.id }).from(chatActions)
.innerJoin(agentWakeupRequests, and(eq(agentWakeupRequests.companyId, chatActions.companyId),
eq(agentWakeupRequests.id, chatActions.id))).where(and(
eq(chatActions.companyId, input.companyId), eq(chatActions.conversationId, conversation.id),
eq(chatActions.kind, "inbound_wakeup"), eq(chatActions.status, "processed"),
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
isNull(agentWakeupRequests.runId),
)).orderBy(asc(chatActions.createdAt), asc(chatActions.id)).for("update", { of: agentWakeupRequests });
return taskRetryInputSchema.parse({ sourceActionIds: pending.map(row => row.id) });
}
export async function claimTaskRetryInput(tx: DbOrTransaction, input: {
companyId: string; retryActionId: string; sourceActionIds: string[];
}) {
if (!input.sourceActionIds.length) return;
await tx.update(agentWakeupRequests).set({ status: "coalesced", updatedAt: new Date(),
payload: sql`coalesce(${agentWakeupRequests.payload}, '{}'::jsonb) || ${JSON.stringify({
coalescedIntoWakeupRequestId: input.retryActionId })}::jsonb`,
}).where(and(eq(agentWakeupRequests.companyId, input.companyId),
inArray(agentWakeupRequests.id, input.sourceActionIds), eq(agentWakeupRequests.status, "deferred_issue_execution"),
isNull(agentWakeupRequests.runId)));
}
export async function settleTaskRetryInput(tx: DbOrTransaction, input: { companyId: string; retryActionId: string; runId: string | null }) {
if (!input.runId) return;
await tx.update(agentWakeupRequests).set({ runId: input.runId, updatedAt: new Date() }).where(and(
eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.status, "coalesced"),
isNull(agentWakeupRequests.runId),
sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${input.retryActionId}`,
));
}
export async function releaseUnadmittedTaskRetryInput(db: Db, input: { companyId: string; retryActionId: string }) {
await db.transaction(async tx => {
const [snapshot] = await tx.select({ payload: chatActions.payload }).from(chatActions).where(and(
eq(chatActions.companyId, input.companyId), eq(chatActions.id, input.retryActionId),
));
const issueId = z.string().uuid().safeParse(snapshot?.payload.issueId);
if (!issueId.success) return;
const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(
eq(issues.companyId, input.companyId), eq(issues.id, issueId.data),
)).for("update");
if (!issue) return;
const [action] = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, input.companyId),
eq(chatActions.id, input.retryActionId), eq(chatActions.kind, "failed_run_retry"),
sql`${chatActions.payload}->>'issueId' = ${issue.id}`,
inArray(chatActions.status, ["failed", "cancelled"]))).for("update");
if (!action || typeof action.payload.taskRetryIntentId !== "string") return;
const [receipt] = await tx.select({ runId: agentWakeupRequests.runId }).from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.retryActionId),
)).for("update");
if (receipt?.runId) return;
await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", finishedAt: null,
payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(),
}).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.status, "coalesced"),
isNull(agentWakeupRequests.runId),
sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${input.retryActionId}`));
await tx.update(agentWakeupRequests).set({ status: "failed", finishedAt: new Date(), updatedAt: new Date() }).where(and(
eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, action.payload.taskRetryIntentId),
eq(agentWakeupRequests.reason, "task_retry"), isNull(agentWakeupRequests.runId),
inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued", "claimed"]),
sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}' = ${action.id}`,
));
});
}
+22 -1
View File
@@ -1,6 +1,6 @@
import { createHash } from "node:crypto";
import { agentWakeupRequests, chatActions, chatDeliveries, chatEndpoints, type Db } from "@paperclipai/db";
import { and, eq, or, sql, type SQLWrapper } from "drizzle-orm";
import { and, eq, inArray, or, sql, type SQLWrapper } from "drizzle-orm";
import { HttpError } from "../errors.js";
export interface CommittedChatResponseAuthorizationInput {
@@ -211,6 +211,7 @@ export interface DurableChatWakeupRequest {
readonly idempotencyKey: string;
readonly failedRunRetry?: { readonly failedRunId: string };
readonly authorize: (tx: Db) => Promise<void>;
readonly onAdmitted?: (tx: Db, receipt: { runId: string | null }) => Promise<void>;
}
const trustedRequests = new WeakSet<object>();
@@ -290,6 +291,26 @@ export function assertDurableChatWakeupReceipt(
}
}
export function withDurableChatAdmission<T extends { kind: string; admitted?: boolean }>(request: DurableChatWakeupRequest | undefined,
admit: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => Promise<T>) {
return async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
const outcome = await admit(tx);
// An existing deferred receipt is not proof this transaction passed its
// gates. Only dispatch or an explicitly persisted wait admits the input.
if (request?.onAdmitted && (outcome.admitted === true || ["queued", "durable", "coalesced"].includes(outcome.kind))) {
const [receipt] = await tx.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, request.companyId), eq(agentWakeupRequests.id, request.id),
inArray(agentWakeupRequests.status, ["queued", "claimed", "deferred_issue_execution", "coalesced"]),
));
if (receipt) {
assertDurableChatWakeupReceipt(request, receipt);
await request.onAdmitted(tx as unknown as Db, { runId: receipt.runId });
}
}
return outcome;
};
}
/** Work-start policy only: ongoing tool and publication authorization must not
* revoke an already-started run when the operator enables mention gating. */
export async function slackMentionAllowsRunStart(
+11 -17
View File
@@ -1,5 +1,4 @@
import { authorizationService } from "../authorization.js";
import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry } from "../task-retry.js";
import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry, taskRetryMutationAllowed } from "../task-retry.js";
import { insertIssueComment } from "../issue-comment-writes.js";
import { withExternalAdmissionGuard, assertNoExternalOverlap } from "../../modules/external-agents/index.js";
import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js";
@@ -50,6 +49,7 @@ import {
slackMentionAllowsRunStart,
assertDurableChatWakeupReceipt,
assertDurableChatWakeupRequest,
withDurableChatAdmission,
authorizeFailedChatRunRetryWake,
FailedChatRunRetryAuthorizationError,
unadmittedChatWakeupCondition,
@@ -2809,7 +2809,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
const cancelledRunsToEmit: (typeof heartbeatRuns.$inferSelect)[] = [];
const outcome = await db
.transaction(async (tx) => {
.transaction(withDurableChatAdmission(durableRequest, async (tx) => {
await tx.execute(
sql`select id from issues where id = ${issueId} and company_id = ${agent.companyId} for update`,
);
@@ -2847,11 +2847,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
if (pending.id !== opts.taskRetryRequestId || retry?.generationId !== pending.id || retry.chatActionId ||
retry.actorId !== opts.requestedByActorId || retry.sourceRunId !== enrichedContextSnapshot.previousRunId)
return { kind: "deferred" as const };
const permission = await authorizationService(tx as unknown as Db).decide({
actor: { type: "board", userId: retry.actorId, source: retry.actorSource }, action: "issue:mutate",
resource: { type: "issue", companyId: agent.companyId, issueId },
});
if (!permission.allowed) return { kind: "deferred" as const };
if (!await taskRetryMutationAllowed(tx as unknown as Db, { companyId: agent.companyId, issueId,
actorId: retry.actorId, actorSource: retry.actorSource })) return { kind: "deferred" as const };
}
if (opts.queuedCommentRequestId) {
const ids = await undeliveredLegacyUserCommentIds(tx as unknown as Db,
@@ -3004,11 +3001,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
if (!failedChatRetry || retry?.chatActionId !== durableRequest.id ||
retry.actorId !== taskRetryActorId || retry.sourceRunId !== durableRequest.failedRunRetry?.failedRunId)
throw new FailedChatRunRetryAuthorizationError();
const permission = await authorizationService(tx as unknown as Db).decide({
actor: { type: "board", userId: retry.actorId, source: retry.actorSource }, action: "issue:mutate",
resource: { type: "issue", companyId: agent.companyId, issueId },
});
if (!permission.allowed) return { kind: "deferred" as const };
if (!await taskRetryMutationAllowed(tx as unknown as Db, { companyId: agent.companyId, issueId,
actorId: retry.actorId, actorSource: retry.actorSource })) return { kind: "deferred" as const };
}
} else {
if (
@@ -3311,7 +3305,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`,
updatedAt: new Date(),
}).where(eq(agentWakeupRequests.id, executionWaitRequestId));
return { kind: "deferred" as const };
return { kind: "deferred" as const, admitted: true };
}
if (durableRequest || wakeCommentId ||
hasInteractionContinuationWakeContext(enrichedContextSnapshot) ||
@@ -3348,7 +3342,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
},
});
}
return { kind: "deferred" as const };
return { kind: "deferred" as const, admitted: true };
};
const explicitContinuationRunId = randomUUID();
const executionBlocker = await getExecutionBlocker(
@@ -3974,7 +3968,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
};
}
if (admission.kind === "deferred") {
return { kind: "deferred" as const };
return { kind: "deferred" as const, admitted: true };
}
// admission.kind === "proceed": no active run absorbed this wake,
// so fall through to the ordinary queue path below.
@@ -4357,7 +4351,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
// transitions to "running" — Fix A (lazy locking).
return { kind: "queued" as const, run: newRun };
})
}))
.catch((error) => {
if (isExternalChatWaitAuthorizationContention(error))
return { kind: "deferred" as const };
+15
View File
@@ -8,6 +8,20 @@ import { createSystemRecovery, readSystemRecovery } from "./system-recovery-stat
import { updateSystemRecovery } from "./system-recovery.js";
import { instanceSettingsService } from "./instance-settings.js";
import { notifyChatActionWork } from "./chat-work-notifications.js";
import { authorizationService } from "./authorization.js";
export async function taskRetryMutationAllowed(db: Db, input: {
companyId: string; issueId: string; actorId: string; actorSource: TaskRetryIntent["actorSource"];
}) {
const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId }).from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId)));
if (!issue) return false;
const permission = await authorizationService(db).decide({
actor: { type: "board", userId: input.actorId, source: input.actorSource }, action: "issue:mutate",
resource: { type: "issue", companyId: input.companyId, issueId: input.issueId, ...issue },
});
return permission.allowed;
}
export async function assertTaskRetryConversationAllowed(
db: Db,
@@ -115,6 +129,7 @@ export async function requestTaskRetry(input: {
const intent: TaskRetryIntent = { schema: "paperclip.task-retry.v1", generationId: active?.id ?? id,
sourceRunId: latest.id, actorId: input.actorId,
actorSource: taskRetryIntentSchema.shape.actorSource.parse(input.actorSource ?? "session"), strategy: "evaluate",
...(active && activeIntent?.chatActionId ? { chatActionId: activeIntent.chatActionId } : {}),
...(active ? { canonicalIntentId: active.id } : {}) };
// Preserve external chat authority through its existing admission owner.
// Generic board Retry must never manufacture a chat-inbound request.