Merge branch 'feature/text-file-embeds' into dev

2 files changed, 339 insertions, 0 deletions

diff --git a/src/features/markdown/adapter/text-file.test.ts b/src/features/markdown/adapter/text-file.test.ts
new file mode 100644
index 0000000..2597271
--- /dev/null
+++ b/src/features/markdown/adapter/text-file.test.ts
@@ -0,0 +1,168 @@
+import { describe, expect, it, vi } from "vitest";
+import type { TextFileState } from "../types";
+import { createTextFileClient } from "./text-file";
+
+const response = (data: unknown, status = 200) => new Response(JSON.stringify(data), { status });
+const context = { conversationId: "c", cwd: "/project", computerId: null };
+const file = (text = "# Plan") => ({
+  path: "/project/a.md",
+  text,
+  format: "markdown",
+  bytes: new TextEncoder().encode(text).length,
+  maxBytes: 262144,
+});
+const flush = async () => {
+  for (let i = 0; i < 20; i++) await Promise.resolve();
+};
+describe("context-bound live-file client", () => {
+  it("preflights once, sends both guards and encodes literal path once; dedups repeated references", async () => {
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockImplementation(async (url) =>
+        response(String(url).endsWith("-context") ? context : file()),
+      );
+    const client = createTextFileClient("/api", "c", fetcher);
+    const a: TextFileState[] = [],
+      b: TextFileState[] = [];
+    const offA = client.subscribe("notes/100%.md", (s) => a.push(s));
+    const offB = client.subscribe("notes/100%.md", (s) => b.push(s));
+    await flush();
+    expect(fetcher).toHaveBeenCalledTimes(2);
+    const url = new URL(String(fetcher.mock.calls[1]?.[0]), "http://localhost");
+    expect(Object.fromEntries(url.searchParams)).toEqual({
+      path: "notes/100%.md",
+      expectedCwd: "/project",
+      expectedComputerId: "",
+    });
+    expect(fetcher.mock.calls[1]?.[1]?.cache).toBe("no-store");
+    expect(a.at(-1)?.kind).toBe("ready");
+    expect(b.at(-1)?.kind).toBe("ready");
+    offA();
+    expect(fetcher.mock.calls[1]?.[1]?.signal?.aborted).toBe(false);
+    offB();
+    expect(fetcher.mock.calls[1]?.[1]?.signal?.aborted).toBe(true);
+    client.subscribe("notes/100%.md", () => {});
+    await flush();
+    expect(fetcher).toHaveBeenCalledTimes(4);
+    client.dispose();
+  });
+  it("never reads a file before context, handles remote inheritance explicitly", async () => {
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockResolvedValue(response({ ...context, computerId: "remote" }));
+    const client = createTextFileClient("", "c", fetcher);
+    const states: TextFileState[] = [];
+    client.subscribe("/same-local-path.md", (s) => states.push(s));
+    await flush();
+    expect(fetcher).toHaveBeenCalledTimes(1);
+    expect(states.at(-1)).toMatchObject({
+      kind: "error",
+      message: expect.stringContaining("remote computers"),
+    });
+    client.dispose();
+  });
+  it.each([
+    400, 404, 409, 413, 422,
+  ])("reports HTTP %i then Retry refreshes authoritative context and live file", async (status) => {
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockResolvedValueOnce(response(context))
+      .mockResolvedValueOnce(response({ error: `failure ${status}` }, status))
+      .mockResolvedValueOnce(response(context))
+      .mockResolvedValueOnce(response(file("changed")));
+    const client = createTextFileClient("", "c", fetcher);
+    const states: TextFileState[] = [];
+    client.subscribe("a.md", (s) => states.push(s));
+    await flush();
+    expect(states.at(-1)).toEqual({ kind: "error", message: `failure ${status}` });
+    client.refresh("a.md");
+    await flush();
+    expect(states.at(-1)).toMatchObject({ kind: "ready", file: { text: "changed" } });
+    expect(fetcher).toHaveBeenCalledTimes(4);
+    client.dispose();
+  });
+  it("guards late responses after refresh and disposal", async () => {
+    let finish: (r: Response) => void = () => {};
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockResolvedValueOnce(response(context))
+      .mockImplementationOnce(
+        () =>
+          new Promise((r) => {
+            finish = r;
+          }),
+      )
+      .mockResolvedValueOnce(response(context))
+      .mockResolvedValueOnce(response(file("new")));
+    const client = createTextFileClient("", "c", fetcher);
+    const states: TextFileState[] = [];
+    client.subscribe("a.md", (s) => states.push(s));
+    await flush();
+    client.refresh("a.md");
+    await flush();
+    finish(response(file("old")));
+    await flush();
+    expect(states.at(-1)).toMatchObject({ file: { text: "new" } });
+    client.dispose();
+    expect(fetcher.mock.calls[3]?.[1]?.signal?.aborted).toBe(true);
+  });
+  it("bounds active cache to32 files/8MiB source and concurrency to4, disposes queue", async () => {
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockImplementation(async (url) =>
+        String(url).endsWith("-context") ? response(context) : new Promise(() => {}),
+      );
+    const client = createTextFileClient("", "c", fetcher);
+    const overflow: TextFileState[] = [];
+    for (let i = 0; i < 40; i++)
+      client.subscribe(`${i}.md`, (s) => {
+        if (i >= 32) overflow.push(s);
+      });
+    await flush();
+    expect(fetcher).toHaveBeenCalledTimes(5);
+    expect(overflow).toHaveLength(8);
+    expect(overflow[0]).toMatchObject({ kind: "error" });
+    client.dispose();
+    for (const call of fetcher.mock.calls) expect(call[1]?.signal?.aborted).toBe(true);
+  });
+  it("rejects oversized or malformed success data instead of truncating", async () => {
+    const fetcher = vi
+      .fn<typeof fetch>()
+      .mockResolvedValueOnce(response(context))
+      .mockResolvedValueOnce(response(file("a".repeat(262145))));
+    const client = createTextFileClient("", "c", fetcher);
+    const states: TextFileState[] = [];
+    client.subscribe("a.md", (s) => states.push(s));
+    await flush();
+    expect(states.at(-1)?.kind).toBe("error");
+    client.dispose();
+  });
+});
+
+it("invalidates every live document and reloads guards before publishing refreshed context", async () => {
+  let generation = 0;
+  const fetcher = vi.fn<typeof fetch>().mockImplementation(async (url) => {
+    if (String(url).endsWith("-context"))
+      return response({ ...context, cwd: generation ? "/new" : "/project" });
+    const params = new URL(String(url), "http://localhost").searchParams;
+    return response({
+      ...file(generation ? "new" : "old"),
+      path: `${params.get("expectedCwd")}/a.md`,
+    });
+  });
+  const client = createTextFileClient("", "c", fetcher);
+  const a: TextFileState[] = [],
+    b: TextFileState[] = [];
+  client.subscribe("a.md", (state) => a.push(state));
+  client.subscribe("b.md", (state) => b.push(state));
+  await flush();
+  generation++;
+  client.refresh("a.md");
+  expect(a.at(-1)?.kind).toBe("loading");
+  expect(b.at(-1)?.kind).toBe("loading");
+  await flush();
+  expect(a.at(-1)).toMatchObject({ kind: "ready", file: { path: "/new/a.md", text: "new" } });
+  expect(b.at(-1)).toMatchObject({ kind: "ready", file: { text: "new" } });
+  expect(fetcher.mock.calls.filter(([url]) => String(url).endsWith("-context"))).toHaveLength(2);
+  client.dispose();
+});
diff --git a/src/features/markdown/adapter/text-file.ts b/src/features/markdown/adapter/text-file.ts
new file mode 100644
index 0000000..05ddb23
--- /dev/null
+++ b/src/features/markdown/adapter/text-file.ts
@@ -0,0 +1,171 @@
+import type {
+  TextFileContextResponse,
+  TextFileQuery,
+  TextFileResponse,
+} from "@dispatch/transport-contract";
+import type { TextFileClient, TextFileState } from "../types";
+
+const MAX_BYTES = 262144;
+const MAX_ENTRIES = 32;
+const MAX_PARALLEL = 4;
+interface Entry {
+  readonly listeners: Set<(state: TextFileState) => void>;
+  state: TextFileState;
+  controller: AbortController;
+}
+/** Per-mounted-chat, bounded cache: shared only by live references, never persisted. */
+export function createTextFileClient(
+  apiBase: string,
+  conversationId: string,
+  fetchImpl: typeof fetch,
+): TextFileClient {
+  const entries = new Map<string, Entry>();
+  let disposed = false;
+  let contextController = new AbortController();
+  let context: Promise<TextFileContextResponse> | null = null;
+  let active = 0;
+  const queue = new Map<string, () => void>();
+  const endpoint = `${apiBase.replace(/\/$/, "")}/conversations/${encodeURIComponent(conversationId)}`;
+  async function json(url: string, signal: AbortSignal): Promise<unknown> {
+    const response = await fetchImpl(url, { signal, cache: "no-store" });
+    const data: unknown = await response.json();
+    if (!response.ok) {
+      const error = data as { error?: string };
+      throw new Error(error.error ?? `Text file request failed (HTTP ${response.status})`);
+    }
+    return data;
+  }
+  function getContext(): Promise<TextFileContextResponse> {
+    context ??= json(`${endpoint}/text-file-context`, contextController.signal).then((data) => {
+      const c = data as TextFileContextResponse;
+      if (
+        c.conversationId !== conversationId ||
+        !(c.cwd === null || typeof c.cwd === "string") ||
+        !(c.computerId === null || typeof c.computerId === "string")
+      ) {
+        throw new Error("Invalid text file context response");
+      }
+      return c;
+    });
+    return context;
+  }
+  function publish(entry: Entry, state: TextFileState): void {
+    entry.state = state;
+    for (const listener of entry.listeners) listener(state);
+  }
+  async function load(path: string, entry: Entry, controller: AbortController): Promise<void> {
+    try {
+      if (controller.signal.aborted || disposed) return;
+      const c = await getContext();
+      if (controller.signal.aborted || disposed) return;
+      if (c.computerId !== null)
+        throw new Error(
+          "Text-file embeds are not available for remote computers. No local file was read.",
+        );
+      const query: TextFileQuery = {
+        path,
+        expectedCwd: c.cwd ?? "",
+        expectedComputerId: c.computerId ?? "",
+      };
+      const params = new URLSearchParams(query as unknown as Record<string, string>);
+      const data = (await json(
+        `${endpoint}/text-file?${params}`,
+        controller.signal,
+      )) as TextFileResponse;
+      if (
+        typeof data.text !== "string" ||
+        typeof data.path !== "string" ||
+        !data.path.startsWith("/") ||
+        !["markdown", "text"].includes(data.format) ||
+        data.maxBytes !== MAX_BYTES ||
+        !Number.isInteger(data.bytes) ||
+        data.bytes < 0 ||
+        data.bytes > MAX_BYTES ||
+        new TextEncoder().encode(data.text).length !== data.bytes
+      ) {
+        throw new Error("Invalid or oversized text file response (maximum 256 KiB)");
+      }
+      if (!controller.signal.aborted && !disposed) publish(entry, { kind: "ready", file: data });
+    } catch (error) {
+      if (!controller.signal.aborted && !disposed)
+        publish(entry, {
+          kind: "error",
+          message: error instanceof Error ? error.message : "Unable to read text file",
+        });
+    }
+  }
+  function drain(): void {
+    while (active < MAX_PARALLEL && queue.size) {
+      const next = queue.entries().next().value;
+      if (!next) return;
+      queue.delete(next[0]);
+      next[1]();
+    }
+  }
+  function start(path: string, entry: Entry): void {
+    entry.controller.abort();
+    const controller = new AbortController();
+    entry.controller = controller;
+    publish(entry, { kind: "loading" });
+    queue.set(path, () => {
+      if (controller.signal.aborted || disposed) return;
+      active++;
+      void load(path, entry, controller).finally(() => {
+        active--;
+        drain();
+      });
+    });
+    drain();
+  }
+  return {
+    subscribe(path, listener) {
+      if (disposed || (!entries.has(path) && entries.size >= MAX_ENTRIES)) {
+        listener({
+          kind: "error",
+          message: "Too many text-file embeds (maximum 32 active files).",
+        });
+        return () => {};
+      }
+      let entry = entries.get(path);
+      if (!entry) {
+        entry = {
+          listeners: new Set(),
+          state: { kind: "loading" },
+          controller: new AbortController(),
+        };
+        entries.set(path, entry);
+        start(path, entry);
+      }
+      entry.listeners.add(listener);
+      listener(entry.state);
+      const current = entry;
+      return () => {
+        current.listeners.delete(listener);
+        if (!current.listeners.size) {
+          current.controller.abort();
+          entries.delete(path);
+          queue.delete(path);
+        }
+        if (!entries.size) {
+          contextController.abort();
+          contextController = new AbortController();
+          context = null;
+        }
+      };
+    },
+    refresh(_path) {
+      // Refresh can reveal a context change: invalidate ALL live content before rereading.
+      contextController.abort();
+      contextController = new AbortController();
+      context = null;
+      for (const [path, entry] of entries) start(path, entry);
+    },
+    dispose() {
+      disposed = true;
+      contextController.abort();
+      for (const entry of entries.values()) entry.controller.abort();
+      entries.clear();
+      queue.clear();
+    },
+  };
+}