mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 14:16:39 +01:00
fix(server): wake governed responses from scoped approval changes
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
9f56f54986
commit
53b26e24f9
9 files changed
+146
-14
No files matched your search
@@ -2375,3 +2375,19 @@ null deadline or unfinished startup batch. Their settlement predicate now includ
|
||||
zero active work. All 18 affected cases pass against the full suite's retained
|
||||
history. Fresh remote acceptance remains pending; these are recorded test repairs,
|
||||
not a waiver of failed gates or a change to production batch behavior.
|
||||
|
||||
## Slice 48 accepted; slice 49 approval producer checkpoint
|
||||
|
||||
#17696 at `fad35f2d7ae6bdbb85e709f1bab4633d197fd108` has all 53 checks
|
||||
passing, Greptile 5/5 at that commit, and zero unresolved review threads. Its PR
|
||||
body now records those results and preserves the earlier local/remote failures.
|
||||
It remains draft and unmerged. Evidence is in /tmp/sleep48-final-gates.json,
|
||||
/tmp/sleep48-final-comments.json and /tmp/sleep48-final-review-threads.json.
|
||||
|
||||
Slice 49 has a guarded recovery owner and scoped worker, plus verified approval
|
||||
producer wakes. The 171-case owner suite and 70 existing approval/budget/queue/
|
||||
retention consumer cases pass, as does server typecheck. See its audit for the
|
||||
reproduced missed-wake failures and repaired test stub. Interaction producer
|
||||
coverage, discovery cost, startup wiring, periodic removal and full acceptance
|
||||
remain outstanding. No slice 49 PR or whole-instance sleep is claimed. Earlier
|
||||
stack gates, master reconciliation and every remaining original duty stay in scope.
|
||||
@@ -160,3 +160,37 @@ startup reconstruction, deleted receipts, and actual quiet SQL after settlement.
|
||||
|
||||
The producer audit, query-cost measurements, startup/periodic wiring, remaining
|
||||
lifecycle acceptance, full verification and remote gates are still outstanding.
|
||||
|
||||
## Approval producer checkpoint
|
||||
|
||||
Approval resolution, cancellation, revision/resubmission, link/unlink and bulk
|
||||
link now pass their existing approval/task identity through writeApprovalChange.
|
||||
The notification owner finds linked tasks before the mutation and registers only
|
||||
those task scopes. Budget resolution supplies its approval identity inside the
|
||||
existing accounting transaction. Creation without a task link and company-wide
|
||||
retention notifications retain their previous behavior and do not reload response
|
||||
history. No additional transaction wrapper or approval execution policy was added.
|
||||
All production approval updates/deletes were inventoried; company deletion leaves
|
||||
no surviving response obligation. Tool-gateway's separate link insertion still
|
||||
needs its explicit task scope during the remaining producer pass.
|
||||
|
||||
Four real-producer regressions (approve, reject, cancel, unlink) failed before
|
||||
wiring and passed afterward. Two further cases exercise nested cancel/unlink
|
||||
with outer rollback/commit and measure no governed-response candidate SQL for an
|
||||
unrelated approval. All 171 owner cases pass. In the combined run two older
|
||||
approval unit cases failed because their ordered stub consumed an approval row
|
||||
for the new issue_approvals read. The stub now distinguishes that table and returns
|
||||
no links. Existing consumers pass 19 approval service/route cases, five budget
|
||||
cases, 25 queue-seeding cases, and 21 retention cases. Server typecheck passes.
|
||||
Logs: /tmp/sleep49-approval-before.log, /tmp/sleep49-approval-after.log,
|
||||
/tmp/sleep49-approval-full.log, /tmp/sleep49-approval-unit.log,
|
||||
/tmp/sleep49-budgets.log, /tmp/sleep49-approval-queue.log,
|
||||
/tmp/sleep49-approval-retention.log, /tmp/sleep49-approval-final-types.log.
|
||||
|
||||
Remaining interaction gaps include connection-intent resolution/expiry, generic
|
||||
interaction administrative settlement, satisfied-connection cleanup, and OAuth
|
||||
confirmation callbacks/denials. Some existing milestone hooks cover only question
|
||||
and confirmation kinds. Payload-only updates that leave a competing card pending
|
||||
do not reactivate a response. Task ownership changes already carry task invalidation,
|
||||
but direct cleanup writers must be individually accounted for. The polling backstop
|
||||
is retained until this audit and the remaining acceptance requirements are complete.
|
||||
@@ -1,6 +1,6 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { approvalService } from "../services/approvals.ts";
|
||||
import { companies } from "@paperclipai/db";
|
||||
import { companies, issueApprovals } from "@paperclipai/db";
|
||||
|
||||
vi.mock("../services/budgets.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../services/budgets.js")>(),
|
||||
@@ -50,7 +50,9 @@ function createDbStub(selectResults: ApprovalRecord[][], updateResults: Approval
|
||||
const pendingSelectResults = [...selectResults];
|
||||
let inTransaction = false;
|
||||
const selectWhere = vi.fn(async () => (inTransaction ? pendingSelectResults.shift() : pendingSelectResults[0]) ?? []);
|
||||
const from = vi.fn((table) => table === companies
|
||||
const from = vi.fn((table) => table === issueApprovals
|
||||
? { where: async () => [] }
|
||||
: table === companies
|
||||
? { where: () => ({ for: async () => [{ id: "company-1" }] }) }
|
||||
: { where: selectWhere });
|
||||
const select = vi.fn(() => ({ from }));
|
||||
|
||||
@@ -49,7 +49,7 @@ export function approvalResolutionRecords(db: Db) {
|
||||
})
|
||||
.where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses)))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null));
|
||||
.then((rows) => rows[0] ?? null), { approvalId: id });
|
||||
|
||||
if (updated) {
|
||||
return { approval: updated, applied: true };
|
||||
|
||||
@@ -1,16 +1,27 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { issueApprovals, type Db } from "@paperclipai/db";
|
||||
import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export const notifyApprovalQueueWork = (transaction: object, companyId: string) => notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalQueueSeeding, companyId);
|
||||
|
||||
export async function notifyApprovalWork(transaction: object, companyId: string) {
|
||||
type ApprovalChangeScope = { approvalId?: string; issueIds?: string[] };
|
||||
|
||||
export async function notifyApprovalWork(transaction: object, companyId: string, scope?: ApprovalChangeScope) {
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention);
|
||||
await notifyApprovalQueueWork(transaction, companyId);
|
||||
const issueIds = new Set(scope?.issueIds);
|
||||
if (scope?.approvalId) {
|
||||
const links = await (transaction as Db).select({ issueId: issueApprovals.issueId }).from(issueApprovals)
|
||||
.where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.approvalId, scope.approvalId)));
|
||||
for (const link of links) issueIds.add(link.issueId);
|
||||
}
|
||||
for (const issueId of issueIds) await notifyNativeReviewResponse(transaction, { companyId, issueId });
|
||||
}
|
||||
|
||||
export function writeApprovalChange<T>(db: Db, companyId: string, write: (transaction: Db) => Promise<T>): Promise<T> {
|
||||
export function writeApprovalChange<T>(db: Db, companyId: string, write: (transaction: Db) => Promise<T>, scope?: ApprovalChangeScope): Promise<T> {
|
||||
return db.transaction(async tx => {
|
||||
await notifyApprovalWork(tx, companyId);
|
||||
await notifyApprovalWork(tx, companyId, scope);
|
||||
return write(tx as unknown as Db);
|
||||
});
|
||||
}
|
||||
@@ -77,7 +77,7 @@ export function approvalRecords(db: Db) {
|
||||
})
|
||||
.where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses)))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null));
|
||||
.then((rows) => rows[0] ?? null), { approvalId: id });
|
||||
return updated;
|
||||
},
|
||||
|
||||
@@ -99,7 +99,7 @@ export function approvalRecords(db: Db) {
|
||||
})
|
||||
.where(eq(approvals.id, id))
|
||||
.returning()
|
||||
.then((rows) => rows[0]));
|
||||
.then((rows) => rows[0]), { approvalId: id });
|
||||
},
|
||||
|
||||
resubmit: async (id: string, payload?: Record<string, unknown>) => {
|
||||
@@ -121,7 +121,7 @@ export function approvalRecords(db: Db) {
|
||||
})
|
||||
.where(eq(approvals.id, id))
|
||||
.returning()
|
||||
.then((rows) => rows[0]));
|
||||
.then((rows) => rows[0]), { approvalId: id });
|
||||
},
|
||||
|
||||
listComments: async (approvalId: string) => {
|
||||
|
||||
@@ -262,7 +262,7 @@ async function markApprovalStatus(
|
||||
if (!approvalId) return;
|
||||
const [approval] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, approvalId));
|
||||
if (!approval) return;
|
||||
await notifyApprovalWork(db, approval.companyId);
|
||||
await notifyApprovalWork(db, approval.companyId, { approvalId });
|
||||
await db
|
||||
.update(approvals)
|
||||
.set({
|
||||
|
||||
@@ -117,7 +117,7 @@ export function issueApprovalService(db: Db) {
|
||||
linkedByAgentId: actor?.agentId ?? null,
|
||||
linkedByUserId: actor?.userId ?? null,
|
||||
})
|
||||
.onConflictDoNothing());
|
||||
.onConflictDoNothing(), { issueIds: [issueId] });
|
||||
|
||||
return db
|
||||
.select()
|
||||
@@ -130,7 +130,7 @@ export function issueApprovalService(db: Db) {
|
||||
const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId);
|
||||
await writeApprovalChange(db, issue.companyId, tx => tx
|
||||
.delete(issueApprovals)
|
||||
.where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId))));
|
||||
.where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId))), { issueIds: [issueId] });
|
||||
},
|
||||
|
||||
linkManyForApproval: async (approvalId: string, issueIds: string[], actor?: LinkActor) => {
|
||||
@@ -169,7 +169,7 @@ export function issueApprovalService(db: Db) {
|
||||
linkedByUserId: actor?.userId ?? null,
|
||||
})),
|
||||
)
|
||||
.onConflictDoNothing());
|
||||
.onConflictDoNothing(), { issueIds: uniqueIssueIds });
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -48,6 +48,8 @@ import type {
|
||||
|
||||
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { finalizeNativeRun, recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js";
|
||||
import { approvalRecords } from "../approvals.js";
|
||||
import { issueApprovalService } from "../issue-approvals.js";
|
||||
import { nativeReviewResponseWork } from "./native-review-response-work.js";
|
||||
import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { idleWorkSnapshot } from "../task-admission.js";
|
||||
@@ -2868,6 +2870,73 @@ describe("native external-chat response wait", () => {
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
|
||||
});
|
||||
|
||||
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(["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("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => {
|
||||
const fixture = await seedRecoverableReviewResponse();
|
||||
const errors: unknown[] = [];
|
||||
|
||||
Reference in new issue
Block a user