mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
fix(server): wake GitHub terminal responses from committed changes
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
272afe9a56
commit
3d7c79c262
15 files changed
+428
-6
No files matched your search
@@ -1937,3 +1937,13 @@ removal wait for the three identified direct issue writers and remaining audit
|
||||
and acceptance requirements in the terminal-response audit. No slice-35 PR or
|
||||
full-stack readiness is claimed yet. Earlier parent CI/review findings, all other
|
||||
inventory duties, and reconciliation with current master remain required.
|
||||
|
||||
Slice 35 now covers the reproduced direct tree-restoration, workspace-admission
|
||||
and exhausted-recovery producers. Real publication-owner tests preserve both
|
||||
known-unsent cancellation and ambiguous outcome parking. Startup registers the
|
||||
new terminal worker and recurring GitHub maintenance no longer invokes that duty.
|
||||
Fourteen new worker cases plus one existing case pass; workspace types/build and
|
||||
migration application/replay/safety pass. The partial candidate index excludes
|
||||
10,000 completed rows in a rollback-only query-plan experiment. Fresh full
|
||||
chat/publication/startup acceptance is running; remote gates remain required.
|
||||
See the terminal-response audit for all retained failures and scoped evidence.
|
||||
@@ -139,3 +139,84 @@ not issue status, and do not independently change this response predicate.
|
||||
Additional dynamic status writers, lookup/index cost, uncertain-publication
|
||||
settlement, page fairness, final startup wiring, full acceptance and remote review
|
||||
still need completion. Do not present this checkpoint as polling removal.
|
||||
|
||||
## Production registration and direct-writer repairs
|
||||
|
||||
The three direct writers above now notify response availability before their
|
||||
writes in the existing transactions. Three real-producer regressions reproduce
|
||||
missing paused response settlement before the change. Tree cancellation continues
|
||||
to use issueService; restoration now notifies each restored issue. Queue admission
|
||||
notifies only a transition into blocked. Exhausted recovery does likewise and
|
||||
retains its existing lock order and failure behavior.
|
||||
|
||||
`/tmp/sleep35-direct-producers-before.log` records the missing tree/recovery wakes.
|
||||
Its workspace case initially failed earlier because the fixture had not enabled
|
||||
isolated workspaces, so it exercised normal admission instead of the intended
|
||||
pre-dispatch refusal. The corrected fixture enables and restores the feature,
|
||||
drains any executions on cleanup, and reproduces the missing workspace wake in
|
||||
`/tmp/sleep35-workspace-producer-before.log`. The real issue state reaches blocked
|
||||
before the missing-notice assertion, so the red result establishes the producer
|
||||
boundary rather than an unrelated setup failure. All eleven producer/lifecycle
|
||||
cases pass after repair in `/tmp/sleep35-direct-producers-fixed.log`.
|
||||
|
||||
The remaining inspected direct status writes in agent-conversations, Slack
|
||||
lifecycle/state, access reassignment, checkout/release, and native workspace
|
||||
reconstruction move away from blocked or remove assignment. They do not create
|
||||
a new paused-response obligation. Execution eligibility is reread before any
|
||||
send, so an outstanding lease-delayed candidate observes these changes. New
|
||||
run/lineage changes still go through the milestone boundary. Generic issue-service
|
||||
assignment/status mutations notify directly. Agent deletion does not restore an
|
||||
owed response. Goal status fields are separate session state.
|
||||
|
||||
Two further cases drive the real review-publication owner through token acquisition,
|
||||
a terminal run transition and pre-send authority denial. A known-unsent publication
|
||||
cancels and wakes terminal fallback through lease release. A saved ambiguous write
|
||||
remains parked, with no repeated SQL over 120 seconds. The latter does not assert
|
||||
that the provider write failed or relax host safety. Both pass in
|
||||
`/tmp/sleep35-publication-settlement.log`. The existing eight-attempt acknowledgement
|
||||
policy remains the owner of actual failed fallback sends.
|
||||
|
||||
The app now registers terminal-response work with warm-standby and idle-drain
|
||||
gates, and recurring GitHub maintenance no longer calls terminal reconciliation.
|
||||
The explicit/manual combined recovery entry point is preserved. Review-check
|
||||
publication and stale-head review repair remain separate duties in maintenance.
|
||||
|
||||
## Candidate lookup and final acceptance
|
||||
|
||||
Generated migration `0391_jittery_norrin_radd.sql` adds a partial index for
|
||||
processed, nonfinal response actions. It runs concurrently through the existing
|
||||
top-level migration registry and is idempotent on replay. Snapshot pruning retains
|
||||
0387 through 0391, exactly five. Application, replay, numbering/safety checks and
|
||||
PostgreSQL index validity pass. No table or queue storage was added.
|
||||
|
||||
The worker pages candidates by ID, then asks the GitHub owner to resolve actual
|
||||
run provenance and reply state. This has per-candidate read cost at startup or
|
||||
an affected scope change; it does not scan while settled. Issue/delivery/repository
|
||||
hints narrow normal wakes. A 101-row prefix of unrelated retained requests cannot
|
||||
hide the owed terminal response and does not publish those unrelated requests.
|
||||
The shared helper provides fair 25-entry dispatch batches after reconstruction.
|
||||
|
||||
A rollback-only experiment inserted 10,000 completed actions. The candidate index
|
||||
query read four eligible rows using an index-only scan and four shared buffers.
|
||||
See `/tmp/sleep35-index-plan.sql` and `/tmp/sleep35-index-plan.log`. This proves
|
||||
completed-history exclusion for the candidate access path, not whole-worker latency
|
||||
or fleet savings. The lifecycle tests prove settled SQL quiet separately.
|
||||
|
||||
Verification:
|
||||
|
||||
- The earlier extracted-owner GitHub acceptance passed 194 cases in 140.71 seconds
|
||||
(`/tmp/sleep35-github-acceptance.log`) before the three direct-writer repairs.
|
||||
- Final focused selection passes 15 cases: fourteen new terminal-worker cases
|
||||
plus one existing matching case (`/tmp/sleep35-wired-focused.log`, 46.15 seconds).
|
||||
The startup source suite did not match this focused selector and is included
|
||||
in the full run below; it is not claimed passed by that focused command.
|
||||
- Workspace typecheck and build pass after final source/wiring/index changes
|
||||
(`/tmp/sleep35-workspace-types.log`, `/tmp/sleep35-workspace-build.log`).
|
||||
- Fresh complete chat/publication/startup acceptance is running on
|
||||
`paperclip_sleep35_full_20261025_test`, log `/tmp/sleep35-full-chat.log`.
|
||||
Its expected count is 1,571. Do not claim completion before its terminal result.
|
||||
- Remote checks/review and any resulting repairs remain required.
|
||||
|
||||
Independent SQL writers and multiple replicas gain no new wake protocol. Cloud
|
||||
alarm coverage and sleep admission are unchanged. The full recurring-server stack,
|
||||
older failed gates/review findings and final master reconciliation remain open.
|
||||
@@ -516,6 +516,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
|
||||
"0388_lying_trish_tilby.sql",
|
||||
"0389_workable_spencer_smythe.sql",
|
||||
"0390_long_cargill.sql",
|
||||
"0391_jittery_norrin_radd.sql",
|
||||
"0380_funny_adam_warlock.sql",
|
||||
]);
|
||||
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_terminal_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_response_comment'
|
||||
and "chat_actions"."status" = 'processed' and "chat_actions"."delivery_id" is not null
|
||||
and coalesce("chat_actions"."result"->>'final', 'false') <> 'true';
|
||||
+174
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"id": "02c5ad8a-cdb9-41ad-acf1-5ab4f7d5bd67",
|
||||
"prevId": "b8b61c60-3d1d-4f3d-9df3-b633a3244150",
|
||||
"id": "5f2e1d87-0147-430a-9bdf-a8df947725f1",
|
||||
"prevId": "887083ec-214e-44ab-a1ab-25bff321118e",
|
||||
"version": "7",
|
||||
"dialect": "postgresql",
|
||||
"tables": {
|
||||
@@ -9616,6 +9616,38 @@
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"chat_actions_github_terminal_work_idx": {
|
||||
"name": "chat_actions_github_terminal_work_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "id",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"status\" = 'processed' and \"chat_actions\".\"delivery_id\" is not null\n and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'",
|
||||
"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": {
|
||||
@@ -12043,6 +12075,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": [
|
||||
@@ -14160,6 +14226,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",
|
||||
@@ -14182,6 +14266,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": [
|
||||
@@ -14934,6 +15034,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": {
|
||||
@@ -30404,6 +30526,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": [
|
||||
@@ -41917,6 +42067,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
@@ -2724,6 +2724,13 @@
|
||||
"when": 1792962182535,
|
||||
"tag": "0390_long_cargill",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 391,
|
||||
"version": "7",
|
||||
"when": 1792965069242,
|
||||
"tag": "0391_jittery_norrin_radd",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -679,6 +679,9 @@ export const chatActions = pgTable(
|
||||
table.endpointId,
|
||||
table.providerActionId,
|
||||
),
|
||||
index("chat_actions_github_terminal_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment'
|
||||
and ${table.status} = 'processed' and ${table.deliveryId} is not null
|
||||
and coalesce(${table.result}->>'final', 'false') <> 'true'`),
|
||||
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'))`),
|
||||
|
||||
@@ -97,7 +97,8 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
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(github).not.toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()");
|
||||
expect(source).toContain("chatChannels.registerGitHubTerminalResponseWork(deliveryWork,");
|
||||
expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,");
|
||||
expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,");
|
||||
});
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js";
|
||||
import { environmentService } from "../services/environments.js";
|
||||
import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
|
||||
import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js";
|
||||
@@ -3714,6 +3715,141 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
} finally { release(); await lease; await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("finds owed terminal GitHub work beyond a full page of unrelated request history", async () => {
|
||||
const f = await toolReplyFixture("github:paperclipai/paperclip:418");
|
||||
const [original] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
|
||||
const [delivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, original.deliveryId!));
|
||||
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
const history = Array.from({ length: 101 }, (_, index) => ({
|
||||
actionId: `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`,
|
||||
deliveryId: randomUUID(), commentId: randomUUID(), providerId: `retained-request-${index}`,
|
||||
}));
|
||||
await db.insert(issueComments).values(history.map(row => ({ id: row.commentId, companyId: f.companyId,
|
||||
issueId: f.conversation.issueId!, authorUserId: "owner-user", body: "A separate retained request" })));
|
||||
await db.insert(chatDeliveries).values(history.map(row => ({ ...delivery, id: row.deliveryId,
|
||||
providerEventId: row.providerId, deduplicationKey: row.providerId, state: "processed" as const })));
|
||||
await db.insert(chatMessageLinks).values(history.map(row => ({ companyId: f.companyId, endpointId: f.endpoint.id,
|
||||
conversationId: f.conversation.id, deliveryId: row.deliveryId, commentId: row.commentId,
|
||||
providerMessageId: row.providerId, direction: "inbound" })));
|
||||
await db.insert(chatActions).values(history.map(row => ({ ...original, id: row.actionId, deliveryId: row.deliveryId,
|
||||
providerActionId: `github_response:${row.deliveryId}`, result: { code: "fast_response_superseded" } })));
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
try {
|
||||
const baseline = f.responseWrites.length;
|
||||
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
|
||||
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
|
||||
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
|
||||
expect(f.responseWrites).toHaveLength(baseline + 1);
|
||||
expect(await db.select().from(chatActions).where(inArray(chatActions.id, history.map(row => row.actionId))))
|
||||
.toEqual(expect.arrayContaining(history.map(row => expect.objectContaining({ id: row.actionId,
|
||||
status: "processed", result: { code: "fast_response_superseded" } }))));
|
||||
expect(errors).toEqual([]);
|
||||
} finally {
|
||||
await coordinator.stop();
|
||||
await db.delete(chatActions).where(inArray(chatActions.id, history.map(row => row.actionId)));
|
||||
}
|
||||
});
|
||||
|
||||
it.each([false, true])("parks terminal GitHub responses behind the publication owner (uncertain: %s)", async uncertain => {
|
||||
const f = await toolReplyFixture();
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
let release!: () => void, entered!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(resolve => { entered = resolve; });
|
||||
let intercepted = false;
|
||||
let sending: Promise<unknown> | undefined;
|
||||
const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"))))[0];
|
||||
try {
|
||||
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
|
||||
f.setSupplementalProviderFetch(async input => {
|
||||
if (intercepted || !String(input).endsWith("/access_tokens")) return undefined;
|
||||
const publication = await action();
|
||||
if (!publication) return undefined;
|
||||
intercepted = true;
|
||||
if (uncertain) await db.update(chatActions).set({ result: { replyWriteStarted: true } }).where(eq(chatActions.id, publication.id));
|
||||
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await issueService(db).update(f.conversation.issueId!, { status: "blocked" });
|
||||
entered(); await hold;
|
||||
return undefined;
|
||||
});
|
||||
const baseline = f.responseWrites.length;
|
||||
sending = githubChatReviewService(db, f.providerFetch).execute(f.session, "comment", { body: "Authored final response", idempotencyKey: "terminal-owner-race" });
|
||||
await ready;
|
||||
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
|
||||
expect(f.responseWrites).toHaveLength(baseline);
|
||||
release(); await sending;
|
||||
expect((await action()).status).toBe("cancelled");
|
||||
if (!uncertain) await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
|
||||
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
|
||||
if (uncertain) expect(f.responseWrites).toHaveLength(baseline);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
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 { release(); await sending; await coordinator.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it.each(["tree restore", "workspace admission", "exhausted recovery"] as const)("wakes a paused GitHub response from %s", async producer => {
|
||||
const f = await toolReplyFixture("github:paperclipai/paperclip:418");
|
||||
const issueId = f.conversation.issueId!;
|
||||
await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await db.update(chatActions).set({ result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || '{"final":true}'::jsonb` })
|
||||
.where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
|
||||
const thread = makeThread({ channelId: "paperclipai/paperclip", id: f.conversation.externalThreadId }).thread;
|
||||
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
|
||||
trigger: "mention", message: makeMessage({ id: "41802", text: "New request", userId: "42", userName: "octocat", mentioned: true }) });
|
||||
const [link] = await db.select().from(chatMessageLinks).where(and(eq(chatMessageLinks.endpointId, f.endpoint.id), eq(chatMessageLinks.providerMessageId, "41802")));
|
||||
const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.deliveryId, link.deliveryId!), eq(chatActions.kind, "github_response_comment"))))[0];
|
||||
let mutate: () => Promise<unknown>;
|
||||
if (producer === "tree restore") {
|
||||
await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issueId));
|
||||
const tree = issueTreeControlService(db);
|
||||
const actor = { actorType: "user" as const, actorId: "owner-user", userId: "owner-user" };
|
||||
const cancelled = await tree.createHold(f.companyId, issueId, { mode: "cancel", actor });
|
||||
await tree.cancelIssueStatusesForHold(f.companyId, issueId, cancelled.hold.id);
|
||||
const restore = await tree.createHold(f.companyId, issueId, { mode: "restore", actor });
|
||||
mutate = () => tree.restoreIssueStatusesForHold(f.companyId, issueId, restore.hold.id, { actor });
|
||||
} else if (producer === "workspace admission") {
|
||||
const settings = instanceSettingsService(db);
|
||||
const previous = await settings.getExperimental();
|
||||
await settings.updateExperimental({ enableIsolatedWorkspaces: true });
|
||||
onTestFinished(() => settings.updateExperimental({ enableIsolatedWorkspaces: previous.enableIsolatedWorkspaces }));
|
||||
const heartbeat = heartbeatService(db);
|
||||
onTestFinished(() => heartbeat.drainActiveRunExecutions());
|
||||
await db.update(issues).set({ status: "todo", workMode: "standard", responsibleUserId: "owner-user",
|
||||
projectId: null, projectWorkspaceId: null, executionWorkspaceId: null, executionRunId: null, checkoutRunId: null,
|
||||
executionWorkspaceSettings: { mode: "isolated_workspace", workspaceStrategy: { type: "git_worktree" } },
|
||||
}).where(eq(issues.id, issueId));
|
||||
await db.update(agents).set({ runtimeConfig: { heartbeat: { wakeOnDemand: true } } }).where(eq(agents.id, f.assignedAgentId));
|
||||
mutate = () => heartbeat.wakeup(f.assignedAgentId, { source: "assignment", triggerDetail: "system",
|
||||
reason: "issue_assigned", payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" } });
|
||||
} else {
|
||||
await db.update(issues).set({ status: "in_progress", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId));
|
||||
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: issueId,
|
||||
kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: f.assignedAgentId,
|
||||
cause: "uncertain_external_action", status: "active", evidence: { runId: f.run.id }, fingerprint: randomUUID(),
|
||||
nextAction: "Preserve the recorded outcome without replay.",
|
||||
});
|
||||
mutate = () => settleUnrecoverableExecutions(db);
|
||||
}
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
try {
|
||||
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
|
||||
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
|
||||
expect((await action()).result?.pausedVersion).toBeUndefined();
|
||||
await mutate();
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].status).toBe("blocked");
|
||||
await expect.poll(async () => (await action()).result?.pausedVersion, { timeout: 5000 }).toEqual(expect.any(String));
|
||||
expect([...f.responseComments.values()][0].body).toContain("paused");
|
||||
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it.each(["release", "restart", "drain", "standby"] as const)("wakes GitHub acknowledgements from a real publication lease %s", async mode => {
|
||||
const f = await reviewBotFixture();
|
||||
const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id));
|
||||
|
||||
@@ -1289,6 +1289,7 @@ export async function createApp(
|
||||
chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerGitHubTerminalResponseWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
|
||||
@@ -3507,6 +3507,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
|
||||
return {
|
||||
registerInteractionContinuationWork,
|
||||
registerGitHubTerminalResponseWork: githubResponseCommentService(db, fetchImpl).registerTerminalResponseWork,
|
||||
registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork,
|
||||
registerTerminalConfirmationWork,
|
||||
loadFailedRetryNoticeScopes,
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { githubResponseCommentService } from "../../chat-github-response-comments.js";
|
||||
import { githubReviewCheckService } from "../../chat-github-checks.js";
|
||||
|
||||
export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fetch, options: {
|
||||
@@ -13,7 +12,6 @@ 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).processTerminalResponses();
|
||||
await githubReviewCheckService(db, fetchImpl).processPending();
|
||||
} finally {
|
||||
githubMaintenancePending = false;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { notifyExecutionRunWork } from "./execution-control-notifications.js";
|
||||
import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
|
||||
@@ -543,6 +544,7 @@ export async function settleUnrecoverableExecutions(
|
||||
let nativeFailureBlock = action.evidence.nativeFailureBlock;
|
||||
let nativeBootstrapFailureBlock = action.evidence.nativeBootstrapFailureBlock;
|
||||
if (current) {
|
||||
if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id });
|
||||
const [projected] = await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyChatResponseAvailability } from "../chat-response-availability.js";
|
||||
import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
|
||||
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
|
||||
import { authorizationService } from "../authorization.js";
|
||||
@@ -3815,6 +3816,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
`- Reason: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_MESSAGE}`,
|
||||
`- Next action: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION}`,
|
||||
].join("\n");
|
||||
if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id });
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
|
||||
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
|
||||
import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -979,7 +980,10 @@ export function issueTreeControlService(db: Db) {
|
||||
const restored: TreeStatusUpdateResult["updatedIssues"] = [];
|
||||
for (const [status, issueIdsForStatus] of issueIdsByStatus) {
|
||||
if (issueIdsForStatus.length === 0) continue;
|
||||
for (const issueId of issueIdsForStatus) await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`);
|
||||
for (const issueId of issueIdsForStatus) {
|
||||
await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`);
|
||||
await notifyChatResponseAvailability(tx, { companyId, issueId });
|
||||
}
|
||||
const rows = await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
|
||||
Reference in new issue
Block a user