1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360// The Reader reads the assembled document a CHUNK_BUDGET window at a time, and every
// round of the review loop re-reads all of it — the whole point of the loop is that the
// Reader confirms what the editor changed. On a long document that is several full text
// calls per round, up to max_review_iterations + 1 rounds, and they used to run strictly
// one after another: a 4-chunk body spent four call latencies per round waiting, for
// calls that share nothing and cannot affect each other.
//
// This pins the two properties that make sending them together safe. They overlap (or
// the change did nothing), and the issues still come back in CHUNK order rather than in
// whichever order the calls happened to finish — the second is what downstream depends
// on: `imagesForIssues` unions the attributions and `unresolved.md` is written in this
// order, so a document's reported issue list must not depend on provider timing.
import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { runReview } from "../src/pipeline/review.ts";
import { cacheableUserPrefix } from "../src/providers/promptCache.ts";
import type { PipelineContext } from "../src/pipeline/context.ts";
import type { Paths } from "../src/store/paths.ts";
// CHUNK_BUDGET is 24000 with a 2000-char overlap, so chunks start every 22000
// characters. A marker at each stride is therefore the first marker inside its own
// chunk, which is what identifies the call.
const STRIDE = 22000;
function markedBody(chunks: number): string {
let body = "";
for (let i = 0; i < chunks; i++) {
const marker = `<p>MARK${i}</p>`;
body += marker + "<p>filler</p>".repeat(Math.ceil((STRIDE - marker.length) / 13));
body = body.slice(0, (i + 1) * STRIDE);
}
return body;
}
function chunkOf(prompt: string): number {
const m = prompt.match(/MARK(\d+)/);
return m ? Number(m[1]) : -1;
}
const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
interface ReaderRun {
order: number[]; // chunk index of each issue reported, in the order returned
maxInFlight: number;
calls: number;
}
// Run one review round against a mock router that answers each chunk with an issue
// naming that chunk, after a delay chosen so completion order is the REVERSE of chunk
// order. `maxReviewIterations: 0` stops the loop after the first read, so the issues it
// returns as `unresolved` are exactly what runReader produced.
async function readerRound(
chunks: number,
concurrency: number,
delayFor: (chunk: number) => number,
onEvent: (type: string, data: Record<string, unknown>) => void = () => {},
): Promise<ReaderRun> {
const dir = mkdtempSync(join(tmpdir(), "iris-review-conc-"));
try {
let inFlight = 0;
let maxInFlight = 0;
let calls = 0;
const ctx = {
sessionId: "ses_test",
images: [],
maxReviewIterations: 0,
extractionConcurrency: concurrency,
recheckSampleSize: 1,
paths: {
agentsDir: join(dir, "agents"),
tmpAgentsDir: () => join(dir, "tmp-agents"),
agentMemory: () => join(dir, "memory", "page.json"),
} as unknown as Paths,
router: {
complete: async (agent: string, _cap: string, messages: { content: string }[]) => {
if (agent !== "reader") return { text: JSON.stringify({ html: "" }) };
calls++;
inFlight++;
maxInFlight = Math.max(maxInFlight, inFlight);
const which = chunkOf(messages.map((m) => m.content).join("\n"));
await sleep(delayFor(which));
inFlight--;
return {
text: JSON.stringify({
issues: [{ issue: `chunk ${which}`, severity: "low", suggested_action: "none", pages: [] }],
}),
};
},
},
log: { event: onEvent, agentCall: () => {} },
} as unknown as PipelineContext;
const result = await runReview(ctx, { body: markedBody(chunks), lint: { ok: true, violations: [] } });
return {
order: result.unresolved.map((i) => Number(i.issue.replace("chunk ", ""))),
maxInFlight,
calls,
};
} finally {
rmSync(dir, { recursive: true, force: true });
}
}
test("the chunks of one read are in flight together", async () => {
const run = await readerRound(3, 4, () => 5);
assert.equal(run.calls, 3, "the body must actually span three chunks for this to prove anything");
assert.equal(run.maxInFlight, 3, "all three chunk calls should be open at once under a limit of 4");
});
test("issues come back in chunk order, not in the order the calls finished", async () => {
// The earliest chunk answers slowest, so a list built from completion order would read
// 2, 1, 0 — which is what the document's unresolved list would then say.
const run = await readerRound(3, 4, (c) => [40, 20, 1][c] ?? 1);
assert.deepEqual(run.order, [0, 1, 2]);
});
test("the run's concurrency knob bounds the read as it bounds extraction", async () => {
const run = await readerRound(4, 2, () => 5);
assert.equal(run.calls, 4);
assert.equal(run.maxInFlight, 2, "a deployment that set 2 must not get 4 calls in flight in the review phase");
});
test("a deployment set to 1 still reads strictly serially", async () => {
const run = await readerRound(3, 1, () => 5);
assert.equal(run.maxInFlight, 1);
assert.deepEqual(run.order, [0, 1, 2]);
});
test("a short document is one call, as it was", async () => {
const run = await readerRound(1, 4, () => 1);
assert.equal(run.calls, 1);
assert.equal(run.maxInFlight, 1);
});
test("the page index leads the prompt, and is declared as its invariant head", async () => {
// The index is the one part of a Reader prompt that is about the document rather than
// the chunk, and it does not change while the loop runs — so every chunk of every round
// was re-sending the same ~1.5k tokens at full price. It now leads the message and is
// declared as the head, which is the only position a cache breakpoint can mark.
const dir = mkdtempSync(join(tmpdir(), "iris-review-head-"));
try {
const sent: { content: string; cachedPrefix?: string }[] = [];
const pages = Array.from({ length: 12 }, (_, i) => ({
order: i + 1,
innerHtml: `<h2>Section ${i + 1}</h2><p>${"content ".repeat(30)}</p>`,
}));
const ctx = {
sessionId: "ses_test",
images: [],
maxReviewIterations: 0,
extractionConcurrency: 4,
recheckSampleSize: 1,
paths: {
agentsDir: join(dir, "agents"),
tmpAgentsDir: () => join(dir, "tmp-agents"),
agentMemory: () => join(dir, "memory", "page.json"),
} as unknown as Paths,
router: {
complete: async (agent: string, _cap: string, messages: { role: string; content: string; cachedPrefix?: string }[]) => {
if (agent === "reader") {
const user = messages.find((m) => m.role === "user")!;
sent.push({ content: user.content, cachedPrefix: user.cachedPrefix });
}
return { text: JSON.stringify({ issues: [] }) };
},
},
log: { event: () => {}, agentCall: () => {} },
} as unknown as PipelineContext;
await runReview(ctx, { body: markedBody(3), lint: { ok: true, violations: [] }, pages });
assert.ok(sent.length > 1, "the body must span more than one chunk for this to prove anything");
const heads = new Set(sent.map((s) => s.cachedPrefix));
assert.equal(heads.size, 1, "every chunk of a round must declare the same head, or nothing is cached");
const [headText] = [...heads];
assert.ok(headText, "no head was declared, so the index is paid for once per chunk");
assert.match(headText, /^## Source pages in this document/);
assert.match(headText, /Section 1/);
// Declaring a head buys nothing unless the adapters will actually mark it: below
// `cacheableUserPrefix`'s minimum the breakpoint is dropped and every chunk pays for
// the index again, with nothing failing. At READER_INDEX_EXCERPT_CHARS a twelve-page
// index clears it with little to spare, so shortening that constant — or the excerpt
// format — would hand the whole saving back silently. This is the tripwire for that.
assert.equal(
cacheableUserPrefix("us.anthropic.claude-sonnet-4-6", headText),
true,
`a 12-page index is ${headText.length} chars — too short for the adapters to cache`,
);
// The head has to be the START of the message — that is what the adapters slice on —
// and the chunk's own material still has to be there, after it.
for (const s of sent) {
assert.ok(s.content.startsWith(headText), "the declared head is not the start of the message");
assert.match(s.content, /## HTML/);
assert.match(s.content, /## Flattened screen-reader view/);
assert.match(s.content, /## axe-core lint/);
assert.equal(s.content.indexOf("## Source pages"), 0, "the index leads, so it can be a prefix");
}
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
test("a document with no page index declares no head at all", async () => {
// An empty head is not a prefix worth naming, and passing "" would ask the adapter to
// slice a message at nothing.
const dir = mkdtempSync(join(tmpdir(), "iris-review-nohead-"));
try {
const sent: { cachedPrefix?: string }[] = [];
const ctx = {
sessionId: "ses_test",
images: [],
maxReviewIterations: 0,
extractionConcurrency: 2,
recheckSampleSize: 1,
paths: {
agentsDir: join(dir, "agents"),
tmpAgentsDir: () => join(dir, "tmp-agents"),
agentMemory: () => join(dir, "memory", "page.json"),
} as unknown as Paths,
router: {
complete: async (agent: string, _cap: string, messages: { role: string; cachedPrefix?: string }[]) => {
if (agent === "reader") sent.push({ cachedPrefix: messages.find((m) => m.role === "user")!.cachedPrefix });
return { text: JSON.stringify({ issues: [] }) };
},
},
log: { event: () => {}, agentCall: () => {} },
} as unknown as PipelineContext;
await runReview(ctx, { body: markedBody(1), lint: { ok: true, violations: [] } });
assert.equal(sent.length, 1);
assert.equal(sent[0].cachedPrefix, undefined);
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
test("the round says how it was read, so a slow one can be diagnosed", async () => {
// A review round that times out on a rate-limited provider is the case this line
// exists for: without it a run log cannot say whether the chunks went out together or
// how many were allowed to, which is the first thing to check.
const events: { type: string; data: Record<string, unknown> }[] = [];
await readerRound(3, 2, () => 1, (type, data) => events.push({ type, data }));
const starts = events.filter((e) => e.type === "reader_start");
assert.equal(starts.length, 1, "one line per round");
assert.deepEqual(starts[0].data, { iteration: 0, chunks: 3, concurrency: 2 });
});
// The same body, against a router whose chunk 0 fails however it likes. Returns which chunks were
// sent — the ones the round paid for — and what it rejected with.
//
// NOTHING HERE IS TIMED, and that is the point of how it is written (issue #386). The property under
// test is an ORDERING — no chunk sends after a failure has been recorded — and the first version
// staged that ordering with `sleep(1)` for chunk 0 against `sleep(30)` for the rest. Under a loaded
// full-suite run a 1ms timer arrives late often enough that a worker pulled chunk 2 before the
// failure landed, and the test failed 2 runs in 6 on unmodified `main` while passing 3 for 3 on its
// own. Widening the count to 3 would have deleted what it checks, since 3 is the first thing the
// guard prevents.
//
// So the ordering is made explicit instead. Chunk 0 fails with no await at all, every other chunk
// is held on a promise this function resolves, and it resolves it only once the round has already
// rejected — which cannot happen before the guard is armed, because the worker that catches the
// failure sets the flag on the line before it rethrows, and `Promise.all` rejects on that rethrow.
// A machine under load cannot reorder any of that: there is no timer in it.
async function failingRound(throwValue: unknown): Promise<{ sent: number[]; rejected: unknown }> {
const dir = mkdtempSync(join(tmpdir(), "iris-review-fail-"));
try {
const sent: number[] = [];
let windows = -1;
// Resolved after the round has rejected, so a chunk released by it is a chunk the guard has
// already had its chance to stop. Held rather than slow: a chunk that merely takes longer than
// chunk 0 is a bet on two timers, which is the defect being fixed.
let release: () => void = () => {};
const held = new Promise<void>((resolve) => {
release = resolve;
});
const ctx = {
sessionId: "ses_test",
images: [],
maxReviewIterations: 0,
extractionConcurrency: 2,
recheckSampleSize: 1,
paths: {
agentsDir: join(dir, "agents"),
tmpAgentsDir: () => join(dir, "tmp-agents"),
agentMemory: () => join(dir, "memory", "page.json"),
} as unknown as Paths,
router: {
complete: async (agent: string, _cap: string, messages: { content: string }[]) => {
if (agent !== "reader") return { text: JSON.stringify({ html: "" }) };
const which = chunkOf(messages.map((m) => m.content).join("\n"));
sent.push(which);
// Synchronously, before any await, so `complete` hands back a promise that is already
// rejected. Every worker but this one is then suspended on `held` and cannot advance.
if (which === 0) throw throwValue;
await held;
return { text: JSON.stringify({ issues: [] }) };
},
},
log: {
event: (type: string, data: Record<string, unknown>) => {
if (type === "reader_start") windows = data.chunks as number;
},
agentCall: () => {},
},
} as unknown as PipelineContext;
let rejected: unknown = "did not reject";
try {
await runReview(ctx, { body: markedBody(5), lint: { ok: true, violations: [] } });
} catch (e) {
rejected = e;
}
// The premise both tests below rest on, asserted rather than assumed: `sent` of `[0, 1]` only
// says the guard fired if there were chunks BEHIND those two for it to stop. Should
// CHUNK_BUDGET or STRIDE ever move so that this body is two windows, `[0, 1]` is what a deleted
// guard produces as well and the pair would pass saying nothing. Read off `reader_start` rather
// than recounted here, so it is the number the round actually used.
assert.equal(windows, 5, "the body must span 5 windows for a guard that stops 3 of them to be visible");
// The guard is armed by now, so let the held chunks go and let their workers pull whatever they
// are going to pull. One turn of the event loop drains all of it: nothing left in that chain
// waits on a timer or on I/O, and the microtask queue runs to empty before a timer callback
// does, whatever else the machine is doing. Without this the workers would still be suspended
// when `sent` was read, and the test would pass with the guard deleted.
release();
await sleep(0);
return { sent, rejected };
} finally {
rmSync(dir, { recursive: true, force: true });
}
}
test("a chunk that fails without an error to show for it still stops the round", async () => {
// The guard is a flag rather than a test on the recorded error, because the thrown
// value is not ours: an adapter or a mock that throws a bare `undefined` is still a
// chunk that failed, and a guard read off the error would be disarmed on exactly that
// call — every queued chunk then paying in full.
const run = await failingRound(undefined);
assert.deepEqual(run.sent, [0, 1], "a nullish failure must arm the guard too");
assert.equal(run.rejected, undefined, "and the round still rejects with what was thrown");
});
test("a chunk that fails stops the round paying for the chunks behind it", async () => {
// mapWithConcurrency rejects with the first error, matching the serial loop — but its
// workers keep pulling items, so without the guard a chunk-0 failure on a 5-chunk body
// at a limit of 2 still buys three more full-price reader calls for a round whose
// result is already discarded.
const boom = new Error("provider said no");
const run = await failingRound(boom);
// The chunks rather than a count of them, which pins the property in both directions: only the
// two already in flight were paid for, AND both of them were, so a guard that stopped sending
// altogether — or one that never let the round overlap in the first place — fails here too.
assert.deepEqual(run.sent, [0, 1], "only the calls already in flight should have been paid for");
// The error the round rejects with is the one that actually happened, not a stand-in
// raised by a chunk that read the flag.
assert.equal(run.rejected, boom);
});