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 2c82c8b14f1..b029c22865a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1892,3 +1892,29 @@ projection repair. Those are separate reasons and remain separate planned PRs. See `2026-10-25-github-chat-maintenance-sleep-audit.md` for the inspected owners, producer boundaries, and response-slot dependencies. No GitHub migration is yet claimed implemented or verified. + +## Slice 34: initial GitHub acknowledgement delivery + +Based on accepted slice 33 (`276ef23d2a0b01b15e6bfb8ff31ab13495741b19`). +The implementation moves only initial response-comment acknowledgement out of +recurring GitHub maintenance. The existing response actions, fast-response +records, and publication leases own durability. Committed response-slot changes +and actual saved deadlines drive recovery. Terminal responses, review checks, +and stale-head projection remain separate upcoming reasons. + +Fourteen new real-PostgreSQL lifecycle cases and the existing startup guard pass. +The producer-removal regression fails as expected when fast-response settlement +has no notification. Migration application/replay pass after adding the generated +concurrent-index migration to the existing top-level execution registry. Broader +local and remote acceptance is pending. See the detailed +[GitHub owner audit](2026-10-25-github-chat-maintenance-sleep-audit.md). + +All earlier failing CI, review findings, stack/master reconciliation, and remaining +inventory duties retain their existing requirements. This slice does not claim +that the stack is ready or authorize a merge. + +Slice-34 final focused acceptance passes 17 new cases plus the startup guard. +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. diff --git a/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md b/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md index 2ccc983898d..aa8b2136282 100644 --- a/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md +++ b/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md @@ -146,3 +146,129 @@ Early failure/skip and adding a publication binding require producer coverage. Do not blindly wrap every write or introduce GitHub branches in fast-responses. Use a normalized response-slot change contract if these shared owners need a new notification capability, with existing transactions reused wherever present. + +## Slice 34 implementation and boundaries + +Initial response acknowledgement now registers with the existing delivery +coordinator. This is a separate worker from terminal-response reconciliation, +review-check delivery, and stale-head repair. The latter three remain in the +GitHub maintenance owner. The explicit/manual combined `processPending` entry +point still performs both initial and terminal response recovery. + +The durable source remains `chat_actions.kind = github_response_comment`. +Received actions and retryable failed actions are reconstructed on startup in +100-row ID pages. The shared scoped-deadline helper dispatches a fair batch of +25 across companies. Exact delivery, conversation, issue, and repository-lease +hints restrict reconstruction. Processed/cancelled actions and exhausted failures +leave this worker's inventory; retained response history does not arm a timer. +No new queue table or changed automatic retry limit is introduced. + +The next deadline is derived from existing saved retry state, a competing +repository publication lease, or a pending fast response's expiration. A +streaming fast response parks acknowledgement work until publication settlement +or quarantine changes its state. The publication worker still owns stale-stream +recovery. This worker does not invent a second timeout or replay an uncertain +publication. Published or delivery-unknown fast responses suppress the ordinary +acknowledgement through the existing provider owner. + +Producer coverage: + +- `stageResponse` signals its insert transaction. Existing response lookups do + not emit another signal. This prevents acknowledgement itself from continuously + waking an unchanged received action. +- Failed/cancelled acknowledgement settlement signals the existing action's + delivery before its transaction writes the result. +- Repository publication lease release signals the precise endpoint/repository/ + pull-or-issue number. The signal covers receipt changes made under that lease, + and releases waiting acknowledgements when another publication finishes. +- Fast-response admission, early skip, accounting settlement, and publication + binding signal the existing response slot. Pending-to-running claims retain + the same wait predicate; pending expiry already has a saved deadline. +- `supersedeFastResponse` is used only by Board chat, which has no external + acknowledgement delivery. It was audited and left unchanged. +- Publication completion, failure, retry, ambiguous delivery, preparation failure, + pause-race reset, and stale-stream quarantine notify response availability. + Task-control cancellation covers the affected conversation. Content preparation + and lease bookkeeping do not create response-availability debt. +- Endpoint removal deletes its response actions. Other authority checks remain + in the response service before any provider write. The worker does not cache + authorization or infer authority from a queued action. + +The shared capability is `notifyChatResponseAvailability`: an identifier-only +hint that an existing response slot may have changed. Fast-response and ordinary +publication owners do not inspect GitHub payloads or choose GitHub behavior. +Only the GitHub acknowledgement owner parses repository thread identity and +applies its response protocol. Notification serialization explicitly picks IDs +so passing a publication row cannot carry its payload into scheduler hints. + +Provider requests remain outside database transactions. The existing repository +lease, current source/principal checks, response markers, shared pull-response +identity, version fencing, and final-receipt preservation remain in charge of +provider effects. A recovered known receipt is marked processed so it cannot +remain permanently in the pending acknowledgement inventory. Admission is checked +at the actual fetch boundary after lease renewal; an in-flight request may settle, +but a stopped worker cannot start the next provider request. Stop-related failures +do not consume the provider retry budget. + +The migration adds three concurrent, idempotent indexes: pending acknowledgement +IDs, fast-response delivery lookup, and publication comment/destination lookup. +It uses the existing top-level migration mechanism and snapshot pruning rules. +These indexes avoid rebuilding pending work by scanning completed history. + +Supported writes are the service's process-local transaction notification model. +Independent SQL writers or a second instance do not acquire a new wake guarantee +from this change. Host sleep eligibility and external alarm coverage are unchanged. +A local deadline is not evidence that a stopped host can wake itself. + +## Verification evidence retained during development + +- The first eight lifecycle tests failed on a PostgreSQL JSON-expression precedence + error in the lease lookup. Parenthesizing the repository-ID expression fixed + the query; all eight then passed. +- Removing the real fast-response settlement notification made the regression + fail: fallback waited until the response expired. Restoring that signal is the + implementation under test. The assertion measures when the action settles, + separately from the coordinator's final coalesced bookkeeping pass. +- The initial concurrent-index application failed because the migration was not + in the existing top-level statement registry. Registering it fixed application; + a second application also completed successfully. +- Expanded tests exposed provider error normalization at the fetch boundary. + The worker now preserves a stopped attempt without charging a provider failure. + The fault fixture was also narrowed to response-comment POSTs so installation + access-token requests were not mistaken for acknowledgement attempts. +- Fourteen new real-PostgreSQL cases and the startup guard pass. They cover empty + and retained-history quiet, lease release, restart, drain, standby, fast-response + failure/expiry/publication/unknown outcome, stale-stream quarantine, outer + rollback/commit, saved provider retry, and stop between provider requests. +- Full chat/fast-response acceptance, workspace typecheck/build, remote CI and + review remain pending. This document does not yet claim PR readiness. + +Workspace typecheck/build now pass. The fast-response test harness supports an +explicit fresh `_test` PostgreSQL database, retaining embedded PostgreSQL by +default. Its first run passed the 69 test bodies but failed cleanup on an invalid +client-close call; the corrected setup then hit retained fixed-prefix fixtures +when the same database was reused. A fresh database passes all 69 tests and cleanup. +Final server typecheck passes. All three migration indexes are valid in PostgreSQL. +The complete affected chat suites are still running; remote checks/review have not +been claimed complete. + +A rollback-only PostgreSQL plan experiment seeded 10,000 completed response +actions and 10,000 unrelated fast-response records alongside one pending action +and its matching request. The pending-ID lookup used +`chat_actions_github_ack_work_idx` and read one row (two shared buffers); the +request lookup used `fast_response_requests_delivery_idx` and read one row +(three shared buffers). This checks the two lookup access paths, not whole-worker +latency or deployed fleet savings. All experimental writes were rolled back. + +The first complete chat acceptance passed all 1,554 tests without skips in +496.42 seconds. Final review then added three receipt-lifecycle cases: confirmed +received/failed receipts settle, but cancelled receipts stay cancelled. The new +cancelled case failed before correction. Settlement now applies only to pending +states and compares the observed status before writing. The final version of the +producer-removal test also failed as expected without the settlement notification. +The corrected implementation is undergoing a fresh full acceptance run with the +three additional cases; earlier full results are not substituted for that run. + +After the final receipt correction, all 17 new lifecycle cases and the startup +guard pass. Workspace typecheck and build pass again. The final complete chat +run remains in progress; source and branch remain fixed during acceptance. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 72b9c1d9f2b..fbd2648e2c8 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -515,6 +515,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0387_huge_skrulls.sql", "0388_lying_trish_tilby.sql", "0389_workable_spencer_smythe.sql", + "0390_long_cargill.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0390_long_cargill.sql b/packages/db/src/migrations/0390_long_cargill.sql new file mode 100644 index 00000000000..ca4eafbda46 --- /dev/null +++ b/packages/db/src/migrations/0390_long_cargill.sql @@ -0,0 +1,5 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_ack_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_response_comment' + and "chat_actions"."delivery_id" is not null and coalesce("chat_actions"."result"->>'final', 'false') <> 'true' + and ("chat_actions"."status" = 'received' or ("chat_actions"."status" = 'failed' and "chat_actions"."result"->>'retryable' = 'true'));--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_publications_comment_destination_idx" ON "chat_publications" USING btree ("company_id","comment_id","endpoint_id","conversation_id") WHERE "chat_publications"."comment_id" is not null;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "fast_response_requests_delivery_idx" ON "fast_response_requests" USING btree ("company_id","endpoint_id","delivery_id") WHERE "fast_response_requests"."delivery_id" is not null; diff --git a/packages/db/src/migrations/meta/0385_snapshot.json b/packages/db/src/migrations/meta/0390_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0385_snapshot.json rename to packages/db/src/migrations/meta/0390_snapshot.json index 0151330ac15..083cc645c95 100644 --- a/packages/db/src/migrations/meta/0385_snapshot.json +++ b/packages/db/src/migrations/meta/0390_snapshot.json @@ -1,6 +1,6 @@ { - "id": "b8b61c60-3d1d-4f3d-9df3-b633a3244150", - "prevId": "45d3a35e-2d6d-4fc1-8445-c35eef1ea173", + "id": "887083ec-214e-44ab-a1ab-25bff321118e", + "prevId": "8bd7f279-1063-4ba6-83bb-b85cba334eb3", "version": "7", "dialect": "postgresql", "tables": { @@ -5721,6 +5721,28 @@ } }, "indexes": { + "approvals_pending_retention_idx": { + "name": "approvals_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"approvals\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "approvals_company_status_type_idx": { "name": "approvals_company_status_type_idx", "columns": [ @@ -9594,6 +9616,22 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_actions_github_ack_work_idx": { + "name": "chat_actions_github_ack_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"delivery_id\" is not null and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -12021,6 +12059,40 @@ "method": "btree", "with": {} }, + "chat_publications_comment_destination_idx": { + "name": "chat_publications_comment_destination_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "comment_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "conversation_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_publications\".\"comment_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_publications_idempotency_uq": { "name": "chat_publications_idempotency_uq", "columns": [ @@ -14138,6 +14210,24 @@ "primaryKey": false, "notNull": false }, + "provider_account_id": { + "name": "provider_account_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "cleanup_confirmed_at": { + "name": "cleanup_confirmed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "cleanup_next_attempt_at": { + "name": "cleanup_next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "expires_at": { "name": "expires_at", "type": "timestamp with time zone", @@ -14160,6 +14250,22 @@ } }, "indexes": { + "chat_voice_inbound_cleanup_idx": { + "name": "chat_voice_inbound_cleanup_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_voice_inbound_expiry_idx": { "name": "chat_voice_inbound_expiry_idx", "columns": [ @@ -14912,6 +15018,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_voice_sessions_report_window_idx": { + "name": "chat_voice_sessions_report_window_idx", + "columns": [ + { + "expression": "ended_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -30382,6 +30510,34 @@ "method": "btree", "with": {} }, + "fast_response_requests_delivery_idx": { + "name": "fast_response_requests_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "delivery_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"fast_response_requests\".\"delivery_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "fast_response_requests_history_idx": { "name": "fast_response_requests_history_idx", "columns": [ @@ -41895,6 +42051,28 @@ } }, "indexes": { + "join_requests_pending_retention_idx": { + "name": "join_requests_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"join_requests\".\"status\" = 'pending_approval'", + "concurrently": false, + "method": "btree", + "with": {} + }, "join_requests_invite_unique_idx": { "name": "join_requests_invite_unique_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 20ef0905b3b..f974a955419 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2717,6 +2717,13 @@ "when": 1792926867183, "tag": "0389_workable_spencer_smythe", "breakpoints": true + }, + { + "idx": 390, + "version": "7", + "when": 1792962182535, + "tag": "0390_long_cargill", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_channels.ts b/packages/db/src/schema/chat_channels.ts index 9f83918b3d8..b57c5b06a3a 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -564,6 +564,8 @@ export const chatPublications = pgTable( name: "chat_publications_company_comment_fk", }), index("chat_publications_work_idx").on(table.state, table.nextAttemptAt), + index("chat_publications_comment_destination_idx").on(table.companyId, table.commentId, table.endpointId, table.conversationId) + .where(sql`${table.commentId} is not null`), uniqueIndex("chat_publications_idempotency_uq").on( table.companyId, table.idempotencyKey, @@ -677,6 +679,9 @@ export const chatActions = pgTable( table.endpointId, table.providerActionId, ), + index("chat_actions_github_ack_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment' + and ${table.deliveryId} is not null and coalesce(${table.result}->>'final', 'false') <> 'true' + and (${table.status} = 'received' or (${table.status} = 'failed' and ${table.result}->>'retryable' = 'true'))`), foreignKey({ columns: [table.companyId, table.deliveryId], foreignColumns: [chatDeliveries.companyId, chatDeliveries.id], diff --git a/packages/db/src/schema/fast_responses.ts b/packages/db/src/schema/fast_responses.ts index 2dfdb72ef17..a7fb859d33b 100644 --- a/packages/db/src/schema/fast_responses.ts +++ b/packages/db/src/schema/fast_responses.ts @@ -81,6 +81,8 @@ export const fastResponseRequests = pgTable( t.companyId, t.sourceKey, ), + delivery: index("fast_response_requests_delivery_idx").on(t.companyId, t.endpointId, t.deliveryId) + .where(sql`${t.deliveryId} is not null`), history: index("fast_response_requests_history_idx").on( t.companyId, t.acceptedAt, diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 7123ae6c78e..c282a0dfd7f 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -94,6 +94,10 @@ describe("createChatReconciliationCoordinator", () => { expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications("); expect(publicationMaintenance).not.toContain("reconcileTerminalConfirmationActions("); expect(source).toContain("chatChannels.registerInteractionContinuationWork(deliveryWork,"); + expect(source).toContain("chatChannels.registerGitHubAcknowledgementWork(deliveryWork,"); + const github = readFileSync(new URL("../services/chat-channels/github/maintenance.ts", import.meta.url), "utf8"); + expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processPending()"); + expect(github).toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()"); expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,"); expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,"); }); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 437ea0daf5d..8e86aa0b726 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -33,7 +33,7 @@ import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-wor import { aiConnectionService } from "../services/ai-connections.js"; import { fastResponseService } from "../services/fast-responses.js"; import { fastResponseReceipt } from "../services/fast-response-provider.js"; -import { fastResponseRequests } from "@paperclipai/db"; +import { companyFastResponses, fastResponseRequests } from "@paperclipai/db"; import { toolAccessService } from "../services/tool-access.js"; import { toolActionRequests, toolInvocations, toolPolicies } from "@paperclipai/db"; import { GitHubPublicationLeaseLost, withGitHubPublicationLease } from "../services/chat-github-publication-lease.js"; @@ -3554,6 +3554,240 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(attempts).toBe(2); expect(f.responseWrites).toHaveLength(writes); }); + 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)); + const errors: unknown[] = []; + let canRun = true; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + const register = () => f.service.registerGitHubAcknowledgementWork(coordinator, () => canRun); + let worker = register(); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const claimed = new Promise(resolve => { entered = resolve; }); + let lease: Promise | undefined; + try { + await worker.ready; + expect(errors).toEqual([]); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + expect(coordinator.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + vi.setSystemTime(Date.now() - 120_000); + lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id, + repositoryId: String(resource.metadata.providerRepositoryId), number: 418 }, f.providerFetch, async () => { entered(); await hold; }); + await claimed; + if (mode === "restart") await coordinator.stop(); + if (mode === "drain" || mode === "standby") canRun = false; + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("received"); expect(f.responseWrites).toEqual([]); + release(); await lease; + if (mode === "drain" || mode === "standby") { + await vi.advanceTimersByTimeAsync(1000); + expect(f.responseWrites).toEqual([]); + canRun = true; worker.wake(); + } + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = register(); await worker.ready; + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].status).toBe("processed"); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(f.responseWrites).toHaveLength(1); + expect(f.responseWrites[0].body).toContain("Working on this"); + const settledReads = vi.spyOn(db, "select"), settledWrites = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(settledReads).not.toHaveBeenCalled(); expect(settledWrites).not.toHaveBeenCalled(); + settledReads.mockRestore(); settledWrites.mockRestore(); + expect(errors).toEqual([]); + } finally { + release(); await lease; await coordinator.stop(); vi.useRealTimers(); + } + }); + + it.each(["failure", "expiry", "published", "delivery_unknown", "streaming quarantine"] as const)("releases a GitHub acknowledgement after fast response %s without maintenance polling", async outcome => { + await instanceSettingsService(db).updateExperimental({ enableFastResponses: true }); + const f = await reviewBotFixture(); + const binding = await aiConnectionService(db).save(f.companyId, "owner-user", { provider: "openrouter", method: "api_key", name: "Fast GitHub", ownership: "shared", apiKey: "fixture-key", agentIds: [], allAgents: true }, "fixture-key"); + const fast = fastResponseService(db, { authorizeExternal: f.service.authorizeFastResponse, + provider: async () => ({ ...(outcome === "failure" ? { errorCode: "invalid_output", noProviderWork: true } : { text: "I’ll check it." }), receipt: fastResponseReceipt({ usage: { inputTokens: 80, outputTokens: 8 } }) }) }); + await fast.configure(f.companyId, "owner-user", { enabled: true, ...binding, model: "openai/gpt-oss-120b", allowSponsored: true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [job] = await db.select().from(fastResponseRequests).where(eq(fastResponseRequests.companyId, f.companyId)); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime()); + }, { timeout: 5000 }); + expect(f.responseWrites).toEqual([]); + if (outcome === "expiry") await vi.advanceTimersByTimeAsync(job.expiresAt.getTime() - Date.now()); + else { + await fast.process(job); + if (outcome === "streaming quarantine") { + await db.update(chatPublications).set({ state: "streaming", updatedAt: new Date(0) }) + .where(eq(chatPublications.idempotencyKey, `fast-response:${job.id}`)); + await vi.advanceTimersByTimeAsync(Math.max(0, job.expiresAt.getTime() - Date.now())); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + expect(f.responseWrites).toEqual([]); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + await f.service.processPendingPublications(); + } else if (outcome === "delivery_unknown") { + await db.transaction(async tx => { + await notifyChatPublicationWork(tx, { companyId: job.companyId, issueId: job.issueId }); + await tx.update(chatPublications).set({ state: "delivery_unknown" }).where(eq(chatPublications.idempotencyKey, `fast-response:${job.id}`)); + }); + } else if (outcome === "published") await f.service.processPendingPublications(); + } + let processedAt: number | undefined; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("processed"); + processedAt ??= Date.now(); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + if (outcome !== "expiry" && outcome !== "streaming quarantine") expect(processedAt).toBeLessThan(job.expiresAt.getTime()); + expect(f.responseWrites).toHaveLength(["failure", "expiry"].includes(outcome) ? 1 : 0); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("wakes a GitHub acknowledgement only after the outer fast-response transaction commits", async () => { + await instanceSettingsService(db).updateExperimental({ enableFastResponses: true }); + const f = await reviewBotFixture(); + const binding = await aiConnectionService(db).save(f.companyId, "owner-user", { provider: "openrouter", method: "api_key", name: "Fast GitHub", ownership: "shared", apiKey: "fixture-key", agentIds: [], allAgents: true }, "fixture-key"); + const fast = fastResponseService(db, { authorizeExternal: f.service.authorizeFastResponse }); + await fast.configure(f.companyId, "owner-user", { enabled: true, ...binding, model: "openai/gpt-oss-120b", allowSponsored: true }); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [job] = await db.update(fastResponseRequests).set({ expiresAt: new Date(Date.now() + 60_000) }) + .where(eq(fastResponseRequests.companyId, f.companyId)).returning(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime()); + const skip = async (tx: Parameters[0]>[0]) => { + await tx.update(companyFastResponses).set({ enabled: false }).where(eq(companyFastResponses.companyId, f.companyId)); + expect(await fastResponseService(tx as unknown as Db).process(job)).toMatchObject({ status: "unavailable" }); + await vi.advanceTimersByTimeAsync(100); + expect(f.responseWrites).toEqual([]); + expect((await db.select().from(fastResponseRequests).where(eq(fastResponseRequests.id, job.id)))[0].status).toBe("pending"); + }; + await expect(db.transaction(async tx => { await skip(tx); throw new Error("rollback fast response skip"); })).rejects.toThrow("rollback fast response skip"); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(idleWorkSnapshot().active).toBe(0); }); + expect(f.responseWrites).toEqual([]); + expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime()); + await db.transaction(skip); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(f.responseWrites).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + expect(Date.now()).toBeLessThan(job.expiresAt.getTime()); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["drain", "standby", "shutdown"] as const)("stops GitHub acknowledgement sends at %s between provider requests", async mode => { + const f = await reviewBotFixture(); + f.setSupplementalProviderFetch(async (input, init) => String(input).includes("/issues/418/comments") && init?.method === "POST" ? new Response("unavailable", { status: 503 }) : undefined); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("failed"); expect(action.result?.attempts).toBe(1); + await db.update(chatActions).set({ result: { ...action.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, action.id)); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + f.setSupplementalProviderFetch(async input => { + if (String(input).includes("/issues/418/comments?")) { entered(); await held; } + return undefined; + }); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + let worker = f.service.registerGitHubAcknowledgementWork(coordinator, () => canRun); + let stopping: Promise | undefined; + try { + await started; + if (mode === "shutdown") stopping = coordinator.stop(); else canRun = false; + release(); await worker.ready; await stopping; + expect(f.responseWrites).toEqual([]); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].result?.attempts).toBe(1); + canRun = true; + if (mode === "shutdown") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = f.service.registerGitHubAcknowledgementWork(coordinator, () => true); await worker.ready; + } else worker.wake(); + await vi.waitFor(() => { expect(f.responseWrites).toHaveLength(1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { release(); await stopping; await coordinator.stop(); } + }); + + it("retries a failed GitHub acknowledgement at its saved deadline", async () => { + const f = await reviewBotFixture(); + let unavailable = true, attempts = 0; + f.setSupplementalProviderFetch(async (input, init) => { + if (String(input).includes("/issues/418/comments") && init?.method === "POST") { attempts++; if (unavailable) return new Response("unavailable", { status: 503 }); } + return undefined; + }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("failed"); expect(attempts).toBe(1); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(coordinator.nextWakeAt()).toBe(Date.parse(String(action.result?.retryAt))); }); + unavailable = false; + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(f.responseWrites).toHaveLength(1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 }); + expect(attempts).toBe(2); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["received", "failed", "cancelled"] as const)("preserves the %s GitHub receipt lifecycle during acknowledgement recovery", async status => { + const f = await toolReplyFixture(); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.result?.id).toBeTruthy(); + await db.update(chatActions).set({ status }).where(eq(chatActions.id, action.id)); + const writes = f.responseWrites.length; + await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(action.deliveryId!); + const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, action.id)); + expect(settled.status).toBe(status === "cancelled" ? "cancelled" : "processed"); + expect(settled.result).toEqual(action.result); + expect(f.responseWrites).toHaveLength(writes); + }); + it("does not post a GitHub working comment when the scheduler declines the request", async () => { const f = await reviewBotFixture(); f.wakeup.mockImplementation(async (agentId, opts) => { diff --git a/server/src/__tests__/fast-responses.test.ts b/server/src/__tests__/fast-responses.test.ts index 19eae5d692f..b08e4aa3a99 100644 --- a/server/src/__tests__/fast-responses.test.ts +++ b/server/src/__tests__/fast-responses.test.ts @@ -37,7 +37,7 @@ import { instanceSettingsService } from "../services/instance-settings.js"; import { settleConversationTurn } from "../services/agent-conversations.js"; import { buildLowTrustSourceTrust, LOW_TRUST_QUARANTINED_BODY } from "../services/source-trust.js"; import { deliverBudgetEnforcement } from "../services/budgets.js"; -let database: Awaited>, +let database: Awaited> | undefined, db: ReturnType, home: string; beforeAll(async () => { @@ -46,12 +46,15 @@ beforeAll(async () => { ); vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_INSTANCE_ID", "fast-response-fixture"); - database = await startEmbeddedPostgresTestDatabase( + const external = process.env.PAPERCLIP_TEST_DATABASE_URL; + if (external && !new URL(external).pathname.endsWith("_test")) throw new Error("Fast-response tests require a database ending in _test"); + database = external ? undefined : await startEmbeddedPostgresTestDatabase( "paperclip-fast-response-db-", ); - db = createDb(database.connectionString); + db = createDb(external ?? database!.connectionString); }, 90000); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); diff --git a/server/src/app.ts b/server/src/app.ts index 60f666fd2c8..40f402f70d0 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1289,6 +1289,7 @@ export async function createApp( chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index ef9dfe40f62..6857b11e762 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -10,6 +10,7 @@ import { notifyChatConnectionHealth } from "./chat-channels/connection-health-no import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js"; +import { githubResponseCommentService } from "./chat-github-response-comments.js"; import { createGitHubMaintenance } from "./chat-channels/github/maintenance.js"; import { createSpekoChannelService, authorizeSpekoPublicVoiceTask, replaceSpekoChannelRuntime, SPEKO_CONVERSATION_IS_LINEAR } from "./chat-channels/speko/runtime.js"; import { prepareDiscordRuntimeCommand } from "./chat-channels/discord/runtime-commands.js"; @@ -3506,6 +3507,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { registerInteractionContinuationWork, + registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork, registerTerminalConfirmationWork, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices, diff --git a/server/src/services/chat-channels/github/acknowledgement-work.ts b/server/src/services/chat-channels/github/acknowledgement-work.ts new file mode 100644 index 00000000000..2194785b8d6 --- /dev/null +++ b/server/src/services/chat-channels/github/acknowledgement-work.ts @@ -0,0 +1,96 @@ +import { and, asc, eq, gt, isNotNull, or, sql } from "drizzle-orm"; +import { chatActions, chatConversations, chatEndpointLeases, chatEndpointResources, + chatPublications, fastResponseRequests, 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"; + +type Scope = ChatResponseScope & { endpointId?: string; repositoryId?: string; number?: number }; +type Entry = { id: string; companyId: string; endpointId: string; deliveryId: string; + issueId: string | null; conversationId: string | null; repositoryId: string | null; number: number | null; dueAt: number }; +type FastResponse = { status: string; expiresAt: string; publicationState: string | null }; + +export function registerGitHubAcknowledgementWork( + 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]*)?$')`; + const selection = { + id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId, + deliveryId: chatActions.deliveryId, conversationId: chatActions.conversationId, issueId: chatConversations.issueId, + repositoryId, number, status: chatActions.status, result: chatActions.result, + 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})`, + fast: sql`(select jsonb_build_object('status', f.status, 'expiresAt', f.expires_at, + 'publicationState', p.state) from ${fastResponseRequests} f left join ${chatPublications} p + on p.company_id = f.company_id and p.comment_id = f.comment_id + and p.endpoint_id = f.endpoint_id and p.conversation_id = f.conversation_id + where f.company_id = ${chatActions.companyId} and f.endpoint_id = ${chatActions.endpointId} + and f.delivery_id = ${chatActions.deliveryId} limit 1)`, + }; + async function page(scope?: Scope, cursor?: string, actionId?: string) { + return db.select(selection).from(chatActions) + .leftJoin(chatConversations, and(eq(chatConversations.id, chatActions.conversationId), eq(chatConversations.companyId, chatActions.companyId))) + .leftJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), eq(chatEndpointResources.companyId, chatActions.companyId))) + .where(and(eq(chatActions.kind, "github_response_comment"), isNotNull(chatActions.deliveryId), + sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`, + or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)), + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(chatConversations.issueId, 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); + } + function entry(row: Awaited>[number]): Entry | null { + let dueAt = row.status === "failed" ? Date.parse(String(row.result?.retryAt)) : 0; + if (!Number.isFinite(dueAt)) throw new Error("GitHub acknowledgement has an invalid retry deadline"); + const fast = row.fast; + if (fast?.publicationState === "streaming") return null; + if (fast && !["published", "delivery_unknown"].includes(fast.publicationState ?? "") && + (["pending", "running"].includes(fast.status) || ["pending", "retry"].includes(fast.publicationState ?? ""))) { + dueAt = Math.max(dueAt, Date.parse(fast.expiresAt)); + } + if (row.leaseExpiresAt) dueAt = Math.max(dueAt, Date.parse(row.leaseExpiresAt)); + 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 }; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubAcknowledgements, + 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) { const value = 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 acknowledgement recovery stopped"); + }, + check: async (value: Entry, signal?: AbortSignal) => { + const [row] = await page(undefined, undefined, value.id); + const current = row ? 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-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts index 8d09defd0f3..5ae081cda53 100644 --- a/server/src/services/chat-channels/github/maintenance.ts +++ b/server/src/services/chat-channels/github/maintenance.ts @@ -13,7 +13,7 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet options.scheduleMessageProcessing(async () => { try { // Provider I/O must not block Slack, Teams, or ordinary message receipts. - await githubResponseCommentService(db, fetchImpl).processPending(); + await githubResponseCommentService(db, fetchImpl).processTerminalResponses(); await githubReviewCheckService(db, fetchImpl).processPending(); } finally { githubMaintenancePending = false; diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index 8a4c8f7ed7a..5f3507d5473 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; @@ -121,6 +122,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur ? "cancelled" : "failed"; await notifyChatPublicationWork(tx, state === "failed" || state === "cancelled" ? publication : undefined); + if (state === "retry" || state === "delivery_unknown") await notifyChatResponseAvailability(tx, publication); const [ownedPublication] = await tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-channels/publications/queue.ts b/server/src/services/chat-channels/publications/queue.ts index a1b3b3e7bf3..f71f8725973 100644 --- a/server/src/services/chat-channels/publications/queue.ts +++ b/server/src/services/chat-channels/publications/queue.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; import { unresolvedSlackPublicationCondition } from "../../connectors/slack-publication.js"; @@ -242,9 +243,11 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti ), ), ); - const candidates = await tx.select({ id: chatPublications.id }).from(chatPublications).where(condition).for("update", { skipLocked: true }); + const candidates = await tx.select({ id: chatPublications.id, companyId: chatPublications.companyId, issueId: chatPublications.issueId }) + .from(chatPublications).where(condition).for("update", { skipLocked: true }); if (!candidates.length) return []; await notifyChatPublicationWork(tx); + for (const candidate of candidates) await notifyChatResponseAvailability(tx, candidate); return tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-channels/publications/task-control.ts b/server/src/services/chat-channels/publications/task-control.ts index ebe1a2b4ee9..040ea993dd6 100644 --- a/server/src/services/chat-channels/publications/task-control.ts +++ b/server/src/services/chat-channels/publications/task-control.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import type { DbOrTransaction } from "../runtime-types.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; @@ -116,6 +117,7 @@ export function createChatPublicationTaskControl(db: Db) { // finishes after /new or /close. A racing insert that commits after this // transaction is rejected by the completed-conversation send guard. await notifyChatPublicationWork(tx); + await notifyChatResponseAvailability(tx, publication); await tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-channels/publications/worker.ts b/server/src/services/chat-channels/publications/worker.ts index 013ef9b3327..975b20a7b6f 100644 --- a/server/src/services/chat-channels/publications/worker.ts +++ b/server/src/services/chat-channels/publications/worker.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import { notifyVoicePublication } from "../../voice/voice-reply-notifications.js"; import { createGitHubPublicationPolicy } from "../github/publication-policy.js"; @@ -155,6 +156,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp : Math.min(60_000, 2 ** Math.max(0, attempts) * 1_000); const result = await db.transaction(async (tx) => { await notifyChatPublicationWork(tx, terminal ? publication : undefined); + if (!terminal) await notifyChatResponseAvailability(tx, publication); const [settled] = await tx .update(chatPublications) .set({ @@ -475,6 +477,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp await publicationLease.assertOwned(); await db.transaction(async (tx) => { await notifyChatPublicationWork(tx); + await notifyChatResponseAvailability(tx, publication); return tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-github-publication-lease.ts b/server/src/services/chat-github-publication-lease.ts index 36cf8a9880d..46d9e3b862b 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.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 8caff8894b9..3fbd7988063 100644 --- a/server/src/services/chat-github-response-comments.ts +++ b/server/src/services/chat-github-response-comments.ts @@ -5,6 +5,9 @@ import { fastResponseRequests, chatPublications, type Db, } from "@paperclipai/db"; +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; +import { registerGitHubAcknowledgementWork } from "./chat-channels/github/acknowledgement-work.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { conflict, forbidden, HttpError } from "../errors.js"; import { githubChatPrincipalAccess } from "./chat-github-access.js"; import { githubAutomaticAdmission } from "./chat-github-events.js"; @@ -42,6 +45,7 @@ const responseRunColumns = { )`, }; const generation = (endpoint: Source["endpoint"]) => (endpoint.setup as { runtimeGeneration?: number }).runtimeGeneration ?? 0; +const stoppedAcknowledgement = Symbol("stopped GitHub acknowledgement"); const actionKey = (deliveryId: string) => `github_response:${deliveryId}`; const responseMarker = (source: Source) => isMainPullResponse(source) ? pullResponseMarker(source) : requestResponseMarker(source); @@ -63,18 +67,23 @@ async function findResponseComment(source: Source, api: Api, legacyMarkers: stri * Reconnect epochs fence publication attempts, not the durable comment identity. * Legacy receipts are also checked against the App-owned provider comment. */ async function stageResponse(db: Db, source: Source) { - const [inserted] = await db.insert(chatActions).values({ - companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - conversationId: source.conversation.id, deliveryId: source.delivery.id, - principalId: source.delivery.principalId, kind: "github_response_comment", - providerActionId: actionKey(source.delivery.id), - payload: { botExternalId: source.endpoint.botExternalId, agentId: source.endpoint.assignedAgentId }, - status: "received", - }).onConflictDoNothing().returning(); - const response = inserted ?? (await db.select().from(chatActions).where(and( + const lookup = () => db.select().from(chatActions).where(and( eq(chatActions.companyId, source.endpoint.companyId), eq(chatActions.endpointId, source.endpoint.id), eq(chatActions.providerActionId, actionKey(source.delivery.id)), - )))[0]; + )).then(rows => rows[0]); + const existing = await lookup(); + const response = existing ?? await db.transaction(async tx => { + await notifyChatResponseAvailability(tx, { companyId: source.endpoint.companyId, deliveryId: source.delivery.id }); + const [inserted] = await tx.insert(chatActions).values({ + companyId: source.endpoint.companyId, endpointId: source.endpoint.id, + conversationId: source.conversation.id, deliveryId: source.delivery.id, + principalId: source.delivery.principalId, kind: "github_response_comment", + providerActionId: actionKey(source.delivery.id), + payload: { botExternalId: source.endpoint.botExternalId, agentId: source.endpoint.assignedAgentId }, + status: "received", + }).onConflictDoNothing().returning(); + return inserted; + }) ?? await lookup(); if (!response || response.conversationId !== source.conversation.id || (response.payload.botExternalId !== undefined && response.payload.botExternalId !== source.endpoint.botExternalId) || response.payload.agentId !== source.endpoint.assignedAgentId) @@ -107,6 +116,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() }) + .where(and(eq(chatActions.id, response.id), eq(chatActions.status, response.status))); + }); return { id: String(response.result.id), url: String(response.result.url) }; } const marker = responseMarker(source); @@ -185,13 +198,18 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { throw forbidden("Invalid GitHub request repository"); return { ...row, repositoryId, repository: match[1]!, number: Number(match[2]), replyId: match[3] ?? null }; } - async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise) { + async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise, canContinue = () => true) { + if (!canContinue()) return; const source = await context(deliveryId); // Persist before competing for the publication lane. A busy lane leaves // durable acknowledgement work for maintenance instead of losing it. await stageResponse(db, source); + if (!canContinue()) return; return withGitHubPublicationLease(db, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - repositoryId: source.repositoryId, number: source.number }, fetchImpl, async lease => { + repositoryId: source.repositoryId, number: source.number }, (input, init) => { + if (!canContinue()) throw stoppedAcknowledgement; + return fetchImpl(input, init); + }, async lease => { const assertCurrent = async () => { await assertRun?.(); const current = await context(deliveryId); @@ -247,7 +265,10 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { (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({ updatedAt: new Date() }).where(eq(chatActions.id, existing.id)); }); + await lease.commit(async tx => { await tx.update(chatActions).set({ + status: ["received", "failed"].includes(existing.status) ? "processed" : existing.status, + updatedAt: new Date(), + }).where(and(eq(chatActions.id, existing.id), eq(chatActions.status, existing.status))); }); return { id: String(existing.result.id), url: String(existing.result.url) }; } if (finalBody === undefined) { @@ -395,6 +416,9 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`, sql`(${chatActions.result}->>'retryAt')::timestamptz <= now()`)))).orderBy(asc(chatActions.createdAt)).limit(limit); for (const row of rows) if (row.deliveryId) await tryAcknowledge(row.deliveryId); + return rows.length + await processTerminalResponses(limit); + } + async function processTerminalResponses(limit = 10) { // Recovery can stop execution before a milestone publication is staged. // Revisit acknowledged requests with terminal execution, even after a // restart; successful acknowledgement is not successful task completion. @@ -419,21 +443,28 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { or coalesce(r.context_snapshot->'wakeCommentIds', '[]'::jsonb) ? l.comment_id::text))))`, )).orderBy(asc(chatActions.updatedAt)).limit(limit); for (const row of stopped) if (row.deliveryId) await tryAcknowledge(row.deliveryId); - return rows.length + stopped.length; + return stopped.length; } - async function tryAcknowledge(deliveryId: string) { - try { return await acknowledge(deliveryId); } + async function tryAcknowledge(deliveryId: string, canContinue = () => true) { + try { return await acknowledge(deliveryId, undefined, undefined, canContinue); } catch (error) { + if (error === stoppedAcknowledgement || !canContinue()) return; const [row] = await db.select().from(chatActions).where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.deliveryId, deliveryId))); if (!row || row.result?.final === true) return; const attempts = Number(row.result?.attempts ?? 0) + 1; const denied = error instanceof Error && "status" in error && error.status === 403; - await db.update(chatActions).set({ status: denied ? "cancelled" : "failed", result: { - ...row.result, attempts, retryable: !denied && attempts < 8, - retryAt: new Date(Date.now() + Math.min(300000, 1000 * 2 ** attempts)).toISOString(), - code: denied ? "authorization_changed" : "acknowledgement_unconfirmed", - }, updatedAt: new Date() }).where(and(eq(chatActions.id, row.id), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); + await db.transaction(async tx => { + await notifyChatResponseAvailability(tx, { companyId: row.companyId, deliveryId }); + await tx.update(chatActions).set({ status: denied ? "cancelled" : "failed", result: { + ...row.result, attempts, retryable: !denied && attempts < 8, + retryAt: new Date(Date.now() + Math.min(300000, 1000 * 2 ** attempts)).toISOString(), + code: denied ? "authorization_changed" : "acknowledgement_unconfirmed", + }, updatedAt: new Date() }).where(and(eq(chatActions.id, row.id), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); + }); } } - return { tryAcknowledge, processPending, failRun, finishRun, hasWorkingResponseForRun }; + return { tryAcknowledge, processPending, processTerminalResponses, failRun, finishRun, hasWorkingResponseForRun, + registerAcknowledgementWork: (coordinator: ReturnType, canRun: () => boolean) => + registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun), + }; } diff --git a/server/src/services/chat-response-availability.ts b/server/src/services/chat-response-availability.ts new file mode 100644 index 00000000000..29de7ca303e --- /dev/null +++ b/server/src/services/chat-response-availability.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string }; + +export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) { + 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-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 5e0d05262c9..9b878913e8d 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -17,6 +18,7 @@ export async function notifyChatPublicationWork( // release its provider response slot. Callers supply the affected issue only // for those changes; transport claim/lease bookkeeping has no projection debt. if (milestoneScope?.issueId) { + await notifyChatResponseAvailability(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId }); await notifyChatRunMilestones(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId }); } } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 720df8a8215..ccdf2c0a356 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", + chatGitHubAcknowledgements: "chat-github-acknowledgements", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", chatReceivers: "chat-receivers", diff --git a/server/src/services/fast-responses.ts b/server/src/services/fast-responses.ts index 8ba6665dea6..4f0d2dddb6d 100644 --- a/server/src/services/fast-responses.ts +++ b/server/src/services/fast-responses.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { insertIssueComment } from "./issue-comment-writes.js"; import { randomUUID } from "node:crypto"; import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js"; @@ -195,6 +196,7 @@ export async function enqueueFastResponse(tx: Db, source: FastResponseSource) { if (source.acceptedAt.getTime() + FAST_RESPONSE_DEADLINE_MS <= Date.now()) return; await notifyDeliveryWork(tx, DELIVERY_QUEUES.fastResponse); + if (source.endpointId && source.deliveryId) await notifyChatResponseAvailability(tx, { companyId: source.companyId, deliveryId: source.deliveryId }); await tx .insert(fastResponseRequests) .values({ @@ -549,20 +551,23 @@ export function fastResponseService( } } async function skip(request: Request, reason: string) { - await db - .update(fastResponseRequests) - .set({ - status: "skipped", - publicationStatus: "suppressed", - errorCode: reason, - finishedAt: new Date(), - }) - .where( - and( - eq(fastResponseRequests.id, request.id), - eq(fastResponseRequests.status, "pending"), - ), - ); + await db.transaction(async tx => { + if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId }); + await tx + .update(fastResponseRequests) + .set({ + status: "skipped", + publicationStatus: "suppressed", + errorCode: reason, + finishedAt: new Date(), + }) + .where( + and( + eq(fastResponseRequests.id, request.id), + eq(fastResponseRequests.status, "pending"), + ), + ); + }); } async function settle(request: Request, outcome: FastResponseOutcome) { await withBudgetSpan("budget.accounting.settlement", () => withAccountingTransaction( @@ -625,6 +630,7 @@ export function fastResponseService( }, ); const finishedAt = new Date(); + if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId }); await tx .update(fastResponseRequests) .set({ @@ -993,6 +999,7 @@ export function fastResponseService( idempotencyKey: `fast-response:${request.id}`, payload: safe, }); + if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId }); await tx .update(fastResponseRequests) .set({