A realtime front, a durable researcher

{
  "name": "glove-foundry-durable-research",
  "version": "0.0.6",
  "private": true,
  "type": "module",
  "scripts": {
    "start": "node --env-file-if-exists=.env --import tsx scripts/start.ts",
    "typecheck": "tsc --noEmit",
    "verify": "node --import tsx scripts/verify.ts",
    "verify:live": "node --env-file-if-exists=../../.env.local --import tsx scripts/verify-live.ts"
  },
  "dependencies": {
    "effect": "^3.22.1",
    "glove-core": "workspace:*",
    "glove-foundry": "workspace:*",
    "glove-memory": "workspace:*",
    "glove-realtime": "workspace:*",
    "zod": "^4.3.6"
  },
  "devDependencies": {
    "@types/node": "^25.2.3",
    "tsx": "^4.21.0",
    "typescript": "^5.9.3"
  }
}

{
  "compilerOptions": { "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext", "strict": true, "skipLibCheck": true, "noEmit": true, "types": ["node"] },
  "include": ["agents", "lib", "scripts", "foundry.application.ts", "foundry.config.ts"]
}

import { defineApplication, FileFoundryDataAdapter } from "glove-foundry";
import { FileStore } from "./lib/file-store.js";
import { dataPath } from "./lib/paths.js";

export default defineApplication({
  name: "durable-research",
  data: new FileFoundryDataAdapter({ file: dataPath("foundry.json") }),
  conversationStore: ({ conversationId }) => FileStore.open(conversationId),
});

import { defineConfig } from "glove-foundry/config";
export default defineConfig({
  server: { host: "127.0.0.1", port: 4143 },
  execution: { maxConcurrent: 4, pollIntervalMs: 50 },
});

import { mkdirSync } from "node:fs";
import { Effect } from "effect";
import { defineMemory } from "glove-foundry";
import { MemorySchema } from "glove-memory/core";
import { createSqliteMemoryAdapters } from "glove-memory/sqlite";
import { dataDir, dataPath } from "../../lib/paths.js";

export default defineMemory({
  description: "Persistent preferences scoped to the parent instance",
  context: {
    adapter: ({ workspaceId, agentId }) => Effect.sync(() => {
      mkdirSync(dataDir, { recursive: true });
      return createSqliteMemoryAdapters({
        file: dataPath("personal.sqlite"),
        namespace: JSON.stringify([workspaceId, agentId]),
        schema: new MemorySchema(),
      }).context;
    }),
  },
});

import { createHash } from "node:crypto";
import { mkdir, writeFile } from "node:fs/promises";
import { Effect } from "effect";
import { defineApp } from "glove-foundry";
import { z } from "zod";
import { documents } from "../../lib/corpus.js";
import { dataPath } from "../../lib/paths.js";

export default defineApp({
  description: "Read demo source documents and save workspace notes",
  install: ({ workspaceId }) => Effect.succeed({ tools: [
    {
      name: "read_document", description: "Read one vendor brief by document ID.",
      inputSchema: z.object({ id: z.string() }),
      async do({ id }: { id: string }) {
        // Optional latency for exercising a slow external service without calling one.
        const delay = Number(process.env.RESEARCH_DOCUMENT_DELAY_MS ?? 0);
        if (!Number.isFinite(delay) || delay < 0 || delay > 10_000) throw new Error("Document delay must be 0–10000 ms.");
        if (delay) await new Promise((resolve) => setTimeout(resolve, delay));
        const document = documents.find((doc) => doc.id === id);
        if (!document) return { status: "error" as const, data: null, message: "Document not found. Use search_documents to obtain an exact ID." };
        return { status: "success" as const, data: document };
      },
    },
    {
      name: "write_note", description: "Save a note for this workspace. Requires approval.",
      requiresPermission: true,
      inputSchema: z.object({ text: z.string() }),
      async do({ text }: { text: string }) {
        const namespace = createHash("sha256").update(workspaceId).digest("hex");
        const directory = dataPath("notes", namespace);
        await mkdir(directory, { recursive: true });
        await writeFile(`${directory}/note.txt`, text);
        return { status: "success" as const, data: { saved: true } };
      },
    },
  ] }),
});

import { z } from "zod";
import { documents } from "../../lib/corpus.js";

export const searchDocuments = {
  name: "search_documents",
  description: "Search the demo vendor briefs. An empty query lists all documents.",
  inputSchema: z.object({ query: z.string() }),
  async do({ query }: { query: string }) {
    const words = query.toLowerCase().split(/\s+/).filter(Boolean);
    const matches = documents.filter((doc) => !words.length || words.some((word) => `${doc.title} ${doc.text}`.toLowerCase().includes(word)));
    return { status: "success" as const, data: matches.map(({ id, title }) => ({ id, title })) };
  },
};

import { createAdapter } from "glove-core/models/providers";
import { realtimeModel } from "glove-realtime";

export function researchProvider(): "gemini" | "openai" {
  const provider = process.env.RESEARCH_PROVIDER ?? "gemini";
  if (provider !== "gemini" && provider !== "openai") throw new Error("RESEARCH_PROVIDER must be gemini or openai.");
  return provider;
}
export const frontModel = () => realtimeModel({ provider: researchProvider() });
export const textModel = () => createAdapter({
  provider: researchProvider(),
  model: process.env.RESEARCH_TEXT_MODEL ?? (researchProvider() === "openai" ? "gpt-4.1-mini" : "gemini-2.5-flash"),
  apiKey: researchProvider() === "openai" ? process.env.OPENAI_API_KEY : process.env.GEMINI_API_KEY,
  maxTokens: 2048,
});

// A small, explicitly fictional corpus makes the example quick to try.
// Replace this catalogue with your application's authenticated document service.
export const documents = [
  { id: "atlas", title: "Atlas vendor brief", text: "Atlas costs $40/user/month. EU hosting, SSO and audit logs are included. Support SLA: 4 hours." },
  { id: "birch", title: "Birch vendor brief", text: "Birch costs $25/user/month. US hosting only. SSO is an add-on. Support SLA: 1 business day." },
  { id: "cedar", title: "Cedar vendor brief", text: "Cedar costs $55/user/month. EU hosting and SSO are included. Support SLA: 1 hour. Audit-log retention: 2 years." },
] as const;

import { join } from "node:path";
import { fileURLToPath } from "node:url";
export const dataDir = process.env.RESEARCH_DATA_DIR ?? fileURLToPath(new URL("../.data/", import.meta.url));
export const dataPath = (...parts: string[]) => join(dataDir, ...parts);

import { createHash } from "node:crypto";
import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises";
import { dirname, join } from "node:path";
import {
  MemoryStore,
  type InboxItem,
  type Message,
  type Task,
  type TokenConsumptionCounter,
} from "glove-core";

import { dataPath } from "./paths.js";

/** Transcripts live in `.data/transcripts`. */
const TRANSCRIPTS = dataPath("transcripts");

interface SavedConversation {
  readonly messages: Message[];
  readonly turns: number;
  readonly tokens: TokenConsumptionCounter;
  readonly tasks: Task[];
  readonly inbox: InboxItem[];
}

/**
 * Foundry runs every message in a fresh process, so a plain `MemoryStore`
 * would start each turn with no history. This one loads the conversation from
 * a JSON file and rewrites it after every change.
 *
 * One host, one active run per conversation. Replace it with a database-backed
 * StoreAdapter when several hosts share conversations.
 */
export class FileStore extends MemoryStore {
  private restoring = false;
  private writes = Promise.resolve();

  private constructor(conversationId: string, private readonly file: string) {
    super(conversationId);
  }

  static async open(conversationId: string, directory = TRANSCRIPTS): Promise<FileStore> {
    const name = createHash("sha256").update(conversationId).digest("hex").slice(0, 32);
    const store = new FileStore(conversationId, join(directory, `${name}.json`));
    await store.restore();
    return store;
  }

  static async remove(conversationId: string): Promise<void> {
    const name = createHash("sha256").update(conversationId).digest("hex").slice(0, 32);
    await rm(join(TRANSCRIPTS, `${name}.json`), { force: true });
  }

  private async restore(): Promise<void> {
    let saved: SavedConversation;
    try {
      saved = JSON.parse(await readFile(this.file, "utf8")) as SavedConversation;
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
      throw error;
    }
    this.restoring = true;
    try {
      await this.appendMessages(saved.messages);
      await this.addTokens(saved.tokens);
      for (let turn = 0; turn < saved.turns; turn++) await this.incrementTurn();
      await this.addTasks(saved.tasks);
      for (const item of saved.inbox) await this.addInboxItem(item);
    } finally {
      this.restoring = false;
    }
  }

