fix(logs): drain existing admissions before finalization (#19055)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Run logs retain execution output for operators.
> - Object storage uses bounded admission and asynchronous uploads.
> - A diagnostic can wait for capacity when its native execution ends.
> - Finalization closed the writer before that existing admission could
finish.
> - This pull request joins those admissions before it closes the log.
> - The benefit is complete accepted output with the existing storage
bounds.

## Linked Issues or Issue Description

Refs #18944 and #18327.

**What happened?**

A runner close diagnostic began object-log admission before its recorder
retired. The call waited at the 8 MiB pending-byte bound. Completion
then closed the writer. The waiting call failed with `Object run log
writer is closed`. A valid final log hash covered the stored prefix, but
the diagnostic was absent.

**Expected behavior**

Finalization must retain calls that started before closure. It must
reject new calls after closure. Cancellation and storage failures must
still settle without a success claim.

**Steps to reproduce**

1. Use the actual recorder and segmented object store with a controlled
conditional upload.
2. Fill the pending-byte bound. Start native warm-session retirement
while the diagnostic waits for capacity.
3. Complete the run through the real completion handler. Release the
upload.
4. Check the diagnostic bytes, final hash, source coverage, and warning
rows.

## What Changed

- Track active admission promises in each object-log writer.
- Reject new producers when finalization starts. Join existing bounded
admissions, then flush their tail before manifest closure.
- Add actual completion coverage and store controls for failure, Stop,
global capacity, and independent writers.
- Clarify the finalization contract in the storage documentation. Keep
feature maps unchanged for this routine fix.

## Verification

- Before the fix, the actual completion probe lost a pre-retirement
close diagnostic and emitted `object_log_persistence_failed`. Useful
canonical output, source coverage, and the final prefix hash remained
valid.
- After the fix, the exact probe retains the diagnostic and checks the
stored bytes and full hash. Ordinary completion and stalled local append
controls also pass.
- Focused store controls pass, including shared 64 MiB capacity,
new-producer rejection, storage failure, Stop, and independent writer
progress.
- Production and test source at
`bc03f8d53a721f01f80a12105a69d8583865e651`: 3 actual completion controls
and 53 store, recorder, journal, and diagnostic tests pass. Run with
`node node_modules/vitest/vitest.mjs run` from `server/`; select the
four safety files plus the completion file with `-t "retired recorder"`
in a separate invocation.
- Initial treatment repository typechecks and build pass. Current map
and module checks pass. A broader four-suite run had 99 passing
assertions but failed its existing 30-second completion teardown hook. A
later isolated completion run passed all 61 cases at the same limits.
The original teardown cause remains unresolved; no timeout was changed.
- Repository typechecks and build pass at that source commit. Current
head `6652adc7f3379ae250f7b726dad6c58c8482410f` removes only the
incidental map paragraph; production and test source are identical. All
current-head checks are terminal: 54 successful check runs, one
successful legacy status, and three skipped checks. Required `ci /
verify` and `ci / e2e` pass. Greptile reviewed exact current head at
5/5. Both review threads are resolved. GitHub reports the PR clean and
mergeable. Full repository tests stopped in chat shard 2: a GitHub
recovery fair-batch test expected no provider requests but observed `GET
/app/hook/config`. Its exact unchanged case passes in isolation. The
original cause remains unresolved; this full-suite run is red.
First-head remote server lane 2 failed a native-reassessment backlog
test: its quiet check observed two active work items instead of zero.
The exact unchanged case passes locally. That original CI cause remains
unresolved; no job retry or timeout change was made. Other first-head CI
lanes passed. The first native review rated the earlier source head 4/5
for the same removed map note. That is not a current-head assessment.
Current GitHub rules require zero approving reviews and only the two CI
checks above; a fresh native advisory re-review is optional and remains
unavailable while its earlier run has an unresolved workspace/accounting
obligation. That obligation is tracked separately and is not claimed
recovered.
- Removed the incidental map paragraph identified by Greptile and the
native review. The follow-up contains no runtime or test changes.
- This proves the local production-path mechanism. It does not attribute
any specific hosted warning to this cause. Hosted replay and served
revision checks remain separate acceptance work.

## Risks

Finalization now waits for existing bounded capacity admissions. Storage
cancellation and the existing operation deadlines still apply. Tests
check that failed uploads and Stop settle, and that a separate writer
can make progress. No capacity limit, timeout, or execution authority
changes.

## Model Used

OpenAI `gpt-6.1-sol`, through Codex, with reasoning and tool execution.
Context window size is not exposed by the host.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass (56 focused checks; the
full-suite red remains documented above)
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

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

doc/run-evidence-storage.md

@@ -89,8 +89,9 @@ the bounded waiting allowance is rejected without poisoning accepted batches.
 Transient network, 409, 429 and server errors receive bounded retries. A 412
 or a changed head fences the stale writer; a 409 alone does not prove ownership
 loss or conditional-write support. Live log messages and progress are published
-after storage confirmation. Finalization drains admitted writes before closing
-and hashing the stream. Logging failure reports incomplete persistence and
+after storage confirmation. Finalization rejects new enqueue calls and drains
+already-started bounded admissions, including calls waiting for capacity, before
+closing and hashing the stream. Logging failure reports incomplete persistence and
 leaves the final size/digest unknown rather than claiming a complete snapshot.
 The original progress projection remains coarsely throttled; concurrent batch
 completions share a single progress write.

server/src/services/heartbeat/__tests__/object-log-fixture.ts

@@ -3,13 +3,14 @@ import { Readable } from "node:stream";
 import type { StorageProvider } from "../../../storage/types.js";
 import { createObjectRunLogStore } from "../../object-run-log-store.js";
 
-export function objectLogFixture() {
+export function objectLogFixture(beforeConditionalPut?: (input: Parameters<NonNullable<StorageProvider["putObjectConditional"]>>[0]) => Promise<void>) {
   const objects = new Map<string, Buffer>();
   const digest = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex");
   const provider: StorageProvider = {
     id: "s3",
     async putObject() { throw new Error("unexpected unconditional write"); },
     async putObjectConditional(input) {
+      await beforeConditionalPut?.(input);
       const old = objects.get(input.objectKey);
       if (input.ifNoneMatch ? old : !old || digest(old) !== input.ifMatch)
         throw Object.assign(new Error("conflict"), { name: "PreconditionFailed" });

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

@@ -44,6 +44,7 @@ vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() }));
 const diagnosticNativeFixture = vi.hoisted(() => ({
   enabled: false,
   retireWarm: false,
+  beforeWarmClose: undefined as (() => Promise<void>) | undefined,
   backendOptions: undefined as Parameters<typeof import("../../vendor/paperclip-runner/index.js").createNativeSessionBackend>[1],
 }));
 vi.mock("../native-runtime/native-codex-runner.js", async importOriginal => {
@@ -63,6 +64,7 @@ vi.mock("../../vendor/paperclip-runner/index.js", async importOriginal => {
       const transport = diagnosticNativeFixture.backendOptions!.codexTransportFactory!();
       const session = { close: async ({ reason }: { reason: string }) => { await transport.close(reason); } };
       await options.onSession?.(session as never);
+      await diagnosticNativeFixture.beforeWarmClose?.();
       if (diagnosticNativeFixture.retireWarm) await closeIdleWarmNativeSessionsForRestart();
       return { result: { summary: "accepted result" }, terminal: { runTerminalState: "succeeded" }, turnId: "test-turn",
         normalizedSessionId: options.input.session.normalizedSessionId, providerSessionId: "synthetic-provider",
@@ -173,12 +175,52 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
   }
   const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!);
 
-  it.each([false, true])("finishes accepted output before ordinary lease teardown without writing to a retired recorder (stalled append=%s)", async stalledAppend => {
+  it.each(["ordinary", "stalled append", "object admission"] as const)("finishes accepted output before ordinary lease teardown without writing to a retired recorder (%s)", async mode => {
+    const stalledAppend = mode === "stalled append", stalledAdmission = mode === "object admission";
     const f = await fixture({ runtimeMode: "native",
       resultJson: { durable: "accepted result" } });
     const workspace = await realpath(await mkdtemp(join(tmpdir(), "diagnostic-completion-")));
     onTestFinished(() => rm(workspace, { recursive: true, force: true }));
-    const store = stalledAppend ? createDurableRunLogStore({ basePath: join(workspace, "run-logs") }) : objectLogFixture();
+    let holdUploads = false, uploadEntered = false, finalizeEntered = false;
+    let releaseUpload!: () => void;
+    const uploadGate = new Promise<void>(resolve => { releaseUpload = resolve; });
+    const store = stalledAppend ? createDurableRunLogStore({ basePath: join(workspace, "run-logs") }) : objectLogFixture(async input => {
+      if (stalledAdmission && holdUploads && input.objectKey.includes("/objects/")) {
+        uploadEntered = true;
+        await uploadGate;
+      }
+    });
+    let latestOutputSeq = 0, admittedBytes = 0;
+    const diagnosticAdmissions: Array<{ chunk: string; admitted?: boolean; error?: string }> = [];
+    if (store.enqueue) {
+      const originalEnqueue = store.enqueue.bind(store);
+      vi.spyOn(store, "enqueue").mockImplementation(async (handle, event) => {
+        latestOutputSeq = event.seq ?? latestOutputSeq;
+        const observation = event.chunk.includes("runnerd diagnostic") ? { chunk: event.chunk } as typeof diagnosticAdmissions[number] : undefined;
+        if (observation) diagnosticAdmissions.push(observation);
+        let receipt;
+        try { receipt = await originalEnqueue(handle, event); if (observation) observation.admitted = true; }
+        catch (error) { if (observation) observation.error = error instanceof Error ? error.message : String(error); throw error; }
+        if (holdUploads) admittedBytes += Buffer.byteLength(JSON.stringify({ ts: event.ts,
+          attemptId: handle.attemptId, stream: event.stream, chunk: event.chunk, seq: event.seq }) + "\n");
+        return receipt;
+      });
+    }
+    const originalFinalize = store.finalize.bind(store);
+    vi.spyOn(store, "finalize").mockImplementation((...args) => {
+      finalizeEntered = true;
+      return originalFinalize(...args);
+    });
+    const readLog = async () => {
+      let content = "", offset = 0;
+      do {
+        const page = await store.read(output.handle!, { offset, limitBytes: 4 * 1024 * 1024 });
+        content += page.content;
+        if (page.nextOffset === undefined) break;
+        offset = page.nextOffset;
+      } while (true);
+      return { content };
+    };
     let releaseAppend!: () => void, observeAppend!: () => void;
     const appendGate = new Promise<void>(resolve => { releaseAppend = resolve; });
     const appendStarted = new Promise<void>(resolve => { observeAppend = resolve; });
@@ -207,17 +249,23 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
     await new Promise<void>((resolve, reject) => { api.once("error", reject); api.listen(0, "127.0.0.1", resolve); });
     onTestFinished(() => new Promise<void>(resolve => api.close(() => resolve())));
     diagnosticNativeFixture.enabled = true;
-    diagnosticNativeFixture.retireWarm = stalledAppend;
+    diagnosticNativeFixture.retireWarm = stalledAppend || stalledAdmission;
+    if (stalledAdmission) diagnosticNativeFixture.beforeWarmClose = async () => {
+      await store.flush!(output.handle!);
+      holdUploads = true;
+      const target = 8 * 1024 * 1024 - 1;
+      while (admittedBytes < target) {
+        const overhead = Buffer.byteLength(JSON.stringify({ ts: new Date().toISOString(),
+          attemptId: output.handle!.attemptId, stream: "stdout", chunk: "", seq: latestOutputSeq + 1 }) + "\n");
+        const chunk = "x".repeat(Math.min(64 * 1024, target - admittedBytes - overhead));
+        await handlers.onLog("stdout", chunk);
+      }
+      expect(admittedBytes).toBe(target);
+      expect(uploadEntered).toBe(true);
+    };
     let completed = false;
     const diagnosticWrites: Promise<void>[] = [];
     const finishing = (async () => {
-      const adapter = await executePaperclipNativeSession({ db, execution: execution as never,
-        runnerInstanceId: "synthetic-controller", useRunnerd: true, onLog: (stream, chunk) => {
-          const write = handlers.onLog(stream, chunk);
-          if (chunk.includes("runnerd diagnostic")) diagnosticWrites.push(write);
-          return write;
-        },
-        runnerEnvironment: { HOME: workspace, CODEX_HOME: join(workspace, "synthetic-codex-home") } });
       const event: PrpEvent = { schema: "paperclip.prp.event.v1", schemaVersion: 1,
         sourceInstanceId: randomUUID(), sourceEventId: randomUUID(), sourceSeq: 1, sourceKind: "runner",
         runId: f.run.id, normalizedSessionId: execution.session.normalizedSessionId,
@@ -227,6 +275,13 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
         nativeSource: { sourceInstanceId: event.sourceInstanceId, sourceEventId: event.sourceEventId,
           sourceSeq: event.sourceSeq, protocolSchemaVersion: 1, canonicalPayload: event as unknown as Record<string, unknown> } });
       if (!stalledAppend) await handlers.onPersistedNativeLog!(persisted.row.seq, formatNativeRunLogEvent(event));
+      const adapter = await executePaperclipNativeSession({ db, execution: execution as never,
+        runnerInstanceId: "synthetic-controller", useRunnerd: true, onLog: (stream, chunk) => {
+          const write = handlers.onLog(stream, chunk);
+          if (chunk.includes("runnerd diagnostic")) diagnosticWrites.push(write);
+          return write;
+        },
+        runnerEnvironment: { HOME: workspace, CODEX_HOME: join(workspace, "synthetic-codex-home") } });
       await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "committed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
       Object.assign(f.run, await read(f.run.id));
       f.agent.adapterType = "paperclip_runner";
@@ -241,12 +296,17 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
     })();
     void finishing.catch(() => undefined);
     try {
+      if (stalledAdmission) {
+        await vi.waitFor(() => expect(finalizeEntered).toBe(true), { timeout: 10_000 });
+        expect(diagnosticWrites.length).toBeGreaterThan(0);
+        releaseUpload();
+      }
       if (stalledAppend) {
         await appendStarted;
         await vi.waitFor(() => expect(completed).toBe(true), { timeout: 10_000 });
       }
       const { persisted, event } = await finishing;
-      const before = await store.read(output.handle!, { limitBytes: 65536 });
+      const before = await readLog();
       expect(before.content).toContain("accepted output for this exact run");
       const settled = await read(f.run.id);
       expect(settled).toMatchObject({ companyId: f.run.companyId, status: "succeeded", resultJson: { durable: "accepted result" } });
@@ -270,23 +330,35 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
       await createHeartbeatRunControl(db, deps).releaseEnvironmentLeasesForRun({ runId: f.run.id, companyId: f.run.companyId,
         agentId: f.run.agentId, status: "succeeded", providerResourceDisposition: "destroy" });
       expect(deps.options.closeWarmNativeSessionsForRun).toHaveBeenCalledOnce();
-      const after = await store.read(output.handle!, { limitBytes: 65536 });
+      const after = await readLog();
       expect(after.content).toBe(before.content);
       if (!stalledAppend) {
         const coverage = createNativeRunLogCoverageVerifier(db, f.run.companyId, f.run.id, { redactChunks: chunk => [chunk] });
         await coverage.write(Buffer.from(after.content));
         expect(await coverage.finish()).toBe(persisted.row.seq);
       }
+      if (stalledAdmission) {
+        await Promise.allSettled(diagnosticWrites);
+        expect(diagnosticAdmissions.length).toBeGreaterThan(0);
+        const storedChunks = before.content.trimEnd().split("\n").map(line => JSON.parse(line).chunk);
+        for (const admission of diagnosticAdmissions) {
+          expect(admission).toMatchObject({ admitted: true });
+          expect(admission.error).toBeUndefined();
+          expect(storedChunks).toContain(admission.chunk);
+        }
+      }
       const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id));
       expect(events.filter(event => (event.payload as { reason?: string } | null)?.reason === "object_log_persistence_failed")).toEqual([]);
     } finally {
       releaseAppend();
+      releaseUpload();
       await Promise.allSettled(diagnosticWrites);
       await finishing;
       if (stalledAppend) expect(await read(f.run.id)).toMatchObject({ logBytes: null, logSha256: null });
       await closeIdleWarmNativeSessionsForRestart();
       diagnosticNativeFixture.enabled = false;
       diagnosticNativeFixture.retireWarm = false;
+      diagnosticNativeFixture.beforeWarmClose = undefined;
     }
   });
 

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

@@ -318,6 +318,93 @@ describe("S3 run log store", () => {
     const [size, result] = await Promise.all([pending, finalizing]);
     expect(result.bytes).toBe(size);
   });
+  it.each(["commit", "storage failure", "Stop"] as const)("settles pre-close bounded admission through %s and fences new producers", async mode => {
+    const m = memory(), original = m.provider.putObjectConditional!;
+    let release!: () => void, entered = false;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const cancellation = new AbortController();
+    m.provider.putObjectConditional = async value => {
+      if (value.objectKey.includes("/objects/") && !entered) {
+        entered = true;
+        await gate;
+        value.signal?.throwIfAborted();
+        if (mode === "storage failure") throw new Error("controlled upload failure");
+      }
+      return original(value);
+    };
+    const store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined });
+    const handle = await store.begin({ ...input, signal: cancellation.signal });
+    const first = await store.enqueue!(handle, { ...event(1), chunk: "a".repeat(7 * 1024 * 1024) });
+    const committed = first.committed.catch(error => error);
+    await vi.waitFor(() => expect(entered).toBe(true));
+    let admitted = false;
+    const waiting = store.enqueue!(handle, { ...event(2), chunk: "b".repeat(2 * 1024 * 1024) })
+      .then(receipt => { admitted = true; return receipt.committed; }).catch(error => error);
+    let finalized = false;
+    const finalizing = store.finalize(handle).then(value => { finalized = true; return value; });
+    try {
+      await expect(store.enqueue!(handle, event(3))).rejects.toThrow("closed");
+      const peer = await store.begin({ ...input, runId: "peer" });
+      await store.append(peer, { ...event(1), chunk: "peer output" });
+      expect((await store.finalize(peer)).bytes).toBeGreaterThan(0);
+      expect(admitted).toBe(false);
+      expect(finalized).toBe(false);
+      if (mode === "Stop") cancellation.abort(new Error("controlled Stop"));
+    } finally { release(); }
+    const [firstResult, secondResult, summary] = await Promise.all([committed, waiting, finalizing]);
+    if (mode === "commit") {
+      expect(admitted).toBe(true);
+      expect(summary.bytes).toBe(Number(firstResult) + Number(secondResult));
+      const hash = createHash("sha256");
+      let offset = 0, chunks = 0;
+      for (;;) {
+        const page = await store.read(handle, { offset, limitBytes: 1024 * 1024 });
+        hash.update(page.content);
+        chunks += (page.content.match(/"chunk":/g) ?? []).length;
+        if (page.nextOffset === undefined) break;
+        offset = page.nextOffset;
+      }
+      expect(chunks).toBe(2);
+      expect(summary.sha256).toBe(hash.digest("hex"));
+    } else {
+      expect(firstResult).toBeInstanceOf(Error);
+      expect(secondResult).toBeInstanceOf(Error);
+      expect(admitted).toBe(false);
+      expect(summary).toMatchObject({ bytes: null, sha256: null });
+    }
+  });
+  it("drains an admission waiting on global capacity while other writers flush", async () => {
+    const m = memory(), original = m.provider.putObjectConditional!;
+    let release!: () => void, uploads = 0;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    m.provider.putObjectConditional = async value => {
+      if (value.objectKey.includes("/objects/")) { uploads++; await gate; }
+      return original(value);
+    };
+    const store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined });
+    const handles = [], committed = [];
+    for (let index = 0; index < 9; index++) {
+      const handle = await store.begin({ ...input, runId: `full-${index}` });
+      handles.push(handle);
+      committed.push((await store.enqueue!(handle, { ...event(1), chunk: "x".repeat(7 * 1024 * 1024) })).committed);
+    }
+    await vi.waitFor(() => expect(uploads).toBe(4));
+    const target = await store.begin({ ...input, runId: "waiting" });
+    let admitted = false, finalized = false;
+    const waiting = store.enqueue!(target, { ...event(1), chunk: "y".repeat(2 * 1024 * 1024) })
+      .then(receipt => { admitted = true; return receipt.committed; });
+    const finalizing = store.finalize(target).then(value => { finalized = true; return value; });
+    try {
+      await expect(store.enqueue!(target, event(2))).rejects.toThrow("closed");
+      expect(admitted).toBe(false);
+      expect(finalized).toBe(false);
+    } finally { release(); }
+    const [bytes, summary] = await Promise.all([waiting, finalizing]);
+    expect(admitted).toBe(true);
+    expect(summary.bytes).toBe(bytes);
+    await Promise.all(committed);
+    await Promise.all(handles.map(handle => store.finalize(handle)));
+  });
   it("does not acknowledge a batch when storage fails", async () => {
     const m = memory(),
       store = createObjectRunLogStore({

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

@@ -31,6 +31,7 @@ interface WriterState {
   timer?: NodeJS.Timeout;
   flushing?: Promise<void>;
   closed: boolean;
+  admitting: Set<Promise<unknown>>;
   failure?: unknown;
   signal?: AbortSignal;
   sourcePrefixVerified: boolean;
@@ -167,6 +168,12 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
   async function enqueue(handle: RunLogHandle, event: Parameters<RunLogStore["append"]>[1]) {
     const state = writers.get(handle);
     if (!state || state.closed) throw new Error("Object run log writer is closed");
+    const admission = admit(state, handle, event);
+    state.admitting.add(admission);
+    try { return await admission; }
+    finally { state.admitting.delete(admission); }
+  }
+  async function admit(state: WriterState, handle: RunLogHandle, event: Parameters<RunLogStore["append"]>[1]) {
     state.signal?.throwIfAborted();
     if (state.failure) throw state.failure;
     const bytes = Buffer.from(
@@ -186,7 +193,6 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         state.pendingBytes + bytes.length > MAX_PENDING_BYTES ||
         totalPendingBytes + bytes.length > 64 * MAX_SEGMENT_BYTES
       ) {
-        if (state.closed) throw new Error("Object run log writer is closed");
         if (state.failure) throw state.failure;
         if (!waiting) {
           if (admissions >= 32 || admissionBytes + bytes.length > MAX_PENDING_BYTES)
@@ -210,7 +216,6 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
           capacityWaiters.add(wake);
         });
       }
-      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 &&
@@ -267,6 +272,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         pending: [],
         pendingBytes: 0,
         closed: false,
+        admitting: new Set(),
         signal: input.signal,
         sourcePrefixVerified,
         coverageEligible: true,
@@ -322,6 +328,8 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
       state.closed = true;
       capacityChanged();
       try {
+        await flush(state);
+        await Promise.allSettled([...state.admitting]);
         await flush(state);
         state.writer = await state.journal.close(state.writer, state.signal);
         if (finalizeOptions?.computeDigest === false)