feat(record): read record chains in seq pages (TASK-561)

backend/__tests__/service/podRecords.realPg.test.js

@@ -176,6 +176,36 @@ real('task record route on real PostgreSQL and MongoDB', () => {
     ]);
   });
 
+  it('serves the record in seq pages, and refuses a malformed cursor before any read', async () => {
+    expect((await claim()).status).toBe(200);
+    expect((await request(app).patch(`/api/v1/tasks/${pod._id}/TASK-1`).set(as(AGENT_TOKEN)).send({ status: 'blocked' })).status).toBe(200);
+    expect((await claim()).status).toBe(200);
+    await flushRecordWrites();
+    const asOwner = generateTestToken(owner._id);
+    const whole = await read(asOwner);
+    expect(whole.headers['x-record-next-after']).toBeUndefined();
+
+    const first = await read(asOwner, pod._id, 'TASK-1', '?limit=2');
+    expect(first.status).toBe(200);
+    expect(first.headers['x-record-next-after']).toBe('2');
+    const rest = await read(asOwner, pod._id, 'TASK-1', '?after=2&limit=2');
+    expect(rest.headers['x-record-next-after']).toBeUndefined();
+    // Both pages name the whole chain's head, and together are the whole chain.
+    expect([first, rest].map((page) => [page.headers['x-record-entries'], page.headers['x-record-head']]))
+      .toEqual([['3', whole.headers['x-record-head']], ['3', whole.headers['x-record-head']]]);
+    expect(first.text + rest.text).toBe(whole.text);
+    expect(verifyRecord(first.text + rest.text)).toMatchObject({ ok: true, entries: 3 });
+    const otlp = await read(asOwner, pod._id, 'TASK-1', '?format=otlp&after=2');
+    expect(otlp.status).toBe(200);
+    expect(otlp.headers['x-record-next-after']).toBeUndefined();
+
+    const readChain = jest.spyOn(repository, 'readChain');
+    const bad = ['?after=-1', '?after=1.5', '?after=abc', '?after=1&after=2', '?limit=0', '?limit=1001', '?limit='];
+    const answers = await Promise.all(bad.map((query) => read(asOwner, pod._id, 'TASK-1', query)));
+    expect(answers.map((res) => [res.status, res.body.error])).toEqual(bad.map(() => [400, 'record_bad_cursor']));
+    expect(readChain).not.toHaveBeenCalled();
+  });
+
   it('refuses everyone else before any record byte is read', async () => {
     await claim();
     await flushRecordWrites();

backend/__tests__/service/recordChains.realPg.test.js

@@ -10,7 +10,7 @@ const path = require('path');
 const { runPgMigrations } = require('../../config/run-pg-migrations');
 const { createRealPgSchema } = require('../utils/realPgSchema');
 const { appendRecord, readRecordExport, BOARD_ACTOR } = require('../../services/recordService');
-const { appendToChain } = require('../../repositories/recordChainRepository');
+const { appendToChain, MAX_PAGE_LINES } = require('../../repositories/recordChainRepository');
 const { verifyRecord } = require('../../vendor/seats/src/record/format');
 
 const real = process.env.INTEGRATION_TEST === 'true' ? describe : describe.skip;
@@ -150,11 +150,41 @@ real('record chains on real PostgreSQL', () => {
 
     it('reads an unwritten chain as empty, and a missing store as an error', async () => {
       expect(await readRecordExport({ podId: POD, taskId: 'TASK-never' }, pool))
-        .toEqual({ runId: null, entries: 0, head: null, jsonl: '' });
+        .toEqual({
+          runId: null, entries: 0, head: null, jsonl: '', nextAfter: null,
+        });
       const { readChain, RecordStoreUnavailableError } = require('../../repositories/recordChainRepository');
       await expect(readChain({ podId: POD }, null)).rejects.toBeInstanceOf(RecordStoreUnavailableError);
     });
 
+    it('reads a long chain in pages that each name the whole chain\'s head and join into one that verifies', async () => {
+      const ref = { podId: POD, taskId: 'TASK-7' };
+      const total = MAX_PAGE_LINES + 2;
+      await appendRecord(ref, 'bulk', Array.from({ length: total }, () => ({ kind: 'task.claimed', actor: AGENT })), pool);
+
+      // No cursor still reads one bounded page, never the whole history.
+      const first = await readRecordExport(ref, pool);
+      expect(kinds(first.jsonl)).toHaveLength(MAX_PAGE_LINES);
+      expect(first).toMatchObject({ entries: total, nextAfter: MAX_PAGE_LINES });
+      expect(kinds((await readRecordExport(ref, pool, { limit: MAX_PAGE_LINES * 5 })).jsonl)).toHaveLength(MAX_PAGE_LINES);
+
+      const pages = [];
+      let after = 0;
+      while (after !== null) {
+        // eslint-disable-next-line no-await-in-loop
+        const page = await readRecordExport(ref, pool, { after, limit: 400 });
+        expect(page).toMatchObject({ entries: total, head: first.head, runId: first.runId });
+        pages.push(page.jsonl);
+        after = page.nextAfter;
+      }
+      expect(pages.map((jsonl) => kinds(jsonl).length)).toEqual([400, 400, total - 800]);
+      expect(kinds(pages[1])[0].seq).toBe(401);
+      expect(verifyRecord(pages.join(''))).toMatchObject({ ok: true, entries: total, head: first.head });
+
+      expect(await readRecordExport(ref, pool, { after: total, limit: 10 }))
+        .toMatchObject({ entries: total, jsonl: '', nextAfter: null });
+    });
+
     it('stores lines byte for byte, so an edited row fails verification', async () => {
       const ref = { podId: POD, taskId: 'TASK-6' };
       await appendRecord(ref, 'a', [{ kind: 'task.claimed', actor: AGENT }], pool);

backend/__tests__/unit/routes/podRecords.otlpFullChain.test.js

@@ -0,0 +1,72 @@
+/**
+ * The record route pages the NDJSON read, but ?format=otlp has no paging of
+ * its own. An OTLP read with no cursor must cover the whole chain, not the
+ * first page, or a long chain exports a silently short trace.
+ */
+const express = require('express');
+const request = require('supertest');
+
+const POD = 'aaaaaaaaaaaaaaaaaaaaaa01';
+const USER = 'bbbbbbbbbbbbbbbbbbbbbbb1';
+
+const mockReadRecordExport = jest.fn();
+const mockToOtlp = jest.fn(() => ({ resourceSpans: [] }));
+
+jest.mock('../../../services/recordService', () => ({
+  isRecordEnabled: () => true,
+  readRecordExport: (...args) => mockReadRecordExport(...args),
+  toOtlp: (...args) => mockToOtlp(...args),
+  RecordStoreUnavailableError: class RecordStoreUnavailableError extends Error {},
+}));
+jest.mock('../../../middleware/dualAuth', () => (req, res, next) => {
+  req.userId = USER;
+  next();
+});
+jest.mock('../../../models/Pod', () => ({
+  findById: () => ({ select: () => ({ lean: async () => ({ _id: POD, type: 'chat', members: [USER] }) }) }),
+}));
+jest.mock('../../../models/Task', () => ({ findOne: jest.fn() }));
+jest.mock('../../../services/dmService', () => ({ canViewPod: async () => true }));
+jest.mock('../../../middleware/ipRateLimit', () => ({
+  cloudflareIpRateLimitKeyGenerator: () => 'test',
+}));
+
+const router = require('../../../routes/podRecords');
+
+const app = express();
+app.use('/api/pods', router);
+
+// Three pages of one line each, as readRecordExport returns them.
+const pages = {
+  start: { jsonl: '{"seq":1}\n', nextAfter: 1 },
+  1: { jsonl: '{"seq":2}\n', nextAfter: 2 },
+  2: { jsonl: '{"seq":3}\n', nextAfter: null },
+};
+
+beforeEach(() => {
+  jest.clearAllMocks();
+  mockReadRecordExport.mockImplementation(async (_ref, _db, page = {}) => (
+    page.after === undefined ? pages.start : pages[page.after]
+  ));
+});
+
+describe('GET /api/pods/:podId/record?format=otlp', () => {
+  test('an OTLP read with no cursor exports every page of the chain', async () => {
+    const res = await request(app).get(`/api/pods/${POD}/record?format=otlp`);
+    expect(res.status).toBe(200);
+    expect(mockToOtlp).toHaveBeenCalledWith('{"seq":1}\n{"seq":2}\n{"seq":3}\n');
+  });
+
+  test('a paged NDJSON read still returns one page with its cursor', async () => {
+    const res = await request(app).get(`/api/pods/${POD}/record`);
+    expect(res.status).toBe(200);
+    expect(res.text).toBe('{"seq":1}\n');
+    expect(res.headers['x-record-next-after']).toBe('1');
+    expect(mockReadRecordExport).toHaveBeenCalledTimes(1);
+  });
+
+  test('an OTLP read with an explicit cursor stays one page', async () => {
+    await request(app).get(`/api/pods/${POD}/record?format=otlp&after=1`);
+    expect(mockToOtlp).toHaveBeenCalledWith('{"seq":2}\n');
+  });
+});

backend/repositories/recordChainRepository.ts

@@ -24,8 +24,20 @@ export type StoredChain = {
   headSeq: number;
   headHash: string | null;
   lines: string[];
+  /** The seq to pass as `after` for the next page; null when this page reached the head. */
+  nextAfter: number | null;
 };
 
+/** The most lines one read returns. A chain longer than this is read in pages. */
+export const MAX_PAGE_LINES = 1000;
+
+/**
+ * A seq cursor: the lines after seq `after`, at most `limit` of them. A page's
+ * first line carries `prev`, the hash of line `after`, so pages read in order
+ * concatenate into one chain that still verifies.
+ */
+export type ChainPage = { after?: number; limit?: number };
+
 /** Thrown when no PostgreSQL is configured; a reader must not answer "empty". */
 export class RecordStoreUnavailableError extends Error {
   constructor() {
@@ -111,25 +123,33 @@ export async function appendToChain(
 }
 
 /**
- * Read a chain's lines up to its stored head. Lines and head commit together,
- * so bounding by the head read first gives a consistent prefix without a
- * transaction. A chain that was never written returns null.
+ * Read one page of a chain's lines, bounded by its stored head. Lines and head
+ * commit together, so bounding by the head read first gives a consistent
+ * prefix without a transaction. A chain that was never written returns null.
  */
-export async function readChain(ref: ChainRef, db: PgPool | null = pool): Promise<StoredChain | null> {
+export async function readChain(
+  ref: ChainRef,
+  db: PgPool | null = pool,
+  page: ChainPage = {},
+): Promise<StoredChain | null> {
   const store = requirePool(db);
+  const after = Math.max(0, Math.trunc(Number(page.after) || 0));
+  const limit = Math.min(MAX_PAGE_LINES, Math.max(1, Math.trunc(Number(page.limit) || MAX_PAGE_LINES)));
   const chain = (await store.query(
     'SELECT id, run_id, head_seq, head_hash FROM record_chains WHERE pod_id = $1 AND task_id = $2',
     [ref.podId, taskKey(ref)],
   )).rows[0] as { id: string; run_id: string; head_seq: number; head_hash: string | null } | undefined;
   if (!chain) return null;
   const rows = (await store.query(
-    'SELECT line FROM record_lines WHERE chain_id = $1 AND seq <= $2 ORDER BY seq',
-    [chain.id, chain.head_seq],
-  )).rows as Array<{ line: string }>;
+    'SELECT seq, line FROM record_lines WHERE chain_id = $1 AND seq > $2 AND seq <= $3 ORDER BY seq LIMIT $4',
+    [chain.id, after, chain.head_seq, limit],
+  )).rows as Array<{ seq: number; line: string }>;
+  const last = rows.length > 0 ? Number(rows[rows.length - 1].seq) : after;
   return {
     runId: chain.run_id,
     headSeq: chain.head_seq,
     headHash: chain.head_hash,
     lines: rows.map((row) => row.line),
+    nextAfter: rows.length > 0 && last < chain.head_seq ? last : null,
   };
 }

backend/routes/podRecords.ts

@@ -2,8 +2,9 @@ import express from 'express';
 import rateLimit from 'express-rate-limit';
 import mongoose from 'mongoose';
 import { cloudflareIpRateLimitKeyGenerator } from '../middleware/ipRateLimit';
+import { MAX_PAGE_LINES } from '../repositories/recordChainRepository';
 import {
-  isRecordEnabled, readRecordExport, RecordStoreUnavailableError, toOtlp,
+  isRecordEnabled, readRecordExport, RecordStoreUnavailableError, toOtlp, type ChainPage,
 } from '../services/recordService';
 
 /* eslint-disable @typescript-eslint/no-require-imports, global-require */
@@ -30,6 +31,19 @@ type RecordReq = express.Request & {
 
 const notFound = (res: express.Response) => res.status(404).json({ error: 'record_not_found' });
 
+// A page cursor: `after` is a seq (0 is the start), `limit` 1..MAX_PAGE_LINES.
+const SEQ = /^\d{1,9}$/;
+const readPage = (query: express.Request['query']): ChainPage | null => {
+  const { after, limit } = query;
+  if (after !== undefined && (typeof after !== 'string' || !SEQ.test(after))) return null;
+  if (limit !== undefined && (typeof limit !== 'string' || !SEQ.test(limit)
+    || Number(limit) < 1 || Number(limit) > MAX_PAGE_LINES)) return null;
+  return {
+    ...(after !== undefined ? { after: Number(after) } : {}),
+    ...(limit !== undefined ? { limit: Number(limit) } : {}),
+  };
+};
+
 /**
  * A chain as record-v1 JSONL, or `?format=otlp` for the spec's OTLP/JSON
  * export. `taskId` null is the pod's taskless chain (grants and the calls
@@ -41,11 +55,17 @@ const notFound = (res: express.Response) => res.status(404).json({ error: 'recor
  * credential-owner ids verbatim, and a hash-chained line can't be redacted
  * without breaking verification. That is the same set runTraces shows those
  * ids to, so admins and agent-DM observers who pass canViewPod get a 403.
+ *
+ * One read returns at most MAX_PAGE_LINES lines after `?after=<seq>`. While
+ * more remain, `X-Record-Next-After` names the cursor for the next page; the
+ * head headers always describe the whole chain.
  */
 const serveRecord = async (req: RecordReq, res: express.Response, taskId: string | null) => {
   try {
     const { podId } = req.params;
     if (!mongoose.isObjectIdOrHexString(podId) || (taskId !== null && !TASK_ID.test(taskId))) return notFound(res);
+    const page = readPage(req.query);
+    if (!page) return res.status(400).json({ error: 'record_bad_cursor' });
     if (!isRecordEnabled(podId)) return notFound(res);
     const userId = req.userId || req.user?._id || req.agentUser?._id;
     const pod = await Pod.findById(podId).select('members type').lean();
@@ -57,7 +77,19 @@ const serveRecord = async (req: RecordReq, res: express.Response, taskId: string
 
     let record;
     try {
-      record = await readRecordExport({ podId: String(pod._id), ...(taskId !== null ? { taskId } : {}) });
+      const ref = { podId: String(pod._id), ...(taskId !== null ? { taskId } : {}) };
+      record = await readRecordExport(ref, undefined, page);
+      // OTLP has no paging of its own: an export of one page would be a
+      // silently short trace. An OTLP read with no cursor covers the chain.
+      if (req.query.format === 'otlp' && page.after === undefined && page.limit === undefined) {
+        const lines = [record.jsonl];
+        for (let next = record.nextAfter; next !== null;) {
+          const more = await readRecordExport(ref, undefined, { after: next });
+          lines.push(more.jsonl);
+          next = more.nextAfter;
+        }
+        record = { ...record, jsonl: lines.join(''), nextAfter: null };
+      }
     } catch (error) {
       // A store outage is an error, never an apparently empty record.
       if (!(error instanceof RecordStoreUnavailableError)) {
@@ -71,6 +103,7 @@ const serveRecord = async (req: RecordReq, res: express.Response, taskId: string
     // The spec's outside record of the head (§4): it covers the last line.
     if (record.head) res.set('X-Record-Head', record.head);
     if (record.runId) res.set('X-Record-Run-Id', record.runId);
+    if (record.nextAfter !== null) res.set('X-Record-Next-After', String(record.nextAfter));
     if (req.query.format === 'otlp') return res.json(toOtlp(record.jsonl));
     res.type('application/x-ndjson; charset=utf-8');
     return res.send(record.jsonl);

backend/services/recordService.ts

@@ -13,7 +13,7 @@
  */
 import { createHash } from 'crypto';
 import {
-  appendToChain, readChain, RecordStoreUnavailableError, type ChainRef,
+  appendToChain, readChain, RecordStoreUnavailableError, type ChainPage, type ChainRef,
 } from '../repositories/recordChainRepository';
 
 type Store = Parameters<typeof readChain>[1];
@@ -42,7 +42,7 @@ export type RecordLineInput = {
   actor?: RecordActor | null;
   data?: Record<string, unknown>;
 };
-export { RecordStoreUnavailableError, type ChainRef };
+export { RecordStoreUnavailableError, type ChainPage, type ChainRef };
 
 export const BOARD_ACTOR: RecordActor = format.BOARD_ACTOR;
 /**
@@ -190,17 +190,24 @@ export type RecordExport = {
   entries: number;
   head: string | null;
   jsonl: string;
+  /** Pass as `after` for the next page; null once a page reaches the head. */
+  nextAfter: number | null;
 };
 
-/** A chain as stored, in record-v1's file form: one line per entry, each ending in \n. */
-export async function readRecordExport(ref: ChainRef, db?: Store): Promise<RecordExport> {
-  const chain = await readChain(ref, db);
-  if (!chain) return { runId: null, entries: 0, head: null, jsonl: '' };
+/**
+ * One page of a chain as stored, in record-v1's file form: one line per entry,
+ * each ending in \n. `entries` and `head` describe the whole chain, not the
+ * page, so every page names the same head.
+ */
+export async function readRecordExport(ref: ChainRef, db?: Store, page: ChainPage = {}): Promise<RecordExport> {
+  const chain = await readChain(ref, db, page);
+  if (!chain) return { runId: null, entries: 0, head: null, jsonl: '', nextAfter: null };
   return {
     runId: chain.runId,
-    entries: chain.lines.length,
+    entries: chain.headSeq,
     head: chain.headHash,
     jsonl: chain.lines.map((line) => `${line}\n`).join(''),
+    nextAfter: chain.nextAfter,
   };
 }
 

docs/plans/task-513-server-record.md

@@ -83,6 +83,20 @@ endpoint. Every writer and the endpoint are off unless the pod is listed in
 - **Reads.** `GET /api/pods/:podId/record` serves the pod's taskless chain
   under the same gate as the task route.
 
+### Decisions taken in slice 3 (TASK-561)
+
+- **Reads are paged by seq.** One read returns at most 1000 lines
+  (`MAX_PAGE_LINES`) after `?after=<seq>`, fewer with `?limit=`. A read with
+  no cursor is the first page, never the whole history. While lines remain,
+  `X-Record-Next-After` names the next cursor. `X-Record-Entries` and
+  `X-Record-Head` describe the whole chain on every page, so a client knows
+  where the chain ends before it reads that far. A page's first line carries
+  `prev`, the hash of line `after`. Pages read in order join into one chain
+  that `verifyRecord` accepts, and a single page on its own does not verify
+  unless it starts at seq 1. A malformed cursor gets a 400 before any record
+  is read. The v2 view (slice 4) calls the API from another origin. It can
+  read these headers only after CORS exposes them.
+
 ## Measured inputs (2026-10-25)
 
 - TASK-504 landed in #2684, merge `ab5132313ad399d8fb990d69bc39db0ccf99e055`.