mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
fix(github): retry and recover review publication
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
6cb6ac3b8e
commit
ce5e790174
16 files changed
+678
-45
No files matched your search
@@ -188,11 +188,8 @@ old runs cannot publish over the latest head. One current summary is updated in
|
||||
place, with history and task/run links retained. Stable finding keys prevent
|
||||
duplicate inline comments on repeated reviews.
|
||||
|
||||
GitHub bot tools allow up to 60 seconds by default for their provider requests
|
||||
and publication receipts. An explicit shorter tool deadline still applies.
|
||||
The assessment and publication intent are saved before publication. A timeout
|
||||
does not cancel that durable publication or prove that GitHub rejected it;
|
||||
check the review's publication links before starting another review.
|
||||
See [Publication recovery](#publication-recovery) for deadlines, retries, and
|
||||
lost-response receipt recovery.
|
||||
|
||||
The check's **Details** link opens its Paperclip task on the current instance
|
||||
hostname, or the connector's Reviews page when no task has been created yet.
|
||||
@@ -212,6 +209,37 @@ where supported. Paperclip does not change repository rules. GitHub account and
|
||||
repository plan restrictions may limit required-check enforcement. If automatic
|
||||
execution is disallowed, a gated head requests an authorized manual review.
|
||||
|
||||
## Publication recovery
|
||||
|
||||
GitHub bot tools have a sixty-second default deadline. A caller can request a
|
||||
shorter deadline. The assessment and publication intent are saved before provider
|
||||
writes, so a caller timeout does not prove that publication failed.
|
||||
|
||||
Temporary network errors, HTTP 408/429, and GitHub server errors use bounded
|
||||
backoff with jitter. Reads and installation-token issuance can retry within the
|
||||
call. Publication writes retry through the durable outbox, where App-owned
|
||||
comment markers and check identities recover lost responses. An unknown write
|
||||
is never repeated without checking its existing provider receipt.
|
||||
|
||||
The caller waits for up to three publication attempts within a forty-five-second
|
||||
retry window. Longer delays remain queued. A dedicated worker restores received
|
||||
actions and future retry deadlines after restart without another inbound event.
|
||||
The durable limit is eight failed attempts. Authorization, configuration, and
|
||||
the current PR head are checked again before dispatch. Permanent permission or
|
||||
validation failures, changed authority, and superseded reviews stop retries.
|
||||
|
||||
Rate-limit retries respect GitHub's `Retry-After` and primary reset headers;
|
||||
secondary limits without a deadline wait at least one minute. See
|
||||
[GitHub's retry guidance](https://docs.github.com/en/rest/using-the-rest-api/best-practices-for-using-the-rest-api#handle-rate-limit-errors-appropriately).
|
||||
|
||||
Repeat the exact `submit_review`, `comment`, or `update_comment` call to receive
|
||||
the current durable receipt after a timeout. Preserve the assessment and
|
||||
idempotency key; do not start another review to recover publication. A result
|
||||
with `status: failed` and `receipt.retryable: true`, or `status: received`, is
|
||||
still pending. Claim publication only after `status: processed`. Receipt recovery
|
||||
does not itself mark a task Done; the agent must finish its task after confirming
|
||||
the receipt.
|
||||
|
||||
## Hosted ingress
|
||||
|
||||
Dedicated Apps use Cloud as a sealed transport gateway. Cloud routes opaque
|
||||
|
||||
@@ -87,6 +87,17 @@ remaining operations before qualifying the whole capability.
|
||||
Gaps: Normal agent replies work through server delivery. Explicit endpoint publication
|
||||
APIs are board management paths.
|
||||
|
||||
GitHub publication recovery (2026-10-14): assigned-agent `submit_review`, `comment`,
|
||||
and `update_comment` calls reuse their run-, task-, endpoint-, and argument-bound
|
||||
publication on an exact retry. Fresh gateway policy and provider authority remain
|
||||
required. The durable worker resumes transient failures after restart and stops
|
||||
at its retry cap, revoked authority, or a superseded head. A pending receipt is
|
||||
not task completion. Source:
|
||||
[publication service](../../server/src/services/chat-github-reviews.ts) and
|
||||
[gateway](../../server/src/services/tool-gateway.ts). Verification uses
|
||||
PostgreSQL-backed provider fixtures; this entry does not claim live provider
|
||||
qualification for the new retry path.
|
||||
|
||||
Speko browser sessions use `POST /api/companies/{companyId}/voice-sessions`,
|
||||
with GET inspection/notification and POST `/{sessionId}/end` operations below
|
||||
that collection. These routes require board identity; ordinary agent credentials
|
||||
|
||||
@@ -58,6 +58,13 @@ For GitHub, submit a disposable PR and verify the summary heading uses the bot's
|
||||
display name and the check remains **Paperclip Review**. For two separate bots on one PR, verify
|
||||
their App-scoped checks and comments remain independent.
|
||||
|
||||
For GitHub publication recovery, interrupt a response after GitHub accepts a
|
||||
write and verify one summary, finding, and App-scoped check. Retry the same tool
|
||||
call and verify it returns the saved receipt. Leave a transient failure queued,
|
||||
restart the instance, and verify publication resumes at its saved deadline
|
||||
without another inbound message. Revoked access and permanent rejections must
|
||||
stop retries. These recipes are separate from live provider proof.
|
||||
|
||||
Configure optional project and workspace defaults at the connection and destination
|
||||
levels. Start a new conversation and inspect its task workspace. Verify that
|
||||
inheritance, an explicit selection, and clearing a default remain distinct.
|
||||
|
||||
@@ -15,7 +15,7 @@ import { fastResponseService } from "../services/fast-responses.js";
|
||||
import { fastResponseReceipt } from "../services/fast-response-provider.js";
|
||||
import { fastResponseRequests } from "@paperclipai/db";
|
||||
import { toolAccessService } from "../services/tool-access.js";
|
||||
import { toolActionRequests, toolInvocations } from "@paperclipai/db";
|
||||
import { toolActionRequests, toolInvocations, toolPolicies } from "@paperclipai/db";
|
||||
import { GitHubPublicationLeaseLost, withGitHubPublicationLease } from "../services/chat-github-publication-lease.js";
|
||||
import { secretService } from "../services/secrets.js";
|
||||
import { githubChatWizardService } from "../services/chat-github-wizard.js";
|
||||
@@ -2555,10 +2555,199 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
const session = { companyId: f.companyId, agentId: f.assignedAgentId, issueId: conversation.issueId, runId: run.id };
|
||||
return { ...f, conversation, run, session };
|
||||
}
|
||||
async function publicationRetryFixture() {
|
||||
const f = await toolReplyFixture("github:paperclipai/paperclip:418", true);
|
||||
const head = "b".repeat(40);
|
||||
const inline: Array<{ id: number; body: string; html_url: string; user: { login: string } }> = [];
|
||||
const checks: Array<{ id: number; external_id: string; app: { id: number }; html_url: string }> = [];
|
||||
const writes: Array<{ path: string; method: string }> = [];
|
||||
const provider = async (input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const url = String(input);
|
||||
if (url.endsWith("/pulls/418")) return Response.json({ number: 418, state: "open", draft: false,
|
||||
head: { sha: head }, base: { sha: "a".repeat(40), ref: "master" }, user: { id: 42, login: "octocat", type: "User" }, labels: [] });
|
||||
if (url.includes("/files?")) return Response.json([{ filename: "src/math.ts", status: "modified", patch: "@@ -1 +1 @@\n-old\n+new" }]);
|
||||
if (url.includes("/pulls/418/comments?")) return Response.json(inline);
|
||||
if (url.includes("/check-runs?")) return Response.json({ check_runs: checks });
|
||||
if (init?.method === "POST" && url.endsWith("/pulls/418/comments")) {
|
||||
writes.push({ path: url, method: init.method });
|
||||
const item = { id: 420 + inline.length, body: JSON.parse(String(init.body)).body,
|
||||
html_url: "https://github.com/acme/repo/pull/418#discussion_r420", user: { login: f.endpoint.botUsername! } };
|
||||
inline.push(item);
|
||||
return Response.json(item);
|
||||
}
|
||||
if (url.includes("/check-runs") && init?.method) {
|
||||
writes.push({ path: url, method: init.method });
|
||||
const body = JSON.parse(String(init.body));
|
||||
const check = checks[0] ?? { id: 419, external_id: body.external_id, app: { id: Number(f.endpoint.botExternalId) }, html_url: "https://github.com/checks/419" };
|
||||
if (!checks.length) checks.push(check);
|
||||
return Response.json(check);
|
||||
}
|
||||
return undefined;
|
||||
};
|
||||
f.setSupplementalProviderFetch(provider);
|
||||
const service = githubChatReviewService(db, f.providerFetch);
|
||||
await service.execute(f.session, "begin_review", { reviewedCommit: head });
|
||||
const assessment = { reviewedCommit: head, score: 3, complete: true, summary: "Advisory review", rationale: "Reviewed the file",
|
||||
coverage: { reviewedPaths: ["src/math.ts"], omittedPaths: [], limitations: [] }, findings: [] };
|
||||
const action = async () => (await db.select().from(chatActions).where(and(
|
||||
eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"),
|
||||
)))[0];
|
||||
return { ...f, service, head, assessment, provider, action, inline, checks, writes };
|
||||
}
|
||||
it("recovers lost inline and check responses without duplicate GitHub publications", async () => {
|
||||
const f = await publicationRetryFixture();
|
||||
const lost = new Set<string>();
|
||||
f.setSupplementalProviderFetch(async (input, init) => {
|
||||
const response = await f.provider(input, init);
|
||||
if (init?.method === "POST" && /\/(?:comments|check-runs)$/.test(String(input)) && !lost.has(String(input))) {
|
||||
lost.add(String(input));
|
||||
throw new Error("Provider accepted the write but the response was lost");
|
||||
}
|
||||
return response;
|
||||
});
|
||||
const result = await f.service.execute(f.session, "submit_review", { ...f.assessment,
|
||||
findings: [{ key: "note", path: "src/math.ts", line: 1, side: "RIGHT", severity: "error", category: "correctness", body: "Example finding" }],
|
||||
}, "lost-receipts");
|
||||
expect(result).toMatchObject({ status: "processed", score: 3, conclusion: "neutral" });
|
||||
expect(f.inline).toHaveLength(1);
|
||||
expect(f.checks).toHaveLength(1);
|
||||
expect(f.writes.filter(write => write.method === "POST")).toHaveLength(2);
|
||||
expect(f.responseComments.size).toBe(1);
|
||||
expect((await f.action()).status).toBe("processed");
|
||||
}, 25_000);
|
||||
it("resumes a delayed GitHub publication after coordinator restart without another inbound event", async () => {
|
||||
const f = await publicationRetryFixture();
|
||||
let unavailable = true;
|
||||
let attempts = 0;
|
||||
f.setSupplementalProviderFetch(async (input, init) => {
|
||||
if (String(input).includes("/issues/comments/") && init?.method === "PATCH") {
|
||||
attempts++;
|
||||
if (unavailable) return new Response("Unavailable", { status: 503, headers: { "retry-after": "120" } });
|
||||
}
|
||||
return f.provider(input, init);
|
||||
});
|
||||
const result = await f.service.execute(f.session, "submit_review", f.assessment, "scheduled-review");
|
||||
expect(result).toMatchObject({ status: "failed", receipt: { retryable: true, attempts: 1 } });
|
||||
const action = await f.action();
|
||||
expect(await f.service.nextPendingAt()).toBe(Date.parse(String(action.result?.retryAt)));
|
||||
expect(await f.service.processPending()).toBe(0);
|
||||
const due = new Date(Date.now() + 1500);
|
||||
await db.update(chatActions).set({ result: { ...action.result, retryAt: due.toISOString() } }).where(eq(chatActions.id, action.id));
|
||||
unavailable = false;
|
||||
const { createDeliveryWorkCoordinator } = await import("../services/delivery-work-coordinator.js");
|
||||
const { registerChatQueueWork } = await import("../services/chat-delivery-work.js");
|
||||
const { DELIVERY_QUEUES } = await import("../services/delivery-work-notifications.js");
|
||||
const restarted = githubChatReviewService(db, f.providerFetch);
|
||||
const errors = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: errors });
|
||||
try {
|
||||
await registerChatQueueWork(coordinator, { queue: DELIVERY_QUEUES.chatGitHubPublications,
|
||||
run: restarted.processPending, next: restarted.nextPendingAt }, () => true).ready;
|
||||
expect(coordinator.nextWakeAt()).toBe(due.getTime());
|
||||
expect(attempts).toBe(1);
|
||||
await expect.poll(async () => (await f.action()).status, { timeout: 10_000 }).toBe("processed");
|
||||
expect(attempts).toBe(2);
|
||||
expect(errors).not.toHaveBeenCalled();
|
||||
expect(await restarted.nextPendingAt()).toBeNull();
|
||||
expect(f.responseComments.size).toBe(1);
|
||||
} finally {
|
||||
await coordinator.stop();
|
||||
}
|
||||
}, 25_000);
|
||||
it.each([403, 422])("does not resubmit a permanent GitHub publication rejection (HTTP %s)", async status => {
|
||||
const f = await publicationRetryFixture();
|
||||
let attempts = 0;
|
||||
f.setSupplementalProviderFetch(async (input, init) => {
|
||||
if (String(input).includes("/issues/comments/") && init?.method === "PATCH") {
|
||||
attempts++;
|
||||
return new Response("Rejected", { status });
|
||||
}
|
||||
return f.provider(input, init);
|
||||
});
|
||||
const result = await f.service.execute(f.session, "submit_review", f.assessment, "permanent-failure");
|
||||
expect(result).toMatchObject({ status: "failed", receipt: { retryable: false, attempts: 1, providerStatus: status } });
|
||||
await f.service.execute(f.session, "submit_review", f.assessment, "permanent-failure");
|
||||
expect(await f.service.processPending()).toBe(0);
|
||||
expect(await f.service.nextPendingAt()).toBeNull();
|
||||
expect(attempts).toBe(1);
|
||||
});
|
||||
it.each(["revoked", "exhausted"] as const)("stops a queued GitHub publication after %s", async mode => {
|
||||
const f = await publicationRetryFixture();
|
||||
let attempts = 0;
|
||||
f.setSupplementalProviderFetch(async (input, init) => {
|
||||
if (String(input).includes("/issues/comments/") && init?.method === "PATCH") {
|
||||
attempts++;
|
||||
return new Response("Unavailable", { status: 503, headers: { "retry-after": "120" } });
|
||||
}
|
||||
return f.provider(input, init);
|
||||
});
|
||||
await f.service.execute(f.session, "submit_review", f.assessment, "bounded-retry");
|
||||
const action = await f.action();
|
||||
await db.update(chatActions).set({ result: { ...action.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, action.id));
|
||||
if (mode === "revoked") await db.update(chatIdentityLinks).set({ status: "revoked" }).where(eq(chatIdentityLinks.endpointId, f.endpoint.id));
|
||||
for (let attempt = 1; attempt < (mode === "exhausted" ? 8 : 2); attempt++) {
|
||||
const current = await f.action();
|
||||
await db.update(chatActions).set({ result: { ...current.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, current.id));
|
||||
await f.service.processPending();
|
||||
}
|
||||
await f.service.processPending();
|
||||
expect((await f.action()).result?.retryable).toBe(false);
|
||||
expect((await f.action()).status).toBe(mode === "revoked" ? "cancelled" : "failed");
|
||||
expect(await f.service.nextPendingAt()).toBeNull();
|
||||
expect(attempts).toBe(mode === "exhausted" ? 8 : 1);
|
||||
});
|
||||
it.each([503, 429])("automatically retries a transient GitHub review publication (HTTP %s)", async providerStatus => {
|
||||
const f = await toolReplyFixture("github:paperclipai/paperclip:418", true);
|
||||
const head = "b".repeat(40);
|
||||
let summaryAttempts = 0;
|
||||
const checks: Record<string, unknown>[] = [];
|
||||
f.setSupplementalProviderFetch(async (input, init) => {
|
||||
const url = String(input);
|
||||
if (url.endsWith("/pulls/418")) return Response.json({ number: 418, state: "open", draft: false,
|
||||
head: { sha: head }, base: { sha: "a".repeat(40), ref: "master" }, user: { id: 42, login: "octocat", type: "User" }, labels: [] });
|
||||
if (url.includes("/files?")) return Response.json([{ filename: "src/math.ts", status: "modified", patch: "@@ -1 +1 @@\n-old\n+new" }]);
|
||||
if (url.includes("/check-runs?")) return Response.json({ check_runs: [] });
|
||||
if (url.includes("/check-runs") && init?.method) {
|
||||
checks.push(JSON.parse(String(init.body)));
|
||||
return Response.json({ id: 419, html_url: "https://github.com/checks/419" });
|
||||
}
|
||||
if (url.includes("/issues/comments/") && init?.method === "PATCH" && ++summaryAttempts === 1)
|
||||
return new Response("Temporary provider failure", { status: providerStatus, headers: { "retry-after": "1" } });
|
||||
return undefined;
|
||||
});
|
||||
vi.stubGlobal("fetch", f.providerFetch);
|
||||
onTestFinished(() => { vi.unstubAllGlobals(); });
|
||||
const { createToolGatewayService } = await import("../services/tool-gateway.js");
|
||||
const gateway = createToolGatewayService(db);
|
||||
const session = await gateway.createSession({ companyId: f.companyId, agentId: f.assignedAgentId, runId: f.run.id });
|
||||
const tool = (await githubBotToolsForSession(db, f.session)).find(tool => tool.upstreamToolName === "submit_review")!;
|
||||
await githubChatReviewService(db, f.providerFetch).execute(f.session, "begin_review", { reviewedCommit: head });
|
||||
const assessment = { reviewedCommit: head, score: 3, complete: true, summary: "Advisory review", rationale: "Reviewed the file",
|
||||
coverage: { reviewedPaths: ["src/math.ts"], omittedPaths: [], limitations: [] }, findings: [] };
|
||||
const idempotencyKey = randomUUID();
|
||||
const result = await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey });
|
||||
expect(result).toMatchObject({ status: "completed", result: { data: { status: "processed", score: 3, conclusion: "neutral" } } });
|
||||
expect(summaryAttempts).toBe(2);
|
||||
expect(f.responseComments.size).toBe(1);
|
||||
expect(checks).toEqual([expect.objectContaining({ conclusion: "neutral", status: "completed" })]);
|
||||
const replay = await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey });
|
||||
expect(replay).toMatchObject({ result: { data: { status: "processed", score: 3, conclusion: "neutral" } } });
|
||||
expect(summaryAttempts).toBe(2);
|
||||
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication")));
|
||||
expect(action).toMatchObject({ status: "processed" });
|
||||
expect((await readChatGitHubReviewsByEndpointId(f.endpoint.id))).toHaveLength(1);
|
||||
await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: { ...assessment, score: 5 }, idempotencyKey }))
|
||||
.rejects.toMatchObject({ status: 409, reasonCode: "github_idempotency_conflict" });
|
||||
await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Revoke review publication", policyType: "block", priority: 0,
|
||||
selectors: { catalogEntryId: tool.catalogEntryId! },
|
||||
});
|
||||
await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey })).rejects.toThrow();
|
||||
expect(summaryAttempts).toBe(2);
|
||||
}, 25_000);
|
||||
it.each([
|
||||
{ label: "returns the confirmed GitHub assessment when publication takes longer than ten seconds", delayMs: 11_000, timeoutMs: undefined },
|
||||
{ label: "preserves an explicit shorter GitHub tool deadline and reconciles without duplicate publication", delayMs: 200, timeoutMs: 50 },
|
||||
])("$label", async ({ delayMs, timeoutMs }) => {
|
||||
])("recovers a durable GitHub publication receipt ($label)", async ({ delayMs, timeoutMs }) => {
|
||||
const f = await toolReplyFixture("github:paperclipai/paperclip:418", true);
|
||||
const head = "b".repeat(40);
|
||||
const checks: Record<string, unknown>[] = [];
|
||||
@@ -2602,7 +2791,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
expect(result).toMatchObject({ error: { status: 504, reasonCode: "tool_timeout" } });
|
||||
}
|
||||
const writes = f.responseWrites.length;
|
||||
await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment });
|
||||
const replay = await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment });
|
||||
expect(replay).toMatchObject({ result: { data: { status: "processed", score: 5, conclusion: "neutral" } } });
|
||||
const [invocation] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, replay.invocationId));
|
||||
expect(invocation).toMatchObject({ status: "succeeded", errorCode: null, errorMessage: null });
|
||||
expect(f.responseWrites).toHaveLength(writes);
|
||||
}, 40_000);
|
||||
it("does not post a GitHub working comment when the scheduler declines the request", async () => {
|
||||
|
||||
@@ -3,6 +3,7 @@ import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signal
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
|
||||
import { notifyChatDeliveryWork, notifyChatPublicationWork } from "../services/chat-work-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
|
||||
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
|
||||
afterEach(async () => {
|
||||
@@ -27,6 +28,8 @@ function setup() {
|
||||
nextInboundDeliveryAt: vi.fn(async (): Promise<number | null> => null),
|
||||
nextPublicationAt: vi.fn(async (): Promise<number | null> => null),
|
||||
nextSlackReceiptAt: vi.fn(async (): Promise<number | null> => null),
|
||||
processPendingGitHubReviewPublications: vi.fn(async () => {}),
|
||||
nextGitHubReviewPublicationAt: vi.fn(async (): Promise<number | null> => null),
|
||||
onPublicationsSettled: vi.fn(),
|
||||
};
|
||||
const start = () => registerChatDeliveryWork(coordinator, service, enabled);
|
||||
@@ -38,7 +41,7 @@ describe("chat delivery work", () => {
|
||||
const s = setup();
|
||||
await s.start().ready;
|
||||
await vi.advanceTimersByTimeAsync(3_600_000);
|
||||
for (const run of [s.service.processQueuedDeliveries, s.service.scheduleQueuedPublications, s.service.processPendingSlackFileUploadReceipts]) expect(run).toHaveBeenCalledTimes(1);
|
||||
for (const run of [s.service.processQueuedDeliveries, s.service.scheduleQueuedPublications, s.service.processPendingSlackFileUploadReceipts, s.service.processPendingGitHubReviewPublications]) expect(run).toHaveBeenCalledTimes(1);
|
||||
expect(s.coordinator.nextWakeAt()).toBeNull();
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
@@ -100,6 +103,23 @@ describe("chat delivery work", () => {
|
||||
expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("restores GitHub retry deadlines on startup and wakes after a publication commit", async () => {
|
||||
const s = setup();
|
||||
const due = Date.now() + 60_000;
|
||||
s.service.nextGitHubReviewPublicationAt.mockResolvedValue(due);
|
||||
await s.start().ready;
|
||||
expect(s.coordinator.nextWakeAt()).toBe(due);
|
||||
await vi.advanceTimersByTimeAsync(59_999);
|
||||
expect(s.service.processPendingGitHubReviewPublications).toHaveBeenCalledTimes(1);
|
||||
s.service.nextGitHubReviewPublicationAt.mockResolvedValue(null);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.service.processPendingGitHubReviewPublications).toHaveBeenCalledTimes(2);
|
||||
await s.owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications));
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(s.service.processPendingGitHubReviewPublications).toHaveBeenCalledTimes(3);
|
||||
expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("retries failed selection without losing its startup recovery", async () => {
|
||||
const s = setup();
|
||||
s.service.nextSlackReceiptAt.mockRejectedValueOnce(new Error("connection lost"));
|
||||
|
||||
@@ -55,6 +55,7 @@ export { createChatSdkStatePersistence } from "./chat-channels/sdk-state.js";
|
||||
import { createChatInteractionCallbacks } from "./chat-channels/interactions/callbacks.js";
|
||||
import { createChatInteractionContinuations } from "./chat-channels/interactions/continuations.js";
|
||||
import { createGitHubInstallationLifecycle } from "./chat-channels/github/installation-lifecycle.js";
|
||||
import { githubChatReviewService } from "./chat-github-reviews.js";
|
||||
import { createGitHubAppSetup } from "./chat-channels/github/app-setup.js";
|
||||
import { createSlackTaskAuthorityRegistration } from "./chat-channels/slack/task-authority.js";
|
||||
import { createGitHubPublicationPolicy } from "./chat-channels/github/publication-policy.js";
|
||||
@@ -3370,6 +3371,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
return publicationQueue.processPendingPublications(...args);
|
||||
}
|
||||
|
||||
const githubReviews = githubChatReviewService(db, fetchImpl);
|
||||
const { scheduleGitHubMaintenance } = createGitHubMaintenance(db, fetchImpl, {
|
||||
scheduleMessageProcessing, isShuttingDown: () => shuttingDown,
|
||||
});
|
||||
@@ -3592,6 +3594,8 @@ return {
|
||||
processPublicationMaintenance,
|
||||
nextInboundDeliveryAt,
|
||||
nextPublicationAt,
|
||||
processPendingGitHubReviewPublications: githubReviews.processPending,
|
||||
nextGitHubReviewPublicationAt: githubReviews.nextPendingAt,
|
||||
nextSlackReceiptAt,
|
||||
onPublicationsSettled: (wake: () => void) => { wakePublications = wake; },
|
||||
prepareFailedChatRunRetry,
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { githubChatReviewService } from "../../chat-github-reviews.js";
|
||||
import { githubResponseCommentService } from "../../chat-github-response-comments.js";
|
||||
import { githubReviewCheckService } from "../../chat-github-checks.js";
|
||||
|
||||
@@ -14,7 +13,6 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet
|
||||
options.scheduleMessageProcessing(async () => {
|
||||
try {
|
||||
// Provider I/O must not block Slack, Teams, or ordinary message receipts.
|
||||
await githubChatReviewService(db, fetchImpl).processPending();
|
||||
await githubResponseCommentService(db, fetchImpl).processPending();
|
||||
await githubReviewCheckService(db, fetchImpl).processPending();
|
||||
} finally {
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { githubBotRequest } from "../../chat-github-client.js";
|
||||
import { forbidden, unprocessable } from "../../../errors.js";
|
||||
import { githubProviderRetry, githubPublicationRetry } from "./retry.js";
|
||||
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
describe("GitHub publication retries", () => {
|
||||
it.each([408, 500, 502, 503, 504])(
|
||||
"retries a safe GitHub read after HTTP %s",
|
||||
async (status) => {
|
||||
vi.useFakeTimers();
|
||||
const provider = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(new Response("Unavailable", { status }))
|
||||
.mockResolvedValueOnce(Response.json({ head: "current" }));
|
||||
const request = githubBotRequest(
|
||||
provider,
|
||||
"private-token",
|
||||
"/repos/acme/repo/pulls/1",
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(2500);
|
||||
await expect(request).resolves.toEqual({ head: "current" });
|
||||
expect(provider).toHaveBeenCalledTimes(2);
|
||||
},
|
||||
);
|
||||
it("retries installation token issuance without retrying arbitrary provider writes", async () => {
|
||||
vi.useFakeTimers();
|
||||
const provider = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce(new Response("Unavailable", { status: 503 }))
|
||||
.mockResolvedValueOnce(Response.json({ token: "issued" }));
|
||||
const request = githubBotRequest(
|
||||
provider,
|
||||
null,
|
||||
"/app/installations/42/access_tokens",
|
||||
{ method: "POST" },
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(2500);
|
||||
await expect(request).resolves.toEqual({ token: "issued" });
|
||||
provider.mockResolvedValue(new Response("Unavailable", { status: 503 }));
|
||||
await expect(
|
||||
githubBotRequest(provider, null, "/repos/acme/repo/pulls/1/comments", {
|
||||
method: "POST",
|
||||
}),
|
||||
).rejects.toMatchObject({
|
||||
details: { providerStatus: 503, retryable: true },
|
||||
});
|
||||
expect(provider).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
it.each([401, 403, 404, 422])(
|
||||
"does not retry permission or validation rejection HTTP %s",
|
||||
async (status) => {
|
||||
const provider = vi.fn(async () =>
|
||||
Response.json({ message: "Rejected private-token" }, { status }),
|
||||
);
|
||||
const error = await githubBotRequest(
|
||||
provider,
|
||||
"private-token",
|
||||
"/repos/acme/repo/pulls/1",
|
||||
).catch((error) => error);
|
||||
expect(error).toMatchObject({
|
||||
details: { providerStatus: status, retryable: false },
|
||||
});
|
||||
expect(String(error)).not.toContain("private-token");
|
||||
expect(provider).toHaveBeenCalledTimes(1);
|
||||
},
|
||||
);
|
||||
it("honors provider retry and primary rate-limit reset deadlines", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-10-14T12:00:00Z"));
|
||||
expect(
|
||||
githubProviderRetry(
|
||||
new Response(null, { status: 429, headers: { "retry-after": "120" } }),
|
||||
),
|
||||
).toMatchObject({ retryable: true, retryAfterMs: 120_000 });
|
||||
expect(
|
||||
githubProviderRetry(
|
||||
new Response(null, {
|
||||
status: 403,
|
||||
headers: {
|
||||
"x-ratelimit-remaining": "0",
|
||||
"x-ratelimit-reset": String((Date.now() + 300_000) / 1000),
|
||||
},
|
||||
}),
|
||||
),
|
||||
).toMatchObject({ retryable: true, retryAfterMs: 300_000 });
|
||||
expect(
|
||||
githubProviderRetry(
|
||||
new Response(null, {
|
||||
status: 503,
|
||||
headers: { "retry-after": "Wed, 14 Oct 2026 12:02:00 GMT" },
|
||||
}),
|
||||
),
|
||||
).toMatchObject({ retryAfterMs: 120_000 });
|
||||
});
|
||||
it("defers a secondary rate limit beyond the inline window without sleeping or resending", async () => {
|
||||
const provider = vi.fn(async () =>
|
||||
Response.json(
|
||||
{ message: "You have exceeded a secondary rate limit" },
|
||||
{ status: 403 },
|
||||
),
|
||||
);
|
||||
await expect(
|
||||
githubBotRequest(provider, null, "/repos/acme/repo/pulls/1"),
|
||||
).rejects.toMatchObject({
|
||||
details: {
|
||||
retryable: true,
|
||||
retryAfterMs: 60_000,
|
||||
code: "github_bot_rate_limited",
|
||||
},
|
||||
});
|
||||
expect(provider).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
it.each(["headers", "body", "json"])(
|
||||
"bounds retries after an interrupted %s response",
|
||||
async (phase) => {
|
||||
vi.useFakeTimers();
|
||||
const provider = vi.fn(async () => {
|
||||
if (phase === "headers")
|
||||
throw new Error("private-token transport failed");
|
||||
if (phase === "json") return new Response("{truncated");
|
||||
return new Response(
|
||||
new ReadableStream({
|
||||
pull(controller) {
|
||||
controller.error(new Error("private-token body interrupted"));
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
const error = githubBotRequest(
|
||||
provider,
|
||||
"private-token",
|
||||
"/repos/acme/repo/pulls/1",
|
||||
).catch((error) => error);
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
expect(await error).toMatchObject({
|
||||
status: 422,
|
||||
details: { retryable: true },
|
||||
});
|
||||
expect(String(await error)).not.toContain("private-token");
|
||||
expect(provider).toHaveBeenCalledTimes(3);
|
||||
},
|
||||
);
|
||||
it("stops durable retries at the attempt cap and for revoked authority", () => {
|
||||
const temporary = unprocessable("Unavailable", {
|
||||
retryable: true,
|
||||
providerStatus: 503,
|
||||
retryAfterMs: 120_000,
|
||||
});
|
||||
const before = Date.now();
|
||||
expect(githubPublicationRetry(temporary, 7).retryable).toBe(true);
|
||||
expect(
|
||||
Date.parse(githubPublicationRetry(temporary, 7).retryAt),
|
||||
).toBeGreaterThanOrEqual(before + 120_000);
|
||||
expect(githubPublicationRetry(temporary, 8).retryable).toBe(false);
|
||||
expect(githubPublicationRetry(forbidden("Revoked"), 1).retryable).toBe(
|
||||
false,
|
||||
);
|
||||
expect(
|
||||
githubPublicationRetry(unprocessable("Invalid assessment"), 1).retryable,
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,79 @@
|
||||
import { HttpError } from "../../../errors.js";
|
||||
|
||||
export const GITHUB_PUBLICATION_MAX_ATTEMPTS = 8;
|
||||
export const GITHUB_INLINE_RETRY_WINDOW_MS = 45_000;
|
||||
export const GITHUB_INLINE_MAX_ATTEMPTS = 3;
|
||||
|
||||
export function githubRetryDelay(attempt: number) {
|
||||
return (
|
||||
Math.min(300_000, 1000 * 2 ** attempt) + Math.floor(Math.random() * 250)
|
||||
);
|
||||
}
|
||||
|
||||
export function githubProviderRetry(
|
||||
response: Response,
|
||||
secondaryRateLimit = false,
|
||||
) {
|
||||
const rateLimited =
|
||||
response.status === 429 ||
|
||||
(response.status === 403 &&
|
||||
(response.headers.has("retry-after") ||
|
||||
response.headers.get("x-ratelimit-remaining") === "0" ||
|
||||
secondaryRateLimit));
|
||||
const retryable =
|
||||
rateLimited || response.status === 408 || response.status >= 500;
|
||||
const value = response.headers.get("retry-after");
|
||||
const seconds = value === null ? NaN : Number(value);
|
||||
const retryAfter = value === null ? NaN : Date.parse(value);
|
||||
const reset = Number(response.headers.get("x-ratelimit-reset"));
|
||||
const retryAfterMs =
|
||||
Number.isFinite(seconds) && seconds >= 0
|
||||
? seconds * 1000
|
||||
: Number.isFinite(retryAfter)
|
||||
? Math.max(0, retryAfter - Date.now())
|
||||
: 0;
|
||||
const resetMs =
|
||||
response.headers.get("x-ratelimit-remaining") === "0" &&
|
||||
Number.isFinite(reset)
|
||||
? Math.max(0, reset * 1000 - Date.now())
|
||||
: 0;
|
||||
return {
|
||||
code: rateLimited
|
||||
? "github_bot_rate_limited"
|
||||
: "github_bot_operation_failed",
|
||||
providerStatus: response.status,
|
||||
retryable,
|
||||
retryAfterMs: Math.max(
|
||||
retryAfterMs,
|
||||
resetMs,
|
||||
rateLimited && !retryAfterMs && !resetMs ? 60_000 : 0,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
export function githubPublicationRetry(error: unknown, attempts: number) {
|
||||
const details =
|
||||
error instanceof HttpError &&
|
||||
error.details &&
|
||||
typeof error.details === "object"
|
||||
? (error.details as {
|
||||
retryable?: boolean;
|
||||
retryAfterMs?: number;
|
||||
providerStatus?: number;
|
||||
})
|
||||
: null;
|
||||
const retryable =
|
||||
attempts < GITHUB_PUBLICATION_MAX_ATTEMPTS &&
|
||||
(details?.retryable ?? !(error instanceof HttpError && error.status < 500));
|
||||
return {
|
||||
attempts,
|
||||
retryable,
|
||||
retryAt: new Date(
|
||||
Date.now() +
|
||||
Math.max(githubRetryDelay(attempts), details?.retryAfterMs ?? 0),
|
||||
).toISOString(),
|
||||
...(details?.providerStatus
|
||||
? { providerStatus: details.providerStatus }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
@@ -8,6 +8,8 @@ type ChatDeliveryService = {
|
||||
nextInboundDeliveryAt(): Promise<number | null>;
|
||||
nextPublicationAt(): Promise<number | null>;
|
||||
nextSlackReceiptAt(): Promise<number | null>;
|
||||
processPendingGitHubReviewPublications(): Promise<unknown>;
|
||||
nextGitHubReviewPublicationAt(): Promise<number | null>;
|
||||
onPublicationsSettled(wake: () => void): void;
|
||||
};
|
||||
|
||||
@@ -21,6 +23,7 @@ export function registerChatDeliveryWork(
|
||||
{ queue: DELIVERY_QUEUES.chatInbound, run: service.processQueuedDeliveries, next: service.nextInboundDeliveryAt },
|
||||
{ queue: DELIVERY_QUEUES.chatPublications, run: service.scheduleQueuedPublications, next: service.nextPublicationAt },
|
||||
{ queue: DELIVERY_QUEUES.chatReceipts, run: service.processPendingSlackFileUploadReceipts, next: service.nextSlackReceiptAt },
|
||||
{ queue: DELIVERY_QUEUES.chatGitHubPublications, run: service.processPendingGitHubReviewPublications, next: service.nextGitHubReviewPublicationAt },
|
||||
];
|
||||
const workers = tasks.map(task => {
|
||||
const worker = registerChatQueueWork(coordinator, task, canRun);
|
||||
|
||||
@@ -3,6 +3,7 @@ import { and, eq } from "drizzle-orm";
|
||||
import { chatEndpoints, toolConnections, type Db } from "@paperclipai/db";
|
||||
import { secretService } from "./secrets.js";
|
||||
import { conflict, forbidden, unprocessable } from "../errors.js";
|
||||
import { githubPublicationRetry, githubProviderRetry, GITHUB_INLINE_RETRY_WINDOW_MS, GITHUB_INLINE_MAX_ATTEMPTS } from "./chat-channels/github/retry.js";
|
||||
|
||||
export function githubAppJwt(
|
||||
appId: string,
|
||||
@@ -25,6 +26,28 @@ export async function githubBotRequest<T>(
|
||||
token: string | null,
|
||||
path: string,
|
||||
options: { method?: string; body?: unknown; accept?: string } = {},
|
||||
): Promise<T> {
|
||||
const deadline = Date.now() + GITHUB_INLINE_RETRY_WINDOW_MS;
|
||||
const safeToRetry = (options.method ?? "GET") === "GET" ||
|
||||
(options.method === "POST" && /^\/app\/installations\/[1-9][0-9]*\/access_tokens$/.test(path));
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
try {
|
||||
return await githubBotRequestOnce<T>(fetchImpl, token, path, options, Math.min(25_000, Math.max(1, deadline - Date.now())));
|
||||
} catch (error) {
|
||||
const retry = githubPublicationRetry(error, attempt);
|
||||
const delay = Date.parse(retry.retryAt) - Date.now();
|
||||
if (!safeToRetry || !retry.retryable || attempt >= GITHUB_INLINE_MAX_ATTEMPTS || Date.now() + delay >= deadline) throw error;
|
||||
await new Promise(resolve => setTimeout(resolve, delay));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function githubBotRequestOnce<T>(
|
||||
fetchImpl: typeof fetch,
|
||||
token: string | null,
|
||||
path: string,
|
||||
options: { method?: string; body?: unknown; accept?: string },
|
||||
timeoutMs: number,
|
||||
): Promise<T> {
|
||||
if (
|
||||
!path.startsWith("/") ||
|
||||
@@ -38,7 +61,7 @@ export async function githubBotRequest<T>(
|
||||
response = await fetchImpl(`https://api.github.com${path}`, {
|
||||
method: options.method ?? "GET",
|
||||
redirect: "error",
|
||||
signal: AbortSignal.timeout(25_000),
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
headers: {
|
||||
accept: options.accept ?? "application/vnd.github+json",
|
||||
"x-github-api-version": "2022-11-28",
|
||||
@@ -54,18 +77,39 @@ export async function githubBotRequest<T>(
|
||||
} catch {
|
||||
throw unprocessable(
|
||||
"GitHub is temporarily unavailable. Retry this operation.",
|
||||
{ code: "github_bot_temporarily_unavailable", retryable: true },
|
||||
);
|
||||
}
|
||||
if (!response.ok) {
|
||||
let secondaryRateLimit = false;
|
||||
if (response.status === 403) {
|
||||
const reader = response.body?.getReader();
|
||||
try {
|
||||
const chunks: Uint8Array[] = [];
|
||||
let size = 0;
|
||||
while (reader) {
|
||||
const part = await reader.read();
|
||||
if (part.done) break;
|
||||
size += part.value.byteLength;
|
||||
if (size > 8192) break;
|
||||
chunks.push(part.value);
|
||||
}
|
||||
secondaryRateLimit = /secondary rate limit|rate limit exceeded|abuse detection/i.test(Buffer.concat(chunks).toString("utf8"));
|
||||
} catch {
|
||||
} finally {
|
||||
await reader?.cancel().catch(() => {});
|
||||
reader?.releaseLock();
|
||||
}
|
||||
}
|
||||
await response.body?.cancel();
|
||||
throw unprocessable(
|
||||
`GitHub rejected this operation (HTTP ${response.status}). Check the App installation and repository permissions.`,
|
||||
{ code: "github_bot_operation_failed", providerStatus: response.status },
|
||||
`GitHub rejected this operation (HTTP ${response.status}).`,
|
||||
githubProviderRetry(response, secondaryRateLimit),
|
||||
);
|
||||
}
|
||||
if (response.status === 204) return undefined as T;
|
||||
const reader = response.body?.getReader();
|
||||
if (!reader) throw unprocessable("GitHub returned an empty response");
|
||||
if (!reader) throw unprocessable("GitHub returned an empty response", { code: "github_bot_invalid_response", retryable: true });
|
||||
const chunks: Uint8Array[] = [];
|
||||
let size = 0;
|
||||
try {
|
||||
@@ -80,7 +124,14 @@ export async function githubBotRequest<T>(
|
||||
chunks.push(part.value);
|
||||
}
|
||||
const body = Buffer.concat(chunks).toString("utf8");
|
||||
return (options.accept?.includes("diff") ? body : JSON.parse(body)) as T;
|
||||
try {
|
||||
return (options.accept?.includes("diff") ? body : JSON.parse(body)) as T;
|
||||
} catch {
|
||||
throw unprocessable("GitHub returned an invalid response", { code: "github_bot_invalid_response", retryable: true });
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof Error && "status" in error) throw error;
|
||||
throw unprocessable("GitHub response was interrupted. Retry this operation.", { code: "github_bot_temporarily_unavailable", retryable: true });
|
||||
} finally {
|
||||
await reader.cancel().catch(() => {});
|
||||
reader.releaseLock();
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, gt, lte, sql } from "drizzle-orm";
|
||||
import { chatEndpointLeases, type Db } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
const TTL_SECONDS = 90;
|
||||
@@ -113,13 +114,9 @@ export async function withGitHubPublicationLease<T>(
|
||||
} finally {
|
||||
clearInterval(timer);
|
||||
await renewal;
|
||||
await db
|
||||
.delete(chatEndpointLeases)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointLeases.id, claimed.id),
|
||||
eq(chatEndpointLeases.token, token),
|
||||
),
|
||||
);
|
||||
await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
|
||||
await tx.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.id, claimed.id), eq(chatEndpointLeases.token, token)));
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import { runtimePublicOrigin } from "./cloud-runtime-identity.js";
|
||||
import { projectSafeChatPublicationText } from "./chat-publication-projection.js";
|
||||
import { githubReviewCheckService } from "./chat-github-checks.js";
|
||||
import { githubReviewHeading } from "./chat-channels/github/review-heading.js";
|
||||
import { githubPublicationRetry, GITHUB_INLINE_RETRY_WINDOW_MS, GITHUB_INLINE_MAX_ATTEMPTS } from "./chat-channels/github/retry.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { githubResponseCommentService, writeGitHubResponseComment } from "./chat-github-response-comments.js";
|
||||
import { and, desc, eq, gt, isNull, or, sql } from "drizzle-orm";
|
||||
@@ -59,6 +60,7 @@ import {
|
||||
githubPreviousAssessment,
|
||||
} from "./chat-github-events.js";
|
||||
import { logActivity } from "./activity-log.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
type Session = {
|
||||
companyId: string;
|
||||
@@ -457,8 +459,10 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
toolName: string,
|
||||
parameters: unknown,
|
||||
writer: Db | Parameters<Parameters<Db["transaction"]>[0]>[0] = db,
|
||||
) {
|
||||
): Promise<string> {
|
||||
if (writer === db) return db.transaction(tx => stage(source, operation, payload, key, invocationId, toolName, parameters, tx));
|
||||
const actionKey = `github_publication:${source.issue.id}:${operation}:${hash(key)}`;
|
||||
await notifyDeliveryWork(writer, DELIVERY_QUEUES.chatGitHubPublications);
|
||||
const values = {
|
||||
companyId: source.endpoint.companyId,
|
||||
endpointId: source.endpoint.id,
|
||||
@@ -524,13 +528,53 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
receipt: action?.result ?? null,
|
||||
};
|
||||
}
|
||||
function publicationIsDue(action: { status: string; result: Record<string, unknown> | null }) {
|
||||
return action.status === "received" || (action.status === "failed" && action.result?.retryable === true &&
|
||||
Date.parse(String(action.result.retryAt)) <= Date.now());
|
||||
}
|
||||
async function publishAndWait(actionId: string, deadline: number) {
|
||||
let failures = 0;
|
||||
while (Date.now() < deadline) {
|
||||
await processPublication(actionId);
|
||||
const result = await receipt(actionId);
|
||||
if (result.status !== "received" && !(result.status === "failed" && result.receipt?.retryable === true)) return result;
|
||||
if (result.status === "failed") failures++;
|
||||
const delay = result.status === "received" ? 1000 : Math.max(0, Date.parse(String(result.receipt?.retryAt)) - Date.now());
|
||||
if (!Number.isFinite(delay) || Date.now() + delay >= deadline || failures >= GITHUB_INLINE_MAX_ATTEMPTS) return result;
|
||||
await new Promise(resolve => setTimeout(resolve, delay));
|
||||
}
|
||||
return receipt(actionId);
|
||||
}
|
||||
async function execute(
|
||||
session: Session,
|
||||
name: string,
|
||||
input: unknown,
|
||||
invocationId?: string,
|
||||
) {
|
||||
const deadline = Date.now() + GITHUB_INLINE_RETRY_WINDOW_MS;
|
||||
const source = await scope(session);
|
||||
if (invocationId && ["comment", "update_comment", "submit_review"].includes(name)) {
|
||||
const [existing] = await db.select().from(chatActions).where(and(
|
||||
eq(chatActions.companyId, session.companyId), eq(chatActions.endpointId, source.endpoint.id),
|
||||
eq(chatActions.conversationId, source.conversation.id), eq(chatActions.kind, "github_review_publication"),
|
||||
sql`${chatActions.payload}->>'invocationId' = ${invocationId}`,
|
||||
sql`${chatActions.payload}->>'toolName' = ${name}`,
|
||||
));
|
||||
if (existing) {
|
||||
const binding = existing.payload.session as Session;
|
||||
if (binding.companyId !== session.companyId || binding.agentId !== session.agentId ||
|
||||
binding.runId !== session.runId || binding.issueId !== session.issueId || hash(existing.payload.parameters) !== hash(input))
|
||||
throw conflict("This GitHub invocation is bound to a different publication");
|
||||
const result = await publishAndWait(existing.id, deadline);
|
||||
if (name !== "submit_review") return result;
|
||||
const [review] = await db.select().from(chatGitHubReviews).where(and(
|
||||
eq(chatGitHubReviews.companyId, session.companyId), eq(chatGitHubReviews.id, String(existing.payload.reviewId)),
|
||||
eq(chatGitHubReviews.runId, session.runId!), eq(chatGitHubReviews.endpointId, source.endpoint.id),
|
||||
));
|
||||
if (!review?.assessment) throw conflict("GitHub publication assessment is unavailable");
|
||||
return { reviewId: review.id, ...result, score: review.assessment.score, conclusion: review.conclusion };
|
||||
}
|
||||
}
|
||||
const api = await client(source);
|
||||
if (name === "comment" || name === "update_comment") {
|
||||
const parsed = commentSchema.parse(input);
|
||||
@@ -543,8 +587,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
name,
|
||||
input,
|
||||
);
|
||||
await processPublication(actionId);
|
||||
return receipt(actionId);
|
||||
return publishAndWait(actionId, deadline);
|
||||
}
|
||||
if (source.isIssue)
|
||||
throw conflict(
|
||||
@@ -807,10 +850,9 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
tx,
|
||||
);
|
||||
});
|
||||
await processPublication(actionId);
|
||||
return {
|
||||
reviewId: review.id,
|
||||
...(await receipt(actionId)),
|
||||
...(await publishAndWait(actionId, deadline)),
|
||||
score: assessment.score,
|
||||
conclusion: githubReviewConclusion(
|
||||
assessment,
|
||||
@@ -846,8 +888,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
name,
|
||||
input,
|
||||
);
|
||||
await processPublication(actionId);
|
||||
return receipt(actionId);
|
||||
return publishAndWait(actionId, deadline);
|
||||
}
|
||||
throw notFound("Unknown GitHub bot tool");
|
||||
}
|
||||
@@ -957,8 +998,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
.limit(1);
|
||||
if (
|
||||
!action ||
|
||||
action.status === "processed" ||
|
||||
action.status === "cancelled"
|
||||
!publicationIsDue(action)
|
||||
)
|
||||
return;
|
||||
const [binding] = await db
|
||||
@@ -981,7 +1021,14 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
const number = binding?.thread.match(
|
||||
/^github:[^:]+:(?:issue:)?([1-9][0-9]*)(?::rc:[1-9][0-9]*)?$/,
|
||||
)?.[1];
|
||||
if (!number || !binding?.repository.providerRepositoryId) return;
|
||||
if (!number || !binding?.repository.providerRepositoryId) {
|
||||
await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
|
||||
await tx.update(chatActions).set({ status: "cancelled", result: { code: "publication_binding_unavailable", retryable: false }, updatedAt: new Date() })
|
||||
.where(and(eq(chatActions.id, actionId), eq(chatActions.companyId, action.companyId)));
|
||||
});
|
||||
return;
|
||||
}
|
||||
await withGitHubPublicationLease(
|
||||
db,
|
||||
{
|
||||
@@ -996,11 +1043,11 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, actionId));
|
||||
if (!fresh || ["processed", "cancelled"].includes(fresh.status)) return;
|
||||
if (!fresh || !publicationIsDue(fresh)) return;
|
||||
const session = action.payload.session as Session;
|
||||
if (!session || session.companyId !== action.companyId)
|
||||
throw forbidden("Invalid GitHub publication binding");
|
||||
try {
|
||||
if (!session || session.companyId !== action.companyId)
|
||||
throw forbidden("Invalid GitHub publication binding");
|
||||
const source = await scope(session, true);
|
||||
if (
|
||||
source.endpoint.id !== action.endpointId ||
|
||||
@@ -1339,6 +1386,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
};
|
||||
} else throw forbidden("Unknown GitHub publication");
|
||||
await lease.commit(async (tx) => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -1405,16 +1453,14 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
);
|
||||
});
|
||||
await lease.commit(async (tx) => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: superseded || denied ? "cancelled" : "failed",
|
||||
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify({
|
||||
attempts,
|
||||
retryable: !superseded && !denied && attempts < 8,
|
||||
retryAt: new Date(
|
||||
Date.now() + Math.min(300000, 1000 * 2 ** attempts),
|
||||
).toISOString(),
|
||||
...githubPublicationRetry(error, attempts),
|
||||
...(superseded || denied ? { retryable: false } : {}),
|
||||
code: superseded
|
||||
? "stale_head"
|
||||
: denied
|
||||
@@ -1440,6 +1486,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.kind, "github_review_publication"),
|
||||
sql`coalesce(${publicationLeaseExpiresAt()}, now()) <= now()`,
|
||||
or(
|
||||
eq(chatActions.status, "received"),
|
||||
and(
|
||||
@@ -1461,5 +1508,24 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
|
||||
await processPublication(row.id);
|
||||
return rows.length;
|
||||
}
|
||||
return { execute, processPending, scope };
|
||||
function publicationLeaseExpiresAt() {
|
||||
return sql`(select max(l.expires_at) from chat_endpoint_leases l
|
||||
inner join chat_conversations c on c.id = ${chatActions.conversationId} and c.company_id = ${chatActions.companyId}
|
||||
inner join chat_endpoint_resources r on r.id = c.resource_id and r.company_id = ${chatActions.companyId}
|
||||
where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId}
|
||||
and l.lease_key = 'github-publication:' || (r.metadata->>'providerRepositoryId') || ':' ||
|
||||
substring(c.external_thread_id from '^github:[^:]+:(?:issue:)?([1-9][0-9]*)'))`;
|
||||
}
|
||||
async function nextPendingAt(): Promise<number | null> {
|
||||
const [row] = await db.select({ at: sql<string | null>`min(greatest(
|
||||
case when ${chatActions.status} = 'received' then now()
|
||||
else (${chatActions.result}->>'retryAt')::timestamptz end,
|
||||
${publicationLeaseExpiresAt()}
|
||||
))` }).from(chatActions).where(and(
|
||||
eq(chatActions.kind, "github_review_publication"),
|
||||
or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)),
|
||||
));
|
||||
return row?.at ? new Date(row.at).getTime() : null;
|
||||
}
|
||||
return { execute, processPending, nextPendingAt, scope };
|
||||
}
|
||||
@@ -100,7 +100,7 @@ export const GITHUB_BOT_TOOLS = [
|
||||
name: "submit_review",
|
||||
title: "Submit a review assessment",
|
||||
description:
|
||||
"Submit a structured assessment for this task's exact PR head. Use begin_review before starting an explicitly requested review. Paperclip validates coverage and score, publishes allowed summary/findings, and computes the Paperclip Review check. Coverage reviewedPaths and omittedPaths name only allowed changed files from read_pull_request(files); describe additional context in the rationale. Follow the schema length limits. Incomplete analysis cannot pass. This never formally approves a PR.",
|
||||
"Submit a structured assessment for this task's exact PR head. Use begin_review before starting an explicitly requested review. Paperclip validates coverage and score, publishes allowed summary/findings, and computes the Paperclip Review check. Coverage reviewedPaths and omittedPaths name only allowed changed files from read_pull_request(files); describe additional context in the rationale. Follow the schema length limits. Incomplete analysis cannot pass. This never formally approves a PR. Transient publication failures retry automatically. A received result or failed result with receipt.retryable=true remains queued; do not claim publication yet. After a timeout, repeat the exact call to recover its durable receipt. Do not begin another review or revise the assessment just to retry publication.",
|
||||
risk: "write",
|
||||
// Share the input contract with server validation so discovery includes every
|
||||
// length/array bound; hidden limits caused real agents to abandon publication.
|
||||
|
||||
@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
|
||||
chatSessionStops: "chat-session-stops",
|
||||
chatProviderEffects: "chat-provider-effects",
|
||||
chatGitHubIngress: "chat-github-ingress",
|
||||
chatGitHubPublications: "chat-github-publications",
|
||||
chatFailedRunRetries: "chat-failed-run-retries",
|
||||
chatTelegramMaintenance: "chat-telegram-maintenance",
|
||||
chatTelegramSubscriptions: "chat-telegram-subscriptions",
|
||||
|
||||
@@ -10523,7 +10523,18 @@ export function createToolGatewayService(
|
||||
const retryingSlackRateLimit = recorded.replayed && accessDecision.allowed && tool.providerType === "paperclip_slack_chat" && session.agentId && session.runId && session.issueId
|
||||
? await claimSlackRateLimitRetry(db, { companyId: session.companyId, agentId: session.agentId, runId: session.runId, issueId: session.issueId, endpointId: String(asRecord(tool.providerMetadata)?.endpointId ?? ""), identityContextId: session.identityContextId }, invocationId)
|
||||
: false;
|
||||
if (recorded.replayed && !retryingSlackRateLimit) {
|
||||
const gitHubPublicationReplay = recorded.replayed &&
|
||||
tool.providerType === "paperclip_github_chat" &&
|
||||
["comment", "update_comment", "submit_review"].includes(tool.upstreamToolName ?? "") &&
|
||||
recorded.invocation.approvalState === "not_required";
|
||||
const retryingGitHubPublication = gitHubPublicationReplay && accessDecision.allowed &&
|
||||
recorded.invocation.agentId === session.agentId && recorded.invocation.runId === session.runId &&
|
||||
recorded.invocation.issueId === session.issueId && recorded.invocation.connectionId === tool.connectionId &&
|
||||
recorded.invocation.toolName === tool.name && recorded.invocation.argumentsHash === effectiveArgumentsSummary.sha256 &&
|
||||
["authorized", "executing", "succeeded", "failed", "timed_out"].includes(recorded.invocation.status);
|
||||
if (gitHubPublicationReplay && accessDecision.allowed && !retryingGitHubPublication)
|
||||
throw new ToolGatewayHttpError(409, "Idempotency key belongs to a different GitHub publication", "github_idempotency_conflict");
|
||||
if (recorded.replayed && !retryingSlackRateLimit && !gitHubPublicationReplay) {
|
||||
await writeAudit({
|
||||
session,
|
||||
companyId: session.companyId,
|
||||
@@ -10599,6 +10610,7 @@ export function createToolGatewayService(
|
||||
.update(toolInvocations)
|
||||
.set({
|
||||
status: "executing",
|
||||
...(retryingGitHubPublication ? { errorCode: null, errorMessage: null, completedAt: null } : {}),
|
||||
startedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
|
||||
Reference in new issue
Block a user