  private async save(): Promise<void> {
    if (this.restoring) return;
    // Realtime transcript and usage callbacks can arrive concurrently.
    this.writes = this.writes.then(async () => {
      const saved: SavedConversation = {
        messages: await this.getMessages(),
        turns: await this.getTurnCount(),
        tokens: await this.getTokenConsumption(),
        tasks: await this.getTasks(),
        inbox: await this.getInboxItems(),
      };
      await mkdir(dirname(this.file), { recursive: true });
      const temporary = `${this.file}.${process.pid}.tmp`;
      await writeFile(temporary, JSON.stringify(saved));
      await rename(temporary, this.file);
    });
    await this.writes;
  }

  override async appendMessages(messages: Message[]) { await super.appendMessages(messages); await this.save(); }
  override async addTokens(usage: TokenConsumptionCounter) { await super.addTokens(usage); await this.save(); }
  override async incrementTurn() { await super.incrementTurn(); await this.save(); }
  override async resetCounters() { await super.resetCounters(); await this.save(); }
  override async addTasks(tasks: Task[]) { await super.addTasks(tasks); await this.save(); }
  override async updateTask(id: string, updates: Partial<Pick<Task, "status" | "content" | "activeForm">>) {
    await super.updateTask(id, updates);
    await this.save();
  }
  override async addInboxItem(item: InboxItem) { await super.addInboxItem(item); await this.save(); }
  override async updateInboxItem(id: string, updates: Partial<Pick<InboxItem, "status" | "response" | "resolved_at">>) {
    await super.updateInboxItem(id, updates);
    await this.save();
  }
}

import { createServer, type ServerResponse } from "node:http";
import { readFile } from "node:fs/promises";
import type { RealtimeSession } from "glove-realtime";

/** Local, single-session text transport. RealtimeSession still owns the model loop. */
export async function openRoom(session: Pick<RealtimeSession, "say">, token: string, end: () => void, port = 8787) {
  const page = await readFile(new URL("./room.html", import.meta.url));
  const streams = new Set<ServerResponse>();
  const publish = (role: string, text: string) => {
    for (const stream of streams) stream.write(`${JSON.stringify({ role, text })}\n`);
  };
  const server = createServer((request, response) => {
    void (async () => {
      if (request.method === "GET" && request.url === "/") {
        response.writeHead(200, { "content-type": "text/html; charset=utf-8" }); response.end(page); return;
      }
      if (request.headers.authorization !== `Bearer ${token}`) {
        response.writeHead(401); response.end(); return;
      }
      if (request.method === "GET" && request.url === "/events") {
        response.writeHead(200, { "content-type": "application/x-ndjson", "cache-control": "no-store" });
        response.flushHeaders(); streams.add(response);
        response.write(`${JSON.stringify({ role: "system", text: "Connected. Ask for vendor research, then keep chatting." })}\n`);
        request.on("close", () => streams.delete(response)); return;
      }
      if (request.method === "POST" && request.url === "/say") {
        let body = "";
        for await (const chunk of request) {
          body += chunk;
          if (body.length > 16_000) { response.writeHead(413); response.end(); return; }
        }
        const { text } = JSON.parse(body) as { text?: unknown };
        if (typeof text !== "string" || !text.trim()) { response.writeHead(400); response.end(); return; }
        await session.say(text);
        response.writeHead(202); response.end(); return;
      }
      if (request.method === "POST" && request.url === "/end") {
        response.writeHead(200); response.end(); end(); return;
      }
      response.writeHead(404); response.end();
    })().catch((error: unknown) => {
      if (!response.headersSent) response.writeHead(500);
      response.end("The request failed.");
      publish("error", error instanceof Error ? error.message : String(error));
    });
  });
  await new Promise<void>((resolve, reject) => {
    server.once("error", reject);
    server.listen(port, "127.0.0.1", resolve);
  });
  return {
    port: (server.address() as import("node:net").AddressInfo).port,
    publish,
    async close() {
      for (const stream of streams) stream.end();
      streams.clear();
      await new Promise<void>((resolve, reject) => {
        server.close((error) => error ? reject(error) : resolve()); server.closeAllConnections();
      });
    },
  };
}

