2969845 · 44baecf · cb2e92a
@@ -74,7 +74,11 @@
FORGE_APP_DATA=./data bun run src/tty.ts                # the terminal interface
  the 503 this same server answers with once a board has 100 tabs open
  on it (`MAX_SUBSCRIBERS_PER_BOARD` in `src/limits.ts`). The button
  opens a fresh connection and, once it's up, reconciles the same way
  an automatic reconnect does.
  an automatic reconnect does. A move made in the terminal (`src/tty.ts`,
  a separate process) reaches the same open page too — see `src/watch.ts`
  below — arriving as a whole-board reconcile rather than one placed
  card, since the poller that notices it only knows the board changed,
  not what changed.
- **Agents**: `POST /mcp` speaks MCP over plain JSON-RPC 2.0
  (`initialize`, `tools/list`, `tools/call`) with five tools —
  `list_boards`, `list_columns`, `list_cards`, `add_card`, `move_card` —
@@ -147,12 +151,12 @@
background, reverting if it fails.
  came from the page or another terminal. A redraw never disturbs what
  you're doing: an open prompt keeps its text, the filter stays, and the
  cursor rides the same card or column (found by id) rather than jumping
  when the list changes under it. It's one-way: only `src/routes.ts` and
  `src/mcp.ts` call `bus.publish`, and the terminal writes straight to
  `src/db.ts`, so a change made *in* the terminal still doesn't reach an
  open web page until its next load — the revision table makes that cheap
  to close (the server could poll it and publish too), but that's outside
  this issue.
  when the list changes under it. The other direction rides the same
  revision counter: `src/watch.ts` polls it, about twice a second, for
  every board with an open `/events` listener, and publishes one `external`
  event (`by: null`) when it moved without this process having published it
  itself — so a card moved *in* the terminal reaches an open web page within
  a second too, the same way a live edit from another tab does.
## Limits
@@ -15,7 +15,8 @@
export type EventKind =
  | "column.moved"
  | "column.deleted"
  | "board.renamed"
  | "undo";
  | "undo"
  | "external";
