fix(server): schedule archived runner evidence from saved deadlines (#17925)

<!-- entire-trail-link-start -->
https://entire.com/gh/paperclipai/-/trails/85/changes/paperclip/85
<!-- entire-trail-link-end -->

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Retired native sessions leave continuity and quarantine evidence on
disk.
> - The old hourly loop mixes evidence retention with backup expiry,
which has different database eligibility rules.
> - Evidence already has saved creation timestamps and local publication
signals.
> - This pull request gives evidence its own registered worker and
actual next deadline.
> - The worker stops filesystem and SQL activity when its evidence
obligations settle.

## Linked Issues or Issue Description

Stack slice 65. Parent: #17923
(`codex/sleep-64-skill-version-retention`). Related foundation: #15625.

**What happened?**

At parent `7b94aa185c29673372646bbf2e07a2fbd16bc394`, an hourly
backup/evidence tick still inspects native state roots. Evidence's
private due time is checked by that recurring tick instead of the shared
scheduler. Empty evidence and future retention windows have no reason to
be rediscovered every hour.

**Expected behavior**

Reconstruct evidence once. Wake from the existing publication signal.
Register the earliest saved retention deadline. Keep busy, failed, or
incomplete evidence recoverable. Preserve backup eligibility as a
separate duty.

**Steps to reproduce**

Start retention with an empty evidence root or archives inside the
seven-day window. The old timer still visits the combined owner. The new
evidence worker disarms when empty, or registers the saved expiry with
no evidence filesystem or SQL calls between relevant changes and that
deadline.

## What Changed

- Add an evidence worker through the delivery coordinator and shared
scheduler. Track startup reconstruction without delaying HTTP readiness.
Unsubscribe local publication before delivery shutdown.
- Select `backups` for the existing backup timer and `evidence` for the
new worker. Existing callers that omit the selection retain combined
behavior. Backup expiration remains the next separate migration.
- Use hourly retries for concrete busy or failed evidence, with earlier
saved expiries and one-second continuations independent. Restore
candidate/removal/byte summaries. Schedule publication at the first
removable instant; preserve legacy manual-tick semantics.
- Expose the existing earliest evidence deadline. Preserve one-second
bounded continuation, concrete failure/busy retries, cleanup-source
archive exclusions, symlink protection, late activity checks, atomic
rename, and interrupted removal recovery.
- Preserve non-ENOENT inspection errors as recovery obligations. Retry a
failed session root in a new evidence pass instead of continuing past it
and forgetting it.
- Register the first removable instant after the inclusive retention
boundary. Keep the existing manual dry-run publication semantics.
- Add real filesystem/isolated PostgreSQL lifecycle tests and
disabled/held-startup wiring cases. Add no table or migration.

The archive and quarantine writers already call
`noteNativeRunnerEvidence`; this PR connects that existing signal to the
scheduler. Shared code owns scheduling, while evidence rules stay with
the retention service. See
[`doc/plans/2026-10-27-runner-evidence-retention-sleep-audit.md`](doc/plans/2026-10-27-runner-evidence-retention-sleep-audit.md).

## Verification

- Adopted actual prerequisite squash
`b06ea111a254c62e0a55199b010d82e0adb33bf0` from #17923 through normal
merges. Qualified runtime/package/UI/map trees remain unchanged. The
final nine-path diff includes one browser-spec readiness repair.
- Runtime qualification remains **652 passing TypeScript cases across 22
complete files**, plus **29 passing Rust cases** in native-backend,
runtime-notification and configured-environment targets. One actual
closed Pi snapshot launch is intentionally skipped because its external
distribution root is not provided. Accepted source is
`b80eee32f344ae0218d8b791c14f2ea174da09cc`; all runtime and earlier
sleep-test bytes are preserved.
- The complete outage-recovery browser file passes **all five journeys**
on a fresh isolated database after four additive rendered-history
readiness assertions. Every original assertion, outage control and
timeout remains byte-identical when those new lines are removed. The
intentionally unresolved first-session scenario keeps its pre-auth
interaction and delayed-owner coverage. The prior CI/local timeout
evidence and one rejected broader precondition remain in the audit.
- Adapter/Runner/database builds, server/Runner typechecking, Rust
formatting/workspace checks, server build, module boundaries, migration
order and generated TypeScript/Rust profiles pass. PostgreSQL 17
fixtures apply all 413 journal entries through index 414; source hashes
and temporary fixture imports restore exactly.
- **Inherited validation exception:** Copilot profile 81's separate
source-identity check fails on the merged Rust configured-environment
hash. Its inputs and attestations equal upstream and stay unchanged
here. Grok's identity check passes. No release attestation is
regenerated or new live-provider qualification claimed. Required
exact-head CI/security and Greptile 5/5 still must pass.
- [The evidence
audit](doc/plans/2026-10-27-runner-evidence-retention-sleep-audit.md)
retains historical failures, counts and limitations. No full local
repository-suite, live provider, external-alarm, measured suspension or
fleet-savings claim. Fresh final-head CI/security, exact-head Greptile
5/5 and actionable Ponytail findings remain separate merge gates.

## Risks

- Live publication is the existing server-owned executor in one process.
Independent filesystem writers need an explicit signal contract. Startup
reconstructs restart state.
- Busy sessions and concrete inspection/removal failures retain retries.
Unknown cleanup-source archives remain protected.
- The existing scan enumerates directory names before applying its
entry/removal budget. This PR preserves that behavior and does not claim
a bounded total directory-listing allocation.
- An in-process deadline cannot wake stopped compute. Cloud sleep-safety
and external-alarm coverage stay unchanged. Physical evidence
reclamation can wait for a supported host wake.
- Backup expiration still has an hourly database eligibility scan. This
slice removes only the archived-evidence duty from that loop and does
not claim full instance suspension.

## Model Used

OpenAI Codex (GPT-6-based coding agent), using repository inspection,
normal Git merges, composition proofs and complete
PostgreSQL/filesystem/startup suites. No subagents were used in this
reconciliation.

## 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
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
authored and GitHub committed 2026-10-29 11:12:16 -05:00
1 parent 81dcc4f6bf
commit 07c6df260b
9 files changed
+596 -26

No files matched your search

@@ -1,3 +1,8 @@
import { applyPendingMigrations } from "@paperclipai/db";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { registerNativeRunnerEvidenceRetentionWork } from "../services/native-runtime/native-runner-evidence-retention-work.js";
import { noteNativeRunnerEvidence } from "../services/native-runtime/native-runner-evidence-signal.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { createHash, randomUUID } from "node:crypto";
import os from "node:os";
import path from "node:path";
@@ -20,9 +25,10 @@ import { nativeHarnessBackupExpirationService } from "../services/native-runtime
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { logger } from "../middleware/logger.js";
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RETENTION_TEST_DATABASE_URL");
const describeEmbeddedPostgres = external || embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
if (!external && !embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres native harness backup expiration tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
@@ -41,11 +47,17 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
const now = new Date();
beforeAll(async () => {
tempDb = await startRecoveryPostgres("paperclip-native-backup-expiration-");
db = createDb(tempDb.connectionString);
if (!external) tempDb = await startRecoveryPostgres("paperclip-native-backup-expiration-");
if (external) await applyPendingMigrations(external);
db = createDb(external ?? tempDb!.connectionString);
}, 20_000);
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let worker: ReturnType<typeof registerNativeRunnerEvidenceRetentionWork> | undefined;
let admitted = true;
afterEach(async () => {
worker?.unsubscribe(); await coordinator?.stop(); coordinator = undefined; worker = undefined; admitted = true;
vi.useRealTimers(); vi.restoreAllMocks();
await db.delete(agentTaskSessions);
await db.delete(nativeRunFinalizations);
await db.delete(environmentLeases);
@@ -58,6 +70,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
@@ -522,6 +535,146 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
return createHash("sha256").update(randomUUID()).digest("hex");
}
async function startEvidence(options: Parameters<typeof nativeHarnessBackupExpirationService>[1] = {}) {
if (!vi.isFakeTimers()) {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
vi.setSystemTime(now);
}
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => true,
onError: error => { throw error; } });
worker = registerNativeRunnerEvidenceRetentionWork(coordinator, db, { stateBase, isSessionBusy: () => false, ...options }, () => admitted);
await worker.ready;
}
async function settled(deadline: number | null = null) {
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000);
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBe(deadline);
}, { timeout: 5000 });
}
async function quiet(ms = 3 * 60 * 60_000) {
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), directories = vi.spyOn(fs, "readdir");
await vi.advanceTimersByTimeAsync(ms);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(directories).not.toHaveBeenCalled();
reads.mockRestore(); writes.mockRestore(); directories.mockRestore();
}
it("leaves empty evidence and retained backups quiet without invoking backup SQL", async () => {
const session = await seedSession();
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
await startEvidence(); await settled(); expect(await exists(session.current)).toBe(true);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
reads.mockRestore(); writes.mockRestore(); await quiet();
});
it("reconstructs and replaces the saved evidence deadline after a publication signal", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
const first = await seedArchive(session.root, { stampedDaysAgo: 2 });
const firstDeadline = now.getTime() + 5 * DAY_MS + 1;
await startEvidence(); await settled(firstDeadline); await quiet();
const writtenAt = Date.now();
const next = await seedArchive(session.root, { name: `${writtenAt}-${randomUUID()}` });
noteNativeRunnerEvidence(writtenAt);
await settled(firstDeadline);
await vi.advanceTimersByTimeAsync(firstDeadline - Date.now());
await settled(writtenAt + 7 * DAY_MS + 1); expect(await exists(first)).toBe(false);
await vi.advanceTimersByTimeAsync(writtenAt + 7 * DAY_MS + 1 - Date.now()); await settled();
expect(await exists(next)).toBe(false); expect(await exists(session.current)).toBe(true); await quiet();
});
it.each(["session root", "archive directory"])("preserves an evidence inspection failure at %s until its concrete retry succeeds", async location => {
const session = await seedSession({ issueStatus: "in_progress" });
const archive = await seedArchive(session.root);
const original = fs.lstat;
const denied = vi.spyOn(fs, "lstat").mockImplementation(async (...args) => {
if (String(args[0]) === (location === "session root" ? session.root : path.join(session.root, "continuity-breaks"))) throw Object.assign(new Error("unavailable"), { code: "EACCES" });
return original(...args);
});
await startEvidence(); expect(coordinator!.nextWakeAt()).not.toBeNull();
const retryAt = coordinator!.nextWakeAt()!; expect(await exists(archive)).toBe(true);
denied.mockRestore(); await quiet(5000);
await vi.advanceTimersByTimeAsync(retryAt - Date.now()); await settled();
expect(await exists(archive)).toBe(false); await quiet();
});
it("retries a busy evidence obligation and disarms after the session releases it", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
const archive = await seedArchive(session.root); let busy = true;
await startEvidence({ isSessionBusy: () => busy });
const retryAt = coordinator!.nextWakeAt()!; expect(retryAt - Date.now()).toBeGreaterThanOrEqual(60 * 60_000);
await quiet(5000); busy = false;
await vi.advanceTimersByTimeAsync(retryAt - Date.now()); await settled();
expect(await exists(archive)).toBe(false); await quiet();
});
it("retains the first publication during drain and cleans up on resume", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
await startEvidence(); await settled(); admitted = false;
const archive = await seedArchive(session.root);
noteNativeRunnerEvidence(now.getTime() - 10 * DAY_MS);
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); });
await quiet(60_000); expect(await exists(archive)).toBe(true);
admitted = true; await settled(); expect(await exists(archive)).toBe(false); await quiet();
});
it("reconstructs the same retained deadline after restart", async () => {
const scope = await emptyStateBase(); await seedQuarantine(scope, { stampedDaysAgo: 6 });
const deadline = now.getTime() + DAY_MS + 1;
await startEvidence(); await settled(deadline);
worker!.unsubscribe(); await coordinator!.stop();
await startEvidence(); await settled(deadline); await quiet(60_000);
expect(coordinator!.nextWakeAt()).toBe(deadline);
});
it("backs off a failed state-base inspection and lets publication reactivate it", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
const archive = await seedArchive(session.root), read = fs.readdir;
const denied = vi.spyOn(fs, "readdir").mockImplementation(async (...args) => {
if (args[0] === stateBase) throw Object.assign(new Error("state base unavailable"), { code: "EACCES" });
return read(...args);
});
await startEvidence(); expect(coordinator!.nextWakeAt()! - Date.now()).toBeGreaterThanOrEqual(60 * 60_000);
denied.mockRestore(); await quiet(5 * 60_000); expect(await exists(archive)).toBe(true);
noteNativeRunnerEvidence(now.getTime() - 10 * DAY_MS);
await settled(); expect(await exists(archive)).toBe(false); await quiet();
});
it("lets healthy saved expiry preempt a long-busy evidence retry", async () => {
const busy = await seedSession({ issueStatus: "in_progress" }), healthy = await seedSession({ issueStatus: "in_progress" });
const held = await seedArchive(busy.root);
const dueAt = now.getTime() + 10_001;
const available = await seedArchive(healthy.root, { name: `${dueAt - 7 * DAY_MS - 1}-${randomUUID()}` });
await startEvidence({ isSessionBusy: scope => scope === busy.scope }); await settled(dueAt);
await vi.advanceTimersByTimeAsync(dueAt - Date.now());
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(100); expect(await exists(available)).toBe(false); });
expect(await exists(held)).toBe(true); expect(coordinator!.nextWakeAt()! - Date.now()).toBeGreaterThan(59 * 60_000);
await quiet(60_000);
});
it("reports reclaimed evidence bytes and dry-run mode", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
const archive = await seedArchive(session.root); const reported = vi.spyOn(logger, "info");
await startEvidence(); await settled();
expect(reported.mock.calls).toEqual(expect.arrayContaining([[expect.objectContaining({ evidenceCandidates: 1,
evidenceRemoved: 1, evidenceReclaimedBytes: expect.any(Number), dryRun: false }), "native runner evidence retention"]]));
expect(await exists(archive)).toBe(false); await quiet();
});
it("schedules publication at the first removable instant after its retention boundary", async () => {
const scope = await emptyStateBase(); await startEvidence(); await settled(); await quiet();
const writtenAt = Date.now();
await seedQuarantine(scope, { name: `${scope}.identity_indeterminate.${writtenAt}.4242.${randomUUID()}` });
noteNativeRunnerEvidence(writtenAt); await settled(writtenAt + 7 * DAY_MS + 1);
});
it("uses bounded continuations without recounting an evidence session forever", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
const archives = [await seedArchive(session.root), await seedArchive(session.root)];
const quarantine = await seedQuarantine(session.scope);
await startEvidence({ maxEntries: 1 }); await settled();
for (const archive of [...archives, quarantine]) expect(await exists(archive)).toBe(false);
await quiet();
});
it("removes continuity archives older than the retention window and keeps recent ones", async () => {
const session = await seedSession({ issueStatus: "in_progress" });
const old = await seedArchive(session.root, { stampedDaysAgo: 9 });
@@ -1,3 +1,4 @@
import { nativeHarnessBackupExpirationService } from "../services/native-runtime/native-harness-backup-expiration.js";
import { skillVersionRetentionService } from "../services/skill-version-retention.js";
import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js";
import { joinRequestRetentionService } from "../services/join-request-retention.js";
@@ -539,6 +540,18 @@ vi.mock("../board-claim.js", () => ({
initializeBoardClaimChallenge: vi.fn(async () => undefined),
}));
vi.mock("../services/native-runtime/native-harness-backup-expiration.js", () => ({
NATIVE_HARNESS_BACKUP_EXPIRATION_INTERVAL_MS: 60 * 60_000,
NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS: 2 * 60_000,
NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS: 7,
NATIVE_RUNNER_EVIDENCE_RETENTION_DEFAULT_DAYS: 7,
nativeHarnessBackupExpirationService: vi.fn(() => ({
tick: vi.fn(async () => ({ checked: 0, candidates: 0, evidenceCandidates: 0, failed: 0, truncated: false })),
noteEvidence: vi.fn(), nextEvidenceAt: vi.fn(() => null),
tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn(),
})),
}));
vi.mock("../services/skill-version-retention.js", () => ({
skillVersionRetentionService: vi.fn(() => ({
listCompanies: vi.fn(async () => []), pruneCompany: vi.fn(async () => ({ checked: 0, truncated: false, nextRunAt: null })),
@@ -582,6 +595,31 @@ describe("startServer feedback export wiring", () => {
process.env.BETTER_AUTH_SECRET = "test-secret";
});
it.each([false, true])("registers runner evidence independently from the backup scan (%s)", async enabled => {
loadConfigMock.mockReturnValue(buildTestConfig({ nativeHarnessBackupExpirationEnabled: enabled }));
await startServer(); await settleStartupHousekeeping();
expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.nativeRunnerEvidenceRetention)).toHaveLength(enabled ? 1 : 0);
expect(vi.mocked(nativeHarnessBackupExpirationService).mock.calls.map(([, options]) => options?.duty)).toEqual(enabled ? ["backups", "evidence"] : ["backups"]);
const backups = vi.mocked(nativeHarnessBackupExpirationService).mock.results[0]!.value;
expect(backups.registerCheckpointExpiration).toHaveBeenCalledTimes(enabled ? 1 : 0);
});
it("tracks held evidence reconstruction after HTTP readiness", async () => {
loadConfigMock.mockReturnValue(buildTestConfig({ nativeHarnessBackupExpirationEnabled: true }));
const pending = Promise.withResolvers<{ truncated: boolean }>();
vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ tick: vi.fn(), noteEvidence: vi.fn(), nextEvidenceAt: () => null,
tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn() } as never);
vi.mocked(nativeHarnessBackupExpirationService).mockReturnValueOnce({ tick: vi.fn(() => pending.promise), noteEvidence: vi.fn(), nextEvidenceAt: () => null,
tickCheckpoints: vi.fn(), registerCheckpointExpiration: vi.fn() } as never);
try {
await startServer(); await settleStartupHousekeeping();
expect(getStartupRecoveryState().phase).toBe("ready"); expect(idleWorkSnapshot().active).toBe(1);
expect(vi.mocked(nativeHarnessBackupExpirationService).mock.results[0]!.value.registerCheckpointExpiration).toHaveBeenCalledOnce();
} finally { pending.resolve({ truncated: false }); await settleStartupHousekeeping(); }
expect(idleWorkSnapshot().active).toBe(0);
expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.nativeRunnerEvidenceRetention)).toEqual([]);
});
it.each([false, true])("registers skill cache retention only when enabled (%s)", async enabled => {
loadConfigMock.mockReturnValue(buildTestConfig({ skillVersionRetentionEnabled: enabled }));
await startServer(); await settleStartupHousekeeping();
+10 -3
View File
@@ -32,7 +32,7 @@ import {
NATIVE_HARNESS_BACKUP_EXPIRATION_STARTUP_DELAY_MS,
nativeHarnessBackupExpirationService,
} from "./services/native-runtime/native-harness-backup-expiration.js";
import { onNativeRunnerEvidence } from "./services/native-runtime/native-runner-evidence-signal.js";
import { registerNativeRunnerEvidenceRetentionWork } from "./services/native-runtime/native-runner-evidence-retention-work.js";
import { gitRestorePointRetentionService, registerGitRestorePointWork } from "./services/git-restore-point-ledger.js";
import {
nativeWorkspaceSyncRetentionService,
@@ -1259,12 +1259,12 @@ async function startServerWithDatabaseTeardown(
};
const skillSourceSync = skillSourceSyncService(db as any);
const nativeHarnessBackupExpiration = nativeHarnessBackupExpirationService(db as any, {
duty: "backups",
dryRun: config.nativeHarnessBackupExpirationDryRun,
closedForDays: config.nativeHarnessBackupExpirationDays,
evidenceRetentionDays: config.nativeRunnerEvidenceRetentionDays,
shouldContinue: () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby(),
});
const unsubscribeNativeRunnerEvidence = onNativeRunnerEvidence(nativeHarnessBackupExpiration.noteEvidence);
let nativeHarnessBackupExpirationInFlight = false;
let nativeHarnessBackupExpirationContinuation: ReturnType<typeof setTimeout> | undefined;
const sweepNativeHarnessBackups = () => {
@@ -1344,6 +1344,13 @@ async function startServerWithDatabaseTeardown(
server.on("close", () => { void taskDrainExpiryWork.stop(); });
const executionSchedule = executionControlSchedule(db);
const workspaceRecoveryRuntime = environmentRuntimeService(db, { pluginWorkerManager });
const nativeRunnerEvidenceRetentionWorker = config.nativeHarnessBackupExpirationEnabled
? registerNativeRunnerEvidenceRetentionWork(deliveryWork, db as any, {
dryRun: config.nativeHarnessBackupExpirationDryRun,
evidenceRetentionDays: config.nativeRunnerEvidenceRetentionDays,
}, () => !heartbeatSchedulerStopped && !isIdleTaskDrainActive() && !isWarmStandby())
: null;
if (nativeRunnerEvidenceRetentionWorker) trackHeartbeatSchedulerWork(nativeRunnerEvidenceRetentionWorker.ready);
const skillVersionRetentionWorker = config.skillVersionRetentionEnabled
? registerSkillVersionRetentionWork(deliveryWork, skillVersionRetentionService(db as any, {
dryRun: config.skillVersionRetentionDryRun,
@@ -2191,7 +2198,7 @@ async function startServerWithDatabaseTeardown(
clearInterval(nativeHarnessBackupExpirationInterval);
clearTimeout(nativeHarnessBackupExpirationStartup);
clearTimeout(nativeHarnessBackupExpirationContinuation);
unsubscribeNativeRunnerEvidence();
nativeRunnerEvidenceRetentionWorker?.unsubscribe();
await skillSourceSync.stop();
unsubscribeChatCompletions();
unsubscribeNativeResponsePresentation?.();
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
/** Topics for existing durable queues; register intent before their writes. */
export const DELIVERY_QUEUES = {
nativeRunnerEvidenceRetention: "native-runner-evidence-retention",
skillVersionRetention: "skill-version-retention",
databaseBackups: "database-backups",
importSpoolCleanup: "import-spool-cleanup",
@@ -40,6 +40,7 @@ const MAX_CONTROL_PLANE_STATE_BYTES = 4 * 1024 * 1024;
const CHECKPOINT_EXPIRATION_MAX_FAILED_PASSES = 3;
export type NativeHarnessBackupExpirationOptions = {
duty?: "backups" | "evidence";
closedForDays?: number;
evidenceRetentionDays?: number;
dryRun?: boolean;
@@ -77,6 +78,7 @@ type EvidenceContext = {
retainedBefore: number;
canInspect: () => boolean;
dueAt: (at: number) => void;
retry: () => void;
};
type SessionIdentity = {
@@ -105,8 +107,13 @@ function emptyResult(): NativeHarnessBackupExpirationResult {
};
}
function missingPath(error: NodeJS.ErrnoException) {
if (error.code === "ENOENT") return null;
throw error;
}
async function isRealDirectory(target: string): Promise<boolean> {
const stat = await fs.lstat(target).catch(() => null);
const stat = await fs.lstat(target).catch(missingPath);
return Boolean(stat?.isDirectory() && !stat.isSymbolicLink());
}
@@ -262,6 +269,8 @@ export function nativeHarnessBackupExpirationService(
db: Db,
defaults: NativeHarnessBackupExpirationOptions = {},
) {
const includeBackups = defaults.duty !== "evidence";
const includeEvidence = defaults.duty !== "backups";
let cursor: string | null = null;
let resume: { scope: string; archive: string } | null = null;
let quarantineCursor: string | null = null;
@@ -421,7 +430,10 @@ export function nativeHarnessBackupExpirationService(
}
async function removeExpiredLeftovers(parent: string, context: EvidenceContext, result: NativeHarnessBackupExpirationResult) {
for (const name of await fs.readdir(parent).catch(() => [] as string[])) {
for (const name of await fs.readdir(parent).catch((error: NodeJS.ErrnoException) => {
if (error.code === "ENOENT") return [] as string[];
throw error;
})) {
if (!name.startsWith(EXPIRED_PREFIX) || !UUID_PATTERN.test(name.slice(EXPIRED_PREFIX.length))) continue;
if (!context.canInspect()) {
context.dueAt(context.now);
@@ -444,17 +456,17 @@ export function nativeHarnessBackupExpirationService(
result.skipped[reason] = (result.skipped[reason] ?? 0) + 1;
};
const target = path.join(parent, name);
const stat = await fs.lstat(target).catch(() => null);
const stat = await fs.lstat(target).catch(missingPath);
if (path.dirname(target) !== parent || !stat?.isDirectory()) {
return skip("unsafe_path");
}
const createdAt = stampedAt > 0 && stampedAt <= context.now ? stampedAt : stat.mtimeMs;
if (createdAt >= context.retainedBefore) {
context.dueAt(createdAt + context.retentionMs);
context.dueAt(createdAt + context.retentionMs + 1);
return skip("recent_evidence");
}
if (context.isSessionBusy(scopeDigest)) {
context.dueAt(context.now);
context.retry();
return skip("session_active");
}
const bytes = await allocatedBytes(target);
@@ -462,7 +474,7 @@ export function nativeHarnessBackupExpirationService(
result.evidenceCandidateBytes += bytes;
if (context.dryRun) return;
if (context.isSessionBusy(scopeDigest)) {
context.dueAt(context.now);
context.retry();
return skip("session_active");
}
const trash = path.join(parent, `${EXPIRED_PREFIX}${randomUUID()}`);
@@ -480,7 +492,7 @@ export function nativeHarnessBackupExpirationService(
result: NativeHarnessBackupExpirationResult,
): Promise<string | null> {
const parent = path.join(root, "continuity-breaks");
const stat = await fs.lstat(parent).catch(() => null);
const stat = await fs.lstat(parent).catch(missingPath);
if (!stat) return null;
if (!stat.isDirectory()) {
result.skipped.unsafe_path = (result.skipped.unsafe_path ?? 0) + 1;
@@ -499,7 +511,7 @@ export function nativeHarnessBackupExpirationService(
await retireEvidence(parent, name, scopeDigest, Number(match[1]), context, result);
} catch (err) {
result.failed++;
context.dueAt(context.now);
context.retry();
logger.warn({ err, scope: scopeDigest }, "native runner evidence retention skipped a continuity archive");
}
}
@@ -513,7 +525,7 @@ export function nativeHarnessBackupExpirationService(
result: NativeHarnessBackupExpirationResult,
): Promise<boolean> {
const parent = path.join(base, "quarantine");
const stat = await fs.lstat(parent).catch(() => null);
const stat = await fs.lstat(parent).catch(missingPath);
if (!stat) return true;
if (!stat.isDirectory()) {
result.skipped.unsafe_path = (result.skipped.unsafe_path ?? 0) + 1;
@@ -537,7 +549,7 @@ export function nativeHarnessBackupExpirationService(
await retireEvidence(parent, name, match[1]!, Number(match[2]), context, result);
} catch (err) {
result.failed++;
context.dueAt(context.now);
context.retry();
logger.warn({ err, scope: match[1] }, "native runner evidence retention skipped a quarantine entry");
}
}
@@ -546,7 +558,7 @@ export function nativeHarnessBackupExpirationService(
}
function noteEvidence(writtenAt: number) {
evidenceDueAt = Math.min(evidenceDueAt, writtenAt + evidenceRetentionMs());
evidenceDueAt = Math.min(evidenceDueAt, writtenAt + evidenceRetentionMs() + (defaults.duty === "evidence" ? 1 : 0));
}
function evidenceRetentionMs(options: NativeHarnessBackupExpirationOptions = {}) {
@@ -572,9 +584,10 @@ export function nativeHarnessBackupExpirationService(
const closedBefore = new Date(now - closedForDays * DAY_MS);
const result = emptyResult();
const hasBudget = () => shouldContinue() && result.candidates + result.evidenceCandidates < maxRemovals;
if (!passActive && !includeBackups && evidenceDueAt > now) return result;
if (!passActive) {
passActive = true;
evidencePass = evidenceDueAt <= now;
evidencePass = includeEvidence && evidenceDueAt <= now;
if (evidencePass) {
evidenceDueAt = Number.POSITIVE_INFINITY;
passDueAt = Number.POSITIVE_INFINITY;
@@ -590,10 +603,16 @@ export function nativeHarnessBackupExpirationService(
dueAt: (at) => {
passDueAt = Math.min(passDueAt, at);
},
retry: () => {
passDueAt = Math.min(passDueAt, now + (defaults.duty === "evidence" ? 60 * 60_000 : 0));
},
};
if (phase === "sessions") {
const names = (await fs.readdir(base).catch(() => [] as string[]))
const names = (await fs.readdir(base).catch((error: NodeJS.ErrnoException) => {
if (error.code === "ENOENT") return [] as string[];
throw error;
}))
.filter((name) => SCOPE_PATTERN.test(name))
.sort();
const pending = cursor ? names.filter((name) => name > cursor!) : names;
@@ -608,11 +627,18 @@ export function nativeHarnessBackupExpirationService(
const previous = cursor;
cursor = name;
const root = path.join(base, name);
if (path.dirname(root) !== base || !(await isRealDirectory(root))) {
result.skipped.unsafe_path = (result.skipped.unsafe_path ?? 0) + 1;
try {
if (path.dirname(root) !== base || !(await isRealDirectory(root))) {
result.skipped.unsafe_path = (result.skipped.unsafe_path ?? 0) + 1;
continue;
}
} catch (err) {
result.failed++;
context.retry();
logger.warn({ err, scope: name }, "native runner retention could not inspect a session root");
continue;
}
if (resumeAfter === null) {
if (resumeAfter === null && includeBackups) {
try {
await expireSession(root, name, { dryRun, isSessionBusy, closedBefore }, result);
} catch (err) {
@@ -626,7 +652,7 @@ export function nativeHarnessBackupExpirationService(
stoppedAt = await retireContinuityBreaks(root, name, resumeAfter, context, result);
} catch (err) {
result.failed++;
context.dueAt(now);
context.retry();
logger.warn({ err, scope: name }, "native runner evidence retention skipped a session");
}
if (stoppedAt !== null) {
@@ -651,7 +677,7 @@ export function nativeHarnessBackupExpirationService(
}
} catch (err) {
result.failed++;
context.dueAt(now);
context.retry();
quarantineCursor = null;
logger.warn({ err }, "native runner evidence retention skipped quarantine");
}
@@ -662,6 +688,11 @@ export function nativeHarnessBackupExpirationService(
return result;
}
function nextEvidenceAt() {
if (passActive) return Date.now();
return Number.isFinite(evidenceDueAt) ? evidenceDueAt : null;
}
async function tickCheckpoints(options: NativeHarnessBackupExpirationOptions = {}) {
const closedForDays = Math.max(1, options.closedForDays ?? defaults.closedForDays ?? NATIVE_HARNESS_BACKUP_EXPIRATION_DEFAULT_DAYS);
const result = emptyResult();
@@ -714,5 +745,5 @@ export function nativeHarnessBackupExpirationService(
});
}
return { tick, noteEvidence, tickCheckpoints, registerCheckpointExpiration };
return { tick, noteEvidence, nextEvidenceAt, tickCheckpoints, registerCheckpointExpiration };
}
@@ -0,0 +1,34 @@
import { logger } from "../../middleware/logger.js";
import { registerDeadlineWork } from "../deadline-work.js";
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
import { nativeHarnessBackupExpirationService, type NativeHarnessBackupExpirationOptions } from "./native-harness-backup-expiration.js";
import { onNativeRunnerEvidence } from "./native-runner-evidence-signal.js";
export function registerNativeRunnerEvidenceRetentionWork(
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
db: Parameters<typeof nativeHarnessBackupExpirationService>[0],
options: NativeHarnessBackupExpirationOptions,
canRun: () => boolean,
) {
const service = nativeHarnessBackupExpirationService(db, { ...options, duty: "evidence" });
let continuation: number | null = null;
const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.nativeRunnerEvidenceRetention, {
retryMs: 30_000, canRun,
run: async signal => {
try {
const result = await service.tick({ shouldContinue: () => !signal.aborted && canRun() });
if (result.evidenceCandidates > 0 || result.failed > 0) {
logger.info({ ...result, dryRun: options.dryRun ?? false }, "native runner evidence retention");
}
continuation = result.truncated ? Date.now() + 1000 : null;
} catch (err) {
continuation = Date.now() + 60 * 60_000;
logger.warn({ err }, "native runner evidence inspection failed");
}
},
next: async () => continuation ?? service.nextEvidenceAt(),
});
const unsubscribe = onNativeRunnerEvidence(at => { service.noteEvidence(at); worker.wake(); });
return { ...worker, unsubscribe };
}