diff --git a/doc/run-evidence-storage.md b/doc/run-evidence-storage.md index b74e10e852a..bbb91b70e1c 100644 --- a/doc/run-evidence-storage.md +++ b/doc/run-evidence-storage.md @@ -254,6 +254,25 @@ switching the run's reference. It supports terminal `interrupted` runs as well. It does not treat an unproven mirror as an original file or ignore a recorded size/digest mismatch. +For online preservation of settled native files, use +`server/scripts/migrate-run-evidence.ts --preserve-settled-native-files` with +the existing company, cursor, limit and apply options. This separate mode does +not require global idle. It requires a native run with a finished status and +an explicitly released finalizer in `committed` or `terminal_failure`, no pending +attempt, no physical finalization owner, no retained allocation for that run, +and no task execution or checkout bound to it. A stale controller deadline or +historical PID does not establish this authority. Unknown recovery states stay +blocked; a terminal failure's existing blocked warning remains unchanged. + +The online path locks the native finalization scope and rechecks the source +row, owner generation and reference before publication. The source must use its +canonical company/agent/run filename, with no aliases, symlinks or hard links. +It streams and verifies the exact local bytes, then publishes the read-only +archive and an activity receipt in one guarded transaction. Originals remain +on disk. A changed source or owner leaves the original reference unchanged. +This mode keeps raw history separate from canonical native-event coverage, +including when the original source has an integrity warning. + An archive preserves historical evidence; it does not certify a complete native transcript. The run's status, canonical events, and original completion metadata remain unchanged. Archive readers do not append missing native events, and diff --git a/server/scripts/migrate-run-evidence.ts b/server/scripts/migrate-run-evidence.ts index 71454359053..cba42ca3b13 100644 --- a/server/scripts/migrate-run-evidence.ts +++ b/server/scripts/migrate-run-evidence.ts @@ -1,14 +1,18 @@ import { createDb } from "@paperclipai/db"; import { migrateRunEvidence } from "../src/services/run-evidence-migration.js"; +import { isCloudManagedInstance } from "../src/services/cloud-instance.js"; +import { cloudRuntimeConfigService } from "../src/services/cloud-runtime-config.js"; +import { setCloudRuntimeConfigSnapshot } from "../src/services/cloud-runtime-config-snapshot.js"; if (!process.env.DATABASE_URL) throw new Error("Set DATABASE_URL to the intended instance; run with legacy writers drained"); const args = process.argv.slice(2); -const flags = new Set(["--apply", "--company", "--after-run", "--after-trace", "--limit"]); +const booleans = new Set(["--apply", "--preserve-settled-native-files"]); +const flags = new Set([...booleans, "--company", "--after-run", "--after-trace", "--limit"]); for (let index = 0; index < args.length; index++) { const flag = args[index]!; if (!flags.has(flag)) throw new Error(`Unknown migration option: ${flag}`); - if (flag !== "--apply" && (!args[++index] || args[index]!.startsWith("--"))) { + if (!booleans.has(flag) && (!args[++index] || args[index]!.startsWith("--"))) { throw new Error(`Missing value for ${flag}`); } } @@ -18,12 +22,18 @@ const value = (flag: string) => { }; const db = createDb(process.env.DATABASE_URL); try { + if (isCloudManagedInstance()) { + const stored = await cloudRuntimeConfigService(db).inspectStored(); + if (!stored) throw new Error("cloud_runtime_configuration_missing"); + setCloudRuntimeConfigSnapshot(stored); + } const result = await migrateRunEvidence(db, { apply: args.includes("--apply"), companyId: value("--company"), afterRunId: value("--after-run"), afterTraceId: value("--after-trace"), limit: Number(value("--limit") || 50), + preserveSettledNativeFiles: args.includes("--preserve-settled-native-files"), }); console.log(JSON.stringify(result, null, 2)); if (result.items.some((item) => item.state === "blocked")) process.exitCode = 1; diff --git a/server/src/services/native-log-archive.integration.test.ts b/server/src/services/native-log-archive.integration.test.ts new file mode 100644 index 00000000000..863431e91fb --- /dev/null +++ b/server/src/services/native-log-archive.integration.test.ts @@ -0,0 +1,176 @@ +import { createHash, randomUUID } from "node:crypto"; +import { appendFile, link, mkdir, readFile, rm, symlink, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { eq, sql } from "drizzle-orm"; +import { afterEach, expect, it, vi } from "vitest"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "../__tests__/helpers/native-restart.js"; +import { createMaintenanceS3Boundary } from "../__tests__/helpers/maintenance/s3-boundary.js"; +import { withMaintenanceProvider } from "../__tests__/helpers/maintenance/server-harness.js"; +import { migrateRunEvidence } from "./run-evidence-migration.js"; +import { getRunLogStore } from "./run-log-store.js"; +import { createObjectByteJournal } from "./object-byte-journal.js"; +import { readRecoverableRunLog } from "./native-runtime/native-run-log-recovery.js"; +import { tryLockNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalization-ownership.js"; +import * as storage from "./run-evidence-storage.js"; + +function beforeUuid(id: string) { + const hex = (BigInt(`0x${id.replaceAll("-", "")}`) - 1n).toString(16).padStart(32, "0"); + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`; +} + +describeEmbeddedPostgres("online native historical log preservation", () => { + const environment = nativeRestartFixture(); + afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); }); + async function fixture() { + const f = await environment.seedRun(randomUUID()); + const { companyId, agentId, paperclipHome } = environment.current; + const ref = `${companyId}/${agentId}/${f.runId}.ndjson`, base = join(paperclipHome, `archive-test-${f.runId}`); + const file = join(base, ref), bytes = Buffer.from('{"seq":1,"chunk":"surviving 再開 output"}\n{"partial":'); + await mkdir(dirname(file), { recursive: true }); await writeFile(file, bytes); + await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), logStore: "local_file", logRef: ref, + logBytes: null, logSha256: null, controllerLeaseExpiresAt: new Date(0), processPid: 123456, + processStartedAt: new Date(0), errorCode: "native_run_log_source_integrity_error", error: "original warning" }) + .where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, + nextAttemptAt: null, recoveryState: "blocked" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + const options = { apply: true, companyId, afterRunId: beforeUuid(f.runId), limit: 1, + kinds: ["log"] as const, preserveSettledNativeFiles: true }; + const s3 = createMaintenanceS3Boundary(); + return { ...f, companyId, agentId, ref, base, file, bytes, options, s3 }; + } + async function withSource(action: (f: Awaited>) => Promise) { + const f = await fixture(); + await withMaintenanceProvider(f.s3.server, async endpoint => { + for (const [key, value] of Object.entries({ PAPERCLIP_STORAGE_PROVIDER: "s3", PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, + PAPERCLIP_STORAGE_S3_BUCKET: "maintenance-test", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", + PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "", + AWS_ACCESS_KEY_ID: "fixture-only", AWS_SECRET_ACCESS_KEY: "fixture-only", + RUN_LOG_S3_ENDPOINT: endpoint, RUN_LOG_S3_BUCKET: "maintenance-test", RUN_LOG_S3_REGION: "us-east-1", + RUN_LOG_S3_FORCE_PATH_STYLE: "true", RUN_LOG_S3_PREFIX: "run-logs", RUN_LOG_BASE_PATH: f.base, + PAPERCLIP_EPHEMERAL_MIGRATION: "false" })) vi.stubEnv(key, value); + await action(f); + expect(f.s3.snapshot().errors).toEqual([]); + }); + } + + it("archives a settled failed source while unrelated work runs and preserves its original failure", async () => { + await withSource(async f => { + const unrelated = await environment.seedRun(randomUUID()); + const [before] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [ownerBefore] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const result = await migrateRunEvidence(f.db, f.options); + expect(result.items).toMatchObject([{ id: f.runId, state: "migrated", historicalArchive: true, + bytes: f.bytes.length, sha256: createHash("sha256").update(f.bytes).digest("hex") }]); + const [after] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(after).toEqual({ ...before, logStore: "s3_archive", logRef: `${f.companyId}/${f.agentId}/${f.runId}.archive` }); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(ownerBefore); + expect((await unrelated.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0].status).toBe("running"); + expect(await readFile(f.file)).toEqual(f.bytes); + await rm(f.file); + const store = getRunLogStore(), handle = { store: "s3_archive" as const, logRef: after.logRef! }; + expect((await readRecoverableRunLog(f.db, store, handle)).content).toBe(f.bytes.toString()); + expect(() => store.begin({ companyId: f.companyId, agentId: f.agentId, runId: f.runId, previousStore: "s3_archive" })).toThrow("read-only"); + }); + }); + + it("preserves a committed successful run without changing its result or completeness metadata", async () => { + await withSource(async f => { + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: run.completionContractId!, serverFingerprint: f.runId, schemaStatus: "accepted", + resultJson: {}, canonicalSha256: f.runId }).returning(); + await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, recoveryState: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "migrated", historicalArchive: true }]); + expect((await f.db.select().from(nativeRunResults).where(eq(nativeRunResults.id, result.id)))[0]).toEqual(result); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "succeeded", logBytes: null, logSha256: null, error: "original warning" }); + }); + }); + + it("preserves exact bytes across the bounded archive segment boundary", async () => { + await withSource(async f => { + const bytes = Buffer.alloc(4 * 1024 * 1024 + 7, 97); + await writeFile(f.file, bytes); + expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "migrated", bytes: bytes.length, + sha256: createHash("sha256").update(bytes).digest("hex") }]); + await rm(f.file); + const archive = createObjectByteJournal({ provider: storage.resolveRunEvidenceStorage()!.provider, + prefix: `run-logs/segments-v1/${f.companyId}/${f.agentId}/${f.runId}.archive` }); + const head = (await archive.head())!.head; + expect(head).toMatchObject({ closed: true, sourceCursor: 0 }); + expect(Buffer.concat([await archive.readRange(head, 0, 4 * 1024 * 1024), + await archive.readRange(head, 4 * 1024 * 1024, 7)])).toEqual(bytes); + }); + }); + + it.each(["expired_owner", "pending_attempt", "pending_recovery", "allocation", "process_group", "physical_owner", "task_bound"])("refuses unresolved source authority: %s", async kind => { + await withSource(async f => { + if (kind === "expired_owner") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "not-a-stop-proof", leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (kind === "pending_attempt") await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (kind === "pending_recovery") await f.db.update(nativeRunFinalizations).set({ recoveryState: "awaiting_runner_reattach" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (kind === "allocation") await f.db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, providerLeaseId: f.runId, status: "active" }); + if (kind === "process_group") await f.db.update(heartbeatRuns).set({ processGroupId: 999999 }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "physical_owner") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: { token: "unknown-owner" } } }) + .where(eq(heartbeatRuns.id, f.runId)); + if (kind === "task_bound") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked", reason: "native_log_archive_source_unsettled" }]); + expect(f.s3.snapshot().requests).toEqual([]); + expect(await readFile(f.file)).toEqual(f.bytes); + }); + }); + + it("honors the physical native finalizer lock and permits a read-only dry run", async () => { + await withSource(async f => { + let release!: () => void, ready!: () => void; + const acquired = new Promise(resolve => { ready = resolve; }), held = new Promise(resolve => { release = resolve; }); + const owner = f.db.transaction(async tx => { + expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: f.runId })).toBe(true); + ready(); await held; + }); + try { + await acquired; + expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked", reason: "native_log_archive_source_busy" }]); + } finally { release(); await owner; } + const result = await f.db.transaction(tx => migrateRunEvidence(tx as unknown as Db, { ...f.options, apply: false }), { accessMode: "read only" }); + expect(result.items).toMatchObject([{ state: "inventoried", historicalArchive: true }]); + expect(f.s3.snapshot().requests).toEqual([]); + }); + }); + + it.each(["owner", "file"])("rechecks %s after the remote archive closes", async kind => { + await withSource(async f => { + const config = storage.resolveRunEvidenceStorage()!; + vi.spyOn(storage, "resolveRunEvidenceStorage").mockReturnValue(config); + const original = config.provider.putObjectConditional!.bind(config.provider); let changed = false; + vi.spyOn(config.provider, "putObjectConditional").mockImplementation(async input => { + const result = await original(input); + if (!changed && input.objectKey.endsWith("/head.json") && Buffer.isBuffer(input.body) && JSON.parse(input.body.toString()).closed) { + changed = true; + if (kind === "file") await appendFile(f.file, "late source bytes"); + else await f.db.update(nativeRunFinalizations).set({ controllerGeneration: sql`${nativeRunFinalizations.controllerGeneration} + 1` }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + } + return result; + }); + expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked" }]); + expect(changed).toBe(true); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].logStore).toBe("local_file"); + expect(f.s3.snapshot().objects.length).toBeGreaterThan(0); + }); + }); + + it.each(["symlink", "hardlink"])("refuses aliased source files: %s", async kind => { + await withSource(async f => { + const other = join(f.base, "other"); + if (kind === "hardlink") await link(f.file, other); + else { await writeFile(other, f.bytes); await rm(f.file); await symlink(other, f.file); } + expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked" }]); + expect(f.s3.snapshot().requests).toEqual([]); + }); + }); +}); diff --git a/server/src/services/native-runtime/native-log-archive-source.ts b/server/src/services/native-runtime/native-log-archive-source.ts new file mode 100644 index 00000000000..85ed0c35804 --- /dev/null +++ b/server/src/services/native-runtime/native-log-archive-source.ts @@ -0,0 +1,69 @@ +import { and, eq, ne, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { nativeSha256 } from "./canonical.js"; +import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; + +export type NativeLogArchiveSource = { + id: string; companyId: string; agentId: string; status: string; + logRef: string | null; bytes: number | null; sha256: string | null; +}; + +async function readSource(db: Db, source: NativeLogArchiveSource, lock: boolean, expected?: string) { + function fail(): never { throw new Error("native_log_archive_source_unsettled"); } + if (source.logRef !== `${source.companyId}/${source.agentId}/${source.id}.ndjson`) fail(); + if (lock && !await tryLockNativeWorkspaceFinalization(db, { companyId: source.companyId, runId: source.id })) + throw new Error("native_log_archive_source_busy"); + const scope = and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, source.companyId)); + const owners = db.select({ issueId: nativeRunFinalizations.issueId, phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, recoveryState: nativeRunFinalizations.recoveryState, + leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + nextAttemptAt: nativeRunFinalizations.nextAttemptAt, controllerGeneration: nativeRunFinalizations.controllerGeneration, + attempt: nativeRunFinalizations.attempt, controllerBootId: nativeRunFinalizations.controllerBootId, + controllerPid: nativeRunFinalizations.controllerPid, controllerProcessStartedAt: nativeRunFinalizations.controllerProcessStartedAt, + }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, source.id), eq(nativeRunFinalizations.companyId, source.companyId))); + const [owner] = await (lock ? owners.for("update") : owners); + const rows = db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, runtimeMode: heartbeatRuns.runtimeMode, + issueId: heartbeatRuns.nativeIssueId, sessionId: heartbeatRuns.nativeSessionId, + logStore: heartbeatRuns.logStore, logRef: heartbeatRuns.logRef, bytes: heartbeatRuns.logBytes, sha256: heartbeatRuns.logSha256, + processGroupId: heartbeatRuns.processGroupId, + physicalOwner: sql`coalesce(${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceFinalizationOwner' <> 'null'::jsonb, false)`, + }).from(heartbeatRuns).where(scope); + const [run] = await (lock ? rows.for("update") : rows); + if (!run || !owner || !run.issueId || !run.sessionId || run.runtimeMode !== "native" || !run.finishedAt + || run.agentId !== source.agentId || run.status !== source.status + || !["succeeded", "failed", "cancelled", "timed_out", "interrupted"].includes(run.status) + || run.logStore !== "local_file" || run.logRef !== source.logRef || run.bytes !== source.bytes || run.sha256 !== source.sha256 + || run.processGroupId !== null || run.physicalOwner || owner.issueId !== run.issueId + || owner.leaseOwner !== null || owner.leaseExpiresAt !== null || owner.nextAttemptAt !== null + || !(owner.phase === "committed" && owner.resultId !== null && owner.recoveryState === null + || owner.phase === "terminal_failure" && run.status !== "succeeded" + && (owner.recoveryState === null || owner.recoveryState === "blocked"))) fail(); + const [task] = await db.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }) + .from(issues).where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId))); + const [allocation] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + sql`${environmentLeases.releasedAt} is null`)).limit(1); + const [alias] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + ne(heartbeatRuns.id, run.id), eq(heartbeatRuns.logStore, "local_file"), eq(heartbeatRuns.logRef, run.logRef!))).limit(1); + if (!task || task.executionRunId === run.id || task.checkoutRunId === run.id || allocation || alias) fail(); + const fence = nativeSha256({ ...run, finishedAt: run.finishedAt.toISOString(), owner: { phase: owner.phase, generation: owner.controllerGeneration, + attempt: owner.attempt, bootId: owner.controllerBootId, pid: owner.controllerPid, + startedAt: owner.controllerProcessStartedAt?.toISOString() ?? null, resultId: owner.resultId, recoveryState: owner.recoveryState } }); + if (expected && fence !== expected) throw new Error("native_log_archive_source_changed"); + return fence; +} + +export async function inspectNativeLogArchiveSource(db: Db, source: NativeLogArchiveSource) { + return readSource(db, source, false); +} + +export async function withNativeLogArchiveSource(db: Db, source: NativeLogArchiveSource, expected: string | undefined, + action: (tx: Db, fence: string) => Promise): Promise { + return db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout='1s'`); + const fence = await readSource(tx as unknown as Db, source, true, expected); + return action(tx as unknown as Db, fence); + }); +} diff --git a/server/src/services/run-evidence-migration.ts b/server/src/services/run-evidence-migration.ts index 41d636a2f55..50d0ba84ebc 100644 --- a/server/src/services/run-evidence-migration.ts +++ b/server/src/services/run-evidence-migration.ts @@ -23,6 +23,9 @@ import { createNativeRunLogCoverageVerifier, nativeRunLogSourceWhere, } from "./native-runtime/native-run-log-export.js"; +import { inspectNativeLogArchiveSource, withNativeLogArchiveSource } from "./native-runtime/native-log-archive-source.js"; +import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; +import { logger } from "../middleware/logger.js"; const SETTLED = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; export interface RunEvidenceMigrationItem { @@ -48,6 +51,7 @@ export async function migrateRunEvidence( /** Offline enrollment only: preserve unfinalized local bytes without * claiming canonical native-event coverage or repairing the old run. */ preserveHistoricalFiles?: boolean; + preserveSettledNativeFiles?: boolean; }, ): Promise<{ items: RunEvidenceMigrationItem[]; @@ -61,6 +65,9 @@ export async function migrateRunEvidence( const config = resolveRunEvidenceStorage(); if (!config) throw new Error("Run evidence migration requires instance S3 storage"); const preserveHistoricalFiles = options.preserveHistoricalFiles === true; + const preserveSettledNativeFiles = options.preserveSettledNativeFiles === true; + if (preserveHistoricalFiles && preserveSettledNativeFiles) + throw new Error("historical_log_archive_authority_conflict"); async function assertNoActiveRuns() { options.signal?.throwIfAborted(); const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) @@ -133,9 +140,14 @@ export async function migrateRunEvidence( try { if (!SETTLED.includes(run.status)) throw new Error("legacy_writer_not_settled"); if (!run.logRef) throw new Error("legacy_reference_missing"); + const originalLogRef = run.logRef; await assertDrained(run.id, run.companyId); - const historicalArchive = preserveHistoricalFiles && run.runtimeMode === "native" && + const historicalArchive = (preserveHistoricalFiles || preserveSettledNativeFiles) && run.runtimeMode === "native" && (run.bytes === null || run.sha256 === null); + const scopedArchive = historicalArchive && preserveSettledNativeFiles; + const sourceFence = scopedArchive ? options.apply + ? await withNativeLogArchiveSource(db, run, undefined, async (_tx, fence) => fence) + : await inspectNativeLogArchiveSource(db, run) : undefined; // Keep raw history separate from the canonical log namespace. A later // writer must never acquire this archive as a native source prefix. const ref = `${run.companyId}/${run.agentId}/${run.id}${historicalArchive ? ".archive" : ""}`; @@ -143,11 +155,16 @@ export async function migrateRunEvidence( provider: config.provider, prefix: `${config.keyPrefix || "run-logs"}/segments-v1/${ref}`, }); - let writer = options.apply + const archiveSource = scopedArchive + ? await openLegacyRunLogForMigration(run.logRef, { requireCanonicalLocalFile: true }) : undefined; + let writer: Awaited> | null = null; + try { writer = options.apply ? await journal.claim({ expectedOwner: (await journal.head(options.signal))?.head.owner ?? null, signal: options.signal, assertAuthority: async () => { + if (scopedArchive) + await withNativeLogArchiveSource(db, run, sourceFence, async () => undefined); const [current] = await db .select({ id: heartbeatRuns.id }) .from(heartbeatRuns) @@ -164,7 +181,8 @@ export async function migrateRunEvidence( }, }) : null; - const source = await openLegacyRunLogForMigration(run.logRef); + } catch (error) { archiveSource?.stream.destroy(); throw error; } + const source = archiveSource ?? await openLegacyRunLogForMigration(run.logRef); if (source.source === "mirror" && (run.bytes === null || !run.sha256)) { source.stream.destroy(); throw new Error("mirror_completeness_unproven"); @@ -233,9 +251,10 @@ export async function migrateRunEvidence( if (writer.head.end !== offset) throw new Error("migration_prefix_exceeds_source"); const remoteHash = createHash("sha256"); - for (let start = 0; start < offset; start += 1024 * 1024) + const verificationBytes = scopedArchive ? 4 * 1024 * 1024 : 1024 * 1024; + for (let start = 0; start < offset; start += verificationBytes) remoteHash.update( - await journal.readRange(writer.head, start, 1024 * 1024, options.signal), + await journal.readRange(writer.head, start, verificationBytes, options.signal), ); if (remoteHash.digest("hex") !== digest) throw new Error("migration_remote_digest_mismatch"); @@ -243,7 +262,7 @@ export async function migrateRunEvidence( await journal.close(writer, options.signal); options.signal?.throwIfAborted(); if (source.source === "local") await source.assertUnchanged(); - const [updated] = await db + const publish = (database: Db) => database .update(heartbeatRuns) .set({ logStore: historicalArchive ? "s3_archive" : "s3_segments", logRef: ref }) .where( @@ -251,7 +270,7 @@ export async function migrateRunEvidence( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.logStore, "local_file"), - eq(heartbeatRuns.logRef, run.logRef), + eq(heartbeatRuns.logRef, originalLogRef), run.bytes === null ? isNull(heartbeatRuns.logBytes) : eq(heartbeatRuns.logBytes, run.bytes), @@ -268,8 +287,25 @@ export async function migrateRunEvidence( ), ) .returning({ id: heartbeatRuns.id }); + const publications: ActivityPublication[] = []; + const [updated] = scopedArchive + ? await withNativeLogArchiveSource(db, run, sourceFence, async tx => { + if (source.source !== "local") throw new Error("native_log_archive_local_source_required"); + await source.assertUnchanged(); + const result = await publish(tx); + if (!result.length) throw new Error("migration_publication_conflict"); + await logActivity(tx, { companyId: run.companyId, actorType: "system", actorId: "run-evidence-migration", + action: "run.historical_log_archived", entityType: "heartbeat_run", entityId: run.id, + agentId: run.agentId, runId: run.id, details: { bytes: offset, sha256: digest, sourceFence } }, publications); + return result; + }) + : await publish(db); if (!updated) throw new Error("migration_publication_conflict"); item.state = "migrated"; + for (const publication of publications) { + try { publishActivity(publication); } + catch { logger.warn({ runId: run.id }, "Historical log archive notification failed after publication"); } + } } } catch (error) { options.signal?.throwIfAborted(); diff --git a/server/src/services/run-log-store.ts b/server/src/services/run-log-store.ts index 1e9120a21b8..2138026afc7 100644 --- a/server/src/services/run-log-store.ts +++ b/server/src/services/run-log-store.ts @@ -1,4 +1,4 @@ -import { createReadStream, createWriteStream, promises as fs } from "node:fs"; +import { constants, createReadStream, createWriteStream, promises as fs } from "node:fs"; import path from "node:path"; import { pipeline } from "node:stream/promises"; import { createHash, randomUUID } from "node:crypto"; @@ -617,9 +617,28 @@ export async function flushInFlightRunLogMirrors(): Promise { } -export async function openLegacyRunLogForMigration(logRef: string) { +export async function openLegacyRunLogForMigration(logRef: string, options: { requireCanonicalLocalFile?: boolean } = {}) { const basePath = process.env.RUN_LOG_BASE_PATH ?? path.resolve(resolvePaperclipInstanceRoot(), "data", "run-logs"); - const file = resolveWithin(basePath, logRef); + const base = options.requireCanonicalLocalFile ? await fs.realpath(basePath) : basePath; + const file = resolveWithin(base, logRef); + if (options.requireCanonicalLocalFile) { + if (await fs.realpath(file) !== file) throw new Error("native_log_archive_source_alias"); + const descriptor = await fs.open(file, constants.O_RDONLY | constants.O_NOFOLLOW); + try { + const stat = await descriptor.stat(); + if (!stat.isFile() || stat.nlink !== 1) throw new Error("native_log_archive_source_not_private"); + return { + stream: descriptor.createReadStream({ highWaterMark: 4 * 1024 * 1024 }), bytes: stat.size, source: "local" as const, + async assertUnchanged() { + const current = await fs.lstat(file); + if (await fs.realpath(file) !== file || !current.isFile() || current.nlink !== 1 + || current.dev !== stat.dev || current.ino !== stat.ino || current.size !== stat.size + || current.mtimeMs !== stat.mtimeMs || current.ctimeMs !== stat.ctimeMs) + throw new Error("legacy_log_source_changed"); + }, + }; + } catch (error) { await descriptor.close(); throw error; } + } try { const stat = await fs.stat(file); if (!stat.isFile()) throw new Error("Legacy run log is not a regular file");