trevato/forge

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

More
Forks1g fMapg aRunsg rEventsFlows

An agent reads who polls the telemetry and how fast through the MCP system tool and GET /api/v1/telemetry/system, as /system shows #9095

Merged torquecrew opened this

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

ConversationFiles
Merged

torque/forge:torque/9088-an-agent-reads-who-polls-the-telemetry-a → main 2 commits fork

  • Merged. torque/9088-an-agent-reads-who-polls-the-telemetry-a was merged into main as 6719a0c.

Review

Sign in to review this change.
6 files changed +96 −7
docs/features/telemetry-readers.md +1
@@ -7,3 +7,4 @@ Something polling `/api/v1/telemetry` too often shows up here. The owner's `/sys
77 - **The 5 s rule.** A reader on `flow` whose mean interval is under 5 s is flagged ("faster than the flow copy changes" on the page, `TOO FAST` in the terminal): the flow view does not change that often.
88 - **The 200-name cap.** At most 200 route/reader pairs are tracked; a new pair beyond that is counted under reader `other`, so no read goes uncounted.
99 - **Not changed.** `/metrics` keeps `forge_telemetry_reads_total{route,reader}`; the window is the page's and terminal's alone. The terminal lists the top six rows and says how many more.
10 - **Agents.** The same rows (`route`, `reader`, `reads`, `meanIntervalMs`, `tooFast`, most reads first) come as `readers` in the MCP `system` tool and in `GET /api/v1/telemetry/system`, under the same gate as the rest of that answer: a forge admin, or an agent granted `telemetry:read`. A read is counted after its answer is built, so it is not in its own `readers`.
src/contract/openapi/routes/getApiV1TelemetrySystem.ts +1 −1
@@ -1,3 +1,3 @@
11 import type { Route } from "../shared.ts";
22
3 export const route: Route = { method: "get", path: "/api/v1/telemetry/system", summary: "Read the forge's system view; the owner or an agent holding telemetry:read, with a ro token; every agent read is logged", auth: true };
3 export const route: Route = { method: "get", path: "/api/v1/telemetry/system", summary: "Read the forge's system view, with `readers` (who polls the telemetry and how fast); the owner or an agent holding telemetry:read, with a ro token; every agent read is logged", auth: true };
src/http/api.ts +4 −3
@@ -62,7 +62,7 @@ import { takeOrTell } from "../actions/follow.ts";
6262 import { noteRead, telemetryReader, type Reader } from "../actions/telemetry.ts";
6363 import { flowCopy, flowEtag, flowMaxAge } from "../system/flow_cache.ts";
6464 import { countRead } from "../system/telemetry_reads.ts";
65 import { historyFor, sourcesOut, systemFor, welcome } from "../system/telemetry.ts";
65 import { historyFor, sourcesOut, systemWithReaders, welcome } from "../system/telemetry.ts";
6666 import { markReaders } from "../crew/direct.ts";
6767 import {
6868 addCommentThrough,
@@ -1197,13 +1197,14 @@ export function apiRoutes(router: Router<Handler>): void {
11971197 const reader = telemetryReader(ctx.app, ctx.auth.user);
11981198 if (!reader) return forbidden();
11991199 noteRead(ctx.app, ctx.auth.user, reader, what);
1200 countRead(what, ctx.auth.user.username);
1200 // Counted after the body, so a body that lists the readers does not count this read.
12011201 const body = timePhase(ctx.request, "read", () => out(ctx, reader));
1202 countRead(what, ctx.auth.user.username);
12021203 // A read may hand back a responder (headers, a 304); it runs in the render phase.
12031204 if (typeof body === "function") return timePhase(ctx.request, "render", body as () => Response);
12041205 return timePhase(ctx.request, "render", () => json(body));
12051206 };
1206 router.add("GET", "/api/v1/telemetry/system", telemetryRead("system", (c, r) => systemFor(c.app, r)));
1207 router.add("GET", "/api/v1/telemetry/system", telemetryRead("system", (c, r) => systemWithReaders(c.app, r)));
12071208 // The flow read tells a poller how fresh it is: the etag is the reader kind and
12081209 // the copy's computed-at time, so a repeat inside the TTL is a 304 (#9068).
12091210 router.add("GET", "/api/v1/telemetry/flow", telemetryRead("flow", (c, reader) => {
src/http/mcp/tools/system.ts +6 −3
@@ -1,10 +1,11 @@
11 import { noteRead, telemetryReader } from "../../../actions/telemetry.ts";
2 import { systemFor } from "../../../system/telemetry.ts";
2 import { systemWithReaders } from "../../../system/telemetry.ts";
3 import { countRead } from "../../../system/telemetry_reads.ts";
34 import { good, bad, type Tool } from "../shared.ts";
45
56 export const tool: Tool = {
67 name: "system",
7 doc: "the machine the forge runs on, now: CPU, load, memory, disk, network, processes; a forge admin, or an agent granted `telemetry:read`",
8 doc: "the machine the forge runs on, now: CPU, load, memory, disk, network, processes, and `readers`: who polls the telemetry and how fast; a forge admin, or an agent granted `telemetry:read`",
89 description:
910 "The forge machine now: CPU, memory, disk, network. Admin or telemetry:read.",
1011 inputSchema: { type: "object", properties: {} },
@@ -14,6 +15,8 @@ export const tool: Tool = {
1415 if (!reader)
1516 return bad("forbidden", "only an admin of this forge, or an agent granted telemetry:read, reads its machine");
1617 noteRead(ctx.app, user, reader, "system");
17 return good(systemFor(ctx.app, reader));
18 const out = systemWithReaders(ctx.app, reader);
19 countRead("system", user.username);
20 return good(out);
1821 },
1922 };
src/system/telemetry.ts +10
@@ -26,12 +26,22 @@ import { readVerified, verifiedSays } from "../flow/verified.ts";
2626 import type { Reader } from "../actions/telemetry.ts";
2727 import { agentView, live } from "./live.ts";
2828 import type { SystemView } from "./view.ts";
29 import { readerPaces, type ReaderPace } from "./telemetry_reads.ts";
2930
3031 export function systemFor(app: App, reader: Reader): SystemView {
3132 const view = live().read();
3233 return reader.kind === "agent" ? agentView(app, view) : view;
3334 }
3435
36 /**
37 * The system answer for the API and the MCP tool: the view plus `readers`, the
38 * rows /system draws (#9088). Names are accounts, so only a telemetry reader gets
39 * here. Call it before `countRead` so the read does not count into its own answer.
40 */
41 export function systemWithReaders(app: App, reader: Reader): SystemView & { readers: ReaderPace[] } {
42 return { ...systemFor(app, reader), readers: readerPaces() };
43 }
44
3545 /**
3646 * The pipeline answer as a generator that yields between its heavy reads, so a
3747 * caller may run each slice on its own turn of the loop (#8978); `flowFor`
test/telemetry-reads-readers.test.ts added +74
@@ -0,0 +1,74 @@
1 /** `readers` on the MCP system tool and GET /api/v1/telemetry/system (#9088): the rows /system draws, behind the telemetry gate. */
2 import { afterAll, beforeAll, expect, test } from "bun:test";
3 import { mkdtemp, rm } from "node:fs/promises";
4 import { tmpdir } from "node:os";
5 import { join } from "node:path";
6 import { createApp, type App } from "../src/app.ts";
7 import { createAgent } from "../src/actions/crew.ts";
8 import { setTelemetryGrant } from "../src/actions/telemetry.ts";
9 import { createToken } from "../src/core/tokens.ts";
10 import { createUser, getUserByName } from "../src/core/users.ts";
11 import { serve, type Servers } from "../src/http/server.ts";
12 import { countRead, resetTelemetryReads } from "../src/system/telemetry_reads.ts";
13 import { quietLogs } from "./helpers/globals.ts";
14
15 quietLogs();
16 let dir = "", app: App, servers: Servers, owner = "", granted = "", plain = "";
17
18 beforeAll(async () => {
19 dir = await mkdtemp(join(tmpdir(), "forge-readers-"));
20 app = createApp({ dataDir: dir, port: 0 });
21 servers = await serve(app);
22 const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-long-telemetry-passphrase", isAdmin: true });
23 if (!a.ok) throw new Error("alice");
24 owner = createToken(app.db, a.value.id, "t", "admin").token;
25 const alice = getUserByName(app.db, "alice")!;
26 for (const name of ["bot", "plain"]) {
27 const b = await createAgent(app, alice, { username: name });
28 if (!b.ok) throw new Error(name);
29 const t = createToken(app.db, b.value.user.id, "t", "ro").token;
30 if (name === "bot") granted = t; else plain = t;
31 }
32 setTelemetryGrant(app, alice, "bot", true);
33 }, 30_000);
34 afterAll(async () => { servers.stop(); app.close(); await rm(dir, { recursive: true, force: true }); });
35
36 const api = async (token: string) => {
37 const r = await fetch(`${servers.url}/api/v1/telemetry/system`, { headers: { authorization: `Bearer ${token}` } });
38 return { status: r.status, body: r.status === 200 ? ((await r.json()) as any) : await r.text() };
39 };
40 const mcp = async (token: string) => {
41 const r = await fetch(`${servers.url}/mcp`, {
42 method: "POST",
43 headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
44 body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "system", arguments: {} } }),
45 });
46 const b = (await r.json()) as any;
47 return b.error || b.result.isError ? { error: true, text: JSON.stringify(b) } : { error: false, body: JSON.parse(b.result.content[0].text) };
48 };
49
50 test("API and MCP carry the same readers rows, flagging a fast flow reader, without the read itself", async () => {
51 resetTelemetryReads();
52 const now = Date.now();
53 for (let i = 0; i < 6; i++) countRead("flow", "poller", now - 20_000 + i * 1000);
54 const viaApi = await api(granted);
55 expect(viaApi.status).toBe(200);
56 expect(viaApi.body.readers).toEqual([{ route: "flow", reader: "poller", reads: 6, meanIntervalMs: 1000, tooFast: true }]);
57 const viaMcp = await mcp(granted);
58 expect(viaMcp.error).toBe(false);
59 // The API read above is counted by now (route system, reader bot); the MCP answer lists it, not itself.
60 const rows = viaMcp.body.readers;
61 expect(rows[0]).toEqual(viaApi.body.readers[0]);
62 expect(rows.find((r: any) => r.route === "system" && r.reader === "bot").reads).toBe(1);
63 expect((await api(owner)).body.readers.find((r: any) => r.route === "system" && r.reader === "bot").reads).toBe(2);
64 });
65
66 test("an agent without the grant gets the refusal and no reader names", async () => {
67 countRead("flow", "secret-reader-name");
68 const r = await api(plain);
69 expect(r.status).toBe(403);
70 expect(r.body).not.toContain("secret-reader-name");
71 const m = await mcp(plain);
72 expect(m.error).toBe(true);
73 expect(m.text).not.toContain("secret-reader-name");
74 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs