feat(runner): add experimental Cursor Cloud provider (#18068)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Runner gives each provider durable execution, tool, and completion
contracts.
> - Cursor Cloud currently has a direct adapter outside that native
contract.
> - Cloud work needs exact ownership across HTTP, SSE, Stop, and
recovery.
> - This PR adds a qualified Cloud API driver and verified proxy to
Runner.
> - Product setup and hosted placement remain separate dependent
changes.

## Linked Issues or Issue Description

Refs #5664, #11803, #17299, and #17471. Product integration is #18069.
Managed-only API placement is #18310. The related direct-adapter
model/transcript work is not replaced. No duplicate native Cursor Cloud
integration was found.

## What Changed

- Add native `cursor_cloud` / `cursor_cloud_api` with explicit
model/repository validation, bounded HTTP/SSE, durable per-turn
checkpoints and a qualified Node proxy launched through the Rust facade.
- Commit terminal custody before publication/release; coalesce
consumers, preserve ambiguous create/Stop/recovery fences, reject late
tools, and distinguish definitely-unsent oversized requests from
ambiguous responses.
- Support scoped cold attachment, current authority, reconnect and
verified artifact staging. Keep unsupported warm/external and Windows
paths closed.
- Preserve early source cursors and verify immutable legacy prefixes
before suffix replay, including supported compaction/masking. Optional
proof publication failure falls back to complete cold verification.
- Preserve canonical records at the durable native producer with
JSON-safe privacy and the existing 8MiB bound; drain admitted successor
flushes. Reconcile upstream semantic/pack/schema/lifecycle changes and
retain S3 and legacy CLI checks.

## Verification

Core head `37c442ec8dd4`: all 66 check contexts are terminal
successful/neutral/skipped, including required `ci / verify` and `ci /
e2e`; fresh exact-head Greptile is 5/5 with zero unresolved threads.
Dependent Product `81e828d20a77` and Top `dd2959689627` include the
later client handoff correction and are running their own fresh gates.

Core: 47/47 Rust, 4/4 compiled-native retirement, 19/19 settlement and
normal Runner build passed. Runtime Top `9af1a5de42a2`: build and
recursive typecheck passed. Earlier source-scoped evidence includes 398
focused tests, five relay tests and token/type gates. The 251-assertion,
7/8-suite run failed physical PostgreSQL teardown; the unchanged stable
WSP suite passed 128 through the documented wrapper.

The downstream fixture corrections passed 7/7 browser and 121/121
startup tests without timeout/oracle relaxation; a broken-unsubscribe
control fails. Hosted runtime `9af1a5de42a2` has a passing normal build
and recursive typecheck. Latest dependent heads additionally contain a
pure client fix for stale live/history overlap, with 48 focused tests,
UI typecheck and token gates passing. Its new client is not deployed in
the frozen hosted image; server/native/API/provider source is unchanged
by these four downstream files.

Fresh hosted Linux admission on `9af1a5de42a2` retired its owned
runner/proxy/socket with zero prompts. The saved configuration survived
reopening, and Test authenticated the bound key/catalog/selected model
without a prompt. Artifact with `claude-sonnet-4-5` completed,
downloaded the exact 27 bytes, matched 121 canonical identities against
complete logs and independently reached FINISHED. The active Stop retry
was clicked within its immutable boundary, produced one canonical
interrupted turn and acknowledged run-only cancellation, retained a
valid 35.076s quiet window (maximum gap 4.584s), and independently
remained CANCELLED/artifact-absent beyond the original 180-second
command deadline. All provider-observer credential callbacks, DB clients
and loaders closed; accounting is settled and explicitly unpriced. The
first fresh Stop retains its failed 10.373s sampling gap and every
sample.

On the upgraded9af server, the historical F226 artifact was reopened and
downloaded with its exact original hash after its provider agent had
been deleted/GET404 earlier today. Historical F226 full Artifact/Stop
cleanup, Question, corrected Plan and reconnect retain their original
identities. Deleting the three newly created disposable provider agents
and the new-case post-delete grade remain blocked on explicit user
approval after automatic review rejected expanding the two original
approved IDs. No new-case cleanup or complete fresh matrix pass is
claimed; this is separate from the directly observed task/Stop behavior
and retained exact cleanup proof.

Retained failures: stale fixture 132/133; E8 48/135 with 87 reset-hook
failures; Mac 17 SQL failures/skips; earlier startup-title, title-save,
custody/Stop and deletion failures with unknown causes and bounded
recoveries. `dea023` Artifact passed 17 checks but Stop failed its false
warning and 50 duplicate log identities. Pre-repair ACK failure and
original Core CI failure remain recorded; retry alone establishes no
cause or full local-suite pass. Provider dollar cost stays UNKNOWN.

## Risks

Experimental: Cursor independently acknowledges neither the effective
model nor instruction digest before the first prompt. The caller’s
explicit values are submitted and behavior checked live. Unpriced
provider dollar cost stays UNKNOWN.

Only qualified per-turn controllers are admitted. Uncertain work and
incomplete cleanup stay fenced. Existing damaged/duplicate archives
remain failed evidence. Prefix proof adds bounded object-store I/O; the
8MiB producer bound does not guarantee future providers’ output fits.

## Model Used

OpenAI Codex, GPT-6 family, assisted with reasoning, repository tools,
code execution, browser verification, and delegated code review. The
serving model’s exact API ID and context window are not exposed. No
value is inferred from the model catalog.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used and the limits of available model
metadata
- [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 linked public issues or described the issue in this PR
- [x] I have not referenced internal or instance-local Paperclip issues
or links
- [x] My branch describes the change and contains no internal ticket ID
- [x] I have run the scoped local tests reported above; pending gates
are stated separately
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation
- [x] I have considered and documented the risks
- [x] All required CI gates pass on the current head
- [x] Fresh Greptile is 5/5 with no unresolved findings on the current
head
- [x] I will address all reviewer comments before requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/__tests__/managed-skill-content.test.ts

@@ -177,13 +177,14 @@ describe.skipIf(!support.supported)("managed skills use committed versions", ()
     const local = await skillInstallationMaintenance(db).verifyVersions({ companyId, limit: 50 });
     expect(local.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "local_storage" })]);
     const key = "cloud-runtime-config:v1";
-    const run = (phase: string) => promisify(execFile)(process.execPath, [
+    const run = (phase: string, storageProvider?: "s3" | "local_disk") => promisify(execFile)(process.execPath, [
       "--import", fileURLToPath(import.meta.resolve("tsx")),
       fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)),
       `--phase=${phase}`, `--company=${companyId}`, "--limit=50",
     ], { env: { ...process.env, DATABASE_URL: database.connectionString,
       PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture",
-      PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects"), ...env },
+      PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects"), ...env,
+      ...(storageProvider === undefined ? {} : { PAPERCLIP_STORAGE_PROVIDER: storageProvider }) },
       timeout: 60_000, maxBuffer: 1024 * 1024 });
     await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration");
     expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0);
@@ -193,6 +194,10 @@ describe.skipIf(!support.supported)("managed skills use committed versions", ()
       expect(eligibility.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "version_backed" })]);
       const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
       expect(verification.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: expectedStatus })]);
+      if (env.PAPERCLIP_STORAGE_PROVIDER === "s3") {
+        const legacyVerification = JSON.parse((await run("verification", "local_disk")).stdout.trim().split("\n").at(-1)!);
+        expect(legacyVerification.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "local_storage" })]);
+      }
       expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toEqual([persisted]);
     } finally {
       await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key));

server/src/services/heartbeat/run-log.test.ts

@@ -1,8 +1,10 @@
 import { describe, expect, it } from "vitest";
+import { createAgentIdentityRedactor } from "../agent-identity-redaction.js";
 import {
   appendExcerpt,
   boundHeartbeatRunEventPayloadForStorage,
   compactRunLogChunk,
+  redactCanonicalNativeRunLogChunk,
 } from "./run-log.js";
 
 describe("heartbeat run-log excerpts", () => {
@@ -173,3 +175,40 @@ describe("heartbeat run-log chunk formatting", () => {
     expect(compacted).not.toContain("[paperclip truncated");
   });
 });
+
+describe("complete canonical native log privacy", () => {
+  it.each([false, true])("keeps structured bytes and applies complete-value secrets, images, and username masking: %s", enabled => {
+    const identity = createAgentIdentityRedactor("fixture-configured-private-canary");
+    const envelope = { type: "paperclip.prp.event", event: {
+      schema: "paperclip.prp.event.v1", schemaVersion: 1, eventType: "item.delta",
+      payload: { header: 'Authorization: Bearer fixture-bearer-canary', api_key: "fixture-api-canary",
+        text: 'quoted "fixture-configured-private-canary" by fixture-operator',
+        image: { source: { data: "A".repeat(70000), type: "base64" }, type: "image" },
+        parts: Array.from({ length: 20 }, () => "full content ".repeat(400)),
+      },
+    } };
+    const chunk = redactCanonicalNativeRunLogChunk(JSON.stringify(envelope), {
+      redactIdentity: identity.redact, currentUser: { enabled, userNames: ["fixture-operator"], homeDirs: [] },
+    });
+    const result = JSON.parse(chunk).event;
+    expect(result.schema).toBe(envelope.event.schema);
+    expect(result.payload.header).toBe("Authorization: ***REDACTED***");
+    expect(result.payload.api_key).toBe("***REDACTED***");
+    expect(result.payload.text).toContain('quoted "***REDACTED***" by ');
+    expect(result.payload.text.includes("fixture-operator")).toBe(!enabled);
+    expect(result.payload.parts).toEqual(envelope.event.payload.parts);
+    expect(result.payload.image.source.data).toBe("[omitted base64 image data: 70000 chars]");
+    for (const secret of ["fixture-configured-private-canary", "fixture-bearer-canary", "fixture-api-canary"])
+      expect(chunk).not.toContain(secret);
+    expect(chunk).not.toContain("data:image");
+    expect(chunk.length).toBeGreaterThan(64 * 1024);
+    expect(chunk).not.toContain("[paperclip truncated");
+  });
+
+  it("rejects malformed owned envelopes instead of treating them as coverage-preserving diagnostics", () => {
+    const options = { redactIdentity: <T>(value: T) => value, currentUser: { enabled: false } };
+    for (const chunk of ['{"type":"paperclip.prp.event",', '{"type":"diagnostic","event":{}}',
+      '{"type":"paperclip.prp.event","event":[]}'])
+      expect(() => redactCanonicalNativeRunLogChunk(chunk, options)).toThrow();
+  });
+});

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

@@ -1,5 +1,6 @@
 import { appendWithByteCap, MAX_EXCERPT_BYTES, parseObject } from "../../adapters/utils.js";
-import { redactSensitiveText } from "../../redaction.js";
+import { redactSensitiveText, sanitizeRecord } from "../../redaction.js";
+import { redactCurrentUserText, redactCurrentUserValue, type CurrentUserRedactionOptions } from "../../log-redaction.js";
 
 const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024;
 const MAX_RUN_EVENT_PAYLOAD_STRING_CHARS = 16 * 1024;
@@ -118,3 +119,53 @@ export function compactRunLogChunk(
   return `${normalized.slice(0, headChars)}${marker}${normalized.slice(normalized.length - tailChars)}`;
 }
 
