mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-15 01:52:55 +01:00
fix(heartbeat): keep an agent's explicit resume comment that a stale continuation dropped (#20856)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agents coordinate on tasks through comments. A comment posted with `resume: true` is an explicit request for more work; on a finished task it reopens the task, for a person or for another agent. > - A comment that arrives while the assignee's run is still settling queues its own run. If the settling run then closes the task, that queued run is cancelled before dispatch as a stale continuation (`continuation_task_ownership_changed`). > - For a person's comment, the outbox saves the unread message first and the release drain reopens the task (chaos s9401/t72). For another agent's explicit `resume: true` comment, nothing was saved. The request was dropped and the task stayed `done`. > - This pull request saves that agent comment too, with its explicit resume intent. The existing release-drain rule for an agent's explicit resume reopens the task and delivers the comment once. > - The benefit is that a parent agent's "do this next turn" request is not lost to a timing race. An agent comment without `resume: true` stays communicative, as before. ## Linked Issues or Issue Description **What happened?** On a hosted staging instance, a parent agent posted a comment with `resume: true` on its child's task: "a changed requirement for your NEXT turn". The child's provider turn had finished 90 s earlier, but its run was still settling (workspace copy-back). The comment queued a new run. 40 s later the settling run applied its `done` decision. The queued run was then cancelled with `continuation_task_ownership_changed`. The task stayed `done` and no run ever read the comment. The cancelled run itself is not shown as a failure (the task is terminal, so task chat hides it, and Sentry skips expected cancellations), so nothing told anyone the request was lost. **Steps to reproduce** 1. Run an agent on a task. While its run is still settling, post a comment from another agent with `resume: true`. 2. Let the run close the task (`done`). 3. The comment's queued run is cancelled as a stale continuation. The task is not reopened and the comment is never delivered. **Expected behavior** The task reopens for the explicit resume request, and the comment is delivered exactly once, as for a person's comment. ## What Changed - `deferUndeliveredCommentsOfStaleRun` finds, per owed comment, the comment's own `issue.comment_added` activity record: written by its author (another agent, not the assignee) with `resumeIntent: true`. Merged runs, whose wake is rewritten as comments join, therefore keep each comment's own intent. - `saveUndeliveredUserComments` also admits those comments (`explicitResumeCommentIds`). When only such comments remain, the deferred receipt records the agent as requester and carries `resumeIntent`/`followUpRequested` in its payload and wake context. The release drain's existing rule for an agent's explicit resume on a `done` task then reopens and delivers it. Person-comment behaviour is unchanged. - `doc/run-log-events.md` describes the rule. ## Verification - RED on `origin/master` (`1d67e309e0`): `pnpm --filter @paperclipai/server exec vitest run src/__tests__/heartbeat-held-comment-wake.test.ts -t "explicit resume comment"` fails: `expected [] to have a length of 1 but got +0` (the comment is never delivered after the stale cancellation). - GREEN: the whole file passes (29/29), including two-requester merged-run tests in both orders, the new negative test (`does not reopen a finished task for another agent's comment without resume intent`) and the existing person-comment and exactly-once tests. - Related suites `ownership-reconciliation-report`, `execution-continuation`, `chat-wake-conversation-batch` plus this file: 118/118 tests pass locally. One `afterAll` cleanup hook in `ownership-reconciliation-report` timed out under local machine load; CI runs the suite fresh. - `pnpm --filter @paperclipai/server exec tsc --noEmit` passes. - Not done: no restart-chaos cell. The harness posts follow-ups as the board user, which s9401/t72 already covers; an agent-authored `resume: true` comment needs a two-agent workload the harness does not have. ## Risks - Low risk. Only a comment whose own wake was an agent's explicit `resume: true` qualifies, only on a task an agent run closed, and only when no run read the comment. The assignee's own comments never qualify. - The ownership reconciler's durable backstop (`checkStaleDroppedComments`) still covers person comments only. If the fast path fails, an agent's resume comment is not recovered later. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, read-only staging API reads). ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
a219dd7499
commit
e0e6462a73
4 files changed
+182
-13
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<string | null> {
|
||||
@@ -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) });
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
reopenAgentClosedDone?: boolean;
|
||||
explicitResumeCommentIds?: readonly string[];
|
||||
}): Promise<string | null> {
|
||||
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");
|
||||
|
||||
Reference in new issue
Block a user