mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(server): schedule governed review response recovery (#17702)
## Thinking Path > - Paperclip manages agents, tasks and their external chat responses. > - A completed chat turn can owe a response while an earlier review gate remains pending. > - Periodic recovery revisits that response history even when no work can advance. > - This change uses the existing delivery coordinator and saved retry deadlines. > - Approval, interaction and authority changes wake the affected issue or run after commit. > - The existing presentation owner verifies the durable result and current authority. > - Settled and dormant history becomes quiet without losing future eligible responses. ## Linked Issues or Issue Description Refs #17696 (merged) and #15625. This is slice 49, now based on master. It owns governed-review response presentation. Accepted responses and pending decision effects keep their existing owners. **What happened?** Native finalization recovery repeatedly inspects committed review responses. Removing the scan without covering approval, interaction, task-policy and authority changes would strand responses that become eligible later. **Expected behavior** Recover an owed response once from existing durable facts. Enforce authority at publication, wake after relevant outer transactions commit, retain real retry deadlines and restart/drain recovery, and stop recurring SQL when no response can advance. **Steps to reproduce** Complete an external chat turn while an earlier completion review remains pending. Block its response behind an approval or interaction, then resolve that blocker. The response must appear once without a periodic scan or resolving the original review gate. ## What Changed - Add a scoped governed-review worker using the existing coordinator, finalizations, decisions and comment receipts. No queue table or transaction wrapper is added. - Keep recovery and publication policy in the native finalizer. Preserve binding, source/session authority, ownership, deadline and deleted-comment fences; match coordinator-before-issue lock ordering. - Notify through actual approval, interaction, OAuth, task-policy and lifecycle producers before their owned writes. Serialize approval resolution with task linking so a concurrent link retains its wake. Nested transactions still settle at outer commit. - Scope ordinary milestones, finalization and coalesced input to their issue/run. Use the explicit company authority boundary for genuine global restoration, including membership changes. Both response workers share this boundary. - Register under maintenance and drain gates, disable the two periodic repair entrances, and retain fresh finalization and explicit recovery. Add the concurrent 0413 comment-receipt index and migration registration; retain five snapshots. - Preserve the producer inventory, source qualifications and earlier failures in the owner audit. Retain a real Photon regression proving routine ownership assertions stay quiet and concurrent authority restoration wakes each worker once. ## Verification - Scoped repair source `2341714720b5c273f0ce011b48345af0ef39227f` passes **2053 distinct cases**: 1703 in 15 complete producer/authority/recovery files, including all 273 governed-response cases and 95 wake-queue adapter cases, plus 350 selected chat cases. Another 1463 chat cases are outside that selection. Thirteen fresh PostgreSQL 17 fixtures migrate through unchanged 0413; the one temporary queue-fixture import restores exactly. All sixteen captured hashes match. - Two actual repeated coarse-hint SQL regressions fail before repair and pass afterward. Actual membership restoration passes before and after. The same assertions and time limits prove dormant work stays quiet and real authority restoration still publishes once. Real coalesced input covers both queues after commit. - Combined source `586f5674cd9798b33cd22437255e0480b5726d20`, including independent master #18129, passes **1297 distinct cases**: 1093 in five complete server files, 185 in five runner files, 14 Node protocol cases and five UI registry cases. This includes the complete 22-case Photon file for both response workers. The external Photon transport is mocked; Paperclip callbacks, transaction fences and observed SQL are real. The earlier and combined case counts are separate, with no cumulative double-counting. - A normal merge retains both complete owned and independent master binary patches. All twelve new-context hashes and sixteen earlier repair hashes match. Three dedicated fresh 0413 fixtures plus the live-event helper's own fresh database use PostgreSQL 17 without source adaptations. Database build, server types/build, boundaries, full-SHA migration order and generated TS/Rust profiles pass. The new inherited Runner vendor inputs rebuild successfully. - Database bytes remain identical to the previously qualified 0413 source. The recorded populated 0412-to0413 upgrade retains all 411 prior receipts and 38 companies, appends one receipt and none on replay; the index is valid and ready. The earlier 2828-case acceptance, CLI journey, setup failures and their repairs retain their recorded source scope in `doc/plans/2026-10-26-governed-review-response-sleep-audit.md`. - Parent #17696 merged with 58 clean checks, Greptile 5/5 and Ponytail 5/5. Its Photon warning was disproved by real callback/SQL evidence and the security check became successful. This PR still requires its own fresh exact-head CI and review gates. No deployed sleep, external alarm or fleet savings claim is made. ## Risks - Notifications use the existing process-local writer contract. Independent SQL writers and uncoordinated replicas are outside its coverage. - Genuine startup and company authority reconstruction remain proportional to the owed backlog. Queries page 100 rows, but discovery accumulates the candidates; this is not a constant-cost global operation. - Current publication authorization, durable binding, ownership and lock ordering remain mandatory. The worker does not resolve the original review gate. - The concurrent index uses existing migration replay handling. Other recovery duties and Cloud sleep eligibility remain separate. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository tools, shell execution and PostgreSQL tests. The runtime does not expose a more specific serving snapshot or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
40 files changed
+1953
-145
No files matched your search
@@ -1,3 +1,6 @@
|
||||
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
|
||||
import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
|
||||
import { accessService } from "../access.js";
|
||||
import { initializeRunIdentity } from "../run-identity.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
@@ -28,6 +31,7 @@ import {
|
||||
completionContracts,
|
||||
closeRegisteredClients,
|
||||
createDb,
|
||||
subscribeDatabaseWork,
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
issueAttachments,
|
||||
@@ -41,6 +45,7 @@ import {
|
||||
statusDecisions,
|
||||
toolApplications,
|
||||
toolConnections,
|
||||
toolOauthStates,
|
||||
chatVoiceSessions, chatVoiceReplies, chatVoiceToolCalls,
|
||||
workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
@@ -51,7 +56,17 @@ import type {
|
||||
} from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { finalizeNativeRun } from "./native-run-finalizer.js";
|
||||
import { finalizeNativeRun, recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js";
|
||||
import { commitNativeStatusDecision } from "./status-decision-committer.js";
|
||||
import { configureAgentLifecycle, createAgentLifecycle, processAgentLifecycle } from "../agent-lifecycle.js";
|
||||
import { createOAuthStateStore } from "../tool-access/oauth/state.js";
|
||||
import { approvalRecords } from "../approvals.js";
|
||||
import { issueApprovalService } from "../issue-approvals.js";
|
||||
import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js";
|
||||
import { nativeReviewResponseWork } from "./native-review-response-work.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { idleWorkSnapshot } from "../task-admission.js";
|
||||
import { nativeCompletionFeedback } from "./native-completion-feedback.js";
|
||||
import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js";
|
||||
import * as chatWaitAuthority from "./chat-attachment-reuse.js";
|
||||
@@ -422,6 +437,10 @@ describe("native external-chat response wait", () => {
|
||||
userId,
|
||||
commentId,
|
||||
principalId,
|
||||
storeBinding: { companyId, issueId, runId, agentId, normalizedSessionId: sessionId,
|
||||
runnerSourceInstanceId: runnerInstanceId, completionContractId: contractId,
|
||||
completionContractSha256: contractSha256, completionContractRevision: "external-chat-wait-v1",
|
||||
completionContractCriterionIds: ["response"] },
|
||||
};
|
||||
}
|
||||
|
||||
@@ -2869,6 +2888,653 @@ describe("native external-chat response wait", () => {
|
||||
return run!.resultJson!;
|
||||
}
|
||||
|
||||
async function seedRecoverableReviewResponse() {
|
||||
const fixture = await seedWaitTurn();
|
||||
const gate = await seedPriorCompletionReview(fixture);
|
||||
await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, fixture.runId));
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(coordinator!.phase).toBe("committed");
|
||||
const input = { companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId,
|
||||
decisionId: coordinator!.decisionId!, resultId: coordinator!.resultId!, assessmentId: coordinator!.assessmentId!, onlyWhenDue: true };
|
||||
return { ...fixture, gate, input };
|
||||
}
|
||||
|
||||
it.each(["reconciler", "finalizer"] as const)("defers periodic governed-response %s repair to its registered worker", async path => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` })
|
||||
.where(eq(heartbeatRuns.id, fixture.runId));
|
||||
if (path === "reconciler") await reconcileNativeFinalizations(db, [fixture.runId], { skipNativeReviewResponse: true });
|
||||
else await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded",
|
||||
projectRunStatus: true, skipNativeReviewResponse: true });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined();
|
||||
const work = nativeReviewResponseWork(db);
|
||||
const [entry] = await work.load({ companyId: fixture.companyId, runId: fixture.runId });
|
||||
expect(entry?.id).toBe(fixture.runId);
|
||||
expect(await work.check(entry)).toBeNull();
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("recovers scheduled governed responses once under concurrent callers and preserves deleted receipts", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const work = nativeReviewResponseWork(db);
|
||||
const [entry] = await work.load({ companyId: fixture.companyId });
|
||||
expect(entry?.id).toBe(fixture.runId);
|
||||
expect(await Promise.all([recoverCommittedNativeReviewResponse(db, fixture.input),
|
||||
recoverCommittedNativeReviewResponse(db, fixture.input)])).toEqual(expect.arrayContaining([true, false]));
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
expect(comments).toHaveLength(1);
|
||||
expect(await db.select().from(chatPublications).where(eq(chatPublications.commentId, comments[0].id))).toHaveLength(1);
|
||||
expect(await work.load({ companyId: fixture.companyId })).toEqual([]);
|
||||
await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comments[0].id));
|
||||
expect(await work.load({ companyId: fixture.companyId })).toEqual([]);
|
||||
expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(false);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("lets status replay finish while governed response recovery waits for its coordinator", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, fixture.issueId));
|
||||
const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, fixture.input.decisionId));
|
||||
let recovery: Promise<{ value?: boolean; error?: unknown }> | undefined;
|
||||
try {
|
||||
await db.transaction(async tx => {
|
||||
await tx.select().from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId)).for("update");
|
||||
const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`);
|
||||
recovery = recoverCommittedNativeReviewResponse(db, fixture.input)
|
||||
.then(value => ({ value }), error => ({ error }));
|
||||
await vi.waitFor(async () => {
|
||||
const [waiting] = await db.execute<{ count: number }>(sql`
|
||||
select count(*)::int as count from pg_stat_activity
|
||||
where datname = current_database() and wait_event_type = 'Lock'
|
||||
and ${backend.pid} = any(pg_blocking_pids(pid))`);
|
||||
expect(waiting.count).toBe(1);
|
||||
});
|
||||
|
||||
|
||||
await tx.execute(sql`set local lock_timeout = '500ms'`);
|
||||
const replay = await commitNativeStatusDecision({
|
||||
db: tx as unknown as typeof db, companyId: fixture.companyId,
|
||||
issueId: fixture.issueId, runId: fixture.runId,
|
||||
assessmentId: fixture.input.assessmentId,
|
||||
priorStatus: issue.status, priorStatusVersion: issue.statusVersion,
|
||||
priorDecisionId: issue.lastStatusDecisionId,
|
||||
decision: decision.decisionJson as unknown as Parameters<typeof commitNativeStatusDecision>[0]["decision"],
|
||||
});
|
||||
expect(replay.replayed).toBe(true);
|
||||
});
|
||||
} finally {
|
||||
if (recovery) {
|
||||
const result = await recovery;
|
||||
expect(result.error).toBeUndefined();
|
||||
expect(result.value).toBe(true);
|
||||
}
|
||||
}
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1);
|
||||
expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(false);
|
||||
});
|
||||
|
||||
it.each(["future", "claim"])("defers scheduled governed responses behind a new %s under the owner lock", async kind => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const work = nativeReviewResponseWork(db);
|
||||
const [entry] = await work.load({ companyId: fixture.companyId });
|
||||
const at = new Date(Date.now() + 60_000);
|
||||
const transaction = db.transaction.bind(db);
|
||||
const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => {
|
||||
await transaction(async tx => {
|
||||
await tx.update(nativeRunFinalizations).set(kind === "future" ? { nextAttemptAt: at }
|
||||
: { leaseOwner: "another-finalizer", leaseExpiresAt: at }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
});
|
||||
return transaction(callback, options);
|
||||
}) as typeof db.transaction);
|
||||
try {
|
||||
expect(await work.check(entry)).toEqual(kind === "future" ? { ...entry, dueAt: at.getTime() } : null);
|
||||
} finally { intercept.mockRestore(); }
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true);
|
||||
});
|
||||
|
||||
it("propagates scheduled governed response contention and rolls back its proof restoration", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` })
|
||||
.where(eq(heartbeatRuns.id, fixture.runId));
|
||||
let release!: () => void, observed!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { observed = resolve; });
|
||||
const holder = db.transaction(async tx => {
|
||||
await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update");
|
||||
observed(); await hold;
|
||||
});
|
||||
await locked;
|
||||
try {
|
||||
await expect(recoverCommittedNativeReviewResponse(db, fixture.input)).rejects.toBeDefined();
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined();
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
} finally { release(); await holder; }
|
||||
expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true);
|
||||
});
|
||||
|
||||
it.each(["gate", "agent", "principal", "company"])("leaves scheduled governed response authority with its owner after %s changes", async kind => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
if (kind === "gate") await db.update(issueThreadInteractions).set({ status: "accepted" }).where(eq(issueThreadInteractions.id, fixture.gate.id));
|
||||
if (kind === "agent") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, fixture.agentId));
|
||||
if (kind === "principal") await db.update(companyMemberships).set({ status: "suspended" })
|
||||
.where(and(eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId)));
|
||||
expect(await recoverCommittedNativeReviewResponse(db, { ...fixture.input,
|
||||
...(kind === "company" ? { companyId: randomUUID() } : {}) })).toBe(false);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
});
|
||||
|
||||
it.each(["milestone", "finalization"] as const)("ignores repeated unscoped routine %s hints without reconstructing dormant governed responses", async source => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
await db.update(companyMemberships).set({ status: "suspended" }).where(and(
|
||||
eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId)));
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 10_000 })
|
||||
.toEqual({ active: 0, next: null });
|
||||
const debug = db.$client.options.debug;
|
||||
const queries: string[] = [];
|
||||
db.$client.options.debug = (_connection, query) => {
|
||||
if (query.startsWith("select") && query.includes("native_run_finalizations")) queries.push(query);
|
||||
};
|
||||
try {
|
||||
for (let i = 0; i < 3; i++) {
|
||||
await db.transaction(tx => source === "milestone" ? notifyChatRunMilestones(tx, { companyId: fixture.companyId })
|
||||
: notifyNativeFinalizationWork(tx, { companyId: fixture.companyId }));
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 })
|
||||
.toEqual({ active: 0, next: null });
|
||||
}
|
||||
expect(queries).toEqual([]);
|
||||
} finally { db.$client.options.debug = debug; }
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("wakes dormant governed responses through the real company membership restoration owner", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const [member] = await db.select().from(companyMemberships).where(and(
|
||||
eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId)));
|
||||
await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.id, member.id));
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 10_000 })
|
||||
.toEqual({ active: 0, next: null });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
await accessService(db).updateMember(fixture.companyId, member.id, { status: "active" });
|
||||
await expect.poll(async () => ({ count: (await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).length,
|
||||
active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 })
|
||||
.toEqual({ count: 1, active: 0, next: null });
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it.each(["approve", "reject", "cancel", "unlink"] as const)("reactivates dormant governed responses after approval %s", async action => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId,
|
||||
type: "hire_agent", status: "pending", payload: {} }).returning();
|
||||
await issueApprovalService(db).link(fixture.issueId, approval.id);
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
if (action === "unlink") await issueApprovalService(db).unlink(fixture.issueId, approval.id);
|
||||
else if (action === "cancel") await approvalRecords(db).cancel(approval.id);
|
||||
else await approvalRecords(db).resolveApproval(approval.id, action === "approve" ? "approved" : "rejected", fixture.userId, null);
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it.each([
|
||||
["resolver", "single"], ["resolver", "bulk"],
|
||||
["linker", "single"], ["linker", "bulk"],
|
||||
] as const)("retains a settled task wake when %s starts first during concurrent %s approval linking", async (first, kind) => {
|
||||
const fixture = await seedWaitTurn();
|
||||
const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId,
|
||||
type: "request_board_approval", status: "pending", payload: {} }).returning();
|
||||
const observed: string[] = [];
|
||||
const reads: Promise<void>[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.nativeReviewResponse, (event, hints) => {
|
||||
if (event !== "settled" || ![...(hints ?? [])].some(hint => {
|
||||
const scope = JSON.parse(hint);
|
||||
return scope.companyId === fixture.companyId && scope.issueId === fixture.issueId;
|
||||
})) return;
|
||||
reads.push(db.select({ status: approvals.status }).from(issueApprovals)
|
||||
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
|
||||
.where(and(eq(issueApprovals.issueId, fixture.issueId), eq(approvals.id, approval.id)))
|
||||
.then(rows => { for (const row of rows) observed.push(row.status); }));
|
||||
});
|
||||
let release!: () => void, ready!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const held = new Promise<void>(resolve => { ready = resolve; });
|
||||
let holderPid = 0, resolverPid = 0, linkerPid = 0;
|
||||
let linked = false, resolved = false;
|
||||
const link = async (tx: typeof db) => {
|
||||
if (kind === "single") await issueApprovalService(tx).link(fixture.issueId, approval.id);
|
||||
else await issueApprovalService(tx).linkManyForApproval(approval.id, [fixture.issueId]);
|
||||
};
|
||||
const holder = db.transaction(async tx => {
|
||||
const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`);
|
||||
holderPid = Number(backend.pid);
|
||||
if (first === "resolver") {
|
||||
|
||||
|
||||
await tx.execute(sql`lock table approvals in share mode`);
|
||||
} else {
|
||||
await link(tx as unknown as typeof db);
|
||||
}
|
||||
ready();
|
||||
await hold;
|
||||
});
|
||||
let resolver: Promise<unknown> | undefined, linker: Promise<unknown> | undefined;
|
||||
const blocked = async (pid: number, by: number) => {
|
||||
if (!pid || !by) return false;
|
||||
const [row] = await db.execute(sql`select ${by} = any(pg_blocking_pids(${pid})) as blocked`);
|
||||
return row.blocked === true;
|
||||
};
|
||||
try {
|
||||
await held;
|
||||
resolver = db.transaction(async tx => {
|
||||
const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`);
|
||||
resolverPid = Number(backend.pid);
|
||||
await approvalRecords(tx as unknown as typeof db).resolveApproval(approval.id, "approved", fixture.userId, null);
|
||||
}).then(() => { resolved = true; });
|
||||
await expect.poll(async () => resolved || await blocked(resolverPid, holderPid), { timeout: 5000 }).toBe(true);
|
||||
if (first === "resolver") {
|
||||
linker = db.transaction(async tx => {
|
||||
const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`);
|
||||
linkerPid = Number(backend.pid);
|
||||
await link(tx as unknown as typeof db);
|
||||
}).then(() => { linked = true; });
|
||||
await expect.poll(async () => linked || await blocked(linkerPid, resolverPid), { timeout: 5000 }).toBe(true);
|
||||
}
|
||||
|
||||
|
||||
await Promise.all(reads);
|
||||
release();
|
||||
await Promise.all([holder, resolver, linker]);
|
||||
await Promise.all(reads);
|
||||
expect(observed).toContain("approved");
|
||||
} finally {
|
||||
release();
|
||||
await Promise.allSettled([holder, resolver, linker]);
|
||||
await Promise.all(reads);
|
||||
unsubscribe();
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["cancel", "unlink"] as const)("keeps governed response approval %s behind outer commit and scoped to linked tasks", async action => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const [approval, unrelated] = await db.insert(approvals).values([
|
||||
{ companyId: fixture.companyId, type: "hire_agent", status: "pending", payload: {} },
|
||||
{ companyId: fixture.companyId, type: "hire_agent", status: "pending", payload: {} },
|
||||
]).returning();
|
||||
await issueApprovalService(db).link(fixture.issueId, approval.id);
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
const mutate = async (tx: typeof db) => {
|
||||
if (action === "cancel") await approvalRecords(tx).cancel(approval.id);
|
||||
else await issueApprovalService(tx).unlink(fixture.issueId, approval.id);
|
||||
};
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
const previousDebug = db.$client.options.debug;
|
||||
const queries: string[] = [];
|
||||
db.$client.options.debug = (_connection, query) => { if (query.includes('"native_run_finalizations"')) queries.push(query); };
|
||||
try {
|
||||
await approvalRecords(db).cancel(unrelated.id);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(queries).toEqual([]);
|
||||
} finally { db.$client.options.debug = previousDebug; }
|
||||
const rollback = new Error("rollback governed approval change");
|
||||
await expect(db.transaction(async tx => {
|
||||
await mutate(tx as unknown as typeof db);
|
||||
expect(await comments()).toHaveLength(0);
|
||||
throw rollback;
|
||||
})).rejects.toBe(rollback);
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
await db.transaction(async tx => {
|
||||
await mutate(tx as unknown as typeof db);
|
||||
expect(await comments()).toHaveLength(0);
|
||||
});
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("wakes dormant governed response recovery after real agent resume", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
await db.update(agents).set({ status: "paused", lifecycleState: "paused", lifecycleHolds: ["manual"], lifecycleOperation: null })
|
||||
.where(eq(agents.id, fixture.agentId));
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const lifecycle = configureAgentLifecycle(db, {
|
||||
requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete",
|
||||
});
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
await createAgentLifecycle(db).resumeAgent(fixture.agentId);
|
||||
await processAgentLifecycle(db, fixture.agentId);
|
||||
await expect.poll(async () => (await db.select().from(agents).where(eq(agents.id, fixture.agentId)))[0].status).toBe("idle");
|
||||
await expect.poll(async () => (await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).length,
|
||||
{ timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await lifecycle.stop(); await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it.each(["skip", "withdraw", "reject_suggestions", "complete_verdicts", "decline_connection"] as const)(
|
||||
"reactivates governed responses after competing interaction %s", async action => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const service = issueThreadInteractionService(db);
|
||||
const issue = { id: fixture.issueId, companyId: fixture.companyId };
|
||||
const actor = { userId: fixture.userId };
|
||||
const competing = action === "decline_connection"
|
||||
? await service.createConnectionIntent(issue, { sourceRunId: fixture.runId,
|
||||
addresseeUserId: fixture.userId, idempotencyKey: `review-competing:${randomUUID()}`,
|
||||
payload: { version: 1, serviceSlug: "notion", serviceName: "Notion", serviceLogoUrl: null,
|
||||
requestingAgentId: fixture.agentId, requestingAgentName: "Agent", phase: "requested" } })
|
||||
: action === "complete_verdicts"
|
||||
? await service.create(issue, { kind: "request_item_verdicts", payload: { version: 1,
|
||||
prompt: "Review this item", items: [{ id: "one", label: "One" }] } }, actor)
|
||||
: await service.create(issue, { kind: "suggest_tasks", payload: { version: 1,
|
||||
tasks: [{ clientKey: "child", title: "A suggested task" }] } }, actor);
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
if (action === "skip") await service.skipInteraction(issue, competing.id, {}, actor);
|
||||
else if (action === "withdraw") await service.withdrawInteraction(issue, competing.id, {}, actor);
|
||||
else if (action === "reject_suggestions") await service.rejectInteraction(issue, competing.id, {}, actor);
|
||||
else if (action === "complete_verdicts") await service.submitItemVerdicts(issue, competing.id,
|
||||
{ verdicts: [{ id: "one", verdict: "approve" }] }, actor);
|
||||
else await service.resolveConnectionIntent(issue, competing.id, { version: 1, outcome: "declined" }, actor);
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it.each(["status", "assignment"] as const)("preserves governed response policy through task %s changes", async change => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
await db.update(issues).set(change === "status" ? { status: "blocked" } : { assigneeAgentId: null })
|
||||
.where(eq(issues.id, fixture.issueId));
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
if (change === "assignment") {
|
||||
await expect(issueService(db).update(fixture.issueId, { assigneeAgentId: fixture.agentId }))
|
||||
.rejects.toThrow("Agent assignment cannot change while this task is bound to an external channel");
|
||||
expect(await comments()).toHaveLength(0);
|
||||
} else {
|
||||
await issueService(db).update(fixture.issueId, { status: "in_review" });
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
}
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("reactivates governed responses after OAuth confirmation denial", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.companyId, fixture.companyId));
|
||||
const competing = await issueThreadInteractionService(db).create({ id: fixture.issueId, companyId: fixture.companyId },
|
||||
{ kind: "request_confirmation", payload: { version: 1, prompt: "Connect your account", target: {
|
||||
type: "custom", key: `connection:${connection.id}`, revisionId: randomUUID(), label: "Connect account" } } },
|
||||
{ userId: fixture.userId });
|
||||
const [state] = await db.insert(toolOauthStates).values({ state: randomUUID(), companyId: fixture.companyId,
|
||||
connectionId: connection.id, codeVerifier: "test-verifier", issueId: fixture.issueId, interactionId: competing.id,
|
||||
subjectUserId: fixture.userId, expiresAt: new Date(Date.now() + 60_000) }).returning();
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
const rollback = new Error("rollback OAuth denial");
|
||||
await expect(db.transaction(async tx => {
|
||||
await createOAuthStateStore(tx as unknown as typeof db).rejectPendingOAuthInteraction(state,
|
||||
{ actorType: "user", actorId: fixture.userId });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
throw rollback;
|
||||
})).rejects.toBe(rollback);
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, competing.id)))[0].status).toBe("pending");
|
||||
await db.transaction(async tx => {
|
||||
await createOAuthStateStore(tx as unknown as typeof db).rejectPendingOAuthInteraction(state,
|
||||
{ actorType: "user", actorId: fixture.userId });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
});
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("reactivates governed responses when pending execution policy is cleared without changing task status", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
await db.update(issues).set({ executionState: { status: "pending", currentStageId: randomUUID(),
|
||||
currentStageIndex: 0, currentStageType: "review", currentParticipant: { type: "user", userId: fixture.userId, agentId: null },
|
||||
returnAssignee: { type: "agent", agentId: fixture.agentId, userId: null }, reviewRequest: null,
|
||||
completedStageIds: [], lastDecisionId: null, lastDecisionOutcome: null, monitor: null, changesRequestedCount: 0 } })
|
||||
.where(eq(issues.id, fixture.issueId));
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
await issueService(db).updateForCompany(fixture.issueId, fixture.companyId, { executionState: null });
|
||||
expect((await db.select().from(issues).where(eq(issues.id, fixture.issueId)))[0].status).toBe("in_review");
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("wakes governed responses from the real coordinator retry release", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const store = new NativeRunCoordinatorStore(db, fixture.storeBinding);
|
||||
await store.claimFinalization({ leaseOwner: "review-response-fixture", leaseTtlMs: 30_000 });
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
await store.markFinalizationRetry({ leaseOwner: "review-response-fixture", failureCode: "review_response_retry", retryAfterMs: 1000 });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1);
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("retains the first governed-response wake through idle drain without passive SQL", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId,
|
||||
type: "hire_agent", status: "pending", payload: {} }).returning();
|
||||
await issueApprovalService(db).link(fixture.issueId, approval.id);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
let canRun = true;
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => canRun);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
canRun = false;
|
||||
await approvalRecords(db).cancel(approval.id);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(coordinator.nextWakeAt()).not.toBeNull();
|
||||
const previousDebug = db.$client.options.debug;
|
||||
const queries: string[] = [];
|
||||
db.$client.options.debug = (_connection, query) => { queries.push(query); };
|
||||
try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); }
|
||||
finally { db.$client.options.debug = previousDebug; }
|
||||
canRun = true; worker.wake();
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await comments()).toHaveLength(1);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("retains failed governed-response recovery across drain and retries after resume", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
let release!: () => void, observed!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const locked = new Promise<void>(resolve => { observed = resolve; });
|
||||
const holder = db.transaction(async tx => {
|
||||
await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update");
|
||||
observed(); await hold;
|
||||
});
|
||||
await locked;
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
let canRun = true;
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => canRun);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ errors: errors.length, active: idleWorkSnapshot().active }),
|
||||
{ timeout: 10_000 }).toEqual({ errors: 1, active: 0 });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now());
|
||||
canRun = false;
|
||||
worker.wake();
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
|
||||
const previousDebug = db.$client.options.debug;
|
||||
const queries: string[] = [];
|
||||
db.$client.options.debug = (_connection, query) => { queries.push(query); };
|
||||
try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); }
|
||||
finally { db.$client.options.debug = previousDebug; }
|
||||
release(); await holder;
|
||||
canRun = true; worker.wake();
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await comments()).toHaveLength(1);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
expect(errors).toHaveLength(1);
|
||||
} finally { release(); await holder; await coordinator.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("reconstructs the governed-response deadline after shutdown without early presentation", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const store = new NativeRunCoordinatorStore(db, fixture.storeBinding);
|
||||
await store.claimFinalization({ leaseOwner: "review-future", leaseTtlMs: 30_000 });
|
||||
await store.markFinalizationRetry({ leaseOwner: "review-future", failureCode: "review_fixture", retryAfterMs: 60_000 });
|
||||
const [state] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
const at = state.nextAttemptAt!.getTime();
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: at });
|
||||
await coordinator.stop();
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
const previousDebug = db.$client.options.debug;
|
||||
const queries: string[] = [];
|
||||
db.$client.options.debug = (_connection, query) => { queries.push(query); };
|
||||
try { await vi.advanceTimersByTimeAsync(30_000); expect(queries).toEqual([]); }
|
||||
finally { db.$client.options.debug = previousDebug; }
|
||||
const resumed = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
try {
|
||||
const restarted = nativeReviewResponseWork(db).register(resumed, () => true);
|
||||
await restarted.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: resumed.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: at });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
await vi.advanceTimersByTimeAsync(Math.max(0, at - Date.now()));
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await comments()).toHaveLength(1);
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
expect(resumed.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await resumed.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
|
||||
try {
|
||||
await worker.ready;
|
||||
await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
|
||||
{ timeout: 10_000 }).toEqual({ active: 0, next: null });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
const previousDebug = db.$client.options.debug;
|
||||
const queries: string[] = [];
|
||||
db.$client.options.debug = (_connection, query) => { queries.push(query); };
|
||||
try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); }
|
||||
finally { db.$client.options.debug = previousDebug; }
|
||||
expect(errors).toEqual([]);
|
||||
} finally { await coordinator.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it.each(["telegram", "discord"] as const)(
|
||||
"presents the exact %s response without resolving a pre-existing native completion review",
|
||||
async (provider) => {
|
||||
|
||||
@@ -500,6 +500,7 @@ export async function restoreNativeChatReviewPresentationInTransaction(
|
||||
decisionId: string;
|
||||
resultId: string;
|
||||
assessmentId: string;
|
||||
onlyWhenDue?: boolean;
|
||||
},
|
||||
): Promise<boolean> {
|
||||
const [coordinator] = await db
|
||||
@@ -519,6 +520,8 @@ export async function restoreNativeChatReviewPresentationInTransaction(
|
||||
.for("update")
|
||||
.limit(1);
|
||||
if (!coordinator) return false;
|
||||
if (input.onlyWhenDue && (coordinator.leaseOwner !== null
|
||||
|| coordinator.nextAttemptAt !== null && coordinator.nextAttemptAt.getTime() > Date.now())) return false;
|
||||
const [decision] = await db
|
||||
.select()
|
||||
.from(statusDecisions)
|
||||
|
||||
@@ -645,6 +645,7 @@ type NativeReconciliationOptions = {
|
||||
skipUnsafeWorkspaceExports?: boolean;
|
||||
skipNativePendingEffects?: boolean;
|
||||
skipNativeResponsePresentation?: boolean;
|
||||
skipNativeReviewResponse?: boolean;
|
||||
pluginWorkerManager?: WorkspaceWorkers;
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
onWorkspaceSettled?: (input: {
|
||||
@@ -875,7 +876,8 @@ async function reconcileNativeFinalizationsPass(
|
||||
row.decisionId &&
|
||||
row.resultId &&
|
||||
row.assessmentId &&
|
||||
currentDecisionJson.externalChatReviewPresentation
|
||||
currentDecisionJson.externalChatReviewPresentation &&
|
||||
!options.skipNativeReviewResponse
|
||||
) {
|
||||
// A later chat turn may retain the same pending review. Recover the
|
||||
// earlier response independently before status reconciliation skips
|
||||
@@ -1125,6 +1127,7 @@ async function reconcileNativeFinalizationsPass(
|
||||
workspaceFinalizeStatus,
|
||||
projectRunStatus: true,
|
||||
skipNativeResponsePresentation: options.skipNativeResponsePresentation,
|
||||
skipNativeReviewResponse: options.skipNativeReviewResponse,
|
||||
preserveProviderAttempt: Boolean(
|
||||
readNativeWorkspaceSyncReference(
|
||||
row.nativeWorkspaceSync,
|
||||
@@ -1184,6 +1187,7 @@ async function reconcileNativeFinalizationsPass(
|
||||
workspaceFinalizeStatus: barrier.status as "succeeded" | "failed",
|
||||
projectRunStatus: true,
|
||||
skipNativeResponsePresentation: options.skipNativeResponsePresentation,
|
||||
skipNativeReviewResponse: options.skipNativeReviewResponse,
|
||||
preserveProviderAttempt: Boolean(
|
||||
readNativeWorkspaceSyncReference(
|
||||
row.nativeWorkspaceSync,
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
import { and, asc, eq, gt, isNotNull, isNull, notExists, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issueComments, nativeRunFinalizations, statusDecisions, type Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import type { NativeReviewResponseScope } from "../native-review-response-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
import { recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js";
|
||||
|
||||
export function registerNativeReviewResponseWork(db: Db,
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
return nativeReviewResponseWork(db).register(coordinator, canRun);
|
||||
}
|
||||
|
||||
type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number };
|
||||
|
||||
export function nativeReviewResponseWork(db: Db) {
|
||||
function candidates(scope?: NativeReviewResponseScope, after?: string, runId?: string) {
|
||||
return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId,
|
||||
issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId,
|
||||
decisionId: nativeRunFinalizations.decisionId, resultId: nativeRunFinalizations.resultId,
|
||||
assessmentId: nativeRunFinalizations.assessmentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
|
||||
}).from(nativeRunFinalizations)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
|
||||
eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
|
||||
.innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId),
|
||||
eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId),
|
||||
eq(statusDecisions.runId, nativeRunFinalizations.runId), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId)))
|
||||
.where(and(eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner),
|
||||
isNotNull(nativeRunFinalizations.resultId), isNotNull(nativeRunFinalizations.assessmentId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded"),
|
||||
eq(statusDecisions.reasonCode, "governed_response_waiting"), eq(statusDecisions.applicationState, "applied"),
|
||||
sql`${statusDecisions.decisionJson}->'externalChatReviewPresentation'->>'schema' = 'paperclip.chat_review_response_presentation.v1'`,
|
||||
notExists(db.select({ id: issueComments.id }).from(issueComments).where(and(
|
||||
eq(issueComments.companyId, nativeRunFinalizations.companyId), eq(issueComments.issueId, nativeRunFinalizations.issueId),
|
||||
eq(issueComments.createdByRunId, nativeRunFinalizations.runId), eq(issueComments.authorAgentId, heartbeatRuns.agentId),
|
||||
eq(issueComments.authorType, "agent"), sql`${issueComments.metadata}->>'authorizationReason' = 'allow_chat_run_presentation'`,
|
||||
sql`${heartbeatRuns.resultJson}->'externalChatReviewPresentation'->>'decisionId' = ${nativeRunFinalizations.decisionId}::text`,
|
||||
))),
|
||||
scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined,
|
||||
scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined,
|
||||
scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined,
|
||||
scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined,
|
||||
runId ? eq(nativeRunFinalizations.runId, runId) : undefined,
|
||||
after ? gt(nativeRunFinalizations.runId, after) : undefined,
|
||||
)).orderBy(asc(nativeRunFinalizations.runId)).limit(100);
|
||||
}
|
||||
async function load(scope?: NativeReviewResponseScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
|
||||
const entries: Entry[] = [];
|
||||
let after: string | undefined;
|
||||
while (canRun() && !signal?.aborted) {
|
||||
const rows = await candidates(scope, after);
|
||||
for (const row of rows) entries.push({ id: row.id, companyId: row.companyId,
|
||||
issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
|
||||
if (rows.length < 100) return entries;
|
||||
after = rows[rows.length - 1].id;
|
||||
}
|
||||
throw new Error("Governed review response discovery stopped");
|
||||
}
|
||||
async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
|
||||
if (signal?.aborted) throw new Error("Governed review response recovery stopped");
|
||||
const scope = { companyId: entry.companyId, issueId: entry.issueId };
|
||||
const [current] = await candidates(scope, undefined, entry.id);
|
||||
if (!current) return null;
|
||||
if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() };
|
||||
if (signal?.aborted) throw new Error("Governed review response recovery stopped");
|
||||
const recoveryStartedAt = Date.now();
|
||||
await recoverCommittedNativeReviewResponse(db, { ...scope, runId: entry.id,
|
||||
decisionId: current.decisionId!, resultId: current.resultId!, assessmentId: current.assessmentId!, onlyWhenDue: true });
|
||||
const [remaining] = await candidates(scope, undefined, entry.id);
|
||||
return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt
|
||||
? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null;
|
||||
}
|
||||
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.nativeReviewResponse,
|
||||
decodeScope: (hint: string): NativeReviewResponseScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (entry: Entry, scope?: NativeReviewResponseScope) => !scope || entry.companyId === scope.companyId
|
||||
&& (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId)
|
||||
&& (!scope.agentId || entry.agentId === scope.agentId),
|
||||
load: (scope, signal) => load(scope, signal, canRun),
|
||||
check: (entry, signal) => {
|
||||
if (!canRun()) throw new Error("Governed review response recovery paused");
|
||||
return check(entry, signal);
|
||||
},
|
||||
}, canRun);
|
||||
}
|
||||
return { load, check, register };
|
||||
}
|
||||
@@ -723,65 +723,62 @@ async function projectCommittedRun(input: {
|
||||
}
|
||||
}
|
||||
|
||||
async function materializeCommittedReviewResponse(db: Db, runId: string) {
|
||||
try {
|
||||
const [run] = await db
|
||||
.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.limit(1);
|
||||
if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
|
||||
return;
|
||||
const decisionId = record(
|
||||
run.resultJson.externalChatReviewPresentation,
|
||||
).decisionId;
|
||||
if (
|
||||
typeof decisionId !== "string" ||
|
||||
(await hasMaterializedNativeReviewResponse(db, {
|
||||
companyId: run.companyId,
|
||||
issueId: run.nativeIssueId,
|
||||
runId,
|
||||
decisionId,
|
||||
}))
|
||||
)
|
||||
return;
|
||||
const summary = record(run.resultJson.nativeResult).summary;
|
||||
if (
|
||||
typeof summary !== "string" ||
|
||||
!(await authorizeNativeChatReviewPresentation(db, {
|
||||
companyId: run.companyId,
|
||||
issueId: run.nativeIssueId,
|
||||
runId,
|
||||
resultJson: run.resultJson,
|
||||
}))
|
||||
)
|
||||
return;
|
||||
// The coordinator transaction is already committed. addComment owns a
|
||||
// fresh issue transaction, rechecks the proof and carries selected files;
|
||||
// its same-run/text and publication keys make recovery replay idempotent.
|
||||
await issueService(db).addComment(
|
||||
run.nativeIssueId,
|
||||
summary,
|
||||
{ agentId: run.agentId, runId },
|
||||
{
|
||||
authorizationReason: "allow_chat_run_presentation",
|
||||
},
|
||||
);
|
||||
publishChatPublicationCommitSignal({
|
||||
async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) {
|
||||
const [run] = await dbOrTx
|
||||
.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.limit(1);
|
||||
if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
|
||||
return null;
|
||||
const decisionId = record(
|
||||
run.resultJson.externalChatReviewPresentation,
|
||||
).decisionId;
|
||||
if (
|
||||
typeof decisionId !== "string" ||
|
||||
(await hasMaterializedNativeReviewResponse(dbOrTx, {
|
||||
companyId: run.companyId,
|
||||
issueId: run.nativeIssueId,
|
||||
runId: run.id,
|
||||
agentId: run.agentId,
|
||||
eventType: "run.presentation.resolved",
|
||||
});
|
||||
runId,
|
||||
decisionId,
|
||||
}))
|
||||
)
|
||||
return null;
|
||||
const summary = record(run.resultJson.nativeResult).summary;
|
||||
if (
|
||||
typeof summary !== "string" ||
|
||||
!(await authorizeNativeChatReviewPresentation(dbOrTx, {
|
||||
companyId: run.companyId,
|
||||
issueId: run.nativeIssueId,
|
||||
runId,
|
||||
resultJson: run.resultJson,
|
||||
}))
|
||||
)
|
||||
return null;
|
||||
await issueService(db).addComment(
|
||||
run.nativeIssueId,
|
||||
summary,
|
||||
{ agentId: run.agentId, runId },
|
||||
{
|
||||
authorizationReason: "allow_chat_run_presentation",
|
||||
},
|
||||
dbOrTx,
|
||||
);
|
||||
return run;
|
||||
}
|
||||
|
||||
function publishNativeReviewResponse(run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "nativeIssueId">) {
|
||||
publishChatPublicationCommitSignal({ companyId: run.companyId, issueId: run.nativeIssueId!,
|
||||
runId: run.id, agentId: run.agentId, eventType: "run.presentation.resolved" });
|
||||
}
|
||||
|
||||
async function materializeCommittedReviewResponse(db: Db, runId: string) {
|
||||
try {
|
||||
const run = await materializeNativeReviewResponse(db, runId);
|
||||
if (run) publishNativeReviewResponse(run);
|
||||
} catch (error) {
|
||||
// The durable committed proof remains retryable by the next reconciler
|
||||
// sweep. Presentation failure must not reclassify a successful native run.
|
||||
logger.warn(
|
||||
{ runId, error },
|
||||
"Committed chat review response is awaiting presentation retry",
|
||||
);
|
||||
logger.warn({ runId, error }, "Committed chat review response is awaiting presentation retry");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1183,6 +1180,28 @@ export async function repairCommittedNativeReviewResponse(
|
||||
);
|
||||
if (restored) await materializeCommittedReviewResponse(db, input.runId);
|
||||
}
|
||||
export async function recoverCommittedNativeReviewResponse(db: Db, input: {
|
||||
companyId: string; issueId: string; runId: string; decisionId: string;
|
||||
resultId: string; assessmentId: string; onlyWhenDue?: boolean;
|
||||
}): Promise<boolean> {
|
||||
const materialized = await db.transaction(async transaction => {
|
||||
const tx = transaction as unknown as Db;
|
||||
|
||||
|
||||
const [coordinator] = await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations)
|
||||
.where(and(eq(nativeRunFinalizations.companyId, input.companyId),
|
||||
eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId))).for("update");
|
||||
if (!coordinator) return null;
|
||||
const [issue] = await tx.select({ id: issues.id }).from(issues)
|
||||
.where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update");
|
||||
if (!issue || await hasMaterializedNativeReviewResponse(tx, input)) return null;
|
||||
if (!await restoreNativeChatReviewPresentationInTransaction(tx, input)) return null;
|
||||
return materializeNativeReviewResponse(db, input.runId, tx);
|
||||
});
|
||||
if (materialized) publishNativeReviewResponse(materialized);
|
||||
return materialized !== null;
|
||||
}
|
||||
|
||||
export async function finalizeNativeRun(input: {
|
||||
db: Db;
|
||||
runId: string;
|
||||
@@ -1190,6 +1209,7 @@ export async function finalizeNativeRun(input: {
|
||||
/** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */
|
||||
projectRunStatus?: boolean;
|
||||
skipNativeResponsePresentation?: boolean;
|
||||
skipNativeReviewResponse?: boolean;
|
||||
/** Workspace-only replay must not consume the provider recovery budget. */
|
||||
preserveProviderAttempt?: boolean;
|
||||
failpoint?: NativeStatusCommitFailpoint;
|
||||
@@ -1214,14 +1234,14 @@ export async function finalizeNativeRun(input: {
|
||||
const coordinator = claim.coordinator;
|
||||
if (!claim.leaseOwner && coordinator.phase === "committed") {
|
||||
if (isNativeRunnerOwnershipHeld(admittedRun)) return coordinator;
|
||||
const presentationAlreadyMaterialized = coordinator.decisionId
|
||||
const presentationAlreadyMaterialized = input.skipNativeReviewResponse || (coordinator.decisionId
|
||||
? await hasMaterializedNativeReviewResponse(input.db, {
|
||||
companyId: admittedRun.companyId,
|
||||
issueId: coordinator.issueId,
|
||||
runId: admittedRun.id,
|
||||
decisionId: coordinator.decisionId,
|
||||
})
|
||||
: false;
|
||||
: false);
|
||||
if (
|
||||
!presentationAlreadyMaterialized &&
|
||||
coordinator.decisionId &&
|
||||
|
||||
Reference in new issue
Block a user