+
+interface CanonicalNativeLogRedaction {
+  redactIdentity: <T>(value: T) => T;
+  currentUser: CurrentUserRedactionOptions;
+}
+
+function omitCanonicalImageData(value: unknown): unknown {
+  if (Array.isArray(value)) return value.map(omitCanonicalImageData);
+  if (!value || typeof value !== "object") return value;
+  const object = value as Record<string, unknown>;
+  const result = Object.fromEntries(Object.entries(object).map(([key, child]) =>
+    [key, omitCanonicalImageData(child)]));
+  const source = object.source as Record<string, unknown> | undefined;
+  if (object.type === "image" && source?.type === "base64" &&
+    typeof source.data === "string" && /^[A-Za-z0-9+/=]{1024,}$/.test(source.data)) {
+    result.source = { ...(result.source as Record<string, unknown>),
+      data: `[omitted base64 image data: ${source.data.length} chars]` };
+  }
+  return result;
+}
+
+/** Only complete envelopes supplied by the durable native exporter use this path.
+ * Redact complete JSON values before encoding so quoted secrets cannot damage
+ * the record's structure. Image omission uses the existing field/length policy.
+ * The object writer separately bounds the complete encoded outer record.
+ */
+export function redactCanonicalNativeRunLogChunk(chunk: string, options: CanonicalNativeLogRedaction) {
+  const envelope = JSON.parse(chunk);
+  if (envelope?.type !== "paperclip.prp.event" || !envelope.event ||
+    typeof envelope.event !== "object" || Array.isArray(envelope.event))
+    throw new Error("invalid_canonical_native_run_log_envelope");
+  const projected = redactCurrentUserValue(
+    options.redactIdentity(omitCanonicalImageData(envelope)), options.currentUser,
+  );
+  return `${JSON.stringify(sanitizeRecord(projected as Record<string, unknown>))}\n`;
+}
+
+/** New complete-value forms first; old text projections are computed only when
+ * an archive needs them. An old malformed projection never repairs evidence.
+ */
+export function* canonicalNativeRunLogRedactionForms(chunk: string, options: CanonicalNativeLogRedaction) {
+  for (const enabled of [false, true])
+    yield redactCanonicalNativeRunLogChunk(chunk, {
+      ...options, currentUser: { ...options.currentUser, enabled },
+    });
+  for (const enabled of [false, true])
+    yield compactRunLogChunk(redactCurrentUserText(options.redactIdentity(chunk), {
+      ...options.currentUser, enabled,
+    }), Number.MAX_SAFE_INTEGER);
+}

server/src/services/heartbeat/run-output.test.ts

@@ -156,11 +156,11 @@ describe("heartbeat run output log recovery", () => {
 });
 
 describe("object-backed diagnostic recording", () => {
-  it("applies identity redaction to a native key-bearing event before S3 admission", async () => {
+  it.each(["legacy", "native"])("applies identity redaction before S3 admission when output begins in %s mode", async (runtimeMode) => {
     const privateKeyPem = generateKeyPairSync("ed25519")
       .privateKey.export({ type: "pkcs8", format: "pem" })
       .toString();
-    const row = { ...newRunRow, runtimeMode: "native" } as Run;
+    const row = { ...newRunRow, runtimeMode } as Run;
     const db = createRunRowDb(row) as Db & { select: () => unknown };
     db.select = (() => ({
       from: () => ({ where: () => ({ orderBy: () => ({ limit: async () => [] }) }) }),

server/src/services/heartbeat/run-output.ts

@@ -7,8 +7,8 @@ import type { createAgentIdentityRedactor } from "../agent-identity-redaction.js
 import { touchHeartbeatRunRuntimeStatus } from "../heartbeat-run-runtime-status.js";
 import { publishLiveEvent } from "../live-events.js";
 import { isRunLogStoreType, type RunLogHandle, type RunLogStore } from "../run-log-store.js";
-import { appendExcerpt, compactRunLogChunk } from "./run-log.js";
-import { createNativeRunLogExporter } from "../native-runtime/native-run-log-export.js";
+import { appendExcerpt, canonicalNativeRunLogRedactionForms, compactRunLogChunk, redactCanonicalNativeRunLogChunk } from "./run-log.js";
+import { createNativeRunLogExporter, nativeRunLogPrefixCursor } from "../native-runtime/native-run-log-export.js";
 import { createRunSecretRedactionRegistry } from "../run-secret-redaction.js";
 import {
   isHeartbeatRunRuntimeStatusActive,
@@ -21,12 +21,13 @@ export const ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS = 60 * 1000;
 export const ACTIVE_RUN_LOG_RUNTIME_STATUS_REFRESH_INTERVAL_MS = 5 * 1000;
 
 type Run = typeof heartbeatRuns.$inferSelect;
+type CanonicalNativeLog = { kind: "canonical_native"; sourceCursor: number };
 
 export interface HeartbeatRunOutputInput {
   run: Run;
   runId: string;
   issueId: string | null;
-  runLogStore: Pick<RunLogStore, "begin" | "append" | "enqueue" | "flush">;
+  runLogStore: Pick<RunLogStore, "begin" | "append" | "enqueue" | "flush"> & Partial<Pick<RunLogStore, "read" | "recordVerifiedSourcePrefix">>;
   identityRedactor: Pick<ReturnType<typeof createAgentIdentityRedactor>, "chunk" | "redact">;
   secretKeys: Set<string>;
   appendRunEvent: ReturnType<typeof createHeartbeatLifecycle>["appendRunEvent"];
@@ -145,15 +146,6 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
       .where(eq(heartbeatRuns.id, runId));
 
     const currentUserRedactionOptions = await getCurrentUserRedactionOptions();
-    if (
-      handle.store === "s3_segments" &&
-      run.runtimeMode === "native" &&
-      input.runtimeLogSecretValues?.length
-    ) {
-      const registry = createRunSecretRedactionRegistry(db);
-      for (const value of new Set(input.runtimeLogSecretValues))
-        await registry.register(run.companyId, runId, value);
-    }
     const reportRecordingFailure = async () => {
       if (recordingFailureReported) return;
       recordingFailureReported = true;
@@ -169,11 +161,15 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
       stream: "stdout" | "stderr",
       chunk: string,
       waitForCommit = true,
-      sourceCursor?: number,
+      source?: CanonicalNativeLog,
     ) => {
-      const sanitizedChunk = compactRunLogChunk(
-        redactCurrentUserText(chunk, currentUserRedactionOptions),
-      );
+      const sourceCursor = source?.sourceCursor;
+      const sanitizedChunk = source?.kind === "canonical_native"
+        ? redactCanonicalNativeRunLogChunk(chunk, {
+            redactIdentity: identityRedactor.redact,
+            currentUser: currentUserRedactionOptions,
+          })
+        : compactRunLogChunk(redactCurrentUserText(chunk, currentUserRedactionOptions));
       if (stream === "stdout") stdoutExcerpt = appendExcerpt(stdoutExcerpt, sanitizedChunk);
       if (stream === "stderr") stderrExcerpt = appendExcerpt(stderrExcerpt, sanitizedChunk);
       const ts = new Date().toISOString();
@@ -182,12 +178,15 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
       const chunkSeq = outputSeq;
       let committed: Promise<number> = Promise.resolve(0);
       if (handle) {
+        let nativeLooking = chunk.includes('"paperclip.prp.event"');
+        try { nativeLooking ||= JSON.parse(chunk)?.type === "paperclip.prp.event"; } catch { /* Diagnostics need not be JSON. */ }
         const event = {
           stream,
           chunk: sanitizedChunk,
           ts,
           seq: chunkSeq,
           sourceCursor,
+          preservesSourceCoverage: source?.kind === "canonical_native" || !nativeLooking,
         };
         committed = runLogStore.enqueue
           ? (await runLogStore.enqueue(handle, event)).committed
@@ -259,27 +258,44 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
       );
       return external ? recorded.catch(reportRecordingFailure) : recorded;
     };
-    const nativeExporter =
-      handle.store === "s3_segments" && run.runtimeMode === "native"
-        ? createNativeRunLogExporter({
-            db,
-            companyId: run.companyId,
-            runId,
-            afterCursor: handle.initialSourceCursor ?? 0,
-            enqueue: (cursor, chunk) =>
-              appendIdentityRedactedLog(
-                "stdout",
-                identityRedactor.chunk("stdout", chunk),
-                false,
-                cursor,
-              ),
-            flush: () => runLogStore.flush!(handle!),
-          })
-        : undefined;
-    if (nativeExporter) await nativeExporter.recover();
-    const onPersistedNativeLog = nativeExporter
+    let nativeExporter: Promise<ReturnType<typeof createNativeRunLogExporter>> | undefined;
+    const getNativeExporter = () => nativeExporter ??= (async () => {
+      if (input.runtimeLogSecretValues?.length) {
+        const registry = createRunSecretRedactionRegistry(db);
+        for (const value of new Set(input.runtimeLogSecretValues))
+          await registry.register(run.companyId, runId, value);
+      }
+      const afterCursor = await nativeRunLogPrefixCursor({
+        db, companyId: run.companyId, runId, store: runLogStore, handle: handle!,
+        // Verify the two supported historical username-masking settings. Keep
+        // the full private-data projection here; compaction is proved separately.
+        redactChunks: (chunk) => canonicalNativeRunLogRedactionForms(chunk, {
+          redactIdentity: identityRedactor.redact, currentUser: currentUserRedactionOptions,
+        }),
+      });
+      return createNativeRunLogExporter({
+        db,
+        companyId: run.companyId,
+        runId,
+        afterCursor,
+        enqueue: (cursor, chunk) =>
+          appendIdentityRedactedLog(
+            "stdout",
+            chunk,
+            false,
+            { kind: "canonical_native", sourceCursor: cursor },
+          ),
+        flush: () => runLogStore.flush!(handle!),
+      });
+    })();
+    if (handle.store === "s3_segments" && run.runtimeMode === "native")
+      await (await getNativeExporter()).recover();
+    // Output begins before runtime selection persists native mode. Keep the
+    // source-aware callback available without querying native sources for
+    // ordinary legacy output.
+    const onPersistedNativeLog = handle.store === "s3_segments"
       ? (cursor: number, chunk: string) =>
-          nativeExporter.record(cursor, chunk).catch(reportRecordingFailure)
+          getNativeExporter().then((exporter) => exporter.record(cursor, chunk)).catch(reportRecordingFailure)
       : undefined;
     const onAdapterMeta = async (meta: AdapterInvocationMeta) => {
       meta = identityRedactor.redact(meta);

server/src/services/native-runtime/native-run-log-export.ts

@@ -3,6 +3,9 @@ import { StringDecoder } from "node:string_decoder";
 import { heartbeatRunEvents, type Db } from "@paperclipai/db";
 import { validatePrpEvent, type PrpEvent } from "../../vendor/paperclip-runner/index.js";
 import { nativeSha256 } from "./canonical.js";
+import type { RunLogHandle, RunLogStore } from "../run-log-store.js";
+import { compactRunLogChunk } from "../heartbeat/run-log.js";
+import { MAX_RUN_LOG_RECORD_BYTES } from "../run-log-record-limits.js";
 
 export function formatNativeRunLogEvent(event: PrpEvent) {
   return `${JSON.stringify({ type: "paperclip.prp.event", event })}\n`;
@@ -77,35 +80,144 @@ async function sourcePage(
   });
 }
 
-export function createNativeRunLogCoverageVerifier(db: Db, companyId: string, runId: string) {
+function matchPublishedImageOmissions(source: unknown, published: unknown): unknown {
+  if (Array.isArray(source))
+    return source.map((value, index) => matchPublishedImageOmissions(value,
+      Array.isArray(published) ? published[index] : undefined));
+  if (!source || typeof source !== "object") return source;
+  const original = source as Record<string, unknown>;
+  const existing = published && typeof published === "object"
+    ? published as Record<string, unknown> : {};
+  const result = Object.fromEntries(Object.entries(original).map(([key, value]) =>
+    [key, matchPublishedImageOmissions(value, existing[key])]));
+  const imageSource = original.source as Record<string, unknown> | undefined;
+  const publishedSource = existing.source as Record<string, unknown> | undefined;
+  if (original.type === "image" && imageSource?.type === "base64" &&
+    typeof imageSource.data === "string" && typeof publishedSource?.data === "string") {
+    const marker = /^\[omitted base64 image data: ([0-9]+) chars\]$/.exec(publishedSource.data);
+    if (marker && marker[1] === String(imageSource.data.length) && imageSource.data.length >= 1024 &&
+      /^[A-Za-z0-9+/=]+$/.test(imageSource.data)) {
+      // The existing scrubber depends on serialized key order, which JSONB
+      // does not retain. Honor only an omission already present at this path;
+      // published raw data still has to match the source exactly.
+      result.source = { ...(result.source as Record<string, unknown>), data: publishedSource.data };
+    }
+  }
+  return result;
+}
+
+function matchesPublishedEvent(source: unknown, published: unknown) {
+  return nativeSha256(published) === nativeSha256(matchPublishedImageOmissions(source, published));
+}
+
+/** Recover a compactor omission contained within one source-backed JSON string.
+ * Retained properties may have any order: JSONB does not retain writer ordering.
+ * Reconstruct the complete envelope, prove the exact compactor output, and then
+ * compare every field semantically. A marker or matching substring alone is not proof.
+ */
+function restoreCompactedEvent(chunk: string, source: unknown): unknown {
+  const marker = /\n\[paperclip truncated run log chunk: omitted ([1-9][0-9]*) chars\]\n/g;
+  const found = marker.exec(chunk);
+  if (!found || marker.exec(chunk)) return;
+  const omitted = Number(found[1]);
+  if (!Number.isSafeInteger(omitted)) return;
+  const head = chunk.slice(0, found.index), tail = chunk.slice(found.index + found[0].length);
+  let quoted = false, escaped = false, start = -1;
+  for (let index = 0; index < head.length; index++) {
+    const character = head[index];
+    if (!quoted) {
+      if (character === '"') { quoted = true; start = index; }
+    } else if (escaped) escaped = false;
+    else if (character === "\\") escaped = true;
+    else if (character === '"') quoted = false;
+  }
+  if (!quoted) return;
+  const leading = head.slice(start);
+  function* strings(value: unknown): Generator<string> {
+    if (typeof value === "string") yield value;
+    else if (Array.isArray(value)) for (const child of value) yield* strings(child);
+    else if (value && typeof value === "object")
+      for (const child of Object.values(value)) yield* strings(child);
+  }
+  for (const value of strings(source)) {
+    const encoded = JSON.stringify(value), end = leading.length + omitted;
+    if (end >= encoded.length || !encoded.startsWith(leading) || !tail.startsWith(encoded.slice(end))) continue;
+    const restored = head + encoded.slice(leading.length, end) + tail;
+    if (compactRunLogChunk(restored) !== chunk) continue;
+    try {
+      const envelope = JSON.parse(restored);
+      if (envelope.type === "paperclip.prp.event" && matchesPublishedEvent(source, envelope.event))
+        return envelope;
+    } catch { /* Another candidate cannot make malformed evidence authoritative. */ }
+  }
+}
+
+export function createNativeRunLogCoverageVerifier(
+  db: Db,
+  companyId: string,
+  runId: string,
+  options?: {
+    allowSourceSuffix?: boolean;
+    redactChunks?: (chunk: string) => Iterable<string>;
+    signal?: AbortSignal;
+  },
+) {
   const decoder = new StringDecoder("utf8");
   let text = "",
     cursor = 0;
   let page: Awaited<ReturnType<typeof sourcePage>> = [];
   const unproven = () => new Error("native_log_source_coverage_unproven");
   async function next() {
-    if (!page.length) page = await sourcePage(db, companyId, runId, cursor);
+    if (!page.length) page = await sourcePage(db, companyId, runId, cursor, undefined, options?.signal);
     return page[0];
   }
-  async function line(value: string) {
+  async function line(value: string, newlineBytes = 1) {
     if (!value.trim()) return;
-    if (Buffer.byteLength(value) > 1024 * 1024) throw unproven();
+    if (Buffer.byteLength(value) + newlineBytes > MAX_RUN_LOG_RECORD_BYTES) throw unproven();
     let chunk: unknown;
     try {
       chunk = JSON.parse(value).chunk;
     } catch {
       throw unproven();
     }
-    if (typeof chunk !== "string" || !chunk.startsWith('{"type":"paperclip.prp.event"')) return;
+    if (typeof chunk !== "string") throw unproven();
+    let expected: Awaited<ReturnType<typeof next>>;
+    async function* sourceForms() {
+      expected ??= await next();
+      if (!expected) throw unproven();
+      if (!options?.redactChunks) { yield expected.event; return; }
+      for (const form of options.redactChunks(formatNativeRunLogEvent(expected.event))) {
+        let candidate;
+        try { candidate = JSON.parse(form); }
+        catch { continue; } // A malformed historical privacy projection proves nothing.
+        if (candidate?.type === "paperclip.prp.event" && candidate.event &&
+          typeof candidate.event === "object" && !Array.isArray(candidate.event))
+          yield candidate.event;
+      }
+    }
+    let envelope: { type?: unknown; event?: unknown };
+    try {
+      envelope = JSON.parse(chunk);
+    } catch {
+      if (!chunk.includes('"paperclip.prp.event"')) return;
+      let restored;
+      for await (const form of sourceForms()) {
+        restored = restoreCompactedEvent(chunk, form);
+        if (restored) break;
+      }
+      if (!restored) throw unproven();
+      envelope = restored;
+    }
+    if (envelope?.type !== "paperclip.prp.event") return;
     let event: PrpEvent;
     try {
-      const parsed = validatePrpEvent(JSON.parse(chunk).event);
+      const parsed = validatePrpEvent(envelope.event);
       if (!parsed.ok) throw unproven();
       event = parsed.event;
     } catch {
       throw unproven();
     }
-    const expected = await next();
+    expected ??= await next();
     if (
       !expected ||
       event.runId !== runId ||
@@ -117,6 +229,19 @@ export function createNativeRunLogCoverageVerifier(db: Db, companyId: string, ru
       event.eventType !== expected.event.eventType
     )
       throw unproven();
+    if (options?.redactChunks) {
+      // JSONB reorders object keys. Compare the complete redacted event
+      // semantically while retaining the original published bytes.
+      try {
+        let matched = false;
+        for await (const form of sourceForms()) {
+          if (matchesPublishedEvent(form, event)) { matched = true; break; }
+        }
+        if (!matched) throw unproven();
+      } catch {
+        throw unproven();
+      }
+    }
     cursor = expected.cursor;
     page.shift();
   }
@@ -128,17 +253,75 @@ export function createNativeRunLogCoverageVerifier(db: Db, companyId: string, ru
         await line(text.slice(0, boundary));
         text = text.slice(boundary + 1);
       }
-      if (Buffer.byteLength(text) > 1024 * 1024) throw unproven();
+      if (Buffer.byteLength(text) > MAX_RUN_LOG_RECORD_BYTES) throw unproven();
     },
     async finish() {
       text += decoder.end();
-      if (text) await line(text);
-      if (await next()) throw unproven();
+      if (text) await line(text, 0);
+      if (!options?.allowSourceSuffix && await next()) throw unproven();
       return cursor;
     },
   };
 }
 
+/** Validate committed provenance, or prove an older unwatermarked source prefix. */
+export async function nativeRunLogPrefixCursor(input: {
+  db: Db;
+  companyId: string;
+  runId: string;
+  store: Partial<Pick<RunLogStore, "read" | "recordVerifiedSourcePrefix">>;
+  handle: RunLogHandle;
+  redactChunks: (chunk: string) => Iterable<string>;
+  signal?: AbortSignal;
+}) {
+  const bytes = input.handle.initialBytes ?? 0;
+  const watermark = input.handle.initialSourceCursor ?? 0;
+  const unproven = () => new Error("native_log_source_coverage_unproven");
+  if (!Number.isSafeInteger(bytes) || bytes < 0 || !Number.isSafeInteger(watermark) || watermark < 0)
+    throw unproven();
+  if (!bytes && !watermark) {
+    await input.store.recordVerifiedSourcePrefix?.(input.handle, 0);
+    return 0;
+  }
+  if (input.handle.initialSourcePrefixVerified) {
+    const sequence = input.handle.initialSequence ?? 0;
+    const [company, , run, extra] = input.handle.logRef.split("/");
+    if (input.handle.store !== "s3_segments" || company !== input.companyId || run !== input.runId || extra ||
+      !bytes || !Number.isSafeInteger(sequence) || sequence < 1) throw unproven();
+    // A separate immutable proof binds this exact journal index/end/sequence/cursor.
+    // Legacy maximum-cursor metadata alone never establishes ordered coverage.
+    // Only explicitly coverage-preserving appends can carry the proof forward;
+    // an older writer's changed prefix gets no matching proof and must scan once.
+    if (watermark) {
+      const [source] = await sourcePage(input.db, input.companyId, input.runId,
+        watermark - 1, watermark, input.signal);
+      if (!source || source.cursor !== watermark) throw unproven();
+    }
+    return watermark;
+  }
+  if (!input.store.read) throw unproven();
+  const verifier = createNativeRunLogCoverageVerifier(input.db, input.companyId, input.runId, {
+    allowSourceSuffix: true,
+    redactChunks: input.redactChunks,
+    signal: input.signal,
+  });
+  let offset = 0;
+  while (offset < bytes) {
+    input.signal?.throwIfAborted();
+    const page = await input.store.read(input.handle, {
+      offset, limitBytes: Math.min(256_000, bytes - offset), signal: input.signal,
+    });
+    const content = Buffer.from(page.content);
+    if (!content.length || offset + content.length > bytes) throw unproven();
+    await verifier.write(content);
+    offset += content.length;
+  }
+  const cursor = await verifier.finish();
+  if (watermark > cursor) throw unproven();
+  await input.store.recordVerifiedSourcePrefix?.(input.handle, cursor);
+  return cursor;
+}
+
 export function createNativeRunLogExporter(input: {
   db: Db;
   companyId: string;
@@ -188,7 +371,7 @@ export function createNativeRunLogExporter(input: {
       if (!Number.isSafeInteger(sourceCursor) || sourceCursor < 1)
         throw new Error("invalid_native_run_log_cursor");
       const bytes = Buffer.byteLength(chunk);
-      if (queuedRecords >= 32 || queuedBytes + bytes > 8 * 1024 * 1024)
+      if (queuedRecords >= 32 || queuedBytes + bytes > MAX_RUN_LOG_RECORD_BYTES)
         throw new Error("native_run_log_admission_capacity_exceeded");
       queuedRecords++;
       queuedBytes += bytes;

server/src/services/native-runtime/native-run-log-recovery.ts

@@ -5,16 +5,16 @@ import {
   nativeRunFinalizations,
   type Db,
 } from "@paperclipai/db";
-import { redactCurrentUserText, type CurrentUserRedactionOptions } from "../../log-redaction.js";
+import type { CurrentUserRedactionOptions } from "../../log-redaction.js";
 import type { RunLogHandle, RunLogReadOptions, RunLogStore } from "../run-log-store.js";
 import { instanceSettingsService } from "../instance-settings.js";
-import { compactRunLogChunk } from "../heartbeat/run-log.js";
+import { canonicalNativeRunLogRedactionForms, redactCanonicalNativeRunLogChunk } from "../heartbeat/run-log.js";
 import { beginIdleTrackedWork } from "../task-admission.js";
 import { agentIdentityService } from "../agent-identity.js";
 import { createRunSecretRedactionRegistry } from "../run-secret-redaction.js";
 import { createAgentIdentityRedactor } from "../agent-identity-redaction.js";
 import { logger } from "../../middleware/logger.js";
-import { createNativeRunLogExporter, nativeRunLogSourceWhere } from "./native-run-log-export.js";
+import { createNativeRunLogExporter, nativeRunLogPrefixCursor, nativeRunLogSourceWhere } from "./native-run-log-export.js";
 
 interface Recovery {
   work: Promise<void>;
@@ -148,37 +148,44 @@ export async function recoverSettledNativeRunLog(
     let lastSourceCursor = handle.initialSourceCursor ?? 0;
     const receipts: Promise<number>[] = [];
     let failure: unknown;
-    const exporter = createNativeRunLogExporter({
-      db,
-      companyId,
-      runId,
-      afterCursor: handle.initialSourceCursor ?? 0,
-      signal,
-      enqueue: async (sourceCursor, chunk) => {
-        const receipt = await store.enqueue!(handle, {
-          stream: "stdout",
-          ts: new Date().toISOString(),
-          seq: ++sequence,
-          sourceCursor,
-          chunk: compactRunLogChunk(
-            redactCurrentUserText(identityRedactor.chunk("stdout", chunk), redaction),
-          ),
-        });
-        receipts.push(
-          receipt.committed.catch((error) => {
-            failure ??= error;
-            return 0;
-          }),
-        );
-        lastSourceCursor = sourceCursor;
-      },
-      flush: async () => {
-        await store.flush!(handle);
-        await Promise.all(receipts.splice(0));
-        if (failure) throw failure;
-      },
-    });
     try {
+      lastSourceCursor = await nativeRunLogPrefixCursor({
+        db, companyId, runId, store, handle, signal,
+        redactChunks: (chunk) => canonicalNativeRunLogRedactionForms(chunk, {
+          redactIdentity: identityRedactor.redact, currentUser: redaction,
+        }),
+      });
+      const exporter = createNativeRunLogExporter({
+        db,
+        companyId,
+        runId,
+        afterCursor: lastSourceCursor,
+        signal,
+        enqueue: async (sourceCursor, chunk) => {
+          const receipt = await store.enqueue!(handle, {
+            stream: "stdout",
+            ts: new Date().toISOString(),
+            seq: ++sequence,
+            sourceCursor,
+            preservesSourceCoverage: true,
+            chunk: redactCanonicalNativeRunLogChunk(chunk, {
+              redactIdentity: identityRedactor.redact, currentUser: redaction,
+            }),
+          });
+          receipts.push(
+            receipt.committed.catch((error) => {
+              failure ??= error;
+              return 0;
+            }),
+          );
+          lastSourceCursor = sourceCursor;
+        },
+        flush: async () => {
+          await store.flush!(handle);
+          await Promise.all(receipts.splice(0));
+          if (failure) throw failure;
+        },
+      });
       await exporter.recover();
     } catch (error) {
       failure ??= error;

server/src/services/native-runtime/native-run-log-watermark.test.ts

@@ -0,0 +1,610 @@
+import { createHash, randomUUID } from "node:crypto";
+import { Readable } from "node:stream";
+import { eq, sql } from "drizzle-orm";
+import { closeRegisteredClients, createDb, heartbeatRunEvents, heartbeatRuns, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, type Db } from "@paperclipai/db";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
+import { createNativeRunLogTestTables } from "../../__tests__/helpers/native-run-log-database.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import { canonicalProviderEventsFromCodex } from "@paperclipai/paperclip-runner";
+import type { RunLogHandle, RunLogStore } from "../run-log-store.js";
+import type { StorageProvider } from "../../storage/types.js";
+import { validatePrpEvent, type PrpEvent } from "../../vendor/paperclip-runner/index.js";
+import { instanceSettingsService } from "../instance-settings.js";
+import type { CurrentUserRedactionOptions } from "../../log-redaction.js";
+import { createHeartbeatRunOutput } from "../heartbeat/run-output.js";
+import { compactRunLogChunk } from "../heartbeat/run-log.js";
+import { createObjectRunLogStore } from "../object-run-log-store.js";
+import { nativeSha256 } from "./canonical.js";
+import { createNativeRunLogCoverageVerifier, formatNativeRunLogEvent } from "./native-run-log-export.js";
+import { readRecoverableRunLog, recoverSettledNativeRunLog } from "./native-run-log-recovery.js";
+
+// Use an explicit guarded test database or a new owned embedded fixture.
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_LOG_WATERMARK_DATABASE_URL");
+const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
+
+function memoryProvider(): StorageProvider {
+  const objects = new Map<string, Buffer>();
+  const etag = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex");
+  return {
+    id: "s3",
+    async putObject() { throw new Error("unconditional object write"); },
+    async putObjectConditional(input) {
+      const old = objects.get(input.objectKey);
+      if (input.ifNoneMatch ? old : !old || etag(old) !== input.ifMatch)
+        throw Object.assign(new Error("conflict"), { name: "PreconditionFailed" });
+      const chunks: Buffer[] = [];
+      for await (const chunk of Readable.from(input.body)) chunks.push(Buffer.from(chunk));
+      const bytes = Buffer.concat(chunks);
+      objects.set(input.objectKey, bytes);
+      return { etag: etag(bytes) };
+    },
+    async getObject(input) {
+      const bytes = objects.get(input.objectKey);
+      if (!bytes) throw Object.assign(new Error("missing"), { name: "NoSuchKey" });
+      return { stream: Readable.from([bytes]), etag: etag(bytes) };
+    },
+    async headObject(input) { return { exists: objects.has(input.objectKey) }; },
+    async deleteObject(input) { objects.delete(input.objectKey); },
+    async listObjects(input) {
+      return { keys: [...objects.keys()].filter((key) => key.startsWith(input.prefix)) };
+    },
+  };
+}
+
+describe.skipIf(!support.supported)("native run log source watermarks", () => {
+  const schema = `native_watermark_${randomUUID().replaceAll("-", "")}`;
+  let db: Db;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  beforeAll(async () => {
+    database = externalUrl ? { connectionString: externalUrl, cleanup: async () => {} }
+      : await startEmbeddedPostgresTestDatabase("native-log-watermark-");
+    db = createDb(database.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0 });
+    await db.execute(sql.raw(`create schema ${schema}`));
+    await db.execute(sql.raw(`set search_path=${schema}`));
+    await createNativeRunLogTestTables(db);
+    await db.execute(sql`create table instance_settings(id uuid primary key default gen_random_uuid(),
+      singleton_key text unique default 'default', default_environment_id uuid,
+      general jsonb not null default '{}', experimental jsonb not null default '{}',
+      created_at timestamptz default now(), updated_at timestamptz default now())`);
+    // Match the production queued-run default instead of the older native fixture default.
+    await db.execute(sql`alter table heartbeat_runs alter column runtime_mode set default 'legacy'`);
+  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+  afterAll(async () => {
+    try {
+      if (db) await db.execute(sql.raw(`drop schema ${schema} cascade`));
+    } finally {
+      if (database) {
+        await closeRegisteredClients(database.connectionString, { scope: "database" });
+        await database.cleanup();
+      }
+    }
+  });
+
+  async function fixture() {
+    const binding = { companyId: randomUUID(), agentId: randomUUID(), runId: randomUUID() };
+    await db.execute(sql`insert into heartbeat_runs(id,company_id,agent_id,status)
+      values(${binding.runId},${binding.companyId},${binding.agentId},'running')`);
+    const provider = memoryProvider();
+    const store = () => createObjectRunLogStore({
+      provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async (input) => { await input.assertWriterAuthority?.(); },
+    });
+    const event = (seq: number): PrpEvent => ({
+      schema: "paperclip.prp.event.v1", schemaVersion: 1,
+      runId: binding.runId, sourceInstanceId: "fixture-source", sourceEventId: `source:${seq}`,
+      sourceSeq: seq, sourceKind: "runner", normalizedSessionId: "fixture-session",
+      eventType: seq === 2 ? "turn.interrupted" : "item.delta", priority: 2,
+      emittedAt: "2026-10-31T00:00:00.000Z", payload: { text: `step ${seq} 再開` },
+    });
+    const persist = async (value: PrpEvent, cursor: number) => {
+      expect(validatePrpEvent(value).ok).toBe(true);
+      await db.execute(sql`insert into heartbeat_run_events(company_id,run_id,agent_id,seq,event_type,
+        payload,source_instance_id,source_event_id,source_seq,source_payload_sha256,protocol_schema_version)
+        values(${binding.companyId},${binding.runId},${binding.agentId},${cursor},${value.eventType},
+        ${JSON.stringify({ prpEvent: value })}::jsonb,${value.sourceInstanceId},${value.sourceEventId},
+        ${value.sourceSeq},${nativeSha256(value)},${value.schemaVersion})`);
+    };
+    const settle = async (logRef: string) => {
+      await db.update(heartbeatRuns).set({ runtimeMode: "native", status: "cancelled",
+        logStore: "s3_segments", logRef, logBytes: null }).where(eq(heartbeatRuns.id, binding.runId));
+    };
+    return { binding, store, event, persist, settle, provider };
+  }
+
+  function toolEvent(f: Awaited<ReturnType<typeof fixture>>, output: string): PrpEvent {
+    return { ...f.event(1), itemId: "tool-1", priority: 1, eventType: "tool.execution.completed", payload: {
+      schema: "paperclip.tool.execution.v1", executionId: "tool-1", transport: "process",
+      operation: "execute", status: "completed", output,
+      outputBytes: Buffer.byteLength(output), outputTruncated: false, outputDigest: null, exitCode: 0,
+    } };
+  }
+
+  async function readAll(store: Pick<RunLogStore, "read">, handle: RunLogHandle) {
+    let offset = 0, content = "";
+    for (let page = 0; page < 16; page++) {
+      const result = await store.read(handle, { offset, limitBytes: 4 * 1024 * 1024 });
+      content += result.content;
+      if (result.nextOffset === undefined) return content;
+      expect(result.nextOffset).toBeGreaterThan(offset);
+      offset = result.nextOffset;
+    }
+    throw new Error("fixture log exceeded bounded read pages");
+  }
+
+  function planEvent(f: Awaited<ReturnType<typeof fixture>>, count: number, text: string, seq = 1): PrpEvent {
+    const [mapped] = canonicalProviderEventsFromCodex("turn/plan/updated", { turnId: "fixture-plan-turn",
+      revision: seq, plan: Array.from({ length: count }, () => ({ step: text, status: "pending" })) });
+    expect(mapped?.eventType).toBe("plan.updated");
+    return { ...f.event(seq), ...mapped! };
+  }
+
+  function failPublishedPrefixProof(f: Awaited<ReturnType<typeof fixture>>) {
+    const put = f.provider.putObjectConditional!.bind(f.provider);
+    let initialProof: string | undefined, enabled = true, failures = 0;
+    f.provider.putObjectConditional = async input => {
+      if (input.objectKey.includes("/objects/source-prefix-proof-v1-")) {
+        initialProof ??= input.objectKey;
+        if (enabled && input.objectKey !== initialProof) {
+          failures++;
+          throw Object.assign(new Error("proof storage unavailable"), { $metadata: { httpStatusCode: 503 } });
+        }
+      }
+      return put(input);
+    };
+    return { failures: () => failures, restore: () => { enabled = false; } };
+  }
+
+  async function publishProductionPrefix(
+    f: Awaited<ReturnType<typeof fixture>>, event: PrpEvent,
+    options: { watermark?: boolean; redaction?: () => Promise<CurrentUserRedactionOptions>; diagnostic?: string } = {},
+  ) {
+    const [saved] = await db.select({ runtimeMode: heartbeatRuns.runtimeMode }).from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, f.binding.runId));
+    const row = { id: f.binding.runId, companyId: f.binding.companyId, agentId: f.binding.agentId,
+      status: "running", runtimeMode: saved!.runtimeMode, logStore: null, logRef: null,
+      logBytes: null, lastOutputSeq: 0, lastOutputAt: null } as typeof heartbeatRuns.$inferSelect;
+    const store = f.store();
+    const output = createHeartbeatRunOutput(db, {
+      run: row!, runId: row!.id, issueId: null, runLogStore: store,
+      identityRedactor: { chunk: (_stream, chunk) => chunk, redact: value => value },
+      secretKeys: new Set(), appendRunEvent: async () => undefined,
+      getCurrentUserRedactionOptions: options.redaction ?? (async () => ({ enabled: false })),
+    });
+    const handlers = await output.begin(row!);
+    await db.update(heartbeatRuns).set({ runtimeMode: "native" }).where(eq(heartbeatRuns.id, row!.id));
+    await f.persist(event, 2);
+    // Historical queued-run writers used onLog before source cursors were retained.
+    if (options.watermark === false) await handlers.onLog("stdout", formatNativeRunLogEvent(event));
+    else await handlers.onPersistedNativeLog!(2, formatNativeRunLogEvent(event));
+    if (options.diagnostic) await handlers.onLog("stderr", options.diagnostic);
+    await store.finalize(output.handle!);
+    return { store, handle: output.handle!, prefix: await readAll(store, output.handle!) };
+  }
+
+  it.each([true, false])("cold-recovers an explicit legacy-compacted 64KiB tool event with committed cursor: %s", async (watermark) => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const event = toolEvent(f, "retained output 再開 ".repeat(4000).slice(0, 65_536));
+    await f.persist(event, 2);
+    // Immutable old producer format, even when the current canonical writer becomes lossless.
+    await store.append(handle, { stream: "stdout", ts: "legacy", seq: 1,
+      ...(watermark ? { sourceCursor: 2 } : {}), chunk: compactRunLogChunk(formatNativeRunLogEvent(event)) });
+    await store.finalize(handle);
+    const prefix = await readAll(store, handle);
+    const chunk = JSON.parse(prefix).chunk as string;
+    expect(chunk).toContain("[paperclip truncated run log chunk: omitted ");
+    expect(() => JSON.parse(chunk)).toThrow();
+    const [durable] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents)
+      .where(eq(heartbeatRunEvents.runId, f.binding.runId));
+    expect(JSON.stringify(durable!.payload!.prpEvent)).not.toBe(JSON.stringify(event));
+    await f.persist(f.event(2), 5);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    const content = (await cold.read(handle)).content;
+    expect(content.startsWith(prefix)).toBe(true);
+    const lines = content.trim().split("\n").map(line => JSON.parse(line));
+    expect(lines).toHaveLength(2);
+    expect(lines[0].chunk).toBe(chunk);
+    expect(JSON.parse(lines[1].chunk).event.sourceEventId).toBe("source:2");
+    const [run] = await db.select({ bytes: heartbeatRuns.logBytes }).from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, f.binding.runId));
+    expect(run!.bytes).toBe(Buffer.byteLength(content));
+  });
+
+  it.each([[20, "A"], [256, "界"]] as const)("retains a current %i-step %s plan through proof outage and cold recovery", async (count, character) => {
+    const f = await fixture(), event = planEvent(f, count, character.repeat(4000));
+    const outage = failPublishedPrefixProof(f);
+    const { store, handle, prefix } = await publishProductionPrefix(f, event, { diagnostic: "diagnostic ".repeat(10_000) });
+    expect(outage.failures()).toBeGreaterThan(0);
+    outage.restore();
+    const snapshot = await store.begin(f.binding);
+    expect(snapshot.initialSourceCursor).toBe(2);
+    expect(snapshot.initialSourcePrefixVerified).toBe(false);
+    await store.finalize(snapshot);
+    const suffix = planEvent(f, count, character.repeat(4000), 2);
+    await f.persist(suffix, 5);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    const content = await readAll(cold, handle);
+    expect(content.startsWith(prefix)).toBe(true);
+    const entries = content.trim().split("\n").map(line => JSON.parse(line));
+    expect(entries).toHaveLength(3);
+    expect(nativeSha256(JSON.parse(entries[0].chunk).event)).toBe(nativeSha256(event));
+    expect(nativeSha256(JSON.parse(entries[2].chunk).event)).toBe(nativeSha256(suffix));
+    expect(entries[0].chunk).not.toContain("[paperclip truncated");
+    expect(entries[2].chunk).not.toContain("[paperclip truncated");
+    expect(entries[1].chunk).toContain("[paperclip truncated run log chunk:");
+    expect(entries[1].chunk.length).toBeLessThan(64 * 1024);
+    if (count === 256) expect(Buffer.byteLength(entries[0].chunk)).toBeGreaterThan(1024 * 1024);
+    expect(entries.map(entry => entry.seq)).toEqual([1, 2, 3]);
+    const resumed = await cold.begin(f.binding);
+    expect(resumed.initialSourceCursor).toBe(5);
+    expect(resumed.initialSourcePrefixVerified).toBe(true);
+    await cold.finalize(resumed);
+  });
+
+  it("scrubs private canaries and image data before lossless canonical publication and recovery", async () => {
+    const f = await fixture(), canary = "fixture-private-bearer-canary";
+    const event = { ...f.event(1), payload: { header: `Authorization: Bearer ${canary}`, api_key: canary,
+      image: { type: "image", source: { type: "base64", data: "A".repeat(70000) } },
+      parts: Array.from({ length: 20 }, () => "preserved ".repeat(400)),
+    } };
+    const outage = failPublishedPrefixProof(f);
+    const { handle, prefix } = await publishProductionPrefix(f, event);
+    expect(outage.failures()).toBeGreaterThan(0);
+    outage.restore();
+    expect(prefix).not.toContain(canary);
+    expect(prefix).toContain("[omitted base64 image data: 70000 chars]");
+    const suffix: PrpEvent = { ...f.event(2), eventType: "item.delta", payload: event.payload };
+    await f.persist(suffix, 5);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    const content = await readAll(cold, handle);
+    expect(content.startsWith(prefix)).toBe(true);
+    expect(content).not.toContain(canary);
+    const entries = content.trim().split("\n").map(line => JSON.parse(JSON.parse(line).chunk).event);
+    expect(entries.map(event => event.sourceEventId)).toEqual(["source:1", "source:2"]);
+    for (const entry of entries) {
+      expect(entry.payload.parts).toEqual(event.payload.parts);
+      expect(entry.payload.image.source.data).toBe("[omitted base64 image data: 70000 chars]");
+      expect(entry.payload.api_key).toBe("***REDACTED***");
+      expect(entry.payload.header).toBe("Authorization: ***REDACTED***");
+    }
+  });
+
+  it("leaves old unprovable multi-string compaction failed and byte-identical", async () => {
+    const f = await fixture(), event = planEvent(f, 20, "A".repeat(4000));
+    await f.persist(event, 2);
+    await f.persist(f.event(2), 5);
+    const store = f.store(), handle = await store.begin(f.binding);
+    const chunk = compactRunLogChunk(formatNativeRunLogEvent(event));
+    expect(chunk).toContain("[paperclip truncated run log chunk:");
+    expect(() => JSON.parse(chunk)).toThrow();
+    await store.append(handle, { stream: "stdout", ts: "immutable-old-format", seq: 1, sourceCursor: 2, chunk });
+    await store.finalize(handle);
+    const prefix = await readAll(store, handle);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+      async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+    expect(await readAll(cold, handle)).toBe(prefix);
+    const [run] = await db.select({ bytes: heartbeatRuns.logBytes }).from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, f.binding.runId));
+    expect(run!.bytes).toBeNull();
+  });
+
+  it("leaves an old privacy-mangled canonical prefix failed and byte-identical", async () => {
+    const f = await fixture();
+    const event = { ...f.event(1), payload: { header: "Authorization: Bearer fixture-old-bearer-canary" } };
+    await f.persist(event, 2);
+    const store = f.store(), handle = await store.begin(f.binding);
+    const chunk = compactRunLogChunk(formatNativeRunLogEvent(event));
+    expect(() => JSON.parse(chunk)).toThrow();
+    await store.append(handle, { stream: "stdout", ts: "immutable-old-privacy-format", seq: 1, sourceCursor: 2, chunk });
+    await store.finalize(handle);
+    const prefix = await readAll(store, handle);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+      async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+    expect(await readAll(cold, handle)).toBe(prefix);
+    const [run] = await db.select({ bytes: heartbeatRuns.logBytes }).from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, f.binding.runId));
+    expect(run!.bytes).toBeNull();
+  });
+
+  it("rejects a verifier record exceeding the accepted 8MiB writer bound", async () => {
+    const f = await fixture();
+    const verifier = createNativeRunLogCoverageVerifier(db, f.binding.companyId, f.binding.runId);
+    const bytes = Buffer.from(`${JSON.stringify({ chunk: "x".repeat(8 * 1024 * 1024) })}\n`);
+    await expect(verifier.write(bytes)).rejects.toThrow("native_log_source_coverage_unproven");
+  });
+
+  it.each([true, false])("retains a zero-watermark production prefix when username masking changes from %s", async (enabled) => {
+    const f = await fixture();
+    const setMasking = async (value: boolean) => db.execute(sql`insert into instance_settings(general)
+      values(${JSON.stringify({ censorUsernameInLogs: value })}::jsonb)
+      on conflict(singleton_key) do update set general=excluded.general`);
+    const redaction = async () => ({ enabled: (await instanceSettingsService(db).getGeneral()).censorUsernameInLogs,
+      userNames: ["fixture-operator"], homeDirs: ["/home/fixture-operator"] });
+    await setMasking(enabled);
+    const event = { ...f.event(1), payload: { text: "fixture-operator owns /home/fixture-operator/file.txt" } };
+    const { handle, prefix } = await publishProductionPrefix(f, event, { watermark: false, redaction });
+    expect(prefix.includes("fixture-operator")).toBe(!enabled);
+    await setMasking(!enabled);
+    await f.persist({ ...f.event(2), payload: event.payload }, 5);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, redaction);
+    const content = (await cold.read(handle)).content;
+    expect(content.startsWith(prefix)).toBe(true);
+    const lines = content.trim().split("\n").map(line => JSON.parse(line));
+    expect(lines).toHaveLength(2);
+    expect(lines[1].chunk.includes("fixture-operator")).toBe(enabled);
+  });
+
+  it("rejects a historical duplicate prefix even when the second copy advanced the journal cursor", async () => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const event = f.event(1), chunk = formatNativeRunLogEvent(event);
+    await f.persist(event, 2);
+    await f.persist(f.event(2), 5);
+    await store.append(handle, { stream: "stdout", ts: "old-unwatermarked", seq: 1, chunk });
+    await store.append(handle, { stream: "stdout", ts: "old-repair", seq: 2, sourceCursor: 2, chunk });
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    const oldHead = await store.begin(f.binding);
+    expect(oldHead.initialSourceCursor).toBe(2);
+    await store.finalize(oldHead);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+      async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+    expect((await cold.read(handle)).content).toBe(prefix);
+    const [run] = await db.select({ bytes: heartbeatRuns.logBytes }).from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, f.binding.runId));
+    expect(run!.bytes).toBeNull();
+  });
+
+  it("fully verifies a valid legacy nonzero prefix once, then resumes with bounded verified provenance", async () => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    await f.persist(f.event(1), 2);
+    await store.append(handle, { stream: "stdout", ts: "legacy", seq: 1, sourceCursor: 2,
+      chunk: formatNativeRunLogEvent(f.event(1)) });
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    await f.persist(f.event(2), 5);
+    await f.settle(handle.logRef);
+    const cold = f.store(), read = cold.read;
+    let bytesRead = 0;
+    cold.read = async (...args) => {
+      const result = await read(...args); bytesRead += Buffer.byteLength(result.content); return result;
+    };
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    expect(bytesRead).toBeGreaterThanOrEqual(Buffer.byteLength(prefix));
+    const repaired = (await read(handle)).content;
+    expect(repaired.startsWith(prefix)).toBe(true);
+    expect(repaired.trim().split("\n")).toHaveLength(2);
+    await f.settle(handle.logRef);
+    bytesRead = 0;
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    expect(bytesRead).toBe(0);
+    expect((await read(handle)).content).toBe(repaired);
+  });
+
+  it("uses bounded committed-cursor proof even after a long non-native diagnostic tail", async () => {
+    const f = await fixture();
+    const { store, handle } = await publishProductionPrefix(f, f.event(1));
+    const writer = await store.begin(f.binding);
+    for (let index = 0; index < 50; index++) await store.append(writer, {
+      stream: "stderr", ts: "diagnostic", seq: index + 2, chunk: "diagnostic ".repeat(3000),
+      preservesSourceCoverage: true,
+    });
+    await store.finalize(writer);
+    const before = await store.begin(f.binding);
+    expect(before.initialBytes).toBeGreaterThan(1_000_000);
+    expect(before.initialSourceCursor).toBe(2);
+    await store.finalize(before);
+    await f.persist(f.event(2), 5);
+    await f.settle(handle.logRef);
+    const cold = f.store(), originalRead = cold.read;
+    let prefixBytesRead = 0;
+    cold.read = async (...args) => {
+      const result = await originalRead(...args);
+      prefixBytesRead += Buffer.byteLength(result.content);
+      return result;
+    };
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    expect(prefixBytesRead).toBe(0);
+    const content = (await originalRead(handle, { limitBytes: 4 * 1024 * 1024 })).content;
+    const chunks = content.trim().split("\n").map(line => JSON.parse(line).chunk as string);
+    expect(chunks).toHaveLength(52);
+    expect(JSON.parse(chunks[0]!).event.sourceEventId).toBe("source:1");
+    expect(chunks.slice(1, -1).every(chunk => chunk.startsWith("diagnostic "))).toBe(true);
+    expect(JSON.parse(chunks.at(-1)!).event.sourceEventId).toBe("source:2");
+  });
+
+  it.each(["visible-content", "foreign", "order"] as const)("rejects a compacted zero-watermark prefix with %s corruption", async kind => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const event = toolEvent(f, "retained output ".repeat(5000).slice(0, 65_536));
+    await f.persist(event, 2);
+    await f.persist(f.event(2), 5);
+    const published = kind === "foreign" ? { ...event, runId: randomUUID() } : event;
+    let chunk = compactRunLogChunk(formatNativeRunLogEvent(published));
+    if (kind === "visible-content") chunk = chunk.replace("retained output", "changed! output");
+    if (kind === "order") await store.append(handle, { stream: "stdout", ts: "original", seq: 1,
+      chunk: formatNativeRunLogEvent(f.event(2)) });
+    await store.append(handle, { stream: "stdout", ts: "original", seq: kind === "order" ? 2 : 1, chunk });
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+      async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+    expect((await cold.read(handle)).content).toBe(prefix);
+  });
+
+  it("records native cursors when output begins with the queued legacy default, then cancels and cold-reads without replay", async () => {
+    const f = await fixture();
+    const [persisted] = await db.select({ runtimeMode: heartbeatRuns.runtimeMode })
+      .from(heartbeatRuns).where(eq(heartbeatRuns.id, f.binding.runId));
+    expect(persisted!.runtimeMode).toBe("legacy");
+    const initialRow: Partial<typeof heartbeatRuns.$inferSelect> = {
+      companyId: f.binding.companyId, agentId: f.binding.agentId,
+      id: f.binding.runId, status: "running", ...persisted,
+      logStore: null, logRef: null, logBytes: null, lastOutputSeq: 0, lastOutputAt: null,
+    };
+    const run = initialRow as typeof heartbeatRuns.$inferSelect;
+    const store = f.store();
+    const output = createHeartbeatRunOutput(db, {
+      run, runId: run.id, issueId: null, runLogStore: store,
+      identityRedactor: { chunk: (_stream, chunk) => chunk, redact: (value) => value },
+      secretKeys: new Set(), appendRunEvent: async () => undefined,
+      getCurrentUserRedactionOptions: async () => ({ enabled: false }),
+    });
+    const handlers = await output.begin(run);
+    // Runtime selection persists a new row; the output closure retains the earlier row.
+    await db.update(heartbeatRuns).set({ runtimeMode: "native" }).where(eq(heartbeatRuns.id, run.id));
+    for (const cursor of [2, 5]) {
+      const event = f.event(cursor === 2 ? 1 : 2);
+      await f.persist(event, cursor);
+      // This is the executor's persisted-native callback/fallback dispatch.
+      const chunk = formatNativeRunLogEvent(event);
+      if (handlers.onPersistedNativeLog) await handlers.onPersistedNativeLog(cursor, chunk);
+      else await handlers.onLog("stdout", chunk);
+    }
+    await store.finalize(output.handle!);
+    const prefix = (await store.read(output.handle!)).content;
+    await f.settle(output.handle!.logRef);
+    const cold = f.store();
+    const read = await readRecoverableRunLog(db, cold, output.handle!, undefined,
+      { runtimeMode: "native", status: "cancelled", logBytes: null });
+    expect(read.content).toBe(prefix);
+    const entries = read.content.trim().split("\n").map(line => JSON.parse(JSON.parse(line).chunk).event);
+    expect(entries.map(event => event.sourceEventId)).toEqual(["source:1", "source:2"]);
+    const resumed = await cold.begin(f.binding);
+    expect(resumed.initialSourceCursor).toBe(5);
+    await cold.finalize(resumed);
+  });
+
+  it("proves a zero-watermark prefix across JSONB key order and appends only the missing suffix", async () => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const events = Array.from({ length: 50 }, (_, index) => ({
+      ...f.event(index + 1), payload: { text: `${index + 1} ${"再開".repeat(1000)}` },
+    }));
+    for (const [index, event] of events.entries()) await f.persist(event, (index + 1) * 2);
+    const [durable] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents)
+      .where(eq(heartbeatRunEvents.runId, f.binding.runId)).limit(1);
+    expect(JSON.stringify(durable!.payload!.prpEvent)).not.toBe(JSON.stringify(events[0]));
+    await store.append(handle, { stream: "stdout", ts: "original", seq: 1, chunk: "preparation diagnostic\n" });
+    for (const [index, event] of events.slice(0, 49).entries()) {
+      await store.append(handle, { stream: "stdout", ts: "original", seq: index + 2,
+        chunk: formatNativeRunLogEvent(event) });
+    }
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    // The full prefix spans multiple byte pages and multiple 16-event source pages.
+    let prefixOffset = Buffer.byteLength(prefix);
+    let fullPrefix = prefix;
+    for (;;) {
+      const page = await store.read(handle, { offset: prefixOffset });
+      fullPrefix += page.content;
+      if (page.nextOffset === undefined) break;
+      prefixOffset = page.nextOffset;
+    }
+    expect(Buffer.byteLength(fullPrefix)).toBeGreaterThan(256_000);
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    const content = (await cold.read(handle, { limitBytes: 1024 * 1024 })).content;
+    expect(content.startsWith(fullPrefix)).toBe(true);
+    const entries = content.trim().split("\n").map(line => JSON.parse(line));
+    expect(entries.slice(1).map(entry => JSON.parse(entry.chunk).event.sourceSeq))
+      .toEqual(Array.from({ length: 50 }, (_, index) => index + 1));
+    expect(entries.map(entry => entry.seq)).toEqual(Array.from({ length: 51 }, (_, index) => index + 1));
+    const resumed = await cold.begin(f.binding);
+    expect(resumed.initialSourceCursor).toBe(100);
+    await cold.finalize(resumed);
+  });
+
+  it.each([true, false])("preserves an existing image prefix when base64 was omitted: %s", async (omitted) => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const data = "A".repeat(1024);
+    const event = { ...f.event(1), payload: { image: { type: "image",
+      source: omitted ? { type: "base64", data } : { data, type: "base64" },
+    } } };
+    await f.persist(event, 1);
+    await f.persist(f.event(2), 2);
+    const chunk = compactRunLogChunk(formatNativeRunLogEvent(event));
+    expect(chunk.includes("[omitted base64 image data: 1024 chars]")).toBe(omitted);
+    await store.append(handle, { stream: "stdout", ts: "original", seq: 1, chunk });
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await recoverSettledNativeRunLog(db, cold, handle.logRef, async () => ({ enabled: false }));
+    const content = (await cold.read(handle)).content;
+    expect(content.startsWith(prefix)).toBe(true);
+    expect(content.trim().split("\n")).toHaveLength(2);
+  });
+
+  it("rejects conflicting raw image data even when the base64 length matches", async () => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const image = (data: string) => ({ ...f.event(1), payload: { image: {
+      type: "image", source: { data, type: "base64" },
+    } } });
+    await f.persist(image("A".repeat(1024)), 1);
+    await store.append(handle, { stream: "stdout", ts: "original", seq: 1,
+      chunk: compactRunLogChunk(formatNativeRunLogEvent(image("B".repeat(1024)))) });
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+      async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+    expect((await cold.read(handle)).content).toBe(prefix);
+  });
+
+  it("rejects an image omission marker in a non-image payload", async () => {
+    const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+    const event = (data: string) => ({ ...f.event(1), payload: { file: {
+      type: "file", source: { type: "base64", data },
+    } } });
+    await f.persist(event("A".repeat(1024)), 1);
+    await store.append(handle, { stream: "stdout", ts: "original", seq: 1,
+      chunk: formatNativeRunLogEvent(event("[omitted base64 image data: 1024 chars]")) });
+    await store.finalize(handle);
+    const prefix = (await store.read(handle)).content;
+    await f.settle(handle.logRef);
+    const cold = f.store();
+    await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+      async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+    expect((await cold.read(handle)).content).toBe(prefix);
+  });
+
+  for (const kind of ["conflicting", "malformed", "duplicate"] as const) {
+    it(`leaves a ${kind} zero-watermark prefix unchanged and incomplete`, async () => {
+      const f = await fixture(), store = f.store(), handle = await store.begin(f.binding);
+      await f.persist(f.event(1), 1);
+      await f.persist(f.event(2), 2);
+      const chunk = kind === "conflicting"
+        ? formatNativeRunLogEvent({ ...f.event(1), payload: { text: "different" } })
+        : kind === "malformed" ? '{"type":"paperclip.prp.event","event":broken\n'
+        : formatNativeRunLogEvent(f.event(1));
+      await store.append(handle, { stream: "stdout", ts: "original", seq: 1, chunk });
+      if (kind === "duplicate") await store.append(handle, { stream: "stdout", ts: "original", seq: 2, chunk });
+      await store.finalize(handle);
+      const prefix = (await store.read(handle)).content;
+      await f.settle(handle.logRef);
+      const cold = f.store();
+      await expect(recoverSettledNativeRunLog(db, cold, handle.logRef,
+        async () => ({ enabled: false }))).rejects.toThrow("native_log_source_coverage_unproven");
+      expect((await cold.read(handle)).content).toBe(prefix);
+      const [row] = await db.select({ bytes: heartbeatRuns.logBytes }).from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, f.binding.runId));
+      expect(row!.bytes).toBeNull();
+    });
+  }
+});

server/src/services/native-runtime/native-session-executor.ts

@@ -6129,6 +6129,7 @@ function runnerProviderStateFilename(execution: NativeExecutionInput): string {
     case "grok_bot":
     case "openai_dot":
       return "dot-provider-state.json";
+    case "cursor_cloud":
     case "codex":
     case "opencode":
       return "codex-provider-state.json";
@@ -6229,11 +6230,12 @@ export function providerSessionIdentityFromDurableProviderState(input: {
         providerSessionIdentity: structuredClone(identity),
       };
     }
+    case "cursor_cloud":
     case "codex":
     case "opencode": {
       const config = record(state.config);
       const expectedDriver =
-        input.execution.provider.kind === "codex"
+        input.execution.provider.kind === "cursor_cloud" ? "cursor_cloud_api" : input.execution.provider.kind === "codex"
           ? "codex_app_server"
           : "opencode_server";
       if (

server/src/services/object-byte-journal.ts

@@ -53,6 +53,14 @@ export class ObjectJournalConflictError extends Error {
     super("The output journal is owned by another writer");
   }
 }
+export class ObjectJournalSourcePrefixProofUnavailableError extends Error {
+  constructor(readonly writer: ObjectJournalWriter) {
+    super("The committed log prefix has no available source proof");
+  }
+}
+class ObjectJournalSourcePrefixProofInvalidError extends Error {
+  constructor() { super("object_journal_source_prefix_proof_invalid"); }
+}
 export function isObjectPreconditionFailure(error: unknown) {
   const e = error as { name?: string; $metadata?: { httpStatusCode?: number } };
   return e?.name === "PreconditionFailed" || e?.$metadata?.httpStatusCode === 412;
@@ -292,6 +300,69 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
     }
     return { head: value, etag: result.etag };
   }
+  function sourcePrefixProof(snapshot: ObjectJournalHead) {
+    // Leave strict head v1 unchanged for older readers/rollback. Ownership and
+    // close/claim state may change without changing this immutable byte prefix.
+    const fingerprint = sha256(Buffer.from(JSON.stringify({
+      end: snapshot.end, sequence: snapshot.sequence, sourceCursor: snapshot.sourceCursor,
+      root: snapshot.root, roots: snapshot.roots ?? null,
+    })));
+    return {
+      // Existing scoped object cleanup also removes these unreferenced metadata objects.
+      key: `${objectPrefix}source-prefix-proof-v1-${fingerprint}`,
+      bytes: Buffer.from(JSON.stringify({ schema: "paperclip.run-log.source-prefix-proof.v1", fingerprint })),
+    };
+  }
+  async function hasVerifiedSourcePrefix(snapshot: ObjectJournalHead, signal?: AbortSignal) {
+    const proof = sourcePrefixProof(snapshot);
+    let stored: Awaited<ReturnType<typeof getBytes>>;
+    try { stored = await getBytes(proof.key, 4096, signal); }
+    catch (error) { if (isMissing(error)) return false; throw error; }
+    if (!stored.bytes.equals(proof.bytes)) throw new ObjectJournalSourcePrefixProofInvalidError();
+    return true;
+  }
+  async function recordVerifiedSourcePrefix(writer: ObjectJournalWriter, sourceCursor: number, signal?: AbortSignal) {
+    const callerSignal = signal;
+    signal = signal
+      ? AbortSignal.any([signal, AbortSignal.timeout(15_000)])
+      : AbortSignal.timeout(15_000);
+    signal.throwIfAborted();
+    if (!Number.isSafeInteger(sourceCursor) || sourceCursor < writer.head.sourceCursor)
+      throw new Error("object_journal_invalid_source_prefix_cursor");
+    const current = await head(signal);
+    if (!current || current.etag !== writer.etag || current.head.closed || current.head.deleted)
+      throw new ObjectJournalConflictError();
+    // One-time verification can repair legacy max-cursor metadata, never payload bytes.
+    if (sourceCursor !== writer.head.sourceCursor)
+      writer = await publish({ ...writer.head, sourceCursor }, writer.etag, signal);
+    const proof = sourcePrefixProof(writer.head);
+    try {
+      await retryObjectOperation(() => putConditional({
+        objectKey: proof.key, body: proof.bytes, contentLength: proof.bytes.length,
+        contentType: "application/json", ifNoneMatch: "*", signal,
+      }), signal);
+    } catch (error) {
+      callerSignal?.throwIfAborted();
+      if (error instanceof ObjectJournalConflictError) throw error;
+      // A lost reply or an already published identical immutable proof is safe.
+      try {
+        if (await hasVerifiedSourcePrefix(writer.head, signal)) return writer;
+      } catch (readError) {
+        callerSignal?.throwIfAborted();
+        if (readError instanceof ObjectJournalSourcePrefixProofInvalidError ||
+          readError instanceof ObjectJournalConflictError) throw readError;
+        if (signal.aborted || isRetryable(readError))
+          throw new ObjectJournalSourcePrefixProofUnavailableError(writer);
+        throw readError;
+      }
+      // Only this immutable proof publication is optional. The ownership read
+      // and any cursor/head CAS above retain their fatal failure boundary.
+      if (signal.aborted || isRetryable(error))
+        throw new ObjectJournalSourcePrefixProofUnavailableError(writer);
+      throw error;
+    }
+    return writer;
+  }
   async function publish(value: ObjectJournalHead, etag: string | null, signal?: AbortSignal) {
     value = headSchema.parse(value);
     signal = signal
@@ -569,6 +640,6 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
       continuationToken = page.continuationToken;
     } while (continuationToken);
   }
