perf(run-logs): read object-store log tails in parallel, once, and cached (#20284)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Hosted instances keep run logs in object storage as a journal of
content-addressed segments and binary index nodes
> - The task chat reads the last 32 KB of a run's log, aligned to a
record, to show what the agent did
> - A run that flushes often has many small segments. The tail read
fetched each object one after another, and record alignment walked the
index a second time
> - On staging, one tail read took 11–12 s, and the task chat waited for
it (see #20277)
> - This pull request reads the tail in parallel, in one pass, and
caches the objects that cannot change
> - The benefit is that the first tail read is about 6.6x faster, and
repeated reads (live polling, reopening a task) use only the mutable
head

## Linked Issues or Issue Description

**What happened?**
`GET
/api/heartbeat-runs/:runId/log?offset=…&limitBytes=32000&alignStartToRecord=true`
took 11–12 s on staging for a 400 KB object-store log. Without alignment
it took 6.1 s. A read from offset 0 took 2.0 s. The task chat waited for
this read before it showed the run's output.

**Expected behavior**
A 32 KB tail read returns in well under a second, and a repeated read of
the same tail costs about one storage round trip.

**Steps to reproduce**
1. Use object-backed run logs (`s3_segments`).
2. Write a run log in many small flushes, as a live run does.
3. Read its last 32 KB with `alignStartToRecord=true`, and count the
object GETs and how many run at the same time.

**Paperclip version or commit**
`master` at 6459488984.

**Deployment mode**
Hosted (authenticated) deployment with S3-compatible run-log storage.

Refs #20276 (measurement and plan), #20277 (chat no longer waits on a
slow transcript).

## What Changed

- `object-byte-journal.ts`, `readRange`:
- Visits the index tree and fetches sibling objects concurrently, with
at most 4 fetches for each read.
  - Reassembles the pieces by offset.
  - Cancels the remaining fetches when one fails.
- The shared `objectIO` limiter still caps the process at 8 slots, and
background persistence still gets freed slots first (#19176).
- `object-byte-journal.ts`, `load`:
- Keeps segments and index nodes in a process-wide LRU, limited to 32 MB
in total and 1 MB per object. The budget counts each entry's key and a
fixed 256-byte overhead with its bytes.
- These objects cannot change: each key ends in the SHA-256 of its
bytes, and `load` verifies that digest before it caches the object.
  - The mutable head is never cached.
- The cache and the shared fetches are scoped to the storage provider
instance. Production uses one provider in the run-log store singleton.
- Concurrent fetches of the same object are shared. A shared fetch has
its own controller. A reader that aborts only stops its own wait. The
fetch is aborted only when no reader still waits for it, so the other
readers never start a second GET for that object.
- `object-run-log-store.ts`, `read`:
- With `alignStartToRecord`, the first 4 KB alignment window is read in
the same range as the body.
  - The existing 32 KB backward windows still handle longer records.
  - The output is unchanged.
- New test `object-run-log-store-tail-read.test.ts`, written against a
fake provider that counts GETs:
  - correct aligned and unaligned content
  - concurrent fetches
- alignment costs at most 30 extra GETs, with no second walk of the
index
  - a repeated read costs 1 GET (the head)
  - appended output is visible after cached reads
  - a reader that shares a fetch survives another reader's abort
  - a reader that joined a shared fetch stops at once on its own abort
- when the reader that started a fetch aborts, no object is fetched
twice
  - a shared fetch stops when its last reader aborts
- one store's cached objects are not served to a reader of another store
- a tail that starts inside a record longer than the 4 KB lookback (6 KB
and 90 KB records)
  - the cache budget counts per-entry overhead
- the shared-fetch tests hold object GETs open with a gate instead of
wall-clock delays

## Verification

- `cd server && npx vitest run
src/services/object-run-log-store-tail-read.test.ts
src/services/object-run-log-store.test.ts
src/services/object-byte-journal.test.ts
src/services/run-log-store.test.ts`: 4 files, 74 tests pass. The
tail-read file passed 5 of 5 repeated runs.
- The 18 server test files that use the journal, the run-log store or
`readLog`: 381 tests pass. Two skill-materialization tests timed out
under load and pass when run alone (4/4).
- Red checks:
- On `master`, the aligned tail read made 357 GETs with at most 1
running at a time, and a repeated read made 356 GETs.
- When I removed the abort handling for shared fetches, the shared-abort
test failed 3 times out of 3.
- The tests for the review fixes (joiner abort, no duplicate GET after
the first reader aborts, last-reader abort, per-store cache scope, cache
entry overhead) fail without the fix.
- An off-by-one in the long-record lookback arithmetic makes both
long-record tests fail.
- Benchmark: same fixture, with 30 ms latency added to each GET, a 32 KB
tail of 1,200 single-record segments.

| Read | `master` | This PR |
|---|---|---|
| Aligned | 11,753 ms, 365 GETs, 1 at a time | 1,778 ms, 205 GETs, 4 at
a time |
| Unaligned | 5,979 ms, 184 GETs | 1,607 ms, 184 GETs |
| Repeat aligned | 11,686 ms, 365 GETs | 34 ms, 1 GET |

- `pnpm typecheck` in `server` passes.

## Risks

- **Memory:** the cache can hold up to 32 MB of decoded log bytes in
each server process. Authorization still happens in the routes before
any read. The cache key is the full object key, which includes the run's
journal prefix.
- **Storage concurrency:** one read now holds up to 4 I/O slots instead
of 1. Background persistence still gets freed slots first. Under many
simultaneous cold reads, interactive reads can reach the 32-waiter limit
sooner and receive the existing retryable 503 `object_journal_io_busy`.
The cache makes repeated reads cheap, which offsets this.
- **Remaining cost:** the first read of a cold tail still walks a binary
index over small segments, about 1.8 s at 30 ms per GET. A larger gain
needs writer-side changes, such as wider index nodes or compaction when
the log closes. Those change the persisted shape, so they are not in
this PR.
- No migration, configuration change, persisted-shape change or new
dependency.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used in Claude Code
with tool use (shell, browser automation, file editing).

## Checklist

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

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -255,6 +255,109 @@ async function objectIO<T>(
   }
 }
 
+// Segments and index nodes are content-addressed: each key ends in the SHA-256
+// of the object's bytes, and load() verifies that digest. Their bytes cannot
+// change, so repeated reads of a run's tail (live polling, reopening a task)
+// reuse them instead of paying one object-store round trip per object. Only the
+// mutable head is read again. The cache is bounded by total bytes (payload plus
+// per-entry overhead) and evicts the least recently used objects first. Readers authorize before any read.
+// Entries are scoped to the storage provider instance, so a different store
+// (or a test's fresh store) never sees objects that another store returned.
+const READ_CACHE_MAX_BYTES = 32 * 1024 * 1024;
+const READ_CACHE_MAX_OBJECT_BYTES = 1024 * 1024;
+const readCache = new Map<string, Buffer>();
+let readCacheBytes = 0;
+const providerScopes = new WeakMap<StorageProvider, number>();
+let nextProviderScope = 0;
+function providerCacheScope(provider: StorageProvider) {
+  let scope = providerScopes.get(provider);
+  if (scope === undefined) providerScopes.set(provider, (scope = nextProviderScope++));
+  return `${scope}:`;
+}
+
+// Concurrent readers of one object share one fetch. The fetch has its own
+// controller: a reader that aborts only stops its own wait, and the fetch is
+// aborted only when no reader is still waiting for it.
+interface SharedObjectRead {
+  work: Promise<Buffer>;
+  controller: AbortController;
+  waiters: number;
+}
+const readsInFlight = new Map<string, SharedObjectRead>();
+function cachedObject(key: string) {
+  const bytes = readCache.get(key);
+  if (bytes) {
+    readCache.delete(key);
+    readCache.set(key, bytes);
+  }
+  return bytes;
+}
+// The budget counts each entry's key and a fixed bookkeeping overhead with its
+// bytes, so many small segments cannot hold much more memory than the cap.
+const READ_CACHE_ENTRY_OVERHEAD_BYTES = 256;
+const readCacheEntryCost = (key: string, bytes: Buffer) =>
+  bytes.length + key.length * 2 + READ_CACHE_ENTRY_OVERHEAD_BYTES;
+function cacheObject(key: string, bytes: Buffer) {
+  if (bytes.length > READ_CACHE_MAX_OBJECT_BYTES || readCache.has(key)) return;
+  readCache.set(key, bytes);
+  readCacheBytes += readCacheEntryCost(key, bytes);
+  for (const [oldestKey, oldest] of readCache) {
+    if (readCacheBytes <= READ_CACHE_MAX_BYTES) break;
+    readCache.delete(oldestKey);
+    readCacheBytes -= readCacheEntryCost(oldestKey, oldest);
+  }
+}
+/** Bytes the read cache counts against its budget, including entry overhead. */
+export function objectJournalReadCacheBytes() {
+  return readCacheBytes;
+}
+export function clearObjectJournalReadCache() {
+  readCache.clear();
+  readCacheBytes = 0;
+  // Readers already waiting keep their fetch; new readers start a new one.
+  readsInFlight.clear();
+}
+
+// One range read fetches its objects with bounded overlap. The shared objectIO
+// limiter still caps the process; this keeps one tail read from taking all of
+// its slots or filling the interactive wait queue.
+const READ_RANGE_CONCURRENCY = 4;
+function concurrencyLimit(max: number) {
+  let active = 0;
+  const waiting: Array<() => void> = [];
+  return async <T>(operation: () => Promise<T>): Promise<T> => {
+    if (active >= max) await new Promise<void>((resolve) => waiting.push(resolve));
+    else active++;
+    try {
+      return await operation();
+    } finally {
+      const next = waiting.shift();
+      if (next) next();
+      else active--;
+    }
+  };
+}
+
+// Wait for a shared operation, but stop waiting when this caller's signal aborts.
+function untilAborted<T>(operation: Promise<T>, signal?: AbortSignal): Promise<T> {
+  if (!signal) return operation;
+  signal.throwIfAborted();
+  return new Promise<T>((resolve, reject) => {
+    const onAbort = () => reject(signal.reason);
+    signal.addEventListener("abort", onAbort, { once: true });
+    operation.then(
+      (value) => {
+        signal.removeEventListener("abort", onAbort);
+        resolve(value);
+      },
+      (error) => {
+        signal.removeEventListener("abort", onAbort);
+        reject(error);
+      },
+    );
+  });
+}
+
 export interface ObjectByteJournalOptions {
   provider: StorageProvider;
   prefix: string;
@@ -280,6 +383,7 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
   const headKey = `${prefix}/head.json`;
   const objectPrefix = `${prefix}/objects/`;
   const readPriority = options.readPriority ?? "background";
+  const cacheScope = providerCacheScope(provider);
   const putConditional = (
     input: Parameters<NonNullable<StorageProvider["putObjectConditional"]>>[0],
   ) => objectIO(() => provider.putObjectConditional!(input), "background", input.signal);
@@ -462,11 +566,42 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
   }
   async function load(ref: Reference, signal?: AbortSignal) {
     validateReference(ref);
-    const stored = await getBytes(ref.key, MAX_OBJECT_BYTES * 2, signal);
-    const bytes = options.decode ? await options.decode(stored.bytes) : stored.bytes;
-    if (bytes.length > MAX_OBJECT_BYTES || sha256(bytes) !== ref.sha256)
-      throw new Error("object_journal_integrity_error");
-    return bytes;
+    signal?.throwIfAborted();
+    const cacheKey = `${cacheScope}${ref.key}`;
+    const cached = cachedObject(cacheKey);
+    if (cached) return cached;
+    let shared = readsInFlight.get(cacheKey);
+    if (!shared) {
+      const controller = new AbortController();
+      const work = (async () => {
+        const stored = await getBytes(ref.key, MAX_OBJECT_BYTES * 2, controller.signal);
+        const bytes = options.decode ? await options.decode(stored.bytes) : stored.bytes;
+        if (bytes.length > MAX_OBJECT_BYTES || sha256(bytes) !== ref.sha256)
+          throw new Error("object_journal_integrity_error");
+        cacheObject(cacheKey, bytes);
+        return bytes;
+      })();
+      const entry: SharedObjectRead = { work, controller, waiters: 0 };
+      const release = () => {
+        if (readsInFlight.get(cacheKey) === entry) readsInFlight.delete(cacheKey);
+      };
+      void work.then(release, release);
+      readsInFlight.set(cacheKey, entry);
+      shared = entry;
+    }
+    const entry = shared;
+    entry.waiters++;
+    try {
+      return await untilAborted(entry.work, signal);
+    } finally {
+      entry.waiters--;
+      // The last reader left before the fetch finished: stop the fetch, and
+      // let a later reader start a new one.
+      if (!entry.waiters && signal?.aborted) {
+        if (readsInFlight.get(cacheKey) === entry) readsInFlight.delete(cacheKey);
+        entry.controller.abort(signal.reason);
+      }
+    }
   }
   async function save(
     bytes: Buffer,
@@ -540,33 +675,39 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
     if ((!snapshot.root && !snapshot.roots?.length) || limit === 0 || start >= snapshot.end)
       return Buffer.alloc(0);
     const end = Math.min(snapshot.end, start + limit),
-      chunks: Buffer[] = [];
+      pieces: Array<{ start: number; bytes: Buffer }> = [],
+      limited = concurrencyLimit(READ_RANGE_CONCURRENCY),
+      failed = new AbortController(),
+      readSignal = signal ? AbortSignal.any([signal, failed.signal]) : failed.signal;
+    const overlaps = (ref: Reference) => ref.end > start && ref.start < end;
+    // Siblings are fetched concurrently; pieces are reassembled by offset.
     async function readPayload(ref: Reference) {
-      const bytes = await load(ref, signal);
+      const bytes = await limited(() => load(ref, readSignal));
       if (bytes.length !== ref.end - ref.start)
         throw new Error("object_journal_invalid_segment_length");
-      chunks.push(
-        bytes.subarray(Math.max(0, start - ref.start), Math.min(bytes.length, end - ref.start)),
-      );
+      pieces.push({
+        start: ref.start,
+        bytes: bytes.subarray(Math.max(0, start - ref.start), Math.min(bytes.length, end - ref.start)),
+      });
     }
-    async function visit(ref: Reference, expectedLevel?: number) {
-      const node = await loadNode(ref, signal);
+    async function visit(ref: Reference, expectedLevel?: number): Promise<void> {
+      const node = await limited(() => loadNode(ref, readSignal));
       if (expectedLevel !== undefined && node.level !== expectedLevel)
         throw new Error("object_journal_invalid_index_level");
-      for (const child of node.children) {
-        if (child.end <= start || child.start >= end) continue;
-        if (node.level > 0) await visit(child, node.level - 1);
-        else await readPayload(child);
-      }
+      await Promise.all(node.children.filter(overlaps).map((child) =>
+        node.level > 0 ? visit(child, node.level - 1) : readPayload(child)));
     }
-    if (snapshot.roots) {
-      for (const root of snapshot.roots) {
-        if (root.end <= start || root.start >= end) continue;
-        if (root.level === 0) await readPayload(root);
-        else await visit(root, root.level - 1);
-      }
-    } else if (snapshot.root) await visit(snapshot.root);
-    const bytes = Buffer.concat(chunks);
+    try {
+      if (snapshot.roots) {
+        await Promise.all(snapshot.roots.filter(overlaps).map((root) =>
+          root.level === 0 ? readPayload(root) : visit(root, root.level - 1)));
+      } else if (snapshot.root) await visit(snapshot.root);
+    } catch (error) {
+      failed.abort(error);
+      throw error;
+    }
+    pieces.sort((a, b) => a.start - b.start);
+    const bytes = Buffer.concat(pieces.map((piece) => piece.bytes));
     if (bytes.length !== end - start) throw new Error("object_journal_missing_range");
     return bytes;
   }

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

@@ -0,0 +1,288 @@
+import { createHash } from "node:crypto";
+import { Readable } from "node:stream";
+import { setTimeout as delay } from "node:timers/promises";
+import { describe, expect, it, vi } from "vitest";
+import type { StorageProvider } from "../storage/types.js";
+import { clearObjectJournalReadCache, objectJournalReadCacheBytes } from "./object-byte-journal.js";
+import { createObjectRunLogStore } from "./object-run-log-store.js";
+
+// A live run flushes its log in many small segments, so the task page's tail
+// read (the last 32 KB, aligned to a record) crosses many objects. Each object
+// GET costs a storage round trip, so the number of GETs that must run one after
+// another decides how long the chat waits for the run's latest output.
+function countingStorage(getLatencyMs: number | ((objectKey: string) => number) = 0) {
+  const objects = new Map<string, Buffer>();
+  const stats = { gets: 0, inFlight: 0, maxInFlight: 0, completed: 0 };
+  const etag = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex");
+  const provider: StorageProvider = {
+    id: "s3",
+    async putObject() {
+      throw new Error("unconditional PUT forbidden");
+    },
+    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[] = [];
+      if (Buffer.isBuffer(input.body)) chunks.push(input.body);
+      else for await (const chunk of input.body) chunks.push(Buffer.from(chunk));
+      const bytes = Buffer.concat(chunks);
+      objects.set(input.objectKey, bytes);
+      return { etag: etag(bytes) };
+    },
+    async getObject({ objectKey }) {
+      stats.gets++;
+      stats.inFlight++;
+      stats.maxInFlight = Math.max(stats.maxInFlight, stats.inFlight);
+      try {
+        await delay(typeof getLatencyMs === "number" ? getLatencyMs : getLatencyMs(objectKey));
+        const bytes = objects.get(objectKey);
+        if (!bytes) throw Object.assign(new Error("missing"), { name: "NoSuchKey" });
+        return { stream: Readable.from([bytes]), etag: etag(bytes), contentLength: bytes.length };
+      } finally {
+        stats.inFlight--;
+        stats.completed++;
+      }
+    },
+    async headObject({ objectKey }) {
+      return { exists: objects.has(objectKey) };
+    },
+    async deleteObject({ objectKey }) {
+      objects.delete(objectKey);
+    },
+    async listObjects({ prefix }) {
+      return { keys: [...objects.keys()].filter((key) => key.startsWith(prefix)) };
+    },
+  };
+  return { provider, stats, objects };
+}
+
+// Holds every content-addressed object GET until released; the head answers.
+function holdObjectGets(provider: StorageProvider) {
+  let release!: () => void;
+  const held = new Promise<void>((resolve) => { release = resolve; });
+  const gate = { keys: [] as string[], signals: [] as Array<AbortSignal | undefined>, headReads: 0, release: () => release() };
+  const getObject = provider.getObject.bind(provider);
+  provider.getObject = async (input) => {
+    if (input.objectKey.includes("/objects/")) {
+      gate.keys.push(input.objectKey);
+      gate.signals.push(input.signal);
+      await held;
+    }
+    const result = await getObject(input);
+    if (input.objectKey.endsWith("/head.json")) gate.headReads++;
+    return result;
+  };
+  return gate;
+}
+
+const TAIL_BYTES = 32_000;
+
+async function writeManySegmentLog(
+  provider: StorageProvider,
+  records: number,
+  chunk = (seq: number) => `line ${seq} ${"x".repeat(240)}\n`,
+) {
+  const store = createObjectRunLogStore({
+    provider,
+    keyPrefix: "logs",
+    assertWriterAuthority: async () => undefined,
+  });
+  const handle = await store.begin({ companyId: "company", agentId: "agent", runId: "run" });
+  for (let seq = 1; seq <= records; seq++) {
+    // One append is one flushed segment, like a live run's frequent flushes.
+    void store.enqueue!(handle, {
+      stream: "stdout",
+      chunk: chunk(seq),
+      ts: "2026-11-09T00:00:00Z",
+      seq,
+    });
+    await store.flush!(handle);
+  }
+  const full = Buffer.from((await store.read(handle)).content);
+  return { store, handle, full };
+}
+
+describe("object run log tail reads", () => {
+  it("reads an aligned tail across many segments in parallel and without a second traversal", async () => {
+    clearObjectJournalReadCache();
+    const { provider, stats } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 400);
+    const offset = full.length - TAIL_BYTES;
+    clearObjectJournalReadCache();
+    stats.gets = 0;
+    stats.maxInFlight = 0;
+
+    const tail = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+
+    // Correctness: the tail starts at the first complete record before offset.
+    const recordStart = full.lastIndexOf(10, offset - 1) + 1;
+    expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart));
+    expect(tail.nextOffset).toBe(full.length);
+    // Every object in the aligned range is fetched once, and fetches overlap.
+    const singlePassGets = stats.gets;
+    expect(stats.maxInFlight).toBeGreaterThan(1);
+
+    clearObjectJournalReadCache();
+    stats.gets = 0;
+    const unaligned = await store.read(handle, { offset, limitBytes: TAIL_BYTES });
+    expect(Buffer.from(unaligned.content)).toEqual(full.subarray(offset));
+    // Alignment adds only a short record prefix, not a second walk of the index.
+    expect(singlePassGets).toBeLessThanOrEqual(stats.gets + 30);
+  });
+
+  it("serves a repeated tail read from immutable objects without fetching them again", async () => {
+    clearObjectJournalReadCache();
+    const { provider, stats } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 200);
+    const offset = full.length - TAIL_BYTES;
+    const first = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+    stats.gets = 0;
+
+    const second = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+
+    expect(second).toEqual(first);
+    // Only the mutable head is read again; segments and index nodes are
+    // content-addressed by their SHA-256, so they cannot change.
+    expect(stats.gets).toBe(1);
+  });
+
+  it("keeps appended output visible to a tail read after earlier reads were cached", async () => {
+    clearObjectJournalReadCache();
+    const { provider } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 50);
+    await store.read(handle, { offset: Math.max(0, full.length - TAIL_BYTES), limitBytes: TAIL_BYTES, alignStartToRecord: true });
+    void store.enqueue!(handle, { stream: "stdout", chunk: "newest line\n", ts: "2026-11-09T00:00:01Z", seq: 51 });
+    await store.flush!(handle);
+
+    const tail = await store.read(handle, { offset: full.length - 300, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+
+    expect(tail.content).toContain("newest line");
+  });
+
+  it("stops a reader that navigates away while it waits on another reader's fetch", async () => {
+    clearObjectJournalReadCache();
+    const { provider } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 60);
+    clearObjectJournalReadCache();
+    const gets = holdObjectGets(provider);
+    const offset = full.length - TAIL_BYTES / 2;
+    const leaving = new AbortController();
+
+    const staying = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+    await vi.waitFor(() => expect(gets.keys.length).toBeGreaterThan(0));
+    const joining = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true, signal: leaving.signal });
+    // Let the joining reader read the head and reach the held shared fetch.
+    await vi.waitFor(() => expect(gets.headReads).toBe(2));
+    await new Promise((resolve) => setImmediate(resolve));
+    leaving.abort(new Error("navigated away"));
+
+    // Every object GET is still held, so only the reader's own abort can end it.
+    await expect(joining).rejects.toThrow("navigated away");
+    expect(gets.signals.every((signal) => !signal?.aborted)).toBe(true);
+    gets.release();
+    const tail = await staying;
+    const recordStart = full.lastIndexOf(10, offset - 1) + 1;
+    expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart));
+    expect(new Set(gets.keys).size).toBe(gets.keys.length);
+  });
+
+  it("keeps one fetch per object when the reader that started it navigates away", async () => {
+    clearObjectJournalReadCache();
+    const { provider } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 60);
+    clearObjectJournalReadCache();
+    const gets = holdObjectGets(provider);
+    const offset = full.length - TAIL_BYTES / 2;
+    const leaving = new AbortController();
+
+    const first = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true, signal: leaving.signal });
+    await vi.waitFor(() => expect(gets.keys.length).toBeGreaterThan(0));
+    const joining = store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+    // Let the joining reader read the head and reach the held shared fetch.
+    await vi.waitFor(() => expect(gets.headReads).toBe(2));
+    await new Promise((resolve) => setImmediate(resolve));
+    leaving.abort(new Error("navigated away"));
+    await expect(first).rejects.toThrow("navigated away");
+    // The remaining reader still waits on the started fetch, so it stays open.
+    expect(gets.signals.every((signal) => !signal?.aborted)).toBe(true);
+    gets.release();
+
+    const tail = await joining;
+    const recordStart = full.lastIndexOf(10, offset - 1) + 1;
+    expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart));
+    // No object was fetched twice: the joining reader did not start its own GET.
+    expect(new Set(gets.keys).size).toBe(gets.keys.length);
+  });
+
+  it("stops a shared fetch when its last reader navigates away", async () => {
+    clearObjectJournalReadCache();
+    const { provider } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 20);
+    clearObjectJournalReadCache();
+    const gets = holdObjectGets(provider);
+    const leaving = new AbortController();
+
+    const read = store.read(handle, { offset: Math.max(0, full.length - TAIL_BYTES), limitBytes: TAIL_BYTES, signal: leaving.signal });
+    await vi.waitFor(() => expect(gets.keys.length).toBeGreaterThan(0));
+    leaving.abort(new Error("navigated away"));
+
+    await expect(read).rejects.toThrow("navigated away");
+    expect(gets.signals[0]?.aborted).toBe(true);
+    gets.release();
+  });
+
+  it.each([
+    ["just over the first lookback window", 6_000],
+    ["across several backward windows", 90_000],
+  ])("aligns a tail that starts inside a record %s", async (_label, recordBytes) => {
+    clearObjectJournalReadCache();
+    const { provider } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 6, (seq) =>
+      seq === 5 ? `long ${"y".repeat(recordBytes)}\n` : `line ${seq}\n`);
+    const longStart = full.indexOf("long ");
+    const longEnd = full.indexOf(10, longStart);
+    // Start near the end of the long record, so the 4 KB lookback has no newline.
+    const offset = longEnd - 100;
+    expect(offset - longStart).toBeGreaterThan(4096);
+    const recordStart = full.lastIndexOf(10, offset - 1) + 1;
+    clearObjectJournalReadCache();
+
+    const tail = await store.read(handle, { offset, limitBytes: TAIL_BYTES, alignStartToRecord: true });
+
+    expect(Buffer.from(tail.content)).toEqual(full.subarray(recordStart));
+    expect(tail.nextOffset).toBe(full.length);
+  });
+
+  it("counts per-entry overhead against the cache budget, not only the object bytes", async () => {
+    clearObjectJournalReadCache();
+    const { provider, objects } = countingStorage();
+    const { store, handle, full } = await writeManySegmentLog(provider, 100);
+    clearObjectJournalReadCache();
+    const gets = holdObjectGets(provider);
+    gets.release();
+
+    await store.read(handle, { offset: Math.max(0, full.length - TAIL_BYTES), limitBytes: TAIL_BYTES });
+
+    const cached = [...new Set(gets.keys)];
+    const payload = cached.reduce((sum, key) => sum + objects.get(key)!.length, 0);
+    // Many small segments: the bookkeeping per entry is a real part of memory.
+    expect(objectJournalReadCacheBytes()).toBeGreaterThanOrEqual(payload + cached.length * 256);
+    clearObjectJournalReadCache();
+    expect(objectJournalReadCacheBytes()).toBe(0);
+  });
+
+  it("does not serve one store's cached objects to a reader of another store", async () => {
+    clearObjectJournalReadCache();
+    const first = countingStorage();
+    const { handle, full } = await writeManySegmentLog(first.provider, 20);
+    const offset = Math.max(0, full.length - TAIL_BYTES);
+    // A second store has the same head, but none of the objects.
+    const second = countingStorage();
+    for (const [key, bytes] of first.objects) if (!key.includes("/objects/")) second.objects.set(key, bytes);
+    const secondStore = createObjectRunLogStore({ provider: second.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined });
+
+    await expect(secondStore.read(handle, { offset, limitBytes: TAIL_BYTES })).rejects.toThrow();
+    expect(second.stats.gets).toBeGreaterThan(1);
+  });
+});

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