<!doctype html>
<html lang="en">
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Realtime research front</title>
<style>
body { font: 17px/1.6 system-ui; max-width: 760px; margin: 3rem auto; padding: 0 1rem; }
#messages { white-space: pre-wrap; min-height: 240px; } form { display: flex; gap: 8px; }
input { flex: 1; min-width: 0; } input, button { font: inherit; padding: .6rem; }
</style>
<h1>Realtime research front</h1>
<p>Try: “Compare Atlas, Birch and Cedar. I need EU hosting. Delegate the research and keep talking to me.”</p>
<div id="messages" role="log" aria-live="polite"></div>
<form><input aria-label="Message" required autocomplete="off"><button>Send</button></form>
<button id="end" type="button">End live session</button>
<script type="module">
const token = location.hash.slice(1);
history.replaceState(null, "", location.pathname);
const headers = { authorization: `Bearer ${token}`, "content-type": "application/json" };
const log = (role, text) => {
  const line = document.createElement("p"); line.textContent = `${role}: ${text}`;
  document.querySelector("#messages").append(line);
};
const form = document.querySelector("form"), input = form.querySelector("input");
form.onsubmit = async (event) => {
  event.preventDefault(); const text = input.value; input.value = "";
  try {
    const response = await fetch("/say", { method: "POST", headers, body: JSON.stringify({ text }) });
    if (!response.ok) throw new Error(`Send failed (${response.status})`);
    log("you", text);
  } catch (error) { log("error", error.message); }
};
document.querySelector("#end").onclick = () => fetch("/end", { method: "POST", headers }).catch((error) => log("error", error.message));
try {
  const response = await fetch("/events", { headers });
  if (!response.ok) throw new Error("Open the URL printed by the start script, including its session token.");
  const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = "";
  for (;;) {
    const { value, done } = await reader.read(); if (done) break;
    buffer += value;
    let newline;
    while ((newline = buffer.indexOf("\n")) >= 0) {
      const event = JSON.parse(buffer.slice(0, newline)); buffer = buffer.slice(newline + 1);
      log(event.role, event.text);
    }
  }
  log("system", "Live session ended. The Foundry controller and researcher can keep running.");
} catch (error) { log("error", error.message); }
</script>
</html>

import assert from "node:assert/strict";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { Effect } from "effect";
import { FileFoundryDataAdapter, FoundryRuntime } from "glove-foundry";
import { openRoom } from "../lib/room.js";

// No model is called, no API key is read, and no user data is changed.
const temporary = await mkdtemp(join(tmpdir(), "durable-research-example-"));
process.env.RESEARCH_DATA_DIR = temporary;
const { default: application } = await import("../foundry.application.js");
const { FileStore } = await import("../lib/file-store.js");
const runtime = await FoundryRuntime.discover({
  rootDir: fileURLToPath(new URL("../", import.meta.url)),
  agentsDir: fileURLToPath(new URL("../agents/", import.meta.url)), application,
});
try {
  const front = runtime.agents.find((agent) => agent.route === "front")!.definition;
  const researcher = runtime.agents.find((agent) => agent.route === "front/workers/research")!.definition;
  assert.equal(researcher.runTimeoutMs, 4 * 60 * 60 * 1000);
  assert.deepEqual(researcher.eventDelivery, { whenIdle: "defer", deferTimeoutMs: 30_000 });
  assert.deepEqual(researcher.components!.capabilities.applications[0]!.restrictions?.tools, ["read_document"]);
  assert.deepEqual(front.durableSubagents!.research!.inherit!.memory, { scope: "shared", access: "reader", profiles: ["personal"] });
  const parent = await runtime.createAgent("front", { id: "offline-parent", context: { preference: "EU" }, installations: [{ kind: "application", id: "library" }] });
  const worker = await runtime.subagent(parent.id, "research");
  assert.equal(worker.context.preference, "EU");
  assert.equal((await runtime.subagent(parent.id, "research")).id, worker.id);
  const receipt = await runtime.messageSubagent(parent.id, "research", "deferred test", { id: "verify-event", whenIdle: "defer" });
  assert.equal(receipt.runId, undefined);
  assert.equal((await Effect.runPromise(runtime.data.listEvents!()))[0]!.status, "pending");
  const store = await FileStore.open("verify-transcript");
  await Promise.all([
    store.appendMessages([{ sender: "user", text: "remember this" }]),
    store.addTokens({ tokens_in: 12, tokens_out: 3 }),
    store.incrementTurn(),
  ]);
  const reopened = await FileStore.open("verify-transcript");
  assert.equal((await reopened.getMessages())[0]!.text, "remember this");
  assert.equal((await reopened.getTokenConsumption()).tokens_in, 12);
  assert.equal(await reopened.getTurnCount(), 1);
  await FileStore.remove("verify-transcript");
  assert.equal((await (await FileStore.open("verify-transcript")).getMessages()).length, 0);

  // Reconstruct the runtime with a fresh adapter, without starting model execution.
  await runtime.stop();
  const recovered = await FoundryRuntime.discover({
    rootDir: fileURLToPath(new URL("../", import.meta.url)),
    agentsDir: fileURLToPath(new URL("../agents/", import.meta.url)),
    application: { ...application, data: new FileFoundryDataAdapter({ file: join(temporary, "foundry.json") }) },
  });
  try {
    assert.equal((await recovered.subagent(parent.id, "research")).id, worker.id);
    const events = await Effect.runPromise(recovered.data.listEvents!());
    assert.equal(events[0]!.event.id, "verify-event");
    assert.equal(events[0]!.status, "pending");
    assert.equal(events[0]!.target.conversationId, receipt.conversationId);
  } finally { await recovered.stop(); }

  const said: string[] = []; let ended = false;
  const room = await openRoom({ say: async (text) => { said.push(text); } }, "offline-test-token", () => { ended = true; }, 0);
  try {
    const base = `http://127.0.0.1:${room.port}`;
    assert.match(await (await fetch(base)).text(), /Realtime research front/);
    assert.equal((await fetch(`${base}/say`, { method: "POST" })).status, 401);
    const headers = { authorization: "Bearer offline-test-token", "content-type": "application/json" };
    const stream = await fetch(`${base}/events`, { headers });
    const reader = stream.body!.getReader();
    assert.match(new TextDecoder().decode((await reader.read()).value), /Connected/);
    room.publish("assistant", "A finding arrived.");
    assert.match(new TextDecoder().decode((await reader.read()).value), /A finding arrived/);
    await reader.cancel();
    assert.equal((await fetch(`${base}/say`, { method: "POST", headers, body: JSON.stringify({ text: "Research Atlas" }) })).status, 202);
    assert.deepEqual(said, ["Research Atlas"]);
    await fetch(`${base}/end`, { method: "POST", headers }); assert.equal(ended, true);
  } finally { await room.close(); }
  console.log("Durable research example verified offline: discovery, inheritance, event persistence, transcripts and room transport.");
} finally { await runtime.stop(); await rm(temporary, { recursive: true, force: true }); }

