From 9cac22ecbde8e93003e44bcbaa39c68f1d4c78be Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 24 Apr 2026 09:05:43 -0500 Subject: [PATCH] Harden pause-hold wake provenance Co-Authored-By: Paperclip --- .../heartbeat-dependency-scheduling.test.ts | 52 ++++++- .../issue-tree-control-service.test.ts | 107 ++++++++++++- server/src/services/heartbeat.ts | 28 +++- server/src/services/issue-tree-control.ts | 146 ++++++++++++++++++ server/src/services/issues.ts | 39 ++--- 5 files changed, 341 insertions(+), 31 deletions(-) diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index c08c2008334..ca280dc8c02 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -424,12 +424,39 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = .then((rows) => rows[0] ?? null); expect(skippedWake).toMatchObject({ status: "skipped", reason: "issue_tree_hold_active" }); + const childCommentId = randomUUID(); + await db.insert(issueComments).values({ + id: childCommentId, + companyId, + issueId: childIssueId, + authorUserId: "board-user", + body: "Please respond while this hold is active.", + }); + + const forgedChildCommentWake = await heartbeat.wakeup(agentId, { + source: "on_demand", + triggerDetail: "manual", + reason: "issue_commented", + payload: { issueId: childIssueId, commentId: childCommentId }, + requestedByActorType: "agent", + requestedByActorId: agentId, + }); + expect(forgedChildCommentWake).toBeNull(); + const childCommentWake = await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", - payload: { issueId: childIssueId, commentId: randomUUID() }, - contextSnapshot: { issueId: childIssueId, wakeReason: "issue_commented" }, + payload: { issueId: childIssueId, commentId: childCommentId }, + requestedByActorType: "user", + requestedByActorId: "board-user", + contextSnapshot: { + issueId: childIssueId, + commentId: childCommentId, + wakeCommentId: childCommentId, + wakeReason: "issue_commented", + source: "issue.comment", + }, }); expect(childCommentWake).not.toBeNull(); @@ -493,12 +520,29 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = releasePolicy: { strategy: "manual", note: "full_pause" }, }); + const rootCommentId = randomUUID(); + await db.insert(issueComments).values({ + id: rootCommentId, + companyId, + issueId: rootIssueId, + authorUserId: "board-user", + body: "Please respond while this hold is active.", + }); + const rootCommentWake = await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", - payload: { issueId: rootIssueId, commentId: randomUUID() }, - contextSnapshot: { issueId: rootIssueId, wakeReason: "issue_commented" }, + payload: { issueId: rootIssueId, commentId: rootCommentId }, + requestedByActorType: "user", + requestedByActorId: "board-user", + contextSnapshot: { + issueId: rootIssueId, + commentId: rootCommentId, + wakeCommentId: rootCommentId, + wakeReason: "issue_commented", + source: "issue.comment", + }, }); expect(rootCommentWake).not.toBeNull(); diff --git a/server/src/__tests__/issue-tree-control-service.test.ts b/server/src/__tests__/issue-tree-control-service.test.ts index d9db94724bc..60953c7c05c 100644 --- a/server/src/__tests__/issue-tree-control-service.test.ts +++ b/server/src/__tests__/issue-tree-control-service.test.ts @@ -3,9 +3,11 @@ import { eq, inArray } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { agents, + agentWakeupRequests, companies, createDb, heartbeatRuns, + issueComments, issueTreeHoldMembers, issueTreeHolds, issues, @@ -38,8 +40,10 @@ describeEmbeddedPostgres("issueTreeControlService", () => { afterEach(async () => { await db.delete(issueTreeHoldMembers); await db.delete(issueTreeHolds); + await db.delete(issueComments); await db.delete(issues); await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); await db.delete(agents); await db.delete(companies); }); @@ -340,6 +344,12 @@ describeEmbeddedPostgres("issueTreeControlService", () => { const childIssueId = randomUUID(); const rootRunId = randomUUID(); const childRunId = randomUUID(); + const forgedRunId = randomUUID(); + const rootWakeupRequestId = randomUUID(); + const childWakeupRequestId = randomUUID(); + const forgedWakeupRequestId = randomUUID(); + const rootCommentId = randomUUID(); + const childCommentId = randomUUID(); await db.insert(companies).values({ id: companyId, @@ -377,6 +387,63 @@ describeEmbeddedPostgres("issueTreeControlService", () => { assigneeAgentId: agentId, }, ]); + await db.insert(issueComments).values([ + { + id: rootCommentId, + companyId, + issueId: rootIssueId, + authorUserId: "board-user", + body: "Please answer this root issue question.", + }, + { + id: childCommentId, + companyId, + issueId: childIssueId, + authorUserId: "board-user", + body: "Please answer this child issue question.", + }, + ]); + await db.insert(agentWakeupRequests).values([ + { + id: rootWakeupRequestId, + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId: rootIssueId, commentId: rootCommentId }, + status: "queued", + requestedByActorType: "user", + requestedByActorId: "board-user", + runId: rootRunId, + }, + { + id: forgedWakeupRequestId, + companyId, + agentId, + source: "on_demand", + triggerDetail: "manual", + reason: "issue_commented", + payload: { issueId: childIssueId, commentId: childCommentId }, + status: "queued", + requestedByActorType: "agent", + requestedByActorId: agentId, + runId: forgedRunId, + }, + { + id: childWakeupRequestId, + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId: childIssueId, commentId: childCommentId }, + status: "queued", + requestedByActorType: "user", + requestedByActorId: "board-user", + runId: childRunId, + }, + ]); await db.insert(heartbeatRuns).values([ { id: rootRunId, @@ -385,7 +452,29 @@ describeEmbeddedPostgres("issueTreeControlService", () => { invocationSource: "automation", triggerDetail: "system", status: "queued", - contextSnapshot: { issueId: rootIssueId, wakeReason: "issue_commented", commentId: randomUUID() }, + wakeupRequestId: rootWakeupRequestId, + contextSnapshot: { + issueId: rootIssueId, + wakeReason: "issue_commented", + commentId: rootCommentId, + wakeCommentId: rootCommentId, + source: "issue.comment", + }, + }, + { + id: forgedRunId, + companyId, + agentId, + invocationSource: "on_demand", + triggerDetail: "manual", + status: "queued", + wakeupRequestId: forgedWakeupRequestId, + contextSnapshot: { + issueId: childIssueId, + wakeReason: "issue_commented", + commentId: childCommentId, + wakeCommentId: childCommentId, + }, }, { id: childRunId, @@ -394,7 +483,14 @@ describeEmbeddedPostgres("issueTreeControlService", () => { invocationSource: "automation", triggerDetail: "system", status: "queued", - contextSnapshot: { issueId: childIssueId, wakeReason: "issue_commented", commentId: randomUUID() }, + wakeupRequestId: childWakeupRequestId, + contextSnapshot: { + issueId: childIssueId, + wakeReason: "issue_commented", + commentId: childCommentId, + wakeCommentId: childCommentId, + source: "issue.comment", + }, }, ]); @@ -413,6 +509,13 @@ describeEmbeddedPostgres("issueTreeControlService", () => { mode: "pause", }), }); + await expect(issueSvc.checkout(childIssueId, agentId, ["todo"], forgedRunId)).rejects.toMatchObject({ + status: 409, + details: expect.objectContaining({ + rootIssueId, + mode: "pause", + }), + }); const checkedOutChild = await issueSvc.checkout(childIssueId, agentId, ["todo"], childRunId); expect(checkedOutChild.status).toBe("in_progress"); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index a432fec602f..aafff481b5f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -85,6 +85,7 @@ import { import { issueService } from "./issues.js"; import { ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, + isVerifiedIssueTreeControlInteractionWake, issueTreeControlService, } from "./issue-tree-control.js"; import { @@ -3662,7 +3663,14 @@ export function heartbeatService(db: Db) { const issueId = readNonEmptyString(context.issueId); if (issueId) { const activePauseHold = await treeControlSvc.getActivePauseHoldGate(run.companyId, issueId); - const treeHoldInteractionWake = activePauseHold && allowsIssueInteractionWake(context); + const treeHoldInteractionWake = activePauseHold && await isVerifiedIssueTreeControlInteractionWake(db, { + companyId: run.companyId, + issueId, + agentId: run.agentId, + runId: run.id, + wakeupRequestId: run.wakeupRequestId, + contextSnapshot: context, + }); if (activePauseHold && !treeHoldInteractionWake) { await cancelRunInternal(run.id, "Cancelled because issue is held by an active subtree pause hold"); await logActivity(db, { @@ -5701,7 +5709,14 @@ export function heartbeatService(db: Db) { const deferredPayload = parseObject(deferred.payload); const deferredContextSeed = parseObject(deferredPayload[DEFERRED_WAKE_CONTEXT_KEY]); const activePauseHold = await treeControlSvc.getActivePauseHoldGate(issue.companyId, issue.id); - const treeHoldInteractionWake = activePauseHold && allowsIssueInteractionWake(deferredContextSeed); + const treeHoldInteractionWake = activePauseHold && await isVerifiedIssueTreeControlInteractionWake(tx, { + companyId: issue.companyId, + issueId: issue.id, + agentId: deferred.agentId, + contextSnapshot: deferredContextSeed, + requestedByActorType: deferred.requestedByActorType, + requestedByActorId: deferred.requestedByActorId, + }); if (activePauseHold && !treeHoldInteractionWake) { await tx .update(agentWakeupRequests) @@ -6108,7 +6123,14 @@ export function heartbeatService(db: Db) { if (issueId) { const activePauseHold = await treeControlSvc.getActivePauseHoldGate(agent.companyId, issueId); if (activePauseHold) { - const treeHoldInteractionWake = allowsIssueInteractionWake(enrichedContextSnapshot); + const treeHoldInteractionWake = await isVerifiedIssueTreeControlInteractionWake(db, { + companyId: agent.companyId, + issueId, + agentId, + contextSnapshot: enrichedContextSnapshot, + requestedByActorType: opts.requestedByActorType, + requestedByActorId: opts.requestedByActorId, + }); if (!treeHoldInteractionWake) { await writeSkippedRequest("issue_tree_hold_active"); diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index ca57ebb4b0e..04de642e1d5 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -3,6 +3,7 @@ import type { Db } from "@paperclipai/db"; import { agentWakeupRequests, heartbeatRuns, + issueComments, issueTreeHoldMembers, issueTreeHolds, issues, @@ -76,6 +77,151 @@ export const ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS: ReadonlySet = "issue_reopened_via_comment", "issue_comment_mentioned", ] as const); +const ISSUE_TREE_CONTROL_INTERACTION_WAKE_SOURCES: Readonly>> = { + issue_commented: new Set(["issue.comment"]), + issue_reopened_via_comment: new Set(["issue.comment.reopen"]), + issue_comment_mentioned: new Set(["comment.mention"]), +}; + +type VerifiedInteractionActor = { + requestedByActorType?: string | null; + requestedByActorId?: string | null; +}; + +function readNonEmptyStringFromRecord(record: unknown, key: string) { + if (!record || typeof record !== "object") return null; + const value = (record as Record)[key]; + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function readInteractionWakeCommentId(record: unknown) { + if (!record || typeof record !== "object") return null; + const value = (record as Record).wakeCommentIds; + if (Array.isArray(value)) { + const latest = value + .filter((entry): entry is string => typeof entry === "string" && entry.trim().length > 0) + .at(-1); + if (latest) return latest.trim(); + } + return readNonEmptyStringFromRecord(record, "wakeCommentId") ?? readNonEmptyStringFromRecord(record, "commentId"); +} + +function hasVerifiedInteractionSource(wakeReason: string, contextSnapshot: Record) { + const source = readNonEmptyStringFromRecord(contextSnapshot, "source"); + if (!source) return false; + return ISSUE_TREE_CONTROL_INTERACTION_WAKE_SOURCES[wakeReason]?.has(source) ?? false; +} + +function actorMatchesComment( + actor: VerifiedInteractionActor, + comment: { authorAgentId: string | null; authorUserId: string | null }, +) { + if (!actor.requestedByActorType) return false; + if (actor.requestedByActorType === "system") return true; + if (!actor.requestedByActorId) return false; + if (actor.requestedByActorType === "agent") return comment.authorAgentId === actor.requestedByActorId; + if (actor.requestedByActorType === "user") return comment.authorUserId === actor.requestedByActorId; + return false; +} + +async function hasVerifiedInteractionWakeRequest( + dbOrTx: Pick, + input: { + companyId: string; + agentId?: string | null; + runId?: string | null; + wakeupRequestId?: string | null; + issueId: string; + commentId: string; + comment: { authorAgentId: string | null; authorUserId: string | null }; + }, +) { + if (!input.runId && !input.wakeupRequestId) return false; + const predicates = [ + eq(agentWakeupRequests.companyId, input.companyId), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.issueId}`, + sql`${agentWakeupRequests.payload} ->> 'commentId' = ${input.commentId}`, + ]; + if (input.agentId) predicates.push(eq(agentWakeupRequests.agentId, input.agentId)); + if (input.runId && input.wakeupRequestId) { + const requestScope = or( + eq(agentWakeupRequests.runId, input.runId), + eq(agentWakeupRequests.id, input.wakeupRequestId), + ); + if (requestScope) predicates.push(requestScope); + } else if (input.runId) { + predicates.push(eq(agentWakeupRequests.runId, input.runId)); + } else if (input.wakeupRequestId) { + predicates.push(eq(agentWakeupRequests.id, input.wakeupRequestId)); + } + + const requests = await dbOrTx + .select({ + requestedByActorType: agentWakeupRequests.requestedByActorType, + requestedByActorId: agentWakeupRequests.requestedByActorId, + }) + .from(agentWakeupRequests) + .where(and(...predicates)); + + return requests.some((request) => actorMatchesComment(request, input.comment)); +} + +export async function isVerifiedIssueTreeControlInteractionWake( + dbOrTx: Pick, + input: { + companyId: string; + issueId: string; + agentId?: string | null; + contextSnapshot: Record | null | undefined; + requestedByActorType?: "user" | "agent" | "system" | string | null; + requestedByActorId?: string | null; + runId?: string | null; + wakeupRequestId?: string | null; + }, +) { + const contextSnapshot = input.contextSnapshot ?? null; + const wakeReason = + readNonEmptyStringFromRecord(contextSnapshot, "wakeReason") ?? + readNonEmptyStringFromRecord(contextSnapshot, "reason"); + if (!wakeReason || !ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS.has(wakeReason)) return false; + if (!contextSnapshot || !hasVerifiedInteractionSource(wakeReason, contextSnapshot)) return false; + + const commentId = readInteractionWakeCommentId(contextSnapshot); + if (!commentId) return false; + + const comment = await dbOrTx + .select({ + id: issueComments.id, + authorAgentId: issueComments.authorAgentId, + authorUserId: issueComments.authorUserId, + }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, input.companyId), + eq(issueComments.issueId, input.issueId), + eq(issueComments.id, commentId), + ), + ) + .then((rows) => rows[0] ?? null); + if (!comment) return false; + + const directActor = { + requestedByActorType: input.requestedByActorType, + requestedByActorId: input.requestedByActorId, + }; + if (actorMatchesComment(directActor, comment)) return true; + + return hasVerifiedInteractionWakeRequest(dbOrTx, { + companyId: input.companyId, + agentId: input.agentId, + runId: input.runId, + wakeupRequestId: input.wakeupRequestId, + issueId: input.issueId, + commentId, + comment, + }); +} function normalizeReleasePolicy( releasePolicy: IssueTreeHoldReleasePolicy | null | undefined, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 36e9ac97afe..1f1254775d2 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -38,7 +38,7 @@ import { redactCurrentUserText } from "../log-redaction.js"; import { resolveIssueGoalId, resolveNextIssueGoalId } from "./issue-goal-fallback.js"; import { getDefaultCompanyGoal } from "./goals.js"; import { - ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, + isVerifiedIssueTreeControlInteractionWake, issueTreeControlService, type ActiveIssueTreePauseHoldGate, } from "./issue-tree-control.js"; @@ -82,18 +82,6 @@ function readStringFromRecord(record: unknown, key: string) { return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; } -function readLatestWakeCommentId(record: unknown) { - if (!record || typeof record !== "object") return null; - const value = (record as Record).wakeCommentIds; - if (Array.isArray(value)) { - const latest = value - .filter((entry): entry is string => typeof entry === "string" && entry.trim().length > 0) - .at(-1); - if (latest) return latest.trim(); - } - return readStringFromRecord(record, "wakeCommentId") ?? readStringFromRecord(record, "commentId"); -} - export interface IssueFilters { status?: string; assigneeAgentId?: string; @@ -956,18 +944,25 @@ export function issueService(db: Db) { ) { if (!checkoutRunId) return false; const run = await db - .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .select({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + contextSnapshot: heartbeatRuns.contextSnapshot, + }) .from(heartbeatRuns) .where(and(eq(heartbeatRuns.id, checkoutRunId), eq(heartbeatRuns.companyId, companyId))) .then((rows) => rows[0] ?? null); - const wakeReason = - readStringFromRecord(run?.contextSnapshot, "wakeReason") ?? - readStringFromRecord(run?.contextSnapshot, "reason"); - return Boolean( - wakeReason && - ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS.has(wakeReason) && - readLatestWakeCommentId(run?.contextSnapshot), - ); + const issueId = readStringFromRecord(run?.contextSnapshot, "issueId"); + if (!run || !issueId) return false; + return isVerifiedIssueTreeControlInteractionWake(db, { + companyId, + issueId, + agentId: run.agentId, + runId: run.id, + wakeupRequestId: run.wakeupRequestId, + contextSnapshot: run.contextSnapshot as Record | null | undefined, + }); } async function assertAssignableUser(companyId: string, userId: string) {