mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 14:16:39 +01:00
fix(server): wake GitHub acknowledgements from committed changes
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
3fcf41fdfb
commit
43900c1f94
25 files changed
+793
-44
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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",
|
||||
]);
|
||||
|
||||
|
||||
@@ -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;
|
||||
+180
-2
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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],
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,");
|
||||
});
|
||||
|
||||
@@ -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<void>(resolve => { release = resolve; });
|
||||
const claimed = new Promise<void>(resolve => { entered = resolve; });
|
||||
let lease: Promise<unknown> | 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<Parameters<Db["transaction"]>[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<void>(resolve => { release = resolve; });
|
||||
const started = new Promise<void>(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<void> | 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) => {
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>,
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined,
|
||||
db: ReturnType<typeof createDb>,
|
||||
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 });
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
acknowledge: (deliveryId: string, canContinue: () => boolean) => Promise<unknown>,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
const repositoryId = sql<string | null>`${chatEndpointResources.metadata}->>'providerRepositoryId'`;
|
||||
const number = sql<string | null>`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<string | null>`(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<FastResponse | null>`(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<ReturnType<typeof page>>[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);
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -116,6 +116,7 @@ export async function withGitHubPublicationLease<T>(
|
||||
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)));
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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<void>) {
|
||||
async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise<void>, 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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) =>
|
||||
registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun),
|
||||
};
|
||||
}
|
||||
@@ -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 }));
|
||||
}
|
||||
@@ -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 });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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({
|
||||
|
||||
Reference in new issue
Block a user