@@ -8,6 +8,10 @@ import { MAX_RUN_LOG_RECORD_BYTES } from "./run-log-record-limits.js";
 const MAX_SEGMENT_BYTES = 1024 * 1024;
 const MAX_PENDING_BYTES = MAX_RUN_LOG_RECORD_BYTES;
 const MAX_READ_BYTES = 4 * MAX_SEGMENT_BYTES;
+const ALIGN_WINDOW_BYTES = 32_000;
+// Most records are short, so the first alignment window rides along with the
+// body read. Longer records continue with ALIGN_WINDOW_BYTES windows.
+const ALIGN_FIRST_LOOKBACK_BYTES = 4096;
 export interface ObjectRunLogStoreOptions {
   provider: StorageProvider;
   keyPrefix: string;
@@ -360,12 +364,20 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         limit = opts.limitBytes ?? 256_000;
       if (!Number.isSafeInteger(limit) || limit < 1 || limit > MAX_READ_BYTES)
         throw new Error("Invalid object log read limit");
-      const fetched = await store.readRange(
+      const bodyLimit = Math.min(MAX_READ_BYTES, limit + 3);
+      // Read the alignment window and the body in one range, so a tail read
+      // walks the index once instead of once for each part.
+      const lookback = opts.alignStartToRecord && offset > 0
+        ? Math.min(offset, ALIGN_FIRST_LOOKBACK_BYTES, MAX_READ_BYTES - bodyLimit)
+        : 0;
+      const window = await store.readRange(
         snapshot.head,
-        offset,
-        Math.min(MAX_READ_BYTES, limit + 3),
+        offset - lookback,
+        lookback + bodyLimit,
         opts.signal,
       );
+      const lookbackBytes = window.subarray(0, lookback);
+      const fetched = window.subarray(lookback);
       let end = Math.min(limit, fetched.length);
       while (end < fetched.length && (fetched[end]! & 0xc0) === 0x80) end++;
       if (
@@ -386,9 +398,16 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
         return { content: bytes.toString("utf8"), nextOffset };
       const parts = [bytes];
       let before = offset;
+      if (lookback > 0) {
+        const boundary = lookbackBytes.lastIndexOf(10);
+        parts.unshift(lookbackBytes.subarray(boundary + 1));
+        before = boundary >= 0 ? 0 : offset - lookback;
+        if (before > 0 && parts.reduce((n, part) => n + part.length, 0) > MAX_PENDING_BYTES)
+          throw new Error("Object log record exceeds read alignment limit");
+      }
 
       while (before > 0) {
-        const start = Math.max(0, before - 32_000);
+        const start = Math.max(0, before - ALIGN_WINDOW_BYTES);
         const prefix = await store.readRange(snapshot.head, start, before - start, opts.signal);
         const boundary = prefix.lastIndexOf(10);
         parts.unshift(prefix.subarray(boundary + 1));