import assert from "node:assert/strict";
import { fork, type ChildProcess } from "node:child_process";
import { randomUUID } from "node:crypto";
import { mkdtemp, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { Effect } from "effect";
import { FileFoundryDataAdapter, type FoundryDelegation } from "glove-foundry";
import { createFoundryClient } from "glove-foundry/client";

// Opt-in live test: only models are remote; all application data is fictional.
assert.ok(process.env.OPENAI_API_KEY, "OPENAI_API_KEY is required for verify:live.");
const root = await mkdtemp(join(tmpdir(), "durable-research-live-"));
const token = randomUUID();
const data = new FileFoundryDataAdapter({ file: join(root, "foundry.json") });
const env = {
  ...process.env, RESEARCH_PROVIDER: "openai", RESEARCH_DATA_DIR: root,
  RESEARCH_LIVE_TEST_ROOT: root, RESEARCH_ROOM_TOKEN: token, RESEARCH_ROOM_PORT: "0",
  RESEARCH_DOCUMENT_DELAY_MS: "3000", REALTIME_MODEL: process.env.REALTIME_MODEL ?? "gpt-realtime",
  RESEARCH_TEXT_MODEL: process.env.RESEARCH_TEXT_MODEL ?? "gpt-4.1-mini",
};
const redact = (text: string) => text.replaceAll(process.env.OPENAI_API_KEY!, "[REDACTED]");
const checks: { name: string; at: string; detail?: unknown }[] = [];
const roomMessages: { role: string; text: string; receivedAt: number }[] = [];
const controllerPids: number[] = [];
let logs = "";
let child: ChildProcess | undefined;
let client!: ReturnType<typeof createFoundryClient>;
let streamAbort: AbortController | undefined;
let readRoom: Promise<void> | undefined;
const journal = () => Effect.runPromise(data.listEvents());
const pass = (name: string, detail?: unknown) => {
  checks.push({ name, at: new Date().toISOString(), detail });
  console.log(`PASS ${name}${detail ? ` ${JSON.stringify(detail)}` : ""}`);
};
async function eventually<T>(label: string, read: () => Promise<T | undefined>, timeoutMs = 120_000): Promise<T> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (child?.exitCode !== null && child?.exitCode !== undefined) throw new Error(`Controller exited during ${label}.`);
    const value = await read();
    if (value !== undefined) return value;
    await new Promise((resolve) => setTimeout(resolve, 150));
  }
  throw new Error(`Timed out: ${label}`);
}
async function startController() {
  child = fork(fileURLToPath(new URL("./live-controller.ts", import.meta.url)), [], {
    env, execArgv: ["--import", "tsx"], stdio: ["ignore", "pipe", "pipe", "ipc"],
  });
  child.stdout!.on("data", (chunk) => { logs += redact(String(chunk)); });
  child.stderr!.on("data", (chunk) => { logs += redact(String(chunk)); });
  const ready = await new Promise<{ url: string; pid: number }>((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error("Controller startup timed out")), 30_000);
    child!.once("message", (message) => { clearTimeout(timer); resolve(message as { url: string; pid: number }); });
    child!.once("error", reject);
    child!.once("exit", (code) => { clearTimeout(timer); reject(new Error(`Controller exited: ${code}`)); });
  });
  controllerPids.push(ready.pid);
  client = createFoundryClient({ baseUrl: ready.url });
}
async function stopController() {
  const stopping = child;
  if (!stopping || stopping.exitCode !== null || stopping.signalCode !== null) return;
  await new Promise<void>((resolve, reject) => {
    const timer = setTimeout(() => { stopping.kill("SIGKILL"); reject(new Error("Controller shutdown timed out")); }, 30_000);
    stopping.once("exit", () => { clearTimeout(timer); resolve(); });
    stopping.kill("SIGTERM");
  });
  child = undefined;
}
async function assistantSays(marker: string, after = 0) {
  return eventually(`live front says ${marker}`, async () => roomMessages.find((message) => message.role === "assistant" && message.text.includes(marker) && message.receivedAt >= after));
}
async function acknowledged(id: string) {
  return eventually(`${id} acknowledged`, async () => {
    const entry = (await journal()).find((entry) => entry.event.id === id);
    if (entry?.status === "acknowledged") return entry;
    if (entry?.runId) {
      // Until recovery reassigns it, a pending receipt can reference a run
      // from the previous controller's execution store.
      const run = (await client.runs()).find((candidate) => candidate.id === entry.runId);
      if (run) assert.notEqual(run.status, "failed", redact(`Delivery failed: ${JSON.stringify(run.error ?? run.output)}`));
    }
    return undefined;
  });
}
async function completedReply(marker: string) {
  return eventually(`parent handles ${marker}`, async () => {
    const entries = await journal();
    const replies = entries.filter((entry) => entry.target.agentId === "test-front" && entry.target.conversationId === "test-live"
      && JSON.stringify(entry.event.content).includes(marker));
    // At-least-once recovery may repeat the worker's outbound side effect.
    // Wait for every accepted copy, not just an earlier acknowledged reply.
    return replies.length && replies.every((entry) => entry.status === "acknowledged") ? replies.at(-1) : undefined;
  });
}
async function tasks() {
  return (await client.subagent("test-front", "research").data()).conversations
    .map((conversation) => conversation.context.delegation as FoundryDelegation | undefined).filter((task) => task !== undefined);
}
let failure: string | undefined;
try {
  await startController();
  await client.createAgent("front", { id: "test-front", workspaceId: "live-test", context: { preference: "EU hosting" }, installations: [{ kind: "application", id: "library" }] });
  await client.createConversation("test-front", { id: "test-live" });
  const worker = await client.subagent("test-front", "research").ensure();
  assert.equal(worker.context.preference, "EU hosting");
  const live = await client.send("test-front", "test-live", "Start the test session.", { payload: { mode: "live" } });
  const ready = await eventually("realtime room", async () => {
    const run = await client.getRun(live.id);
    assert.ok(!["failed", "cancelled", "completed"].includes(run.status), `Live run ended: ${JSON.stringify(run.error ?? run.output)}`);
    return (await client.getEvents({ runId: live.id })).find((event) => event.type === "agent.research.front.ready");
  });
  const roomUrl = `http://127.0.0.1:${(ready.data as { port: number }).port}`;
  const headers = { authorization: `Bearer ${token}`, "content-type": "application/json" };
  const say = async (text: string) => {
    const response = await fetch(`${roomUrl}/say`, { method: "POST", headers, body: JSON.stringify({ text }) });
    assert.equal(response.status, 202, "room accepts user input");
  };
  streamAbort = new AbortController();
  const stream = await fetch(`${roomUrl}/events`, { headers, signal: streamAbort.signal });
  readRoom = (async () => {
    let pending = ""; const decoder = new TextDecoder();
    for await (const chunk of stream.body!) {
      pending += decoder.decode(chunk, { stream: true });
      let end: number;
      while ((end = pending.indexOf("\n")) >= 0) {
        const message = { ...JSON.parse(pending.slice(0, end)), receivedAt: Date.now() } as { role: string; text: string; receivedAt: number };
        roomMessages.push(message); pending = pending.slice(end + 1);
        if (message.role === "assistant" || message.role === "error") console.log(redact(`${message.role}: ${message.text}`).slice(0, 600));
      }
    }
  })().catch((error) => { if (!streamAbort?.signal.aborted) throw error; });
  await say('Reply with READY_LIVE. Keep all replies brief throughout this integration test.');
  await assistantSays("READY_LIVE");
  pass("native realtime front responds");

  await say('Call glove_foundry_message with exactly this JSON: {"to":"research","message":"Protocol test: send parent a direct message containing PONG_DIRECT, targeting parent conversation test-live with whenIdle=wake. Do not research or delegate.","delivery":{"id":"ping-direct","whenIdle":"wake"}}. Omit conversationId in YOUR outgoing delivery: test-live is only the WORKER reply target. When the worker reply arrives, tell me PONG_DIRECT.');
  await acknowledged("ping-direct");
  const pong = (await journal()).find((entry) => entry.target.conversationId === "test-live" && JSON.stringify(entry.event.content).includes("PONG_DIRECT"));
  assert.ok(pong, "worker actually sent the reply, not just a front-model claim");
  await assistantSays("PONG_DIRECT", Date.parse(pong.createdAt));
  assert.equal(pong.runId, live.id);
  assert.equal((await client.runs("front")).filter((run) => run.conversationId === "test-live").length, 1);
  pass("front → worker → same live front via durable direct messages");

  await say('Delegate exactly one task to research: "Before reading any source, call glove_foundry_ask to ask What budget should I use? Wait for the answer. Then search for the Atlas brief, read its exact document ID and state whether it fits EU hosting and that budget. Include TASK_DONE in your final answer." The budget is not known yet: wait for the worker question and ask me. Do not send a reply or invent a budget before I answer. Tell me TASK_DONE only after its result arrives.');
  const task = await eventually("task accepted", async () => (await tasks()).find((item) => item.task.includes("TASK_DONE")));
  assert.equal(task.workerAgentId, worker.id);
  await say('While research is pending, reply immediately with STILL_HERE; leave the task running.');
  await assistantSays("STILL_HERE");
  assert.notEqual((await tasks()).find((item) => item.id === task.id)?.status, "completed", "front responds before background work finishes");
  pass("front remains responsive during pending research");
  const question = await eventually("worker question reaches the live front", async () => (await journal()).find((entry) => entry.delivery.kind === "question" && entry.delivery.delegationId === task.id && entry.runId === live.id));
  await assistantSays("budget", Date.parse(question.createdAt));
  await say('My budget is $50/user/month. Answer the research worker with glove_foundry_reply and let the original task finish.');
  const finished = await eventually("delegated task completion", async () => (await tasks()).find((item) => item.id === task.id && item.status === "completed"));
  assert.match(finished.result ?? "", /TASK_DONE/);
  const resultEvent = await eventually("task result assigned to a receiving run", async () => (await journal()).find((entry) => entry.delivery.kind === "result" && entry.delivery.delegationId === task.id && entry.runId));
  const taskEvents = (await journal()).filter((entry) => "delegationId" in entry.delivery && entry.delivery.delegationId === task.id);
  for (const kind of ["question", "reply", "result"]) assert.ok(taskEvents.some((entry) => entry.delivery.kind === kind), `journal contains ${kind}`);
  assert.equal(resultEvent.runId, live.id);
  await assistantSays("TASK_DONE", Date.parse(resultEvent.createdAt));
  const workerTranscript = await client.conversationTranscript(worker.id, task.id);
  assert.ok(JSON.stringify(workerTranscript.messages).includes("read_document"), "worker used the inherited mock app");
  assert.ok(!JSON.stringify(workerTranscript.messages).includes('"name":"write_note"'), "worker did not invoke the write tool");
  pass("delegation asks, receives an answer and returns a result", { taskId: task.id });

  await fetch(`${roomUrl}/end`, { method: "POST", headers });
  assert.equal((await live.wait({ timeoutMs: 30_000 })).status, "completed");
  streamAbort.abort(); await readRoom;
  await eventually("live run acknowledges consumed events", async () => (await journal()).find((entry) => entry.id === pong.id && entry.status === "acknowledged"));
  pass("live-session consumption acknowledged on successful close");

  const restartChat = await client.createConversation(worker.id, { id: "restart-target" });
  const target = { agentId: worker.id, conversationId: restartChat.id };
  const deferred = { id: "restart-deferred", source: "test:app", content: 'Read document id atlas, then call glove_foundry_message with to="parent", delivery={"conversationId":"test-live","whenIdle":"wake"}, and message containing RESTART_DELIVERED plus the price. The parent conversation ID is test-live; it must be in delivery.conversationId. Do not delegate.' };
  const policy = { whenIdle: "defer" as const, deferTimeoutMs: 5000 };
  await client.emitEvent(target, deferred, policy);
  await client.emitEvent(target, { id: "restart-expired", source: "test:app", content: "NEVER_DELIVER_THIS", expiresAt: "2000-01-01T00:00:00Z" }, policy);
  assert.equal((await journal()).find((entry) => entry.event.id === deferred.id)?.runId, undefined);
  await stopController();
  await startController();
  assert.equal((await client.subagent("test-front", "research").ensure()).id, worker.id);
  const recovered = await acknowledged(deferred.id);
  assert.deepEqual(recovered.target, target);
  await completedReply("RESTART_DELIVERED");
  assert.equal((await journal()).find((entry) => entry.event.id === "restart-expired")?.status, "expired");
  const afterRestart = await client.conversationTranscript(worker.id, restartChat.id);
  assert.ok(!JSON.stringify(afterRestart.messages).includes("NEVER_DELIVER_THIS"));
  const parentHistory = await client.conversationTranscript("test-front", "test-live");
  const receivedAt = parentHistory.messages.findIndex((message) => message.framework_context === "event" && message.text?.includes("RESTART_DELIVERED"));
  assert.ok(receivedAt >= 0, "the exact parent conversation persisted the event");
  assert.ok(parentHistory.messages.slice(receivedAt + 1).some((message) => message.sender === "agent" && message.text), "idle parent produced a text response after consuming the event");
  pass("deferred event survives controller restart; exact target, expiry and idle parent fallback verified");

  // Stop a real worker while its simulated external read is in progress.
  const interruptChat = await client.createConversation(worker.id, { id: "interrupted-target" });
  const interrupted = await client.subagent("test-front", "research").send(
    'Read document id cedar now, then call glove_foundry_message with to="parent", delivery={"conversationId":"test-live","whenIdle":"wake"}, and message containing RECOVERED_INFLIGHT plus the price. The parent conversation ID is test-live; it must be in delivery.conversationId. Do not delegate.',
    { id: "interrupted-event", conversationId: interruptChat.id, whenIdle: "wake" },
  );
  await eventually("mock document read starts", async () => (await client.getEvents({ runId: interrupted.runId })).find((event) => event.type.includes("tool") && JSON.stringify(event.data).includes("read_document")));
  await stopController();
  assert.equal((await journal()).find((entry) => entry.event.id === "interrupted-event")?.status, "pending");
  await startController();
  const replayed = await acknowledged("interrupted-event");
  assert.notEqual(replayed.runId, interrupted.runId);
  await completedReply("RECOVERED_INFLIGHT");
  pass("in-flight event replays in a new worker process after restart");

  const before = (await client.runs()).length;
  await client.emitEvent(target, deferred, policy);
  await new Promise((resolve) => setTimeout(resolve, 700));
  assert.equal((await client.runs()).length, before, "acknowledged retry must not create a new run");
  await assert.rejects(client.emitEvent({ agentId: worker.id, conversationId: "test-live" }, { id: "wrong-target", source: "test", content: "reject" }), /owned/);
  pass("deduplication and conversation ownership enforced after restart");
  assert.deepEqual(roomMessages.filter((message) => message.role === "error"), [], "no realtime provider errors");
  assert.equal((await journal()).filter((entry) => entry.status === "pending").length, 0, "all accepted test events are settled before teardown");
} catch (error) {
  failure = redact(error instanceof Error ? error.stack ?? error.message : String(error));
  console.error(failure);
  process.exitCode = 1;
} finally {
  streamAbort?.abort(); await readRoom?.catch(() => {});
  await stopController();
  await writeFile(join(root, "report.json"), redact(JSON.stringify({ checks, failure, controllerPids, models: { realtime: env.REALTIME_MODEL, text: env.RESEARCH_TEXT_MODEL }, roomMessages, events: await journal() }, null, 2)));
  await writeFile(join(root, "controller.log"), logs);
  console.log(`Test artifacts: ${root}`);
}