mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 02:26:12 +02:00
fix(server): dedupe run-scoped comment retries
This commit is contained in:
1 parent
d104895c41
commit
498ccfd243
3 files changed
+120
-3
No files matched your search
@@ -0,0 +1,61 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agents, companies, createDb, heartbeatRuns, instanceSettings, issueComments, issues } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
describeEmbeddedPostgres("issueService.addCommentWithRunRetryDedup", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-comment-run-retry-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(issueComments);
|
||||
await db.delete(issues);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
await db.delete(instanceSettings);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
it("collapses repeated writes from the same run, author, and body", async () => {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "TDEDUP" });
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Coder",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Retry target", status: "in_progress", priority: "medium" });
|
||||
|
||||
const svc = issueService(db);
|
||||
const first = await svc.addCommentWithRunRetryDedup(issueId, "Same update", { agentId, runId });
|
||||
const second = await svc.addCommentWithRunRetryDedup(issueId, "Same update", { agentId, runId });
|
||||
|
||||
expect(first.reused).toBe(false);
|
||||
expect(second.reused).toBe(true);
|
||||
expect(second.comment.id).toBe(first.comment.id);
|
||||
await expect(db.select().from(issueComments)).resolves.toHaveLength(1);
|
||||
});
|
||||
});
|
||||
@@ -9735,6 +9735,7 @@ export function issueRoutes(
|
||||
// Without a single transaction, a 422 (or any error) thrown by the status update after the
|
||||
// comment is inserted would leave an orphan comment without the corresponding state change.
|
||||
let comment: Awaited<ReturnType<typeof svc.addComment>>;
|
||||
let reusedComment = false;
|
||||
if (shouldAutoApproveReviewComment) {
|
||||
const transition = applyIssueExecutionPolicyTransition({
|
||||
issue: currentIssue,
|
||||
@@ -9848,16 +9849,27 @@ export function issueRoutes(
|
||||
requestedByActorId: actor.actorId,
|
||||
});
|
||||
} else {
|
||||
comment = await svc.addComment(id, req.body.body, {
|
||||
const commentActor = {
|
||||
agentId: actor.agentId ?? undefined,
|
||||
userId: actor.actorType === "user" ? actor.actorId : undefined,
|
||||
runId: actor.runId,
|
||||
}, {
|
||||
};
|
||||
const commentOptions = {
|
||||
authorType: req.body.authorType ?? (actor.actorType === "agent" ? "agent" : "user"),
|
||||
presentation: req.body.presentation ?? null,
|
||||
metadata: req.body.metadata ?? null,
|
||||
sourceTrust: await sourceTrustForActorWrite(currentIssue, actor),
|
||||
});
|
||||
};
|
||||
const result = typeof svc.addCommentWithRunRetryDedup === "function"
|
||||
? await svc.addCommentWithRunRetryDedup(id, req.body.body, commentActor, commentOptions)
|
||||
: { comment: await svc.addComment(id, req.body.body, commentActor, commentOptions), reused: false };
|
||||
comment = result.comment;
|
||||
reusedComment = result.reused;
|
||||
}
|
||||
|
||||
if (reusedComment) {
|
||||
res.status(200).json(comment);
|
||||
return;
|
||||
}
|
||||
|
||||
await issueReferencesSvc.syncComment(comment.id);
|
||||
|
||||
@@ -7362,6 +7362,50 @@ export function issueService(db: Db) {
|
||||
return redactIssueComment(comment, currentUserRedactionOptions.enabled);
|
||||
},
|
||||
|
||||
addCommentWithRunRetryDedup: async (
|
||||
issueId: string,
|
||||
body: string,
|
||||
actor: { agentId?: string; userId?: string; runId?: string | null },
|
||||
options?: {
|
||||
authorType?: IssueCommentAuthorType | null;
|
||||
presentation?: IssueCommentPresentation | null;
|
||||
metadata?: IssueCommentMetadata | null;
|
||||
sourceTrust?: typeof issueComments.$inferInsert.sourceTrust;
|
||||
createdAt?: Date | string | null;
|
||||
},
|
||||
) => {
|
||||
if (!actor.runId) {
|
||||
return { comment: await issueService(db).addComment(issueId, body, actor, options), reused: false };
|
||||
}
|
||||
const runId = actor.runId;
|
||||
return db.transaction(async (tx) => {
|
||||
const guardKey = `issue-comment:run-retry:${issueId}:${runId}:${actor.agentId ?? actor.userId ?? "system"}:${body}`;
|
||||
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${guardKey}, 0))`);
|
||||
const currentUserRedactionOptions = {
|
||||
enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs,
|
||||
};
|
||||
const redactedBody = redactCurrentUserText(body, currentUserRedactionOptions);
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(issueComments)
|
||||
.where(and(
|
||||
eq(issueComments.issueId, issueId),
|
||||
eq(issueComments.createdByRunId, runId),
|
||||
actor.agentId ? eq(issueComments.authorAgentId, actor.agentId) : isNull(issueComments.authorAgentId),
|
||||
actor.userId ? eq(issueComments.authorUserId, actor.userId) : isNull(issueComments.authorUserId),
|
||||
eq(issueComments.body, redactedBody),
|
||||
isNull(issueComments.deletedAt),
|
||||
))
|
||||
.orderBy(asc(issueComments.createdAt))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (existing) {
|
||||
return { comment: redactIssueComment(existing, currentUserRedactionOptions.enabled), reused: true };
|
||||
}
|
||||
const comment = await issueService(db).addComment(issueId, body, actor, options, tx);
|
||||
return { comment, reused: false };
|
||||
});
|
||||
},
|
||||
|
||||
createAttachment: async (input: {
|
||||
issueId: string;
|
||||
issueCommentId?: string | null;
|
||||
|
||||
Reference in new issue
Block a user