fix(server): wake native reassessment from task and artifact writes

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-26 02:49:23 -05:00
1 parent 27d0ca8264
commit 2d1b61c75f
9 files changed
+188 -10

No files matched your search

@@ -2481,3 +2481,12 @@ and their synthetic-fixture limits are recorded in the owner audit. The worker i
not registered in startup and periodic reassessment remains. Producer coverage,
concurrent authority audit, selected-current/backlog cost and full acceptance are
still pending; no slice 50 PR or merge, and the full original scope stays open.
Slice 50 producer checkpoint: six real work-product/attachment transaction misses
and three task mutation misses were reproduced and repaired. The real promotion
HTTP route also wakes reassessment. All 13 owner lifecycle tests, 79 affected
consumer cases across their documented runs, and server types pass. Initial
embedded-PostgreSQL failures and the external-database test repairs are preserved
in the owner audit. Approval/interaction/run evidence and workspace producers,
concurrent authority, backlog cost, startup wiring and periodic removal remain.
Master remains 1fbbef45a8; older stack gates and the full original scope stay open.
@@ -209,3 +209,52 @@ Final current-source local rerun passes 116 tests across the status corpus and
isolated worker lifecycle files (/tmp/sleep50-final-local.log). Together with the
187 governed-response cases this is 303 distinct passing cases, not whole-workspace
or remote PR acceptance. The last server typecheck also passes.
## Work-product and task producer checkpoint
Six real writer regressions failed before notifications: artifact update, create,
primary replacement, import, attachment creation, and nested transaction commit.
Their existing write transactions now signal task-scoped reassessment before the
mutation. Import coalesces its scopes per task. Updates signal both old and new
scope if their supported patch moves a product. Removal also signals its old task.
Attachment creation signals before asset/attachment/product writes, including
attachment evidence without a registered run. Native file handoff and attachment
reuse call that service inside their tool-authority transaction; their subsequent
same-transaction title/metadata edits share its outer settlement. Those fields
are not classifier evidence, so no second recovery wrapper was added there.
The low-trust promotion route signals inside its own atomic source/product write.
A real HTTP promotion now also proves completion through the registered worker.
Three further regressions failed for task status changes, assignment changes,
and reasserting blocked state. The task mutation owner now signals relevant
status/version, execution, assignment and blocker writes. The regression checks
that the new assessment reflects the task's actual version and does not overwrite
the board's status. The nested work-product case verifies no decision before the
outer commit, no decision after rollback, then successful reassessment after a
later commit. All paths return to driver-observed SQL quiet.
Final owner lifecycle tests pass all 13 cases, including the real promotion route,
and server types pass. Evidence: /tmp/sleep50-products-before.log (six failures),
/tmp/sleep50-products-after.log (nine passes), /tmp/sleep50-tasks-before.log
(three failures), /tmp/sleep50-tasks-after.log (12 passes), and
/tmp/sleep50-products-final.log (13 passes) plus its final typecheck log.
Affected consumers pass 79 cases across five files: work products, attachment
routes, low-trust HTTP routes, attachment reuse, and runner task attachments.
The first combined run passed 53 but failed two suites during embedded PostgreSQL
bootstrap, leaving 26 cases unrun. Those suites now accept validated dedicated
external test database URLs and explicitly close their clients. Their 26 cases
then pass on separate PostgreSQL databases. Low-trust route tests use their own
dedicated database because teardown deletes shared tables. Preserve the initial
failure in /tmp/sleep50-consumers.log and the successful rerun in
/tmp/sleep50-attachment-consumers-after.log. This is 79 passing consumer cases
across the two runs, not a green initial run or a whole-workspace result.
Master was refreshed and remains 1fbbef45a8. The stack has not been reconciled
onto it yet. The worker remains unregistered and periodic reassessment remains.
Remaining producer coverage includes approval/interaction and run-event evidence,
case handoffs, native ownership and run-status transitions, completion-contract
changes and workspace-finalization barriers. Workspace recovery writes a separate
workspace_finalize success receipt; snapshot-only rows are not that barrier.
Continue the concurrent authority audit, independent-current-history/backlog cost,
lifecycle/failure evidence, startup/periodic wiring and full PR acceptance.
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { randomUUID } from "node:crypto";
import { createServer } from "node:http";
import express from "express";
@@ -55,12 +56,13 @@ import { issueService } from "../services/issues.js";
import { heartbeatService } from "../services/heartbeat.js";
import { LOW_TRUST_QUARANTINED_BODY } from "../services/source-trust.js";
const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_LOW_TRUST_TEST_DATABASE_URL");
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported
const describeEmbeddedPostgres = externalTestDatabaseUrl || embeddedPostgresSupport.supported
? describe
: describe.skip;
if (!embeddedPostgresSupport.supported) {
if (!externalTestDatabaseUrl && !embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres low-trust route tests on this host: ${
embeddedPostgresSupport.reason ?? "unsupported environment"
@@ -885,10 +887,10 @@ describeEmbeddedPostgres(
}
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase(
if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase(
"paperclip-low-trust-red-team-routes-",
);
db = createDb(tempDb.connectionString);
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString);
}, 20_000);
afterEach(async () => {
@@ -927,6 +929,7 @@ describeEmbeddedPostgres(
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
@@ -1,3 +1,10 @@
import express from "express";
import request from "supertest";
import { LOW_TRUST_REVIEW_PRESET } from "@paperclipai/shared";
import { issueRoutes } from "../routes/issues.js";
import { errorHandler } from "../middleware/index.js";
import { workProductService } from "../services/work-products.js";
import { issueService } from "../services/issues.js";
import { randomUUID } from "node:crypto";
import { setTimeout as delay } from "node:timers/promises";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -96,9 +103,92 @@ describe("native reassessment worker lifecycle", () => {
prompt: "Review the persisted native-run evidence and confirm whether this issue may be completed." }] } });
await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: runId, issueId, phase: "workspace_finalize",
status: "succeeded", exitCode: 0, cwd: process.cwd(), finishedAt: new Date() });
return { issueId, runId, decisionId: committed.decision.id };
return { issueId, runId, workProductId, decisionId: committed.decision.id };
}
it.each(["update", "create", "primary replacement", "import", "attachment", "source promotion"] as const)("wakes from the real work-product %s writer", async writer => {
const seeded = await seedReview();
if (writer === "update") await db.update(issueWorkProducts).set({ reviewState: "none" }).where(eq(issueWorkProducts.id, seeded.workProductId));
if (writer === "create") await db.delete(issueWorkProducts).where(eq(issueWorkProducts.id, seeded.workProductId));
if (writer === "source promotion") await db.update(issueWorkProducts).set({ sourceTrust: {
preset: LOW_TRUST_REVIEW_PRESET, disposition: "quarantined", sourceIssueId: seeded.issueId,
sourceRunId: seeded.runId, sourceAgentId: agentId,
} }).where(eq(issueWorkProducts.id, seeded.workProductId));
await start(); await flush(); await quiet();
expect(await status(seeded.issueId)).toBe("in_review");
const products = workProductService(db);
if (writer === "update") await products.update(seeded.workProductId, { reviewState: "approved" });
if (writer === "create" || writer === "primary replacement") await products.createForIssue(seeded.issueId, companyId, {
...(writer === "create" ? { id: seeded.workProductId } : {}),
type: "artifact", provider: "paperclip", title: "Verified artifact", status: "approved", reviewState: "approved",
isPrimary: writer === "primary replacement",
});
if (writer === "import") await products.createManyForImport([{ companyId, issueId: seeded.issueId, projectId: null,
type: "artifact", provider: "paperclip", externalId: null, title: "Imported evidence", url: null, status: "approved",
reviewState: "approved", isPrimary: false, healthStatus: "unknown", summary: null, metadata: null,
executionWorkspaceId: null, runtimeServiceId: null, createdByRunId: null, sourceTrust: null }]);
if (writer === "attachment") await issueService(db).createAttachment({ issueId: seeded.issueId, provider: "local_disk",
objectKey: randomUUID(), contentType: "text/plain", byteSize: 4, sha256: "a".repeat(64), originalFilename: "result.txt",
createdByAgentId: agentId, createdByRunId: seeded.runId });
if (writer === "source promotion") {
const app = express(); app.use(express.json());
app.use((req, _res, next) => {
req.actor = { type: "board", userId: "board-user", companyIds: [companyId],
memberships: [{ companyId, membershipRole: "operator", status: "active" }],
isInstanceAdmin: true, source: "local_implicit" }; next();
});
app.use("/api", issueRoutes(db, {} as any)); app.use(errorHandler);
const promoted = await request(app).post(`/api/issues/${seeded.issueId}/low-trust/promotions`).send({
sourceArtifactKind: "work_product", sourceArtifactId: seeded.workProductId,
title: "Reviewed artifact", summary: "Reviewed completion evidence",
});
expect(promoted.status, JSON.stringify(promoted.body)).toBe(201);
}
await flush();
expect(await status(seeded.issueId)).toBe("done");
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2);
await quiet();
});
it("waits for an outer work-product transaction and ignores rollback", async () => {
const seeded = await seedReview();
await db.update(issueWorkProducts).set({ reviewState: "none" }).where(eq(issueWorkProducts.id, seeded.workProductId));
await start(); await flush(); await quiet();
await expect(db.transaction(async tx => {
await workProductService(tx as unknown as typeof db).update(seeded.workProductId, { reviewState: "approved" });
await vi.advanceTimersByTimeAsync(1000);
expect(await status(seeded.issueId)).toBe("in_review");
throw new Error("rollback evidence approval");
})).rejects.toThrow("rollback evidence approval");
await flush(); await quiet();
expect(await status(seeded.issueId)).toBe("in_review");
await db.transaction(async tx => {
await workProductService(tx as unknown as typeof db).update(seeded.workProductId, { reviewState: "approved" });
await vi.advanceTimersByTimeAsync(1000);
expect(await status(seeded.issueId)).toBe("in_review");
});
await flush();
expect(await status(seeded.issueId)).toBe("done");
await quiet();
});
it.each(["status", "blocked reassertion", "assignment"] as const)("reassesses a real task %s mutation after quiet", async mutation => {
const seeded = await seedReview();
if (mutation === "blocked reassertion") await issueService(db).update(seeded.issueId, { status: "blocked" });
await start(); await flush(); await quiet();
const before = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId));
await issueService(db).update(seeded.issueId, mutation === "assignment" ? { assigneeAgentId: null } : { status: "blocked" });
await flush();
const after = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId));
expect(after).toHaveLength(before.length + 1);
expect(await status(seeded.issueId)).toBe(mutation === "assignment" ? "in_review" : "blocked");
const [current] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seeded.runId));
const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, current.assessmentId!));
const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId));
expect(assessment.priorStatusVersion).toBe(issue.statusVersion);
await quiet();
});
it("retains the first review-retirement wake through idle drain", async () => {
const seeded = await seedReview();
await start(); await flush(); await quiet();
+2
View File
@@ -1,3 +1,4 @@
import { notifyNativeStatusReassessment } from "../services/native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../services/unsafe-workspace-export-notifications.js";
import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js";
import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js";
@@ -11609,6 +11610,7 @@ export function issueRoutes(
promotedAt,
});
const product = await db.transaction(async (tx) => {
await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id });
const markPromoted = {
sourceTrust: promotionTrust,
updatedAt: promotedAt,
+7
View File
@@ -1,3 +1,4 @@
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
@@ -11332,6 +11333,11 @@ export function issueService(db: Db) {
|| (patch.checkoutRunId !== undefined && patch.checkoutRunId !== receiptExisting.checkoutRunId)) {
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: receiptExisting.companyId, issueId: id });
}
if (patch.status !== undefined || patch.statusVersion !== undefined || patch.executionState !== undefined
|| patch.executionRunId !== undefined || patch.checkoutRunId !== undefined
|| patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || blockedByIssueIds !== undefined) {
await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id });
}
const updated = await tx
.update(issues)
.set(patch)
@@ -13349,6 +13355,7 @@ export function issueService(db: Db) {
}
}
}
await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id });
const [asset] = await tx
.insert(assets)
.values({
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
import { createHash, randomUUID } from "node:crypto";
import { mkdir, mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
@@ -68,8 +69,9 @@ describe("native same-conversation chat attachment reuse", () => {
);
beforeAll(async () => {
temporary = await startEmbeddedPostgresTestDatabase("native-chat-reuse-");
db = createDb(temporary.connectionString);
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CHAT_REUSE_TEST_DATABASE_URL");
if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-chat-reuse-");
db = createDb(external ?? temporary!.connectionString);
const storageRoot = await mkdtemp(
path.join(tmpdir(), "paperclip-chat-reuse-"),
);
@@ -293,6 +295,7 @@ describe("native same-conversation chat attachment reuse", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await temporary?.cleanup();
});
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
import { createHash, randomUUID } from "node:crypto";
import { Readable } from "node:stream";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
@@ -11,10 +12,14 @@ import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.
const sha = (body: Buffer) => createHash("sha256").update(body).digest("hex");
describe("Dot assigned-task attachment reading", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>;
beforeAll(async () => { temporary = await startEmbeddedPostgresTestDatabase("dot-task-files-"); db = createDb(temporary.connectionString); });
afterAll(async () => { await temporary?.cleanup(); });
beforeAll(async () => {
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUNNER_ATTACHMENTS_TEST_DATABASE_URL");
if (!external) temporary = await startEmbeddedPostgresTestDatabase("dot-task-files-");
db = createDb(external ?? temporary!.connectionString);
});
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
afterEach(() => { vi.unstubAllEnvs(); vi.unstubAllGlobals(); });
async function fixture(body = Buffer.from("only in the file 🌍"), enabled = true) {
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+10
View File
@@ -1,3 +1,4 @@
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyApprovalQueueWork } from "./approval-work-notifications.js";
import { and, desc, eq, inArray } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@@ -274,6 +275,7 @@ export function workProductService(
createForIssue: async (issueId: string, companyId: string, data: Omit<typeof issueWorkProducts.$inferInsert, "issueId" | "companyId">) => {
const row = await db.transaction(async (tx) => {
await notifyNativeStatusReassessment(tx, { companyId, issueId });
if (data.type === "pull_request") {
await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`);
await notifyApprovalQueueWork(tx, companyId);
@@ -312,6 +314,10 @@ export function workProductService(
.where(eq(issueWorkProducts.id, id))
.then((rows) => rows[0] ?? null);
if (!existing) return null;
await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: existing.issueId });
if ((patch.companyId && patch.companyId !== existing.companyId) || (patch.issueId && patch.issueId !== existing.issueId)) {
await notifyNativeStatusReassessment(tx, { companyId: patch.companyId ?? existing.companyId, issueId: patch.issueId ?? existing.issueId });
}
if (existing.type === "pull_request" || patch.type === "pull_request") {
await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`);
@@ -385,6 +391,9 @@ export function workProductService(
// prefix behind (which a retry would then duplicate). Mirrors the
// per-writer transaction the batched issue/document writers use.
await db.transaction(async (tx) => {
for (const scope of new Map(values.map(value => [value.issueId, { companyId: value.companyId, issueId: value.issueId }])).values()) {
await notifyNativeStatusReassessment(tx, scope);
}
for (const companyId of new Set(values.filter(value => value.type === "pull_request").map(value => value.companyId))) {
await notifyApprovalQueueWork(tx, companyId);
}
@@ -399,6 +408,7 @@ export function workProductService(
const row = await db.transaction(async tx => {
const [existing] = await tx.select({ type: issueWorkProducts.type, issueId: issueWorkProducts.issueId, companyId: issueWorkProducts.companyId })
.from(issueWorkProducts).where(eq(issueWorkProducts.id, id));
if (existing) await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: existing.issueId });
if (existing?.type === "pull_request") {
await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`);
await notifyApprovalQueueWork(tx, existing.companyId);