mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 09:37:19 +01:00
Merge branch 'codex/sleep-09-aggregate' into codex/sleep-stack-09
* codex/sleep-09-aggregate: test: prove scheduler admission and ordinary subtree restore test: preserve recorded lease authority in startup acceptance test: exercise actual goal-startup failure isolation
This commit is contained in:
commit
f44da3eed8
4 files changed
+138
-4
No files matched your search
@@ -87,11 +87,21 @@ the raw/prepared real-PostgreSQL calibration remains.
|
||||
|
||||
Current assembled cases cover:
|
||||
|
||||
- A fixture-only goal-control INSERT error that previously aborted actual boot.
|
||||
With owner isolation, the goal row stays pending, a new real residual lease
|
||||
obligation is discovered within one second of child time and is cleaned by its
|
||||
separate owner, and ordinary attention ingestion still archives its source.
|
||||
Native classification remains fail-closed.
|
||||
- Actual readiness while a fixture-owned PostgreSQL lock holds the real startup
|
||||
retention INSERT. Health becomes ready with active work and pending SQL;
|
||||
releasing the lock produces the archive and zero active work. Readiness alone
|
||||
therefore does not stand in for idle eligibility.
|
||||
- Disabled-heartbeat empty boot across two simulated hourly windows.
|
||||
- Actual worktree suppression with a committed residual obligation and no
|
||||
recurring SQL over two minutes of child time. A normal operator settings
|
||||
activation reconstructs that obligation and produces a durable cleanup result.
|
||||
Activation uses the stored instance identity and timestamp; no direct worker
|
||||
wake or safety-gate bypass is used.
|
||||
- Ordinary HTTP company/settings/completed-task creation, independent durable
|
||||
inspection, drain/resume and two additional hourly windows without owner work.
|
||||
- An ordinary HTTP issue insertion held by an isolated PostgreSQL lock while
|
||||
|
||||
@@ -47,7 +47,7 @@ try {
|
||||
const server = await startServer();
|
||||
const app = server.server.listeners("request")[0] as unknown as { locals: { paperclipDb: Db } };
|
||||
process.send?.({ kind: "ready", port: server.listenPort });
|
||||
process.on("message", async (message: { id: number; command: string; ms?: number; issueUpdate?: { id: string; status: "done" }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } }) => {
|
||||
process.on("message", async (message: { id: number; command: string; ms?: number; issueUpdate?: { id: string; status: "done" }; orphanLease?: { companyId: string }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } }) => {
|
||||
try {
|
||||
if (message.command === "snapshot" || message.command === "advance" || message.command === "drain" || message.command === "resume" || message.command === "freeze" || message.command === "realtime") {
|
||||
if (message.command === "freeze" && bootClock) { clearRealInterval(bootClock); bootClock = realInterval(() => mock.timers.tick(0), 10); }
|
||||
@@ -60,6 +60,17 @@ try {
|
||||
if (message.command === "resume") stopTaskDrain();
|
||||
if (fakeClock) mock.timers.tick(0);
|
||||
process.send?.({ id: message.id, kind: "snapshot", idle: idleWorkSnapshot(), startup: getStartupRecoveryState(), outgoingHttpRequests, network: network(), now: Date.now() });
|
||||
} else if (message.command === "createOrphanLease") {
|
||||
if (!message.orphanLease) throw new Error("Missing orphan lease fixture");
|
||||
const { environmentLeases } = await import("@paperclipai/db");
|
||||
const { writeEnvironmentCleanup } = await import("../../../services/environment-cleanup-notifications.js");
|
||||
const old = new Date(Date.now() - 6 * 60_000);
|
||||
const companyId = message.orphanLease.companyId;
|
||||
const [lease] = await writeEnvironmentCleanup(app.locals.paperclipDb, tx => tx.insert(environmentLeases).values({
|
||||
companyId, provider: "local", providerLeaseId: null, status: "active", leasePolicy: "ephemeral", metadata: { driver: "local" },
|
||||
acquiredAt: old, lastUsedAt: old, createdAt: old, updatedAt: old,
|
||||
}).returning(), { companyId });
|
||||
process.send?.({ id: message.id, kind: "leaseCreated", leaseId: lease!.id });
|
||||
} else if (message.command === "updateIssue") {
|
||||
if (!message.issueUpdate) throw new Error("Missing issue update");
|
||||
const { issueService } = await import("../../../services/issues.js");
|
||||
|
||||
@@ -93,7 +93,7 @@ export async function startMaintenanceServer(databaseUrl: string, options: { ena
|
||||
for (const request of requests.values()) request.reject(error);
|
||||
requests.clear();
|
||||
});
|
||||
async function request<T>(command: string, payload: { ms?: number; issueUpdate?: { id: string; status: "done" }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } } = {}): Promise<T> {
|
||||
async function request<T>(command: string, payload: { ms?: number; issueUpdate?: { id: string; status: "done" }; orphanLease?: { companyId: string }; publication?: { companyId: string; scopeDigest: string; backupRoot: string; pendingRoot: string } } = {}): Promise<T> {
|
||||
const id = ++requestId;
|
||||
const result = new Promise<T>((resolve, reject) => requests.set(id, { resolve, reject }));
|
||||
child.send({ id, command, ...payload });
|
||||
@@ -124,6 +124,11 @@ export async function startMaintenanceServer(databaseUrl: string, options: { ena
|
||||
return value as T;
|
||||
} finally { if (options.clock) await request("freeze"); }
|
||||
},
|
||||
async createOrphanLease(companyId: string) {
|
||||
if (options.clock) await request("realtime");
|
||||
try { return await request<{ leaseId: string }>("createOrphanLease", { orphanLease: { companyId } }); }
|
||||
finally { if (options.clock) await request("freeze"); }
|
||||
},
|
||||
async updateIssue(issueUpdate: { id: string; status: "done" }) {
|
||||
if (options.clock) await request("realtime");
|
||||
try { return await request("updateIssue", { issueUpdate }); }
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createCipheriv, createHash, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, sign, verify } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
import { agents, agentApiKeys, agentTaskSessions, authUsers, companies, companyMemberships, connectionGrants, createDb, heartbeatRuns, issues, issueThreadInteractions, toolApplications, toolConnections } from "@paperclipai/db";
|
||||
import { agents, agentApiKeys, agentSessionGoalActions, agentTaskSessions, authUsers, companies, companyMemberships, connectionGrants, createDb, heartbeatRuns, issues, issueThreadInteractions, toolApplications, toolConnections } from "@paperclipai/db";
|
||||
import { access, mkdir, readdir, utimes, writeFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
@@ -65,6 +65,100 @@ describe("assembled maintenance server production wiring", () => {
|
||||
repeatedQueries: repeated.length, repeated: receipt(repeated), initialFrames, finalFrames: server.proxy.frames, outgoingHttpRequests: 0 }, null, 2));
|
||||
} finally { await server.stop(); }
|
||||
}, 150_000);
|
||||
it("reconstructs residual work from a normal scheduler-admission change after suppression", async () => {
|
||||
await clearDatabase();
|
||||
const server = await startMaintenanceServer(databaseUrl!, { clock: true, suppressed: true });
|
||||
const fixture = createDb(databaseUrl!, { maxConnections: 1 });
|
||||
try {
|
||||
await server.settled();
|
||||
const company = await server.api<{ id: string }>("POST", "/companies", { name: "Admission reconstruction" });
|
||||
const { leaseId } = await server.createOrphanLease(company.id);
|
||||
await server.advance(1_000); await settleClock(server);
|
||||
const [suppressed] = await fixture.$client`select status, cleanup_status from environment_leases where id=${leaseId}`;
|
||||
expect(suppressed).toMatchObject({ status: "active", cleanup_status: null });
|
||||
const baseline = server.proxy.queries.length, before = await server.snapshot();
|
||||
for (let step = 0; step < 4; step++) { await server.advance(30_000); await server.settled(10); }
|
||||
expect(server.proxy.queries.length).toBe(baseline);
|
||||
expect((await server.snapshot()).outgoingHttpRequests).toBe(before.outgoingHttpRequests);
|
||||
await server.api("PATCH", "/instance/settings/experimental", { enableWorktreeRunExecution: true });
|
||||
await server.advance(1_000); await settleClock(server);
|
||||
const [admitted] = await fixture.$client`select experimental from instance_settings where singleton_key='default'`;
|
||||
expect(admitted!.experimental).toMatchObject({ enableWorktreeRunExecution: true,
|
||||
worktreeRunExecutionActivatedAt: expect.any(String), worktreeRunExecutionActivationInstanceId: "maintenance-test" });
|
||||
const [reconstructed] = await fixture.$client`select status, failure_reason, cleanup_status from environment_leases where id=${leaseId}`;
|
||||
expect(reconstructed).toMatchObject({ status: "expired", failure_reason: "pending_cleanup_retry", cleanup_status: "success" });
|
||||
await server.advance(30_000); await settleClock(server);
|
||||
const [cleaned] = await fixture.$client`select status, cleanup_status from environment_leases where id=${leaseId}`;
|
||||
expect(cleaned).toMatchObject({ status: "expired", cleanup_status: "success" });
|
||||
await writeFile(path.join(server.directory, "admission-measurement.json"), JSON.stringify({ phase: "ordinary-settings-admission-reconstruction",
|
||||
companyId: company.id, leaseId, suppressed, suppressedClockMs: 120_000, suppressedQueries: 0,
|
||||
admitted: admitted!.experimental, reconstructed, cleaned, frames: server.proxy.frames,
|
||||
boundary: "actual suppressed runtime, committed environment-cleanup ingress, normal operator activation; no direct worker wake or safety-gate bypass",
|
||||
}, null, 2));
|
||||
} finally { await server.stop(); await fixture.$client.end(); }
|
||||
}, 150_000);
|
||||
|
||||
it("isolates a real goal-startup write failure and still registers residual and attention owners", async () => {
|
||||
await clearDatabase();
|
||||
const fixture = createDb(databaseUrl!, { maxConnections: 1 });
|
||||
const companyId = randomUUID(), agentId = randomUUID(), userId = randomUUID(), issueId = randomUUID(), requestId = randomUUID();
|
||||
await fixture.insert(authUsers).values({ id: userId, name: "Goal control owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await fixture.insert(companies).values({ id: companyId, name: "Startup failure isolation", issuePrefix: "MFI", defaultResponsibleUserId: userId });
|
||||
await fixture.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" });
|
||||
await fixture.insert(agents).values({ id: agentId, companyId, name: "Goal owner", adapterType: "codex_local", status: "idle", lifecycleState: "ready" });
|
||||
await fixture.insert(issues).values({ id: issueId, companyId, title: "Retained terminal goal task", status: "done", assigneeAgentId: agentId, responsibleUserId: userId });
|
||||
const [session] = await fixture.insert(agentTaskSessions).values({ companyId, agentId, taskKey: issueId, adapterType: "codex_local" }).returning();
|
||||
const [action] = await fixture.insert(agentSessionGoalActions).values({ companyId, sessionId: session!.id, requestId,
|
||||
action: "clear", payloadJson: { requestId, action: "clear" }, status: "pending" }).returning();
|
||||
await fixture.$client.unsafe(`create or replace function maintenance_fail_goal_startup() returns trigger language plpgsql as $$begin
|
||||
if new.reason = 'goal_control' then raise exception 'maintenance_goal_startup_injected'; end if; return new; end$$`);
|
||||
await fixture.$client.unsafe(`create trigger maintenance_fail_goal_startup before insert on agent_wakeup_requests for each row execute function maintenance_fail_goal_startup()`);
|
||||
let server: Awaited<ReturnType<typeof startMaintenanceServer>> | undefined;
|
||||
try {
|
||||
server = await startMaintenanceServer(databaseUrl!, { clock: true });
|
||||
await settleClock(server);
|
||||
expect((await server.snapshot()).startup.phase).toBe("ready");
|
||||
expect(server.output).toContain("maintenance_goal_startup_injected");
|
||||
const [retained] = await fixture.$client`select status from agent_session_goal_actions where id=${action!.id}`;
|
||||
expect(retained!.status).toBe("pending");
|
||||
const [failedWake] = await fixture.$client`select count(*)::integer as count from agent_wakeup_requests where reason='goal_control'`;
|
||||
expect(failedWake!.count).toBe(0);
|
||||
const start = (await server.snapshot()).now;
|
||||
const { leaseId } = await server.createOrphanLease(companyId);
|
||||
let discovered: Record<string, unknown> | undefined;
|
||||
for (let step = 0; step < 10; step++) {
|
||||
await server.advance(1_000); await delay(50);
|
||||
[discovered] = await fixture.$client`select status, failure_reason, provider, provider_lease_id from environment_leases where id=${leaseId}`;
|
||||
if (discovered?.status === "pending_cleanup") break;
|
||||
}
|
||||
expect(discovered).toMatchObject({ status: "pending_cleanup", failure_reason: "orphaned_active_lease_recovered", provider: "local", provider_lease_id: null });
|
||||
const elapsedClockMs = (await server.snapshot()).now - start;
|
||||
expect(elapsedClockMs).toBeLessThan(30_000);
|
||||
await server.advance(30_000); await settleClock(server);
|
||||
const [lease] = await fixture.$client`select status, cleanup_status, provider, provider_lease_id from environment_leases where id=${leaseId}`;
|
||||
expect(lease).toMatchObject({ status: "expired", cleanup_status: "success", provider: "local", provider_lease_id: null });
|
||||
await fixture.$client.unsafe(`create or replace function maintenance_age_isolation_attention() returns trigger language plpgsql as $$begin
|
||||
if new.title = 'Attention after goal failure' then new.updated_at = now() - interval '91 days'; end if; return new; end$$`);
|
||||
await fixture.$client.unsafe(`create trigger maintenance_age_isolation_attention before insert on issue_thread_interactions for each row execute function maintenance_age_isolation_attention()`);
|
||||
const issue = await server.api<{ id: string }>("POST", `/companies/${companyId}/issues`, { title: "Independent attention owner", status: "todo" });
|
||||
const card = await server.api<{ id: string }>("POST", `/issues/${issue.id}/interactions`, { kind: "request_confirmation", title: "Attention after goal failure",
|
||||
resolverPolicy: "human_only", continuationPolicy: "none", payload: { version: 1, prompt: "Choose a color?" } });
|
||||
await server.advance(1_000); await settleClock(server);
|
||||
const [archived] = await fixture.$client`select archived_reason, archived_at is not null as archived from decision_retention where source_id=${card.id}`;
|
||||
expect(archived).toMatchObject({ archived_reason: "idle_ttl", archived: true });
|
||||
await writeFile(path.join(server.directory, "startup-failure-measurement.json"), JSON.stringify({ phase: "actual-goal-startup-failure-isolation",
|
||||
companyId, actionId: action!.id, retained, failedWake, leaseId, discovered, lease, elapsedClockMs, archived, frames: server.proxy.frames,
|
||||
boundary: "fixture-only conditional goal_control INSERT failure; actual native classification untouched; actual environment cleanup ingress and ordinary attention HTTP producer",
|
||||
}, null, 2));
|
||||
} finally {
|
||||
await fixture.$client.unsafe("drop trigger if exists maintenance_fail_goal_startup on agent_wakeup_requests").catch(() => {});
|
||||
await fixture.$client.unsafe("drop function if exists maintenance_fail_goal_startup()").catch(() => {});
|
||||
await fixture.$client.unsafe("drop trigger if exists maintenance_age_isolation_attention on issue_thread_interactions").catch(() => {});
|
||||
await fixture.$client.unsafe("drop function if exists maintenance_age_isolation_attention()").catch(() => {});
|
||||
await server?.stop(); await fixture.$client.end();
|
||||
}
|
||||
}, 150_000);
|
||||
|
||||
it("exposes readiness while real deferred housekeeping is held and settles its work after release", async () => {
|
||||
await clearDatabase();
|
||||
const fixture = createDb(databaseUrl!, { maxConnections: 1 });
|
||||
@@ -610,12 +704,26 @@ describe("assembled maintenance server production wiring", () => {
|
||||
const [projected] = await fixture.$client`select rr.company_id, rr.status, rr.completed_at is not null as completed,
|
||||
i.status as issue_status from routine_runs rr join issues i on i.id=rr.linked_issue_id where rr.id=${run.id}`;
|
||||
expect(projected).toMatchObject({ company_id: companyId, status: "completed", completed: true, issue_status: "done" });
|
||||
const restoredRun = await server.api<{ id: string; linkedIssueId: string }>("POST", `/routines/${routine.id}/run`, {
|
||||
source: "manual", idempotencyKey: randomUUID(),
|
||||
});
|
||||
const cancel = await server.api<{ hold: { id: string } }>("POST", `/issues/${restoredRun.linkedIssueId}/tree-holds`, { mode: "cancel", reason: "Ordinary cancel before restore" });
|
||||
await server.advance(1_000); await settleClock(server);
|
||||
const [cancelled] = await fixture.$client`select status from routine_runs where id=${restoredRun.id}`;
|
||||
expect(cancelled!.status).toBe("failed");
|
||||
await server.api("POST", `/issues/${restoredRun.linkedIssueId}/tree-holds`, { mode: "restore", reason: "Ordinary restore" });
|
||||
await server.advance(1_000); await settleClock(server);
|
||||
const [restored] = await fixture.$client`select rr.status, rr.completed_at, rr.failure_reason, i.status as issue_status
|
||||
from routine_runs rr join issues i on i.id=rr.linked_issue_id where rr.id=${restoredRun.id}`;
|
||||
expect(restored).toMatchObject({ status: "issue_created", completed_at: null, failure_reason: null, issue_status: "todo" });
|
||||
const [cancelHold] = await fixture.$client`select status from issue_tree_holds where id=${cancel.hold.id}`;
|
||||
expect(cancelHold!.status).toBe("released");
|
||||
const [executions] = await fixture.$client`select count(*)::integer as count from heartbeat_runs where company_id=${companyId}`;
|
||||
expect(executions!.count).toBe(0);
|
||||
expect(storageRequests).toEqual([]);
|
||||
await writeFile(path.join(server.directory, "routine-measurement.json"), JSON.stringify({
|
||||
phase: "ordinary-routine-service-completion-during-drain", completionBoundary: "actual issueService.update with runtime-owned DB; excludes redundant route projection hook", companyId, routineId: routine.id, runId: run.id,
|
||||
before, drained, projected, executions: executions!.count, storageRequests,
|
||||
before, drained, projected, restoredRun, cancelled, restored, cancelHold, executions: executions!.count, storageRequests,
|
||||
suppression: "PAPERCLIP_IN_WORKTREE; actual production policy", frames: server.proxy.frames,
|
||||
}, null, 2));
|
||||
} finally {
|
||||
|
||||
Reference in new issue
Block a user