-  return { head, claim, append, close, readRange, remove };
+  return { head, claim, append, close, readRange, remove, hasVerifiedSourcePrefix, recordVerifiedSourcePrefix };
 }
 export type ObjectByteJournal = ReturnType<typeof createObjectByteJournal>;

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

@@ -47,6 +47,7 @@ function memory() {
   };
   return {
     provider,
+    objects,
     puts: () => puts,
     offline: (value: boolean) => {
       unavailable = value;
@@ -62,6 +63,215 @@ const event = (seq: number) => ({
 });
 
 describe("S3 run log store", () => {
+  it("awaits a successor flush started by another drain waiter before finalizing", async () => {
+    const m = memory(), store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", flushMs: 1000,
+      segmentBytes: 1024, assertWriterAuthority: async () => undefined });
+    const handle = await store.begin(input), put = m.provider.putObjectConditional!.bind(m.provider);
+    let firstEntered!: () => void, secondEntered!: () => void, releaseFirst!: () => void, releaseSecond!: () => void;
+    const firstStarted = new Promise<void>(resolve => { firstEntered = resolve; });
+    const secondStarted = new Promise<void>(resolve => { secondEntered = resolve; });
+    const firstGate = new Promise<void>(resolve => { releaseFirst = resolve; });
+    const secondGate = new Promise<void>(resolve => { releaseSecond = resolve; });
+    let publications = 0;
+    m.provider.putObjectConditional = async value => {
+      if (value.objectKey.endsWith("/head.json") && Buffer.isBuffer(value.body)) {
+        const head = JSON.parse(value.body.toString());
+        if (!head.closed && head.end > 0) {
+          publications++;
+          if (publications === 1) { firstEntered(); await firstGate; }
+          else if (publications === 2) { secondEntered(); await secondGate; }
+        }
+      }
+      return put(value);
+    };
+    const first = await store.enqueue!(handle, { ...event(1), chunk: "first ".repeat(200) });
+    const firstCommitted = first.committed.catch(error => error);
+    await firstStarted;
+    const second = await store.enqueue!(handle, { ...event(2), chunk: "second" });
+    const secondCommitted = second.committed.catch(error => error);
+    let finalized = false;
+    const finalizing = store.finalize(handle).then(result => { finalized = true; return result; });
+    releaseFirst();
+    await secondStarted;
+    await new Promise<void>(resolve => setImmediate(resolve));
+    const finalizedBeforeSecondCommit = finalized;
+    releaseSecond();
+    const summary = await finalizing;
+    const committed = await Promise.all([firstCommitted, secondCommitted]);
+    const content = (await store.read(handle)).content;
+    expect(finalizedBeforeSecondCommit).toBe(false);
+    expect(committed.every(value => typeof value === "number")).toBe(true);
+    expect(summary.bytes).toBe(Buffer.byteLength(content));
+    expect(content.trim().split("\n").map(line => JSON.parse(line).seq)).toEqual([1, 2]);
+  });
+
+  it("rejects outer records exceeding 8MiB without losing later bounded appends", async () => {
+    const m = memory(), store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async () => undefined });
+    const handle = await store.begin(input);
+    await expect(store.append(handle, { ...event(1), chunk: "x".repeat(8 * 1024 * 1024) }))
+      .rejects.toThrow("Object run log record exceeds pending-byte limit");
+    await store.append(handle, event(1));
+    const content = (await store.read(handle)).content;
+    expect(content.trim().split("\n")).toHaveLength(1);
+    expect((await store.finalize(handle)).bytes).toBe(Buffer.byteLength(content));
+  });
+
+  it("keeps live byte appends deliverable after a proof-only outage and requires cold prefix verification", async () => {
+    const m = memory(), options = { provider: m.provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async () => undefined };
+    const put = m.provider.putObjectConditional!.bind(m.provider);
+    let proofUnavailable = false, proofFailures = 0;
+    m.provider.putObjectConditional = async value => {
+      if (proofUnavailable && value.objectKey.includes("/objects/source-prefix-proof-v1-")) {
+        proofFailures++;
+        throw Object.assign(new Error("proof storage unavailable"), { $metadata: { httpStatusCode: 503 } });
+      }
+      return put(value);
+    };
+    const store = createObjectRunLogStore(options), handle = await store.begin(input);
+    await store.recordVerifiedSourcePrefix!(handle, 0);
+    proofUnavailable = true;
+    const first = await store.append(handle, { ...event(1), sourceCursor: 2, preservesSourceCoverage: true })
+      .catch(error => error);
+    // The existing byte/index/head publication succeeded despite the proof PUT failure.
+    expect((await store.read(handle)).content.trim().split("\n")).toHaveLength(1);
+    proofUnavailable = false;
+    const second = await store.append(handle, { ...event(2), sourceCursor: 5, preservesSourceCoverage: true })
+      .catch(error => error);
+    const summary = await store.finalize(handle);
+    const cold = createObjectRunLogStore(options), resumed = await cold.begin(input);
+    const content = (await cold.read(resumed)).content;
+    expect(resumed.initialSourceCursor).toBe(5);
+    // Absence selects the maintained consumer's mandatory full source-prefix scan.
+    expect(resumed.initialSourcePrefixVerified).toBe(false);
+    proofUnavailable = true;
+    // A complete legacy proof can advance the old max cursor before its optional
+    // proof PUT fails. The public recorder must retain that known committed head.
+    await cold.recordVerifiedSourcePrefix!(resumed, 8);
+    expect(resumed.initialSourcePrefixVerified).toBe(false);
+    proofUnavailable = false;
+    const third = await cold.append(resumed, { ...event(3), sourceCursor: 11, preservesSourceCoverage: true });
+    await cold.finalize(resumed);
+    const final = await store.begin(input);
+    const finalContent = (await store.read(final)).content;
+    await store.finalize(final);
+    expect(proofFailures).toBeGreaterThan(0);
+    expect(first).toBeTypeOf("number");
+    expect(second).toBeTypeOf("number");
+    expect(third).toBeTypeOf("number");
+    expect(summary.bytes).toBe(Buffer.byteLength(content));
+    expect(finalContent.trim().split("\n").map(line => JSON.parse(line).seq)).toEqual([1, 2, 3]);
+    expect(final.initialSourceCursor).toBe(11);
+    expect(final.initialSourcePrefixVerified).toBe(false);
+  });
+
+  it.each(["internal-deadline", "caller-abort"] as const)("distinguishes proof %s after the byte head committed", async kind => {
+    const m = memory(), caller = new AbortController();
+    const options = { provider: m.provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async () => undefined };
+    const store = createObjectRunLogStore(options), handle = await store.begin({ ...input, signal: caller.signal });
+    await store.recordVerifiedSourcePrefix!(handle, 0);
+    const put = m.provider.putObjectConditional!.bind(m.provider);
+    const deadline = new AbortController();
+    let entered!: () => void;
+    const proofStarted = new Promise<void>(resolve => { entered = resolve; });
+    m.provider.putObjectConditional = async value => {
+      if (!value.objectKey.includes("/objects/source-prefix-proof-v1-")) return put(value);
+      entered();
+      return new Promise((_resolve, reject) => {
+        if (value.signal!.aborted) reject(value.signal!.reason);
+        else value.signal!.addEventListener("abort", () => reject(value.signal!.reason), { once: true });
+      });
+    };
+    const timeout = vi.spyOn(AbortSignal, "timeout").mockReturnValue(deadline.signal);
+    let first: unknown;
+    const callerReason = new Error("caller revoked writer");
+    try {
+      const pending = store.append(handle, { ...event(1), sourceCursor: 2, preservesSourceCoverage: true })
+        .catch(error => error);
+      await proofStarted;
+      if (kind === "internal-deadline") deadline.abort(new DOMException("proof deadline", "TimeoutError"));
+      else caller.abort(callerReason);
+      first = await pending;
+    } finally { timeout.mockRestore(); m.provider.putObjectConditional = put; }
+    const second = await store.append(handle, { ...event(2), sourceCursor: 5, preservesSourceCoverage: true })
+      .catch(error => error);
+    const summary = await store.finalize(handle);
+    const cold = createObjectRunLogStore(options), resumed = await cold.begin(input);
+    await cold.finalize(resumed);
+    expect(resumed.initialSourcePrefixVerified).toBe(false);
+    if (kind === "internal-deadline") {
+      expect(first).toBeTypeOf("number");
+      expect(second).toBeTypeOf("number");
+      expect(summary.bytes).not.toBeNull();
+      expect(resumed.initialSourceCursor).toBe(5);
+    } else {
+      expect(first).toBe(callerReason);
+      expect(second).toBe(callerReason);
+      expect(summary.bytes).toBeNull();
+      expect(resumed.initialSourceCursor).toBe(2);
+    }
+  });
+
+  it("keeps an unknown head publication failure fatal", async () => {
+    const m = memory(), store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async () => undefined });
+    const handle = await store.begin(input);
+    await store.recordVerifiedSourcePrefix!(handle, 0);
+    const put = m.provider.putObjectConditional!.bind(m.provider);
+    m.provider.putObjectConditional = async value => {
+      if (value.objectKey.endsWith("/head.json")) throw new Error("unknown head publication failure");
+      return put(value);
+    };
+    await expect(store.append(handle, { ...event(1), sourceCursor: 2, preservesSourceCoverage: true }))
+      .rejects.toThrow("unknown head publication failure");
+    await expect(store.append(handle, { ...event(2), sourceCursor: 5, preservesSourceCoverage: true }))
+      .rejects.toThrow("unknown head publication failure");
+    expect((await store.read(handle)).content).toBe("");
+    await store.finalize(handle);
+  });
+
+  it("binds source proof to exact immutable bytes and invalidates it on ordinary appends", async () => {
+    const m = memory(), options = { provider: m.provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async () => undefined };
+    const store = createObjectRunLogStore(options), handle = await store.begin(input);
+    expect(handle.initialSourcePrefixVerified).toBe(false);
+    await store.recordVerifiedSourcePrefix!(handle, 0);
+    await store.append(handle, { ...event(1), sourceCursor: 2, preservesSourceCoverage: true });
+    await store.finalize(handle);
+    const replacement = createObjectRunLogStore(options), verified = await replacement.begin(input);
+    expect(verified.initialSourceCursor).toBe(2);
+    expect(verified.initialSourcePrefixVerified).toBe(true);
+    const head = JSON.parse([...m.objects].find(([key]) => key.endsWith("/head.json"))![1].toString());
+    expect(Object.keys(head).sort()).toEqual([
+      "cleanupAfter", "closed", "deleted", "end", "generation", "owner", "root", "roots", "sequence", "sourceCursor", "v",
+    ]);
+    // An older/generic caller cannot carry the attestation across new bytes,
+    // even when it reuses the old max cursor or calls its payload a diagnostic.
+    await replacement.append(verified, { ...event(2), sourceCursor: 2,
+      chunk: '{"type":"paperclip.prp.event","event":{"duplicate":true}}' });
+    await replacement.finalize(verified);
+    const legacy = await store.begin(input);
+    expect(legacy.initialSourceCursor).toBe(2);
+    expect(legacy.initialSourcePrefixVerified).toBe(false);
+    await store.finalize(legacy);
+  });
+
+  it("fails closed when an existing exact-prefix proof is malformed or changed", async () => {
+    const m = memory(), options = { provider: m.provider, keyPrefix: "logs", flushMs: 1,
+      assertWriterAuthority: async () => undefined };
+    const store = createObjectRunLogStore(options), handle = await store.begin(input);
+    await store.recordVerifiedSourcePrefix!(handle, 0);
+    await store.append(handle, { ...event(1), sourceCursor: 2, preservesSourceCoverage: true });
+    await store.finalize(handle);
+    const proofs = [...m.objects.keys()].filter(key => key.includes("/objects/source-prefix-proof-v1-"));
+    expect(proofs.length).toBeGreaterThan(0);
+    for (const key of proofs) m.objects.set(key, Buffer.from('{"schema":"changed-proof"}'));
+    await expect(createObjectRunLogStore(options).begin(input))
+      .rejects.toThrow("object_journal_source_prefix_proof_invalid");
+  });
+
   it("batches concurrent appends, confirms durability, and resumes offsets/sequence on a replacement", async () => {
     const m = memory(),
       options = {

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

@@ -2,10 +2,11 @@ import { beginIdleTrackedWork } from "./task-admission.js";
 import { createHash, randomUUID } from "node:crypto";
 import type { StorageProvider } from "../storage/types.js";
 import type { RunLogStore, RunLogHandle, RunLogBeginInput } from "./run-log-store.js";
-import { createObjectByteJournal, type ObjectJournalWriter } from "./object-byte-journal.js";
+import { createObjectByteJournal, ObjectJournalSourcePrefixProofUnavailableError, type ObjectJournalWriter } from "./object-byte-journal.js";
+import { MAX_RUN_LOG_RECORD_BYTES } from "./run-log-record-limits.js";
 
 const MAX_SEGMENT_BYTES = 1024 * 1024;
-const MAX_PENDING_BYTES = 8 * MAX_SEGMENT_BYTES;
+const MAX_PENDING_BYTES = MAX_RUN_LOG_RECORD_BYTES;
 const MAX_READ_BYTES = 4 * MAX_SEGMENT_BYTES;
 export interface ObjectRunLogStoreOptions {
   provider: StorageProvider;
@@ -32,6 +33,9 @@ interface WriterState {
   closed: boolean;
   failure?: unknown;
   signal?: AbortSignal;
+  sourcePrefixVerified: boolean;
+  coverageEligible: boolean;
+  admittedSourceCursor: number;
 }
 
 export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunLogStore {
@@ -103,8 +107,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
     }
     if (state.flushing) {
       await state.flushing;
-      if (state.pending.length) return flush(state);
-      return;
+      return flush(state);
     }
     if (state.failure) throw state.failure;
     if (!state.pending.length) return;
@@ -127,6 +130,20 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
             state.writer.head.sourceCursor,
           ),
         );
+        if (state.sourcePrefixVerified) {
+          try {
+            state.writer = await state.journal.recordVerifiedSourcePrefix(
+              state.writer, state.writer.head.sourceCursor, state.signal,
+            );
+          } catch (error) {
+            if (!(error instanceof ObjectJournalSourcePrefixProofUnavailableError)) throw error;
+            // The byte/index/head append is already committed. Lose only the
+            // optional provenance optimization; later appends remain deliverable.
+            state.writer = error.writer;
+            state.sourcePrefixVerified = false;
+            state.coverageEligible = false;
+          }
+        }
         for (const item of batch) item.resolve(item.bytes.length);
       } catch (error) {
         state.failure = error;
@@ -196,6 +213,12 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
       if (state.closed) throw new Error("Object run log writer is closed");
       state.signal?.throwIfAborted();
       if (state.failure) throw state.failure;
+      if (event.preservesSourceCoverage !== true || event.sourceCursor !== undefined &&
+        (!Number.isSafeInteger(event.sourceCursor) || event.sourceCursor <= state.admittedSourceCursor)) {
+        state.sourcePrefixVerified = false;
+        state.coverageEligible = false;
+      }
+      state.admittedSourceCursor = Math.max(state.admittedSourceCursor, event.sourceCursor ?? 0);
       state.pendingBytes += bytes.length;
       totalPendingBytes += bytes.length;
       const finishIdle = beginIdleTrackedWork();
@@ -228,6 +251,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         assertAuthority: () => options.assertWriterAuthority(input),
         signal: input.signal,
       });
