eb39577
@@ -78,7 +78,11 @@
FORGE_APP_DATA=./data bun run src/tty.ts                # the terminal interface
  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.
  not what changed. `src/live-e2e.test.ts` is where the direction's own
  sentence — two people and one agent on one board, each seeing the
  others' moves as they happen — is checked end to end, over real HTTP
  and a real `/events` stream rather than the bus directly; it's the
  first place to look when live updates stop working.
- **Presence**: the header shows who *else* has this board's `/events`
  stream open right now — a forge username for a signed-in visitor (once,
  even if they have it open in two tabs), "Guest" for one anonymous tab,
@@ -0,0 +1,139 @@
/**
 * The direction's own sentence, asserted end to end: two people and one
 * agent on one board, each seeing the others' moves as they happen. This
 * file drives only public surfaces — HTTP requests and the SSE stream —
 * and imports nothing from src/bus.ts, so it keeps failing the day
 * something stops publishing even if the bus itself is rewritten.
 */
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-"));
process.env.PORT = "0";
const { server } = await import("./server.ts");
// Only used to seed a board and its default columns/cards directly — the
// same shortcut src/server.test.ts takes to skip the board-creation rate
// limit — never to read the bus or the events this test is checking for.
const db = await import("./db.ts");
const base = `http://${server.hostname}:${server.port}`;
type Frame = {
  kind?: string;
  boardId?: string;
  by?: string | null;
  client?: string | null;
  ids?: Record<string, string>;
};
/**
 * One open `/events` connection, named for the assertion messages: a
 * timeout here says which seat stopped hearing the bus, rather than
 * failing with a bare "timeout" that could be any of the three.
 */
function seat(name: string, stream: Response) {
  const reader = stream.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  async function rawFrame(timeoutMs: number): Promise<string> {
    let timer: ReturnType<typeof setTimeout>;
    const timeout = new Promise<never>((_, reject) => {
      timer = setTimeout(() => reject(new Error(`${name}'s stream went silent — the bus stopped reaching this seat`)), timeoutMs);
    });
    try {
      while (!buffer.includes("\n\n")) {
        const { value, done } = await Promise.race([reader.read(), timeout]);
        if (done) throw new Error(`${name}'s stream ended`);
        buffer += decoder.decode(value, { stream: true });
      }
    } finally {
      clearTimeout(timer!);
    }
    const at = buffer.indexOf("\n\n");
    const frame = buffer.slice(0, at);
    buffer = buffer.slice(at + 2);
    return frame;
  }
  // The next board-change frame, skipping heartbeats and presence frames,
  // failing loudly (naming this seat) if none lands inside timeoutMs.
  async function next(timeoutMs: number): Promise<Frame> {
    const deadline = Date.now() + timeoutMs;
    for (;;) {
      const remaining = deadline - Date.now();
      if (remaining <= 0) throw new Error(`${name} never saw the change within ${timeoutMs}ms — the bus stopped reaching this seat`);
      const frame = await rawFrame(remaining);
      if (frame.startsWith(":") || frame.startsWith("retry:") || frame.startsWith("event: presence")) continue;
      const data = frame.match(/^data: (.*)$/m)?.[1];
      if (!data) continue;
      return JSON.parse(data) as Frame;
    }
  }
  return { next, cancel: () => reader.cancel() };
}
// The direction's number is a second; this test waits up to two so a
// loaded machine doesn't flake, without letting a genuinely broken bus
// hang the suite.
const WITHIN = 2000;
test("two people and an agent on one board each see the others' moves within a second", async () => {
  const board = db.createBoard("Live e2e board");
  const [todo, doing] = db.listColumns(board.id);
  const card = db.createCard(board.id, todo.id, "Move me");
  if (card === "not_found" || card === "full") throw new Error("test setup failed to create a card");
  const streamA = await fetch(`${base}/b/${board.id}/events`);
  const streamB = await fetch(`${base}/b/${board.id}/events`);
  expect(streamA.headers.get("content-type")).toBe("text/event-stream");
  expect(streamB.headers.get("content-type")).toBe("text/event-stream");
  const a = seat("viewer A", streamA);
  const b = seat("viewer B", streamB);
  try {
    // Person A moves a card. Both A's own stream and B's must see it — A's
    // own frame identifies A as the writer, so A's tab can skip work it
    // already did, and B's frame is how B's page redraws without a reload.
    const moved = await fetch(`${base}/b/${board.id}/cards/${card.id}/move`, {
      method: "POST",
      redirect: "manual",
      headers: { "content-type": "application/x-www-form-urlencoded", "x-tempo-client": "A" },
      body: new URLSearchParams({ column_id: doing.id, index: "0" }),
    });
    expect(moved.status).toBe(303);
    const [movedForA, movedForB] = await Promise.all([a.next(WITHIN), b.next(WITHIN)]);
    expect(movedForA.kind).toBe("card.moved");
    expect(movedForA.client).toBe("A");
    expect(movedForA.ids?.cardId).toBe(card.id);
    expect(movedForB.kind).toBe("card.moved");
    expect(movedForB.ids?.cardId).toBe(card.id);
    // The agent takes its turn over MCP, with no tab and so no client id
    // of its own — both open seats still hear it within the same window.
    const call = 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: todo.id, title: "From the agent" } },
      }),
    });
    expect(call.status).toBe(200);
    const rpcBody = (await call.json()) as { result?: { isError?: boolean } };
    expect(rpcBody.result?.isError).toBeUndefined();
    const [addedForA, addedForB] = await Promise.all([a.next(WITHIN), b.next(WITHIN)]);
    expect(addedForA.kind).toBe("card.added");
    expect(addedForB.kind).toBe("card.added");
  } finally {
    await a.cancel();
    await b.cancel();
  }
});
		@@ -78,7 +78,11 @@ FORGE_APP_DATA=./data bun run src/tty.ts # the terminal interface
