a329e1c · 7b4b057 · 978dfb0
@@ -323,6 +323,8 @@
and `FORGE_APP_USER`, the forge account at the keyboard or empty for a guest
it runs on: `POST $FORGE_URL/mcp` (`now`, `map`, `apps`, `usage` — every tool that
does not write), the JSON API, and `GET $FORGE_URL/events`, the live stream
of everything the owner may see, with `Authorization: Bearer $FORGE_TOKEN`.
It also sends `run.log` frames (`{ run, lines, truncated }`), a run's new log
lines: no id, never replayed, and dropped while the reader is behind.
`examples/forge-live` is the worked example, and the acceptance test
(`test/apps-token.test.ts`) deploys it.
@@ -413,7 +413,10 @@
carry is the no-script fallback, written inside `<noscript>` so a browser
running the script never even schedules it (removing the tag afterwards
would not cancel a refresh the parser had already set). Every live region
listens to one `EventSource` (`/events`,
everything the viewer may see) and works like this:
everything the viewer may see) and works like this. (The stream also carries
`run.log` frames, `{ run, lines, truncated }`, a run's new log lines to those
who may read its log: no id, never replayed, skipped for a reader that is
behind. A late tab reads the log file.)
- **An address of its own.** A region carries `data-live-src` — `/live/apps`
  and `/live/on-you` on the home, `<repo>/app/now` for _Now_ on the app
@@ -102,6 +102,22 @@
export function mayReadRunLog(
  );
}
/**
 * Whether `actor` may read this run's log: the run is theirs to see (app
 * gate, sealed report) and the log is theirs to read. The one question both
 * the log door and the live `run.log` frames ask.
 */
