From 272afe9a5666280eb78ee657fdaf97632203d30a Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 16:43:52 -0500 Subject: [PATCH] refactor(server): model GitHub terminal response recovery obligations Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 19 +++ ...25-github-terminal-response-sleep-audit.md | 141 +++++++++++++++ .../chat-channels.integration.test.ts | 160 ++++++++++++++++++ .../github/response-execution.ts | 73 ++++++++ .../github/terminal-response-work.ts | 97 +++++++++++ .../services/chat-github-publication-lease.ts | 1 + .../services/chat-github-response-comments.ts | 74 ++++---- server/src/services/chat-github-reviews.ts | 3 + .../services/chat-response-availability.ts | 1 + .../chat-run-milestone-notifications.ts | 1 + .../services/delivery-work-notifications.ts | 1 + server/src/services/issues.ts | 5 + 12 files changed, 532 insertions(+), 44 deletions(-) create mode 100644 doc/plans/2026-10-25-github-terminal-response-sleep-audit.md create mode 100644 server/src/services/chat-channels/github/response-execution.ts create mode 100644 server/src/services/chat-channels/github/terminal-response-work.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index b029c22865a..0c5528912f1 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1918,3 +1918,22 @@ Workspace typecheck/build pass again after the receipt-lifecycle correction. The fast-response suites pass 69 tests. The initial complete chat run passed 1,554 tests; a fresh complete run with the three added lifecycle regressions is in progress. No remote gate or full-stack readiness claim is made yet. + +## Slice 34 accepted head and slice 35 checkpoint + +[#17656](https://github.com/paperclipai/paperclip/pull/17656) at +`43900c1f94500af4c4de1987f9608ac4e12feae6` passes all 53 remote checks, +fresh Greptile 5/5, and has zero review threads. Final chat/publication/startup +acceptance passes all 1,557 tests with no skips in 514.70 seconds. Workspace +types/build and 69 fast-response cases pass. The PR remains draft and unmerged. + +Branch `codex/sleep-35-github-terminal-responses` starts from that head. +It extracts provider-owned terminal-state decisions, records successful paused +presentation versions in existing receipts, and adds a scoped worker plus +committed issue/run/response-slot notifications. Eight new real-database +lifecycle cases and nine existing owner cases pass; server types/build pass. +The broader GitHub workflow suite is running. Startup registration and polling +removal wait for the three identified direct issue writers and remaining audit +and acceptance requirements in the terminal-response audit. No slice-35 PR or +full-stack readiness is claimed yet. Earlier parent CI/review findings, all other +inventory duties, and reconciliation with current master remain required. diff --git a/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md b/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md new file mode 100644 index 00000000000..38b884a2194 --- /dev/null +++ b/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md @@ -0,0 +1,141 @@ +# GitHub terminal-response recovery + +Slice 35 starts from acknowledgement PR #17656 at +`43900c1f94500af4c4de1987f9608ac4e12feae6`. +The final parent acceptance passed all 1,557 chat/publication/startup tests without +skips in 514.70 seconds. Parent workspace types/build and 69 fast-response tests +also passed. Greptile reviewed that head at 5/5 without findings. Remote CI was +still completing when this slice started; no merge or full-stack completion is +claimed. + +## Obligation and ownership + +A received acknowledgement and an owed terminal response are different states +of the existing `github_response_comment` action. Slice 34 owns received actions +and retryable failed deliveries. This slice owns processed, nonfinal actions +whose bound request has a terminal outcome or needs a paused notice. Review-check +publication and stale-head projection repair remain independent duties. + +The provider owner must determine exact source/run provenance, latest-run +ownership, and whether a real reply is confirmed or uncertain. Keep those queries +and presentation decisions in `chat-channels/github/response-execution.ts`; shared +notification and scheduler code must not interpret GitHub payloads. Existing +principal/configuration checks, provider markers and repository publication leases +remain the authority for sends and receipt writes. + +## Passive history is not unfinished delivery + +A paused response deliberately remains nonfinal so a later resumed run can answer +it. The old scan repeatedly revisits it, even when it makes no provider mutation. +The replacement needs a saved paused-presentation version in the existing receipt, +written only with confirmed receipt persistence. A changed run or pause state +invalidates that version. No table or separate queue is needed. + +An unrelated newer terminal run does not automatically authorize closing every +older request. Resolve native identity and legacy wake-comment provenance using +the existing owner. Passive non-owning rows must be excluded without acquiring +and releasing a publication lease on every wake. That would otherwise create a +self-wake loop through the release notification. + +## Producer audit still required + +- Run and issue transitions, identity/lineage changes and restored authority use + the existing milestone notification boundary; extend it once with the narrow + response-state topic after verifying coverage. +- Response action completion and provider publication/assessment settlement must + wake affected terminal candidates. Repository lease release is scoped to its + company, endpoint, repository and PR/issue number. +- An unresolved competing lease owns its saved expiration, not a permanent scan. +- Failed terminal sends remain with the existing acknowledgement retry owner, + including the current attempt budget and ambiguous-outcome policy. +- Startup reconstructs eligible work; drain/standby/shutdown retain skipped work + and prevent starting subsequent provider requests. + +## Acceptance required before polling removal + +Use real PostgreSQL and the existing controlled GitHub fixture. Cover terminal +run producer commit/rollback, native identity, blocked-but-non-owning requests, +paused-history quiet and resume, final/superseded history, lease release and +restart, uncertain reply settlement, admission changes between provider requests, +changes during a sweep, and candidate prefixes across the bounded page size. +Measure actual SQL and HTTP effects while idle, not only timer registration. + +Current state: extracting the provider-owned execution decision and persisting +confirmed paused presentation versions. The terminal worker, complete producer +audit, lifecycle tests, startup wiring and recurring-call removal remain pending. + +## Implemented worker checkpoint + +The provider execution resolver is extracted. Successful paused receipt writes +save `pausedVersion` in the existing result. The scoped worker reconstructs +processed nonfinal candidates in 100-row ID pages, excludes unrelated run +history and unresolved reply owners, and uses real publication-lease expiration +when another writer owns the lane. A settled paused receipt is passive. New +run state invalidates it. The owning acknowledgement service still checks +current authority and stops subsequent provider requests at admission/shutdown. +Failed sends transfer to the existing acknowledgement retry worker. + +The response-availability helper now wakes this worker as well as initial +acknowledgements. The milestone boundary wakes it for existing run, identity, +lineage and authority producers. Issue-service status/assignee changes notify +inside their owning transaction. GitHub publication staging, missing-binding +cancellation and scoped lease release also notify. Registration is exposed by +the provider service for tests, but app startup still does not register it and +the old terminal maintenance call remains until the audit below is complete. + +New acceptance covers eight cases: startup terminal recovery and settled-history +quiet; outer issue transaction rollback/commit; paused-history quiet and a later +run; restart; drain; standby; native-identity rollback/commit; real repository +lease contention/release. All eight pass in `/tmp/sleep35-lifecycle-coalesced.log` +(21.88 seconds). The SQL-quiet checks advance 120 seconds and inspect real +`select` and `transaction` calls. GitHub HTTP remains controlled. + +Server typecheck and build pass at this checkpoint, with logs +`/tmp/sleep35-final-server-types.log` and `/tmp/sleep35-final-server-build.log`. +All GitHub review-workflow acceptance is running in +`/tmp/sleep35-github-acceptance.log`; no broad result is claimed yet. + +Material failures retained: + +- `/tmp/sleep35-before-producer.log`: all three first cases failed because the + test coordinator omitted its required `canRun` function. Corrected the fixture. +- `/tmp/sleep35-before-producer-fixed-fixture.log`: startup and paused recovery + passed, while the real issue commit failed to wake terminal response work. + The issue owner now signals the response slot before writing its status. +- `/tmp/sleep35-producer-fixed.log`: the producer regression and 15 adjacent + cases passed; the paused restart assertion observed the legitimate pending + coalesced lease-release pass rather than waiting for settlement. +- `/tmp/sleep35-lifecycle.log`: four passed and four assertions failed at the + default one-second poll timeout or before the coalesced pass. The coordinator + itself coalesces at one second. Explicit five-second observation bounds now + allow that existing behavior; `/tmp/sleep35-lifecycle-coalesced.log` passes all + eight. No production delay or retry policy was widened to make tests pass. +- `/tmp/sleep35-owner-extraction.log`: nine existing ownership/paused/native/ + legacy-placeholder cases passed after extraction. + +## Direct issue writers found before startup wiring + +These are unresolved producer-audit items, not exemptions: + +1. `issue-tree-control.ts::restoreIssueStatusesForHold` restores issue status + directly. Its cancellation path already calls `issueService.update`, but the + restoration path does not. Test a restored blocked request with existing + response history and cover its transaction before removing the old sweep. +2. `heartbeat/queue.ts` directly blocks a task whose isolated worktree lacks a + usable project. This can happen before any run is admitted, so a run-terminal + notification cannot cover it. Existing assembled coverage is in + `heartbeat-workspace-branch-containment.test.ts` around the projectless + isolated-worktree rejection case. Verify this producer's response notification. +3. `execution-recovery-resolution.ts::settleUnrecoverableExecutions` can set + status blocked after a run was already terminal. It does not currently send + the response notification. Existing real-database coverage is in + `native-safe-replacement.test.ts` and run-dispatch adapter tests. Preserve its + task/coordinator/run lock order and failure rollback when adding the signal. + +The native-finalization reconciler's blocked transition calls issueService and +is covered by the new owning boundary. Recovery/service's inspected blocked +transitions also call issueService. Runner goal statuses are session-goal data, +not issue status, and do not independently change this response predicate. +Additional dynamic status writers, lookup/index cost, uncertain-publication +settlement, page fairness, final startup wiring, full acceptance and remote review +still need completion. Do not present this checkpoint as polling removal. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 8e86aa0b726..8cee0a7b041 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -3554,6 +3554,166 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(attempts).toBe(2); expect(f.responseWrites).toHaveLength(writes); }); + it("restores terminal GitHub responses and leaves finalized history quiet", async () => { + const f = await toolReplyFixture(); + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + const providerWrites = f.responseWrites.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + expect(f.responseWrites).toHaveLength(providerWrites); + reads.mockRestore(); writes.mockRestore(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("wakes terminal GitHub responses after the outer issue transaction commits", async () => { + const f = await toolReplyFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true); + try { + await worker.ready; + const baseline = f.responseWrites.length; + const change = async (tx: Parameters[0]>[0]) => { + await tx.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(tx as unknown as Db).update(f.conversation.issueId!, { status: "blocked" }); + }; + await expect(db.transaction(async tx => { await change(tx); throw new Error("rollback terminal issue"); })) + .rejects.toThrow("rollback terminal issue"); + expect(f.responseWrites).toHaveLength(baseline); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const pending = db.transaction(async tx => { await change(tx); entered(); await hold; }); + try { await ready; expect(f.responseWrites).toHaveLength(baseline); } + finally { release(); await pending; } + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("keeps an acknowledged paused GitHub request quiet until its new run finishes", async () => { + const f = await toolReplyFixture("github:paperclipai/paperclip:418"); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id)); + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.conversation.issueId!)); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: f.conversation.externalThreadId }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41802", text: "New request", userId: "42", userName: "octocat", mentioned: true }) }); + const responses = githubResponseCommentService(db, f.providerFetch); + await responses.processPending(); + expect([...f.responseComments.values()][0].body).toContain("paused"); + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await responses.registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + const providerWrites = f.responseWrites.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + expect(f.responseWrites).toHaveLength(providerWrites); + reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); + await coordinator.stop(); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.conversation.issueId!)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, + status: "failed", runtimeMode: "native", contextSnapshot: await chatWakeContext({ + endpointId: f.endpoint.id, issueId: f.conversation.issueId, provider: "github", providerMessageId: "41802", + }) }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await responses.registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["restart", "drain", "standby"] as const)("retains terminal GitHub response work across %s", async mode => { + const f = await toolReplyFixture(); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + const responses = githubResponseCommentService(db, f.providerFetch); + let worker = responses.registerTerminalResponseWork(coordinator, () => canRun); + try { + await worker.ready; + const baseline = f.responseWrites.length; + if (mode === "restart") await coordinator.stop(); else canRun = false; + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(db).update(f.conversation.issueId!, { status: "blocked" }); + expect(f.responseWrites).toHaveLength(baseline); + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = responses.registerTerminalResponseWork(coordinator, () => true); + await worker.ready; + } else { canRun = true; worker.wake(); } + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("reactivates terminal GitHub response recovery from committed native identity", async () => { + const f = await toolReplyFixture(); + const messageId = String(f.run.contextSnapshot?.wakeCommentId); + await db.update(heartbeatRuns).set({ status: "failed", nativeIssueId: f.conversation.issueId, + contextSnapshot: {}, updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + const baseline = f.responseWrites.length; + await expect(db.transaction(async tx => { + await initializeRunIdentity(tx as unknown as Db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", messageIds: [messageId], cause: "instruction" }); + throw new Error("rollback identity"); + })).rejects.toThrow("rollback identity"); + expect(f.responseWrites).toHaveLength(baseline); + await initializeRunIdentity(db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", messageIds: [messageId], cause: "instruction" }); + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("waits for the existing GitHub publication lease before a terminal response", async () => { + const f = await toolReplyFixture(); + const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + let lease: Promise | undefined; + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id, + repositoryId: String(resource.metadata.providerRepositoryId), number: 418 }, f.providerFetch, async () => { entered(); await hold; }); + await ready; + const baseline = f.responseWrites.length; + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(db).update(f.conversation.issueId!, { status: "blocked" }); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).not.toBeNull(); + expect(f.responseWrites).toHaveLength(baseline); + release(); await lease; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { release(); await lease; await coordinator.stop(); } + }); + it.each(["release", "restart", "drain", "standby"] as const)("wakes GitHub acknowledgements from a real publication lease %s", async mode => { const f = await reviewBotFixture(); const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id)); diff --git a/server/src/services/chat-channels/github/response-execution.ts b/server/src/services/chat-channels/github/response-execution.ts new file mode 100644 index 00000000000..64e0ced4140 --- /dev/null +++ b/server/src/services/chat-channels/github/response-execution.ts @@ -0,0 +1,73 @@ +import { and, desc, eq, or, sql } from "drizzle-orm"; +import { chatMessageLinks, heartbeatRuns, type Db } from "@paperclipai/db"; +import { githubRunReplyState } from "../../chat-run-publications.js"; +import { chatRunSourceMessageId } from "../run-origin.js"; + +const runColumns = { + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + responsibleUserId: heartbeatRuns.responsibleUserId, + activeIdentityContextId: heartbeatRuns.activeIdentityContextId, + nativeIssueId: heartbeatRuns.nativeIssueId, + issueId: heartbeatRuns.issueId, + status: heartbeatRuns.status, + updatedAt: heartbeatRuns.updatedAt, + contextSnapshot: sql>`jsonb_build_object( + 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', + 'taskId', ${heartbeatRuns.contextSnapshot}->'taskId', + 'wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId', + 'wakeCommentIds', ${heartbeatRuns.contextSnapshot}->'wakeCommentIds' + )`, +}; + +export type GitHubResponseExecutionInput = { + companyId: string; + endpointId: string; + agentId: string; + issueId: string; + deliveryId: string; + sourceMessageId: string | null; + issueStatus: string; + deliveryCreatedAt: Date; +}; + +export async function githubResponseExecution(db: Db, input: GitHubResponseExecutionInput) { + const [requestRun] = await db.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).innerJoin(chatMessageLinks, and( + eq(chatMessageLinks.companyId, input.companyId), eq(chatMessageLinks.endpointId, input.endpointId), + eq(chatMessageLinks.deliveryId, input.deliveryId), eq(chatMessageLinks.direction, "inbound"), + or(sql`${chatMessageLinks.commentId}::text = ${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`, + sql`coalesce(${heartbeatRuns.contextSnapshot}->'wakeCommentIds', '[]'::jsonb) ? ${chatMessageLinks.commentId}::text`), + )).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${input.issueId}`)) + .orderBy(desc(heartbeatRuns.createdAt)).limit(1); + const [latestRun] = await db.select(runColumns).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${input.issueId}`, + )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + const originMessageId = latestRun?.activeIdentityContextId ? await chatRunSourceMessageId(db, latestRun, + { allowLegacyWakeComment: true, provider: "github" }) : null; + const ownsLatestRequest = !!latestRun && (requestRun?.runId === latestRun.id || + (!!input.sourceMessageId && originMessageId === input.sourceMessageId)); + const requestRunning = ownsLatestRequest && latestRun?.status === "running"; + const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status); + const run = terminal && (ownsLatestRequest || input.issueStatus === "blocked") ? latestRun : null; + + const reply = run && ownsLatestRequest ? await githubRunReplyState(db, { + companyId: input.companyId, endpointId: input.endpointId, issueId: input.issueId, runId: run.id, + }) : "none"; + const paused = !!run ? input.issueStatus === "blocked" && !ownsLatestRequest + : input.issueStatus === "blocked" && latestRun?.status !== "running"; + const body = run ? reply === "confirmed" ? "This request has been handled. See the response in this conversation." + : paused ? "Work is paused. Your request is saved. See the Paperclip task for details." + : run.status === "succeeded" ? "This turn ended without publishing a final response. See the Paperclip task for details." + : "This turn stopped before completing. See the Paperclip task for details." + : paused ? "Work is paused. Your request is saved. See the Paperclip task for details." : undefined; + return { + body, paused, reply, running: requestRunning, + versionAt: run ? new Date(Math.max(run.updatedAt.getTime(), input.deliveryCreatedAt.getTime())) + : paused ? input.deliveryCreatedAt : undefined, + version: JSON.stringify([latestRun?.id ?? null, latestRun?.status ?? null, + latestRun?.updatedAt.toISOString() ?? null, ownsLatestRequest, input.issueStatus === "blocked", reply]), + }; +} diff --git a/server/src/services/chat-channels/github/terminal-response-work.ts b/server/src/services/chat-channels/github/terminal-response-work.ts new file mode 100644 index 00000000000..68068daa701 --- /dev/null +++ b/server/src/services/chat-channels/github/terminal-response-work.ts @@ -0,0 +1,97 @@ +import { and, asc, eq, gt, isNotNull, sql } from "drizzle-orm"; +import { chatActions, chatConversations, chatDeliveries, chatEndpointLeases, chatEndpointResources, + chatEndpoints, chatMessageLinks, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import type { ChatResponseScope } from "../../chat-response-availability.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; +import { githubResponseExecution } from "./response-execution.js"; + +type Scope = ChatResponseScope & { endpointId?: string; repositoryId?: string; number?: number }; +type Entry = { id: string; companyId: string; endpointId: string; deliveryId: string; + issueId: string; conversationId: string; repositoryId: string | null; number: number | null; dueAt: number }; + +export function registerGitHubTerminalResponseWork( + db: Db, + coordinator: ReturnType, + acknowledge: (deliveryId: string, canContinue: () => boolean) => Promise, + canRun: () => boolean, +) { + const repositoryId = sql`${chatEndpointResources.metadata}->>'providerRepositoryId'`; + const number = sql`substring(${chatConversations.externalThreadId} + from '^github:[^:]+:(?:issue:)?([1-9][0-9]*)(?::rc:[1-9][0-9]*)?$')`; + async function page(scope?: Scope, cursor?: string, actionId?: string) { + return db.select({ + id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId, + deliveryId: chatActions.deliveryId, conversationId: chatConversations.id, issueId: issues.id, + agentId: chatEndpoints.assignedAgentId, issueStatus: issues.status, + deliveryCreatedAt: chatDeliveries.createdAt, repositoryId, number, result: chatActions.result, + sourceMessageId: sql`(select l.comment_id from ${chatMessageLinks} l + where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId} + and l.delivery_id = ${chatActions.deliveryId} and l.direction = 'inbound' limit 1)`, + leaseExpiresAt: sql`(select max(l.expires_at) from ${chatEndpointLeases} l + where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId} + and l.lease_key = 'github-publication:' || (${repositoryId}) || ':' || ${number})`, + }).from(chatActions) + .innerJoin(chatConversations, and(eq(chatConversations.id, chatActions.conversationId), eq(chatConversations.companyId, chatActions.companyId))) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId))) + .innerJoin(chatDeliveries, and(eq(chatDeliveries.id, chatActions.deliveryId), eq(chatDeliveries.companyId, chatActions.companyId))) + .innerJoin(issues, and(eq(issues.id, chatConversations.issueId), eq(issues.companyId, chatActions.companyId))) + .leftJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), eq(chatEndpointResources.companyId, chatActions.companyId))) + .where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.status, "processed"), + isNotNull(chatActions.deliveryId), isNotNull(chatEndpoints.assignedAgentId), + sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`, + sql`(${issues.status} = 'blocked' or (select r.status from ${heartbeatRuns} r + where r.company_id = ${chatActions.companyId} and r.agent_id = ${chatEndpoints.assignedAgentId} + and coalesce(r.native_issue_id::text, r.context_snapshot->>'issueId') = ${chatConversations.issueId}::text + order by r.created_at desc, r.id desc limit 1) in ('succeeded', 'failed', 'interrupted', 'timed_out', 'cancelled'))`, + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(issues.id, scope.issueId) : undefined, + scope?.conversationId ? eq(chatActions.conversationId, scope.conversationId) : undefined, + scope?.deliveryId ? eq(chatActions.deliveryId, scope.deliveryId) : undefined, + scope?.endpointId ? eq(chatActions.endpointId, scope.endpointId) : undefined, + scope?.repositoryId ? eq(repositoryId, scope.repositoryId) : undefined, + scope?.number ? eq(number, String(scope.number)) : undefined, + cursor ? gt(chatActions.id, cursor) : undefined, actionId ? eq(chatActions.id, actionId) : undefined, + )).orderBy(asc(chatActions.id)).limit(100); + } + async function entry(row: Awaited>[number]): Promise { + const execution = await githubResponseExecution(db, { ...row, agentId: row.agentId!, deliveryId: row.deliveryId! }); + if (execution.body === undefined || execution.reply === "unsettled" || + (execution.paused && row.result?.pausedVersion === execution.version)) return null; + return { id: row.id, companyId: row.companyId, endpointId: row.endpointId, deliveryId: row.deliveryId!, + issueId: row.issueId, conversationId: row.conversationId, repositoryId: row.repositoryId, + number: row.number ? Number(row.number) : null, dueAt: row.leaseExpiresAt ? Date.parse(row.leaseExpiresAt) : 0 }; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubTerminalResponses, + decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (value: Entry, scope?: Scope) => !scope || (value.companyId === scope.companyId && + (!scope.issueId || value.issueId === scope.issueId) && (!scope.deliveryId || value.deliveryId === scope.deliveryId) && + (!scope.conversationId || value.conversationId === scope.conversationId) && + (!scope.endpointId || value.endpointId === scope.endpointId) && (!scope.repositoryId || value.repositoryId === scope.repositoryId) && + (!scope.number || value.number === scope.number)), + load: async (scope?: Scope, signal?: AbortSignal) => { + const entries: Entry[] = []; + let cursor: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await page(scope, cursor); + for (const row of rows) { + if (!canRun() || signal?.aborted) throw signal?.reason ?? new Error("GitHub terminal-response recovery stopped"); + const value = await entry(row); if (value) entries.push(value); + } + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw signal?.reason ?? new Error("GitHub terminal-response recovery stopped"); + }, + check: async (value: Entry, signal?: AbortSignal) => { + const [row] = await page(undefined, undefined, value.id); + const current = row ? await entry(row) : null; + if (!current || current.dueAt > Date.now() || !canRun() || signal?.aborted) return current; + await acknowledge(current.deliveryId, () => canRun() && !signal?.aborted); + const [remaining] = await page(undefined, undefined, value.id); + return remaining ? entry(remaining) : null; + }, + }, canRun); +} diff --git a/server/src/services/chat-github-publication-lease.ts b/server/src/services/chat-github-publication-lease.ts index 46d9e3b862b..994bb8e9174 100644 --- a/server/src/services/chat-github-publication-lease.ts +++ b/server/src/services/chat-github-publication-lease.ts @@ -116,6 +116,7 @@ export async function withGitHubPublicationLease( await renewal; await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify(scope)); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify(scope)); await tx.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.id, claimed.id), eq(chatEndpointLeases.token, token))); }); diff --git a/server/src/services/chat-github-response-comments.ts b/server/src/services/chat-github-response-comments.ts index 3fbd7988063..0f5bb35222f 100644 --- a/server/src/services/chat-github-response-comments.ts +++ b/server/src/services/chat-github-response-comments.ts @@ -5,6 +5,8 @@ import { fastResponseRequests, chatPublications, type Db, } from "@paperclipai/db"; +import { registerGitHubTerminalResponseWork } from "./chat-channels/github/terminal-response-work.js"; +import { githubResponseExecution } from "./chat-channels/github/response-execution.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { registerGitHubAcknowledgementWork } from "./chat-channels/github/acknowledgement-work.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -93,7 +95,7 @@ async function stageResponse(db: Db, source: Source) { export async function writeGitHubResponseComment( db: Db, source: Source, api: Api, lease: Lease, - input: { body: string; final: boolean; versionAt?: Date; acknowledgement?: boolean; preserveFinal?: boolean }, + input: { body: string; final: boolean; versionAt?: Date; acknowledgement?: boolean; preserveFinal?: boolean; pausedVersion?: string }, assertCurrent: () => Promise, ) { await assertCurrent(); @@ -116,8 +118,10 @@ export async function writeGitHubResponseComment( (input.versionAt && Date.parse(String(response.result.versionAt)) > input.versionAt.getTime()))) { if (!input.acknowledgement && response.result.final === true && !input.final) throw conflict("This GitHub response is already final"); - if (["received", "failed"].includes(response.status)) await lease.commit(async tx => { - await tx.update(chatActions).set({ status: "processed", updatedAt: new Date() }) + if (["received", "failed"].includes(response.status) || (response.status === "processed" && + input.pausedVersion && response.result.pausedVersion !== input.pausedVersion)) await lease.commit(async tx => { + await tx.update(chatActions).set({ status: "processed", + result: { ...response.result, ...(input.pausedVersion ? { pausedVersion: input.pausedVersion } : {}) }, updatedAt: new Date() }) .where(and(eq(chatActions.id, response.id), eq(chatActions.status, response.status))); }); return { id: String(response.result.id), url: String(response.result.url) }; @@ -161,6 +165,7 @@ export async function writeGitHubResponseComment( }, updatedAt: new Date() }).where(eq(chatActions.id, target.id)); await tx.update(chatActions).set({ status: "processed", result: { ...receipt, final: input.final || alreadyFinal, versionAt: versionAt.toISOString(), + ...(input.pausedVersion ? { pausedVersion: input.pausedVersion } : {}), }, updatedAt: new Date() }).where(eq(chatActions.id, response.id)); await logActivity(tx as unknown as Db, { companyId: source.endpoint.companyId, actorType: "system", actorId: "github-response", action: "chat_github.response_updated", @@ -222,49 +227,28 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { let requestRunning = false; let requestPaused = false; let preserveFinal = false; + let pausedVersion: string | undefined; if (finalBody === undefined) { const existing = await stageResponse(db, source); - const [requestRun] = await db.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).innerJoin(chatMessageLinks, and( - eq(chatMessageLinks.companyId, source.endpoint.companyId), eq(chatMessageLinks.endpointId, source.endpoint.id), - eq(chatMessageLinks.deliveryId, deliveryId), eq(chatMessageLinks.direction, "inbound"), - or(sql`${chatMessageLinks.commentId}::text = ${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`, - sql`coalesce(${heartbeatRuns.contextSnapshot}->'wakeCommentIds', '[]'::jsonb) ? ${chatMessageLinks.commentId}::text`), - )).where(and(eq(heartbeatRuns.companyId, source.endpoint.companyId), eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), - sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${source.conversation.issueId}`)) - .orderBy(desc(heartbeatRuns.createdAt)).limit(1); - const [latestRun] = await db.select(responseRunColumns).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, source.endpoint.companyId), eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), - sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${source.conversation.issueId}`, - )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); - const originMessageId = latestRun?.activeIdentityContextId ? await chatRunSourceMessageId(db, latestRun, - { allowLegacyWakeComment: true, provider: "github" }) : null; - const ownsLatestRequest = !!latestRun && (requestRun?.runId === latestRun.id || - (!!source.sourceMessageId && originMessageId === source.sourceMessageId)); - requestRunning = ownsLatestRequest && latestRun?.status === "running"; - const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status); - const run = terminal && (ownsLatestRequest || source.issueStatus === "blocked") ? latestRun : null; - if (run) { - const reply = ownsLatestRequest ? await githubRunReplyState(db, { - companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - issueId: source.conversation.issueId!, runId: run.id }) : "none"; - if (reply === "unsettled") throw conflict("GitHub reply is still being resolved"); - if (existing.result?.final === true) return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined; - preserveFinal = reply === "confirmed"; - onlyExisting = reply === "confirmed" && !existing.result?.id; - requestPaused = source.issueStatus === "blocked" && !ownsLatestRequest; - finalBody = reply === "confirmed" ? "This request has been handled. See the response in this conversation." - : requestPaused ? "Work is paused. Your request is saved. See the Paperclip task for details." - : run.status === "succeeded" ? "This turn ended without publishing a final response. See the Paperclip task for details." - : "This turn stopped before completing. See the Paperclip task for details."; - finalVersionAt = new Date(Math.max(run.updatedAt.getTime(), source.delivery.createdAt.getTime())); - } else if (source.issueStatus === "blocked" && latestRun?.status !== "running") { - requestPaused = true; - finalBody = "Work is paused. Your request is saved. See the Paperclip task for details."; - finalVersionAt = source.delivery.createdAt; - } else if (existing.result?.id && (!isMainPullResponse(source) || + const execution = await githubResponseExecution(db, { + companyId: source.endpoint.companyId, endpointId: source.endpoint.id, + agentId: source.endpoint.assignedAgentId!, issueId: source.conversation.issueId!, deliveryId, + sourceMessageId: source.sourceMessageId, issueStatus: source.issueStatus, + deliveryCreatedAt: source.delivery.createdAt, + }); + if (execution.reply === "unsettled") throw conflict("GitHub reply is still being resolved"); + if (existing.result?.final === true) return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined; + if (execution.paused && existing.status === "processed" && existing.result?.id && existing.result?.pausedVersion === execution.version) + return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined; + pausedVersion = execution.paused ? execution.version : undefined; + requestRunning = execution.running; + requestPaused = execution.paused; + preserveFinal = execution.reply === "confirmed"; + onlyExisting = preserveFinal && !existing.result?.id; + finalBody = execution.body; + finalVersionAt = execution.versionAt; + if (finalBody === undefined && existing.result?.id && (!isMainPullResponse(source) || (await stagePullResponse(db, source)).result?.id === existing.result.id)) { - // Rotate no-op maintenance checks so a page of newer queued asks - // cannot hide other conversations that need terminal reconciliation. await lease.commit(async tx => { await tx.update(chatActions).set({ status: ["received", "failed"].includes(existing.status) ? "processed" : existing.status, updatedAt: new Date(), @@ -305,7 +289,7 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { } return writeGitHubResponseComment(db, source, api, lease, { body: finalBody ?? (isMainPullResponse(source) && !requestRunning ? "Review queued…" : "Working on this…"), final: finalBody !== undefined && !requestPaused, - acknowledgement: finalBody === undefined, preserveFinal, + acknowledgement: finalBody === undefined, preserveFinal, pausedVersion, ...(finalBody !== undefined ? { versionAt: finalVersionAt ?? new Date() } : {}) }, assertCurrent); }); } @@ -464,6 +448,8 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { } } return { tryAcknowledge, processPending, processTerminalResponses, failRun, finishRun, hasWorkingResponseForRun, + registerTerminalResponseWork: (coordinator: ReturnType, canRun: () => boolean) => + registerGitHubTerminalResponseWork(db, coordinator, tryAcknowledge, canRun), registerAcknowledgementWork: (coordinator: ReturnType, canRun: () => boolean) => registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun), }; diff --git a/server/src/services/chat-github-reviews.ts b/server/src/services/chat-github-reviews.ts index 5fb2be6d1a6..06102e8df4b 100644 --- a/server/src/services/chat-github-reviews.ts +++ b/server/src/services/chat-github-reviews.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { githubRunReplyState } from "./chat-run-publications.js"; import { stageReceiptReactionRemovals } from "./chat-receipt-reactions.js"; import { chatRunSourceMessageId } from "./chat-channels/run-origin.js"; @@ -461,6 +462,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { ): Promise { if (writer === db) return db.transaction(tx => stage(source, operation, payload, key, invocationId, toolName, parameters, tx)); const actionKey = `github_publication:${source.issue.id}:${operation}:${hash(key)}`; + await notifyChatResponseAvailability(writer, { companyId: source.endpoint.companyId, issueId: source.issue.id }); await notifyDeliveryWork(writer, DELIVERY_QUEUES.chatGitHubPublications); const values = { companyId: source.endpoint.companyId, @@ -1046,6 +1048,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { )?.[1]; if (!number || !binding?.repository.providerRepositoryId) { await db.transaction(async tx => { + await notifyChatResponseAvailability(tx, { companyId: action.companyId, conversationId: action.conversationId ?? undefined }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); await tx.update(chatActions).set({ status: "cancelled", result: { code: "publication_binding_unavailable", retryable: false }, updatedAt: new Date() }) .where(and(eq(chatActions.id, actionId), eq(chatActions.companyId, action.companyId))); diff --git a/server/src/services/chat-response-availability.ts b/server/src/services/chat-response-availability.ts index 29de7ca303e..d1459d65d9d 100644 --- a/server/src/services/chat-response-availability.ts +++ b/server/src/services/chat-response-availability.ts @@ -3,5 +3,6 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string }; export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); } diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 972d6789bb7..71892270ebb 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -8,6 +8,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatRunMilestoneScope = { companyId: string; issueId?: string }; export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); // Lifecycle and coarse native progress share lineage and authority. Extend // this eligibility boundary once rather than adding wrappers to its callers. diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index ccdf2c0a356..762e377ebbb 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -38,6 +38,7 @@ export const DELIVERY_QUEUES = { chatProviderEffects: "chat-provider-effects", chatGitHubIngress: "chat-github-ingress", chatGitHubPublications: "chat-github-publications", + chatGitHubTerminalResponses: "chat-github-terminal-responses", chatGitHubAcknowledgements: "chat-github-acknowledgements", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 7ec92cff370..c72bbab8fbd 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; @@ -11307,6 +11308,10 @@ export function issueService(db: Db) { if (patch.status === "done" || patch.status === "cancelled") { await notifyDecisionTargetWork(tx, `issue:${id}`); } + if ((patch.status !== undefined && patch.status !== receiptExisting.status) || + (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId)) { + await notifyChatResponseAvailability(tx, { companyId: receiptExisting.companyId, issueId: id }); + } if (patch.status !== undefined && patch.status !== receiptExisting.status) { await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId, issueId: id }); }