78	78	a separate process) reaches the same open page too — see `src/watch.ts`
79	79	below — arriving as a whole-board reconcile rather than one placed
80	80	card, since the poller that notices it only knows the board changed,
81		not what changed.
	81	not what changed. `src/live-e2e.test.ts` is where the direction's own
	82	sentence — two people and one agent on one board, each seeing the
	83	others' moves as they happen — is checked end to end, over real HTTP
	84	and a real `/events` stream rather than the bus directly; it's the
	85	first place to look when live updates stop working.
82	86	- Presence: the header shows who else has this board's `/events`
83	87	stream open right now — a forge username for a signed-in visitor (once,
84	88	even if they have it open in two tabs), "Guest" for one anonymous tab,
		@@ -0,0 +1,139 @@
	1	/**
	2	* The direction's own sentence, asserted end to end: two people and one
	3	* agent on one board, each seeing the others' moves as they happen. This
	4	* file drives only public surfaces — HTTP requests and the SSE stream —
	5	* and imports nothing from src/bus.ts, so it keeps failing the day
	6	* something stops publishing even if the bus itself is rewritten.
	7	*/
	8	import { expect, test } from "bun:test";
	9	import { mkdtempSync } from "node:fs";
	10	import { tmpdir } from "node:os";
	11	import { join } from "node:path";
	12
	13	process.env.FORGE_APP_DATA = mkdtempSync(join(tmpdir(), "app-"));
	14	process.env.PORT = "0";
	15	const { server } = await import("./server.ts");
	16	// Only used to seed a board and its default columns/cards directly — the
	17	// same shortcut src/server.test.ts takes to skip the board-creation rate
	18	// limit — never to read the bus or the events this test is checking for.
	19	const db = await import("./db.ts");
	20
	21	const base = `http://${server.hostname}:${server.port}`;
	22
	23	type Frame = {
	24	kind?: string;
	25	boardId?: string;
	26	by?: string \| null;
	27	client?: string \| null;
	28	ids?: Record<string, string>;
	29	};
	30
	31	/**
	32	* One open `/events` connection, named for the assertion messages: a
	33	* timeout here says which seat stopped hearing the bus, rather than
	34	* failing with a bare "timeout" that could be any of the three.
	35	*/
	36	function seat(name: string, stream: Response) {
	37	const reader = stream.body!.getReader();
	38	const decoder = new TextDecoder();
	39	let buffer = "";
	40
	41	async function rawFrame(timeoutMs: number): Promise<string> {
	42	let timer: ReturnType<typeof setTimeout>;
	43	const timeout = new Promise<never>((_, reject) => {
	44	timer = setTimeout(() => reject(new Error(`${name}'s stream went silent — the bus stopped reaching this seat`)), timeoutMs);
	45	});
	46	try {
	47	while (!buffer.includes("\n\n")) {
	48	const { value, done } = await Promise.race([reader.read(), timeout]);
	49	if (done) throw new Error(`${name}'s stream ended`);
	50	buffer += decoder.decode(value, { stream: true });
	51	}
	52	} finally {
	53	clearTimeout(timer!);
	54	}
	55	const at = buffer.indexOf("\n\n");
	56	const frame = buffer.slice(0, at);
	57	buffer = buffer.slice(at + 2);
	58	return frame;
	59	}
	60
	61	// The next board-change frame, skipping heartbeats and presence frames,
	62	// failing loudly (naming this seat) if none lands inside timeoutMs.
	63	async function next(timeoutMs: number): Promise<Frame> {
	64	const deadline = Date.now() + timeoutMs;
	65	for (;;) {
	66	const remaining = deadline - Date.now();
	67	if (remaining <= 0) throw new Error(`${name} never saw the change within ${timeoutMs}ms — the bus stopped reaching this seat`);
	68	const frame = await rawFrame(remaining);
	69	if (frame.startsWith(":") \|\| frame.startsWith("retry:") \|\| frame.startsWith("event: presence")) continue;
	70	const data = frame.match(/^data: (.*)$/m)?.[1];
	71	if (!data) continue;
	72	return JSON.parse(data) as Frame;
	73	}
	74	}
	75
	76	return { next, cancel: () => reader.cancel() };
	77	}
	78
	79	// The direction's number is a second; this test waits up to two so a
	80	// loaded machine doesn't flake, without letting a genuinely broken bus
	81	// hang the suite.
	82	const WITHIN = 2000;
	83
	84	test("two people and an agent on one board each see the others' moves within a second", async () => {
	85	const board = db.createBoard("Live e2e board");
	86	const [todo, doing] = db.listColumns(board.id);
	87	const card = db.createCard(board.id, todo.id, "Move me");
	88	if (card === "not_found" \|\| card === "full") throw new Error("test setup failed to create a card");
	89
	90	const streamA = await fetch(`${base}/b/${board.id}/events`);
	91	const streamB = await fetch(`${base}/b/${board.id}/events`);
	92	expect(streamA.headers.get("content-type")).toBe("text/event-stream");
	93	expect(streamB.headers.get("content-type")).toBe("text/event-stream");
	94	const a = seat("viewer A", streamA);
	95	const b = seat("viewer B", streamB);
	96
	97	try {
	98	// Person A moves a card. Both A's own stream and B's must see it — A's
	99	// own frame identifies A as the writer, so A's tab can skip work it
	100	// already did, and B's frame is how B's page redraws without a reload.
	101	const moved = await fetch(`${base}/b/${board.id}/cards/${card.id}/move`, {
	102	method: "POST",
	103	redirect: "manual",
	104	headers: { "content-type": "application/x-www-form-urlencoded", "x-tempo-client": "A" },
	105	body: new URLSearchParams({ column_id: doing.id, index: "0" }),
	106	});
	107	expect(moved.status).toBe(303);
	108
	109	const [movedForA, movedForB] = await Promise.all([a.next(WITHIN), b.next(WITHIN)]);
	110	expect(movedForA.kind).toBe("card.moved");
	111	expect(movedForA.client).toBe("A");
	112	expect(movedForA.ids?.cardId).toBe(card.id);
	113	expect(movedForB.kind).toBe("card.moved");
	114	expect(movedForB.ids?.cardId).toBe(card.id);
	115
	116	// The agent takes its turn over MCP, with no tab and so no client id
	117	// of its own — both open seats still hear it within the same window.
	118	const call = await fetch(`${base}/mcp`, {
	119	method: "POST",
	120	headers: { "content-type": "application/json" },
	121	body: JSON.stringify({
	122	jsonrpc: "2.0",
	123	id: 1,
	124	method: "tools/call",
	125	params: { name: "add_card", arguments: { board_id: board.id, column_id: todo.id, title: "From the agent" } },
	126	}),
	127	});
	128	expect(call.status).toBe(200);
	129	const rpcBody = (await call.json()) as { result?: { isError?: boolean } };
	130	expect(rpcBody.result?.isError).toBeUndefined();
	131
	132	const [addedForA, addedForB] = await Promise.all([a.next(WITHIN), b.next(WITHIN)]);
	133	expect(addedForA.kind).toBe("card.added");
	134	expect(addedForB.kind).toBe("card.added");
	135	} finally {
	136	await a.cancel();
	137	await b.cancel();
	138	}
	139	});