mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
Merge branch 'codex/sleep-stack-04' into codex/sleep-09-aggregate
* codex/sleep-stack-04: fix(ui): supersede startup jobs when newer edits are admitted fix(ui): retain startup autosave through session recovery fix(ui): retain startup autosave through session recovery fix(server): bound and coalesce issue retention invalidations perf(runner): cache verified native executable digests (#18390) fix(server): reconcile failed sandbox cleanup with signed absence (#18407) Reuse signed master images for npm canaries (#18404) perf(server): overlap independent heartbeat preparation reads (#18372) test(e2e): add a portable active follow-up journey (#18345) fix(server): preserve budget history for deleted resources (#18389)
This commit is contained in:
Dotta committed
commit
7add4d4c5c
57 files changed
+2024
-361
No files matched your search
@@ -1,5 +1,6 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { and, eq, sql, type SQL } from "drizzle-orm";
|
||||
import { PgDialect } from "drizzle-orm/pg-core";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox,
|
||||
decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
|
||||
@@ -425,6 +426,106 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("bounds high-fanin ordinary issue writes to one coalescible hint", async () => {
|
||||
const service = attentionRetentionService(db), original = service.load, load = vi.spyOn(service, "load");
|
||||
await start(service); await settle(); load.mockClear();
|
||||
const leaf = await insertIssue("in_review", { updatedAt: new Date(), assigneeUserId: "board" });
|
||||
await db.execute(sql`with parents as (
|
||||
insert into issues (id, company_id, title, status)
|
||||
select gen_random_uuid(), ${companyId}::uuid, 'fan-in-' || n, 'todo'
|
||||
from generate_series(1, 2100) n returning id)
|
||||
insert into issue_relations (company_id, issue_id, related_issue_id, type)
|
||||
select ${companyId}::uuid, ${leaf.id}::uuid, id, 'blocks' from parents`);
|
||||
const hints: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => {
|
||||
if (event === "settled") hints.push(...values ?? []);
|
||||
});
|
||||
let release!: () => void, entered = false;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
load.mockImplementationOnce(async (...args) => { const page = await original(...args); entered = true; await gate; return page; });
|
||||
statements.length = 0;
|
||||
try {
|
||||
await issueService(db).update(leaf.id, { title: "High fan-in ordinary edit 0" });
|
||||
await vi.waitFor(() => expect(entered).toBe(true), { timeout: 15_000 });
|
||||
for (let n = 1; n < 8; n++) await issueService(db).update(leaf.id, { title: `High fan-in ordinary edit ${n}` });
|
||||
} finally { release(); unsubscribe(); }
|
||||
expect(statements.filter(query => query.includes("with recursive affected"))).toHaveLength(8);
|
||||
expect(hints).toHaveLength(8);
|
||||
expect(new Set(hints)).toEqual(new Set([JSON.stringify({ companyId })]));
|
||||
await settle();
|
||||
expect(load.mock.calls.length).toBeLessThanOrEqual(3);
|
||||
expect(load.mock.calls.every(([scope]) => scope?.companyId === companyId && !scope.issueId && !scope.issueIds)).toBe(true);
|
||||
let query!: Parameters<typeof db.execute>[0], returned = 0;
|
||||
await db.transaction(async tx => {
|
||||
const execute = tx.execute.bind(tx);
|
||||
vi.spyOn(tx, "execute").mockImplementation(async statement => {
|
||||
const result = await execute(statement);
|
||||
if (new PgDialect().sqlToQuery(statement as SQL).sql.includes("with recursive affected")) {
|
||||
query = statement; returned = result.length;
|
||||
}
|
||||
return result;
|
||||
});
|
||||
await notifyIssueAttentionRetentionWork(tx, { companyId, issueId: leaf.id });
|
||||
});
|
||||
expect(returned).toBe(101);
|
||||
const plans = await db.execute(sql`explain (analyze, format json) ${query as SQL}`);
|
||||
const plan = (plans[0]["QUERY PLAN"] as any)[0].Plan;
|
||||
expect(plan["Node Type"]).toBe("Limit");
|
||||
expect(plan["Actual Rows"]).toBe(101);
|
||||
const nodes: any[] = [], collect = (node: any) => { nodes.push(node); for (const child of node.Plans ?? []) collect(child); };
|
||||
collect(plan);
|
||||
const recursive = nodes.filter(node => node["Node Type"] === "Recursive Union");
|
||||
const relations = nodes.filter(node => node["Relation Name"] === "issue_relations");
|
||||
expect(recursive).toHaveLength(1);
|
||||
expect(recursive[0]["Actual Rows"]).toBe(101);
|
||||
expect(relations.length).toBeGreaterThan(0);
|
||||
expect(relations.every(node => node["Actual Rows"] * node["Actual Loops"] <= 101)).toBe(true);
|
||||
process.stdout.write(JSON.stringify({ reverseFanInPlan: plan }) + "\n");
|
||||
});
|
||||
|
||||
it("retains far ancestor deadlines when a cyclic chain exceeds the precise hint budget", async () => {
|
||||
const service = attentionRetentionService(db), load = vi.spyOn(service, "load");
|
||||
await start(service); await settle(); load.mockClear();
|
||||
await db.execute(sql`insert into issues (id, company_id, title, status, updated_at, assignee_user_id)
|
||||
select gen_random_uuid(), ${companyId}::uuid, 'chain-' || n,
|
||||
case when n = 150 then 'in_review' else 'todo' end,
|
||||
case when n = 150 then ${new Date(Date.now() - delay + 15_000).toISOString()}::timestamptz else now() end,
|
||||
case when n = 150 then 'board' else null end
|
||||
from generate_series(0, 150) n`);
|
||||
await db.execute(sql`update issues child set parent_id = parent.id from issues parent
|
||||
where child.company_id = ${companyId}::uuid and parent.company_id = child.company_id
|
||||
and parent.title = 'chain-' || ((substring(child.title from 7)::integer + 1) % 151)`);
|
||||
const [leaf, far] = await Promise.all([db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.title, "chain-0"))).then(rows => rows[0]),
|
||||
db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.title, "chain-150"))).then(rows => rows[0])]);
|
||||
await issueService(db).update(leaf.id, { title: "Ordinary deep chain edit" });
|
||||
await settle();
|
||||
expect(load.mock.calls.some(([scope]) => scope?.companyId === companyId && !scope.issueId && !scope.issueIds)).toBe(true);
|
||||
expect(await state("review", far.id)).toBeNull();
|
||||
expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
await archived("review", far.id);
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it("canonicalizes a complete cyclic affected set into one bounded batch hint", async () => {
|
||||
const a = await insertIssue(), b = await insertIssue(), c = await insertIssue();
|
||||
await db.update(issues).set({ parentId: b.id }).where(eq(issues.id, a.id));
|
||||
await db.update(issues).set({ parentId: c.id }).where(eq(issues.id, b.id));
|
||||
await db.update(issues).set({ parentId: a.id }).where(eq(issues.id, c.id));
|
||||
const hints: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => {
|
||||
if (event === "settled") hints.push(...values ?? []);
|
||||
});
|
||||
try {
|
||||
await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, issueId: a.id }));
|
||||
await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, issueId: c.id }));
|
||||
const wide = { companyId, issueIds: [b.id, a.id, c.id, a.id], ignored: {} as unknown };
|
||||
wide.ignored = wide;
|
||||
await db.transaction(tx => notifyAttentionRetentionWork(tx, wide));
|
||||
} finally { unsubscribe(); }
|
||||
expect(hints).toHaveLength(3);
|
||||
expect(new Set(hints)).toEqual(new Set([JSON.stringify({ companyId, issueIds: [a.id, b.id, c.id].sort() })]));
|
||||
});
|
||||
|
||||
it("canonicalizes wide circular notification variables to the supported identity fields", async () => {
|
||||
const sourceId = randomUUID(), issueId = randomUUID(), hints: string[] = [];
|
||||
const wide = { companyId, sourceKind: "review" as const, sourceId, issueId, ignored: {} as unknown };
|
||||
@@ -487,14 +588,14 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
const terminal = await insertIssue("backlog", { parentId: oldParent.id, hiddenAt: old(), assigneeAgentId: agentId, responsibleUserId: "board" });
|
||||
const service = attentionRetentionService(db), load = vi.spyOn(service, "load"); await start(service); await settle(); load.mockClear();
|
||||
await issueService(db).update(terminal.id, { parentId: newParent.id }); await settle();
|
||||
const ids = load.mock.calls.map(([scope]) => scope?.issueId);
|
||||
const ids = load.mock.calls.flatMap(([scope]) => scope?.issueIds ?? (scope?.issueId ? [scope.issueId] : []));
|
||||
expect(ids).toEqual(expect.arrayContaining([oldParent.id, newParent.id, terminal.id]));
|
||||
load.mockClear();
|
||||
await issueService(db).update(terminal.id, { hiddenAt: null }); await settle();
|
||||
expect(await state("blocker_attention", terminal.id)).toBeNull();
|
||||
const visible = await service.load({ companyId, issueId: newParent.id });
|
||||
expect(visible.rows!.some(entry => entry.sourceId === terminal.id && entry.dueAt > Date.now())).toBe(true);
|
||||
expect(load.mock.calls.some(([scope]) => scope?.issueId === newParent.id)).toBe(true);
|
||||
expect(load.mock.calls.some(([scope]) => scope?.issueId === newParent.id || scope?.issueIds?.includes(newParent.id))).toBe(true);
|
||||
const next = await insertIssue("backlog", { assigneeAgentId: agentId });
|
||||
await issueService(db).update(oldParent.id, { blockedByIssueIds: [next.id] }); await archived("blocker_attention", next.id);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, approvals, budgetIncidents, budgetPolicies, companies, costEvents, createDb, projects, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
vi.mock("../services/activity-log.js", () => ({ logActivity: vi.fn() }));
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
|
||||
describe.skipIf(!support.supported)("budget history for deleted scopes", () => {
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-budget-history-");
|
||||
db = createDb(database.connectionString);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterEach(async () => {
|
||||
await db.delete(budgetIncidents); await db.delete(approvals); await db.delete(budgetPolicies); await db.delete(costEvents);
|
||||
await db.delete(projects); await db.delete(agents); await db.delete(companies);
|
||||
});
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
|
||||
|
||||
async function fixture() {
|
||||
const companyId = randomUUID(), agentId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "History company", issuePrefix: `B${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Current agent", role: "engineer" });
|
||||
const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType: "agent", scopeId: agentId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc", amount: 300 }).returning();
|
||||
await db.insert(costEvents).values({ companyId, agentId, provider: "openai", model: "history-fixture", inputTokens: 1, outputTokens: 1, costCents: 17, occurredAt: new Date() });
|
||||
return { companyId, agentId, policy: policy! };
|
||||
}
|
||||
|
||||
it.each(["agent", "project"] as const)("keeps current budgets and deleted %s policy/incident history readable without changing stored rows", async scopeType => {
|
||||
const { companyId, policy } = await fixture(), deletedId = randomUUID();
|
||||
if (scopeType === "agent") await db.insert(agents).values({ id: deletedId, companyId, name: "Former agent", role: "engineer" });
|
||||
else await db.insert(projects).values({ id: deletedId, companyId, name: "Former project" });
|
||||
const [historical] = await db.insert(budgetPolicies).values({ companyId, scopeType, scopeId: deletedId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc", amount: 200 }).returning();
|
||||
await db.insert(budgetIncidents).values({ companyId, policyId: historical!.id, scopeType, scopeId: deletedId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc", windowStart: new Date("2020-01-01"), windowEnd: new Date("2020-02-01"),
|
||||
thresholdType: "hard", amountLimit: 200, amountObserved: 201 });
|
||||
if (scopeType === "agent") await db.delete(agents).where(eq(agents.id, deletedId));
|
||||
else await db.delete(projects).where(eq(projects.id, deletedId));
|
||||
const policiesBefore = await db.select().from(budgetPolicies), incidentsBefore = await db.select().from(budgetIncidents), costsBefore = await db.select().from(costEvents);
|
||||
|
||||
const overview = await budgetService(db).overview(companyId);
|
||||
expect(overview.policies).toHaveLength(2);
|
||||
expect(overview.policies.find(row => row.policyId === policy.id)).toMatchObject({ companyId, scopeName: "Current agent", amount: 300, observedAmount: 17 });
|
||||
expect(overview.policies.find(row => row.policyId === historical!.id)).toMatchObject({ companyId, scopeType, scopeId: deletedId, scopeName: `Deleted ${scopeType}`, amount: 200 });
|
||||
expect(overview.activeIncidents).toMatchObject([{ companyId, policyId: historical!.id, scopeId: deletedId, scopeName: `Deleted ${scopeType}`, status: "open", amountObserved: 201 }]);
|
||||
expect(await db.select().from(budgetPolicies)).toEqual(policiesBefore);
|
||||
expect(await db.select().from(budgetIncidents)).toEqual(incidentsBefore);
|
||||
expect(await db.select().from(costEvents)).toEqual(costsBefore);
|
||||
});
|
||||
it("rejects missing-scope budget grants but can dismiss their historical incidents", async () => {
|
||||
const { companyId, agentId } = await fixture(), missingId = randomUUID(), service = budgetService(db);
|
||||
await expect(service.evaluateAdmission(companyId, missingId)).rejects.toThrow("Agent not found");
|
||||
await expect(service.evaluateAdmission(companyId, agentId, { projectId: missingId })).rejects.toThrow("Project not found");
|
||||
for (const scopeType of ["agent", "project"] as const) {
|
||||
await expect(service.upsertPolicy(companyId, { scopeType, scopeId: missingId, amount: 100 }, "board")).rejects.toThrow(`${scopeType === "agent" ? "Agent" : "Project"} not found`);
|
||||
const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType, scopeId: missingId,
|
||||
windowKind: "calendar_month_utc", amount: 100 }).returning();
|
||||
const [approval] = await db.insert(approvals).values({ companyId, type: "budget_override_required", payload: {} }).returning();
|
||||
const [incident] = await db.insert(budgetIncidents).values({ companyId, policyId: policy!.id, approvalId: approval!.id, scopeType, scopeId: missingId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc", windowStart: new Date("2020-01-01"), windowEnd: new Date("2020-02-01"),
|
||||
thresholdType: "hard", amountLimit: 100, amountObserved: 101 }).returning();
|
||||
const before = { policies: await db.select().from(budgetPolicies), incidents: await db.select().from(budgetIncidents),
|
||||
agents: await db.select().from(agents), costs: await db.select().from(costEvents), approvals: await db.select().from(approvals) };
|
||||
await expect(service.resolveIncident(companyId, incident!.id, { action: "raise_budget_and_resume", amount: 200 }, "board"))
|
||||
.rejects.toThrow(`${scopeType === "agent" ? "Agent" : "Project"} not found`);
|
||||
expect({ policies: await db.select().from(budgetPolicies), incidents: await db.select().from(budgetIncidents),
|
||||
agents: await db.select().from(agents), costs: await db.select().from(costEvents), approvals: await db.select().from(approvals) }).toEqual(before);
|
||||
const dismissed = await service.resolveIncident(companyId, incident!.id, { action: "keep_paused" }, "board");
|
||||
expect(dismissed).toMatchObject({ companyId, policyId: policy!.id, scopeId: missingId, scopeName: `Deleted ${scopeType}`, status: "dismissed" });
|
||||
const { status: _status, resolvedAt: _resolvedAt, updatedAt: _updatedAt, ...preserved } = incident!;
|
||||
const [saved] = await db.select().from(budgetIncidents).where(eq(budgetIncidents.id, incident!.id));
|
||||
expect(saved).toMatchObject({ ...preserved, status: "dismissed", resolvedAt: expect.any(Date) });
|
||||
expect((await db.select().from(approvals).where(eq(approvals.id, approval!.id)))[0]).toMatchObject({ status: "rejected" });
|
||||
expect(await db.select().from(budgetPolicies)).toEqual(before.policies);
|
||||
expect(await db.select().from(agents)).toEqual(before.agents);
|
||||
expect(await db.select().from(costEvents)).toEqual(before.costs);
|
||||
expect((await service.overview(companyId)).activeIncidents.some(row => row.id === incident!.id)).toBe(false);
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects a corrupted policy pointing to another company's scope instead of exposing its name", async () => {
|
||||
const { companyId } = await fixture(), foreignCompanyId = randomUUID(), foreignAgentId = randomUUID();
|
||||
await db.insert(companies).values({ id: foreignCompanyId, name: "Other company", issuePrefix: `F${foreignCompanyId.replace(/-/g, "").slice(0, 6).toUpperCase()}` });
|
||||
await db.insert(agents).values({ id: foreignAgentId, companyId: foreignCompanyId, name: "Private other-company agent", role: "engineer" });
|
||||
const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType: "agent", scopeId: foreignAgentId, windowKind: "calendar_month_utc", amount: 200 }).returning();
|
||||
const [incident] = await db.insert(budgetIncidents).values({ companyId, policyId: policy!.id, scopeType: "agent", scopeId: foreignAgentId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc", windowStart: new Date("2020-01-01"), windowEnd: new Date("2020-02-01"),
|
||||
thresholdType: "hard", amountLimit: 200, amountObserved: 201 }).returning();
|
||||
const service = budgetService(db);
|
||||
await expect(service.overview(companyId)).rejects.toThrow("Budget scope not found");
|
||||
for (const scopeId of [foreignAgentId, randomUUID()]) {
|
||||
await db.update(budgetIncidents).set({ scopeId }).where(eq(budgetIncidents.id, incident!.id));
|
||||
const before = { policies: await db.select().from(budgetPolicies), incidents: await db.select().from(budgetIncidents),
|
||||
agents: await db.select().from(agents), costs: await db.select().from(costEvents) };
|
||||
for (const action of ["raise_budget_and_resume", "keep_paused"] as const) {
|
||||
await expect(service.resolveIncident(companyId, incident!.id, { action, amount: 300 }, "board")).rejects.toThrow("Budget scope not found");
|
||||
expect({ policies: await db.select().from(budgetPolicies), incidents: await db.select().from(budgetIncidents),
|
||||
agents: await db.select().from(agents), costs: await db.select().from(costEvents) }).toEqual(before);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
});
|
||||
@@ -19,11 +19,20 @@ import { secretService } from "../services/secrets.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { aiConnectionService } from "../services/ai-connections.js";
|
||||
import * as aiRuntime from "../services/ai-connection-runtime.js";
|
||||
import * as connectorRuntime from "../services/connector-runtime.js";
|
||||
import * as assignedMcp from "../services/native-runtime/assigned-mcp-tools.js";
|
||||
import * as sentry from "../sentry.js";
|
||||
import { waitForPendingRunFailureReports } from "../services/run-failure-report.js";
|
||||
import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recovery-postgres.js";
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>(complete => { resolve = complete; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
const skillFailures = vi.hoisted(() => new Map<string, string>());
|
||||
const skillWaits = vi.hoisted(() => new Map<string, () => Promise<void>>());
|
||||
|
||||
describe("Runner home mode during recovery preparation", () => {
|
||||
it.each([
|
||||
@@ -52,6 +61,7 @@ vi.mock("../services/company-skills.js", async importOriginal => {
|
||||
return { ...service, withRuntimeSkillEntries: async <T>(companyId: string,
|
||||
options: Parameters<typeof service.withRuntimeSkillEntries>[1],
|
||||
consume: (entries: Awaited<ReturnType<typeof service.listRuntimeSkillEntries>>) => T | Promise<T>) => {
|
||||
await skillWaits.get(companyId)?.();
|
||||
const failure = skillFailures.get(companyId);
|
||||
if (failure) throw new Error(failure);
|
||||
return service.withRuntimeSkillEntries(companyId, options, consume);
|
||||
@@ -91,7 +101,7 @@ describePostgres("heartbeat execution configuration through adapter dispatch", (
|
||||
testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }),
|
||||
});
|
||||
}, 60_000);
|
||||
afterEach(() => { vi.restoreAllMocks(); vi.unstubAllEnvs(); skillFailures.clear(); });
|
||||
afterEach(() => { vi.restoreAllMocks(); vi.unstubAllEnvs(); skillFailures.clear(); skillWaits.clear(); });
|
||||
afterAll(async () => {
|
||||
await heartbeat?.drainActiveRunExecutions();
|
||||
unregisterServerAdapter(adapterType);
|
||||
@@ -288,6 +298,45 @@ describePostgres("heartbeat execution configuration through adapter dispatch", (
|
||||
const [unchangedAgent] = await db.select().from(agents).where(eq(agents.id, f.agentId));
|
||||
expect(unchangedAgent.adapterConfig).toMatchObject({ paperclipSkillSync: { desiredSkills: [other.key] } });
|
||||
}, 60_000);
|
||||
it("overlaps skills with assignments and dependent connection instructions before composition", async () => {
|
||||
const f = await seed();
|
||||
const skillGate = deferred<void>();
|
||||
skillWaits.set(f.companyId, () => skillGate.promise);
|
||||
const assignments = vi.spyOn(connectorRuntime, "resolveConnectorAssignments");
|
||||
const instructions = vi.spyOn(assignedMcp, "resolveAssignedConnectionInstructionsForRun");
|
||||
const configured = await configuration(f, { provider: "codex", model: "gpt-6.1-sol", homeMode: "isolated" });
|
||||
const opening = configured.prepare();
|
||||
try {
|
||||
await vi.waitFor(() => {
|
||||
expect(assignments).toHaveBeenCalledOnce();
|
||||
expect(instructions).toHaveBeenCalledOnce();
|
||||
});
|
||||
} finally {
|
||||
skillGate.resolve();
|
||||
}
|
||||
expect(await opening).toMatchObject({ prepared: true, runtimeConfig: { homeMode: "isolated", model: "gpt-6.1-sol" } });
|
||||
});
|
||||
|
||||
it("settles concurrent configuration reads before exposing the first preparation failure", async () => {
|
||||
const f = await seed();
|
||||
const assignmentGate = deferred<Awaited<ReturnType<typeof connectorRuntime.resolveConnectorAssignments>>>();
|
||||
const assignments = vi.spyOn(connectorRuntime, "resolveConnectorAssignments").mockReturnValue(assignmentGate.promise);
|
||||
const instructions = vi.spyOn(assignedMcp, "resolveAssignedConnectionInstructionsForRun").mockRejectedValue(new Error("later instruction failure"));
|
||||
skillFailures.set(f.companyId, "first skill failure");
|
||||
const configured = await configuration(f, { provider: "codex", homeMode: "isolated" });
|
||||
let settled = false;
|
||||
const opening = configured.prepare();
|
||||
void opening.then(() => { settled = true; }, () => { settled = true; });
|
||||
try {
|
||||
await vi.waitFor(() => expect(assignments).toHaveBeenCalledOnce());
|
||||
await Promise.resolve();
|
||||
expect(settled).toBe(false);
|
||||
expect(instructions).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
assignmentGate.resolve([]);
|
||||
}
|
||||
await expect(opening).rejects.toThrow("first skill failure");
|
||||
});
|
||||
|
||||
it("does not install new environment declarations when recovering a saved skill selection", async () => {
|
||||
const f = await seed(), pluginId = randomUUID(), pluginKey = `fixture.${randomUUID()}`;
|
||||
@@ -445,11 +494,11 @@ describePostgres("heartbeat execution configuration through adapter dispatch", (
|
||||
const runIdHash = createHash("sha256").update(finished.id).digest("hex").slice(0, 12);
|
||||
const steps = exporter.getFinishedSpans().filter(span => span.name.startsWith("heartbeat.prepare.configuration_")
|
||||
&& span.attributes["paperclip.task.run.run_id"] === runIdHash);
|
||||
expect(steps.map(span => span.name)).toEqual([
|
||||
expect(steps.map(span => span.name).sort()).toEqual([
|
||||
"configuration_mentions", "configuration_github_identity", "configuration_secrets", "configuration_ai_runtime",
|
||||
"configuration_skills", "configuration_connector_assignments", "configuration_connector_skills",
|
||||
"configuration_connector_delivery", "configuration_connection_instructions", "configuration_agent_identity",
|
||||
].map(step => `heartbeat.prepare.${step}`));
|
||||
].map(step => `heartbeat.prepare.${step}`).sort());
|
||||
for (const span of steps) {
|
||||
expect(span.status).toEqual({ code: 0 });
|
||||
expect(span.attributes).toEqual({ "paperclip.task.run.run_id": runIdHash });
|
||||
@@ -494,7 +543,7 @@ describePostgres("heartbeat execution configuration through adapter dispatch", (
|
||||
&& span.attributes["paperclip.task.run.run_id"] === runIdHash);
|
||||
expect(steps.find(span => span.name === "heartbeat.prepare.configuration_ai_runtime")?.status.code).toBe(0);
|
||||
expect(steps.find(span => span.name === "heartbeat.prepare.configuration_skills")?.status).toEqual({ code: 2 });
|
||||
expect(steps.some(span => span.name === "heartbeat.prepare.configuration_connector_assignments")).toBe(false);
|
||||
expect(steps.some(span => span.name === "heartbeat.prepare.configuration_connector_assignments")).toBe(true);
|
||||
for (const span of steps) {
|
||||
expect(span.attributes).toEqual({ "paperclip.task.run.run_id": runIdHash });
|
||||
expect(span.events).toEqual([]);
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { FAST_RESPONSE_AGENT_GUIDANCE } from "@paperclipai/shared";
|
||||
import { registerServerAdapter, unregisterServerAdapter, type AdapterExecutionContext } from "../adapters/index.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { prepareHeartbeatTaskContext } from "../services/heartbeat/task-context.js";
|
||||
import { documentService } from "../services/documents.js";
|
||||
import { ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY } from "../services/issue-continuation-summary.js";
|
||||
import { createRunSecretRedactionRegistry } from "../services/run-secret-redaction.js";
|
||||
@@ -19,6 +20,7 @@ if (!support.supported) console.warn(`Skipping heartbeat task context tests: ${s
|
||||
describePostgres("heartbeat task context delivered to an adapter", () => {
|
||||
const adapterType = "task_context_test";
|
||||
const captured = new Map<string, AdapterExecutionContext>();
|
||||
const inspectPreparation = new Map<string, () => void>();
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let heartbeat: ReturnType<typeof heartbeatService>;
|
||||
@@ -33,6 +35,7 @@ describePostgres("heartbeat task context delivered to an adapter", () => {
|
||||
type: adapterType,
|
||||
supportsLocalAgentJwt: false,
|
||||
execute: async (input) => {
|
||||
inspectPreparation.get(input.agent.id)?.();
|
||||
captured.set(input.runId, input);
|
||||
return { exitCode: 0, signal: null, timedOut: false, sessionId: "context-session", resultJson: {} };
|
||||
},
|
||||
@@ -118,6 +121,101 @@ describePostgres("heartbeat task context delivered to an adapter", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("reads selected comments once for both wake projections", async () => {
|
||||
const f = await seed();
|
||||
const commentId = randomUUID();
|
||||
await db.insert(issueComments).values({
|
||||
id: commentId, companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "context-user",
|
||||
body: "selected wake canary",
|
||||
});
|
||||
const select = vi.spyOn(db, "select");
|
||||
inspectPreparation.set(f.agentId, () => {
|
||||
const commentReads = select.mock.calls.filter(([projection]) => projection
|
||||
&& projection.body === issueComments.body && projection.id === issueComments.id && projection.deletedAt === issueComments.deletedAt);
|
||||
expect(commentReads).toHaveLength(1);
|
||||
});
|
||||
try {
|
||||
const { context } = await execute(f, { wakeCommentId: commentId, wakeCommentIds: [commentId] });
|
||||
expect(context.paperclipWakeComment).toMatchObject({ id: commentId, body: "selected wake canary" });
|
||||
expect(context.paperclipWake).toMatchObject({ comments: [{ id: commentId, body: "selected wake canary" }] });
|
||||
} finally {
|
||||
inspectPreparation.delete(f.agentId);
|
||||
select.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("cancels deleted-only queued comments before adapter dispatch", async () => {
|
||||
const f = await seed(), commentId = randomUUID();
|
||||
await db.insert(issueComments).values({ id: commentId, companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: "context-user", body: "discarded wake", deletedAt: new Date() });
|
||||
const run = await heartbeat.wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
|
||||
payload: { issueId: f.issueId }, contextSnapshot: { issueId: f.issueId, wakeCommentId: commentId, wakeCommentIds: [commentId] } });
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const finished = await heartbeat.getRun(run!.id);
|
||||
expect(finished).toMatchObject({ status: "cancelled", error: "Queued messages were discarded before dispatch" });
|
||||
expect(captured.has(run!.id)).toBe(false);
|
||||
});
|
||||
|
||||
it("preserves deleted comment metadata without delivering its body", async () => {
|
||||
const f = await seed();
|
||||
const commentId = randomUUID();
|
||||
await db.insert(issueComments).values({ id: commentId, companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: "context-user", body: "deleted wake canary", deletedAt: new Date(),
|
||||
presentation: { fixture: "deleted" }, metadata: { fixture: "deleted" } });
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "running", issueId: f.issueId }).returning();
|
||||
const [agent] = await db.select().from(agents).where(eq(agents.id, f.agentId));
|
||||
const [issueRef] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
const context: Record<string, unknown> = { issueId: f.issueId, wakeCommentId: commentId, wakeCommentIds: [commentId] };
|
||||
await prepareHeartbeatTaskContext(db, {
|
||||
run, agent, task: { issueRef, issueContext: null, context, taskSession: null },
|
||||
policy: { trustPreset: { kind: "standard" }, experimentalInstanceSettings: {} },
|
||||
services: { issuesSvc: { getAncestors: async () => [] }, getPinnedSkillTestContext: async () => null },
|
||||
});
|
||||
expect(context.paperclipWakeComment).toMatchObject({ id: commentId, body: "", presentation: null, metadata: null });
|
||||
expect(context.paperclipWake).toMatchObject({ comments: [{ id: commentId, body: "", presentation: null, metadata: null }] });
|
||||
expect(context.paperclipTaskMarkdown).not.toContain("deleted wake canary");
|
||||
});
|
||||
|
||||
it.each([false, true])("uses admitted skill-test scope when recovery is %s", async (recovering) => {
|
||||
const f = await seed();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "running", issueId: f.issueId }).returning();
|
||||
const [agent] = await db.select().from(agents).where(eq(agents.id, f.agentId));
|
||||
const [issueRef] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
const context: Record<string, unknown> = { issueId: f.issueId, paperclipAgentJwtScope: { kind: "skill_test", issueId: f.issueId } };
|
||||
const pinned = vi.fn(async () => null);
|
||||
const preparing = prepareHeartbeatTaskContext(db, {
|
||||
run, agent, task: { issueRef, issueContext: null, context, taskSession: null },
|
||||
policy: { recovering, trustPreset: { kind: "standard" }, experimentalInstanceSettings: {} },
|
||||
services: { issuesSvc: { getAncestors: async () => [] }, getPinnedSkillTestContext: pinned },
|
||||
});
|
||||
if (recovering) await expect(preparing).rejects.toThrow("skill test's saved version is unavailable");
|
||||
else await preparing;
|
||||
expect(pinned).toHaveBeenCalledTimes(recovering ? 1 : 0);
|
||||
expect(context.paperclipAgentJwtScope).toEqual({ kind: recovering ? "skill_test" : "standard", ...(recovering ? { issueId: f.issueId } : {}) });
|
||||
});
|
||||
|
||||
it("retains the context issue scope when no issue reference is available", async () => {
|
||||
const f = await seed();
|
||||
const otherIssueId = randomUUID(), commentId = randomUUID();
|
||||
await db.insert(issues).values({ id: otherIssueId, companyId: f.companyId, title: "Other issue", issueNumber: 2 });
|
||||
await db.insert(issueComments).values({ id: commentId, companyId: f.companyId, issueId: otherIssueId,
|
||||
authorType: "user", authorUserId: "context-user", body: "other issue canary" });
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "running", issueId: f.issueId }).returning();
|
||||
const [agent] = await db.select().from(agents).where(eq(agents.id, f.agentId));
|
||||
const context: Record<string, unknown> = { issueId: f.issueId, wakeCommentId: commentId, wakeCommentIds: [commentId] };
|
||||
await prepareHeartbeatTaskContext(db, {
|
||||
run, agent, task: { issueRef: null, issueContext: null, context, taskSession: null },
|
||||
policy: { trustPreset: { kind: "standard" }, experimentalInstanceSettings: {} },
|
||||
services: { issuesSvc: { getAncestors: async () => [] }, getPinnedSkillTestContext: async () => null },
|
||||
});
|
||||
expect(JSON.stringify(context.paperclipWake)).not.toContain("other issue canary");
|
||||
expect(context.paperclipWakeComment).toBeUndefined();
|
||||
});
|
||||
|
||||
it("quarantines low-trust summaries and comments before preparing higher-trust prompts", async () => {
|
||||
const f = await seed();
|
||||
const sourceTrust = buildLowTrustSourceTrust({ issueId: f.issueId });
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { generateKeyPairSync, randomUUID, sign } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, companies, createDb, environmentCustomImageSetupSessions, environmentLeases, environments, heartbeatRuns } from "@paperclipai/db";
|
||||
import { historicalCleanupVerificationService } from "../services/historical-cleanup-verification.js";
|
||||
import { environmentRuntimeService } from "../services/environment-runtime.js";
|
||||
import { withoutCleanupVerification } from "../services/cleanup-verification-proof.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js";
|
||||
@@ -16,7 +17,7 @@ if (!support.supported) console.warn(`Skipping historical cleanup database tests
|
||||
(support.supported ? describe : describe.skip)("historical cleanup verification", () => {
|
||||
let db: ReturnType<typeof createDb>, database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("historical-cleanup-"); db = createDb(database.connectionString); }, 60_000);
|
||||
afterEach(async () => { await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies, plugins, environments CASCADE`); });
|
||||
afterEach(async () => { vi.unstubAllEnvs(); vi.unstubAllGlobals(); await db.execute(sql`SET client_min_messages = warning; TRUNCATE companies, plugins, environments CASCADE`); });
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
|
||||
const read = () => readIdleSleepSafety(db, owned, () => now, ownerId, async () => "none", undefined, undefined, true, true);
|
||||
const state = async () => (await read()).backgroundWork;
|
||||
@@ -37,6 +38,55 @@ if (!support.supported) console.warn(`Skipping historical cleanup database tests
|
||||
inspectSetup: async row => ({ providerLeaseId: row.providerLeaseId, state: setupState }),
|
||||
});
|
||||
}
|
||||
it.each([false, true])("reconciles a legacy Cloud lease through signed absence and permits idle sleep (ad-hoc %s)", async adHoc => {
|
||||
const f = await seed(adHoc);
|
||||
const { privateKey, publicKey } = generateKeyPairSync("ed25519");
|
||||
for (const [key, value] of Object.entries({ PAPERCLIP_CLOUD_STACK_ID: "fixture-stack",
|
||||
PAPERCLIP_CLOUD_API_ORIGIN: "https://cloud.example.test", PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "fleet-fixture",
|
||||
PAPERCLIP_PRIMARY_HOST: "fixture.paperclip.test", PAPERCLIP_AUTH_SECRET: "fixture-secret",
|
||||
PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: JSON.stringify({ keys: [{ ...publicKey.export({ format: "jwk" }), kid: "fixture", alg: "EdDSA", use: "sig" }] }) })) vi.stubEnv(key, value);
|
||||
let absent = false;
|
||||
const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => {
|
||||
if (!absent) return Response.json({ absent: false });
|
||||
const binding = JSON.parse(init.body as string), seconds = Math.floor(Date.now() / 1000);
|
||||
expect(binding).toEqual({ purpose: "cleanup", companyId: f.companyId, environmentId: f.environment.id,
|
||||
leaseId: f.lease.id, heartbeatRunId: adHoc ? null : f.runId, provider: "daytona", providerLeaseId: f.lease.providerLeaseId });
|
||||
const header = Buffer.from(JSON.stringify({ alg: "EdDSA", kid: "fixture", typ: "paperclip-sandbox-cleanup-absence+jwt" })).toString("base64url");
|
||||
const claims = Buffer.from(JSON.stringify({ v: 1, iss: "paperclip-cloud", aud: "paperclip-sandbox-cleanup-absence/v1",
|
||||
sub: "fixture-stack", ...binding, absent: true, iat: seconds, exp: seconds + 60 })).toString("base64url");
|
||||
return Response.json({ receipt: `${header}.${claims}.${sign(null, Buffer.from(`${header}.${claims}`), privateKey).toString("base64url")}` });
|
||||
});
|
||||
vi.stubGlobal("fetch", fetchImpl);
|
||||
const runtime = environmentRuntimeService(db);
|
||||
const service = historicalCleanupVerificationService(db, {
|
||||
inspectLease: row => runtime.inspectLeaseResource(row.id), inspectSetup: async () => undefined,
|
||||
});
|
||||
expect(await state()).toBe("present");
|
||||
expect(await service.sweep()).toMatchObject({ checked: 1, deferred: 1, verified: 0 });
|
||||
expect(await state()).toBe("present");
|
||||
absent = true;
|
||||
expect(await service.sweep()).toMatchObject({ checked: 1, verified: 1, failed: 0 });
|
||||
expect(await state()).toBe("none");
|
||||
const [current] = await db.select().from(environmentLeases);
|
||||
expect(current).toMatchObject({ status: "failed", cleanupStatus: "success", heartbeatRunId: adHoc ? null : f.runId });
|
||||
expect(current!.metadata).not.toHaveProperty("remoteExecutionTermination");
|
||||
expect(await service.sweep()).toMatchObject({ checked: 0 });
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
it.each(["active", "stopped", "unknown"] as const)("does not replace an explicit %s provider observation with a Cloud read", async providerState => {
|
||||
await seed(true);
|
||||
for (const [key, value] of Object.entries({ PAPERCLIP_CLOUD_STACK_ID: "fixture-stack",
|
||||
PAPERCLIP_CLOUD_API_ORIGIN: "https://cloud.example.test", PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "fleet-fixture",
|
||||
PAPERCLIP_PRIMARY_HOST: "fixture.paperclip.test", PAPERCLIP_AUTH_SECRET: "fixture-secret" })) vi.stubEnv(key, value);
|
||||
const fetchImpl = vi.fn(); vi.stubGlobal("fetch", fetchImpl);
|
||||
const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox",
|
||||
acquireRunLease: async () => { throw new Error("unexpected acquire"); }, releaseRunLease: async () => null,
|
||||
inspectLeaseResource: async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId!, state: providerState }),
|
||||
}] });
|
||||
const service = historicalCleanupVerificationService(db, { inspectLease: row => runtime.inspectLeaseResource(row.id), inspectSetup: async () => undefined });
|
||||
expect(await service.sweep()).toMatchObject({ checked: 1, verified: 0, deferred: 1 });
|
||||
expect(await state()).toBe("present"); expect(fetchImpl).not.toHaveBeenCalled();
|
||||
});
|
||||
it.each([false, true])("accepts exact deletion proof without changing outcomes or granting run authority (ad-hoc %s)", async adHoc => {
|
||||
const f = await seed(adHoc); expect(await state()).toBe("present");
|
||||
const service = verifier(); expect(await service.sweep()).toMatchObject({ checked: 1, verified: 1, failed: 0 });
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import { and, eq, getTableColumns, inArray, isNull, sql, type SQL } from "drizzle-orm";
|
||||
import { agents, decisionRetention, decisions, heartbeatRuns,
|
||||
issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, approvals, type Db } from "@paperclipai/db";
|
||||
import type { AttentionSourceKind } from "@paperclipai/shared";
|
||||
import { evaluateAgentInvokability } from "./agent-invokability.js";
|
||||
import { HUMAN_RECOVERY_OWNER_TYPES, ATTENTION_BUDGET_WARNING_PERCENT, ATTENTION_AGENT_ERROR_STATUS, collapsePendingConfirmationsToNewest, listFailedRunsWithNewerRun, readRunIssueId, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js";
|
||||
import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
|
||||
@@ -13,12 +12,13 @@ import { issueService } from "./issues.js";
|
||||
import { parseIssueExecutionState } from "./issue-execution-policy.js";
|
||||
import { isProspectiveBlockedTransition } from "./routable-blocked.js";
|
||||
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
|
||||
import type { AttentionRetentionScope } from "./retention-work-notifications.js";
|
||||
|
||||
const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
const kinds = ["decision", "issue_thread_interaction", "recovery_action", "blocker_attention", "review",
|
||||
"failed_run", "budget_alert", "agent_error_alert"] as const;
|
||||
type Kind = typeof kinds[number];
|
||||
type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string; issueId?: string };
|
||||
type Scope = AttentionRetentionScope;
|
||||
type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number; issueId: string | null };
|
||||
type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" };
|
||||
type Cursor = { sourceKind: string; sourceId: string };
|
||||
@@ -66,8 +66,9 @@ const sources: Record<Kind, SQL> = {
|
||||
export function attentionRetentionService(db: Db) {
|
||||
const cursors = new Map<string, Cursor>();
|
||||
async function candidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise<Candidate[]> {
|
||||
const issueIds = scope.issueIds ?? (scope.issueId ? [scope.issueId] : []);
|
||||
const selected = kinds.filter(kind => (!scope.sourceKind || scope.sourceKind === kind)
|
||||
&& (!scope.issueId || !["budget_alert", "agent_error_alert"].includes(kind)));
|
||||
&& (!issueIds.length || !["budget_alert", "agent_error_alert"].includes(kind)));
|
||||
if (!selected.length) return [];
|
||||
const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at, issue_id) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)})
|
||||
select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId",
|
||||
@@ -75,7 +76,7 @@ export function attentionRetentionService(db: Db) {
|
||||
from source_candidates s inner join companies c on c.id = s.company_id
|
||||
left join decision_retention r on r.company_id = s.company_id and r.source_kind = s.source_kind and r.source_id = s.source_id
|
||||
where c.status = 'active'
|
||||
and (${scope.issueId ?? null}::text is null or s.issue_id = ${scope.issueId ?? null})
|
||||
${issueIds.length ? sql`and s.issue_id in (${sql.join(issueIds.map(id => sql`${id}`), sql`, `)})` : sql``}
|
||||
and (${scope.companyId ?? null}::uuid is null or s.company_id = ${scope.companyId ?? null}::uuid)
|
||||
and (${sourceId ?? scope.sourceId ?? null}::text is null or s.source_id = ${sourceId ?? scope.sourceId ?? null})
|
||||
and (s.source_kind = 'blocker_attention' or ((r.id is null or r.keep = false) and r.archived_at is null))
|
||||
@@ -312,7 +313,8 @@ export function registerAttentionRetentionWork(coordinator: ReturnType<typeof cr
|
||||
matches: (entry, scope) => (!scope?.companyId || scope.companyId === entry.companyId)
|
||||
&& (!scope?.sourceKind || scope.sourceKind === entry.sourceKind)
|
||||
&& (!scope?.sourceId || scope.sourceId === entry.sourceId)
|
||||
&& (!scope?.issueId || scope.issueId === entry.issueId),
|
||||
&& (!scope?.issueId || scope.issueId === entry.issueId)
|
||||
&& (!scope?.issueIds?.length || entry.issueId !== null && scope.issueIds.includes(entry.issueId)),
|
||||
load: service.load, check: service.check,
|
||||
}, canRun);
|
||||
}
|
||||
@@ -167,6 +167,19 @@ async function resolveScopeRecord(db: Db, scopeType: BudgetScopeType, scopeId: s
|
||||
};
|
||||
}
|
||||
|
||||
async function resolveSummaryScopeRecord(db: Db, companyId: string, scopeType: BudgetScopeType, scopeId: string): Promise<ScopeRecord> {
|
||||
let scope: ScopeRecord;
|
||||
try {
|
||||
scope = await resolveScopeRecord(db, scopeType, scopeId);
|
||||
} catch (error) {
|
||||
if (!(error instanceof HttpError) || error.status !== 404 || (scopeType !== "agent" && scopeType !== "project")) throw error;
|
||||
return { companyId, name: scopeType === "agent" ? "Deleted agent" : "Deleted project",
|
||||
paused: false, pauseReason: null, budgetHeld: false };
|
||||
}
|
||||
if (scope.companyId !== companyId) throw notFound("Budget scope not found");
|
||||
return scope;
|
||||
}
|
||||
|
||||
// Keep a large dashboard from filling the database pool with queued reads.
|
||||
async function mapBudgetReads<T, R>(rows: T[], read: (row: T) => Promise<R>): Promise<R[]> {
|
||||
const result: R[] = [];
|
||||
@@ -411,7 +424,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
}
|
||||
|
||||
async function buildPolicySummary(policy: PolicyRow): Promise<BudgetPolicySummary> {
|
||||
const scope = await resolveScopeRecord(db, policy.scopeType as BudgetScopeType, policy.scopeId);
|
||||
const scope = await resolveSummaryScopeRecord(db, policy.companyId, policy.scopeType as BudgetScopeType, policy.scopeId);
|
||||
const observed = await computeObservedSpend(db, policy);
|
||||
const { total: observedAmount, totalExact: observedAmountExact, unpricedEventCount, pendingRunCount } = observed;
|
||||
const { windowStart: start, windowEnd: end } = observed;
|
||||
@@ -580,7 +593,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
const approvalStatusById = new Map(approvalRows.map((row) => [row.id, row.status]));
|
||||
|
||||
return mapBudgetReads(rows, async (row) => {
|
||||
const scope = await resolveScopeRecord(db, row.scopeType as BudgetScopeType, row.scopeId);
|
||||
const scope = await resolveSummaryScopeRecord(db, row.companyId, row.scopeType as BudgetScopeType, row.scopeId);
|
||||
return {
|
||||
id: row.id,
|
||||
companyId: row.companyId,
|
||||
@@ -945,6 +958,13 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
if (incident.companyId !== companyId) throw notFound("Budget incident not found");
|
||||
|
||||
const policy = await getPolicyRow(incident.policyId);
|
||||
if (policy.companyId !== companyId || policy.scopeType !== incident.scopeType || policy.scopeId !== incident.scopeId) {
|
||||
throw notFound("Budget scope not found");
|
||||
}
|
||||
const scope = input.action === "keep_paused"
|
||||
? await resolveSummaryScopeRecord(db, companyId, policy.scopeType as BudgetScopeType, policy.scopeId)
|
||||
: await resolveScopeRecord(db, policy.scopeType as BudgetScopeType, policy.scopeId);
|
||||
if (scope.companyId !== companyId) throw notFound("Budget scope not found");
|
||||
if (input.action === "raise_budget_and_resume") {
|
||||
const nextAmount = Math.max(0, Math.floor(input.amount ?? 0));
|
||||
const observed = await computeObservedSpend(db, policy);
|
||||
|
||||
@@ -15,6 +15,7 @@ import { hasStopOnlyCleanup, prepareSandboxStopAndRetain, readStopOnlyCleanup, s
|
||||
import { JsonRpcCallError, readEnvironmentAcquisitionDiagnostic, readEnvironmentCreationCleanupError } from "@paperclipai/plugin-sdk";
|
||||
import { preserveEnvironmentSyncOutErrorDiagnostic } from "./environment-sync-out-error.js";
|
||||
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { inspectCloudSandboxCleanupAbsence } from "./sandbox-absence-proof.js";
|
||||
import { hasNativeWorkspaceExportResume, releaseCompletedNativeWorkspaceExportRetention } from "./native-runtime/native-workspace-export-resume.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { and, eq, inArray, ne, or, sql } from "drizzle-orm";
|
||||
@@ -4021,7 +4022,12 @@ export function environmentRuntimeService(
|
||||
const lease = await environmentsSvc.getLeaseById(leaseId);
|
||||
if (!lease || lease.status !== "failed" || lease.leasePolicy !== "ephemeral" || !lease.providerLeaseId) return undefined;
|
||||
const environment = lease.environmentId ? await environmentsSvc.getById(lease.environmentId) : null;
|
||||
return await getDriver(getLeaseDriverKey(lease, environment))?.inspectLeaseResource?.({ environment, lease });
|
||||
// Legacy Cloud leases can lack a credential version that permits a
|
||||
// provider-local observation. Cloud's manager verifies the historical
|
||||
// organization and signs only exact absence; no replacement key or
|
||||
// cleanup label is treated as proof. Explicit provider answers win.
|
||||
return await getDriver(getLeaseDriverKey(lease, environment))?.inspectLeaseResource?.({ environment, lease })
|
||||
?? await inspectCloudSandboxCleanupAbsence(lease);
|
||||
},
|
||||
|
||||
async retryPendingSandboxTeardown(input: {
|
||||
|
||||
@@ -2006,8 +2006,11 @@ export function heartbeatService(
|
||||
const dispatchIssueId = readNonEmptyString(parseObject(run.contextSnapshot).issueId);
|
||||
const resumingAdmittedConversationTurn = !!runOptions.nativeLeaseOwner
|
||||
&& typeof run.contextSnapshot?.conversationSessionGeneration === "number";
|
||||
if (dispatchIssueId && isConversation(await getIssueExecutionContext(run.companyId, dispatchIssueId))
|
||||
&& !resumingAdmittedConversationTurn && !(await instanceSettings.getExperimental()).enableAgentChat) {
|
||||
const conversationIdentity = dispatchIssueId && !resumingAdmittedConversationTurn
|
||||
? await db.select({ conversationAgentId: issues.conversationAgentId, conversationUserId: issues.conversationUserId })
|
||||
.from(issues).where(and(eq(issues.companyId, run.companyId), eq(issues.id, dispatchIssueId))).then(rows => rows[0] ?? null)
|
||||
: null;
|
||||
if (isConversation(conversationIdentity) && !(await instanceSettings.getExperimental()).enableAgentChat) {
|
||||
await setRunStatus(run.id, "cancelled", { finishedAt: new Date(), error: "Agent Chat is disabled", errorCode: "agent_chat_disabled" });
|
||||
await setWakeupStatus(run.wakeupRequestId, "cancelled", { finishedAt: new Date() });
|
||||
await releaseIssueExecutionAndPromote((await getRun(run.id))!, { suppressImmediateRecovery: true });
|
||||
@@ -2203,44 +2206,6 @@ export function heartbeatService(
|
||||
context[PAPERCLIP_EXTERNAL_CHAT_EXECUTION_BOUND_KEY] = true;
|
||||
}
|
||||
const wakeCommentId = deriveCommentId(context, null);
|
||||
const wakeCommentContext =
|
||||
issueContext && wakeCommentId
|
||||
? await db
|
||||
.select({
|
||||
id: issueComments.id,
|
||||
body: issueComments.body,
|
||||
authorType: issueComments.authorType,
|
||||
authorAgentId: issueComments.authorAgentId,
|
||||
authorUserId: issueComments.authorUserId,
|
||||
presentation: issueComments.presentation,
|
||||
metadata: issueComments.metadata,
|
||||
deletedAt: issueComments.deletedAt,
|
||||
deletedByType: issueComments.deletedByType,
|
||||
deletedByAgentId: issueComments.deletedByAgentId,
|
||||
deletedByUserId: issueComments.deletedByUserId,
|
||||
deletedByRunId: issueComments.deletedByRunId,
|
||||
sourceTrust: issueComments.sourceTrust,
|
||||
})
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.id, wakeCommentId),
|
||||
eq(issueComments.issueId, issueContext.id),
|
||||
eq(issueComments.companyId, agent.companyId),
|
||||
),
|
||||
)
|
||||
.then((rows) => {
|
||||
const row = rows[0] ?? null;
|
||||
return row?.deletedAt
|
||||
? {
|
||||
...row,
|
||||
body: "",
|
||||
presentation: null,
|
||||
metadata: null,
|
||||
}
|
||||
: row;
|
||||
})
|
||||
: null;
|
||||
let issueAssigneeOverrides =
|
||||
issueContext && issueContext.assigneeAgentId === agent.id
|
||||
? parseIssueAssigneeAdapterOverrides(
|
||||
@@ -2609,7 +2574,7 @@ export function heartbeatService(
|
||||
continuationSummary, executionContinuation, paperclipWakePayload, safeWakeComments, safeWakeCommentContext, pinnedSkillTest,
|
||||
} = await timePreparation("task_context", () => prepareHeartbeatTaskContext(db, {
|
||||
run, agent,
|
||||
task: { issueRef, issueContext, wakeCommentContext, context, taskSession },
|
||||
task: { issueRef, issueContext, context, taskSession },
|
||||
policy: { trustPreset, experimentalInstanceSettings, recovering: Boolean(runOptions.nativeRestartRecovery || runOptions.nativeLeaseOwner) },
|
||||
services: { issuesSvc, getPinnedSkillTestContext },
|
||||
}));
|
||||
|
||||
@@ -139,28 +139,27 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
stripWorkspaceRuntimeFromExecutionRunConfig(mergedConfig);
|
||||
const skillActor: AuthorizationActor = { type: "agent", agentId: agent.id, companyId: agent.companyId,
|
||||
runId: run.id, source: "agent_jwt", onBehalfOfUserId: responsibleUserId === "local-board" ? null : responsibleUserId };
|
||||
const runScopedMentionedSkillKeys =
|
||||
preconfiguredHome || pinnedSkillTest ? [] : await time("configuration_mentions", () => resolveRunScopedMentionedSkillKeys({
|
||||
const [mentionedSkills, githubIdentity] = await Promise.allSettled([
|
||||
preconfiguredHome || pinnedSkillTest ? Promise.resolve([]) : time("configuration_mentions", () => resolveRunScopedMentionedSkillKeys({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
issueId,
|
||||
commentIds: extractWakeCommentIds(context),
|
||||
actor: skillActor,
|
||||
}));
|
||||
})),
|
||||
time("configuration_github_identity", () => resolveManagedGitHubIdentitySelection(db, agent.companyId, {
|
||||
agentId: agent.id, responsibleUserId, allowStandingDelegation: false,
|
||||
})),
|
||||
]);
|
||||
if (mentionedSkills.status === "rejected") throw mentionedSkills.reason;
|
||||
if (githubIdentity.status === "rejected") throw githubIdentity.reason;
|
||||
const runScopedMentionedSkillKeys = mentionedSkills.value;
|
||||
const runScopedSkillKeys =
|
||||
!pinnedSkillTest && acceptedPlanContinuationWake &&
|
||||
!acceptedPlanWakeRoutingDecision?.suppressAcceptedContinuation
|
||||
? [...runScopedMentionedSkillKeys, ACCEPTED_PLAN_CONVERSION_SKILL_KEY]
|
||||
: runScopedMentionedSkillKeys;
|
||||
const githubSelection = await time("configuration_github_identity", () => resolveManagedGitHubIdentitySelection(
|
||||
db,
|
||||
agent.companyId,
|
||||
{
|
||||
agentId: agent.id,
|
||||
responsibleUserId,
|
||||
allowStandingDelegation: false,
|
||||
},
|
||||
));
|
||||
const githubSelection = githubIdentity.value;
|
||||
const useHostGitHub =
|
||||
!githubSelection.configured &&
|
||||
trustPreset.kind === "standard" &&
|
||||
@@ -274,6 +273,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
);
|
||||
const nativeRunnerPreparationSpans: NativeRunHistoricalSpan[] = [];
|
||||
const skillsPrepareStartedAtMs = Date.now();
|
||||
let skillsPrepareEndedAtMs = skillsPrepareStartedAtMs;
|
||||
let skillDeliveryConfig = effectiveResolvedConfig;
|
||||
const prepareSkills = async () => {
|
||||
|
||||
@@ -346,24 +346,35 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
const runtimeSkillEntries = preconfiguredHome ? [] : await time("configuration_skills", prepareSkills);
|
||||
const skillsPreparation = (preconfiguredHome ? Promise.resolve([]) : time("configuration_skills", prepareSkills))
|
||||
.finally(() => { skillsPrepareEndedAtMs = Date.now(); });
|
||||
const assignmentsPreparation = preconfiguredHome ? Promise.resolve([]) : time("configuration_connector_assignments", () => resolveConnectorAssignments(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id, issueId: typeof context.issueId === "string" ? context.issueId : undefined }));
|
||||
const [preparedSkills, preparedAssignments, preparedInstructions] = await Promise.allSettled([
|
||||
skillsPreparation,
|
||||
assignmentsPreparation,
|
||||
assignmentsPreparation.then(() => time("configuration_connection_instructions", () => resolveAssignedConnectionInstructionsForRun(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id }))),
|
||||
]);
|
||||
if (preparedSkills.status === "rejected") throw preparedSkills.reason;
|
||||
if (preparedAssignments.status === "rejected") throw preparedAssignments.reason;
|
||||
if (preparedInstructions.status === "rejected") throw preparedInstructions.reason;
|
||||
const runtimeSkillEntries = preparedSkills.value;
|
||||
nativeRunnerPreparationSpans.push({
|
||||
name: "skills.prepare",
|
||||
parentName: "task.prepare",
|
||||
startedAtMs: skillsPrepareStartedAtMs,
|
||||
endedAtMs: Date.now(),
|
||||
endedAtMs: skillsPrepareEndedAtMs,
|
||||
});
|
||||
const connectorAssignments = preconfiguredHome ? [] : await time("configuration_connector_assignments", () => resolveConnectorAssignments(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id, issueId: typeof context.issueId === "string" ? context.issueId : undefined }));
|
||||
const connectorAssignments = preparedAssignments.value;
|
||||
const connectorSkillConfig = await time("configuration_connector_skills", () => applyConnectorSkills(skillDeliveryConfig, runtimeSkillEntries, connectorAssignments));
|
||||
// Both CLI adapters and native context materialization use the same resolved set.
|
||||
runtimeSkillEntries.splice(0, runtimeSkillEntries.length, ...connectorSkillConfig.paperclipRuntimeSkills);
|
||||
const connectorDelivery = await time("configuration_connector_delivery", () => prepareConnectorSkillDelivery(connectorSkillConfig, agent.adapterType));
|
||||
// Always replace this runtime-only field; caller wake data cannot supply skills.
|
||||
context.paperclipWake = { ...parseObject(context.paperclipWake), connectorSkillInstructions: connectorDelivery.instructions };
|
||||
let runtimeConfig = await time("configuration_connection_instructions", () => prepareConnectionInstructionDelivery({
|
||||
resolve: () => resolveAssignedConnectionInstructionsForRun(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id }),
|
||||
let runtimeConfig = await prepareConnectionInstructionDelivery({
|
||||
resolve: async () => preparedInstructions.value,
|
||||
context, config: connectorDelivery.config, native: agent.adapterType === "paperclip_runner",
|
||||
}));
|
||||
});
|
||||
const agentIdentity = supportsManagedAgentIdentity(agent.adapterType, agent.adapterConfig, run.runtimeMode === "native" ? run.driverKind : undefined)
|
||||
? await time("configuration_agent_identity", () => agentIdentityService(db).ensureAgentIdentity(agent.companyId, agent.id))
|
||||
: undefined;
|
||||
|
||||
@@ -14,6 +14,7 @@ import {
|
||||
ConfigurationIncompleteFailure,
|
||||
createHeartbeatRunPreparation,
|
||||
resolveExecutionRunAdapterConfig,
|
||||
readHeartbeatWakeComments,
|
||||
} from "./run-preparation.js";
|
||||
|
||||
// Existing integration suites exercise the real database and dispatch path.
|
||||
@@ -38,6 +39,12 @@ function databaseWithResults(...results: unknown[][]) {
|
||||
return { db: { select } as unknown as Db, select, scopes };
|
||||
}
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>(complete => { resolve = complete; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
const emptyRoutineContext = { routineId: null, env: null, responsibleUserId: null };
|
||||
|
||||
describe("heartbeat run preparation module", () => {
|
||||
@@ -120,6 +127,42 @@ describe("heartbeat run preparation module", () => {
|
||||
expect(database.select).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("scopes one selected-comment snapshot for both wake consumers", async () => {
|
||||
const database = databaseWithResults([{ id: "comment-one", body: "selected wake" }]);
|
||||
expect(await readHeartbeatWakeComments(database.db, "company-one", "issue-one", ["comment-one"]))
|
||||
.toEqual([{ id: "comment-one", body: "selected wake" }]);
|
||||
expect(database.scopes[0]).toEqual(expect.arrayContaining(["company-one", "issue-one", "comment-one"]));
|
||||
expect(await readHeartbeatWakeComments(database.db, "company-one", "issue-one", [])).toEqual([]);
|
||||
expect(database.select).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("settles independent binding checks before surfacing a failure or resolving values", async () => {
|
||||
const projectCheck = deferred<[]>();
|
||||
const firstFailure = new Error("agent binding check failed");
|
||||
const secretsSvc = {
|
||||
resolveAdapterConfigForRuntime: vi.fn(), resolveEnvBindings: vi.fn(),
|
||||
collectMissingRuntimeBindings: vi.fn(async (_company, _env, context) => {
|
||||
if (context.consumerType === "agent") throw firstFailure;
|
||||
return context.consumerType === "project" ? projectCheck.promise : [];
|
||||
}),
|
||||
collectMissingAdapterConfigRuntimeBindings: vi.fn(async () => []),
|
||||
};
|
||||
let settled = false;
|
||||
const opening = resolveExecutionRunAdapterConfig({
|
||||
companyId: "company-one", agentId: "agent-one", projectId: "project-one",
|
||||
environmentId: "environment-one", environmentEnv: { SHARED: "environment" },
|
||||
executionRunConfig: { env: { SHARED: "agent" } }, projectEnv: { SHARED: "project" }, secretsSvc,
|
||||
});
|
||||
void opening.then(() => { settled = true; }, () => { settled = true; });
|
||||
await vi.waitFor(() => expect(secretsSvc.collectMissingRuntimeBindings).toHaveBeenCalledTimes(3));
|
||||
expect(secretsSvc.collectMissingAdapterConfigRuntimeBindings).toHaveBeenCalledOnce();
|
||||
expect(settled).toBe(false);
|
||||
expect(secretsSvc.resolveAdapterConfigForRuntime).not.toHaveBeenCalled();
|
||||
expect(secretsSvc.resolveEnvBindings).not.toHaveBeenCalled();
|
||||
projectCheck.resolve([]);
|
||||
await expect(opening).rejects.toBe(firstFailure);
|
||||
});
|
||||
|
||||
it("keeps missing scoped credentials as a pre-dispatch failure", async () => {
|
||||
const secretsSvc = {
|
||||
resolveAdapterConfigForRuntime: vi.fn(),
|
||||
|
||||
@@ -394,76 +394,31 @@ export async function resolveExecutionRunAdapterConfig(input: {
|
||||
// no binding before resolving any secret value. Missing bindings short-circuit
|
||||
// to a configuration-incomplete blocker routed to a human owner instead of a
|
||||
// dispatched-then-failed run (which previously surfaced as opaque setup_failed).
|
||||
if (typeof input.secretsSvc.collectMissingRuntimeBindings === "function") {
|
||||
const missingBindings: MissingRuntimeBinding[] = [];
|
||||
if (environmentEnv && input.environmentId) {
|
||||
missingBindings.push(
|
||||
...(await input.secretsSvc.collectMissingRuntimeBindings(
|
||||
input.companyId,
|
||||
environmentEnv,
|
||||
{
|
||||
consumerType: "environment",
|
||||
consumerId: input.environmentId,
|
||||
responsibleUserId: input.responsibleUserId ?? null,
|
||||
},
|
||||
)),
|
||||
);
|
||||
}
|
||||
const collectMissing = input.secretsSvc.collectMissingRuntimeBindings;
|
||||
if (typeof collectMissing === "function") {
|
||||
const checks: Array<() => Promise<MissingRuntimeBinding[]>> = [];
|
||||
const checkEnv = (env: unknown, consumerType: "environment" | "agent" | "project" | "routine", consumerId: string) => {
|
||||
checks.push(() => collectMissing(input.companyId, env, {
|
||||
consumerType, consumerId, responsibleUserId: input.responsibleUserId ?? null,
|
||||
}));
|
||||
};
|
||||
if (environmentEnv && input.environmentId) checkEnv(environmentEnv, "environment", input.environmentId);
|
||||
if (input.agentId) {
|
||||
missingBindings.push(
|
||||
...(await input.secretsSvc.collectMissingRuntimeBindings(
|
||||
input.companyId,
|
||||
parseObject(executionRunConfig.env),
|
||||
{
|
||||
consumerType: "agent",
|
||||
consumerId: input.agentId,
|
||||
responsibleUserId: input.responsibleUserId ?? null,
|
||||
},
|
||||
)),
|
||||
);
|
||||
if (
|
||||
typeof input.secretsSvc.collectMissingAdapterConfigRuntimeBindings ===
|
||||
"function"
|
||||
) {
|
||||
missingBindings.push(
|
||||
...(await input.secretsSvc.collectMissingAdapterConfigRuntimeBindings(
|
||||
input.companyId,
|
||||
executionRunConfig,
|
||||
input.adapterType ?? null,
|
||||
{
|
||||
consumerType: "agent",
|
||||
consumerId: input.agentId,
|
||||
responsibleUserId: input.responsibleUserId ?? null,
|
||||
},
|
||||
)),
|
||||
);
|
||||
checkEnv(parseObject(executionRunConfig.env), "agent", input.agentId);
|
||||
const collectAdapterMissing = input.secretsSvc.collectMissingAdapterConfigRuntimeBindings;
|
||||
if (typeof collectAdapterMissing === "function") {
|
||||
checks.push(() => collectAdapterMissing(input.companyId, executionRunConfig, input.adapterType ?? null, {
|
||||
consumerType: "agent", consumerId: input.agentId!, responsibleUserId: input.responsibleUserId ?? null,
|
||||
}));
|
||||
}
|
||||
}
|
||||
if (projectEnv && input.projectId) {
|
||||
missingBindings.push(
|
||||
...(await input.secretsSvc.collectMissingRuntimeBindings(
|
||||
input.companyId,
|
||||
projectEnv,
|
||||
{
|
||||
consumerType: "project",
|
||||
consumerId: input.projectId,
|
||||
responsibleUserId: input.responsibleUserId ?? null,
|
||||
},
|
||||
)),
|
||||
);
|
||||
}
|
||||
if (routineEnv && input.routineId) {
|
||||
missingBindings.push(
|
||||
...(await input.secretsSvc.collectMissingRuntimeBindings(
|
||||
input.companyId,
|
||||
routineEnv,
|
||||
{
|
||||
consumerType: "routine",
|
||||
consumerId: input.routineId,
|
||||
responsibleUserId: input.responsibleUserId ?? null,
|
||||
},
|
||||
)),
|
||||
);
|
||||
if (projectEnv && input.projectId) checkEnv(projectEnv, "project", input.projectId);
|
||||
if (routineEnv && input.routineId) checkEnv(routineEnv, "routine", input.routineId);
|
||||
const results = await Promise.allSettled(checks.map(check => Promise.resolve().then(check)));
|
||||
const missingBindings: MissingRuntimeBinding[] = [];
|
||||
for (const result of results) {
|
||||
if (result.status === "rejected") throw result.reason;
|
||||
missingBindings.push(...result.value);
|
||||
}
|
||||
if (requiredScopedEnvBinding) {
|
||||
const requiredEnvKeys = new Set(requiredScopedEnvBinding.keys);
|
||||
@@ -2244,12 +2199,45 @@ export async function attestReviewedExternalChatRun(input: {
|
||||
throw new Error("reviewed_chat_execution_binding_not_ready");
|
||||
}
|
||||
|
||||
export async function readHeartbeatWakeComments(db: Db, companyId: string, issueId: string | null, commentIds: string[]) {
|
||||
if (commentIds.length === 0) return [];
|
||||
return db
|
||||
.select({
|
||||
id: issueComments.id,
|
||||
companyId: issueComments.companyId,
|
||||
issueId: issueComments.issueId,
|
||||
body: issueComments.body,
|
||||
origin: issueComments.origin,
|
||||
authorType: issueComments.authorType,
|
||||
authorAgentId: issueComments.authorAgentId,
|
||||
authorUserId: issueComments.authorUserId,
|
||||
presentation: issueComments.presentation,
|
||||
metadata: issueComments.metadata,
|
||||
deletedAt: issueComments.deletedAt,
|
||||
deletedByType: issueComments.deletedByType,
|
||||
deletedByAgentId: issueComments.deletedByAgentId,
|
||||
deletedByUserId: issueComments.deletedByUserId,
|
||||
deletedByRunId: issueComments.deletedByRunId,
|
||||
sourceTrust: issueComments.sourceTrust,
|
||||
createdAt: issueComments.createdAt,
|
||||
})
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.companyId, companyId),
|
||||
issueId ? eq(issueComments.issueId, issueId) : undefined,
|
||||
inArray(issueComments.id, commentIds),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
export async function buildPaperclipWakePayload(input: {
|
||||
db: Db;
|
||||
companyId: string;
|
||||
agentId?: string | null;
|
||||
runId?: string | null;
|
||||
contextSnapshot: Record<string, unknown>;
|
||||
wakeCommentRows?: Awaited<ReturnType<typeof readHeartbeatWakeComments>>;
|
||||
continuationSummary?: {
|
||||
key: string;
|
||||
title: string | null;
|
||||
@@ -2314,36 +2302,9 @@ export async function buildPaperclipWakePayload(input: {
|
||||
)
|
||||
return null;
|
||||
|
||||
const commentRows =
|
||||
commentIds.length === 0
|
||||
? []
|
||||
: await input.db
|
||||
.select({
|
||||
id: issueComments.id,
|
||||
issueId: issueComments.issueId,
|
||||
body: issueComments.body,
|
||||
origin: issueComments.origin,
|
||||
authorType: issueComments.authorType,
|
||||
authorAgentId: issueComments.authorAgentId,
|
||||
authorUserId: issueComments.authorUserId,
|
||||
presentation: issueComments.presentation,
|
||||
metadata: issueComments.metadata,
|
||||
deletedAt: issueComments.deletedAt,
|
||||
deletedByType: issueComments.deletedByType,
|
||||
deletedByAgentId: issueComments.deletedByAgentId,
|
||||
deletedByUserId: issueComments.deletedByUserId,
|
||||
deletedByRunId: issueComments.deletedByRunId,
|
||||
sourceTrust: issueComments.sourceTrust,
|
||||
createdAt: issueComments.createdAt,
|
||||
})
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.companyId, input.companyId),
|
||||
issueId ? eq(issueComments.issueId, issueId) : undefined,
|
||||
inArray(issueComments.id, commentIds),
|
||||
),
|
||||
);
|
||||
const commentRows = input.wakeCommentRows
|
||||
? input.wakeCommentRows.filter(row => row.companyId === input.companyId && (!issueId || row.issueId === issueId) && commentIds.includes(row.id))
|
||||
: await readHeartbeatWakeComments(input.db, input.companyId, issueId, commentIds);
|
||||
|
||||
const commentsById = new Map(
|
||||
commentRows.map((comment) => [comment.id, comment]),
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import type { Db, agents, agentTaskSessions, heartbeatRuns, issueComments } from "@paperclipai/db";
|
||||
import type { Db, agents, agentTaskSessions, heartbeatRuns } from "@paperclipai/db";
|
||||
import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared";
|
||||
import type { PaperclipTurnContext } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { parseObject } from "../../adapters/utils.js";
|
||||
@@ -15,8 +15,9 @@ import { getTaskPlanContext } from "../task-plan-context.js";
|
||||
import { caseExecutionMarkdown, getCaseExecutionContext } from "../case-context.js";
|
||||
import type { TrustPresetResolution } from "../trust-preset-resolver.js";
|
||||
import { liveVoiceExecutionGuidance } from "../voice/voice-execution-guidance.js";
|
||||
import { buildPaperclipWakePayload, ConfigurationIncompleteFailure, PAPERCLIP_WAKE_PAYLOAD_KEY, type createHeartbeatRunPreparation } from "./run-preparation.js";
|
||||
import { buildPaperclipWakePayload, ConfigurationIncompleteFailure, readHeartbeatWakeComments, PAPERCLIP_WAKE_PAYLOAD_KEY, type createHeartbeatRunPreparation } from "./run-preparation.js";
|
||||
import { buildPaperclipTaskMarkdown } from "./task-markdown.js";
|
||||
import { deriveCommentId, extractWakeCommentIds } from "../../modules/run-dispatch/index.js";
|
||||
import { workspacePreferenceService } from "../workspace-preferences.js";
|
||||
|
||||
type RunPreparation = ReturnType<typeof createHeartbeatRunPreparation>;
|
||||
@@ -30,8 +31,6 @@ export interface HeartbeatTaskContextInput {
|
||||
"id" | "identifier" | "title" | "status" | "priority" | "workMode" |
|
||||
"conversationAgentId" | "description" | "projectId"> | null;
|
||||
issueContext: IssueContext | null;
|
||||
wakeCommentContext: Pick<typeof issueComments.$inferSelect,
|
||||
"id" | "body" | "sourceTrust" | "presentation" | "metadata"> | null;
|
||||
context: Record<string, unknown>;
|
||||
taskSession: Pick<typeof agentTaskSessions.$inferSelect, "lastRunId"> | null;
|
||||
};
|
||||
@@ -53,35 +52,9 @@ function readNonEmptyString(value: unknown): string | null {
|
||||
/** Rebuilds task prompt fields on the run's context; acquires no execution resources. */
|
||||
export async function prepareHeartbeatTaskContext(db: Db, input: HeartbeatTaskContextInput) {
|
||||
const { run, agent } = input;
|
||||
const { issueRef, issueContext, wakeCommentContext, context, taskSession } = input.task;
|
||||
const { issueRef, issueContext, context, taskSession } = input.task;
|
||||
const { trustPreset, experimentalInstanceSettings } = input.policy;
|
||||
const { issuesSvc, getPinnedSkillTestContext } = input.services;
|
||||
const continuationSummary = issueRef && !isConversation(issueContext)
|
||||
? await getIssueContinuationSummaryDocument(db, issueRef.id)
|
||||
: null;
|
||||
const exposeLowTrustRaw = trustPreset.kind === "low_trust_review";
|
||||
const safeContinuationSummary =
|
||||
continuationSummary && !exposeLowTrustRaw
|
||||
? redactQuarantinedBodyForHigherTrust(continuationSummary)
|
||||
: continuationSummary;
|
||||
const safeWakeCommentContext =
|
||||
wakeCommentContext && !exposeLowTrustRaw
|
||||
? sanitizeQuarantinedCommentForHigherTrust(wakeCommentContext)
|
||||
: wakeCommentContext;
|
||||
const issueAncestors = issueRef
|
||||
? await issuesSvc.getAncestors(issueRef.id)
|
||||
: [];
|
||||
if (continuationSummary) {
|
||||
context.paperclipContinuationSummary = {
|
||||
key: safeContinuationSummary!.key,
|
||||
title: safeContinuationSummary!.title,
|
||||
body: safeContinuationSummary!.body,
|
||||
sourceTrust: safeContinuationSummary!.sourceTrust ?? null,
|
||||
updatedAt: safeContinuationSummary!.updatedAt.toISOString(),
|
||||
};
|
||||
} else {
|
||||
delete context.paperclipContinuationSummary;
|
||||
}
|
||||
const admittedScope = parseObject(context.paperclipAgentJwtScope);
|
||||
if (!input.policy.recovering || (admittedScope.kind !== "standard" && admittedScope.kind !== "skill_test")) {
|
||||
const savedIssue = parseObject(context.paperclipIssue);
|
||||
@@ -93,10 +66,58 @@ export async function prepareHeartbeatTaskContext(db: Db, input: HeartbeatTaskCo
|
||||
: { kind: "standard" };
|
||||
}
|
||||
const isSkillTest = parseObject(context.paperclipAgentJwtScope).kind === "skill_test";
|
||||
const pinnedSkillTestContext =
|
||||
isSkillTest && issueRef
|
||||
? await getPinnedSkillTestContext(agent.companyId, issueRef.id)
|
||||
: null;
|
||||
const wakeCommentId = deriveCommentId(context, null);
|
||||
const commentIds = [...new Set([...extractWakeCommentIds(context), ...(wakeCommentId ? [wakeCommentId] : [])])];
|
||||
const [summary, ancestors, pinnedSkillTestRead, wakeComments] = await Promise.allSettled([
|
||||
issueRef && !isConversation(issueContext) ? getIssueContinuationSummaryDocument(db, issueRef.id) : Promise.resolve(null),
|
||||
issueRef ? issuesSvc.getAncestors(issueRef.id) : Promise.resolve([]),
|
||||
isSkillTest && issueRef ? getPinnedSkillTestContext(agent.companyId, issueRef.id) : Promise.resolve(null),
|
||||
readHeartbeatWakeComments(db, agent.companyId, readNonEmptyString(context.issueId), commentIds),
|
||||
]);
|
||||
if (summary.status === "rejected") throw summary.reason;
|
||||
if (ancestors.status === "rejected") throw ancestors.reason;
|
||||
if (pinnedSkillTestRead.status === "rejected") throw pinnedSkillTestRead.reason;
|
||||
if (wakeComments.status === "rejected") throw wakeComments.reason;
|
||||
const wakeCommentRows = wakeComments.value;
|
||||
const selectedWakeComment = issueRef && wakeCommentId ? wakeCommentRows.find(comment => comment.id === wakeCommentId && comment.issueId === issueRef.id) ?? null : null;
|
||||
const wakeCommentContext = selectedWakeComment ? {
|
||||
id: selectedWakeComment.id,
|
||||
body: selectedWakeComment.deletedAt ? "" : selectedWakeComment.body,
|
||||
authorType: selectedWakeComment.authorType,
|
||||
authorAgentId: selectedWakeComment.authorAgentId,
|
||||
authorUserId: selectedWakeComment.authorUserId,
|
||||
presentation: selectedWakeComment.deletedAt ? null : selectedWakeComment.presentation,
|
||||
metadata: selectedWakeComment.deletedAt ? null : selectedWakeComment.metadata,
|
||||
deletedAt: selectedWakeComment.deletedAt,
|
||||
deletedByType: selectedWakeComment.deletedByType,
|
||||
deletedByAgentId: selectedWakeComment.deletedByAgentId,
|
||||
deletedByUserId: selectedWakeComment.deletedByUserId,
|
||||
deletedByRunId: selectedWakeComment.deletedByRunId,
|
||||
sourceTrust: selectedWakeComment.sourceTrust,
|
||||
} : null;
|
||||
const continuationSummary = summary.value;
|
||||
const issueAncestors = ancestors.value;
|
||||
const pinnedSkillTestContext = pinnedSkillTestRead.value;
|
||||
const exposeLowTrustRaw = trustPreset.kind === "low_trust_review";
|
||||
const safeContinuationSummary =
|
||||
continuationSummary && !exposeLowTrustRaw
|
||||
? redactQuarantinedBodyForHigherTrust(continuationSummary)
|
||||
: continuationSummary;
|
||||
const safeWakeCommentContext =
|
||||
wakeCommentContext && !exposeLowTrustRaw
|
||||
? sanitizeQuarantinedCommentForHigherTrust(wakeCommentContext)
|
||||
: wakeCommentContext;
|
||||
if (continuationSummary) {
|
||||
context.paperclipContinuationSummary = {
|
||||
key: safeContinuationSummary!.key,
|
||||
title: safeContinuationSummary!.title,
|
||||
body: safeContinuationSummary!.body,
|
||||
sourceTrust: safeContinuationSummary!.sourceTrust ?? null,
|
||||
updatedAt: safeContinuationSummary!.updatedAt.toISOString(),
|
||||
};
|
||||
} else {
|
||||
delete context.paperclipContinuationSummary;
|
||||
}
|
||||
if (isSkillTest && !pinnedSkillTestContext) {
|
||||
throw new ConfigurationIncompleteFailure("The skill test's saved version is unavailable.", {
|
||||
configurationIncomplete: { reason: "skill_unavailable", actionUrl: "/skills" },
|
||||
@@ -132,6 +153,7 @@ export async function prepareHeartbeatTaskContext(db: Db, input: HeartbeatTaskCo
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
contextSnapshot: context,
|
||||
wakeCommentRows,
|
||||
continuationSummary,
|
||||
issueSummary: issueRef
|
||||
? {
|
||||
|
||||
@@ -497,7 +497,7 @@ else if(m.id!==undefined) send({id:m.id,result:{}});
|
||||
const [agent] = await server.db.select().from(agents).where(eq(agents.id, fixture.agentId));
|
||||
const context = structuredClone(run.contextSnapshot ?? {});
|
||||
await prepareHeartbeatTaskContext(server.db, { run, agent,
|
||||
task: { issueRef: issue, issueContext: null, wakeCommentContext: null, context, taskSession: null },
|
||||
task: { issueRef: issue, issueContext: null, context, taskSession: null },
|
||||
policy: { recovering: true, trustPreset: { kind: "standard" }, experimentalInstanceSettings: {} },
|
||||
services: { issuesSvc: { getAncestors: async () => [] },
|
||||
getPinnedSkillTestContext: createHeartbeatRunPreparation(server.db).getPinnedSkillTestContext },
|
||||
|
||||
@@ -5,9 +5,16 @@ import type { Db } from "@paperclipai/db";
|
||||
import type { AttentionSourceKind } from "@paperclipai/shared";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; issueId?: string; sourceId?: string } = {}) {
|
||||
export const ATTENTION_RETENTION_MAX_ISSUE_IDS = 100;
|
||||
export type AttentionRetentionScope = { companyId?: string; sourceKind?: AttentionSourceKind;
|
||||
issueId?: string; issueIds?: string[]; sourceId?: string };
|
||||
|
||||
export async function notifyAttentionRetentionWork(transaction: object, scope: AttentionRetentionScope = {}) {
|
||||
const issueIds = [...new Set([...(scope.issueIds ?? []), ...(scope.issueId ? [scope.issueId] : [])])].sort();
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify({
|
||||
companyId: scope.companyId, sourceKind: scope.sourceKind, sourceId: scope.sourceId, issueId: scope.issueId,
|
||||
companyId: scope.companyId, sourceKind: scope.sourceKind, sourceId: scope.sourceId,
|
||||
issueId: issueIds.length === 1 ? issueIds[0] : undefined,
|
||||
issueIds: issueIds.length > 1 && issueIds.length <= ATTENTION_RETENTION_MAX_ISSUE_IDS ? issueIds : undefined,
|
||||
}));
|
||||
}
|
||||
|
||||
@@ -47,18 +54,24 @@ export async function notifyIssueAttentionRetentionWork(transaction: object,
|
||||
and id::text in (${sql.join([...new Set(references)].map(id => sql`${id}`), sql`, `)})`);
|
||||
for (const row of trusted) seeds.push(row.id as string);
|
||||
}
|
||||
// UNION deduplicates visited UUIDs (including cycles). Each relation edge is
|
||||
// unique by schema, so 101 neighbours guarantee overflow of a 100-ID scope.
|
||||
// Bound that adjacency before materialization and stream at most 101 unique
|
||||
// IDs to the caller; an inner UNION DISTINCT would consume the full fan-in.
|
||||
const affected = await database.execute(sql`with recursive affected(id) as (
|
||||
select unnest(array[${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)}])
|
||||
union
|
||||
select parent.id from affected current cross join lateral (
|
||||
select related_issue_id as id from issue_relations where company_id = ${scope.companyId}::uuid
|
||||
and issue_id = current.id and type = 'blocks'
|
||||
union select parent_id from issues where company_id = ${scope.companyId}::uuid
|
||||
(select related_issue_id as id from issue_relations where company_id = ${scope.companyId}::uuid
|
||||
and issue_id = current.id and type = 'blocks' limit ${ATTENTION_RETENTION_MAX_ISSUE_IDS + 1})
|
||||
union all select parent_id from issues where company_id = ${scope.companyId}::uuid
|
||||
and id = current.id and parent_id is not null
|
||||
) parent
|
||||
) select id::text as "issueId" from affected`);
|
||||
for (const row of affected) await notifyAttentionRetentionWork(transaction, {
|
||||
companyId: scope.companyId, issueId: row.issueId as string,
|
||||
) select id::text as "issueId" from affected limit ${ATTENTION_RETENTION_MAX_ISSUE_IDS + 1}`);
|
||||
// The sentinel means the affected set is incomplete. Reconstruct the whole
|
||||
// company through the existing paged authority instead of dropping ancestors.
|
||||
await notifyAttentionRetentionWork(transaction, { companyId: scope.companyId,
|
||||
issueIds: affected.length <= ATTENTION_RETENTION_MAX_ISSUE_IDS ? affected.map(row => row.issueId as string) : undefined,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { createHmac, generateKeyPairSync, randomUUID, sign } from "node:crypto";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { requestSandboxAbsenceProof, verifySandboxAbsenceReceipt } from "./sandbox-absence-proof.js";
|
||||
import { inspectCloudSandboxCleanupAbsence, requestSandboxAbsenceProof, verifySandboxAbsenceReceipt,
|
||||
verifySandboxCleanupAbsenceReceipt } from "./sandbox-absence-proof.js";
|
||||
|
||||
const { privateKey, publicKey } = generateKeyPairSync("ed25519");
|
||||
const binding = { companyId: randomUUID(), issueId: randomUUID(), runId: randomUUID(), leaseId: randomUUID(),
|
||||
@@ -9,10 +10,10 @@ const env = { PAPERCLIP_CLOUD_STACK_ID: "fixture-stack", PAPERCLIP_CLOUD_API_ORI
|
||||
PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "fleet-token", PAPERCLIP_PRIMARY_HOST: "fixture.paperclip.test", PAPERCLIP_AUTH_SECRET: "fixture-secret",
|
||||
PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS: JSON.stringify({ keys: [{ ...publicKey.export({ format: "jwk" }), kid: "fixture", alg: "EdDSA", use: "sig" }] }) };
|
||||
const now = new Date("2099-01-01T00:00:00Z"), seconds = now.getTime() / 1000;
|
||||
function receipt(overrides: Record<string, unknown> = {}) {
|
||||
const header = Buffer.from(JSON.stringify({ alg: "EdDSA", kid: "fixture", typ: "paperclip-sandbox-absence+jwt" })).toString("base64url");
|
||||
const payload = Buffer.from(JSON.stringify({ v: 1, iss: "paperclip-cloud", aud: "paperclip-sandbox-absence/v1", sub: "fixture-stack",
|
||||
...binding, absent: true, iat: seconds, exp: seconds + 60, ...overrides })).toString("base64url");
|
||||
function receipt(overrides: Record<string, unknown> = {}, purpose = "sandbox-absence", source: Record<string, unknown> = binding) {
|
||||
const header = Buffer.from(JSON.stringify({ alg: "EdDSA", kid: "fixture", typ: `paperclip-${purpose}+jwt` })).toString("base64url");
|
||||
const payload = Buffer.from(JSON.stringify({ v: 1, iss: "paperclip-cloud", aud: `paperclip-${purpose}/v1`, sub: "fixture-stack",
|
||||
...source, absent: true, iat: seconds, exp: seconds + 60, ...overrides })).toString("base64url");
|
||||
return `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`), privateKey).toString("base64url")}`;
|
||||
}
|
||||
describe("sandbox absence proof", () => {
|
||||
@@ -45,3 +46,44 @@ describe("sandbox absence proof", () => {
|
||||
expect(fetchImpl).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("cleanup-only sandbox absence", () => {
|
||||
const cleanup = { purpose: "cleanup" as const, companyId: binding.companyId, environmentId: randomUUID(),
|
||||
leaseId: binding.leaseId, heartbeatRunId: null as string | null, provider: binding.provider, providerLeaseId: binding.providerLeaseId };
|
||||
const lease = { id: cleanup.leaseId, ...cleanup };
|
||||
const signed = (overrides: Record<string, unknown> = {}) => receipt(overrides, "sandbox-cleanup-absence", cleanup);
|
||||
it.each([null, binding.runId])("accepts exact absence with nullable run %s without issuing a run receipt", async heartbeatRunId => {
|
||||
const source = { ...cleanup, heartbeatRunId };
|
||||
const fetchImpl = vi.fn().mockResolvedValue(Response.json({ receipt: receipt({}, "sandbox-cleanup-absence", source) }));
|
||||
expect(await inspectCloudSandboxCleanupAbsence({ ...lease, heartbeatRunId }, { env, now, fetchImpl }))
|
||||
.toEqual({ providerLeaseId: cleanup.providerLeaseId, state: "destroyed" });
|
||||
const [url, init] = fetchImpl.mock.calls[0];
|
||||
expect(url).toBe("https://cloud.example.test/v1/tenant/sandbox-absence");
|
||||
expect(JSON.parse(init.body)).toEqual(source);
|
||||
expect(init.headers.authorization).not.toContain(env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN);
|
||||
expect(init.redirect).toBe("error");
|
||||
});
|
||||
it("requires every binding, current stack, signature and freshness, and rejects recovery receipts", () => {
|
||||
expect(verifySandboxCleanupAbsenceReceipt(signed(), cleanup, env, now)).toMatchObject(cleanup);
|
||||
for (const overrides of [
|
||||
...Object.keys(cleanup).map(key => ({ [key]: randomUUID() })),
|
||||
{ sub: "foreign-stack" }, { absent: false }, { aud: "paperclip-sandbox-absence/v1" }, { iss: "other" },
|
||||
{ iat: seconds + 31 }, { exp: seconds }, { exp: seconds + 61 },
|
||||
]) expect(verifySandboxCleanupAbsenceReceipt(signed(overrides), cleanup, env, now)).toBeNull();
|
||||
expect(verifySandboxCleanupAbsenceReceipt(signed().slice(0, -8) + "tampered", cleanup, env, now)).toBeNull();
|
||||
expect(verifySandboxCleanupAbsenceReceipt(receipt(), cleanup, env, now)).toBeNull();
|
||||
expect(verifySandboxAbsenceReceipt(signed(), binding, env, now)).toBeNull();
|
||||
});
|
||||
it("keeps old Clouds, outages, invalid replies and unconfigured instances inconclusive", async () => {
|
||||
for (const answer of [{ absent: false }, { absent: true }, { receipt: "forged" }, { receipt: signed({ heartbeatRunId: binding.runId }) }])
|
||||
expect(await inspectCloudSandboxCleanupAbsence(lease, { env, now, fetchImpl: vi.fn().mockResolvedValue(Response.json(answer)) })).toBeUndefined();
|
||||
for (const status of [400, 401, 403, 503]) expect(await inspectCloudSandboxCleanupAbsence(lease,
|
||||
{ env, now, fetchImpl: vi.fn().mockResolvedValue(new Response(null, { status })) })).toBeUndefined();
|
||||
expect(await inspectCloudSandboxCleanupAbsence(lease, { env, now, fetchImpl: vi.fn().mockRejectedValue(new Error("offline")) })).toBeUndefined();
|
||||
const fetchImpl = vi.fn();
|
||||
expect(await inspectCloudSandboxCleanupAbsence(lease, { env: {}, now, fetchImpl })).toBeUndefined();
|
||||
for (const change of [{ provider: "other" }, { environmentId: null }, { providerLeaseId: "invalid" }])
|
||||
expect(await inspectCloudSandboxCleanupAbsence({ ...lease, ...change }, { env, now, fetchImpl })).toBeUndefined();
|
||||
expect(fetchImpl).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -7,8 +7,18 @@ const bindingSchema = z.object({
|
||||
companyId: z.string().uuid(), issueId: z.string().uuid(), runId: z.string().uuid(),
|
||||
leaseId: z.string().uuid(), ownerToken: z.string().uuid(), provider: z.literal("daytona"),
|
||||
providerLeaseId: z.string().uuid(),
|
||||
});
|
||||
}).strict();
|
||||
const cleanupBindingSchema = z.object({
|
||||
purpose: z.literal("cleanup"), companyId: z.string().uuid(), environmentId: z.string().uuid(),
|
||||
leaseId: z.string().uuid(), heartbeatRunId: z.string().uuid().nullable(),
|
||||
provider: z.literal("daytona"), providerLeaseId: z.string().uuid(),
|
||||
}).strict();
|
||||
export type SandboxAbsenceBinding = z.infer<typeof bindingSchema>;
|
||||
export type SandboxCleanupAbsenceBinding = z.infer<typeof cleanupBindingSchema>;
|
||||
export type SandboxCleanupAbsenceProof = SandboxCleanupAbsenceBinding & {
|
||||
v: 1; iss: "paperclip-cloud"; aud: "paperclip-sandbox-cleanup-absence/v1"; sub: string;
|
||||
absent: true; iat: number; exp: number;
|
||||
};
|
||||
export type SandboxAbsenceProof = SandboxAbsenceBinding & {
|
||||
v: 1; iss: "paperclip-cloud"; aud: "paperclip-sandbox-absence/v1"; sub: string;
|
||||
absent: true; iat: number; exp: number;
|
||||
@@ -16,19 +26,30 @@ export type SandboxAbsenceProof = SandboxAbsenceBinding & {
|
||||
|
||||
export function verifySandboxAbsenceReceipt(receipt: string, binding: SandboxAbsenceBinding,
|
||||
env: NodeJS.ProcessEnv = process.env, now = new Date()): SandboxAbsenceProof | null {
|
||||
const parsed = bindingSchema.safeParse(binding);
|
||||
return parsed.success ? verifyReceipt(receipt, parsed.data, "sandbox-absence", env, now) as SandboxAbsenceProof | null : null;
|
||||
}
|
||||
|
||||
export function verifySandboxCleanupAbsenceReceipt(receipt: string, binding: SandboxCleanupAbsenceBinding,
|
||||
env: NodeJS.ProcessEnv = process.env, now = new Date()): SandboxCleanupAbsenceProof | null {
|
||||
const parsed = cleanupBindingSchema.safeParse(binding);
|
||||
return parsed.success ? verifyReceipt(receipt, parsed.data, "sandbox-cleanup-absence", env, now) as SandboxCleanupAbsenceProof | null : null;
|
||||
}
|
||||
|
||||
function verifyReceipt(receipt: string, binding: Record<string, unknown>, purpose: string,
|
||||
env: NodeJS.ProcessEnv, now: Date): Record<string, unknown> | null {
|
||||
try {
|
||||
const parsed = bindingSchema.safeParse(binding);
|
||||
if (receipt.length > 16_000 || !parsed.success) return null;
|
||||
if (receipt.length > 16_000) return null;
|
||||
const parts = receipt.split(".");
|
||||
if (parts.length !== 3 || parts.some(part => !/^[A-Za-z0-9_-]+$/.test(part))) return null;
|
||||
const header = JSON.parse(Buffer.from(parts[0], "base64url").toString("utf8"));
|
||||
if (header.alg !== "EdDSA" || header.typ !== "paperclip-sandbox-absence+jwt" || typeof header.kid !== "string") return null;
|
||||
if (header.alg !== "EdDSA" || header.typ !== `paperclip-${purpose}+jwt` || typeof header.kid !== "string") return null;
|
||||
if (!verify(null, Buffer.from(`${parts[0]}.${parts[1]}`, "ascii"), cloudRuntimePublicKeyForKid(env, header.kid), Buffer.from(parts[2], "base64url"))) return null;
|
||||
const proof = JSON.parse(Buffer.from(parts[1], "base64url").toString("utf8")) as SandboxAbsenceProof;
|
||||
const proof = JSON.parse(Buffer.from(parts[1], "base64url").toString("utf8"));
|
||||
const seconds = Math.floor(now.getTime() / 1000);
|
||||
if (proof.v !== 1 || proof.iss !== "paperclip-cloud" || proof.aud !== "paperclip-sandbox-absence/v1"
|
||||
if (proof.v !== 1 || proof.iss !== "paperclip-cloud" || proof.aud !== `paperclip-${purpose}/v1`
|
||||
|| !env.PAPERCLIP_CLOUD_STACK_ID || proof.sub !== env.PAPERCLIP_CLOUD_STACK_ID || proof.absent !== true
|
||||
|| Object.keys(parsed.data).some(key => proof[key as keyof SandboxAbsenceBinding] !== parsed.data[key as keyof SandboxAbsenceBinding])
|
||||
|| Object.keys(binding).some(key => proof[key] !== binding[key])
|
||||
|| !Number.isInteger(proof.iat) || !Number.isInteger(proof.exp) || proof.iat > seconds + 30
|
||||
|| proof.exp <= seconds || proof.exp <= proof.iat || proof.exp - proof.iat > 60) return null;
|
||||
return proof;
|
||||
@@ -38,11 +59,35 @@ export function verifySandboxAbsenceReceipt(receipt: string, binding: SandboxAbs
|
||||
export async function requestSandboxAbsenceProof(binding: SandboxAbsenceBinding, options: {
|
||||
env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch; now?: Date;
|
||||
} = {}): Promise<{ proof: SandboxAbsenceProof; receipt: string } | null> {
|
||||
const parsed = bindingSchema.safeParse(binding);
|
||||
if (!parsed.success) return null;
|
||||
const receipt = await requestReceipt(parsed.data, options);
|
||||
if (!receipt) return null;
|
||||
const proof = verifySandboxAbsenceReceipt(receipt, parsed.data, options.env, options.now);
|
||||
return proof ? { proof, receipt } : null;
|
||||
}
|
||||
|
||||
/** Cleanup-only observation: never grants continuation authority to a run. */
|
||||
export async function inspectCloudSandboxCleanupAbsence(lease: {
|
||||
id: string; companyId: string; environmentId: string | null; heartbeatRunId: string | null;
|
||||
provider: string | null; providerLeaseId: string | null;
|
||||
}, options: { env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch; now?: Date } = {}) {
|
||||
const parsed = cleanupBindingSchema.safeParse({ purpose: "cleanup", companyId: lease.companyId,
|
||||
environmentId: lease.environmentId, leaseId: lease.id, heartbeatRunId: lease.heartbeatRunId,
|
||||
provider: lease.provider, providerLeaseId: lease.providerLeaseId });
|
||||
if (!parsed.success) return undefined;
|
||||
const receipt = await requestReceipt(parsed.data, options);
|
||||
if (!receipt || !verifySandboxCleanupAbsenceReceipt(receipt, parsed.data, options.env, options.now)) return undefined;
|
||||
return { providerLeaseId: parsed.data.providerLeaseId, state: "destroyed" as const };
|
||||
}
|
||||
|
||||
async function requestReceipt(binding: Record<string, unknown>, options: {
|
||||
env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch;
|
||||
}): Promise<string | null> {
|
||||
const env = options.env ?? process.env;
|
||||
const context = getCloudStackContext(env);
|
||||
const secret = env.PAPERCLIP_AUTH_SECRET ?? env.BETTER_AUTH_SECRET;
|
||||
const parsed = bindingSchema.safeParse(binding);
|
||||
if (!context?.stackId || !context.primaryHost || !context.cloudOrigin || !secret?.trim() || !parsed.success) return null;
|
||||
if (!context?.stackId || !context.primaryHost || !context.cloudOrigin || !secret?.trim()) return null;
|
||||
try {
|
||||
const origin = new URL(context.cloudOrigin);
|
||||
const host = new URL(`https://${context.primaryHost}`);
|
||||
@@ -51,12 +96,11 @@ export async function requestSandboxAbsenceProof(binding: SandboxAbsenceBinding,
|
||||
const response = await (options.fetchImpl ?? fetch)(`${origin.origin}/v1/tenant/sandbox-absence`, {
|
||||
method: "POST", redirect: "error", signal: AbortSignal.timeout(10_000),
|
||||
headers: { authorization: `Bearer ${capability}`, "x-paperclip-cloud-tenant-host": host.hostname, "content-type": "application/json" },
|
||||
body: JSON.stringify(parsed.data),
|
||||
body: JSON.stringify(binding),
|
||||
});
|
||||
if (!response.ok) return null;
|
||||
const payload = await response.json() as { receipt?: unknown };
|
||||
if (typeof payload.receipt !== "string") return null;
|
||||
const proof = verifySandboxAbsenceReceipt(payload.receipt, binding, env, options.now);
|
||||
return proof ? { proof, receipt: payload.receipt } : null;
|
||||
return payload.receipt;
|
||||
} catch { return null; }
|
||||
}
|
||||
Reference in new issue
Block a user