diff --git a/doc/run-log-events.md b/doc/run-log-events.md index c0e89d5da38..417088809f0 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -862,7 +862,11 @@ or its owner changed, the existing cancellation settlement records `continuation_task_ownership_changed`. The run and wake request become cancelled before adapter dispatch, with the run-log message `stale execution continuation cancelled before dispatch`. Immediate recovery is -suppressed. Missing source context, authorization failures, and other setup +suppressed. A comment the cancelled run carried that no run read is saved +first when an agent run closed the task: a person's comment, or another agent's +comment posted with `resume: true`. The release then reopens the task and +delivers that comment once. Another agent's comment without `resume: true` +stays communicative and does not reopen the task. Missing source context, authorization failures, and other setup errors retain their failure classification. An untyped error with the same message is also still a failure; cancellation requires the typed ownership guard. diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index c11ac218272..ea76358028e 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -124,6 +124,12 @@ describePostgres("saved intents while another run's accounting is pending", () = action: "issue.updated", entityType: "issue", entityId: f.issueId, details: { source: "native_status_decision", fromStatus: "in_progress", toStatus: "done" } }); + /** The comment route's `issue.comment_added` record, which carries the comment's own resume intent. */ + const commentAdded = (f: { companyId: string; issueId: string }, agentId: string, commentId: string, resumeIntent: boolean) => + db.insert(activityLog).values({ companyId: f.companyId, actorType: "agent", actorId: agentId, agentId, + action: "issue.comment_added", entityType: "issue", entityId: f.issueId, + details: { commentId, ...(resumeIntent ? { resumeIntent: true, followUpRequested: true } : {}) } }); + async function settle(f: { companyId: string }) { await heartbeat.resumeQueuedRuns(); await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); @@ -488,6 +494,132 @@ describePostgres("saved intents while another run's accounting is pending", () = expect(deliveries(f.commentId)).toHaveLength(1); }, 180_000); + // Staging run 8c71398f: a parent agent posted a comment with `resume: true` + // ("a changed requirement for your NEXT turn") 90 s after the child's turn + // ended but before its run settled. The comment queued its own run; the + // settling run then closed the task, the queued run was cancelled as a stale + // continuation, and the explicit resume request was never delivered. + it("reopens a finished task for another agent's explicit resume comment whose queued run turned stale", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + const [parent] = await db.select().from(agents).where(and(eq(agents.companyId, f.companyId), eq(agents.name, "Earlier worker"))); + const [parentRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: parent!.id, invocationSource: "automation", + triggerDetail: "system", status: "succeeded", startedAt: new Date(Date.now() - 20_000), finishedAt: new Date(), + contextSnapshot: {} }).returning(); + const [resume] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: parent!.id, + createdByRunId: parentRun!.id, body: "Changed requirement for your next turn: revalidate the verdict." }).returning(); + await commentAdded(f, parent!.id, resume!.id, true); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "queued", requestedByActorType: "agent", requestedByActorId: parent!.id, + payload: { issueId: f.issueId, commentId: resume!.id, mutation: "comment", resumeIntent: true, followUpRequested: true } }).returning(); + const [stale] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "queued", wakeupRequestId: wake!.id, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: resume!.id, wakeCommentId: resume!.id, + wakeCommentIds: [resume!.id], wakeReason: "issue_commented", source: "issue.comment", + resumeIntent: true, followUpRequested: true } }).returning(); + await db.update(agentWakeupRequests).set({ runId: stale!.id }).where(eq(agentWakeupRequests.id, wake!.id)); + // The child's run, which started before the comment, closed the task after it. + await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() + 1_000) }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await closeByAgent(f); + await settle(f); await settle(f); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, stale!.id)))[0]).toMatchObject({ + status: "cancelled", errorCode: "continuation_task_ownership_changed" }); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) + .where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); + await settle(f); await settle(f); + expect(deliveries(resume!.id)).toHaveLength(1); + await settle(f); + expect(deliveries(resume!.id)).toHaveLength(1); + }, 180_000); + + // Queued runs merge comments of several requesters, and the run's wake is + // rewritten as comments join. The resume intent is each comment's own (its + // `issue.comment_added` record): another agent's plain note on the same + // queued run is not promoted, and the resuming agent's comment is not lost, + // whichever comment came first. + it.each(["resume first", "resume second"] as const)("keeps each merged comment's own resume intent when two agents share a stale queued run (%s)", async order => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + const [resumer] = await db.select().from(agents).where(and(eq(agents.companyId, f.companyId), eq(agents.name, "Earlier worker"))); + const [notifier] = await db.insert(agents).values({ companyId: f.companyId, name: "Notifier", adapterType, status: "idle", + adapterConfig: {} }).returning(); + const post = async (authorAgentId: string, body: string, resumeIntent: boolean) => { + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId, body }).returning(); + await commentAdded(f, authorAgentId, comment!.id, resumeIntent); + return comment!; + }; + const first = order === "resume first" ? await post(resumer!.id, "Changed requirement for your next turn.", true) + : await post(notifier!.id, "FYI: unrelated status note.", false); + const second = order === "resume first" ? await post(notifier!.id, "FYI: unrelated status note.", false) + : await post(resumer!.id, "Changed requirement for your next turn.", true); + const resume = order === "resume first" ? first : second, note = order === "resume first" ? second : first; + // The first comment's wake owns the queued run and now names the joining + // comment, as claiming a merged run rewrites it; the other wake coalesced. + const firstAuthor = order === "resume first" ? resumer! : notifier!, secondAuthor = order === "resume first" ? notifier! : resumer!; + const [primary] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "queued", requestedByActorType: "agent", requestedByActorId: firstAuthor.id, + payload: { issueId: f.issueId, commentId: second.id, mutation: "comment", + ...(order === "resume first" ? { resumeIntent: true, followUpRequested: true } : {}) } }).returning(); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "coalesced", requestedByActorType: "agent", requestedByActorId: secondAuthor.id, + payload: { issueId: f.issueId, commentId: second.id, mutation: "comment", + ...(order === "resume second" ? { resumeIntent: true, followUpRequested: true } : {}) } }); + const [stale] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "queued", wakeupRequestId: primary!.id, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: second.id, wakeCommentId: second.id, + wakeCommentIds: [first.id, second.id], wakeReason: "issue_commented", source: "issue.comment", + resumeIntent: true, followUpRequested: true } }).returning(); + await db.update(agentWakeupRequests).set({ runId: stale!.id }).where(eq(agentWakeupRequests.id, primary!.id)); + await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() + 1_000) }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await closeByAgent(f); + await settle(f); await settle(f); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) + .where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); + await settle(f); await settle(f); + expect(deliveries(resume.id)).toHaveLength(1); + expect(deliveries(note.id).filter(turn => !(turn.context.wakeCommentIds as string[]).includes(resume.id))).toHaveLength(0); + const [saved] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.requestedByActorId, resumer!.id), sql`${agentWakeupRequests.payload}->>'commentId' = ${resume.id}`, + sql`${agentWakeupRequests.payload}->'_paperclipWakeContext' is not null`)); + expect(saved?.payload).toMatchObject({ resumeIntent: true }); + expect((saved?.payload as { _paperclipWakeContext?: { wakeCommentIds?: string[] } })._paperclipWakeContext?.wakeCommentIds) + .toEqual([resume.id]); + }, 180_000); + + // Without `resume: true` an agent's comment stays communicative: the closed + // task is not reopened for it (issues route: only human comments reopen + // finished work implicitly). + it("does not reopen a finished task for another agent's comment without resume intent", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + const [parent] = await db.select().from(agents).where(and(eq(agents.companyId, f.companyId), eq(agents.name, "Earlier worker"))); + const [note] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: parent!.id, + body: "FYI: the parent review finished." }).returning(); + await commentAdded(f, parent!.id, note!.id, false); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "queued", requestedByActorType: "agent", requestedByActorId: parent!.id, + payload: { issueId: f.issueId, commentId: note!.id, mutation: "comment" } }).returning(); + const [stale] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "automation", + triggerDetail: "system", status: "queued", wakeupRequestId: wake!.id, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: note!.id, wakeCommentId: note!.id, + wakeCommentIds: [note!.id], wakeReason: "issue_commented", source: "issue.comment" } }).returning(); + await db.update(agentWakeupRequests).set({ runId: stale!.id }).where(eq(agentWakeupRequests.id, wake!.id)); + await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() + 1_000) }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await closeByAgent(f); + await settle(f); await settle(f); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) + .where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); + await settle(f); + expect(deliveries(note!.id)).toHaveLength(0); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]?.status).toBe("done"); + }, 180_000); + // The durable half: the stale check ran before the closing run settled (or // its save failed), so only the cancelled stale wake carries the message. // Once the closing run settles, the reconciler saves it and the drain diff --git a/server/src/services/heartbeat/stale-comment-continuation.ts b/server/src/services/heartbeat/stale-comment-continuation.ts index bedd51564e8..ec905baffdd 100644 --- a/server/src/services/heartbeat/stale-comment-continuation.ts +++ b/server/src/services/heartbeat/stale-comment-continuation.ts @@ -1,5 +1,5 @@ -import { agentWakeupRequests, heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; -import { and, eq, inArray, sql } from "drizzle-orm"; +import { activityLog, agentWakeupRequests, heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; +import { and, eq, inArray, isNotNull, ne, sql } from "drizzle-orm"; import { saveUndeliveredUserComments, taskClosedByAgentSql } from "../work-intent-outbox.js"; const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; @@ -17,7 +17,9 @@ const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; * and the release drain reopens the done task for the user's message and * delivers it once. Only a task that an agent run completed without reading * the message qualifies: that run started before the comment and succeeded - * after it. A task a person marked done ends the message's intent. + * after it. A task a person marked done ends the message's intent. Another + * agent's comment qualifies only when its wake carried an explicit resume + * intent (`resume: true`). * Returns the saved receipt id, or null when nothing is owed. */ export async function deferUndeliveredCommentsOfStaleRun(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise { @@ -40,7 +42,21 @@ export async function deferUndeliveredCommentsOfStaleRun(db: Db, run: typeof hea if (closed?.byAgent !== true) return null; const [wake] = run.wakeupRequestId ? await db.select().from(agentWakeupRequests) .where(and(eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.id, run.wakeupRequestId))) : []; + // Another agent's comment posted with `resume: true` reopens finished work + // like a person's message (staging run 8c71398f dropped one). Without that + // explicit intent an agent's comment stays communicative. A queued run can + // merge comments of several requesters and its wake is rewritten as they + // join, so the proof is each comment's own immutable record: the + // `issue.comment_added` activity its author wrote with the resume intent. + const explicitResume = await db.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, owed), + isNotNull(issueComments.authorAgentId), ne(issueComments.authorAgentId, run.agentId), + sql`exists (select 1 from ${activityLog} added where added.company_id = ${run.companyId} + and added.entity_type = 'issue' and added.entity_id = ${issueId} and added.action = 'issue.comment_added' + and added.actor_type = 'agent' and added.actor_id = ${issueComments.authorAgentId}::text + and added.details->>'commentId' = ${issueComments.id}::text + and added.details->'resumeIntent' = 'true'::jsonb)`)); return saveUndeliveredUserComments(db, { companyId: run.companyId, issueId, agentId: run.agentId, commentIds: owed, allowedStatuses: ["done"], requestedByActorId: wake?.requestedByActorType === "user" ? wake.requestedByActorId : null, - source: wake?.source, triggerDetail: wake?.triggerDetail }); + source: wake?.source, triggerDetail: wake?.triggerDetail, explicitResumeCommentIds: explicitResume.map(row => row.id) }); } diff --git a/server/src/services/work-intent-outbox.ts b/server/src/services/work-intent-outbox.ts index fa94d0cca6d..54d0eef5c39 100644 --- a/server/src/services/work-intent-outbox.ts +++ b/server/src/services/work-intent-outbox.ts @@ -1,4 +1,4 @@ -import { and, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm"; +import { and, eq, inArray, isNotNull, isNull, ne, or, sql } from "drizzle-orm"; import { agentWakeupRequests, issueComments, issues, type Db } from "@paperclipai/db"; import { notifyDeferredExecutionEligibility, notifyDeferredExecutionWork, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js"; import { withQueuedCommentIdsInWakePayload } from "./issue-queued-comment-queue.js"; @@ -105,6 +105,13 @@ export function taskClosedByAgentSql(companyId: unknown, issueId: unknown) { * caller allows (for example `done`, which the release drain reopens for a * user's message) gate it. Normal admission and the release drain deliver * the receipt once. Returns the receipt id, or null when nothing is owed. + * + * `explicitResumeCommentIds` also admits those comments of another agent: each + * was posted with `resume: true`, which reopens finished work as a person's + * comment does. When only such comments remain, the receipt names the latest + * one's author as requester and keeps the explicit resume intent, so the + * release drain's existing agent-resume rule reopens the task. The assignee's + * own comments never qualify. */ export async function saveUndeliveredUserComments(db: Db, input: { companyId: string; issueId: string; agentId: string; commentIds: string[]; @@ -113,6 +120,7 @@ export async function saveUndeliveredUserComments(db: Db, input: { /** Names why the receipt waits and what promotes it. */ executionWait?: Record; reopenAgentClosedDone?: boolean; + explicitResumeCommentIds?: readonly string[]; }): Promise { if (input.commentIds.length === 0) return null; let publication: ActivityPublication | undefined; @@ -128,27 +136,36 @@ export async function saveUndeliveredUserComments(db: Db, input: { const [closed] = await tx.execute<{ byAgent: boolean }>(sql`select ${taskClosedByAgentSql(input.companyId, input.issueId)} as "byAgent"`); if (closed?.byAgent !== true) return null; } - const live = await tx.select({ id: issueComments.id, authorUserId: issueComments.authorUserId }).from(issueComments).where(and( + const resumeIds = input.explicitResumeCommentIds?.filter(id => input.commentIds.includes(id)) ?? []; + // A direct human comment: no agent author and no creating run. + const human = and(isNotNull(issueComments.authorUserId), isNull(issueComments.authorAgentId), isNull(issueComments.createdByRunId)); + const live = await tx.select({ id: issueComments.id, authorUserId: issueComments.authorUserId, + authorAgentId: issueComments.authorAgentId }).from(issueComments).where(and( eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId), - inArray(issueComments.id, input.commentIds), isNotNull(issueComments.authorUserId), isNull(issueComments.deletedAt), - // A direct human comment: no agent author and no creating run. - isNull(issueComments.authorAgentId), isNull(issueComments.createdByRunId))); + inArray(issueComments.id, input.commentIds), isNull(issueComments.deletedAt), + resumeIds.length ? or(human, and(inArray(issueComments.id, resumeIds), isNotNull(issueComments.authorAgentId), + ne(issueComments.authorAgentId, input.agentId))) : human)); const delivered = await deliveredQueuedCommentIds(tx as unknown as Db, { companyId: input.companyId, issueId: input.issueId, agentId: input.agentId, commentIds: live.map(row => row.id) }); const remaining = input.commentIds.filter(id => live.some(row => row.id === id) && !delivered.has(id)); if (remaining.length === 0) return null; const latest = remaining.at(-1)!; + const resumingAgentId = live.find(row => row.id === latest)?.authorAgentId ?? null; + const agentResume = resumingAgentId !== null && remaining.every(id => live.find(row => row.id === id)?.authorAgentId); + const resumeIntent = agentResume ? { resumeIntent: true, followUpRequested: true } : {}; await notifyDeferredExecutionWork(tx, { companyId: input.companyId }); await notifyDeferredWakesForIssue(tx, { companyId: input.companyId, issueId: input.issueId }); const [saved] = await tx.insert(agentWakeupRequests).values({ companyId: input.companyId, agentId: input.agentId, source: input.source ?? "automation", triggerDetail: input.triggerDetail ?? "system", reason: "issue_commented", status: "deferred_issue_execution", - requestedByActorType: "user", - requestedByActorId: input.requestedByActorId ?? live.find(row => row.id === latest)?.authorUserId ?? null, + requestedByActorType: agentResume ? "agent" : "user", + requestedByActorId: agentResume ? resumingAgentId : input.requestedByActorId + ?? [...remaining].reverse().map(id => live.find(row => row.id === id)?.authorUserId).find(Boolean) ?? null, payload: withQueuedCommentIdsInWakePayload({ issueId: input.issueId, mutation: "comment", commentId: latest, + ...resumeIntent, ...(input.executionWait ? { executionWait: input.executionWait } : {}), [DEFERRED_WAKE_CONTEXT_KEY]: { issueId: input.issueId, taskId: input.issueId, wakeReason: "issue_commented", - source: "issue.comment", wakeCommentIds: remaining, wakeCommentId: latest } }, remaining), + source: "issue.comment", wakeCommentIds: remaining, wakeCommentId: latest, ...resumeIntent } }, remaining), }).returning({ id: agentWakeupRequests.id }); if (saved && issue.status === "done" && input.reopenAgentClosedDone) { const { issueService } = await import("./issues.js");