Read release evidence and telemetry without unused output (#17485)

Read execution release evidence and terminal telemetry without unused output. Keep full stored history, in-flight guards, and accepted messages.

Validated with real PostgreSQL tests, build, typecheck, green CI, and Greptile 5/5.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed 2026-10-24 14:01:41 -07:00
1 parent 991c47bc6e
commit 1cfaf973cd
6 files changed
+157 -13

No files matched your search

@@ -1,5 +1,5 @@
import { instanceSettingsService } from "../../../services/instance-settings.js";
import { randomUUID } from "node:crypto";
import { randomBytes, randomUUID } from "node:crypto";
import { eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import type { Db } from "@paperclipai/db";
@@ -177,6 +177,79 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
return id;
}
function recordReleaseRunReads() {
const rows: unknown[][] = [];
function recordSelects<T extends Pick<Db, "select">>(executor: T): T {
return new Proxy(executor, { get(target, key) {
if (key !== "select") return Reflect.get(target, key);
return (...fields: Parameters<Db["select"]>) => {
const builder = target.select(...fields), from = builder.from.bind(builder);
builder.from = ((...args: Parameters<typeof from>) => {
const query = from(...args);
if (args[0] === heartbeatRuns) {
const execute = query.execute.bind(query);
query.execute = async (...executeArgs: Parameters<typeof execute>) => {
const result = await execute(...executeArgs); rows.push(result); return result;
};
}
return query;
}) as typeof builder.from;
return builder;
};
}});
}
const transaction: Db["transaction"] = (callback, config) =>
db.transaction(tx => callback(recordSelects(tx)), config);
const recorded = new Proxy(db, { get(target, key) {
return key === "transaction" ? transaction : Reflect.get(target, key);
}});
return { recorded, rows };
}
it("keeps in-flight work and every recovery receipt while omitting output from release reads", async () => {
const companyId = await seedCompany();
const agentId = await seedAgent({ companyId });
const issueId = await seedIssue({ companyId, assigneeAgentId: agentId });
const runId = await seedRun({ companyId, agentId, status: "running", contextSnapshot: { issueId } });
const wakeId = await seedDeferredWake({ companyId, agentId, issueId });
const payload = randomBytes(512 * 1024).toString("base64");
const evidence = { cancellation: { source: "queued_message", expected: true, initiator: { type: "system" }, reason: "Fixture", recordedAt: "2020-01-01T00:00:00Z" },
workspaceRestoreFailure: "restore_unsafe_archive", workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [randomUUID()] },
nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId, companyId, issueId },
executionCancellation: { state: "acknowledged" }, executionRecovery: { kind: "interrupted", providerStopped: true },
configurationIncomplete: { reason: "workspace_unavailable" }, unknownFutureReceipt: { retained: true } };
await db.update(heartbeatRuns).set({ resultJson: { ...evidence, stdout: payload.repeat(3), stderr: payload } }).where(eq(heartbeatRuns.id, runId));
await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId));
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId));
const { recorded, rows } = recordReleaseRunReads();
const result = await createPostgresWakeQueueAdapter(recorded, stubDeps).withIssueExecutionLock(
{ companyId, runId, now: new Date(), resumeStrandedQueue: true },
async () => { throw new Error("in-flight work must not drain"); },
);
expect(result.outcome.kind).toBe("released");
expect(rows).toHaveLength(1);
expect(rows[0]).toEqual([{ ...run, resultJson: evidence }]);
expect(Buffer.byteLength(JSON.stringify(rows))).toBeLessThan(8 * 1024);
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]).toEqual(run);
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toEqual(issue);
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]).toEqual(wake);
});
it.each([null, ["stdout", { stderr: "array evidence" }], "scalar", 42].map(result => ({ result })))("preserves non-object result evidence $result during release", async ({ result }) => {
const companyId = await seedCompany(), agentId = await seedAgent({ companyId });
const runId = await seedRun({ companyId, agentId, status: "running" });
await db.update(heartbeatRuns).set({ resultJson: result as Record<string, unknown> | null }).where(eq(heartbeatRuns.id, runId));
const { recorded, rows } = recordReleaseRunReads();
await createPostgresWakeQueueAdapter(recorded, stubDeps).withIssueExecutionLock(
{ companyId, runId, now: new Date(), resumeStrandedQueue: true },
async () => { throw new Error("in-flight work must not drain"); },
);
expect(rows).toHaveLength(1);
expect((rows[0][0] as typeof heartbeatRuns.$inferSelect).resultJson).toEqual(result);
});
it.each(["assignee", "mentioned"])("preserves a paused recipient's accepted message until resume (%s)", async kind => {
const companyId = await seedCompany();
const ownerId = await seedAgent({ companyId });