mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 18:19:05 +02:00
refactor(server): move the queued-comment queue mutations into the wake-queue module (#13145)
## 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 <noreply@paperclip.ing>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed
1 parent
889947c238
commit
0d8bbf7cf4
12 files changed
+2077
-407
No files matched your search
@@ -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<void>;
|
||||
/** `issueReferenceService(db).deleteCommentSource`; runs on the module's own transaction. */
|
||||
deleteCommentReferenceSource(commentId: string, tx: Db): Promise<void>;
|
||||
/** `externalObjectService(db, opts).syncCommentSafely`; runs on the module's own transaction. */
|
||||
syncCommentExternalObjectsSafely(commentId: string, tx: Db): Promise<void>;
|
||||
};
|
||||
|
||||
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<IssueQueuedCommentQueue> {
|
||||
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);
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user