+      const sourcePrefixVerified = await store.hasVerifiedSourcePrefix(writer.head, input.signal);
       const handle: RunLogHandle = {
         store: "s3_segments",
         logRef,
@@ -235,6 +259,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         initialBytes: writer.head.end,
         initialSequence: writer.head.sequence,
         initialSourceCursor: writer.head.sourceCursor,
+        initialSourcePrefixVerified: sourcePrefixVerified,
       };
       writers.set(handle, {
         journal: store,
@@ -243,6 +268,9 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         pendingBytes: 0,
         closed: false,
         signal: input.signal,
+        sourcePrefixVerified,
+        coverageEligible: true,
+        admittedSourceCursor: writer.head.sourceCursor,
       });
       return handle;
     },
@@ -255,6 +283,39 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
       if (!state) throw new Error("Object run log writer not found");
       await flush(state);
     },
+    async recordVerifiedSourcePrefix(handle, sourceCursor) {
+      const state = writers.get(handle);
+      if (!state || state.closed) throw new Error("Object run log writer is closed");
+      if (!state.coverageEligible) throw new Error("native_log_source_prefix_changed");
+      await flush(state);
+      if (!state.coverageEligible) throw new Error("native_log_source_prefix_changed");
+      // Serialize metadata publication with byte flushes, including diagnostics
+      // admitted while the one-time source proof is being recorded.
+      state.flushing = (async () => {
+        try {
+          let proofUnavailable = false;
+          try {
+            state.writer = await state.journal.recordVerifiedSourcePrefix(state.writer, sourceCursor, state.signal);
+          } catch (error) {
+            if (!(error instanceof ObjectJournalSourcePrefixProofUnavailableError)) throw error;
+            state.writer = error.writer;
+            proofUnavailable = true;
+          }
+          if (!state.coverageEligible) throw new Error("native_log_source_prefix_changed");
+          state.sourcePrefixVerified = !proofUnavailable;
+          if (proofUnavailable) state.coverageEligible = false;
+          state.admittedSourceCursor = Math.max(state.admittedSourceCursor, sourceCursor);
+          handle.initialSourcePrefixVerified = state.sourcePrefixVerified;
+          handle.initialSourceCursor = sourceCursor;
+          handle.initialBytes = state.writer.head.end;
+          handle.initialSequence = state.writer.head.sequence;
+        } finally {
+          state.flushing = undefined;
+          schedule(state);
+        }
+      })();
+      await state.flushing;
+    },
     async finalize(handle, finalizeOptions) {
       const state = writers.get(handle);
       if (!state) throw new Error("Object run log writer not found");

server/src/services/run-log-record-limits.ts

@@ -0,0 +1,2 @@
+/** Maximum encoded outer NDJSON record, including its newline. */
+export const MAX_RUN_LOG_RECORD_BYTES = 8 * 1024 * 1024;

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

@@ -22,6 +22,8 @@ export interface RunLogHandle {
   initialBytes?: number;
   initialSequence?: number;
   initialSourceCursor?: number;
+  /** Exact immutable prefix has a durable source-coverage proof, not just a max cursor. */
+  initialSourcePrefixVerified?: boolean;
 }
 
 export interface RunLogReadOptions {
@@ -68,13 +70,17 @@ export interface RunLogStore {
   begin(input: RunLogBeginInput): Promise<RunLogHandle>;
   append(
     handle: RunLogHandle,
-    event: { stream: "stdout" | "stderr" | "system"; chunk: string; ts: string; seq?: number; sourceCursor?: number },
+    event: { stream: "stdout" | "stderr" | "system"; chunk: string; ts: string; seq?: number; sourceCursor?: number;
+      /** Caller proved this is ordered canonical output or a non-native diagnostic. */
+      preservesSourceCoverage?: boolean },
   ): Promise<number>;
   enqueue?(
     handle: RunLogHandle,
     event: Parameters<RunLogStore["append"]>[1],
   ): Promise<{ committed: Promise<number> }>;
   flush?(handle: RunLogHandle): Promise<void>;
+  /** Record the caller's complete ordered source-prefix proof without changing log bytes. */
+  recordVerifiedSourcePrefix?(handle: RunLogHandle, sourceCursor: number): Promise<void>;
   finalize(handle: RunLogHandle, options?: { computeDigest?: boolean }): Promise<RunLogFinalizeSummary>;
   read(handle: RunLogHandle, opts?: RunLogReadOptions): Promise<RunLogReadResult>;
   // Optional so existing fakes/fixtures keep compiling: uploads every dirty