From 8c46b64ef1ccd60d00ae34472f881a68af3f8fac Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:48:31 +0100 Subject: [PATCH 1/3] Page back to a bounded read's window instead of filtering one page Zulip's GET /messages selects a page by anchor and count and has no timestamp predicate, so since and until never reached the realm: they filtered whatever a newest-anchored page happened to return. A window lying below that page came back empty while looking authoritative. readWindow walks back one anchored page at a time until the window's lower bound is crossed, the cap is met, or history runs out. It is sequential by construction, dedupes by id because a Zulip anchor is a range hint that returns a neighbour rather than an exact match, and maps each page against the query that fetched it, which is what the rendered-content read has to re-issue. readChannel, readThread and the boot catch-up path all go through it. An unbounded read still asks for one page and gets the newest limit. --- packages/zulip/adapter.test.ts | 164 ++++++++++++++++++++++ packages/zulip/adapter.ts | 136 +++++++++--------- packages/zulip/history-paging.test.ts | 192 ++++++++++++++++++++++++++ packages/zulip/history-paging.ts | 146 ++++++++++++++++++++ 4 files changed, 576 insertions(+), 62 deletions(-) create mode 100644 packages/zulip/history-paging.test.ts create mode 100644 packages/zulip/history-paging.ts diff --git a/packages/zulip/adapter.test.ts b/packages/zulip/adapter.test.ts index aff6462..3bb0eb9 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -2169,6 +2169,147 @@ effectTest('history.readChannel filters by range.until (epoch seconds, inclusive }), ) +const historyRow = (id: number, subject = 'a'): Record => ({ + id, + sender_id: 5, + sender_full_name: 'Robin Reyes', + stream_id: 1234, + display_recipient: 'general', + subject, + content: `m${id}`, + timestamp: 1714000000 + id * 100, +}) + +const rowTs = (id: number) => decodeTimestampSync(1714000000 + id * 100) + +/** + * Answer consecutive `/messages` reads with the given pages, then with an + * empty one. Zulip's anchor is inclusive, so a page anchored on the id below + * the last one re-offers that row — the overlap these pages reproduce. + */ +const seedMessagePages = ( + stub: StubHttpClient, + pages: ReadonlyArray>>, +): Effect.Effect => + stub + .respond('GET', '/api/v1/messages', messagesPage([])) + .pipe(Effect.zipRight(stub.respondSequence('GET', '/api/v1/messages', pages.map(messagesPage)))) + +const historyAnchors = (stub: StubHttpClient) => + stub.captured.pipe( + Effect.map((reqs) => + reqs + .filter((r) => r.method === 'GET' && r.url.pathname === '/api/v1/messages') + .map((r) => r.url.searchParams.get('anchor')), + ), + ) + +// The defect this fix removes. Every message in the window sits below a page +// selected purely by recency, so filtering that page empties it and the read +// returns an authoritative-looking nothing. +effectTest('history.readChannel pages back to a window that lies below the newest page', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedUsers(stub, [HERMES, MAINTAINER]) + yield* seedMessagePages(stub, [ + [7, 8, 9].map((id) => historyRow(id)), + [4, 5, 6, 7].map((id) => historyRow(id)), + [1, 2, 3, 4].map((id) => historyRow(id)), + ]) + const messages = yield* adapter.history.readChannel(generalChannel.name, { + until: rowTs(3), + limit: 3, + }) + expect(messages.map((m) => m.body)).toEqual([ + decodeMessageBodySync('m1'), + decodeMessageBodySync('m2'), + decodeMessageBodySync('m3'), + ]) + expect(yield* historyAnchors(stub)).toEqual(['newest', '7', '4']) + }), +) + +effectTest('history.readThread pages back to a window that lies below the newest page', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedUsers(stub, [HERMES, MAINTAINER]) + yield* seedMessagePages(stub, [ + [7, 8, 9].map((id) => historyRow(id, 'planning')), + [4, 5, 6, 7].map((id) => historyRow(id, 'planning')), + [1, 2, 3, 4].map((id) => historyRow(id, 'planning')), + ]) + const messages = yield* adapter.history.readThread( + generalChannel.name, + decodeThreadNameSync('planning'), + { until: rowTs(3), limit: 3 }, + ) + expect(messages.map((m) => m.body)).toEqual([ + decodeMessageBodySync('m1'), + decodeMessageBodySync('m2'), + decodeMessageBodySync('m3'), + ]) + }), +) + +// A row Zulip re-offers on the next page because the anchor is inclusive must +// not be counted twice, and the window must come back in ascending order. +effectTest('history.readChannel collects a row that spans two pages exactly once', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedUsers(stub, [HERMES, MAINTAINER]) + yield* seedMessagePages(stub, [ + [7, 8, 9].map((id) => historyRow(id)), + [4, 5, 6, 7].map((id) => historyRow(id)), + [1, 2, 3, 4].map((id) => historyRow(id)), + ]) + const messages = yield* adapter.history.readChannel(generalChannel.name, { + since: rowTs(1), + limit: 50, + }) + expect(messages.map((m) => m.body)).toEqual( + [1, 2, 3, 4, 5, 6, 7, 8, 9].map((id) => decodeMessageBodySync(`m${id}`)), + ) + }), +) + +// A read with no bounds is answered by the newest page and nothing more — +// the walk exists to reach a window, and there is no window to reach. +effectTest('history.readChannel with no bounds reads a single page', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedUsers(stub, [HERMES, MAINTAINER]) + yield* seedMessagePages(stub, [[7, 8, 9].map((id) => historyRow(id))]) + const messages = yield* adapter.history.readChannel(generalChannel.name, { limit: 3 }) + expect(messages).toHaveLength(3) + expect(yield* historyAnchors(stub)).toEqual(['newest']) + }), +) + +// `limit` stays a cap that truncates from the old end: the walk stops once it +// holds that many, and keeps the newest of them. +effectTest('history.readChannel keeps the newest messages when the window overruns limit', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedUsers(stub, [HERMES, MAINTAINER]) + yield* seedMessagePages(stub, [ + [7, 8, 9].map((id) => historyRow(id)), + [4, 5, 6, 7].map((id) => historyRow(id)), + ]) + const messages = yield* adapter.history.readChannel(generalChannel.name, { + since: rowTs(1), + limit: 4, + }) + expect(messages.map((m) => m.body)).toEqual( + [6, 7, 8, 9].map((id) => decodeMessageBodySync(`m${id}`)), + ) + }), +) + effectTest('history.readThread narrows by both channel and topic', () => Effect.gen(function* () { const stub = yield* makeStubHttpClient @@ -2640,6 +2781,26 @@ effectTest('inbox.replay(since) returns message-posted events for messages with }), ) +// Boot catch-up carries the same defect independently: a seat offline across +// more realm traffic than one page holds silently lost the overflow. +effectTest('inbox.replay pages back until it reaches the catch-up bound', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedUsers(stub, [HERMES, MAINTAINER]) + yield* seedMessagePages(stub, [ + [7, 8, 9].map((id) => historyRow(id, 'lobby')), + [4, 5, 6, 7].map((id) => historyRow(id, 'lobby')), + [1, 2, 3, 4].map((id) => historyRow(id, 'lobby')), + ]) + const events = yield* adapter.inbox.replay(rowTs(2)) + expect(events.flatMap((e) => (e.kind === 'message-posted' ? [e.message.body] : []))).toEqual( + [2, 3, 4, 5, 6, 7, 8, 9].map((id) => decodeMessageBodySync(`m${id}`)), + ) + expect(yield* historyAnchors(stub)).toEqual(['newest', '7', '4']) + }), +) + effectTest('inbox.replay calls /messages with anchor=newest and a generous num_before', () => Effect.gen(function* () { const stub = yield* makeStubHttpClient @@ -2730,6 +2891,9 @@ effectTest( history_limited: false, }, }) + // Catch-up walks back until a page adds nothing new, so the realm has + // to keep answering past the raw page and its rendered twin. + yield* stub.respond('GET', '/api/v1/messages', messagesPage([])) yield* stub.respondSequence('GET', '/api/v1/messages', [ replayPage('@**hermes-agent** wake up'), replayPage( diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index a24f68c..b5f3c7d 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -67,6 +67,7 @@ import { Duration, Effect, Equivalence, + Fiber, HashMap, HashSet, Option, @@ -92,6 +93,7 @@ import { registerQueue, zulipMessageContentSchema, } from './events.ts' +import { readWindow } from './history-paging.ts' import type { ApiKey as ApiKeyType, BotEmail as BotEmailType, @@ -441,6 +443,13 @@ const RECENT_THREADS_DEFAULT_LIMIT = 10 const RECENT_THREADS_FETCH_LIMIT = 50 const HISTORY_DEFAULT_LIMIT = 100 +/** + * Requests one bounded read may spend walking back to its window. A window + * far enough back to outrun this returns a short read rather than paging the + * whole realm; `limit` already stops the common case well before it. + */ +const HISTORY_MAX_PAGES = 20 + /** * The slice of Zulip's `/register` initial state that carries the realm-wide * editing switch. Zulip exposes realm settings on no GET endpoint at all @@ -478,14 +487,6 @@ type NarrowFilter = // port-facing ThreadName. | { readonly operator: 'topic'; readonly operand: string } -const inRange = - (range: Range) => - (m: HistoricalMessage): boolean => { - if (range.since !== undefined && m.ts < range.since) return false - if (range.until !== undefined && m.ts > range.until) return false - return true - } - const toIdentity = (u: ZulipUser): Effect.Effect => Effect.all({ id: decodeIdentityId(String(u.user_id)), @@ -822,29 +823,37 @@ export const zulipAdapter = ( narrow: ReadonlyArray, ): Effect.Effect, ZulipApiError | ParseResult.ParseError> => Effect.gen(function* () { - const historyQuery = { - anchor: 'newest', - num_before: range.limit ?? HISTORY_DEFAULT_LIMIT, - num_after: 0, - narrow: JSON.stringify(narrow), - } as const - const [directory, res] = yield* Effect.all( - [ - buildDirectoryLookup(), - minterHttp.get('/messages', messagesResponseSchema, { - ...historyQuery, - apply_markdown: false, + const cap = range.limit ?? HISTORY_DEFAULT_LIMIT + const bounded = range.since !== undefined || range.until !== undefined + // The directory read is independent of the walk, so it runs alongside + // the first page rather than in front of it. Joined on the first page + // that has anything to map. + const directoryFiber = yield* Effect.fork(buildDirectoryLookup()) + return yield* readWindow({ + query: { narrow: JSON.stringify(narrow), apply_markdown: false }, + window: { since: range.since, until: range.until }, + // An unbounded read is answered by the newest page, so its page is + // exactly the cap. A bounded one may have to walk to reach its + // window, and every page it walks past costs a request. + pageSize: bounded ? Math.max(cap, HISTORY_DEFAULT_LIMIT) : cap, + cap, + maxPages: bounded ? HISTORY_MAX_PAGES : 1, + fetchPage: (query) => + minterHttp + .get('/messages', messagesResponseSchema, query) + .pipe(Effect.map((res) => res.messages)), + // One rendered read per page, or none at all — never one per + // mention-bearing message. See renderedContentForBatch. + onPage: (query, rows) => + Effect.gen(function* () { + const renderedFor = yield* renderedContentForBatch(minterHttp, query, rows) + const historical = yield* Effect.forEach(rows, toHistoricalMessage) + const directory = yield* Fiber.join(directoryFiber) + return yield* Effect.forEach(historical, (m) => + mapHistoricalMessage(m, directory, renderedFor), + ) }), - ], - { concurrency: 2 }, - ) - // One rendered read for the whole batch, or none at all — never one - // per mention-bearing message. See renderedContentForBatch. - const renderedFor = yield* renderedContentForBatch(minterHttp, historyQuery, res.messages) - const historical = yield* Effect.forEach(res.messages, toHistoricalMessage) - return yield* Effect.forEach(historical.filter(inRange(range)), (m) => - mapHistoricalMessage(m, directory, renderedFor), - ) + }) }) // Zulip constructs bot delivery emails as `-bot@` @@ -2023,40 +2032,43 @@ export const zulipAdapter = ( // subject) never sees PM-shaped rows — any DM in the minter's // recent history would otherwise crash the schema decode. const replayNarrow = JSON.stringify([{ negated: true, operator: 'is', operand: 'dm' }]) - const replayQuery = { - anchor: 'newest', - num_before: REPLAY_NUM_BEFORE, - num_after: 0, - narrow: replayNarrow, - } as const return Effect.gen(function* () { - const [directory, res, current] = yield* Effect.all( - [ - buildDirectoryLookup(), - minterHttp.get('/messages', replayResponseSchema, { - ...replayQuery, - apply_markdown: false, + const [directoryFiber, current] = yield* Effect.all([ + Effect.fork(buildDirectoryLookup()), + SynchronizedRef.get(boundRef), + ]) + // Catch-up has no cap of its own: a seat that was away for a week + // has to be told everything it missed, however many pages that + // spans. + const perMessage = yield* readWindow({ + query: { narrow: replayNarrow, apply_markdown: false }, + window: { since }, + pageSize: REPLAY_NUM_BEFORE, + cap: undefined, + maxPages: HISTORY_MAX_PAGES, + fetchPage: (query) => + minterHttp + .get('/messages', replayResponseSchema, query) + .pipe(Effect.map((res) => res.messages)), + // Catch-up is the burst case — a fleet bounce replaying a + // mention-heavy window. One rendered read covers a whole page, + // and it has to be that page's own query. + onPage: (query, rows) => + Effect.gen(function* () { + const renderedFor = yield* renderedContentForBatch(minterHttp, query, rows) + const directory = yield* Fiber.join(directoryFiber) + return yield* Effect.forEach(rows, (raw) => { + const { flags: _flags, ...message } = raw + return messageToInboundEvents( + message, + directory, + Option.getOrUndefined(current)?.identity, + base, + renderedFor, + ) + }) }), - SynchronizedRef.get(boundRef), - ], - { concurrency: 2 }, - ) - // Catch-up is the burst case — a fleet bounce replaying a - // mention-heavy window. One rendered read covers the whole window. - const renderedFor = yield* renderedContentForBatch(minterHttp, replayQuery, res.messages) - const perMessage = yield* Effect.forEach( - res.messages.filter((raw) => raw.timestamp >= since), - (raw) => { - const { flags: _flags, ...message } = raw - return messageToInboundEvents( - message, - directory, - Option.getOrUndefined(current)?.identity, - base, - renderedFor, - ) - }, - ) + }) const out: InboundEvent[] = [] for (const mapped of perMessage) { for (const ev of mapped) { diff --git a/packages/zulip/history-paging.test.ts b/packages/zulip/history-paging.test.ts new file mode 100644 index 0000000..7a5079a --- /dev/null +++ b/packages/zulip/history-paging.test.ts @@ -0,0 +1,192 @@ +import { expect } from 'bun:test' +import { effectTest } from '@commy/testing/effect-test' +import { Effect, Ref } from 'effect' +import { readWindow } from './history-paging.ts' +import type { ZulipParams, ZulipParamValue } from './http.ts' + +type Row = { readonly id: number; readonly timestamp: number } + +const rowsUpTo = (count: number): ReadonlyArray => + Array.from({ length: count }, (_, i) => ({ id: i + 1, timestamp: 1000 + (i + 1) * 100 })) + +type FakeRealm = { + readonly fetchPage: (query: ZulipParams) => Effect.Effect> + readonly anchors: Effect.Effect> +} + +/** + * A realm answering `anchor` + `num_before` the way Zulip does: `newest` + * takes the newest `num_before` rows, an id takes that row plus `num_before` + * older ones. Rows come back oldest first, and consecutive pages therefore + * overlap on the anchor row. + */ +const fakeRealm = (rows: ReadonlyArray): Effect.Effect => + Effect.gen(function* () { + const seen = yield* Ref.make>([]) + const ascending = [...rows].sort((a, b) => a.id - b.id) + return { + anchors: Ref.get(seen), + fetchPage: (query) => + Ref.update(seen, (all) => [...all, query['anchor'] ?? 'newest']).pipe( + Effect.as( + (() => { + const size = Number(query['num_before']) + if (query['anchor'] === 'newest') return ascending.slice(-size) + const anchorIndex = ascending.findIndex((r) => r.id === Number(query['anchor'])) + if (anchorIndex === -1) return [] + return ascending.slice(Math.max(0, anchorIndex - size), anchorIndex + 1) + })(), + ), + ), + } + }) + +const walk = ( + realm: FakeRealm, + options: { + readonly window?: { readonly since?: number; readonly until?: number } + readonly pageSize?: number + readonly cap?: number + readonly maxPages?: number + readonly onPage?: ( + query: ZulipParams, + rows: ReadonlyArray, + ) => Effect.Effect> + } = {}, +) => + readWindow({ + query: { narrow: '[]' }, + window: options.window ?? {}, + pageSize: options.pageSize ?? 3, + cap: options.cap, + maxPages: options.maxPages ?? 10, + fetchPage: realm.fetchPage, + onPage: options.onPage ?? ((_query, rows) => Effect.succeed(rows.map((r) => r.id))), + }) + +// The defect this module exists to remove. Every message in the window sits +// below a page selected purely by recency, so a post-filter on that page +// returns nothing while looking authoritative. +effectTest('a window below the newest page is reached by paging back to it', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + expect(yield* walk(realm, { window: { until: 1300 }, cap: 3 })).toEqual([1, 2, 3]) + }), +) + +// Each page is mapped against the query that fetched it, because the +// rendered-content read re-issues that same query. +effectTest('a page is mapped with the query that fetched it', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + const queries = yield* Ref.make>([]) + yield* walk(realm, { + window: { since: 1400 }, + onPage: (query, rows) => + Ref.update(queries, (all) => [...all, query['anchor'] ?? 'newest']).pipe( + Effect.as(rows.map((r) => r.id)), + ), + }) + expect(yield* Ref.get(queries)).toEqual(['newest', 7]) + }), +) + +// A page holding nothing in the window is never mapped, so it costs no +// rendered-content read. +effectTest('a page entirely outside the window is not mapped', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + const mappedPages = yield* Ref.make(0) + yield* walk(realm, { + window: { until: 1300 }, + cap: 3, + onPage: (_query, rows) => + Ref.update(mappedPages, (n) => n + 1).pipe(Effect.as(rows.map((r) => r.id))), + }) + expect(yield* Ref.get(mappedPages)).toBe(1) + expect(yield* realm.anchors).toEqual(['newest', 7, 4]) + }), +) + +// Zulip's anchor is inclusive, so consecutive pages share their boundary row. +effectTest('a row that appears on two pages is collected once', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + expect(yield* walk(realm, { window: { since: 1100 } })).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9]) + }), +) + +// The anchor is a range hint, not an exact match — a range query around a +// deleted id returns its neighbours instead. The next anchor has to come from +// what actually returned, never from arithmetic on what was asked for. +effectTest('the next anchor comes from the rows that returned, not the one requested', () => + Effect.gen(function* () { + const realm = yield* fakeRealm([ + { id: 2, timestamp: 1200 }, + { id: 3, timestamp: 1300 }, + { id: 7, timestamp: 1700 }, + { id: 8, timestamp: 1800 }, + { id: 9, timestamp: 1900 }, + ]) + const ids = yield* walk(realm, { window: { since: 1000 } }) + expect(yield* realm.anchors).toEqual(['newest', 7, 2]) + expect(ids).toEqual([2, 3, 7, 8, 9]) + }), +) + +// An anchor naming no live message answers success with an empty list rather +// than an error, so an exhausted walk ends quietly. +effectTest('an empty page ends the walk', () => + Effect.gen(function* () { + const realm = yield* fakeRealm([]) + expect(yield* walk(realm, { window: { since: 1000 } })).toEqual([]) + expect(yield* realm.anchors).toEqual(['newest']) + }), +) + +effectTest('a page that adds nothing new ends the walk', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(2)) + yield* walk(realm, { window: { since: 1000 } }) + // The second page re-offers the row the first already had, and there is + // nothing older behind it. + expect(yield* realm.anchors).toEqual(['newest', 1]) + }), +) + +effectTest('crossing the lower bound ends the walk without reading further', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + const ids = yield* walk(realm, { window: { since: 1600 } }) + expect(yield* realm.anchors).toEqual(['newest', 7]) + expect(ids).toEqual([6, 7, 8, 9]) + }), +) + +effectTest('the cap stops the walk and keeps the newest rows in the window', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + const ids = yield* walk(realm, { window: { since: 1000 }, cap: 4 }) + expect(ids).toEqual([6, 7, 8, 9]) + expect(yield* realm.anchors).toEqual(['newest', 7]) + }), +) + +effectTest('an unbounded read of one page asks the realm exactly once', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(9)) + const ids = yield* walk(realm, { pageSize: 5, cap: 5, maxPages: 1 }) + expect(ids).toEqual([5, 6, 7, 8, 9]) + expect(yield* realm.anchors).toEqual(['newest']) + }), +) + +// A window far enough back to outrun the page budget returns a short read +// rather than walking the realm without limit. +effectTest('the page budget bounds how many requests one read can make', () => + Effect.gen(function* () { + const realm = yield* fakeRealm(rowsUpTo(20)) + yield* walk(realm, { window: { since: 1000 }, maxPages: 2 }) + expect(yield* realm.anchors).toEqual(['newest', 18]) + }), +) diff --git a/packages/zulip/history-paging.ts b/packages/zulip/history-paging.ts new file mode 100644 index 0000000..dc6aad1 --- /dev/null +++ b/packages/zulip/history-paging.ts @@ -0,0 +1,146 @@ +import { Array as Arr, Effect, HashSet, Order } from 'effect' +import type { ZulipParams } from './http.ts' + +/** + * Reading a time window out of Zulip history. + * + * `GET /messages` selects a page by anchor and count and has no timestamp + * predicate at all, so a caller holding a `since`/`until` window cannot ask + * for it. Asking for the newest page and filtering it is what produces the + * silent false zero this module exists to remove: when every message on the + * recency-selected page lies outside the window, the filter empties it and + * the caller sees an authoritative-looking nothing. + * + * So the window is reached by walking back to it, one anchored page at a + * time, until the window's lower bound is crossed, the caller's cap is met, + * or history runs out. + */ + +/** The only two fields the walk reads off a message row. */ +export type PagedRow = { + readonly id: number + readonly timestamp: number +} + +/** Inclusive bounds, both optional. */ +export type HistoryWindow = { + readonly since?: number | undefined + readonly until?: number | undefined +} + +const inWindow = + (window: HistoryWindow) => + (row: PagedRow): boolean => { + if (window.since !== undefined && row.timestamp < window.since) return false + if (window.until !== undefined && row.timestamp > window.until) return false + return true + } + +type Walk = { + readonly anchor: ZulipParams[string] + readonly pagesLeft: number + readonly seen: HashSet.HashSet + /** One entry per page, newest page first — reversed on the way out. */ + readonly byPage: ReadonlyArray> + readonly collected: number + readonly more: boolean +} + +/** + * Read the messages in a time window, paging backwards to reach it. + * + * Results come back in ascending order. The walk is sequential by + * construction — every anchor is derived from the page before it — so it + * cannot fan out onto a rate-limited realm. + * + * `onPage` turns one page's in-window rows into results, and is handed the + * query that fetched them: Zulip returns raw content or rendered content and + * never both, so resolving a page's mentions means re-issuing that page's own + * query. A walk that rendered once for the whole thing would index a single + * page and quietly report no mentions on every other. + * + * The walk stops on the first of: a page carrying a row older than `since`, + * so the window's start has been passed; `cap` results collected; a page that + * adds no row the walk has not already seen, which covers both exhausted + * history and an anchor naming no live message; or the `maxPages` budget. The + * last two make a short read possible, and a short read is truthful in a way + * the post-filtered page was not. + * + * `cap` truncates from the old end, keeping the newest results — the same + * thing a `limit` meant before there was a walk. + */ +export const readWindow = (options: { + /** Query fields shared by every page — the narrow, and nothing anchored. */ + readonly query: ZulipParams + readonly window: HistoryWindow + readonly pageSize: number + /** Stop once this many results are held. Omit for no cap. */ + readonly cap: number | undefined + /** Hard ceiling on requests, so a far-back window cannot walk the realm. */ + readonly maxPages: number + readonly fetchPage: (query: ZulipParams) => Effect.Effect, E, R> + readonly onPage: ( + query: ZulipParams, + rows: ReadonlyArray, + ) => Effect.Effect, E, R> +}): Effect.Effect, E, R> => { + const keep = inWindow(options.window) + + const step = (state: Walk): Effect.Effect, E, R> => { + const query: ZulipParams = { + ...options.query, + anchor: state.anchor, + num_before: options.pageSize, + num_after: 0, + } + const stop: Walk = { ...state, more: false } + return options.fetchPage(query).pipe( + Effect.flatMap((rows): Effect.Effect, E, R> => { + if (!Arr.isNonEmptyReadonlyArray(rows)) return Effect.succeed(stop) + const fresh = rows.filter((r) => !HashSet.has(state.seen, r.id)) + if (!Arr.isNonEmptyReadonlyArray(fresh)) return Effect.succeed(stop) + const kept = fresh.filter(keep) + const oldestId = Arr.min(Order.number)(Arr.map(rows, (r) => r.id)) + const oldestTs = Arr.min(Order.number)(Arr.map(rows, (r) => r.timestamp)) + // A page holding nothing in the window costs no mapping, and for the + // rendered-content read that means no request. + const mapped = Arr.isEmptyReadonlyArray(kept) + ? Effect.succeed>([]) + : options.onPage(query, kept) + return mapped.pipe( + Effect.map((out) => { + const collected = state.collected + out.length + return { + anchor: oldestId, + pagesLeft: state.pagesLeft - 1, + seen: rows.reduce((s, r) => HashSet.add(s, r.id), state.seen), + byPage: [...state.byPage, out], + collected, + more: + state.pagesLeft > 1 && + !(options.window.since !== undefined && oldestTs < options.window.since) && + !(options.cap !== undefined && collected >= options.cap), + } + }), + ) + }), + ) + } + + return Effect.iterate( + { + anchor: 'newest', + pagesLeft: options.maxPages, + seen: HashSet.empty(), + byPage: [], + collected: 0, + more: options.maxPages > 0, + } satisfies Walk as Walk, + { while: (state) => state.more, body: step }, + ).pipe( + Effect.map((state) => { + const all = Arr.flatten(Arr.reverse(state.byPage)) + return options.cap === undefined ? all : Arr.takeRight(all, options.cap) + }), + ) +} From 08505d35b615825be7b86fc054ed5191b6e6976a Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:50:40 +0100 Subject: [PATCH 2/3] The bounds are honest now, so say what limit still does to a window read_channel and read_thread promised a result "bounded by optional since/until/limit". The since/until half is true now that a window is read out of history. The limit half is not what it sounds like: it cuts from the old end, so a result holding exactly limit messages may be missing older ones inside the window it was asked for, and looks identical to a complete read. Both tool descriptions and the limit field now say that, and name the check that separates the two cases: compare the oldest message returned against since. agent-experience.md gains the divergence rather than losing one. A human scrolling back sees where they stopped; an agent whose read is capped does not. --- docs/agent-experience.md | 6 ++++++ packages/mcp/tools.ts | 10 +++++++--- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/docs/agent-experience.md b/docs/agent-experience.md index ed6f760..a8cd934 100644 --- a/docs/agent-experience.md +++ b/docs/agent-experience.md @@ -134,6 +134,12 @@ Places the current implementation fails this reference. `read_thread` and `list_channels`; a human member also gets search and unread state. Principle 1, prospectively — this is a gap to fill, not something to cut. +- **A capped read does not say it was capped.** A human scrolling back sees + where they stopped. An agent whose read hits `limit` gets a page that looks + the same as a complete one, with the old end of its window silently missing. + The bounds themselves are honest — a window is read out of history rather + than filtered off the newest page — but the cap is not. Principle 1, and a + gap to fill. ## Worked example: lazy acquire diff --git a/packages/mcp/tools.ts b/packages/mcp/tools.ts index 09a5635..2c1bccd 100644 --- a/packages/mcp/tools.ts +++ b/packages/mcp/tools.ts @@ -426,7 +426,11 @@ const parseRange = (range: { const rangeSchemaFields = { since: { type: 'number', description: 'Inclusive lower bound in epoch seconds' }, until: { type: 'number', description: 'Inclusive upper bound in epoch seconds' }, - limit: { type: 'number', description: 'Hard cap on returned messages' }, + limit: { + type: 'number', + description: + "Hard cap on returned messages, cut from the window's old end. A result holding exactly this many may omit older messages inside the window — compare the oldest message returned against since to tell one from the other.", + }, } as const /** @@ -996,7 +1000,7 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA { name: 'read_channel', description: - 'Read recent messages from a channel by name. Returns {messages: Message[]} bounded by optional since/until/limit. Each message carries a clickable permalink (and channel.permalink / thread.permalink) — when you cite one of these to a human, render it as that permalink, not a bare name or id.', + "Read messages from a channel by name. Returns {messages: Message[]}. since/until name a window and are read out of history, so a window further back than the newest messages still comes back; limit then caps the result and cuts from the window's old end. Each message carries a clickable permalink (and channel.permalink / thread.permalink) — when you cite one of these to a human, render it as that permalink, not a bare name or id.", inputSchema: { type: 'object', properties: { @@ -1025,7 +1029,7 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA { name: 'read_thread', description: - 'Read recent messages from a thread (topic) within a channel. Returns {messages: Message[]} bounded by optional since/until/limit. Each message carries a clickable permalink (and channel.permalink / thread.permalink) — when you cite one of these to a human, render it as that permalink, not a bare name or id.', + "Read messages from a thread (topic) within a channel. Returns {messages: Message[]}. since/until name a window and are read out of history, so a window further back than the newest messages still comes back; limit then caps the result and cuts from the window's old end. Each message carries a clickable permalink (and channel.permalink / thread.permalink) — when you cite one of these to a human, render it as that permalink, not a bare name or id.", inputSchema: { type: 'object', properties: { From a34f5c66c4c5ed2f54d1d6510422cc6862dac4a0 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:53:32 +0100 Subject: [PATCH 3/3] Wait on the directory read even when the window came back empty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The directory read is forked so it overlaps the walk rather than sitting in front of it, and the join lives where the mapping needs it. A window that turns out empty maps no page, so nothing joined and a /users failure was dropped with the interrupted fiber. The read then answered [] — the same shape as a channel with nothing in the window. Both call sites now join unconditionally before returning. --- packages/zulip/adapter.test.ts | 19 +++++++++++ packages/zulip/adapter.ts | 11 ++++++- packages/zulip/history-paging.ts | 56 ++++++++++++++------------------ 3 files changed, 53 insertions(+), 33 deletions(-) diff --git a/packages/zulip/adapter.test.ts b/packages/zulip/adapter.test.ts index 3bb0eb9..2e45c45 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -2310,6 +2310,25 @@ effectTest('history.readChannel keeps the newest messages when the window overru }), ) +// The directory read runs alongside the walk, so a window that turns out +// empty never consumes it. It must still be waited on, or a realm that has +// stopped answering /users reads as a channel with nothing in the window. +effectTest('history.readChannel surfaces a directory failure on an empty window', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedMessagePages(stub, []) + yield* stub.respond('GET', '/api/v1/users', { + body: { result: 'error', msg: 'realm unavailable' }, + status: 503, + }) + const err = yield* Effect.flip( + adapter.history.readChannel(generalChannel.name, { since: rowTs(1), limit: 10 }), + ) + expect(err._tag).toBe('HistoryError') + }), +) + effectTest('history.readThread narrows by both channel and topic', () => Effect.gen(function* () { const stub = yield* makeStubHttpClient diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index b5f3c7d..6c8fc83 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -829,7 +829,7 @@ export const zulipAdapter = ( // the first page rather than in front of it. Joined on the first page // that has anything to map. const directoryFiber = yield* Effect.fork(buildDirectoryLookup()) - return yield* readWindow({ + const messages = yield* readWindow({ query: { narrow: JSON.stringify(narrow), apply_markdown: false }, window: { since: range.since, until: range.until }, // An unbounded read is answered by the newest page, so its page is @@ -854,6 +854,11 @@ export const zulipAdapter = ( ) }), }) + // A walk that found nothing never mapped a page, so nothing has + // joined the directory yet. Join it anyway: a failed directory read + // must not pass for an empty window. + yield* Fiber.join(directoryFiber) + return messages }) // Zulip constructs bot delivery emails as `-bot@` @@ -2069,6 +2074,10 @@ export const zulipAdapter = ( }) }), }) + // A catch-up that found nothing never mapped a page, so nothing has + // joined the directory yet. Join it anyway: a failed directory read + // must not pass for a seat that missed nothing. + yield* Fiber.join(directoryFiber) const out: InboundEvent[] = [] for (const mapped of perMessage) { for (const ev of mapped) { diff --git a/packages/zulip/history-paging.ts b/packages/zulip/history-paging.ts index dc6aad1..5a0d1ac 100644 --- a/packages/zulip/history-paging.ts +++ b/packages/zulip/history-paging.ts @@ -6,14 +6,10 @@ import type { ZulipParams } from './http.ts' * * `GET /messages` selects a page by anchor and count and has no timestamp * predicate at all, so a caller holding a `since`/`until` window cannot ask - * for it. Asking for the newest page and filtering it is what produces the - * silent false zero this module exists to remove: when every message on the - * recency-selected page lies outside the window, the filter empties it and - * the caller sees an authoritative-looking nothing. - * - * So the window is reached by walking back to it, one anchored page at a - * time, until the window's lower bound is crossed, the caller's cap is met, - * or history runs out. + * for it. Filtering the newest page instead is what produces a silent false + * zero: when every message on a recency-selected page lies outside the + * window, the filter empties it and the caller sees an authoritative-looking + * nothing. */ /** The only two fields the walk reads off a message row. */ @@ -62,12 +58,9 @@ type Walk = { * The walk stops on the first of: a page carrying a row older than `since`, * so the window's start has been passed; `cap` results collected; a page that * adds no row the walk has not already seen, which covers both exhausted - * history and an anchor naming no live message; or the `maxPages` budget. The - * last two make a short read possible, and a short read is truthful in a way - * the post-filtered page was not. + * history and an anchor naming no live message; or the `maxPages` budget. * - * `cap` truncates from the old end, keeping the newest results — the same - * thing a `limit` meant before there was a walk. + * `cap` truncates from the old end, keeping the newest results. */ export const readWindow = (options: { /** Query fields shared by every page — the narrow, and nothing anchored. */ @@ -93,16 +86,18 @@ export const readWindow = (options: { num_before: options.pageSize, num_after: 0, } - const stop: Walk = { ...state, more: false } return options.fetchPage(query).pipe( Effect.flatMap((rows): Effect.Effect, E, R> => { - if (!Arr.isNonEmptyReadonlyArray(rows)) return Effect.succeed(stop) + const exhausted: Walk = { ...state, more: false } + if (!Arr.isNonEmptyReadonlyArray(rows)) return Effect.succeed(exhausted) const fresh = rows.filter((r) => !HashSet.has(state.seen, r.id)) - if (!Arr.isNonEmptyReadonlyArray(fresh)) return Effect.succeed(stop) + if (!Arr.isNonEmptyReadonlyArray(fresh)) return Effect.succeed(exhausted) const kept = fresh.filter(keep) const oldestId = Arr.min(Order.number)(Arr.map(rows, (r) => r.id)) const oldestTs = Arr.min(Order.number)(Arr.map(rows, (r) => r.timestamp)) - // A page holding nothing in the window costs no mapping, and for the + const crossedLowerBound = + options.window.since !== undefined && oldestTs < options.window.since + // A page holding nothing in the window is not mapped, and for the // rendered-content read that means no request. const mapped = Arr.isEmptyReadonlyArray(kept) ? Effect.succeed>([]) @@ -110,16 +105,14 @@ export const readWindow = (options: { return mapped.pipe( Effect.map((out) => { const collected = state.collected + out.length + const capMet = options.cap !== undefined && collected >= options.cap return { anchor: oldestId, pagesLeft: state.pagesLeft - 1, seen: rows.reduce((s, r) => HashSet.add(s, r.id), state.seen), byPage: [...state.byPage, out], collected, - more: - state.pagesLeft > 1 && - !(options.window.since !== undefined && oldestTs < options.window.since) && - !(options.cap !== undefined && collected >= options.cap), + more: state.pagesLeft > 1 && !crossedLowerBound && !capMet, } }), ) @@ -127,17 +120,16 @@ export const readWindow = (options: { ) } - return Effect.iterate( - { - anchor: 'newest', - pagesLeft: options.maxPages, - seen: HashSet.empty(), - byPage: [], - collected: 0, - more: options.maxPages > 0, - } satisfies Walk as Walk, - { while: (state) => state.more, body: step }, - ).pipe( + const start: Walk = { + anchor: 'newest', + pagesLeft: options.maxPages, + seen: HashSet.empty(), + byPage: [], + collected: 0, + more: options.maxPages > 0, + } + + return Effect.iterate(start, { while: (state) => state.more, body: step }).pipe( Effect.map((state) => { const all = Arr.flatten(Arr.reverse(state.byPage)) return options.cap === undefined ? all : Arr.takeRight(all, options.cap)