mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
Merge current master into portable controller runtime
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
d6351cc079
7 files changed
+388
-25
No files matched your search
@@ -68,6 +68,13 @@ Runner profiles retain the adapter type and complete native execution input and
|
||||
workspace-sync receipts. Full result evidence stays available to the existing
|
||||
retry and workspace checks. Stored context, profiles, and output are unchanged.
|
||||
|
||||
Conversation ownership checks select process identity and full recovery result
|
||||
evidence without execution prompts or profiles. Committed native reply repair
|
||||
selects its persisted result and identity bindings plus typed presentation flags.
|
||||
It keeps its existing row locks, accepted-result digest checks, publication
|
||||
authorization, and selected-answer protections. Full execution history stays in
|
||||
storage.
|
||||
|
||||
Run-context lookups use expression indexes for both `context_snapshot.issueId`
|
||||
and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own
|
||||
company-scoped expression index that follows the query's native issue ID and
|
||||
|
||||
@@ -791,6 +791,62 @@ describeEmbeddedPostgres("issue list routes assigneeAgentId filter", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
"projectId",
|
||||
"parentId",
|
||||
"parentIssueId",
|
||||
"workspaceId",
|
||||
"executionWorkspaceId",
|
||||
"descendantOf",
|
||||
"labelId",
|
||||
"participantAgentId",
|
||||
])("returns 422 instead of a DB error for malformed %s filters", async (param) => {
|
||||
const companyId = randomUUID();
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: uniqueIssuePrefix(),
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await seedCloudTenantMember(companyId);
|
||||
|
||||
const app = createApp(companyId);
|
||||
const res = await request(app)
|
||||
.get(`/api/companies/${companyId}/issues`)
|
||||
.query({ status: "todo", [param]: "d", limit: "20" });
|
||||
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(422);
|
||||
expect(res.body).toMatchObject({ error: `${param} must be a UUID` });
|
||||
});
|
||||
|
||||
it.each([
|
||||
"projectId",
|
||||
"parentId",
|
||||
"parentIssueId",
|
||||
"workspaceId",
|
||||
"executionWorkspaceId",
|
||||
"descendantOf",
|
||||
"labelId",
|
||||
"participantAgentId",
|
||||
])("returns 422 from issues/count for malformed %s filters", async (param) => {
|
||||
const companyId = randomUUID();
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: uniqueIssuePrefix(),
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await seedCloudTenantMember(companyId);
|
||||
|
||||
const app = createApp(companyId);
|
||||
const res = await request(app)
|
||||
.get(`/api/companies/${companyId}/issues/count`)
|
||||
.query({ attention: "blocked", [param]: "d" });
|
||||
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(422);
|
||||
expect(res.body).toMatchObject({ error: `${param} must be a UUID` });
|
||||
});
|
||||
|
||||
it("returns opt-in live descendant counts for offscreen live descendants only", async () => {
|
||||
const companyId = randomUUID();
|
||||
const otherCompanyId = randomUUID();
|
||||
|
||||
+65
-20
@@ -3190,6 +3190,42 @@ function decrementIssueListActorClientInflight(actorClientKey: string) {
|
||||
else issueListActorClientInflight.set(actorClientKey, next);
|
||||
}
|
||||
|
||||
const ISSUE_LIST_UUID_QUERY_PARAMS = [
|
||||
"participantAgentId",
|
||||
"projectId",
|
||||
"workspaceId",
|
||||
"executionWorkspaceId",
|
||||
"parentId",
|
||||
"parentIssueId",
|
||||
"descendantOf",
|
||||
"createdFromIssueId",
|
||||
"labelId",
|
||||
] as const;
|
||||
|
||||
type IssueListUuidQueryParam = (typeof ISSUE_LIST_UUID_QUERY_PARAMS)[number];
|
||||
|
||||
function parseIssueListUuidQueryParams(
|
||||
query: Record<string, unknown>,
|
||||
):
|
||||
| { ok: true; values: Partial<Record<IssueListUuidQueryParam, string>> }
|
||||
| { ok: false; error: string } {
|
||||
const values: Partial<Record<IssueListUuidQueryParam, string>> = {};
|
||||
for (const name of ISSUE_LIST_UUID_QUERY_PARAMS) {
|
||||
const raw = query[name];
|
||||
if (raw === undefined) continue;
|
||||
if (typeof raw !== "string") {
|
||||
return { ok: false, error: `${name} must be a UUID` };
|
||||
}
|
||||
const trimmed = raw.trim();
|
||||
if (trimmed.length === 0) continue;
|
||||
if (!isUuidLike(trimmed)) {
|
||||
return { ok: false, error: `${name} must be a UUID` };
|
||||
}
|
||||
values[name] = trimmed;
|
||||
}
|
||||
return { ok: true, values };
|
||||
}
|
||||
|
||||
async function coordinateIssueListGet(input: {
|
||||
req: Request;
|
||||
companyId: string;
|
||||
@@ -8445,6 +8481,12 @@ export function issueRoutes(
|
||||
});
|
||||
return;
|
||||
}
|
||||
const uuidQuery = parseIssueListUuidQueryParams(req.query);
|
||||
if (!uuidQuery.ok) {
|
||||
res.status(422).json({ error: uuidQuery.error });
|
||||
return;
|
||||
}
|
||||
const uuidFilters = uuidQuery.values;
|
||||
const offset = parsedOffset ?? 0;
|
||||
|
||||
const listFilters: IssueFilters = {
|
||||
@@ -8452,20 +8494,18 @@ export function issueRoutes(
|
||||
attention: attention === "blocked" ? "blocked" : undefined,
|
||||
status: req.query.status as string | string[] | undefined,
|
||||
assigneeAgentId,
|
||||
participantAgentId: req.query.participantAgentId as string | undefined,
|
||||
participantAgentId: uuidFilters.participantAgentId,
|
||||
assigneeUserId,
|
||||
touchedByUserId,
|
||||
inboxArchivedByUserId,
|
||||
unreadForUserId,
|
||||
projectId: req.query.projectId as string | undefined,
|
||||
workspaceId: req.query.workspaceId as string | undefined,
|
||||
executionWorkspaceId: req.query.executionWorkspaceId as
|
||||
string | undefined,
|
||||
parentId: (req.query.parentId ?? req.query.parentIssueId) as
|
||||
string | undefined,
|
||||
descendantOf: req.query.descendantOf as string | undefined,
|
||||
createdFromIssueId: req.query.createdFromIssueId as string | undefined,
|
||||
labelId: req.query.labelId as string | undefined,
|
||||
projectId: uuidFilters.projectId,
|
||||
workspaceId: uuidFilters.workspaceId,
|
||||
executionWorkspaceId: uuidFilters.executionWorkspaceId,
|
||||
parentId: uuidFilters.parentId ?? uuidFilters.parentIssueId,
|
||||
descendantOf: uuidFilters.descendantOf,
|
||||
createdFromIssueId: uuidFilters.createdFromIssueId,
|
||||
labelId: uuidFilters.labelId,
|
||||
originKind: req.query.originKind as string | undefined,
|
||||
originKindPrefix: req.query.originKindPrefix as string | undefined,
|
||||
originId: req.query.originId as string | undefined,
|
||||
@@ -8686,22 +8726,27 @@ export function issueRoutes(
|
||||
return;
|
||||
}
|
||||
|
||||
const countUuidQuery = parseIssueListUuidQueryParams(req.query);
|
||||
if (!countUuidQuery.ok) {
|
||||
res.status(422).json({ error: countUuidQuery.error });
|
||||
return;
|
||||
}
|
||||
const countUuidFilters = countUuidQuery.values;
|
||||
|
||||
const blockedCountFilters = {
|
||||
readCondition: await issueReadSqlCondition(db, req.actor),
|
||||
attention: "blocked",
|
||||
status: req.query.status as string | string[] | undefined,
|
||||
assigneeAgentId: req.query.assigneeAgentId as string | undefined,
|
||||
participantAgentId: req.query.participantAgentId as string | undefined,
|
||||
participantAgentId: countUuidFilters.participantAgentId,
|
||||
assigneeUserId: req.query.assigneeUserId as string | undefined,
|
||||
projectId: req.query.projectId as string | undefined,
|
||||
workspaceId: req.query.workspaceId as string | undefined,
|
||||
executionWorkspaceId: req.query.executionWorkspaceId as
|
||||
string | undefined,
|
||||
parentId: (req.query.parentId ?? req.query.parentIssueId) as
|
||||
string | undefined,
|
||||
descendantOf: req.query.descendantOf as string | undefined,
|
||||
createdFromIssueId: req.query.createdFromIssueId as string | undefined,
|
||||
labelId: req.query.labelId as string | undefined,
|
||||
projectId: countUuidFilters.projectId,
|
||||
workspaceId: countUuidFilters.workspaceId,
|
||||
executionWorkspaceId: countUuidFilters.executionWorkspaceId,
|
||||
parentId: countUuidFilters.parentId ?? countUuidFilters.parentIssueId,
|
||||
descendantOf: countUuidFilters.descendantOf,
|
||||
createdFromIssueId: countUuidFilters.createdFromIssueId,
|
||||
labelId: countUuidFilters.labelId,
|
||||
originKind: req.query.originKind as string | undefined,
|
||||
originKindPrefix: req.query.originKindPrefix as string | undefined,
|
||||
originId: req.query.originId as string | undefined,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
@@ -237,6 +237,12 @@ if (!support.supported) console.warn(`Monitor publication database unavailable:
|
||||
|
||||
it("recovers the accepted continuation explanation after interrupted finalization without another provider result", async () => {
|
||||
const f = await fixture();
|
||||
const executionPayload = randomBytes(128 * 1024).toString("base64");
|
||||
await db.update(heartbeatRuns).set({
|
||||
contextSnapshot: { ...f.source.contextSnapshot, largePrompt: executionPayload,
|
||||
paperclipWake: { prompt: executionPayload } },
|
||||
runnerProfileJson: { instructions: executionPayload }, stdoutExcerpt: executionPayload,
|
||||
}).where(eq(heartbeatRuns.id, f.source.id));
|
||||
const [contract] = await db.insert(completionContracts).values({ companyId: f.company.id, issueId: f.issue.id,
|
||||
revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
|
||||
completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
|
||||
@@ -270,6 +276,9 @@ if (!support.supported) console.warn(`Monitor publication database unavailable:
|
||||
expect(coordinator).toMatchObject({ phase: "committed" });
|
||||
const [settledRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.source.id));
|
||||
expect(settledRun).toMatchObject({ status: "succeeded", resultJson: { finalizationPhase: "committed" } });
|
||||
expect(settledRun.contextSnapshot?.largePrompt).toBe(executionPayload);
|
||||
expect(settledRun.runnerProfileJson?.instructions).toBe(executionPayload);
|
||||
expect(settledRun.stdoutExcerpt).toBe(executionPayload);
|
||||
const priorComments = await db.select({ body: issueComments.body }).from(issueComments).where(eq(issueComments.createdByRunId, f.source.id));
|
||||
expect(priorComments).toEqual([{ body: summary }]);
|
||||
expect(await repair()).toBe(false);
|
||||
|
||||
@@ -111,7 +111,14 @@ export async function getConversationOwnershipBlocker(db: Db, companyId: string,
|
||||
and (${environmentLeases.releasedAt} is null
|
||||
or ${environmentLeases.status} = 'pending_cleanup'
|
||||
or ${environmentLeases.cleanupStatus} = 'failed'))`;
|
||||
const candidates = await db.select({ run: heartbeatRuns, activeLease }).from(heartbeatRuns)
|
||||
const candidates = await db.select({ run: {
|
||||
id: heartbeatRuns.id,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
processPid: heartbeatRuns.processPid,
|
||||
processGroupId: heartbeatRuns.processGroupId,
|
||||
processStartedAt: heartbeatRuns.processStartedAt,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
}, activeLease }).from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.runtimeMode, "legacy"),
|
||||
conversationRunPredicate(),
|
||||
|
||||
@@ -596,7 +596,7 @@ function recoveredExecutionFailureMetadata() {
|
||||
|
||||
async function projectCommittedRun(input: {
|
||||
db: Db;
|
||||
run: typeof heartbeatRuns.$inferSelect;
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status" | "finishedAt" | "runtimeMode" | "nativePhase">;
|
||||
coordinator: typeof nativeRunFinalizations.$inferSelect;
|
||||
}) {
|
||||
if (!input.coordinator.resultId) return;
|
||||
@@ -764,7 +764,9 @@ async function materializeCommittedReviewResponse(db: Db, runId: string) {
|
||||
|
||||
async function acceptedResponseDigestMatches(
|
||||
db: Db,
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "agentId" | "nativeIssueId" | "nativeSessionId" |
|
||||
"completionContractId" | "completionContractSha256" | "runnerInstanceId">,
|
||||
accepted: typeof nativeRunResults.$inferSelect,
|
||||
): Promise<boolean> {
|
||||
const envelope = record(accepted.resultJson);
|
||||
@@ -842,7 +844,30 @@ export async function repairCommittedNativeChatResponse(
|
||||
.for("update");
|
||||
if (!issue) return false;
|
||||
const [run] = await tx
|
||||
.select()
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
nativeSessionId: heartbeatRuns.nativeSessionId,
|
||||
completionContractId: heartbeatRuns.completionContractId,
|
||||
completionContractSha256: heartbeatRuns.completionContractSha256,
|
||||
runnerInstanceId: heartbeatRuns.runnerInstanceId,
|
||||
runtimeMode: heartbeatRuns.runtimeMode,
|
||||
status: heartbeatRuns.status,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
nativePhase: heartbeatRuns.nativePhase,
|
||||
finishedAt: heartbeatRuns.finishedAt,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
contextSnapshot: sql<Record<string, unknown> | null>`case when ${heartbeatRuns.contextSnapshot} is null then null
|
||||
else jsonb_build_object(
|
||||
'skipIssueComment', ${heartbeatRuns.contextSnapshot}->'skipIssueComment',
|
||||
'source', ${heartbeatRuns.contextSnapshot}->'source',
|
||||
'externalChatContinuation', ${heartbeatRuns.contextSnapshot}->'externalChatContinuation',
|
||||
'paperclipWake', jsonb_build_object('externalInteractionContinuation',
|
||||
${heartbeatRuns.contextSnapshot} #> '{paperclipWake,externalInteractionContinuation}')
|
||||
) end`,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
|
||||
@@ -0,0 +1,214 @@
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import { agents, companies, completionContracts, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { eq, type SQL } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { NativeRunCoordinatorStore } from "./native-runtime/native-run-coordinator-store.js";
|
||||
import { getConversationOwnershipBlocker } from "./conversation-continuation.js";
|
||||
import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js";
|
||||
import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read projections", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("presentation-ownership-read-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 60_000);
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
|
||||
|
||||
async function fixture(patch: Partial<typeof heartbeatRuns.$inferInsert> = {}) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Read fixture", issuePrefix: companyId.slice(0, 8) });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Read agent", adapterType: "codex_local" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Read task", status: "blocked", assigneeAgentId: agentId });
|
||||
const payload = randomBytes(128 * 1024).toString("base64");
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: issueId,
|
||||
status: "failed", runtimeMode: "legacy", finishedAt: new Date("2020-01-01T00:00:00Z"),
|
||||
contextSnapshot: { issueId, source: "chat:fixture", prompt: payload,
|
||||
paperclipWake: { externalInteractionContinuation: true, prompt: payload } },
|
||||
runnerProfileJson: { adapterDispatch: { adapterType: "codex_local", instructions: payload }, nativeExecutionInput: { prompt: payload } },
|
||||
resultJson: { unknownFutureRecoveryReceipt: { preserved: true } }, stdoutExcerpt: payload, stderrExcerpt: payload,
|
||||
...patch,
|
||||
}).returning();
|
||||
return { companyId, agentId, issueId, run };
|
||||
}
|
||||
|
||||
function captureReads(fullRows = false) {
|
||||
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 selection = fields[0];
|
||||
const nested = selection && "run" in selection && "activeLease" in selection;
|
||||
const repairSource = selection && "nativeSessionId" in selection && "completionContractId" in selection && "finishedAt" in selection;
|
||||
const sourceRead = args[0] === heartbeatRuns && (nested || repairSource);
|
||||
const query = fullRows && sourceRead
|
||||
? (nested ? target.select({ run: heartbeatRuns, activeLease: selection.activeLease as SQL }) : target.select()).from(heartbeatRuns)
|
||||
: from(...args);
|
||||
if (sourceRead) {
|
||||
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 root = recordSelects(db);
|
||||
const transaction: Db["transaction"] = (callback, config) => db.transaction(tx => callback(recordSelects(tx)), config);
|
||||
const recorded = new Proxy(root, { get(target, key) { return key === "transaction" ? transaction : Reflect.get(target, key); } });
|
||||
return { recorded, rows };
|
||||
}
|
||||
|
||||
async function compareOwnership(f: Awaited<ReturnType<typeof fixture>>) {
|
||||
const baseline = captureReads(true), projected = captureReads();
|
||||
const expected = await getConversationOwnershipBlocker(baseline.recorded, f.companyId, f.issueId);
|
||||
const actual = await getConversationOwnershipBlocker(projected.recorded, f.companyId, f.issueId);
|
||||
expect(actual).toEqual(expected);
|
||||
expect(projected.rows.flat()).toHaveLength(1);
|
||||
expect(Buffer.byteLength(JSON.stringify(projected.rows))).toBeLessThan(4096);
|
||||
expect(Buffer.byteLength(JSON.stringify(baseline.rows))).toBeGreaterThan(128 * 1024);
|
||||
return actual;
|
||||
}
|
||||
|
||||
it("keeps current process authority without reading execution payloads", async () => {
|
||||
const f = await fixture({ processPid: process.pid });
|
||||
expect(await compareOwnership(f)).toMatchObject({ runId: f.run.id, cause: "execution_owner_active" });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(f.run);
|
||||
expect(await getConversationOwnershipBlocker(db, randomUUID(), f.issueId)).toBeNull();
|
||||
await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
});
|
||||
|
||||
it("retains an active lease until the current database row releases it", async () => {
|
||||
const f = await fixture();
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.run.id,
|
||||
issueId: f.issueId, provider: "daytona", providerLeaseId: "fixture-lease", status: "active" }).returning();
|
||||
expect(await compareOwnership(f)).toMatchObject({ runId: f.run.id, cause: "execution_owner_active" });
|
||||
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0]).toEqual(lease);
|
||||
await db.update(environmentLeases).set({ status: "released", releasedAt: new Date("2020-01-01T00:00:00Z") }).where(eq(environmentLeases.id, lease.id));
|
||||
expect(await getConversationOwnershipBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps unconfirmed remote cleanup and all recovery evidence", async () => {
|
||||
const f = await fixture();
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.run.id,
|
||||
issueId: f.issueId, provider: "daytona", providerLeaseId: "retained-fixture", status: "released",
|
||||
releasedAt: new Date("2020-01-01T00:00:00Z"), cleanupStatus: "success" }).returning();
|
||||
const result = { ...f.run.resultJson, workspaceRestoreFailure: "restore_unsafe_archive",
|
||||
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [lease.id] } };
|
||||
await db.update(heartbeatRuns).set({ resultJson: result }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
expect(await compareOwnership(f)).toMatchObject({ runId: f.run.id, cause: "execution_owner_active" });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].resultJson).toEqual(result);
|
||||
});
|
||||
|
||||
async function compareRepair(f: Awaited<ReturnType<typeof fixture>>) {
|
||||
const baseline = captureReads(true), projected = captureReads();
|
||||
const input = { companyId: f.companyId, issueId: f.issueId, runId: f.run.id };
|
||||
expect(await repairCommittedNativeChatResponse(projected.recorded, input))
|
||||
.toBe(await repairCommittedNativeChatResponse(baseline.recorded, input));
|
||||
expect(projected.rows.flat()).toHaveLength(1);
|
||||
expect(Buffer.byteLength(JSON.stringify(projected.rows))).toBeLessThan(4096);
|
||||
expect(Buffer.byteLength(JSON.stringify(baseline.rows))).toBeGreaterThan(128 * 1024);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(f.run);
|
||||
return projected.rows.flat()[0] as { contextSnapshot: Record<string, unknown> | null; resultJson: unknown };
|
||||
}
|
||||
|
||||
it("preserves selected answers without downloading prompts, profiles, or excerpts", async () => {
|
||||
const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: {
|
||||
presentationDecision: { commentId: randomUUID() }, unknownFutureReceipt: { preserved: true } } });
|
||||
const row = await compareRepair(f);
|
||||
expect(row.resultJson).toEqual(f.run.resultJson);
|
||||
expect(await repairCommittedNativeChatResponse(db, { companyId: randomUUID(), issueId: f.issueId, runId: f.run.id })).toBe(false);
|
||||
const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task" }).returning();
|
||||
expect(await repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: other.id, runId: f.run.id })).toBe(false);
|
||||
});
|
||||
|
||||
async function unpublishedAcceptedReply() {
|
||||
const payload = randomBytes(128 * 1024).toString("base64");
|
||||
const f = await fixture({ runtimeMode: "native", status: "running", finishedAt: null,
|
||||
contextSnapshot: { source: "internal", prompt: payload, paperclipWake: { prompt: payload } } });
|
||||
const [contract] = await db.insert(completionContracts).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
|
||||
completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
|
||||
contractJson: { revision: "test", objective: "Finish the fixture task", criteria: [] },
|
||||
canonicalSha256: "read-fixture-contract", createdByActorType: "system", createdByActorId: "test" }).returning();
|
||||
const sessionId = randomUUID(), runnerId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ nativeSessionId: sessionId, runnerInstanceId: runnerId,
|
||||
completionContractId: contract.id, completionContractSha256: contract.canonicalSha256 }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await db.update(issues).set({ status: "in_progress", executionRunId: f.run.id }).where(eq(issues.id, f.issueId));
|
||||
const port = new NativeRunCoordinatorStore(db, { companyId: f.companyId, issueId: f.issueId,
|
||||
runId: f.run.id, agentId: f.agentId, normalizedSessionId: sessionId, completionContractId: contract.id,
|
||||
completionContractSha256: contract.canonicalSha256, runnerSourceInstanceId: runnerId,
|
||||
completionContractRevision: "test", completionContractCriterionIds: [] });
|
||||
const reply = "The accepted provider reply is preserved exactly.";
|
||||
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.run.id, agentId: f.agentId,
|
||||
seq: 1_000_000, eventType: "item.completed", payload: { prpEvent: { turnId: "reply-turn", sourceEventId: "fixture-final-answer",
|
||||
payload: { kind: "agentMessage", channel: "final", text: reply } } } });
|
||||
await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
|
||||
summary: "The accepted summary differs from its final answer.", completionClaim: { contractRevision: "test",
|
||||
objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] },
|
||||
terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" },
|
||||
turnId: "reply-turn" });
|
||||
await finalizeNativeRun({ db, runId: f.run.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)))[0]).toMatchObject({ phase: "committed" });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:00:00Z") }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id))).toEqual([]);
|
||||
const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
|
||||
expect(before.resultJson?.presentationDecision).toBeUndefined();
|
||||
return { ...f, before, reply };
|
||||
}
|
||||
|
||||
it("publishes the same accepted reply and decision from equivalent full-row and projected starting states", async () => {
|
||||
const outcomes: unknown[] = [];
|
||||
for (const fullRows of [true, false]) {
|
||||
const f = await unpublishedAcceptedReply(), captured = captureReads(fullRows);
|
||||
const input = { companyId: f.companyId, issueId: f.issueId, runId: f.run.id };
|
||||
expect(await repairCommittedNativeChatResponse(captured.recorded, input)).toBe(true);
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id));
|
||||
expect(comments).toHaveLength(1); expect(comments[0].body).toBe(f.reply);
|
||||
const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
|
||||
const decision = after.resultJson!.presentationDecision as Record<string, unknown>;
|
||||
expect(decision.commentId).toBe(comments[0].id);
|
||||
const { commentId: _, ...stableDecision } = decision;
|
||||
outcomes.push({ body: comments[0].body, decision: stableDecision });
|
||||
expect(after.resultJson).toMatchObject(f.before.resultJson!);
|
||||
expect(after.contextSnapshot).toEqual(f.before.contextSnapshot);
|
||||
expect(after.runnerProfileJson).toEqual(f.before.runnerProfileJson);
|
||||
expect(after.stdoutExcerpt).toBe(f.before.stdoutExcerpt);
|
||||
expect(after.stderrExcerpt).toBe(f.before.stderrExcerpt);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.run.id))).toHaveLength(1);
|
||||
expect(await repairCommittedNativeChatResponse(db, input)).toBe(false);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id))).toHaveLength(1);
|
||||
expect(captured.rows.flat()).toHaveLength(1);
|
||||
if (fullRows) expect(Buffer.byteLength(JSON.stringify(captured.rows))).toBeGreaterThan(128 * 1024);
|
||||
else expect(Buffer.byteLength(JSON.stringify(captured.rows))).toBeLessThan(4096);
|
||||
}
|
||||
expect(outcomes[1]).toEqual(outcomes[0]);
|
||||
});
|
||||
|
||||
it.each([null, [], "scalar", 42, { source: 42, skipIssueComment: "true", paperclipWake: [] },
|
||||
{ source: "issue.comment.created", externalChatContinuation: true },
|
||||
{ source: "chat:fixture", skipIssueComment: true, paperclipWake: { externalInteractionContinuation: "true" } },
|
||||
{ source: "internal", paperclipWake: { externalInteractionContinuation: true } },
|
||||
].map(context => ({ context })))("preserves typed publication flags for $context", async ({ context }) => {
|
||||
const f = await fixture({ runtimeMode: "native", status: "succeeded", contextSnapshot: context as Record<string, unknown> | null });
|
||||
const row = await compareRepair(f);
|
||||
expect(isExternalChatPresentationContext(row.contextSnapshot)).toBe(isExternalChatPresentationContext(context));
|
||||
const original = context && typeof context === "object" && !Array.isArray(context) ? context : {};
|
||||
expect(row.contextSnapshot?.skipIssueComment === true).toBe((original as Record<string, unknown>).skipIssueComment === true);
|
||||
});
|
||||
|
||||
it.each([null, ["stdout", { receipt: true }], "scalar", 42].map(result => ({ result })))("preserves non-object result evidence $result", async ({ result }) => {
|
||||
const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: result as Record<string, unknown> | null });
|
||||
expect((await compareRepair(f)).resultJson).toEqual(result);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user