Preserve unfinalized local logs as read-only enrollment archives

doc/run-evidence-storage.md

@@ -239,6 +239,24 @@ chunks whose identities cannot be verified block publication as
 does not establish completeness. Publication checks that the original reference,
 completeness metadata, and source cursor have not changed. Keep the original
 volume and resolve blocked inventory before cutover.
+
+The offline ephemeral-enrollment process can also preserve unfinalized native
+local files as read-only `s3_archive` logs. This path requires
+`PAPERCLIP_EPHEMERAL_MIGRATION=true`, all old deployments stopped, no queued or
+running execution anywhere in the instance, and expired writer/finalizer leases.
+It copies exact surviving bytes to a separate `<run>.archive` journal, checks
+that the local source did not change, and verifies the remote SHA-256 before
+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.
+
+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
+writers cannot reopen or modify the archive. Its source cursor remains zero.
+The normal background importer retains its strict native coverage checks.
+Rollbacks after enrollment require readers that support `s3_archive`.
+
 An interrupted trace import keeps an `s3-import-v1:` reference: reads still use
 the original local sidecars, while deletion and expiry retain responsibility for
 the staged object prefix. The automatic worker resumes the verified switch after restart or retry.

server/src/maintenance/ephemeral-migration-data.ts

@@ -51,6 +51,7 @@ export async function migrateEphemeralServerData(db: Db, input: {
     do {
       signal.throwIfAborted();
       const page = await migrateRunEvidence(db, { apply: true, kinds: [kind], limit: 50, signal,
+        preserveHistoricalFiles: true,
         ...(kind === "log" ? { afterRunId: after } : { afterTraceId: after }) });
       if (page.items.some(item => item.state !== "migrated")) throw new Error(`${kind}_migration_blocked`);
       counts[kind] = (counts[kind] ?? 0) + page.items.length;

server/src/services/object-run-log-store.ts

@@ -343,7 +343,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
       }
     },
     async read(handle, opts = {}) {
-      if (handle.store !== "s3_segments") throw new Error("Invalid object run log backend");
+      if (handle.store !== "s3_segments" && handle.store !== "s3_archive") throw new Error("Invalid object run log backend");
       const store = journal(handle.logRef),
         snapshot = await store.head(opts.signal);
       if (!snapshot || snapshot.head.deleted) throw new Error("Object run log not found");

server/src/services/run-evidence-migration.integration.test.ts

@@ -22,6 +22,8 @@ import { nativeSha256 } from "./native-runtime/canonical.js";
 import { createObjectByteJournal } from "./object-byte-journal.js";
 import type { PrpEvent } from "../vendor/paperclip-runner/index.js";
 import { compactRunLogChunk } from "./heartbeat/run-log.js";
+import { getRunLogStore, openLegacyRunLogForMigration } from "./run-log-store.js";
+import { readRecoverableRunLog } from "./native-runtime/native-run-log-recovery.js";
 
 const endpoint = process.env.PAPERCLIP_LOG_TEST_S3_ENDPOINT;
 const databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_EVIDENCE_MIGRATION_DATABASE_URL");
@@ -84,6 +86,82 @@ describe.skipIf(!endpoint || !databaseUrl)("run evidence migration (Postgres and
     if (directory) await fs.rm(directory, { recursive: true, force: true });
     vi.unstubAllEnvs();
   });
+  it("archives incomplete native history only offline, reads it without the old disk, and never repairs or reopens it", async () => {
+    const companyId = randomUUID(), agentId = randomUUID();
+    const provider = createS3StorageProvider({ endpoint, bucket, region: "us-east-1", forcePathStyle: true });
+    const cases = ["old-format", "interrupted", "corrupt-source", "wrong-digest", "active-lease", "active-finalizer", "mirror"] as const;
+    const runs = new Map<string, { id: string; ref: string; bytes: Buffer }>();
+    for (const name of cases) {
+      const id = randomUUID(), ref = `${companyId}/${agentId}/${id}.ndjson`;
+      // Retain the original old formatter bytes, including its partial last line.
+      const bytes = Buffer.from('{"seq":1,"stream":"stdout","chunk":"surviving 再開 output"}\n{"partial":');
+      runs.set(name, { id, ref, bytes });
+      await fs.mkdir(path.dirname(path.join(directory, "logs", ref)), { recursive: true });
+      if (name === "mirror") await provider.putObject({ objectKey: `run-logs/${ref}`, body: bytes,
+        contentType: "application/x-ndjson", contentLength: bytes.length });
+      else await fs.writeFile(path.join(directory, "logs", ref), bytes);
+      await db.execute(sql`insert into heartbeat_runs(id,company_id,agent_id,status,runtime_mode,log_store,log_ref,log_sha256,controller_lease_expires_at)
+        values(${id},${companyId},${agentId},${name === "interrupted" ? "interrupted" : "failed"},'native','local_file',${ref},
+          ${name === "wrong-digest" ? "0".repeat(64) : null},${name === "active-lease" ? new Date(Date.now()+60_000).toISOString() : null})`);
+      if (name === "corrupt-source") await db.execute(sql`insert into heartbeat_run_events
+        values(${companyId},${id},1,'{}'::jsonb,'old-source','old-event',1,'bad-digest','item.delta')`);
+      if (name === "active-finalizer") await db.execute(sql`insert into native_run_finalizations
+        values(${id},${companyId},${new Date(Date.now()+60_000).toISOString()})`);
+    }
+    const snapshot = await db.$client.unsafe("select * from heartbeat_run_events where company_id=$1", [companyId]);
+    vi.stubEnv("RUN_LOG_S3_BUCKET", bucket);
+    vi.stubEnv("RUN_LOG_S3_ENDPOINT", endpoint!);
+    vi.stubEnv("RUN_LOG_S3_REGION", "us-east-1");
+    vi.stubEnv("RUN_LOG_S3_FORCE_PATH_STYLE", "true");
+    vi.stubEnv("RUN_LOG_S3_PREFIX", "run-logs");
+    vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "false");
+    const options = { apply: true, companyId, kinds: ["log"] as const, preserveHistoricalFiles: true };
+    await expect(migrateRunEvidence(db, options)).rejects.toThrow("historical_log_archive_requires_offline_migration");
+    vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true");
+    const active = randomUUID();
+    // A different company's active execution must also block offline archival.
+    await db.execute(sql`insert into heartbeat_runs(id,company_id,agent_id,status) values(${active},${randomUUID()},${randomUUID()},'running')`);
+    await expect(migrateRunEvidence(db, options)).rejects.toThrow("active_execution_requires_drain");
+    await db.execute(sql`delete from heartbeat_runs where id=${active}`);
+    const result = await migrateRunEvidence(db, options);
+    const store = getRunLogStore();
+    for (const name of cases.slice(0, 3)) {
+      const run = runs.get(name)!;
+      expect(result.items.find(item => item.id === run.id)).toMatchObject({
+        state: "migrated", historicalArchive: true, bytes: run.bytes.length,
+        sha256: createHash("sha256").update(run.bytes).digest("hex"),
+      });
+      const [row] = await db.$client.unsafe("select status,log_store,log_ref,log_bytes,log_sha256 from heartbeat_runs where id=$1", [run.id]);
+      expect(row).toMatchObject({ status: name === "interrupted" ? "interrupted" : "failed",
+        log_store: "s3_archive", log_ref: `${companyId}/${agentId}/${run.id}.archive`, log_bytes: null, log_sha256: null });
+      await fs.rm(path.join(directory, "logs", run.ref));
+      const handle = { store: "s3_archive" as const, logRef: row!.log_ref as string };
+      expect((await readRecoverableRunLog(db, store, handle)).content).toBe(run.bytes.toString("utf8"));
+      expect(() => store.begin({ companyId, agentId, runId: run.id, previousStore: "s3_archive" })).toThrow("read-only");
+      const event = { ts: "now", stream: "stdout" as const, chunk: "must not append" };
+      expect(() => store.append(handle, event)).toThrow("read-only");
+      expect(() => store.enqueue!(handle, event)).toThrow("read-only");
+      expect(() => store.finalize(handle)).toThrow("read-only");
+      await expect(store.flush!(handle)).rejects.toThrow("read-only");
+      const journal = createObjectByteJournal({ provider, prefix: `run-logs/segments-v1/${handle.logRef}` });
+      expect((await journal.head())!.head.sourceCursor).toBe(0);
+      expect((await store.read(handle)).content).toBe(run.bytes.toString("utf8"));
+    }
+    expect(await db.$client.unsafe("select * from heartbeat_run_events where company_id=$1", [companyId])).toEqual(snapshot);
+    for (const [name, reason] of [["wrong-digest", "legacy_length_or_digest_mismatch"],
+      ["active-lease", "legacy_writer_not_drained"], ["active-finalizer", "legacy_writer_not_drained"],
+      ["mirror", "mirror_completeness_unproven"]]) {
+      expect(result.items.find(item => item.id === runs.get(name!)!.id)).toMatchObject({ state: "blocked", reason });
+    }
+    vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "false");
+    // A source that changes after it was opened cannot be published as a stable archive.
+    const sourceRun = runs.get("wrong-digest")!;
+    const source = await openLegacyRunLogForMigration(sourceRun.ref);
+    for await (const _ of source.stream) { /* consume the real file */ }
+    await fs.appendFile(path.join(directory, "logs", sourceRun.ref), "changed");
+    if (source.source !== "local") throw new Error("expected real local source");
+    await expect(source.assertUnchanged()).rejects.toThrow("legacy_log_source_changed");
+  });
   it("preserves exact historical bytes and expiry, while refusing active/missing/corrupt sources", async () => {
     const company = randomUUID(),
       agent = randomUUID(),
@@ -270,6 +348,7 @@ describe.skipIf(!endpoint || !databaseUrl)("run evidence migration (Postgres and
       "finalized",
       "missing-size",
       "missing-digest",
+      "different-source",
       "wrong-digest",
       "wrong-size",
     ] as const;
@@ -304,15 +383,19 @@ describe.skipIf(!endpoint || !databaseUrl)("run evidence migration (Postgres and
       const digest = createHash("sha256").update(bytes).digest("hex");
       await db.execute(sql`insert into heartbeat_runs(id,company_id,agent_id,status,runtime_mode,log_store,log_ref,log_bytes,log_sha256)
         values(${runId},${companyId},${agentId},'succeeded','native','local_file',${logRef},
-        ${state === "missing-size" ? null : bytes.length + (state === "wrong-size" ? 1 : 0)},
+        ${state === "missing-size" || state === "different-source" ? null : bytes.length + (state === "wrong-size" ? 1 : 0)},
         ${state === "missing-digest" ? null : state === "wrong-digest" ? "0".repeat(64) : digest})`);
+      // A similar truncated display string cannot prove a different canonical event.
+      if (state === "different-source") event.payload = { text: "different source ".repeat(12_000) };
       await db.execute(sql`insert into heartbeat_run_events values(${companyId},${runId},1,
         ${JSON.stringify({ prpEvent: event })}::jsonb,${sourceInstanceId},${event.sourceEventId},1,${nativeSha256(event)},${event.eventType})`);
     }
     const result = await migrateRunEvidence(db, { apply: true, companyId });
     const finalized = runs.get("finalized")!;
-    expect(result.items.find((item) => item.id === finalized.runId)?.state).toBe("migrated");
-    for (const state of cases.slice(1)) {
+    for (const state of cases.slice(0, 3)) {
+      expect(result.items.find((item) => item.id === runs.get(state)!.runId)?.state).toBe("migrated");
+    }
+    for (const state of cases.slice(3)) {
       const run = runs.get(state)!;
       expect(result.items.find((item) => item.id === run.runId)).toMatchObject({
         state: "blocked",

server/src/services/run-evidence-migration.ts

@@ -23,13 +23,14 @@ import {
   nativeRunLogSourceWhere,
 } from "./native-runtime/native-run-log-export.js";
 
-const SETTLED = ["succeeded", "failed", "cancelled", "timed_out"];
+const SETTLED = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"];
 export interface RunEvidenceMigrationItem {
   kind: "log" | "trace";
   id: string;
   state: "inventoried" | "migrated" | "blocked";
   bytes?: number;
   sha256?: string;
+  historicalArchive?: boolean;
   reason?: string;
 }
 
@@ -43,6 +44,9 @@ export async function migrateRunEvidence(
     limit?: number;
     kinds?: readonly ("log" | "trace")[];
     signal?: AbortSignal;
+    /** Offline enrollment only: preserve unfinalized local bytes without
+     * claiming canonical native-event coverage or repairing the old run. */
+    preserveHistoricalFiles?: boolean;
   },
 ): Promise<{
   items: RunEvidenceMigrationItem[];
@@ -55,6 +59,16 @@ export async function migrateRunEvidence(
     throw new Error("Migration limit must be between 1 and 100");
   const config = resolveRunEvidenceStorage();
   if (!config) throw new Error("Run evidence migration requires instance S3 storage");
+  const preserveHistoricalFiles = options.preserveHistoricalFiles === true;
+  const noActiveRuns = sql`not exists (select 1 from ${heartbeatRuns} where ${inArray(heartbeatRuns.status, ["queued", "running"])})`;
+  if (preserveHistoricalFiles) {
+    if (process.env.PAPERCLIP_EPHEMERAL_MIGRATION !== "true") {
+      throw new Error("historical_log_archive_requires_offline_migration");
+    }
+    const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
+      .where(inArray(heartbeatRuns.status, ["queued", "running"])).limit(1);
+    if (active) throw new Error("active_execution_requires_drain");
+  }
   function drainedWriter(now = new Date()) {
     return and(
       inArray(heartbeatRuns.status, SETTLED),
@@ -67,6 +81,7 @@ export async function migrateRunEvidence(
         eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId),
         gt(nativeRunFinalizations.leaseExpiresAt, now),
       )})`,
+      preserveHistoricalFiles ? noActiveRuns : undefined,
     );
   }
   async function assertDrained(runId: string, companyId: string) {
@@ -116,7 +131,11 @@ export async function migrateRunEvidence(
       if (!SETTLED.includes(run.status)) throw new Error("legacy_writer_not_settled");
       if (!run.logRef) throw new Error("legacy_reference_missing");
       await assertDrained(run.id, run.companyId);
-      const ref = `${run.companyId}/${run.agentId}/${run.id}`;
+      const historicalArchive = preserveHistoricalFiles && run.runtimeMode === "native" &&
+        (run.bytes === null || run.sha256 === null);
+      // 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" : ""}`;
       const journal = createObjectByteJournal({
         provider: config.provider,
         prefix: `${config.keyPrefix || "run-logs"}/segments-v1/${ref}`,
@@ -149,7 +168,7 @@ export async function migrateRunEvidence(
       }
       const hash = createHash("sha256");
       const nativeCoverage =
-        run.runtimeMode === "native" && (run.bytes === null || run.sha256 === null)
+        !historicalArchive && run.runtimeMode === "native" && (run.bytes === null || run.sha256 === null)
           ? createNativeRunLogCoverageVerifier(db, run.companyId, run.id)
           : undefined;
       let offset = 0,
@@ -180,6 +199,7 @@ export async function migrateRunEvidence(
         source.stream.destroy();
       }
       const digest = hash.digest("hex");
+      if (source.source === "local") await source.assertUnchanged();
       if (
         (source.bytes !== undefined && source.bytes !== offset) ||
         (run.bytes !== null && run.bytes !== offset) ||
@@ -194,8 +214,9 @@ export async function migrateRunEvidence(
       }
       item.bytes = offset;
       item.sha256 = digest;
+      if (historicalArchive) item.historicalArchive = true;
       let sourceCursor = (await nativeCoverage?.finish()) ?? 0;
-      if (run.runtimeMode === "native" && !nativeCoverage) {
+      if (run.runtimeMode === "native" && !nativeCoverage && !historicalArchive) {
         const [source] = await db
           .select({ seq: heartbeatRunEvents.seq })
           .from(heartbeatRunEvents)
@@ -218,9 +239,10 @@ export async function migrateRunEvidence(
         writer.head = { ...writer.head, sequence, sourceCursor };
         await journal.close(writer, options.signal);
         options.signal?.throwIfAborted();
+        if (source.source === "local") await source.assertUnchanged();
         const [updated] = await db
           .update(heartbeatRuns)
-          .set({ logStore: "s3_segments", logRef: ref })
+          .set({ logStore: historicalArchive ? "s3_archive" : "s3_segments", logRef: ref })
           .where(
             and(
               eq(heartbeatRuns.id, run.id),
@@ -234,7 +256,7 @@ export async function migrateRunEvidence(
                 ? isNull(heartbeatRuns.logSha256)
                 : eq(heartbeatRuns.logSha256, run.sha256),
               drainedWriter(),
-              run.runtimeMode === "native"
+              run.runtimeMode === "native" && !historicalArchive
                 ? sql`not exists (select 1 from ${heartbeatRunEvents} where ${and(
                     nativeRunLogSourceWhere(run.companyId, run.id),
                     gt(heartbeatRunEvents.seq, sourceCursor),

server/src/services/run-log-store.ts

@@ -10,9 +10,9 @@ import type { StorageProvider } from "../storage/types.js";
 import { createObjectRunLogStore } from "./object-run-log-store.js";
 import { resolveRunEvidenceStorage } from "./run-evidence-storage.js";
 
-export type RunLogStoreType = "local_file" | "s3_segments";
+export type RunLogStoreType = "local_file" | "s3_segments" | "s3_archive";
 export function isRunLogStoreType(value: unknown): value is RunLogStoreType {
-  return value === "local_file" || value === "s3_segments";
+  return value === "local_file" || value === "s3_segments" || value === "s3_archive";
 }
 
 export interface RunLogHandle {
@@ -571,8 +571,15 @@ export function getRunLogStore() {
     },
   }) : null;
   const writer = external ?? local;
+  function writable(handle: RunLogHandle) {
+    if (handle.store === "s3_archive") throw new Error("Historical run log archives are read-only");
+    if (handle.store === "local_file") return local;
+    if (!external) throw new Error("Object run log storage is not configured");
+    return external;
+  }
   cachedStore = {
     begin: input => {
+      if (input.previousStore === "s3_archive") throw new Error("Historical run log archives are read-only");
       if (input.previousStore === "local_file") return local.begin(input);
       if (input.previousStore === "s3_segments") {
         if (!external) throw new Error("Object run log storage is not configured");
@@ -580,14 +587,15 @@ export function getRunLogStore() {
       }
       return writer.begin(input);
     },
-    append: (handle, event) => handle.store === "s3_segments" ? external!.append(handle, event) : local.append(handle, event),
-    enqueue: (handle, event) => handle.store === "s3_segments"
-      ? external!.enqueue!(handle, event)
-      : Promise.resolve({ committed: local.append(handle, event) }),
-    flush: async handle => { if (handle.store === "s3_segments") await external!.flush!(handle); },
-    finalize: (handle, options) => handle.store === "s3_segments" ? external!.finalize(handle, options) : local.finalize(handle),
+    append: (handle, event) => writable(handle).append(handle, event),
+    enqueue: (handle, event) => {
+      const store = writable(handle);
+      return store.enqueue?.(handle, event) ?? Promise.resolve({ committed: store.append(handle, event) });
+    },
+    flush: async handle => { await writable(handle).flush?.(handle); },
+    finalize: (handle, options) => writable(handle).finalize(handle, options),
     read: (handle, opts) => {
-      if (handle.store === "s3_segments") {
+      if (handle.store === "s3_segments" || handle.store === "s3_archive") {
         if (!external) throw new Error("Object run log storage is not configured");
         return external.read(handle, opts);
       }
@@ -613,7 +621,16 @@ export async function openLegacyRunLogForMigration(logRef: string) {
   try {
     const stat = await fs.stat(file);
     if (!stat.isFile()) throw new Error("Legacy run log is not a regular file");
-    return { stream: createReadStream(file), bytes: stat.size, source: "local" as const };
+    return {
+      stream: createReadStream(file), bytes: stat.size, source: "local" as const,
+      async assertUnchanged() {
+        const current = await fs.stat(file);
+        if (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) {
     if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
   }