export function mayReadLogOfRun(
  app: App,
  actor: User | undefined,
  repo: Repo,
  runId: string,
): boolean {
  return (
    getRepoRun(app, actor, repo, runId).ok && mayReadRunLog(app, actor, repo)
  );
}
export function listRepoRuns(
  app: App,
  actor: User | undefined,
@@ -20,6 +20,9 @@
import type { User } from "../core/users.ts";
import { listEvents, listEventsVisibleTo } from "../core/events.ts";
import { audienceFilter } from "../core/event_audience.ts";
import { canRead, getRepoById } from "../core/repos.ts";
import { mayReadLogOfRun } from "../actions/runs.ts";
import { getRun } from "../core/runs.ts";
import { frameOf, tapRunLogs } from "../runs/log.ts";
import type { ForgeEvent } from "../core/events.ts";
const HEARTBEAT_MS = 25_000;
@@ -215,7 +218,7 @@
export function viewerStream(
  // admins', a person's own choices are theirs (`core/event_audience.ts`).
  const meant = audienceFilter(app.db, viewer);
  const after = resumeFrom(request);
  return eventStream(request, (emit) => {
  return eventStream(request, (emit, _close, ready) => {
    if (after !== undefined) {
      const missed = listEventsVisibleTo(app.db, viewer, {
        afterId: after,
@@ -224,11 +227,37 @@
export function viewerStream(
      for (const event of missed)
        if (meant(event)) emit("forge", named(event), event.id);
    }
    return app.bus.subscribe((event: ForgeEvent) => {
    // New log lines of a run whose log the viewer may read (`mayReadRunLog`). Not
    // history: no id, never replayed — a late tab reads the log file.
    const gone = new AbortController();
    // The verdict is decided once per run per connection, like repository
    // visibility: someone removed from the crew keeps this run's lines until
    // they reconnect.
    const verdicts = new Map<string, boolean>();
    tapRunLogs((runId, lines) => {
      let may = verdicts.get(runId);
      if (may === undefined) {
      
        sentinelcrew
        25d ago
        outdated
      
      High: visible is canRead, but a run's log needs mayReadRunLog + getRepoRun (sealed issue, app gate), the same check as runLog in src/actions/runs.ts. As written, signed-out viewers get every public repo's crew transcripts.
    
      
        sentinelcrew
        25d ago
        outdated
      
      Low: this verdict holds for the life of the connection, so someone removed from the crew keeps receiving this run's lines until they reconnect. Name that here, the way the viewerStream doc comment does for repository visibility.
    
        const run = getRun(app.db, runId);
        const repo = run && getRepoById(app.db, run.repoId);
        may = repo !== undefined && mayReadLogOfRun(app, viewer, repo, runId);
        // Bounded: a tab left open sees many runs go by.
        if (verdicts.size >= 256) verdicts.clear();
        verdicts.set(runId, may);
      
        sentinelcrew
        25d ago
        outdated
      
      High: this lossy frame shares the /events byte budget with the durable forge frames. A burst of up to 16 KiB lines that a slow reader hasn't drained gets the whole stream dropped by eventStream's 256 KiB budget. Skip the frame while the reader is behind: take ready from the register callback and use if (!may || !ready()) return;, the same way system.ts handles samples. Add a test that would fail without it.
    
      }
      // Lossy by design: a reader that is behind skips the frame rather than
      // spend the byte budget the durable `forge` frames need to stay open.
      if (!may || !ready()) return;
      emit("run.log", { run: runId, ...frameOf(lines) });
    }, gone.signal);
    const unsubscribe = app.bus.subscribe((event: ForgeEvent) => {
      if (event.repoId === null || !visible(event.repoId)) return;
      if (!meant(event)) return;
      emit("forge", named(event), event.id);
    });
    return () => {
      gone.abort();
      unsubscribe();
    };
  });
}
@@ -31,6 +31,47 @@
export type LogListener = (line: string) => void;
const listeners = new Map<string, Set<LogListener>>();
/** Who follows every run's log at once: the forge's `/events` streams. */
export type LogTap = (runId: string, lines: string[]) => void;
const taps = new Set<LogTap>();
/** Follow every run's new lines, redacted as written, until `signal` aborts. */
export function tapRunLogs(tap: LogTap, signal: AbortSignal): void {
  taps.add(tap);
  if (signal.aborted) taps.delete(tap);
  else signal.addEventListener("abort", () => taps.delete(tap), { once: true });
}
/** How many taps are open — a test seam for leaks. */
export const runLogTapCount = (): number => taps.size;
/** The most a single live frame carries. */
export const FRAME_MAX_LINES = 64;
      
        sentinelcrew
        25d ago
        
      
      Low: appendRunLog always hands the tap a single line, so FRAME_MAX_LINES never bites today. The PR body says so too. Fine to keep as the frame's contract, but drop it if batching never comes.
    
export const FRAME_MAX_BYTES = 16 * 1024;
/** Lines cut to one frame's size; `truncated` says something was left out. */
export function frameOf(lines: readonly string[]): {
  lines: string[];
  truncated: boolean;
} {
  const out: string[] = [];
  let bytes = 0;
  for (const line of lines) {
    const room = FRAME_MAX_BYTES - bytes;
    const size = Buffer.byteLength(line);
    if (out.length >= FRAME_MAX_LINES || room <= 0)
      return { lines: out, truncated: true };
      
        sentinelcrew
        25d ago
        
      
      Low: Buffer.subarray(0, room).toString("utf8") can cut through a multi-byte character. The partial bytes become U+FFFD (3 bytes), so the frame can go a couple of bytes over FRAME_MAX_BYTES. The test only checks ASCII. If you touch this again, back room off to a character boundary (while (buf[room] & 0xc0) === 0x80 room--).
    
    if (size > room) {
      out.push(Buffer.from(line).subarray(0, room).toString("utf8"));
      return { lines: out, truncated: true };
    }
    out.push(line);
    bytes += size;
  }
  return { lines: out, truncated: false };
}
/** A line without its newline; a trailing newline is added on write. */
const MAX_LINE = 16 * 1024;
@@ -43,6 +84,13 @@
export function appendRunLog(
  const clean = redactHeld(line.replace(/\r?\n/g, " ")).slice(0, MAX_LINE);
  mkdirSync(runsRoot(dataDir), { recursive: true });
  appendFileSync(runLogPath(dataDir, runId), `${clean}\n`);
  for (const tap of taps) {
    try {
      tap(runId, [clean]);
    } catch {
      // A tap's failure is its own; the run keeps logging.
    }
  }
  const watching = listeners.get(runId);
  if (watching === undefined) return;
  for (const listener of watching) {
@@ -0,0 +1,244 @@
/**
 * A run's new log lines ride the `/events` stream as `run.log` frames, to
 * people who can read the run's repository and no one else. They are not
 * history: no id, never replayed.
 */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { addToCrew, createAgent } from "../src/actions/crew.ts";
import { openIssue } from "../src/actions/issues.ts";
import { createRepo } from "../src/actions/repos.ts";
import { createRun, getRun } from "../src/core/runs.ts";
import { createToken } from "../src/core/tokens.ts";
import { createUser, type User } from "../src/core/users.ts";
import { releaseValues, holdValues } from "../src/kernel/log.ts";
import {
  FRAME_MAX_BYTES,
  FRAME_MAX_LINES,
  appendRunLog,
  frameOf,
  runLogTapCount,
} from "../src/runs/log.ts";
import { resetSseDroppedCount, sseDroppedCount, viewerStream } from "../src/http/sse.ts";
import { serve, type Servers } from "../src/http/server.ts";
describe("run.log frames on /events", () => {
  let dataDir: string;
  let app: App;
  let servers: Servers;
  let base: string;
  let alice: User;
  let aliceToken: string;
  let bobToken: string;
  let privateRun: string;
  let publicRun: string;
  let sealedRun: string;
  let crewToken: string;
  beforeAll(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-runlog-"));
    app = createApp({ dataDir, port: 0 });
    servers = await serve(app);
    base = servers.url;
    const mk = async (username: string, first = false) => {
      const u = await createUser(app.db, {
        origin: "seed",
        trusted: true,
        username,
        email: `${username}@forge.local`,
        password: "a-good-passphrase",
        adminIfFirst: first,
      });
      if (!u.ok) throw new Error(u.error);
      return u.value;
    };
    alice = await mk("alice", true);
    const bob = await mk("bob");
    aliceToken = createToken(app.db, alice.id, "t", "rw").token;
    bobToken = createToken(app.db, bob.id, "t", "rw").token;
    for (const [name, visibility] of [
      ["secret", "private"],
      ["open", "public"],
    ] as const) {
      const made = await createRepo(app, alice, { name, visibility });
      if (!made.ok) throw new Error(String(made.error));
      const run = createRun(app.db, { repoId: made.value.id, kind: "check" });
      if (name === "secret") privateRun = run.id;
      else {
        publicRun = run.id;
        const bot = await createAgent(app, alice, { username: "bot" });
        if (!bot.ok) throw new Error(String(bot.error));
        if (!addToCrew(app, alice, made.value, "bot").ok) throw new Error("crew");
        crewToken = createToken(app.db, bot.value.user.id, "t", "rw").token;
        const report = openIssue(app, alice, made.value, {
          title: "Hole",
          body: "detail",
          sealed: true,
        });
        if (!report.ok) throw new Error(String(report.error));
        sealedRun = createRun(app.db, {
          repoId: made.value.id,
          kind: "agent",
          issueId: report.value.id,
        }).id;
      }
    }
  }, 30_000);
  afterAll(async () => {
    servers.stop();
    app.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  /** Open /events, run `act`, and return the text received meanwhile. */
  async function listen(
    headers: Record<string, string>,
    act: () => void,
    path = "/events",
  ): Promise<string> {
    const ctl = new AbortController();
    const res = await fetch(`${base}${path}`, { headers, signal: ctl.signal });
    const reader = res.body!.getReader();
    await reader.read(); // ": connected"
    act();
    let text = "";
    const decoder = new TextDecoder();
    const stop = setTimeout(() => ctl.abort(), 300);
    try {
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break;
        text += decoder.decode(value);
      }
    } catch {
      // aborted: that is the end of the listening
    }
    clearTimeout(stop);
    return text;
  }
  const auth = (t: string) => ({ authorization: `Bearer ${t}` });
  test("a reader receives the frame, without an id", async () => {
    const text = await listen(auth(aliceToken), () =>
      appendRunLog(dataDir, privateRun, "hello from the run"),
      
        sentinelcrew
        25d ago
        outdated
      
      This asserts the leak. On the existing door a signed-out viewer gets 404 for this log (logHidden). Invert it: signed out → no frame. Add a writer/crew case that does get the frame, and a sealed-issue case.
    
    );
    expect(text).toContain("event: run.log\n");
    expect(text).not.toMatch(/^id: .*\n(?=event: run\.log)/m);
    const data = text.split("event: run.log\ndata: ")[1]!.split("\n")[0]!;
    expect(JSON.parse(data)).toEqual({
      run: privateRun,
      lines: ["hello from the run"],
      truncated: false,
    });
  });
  test("someone who cannot read the repository, or is signed out, gets nothing", async () => {
    for (const headers of [auth(bobToken), {}]) {
      const text = await listen(headers, () =>
        appendRunLog(dataDir, privateRun, "private line"),
      );
      expect(text).not.toContain("run.log");
      expect(text).not.toContain("private line");
    }
  });
  test("a public repository's log is for its writers and crew, not readers or the signed out", async () => {
    for (const headers of [{}, auth(bobToken)]) {
      const text = await listen(headers, () =>
        appendRunLog(dataDir, publicRun, "public line"),
      );
      expect(text).not.toContain("run.log");
      expect(text).not.toContain("public line");
    }
    for (const token of [aliceToken, crewToken]) {
      const text = await listen(auth(token), () =>
        appendRunLog(dataDir, publicRun, "public line"),
      );
      expect(text).toContain("public line");
    }
  });
  test("a run on a sealed issue reaches no one outside it", async () => {
    for (const headers of [{}, auth(bobToken)]) {
      const text = await listen(headers, () =>
        appendRunLog(dataDir, sealedRun, "sealed line"),
      );
      expect(text).not.toContain("sealed line");
    }
    const text = await listen(auth(aliceToken), () =>
      appendRunLog(dataDir, sealedRun, "sealed line"),
    );
    expect(text).toContain("sealed line");
  });
  test("a secret the redactor knows never appears in a frame", async () => {
    holdValues("test", ["hunter2-token-value"]);
    try {
      const text = await listen(auth(aliceToken), () =>
        appendRunLog(dataDir, privateRun, "curl -H hunter2-token-value"),
      );
      expect(text).toContain("run.log");
      expect(text).toContain("[redacted]");
      expect(text).not.toContain("hunter2-token-value");
    } finally {
      releaseValues("test");
    }
  });
  test("a reconnect with since replays no log frame", async () => {
    appendRunLog(dataDir, privateRun, "written before the reconnect");
    const text = await listen(auth(aliceToken), () => {}, "/events?since=0");
    expect(text).not.toContain("run.log");
    expect(text).not.toContain("written before");
  });
  test("a closed stream leaves no tap behind", async () => {
    await listen(auth(aliceToken), () => {});
    await Bun.sleep(50);
    expect(runLogTapCount()).toBe(0);
  });
  test("a burst to a reader that is not reading skips frames and keeps the stream open", async () => {
    resetSseDroppedCount();
    const res = viewerStream(app, alice, new Request(`${base}/events`));
    const reader = res.body!.getReader();
    const line = "x".repeat(16 * 1024);
    // Far more than the 256 KiB budget, with nobody reading.
    for (let i = 0; i < 64; i++) appendRunLog(dataDir, privateRun, line);
    expect(sseDroppedCount()).toBe(0);
    app.bus.append({
      kind: "push",
      repoId: (getRun(app.db, privateRun) as { repoId: string }).repoId,
      payload: { after: "burst" },
    });
    let text = "";
    const decoder = new TextDecoder();
    const stop = setTimeout(() => void reader.cancel(), 1000);
    while (!text.includes("event: forge")) {
      const { value, done } = await reader.read();
      if (done) break;
      text += decoder.decode(value);
    }
    clearTimeout(stop);
    await reader.cancel();
    expect(text).toContain("event: forge");
    expect(text).toContain("event: run.log"); // some got through, not all
    expect(sseDroppedCount()).toBe(0);
  });
  test("a frame is bounded", () => {
    const many = Array.from({ length: 200 }, (_, i) => `line ${i}`);
    const framed = frameOf(many);
    expect(framed.lines).toHaveLength(FRAME_MAX_LINES);
    expect(framed.truncated).toBe(true);
    const fat = frameOf(["x".repeat(FRAME_MAX_BYTES * 2)]);
    expect(Buffer.byteLength(fat.lines[0]!)).toBeLessThanOrEqual(FRAME_MAX_BYTES);
    expect(fat.truncated).toBe(true);
    expect(frameOf(["a", "b"])).toEqual({ lines: ["a", "b"], truncated: false });
  });
});
		@@ -323,6 +323,8 @@ and `FORGE_APP_USER`, the forge account at the keyboard or empty for a guest
323	323	it runs on: `POST $FORGE_URL/mcp` (`now`, `map`, `apps`, `usage` — every tool that
324	324	does not write), the JSON API, and `GET $FORGE_URL/events`, the live stream
325	325	of everything the owner may see, with `Authorization: Bearer $FORGE_TOKEN`.
	326	It also sends `run.log` frames (`{ run, lines, truncated }`), a run's new log
	327	lines: no id, never replayed, and dropped while the reader is behind.
326	328	`examples/forge-live` is the worked example, and the acceptance test
327	329	(`test/apps-token.test.ts`) deploys it.
328	330
		@@ -413,7 +413,10 @@ carry is the no-script fallback, written inside `<noscript>` so a browser
413	413	running the script never even schedules it (removing the tag afterwards
414	414	would not cancel a refresh the parser had already set). Every live region
415	415	listens to one `EventSource` (`/events`,
416		everything the viewer may see) and works like this:
	416	everything the viewer may see) and works like this. (The stream also carries
	417	`run.log` frames, `{ run, lines, truncated }`, a run's new log lines to those
	418	who may read its log: no id, never replayed, skipped for a reader that is
	419	behind. A late tab reads the log file.)
417	420
418	421	- An address of its own. A region carries `data-live-src` — `/live/apps`
419	422	and `/live/on-you` on the home, `<repo>/app/now` for _Now_ on the app
		@@ -102,6 +102,22 @@ export function mayReadRunLog(
102	102	);
103	103	}
104	104
	105	/**
	106	* Whether `actor` may read this run's log: the run is theirs to see (app
	107	* gate, sealed report) and the log is theirs to read. The one question both
	108	* the log door and the live `run.log` frames ask.
	109	*/
	110	export function mayReadLogOfRun(
	111	app: App,
	112	actor: User \| undefined,
	113	repo: Repo,
	114	runId: string,
	115	): boolean {
	116	return (
	117	getRepoRun(app, actor, repo, runId).ok && mayReadRunLog(app, actor, repo)
	118	);
	119	}
	120
105	121	export function listRepoRuns(
106	122	app: App,
107	123	actor: User \| undefined,
		@@ -20,6 +20,9 @@ import type { User } from "../core/users.ts";
20	20	import { listEvents, listEventsVisibleTo } from "../core/events.ts";
21	21	import { audienceFilter } from "../core/event_audience.ts";
22	22	import { canRead, getRepoById } from "../core/repos.ts";
	23	import { mayReadLogOfRun } from "../actions/runs.ts";
	24	import { getRun } from "../core/runs.ts";
	25	import { frameOf, tapRunLogs } from "../runs/log.ts";
23	26	import type { ForgeEvent } from "../core/events.ts";
24	27
25	28	const HEARTBEAT_MS = 25_000;
		@@ -215,7 +218,7 @@ export function viewerStream(
215	218	// admins', a person's own choices are theirs (`core/event_audience.ts`).
216	219	const meant = audienceFilter(app.db, viewer);
217	220	const after = resumeFrom(request);
218		return eventStream(request, (emit) => {
	221	return eventStream(request, (emit, _close, ready) => {
219	222	if (after !== undefined) {
220	223	const missed = listEventsVisibleTo(app.db, viewer, {
221	224	afterId: after,
		@@ -224,11 +227,37 @@ export function viewerStream(
224	227	for (const event of missed)
225	228	if (meant(event)) emit("forge", named(event), event.id);
226	229	}
227		return app.bus.subscribe((event: ForgeEvent) => {
	230	// New log lines of a run whose log the viewer may read (`mayReadRunLog`). Not
	231	// history: no id, never replayed — a late tab reads the log file.
	232	const gone = new AbortController();
	233	// The verdict is decided once per run per connection, like repository
	234	// visibility: someone removed from the crew keeps this run's lines until
	235	// they reconnect.
	236	const verdicts = new Map<string, boolean>();
	237	tapRunLogs((runId, lines) => {
	238	let may = verdicts.get(runId);
	239	if (may === undefined) {
		sentinelcrew 25d ago outdated High: `visible` is `canRead`, but a run's log needs `mayReadRunLog` + `getRepoRun` (sealed issue, app gate), the same check as `runLog` in `src/actions/runs.ts`. As written, signed-out viewers get every public repo's crew transcripts. sentinelcrew 25d ago outdated Low: this verdict holds for the life of the connection, so someone removed from the crew keeps receiving this run's lines until they reconnect. Name that here, the way the `viewerStream` doc comment does for repository visibility.
	240	const run = getRun(app.db, runId);
	241	const repo = run && getRepoById(app.db, run.repoId);
	242	may = repo !== undefined && mayReadLogOfRun(app, viewer, repo, runId);
	243	// Bounded: a tab left open sees many runs go by.
	244	if (verdicts.size >= 256) verdicts.clear();
	245	verdicts.set(runId, may);
		sentinelcrew 25d ago outdated High: this lossy frame shares the `/events` byte budget with the durable `forge` frames. A burst of up to 16 KiB lines that a slow reader hasn't drained gets the whole stream dropped by `eventStream`'s 256 KiB budget. Skip the frame while the reader is behind: take `ready` from the register callback and use `if (!may \|\| !ready()) return;`, the same way `system.ts` handles samples. Add a test that would fail without it.
	246	}
	247	// Lossy by design: a reader that is behind skips the frame rather than
	248	// spend the byte budget the durable `forge` frames need to stay open.
	249	if (!may \|\| !ready()) return;
	250	emit("run.log", { run: runId, ...frameOf(lines) });
	251	}, gone.signal);
	252	const unsubscribe = app.bus.subscribe((event: ForgeEvent) => {
228	253	if (event.repoId === null \|\| !visible(event.repoId)) return;
229	254	if (!meant(event)) return;
230	255	emit("forge", named(event), event.id);
231	256	});
	257	return () => {
	258	gone.abort();
	259	unsubscribe();
	260	};
232	261	});
233	262	}
234	263
		@@ -31,6 +31,47 @@ export type LogListener = (line: string) => void;
31	31
32	32	const listeners = new Map<string, Set<LogListener>>();
33	33
	34	/** Who follows every run's log at once: the forge's `/events` streams. */
	35	export type LogTap = (runId: string, lines: string[]) => void;
	36
	37	const taps = new Set<LogTap>();
	38
	39	/** Follow every run's new lines, redacted as written, until `signal` aborts. */
	40	export function tapRunLogs(tap: LogTap, signal: AbortSignal): void {
	41	taps.add(tap);
	42	if (signal.aborted) taps.delete(tap);
	43	else signal.addEventListener("abort", () => taps.delete(tap), { once: true });
	44	}
	45
	46	/** How many taps are open — a test seam for leaks. */
	47	export const runLogTapCount = (): number => taps.size;
	48
	49	/** The most a single live frame carries. */
	50	export const FRAME_MAX_LINES = 64;
		sentinelcrew 25d ago Low: `appendRunLog` always hands the tap a single line, so `FRAME_MAX_LINES` never bites today. The PR body says so too. Fine to keep as the frame's contract, but drop it if batching never comes.
	51	export const FRAME_MAX_BYTES = 16 * 1024;
	52
	53	/** Lines cut to one frame's size; `truncated` says something was left out. */
	54	export function frameOf(lines: readonly string[]): {
	55	lines: string[];
	56	truncated: boolean;
	57	} {
	58	const out: string[] = [];
	59	let bytes = 0;
	60	for (const line of lines) {
	61	const room = FRAME_MAX_BYTES - bytes;
	62	const size = Buffer.byteLength(line);
	63	if (out.length >= FRAME_MAX_LINES \|\| room <= 0)
	64	return { lines: out, truncated: true };
		sentinelcrew 25d ago Low: `Buffer.subarray(0, room).toString("utf8")` can cut through a multi-byte character. The partial bytes become U+FFFD (3 bytes), so the frame can go a couple of bytes over `FRAME_MAX_BYTES`. The test only checks ASCII. If you touch this again, back `room` off to a character boundary (while `(buf[room] & 0xc0) === 0x80` room--).
	65	if (size > room) {
	66	out.push(Buffer.from(line).subarray(0, room).toString("utf8"));
	67	return { lines: out, truncated: true };
	68	}
	69	out.push(line);
	70	bytes += size;
	71	}
	72	return { lines: out, truncated: false };
	73	}
	74
34	75	/** A line without its newline; a trailing newline is added on write. */
35	76	const MAX_LINE = 16 * 1024;
36	77
		@@ -43,6 +84,13 @@ export function appendRunLog(
43	84	const clean = redactHeld(line.replace(/\r?\n/g, " ")).slice(0, MAX_LINE);
44	85	mkdirSync(runsRoot(dataDir), { recursive: true });
45	86	appendFileSync(runLogPath(dataDir, runId), `${clean}\n`);
	87	for (const tap of taps) {
	88	try {
	89	tap(runId, [clean]);
	90	} catch {
	91	// A tap's failure is its own; the run keeps logging.
	92	}
	93	}
46	94	const watching = listeners.get(runId);
47	95	if (watching === undefined) return;
48	96	for (const listener of watching) {
		@@ -0,0 +1,244 @@
	1	/**
	2	* A run's new log lines ride the `/events` stream as `run.log` frames, to
	3	* people who can read the run's repository and no one else. They are not
	4	* history: no id, never replayed.
	5	*/
	6	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	7	import { mkdtemp, rm } from "node:fs/promises";
	8	import { tmpdir } from "node:os";
	9	import { join } from "node:path";
	10	import { createApp, type App } from "../src/app.ts";
	11	import { addToCrew, createAgent } from "../src/actions/crew.ts";
	12	import { openIssue } from "../src/actions/issues.ts";
	13	import { createRepo } from "../src/actions/repos.ts";
	14	import { createRun, getRun } from "../src/core/runs.ts";
	15	import { createToken } from "../src/core/tokens.ts";
	16	import { createUser, type User } from "../src/core/users.ts";
	17	import { releaseValues, holdValues } from "../src/kernel/log.ts";
	18	import {
	19	FRAME_MAX_BYTES,
	20	FRAME_MAX_LINES,
	21	appendRunLog,
	22	frameOf,
	23	runLogTapCount,
	24	} from "../src/runs/log.ts";
	25	import { resetSseDroppedCount, sseDroppedCount, viewerStream } from "../src/http/sse.ts";
	26	import { serve, type Servers } from "../src/http/server.ts";
	27
	28	describe("run.log frames on /events", () => {
	29	let dataDir: string;
	30	let app: App;
	31	let servers: Servers;
	32	let base: string;
	33	let alice: User;
	34	let aliceToken: string;
	35	let bobToken: string;
	36	let privateRun: string;
	37	let publicRun: string;
	38	let sealedRun: string;
	39	let crewToken: string;
	40
	41	beforeAll(async () => {
	42	dataDir = await mkdtemp(join(tmpdir(), "forge-runlog-"));
	43	app = createApp({ dataDir, port: 0 });
	44	servers = await serve(app);
	45	base = servers.url;
	46	const mk = async (username: string, first = false) => {
	47	const u = await createUser(app.db, {
	48	origin: "seed",
	49	trusted: true,
	50	username,
	51	email: `${username}@forge.local`,
	52	password: "a-good-passphrase",
	53	adminIfFirst: first,
	54	});
	55	if (!u.ok) throw new Error(u.error);
	56	return u.value;
	57	};
	58	alice = await mk("alice", true);
	59	const bob = await mk("bob");
	60	aliceToken = createToken(app.db, alice.id, "t", "rw").token;
	61	bobToken = createToken(app.db, bob.id, "t", "rw").token;
	62	for (const [name, visibility] of [
	63	["secret", "private"],
	64	["open", "public"],
	65	] as const) {
	66	const made = await createRepo(app, alice, { name, visibility });
	67	if (!made.ok) throw new Error(String(made.error));
	68	const run = createRun(app.db, { repoId: made.value.id, kind: "check" });
	69	if (name === "secret") privateRun = run.id;
	70	else {
	71	publicRun = run.id;
	72	const bot = await createAgent(app, alice, { username: "bot" });
	73	if (!bot.ok) throw new Error(String(bot.error));
	74	if (!addToCrew(app, alice, made.value, "bot").ok) throw new Error("crew");
	75	crewToken = createToken(app.db, bot.value.user.id, "t", "rw").token;
	76	const report = openIssue(app, alice, made.value, {
	77	title: "Hole",
	78	body: "detail",
	79	sealed: true,
	80	});
	81	if (!report.ok) throw new Error(String(report.error));
	82	sealedRun = createRun(app.db, {
	83	repoId: made.value.id,
	84	kind: "agent",
	85	issueId: report.value.id,
	86	}).id;
	87	}
	88	}
	89	}, 30_000);
	90
	91	afterAll(async () => {
	92	servers.stop();
	93	app.close();
	94	await rm(dataDir, { recursive: true, force: true });
	95	});
	96
	97	/** Open /events, run `act`, and return the text received meanwhile. */
	98	async function listen(
	99	headers: Record<string, string>,
	100	act: () => void,
	101	path = "/events",
	102	): Promise<string> {
	103	const ctl = new AbortController();
	104	const res = await fetch(`${base}${path}`, { headers, signal: ctl.signal });
	105	const reader = res.body!.getReader();
	106	await reader.read(); // ": connected"
	107	act();
	108	let text = "";
	109	const decoder = new TextDecoder();
	110	const stop = setTimeout(() => ctl.abort(), 300);
	111	try {
	112	for (;;) {
	113	const { value, done } = await reader.read();
	114	if (done) break;
	115	text += decoder.decode(value);
	116	}
	117	} catch {
	118	// aborted: that is the end of the listening
	119	}
	120	clearTimeout(stop);
	121	return text;
	122	}
	123
	124	const auth = (t: string) => ({ authorization: `Bearer ${t}` });
	125
	126	test("a reader receives the frame, without an id", async () => {
	127	const text = await listen(auth(aliceToken), () =>
	128	appendRunLog(dataDir, privateRun, "hello from the run"),
		sentinelcrew 25d ago outdated This asserts the leak. On the existing door a signed-out viewer gets 404 for this log (`logHidden`). Invert it: signed out → no frame. Add a writer/crew case that does get the frame, and a sealed-issue case.
	129	);
	130	expect(text).toContain("event: run.log\n");
	131	expect(text).not.toMatch(/^id: .*\n(?=event: run\.log)/m);
	132	const data = text.split("event: run.log\ndata: ")[1]!.split("\n")[0]!;
	133	expect(JSON.parse(data)).toEqual({
	134	run: privateRun,
	135	lines: ["hello from the run"],
	136	truncated: false,
	137	});
	138	});
	139
	140	test("someone who cannot read the repository, or is signed out, gets nothing", async () => {
	141	for (const headers of [auth(bobToken), {}]) {
	142	const text = await listen(headers, () =>
	143	appendRunLog(dataDir, privateRun, "private line"),
	144	);
	145	expect(text).not.toContain("run.log");
	146	expect(text).not.toContain("private line");
	147	}
	148	});
	149
	150	test("a public repository's log is for its writers and crew, not readers or the signed out", async () => {
	151	for (const headers of [{}, auth(bobToken)]) {
	152	const text = await listen(headers, () =>
	153	appendRunLog(dataDir, publicRun, "public line"),
	154	);
	155	expect(text).not.toContain("run.log");
	156	expect(text).not.toContain("public line");
	157	}
	158	for (const token of [aliceToken, crewToken]) {
	159	const text = await listen(auth(token), () =>
	160	appendRunLog(dataDir, publicRun, "public line"),
	161	);
	162	expect(text).toContain("public line");
	163	}
	164	});
	165
	166	test("a run on a sealed issue reaches no one outside it", async () => {
	167	for (const headers of [{}, auth(bobToken)]) {
	168	const text = await listen(headers, () =>
	169	appendRunLog(dataDir, sealedRun, "sealed line"),
	170	);
	171	expect(text).not.toContain("sealed line");
	172	}
	173	const text = await listen(auth(aliceToken), () =>
	174	appendRunLog(dataDir, sealedRun, "sealed line"),
	175	);
	176	expect(text).toContain("sealed line");
	177	});
	178
	179	test("a secret the redactor knows never appears in a frame", async () => {
	180	holdValues("test", ["hunter2-token-value"]);
	181	try {
	182	const text = await listen(auth(aliceToken), () =>
	183	appendRunLog(dataDir, privateRun, "curl -H hunter2-token-value"),
	184	);
	185	expect(text).toContain("run.log");
	186	expect(text).toContain("[redacted]");
	187	expect(text).not.toContain("hunter2-token-value");
	188	} finally {
	189	releaseValues("test");
	190	}
	191	});
	192
	193	test("a reconnect with since replays no log frame", async () => {
	194	appendRunLog(dataDir, privateRun, "written before the reconnect");
	195	const text = await listen(auth(aliceToken), () => {}, "/events?since=0");
	196	expect(text).not.toContain("run.log");
	197	expect(text).not.toContain("written before");
	198	});
	199
	200	test("a closed stream leaves no tap behind", async () => {
	201	await listen(auth(aliceToken), () => {});
	202	await Bun.sleep(50);
	203	expect(runLogTapCount()).toBe(0);
	204	});
	205
	206	test("a burst to a reader that is not reading skips frames and keeps the stream open", async () => {
	207	resetSseDroppedCount();
	208	const res = viewerStream(app, alice, new Request(`${base}/events`));
	209	const reader = res.body!.getReader();
	210	const line = "x".repeat(16 * 1024);
	211	// Far more than the 256 KiB budget, with nobody reading.
	212	for (let i = 0; i < 64; i++) appendRunLog(dataDir, privateRun, line);
	213	expect(sseDroppedCount()).toBe(0);
	214	app.bus.append({
	215	kind: "push",
	216	repoId: (getRun(app.db, privateRun) as { repoId: string }).repoId,
	217	payload: { after: "burst" },
	218	});
	219	let text = "";
	220	const decoder = new TextDecoder();
	221	const stop = setTimeout(() => void reader.cancel(), 1000);
	222	while (!text.includes("event: forge")) {
	223	const { value, done } = await reader.read();
	224	if (done) break;
	225	text += decoder.decode(value);
	226	}
	227	clearTimeout(stop);
	228	await reader.cancel();
	229	expect(text).toContain("event: forge");
	230	expect(text).toContain("event: run.log"); // some got through, not all
	231	expect(sseDroppedCount()).toBe(0);
	232	});
	233
	234	test("a frame is bounded", () => {
	235	const many = Array.from({ length: 200 }, (_, i) => `line ${i}`);
	236	const framed = frameOf(many);
	237	expect(framed.lines).toHaveLength(FRAME_MAX_LINES);
	238	expect(framed.truncated).toBe(true);
	239	const fat = frameOf(["x".repeat(FRAME_MAX_BYTES * 2)]);
	240	expect(Buffer.byteLength(fat.lines[0]!)).toBeLessThanOrEqual(FRAME_MAX_BYTES);
	241	expect(fat.truncated).toBe(true);
	242	expect(frameOf(["a", "b"])).toEqual({ lines: ["a", "b"], truncated: false });
	243	});
	244	});