From 0d8bbf7cf4fb90597e1b352c7626a78e5c196992 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 10 Sep 2026 08:41:59 -0700 Subject: [PATCH] refactor(server): move the queued-comment queue mutations into the wake-queue module (#13145) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The server coordinates issue execution and agent wake events > - Queued comment mutations belong to the wake queue that owns their state > - Route-local database writes split queue rules across two layers > - This pull request moves those mutations into the wake-queue module and keeps route authorization and response mapping > - The benefit is one transaction boundary with company-scoped writes and a shared checked response contract ## Linked Issues or Issue Description **What existing behavior does this improve?** The queued-comment edit, reorder, and discard endpoints write queue state directly from the route layer. **Subsystem affected** server/ — REST API and orchestration services. **Current behavior** The route layer owns database transactions, locks, queue writes, and wake-row writes for queued comments. **Proposed behavior** The wake-queue module owns these operations. The routes keep authorization, input checks, error mapping, and response mapping. **Reason and benefit** The module gives all queued-comment callers one transaction boundary and applies company predicates to every adapter read and write. **Breaking changes** None. The endpoints keep their existing paths and response behavior. ## What Changed - Move queued-comment edit, reorder, and discard operations into the wake-queue module. - Add company predicates to seven queue writes. - Use the shared queue contract type for mutation responses. - Add module tests and route tests for the moved operations. ## Verification - `server/src/modules/wake-queue`: 128 tests pass across 6 files. - `server/src/__tests__/issue-queued-comments-routes.test.ts`: 19 tests pass. - The server TypeScript check reports the same 141 pre-existing errors before and after this change. - GitHub Actions must pass the required pull-request checks. ## Risks The change moves transaction and lock ownership across module boundaries. The new adapter, use-case, and route tests cover the moved behavior. No database schema changes occur. ## Model Used OpenAI Codex, GPT-5, current agent runtime, tool use and code review support. ## 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 --- .../issue-queued-comments-routes.test.ts | 39 ++ .../adapters/queued-comment-postgres.test.ts | 285 ++++++++++ .../adapters/queued-comment-postgres.ts | 287 ++++++++++ .../application/queued-comment-ports.ts | 145 +++++ .../queued-comment-use-cases.test.ts | 374 +++++++++++++ .../application/queued-comment-use-cases.ts | 285 ++++++++++ .../modules/wake-queue/domain/policy.test.ts | 169 ++++++ .../src/modules/wake-queue/domain/policy.ts | 92 +++ server/src/modules/wake-queue/index.ts | 44 ++ server/src/routes/issues.ts | 524 ++++-------------- .../issue-queued-comment-queue.test.ts | 117 ++++ .../services/issue-queued-comment-queue.ts | 123 ++++ 12 files changed, 2077 insertions(+), 407 deletions(-) create mode 100644 server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts create mode 100644 server/src/modules/wake-queue/adapters/queued-comment-postgres.ts create mode 100644 server/src/modules/wake-queue/application/queued-comment-ports.ts create mode 100644 server/src/modules/wake-queue/application/queued-comment-use-cases.test.ts create mode 100644 server/src/modules/wake-queue/application/queued-comment-use-cases.ts create mode 100644 server/src/services/issue-queued-comment-queue.test.ts diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 253efb7e2c..6db7e9432d 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -313,6 +313,45 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { expect(storedIssue?.executionRunId).toBeNull(); }); + it("keeps a mutation response's steering disposition in step with a fresh GET after promotion", async () => { + const seeded = await seedQueue(); + await promoteQueue(seeded); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(initial.body.steeringDisposition).toBe("temporarily_unavailable"); + + const edited = await request(app(seeded.companyId)) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "edited during promotion" }); + expect(edited.status, JSON.stringify(edited.body)).toBe(200); + const afterEdit = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(edited.body.steeringDisposition).toBe(afterEdit.body.steeringDisposition); + + const reordered = await request(app(seeded.companyId)) + .put(`/api/issues/${seeded.issueId}/queued-comments/order`) + .send({ + queueId: seeded.wakeId, + revision: edited.body.revision, + orderedCommentIds: [...seeded.commentIds].reverse(), + }); + expect(reordered.status, JSON.stringify(reordered.body)).toBe(200); + const afterReorder = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(reordered.body.steeringDisposition).toBe(afterReorder.body.steeringDisposition); + }); + + it("returns entry objects with the same keys as the GET endpoint", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + const edited = await request(app(seeded.companyId)) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "edited body" }); + expect(edited.status, JSON.stringify(edited.body)).toBe(200); + expect(Object.keys(edited.body.entries[0]).sort()).toEqual(Object.keys(initial.body.entries[0]).sort()); + }); + it("cancels the deferred wake when the final message is discarded before promotion", async () => { const seeded = await seedQueue(); await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1])); diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts new file mode 100644 index 0000000000..9ca96964b8 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts @@ -0,0 +1,285 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "../../../__tests__/helpers/embedded-postgres.js"; +import { createQueuedCommentIssueLockWriter } from "./queued-comment-postgres.js"; +import type { QueuedCommentQueuePostgresAdapterDeps } from "./queued-comment-postgres.js"; +import { QueuedCommentMutationError } from "../application/queued-comment-use-cases.js"; + +// Proves the same two properties the release-half adapter test proves for +// this module's other transaction: the one company the caller names in +// `issue.companyId` binds every read and write for the whole transaction, so +// a caller-supplied `issue`/`wake` for the wrong company sees nothing, and a +// guarded write that affects no row rolls the transaction back instead of +// leaving a partial write. The decision branching itself is proven against +// plain facts in `domain/policy.test.ts`; the use-case orchestration is +// proven against a mocked port in `application/queued-comment-use-cases.test.ts`. +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres queued-comment adapter tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("queued-comment postgres adapter", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + const noopDeps: QueuedCommentQueuePostgresAdapterDeps = { + syncCommentReferences: async () => {}, + deleteCommentReferenceSource: async () => {}, + syncCommentExternalObjectsSafely: async () => {}, + }; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-queued-comment-postgres-adapter-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(issueComments); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(issues); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedCompany(): Promise { + const companyId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + defaultResponsibleUserId: "responsible-user", + }); + return companyId; + } + + async function seedAgent(input: { companyId: string; adapterType?: string }): Promise { + const agentId = randomUUID(); + await db.insert(agents).values({ + id: agentId, + companyId: input.companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: input.adapterType ?? "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + return agentId; + } + + async function seedIssue(input: { companyId: string; assigneeAgentId: string | null }): Promise { + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId: input.companyId, + title: "Queued-comment adapter fixture issue", + status: "in_progress", + priority: "medium", + assigneeAgentId: input.assigneeAgentId, + }); + return issueId; + } + + async function seedComment(input: { companyId: string; issueId: string; authorUserId: string }): Promise { + const commentId = randomUUID(); + await db.insert(issueComments).values({ + id: commentId, + companyId: input.companyId, + issueId: input.issueId, + authorUserId: input.authorUserId, + body: "queued message", + }); + return commentId; + } + + async function seedDeferredWake(input: { + companyId: string; + agentId: string; + issueId: string; + commentIds: string[]; + }): Promise { + const id = randomUUID(); + await db.insert(agentWakeupRequests).values({ + id, + companyId: input.companyId, + agentId: input.agentId, + source: "automation", + reason: "issue_commented", + status: "deferred_issue_execution", + requestedByActorType: "user", + payload: { + issueId: input.issueId, + _paperclipWakeContext: { wakeCommentIds: input.commentIds }, + }, + }); + return id; + } + + it("scopes the wake lookup to its own company: a foreign-company issue context resolves not_pending and deletes nothing", async () => { + const companyId = await seedCompany(); + const otherCompanyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const commentId = await seedComment({ companyId, issueId, authorUserId: "user-1" }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId, commentIds: [commentId] }); + + const issueLock = createQueuedCommentIssueLockWriter(db, noopDeps); + await expect( + issueLock.withLockedQueue( + { + // The caller mistakenly names the *other* company on the issue + // context; that single value binds every read and write for the + // whole transaction, so it alone must decide what is visible. + issue: { id: issueId, companyId: otherCompanyId, assigneeAgentId: agentId, executionRunId: null }, + actor: { actorType: "user", actorId: "user-1", agentId: null }, + queueId: wakeId, + }, + async () => { + throw new Error("fn must not run when the wake is invisible to the caller's company"); + }, + ), + ).rejects.toMatchObject({ code: "queued_comment_not_pending" }); + + const commentRow = (await db.select().from(issueComments).where(eq(issueComments.id, commentId)))[0]; + expect(commentRow?.deletedAt ?? null).toBeNull(); + const wakeRow = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]; + expect(wakeRow?.status).toBe("deferred_issue_execution"); + }); + + it("rolls back a discard when the comment id given belongs to a different issue than the one this transaction locked", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const otherIssueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const commentId = await seedComment({ companyId, issueId: otherIssueId, authorUserId: "user-1" }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId, commentIds: [commentId] }); + + const issueLock = createQueuedCommentIssueLockWriter(db, noopDeps); + await expect( + issueLock.withLockedQueue( + { + issue: { id: issueId, companyId, assigneeAgentId: agentId, executionRunId: null }, + actor: { actorType: "user", actorId: "user-1", agentId: null }, + queueId: wakeId, + }, + async (_locked, transaction) => { + // The comment belongs to a different issue than the one this + // transaction locked. The write's own `issueId` predicate, not + // just the bound company, must decide what is visible. + const deleted = await transaction.deleteComment({ issueId, commentId }); + if (!deleted) { + throw new QueuedCommentMutationError("queued_comment_not_pending", "The queued message is no longer pending"); + } + return deleted; + }, + ), + ).rejects.toMatchObject({ code: "queued_comment_not_pending" }); + + const commentRow = (await db.select().from(issueComments).where(eq(issueComments.id, commentId)))[0]; + expect(commentRow).toBeDefined(); + expect(commentRow?.deletedAt ?? null).toBeNull(); + }); + + it("edits the comment body, syncs references, and rebuilds the queue snapshot inside one company-scoped transaction", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const commentId = await seedComment({ companyId, issueId, authorUserId: "user-1" }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId, commentIds: [commentId] }); + + let syncedCommentId: string | null = null; + const deps: QueuedCommentQueuePostgresAdapterDeps = { + ...noopDeps, + syncCommentReferences: async (id) => { + syncedCommentId = id; + }, + }; + const issueLock = createQueuedCommentIssueLockWriter(db, deps); + + const queue = await issueLock.withLockedQueue( + { + issue: { id: issueId, companyId, assigneeAgentId: agentId, executionRunId: null }, + actor: { actorType: "user", actorId: "user-1", agentId: null }, + queueId: wakeId, + }, + async (locked, transaction) => { + expect(locked.state).toBe("deferred"); + const updated = await transaction.updateCommentBody({ + issueId, + commentId, + body: "edited body", + updatedAt: new Date(), + }); + expect(updated).toBe(true); + await transaction.syncCommentReferences(commentId); + return transaction.buildQueueSnapshot({ + issue: { id: issueId, companyId, assigneeAgentId: agentId, executionRunId: null }, + actor: { actorType: "user", actorId: "user-1", agentId: null }, + wake: locked.wake, + state: locked.state, + queueRun: locked.queueRun, + activeRun: locked.activeRun, + }); + }, + ); + + expect(syncedCommentId).toBe(commentId); + expect(queue.entries).toHaveLength(1); + expect((queue.entries[0]!.comment as { body: string }).body).toBe("edited body"); + const commentRow = (await db.select().from(issueComments).where(eq(issueComments.id, commentId)))[0]; + expect(commentRow?.body).toBe("edited body"); + }); + + // Pins a fact the database itself cannot persist today: `runtime_mode` is + // a NOT NULL column, so a real active run's own field is never null. The + // port type allows it (`runtimeMode: string | null`), so this test builds + // the fact directly instead of through a database row, to keep this + // branch of the shared steering rule under a regression test. + it("answers the steering question for a deferred paperclip_runner queue whose active run has no persisted runtime mode yet", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId, adapterType: "paperclip_runner" }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const commentId = await seedComment({ companyId, issueId, authorUserId: "user-1" }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId, commentIds: [commentId] }); + + const issueLock = createQueuedCommentIssueLockWriter(db, noopDeps); + const queue = await issueLock.withLockedQueue( + { + issue: { id: issueId, companyId, assigneeAgentId: agentId, executionRunId: null }, + actor: { actorType: "user", actorId: "user-1", agentId: null }, + queueId: wakeId, + }, + async (locked, transaction) => { + expect(locked.state).toBe("deferred"); + return transaction.buildQueueSnapshot({ + issue: { id: issueId, companyId, assigneeAgentId: agentId, executionRunId: null }, + actor: { actorType: "user", actorId: "user-1", agentId: null }, + wake: locked.wake, + state: "deferred", + queueRun: null, + activeRun: { id: randomUUID(), status: "running", runtimeMode: null, contextSnapshot: {} }, + }); + }, + ); + + expect(queue.protocol).toBe("paperclip_runner_v1"); + expect(queue.steeringDisposition).toBe("temporarily_unavailable"); + }); +}); diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts new file mode 100644 index 0000000000..0d8d4e01c8 --- /dev/null +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -0,0 +1,287 @@ +import { and, eq, inArray } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { agentWakeupRequests, agents, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import type { IssueComment, IssueQueuedCommentQueue } from "@paperclipai/shared"; +import { + buildQueuedCommentQueueSnapshot, + decideQueuedCommentQueueSteering, + queuedCommentIdsFromWakePayload, + withQueuedCommentIdsInRunContext, + withQueuedCommentIdsInWakePayload, +} from "../../../services/issue-queued-comment-queue.js"; +import { decideQueuedCommentWakeLookup } from "../domain/policy.js"; +import { parseObject, readNonEmptyString } from "../domain/values.js"; +import { QueuedCommentMutationError } from "../application/queued-comment-use-cases.js"; +import type { + LockedQueuedCommentState, + QueuedCommentIssueLockWriter, + QueuedCommentQueueTransaction, + QueuedCommentRunRow, + QueuedCommentWakeRow, +} from "../application/queued-comment-ports.js"; + +type WakeRow = typeof agentWakeupRequests.$inferSelect; +type RunRow = typeof heartbeatRuns.$inferSelect; + +function toWakeRow(row: WakeRow): QueuedCommentWakeRow { + return { id: row.id, agentId: row.agentId, status: row.status, runId: row.runId, payload: parseObject(row.payload) }; +} + +function toRunRow(row: RunRow): QueuedCommentRunRow { + return { id: row.id, status: row.status, runtimeMode: row.runtimeMode, contextSnapshot: parseObject(row.contextSnapshot) }; +} + +export type QueuedCommentQueuePostgresAdapterDeps = { + /** `issueReferenceService(db).syncComment`; runs on the module's own transaction. */ + syncCommentReferences(commentId: string, tx: Db): Promise; + /** `issueReferenceService(db).deleteCommentSource`; runs on the module's own transaction. */ + deleteCommentReferenceSource(commentId: string, tx: Db): Promise; + /** `externalObjectService(db, opts).syncCommentSafely`; runs on the module's own transaction. */ + syncCommentExternalObjectsSafely(commentId: string, tx: Db): Promise; +}; + +function buildTransaction(tx: Db, companyId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction { + return { + async updateCommentBody({ issueId, commentId, body, updatedAt }) { + const updated = await tx + .update(issueComments) + .set({ body, updatedAt }) + .where(and(eq(issueComments.id, commentId), eq(issueComments.issueId, issueId), eq(issueComments.companyId, companyId))) + .returning({ id: issueComments.id }) + .then((rows) => rows[0] ?? null); + return updated !== null; + }, + + async touchIssueUpdatedAt({ issueId, updatedAt }) { + await tx.update(issues).set({ updatedAt }).where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))); + }, + + async updateWakeQueuedCommentIds({ wakeId, payload, ids, updatedAt }) { + const row = await tx + .update(agentWakeupRequests) + .set({ payload: withQueuedCommentIdsInWakePayload(payload, ids), updatedAt }) + .where(and(eq(agentWakeupRequests.id, wakeId), eq(agentWakeupRequests.companyId, companyId))) + .returning() + .then((rows) => rows[0]); + return toWakeRow(row); + }, + + async updateQueueRunCommentIds({ queueRunId, contextSnapshot, ids, updatedAt }) { + const row = await tx + .update(heartbeatRuns) + .set({ contextSnapshot: withQueuedCommentIdsInRunContext(contextSnapshot, ids), updatedAt }) + .where(and(eq(heartbeatRuns.id, queueRunId), eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))) + .returning() + .then((rows) => rows[0] ?? null); + return row ? toRunRow(row) : null; + }, + + async deleteComment({ issueId, commentId }) { + const row = await tx + .delete(issueComments) + .where(and(eq(issueComments.id, commentId), eq(issueComments.issueId, issueId), eq(issueComments.companyId, companyId))) + .returning() + .then((rows) => rows[0] ?? null); + return row ? (row as IssueComment) : null; + }, + + async cancelWake({ wakeId, reason, now }) { + await tx + .update(agentWakeupRequests) + .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now }) + .where(and(eq(agentWakeupRequests.id, wakeId), eq(agentWakeupRequests.companyId, companyId))); + }, + + async cancelQueueRun({ queueRunId, reason, now }) { + const row = await tx + .update(heartbeatRuns) + .set({ status: "cancelled", finishedAt: now, error: reason, errorCode: "queued_comment_discarded", updatedAt: now }) + .where(and(eq(heartbeatRuns.id, queueRunId), eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))) + .returning({ id: heartbeatRuns.id }) + .then((rows) => rows[0] ?? null); + return row ? { id: row.id } : null; + }, + + async clearExecutionLockAndTouchIssue({ issueId, executionRunId, updatedAt }) { + await tx + .update(issues) + .set({ executionRunId: null, executionAgentNameKey: null, executionLockedAt: null, updatedAt }) + .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId), eq(issues.executionRunId, executionRunId))); + }, + + async buildQueueSnapshot({ issue, actor, wake, state, queueRun, activeRun }): Promise { + const commentIds = queuedCommentIdsFromWakePayload(wake?.payload ?? null); + const rows = + commentIds.length > 0 + ? await tx + .select() + .from(issueComments) + .where( + and( + eq(issueComments.companyId, companyId), + eq(issueComments.issueId, issue.id), + inArray(issueComments.id, commentIds), + ), + ) + : []; + const byId = new Map(rows.map((row) => [row.id, row])); + const comments = commentIds.flatMap((id) => { + const row = byId.get(id); + return row && !row.deletedAt ? [row] : []; + }); + + const assignedAgent = issue.assigneeAgentId + ? await tx + .select({ adapterType: agents.adapterType }) + .from(agents) + .where(and(eq(agents.id, issue.assigneeAgentId), eq(agents.companyId, companyId))) + .limit(1) + .then((agentRows) => agentRows[0] ?? null) + : null; + + // A queue mutation never delivers same-turn steering itself, so this + // adapter never probes the live runner: it answers + // "temporarily_unavailable" wherever the shared rule says a caller + // may probe. Only the read path probes the live provider. + const steering = decideQueuedCommentQueueSteering({ + state, + queueRunRuntimeMode: queueRun?.runtimeMode ?? null, + activeRun, + assignedAgentAdapterType: assignedAgent?.adapterType ?? null, + queuedCommentCount: comments.length, + }); + const steeringDisposition: IssueQueuedCommentQueue["steeringDisposition"] = + steering.kind === "probe" ? "temporarily_unavailable" : steering.kind; + + return buildQueuedCommentQueueSnapshot({ + issueId: issue.id, + queueId: wake?.id ?? null, + state, + activeRunId: activeRun?.id ?? null, + protocol: steering.protocol, + steeringDisposition, + comments, + actorType: actor.actorType, + actorId: actor.actorId, + }); + }, + + async syncCommentReferences(commentId) { + await deps.syncCommentReferences(commentId, tx); + }, + async deleteCommentReferenceSource(commentId) { + await deps.deleteCommentReferenceSource(commentId, tx); + }, + async syncCommentExternalObjectsSafely(commentId) { + await deps.syncCommentExternalObjectsSafely(commentId, tx); + }, + }; +} + +export function createQueuedCommentIssueLockWriter(db: Db, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentIssueLockWriter { + return { + async withLockedQueue(input, fn) { + return db.transaction(async (rawTx) => { + const tx = rawTx as unknown as Db; + const companyId = input.issue.companyId; + + await tx + .select({ id: issues.id }) + .from(issues) + .where(and(eq(issues.id, input.issue.id), eq(issues.companyId, companyId))) + .for("update"); + + const wakeRow = await tx + .select() + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.id, input.queueId), + eq(agentWakeupRequests.companyId, companyId), + input.issue.assigneeAgentId ? eq(agentWakeupRequests.agentId, input.issue.assigneeAgentId) : undefined, + ), + ) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + + const wakePayload = parseObject(wakeRow?.payload); + const lookup = decideQueuedCommentWakeLookup({ + wakePresent: wakeRow !== null, + wakeIssueIdMatches: readNonEmptyString(wakePayload.issueId) === input.issue.id, + hasQueuedCommentIds: queuedCommentIdsFromWakePayload(wakeRow?.payload ?? null).length > 0, + wakeStatus: wakeRow?.status ?? null, + wakeHasRunId: Boolean(wakeRow?.runId), + }); + + if (lookup.kind === "not_pending") { + throw new QueuedCommentMutationError("queued_comment_not_pending", "The queued message is no longer pending"); + } + if (lookup.kind === "already_dispatching") { + throw new QueuedCommentMutationError("queued_comment_already_dispatching", "The queued message is already being dispatched"); + } + + // Unreachable: `decideQueuedCommentWakeLookup` only returns "deferred" or "check_queue_run" when the wake row is present. + if (!wakeRow) throw new Error("wake-queue: queued-comment lookup resolved without a wake row"); + + let state: "deferred" | "queued"; + let queueRunRow: RunRow | null = null; + + if (lookup.kind === "deferred") { + state = "deferred"; + } else { + // check_queue_run: `wakeHasRunId` was true for this branch to have been reached. + queueRunRow = await tx + .select() + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, wakeRow.runId!), + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, wakeRow.agentId), + eq(heartbeatRuns.wakeupRequestId, wakeRow.id), + ), + ) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + if (!queueRunRow || queueRunRow.status !== "queued") { + throw new QueuedCommentMutationError( + "queued_comment_already_dispatching", + "The queued message is already being dispatched", + ); + } + state = "queued"; + } + + const activeRunId = state === "deferred" ? input.issue.executionRunId : null; + const activeRunRow = activeRunId + ? await tx + .select() + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, activeRunId), eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"))) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null) + : null; + + const transaction = buildTransaction(tx, companyId, deps); + const wake = toWakeRow(wakeRow); + const queueRun = queueRunRow ? toRunRow(queueRunRow) : null; + const activeRun = activeRunRow ? toRunRow(activeRunRow) : null; + + const queue = await transaction.buildQueueSnapshot({ + issue: input.issue, + actor: input.actor, + wake, + state, + queueRun, + activeRun, + }); + + const locked: LockedQueuedCommentState = { wake, state, queueRun, activeRun, queue }; + return fn(locked, transaction); + }); + }, + }; +} diff --git a/server/src/modules/wake-queue/application/queued-comment-ports.ts b/server/src/modules/wake-queue/application/queued-comment-ports.ts new file mode 100644 index 0000000000..36ca88c440 --- /dev/null +++ b/server/src/modules/wake-queue/application/queued-comment-ports.ts @@ -0,0 +1,145 @@ +// Ports for the three queued-comment queue mutations (edit, reorder, +// discard). Mirrors the release half's `IssueLockWriter` shape: the adapter +// owns the one transaction each mutation runs in, locks the issue and the +// wake row the caller named, classifies the locked state, and hands the +// caller a `LockedQueuedCommentState` plus a `QueuedCommentQueueTransaction` +// bound to that same transaction for every further read and write. + +import type { IssueComment, IssueQueuedCommentEntry, IssueQueuedCommentQueue } from "@paperclipai/shared"; + +export type QueuedCommentActor = { + actorType: "agent" | "user"; + /** The user id for a user actor, the agent id for an agent actor -- the same value `getActorInfo` names `actorId`. */ + actorId: string; + /** Null for a user actor. */ + agentId: string | null; +}; + +export type QueuedCommentIssueContext = { + id: string; + companyId: string; + assigneeAgentId: string | null; + executionRunId: string | null; +}; + +export type QueuedCommentWakeRow = { + id: string; + agentId: string; + status: string; + runId: string | null; + payload: Record; +}; + +export type QueuedCommentRunRow = { + id: string; + status: string; + runtimeMode: string | null; + contextSnapshot: Record; +}; + +/** The module's entry shape is the shared contract, so the compiler checks it directly; the route needs no cast. */ +export type QueuedCommentEntrySnapshot = IssueQueuedCommentEntry; + +/** The module's queue-snapshot shape is the shared contract, so the compiler checks it directly; the route needs no cast. */ +export type QueuedCommentQueueSnapshot = IssueQueuedCommentQueue; + +/** The locked, transaction-scoped state a mutation reads before it decides what to write. */ +export type LockedQueuedCommentState = { + wake: QueuedCommentWakeRow; + state: "deferred" | "queued"; + queueRun: QueuedCommentRunRow | null; + activeRun: QueuedCommentRunRow | null; + queue: QueuedCommentQueueSnapshot; +}; + +/** + * Every member is bound to the one transaction `withLockedQueue` owns, and + * to the one company that transaction is open for. Every read and every + * write names that bound `companyId` in its own predicate; a + * caller-supplied `issue`/`wake`/`queueRun` value is never trusted as an + * authorization boundary by itself. + */ +export interface QueuedCommentQueueTransaction { + updateCommentBody(input: { + issueId: string; + commentId: string; + body: string; + updatedAt: Date; + }): Promise; + touchIssueUpdatedAt(input: { issueId: string; updatedAt: Date }): Promise; + /** Compare-and-set on `id`; the wake's current status is not re-checked here because the row is already locked for the duration of this transaction. */ + updateWakeQueuedCommentIds(input: { + wakeId: string; + payload: Record; + ids: string[]; + updatedAt: Date; + }): Promise; + /** Guarded on the run's current `queued` status. Returns `null` when a concurrent writer already moved the run off `queued`. */ + updateQueueRunCommentIds(input: { + queueRunId: string; + /** The run's own context snapshot, as already read under lock; the rewrite is derived from this base. */ + contextSnapshot: Record; + ids: string[]; + updatedAt: Date; + }): Promise; + /** Returns the full deleted comment row so the caller can echo it back as the delete route's response body. */ + deleteComment(input: { + issueId: string; + commentId: string; + }): Promise; + cancelWake(input: { wakeId: string; reason: string; now: Date }): Promise; + /** + * Guarded on the run's current `queued` status. Returns `null` when a + * concurrent writer already moved the run off `queued`; returns just the + * cancelled run's id, which is all a post-commit telemetry emission needs. + */ + cancelQueueRun(input: { + queueRunId: string; + reason: string; + now: Date; + }): Promise<{ id: string } | null>; + /** + * Clears the issue's execution-lock columns and performs the `updatedAt` + * touch, in the one update the original route issued for a discard that + * empties the queue with a live queue run. The write is guarded on the + * issue's current `executionRunId`; a lost guard silently skips the whole + * update, matching the pre-existing behavior of this best-effort touch. + */ + clearExecutionLockAndTouchIssue(input: { + issueId: string; + executionRunId: string; + updatedAt: Date; + }): Promise; + buildQueueSnapshot(input: { + issue: QueuedCommentIssueContext; + actor: QueuedCommentActor; + wake: QueuedCommentWakeRow | null; + state: "deferred" | "queued" | null; + queueRun: QueuedCommentRunRow | null; + activeRun: QueuedCommentRunRow | null; + }): Promise; + syncCommentReferences(commentId: string): Promise; + deleteCommentReferenceSource(commentId: string): Promise; + syncCommentExternalObjectsSafely(commentId: string): Promise; +} + +export interface QueuedCommentIssueLockWriter { + /** + * Opens the one transaction a mutation runs in: locks the issue row, locks + * the wake row named by `queueId`, classifies it (reading and locking the + * linked heartbeat run when the classification needs it), and builds the + * queue snapshot the caller's mutation target check compares against. + * Throws `QueuedCommentMutationError` with code `queued_comment_not_pending` + * or `queued_comment_already_dispatching` when the lock step itself cannot + * resolve a live queue; `fn` never runs in that case. + */ + withLockedQueue( + input: { + /** Also carries the company id; every locked read and write binds its `companyId` predicate to `issue.companyId`. */ + issue: QueuedCommentIssueContext; + actor: QueuedCommentActor; + queueId: string; + }, + fn: (locked: LockedQueuedCommentState, transaction: QueuedCommentQueueTransaction) => Promise, + ): Promise; +} diff --git a/server/src/modules/wake-queue/application/queued-comment-use-cases.test.ts b/server/src/modules/wake-queue/application/queued-comment-use-cases.test.ts new file mode 100644 index 0000000000..6be0b8ba2a --- /dev/null +++ b/server/src/modules/wake-queue/application/queued-comment-use-cases.test.ts @@ -0,0 +1,374 @@ +import { describe, expect, it, vi } from "vitest"; +import type { IssueComment } from "@paperclipai/shared"; +import { + createDiscardQueuedComment, + createEditQueuedComment, + createReorderQueuedComments, + QueuedCommentMutationError, + QueuedCommentMutationForbiddenError, +} from "./queued-comment-use-cases.js"; +import type { + LockedQueuedCommentState, + QueuedCommentActor, + QueuedCommentEntrySnapshot, + QueuedCommentIssueContext, + QueuedCommentIssueLockWriter, + QueuedCommentQueueSnapshot, + QueuedCommentQueueTransaction, + QueuedCommentRunRow, + QueuedCommentWakeRow, +} from "./queued-comment-ports.js"; + +const ISSUE: QueuedCommentIssueContext = { + id: "issue-1", + companyId: "company-1", + assigneeAgentId: "agent-1", + executionRunId: null, +}; + +const USER_ACTOR: QueuedCommentActor = { actorType: "user", actorId: "user-1", agentId: null }; +const AGENT_ACTOR: QueuedCommentActor = { actorType: "agent", actorId: "agent-1", agentId: "agent-1" }; + +function wakeRow(overrides: Partial = {}): QueuedCommentWakeRow { + return { id: "wake-1", agentId: "agent-1", status: "deferred_issue_execution", runId: null, payload: {}, ...overrides }; +} + +function runRow(overrides: Partial = {}): QueuedCommentRunRow { + return { id: "run-1", status: "queued", runtimeMode: null, contextSnapshot: {}, ...overrides }; +} + +function commentFixture(overrides: Partial = {}): IssueComment { + return { + id: "comment-1", + companyId: "company-1", + issueId: "issue-1", + authorType: "user", + authorAgentId: null, + authorUserId: "user-1", + body: "queued comment", + presentation: null, + metadata: null, + createdAt: new Date("2026-01-01T00:00:00Z"), + updatedAt: new Date("2026-01-01T00:00:00Z"), + ...overrides, + }; +} + +function entry(overrides: Partial = {}): QueuedCommentEntrySnapshot { + return { + comment: commentFixture(), + position: 0, + canEdit: true, + canDiscard: true, + ...overrides, + }; +} + +function queueSnapshot(overrides: Partial = {}): QueuedCommentQueueSnapshot { + return { + issueId: ISSUE.id, + queueId: "wake-1", + state: "deferred", + targetRunId: null, + revision: "rev-1", + protocol: "legacy", + steeringDisposition: "unsupported", + entries: [entry()], + ...overrides, + }; +} + +function lockedState(overrides: Partial = {}): LockedQueuedCommentState { + return { + wake: wakeRow(), + state: "deferred", + queueRun: null, + activeRun: null, + queue: queueSnapshot(), + ...overrides, + }; +} + +function createFakeTransaction(overrides: Partial = {}): QueuedCommentQueueTransaction { + return { + updateCommentBody: vi.fn(async () => true), + touchIssueUpdatedAt: vi.fn(async () => {}), + updateWakeQueuedCommentIds: vi.fn(async (input) => wakeRow({ id: input.wakeId })), + updateQueueRunCommentIds: vi.fn(async (input) => runRow({ id: input.queueRunId })), + deleteComment: vi.fn(async () => commentFixture()), + cancelWake: vi.fn(async () => {}), + cancelQueueRun: vi.fn(async () => ({ id: "run-1" })), + clearExecutionLockAndTouchIssue: vi.fn(async () => {}), + buildQueueSnapshot: vi.fn(async () => queueSnapshot()), + syncCommentReferences: vi.fn(async () => {}), + deleteCommentReferenceSource: vi.fn(async () => {}), + syncCommentExternalObjectsSafely: vi.fn(async () => {}), + ...overrides, + }; +} + +function createFakeIssueLock(locked: LockedQueuedCommentState, transaction: QueuedCommentQueueTransaction): QueuedCommentIssueLockWriter { + return { + withLockedQueue: vi.fn(async (_input, fn) => fn(locked, transaction)), + }; +} + +describe("editQueuedComment", () => { + it("updates the comment body and rebuilds the queue snapshot", async () => { + const locked = lockedState(); + const transaction = createFakeTransaction(); + const issueLock = createFakeIssueLock(locked, transaction); + const editQueuedComment = createEditQueuedComment({ issueLock }); + + const result = await editQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + body: "updated body", + now: new Date("2026-01-01T00:00:00Z"), + }); + + expect(transaction.updateCommentBody).toHaveBeenCalledWith( + expect.objectContaining({ issueId: "issue-1", commentId: "comment-1", body: "updated body" }), + ); + expect(transaction.syncCommentReferences).toHaveBeenCalledWith("comment-1"); + expect(transaction.syncCommentExternalObjectsSafely).toHaveBeenCalledWith("comment-1"); + expect(result).toEqual(queueSnapshot()); + }); + + it("rejects a stale queue id with queued_comment_stale_queue", async () => { + const locked = lockedState({ queue: queueSnapshot({ queueId: "wake-1" }) }); + const editQueuedComment = createEditQueuedComment({ issueLock: createFakeIssueLock(locked, createFakeTransaction()) }); + + await expect( + editQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-2", + revision: "rev-1", + body: "x", + now: new Date(), + }), + ).rejects.toMatchObject({ code: "queued_comment_stale_queue" }); + }); + + it("rejects a stale revision with queued_comment_revision_conflict", async () => { + const locked = lockedState(); + const editQueuedComment = createEditQueuedComment({ issueLock: createFakeIssueLock(locked, createFakeTransaction()) }); + + await expect( + editQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "stale-rev", + body: "x", + now: new Date(), + }), + ).rejects.toBeInstanceOf(QueuedCommentMutationError); + }); + + it("rejects a comment the actor cannot edit", async () => { + const locked = lockedState({ queue: queueSnapshot({ entries: [entry({ canEdit: false })] }) }); + const editQueuedComment = createEditQueuedComment({ issueLock: createFakeIssueLock(locked, createFakeTransaction()) }); + + await expect( + editQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + body: "x", + now: new Date(), + }), + ).rejects.toBeInstanceOf(QueuedCommentMutationForbiddenError); + }); + + it("surfaces queued_comment_already_dispatching when the linked run left the queued status mid-mutation", async () => { + const locked = lockedState({ queueRun: runRow({ id: "run-1" }) }); + const transaction = createFakeTransaction({ updateQueueRunCommentIds: vi.fn(async () => null) }); + const editQueuedComment = createEditQueuedComment({ issueLock: createFakeIssueLock(locked, transaction) }); + + await expect( + editQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + body: "x", + now: new Date(), + }), + ).rejects.toMatchObject({ code: "queued_comment_already_dispatching" }); + }); +}); + +describe("reorderQueuedComments", () => { + it("rewrites the wake payload with the submitted order", async () => { + const locked = lockedState({ + queue: queueSnapshot({ entries: [ + entry({ comment: commentFixture({ id: "a" }), position: 0 }), + entry({ comment: commentFixture({ id: "b" }), position: 1 }), + ] }), + }); + const transaction = createFakeTransaction(); + const reorderQueuedComments = createReorderQueuedComments({ issueLock: createFakeIssueLock(locked, transaction) }); + + await reorderQueuedComments({ + issue: ISSUE, + actor: USER_ACTOR, + queueId: "wake-1", + revision: "rev-1", + orderedCommentIds: ["b", "a"], + now: new Date(), + }); + + expect(transaction.updateWakeQueuedCommentIds).toHaveBeenCalledWith( + expect.objectContaining({ wakeId: "wake-1", ids: ["b", "a"] }), + ); + }); + + it("rejects an order that is not a permutation of the current queue", async () => { + const locked = lockedState({ queue: queueSnapshot({ entries: [ + entry({ comment: commentFixture({ id: "a" }) }), + entry({ comment: commentFixture({ id: "b" }) }), + ] }) }); + const reorderQueuedComments = createReorderQueuedComments({ issueLock: createFakeIssueLock(locked, createFakeTransaction()) }); + + await expect( + reorderQueuedComments({ + issue: ISSUE, + actor: USER_ACTOR, + queueId: "wake-1", + revision: "rev-1", + orderedCommentIds: ["a"], + now: new Date(), + }), + ).rejects.toMatchObject({ code: "queued_comment_order_mismatch" }); + }); +}); + +describe("discardQueuedComment", () => { + it("cancels the wake and the queued run when the discard empties the queue", async () => { + const locked = lockedState({ state: "queued", queueRun: runRow({ id: "run-1" }) }); + const transaction = createFakeTransaction(); + const discardQueuedComment = createDiscardQueuedComment({ issueLock: createFakeIssueLock(locked, transaction) }); + + const result = await discardQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + now: new Date(), + }); + + expect(transaction.cancelWake).toHaveBeenCalledWith(expect.objectContaining({ wakeId: "wake-1" })); + expect(transaction.cancelQueueRun).toHaveBeenCalledWith(expect.objectContaining({ queueRunId: "run-1" })); + expect(transaction.clearExecutionLockAndTouchIssue).toHaveBeenCalledWith( + expect.objectContaining({ executionRunId: "run-1" }), + ); + expect(result.cancelledRun).toEqual({ id: "run-1" }); + }); + + it("rewrites the remaining ids when other queued comments are left", async () => { + const locked = lockedState({ + queue: queueSnapshot({ entries: [ + entry({ comment: commentFixture({ id: "comment-1" }) }), + entry({ comment: commentFixture({ id: "comment-2" }) }), + ] }), + }); + const transaction = createFakeTransaction(); + const discardQueuedComment = createDiscardQueuedComment({ issueLock: createFakeIssueLock(locked, transaction) }); + + const result = await discardQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + now: new Date(), + }); + + expect(transaction.updateWakeQueuedCommentIds).toHaveBeenCalledWith(expect.objectContaining({ ids: ["comment-2"] })); + expect(transaction.cancelWake).not.toHaveBeenCalled(); + expect(transaction.touchIssueUpdatedAt).toHaveBeenCalledWith(expect.objectContaining({ issueId: ISSUE.id })); + expect(result.cancelledRun).toBeNull(); + }); + + it("skips the mutation-target check when no revision is submitted, matching the comment-delete route's cancellation call site", async () => { + const locked = lockedState({ queue: queueSnapshot({ revision: "some-other-revision" }) }); + const transaction = createFakeTransaction(); + const discardQueuedComment = createDiscardQueuedComment({ issueLock: createFakeIssueLock(locked, transaction) }); + + await expect( + discardQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + now: new Date(), + }), + ).resolves.toBeDefined(); + }); + + it("rejects a discard from an actor who did not author the comment", async () => { + const locked = lockedState({ queue: queueSnapshot({ entries: [ + entry({ comment: commentFixture({ authorUserId: "user-2", authorAgentId: null }) }), + ] }) }); + const discardQueuedComment = createDiscardQueuedComment({ issueLock: createFakeIssueLock(locked, createFakeTransaction()) }); + + await expect( + discardQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + now: new Date(), + }), + ).rejects.toBeInstanceOf(QueuedCommentMutationForbiddenError); + }); + + it("authorizes an agent actor discarding its own queued message", async () => { + const locked = lockedState({ + queue: queueSnapshot({ entries: [ + entry({ comment: commentFixture({ authorUserId: null, authorAgentId: "agent-1" }) }), + ] }), + }); + const transaction = createFakeTransaction(); + const discardQueuedComment = createDiscardQueuedComment({ issueLock: createFakeIssueLock(locked, transaction) }); + + await expect( + discardQueuedComment({ + issue: ISSUE, + actor: AGENT_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + now: new Date(), + }), + ).resolves.toBeDefined(); + }); + + it("rolls back when the queued run to cancel already left the queued status", async () => { + const locked = lockedState({ queueRun: runRow({ id: "run-1" }) }); + const transaction = createFakeTransaction({ cancelQueueRun: vi.fn(async () => null) }); + const discardQueuedComment = createDiscardQueuedComment({ issueLock: createFakeIssueLock(locked, transaction) }); + + await expect( + discardQueuedComment({ + issue: ISSUE, + actor: USER_ACTOR, + commentId: "comment-1", + queueId: "wake-1", + revision: "rev-1", + now: new Date(), + }), + ).rejects.toMatchObject({ code: "queued_comment_already_dispatching" }); + }); +}); diff --git a/server/src/modules/wake-queue/application/queued-comment-use-cases.ts b/server/src/modules/wake-queue/application/queued-comment-use-cases.ts new file mode 100644 index 0000000000..9d9b6a3f94 --- /dev/null +++ b/server/src/modules/wake-queue/application/queued-comment-use-cases.ts @@ -0,0 +1,285 @@ +import type { IssueComment } from "@paperclipai/shared"; +import { + decideQueuedCommentActorOwnsEntry, + decideQueuedCommentReorder, +} from "../domain/policy.js"; +import type { + QueuedCommentActor, + QueuedCommentIssueContext, + QueuedCommentIssueLockWriter, + QueuedCommentQueueSnapshot, + QueuedCommentQueueTransaction, + QueuedCommentRunRow, +} from "./queued-comment-ports.js"; + +export type QueuedCommentMutationErrorCode = + | "queued_comment_not_pending" + | "queued_comment_already_dispatching" + | "queued_comment_stale_queue" + | "queued_comment_revision_conflict" + | "queued_comment_order_mismatch"; + +/** The route maps this 1:1 onto the `conflict(...)` HTTP error it threw before this move, using `code` and `message` unchanged. */ +export class QueuedCommentMutationError extends Error { + constructor( + readonly code: QueuedCommentMutationErrorCode, + message: string, + ) { + super(message); + this.name = "QueuedCommentMutationError"; + } +} + +/** The route maps this onto the `forbidden(...)` HTTP error it threw before this move, using `message` unchanged. */ +export class QueuedCommentMutationForbiddenError extends Error { + constructor(message: string) { + super(message); + this.name = "QueuedCommentMutationForbiddenError"; + } +} + +function requireMutationTarget(queue: QueuedCommentQueueSnapshot, queueId: string, revision: string): void { + if (queue.queueId !== queueId) { + throw new QueuedCommentMutationError("queued_comment_stale_queue", "The queued message targets a stale queue"); + } + if (queue.revision !== revision) { + throw new QueuedCommentMutationError("queued_comment_revision_conflict", "The queued messages changed in another session"); + } +} + +async function updateQueueRunCommentIdsGuarded( + tx: QueuedCommentQueueTransaction, + input: { queueRun: QueuedCommentRunRow | null; ids: string[]; updatedAt: Date }, +): Promise { + if (!input.queueRun) return null; + const updated = await tx.updateQueueRunCommentIds({ + queueRunId: input.queueRun.id, + contextSnapshot: input.queueRun.contextSnapshot, + ids: input.ids, + updatedAt: input.updatedAt, + }); + if (!updated) { + throw new QueuedCommentMutationError("queued_comment_already_dispatching", "The queued message is already being dispatched"); + } + return updated; +} + +export type EditQueuedCommentInput = { + issue: QueuedCommentIssueContext; + actor: QueuedCommentActor; + commentId: string; + queueId: string; + revision: string; + body: string; + now: Date; +}; + +export function createEditQueuedComment(deps: { issueLock: QueuedCommentIssueLockWriter }) { + return async function editQueuedComment(input: EditQueuedCommentInput): Promise { + return deps.issueLock.withLockedQueue( + { issue: input.issue, actor: input.actor, queueId: input.queueId }, + async (locked, tx) => { + requireMutationTarget(locked.queue, input.queueId, input.revision); + const entry = locked.queue.entries.find((candidate) => candidate.comment.id === input.commentId); + if (!entry) { + throw new QueuedCommentMutationError("queued_comment_not_pending", "The queued message is no longer pending"); + } + if (!entry.canEdit) { + throw new QueuedCommentMutationForbiddenError("Only the queued message author can edit it"); + } + + const updated = await tx.updateCommentBody({ + issueId: input.issue.id, + commentId: input.commentId, + body: input.body, + updatedAt: input.now, + }); + if (!updated) { + throw new QueuedCommentMutationError("queued_comment_not_pending", "The queued message is no longer pending"); + } + await tx.touchIssueUpdatedAt({ issueId: input.issue.id, updatedAt: input.now }); + await tx.syncCommentReferences(input.commentId); + await tx.syncCommentExternalObjectsSafely(input.commentId); + + const ids = locked.queue.entries.map((candidate) => candidate.comment.id); + const updatedQueueRun = await updateQueueRunCommentIdsGuarded(tx, { + queueRun: locked.queueRun, + ids, + updatedAt: input.now, + }); + + return tx.buildQueueSnapshot({ + issue: input.issue, + actor: input.actor, + wake: locked.wake, + state: locked.state, + queueRun: updatedQueueRun ?? locked.queueRun, + activeRun: locked.activeRun, + }); + }, + ); + }; +} + +export type ReorderQueuedCommentsInput = { + issue: QueuedCommentIssueContext; + actor: QueuedCommentActor; + queueId: string; + revision: string; + orderedCommentIds: string[]; + now: Date; +}; + +export function createReorderQueuedComments(deps: { issueLock: QueuedCommentIssueLockWriter }) { + return async function reorderQueuedComments(input: ReorderQueuedCommentsInput): Promise { + return deps.issueLock.withLockedQueue( + { issue: input.issue, actor: input.actor, queueId: input.queueId }, + async (locked, tx) => { + requireMutationTarget(locked.queue, input.queueId, input.revision); + + const currentIds = locked.queue.entries.map((entry) => entry.comment.id); + const reorderDecision = decideQueuedCommentReorder({ currentIds, orderedIds: input.orderedCommentIds }); + if (reorderDecision.kind === "mismatch") { + throw new QueuedCommentMutationError( + "queued_comment_order_mismatch", + "The queued message order does not match the current queue", + ); + } + + const updatedWake = await tx.updateWakeQueuedCommentIds({ + wakeId: locked.wake.id, + payload: locked.wake.payload, + ids: input.orderedCommentIds, + updatedAt: input.now, + }); + const updatedQueueRun = await updateQueueRunCommentIdsGuarded(tx, { + queueRun: locked.queueRun, + ids: input.orderedCommentIds, + updatedAt: input.now, + }); + + return tx.buildQueueSnapshot({ + issue: input.issue, + actor: input.actor, + wake: updatedWake, + state: locked.state, + queueRun: updatedQueueRun ?? locked.queueRun, + activeRun: locked.activeRun, + }); + }, + ); + }; +} + +export type DiscardQueuedCommentInput = { + issue: QueuedCommentIssueContext; + actor: QueuedCommentActor; + commentId: string; + queueId: string; + /** Skipped entirely when omitted, matching the comment-delete route's cancellation call site, which does not carry a revision. */ + revision?: string; + now: Date; +}; + +export type DiscardQueuedCommentResult = { + /** The full deleted comment row; the comment-delete route echoes it back as its own response body. */ + deleted: IssueComment; + queue: QueuedCommentQueueSnapshot; + /** Set only when the discard emptied the queue and cancelled a queued run; the caller emits telemetry for it after the transaction commits. */ + cancelledRun: { id: string } | null; +}; + +export function createDiscardQueuedComment(deps: { issueLock: QueuedCommentIssueLockWriter }) { + return async function discardQueuedComment(input: DiscardQueuedCommentInput): Promise { + return deps.issueLock.withLockedQueue( + { issue: input.issue, actor: input.actor, queueId: input.queueId }, + async (locked, tx) => { + if (input.revision !== undefined) { + requireMutationTarget(locked.queue, input.queueId, input.revision); + } + + const entry = locked.queue.entries.find((candidate) => candidate.comment.id === input.commentId); + if (!entry) { + throw new QueuedCommentMutationError("queued_comment_not_pending", "The queued message is no longer pending"); + } + const owns = decideQueuedCommentActorOwnsEntry({ + actorType: input.actor.actorType, + actorId: input.actor.actorId, + actorAgentId: input.actor.agentId, + authorAgentId: entry.comment.authorAgentId, + authorUserId: entry.comment.authorUserId, + }); + if (!owns) { + throw new QueuedCommentMutationForbiddenError("Only the queued message author can discard it"); + } + + const deleted = await tx.deleteComment({ issueId: input.issue.id, commentId: input.commentId }); + if (!deleted) { + throw new QueuedCommentMutationError("queued_comment_not_pending", "The queued message is no longer pending"); + } + await tx.deleteCommentReferenceSource(input.commentId); + await tx.syncCommentExternalObjectsSafely(input.commentId); + + const remainingIds = locked.queue.entries.map((candidate) => candidate.comment.id).filter((id) => id !== input.commentId); + const queueBecomesEmpty = remainingIds.length === 0; + + let cancelledRun: { id: string } | null = null; + let nextWake = locked.wake; + let nextQueueRun = locked.queueRun; + + if (queueBecomesEmpty) { + await tx.cancelWake({ + wakeId: locked.wake.id, + reason: "Queued message discarded before dispatch", + now: input.now, + }); + if (locked.queueRun) { + const cancelled = await tx.cancelQueueRun({ + queueRunId: locked.queueRun.id, + reason: "Queued message discarded before dispatch", + now: input.now, + }); + if (!cancelled) { + throw new QueuedCommentMutationError( + "queued_comment_already_dispatching", + "The queued message is already being dispatched", + ); + } + cancelledRun = cancelled; + await tx.clearExecutionLockAndTouchIssue({ + issueId: input.issue.id, + executionRunId: locked.queueRun.id, + updatedAt: input.now, + }); + } else { + await tx.touchIssueUpdatedAt({ issueId: input.issue.id, updatedAt: input.now }); + } + } else { + nextWake = await tx.updateWakeQueuedCommentIds({ + wakeId: locked.wake.id, + payload: locked.wake.payload, + ids: remainingIds, + updatedAt: input.now, + }); + nextQueueRun = await updateQueueRunCommentIdsGuarded(tx, { + queueRun: locked.queueRun, + ids: remainingIds, + updatedAt: input.now, + }); + await tx.touchIssueUpdatedAt({ issueId: input.issue.id, updatedAt: input.now }); + } + + const queue = await tx.buildQueueSnapshot({ + issue: input.issue, + actor: input.actor, + wake: queueBecomesEmpty ? null : nextWake, + state: queueBecomesEmpty ? null : locked.state, + queueRun: queueBecomesEmpty ? null : (nextQueueRun ?? locked.queueRun), + activeRun: locked.activeRun, + }); + + return { deleted, queue, cancelledRun }; + }, + ); + }; +} diff --git a/server/src/modules/wake-queue/domain/policy.test.ts b/server/src/modules/wake-queue/domain/policy.test.ts index 12119c0138..1d09806c3c 100644 --- a/server/src/modules/wake-queue/domain/policy.test.ts +++ b/server/src/modules/wake-queue/domain/policy.test.ts @@ -2,6 +2,9 @@ import { describe, expect, it } from "vitest"; import { decidePreDrain, decideQueuedCommentAction, + decideQueuedCommentActorOwnsEntry, + decideQueuedCommentReorder, + decideQueuedCommentWakeLookup, decideReleaseRecovery, decideWakeAdmission, decideWakeOutcome, @@ -10,6 +13,9 @@ import { type DeferredWakeQueuedCommentFacts, type ImmediateRecoveryContextLabels, type PreDrainFacts, + type QueuedCommentActorOwnershipFacts, + type QueuedCommentReorderFacts, + type QueuedCommentWakeLookupFacts, type ReleaseRecoveryFacts, type WakeAdmissionFacts, } from "./policy.js"; @@ -544,3 +550,166 @@ describe("decideWakeAdmission", () => { }); } }); + +describe("decideQueuedCommentWakeLookup", () => { + const baseFacts: QueuedCommentWakeLookupFacts = { + wakePresent: true, + wakeIssueIdMatches: true, + hasQueuedCommentIds: true, + wakeStatus: "deferred_issue_execution", + wakeHasRunId: false, + }; + + const cases: Array<{ + name: string; + facts: QueuedCommentWakeLookupFacts; + expected: ReturnType; + }> = [ + { + name: "not_pending: no wake row was found", + facts: { ...baseFacts, wakePresent: false }, + expected: { kind: "not_pending" }, + }, + { + name: "not_pending: the wake's payload names a different issue", + facts: { ...baseFacts, wakeIssueIdMatches: false }, + expected: { kind: "not_pending" }, + }, + { + name: "not_pending: the wake's payload carries no queued comment ids", + facts: { ...baseFacts, hasQueuedCommentIds: false }, + expected: { kind: "not_pending" }, + }, + { + name: "deferred: the wake is still waiting behind an active execution run", + facts: baseFacts, + expected: { kind: "deferred" }, + }, + { + name: "check_queue_run: the wake is queued and carries a linked run id", + facts: { ...baseFacts, wakeStatus: "queued", wakeHasRunId: true }, + expected: { kind: "check_queue_run" }, + }, + { + name: "not_pending: the wake is queued but carries no linked run id", + facts: { ...baseFacts, wakeStatus: "queued", wakeHasRunId: false }, + expected: { kind: "not_pending" }, + }, + { + name: "already_dispatching: the wake was claimed", + facts: { ...baseFacts, wakeStatus: "claimed" }, + expected: { kind: "already_dispatching" }, + }, + { + name: "already_dispatching: the wake is running", + facts: { ...baseFacts, wakeStatus: "running" }, + expected: { kind: "already_dispatching" }, + }, + { + name: "already_dispatching: the wake succeeded and still carries a run id", + facts: { ...baseFacts, wakeStatus: "succeeded", wakeHasRunId: true }, + expected: { kind: "already_dispatching" }, + }, + { + name: "already_dispatching: the wake failed and still carries a run id", + facts: { ...baseFacts, wakeStatus: "failed", wakeHasRunId: true }, + expected: { kind: "already_dispatching" }, + }, + { + name: "not_pending: the wake succeeded but carries no run id", + facts: { ...baseFacts, wakeStatus: "succeeded", wakeHasRunId: false }, + expected: { kind: "not_pending" }, + }, + { + name: "not_pending: the wake was cancelled", + facts: { ...baseFacts, wakeStatus: "cancelled" }, + expected: { kind: "not_pending" }, + }, + ]; + + for (const testCase of cases) { + it(testCase.name, () => { + expect(decideQueuedCommentWakeLookup(testCase.facts)).toEqual(testCase.expected); + }); + } +}); + +describe("decideQueuedCommentReorder", () => { + const cases: Array<{ + name: string; + facts: QueuedCommentReorderFacts; + expected: ReturnType; + }> = [ + { + name: "ok: the submitted order is a permutation of the current ids", + facts: { currentIds: ["a", "b", "c"], orderedIds: ["c", "a", "b"] }, + expected: { kind: "ok" }, + }, + { + name: "mismatch: the submitted order carries a duplicate id", + facts: { currentIds: ["a", "b"], orderedIds: ["a", "a"] }, + expected: { kind: "mismatch" }, + }, + { + name: "mismatch: the submitted order drops an id", + facts: { currentIds: ["a", "b", "c"], orderedIds: ["a", "b"] }, + expected: { kind: "mismatch" }, + }, + { + name: "mismatch: the submitted order adds an id the queue does not have", + facts: { currentIds: ["a", "b"], orderedIds: ["a", "b", "c"] }, + expected: { kind: "mismatch" }, + }, + { + name: "mismatch: the submitted order names an id the current queue does not have", + facts: { currentIds: ["a", "b"], orderedIds: ["a", "c"] }, + expected: { kind: "mismatch" }, + }, + ]; + + for (const testCase of cases) { + it(testCase.name, () => { + expect(decideQueuedCommentReorder(testCase.facts)).toEqual(testCase.expected); + }); + } +}); + +describe("decideQueuedCommentActorOwnsEntry", () => { + const cases: Array<{ + name: string; + facts: QueuedCommentActorOwnershipFacts; + expected: boolean; + }> = [ + { + name: "owns: a user actor authored the comment", + facts: { actorType: "user", actorId: "user-1", actorAgentId: null, authorAgentId: null, authorUserId: "user-1" }, + expected: true, + }, + { + name: "does not own: a user actor did not author the comment", + facts: { actorType: "user", actorId: "user-1", actorAgentId: null, authorAgentId: null, authorUserId: "user-2" }, + expected: false, + }, + { + name: "owns: an agent actor authored the comment as that agent", + facts: { actorType: "agent", actorId: "agent-1", actorAgentId: "agent-1", authorAgentId: "agent-1", authorUserId: null }, + expected: true, + }, + { + name: "does not own: an agent actor authored a different comment", + facts: { actorType: "agent", actorId: "agent-1", actorAgentId: "agent-1", authorAgentId: "agent-2", authorUserId: null }, + expected: false, + }, + { + name: "does not own: an agent actor with no resolved agent id", + facts: { actorType: "agent", actorId: "agent-1", actorAgentId: null, authorAgentId: null, authorUserId: null }, + expected: false, + }, + ]; + + for (const testCase of cases) { + it(testCase.name, () => { + expect(decideQueuedCommentActorOwnsEntry(testCase.facts)).toBe(testCase.expected); + }); + } +}); diff --git a/server/src/modules/wake-queue/domain/policy.ts b/server/src/modules/wake-queue/domain/policy.ts index f19b6e7b34..8ef20f07af 100644 --- a/server/src/modules/wake-queue/domain/policy.ts +++ b/server/src/modules/wake-queue/domain/policy.ts @@ -273,6 +273,98 @@ export function deriveImmediateRecoveryContextLabels(issueStatus: string): Immed }; } +// Pure decision rules for the three queued-comment queue mutations (edit, +// reorder, discard): the queue-mutation target check every mutation runs +// first, the wake and queue-run status classification the initial lock +// step needs, the reorder set-equality check, the queue-entry permission +// fields the response carries, the actor-ownership check discard enforces, +// and the empty-versus-partial outcome a discard resolves to. As with every +// other decision in this file, the caller reads the database and packs the +// result into a facts object; this file only branches on that object. + +export type QueuedCommentWakeLookupFacts = { + /** True when a wake row was found for the submitted queue id. */ + wakePresent: boolean; + /** True when the wake's own payload still names this issue. */ + wakeIssueIdMatches: boolean; + /** True when the wake's payload still carries one or more queued comment ids. */ + hasQueuedCommentIds: boolean; + /** The wake row's own status. Meaningless when `wakePresent` is false. */ + wakeStatus: string | null; + /** True when the wake row carries a linked heartbeat run id. */ + wakeHasRunId: boolean; +}; + +export type QueuedCommentWakeLookupDecision = + | { kind: "not_pending" } + | { kind: "deferred" } + /** The caller must read the linked heartbeat run next and confirm it is still queued. */ + | { kind: "check_queue_run" } + | { kind: "already_dispatching" }; + +/** + * Classifies a locked wake row into the queue state a mutation needs: still + * waiting behind an active run (`deferred`), queued behind a not-yet-running + * turn (needs a second read to confirm, `check_queue_run`), already being + * dispatched, or no longer a pending queue at all. + */ +export function decideQueuedCommentWakeLookup(facts: QueuedCommentWakeLookupFacts): QueuedCommentWakeLookupDecision { + if (!facts.wakePresent || !facts.wakeIssueIdMatches || !facts.hasQueuedCommentIds) { + return { kind: "not_pending" }; + } + if (facts.wakeStatus === "deferred_issue_execution") return { kind: "deferred" }; + if (facts.wakeStatus === "queued" && facts.wakeHasRunId) return { kind: "check_queue_run" }; + if ( + facts.wakeStatus === "claimed" || + facts.wakeStatus === "running" || + (facts.wakeHasRunId && (facts.wakeStatus === "succeeded" || facts.wakeStatus === "failed")) + ) { + return { kind: "already_dispatching" }; + } + return { kind: "not_pending" }; +} + +export type QueuedCommentReorderFacts = { + currentIds: string[]; + orderedIds: string[]; +}; + +export type QueuedCommentReorderDecision = { kind: "ok" } | { kind: "mismatch" }; + +/** Decides whether a submitted order is a permutation of the queue's current comment ids: no duplicates, no drops, no additions. */ +export function decideQueuedCommentReorder(facts: QueuedCommentReorderFacts): QueuedCommentReorderDecision { + const orderedSet = new Set(facts.orderedIds); + if ( + orderedSet.size !== facts.orderedIds.length || + facts.orderedIds.length !== facts.currentIds.length || + facts.currentIds.some((id) => !orderedSet.has(id)) + ) { + return { kind: "mismatch" }; + } + return { kind: "ok" }; +} + +export type QueuedCommentActorOwnershipFacts = { + actorType: "agent" | "user"; + actorId: string; + actorAgentId: string | null; + authorAgentId: string | null; + authorUserId: string | null; +}; + +/** + * Decides whether the actor discarding a queued message authored it. A user + * actor must be the board user who wrote the message. An agent actor must be + * the agent that wrote it, because an agent actor can discard its own queued + * message through the general comment-delete route. + */ +export function decideQueuedCommentActorOwnsEntry(facts: QueuedCommentActorOwnershipFacts): boolean { + if (facts.actorType === "agent") { + return facts.actorAgentId !== null && facts.authorAgentId === facts.actorAgentId; + } + return facts.authorUserId === facts.actorId; +} + /** * Decides the release-recovery outcome once the deferred-wake queue is * empty and no wake was promoted. The review-participant branch and the diff --git a/server/src/modules/wake-queue/index.ts b/server/src/modules/wake-queue/index.ts index 57f02cca2c..7f6e086679 100644 --- a/server/src/modules/wake-queue/index.ts +++ b/server/src/modules/wake-queue/index.ts @@ -5,7 +5,14 @@ import { createWakeAdmissionReader, createWakeAdmissionWriter, } from "./adapters/postgres.js"; +import { createQueuedCommentIssueLockWriter } from "./adapters/queued-comment-postgres.js"; +import type { QueuedCommentQueuePostgresAdapterDeps } from "./adapters/queued-comment-postgres.js"; import { createAdmitWakeBehindIssueExecution, createReleaseIssueExecution } from "./application/use-cases.js"; +import { + createDiscardQueuedComment, + createEditQueuedComment, + createReorderQueuedComments, +} from "./application/queued-comment-use-cases.js"; import type { IssueSnapshot, RecoveryEscalationPort, @@ -29,6 +36,23 @@ export type { TransactionScope, } from "./application/ports.js"; export type { AdmitWakeBehindIssueExecutionInput, AdmitWakeBehindIssueExecutionResult, ReleaseIssueExecutionInput } from "./application/use-cases.js"; +export { + QueuedCommentMutationError, + QueuedCommentMutationForbiddenError, +} from "./application/queued-comment-use-cases.js"; +export type { + DiscardQueuedCommentInput, + DiscardQueuedCommentResult, + EditQueuedCommentInput, + QueuedCommentMutationErrorCode, + ReorderQueuedCommentsInput, +} from "./application/queued-comment-use-cases.js"; +export type { + QueuedCommentActor, + QueuedCommentIssueContext, + QueuedCommentQueueSnapshot, +} from "./application/queued-comment-ports.js"; +export type { QueuedCommentQueuePostgresAdapterDeps } from "./adapters/queued-comment-postgres.js"; export type WakeQueueDeps = { /** Stays in `heartbeat.ts`; resolves the responsible user for a promoted or recovery run seed. */ @@ -80,3 +104,23 @@ export function createWakeQueue(db: Db, deps: WakeQueueDeps) { } export type WakeQueue = ReturnType; + +/** + * Composes the three queued-comment queue mutations (edit, reorder, + * discard): the Postgres adapter, which owns the one transaction each + * mutation runs in, and the three use cases. This is a separate factory + * from `createWakeQueue` because these mutations need none of the release + * or admission host callbacks -- only the small set of comment-reference + * and external-object sync callbacks in `deps`, which a caller outside + * `heartbeat.ts` (the queued-comment route) can supply directly. + */ +export function createQueuedCommentQueue(db: Db, deps: QueuedCommentQueuePostgresAdapterDeps) { + const issueLock = createQueuedCommentIssueLockWriter(db, deps); + return { + editQueuedComment: createEditQueuedComment({ issueLock }), + reorderQueuedComments: createReorderQueuedComments({ issueLock }), + discardQueuedComment: createDiscardQueuedComment({ issueLock }), + }; +} + +export type QueuedCommentQueue = ReturnType; diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index b0b98a2e50..31cabea88e 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -123,6 +123,7 @@ import { type IssueRelationIssueSummary, type IssueReviewPolicy, type IssueThreadInteractionCanonicalResolverPolicy, + type IssueComment, type IssueCommentPresentation, type IssueQueuedCommentQueue, type IssueWatchdogDiscoveryKind, @@ -179,7 +180,13 @@ import { } from "../services/runner-goals.js"; import { queueLiveRunnerPrpCommand } from "../realtime/runner-prp-ws.js"; import { questionResponseDeliveryService } from "../services/question-response-delivery.js"; -import { emitAgentTaskRun } from "../services/agent-task-run-telemetry.js"; +import { emitAgentTaskRunById } from "../services/agent-task-run-telemetry.js"; +import { + createQueuedCommentQueue, + QueuedCommentMutationError, + QueuedCommentMutationForbiddenError, + type QueuedCommentIssueContext, +} from "../modules/wake-queue/index.js"; import { artifactReviewDocumentService } from "../services/artifact-review-documents.js"; import { assertCanResolveProposal } from "../services/secret-proposal-authorization.js"; import { @@ -330,8 +337,9 @@ import { steerNativeSession, } from "../services/native-runtime/native-session-executor.js"; import { + buildQueuedCommentQueueSnapshot, + decideQueuedCommentQueueSteering, queuedCommentIdsFromWakePayload, - withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload, } from "../services/issue-queued-comment-queue.js"; @@ -3689,6 +3697,11 @@ export function issueRoutes( enabled: async () => (await instanceSettings.getExperimental()).enableExternalObjects === true, }); + const queuedCommentQueue = createQueuedCommentQueue(db, { + syncCommentReferences: (commentId, tx) => issueReferencesSvc.syncComment(commentId, tx), + deleteCommentReferenceSource: (commentId, tx) => issueReferencesSvc.deleteCommentSource(commentId, tx), + syncCommentExternalObjectsSafely: (commentId, tx) => externalObjectsSvc.syncCommentSafely(commentId, tx), + }); const routinesSvc = routineService(db, { pluginWorkerManager: opts.pluginWorkerManager, }); @@ -6747,24 +6760,6 @@ export function issueRoutes( queueRun: IssueQueueRun | null; }; - function queueRevision(input: { - wake: IssueQueueWake | null; - comments: Array<{ id: string; updatedAt: Date }>; - }): string { - return createHash("sha256") - .update( - JSON.stringify({ - queueId: input.wake?.id ?? null, - comments: input.comments.map((comment) => [ - comment.id, - comment.updatedAt.toISOString(), - ]), - }), - ) - .digest("hex") - .slice(0, 32); - } - async function findQueuedCommentWake( executor: IssueQueueDb, issue: { id: string; companyId: string; assigneeAgentId: string | null }, @@ -6865,53 +6860,31 @@ export function issueRoutes( .limit(1) .then((rows) => rows[0] ?? null) : null; - const persistedRuntimeMode = - queueState?.state === "queued" && queueState.queueRun - ? queueState.queueRun.runtimeMode - : queueState?.state === "deferred" && input.activeRun - ? input.activeRun.runtimeMode - : null; - const protocol = - persistedRuntimeMode === "native" || - (persistedRuntimeMode === null && - assignedAgent?.adapterType === "paperclip_runner") - ? ("paperclip_runner_v1" as const) - : ("legacy" as const); - const steeringRun = - queueState?.state === "deferred" ? input.activeRun : null; - let steeringDisposition = - input.steeringDisposition ?? - (protocol === "paperclip_runner_v1" && steeringRun - ? await getNativeSessionSteeringState(steeringRun.id) + const steering = decideQueuedCommentQueueSteering({ + state: queueState?.state ?? null, + queueRunRuntimeMode: queueState?.state === "queued" ? queueState.queueRun?.runtimeMode ?? null : null, + activeRun: input.activeRun, + assignedAgentAdapterType: assignedAgent?.adapterType ?? null, + queuedCommentCount: comments.length, + }); + const steeringDisposition: IssueQueuedCommentQueue["steeringDisposition"] = + steering.kind !== "probe" + ? steering.kind + : input.steeringDisposition + ?? (await getNativeSessionSteeringState(steering.steeringRunId) .then((state) => state.disposition) - .catch(() => "temporarily_unavailable" as const) - : ("unsupported" as const)); - if ( - protocol === "paperclip_runner_v1" && - (!steeringRun || comments.length === 0) - ) { - steeringDisposition = "temporarily_unavailable"; - } - return { + .catch(() => "temporarily_unavailable" as const)); + return buildQueuedCommentQueueSnapshot({ issueId: input.issue.id, queueId: wake?.id ?? null, state: queueState?.state ?? null, - targetRunId: steeringRun?.id ?? null, - revision: queueRevision({ wake, comments }), - protocol, + activeRunId: input.activeRun?.id ?? null, + protocol: steering.protocol, steeringDisposition, - entries: comments.map((comment, position) => ({ - comment: - comment as IssueQueuedCommentQueue["entries"][number]["comment"], - position, - canEdit: - input.actor.actorType === "user" && - comment.authorUserId === input.actor.actorId, - canDiscard: - input.actor.actorType === "user" && - comment.authorUserId === input.actor.actorId, - })), - }; + comments, + actorType: input.actor.actorType, + actorId: input.actor.actorId, + }); } function assertQueueMutationTarget(input: { @@ -7063,211 +7036,7 @@ export function issueRoutes( return { activeRun, wake, queueRun, state, queue, queueState }; } - async function updateQueuedRunCommentIds( - tx: IssueQueueTx, - queueRun: IssueQueueRun | null, - ids: string[], - updatedAt: Date, - ) { - if (!queueRun) return null; - const updated = await tx - .update(heartbeatRuns) - .set({ - contextSnapshot: withQueuedCommentIdsInRunContext( - queueRun.contextSnapshot, - ids, - ), - updatedAt, - }) - .where( - and( - eq(heartbeatRuns.id, queueRun.id), - eq(heartbeatRuns.status, "queued"), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); - if (!updated) { - throw conflict("The queued message is already being dispatched", { - code: "queued_comment_already_dispatching", - }); - } - return updated; - } - - async function discardQueuedComment(input: { - issue: { - id: string; - companyId: string; - assigneeAgentId: string | null; - executionRunId?: string | null; - }; - actor: ReturnType; - commentId: string; - queueId: string; - revision?: string; - }) { - let cancelledRunToEmit: typeof heartbeatRuns.$inferSelect | null = null; - const result = await db.transaction(async (tx) => { - const locked = await lockQueuedCommentState({ - tx, - issue: input.issue, - actor: input.actor, - queueId: input.queueId, - }); - if (input.revision) { - assertQueueMutationTarget({ - queue: locked.queue, - queueId: input.queueId, - revision: input.revision, - }); - } - const entry = locked.queue.entries.find( - (candidate) => candidate.comment.id === input.commentId, - ); - if (!entry) { - throw conflict("The queued message is no longer pending", { - code: "queued_comment_not_pending", - }); - } - const actorOwnsEntry = - input.actor.actorType === "agent" - ? entry.comment.authorAgentId === input.actor.agentId - : entry.comment.authorUserId === input.actor.actorId; - if (!actorOwnsEntry) { - throw forbidden("Only the queued message author can discard it"); - } - - const deleted = await tx - .delete(issueComments) - .where( - and( - eq(issueComments.id, input.commentId), - eq(issueComments.issueId, input.issue.id), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); - if (!deleted) { - throw conflict("The queued message is no longer pending", { - code: "queued_comment_not_pending", - }); - } - await issueReferencesSvc.deleteCommentSource(input.commentId, tx); - await externalObjectsSvc.syncCommentSafely(input.commentId, tx); - - const remainingIds = locked.queue.entries - .map((candidate) => candidate.comment.id) - .filter((candidateId) => candidateId !== input.commentId); - const now = new Date(); - let nextQueueState: IssueQueueState | null = null; - - if (remainingIds.length === 0) { - await tx - .update(agentWakeupRequests) - .set({ - status: "cancelled", - finishedAt: now, - error: "Queued message discarded before dispatch", - updatedAt: now, - }) - .where(eq(agentWakeupRequests.id, locked.wake.id)); - - if (locked.queueRun) { - const cancelledRun = await tx - .update(heartbeatRuns) - .set({ - status: "cancelled", - finishedAt: now, - error: "Queued message discarded before dispatch", - errorCode: "queued_comment_discarded", - updatedAt: now, - }) - .where( - and( - eq(heartbeatRuns.id, locked.queueRun.id), - eq(heartbeatRuns.status, "queued"), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); - if (!cancelledRun) { - throw conflict("The queued message is already being dispatched", { - code: "queued_comment_already_dispatching", - }); - } - cancelledRunToEmit = cancelledRun; - } - } else { - const updatedWake = await tx - .update(agentWakeupRequests) - .set({ - payload: withQueuedCommentIdsInWakePayload( - locked.wake.payload, - remainingIds, - ), - updatedAt: now, - }) - .where(eq(agentWakeupRequests.id, locked.wake.id)) - .returning() - .then((rows) => rows[0] ?? locked.wake); - const updatedQueueRun = await updateQueuedRunCommentIds( - tx, - locked.queueRun, - remainingIds, - now, - ); - nextQueueState = { - wake: updatedWake, - state: locked.state, - queueRun: updatedQueueRun ?? locked.queueRun, - }; - } - - await tx - .update(issueRows) - .set({ - ...(locked.queueRun && remainingIds.length === 0 - ? { - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - } - : {}), - updatedAt: now, - }) - .where( - and( - eq(issueRows.id, input.issue.id), - locked.queueRun && remainingIds.length === 0 - ? eq(issueRows.executionRunId, locked.queueRun.id) - : undefined, - ), - ); - - return { - deleted, - queue: await buildQueuedCommentQueue({ - executor: tx, - issue: input.issue, - activeRun: locked.activeRun, - actor: input.actor, - queueState: nextQueueState, - }), - }; - }); - // Telemetry is best-effort background work; it must not delay the - // response with a slow lookup, so fire it and do not await it. - if (cancelledRunToEmit) { - void emitAgentTaskRun(db, cancelledRunToEmit); - } - return result; - } - - function operatorInterruptCancelOptions(input: { - issueId: string; - actor: ReturnType; - }) { + function operatorInterruptCancelOptions(input: { issueId: string; actor: ReturnType }) { return { errorCode: "operator_interrupted", resultJson: { @@ -15300,6 +15069,41 @@ export function issueRoutes( ); }); + /** Maps a wake-queue queued-comment mutation error onto the same HTTP error the route threw before this mutation moved into the module. */ + function throwForQueuedCommentMutationError(error: unknown): never { + if (error instanceof QueuedCommentMutationError) { + throw conflict(error.message, { code: error.code }); + } + if (error instanceof QueuedCommentMutationForbiddenError) { + throw forbidden(error.message); + } + throw error; + } + + /** Builds the four-field issue context every queued-comment queue mutation call site passes, from the route's already-loaded issue. */ + function buildQueuedCommentIssueContext(issue: { + id: string; + companyId: string; + assigneeAgentId: string | null; + executionRunId: string | null | undefined; + }): QueuedCommentIssueContext { + return { + id: issue.id, + companyId: issue.companyId, + assigneeAgentId: issue.assigneeAgentId, + executionRunId: issue.executionRunId ?? null, + }; + } + + /** Runs a queued-comment queue mutation and maps its error onto the route's HTTP error, so each call site is a `const`. */ + async function runQueuedCommentMutation(run: () => Promise): Promise { + try { + return await run(); + } catch (error) { + throwForQueuedCommentMutationError(error); + } + } + router.get("/issues/:id/queued-comments", async (req, res) => { const id = req.params.id as string; const issue = await getAccessibleResource( @@ -15337,70 +15141,18 @@ export function issueRoutes( ); if (!issue) return; const actor = getActorInfo(req); - const queue = await db.transaction(async (tx) => { - const locked = await lockQueuedCommentState({ - tx, - issue, + const queue = await runQueuedCommentMutation(() => + queuedCommentQueue.editQueuedComment({ + issue: buildQueuedCommentIssueContext(issue), actor, - queueId: req.body.queueId, - }); - assertQueueMutationTarget({ - queue: locked.queue, + commentId, queueId: req.body.queueId, revision: req.body.revision, - }); - const entry = locked.queue.entries.find( - (candidate) => candidate.comment.id === commentId, - ); - if (!entry) - throw conflict("The queued message is no longer pending", { - code: "queued_comment_not_pending", - }); - if (!entry.canEdit) - throw forbidden("Only the queued message author can edit it"); - const updatedAt = new Date(); - const updated = await tx - .update(issueComments) - .set({ body: req.body.body, updatedAt }) - .where( - and( - eq(issueComments.id, commentId), - eq(issueComments.issueId, issue.id), - ), - ) - .returning({ id: issueComments.id }) - .then((rows) => rows[0] ?? null); - if (!updated) - throw conflict("The queued message is no longer pending", { - code: "queued_comment_not_pending", - }); - await tx - .update(issueRows) - .set({ updatedAt }) - .where(eq(issueRows.id, issue.id)); - await issueReferencesSvc.syncComment(commentId, tx); - await externalObjectsSvc.syncCommentSafely(commentId, tx); - const updatedQueueRun = await updateQueuedRunCommentIds( - tx, - locked.queueRun, - locked.queue.entries.map((candidate) => candidate.comment.id), - updatedAt, - ); - return buildQueuedCommentQueue({ - executor: tx, - issue, - activeRun: locked.activeRun, - actor, - queueState: { - wake: locked.wake, - state: locked.state, - queueRun: updatedQueueRun ?? locked.queueRun, - }, - }); - }); - res.json( - await runRedactions.redactForIssue(issue.companyId, issue.id, queue), + body: req.body.body, + now: new Date(), + }), ); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); }, ); @@ -15419,69 +15171,17 @@ export function issueRoutes( ); if (!issue) return; const actor = getActorInfo(req); - const queue = await db.transaction(async (tx) => { - const locked = await lockQueuedCommentState({ - tx, - issue, + const queue = await runQueuedCommentMutation(() => + queuedCommentQueue.reorderQueuedComments({ + issue: buildQueuedCommentIssueContext(issue), actor, queueId: req.body.queueId, - }); - assertQueueMutationTarget({ - queue: locked.queue, - queueId: req.body.queueId, revision: req.body.revision, - }); - const currentIds = locked.queue.entries.map( - (entry) => entry.comment.id, - ); - const orderedIds = req.body.orderedCommentIds as string[]; - const orderedSet = new Set(orderedIds); - if ( - orderedSet.size !== orderedIds.length || - orderedIds.length !== currentIds.length || - currentIds.some((commentId) => !orderedSet.has(commentId)) - ) { - throw conflict( - "The queued message order does not match the current queue", - { - code: "queued_comment_order_mismatch", - }, - ); - } - const now = new Date(); - const updatedWake = await tx - .update(agentWakeupRequests) - .set({ - payload: withQueuedCommentIdsInWakePayload( - locked.wake.payload, - orderedIds, - ), - updatedAt: now, - }) - .where(eq(agentWakeupRequests.id, locked.wake.id)) - .returning() - .then((rows) => rows[0] ?? locked.wake); - const updatedQueueRun = await updateQueuedRunCommentIds( - tx, - locked.queueRun, - orderedIds, - now, - ); - return buildQueuedCommentQueue({ - executor: tx, - issue, - activeRun: locked.activeRun, - actor, - queueState: { - wake: updatedWake, - state: locked.state, - queueRun: updatedQueueRun ?? locked.queueRun, - }, - }); - }); - res.json( - await runRedactions.redactForIssue(issue.companyId, issue.id, queue), + orderedCommentIds: req.body.orderedCommentIds as string[], + now: new Date(), + }), ); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); }, ); @@ -15759,16 +15459,22 @@ export function issueRoutes( ); if (!issue) return; const actor = getActorInfo(req); - const { queue } = await discardQueuedComment({ - issue, - actor, - commentId, - queueId: req.body.queueId, - revision: req.body.revision, - }); - res.json( - await runRedactions.redactForIssue(issue.companyId, issue.id, queue), + const result = await runQueuedCommentMutation(() => + queuedCommentQueue.discardQueuedComment({ + issue: buildQueuedCommentIssueContext(issue), + actor, + commentId, + queueId: req.body.queueId, + revision: req.body.revision, + now: new Date(), + }), ); + // Telemetry is best-effort background work; it must not delay the + // response with a slow lookup, so fire it and do not await it. + if (result.cancelledRun) { + void emitAgentTaskRunById(db, { runId: result.cancelledRun.id, companyId: issue.companyId }); + } + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, result.queue)); }, ); @@ -16932,20 +16638,25 @@ export function issueRoutes( return; } - const queueWakeForCancellation = - deleteMode === "cancel" ? authoritativeQueueWake : pendingQueueWake; - const removed = queueWakeForCancellation - ? ( - await discardQueuedComment({ - issue, + const queueWakeForCancellation = deleteMode === "cancel" + ? authoritativeQueueWake + : pendingQueueWake; + let removed: IssueComment | null; + if (queueWakeForCancellation) { + removed = ( + await runQueuedCommentMutation(() => + queuedCommentQueue.discardQueuedComment({ + issue: buildQueuedCommentIssueContext(issue), actor, commentId, queueId: queueWakeForCancellation.id, - }) - ).deleted - : activeRun && isLegacyQueuedComment - ? await svc.removeComment(commentId) - : null; + now: new Date(), + }), + ) + ).deleted; + } else { + removed = activeRun && isLegacyQueuedComment ? await svc.removeComment(commentId) : null; + } if (!removed) { res.status(409).json({ error: activeRun @@ -16954,7 +16665,6 @@ export function issueRoutes( }); return; } - await logActivity(db, { companyId: issue.companyId, actorType: actor.actorType, diff --git a/server/src/services/issue-queued-comment-queue.test.ts b/server/src/services/issue-queued-comment-queue.test.ts new file mode 100644 index 0000000000..a7c771641b --- /dev/null +++ b/server/src/services/issue-queued-comment-queue.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it } from "vitest"; +import { buildQueuedCommentQueueSnapshot, decideQueuedCommentQueueSteering } from "./issue-queued-comment-queue.js"; + +describe("decideQueuedCommentQueueSteering", () => { + it("answers unsupported on the legacy protocol", () => { + const decision = decideQueuedCommentQueueSteering({ + state: "deferred", + queueRunRuntimeMode: null, + activeRun: { id: "run-1", runtimeMode: "legacy" }, + assignedAgentAdapterType: "codex_local", + queuedCommentCount: 1, + }); + + expect(decision).toEqual({ protocol: "legacy", kind: "unsupported" }); + }); + + it("answers temporarily_unavailable for a promoted native queue with no deferred run", () => { + const decision = decideQueuedCommentQueueSteering({ + state: "queued", + queueRunRuntimeMode: "native", + activeRun: null, + assignedAgentAdapterType: "paperclip_runner", + queuedCommentCount: 1, + }); + + expect(decision).toEqual({ protocol: "paperclip_runner_v1", kind: "temporarily_unavailable" }); + }); + + it("answers temporarily_unavailable when the queue holds no live comments", () => { + const decision = decideQueuedCommentQueueSteering({ + state: "deferred", + queueRunRuntimeMode: null, + activeRun: { id: "run-1", runtimeMode: "native" }, + assignedAgentAdapterType: "paperclip_runner", + queuedCommentCount: 0, + }); + + expect(decision).toEqual({ protocol: "paperclip_runner_v1", kind: "temporarily_unavailable" }); + }); + + it("tells the caller it may probe a running deferred turn on the native protocol", () => { + const decision = decideQueuedCommentQueueSteering({ + state: "deferred", + queueRunRuntimeMode: null, + activeRun: { id: "run-1", runtimeMode: "native" }, + assignedAgentAdapterType: "paperclip_runner", + queuedCommentCount: 1, + }); + + expect(decision).toEqual({ protocol: "paperclip_runner_v1", kind: "probe", steeringRunId: "run-1" }); + }); + + // Acceptance-criterion fact pattern: a deferred queue whose active run + // has not resolved a runtime mode yet, for an agent on the + // `paperclip_runner` adapter. The protocol resolves to + // `paperclip_runner_v1` through the adapter-type fallback, and the + // decision hands the run to the caller to probe live — it never answers + // the flat "unsupported" value a duplicated, unshared rule can drift to. + it("resolves the protocol through the adapter-type fallback and asks the caller to probe", () => { + const decision = decideQueuedCommentQueueSteering({ + state: "deferred", + queueRunRuntimeMode: null, + activeRun: { id: "run-1", runtimeMode: null }, + assignedAgentAdapterType: "paperclip_runner", + queuedCommentCount: 1, + }); + + expect(decision).toEqual({ protocol: "paperclip_runner_v1", kind: "probe", steeringRunId: "run-1" }); + }); +}); + +describe("buildQueuedCommentQueueSnapshot entry permissions", () => { + const baseFacts = { + issueId: "issue-1", + queueId: "queue-1", + state: "queued" as const, + activeRunId: null, + protocol: "legacy" as const, + steeringDisposition: "unsupported" as const, + }; + + it("grants edit and discard to the user who authored the queued comment", () => { + const queue = buildQueuedCommentQueueSnapshot({ + ...baseFacts, + actorType: "user", + actorId: "user-1", + comments: [{ id: "comment-1", updatedAt: new Date(), authorUserId: "user-1" }], + }); + + expect(queue.entries[0]?.canEdit).toBe(true); + expect(queue.entries[0]?.canDiscard).toBe(true); + }); + + it("denies edit and discard to a user who did not author the queued comment", () => { + const queue = buildQueuedCommentQueueSnapshot({ + ...baseFacts, + actorType: "user", + actorId: "user-1", + comments: [{ id: "comment-1", updatedAt: new Date(), authorUserId: "user-2" }], + }); + + expect(queue.entries[0]?.canEdit).toBe(false); + expect(queue.entries[0]?.canDiscard).toBe(false); + }); + + it("denies edit and discard to an agent actor even when the comment carries a matching author id", () => { + const queue = buildQueuedCommentQueueSnapshot({ + ...baseFacts, + actorType: "agent", + actorId: "user-1", + comments: [{ id: "comment-1", updatedAt: new Date(), authorUserId: "user-1" }], + }); + + expect(queue.entries[0]?.canEdit).toBe(false); + expect(queue.entries[0]?.canDiscard).toBe(false); + }); +}); diff --git a/server/src/services/issue-queued-comment-queue.ts b/server/src/services/issue-queued-comment-queue.ts index 13700f33ad..0fb59fc625 100644 --- a/server/src/services/issue-queued-comment-queue.ts +++ b/server/src/services/issue-queued-comment-queue.ts @@ -1,3 +1,6 @@ +import { createHash } from "node:crypto"; +import type { IssueComment, IssueQueuedCommentQueue } from "@paperclipai/shared"; + const QUEUE_CONTEXT_KEY = "_paperclipWakeContext"; const QUEUE_IDS_KEY = "wakeCommentIds"; @@ -49,6 +52,27 @@ export function withQueuedCommentIdsInWakePayload( return payload; } +/** + * Fingerprints one queued-comment queue: the wake id plus every comment's id + * and last-updated time. A mutation that changes the queue changes this + * value, so a caller can echo it back to detect a queue it no longer holds + * the latest view of. The read path (`GET /queued-comments`) and every queue + * mutation must call this same function, so a client's fingerprint always + * compares against the same computation. + */ +export function queuedCommentQueueRevision(input: { + queueId: string | null; + comments: Array<{ id: string; updatedAt: Date }>; +}): string { + return createHash("sha256") + .update(JSON.stringify({ + queueId: input.queueId, + comments: input.comments.map((comment) => [comment.id, comment.updatedAt.toISOString()]), + })) + .digest("hex") + .slice(0, 32); +} + export function withQueuedCommentIdsInRunContext( contextValue: unknown, ids: string[], @@ -73,3 +97,102 @@ export function withQueuedCommentIdsInRunContext( delete context.paperclipTaskMarkdownCompact; return context; } + +export type QueuedCommentQueueProtocol = "paperclip_runner_v1" | "legacy"; + +export type QueuedCommentQueueSteeringDecision = + | { protocol: QueuedCommentQueueProtocol; kind: "unsupported" } + | { protocol: QueuedCommentQueueProtocol; kind: "temporarily_unavailable" } + /** Only the caller can probe the live runner. `steeringRunId` names the run to probe. */ + | { protocol: "paperclip_runner_v1"; kind: "probe"; steeringRunId: string }; + +/** + * Decides the queue protocol and the steering answer for one queued-comment + * queue, from plain facts. This is the one place that resolves the + * `paperclip_runner_v1`/`legacy` protocol and the steering answer; every + * caller that builds a queue response must call this function instead of + * repeating the rule. + * + * When the decision is `"probe"`, only the caller can answer the question: + * it must ask the live runner (through a call such as + * `getNativeSessionSteeringState`) and fall back to + * `"temporarily_unavailable"` on failure. A caller that never probes the + * live runner must answer `"temporarily_unavailable"` for a `"probe"` + * decision instead. + */ +export function decideQueuedCommentQueueSteering(facts: { + state: "deferred" | "queued" | null; + /** The queued run's own runtime mode. Read only when `state` is `"queued"`. */ + queueRunRuntimeMode: string | null; + /** The currently running turn, if any. Read only when `state` is `"deferred"`. */ + activeRun: { id: string; runtimeMode: string | null } | null; + assignedAgentAdapterType: string | null; + queuedCommentCount: number; +}): QueuedCommentQueueSteeringDecision { + const persistedRuntimeMode = + facts.state === "queued" + ? facts.queueRunRuntimeMode + : facts.state === "deferred" + ? facts.activeRun?.runtimeMode ?? null + : null; + + const protocol: QueuedCommentQueueProtocol = + persistedRuntimeMode === "native" + || (persistedRuntimeMode === null && facts.assignedAgentAdapterType === "paperclip_runner") + ? "paperclip_runner_v1" + : "legacy"; + + if (protocol !== "paperclip_runner_v1") { + return { protocol, kind: "unsupported" }; + } + + const steeringRun = facts.state === "deferred" ? facts.activeRun : null; + if (!steeringRun || facts.queuedCommentCount === 0) { + return { protocol, kind: "temporarily_unavailable" }; + } + + return { protocol, kind: "probe", steeringRunId: steeringRun.id }; +} + +type QueuedCommentQueueEntryFacts = { + id: string; + updatedAt: Date; + authorUserId: string | null; +}; + +/** + * Assembles the shared `IssueQueuedCommentQueue` response from the already + * resolved facts: the queue identity, the already-decided protocol and + * steering answer, the live comment rows, and the actor who reads the + * queue. This is the one place that builds the response shape; every + * caller that builds a queue response must call this function instead of + * repeating the rule. + */ +export function buildQueuedCommentQueueSnapshot(facts: { + issueId: string; + queueId: string | null; + state: "deferred" | "queued" | null; + /** The currently running turn's id. Read only when `state` is `"deferred"`. */ + activeRunId: string | null; + protocol: QueuedCommentQueueProtocol; + steeringDisposition: IssueQueuedCommentQueue["steeringDisposition"]; + comments: TComment[]; + actorType: "agent" | "user"; + actorId: string; +}): IssueQueuedCommentQueue { + return { + issueId: facts.issueId, + queueId: facts.queueId, + state: facts.state, + targetRunId: facts.state === "deferred" ? facts.activeRunId : null, + revision: queuedCommentQueueRevision({ queueId: facts.queueId, comments: facts.comments }), + protocol: facts.protocol, + steeringDisposition: facts.steeringDisposition, + entries: facts.comments.map((comment, position) => ({ + comment: comment as unknown as IssueComment, + position, + canEdit: facts.actorType === "user" && comment.authorUserId === facts.actorId, + canDiscard: facts.actorType === "user" && comment.authorUserId === facts.actorId, + })), + }; +}