export type BoardEvent = {
  readonly kind: EventKind;
@@ -48,3 +49,8 @@
export function subscribe(boardId: string, fn: (event: BoardEvent) => void): ()
export function subscriberCount(boardId: string): number {
  return boards.get(boardId)?.size ?? 0;
}
/** Every board id with at least one open listener — what `src/watch.ts` needs to know which boards are worth polling. */
export function subscribedBoardIds(): string[] {
  return [...boards.keys()];
}
@@ -114,10 +114,12 @@
function logAction(boardId: string, payload: Payload): void {
}
// The terminal (src/tty.ts) is a separate process on a pty; it cannot hear
// the web server's in-process event bus (src/live.ts), so it polls this
// instead. A scope's revision is a counter bumped in the same transaction
// as every write that changes what that scope shows, never derived from
// the action log — the log is trimmed and undone, so it is not a clock.
// the web server's in-process event bus (src/bus.ts), so it polls this
// instead — and src/watch.ts polls it right back, for the direction a
// change made in the terminal has to travel to reach the bus. A scope's
// revision is a counter bumped in the same transaction as every write that
// changes what that scope shows, never derived from the action log — the
// log is trimmed and undone, so it is not a clock.
// "__boards__" is not a valid board id (those are 8 lowercase/digit
// characters from randomId), so it cannot collide with one.
const BOARDS_SCOPE = "__boards__";
@@ -12,11 +12,13 @@
import * as bus from "./bus.ts";
import * as db from "./db.ts";
import { MAX_CARDS_PER_BOARD, writeBucket } from "./limits.ts";
import { cleanText } from "./text.ts";
import { noteRevision } from "./watch.ts";
import type { Who } from "./who.ts";
// An agent has no tab of its own, so there is nothing to echo-suppress.
function publish(boardId: string, who: Who, kind: bus.EventKind, ids: Record<string, string>): void {
  bus.publish(boardId, { kind, boardId, by: who.role === "anon" ? null : { id: who.id, name: who.name }, client: null, ids });
  noteRevision(boardId, db.boardRevision(boardId));
}
const BOARD_ID = /^[a-z0-9]{8}$/;
@@ -289,6 +289,6 @@
const boardShortcuts: [string, string][] = [
// Named by content — a change to static/board.js gets a new hash here,
// so the year-long cache on /static/ never serves a stale script.
export const BOARD_JS = "board.822340210.js";
export const BOARD_JS = "board.351899687.js";
export const THEME_JS = "theme.1025564933.js";
export const BOARDS_JS = "boards.608180649.js";
@@ -11,6 +11,7 @@
import { boardCreationBucket, callerKey, MAX_CARDS_PER_BOARD, MAX_COLUMNS_PER_BO
import { rpc } from "./mcp.ts";
import { boardPage, landingPage } from "./pages/board.ts";
import { cleanText } from "./text.ts";
import { noteRevision } from "./watch.ts";
import type { Who } from "./who.ts";
const BOARD_ID = /^[a-z0-9]{8}$/;
@@ -32,6 +33,7 @@
function publish(boardId: string, request: Request, who: Who, kind: EventKind, i
    client: request.headers.get(CLIENT_HEADER)?.slice(0, 64) ?? null,
    ids,
  });
  noteRevision(boardId, db.boardRevision(boardId));
}
// Pages carry a visitor's name and sign-out link once they're signed in, so
@@ -1,7 +1,7 @@
import { expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { existsSync, mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { dirname, join } from "node:path";
import { subscriberCount } from "./bus.ts";
process.env.FORGE_APP_DATA = mkdtempSync(join(tmpdir(), "app-"));
@@ -884,3 +884,64 @@
test("list_boards over MCP only ever shows the caller's own boards", async () =>
  const carol = JSON.parse((await (await call(identityHeaders("u5", "carol"))).json()).result.content[0].text);
  expect(carol.some((b: { name: string }) => b.name === "Mine")).toBe(true);
});
// Same shape as db.test.ts's own-process-writes-while-another-holds-the-lock
// tests: a second process, writing through db.ts exactly the way src/tty.ts
// does (never touching this process's in-memory bus), stands in for the
// terminal. src/watch.ts polls boardRevision and republishes for it, so an
// open /events stream hears about a move it could never have heard about
// any other way.
test("a card moved by a second process (standing in for the ssh terminal) reaches an open stream", async () => {
  const id = await newBoard();
  const page = await (await fetch(base + `/b/${id}`)).text();
  const [todoId, doingId] = [...page.matchAll(/data-column-id="([a-z0-9]{12})"/g)].map((m) => m[1]);
  const created = await post(`/b/${id}/columns/${todoId}/cards`, { title: "Moved from the terminal" });
  const { id: cardId } = (await created.json()) as { id: string };
  const stream = await fetch(base + `/b/${id}/events`);
  const reader = stream.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  async function nextFrame(): Promise<string> {
    while (!buffer.includes("\n\n")) {
      const { value, done } = await reader.read();
      if (done) throw new Error("stream ended");
      buffer += decoder.decode(value, { stream: true });
    }
    const at = buffer.indexOf("\n\n");
    const frame = buffer.slice(0, at);
    buffer = buffer.slice(at + 2);
    return frame;
  }
  const script = `
    import * as db from "./src/db.ts";
    db.moveCard(process.env.BOARD_ID, process.env.CARD_ID, process.env.COLUMN_ID, 0);
  `;
  const child = Bun.spawn({
    cmd: [process.execPath, "-e", script],
    cwd: join(import.meta.dir, ".."),
    // The live server's actual data directory, not process.env.FORGE_APP_DATA:
    // bun test shares one module cache (and so one server, one db.ts) across
    // every test file in the run, and each file's own top-level assignment to
    // that env var only wins the race to be read once, by whichever file's
    // import of db.ts happens to run first — reading it back here would as
    // likely name a different file's directory as this server's own.
    env: { ...process.env, FORGE_APP_DATA: dirname(db.db.filename), BOARD_ID: id, CARD_ID: cardId, COLUMN_ID: doingId },
    stderr: "pipe",
  });
  const exitCode = await child.exited;
  if (exitCode !== 0) console.error(await new Response(child.stderr).text());
  expect(exitCode).toBe(0);
  let event: { kind: string; boardId: string; by: unknown } | undefined;
  for (let i = 0; i < 10 && !event; i++) {
    const frame = await nextFrame();
    const data = frame.match(/^data: (.*)$/m)?.[1];
    if (data) event = JSON.parse(data);
  }
  expect(event?.kind).toBe("external");
  expect(event?.boardId).toBe(id);
  expect(event).not.toHaveProperty("by");
  await reader.cancel();
}, 10000);
@@ -11,6 +11,7 @@
import "./db.ts";
import { readBody } from "./body.ts";
import { MAX_REQUEST_BODY_BYTES, MAX_REQUEST_DRAIN_BYTES } from "./limits.ts";
import { route } from "./routes.ts";
import { startWatching } from "./watch.ts";
import { who } from "./who.ts";
const env = process.env.FORGE_APP_ENV ?? "local";
@@ -124,4 +125,8 @@
export const server = Bun.serve({
  },
});
// A move made in src/tty.ts (a separate process on a pty) reaches every
// open board page through this: see src/watch.ts.
startWatching();
console.log(`tempo listening on http://${server.hostname}:${server.port} (${env})`);
@@ -0,0 +1,77 @@
/**
 * Unit-level: pollExternalWrites against bus.ts and db.ts directly, no
 * HTTP server. src/server.test.ts covers the cross-process, over-the-wire
 * shape of the same behaviour.
 */
import { expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
process.env.FORGE_APP_DATA = mkdtempSync(join(tmpdir(), "app-"));
const db = await import("./db.ts");
const bus = await import("./bus.ts");
const watch = await import("./watch.ts");
function setup() {
  const board = db.createBoard("Board");
  const column = db.listColumns(board.id)[0];
  return { boardId: board.id, columnId: column.id };
}
test("a board with no listener is never polled or published to", () => {
  const { boardId, columnId } = setup();
  db.createCard(boardId, columnId, "first");
  watch.pollExternalWrites();
  db.createCard(boardId, columnId, "second");
  const heard: unknown[] = [];
  const stop = bus.subscribe(boardId, (e) => heard.push(e));
  watch.pollExternalWrites(); // first look after subscribing only sets the baseline
  expect(heard).toEqual([]);
  stop();
});
test("a write made outside publish() reaches a listener within one poll, as external with by: null", () => {
  const { boardId, columnId } = setup();
  const heard: unknown[] = [];
  const stop = bus.subscribe(boardId, (e) => heard.push(e));
  watch.pollExternalWrites(); // establishes the baseline for this newly-subscribed board
  expect(heard).toEqual([]);
  db.createCard(boardId, columnId, "written elsewhere"); // e.g. a move made in src/tty.ts
  watch.pollExternalWrites();
  expect(heard).toEqual([{ kind: "external", boardId, by: null, client: null, ids: {} }]);
  stop();
});
test("a write already announced through noteRevision is not announced again by the poller", () => {
  const { boardId, columnId } = setup();
  const heard: unknown[] = [];
  const stop = bus.subscribe(boardId, (e) => heard.push(e));
  watch.pollExternalWrites();
  db.createCard(boardId, columnId, "announced by routes.ts");
  watch.noteRevision(boardId, db.boardRevision(boardId));
  watch.pollExternalWrites();
  expect(heard).toEqual([]);
  stop();
});
test("the tick after everyone leaves forgets the board, so a later resubscribe does not replay a stale change as new", () => {
  const { boardId, columnId } = setup();
  const stop1 = bus.subscribe(boardId, () => {});
  watch.pollExternalWrites(); // baseline set while the first listener is open
  stop1();
  watch.pollExternalWrites(); // the next tick, with no listener left, forgets that baseline
  db.createCard(boardId, columnId, "while nobody was watching");
  const heard: unknown[] = [];
  const stop2 = bus.subscribe(boardId, (e) => heard.push(e));
  watch.pollExternalWrites(); // first look after resubscribing only re-establishes the baseline
  expect(heard).toEqual([]);
  stop2();
});
@@ -0,0 +1,42 @@
/**
 * The bridge from a write made outside this process to the in-process bus
 * (`src/bus.ts`). `src/tty.ts` runs on its own pty process and writes
 * through `src/db.ts` directly — it cannot call `bus.publish`, the
 * listeners live in here. So this polls `db.boardRevision` for every board
 * with an open listener and publishes one `external` event when it moved.
 * A write this process already published itself (see the `publish` helpers
 * in `src/routes.ts` and `src/mcp.ts`) calls `noteRevision` too, so the
 * same change is never announced twice.
 */
import { publish, subscribedBoardIds } from "./bus.ts";
import { boardRevision } from "./db.ts";
const seen = new Map<string, number>();
/** Record the revision a board's own write already announced through the bus. */
export function noteRevision(boardId: string, revision: number): void {
  seen.set(boardId, revision);
}
/** One pass: publish `external` for any watched board whose revision moved since it was last looked at. */
export function pollExternalWrites(): void {
  const active = subscribedBoardIds();
  for (const boardId of active) {
    const revision = boardRevision(boardId);
    const before = seen.get(boardId);
    seen.set(boardId, revision);
    if (before !== undefined && before !== revision) {
      publish(boardId, { kind: "external", boardId, by: null, client: null, ids: {} });
    }
  }
  const activeSet = new Set(active);
  for (const boardId of seen.keys()) if (!activeSet.has(boardId)) seen.delete(boardId);
}
/** Starts polling on an interval; returns a function that stops it. Unref'd so it never keeps the process alive on its own. */
export function startWatching(intervalMs = 500): () => void {
  const timer = setInterval(pollExternalWrites, intervalMs);
  timer.unref?.();
  return () => clearInterval(timer);
}
@@ -7,6 +7,17 @@
 * shares one key with theirs (see callerKey). `afterAll` below undoes the
 * drain once this file is done, so this is safe regardless of what order
 * bun runs test files in.
 *
 * writeBucket refills off the real clock, so a drained bucket only stays
 * empty for as long as no real time passes before the write it's meant to
 * refuse lands. A request sent over an actual socket (like every other test
 * file's `fetch`) hands that landing to the OS's own scheduling, which is
 * plenty of time for a token to trickle back under load — the bucket's
 * 300/minute rate needs only ~200ms to hand back one. So this file calls
 * `route()` directly, the same function the real server's `fetch` handler
 * calls after its own request setup, instead of going over the network:
 * same rate-limiting code, same writeBucket, but no socket round trip
 * between draining and the write actually being checked.
 */
import { afterAll, expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
@@ -17,17 +28,30 @@
process.env.FORGE_APP_DATA = mkdtempSync(join(tmpdir(), "app-"));
process.env.PORT = "0";
process.env.HOST = "127.0.0.1";
const { server } = await import("./server.ts");
const { route } = await import("./routes.ts");
const { callerKey, writeBucket } = await import("./limits.ts");
const { ANON } = await import("./who.ts");
const db = await import("./db.ts");
const base = `http://${server.hostname}:${server.port}`;
// Every request below is anonymous over loopback, so this is the one key
// they all spend from — the same key a real flood from this address would.
const anonKey = callerKey(ANON, "127.0.0.1");
// Every request below is anonymous from this made-up address, so this is
// the one key they all spend from — the same key a real flood from it would.
const anonIp = "127.0.0.1";
const anonKey = callerKey(ANON, anonIp);
const post = (path: string, body: Record<string, string> = {}, headers: Record<string, string> = {}) =>
  fetch(base + path, { method: "POST", headers: { "content-type": "application/x-www-form-urlencoded", ...headers }, body: new URLSearchParams(body) });
function post(path: string, body: Record<string, string> = {}, headers: Record<string, string> = {}): Promise<Response> {
  const request = new Request(base + path, {
    method: "POST",
    headers: { "content-type": "application/x-www-form-urlencoded", ...headers },
    body: new URLSearchParams(body),
  });
  return route(request, new URL(request.url), "test", ANON, anonIp).then((r) => r ?? new Response("not found", { status: 404 }));
}
const mcp = (body: unknown): Promise<Response> => {
  const request = new Request(base + "/mcp", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) });
  return route(request, new URL(request.url), "test", ANON, anonIp).then((r) => r ?? new Response("not found", { status: 404 }));
};
function drainWriteBucket(): void {
  while (writeBucket.take(anonKey)) {
@@ -82,7 +106,11 @@
test("a flood retrying the same refusal without navigating gets the cheap answer
  const refused = await post(`${boardUrl}/rename`, { name: "New name" });
  expect(refused.status).toBe(429);
  expect(refused.headers.get("retry-after")).toMatch(/^\d+$/);
  expect(refused.headers.get("content-type")).not.toContain("text/html");
  // route() is called directly here (see the file comment), so this never
  // reaches Bun's socket writer — the one that would fill in a default
  // text/plain content-type for a string body with none set, the way a real
  // client would see it. Absent is just as much "not html" as text/plain is.
  expect(refused.headers.get("content-type") ?? "").not.toContain("text/html");
  expect(await refused.text()).toBe("too many changes just now — try again shortly");
});
@@ -92,15 +120,11 @@
test("an MCP write past the limit comes back as a readable tool error, not a tra
  drainWriteBucket();
  const response = await fetch(base + "/mcp", {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      jsonrpc: "2.0",
      id: 1,
      method: "tools/call",
      params: { name: "add_card", arguments: { board_id: board.id, column_id: columnId, title: "should be refused" } },
    }),
  const response = await mcp({
    jsonrpc: "2.0",
    id: 1,
    method: "tools/call",
    params: { name: "add_card", arguments: { board_id: board.id, column_id: columnId, title: "should be refused" } },
  });
  expect(response.status).toBe(200); // JSON-RPC over HTTP: the transport succeeded
  const rpc = await response.json();
@@ -111,14 +135,13 @@
test("an MCP write past the limit comes back as a readable tool error, not a tra
test("a drained bucket still lets initialize, tools/list and the read-only tools answer", async () => {
  drainWriteBucket();
  const call = (body: unknown) => fetch(base + "/mcp", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) });
  const init = await (await call({ jsonrpc: "2.0", id: 1, method: "initialize" })).json();
  const init = await (await mcp({ jsonrpc: "2.0", id: 1, method: "initialize" })).json();
  expect(init.result.serverInfo.name).toBe("tempo");
  const listed = await (await call({ jsonrpc: "2.0", id: 2, method: "tools/list" })).json();
  const listed = await (await mcp({ jsonrpc: "2.0", id: 2, method: "tools/list" })).json();
  expect(listed.result.tools.length).toBeGreaterThan(0);
  const boards = await (await call({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "list_boards", arguments: {} } })).json();
  const boards = await (await mcp({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "list_boards", arguments: {} } })).json();
  expect(boards.result.isError).toBeUndefined();
});
@@ -425,6 +425,10 @@
      const focused = root.querySelector(".board :focus");
      return !!focused && (focused.tagName === "INPUT" || focused.tagName === "SELECT");
    }
    // A move made from the terminal (src/tty.ts, on a separate process) is
    // reported the same way as undo: its scope isn't known here either, so
    // it waits out anything mid-type or mid-drag, then reconciles the whole
    // board rather than guessing what changed.
    // An event that would clobber something the user is mid-typing (or
    // mid-drag) waits until they're done rather than overwriting it. A tab
