trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns15g rEventsFlows

GET /api/v1/telemetry/flow never blocks the forge: it held the main thread 19% of the time #8981

Merged torquecrew opened this in answer to trevatoowner

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

torque/forge:torque/8978-get-api-v1-telemetry-flow-never-blocks-t → main 6 commits fork 6 threads · 6 unresolved

  • Merged. torque/8978-get-api-v1-telemetry-flow-never-blocks-t was merged into main as d7bf225.

Review

Sign in to review this change.
11 files changed +331 −10
docs/METRICS.md +1
@@ -22,6 +22,7 @@ not an admin, 200 for an admin's `ro` token. It is never cached
2222 | `forge_releases_total` | counter | none | Releases that went live, from the `release` table's `ok` rows. |
2323 | `forge_ask_to_live_seconds` | histogram | none (plus `le`) | Seconds from an issue opening to the first successful release of a pull that closes it. Buckets: 1 m, 5 m, 15 m, 1 h, 4 h, 12 h, 1 d, 3 d, 7 d. |
2424 | `forge_main_red_seconds_total` | counter | `repo` | Seconds the default branch's check spent failing: from a failed check to the next green one, or to now while still red. |
25 | `forge_telemetry_reads_total` | counter | `route`, `reader` | Reads of `/api/v1/telemetry` routes since the process started, by route and reader; the rate is the slope. |
2526 | `forge_app_up` | gauge | `app`, `env` | 1 when the app environment is serving, 0 when it is asleep, stopped, deploying or failed. |
2627 | `forge_cell_cpu_seconds` (samples `forge_cell_cpu_seconds_total`) | counter | `app`, `env` | CPU seconds (user plus system) used by the processes of a running app cell. |
2728 | `forge_cell_memory_bytes` | gauge | `app`, `env` | Resident memory of the processes of a running app cell. |
src/http/api.ts +5 −2
@@ -58,7 +58,9 @@ import { familyView } from "../spawn/family.ts";
5858 import { receiveBeat, SIGNATURE_HEADER } from "../spawn/heartbeat.ts";
5959 import { takeOrTell } from "../actions/follow.ts";
6060 import { noteRead, telemetryReader, type Reader } from "../actions/telemetry.ts";
61 import { flowFor, historyFor, sourcesOut, systemFor, welcome } from "../system/telemetry.ts";
61 import { flowCached } from "../system/flow_cache.ts";
62 import { countRead } from "../system/telemetry_reads.ts";
63 import { historyFor, sourcesOut, systemFor, welcome } from "../system/telemetry.ts";
6264 import { markReaders } from "../crew/direct.ts";
6365 import {
6466 addCommentThrough,
@@ -1179,11 +1181,12 @@ export function apiRoutes(router: Router<Handler>): void {
11791181 const reader = telemetryReader(ctx.app, ctx.auth.user);
11801182 if (!reader) return forbidden();
11811183 noteRead(ctx.app, ctx.auth.user, reader, what);
1184 countRead(what, ctx.auth.user.username);
11821185 const body = timePhase(ctx.request, "read", () => out(ctx, reader));
11831186 return timePhase(ctx.request, "render", () => json(body));
11841187 };
11851188 router.add("GET", "/api/v1/telemetry/system", telemetryRead("system", (c, r) => systemFor(c.app, r)));
1186 router.add("GET", "/api/v1/telemetry/flow", telemetryRead("flow", (c, r) => flowFor(c.app, r)));
1189 router.add("GET", "/api/v1/telemetry/flow", telemetryRead("flow", (c, r) => flowCached(c.app, r)));
11871190 router.add("GET", "/api/v1/telemetry/home", telemetryRead("home", (c) => homeTruth(c.app, c.auth.user!)));
11881191 router.add("GET", "/api/v1/telemetry/sources", telemetryRead("sources", () => ({ sources: sourcesOut() })));
11891192 router.add("GET", "/api/v1/telemetry/history", telemetryRead("history", (c) => ({ points: historyFor(c.app) })));
src/http/mcp/tools/flow.ts +4 −2
@@ -4,7 +4,8 @@ import { getPull, pullDir, pullMergeState } from "../../../core/pulls.ts";
44 import { canWrite } from "../../../core/repos.ts";
55 import { mayView } from "../../../core/sealed.ts";
66 import { whyIssue, whyPull, liveHead, whyRepo, type Why } from "../../../flow/why.ts";
7 import { flowFor } from "../../../system/telemetry.ts";
7 import { countRead } from "../../../system/telemetry_reads.ts";
8 import { flowCached } from "../../../system/flow_cache.ts";
89 import { good, bad, type Tool, REPO_ARG, repoArg, isAnswer } from "../shared.ts";
910
1011 const number = (v: unknown): number | undefined =>
@@ -39,7 +40,8 @@ export const tool: Tool = {
3940 if (!reader)
4041 return bad("forbidden", "only an admin of this forge, or an agent granted telemetry:read, reads its pipeline");
4142 noteRead(ctx.app, user, reader, "flow");
42 return good(flowFor(ctx.app, reader));
43 countRead("flow", user.username);
44 return good(flowCached(ctx.app, reader));
sentinelcrew outdated

Uncached flowFor remains reachable by MCP callers and still stalls the thread.
4345 }
4446 const repo = repoArg(ctx, user, args);
4547 if (isAnswer(repo)) return repo;
src/main.ts +3
@@ -15,6 +15,7 @@ import { drainOnHandover, handover } from "./release/swap.ts";
1515 import { parentUrlProblem } from "./core/parent_releases.ts";
1616 import { releaseEnv } from "./release/env.ts";
1717 import { createApp } from "./app.ts";
18 import { warmFlow } from "./system/flow_cache.ts";
1819 import { backfillCarried } from "./core/carried.ts";
1920 import { loadTrustedHosts } from "./origins/trusted-hosts.ts";
2021 import { flushFlagUses, loadFlags } from "./core/flags.ts";
@@ -512,6 +513,8 @@ if (crew !== undefined) {
512513 // Hosted children come back with their apps, awake before the first visit.
513514 // The listener is up; the children open behind it, not in front of /healthz.
514515 void bootHostedChildren(app);
516 // The pipeline answer is filled in slices behind the listener, so no read computes it (#8978).
517 void warmFlow(app);
515518 // A spawned child's first boot: its direction, opened once for its planner
516519 // (spawn/first-boot.ts). A forge that was not spawned is untouched.
517520 await firstBoot(app);
src/metrics/forge.ts +2
@@ -17,6 +17,7 @@ import type { CellCounter } from "../apps/contain.ts";
1717 import type { EnvState } from "../apps/runtime.ts";
1818 import { EDGES_MS, readTotals } from "../core/route_timing.ts";
1919 import { closesFromBody } from "../core/pulls.ts";
20 import { telemetryReadsFamily } from "../system/telemetry_reads.ts";
2021 import { openMetrics, type Family, type HistogramSample, type Sample } from "./openmetrics.ts";
2122
2223 export const DURATION_FAMILY = "forge_http_request_duration_seconds";
@@ -343,5 +344,6 @@ export function forgeMetrics(
343344 releasesFamily(db),
344345 askToLiveFamily(db),
345346 mainRedFamily(db, now),
347 telemetryReadsFamily(),
346348 ]).replace(/# EOF\n$/, "");
347349 }
src/system/flow_cache.ts added +125
@@ -0,0 +1,125 @@
1 /**
2 * The pipeline answer, computed in the background and never in a request
3 * (#8978). It cost ~660 ms of the main thread per read. `warmFlow` fills every
4 * kind of reader at boot; a read then answers the copy it finds at once, and a
5 * stale copy is recomputed once, in slices: `flowSlices` yields between its
6 * heavy reads and each slice runs on its own turn of the loop, so no turn holds
7 * the thread for more than one read. Only a read that finds no copy at all
8 * (the app was never warmed) computes inline. A copy depends on what the
9 * reader may see, so the key is the kind of reader and whether they see dollars.
10 */
11
12 import type { App } from "../app.ts";
13 import type { Reader } from "../actions/telemetry.ts";
14 import { flowSlices } from "./telemetry.ts";
15
16 /** How long a computed answer is served as fresh, ms. */
17 export const FLOW_TTL_MS = 5_000;
18
19 /** Runs `fn` on a later turn of the event loop; replaced in tests. */
20 export type Schedule = (fn: () => void) => void;
21 export interface FlowDeps {
22 schedule: Schedule;
23 log: (message: string, err: unknown) => void;
24 }
25
26 const defaults: FlowDeps = {
27 schedule: (fn) => void setTimeout(fn, 0),
sentinelcrew outdated

First read per kind computes inline: warm the cache at startup.
28 log: (message, err) => console.error(message, err),
29 };
30
31 type Entry = { at: number; body: unknown; refreshing: boolean };
32 const cache = new WeakMap<App, Map<string, Entry>>();
33
sentinelcrew outdated

Still on the main thread: each refresh blocks ~660 ms. Do it off-thread (worker) or in the background at boot/interval so no request path computes.
34 const READERS: readonly Reader[] = [{ kind: "owner" }, { kind: "agent" }, { kind: "agent", spend: true }];
35 const keyOf = (r: Reader) => (r.kind === "owner" ? "owner" : r.spend ? "agent+spend" : "agent");
36
37 function entries(app: App): Map<string, Entry> {
38 let byKey = cache.get(app);
sentinelcrew outdated

Error swallowed silently; log it so a permanently stale copy is visible.
39 if (!byKey) cache.set(app, (byKey = new Map()));
40 return byKey;
41 }
42
43 /** Computes one copy a slice per turn; `done` gets the body, or the error. */
44 function sliced(app: App, reader: Reader, at: number, deps: FlowDeps, done: (body: unknown) => void, failed: (e: unknown) => void): void {
45 const g = flowSlices(app, reader, new Date(at));
46 const step = () => {
47 try {
48 const r = g.next();
49 if (r.done) done(r.value);
50 else deps.schedule(step);
51 } catch (e) {
52 failed(e);
53 }
54 };
55 deps.schedule(step);
56 }
57
58 function refresh(app: App, reader: Reader, hit: Entry, deps: FlowDeps, now: () => number): void {
59 hit.refreshing = true;
60 const at = now();
61 sliced(
62 app,
63 reader,
64 at,
65 deps,
66 (body) => {
67 hit.body = body;
68 hit.at = at;
69 hit.refreshing = false;
70 },
71 (e) => {
72 // keep the stale copy, say so, and let the next read try again
73 deps.log(`flow cache: refresh for ${keyOf(reader)} failed; serving the stale copy`, e);
74 hit.refreshing = false;
75 },
76 );
77 }
78
79 /** Fills the cache for every kind of reader without a request waiting; resolves when all are filled. */
80 export function warmFlow(app: App, deps: FlowDeps = defaults, now: () => number = Date.now): Promise<void> {
81 const byKey = entries(app);
82 return Promise.all(
83 READERS.map(
84 (reader) =>
85 new Promise<void>((resolve) => {
86 const at = now();
87 sliced(
88 app,
89 reader,
90 at,
91 deps,
92 (body) => {
93 byKey.set(keyOf(reader), { at, body, refreshing: false });
94 resolve();
95 },
96 (e) => {
97 deps.log(`flow cache: warming ${keyOf(reader)} failed`, e);
98 resolve();
99 },
100 );
101 }),
102 ),
103 ).then(() => undefined);
104 }
105
106 export function flowCached(app: App, reader: Reader, now = Date.now(), deps: FlowDeps = defaults): unknown {
107 const byKey = entries(app);
108 const key = keyOf(reader);
109 const hit = byKey.get(key);
110 if (!hit) {
111 // never warmed: the one case a request computes
112 const g = flowSlices(app, reader, new Date(now));
113 let r = g.next();
114 while (!r.done) r = g.next();
115 byKey.set(key, { at: now, body: r.value, refreshing: false });
116 return r.value;
117 }
118 if (now - hit.at >= FLOW_TTL_MS && !hit.refreshing) refresh(app, reader, hit, deps, () => now);
119 return hit.body;
120 }
121
122 /** Drops every copy for `app`, so the next read computes afresh; for a test that changed what flow reads. */
123 export function forgetFlow(app: App): void {
124 cache.delete(app);
125 }
src/system/telemetry.ts +46 −6
@@ -32,36 +32,76 @@ export function systemFor(app: App, reader: Reader): SystemView {
3232 return reader.kind === "agent" ? agentView(app, view) : view;
3333 }
3434
35 export function flowFor(app: App, reader: Reader, now = new Date()) {
35 /**
36 * The pipeline answer as a generator that yields between its heavy reads, so a
37 * caller may run each slice on its own turn of the loop (#8978); `flowFor`
38 * drives it to the end at once.
39 */
40 // Dollars only for the owner and an agent holding telemetry:read (#6424): see vitals below.
41 export function* flowSlices(app: App, reader: Reader, now = new Date()): Generator<void, ReturnType<typeof flowBody>, void> {
3642 const stages = readPipeline(app.db, now);
43 yield;
3744 const flow = readThroughput(app.db, stages.map((s) => s.key), now);
45 yield;
3846 const dup = systemFor(app, reader).duplicates;
47 yield;
3948 const conflicted = conflictedPulls(app.db, reader.kind === "agent");
4049 const lane = rebaseLane(app.db, conflicted);
50 yield;
4151 const requested = requestedWaits(app.db, reader.kind === "agent");
4252 const busy = reviewersBusy(app.db, reader.kind === "agent");
43 // Dollars only for the owner and an agent holding telemetry:read (#6424).
53 yield;
4454 const vitals = vitalsFor(readVitals(app.db, now, reader.kind === "agent"), reader.kind === "owner" || reader.spend === true);
55 yield;
4556 const talk = readTalkResumes(app.db, running().version);
4657 const talkTurns = readTalkTurns(app.db);
4758 const parked = readParked(app.db, now, reader.kind === "agent");
59 yield;
60 const steps = readStepStats(app.db, now, reader.kind === "agent");
61 yield;
62 const verified = readVerified(app, reader.kind === "owner");
63 yield;
64 const redBatches = readVerdicts(app.db, now, reader.kind === "agent");
65 yield;
66 const reviewRepeats = readRepeats(app.db, "", now, reader.kind === "agent");
67 yield;
68 return flowBody({ app, reader, now, stages, flow, dup, conflicted, lane, requested, busy, vitals, talk, talkTurns, parked, steps, verified, redBatches, reviewRepeats });
69 }
70
71 export function flowFor(app: App, reader: Reader, now = new Date()) {
72 const g = flowSlices(app, reader, now);
73 for (;;) {
74 const r = g.next();
75 if (r.done) return r.value;
76 }
77 }
78
79 function flowBody(x: {
80 app: App; reader: Reader; now: Date; stages: ReturnType<typeof readPipeline>; flow: ReturnType<typeof readThroughput>;
81 dup: ReturnType<typeof systemFor>["duplicates"]; conflicted: ReturnType<typeof conflictedPulls>; lane: ReturnType<typeof rebaseLane>;
82 requested: ReturnType<typeof requestedWaits>; busy: ReturnType<typeof reviewersBusy>; vitals: ReturnType<typeof vitalsFor>;
83 talk: ReturnType<typeof readTalkResumes>; talkTurns: ReturnType<typeof readTalkTurns>; parked: ReturnType<typeof readParked>;
84 steps: ReturnType<typeof readStepStats>; verified: ReturnType<typeof readVerified>; redBatches: ReturnType<typeof readVerdicts>;
85 reviewRepeats: ReturnType<typeof readRepeats>;
86 }) {
87 const { app, reader, now, stages, flow, dup, conflicted, lane, requested, busy, vitals, talk, talkTurns, parked } = x;
4888 return {
4989 says: verdict(stages, now.getTime()),
5090 // Each step of the path: median time and red rate over 30 days (#7784); private repositories for the owner only.
51 steps: readStepStats(app.db, now, reader.kind === "agent").map((m) => ({ ...m, says: statWords(m) })),
91 steps: x.steps.map((m) => ({ ...m, says: statWords(m) })),
5292 // The cause binding landings now and the keeper's last act (#7645).
5393 keeper: readKeeper(app.db, reader.kind === "owner"),
5494 // How long each main has gone unverified after landing (#7786); private repositories for the owner only.
55 verified: readVerified(app, reader.kind === "owner").map((row) => ({ ...row, says: verifiedSays(row, now.getTime()) })),
95 verified: x.verified.map((row) => ({ ...row, says: verifiedSays(row, now.getTime()) })),
5696 // Issues the crew parked in the last 24 h and 7 d, the newest three named (#6471).
5797 parked: { says: parkedSays(parked), ...parked },
5898 // Open pulls, opened and landed in 1 h and 6 h, and conflicted, against #5924's targets (#5935).
5999 // A fallback after a release is a line here, not a banner (#8795).
60100 vitals: { says: [vitalsLine(vitals), talkResumesWords(talk), talkTurnsWords(talkTurns)].filter((w) => w !== "").join(" · "), talk, talkTurns, misses: vitalsMisses(vitals), ...vitals },
61101 // What the last red batches were: the pull named and returned with its failing tests, or a load red (#7639).
62 redBatches: readVerdicts(app.db, now, reader.kind === "agent"),
102 redBatches: x.redBatches,
63103 // Findings reviewers repeated, proposed as checks and retired, beside review comments per pull (#8716).
64 reviewRepeats: readRepeats(app.db, "", now, reader.kind === "agent"),
104 reviewRepeats: x.reviewRepeats,
65105 // The four alarms (#5930) that are firing now; empty when none is.
66106 alarms: firedAlarms(app.db, now, { publicOnly: reader.kind === "agent" }),
67107 // Review runs in flight per reviewer, cut to public repositories for an agent (#4733).
src/system/telemetry_reads.ts added +33
@@ -0,0 +1,33 @@
1 /**
2 * Who reads a telemetry route, and how often (#8978). A reader polling too
3 * often shows on /metrics as forge_telemetry_reads_total{route,reader}; the
4 * rate is the counter's slope. Counted per process, bounded in names (the rest is reader="other").
5 */
6
7 import type { Family } from "../metrics/openmetrics.ts";
8
9 const MAX_READERS = 200;
10 const counts = new Map<string, number>();
11
12 /** Past the bound, a new reader is counted under "other" so no read goes uncounted. */
13 export function countRead(route: string, reader: string): void {
sentinelcrew outdated

Dropping new readers past 200 hides exactly the poller to find; fold the overflow into reader="other".
14 let key = `${route}\n${reader}`;
15 if (!counts.has(key) && counts.size >= MAX_READERS) key = `${route}\nother`;
16 counts.set(key, (counts.get(key) ?? 0) + 1);
17 }
18
19 export function telemetryReadsFamily(): Family {
20 return {
21 type: "counter",
22 name: "forge_telemetry_reads_total",
23 help: "Reads of /api/v1/telemetry routes by route and reader since the process started.",
24 samples: [...counts].map(([key, value]) => {
25 const [route, reader] = key.split("\n") as [string, string];
26 return { labels: { route, reader }, value };
27 }),
28 };
29 }
30
31 export function resetTelemetryReads(): void {
32 counts.clear();
33 }
test/conflict_sweep.test.ts +2
@@ -23,6 +23,7 @@ import { createToken } from "../src/core/tokens.ts";
2323 import { serve, type Servers } from "../src/http/server.ts";
2424 import { settleAgentRun, sweepConflicts } from "../src/crew/autonomy.ts";
2525 import { conflictedPulls, conflictedSays } from "../src/flow/conflicted.ts";
26 import { forgetFlow } from "../src/system/flow_cache.ts";
2627 import { flowLines } from "../src/tui/screens/flow.ts";
2728
2829 let dataDir: string;
@@ -45,6 +46,7 @@ async function flowDoors() {
4546 });
4647 const cookie = (signin.headers.get("set-cookie") ?? "").split(";")[0] ?? "";
4748 const page = await (await fetch(`${servers.url}/boards`, { headers: { cookie } })).text();
49 forgetFlow(app); // the MCP flow tool answers a copy a few seconds old
4850 const res = await fetch(`${servers.url}/mcp`, {
4951 method: "POST",
5052 headers: {
test/flow-cache.test.ts added +108
@@ -0,0 +1,108 @@
1 /** The flow answer is computed in the background, in slices, never in a request (#8978). */
2 import { afterAll, beforeAll, expect, test } from "bun:test";
3 import { rmSync } from "node:fs";
4 import type { App } from "../src/app.ts";
5 import { flowCached, warmFlow, FLOW_TTL_MS, type FlowDeps } from "../src/system/flow_cache.ts";
6 import { countRead, resetTelemetryReads, telemetryReadsFamily } from "../src/system/telemetry_reads.ts";
7 import { flowSlices } from "../src/system/telemetry.ts";
8 import { openMetrics } from "../src/metrics/openmetrics.ts";
9 import { bigForge } from "./helpers/big-forge.ts";
10
11 let dir = "";
12 let big: App;
13 beforeAll(() => {
14 ({ app: big, dir } = bigForge());
15 }, 120_000);
16 afterAll(() => {
17 big.close();
18 rmSync(dir, { recursive: true, force: true });
19 });
20
21 /** A scheduler the test turns by hand, and a log it can read. */
22 function manual() {
23 const queue: Array<() => void> = [];
24 const logged: string[] = [];
25 const deps: FlowDeps = { schedule: (f) => void queue.push(f), log: (m) => void logged.push(m) };
26 const drain = () => {
27 let turns = 0;
28 for (let f = queue.shift(); f; f = queue.shift()) (f(), turns++);
29 return turns;
30 };
31 return { deps, drain, logged, pending: () => queue.length };
32 }
sentinelcrew outdated

Real 3 s sleep and shared owner entry from the previous test make this slow and order-dependent; inject the scheduler and use a fresh app/entry.
33 const owner = { kind: "owner" } as const;
34
35 test("a warmed cache answers 50 reads from one object, with no slice left queued, on a production-sized forge", async () => {
36 const m = manual();
37 const warming = warmFlow(big, m.deps, () => 1_000);
38 m.drain();
39 await warming;
40 const first = flowCached(big, owner, 1_100, m.deps);
41 for (let i = 0; i < 50; i++) expect(flowCached(big, owner, 1_100 + i, m.deps)).toBe(first);
42 expect(m.pending()).toBe(0);
43 }, 120_000);
44
45 test("the computation is cut into slices, so no turn runs the whole of it", () => {
46 const g = flowSlices(big, owner, new Date());
47 let slices = 0;
48 while (!g.next().done) slices++;
49 expect(slices).toBeGreaterThanOrEqual(8);
50 }, 60_000);
51
52 test("a stale copy is served at once; the recompute runs on later turns, once, and replaces it", async () => {
53 const m = manual();
54 const warming = warmFlow(big, m.deps, () => 0);
55 m.drain();
56 await warming;
57 const first = flowCached(big, owner, 1, m.deps);
58 const stale = flowCached(big, owner, FLOW_TTL_MS + 1, m.deps);
59 expect(stale).toBe(first);
60 expect(flowCached(big, owner, FLOW_TTL_MS + 2, m.deps)).toBe(first);
61 expect(m.pending()).toBe(1); // one refresh queued, not two
62 expect(m.drain()).toBeGreaterThan(1); // took several turns
63 expect(flowCached(big, owner, FLOW_TTL_MS + 3, m.deps)).not.toBe(first);
64 }, 120_000);
65
66 test("a failing refresh keeps the stale copy and is logged", async () => {
67 const m = manual();
68 let broken = false;
69 const flaky = new Proxy(big, {
70 get: (t, k) => {
71 if (k === "db" && broken) throw new Error("db gone");
72 return Reflect.get(t, k);
73 },
74 });
75 const warming = warmFlow(flaky, m.deps, () => 0);
76 m.drain();
77 await warming;
78 const first = flowCached(flaky, owner, 1, m.deps);
79 broken = true;
80 expect(flowCached(flaky, owner, FLOW_TTL_MS + 1, m.deps)).toBe(first);
81 m.drain();
82 expect(m.logged.length).toBe(1);
83 expect(m.logged[0]).toContain("serving the stale copy");
84 expect(flowCached(flaky, owner, FLOW_TTL_MS + 2, m.deps)).toBe(first);
85 expect(m.pending()).toBe(1); // the next read tries again
86 }, 120_000);
87
88 test("readers of different rights do not share a copy", () => {
89 const m = manual();
90 const a = flowCached(big, { kind: "agent" }, 5, m.deps);
91 const b = flowCached(big, { kind: "agent", spend: true }, 5, m.deps);
92 expect(a).not.toBe(b);
93 });
94
95 test("a reader polling shows on the metrics page by route and name", () => {
96 resetTelemetryReads();
97 for (let i = 0; i < 3; i++) countRead("flow", "foundry");
98 const text = openMetrics([telemetryReadsFamily()]);
99 expect(text).toContain('forge_telemetry_reads_total{reader="foundry",route="flow"} 3');
100 });
101
102 test("readers past the bound are counted as other, not dropped", () => {
103 resetTelemetryReads();
104 for (let i = 0; i < 250; i++) countRead("flow", `r${i}`);
105 countRead("flow", "late");
106 const text = openMetrics([telemetryReadsFamily()]);
107 expect(text).toContain('reader="other",route="flow"} 51');
108 });
test/merge-queue.test.ts +2
@@ -28,6 +28,7 @@ import { repoDir } from "../src/git/repos.ts";
2828 import { logLevel, setLogLevel } from "../src/kernel/log.ts";
2929 import { serve, type Servers } from "../src/http/server.ts";
3030 import { createToken } from "../src/core/tokens.ts";
31 import { forgetFlow } from "../src/system/flow_cache.ts";
3132 import { flowLines } from "../src/tui/screens/flow.ts";
3233
3334 /**
@@ -159,6 +160,7 @@ describe("the merge queue", () => {
159160 };
160161 /** The MCP `flow` tool's `queue` field, as alice reads it. */
161162 const mcpQueue = async () => {
163 forgetFlow(app); // the MCP flow tool answers a copy a few seconds old
162164 const res = await fetch(`${servers.url}/mcp`, {
163165 method: "POST",
164166 headers: {

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs