mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): recover captured instruction saves from committed deadlines (#17670)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A stopped run can leave captured instruction edits waiting for a persistent save. > - Existing working-copy records retain the candidate bytes, attempt count, and retry deadline. > - The orphan reaper scans those records on every heartbeat, even when no save is due. > - This PR wakes the existing save owner after committed changes and at saved deadlines. > - The worker preserves permissions, conflicts, and bounded retries, then becomes quiet. ## Linked Issues or Issue Description Refs #17667, the merged preceding PR in this recurring-server-work stack. This PR targets master. Related warm-file ownership work: #14695. Its process-stop and retained-directory guarantees remain with their existing owner. The dedup search found no equivalent captured-save scheduling migration. **What happened?** The commit portion of `recoverCaptured` runs inside the periodic orphan sweep. Prepared copies, completed history, conflicts, and exhausted candidates do not need repeated save discovery. A saved candidate does need recovery after restart or a failed foreground save. **Expected behavior** Committed candidate changes reactivate recovery. The worker honors the saved deadline and existing three-attempt bound. It stops querying after settlement or deliberate conflict/exhaustion. Permission checks and canonical save semantics remain in the revision service. **Steps to reproduce** 1. Start a server with heartbeat scheduling enabled and no captured save pending. 2. Observe repeated pending-commit queries from the orphan reaper. 3. Capture edited instructions and fail their first canonical file replacement. The durable candidate must be saved after its retry deadline without starting a provider. **Paperclip version or commit** Reviewed parent `405e533394b26215e22d0452e9b0207d3f784a79`. Actual parent squash and current master baseline `27270e9bf8b7a294cf03f8b84f4857f1b1ff05c7`. **Deployment mode** Self-hosted server with the existing single-process transaction signal model. Local verification uses PostgreSQL 17 and actual managed instruction files. ## What Changed - Preserve foreground instruction-save ownership and its failure deadline before recovery spends another attempt. - Add a bounded captured-commit worker inside the instruction-copy owner. Existing rows and deadlines supply its state. No new table, migration, or external scheduler is added. - Signal writes entering or leaving pending commit from the existing copy mutation boundary. Only these writes gain a short transaction. File collection, file replacement, and provider operations stay outside that new transaction. - Signal company/agent deletion and worktree execution setting changes. Preserve drain, standby, startup reconstruction, shutdown join, and uncertain-transaction handling through the shared coordinator. - Guard attempt claims with current copy ownership and a still-due deadline. A concurrent writer that postpones the retry wins without consuming an attempt. Manual/startup recovery uses the same guarded owner. - Remove only captured commits from periodic reaping. Stopped collection and the separate post-save directory cleanup sweep remain for later slices. - Add real producer, file, transaction, concurrency, and lifecycle tests. Add guarded external test databases to the two filesystem suites so they can run where embedded PostgreSQL is unavailable. ## Verification Published head: `915ef6f20a68fffe89711cc97a6196521ea9885d`. Source acceptance: `9fb75e508f06b05b740099403ce021a61e7c6603`. The actual #17667 squash ancestry is recorded by a normal merge. Its entire result tree equals the prior accepted descendant tree; the final commit changes the audit only. - All 268 cases in six complete files pass, without skips, in 41.89 seconds on the updated parent. Coverage includes captured saves, directory consumers, startup, instruction-copy lifecycle boundaries, recovery and run lifecycle. - Four unused fresh PostgreSQL 17 databases migrated through parent 0406 isolate the destructive fixtures. Three use supported external URL paths. The embedded-only lifecycle fixture temporarily uses its own external database on this IPC-exhausted host, then is restored byte for byte. Assertions and original timing limits are unchanged. CI must still pass with the original embedded fixture. - Actual database build, server typecheck/build, module boundaries and exact-current migration order pass. Commands select each operation by name. The earlier local no-projects filter error remains recorded and is not treated as a build pass. - The normal parent merge preserves both complete added/removed patches and every unrelated upstream blob. The entire database subtree is parent-identical, with no owned schema or migration changes. New parent worker behavior has current consumer/startup acceptance here and separate owner acceptance in #17667. - Earlier foreground ownership, postponed deadline, attempt cap, drain hold, fixture and skipped-test findings remain in the audit. The prior head's Greptile 5/5, Ponytail 5/5 and green CI are historical after this new publication. - Current-head Greptile and Ponytail both give 5/5. All 58 checks pass. Prior stacked-head scores are historical; the new published SHA requires fresh gates. These checks do not establish deployed host/database suspension or external wake delivery. ## Risks - Signals cover supported in-process writers. Independent SQL writers and other replicas require a separate ingress contract. Startup reconstructs persisted obligations. - Exhausted candidates stay preserved under the existing policy; the worker does not increase retry limits or silently discard saved edits. Database reconstruction failures retain shared backoff. - Inventory pages scalar fields but retains candidate identifiers in memory. The query-plan result covers completed history, not a constant-cost promise for arbitrary exhausted-pending history. - File saves can outlive a stop request once admitted. Shutdown joins them. Exact-byte replay and current authorization remain with the revision owner. - Other instruction recovery duties and other stack blockers remain. This PR does not prove host suspension, database suspension, or external wake delivery. An ordinary protected merge is authorized only after current review and CI gates pass. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, shell tools, and local PostgreSQL/filesystem tests. The runtime does not expose a more precise model identifier or context-window size for this turn. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Review follow-up: remove seven newly added comment lines to satisfy CONTRIBUTING.md, preserving the explanations in the owning audit. Both changed files retain the exact non-comment TypeScript token stream; other source bytes are unchanged. The original cffe server-shard-9 run ended after a runner shutdown signal and cancellation, without a final assertion verdict; that failure is retained and not counted as a pass. No old-head retry is requested. Fresh current-head CI and Greptile are required. --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
27270e9bf8
commit
b294de546b
14 files changed
+879
-25
No files matched your search
@@ -1,3 +1,4 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import fs from "node:fs/promises";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
@@ -22,6 +23,8 @@ import { AgentDirectoryReuseInvalidatedError, agentDirectoryWorkingCopyService }
|
||||
import { withDirectoryMergeLock } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { heartbeatRunEvents } from "@paperclipai/db";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_DIRECTORY_TEST_DATABASE_URL");
|
||||
|
||||
describe("persistent agent directories", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
@@ -42,13 +45,14 @@ describe("persistent agent directories", () => {
|
||||
beforeAll(async () => {
|
||||
home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "instruction-working-copies-")));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-");
|
||||
db = createDb(database.connectionString);
|
||||
if (!external) database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-");
|
||||
db = createDb(external ?? database.connectionString);
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
revisions = agentInstructionRevisionService(db);
|
||||
}, 90_000);
|
||||
afterAll(async () => {
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
|
||||
await db?.$client.end({ timeout: 1 });
|
||||
await database?.cleanup();
|
||||
if (home) {
|
||||
const writable = async (dir: string) => {
|
||||
|
||||
@@ -1,3 +1,9 @@
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { notifyInstructionCommit } from "../services/instruction-commit-notifications.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import fs from "node:fs/promises";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
@@ -8,9 +14,9 @@ const execFile = promisify(execFileCallback);
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, agentInstructionWorkingCopies, createDb } from "@paperclipai/db";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { activityLog, agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, agentInstructionWorkingCopies, createDb, withDedicatedDbConnection } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { agentInstructionRevisionService } from "../services/agent-instruction-revisions.js";
|
||||
import { agentInstructionWorkingCopyService } from "../services/agent-instruction-working-copies.js";
|
||||
@@ -18,6 +24,8 @@ import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js";
|
||||
import { buildNativeRuntimeContext } from "../services/native-runtime/runtime-context.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_INSTRUCTION_COPY_TEST_DATABASE_URL");
|
||||
|
||||
describe("registered run instruction copies", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
@@ -38,13 +46,14 @@ describe("registered run instruction copies", () => {
|
||||
beforeAll(async () => {
|
||||
home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "instruction-working-copies-")));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-");
|
||||
db = createDb(database.connectionString);
|
||||
if (!external) database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-");
|
||||
db = createDb(external ?? database.connectionString);
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
revisions = agentInstructionRevisionService(db);
|
||||
}, 90_000);
|
||||
afterAll(async () => {
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
|
||||
await db?.$client.end({ timeout: 1 });
|
||||
await database?.cleanup();
|
||||
if (home) {
|
||||
const writable = async (dir: string) => {
|
||||
@@ -70,6 +79,334 @@ describe("registered run instruction copies", () => {
|
||||
await fs.writeFile(path.join(root, entryFile), initial);
|
||||
});
|
||||
|
||||
describe("captured instruction commit worker", () => {
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers(); vi.restoreAllMocks();
|
||||
|
||||
await db.update(agentInstructionWorkingCopies).set({ state: "saved", nextAttemptAt: null })
|
||||
.where(eq(agentInstructionWorkingCopies.companyId, companyId));
|
||||
});
|
||||
const clock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
async function captured(copy: NonNullable<Awaited<ReturnType<typeof copies.get>>>, dueAt = new Date(), owner = db) {
|
||||
await owner.transaction(async tx => {
|
||||
await notifyInstructionCommit(tx, { companyId: copy.companyId, runId: copy.runId });
|
||||
await tx.update(agentInstructionWorkingCopies).set({ state: "pending_commit", attempts: 1,
|
||||
candidateBase64: Buffer.from("recovered bytes").toString("base64"), nextAttemptAt: dueAt,
|
||||
processStoppedAt: new Date(),
|
||||
}).where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
});
|
||||
}
|
||||
async function saved(runId: string) {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect((await copies.get(companyId, runId))?.state).toBe("saved");
|
||||
}, { timeout: 5000 });
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("recovered bytes");
|
||||
}
|
||||
function coordinator(canRun = () => true, canReconcile = () => true) {
|
||||
return createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile, onError: error => { throw error; } });
|
||||
}
|
||||
it("stays quiet for prepared copies, completed history, conflicts, and exhausted candidates", async () => {
|
||||
for (const state of ["prepared", "saved", "conflict", "pending_commit"]) {
|
||||
const copy = await run();
|
||||
await db.update(agentInstructionWorkingCopies).set({ state, attempts: 3 }).where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
}
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update");
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("recovers a real capture at its saved deadline after a transient save failure", async () => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes");
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
const rename = fs.rename.bind(fs);
|
||||
let fail = true;
|
||||
const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => {
|
||||
if (fail && to === path.join(root, entryFile)) { fail = false; throw new Error("temporary file replacement failure"); }
|
||||
return rename(from, to);
|
||||
});
|
||||
const pending = await copies.collectStopped({ companyId, runId: copy.runId });
|
||||
expect(pending).toMatchObject({ state: "pending_commit", attempts: 1, nextAttemptAt: expect.any(Date) });
|
||||
renameSpy.mockRestore();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBe(pending!.nextAttemptAt!.getTime());
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await saved(copy.runId);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it.each(["restart", "drain", "standby"] as const)("reconstructs a saved candidate across %s", async mode => {
|
||||
const copy = await run();
|
||||
if (mode === "restart") await captured(copy);
|
||||
let admitted = mode === "restart";
|
||||
clock(); const worker = coordinator(() => admitted, () => mode !== "standby");
|
||||
try {
|
||||
await agentInstructionWorkingCopyService(db).registerCapturedCommitWork(worker, () => admitted, async () => true).ready;
|
||||
if (mode !== "restart") {
|
||||
await captured(copy);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
if (mode !== "standby") await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const reads = vi.spyOn(db, "select");
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
admitted = true; await vi.advanceTimersByTimeAsync(30_000);
|
||||
}
|
||||
await saved(copy.runId);
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("preserves the existing attempt cap and becomes quiet with the candidate retained", async () => {
|
||||
const copy = await run(); await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes");
|
||||
clock(); const worker = coordinator();
|
||||
const rename = fs.rename.bind(fs);
|
||||
let release!: () => void, entered!: () => void;
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
const saving = new Promise<void>(resolve => { entered = resolve; });
|
||||
let destinationWrites = 0;
|
||||
const writes = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => {
|
||||
if (to === path.join(root, entryFile)) {
|
||||
if (++destinationWrites === 2) { entered(); await held; }
|
||||
throw new Error("file replacement still unavailable");
|
||||
}
|
||||
return rename(from, to);
|
||||
});
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
expect((await copies.collectStopped({ companyId, runId: copy.runId }))?.attempts).toBe(1);
|
||||
for (const attempts of [2, 3]) {
|
||||
const previous = await copies.get(companyId, copy.runId);
|
||||
expect(previous?.nextAttemptAt).toBeInstanceOf(Date);
|
||||
await vi.advanceTimersByTimeAsync(Math.max(0, previous!.nextAttemptAt!.getTime() - Date.now()) + 1);
|
||||
if (attempts === 2) {
|
||||
await saving;
|
||||
|
||||
|
||||
expect((await copies.get(companyId, copy.runId))?.attempts).toBe(2);
|
||||
await vi.advanceTimersByTimeAsync(31_000);
|
||||
release();
|
||||
}
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
const row = await copies.get(companyId, copy.runId);
|
||||
expect(row?.attempts).toBe(attempts);
|
||||
if (attempts < 3) expect(row!.nextAttemptAt!.getTime()).toBeGreaterThan(Date.now());
|
||||
else expect(row?.nextAttemptAt).toBeNull();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
const count = writes.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(300_000);
|
||||
expect(writes.mock.calls.length).toBe(count);
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_commit", attempts: 3,
|
||||
candidateBase64: Buffer.from("recovered bytes").toString("base64"), nextAttemptAt: null });
|
||||
} finally { release(); writes.mockRestore(); await worker.stop(); }
|
||||
});
|
||||
it("joins an active save during shutdown and reconstructs its remaining receipt", async () => {
|
||||
const copy = await run(); await captured(copy);
|
||||
clock(); const worker = coordinator();
|
||||
const rename = fs.rename.bind(fs);
|
||||
let release!: () => void, entered!: () => void;
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
const reached = new Promise<void>(resolve => { entered = resolve; });
|
||||
const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => {
|
||||
if (to === path.join(root, entryFile)) { entered(); await held; }
|
||||
return rename(from, to);
|
||||
});
|
||||
let stopped: Promise<void> | undefined;
|
||||
try {
|
||||
const registration = copies.registerCapturedCommitWork(worker, () => true, async () => true);
|
||||
await reached;
|
||||
let complete = false;
|
||||
stopped = worker.stop().then(() => { complete = true; });
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(complete).toBe(false);
|
||||
release(); await stopped; await registration.ready;
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("recovered bytes");
|
||||
expect((await copies.get(companyId, copy.runId))?.state).toBe("saved");
|
||||
const restarted = coordinator();
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(restarted, () => true, async () => true).ready;
|
||||
expect(restarted.nextWakeAt()).toBeNull();
|
||||
} finally { await restarted.stop(); }
|
||||
} finally { release(); renameSpy.mockRestore(); await stopped; await worker.stop(); }
|
||||
});
|
||||
it("honors a newer retry deadline written before its attempt claim", async () => {
|
||||
const copy = await run(); await captured(copy);
|
||||
clock(); const worker = coordinator();
|
||||
const later = new Date(Date.now() + 60_000);
|
||||
const transaction = db.transaction.bind(db);
|
||||
let changed = false;
|
||||
const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => {
|
||||
if (!changed) {
|
||||
changed = true;
|
||||
await transaction(async tx => {
|
||||
await notifyInstructionCommit(tx, { companyId, runId: copy.runId });
|
||||
await tx.update(agentInstructionWorkingCopies).set({ nextAttemptAt: later })
|
||||
.where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
});
|
||||
}
|
||||
return transaction(callback, options);
|
||||
}) as typeof db.transaction);
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
expect(changed).toBe(true);
|
||||
expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_commit", attempts: 1, nextAttemptAt: later });
|
||||
transactions.mockRestore();
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBe(later.getTime());
|
||||
await vi.advanceTimersByTimeAsync(60_000); await saved(copy.runId);
|
||||
} finally { transactions.mockRestore(); await worker.stop(); }
|
||||
});
|
||||
it("reactivates after a real worktree setting change without polling disabled settings", async () => {
|
||||
const copy = await run(); await captured(copy);
|
||||
const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "sleep40-instruction-activation" };
|
||||
const settings = instanceSettingsService(db, { runtimeEnv });
|
||||
await settings.updateExperimental({ enableWorktreeRunExecution: false });
|
||||
const service = heartbeatService(db, { runtimeEnv });
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await service.registerCapturedInstructionCommitWork(worker, () => true).ready;
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
const reads = vi.spyOn(db, "select");
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await service.resolveSchedulingSuppression();
|
||||
await settings.updateExperimental({ enableWorktreeRunExecution: true });
|
||||
await saved(copy.runId);
|
||||
} finally { await worker.stop(); await settings.updateExperimental({ enableWorktreeRunExecution: false }); }
|
||||
});
|
||||
it("reconstructs more than one page and drains the saved candidates in bounded batches", async () => {
|
||||
const copy = await run();
|
||||
const ids = Array.from({ length: 103 }, () => randomUUID());
|
||||
await db.insert(heartbeatRuns).values(ids.map(id => ({ id, companyId, agentId,
|
||||
invocationSource: "on_demand", responsibleUserId: userId })));
|
||||
await db.insert(agentInstructionWorkingCopies).values(ids.map(runId => ({ ...copy, runId,
|
||||
state: "pending_commit", attempts: 1, nextAttemptAt: new Date(), processStoppedAt: new Date(),
|
||||
candidateBase64: Buffer.from("recovered bytes").toString("base64"),
|
||||
})));
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
const rows = () => db.select({ state: agentInstructionWorkingCopies.state }).from(agentInstructionWorkingCopies)
|
||||
.where(eq(agentInstructionWorkingCopies.companyId, companyId));
|
||||
expect((await rows()).filter(row => row.state === "saved").length).toBeLessThanOrEqual(20);
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect((await rows()).filter(row => row.state === "saved")).toHaveLength(103);
|
||||
}, { timeout: 15_000 });
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("recovered bytes");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
} finally { await worker.stop(); }
|
||||
}, 20_000);
|
||||
it("requires outer commit and ignores a nested rollback", async () => {
|
||||
const copy = await run(); clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
await expect(db.transaction(async tx => {
|
||||
await captured(copy, new Date(), tx as unknown as typeof db);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
throw new Error("outer rollback");
|
||||
})).rejects.toThrow("outer rollback");
|
||||
expect((await copies.get(companyId, copy.runId))?.state).toBe("prepared");
|
||||
await db.transaction(async tx => {
|
||||
await captured(copy, new Date(), tx as unknown as typeof db);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
});
|
||||
await saved(copy.runId);
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it("preserves a permission conflict instead of retrying it forever", async () => {
|
||||
const copy = await run(); await captured(copy);
|
||||
await db.delete(principalPermissionGrants).where(eq(principalPermissionGrants.companyId, companyId));
|
||||
clock(); const worker = coordinator();
|
||||
try {
|
||||
await copies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect((await copies.get(companyId, copy.runId))?.state).toBe("conflict");
|
||||
});
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
expect((await copies.get(companyId, copy.runId))?.candidateBase64).toBe(Buffer.from("recovered bytes").toString("base64"));
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBeNull();
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
it.each((["shared", "separate", "dedicated"] as const).flatMap(mode =>
|
||||
(["saved", "retry"] as const).map(outcome => ({ mode, outcome }))))("joins foreground ownership without consuming a recovery attempt ($mode, $outcome)", async ({ mode, outcome }) => {
|
||||
const copy = await run(); await fs.writeFile(path.join(copy.localRoot, entryFile), "recovered bytes");
|
||||
const exercise = async (workerDb: typeof db) => {
|
||||
const recoveryCopies = mode === "shared" ? copies : agentInstructionWorkingCopyService(workerDb);
|
||||
clock(); const worker = coordinator();
|
||||
let release!: () => void, entered!: () => void;
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
const reached = new Promise<void>(resolve => { entered = resolve; });
|
||||
let foreground: ReturnType<typeof copies.collectStopped> | undefined;
|
||||
const rename = fs.rename.bind(fs);
|
||||
let holding = true;
|
||||
const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => {
|
||||
if (holding && to === path.join(root, entryFile)) {
|
||||
holding = false; entered(); await held;
|
||||
if (outcome === "retry") throw new Error("temporary foreground replacement failure");
|
||||
}
|
||||
return rename(from, to);
|
||||
});
|
||||
try {
|
||||
await recoveryCopies.registerCapturedCommitWork(worker, () => true, async () => true).ready;
|
||||
foreground = copies.collectStopped({ companyId, runId: copy.runId });
|
||||
await reached;
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(idleWorkSnapshot().active).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_commit", attempts: 1 });
|
||||
release();
|
||||
const result = await foreground;
|
||||
if (outcome === "retry") {
|
||||
expect(result).toMatchObject({ state: "pending_commit", attempts: 1, nextAttemptAt: expect.any(Date) });
|
||||
expect(result!.nextAttemptAt!.getTime()).toBeGreaterThan(Date.now() + 29_000);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await expect.poll(() => idleWorkSnapshot().active).toBe(0);
|
||||
expect(worker.nextWakeAt()).toBe(result!.nextAttemptAt!.getTime());
|
||||
await vi.advanceTimersByTimeAsync(Math.max(0, result!.nextAttemptAt!.getTime() - Date.now() - 5000));
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
expect((await copies.get(companyId, copy.runId))?.attempts).toBe(1);
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
} else expect(result).toMatchObject({ state: "saved", attempts: 1 });
|
||||
await saved(copy.runId);
|
||||
const events = await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), eq(activityLog.action, "agent.files_updated")));
|
||||
expect(events).toHaveLength(1);
|
||||
} finally { release(); await foreground; renameSpy.mockRestore(); await worker.stop(); }
|
||||
};
|
||||
if (mode === "dedicated") await withDedicatedDbConnection(db, exercise);
|
||||
else await exercise(db);
|
||||
});
|
||||
});
|
||||
|
||||
it("persists an atomic editor replacement, ignores repository instructions, and gives the next run exact bytes", async () => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(home, "AGENTS.md"), "Repository text must not become agent instructions");
|
||||
|
||||
@@ -115,6 +115,7 @@ const {
|
||||
reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })),
|
||||
reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
|
||||
registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })),
|
||||
resumeQueuedRuns: vi.fn(async () => undefined),
|
||||
recoverPendingSessionGoalActions: vi.fn(async () => ({ scanned: 0, enqueued: 0, alreadyQueued: 0, invalid: 0 })),
|
||||
@@ -550,6 +551,13 @@ describe("startServer feedback export wiring", () => {
|
||||
process.env.BETTER_AUTH_SECRET = "test-secret";
|
||||
});
|
||||
|
||||
it.each([false, true])("registers captured instruction recovery only when maintenance is enabled (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
|
||||
await startServer(); await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.registerCapturedInstructionCommitWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
|
||||
if (enabled) expect(heartbeatServiceMock.registerCapturedInstructionCommitWork.mock.calls[0][1]()).toBe(true);
|
||||
});
|
||||
|
||||
it.each([false, true])("registers retry cancellation recovery only when maintenance is enabled (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
|
||||
await startServer(); await settleStartupHousekeeping();
|
||||
@@ -941,7 +949,7 @@ describe("startServer feedback export wiring", () => {
|
||||
intervalCallback?.();
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true });
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true });
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(retiredDetector).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
|
||||
+4
-1
@@ -1820,6 +1820,9 @@ async function startServerWithDatabaseTeardown(
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork,
|
||||
() => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
|
||||
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerCapturedInstructionCommitWork(deliveryWork,
|
||||
() => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
|
||||
|
||||
const runRetentionSweep = async () => {
|
||||
const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
|
||||
let archived = 0;
|
||||
@@ -1947,7 +1950,7 @@ async function startServerWithDatabaseTeardown(
|
||||
// Periodically reap orphaned runs (5-min staleness threshold) and make sure
|
||||
// persisted queued work is still being driven forward.
|
||||
runPeriodicRecovery(() => heartbeat
|
||||
.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true })
|
||||
.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true })
|
||||
.then(() => heartbeat.promoteDueScheduledRetries())
|
||||
.then(async (promotion) => {
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
|
||||
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
|
||||
import { and, eq, or, sql } from "drizzle-orm";
|
||||
import { agents, agentApiKeys, agentRuntimeState, agentTaskSessions, agentWakeupRequests, activityLog,
|
||||
@@ -44,6 +45,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
|
||||
await tx.delete(issueExecutionDecisions).where(eq(issueExecutionDecisions.actorAgentId, id));
|
||||
await tx.delete(issueComments).where(eq(issueComments.authorAgentId, id));
|
||||
await notifyScheduledRetryCancellation(tx, { companyId: companyId });
|
||||
await notifyInstructionCommit(tx, { companyId: companyId });
|
||||
await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id));
|
||||
await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id));
|
||||
await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id));
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
|
||||
import { agentDirectoryWorkingCopyService, isAgentDirectoryCopy } from "./agent-directory-working-copies.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { and, asc, desc, eq, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm";
|
||||
import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db";
|
||||
import { and, asc, desc, eq, gt, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm";
|
||||
import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db";
|
||||
import {
|
||||
prepareAdapterExecutionTargetRuntime,
|
||||
runAdapterExecutionTargetShellCommand,
|
||||
@@ -32,6 +36,23 @@ const MAX_COLLECTION_ATTEMPTS = 3;
|
||||
// supply one. These handles are optional: durable captured bytes survive restart.
|
||||
const liveTargets = new Map<string, AdapterExecutionTarget>();
|
||||
const targetKey = (companyId: string, runId: string) => `${companyId}:${runId}`;
|
||||
const copySaves = new WeakMap<object, Map<string, Promise<void>>>();
|
||||
|
||||
function serializeCopySave<T>(db: Db, companyId: string, runId: string, save: () => Promise<T>): Promise<T> {
|
||||
const scope = databaseWorkScopeKey(db);
|
||||
let active = copySaves.get(scope);
|
||||
if (!active) copySaves.set(scope, active = new Map());
|
||||
const key = targetKey(companyId, runId);
|
||||
|
||||
|
||||
const operation = (active.get(key) ?? Promise.resolve()).then(save);
|
||||
const settled = operation.then(() => undefined, () => undefined);
|
||||
active.set(key, settled);
|
||||
return operation.finally(() => {
|
||||
if (active.get(key) === settled) active.delete(key);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
/** Finish only bounded immediate collection work; deferred ownership remains pending. */
|
||||
export async function collectStoppedInstructionCopyWithRetries<T extends Pick<Copy, "state" | "attempts" | "nextAttemptAt">>(collect: () => Promise<T | null>): Promise<T | null> {
|
||||
@@ -73,11 +94,22 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
const [row] = await db.select().from(copies).where(scope(companyId, runId));
|
||||
return row ?? null;
|
||||
}
|
||||
async function tryPatch(row: Copy, values: Partial<typeof copies.$inferInsert>, requireDue = false) {
|
||||
const signalsCommit = row.state === "pending_commit" || values.state === "pending_commit";
|
||||
const write = async (tx: Db) => {
|
||||
if (signalsCommit)
|
||||
await notifyInstructionCommit(tx, { companyId: row.companyId, runId: row.runId });
|
||||
const [updated] = await tx.update(copies).set({ ...values, updatedAt: new Date() }).where(and(
|
||||
scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash),
|
||||
sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`,
|
||||
requireDue ? or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())) : undefined,
|
||||
)).returning();
|
||||
return updated;
|
||||
};
|
||||
return signalsCommit ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
|
||||
}
|
||||
async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
|
||||
const [updated] = await db.update(copies).set({ ...values, updatedAt: new Date() }).where(and(
|
||||
scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash),
|
||||
sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`,
|
||||
)).returning();
|
||||
const updated = await tryPatch(row, values);
|
||||
// A late cleanup must not overwrite an explicit resolution or a newer
|
||||
// baseline. Canonical content has its own independent head CAS.
|
||||
return updated ?? (await get(row.companyId, row.runId))!;
|
||||
@@ -277,7 +309,10 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
}
|
||||
|
||||
/** Caller must have joined the owned provider process before granting this proof. */
|
||||
async function collectStopped(input: { companyId: string; runId: string; target?: AdapterExecutionTarget | null }) {
|
||||
function collectStopped(input: { companyId: string; runId: string; target?: AdapterExecutionTarget | null }): Promise<Copy | null> {
|
||||
return serializeCopySave(db, input.companyId, input.runId, () => collectStoppedOnce(input));
|
||||
}
|
||||
async function collectStoppedOnce(input: Parameters<typeof collectStopped>[0]) {
|
||||
let row = await get(input.companyId, input.runId);
|
||||
if (!row) return row;
|
||||
if (isAgentDirectoryCopy(row)) return directories.collectStopped(row, input.target);
|
||||
@@ -300,16 +335,13 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
}
|
||||
|
||||
/** Restarts can retry persisted bytes without touching a live filesystem or launching a model. */
|
||||
async function recoverCaptured() {
|
||||
const pending = await db.select().from(copies).where(and(
|
||||
async function recoverCaptured(options: { skipCommit?: boolean } = {}) {
|
||||
const pending = options.skipCommit ? [] : await db.select().from(copies).where(and(
|
||||
eq(copies.state, "pending_commit"),
|
||||
or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())),
|
||||
lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1),
|
||||
)).limit(20);
|
||||
for (const row of pending) {
|
||||
const result = await commitCandidate(await patch(row, { attempts: row.attempts + 1 }));
|
||||
if (isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
|
||||
}
|
||||
for (const row of pending) await recoverCommit(commitEntry(row)!, () => true);
|
||||
// A crash between a durable save and release must not retain run snapshots
|
||||
// forever. Only discard copies with recorded stop proof and a terminal receipt.
|
||||
const cleanup = await db.select().from(copies).where(and(
|
||||
@@ -374,6 +406,56 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
return pending.length;
|
||||
}
|
||||
|
||||
type CommitEntry = { id: string; companyId: string; dueAt: number };
|
||||
function commitEntry(row: Pick<Copy, "runId" | "companyId" | "state" | "attempts" | "nextAttemptAt"> | null): CommitEntry | null {
|
||||
return row?.state === "pending_commit" && row.attempts < MAX_COLLECTION_ATTEMPTS
|
||||
? { id: row.runId, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 } : null;
|
||||
}
|
||||
async function loadCommits(scope: InstructionCommitScope | undefined, admitted: () => boolean) {
|
||||
const entries: CommitEntry[] = [];
|
||||
let cursor: string | undefined;
|
||||
while (admitted()) {
|
||||
const rows = await db.select({ runId: copies.runId, companyId: copies.companyId, state: copies.state,
|
||||
attempts: copies.attempts, nextAttemptAt: copies.nextAttemptAt }).from(copies).where(and(
|
||||
eq(copies.state, "pending_commit"), lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1),
|
||||
scope ? eq(copies.companyId, scope.companyId) : undefined,
|
||||
scope?.runId ? eq(copies.runId, scope.runId) : undefined,
|
||||
cursor ? gt(copies.runId, cursor) : undefined,
|
||||
)).orderBy(asc(copies.runId)).limit(100);
|
||||
for (const row of rows) entries.push(commitEntry(row)!);
|
||||
if (rows.length < 100) return entries;
|
||||
cursor = rows[rows.length - 1].runId;
|
||||
}
|
||||
throw new Error("Instruction commit reconstruction stopped");
|
||||
}
|
||||
function recoverCommit(entry: CommitEntry, admitted: () => boolean): Promise<CommitEntry | null> {
|
||||
return serializeCopySave(db, entry.companyId, entry.id, () => recoverCommitOnce(entry, admitted));
|
||||
}
|
||||
async function recoverCommitOnce(entry: CommitEntry, admitted: () => boolean) {
|
||||
const row = await get(entry.companyId, entry.id);
|
||||
const current = commitEntry(row);
|
||||
if (!row || !current || current.dueAt > Date.now()) return current;
|
||||
if (!admitted()) throw new Error("Instruction commit recovery stopped");
|
||||
const claimed = await tryPatch(row, { attempts: row.attempts + 1 }, true);
|
||||
if (claimed) {
|
||||
const result = await commitCandidate(claimed);
|
||||
if (isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
|
||||
}
|
||||
return commitEntry(await get(entry.companyId, entry.id));
|
||||
}
|
||||
function registerCapturedCommitWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
canRun: () => boolean, canRecover: () => Promise<boolean>) {
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.instructionCommit,
|
||||
batchSize: 20,
|
||||
decodeScope: (hint: string): InstructionCommitScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (entry: CommitEntry, scope?: InstructionCommitScope) => !scope ||
|
||||
entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId),
|
||||
load: async (scope, signal) => await canRecover() ? loadCommits(scope, () => canRun() && !signal?.aborted) : [],
|
||||
check: async (entry, signal) => await canRecover() ? recoverCommit(entry, () => canRun() && !signal?.aborted) : null,
|
||||
}, canRun);
|
||||
}
|
||||
|
||||
async function list(companyId: string, agentId: string, actor: AuthorizationActor) {
|
||||
await revisions.readCurrent({ companyId, agentId }, actor);
|
||||
const rows = await db.select().from(copies).where(and(eq(copies.companyId, companyId), eq(copies.agentId, agentId), inArray(copies.state, ["conflict", "pending_collection", "pending_commit", "unavailable"]))).orderBy(desc(copies.createdAt)).limit(50);
|
||||
@@ -422,5 +504,5 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
|
||||
}
|
||||
return row;
|
||||
}
|
||||
return { prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
|
||||
return { registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
|
||||
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
|
||||
import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js";
|
||||
import { chatCompanyDeletion } from "./chat-channels/company-deletion.js";
|
||||
@@ -98,6 +99,7 @@ export async function deleteCompany(db: Db, id: string) {
|
||||
await tx.delete(activityLog).where(eq(activityLog.companyId, id));
|
||||
await tx.delete(runIdentityContexts).where(eq(runIdentityContexts.companyId, id));
|
||||
await notifyScheduledRetryCancellation(tx, { companyId: id });
|
||||
await notifyInstructionCommit(tx, { companyId: id });
|
||||
await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, id));
|
||||
await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, id));
|
||||
await tx.delete(agentApiKeys).where(eq(agentApiKeys.companyId, id));
|
||||
|
||||
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
instructionCommit: "instruction-commit",
|
||||
scheduledRetryCancellation: "scheduled-retry-cancellation",
|
||||
executionFinalization: "execution-finalization",
|
||||
workspaceRecovery: "workspace-recovery",
|
||||
|
||||
@@ -4084,6 +4084,14 @@ export function heartbeatService(
|
||||
triggerIssueMonitor,
|
||||
|
||||
reportRunActivity: clearDetachedRunWarning,
|
||||
registerCapturedInstructionCommitWork: (
|
||||
coordinator: Parameters<typeof instructionCopies.registerCapturedCommitWork>[0], canRun: () => boolean,
|
||||
) => instructionCopies.registerCapturedCommitWork(coordinator,
|
||||
() => canRun() && readTaskDrain(new Date()) === null,
|
||||
async () => {
|
||||
const state = await getSchedulingSuppression(true);
|
||||
return !state.suppressed || state.reason === "task_drain";
|
||||
}),
|
||||
registerScheduledRetryCancellationWork: (
|
||||
coordinator: Parameters<typeof registerScheduledRetryCancellationWork>[0], canRun: () => boolean,
|
||||
) => registerScheduledRetryCancellationWork(coordinator,
|
||||
|
||||
@@ -1947,7 +1947,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
|
||||
let settledProcessesDiscovered = false;
|
||||
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean }) {
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean }) {
|
||||
await preserveCancelledNativeWorkspaces(db);
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
@@ -1957,7 +1957,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
await instructionCopies.recoverStopped().catch(error => {
|
||||
logger.warn({ err: error }, "failed to recover stopped instruction copies");
|
||||
});
|
||||
await instructionCopies.recoverCaptured().catch(error => {
|
||||
await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit }).catch(error => {
|
||||
logger.warn({ err: error }, "failed to retry captured instruction revisions");
|
||||
});
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
|
||||
import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
|
||||
import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
|
||||
import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
|
||||
@@ -617,6 +618,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
|
||||
if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
|
||||
if (patch.enableWorktreeRunExecution !== undefined) await notifyInstructionCommit(tx);
|
||||
if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx);
|
||||
await notifyToolConnectionHealth(tx, "*");
|
||||
return tx
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type InstructionCommitScope = { companyId: string; runId?: string };
|
||||
|
||||
export async function notifyInstructionCommit(transaction: object, scope?: InstructionCommitScope) {
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.instructionCommit, scope ? JSON.stringify(scope) : "*");
|
||||
}
|
||||
Reference in new issue
Block a user