@@ -450,7 +454,7 @@
        (event.kind === "card.moved" && dragging && dragging.el.dataset.cardId === event.ids.cardId) ||
        (event.kind === "column.renamed" && isRenamingColumn(event.ids.columnId)) ||
        (event.kind === "board.renamed" && isRenamingBoard()) ||
        (event.kind === "undo" && (dragging || isTypingSomewhere()))
        ((event.kind === "undo" || event.kind === "external") && (dragging || isTypingSomewhere()))
      ) {
        pending.push(event);
        if (pending.length > PENDING_LIMIT) {
@@ -500,7 +504,7 @@
        for (const opt of root.querySelectorAll(`select[name=column_id] option[value="${CSS.escape(ids.columnId)}"]`)) opt.remove();
        return;
      }
      if (kind === "undo") return reconcile();
      if (kind === "undo" || kind === "external") return reconcile();
      const fresh = await freshBoard();
      if (!fresh) return;
		@@ -74,7 +74,11 @@ FORGE_APP_DATA=./data bun run src/tty.ts # the terminal interface
74	74	the 503 this same server answers with once a board has 100 tabs open
75	75	on it (`MAX_SUBSCRIBERS_PER_BOARD` in `src/limits.ts`). The button
76	76	opens a fresh connection and, once it's up, reconciles the same way
77		an automatic reconnect does.
	77	an automatic reconnect does. A move made in the terminal (`src/tty.ts`,
	78	a separate process) reaches the same open page too — see `src/watch.ts`
	79	below — arriving as a whole-board reconcile rather than one placed
	80	card, since the poller that notices it only knows the board changed,
	81	not what changed.
78	82	- Agents: `POST /mcp` speaks MCP over plain JSON-RPC 2.0
79	83	(`initialize`, `tools/list`, `tools/call`) with five tools —
80	84	`list_boards`, `list_columns`, `list_cards`, `add_card`, `move_card` —
		@@ -147,12 +151,12 @@ background, reverting if it fails.
147	151	came from the page or another terminal. A redraw never disturbs what
148	152	you're doing: an open prompt keeps its text, the filter stays, and the
149	153	cursor rides the same card or column (found by id) rather than jumping
150		when the list changes under it. It's one-way: only `src/routes.ts` and
151		`src/mcp.ts` call `bus.publish`, and the terminal writes straight to
152		`src/db.ts`, so a change made in the terminal still doesn't reach an
153		open web page until its next load — the revision table makes that cheap
154		to close (the server could poll it and publish too), but that's outside
155		this issue.
	154	when the list changes under it. The other direction rides the same
	155	revision counter: `src/watch.ts` polls it, about twice a second, for
	156	every board with an open `/events` listener, and publishes one `external`
	157	event (`by: null`) when it moved without this process having published it
	158	itself — so a card moved in the terminal reaches an open web page within
	159	a second too, the same way a live edit from another tab does.
156	160
157	161	## Limits
158	162
		@@ -15,7 +15,8 @@ export type EventKind =
15	15	\| "column.moved"
16	16	\| "column.deleted"
17	17	\| "board.renamed"
18		\| "undo";
	18	\| "undo"
	19	\| "external";
19	20
20	21	export type BoardEvent = {
21	22	readonly kind: EventKind;
		@@ -48,3 +49,8 @@ export function subscribe(boardId: string, fn: (event: BoardEvent) => void): ()
48	49	export function subscriberCount(boardId: string): number {
49	50	return boards.get(boardId)?.size ?? 0;
50	51	}
	52
	53	/** Every board id with at least one open listener — what `src/watch.ts` needs to know which boards are worth polling. */
	54	export function subscribedBoardIds(): string[] {
	55	return [...boards.keys()];
	56	}
		@@ -114,10 +114,12 @@ function logAction(boardId: string, payload: Payload): void {
114	114	}
115	115
116	116	// The terminal (src/tty.ts) is a separate process on a pty; it cannot hear
117		// the web server's in-process event bus (src/live.ts), so it polls this
118		// instead. A scope's revision is a counter bumped in the same transaction
119		// as every write that changes what that scope shows, never derived from
120		// the action log — the log is trimmed and undone, so it is not a clock.
	117	// the web server's in-process event bus (src/bus.ts), so it polls this
	118	// instead — and src/watch.ts polls it right back, for the direction a
	119	// change made in the terminal has to travel to reach the bus. A scope's
	120	// revision is a counter bumped in the same transaction as every write that
	121	// changes what that scope shows, never derived from the action log — the
	122	// log is trimmed and undone, so it is not a clock.
121	123	// "__boards__" is not a valid board id (those are 8 lowercase/digit
122	124	// characters from randomId), so it cannot collide with one.
123	125	const BOARDS_SCOPE = "__boards__";
		@@ -12,11 +12,13 @@ import * as bus from "./bus.ts";
12	12	import * as db from "./db.ts";
13	13	import { MAX_CARDS_PER_BOARD, writeBucket } from "./limits.ts";
14	14	import { cleanText } from "./text.ts";
	15	import { noteRevision } from "./watch.ts";
15	16	import type { Who } from "./who.ts";
16	17
17	18	// An agent has no tab of its own, so there is nothing to echo-suppress.
18	19	function publish(boardId: string, who: Who, kind: bus.EventKind, ids: Record<string, string>): void {
19	20	bus.publish(boardId, { kind, boardId, by: who.role === "anon" ? null : { id: who.id, name: who.name }, client: null, ids });
	21	noteRevision(boardId, db.boardRevision(boardId));
20	22	}
21	23
22	24	const BOARD_ID = /^[a-z0-9]{8}$/;
		@@ -289,6 +289,6 @@ const boardShortcuts: [string, string][] = [
289	289
290	290	// Named by content — a change to static/board.js gets a new hash here,
291	291	// so the year-long cache on /static/ never serves a stale script.
292		export const BOARD_JS = "board.822340210.js";
	292	export const BOARD_JS = "board.351899687.js";
293	293	export const THEME_JS = "theme.1025564933.js";
294	294	export const BOARDS_JS = "boards.608180649.js";
		@@ -11,6 +11,7 @@ import { boardCreationBucket, callerKey, MAX_CARDS_PER_BOARD, MAX_COLUMNS_PER_BO
11	11	import { rpc } from "./mcp.ts";
12	12	import { boardPage, landingPage } from "./pages/board.ts";
13	13	import { cleanText } from "./text.ts";
	14	import { noteRevision } from "./watch.ts";
14	15	import type { Who } from "./who.ts";
15	16
16	17	const BOARD_ID = /^[a-z0-9]{8}$/;
		@@ -32,6 +33,7 @@ function publish(boardId: string, request: Request, who: Who, kind: EventKind, i
32	33	client: request.headers.get(CLIENT_HEADER)?.slice(0, 64) ?? null,
33	34	ids,
34	35	});
	36	noteRevision(boardId, db.boardRevision(boardId));
35	37	}
36	38
37	39	// Pages carry a visitor's name and sign-out link once they're signed in, so
		@@ -1,7 +1,7 @@
1	1	import { expect, test } from "bun:test";
2		import { mkdtempSync } from "node:fs";
	2	import { existsSync, mkdtempSync, rmSync } from "node:fs";
3	3	import { tmpdir } from "node:os";
4		import { join } from "node:path";
	4	import { dirname, join } from "node:path";
5	5	import { subscriberCount } from "./bus.ts";
6	6
7	7	process.env.FORGE_APP_DATA = mkdtempSync(join(tmpdir(), "app-"));
		@@ -884,3 +884,64 @@ test("list_boards over MCP only ever shows the caller's own boards", async () =>
884	884	const carol = JSON.parse((await (await call(identityHeaders("u5", "carol"))).json()).result.content[0].text);
885	885	expect(carol.some((b: { name: string }) => b.name === "Mine")).toBe(true);
886	886	});
	887
	888	// Same shape as db.test.ts's own-process-writes-while-another-holds-the-lock
	889	// tests: a second process, writing through db.ts exactly the way src/tty.ts
	890	// does (never touching this process's in-memory bus), stands in for the
	891	// terminal. src/watch.ts polls boardRevision and republishes for it, so an
	892	// open /events stream hears about a move it could never have heard about
	893	// any other way.
	894	test("a card moved by a second process (standing in for the ssh terminal) reaches an open stream", async () => {
	895	const id = await newBoard();
	896	const page = await (await fetch(base + `/b/${id}`)).text();
	897	const [todoId, doingId] = [...page.matchAll(/data-column-id="([a-z0-9]{12})"/g)].map((m) => m[1]);
	898	const created = await post(`/b/${id}/columns/${todoId}/cards`, { title: "Moved from the terminal" });
	899	const { id: cardId } = (await created.json()) as { id: string };
	900
	901	const stream = await fetch(base + `/b/${id}/events`);
	902	const reader = stream.body!.getReader();
	903	const decoder = new TextDecoder();
	904	let buffer = "";
	905	async function nextFrame(): Promise<string> {
	906	while (!buffer.includes("\n\n")) {
	907	const { value, done } = await reader.read();
	908	if (done) throw new Error("stream ended");
	909	buffer += decoder.decode(value, { stream: true });
	910	}
	911	const at = buffer.indexOf("\n\n");
	912	const frame = buffer.slice(0, at);
	913	buffer = buffer.slice(at + 2);
	914	return frame;
	915	}
	916
	917	const script = `
	918	import * as db from "./src/db.ts";
	919	db.moveCard(process.env.BOARD_ID, process.env.CARD_ID, process.env.COLUMN_ID, 0);
	920	`;
	921	const child = Bun.spawn({
	922	cmd: [process.execPath, "-e", script],
	923	cwd: join(import.meta.dir, ".."),
	924	// The live server's actual data directory, not process.env.FORGE_APP_DATA:
	925	// bun test shares one module cache (and so one server, one db.ts) across
	926	// every test file in the run, and each file's own top-level assignment to
	927	// that env var only wins the race to be read once, by whichever file's
	928	// import of db.ts happens to run first — reading it back here would as
	929	// likely name a different file's directory as this server's own.
	930	env: { ...process.env, FORGE_APP_DATA: dirname(db.db.filename), BOARD_ID: id, CARD_ID: cardId, COLUMN_ID: doingId },
	931	stderr: "pipe",
	932	});
	933	const exitCode = await child.exited;
	934	if (exitCode !== 0) console.error(await new Response(child.stderr).text());
	935	expect(exitCode).toBe(0);
	936
	937	let event: { kind: string; boardId: string; by: unknown } \| undefined;
	938	for (let i = 0; i < 10 && !event; i++) {
	939	const frame = await nextFrame();
	940	const data = frame.match(/^data: (.*)$/m)?.[1];
	941	if (data) event = JSON.parse(data);
	942	}
	943	expect(event?.kind).toBe("external");
	944	expect(event?.boardId).toBe(id);
	945	expect(event).not.toHaveProperty("by");
	946	await reader.cancel();
	947	}, 10000);
		@@ -11,6 +11,7 @@ import "./db.ts";
11	11	import { readBody } from "./body.ts";
12	12	import { MAX_REQUEST_BODY_BYTES, MAX_REQUEST_DRAIN_BYTES } from "./limits.ts";
13	13	import { route } from "./routes.ts";
	14	import { startWatching } from "./watch.ts";
14	15	import { who } from "./who.ts";
15	16
16	17	const env = process.env.FORGE_APP_ENV ?? "local";
		@@ -124,4 +125,8 @@ export const server = Bun.serve({
124	125	},
125	126	});
126	127
	128	// A move made in src/tty.ts (a separate process on a pty) reaches every
	129	// open board page through this: see src/watch.ts.
	130	startWatching();
	131
127	132	console.log(`tempo listening on http://${server.hostname}:${server.port} (${env})`);
		@@ -0,0 +1,77 @@
	1	/**
	2	* Unit-level: pollExternalWrites against bus.ts and db.ts directly, no
	3	* HTTP server. src/server.test.ts covers the cross-process, over-the-wire
	4	* shape of the same behaviour.
	5	*/
	6	import { expect, test } from "bun:test";
	7	import { mkdtempSync } from "node:fs";
	8	import { tmpdir } from "node:os";
	9	import { join } from "node:path";
	10
	11	process.env.FORGE_APP_DATA = mkdtempSync(join(tmpdir(), "app-"));
	12	const db = await import("./db.ts");
	13	const bus = await import("./bus.ts");
	14	const watch = await import("./watch.ts");
	15
	16	function setup() {
	17	const board = db.createBoard("Board");
	18	const column = db.listColumns(board.id)[0];
	19	return { boardId: board.id, columnId: column.id };
	20	}
	21
	22	test("a board with no listener is never polled or published to", () => {
	23	const { boardId, columnId } = setup();
	24	db.createCard(boardId, columnId, "first");
	25	watch.pollExternalWrites();
	26	db.createCard(boardId, columnId, "second");
	27	const heard: unknown[] = [];
	28	const stop = bus.subscribe(boardId, (e) => heard.push(e));
	29	watch.pollExternalWrites(); // first look after subscribing only sets the baseline
	30	expect(heard).toEqual([]);
	31	stop();
	32	});
	33
	34	test("a write made outside publish() reaches a listener within one poll, as external with by: null", () => {
	35	const { boardId, columnId } = setup();
	36	const heard: unknown[] = [];
	37	const stop = bus.subscribe(boardId, (e) => heard.push(e));
	38
	39	watch.pollExternalWrites(); // establishes the baseline for this newly-subscribed board
	40	expect(heard).toEqual([]);
	41
	42	db.createCard(boardId, columnId, "written elsewhere"); // e.g. a move made in src/tty.ts
	43	watch.pollExternalWrites();
	44
	45	expect(heard).toEqual([{ kind: "external", boardId, by: null, client: null, ids: {} }]);
	46	stop();
	47	});
	48
	49	test("a write already announced through noteRevision is not announced again by the poller", () => {
	50	const { boardId, columnId } = setup();
	51	const heard: unknown[] = [];
	52	const stop = bus.subscribe(boardId, (e) => heard.push(e));
	53	watch.pollExternalWrites();
	54
	55	db.createCard(boardId, columnId, "announced by routes.ts");
	56	watch.noteRevision(boardId, db.boardRevision(boardId));
	57	watch.pollExternalWrites();
	58
	59	expect(heard).toEqual([]);
	60	stop();
	61	});
	62
	63	test("the tick after everyone leaves forgets the board, so a later resubscribe does not replay a stale change as new", () => {
	64	const { boardId, columnId } = setup();
	65	const stop1 = bus.subscribe(boardId, () => {});
	66	watch.pollExternalWrites(); // baseline set while the first listener is open
	67	stop1();
	68	watch.pollExternalWrites(); // the next tick, with no listener left, forgets that baseline
	69
	70	db.createCard(boardId, columnId, "while nobody was watching");
	71
	72	const heard: unknown[] = [];
	73	const stop2 = bus.subscribe(boardId, (e) => heard.push(e));
	74	watch.pollExternalWrites(); // first look after resubscribing only re-establishes the baseline
	75	expect(heard).toEqual([]);
	76	stop2();
	77	});
		@@ -0,0 +1,42 @@
	1	/**
	2	* The bridge from a write made outside this process to the in-process bus
	3	* (`src/bus.ts`). `src/tty.ts` runs on its own pty process and writes
	4	* through `src/db.ts` directly — it cannot call `bus.publish`, the
	5	* listeners live in here. So this polls `db.boardRevision` for every board
	6	* with an open listener and publishes one `external` event when it moved.
	7	* A write this process already published itself (see the `publish` helpers
	8	* in `src/routes.ts` and `src/mcp.ts`) calls `noteRevision` too, so the
	9	* same change is never announced twice.
	10	*/
	11
	12	import { publish, subscribedBoardIds } from "./bus.ts";
	13	import { boardRevision } from "./db.ts";
	14
	15	const seen = new Map<string, number>();
	16
	17	/** Record the revision a board's own write already announced through the bus. */
	18	export function noteRevision(boardId: string, revision: number): void {
	19	seen.set(boardId, revision);
	20	}
	21
	22	/** One pass: publish `external` for any watched board whose revision moved since it was last looked at. */
	23	export function pollExternalWrites(): void {
	24	const active = subscribedBoardIds();
	25	for (const boardId of active) {
	26	const revision = boardRevision(boardId);
	27	const before = seen.get(boardId);
	28	seen.set(boardId, revision);
	29	if (before !== undefined && before !== revision) {
	30	publish(boardId, { kind: "external", boardId, by: null, client: null, ids: {} });
	31	}
	32	}
	33	const activeSet = new Set(active);
	34	for (const boardId of seen.keys()) if (!activeSet.has(boardId)) seen.delete(boardId);
	35	}
	36
	37	/** Starts polling on an interval; returns a function that stops it. Unref'd so it never keeps the process alive on its own. */
	38	export function startWatching(intervalMs = 500): () => void {
	39	const timer = setInterval(pollExternalWrites, intervalMs);
	40	timer.unref?.();
	41	return () => clearInterval(timer);
	42	}