FE-1524: Prove idempotent capture sweep on the live Flue chat path

apps/brunch-agent/README.md

@@ -11,11 +11,18 @@ yarn dev:brunch
 The first step builds the Petrinaut libraries the panel imports (`dist/` and design-system
 codegen). Then it starts the Brunch server at `http://127.0.0.1:4321` and the real Petrinaut
 website at `http://127.0.0.1:4915`. The website proxies `/api/chat` to Brunch. The panel talks to one plain
-Flue chat agent: streamed text and reasoning, one server `ping` tool, and the existing Petrinaut
-`readPetrinautDoc` client tool. There is no elicitation, capture, or `brunch_ask` on this path.
+Flue chat agent: streamed text and reasoning, one server `ping` tool, one stub
+skill (`confirm-path`, activated via `activate_skill`), and the existing Petrinaut
+`readPetrinautDoc` client tool. There is no elicitation loop, sweep tool, or
+`brunch_ask` on this path. Capture is a harness-side pipe: an explicit settled
+range of Flue history is applied into a JSON store beside the conversation
+database, not by the interviewer.
 
 Conversations persist in `apps/brunch-agent/.data-wipe-me/conversations.db`. `BRUNCH_DEV_DB_PATH`
-overrides that local path. Flue history is the conversation log; the browser may cache messages
+overrides that local path. Capture envelopes for one Flue conversation sit beside that sqlite
+file, named by the hashed instance id (`<instanceId>.json`). The hermetic `/api/chat` test uses
+`BRUNCH_CHAT_DB_PATH` and writes the capture file in that same directory. Flue history is the
+conversation log; the capture store is not a second transcript. The browser may cache messages
 but reload hydrates from `GET /api/chat?id=`.
 
 The mounted Flue URL `/agents/chat/:id` requires the same principal and conversation identity (`x-brunch-principal` and `x-brunch-conversation`) as `/api/chat`; the path id is the hash of those, not a bearer token.

apps/brunch-agent/package.json

@@ -20,6 +20,7 @@
     "@flue/react": "2.0.3",
     "@flue/runtime": "2.0.3",
     "@flue/sdk": "2.0.3",
+    "@hashintel/brunch-agent-binding-flue": "workspace:*",
     "@hashintel/brunch-agent-transport-aisdk": "workspace:*",
     "@hashintel/petrinaut-core": "workspace:*",
     "@opentelemetry/api": "1.9.1",

apps/brunch-agent/src/agents/chat-agent.ts

@@ -2,25 +2,39 @@
 /**
  * One plain Flue chat agent for the Petrinaut panel throughline.
  *
- * No elicitation, capture, or plugin. The model can call a server-side ping
- * and a browser-executed Petrinaut doc reader; Flue history is the session log.
+ * Capture is a harness-side pipe, not an interviewer tool. One stub skill is
+ * mounted so activation can appear in Flue history.
  */
 
-import { useModel, useTool } from "@flue/runtime";
+import { defineSkill, useModel, useSkill, useTool } from "@flue/runtime";
 
 import { ping } from "../tools/ping.ts";
 import { readPetrinautDoc } from "../tools/read-petrinaut-doc.ts";
 
 export const CHAT_MODEL_ID =
   process.env["BRUNCH_CHAT_MODEL"] || "claude-haiku-4-5";
 
+export const STUB_SKILL_NAME = "confirm-path";
+
+export const ACTIVATE_SKILL_TOOL_NAME = "activate_skill";
+
+const confirmPath = defineSkill({
+  name: STUB_SKILL_NAME,
+  description:
+    "Confirm how this assistant is mounted. Use when checking the server path or tool layout.",
+  instructions:
+    "Say that ping confirms the server tool path. Then continue helping the user.",
+});
+
 export function ChatAgent() {
   useModel(`anthropic/${CHAT_MODEL_ID}`);
+  useSkill(confirmPath);
   useTool(ping);
   useTool(readPetrinautDoc);
   return [
     "You are a concise assistant inside the Petrinaut editor.",
     "Call ping when you need to confirm the server tool path.",
+    `Activate the \`${STUB_SKILL_NAME}\` skill before calling ping.`,
     "When the user asks how Petrinaut's UI works, call readPetrinautDoc.",
     "A client-tool-result signal is JSON [{ toolCallId, toolName, output }]. Treat output as the browser's result for that call and continue helping the user.",
   ].join("\n");

apps/brunch-agent/src/capture-sweep.ts

@@ -0,0 +1,102 @@
+/**
+ * Harness-side apply-sweep over a named Flue history range.
+ *
+ * The interviewer does not call this. A test or harness fact names the range.
+ * Stub extraction: one envelope per user utterance, quote = that text, payload {}.
+ */
+
+import {
+  createFlueHistoryReader,
+  createLocalCaptureStore,
+  projectFlueHistoryForSweep,
+} from "@hashintel/brunch-agent-binding-flue";
+
+import {
+  agentOwnershipHeaders,
+  flueConversationIdFrom,
+  type ConversationIdentity,
+} from "./conversation-identity.ts";
+import { captureStorePath } from "./db-path.ts";
+import { CHAT_AGENT_ROUTE } from "./routes.ts";
+
+export interface CaptureSweepCapture {
+  readonly id: string;
+  readonly excerpt: string;
+  readonly payload: unknown;
+}
+
+export interface CaptureSweepResult {
+  readonly appliedCaptureIds: readonly string[];
+  readonly skippedDedupKeys: readonly string[];
+  readonly captures: readonly CaptureSweepCapture[];
+}
+
+const conversationUrl = (instanceId: string): string =>
+  `http://brunch.local/agents/${CHAT_AGENT_ROUTE}/${instanceId}`;
+
+const ownedTransport = (identity: ConversationIdentity): typeof fetch => {
+  const ownership = agentOwnershipHeaders(identity);
+  return async (input, init) => {
+    const { default: app } = await import("./app.ts");
+    const headers = new Headers(init?.headers);
+    for (const [key, value] of Object.entries(ownership)) {
+      headers.set(key, value);
+    }
+    return app.fetch(
+      input instanceof Request
+        ? new Request(input, { headers })
+        : new Request(input, { ...init, headers }),
+    );
+  };
+};
+
+export const applyCaptureSweep = async (
+  identity: ConversationIdentity,
+  userEntryIds: readonly string[],
+): Promise<CaptureSweepResult> => {
+  const instanceId = flueConversationIdFrom(identity);
+  const store = createLocalCaptureStore(captureStorePath(instanceId), {
+    ownerKey: identity.principalKey,
+  });
+  const historyReader = createFlueHistoryReader({
+    resolveConversationUrl: conversationUrl,
+    transport: ownedTransport(identity),
+    archive: store,
+  });
+  const snapshot = await historyReader.read(instanceId);
+  const range = new Set(userEntryIds);
+  const proposals = projectFlueHistoryForSweep(snapshot)
+    .filter(
+      (entry) =>
+        entry.kind === "user" && range.has(entry.id) && entry.text.length > 0,
+    )
+    .map((entry) => ({
+      evidence: [{ excerpt: entry.text }],
+      epistemicStatus: "explicit" as const,
+      confidence: "high",
+      content: { value: {} },
+    }));
+  const applied = await store.execute(
+    { type: "apply-sweep", proposals },
+    { sessionId: instanceId },
+  );
+  if (!applied.ok) {
+    throw new Error(
+      `apply-sweep refused: ${applied.refusal.code}: ${applied.refusal.message}`,
+    );
+  }
+  if (!("appliedCaptureIds" in applied.value)) {
+    throw new Error("apply-sweep did not return a sweep value.");
+  }
+  return {
+    appliedCaptureIds: applied.value.appliedCaptureIds,
+    skippedDedupKeys: applied.value.skippedDedupKeys,
+    captures: applied.snapshot.captures.map((capture) => ({
+      id: capture.id,
+      excerpt:
+        "evidence" in capture ? (capture.evidence[0]?.excerpt ?? "") : "",
+      payload:
+        "value" in capture.content ? capture.content.value : capture.content,
+    })),
+  };
+};

apps/brunch-agent/src/db-path.ts

@@ -12,8 +12,12 @@
  * Flue Node runtime and SQLite adapter.
  */
 
+import { dirname, join } from "node:path";
 import { fileURLToPath } from "node:url";
 
+const conversationDbFileFrom = (override: string): string =>
+  override.endsWith(".db") ? override : join(override, "conversations.db");
+
 export function conversationDbPath(): string {
   // Truthiness, not nullish, on purpose: a set-but-empty override would pass
   // '' through to sqlite(), which opens an anonymous temporary database
@@ -25,3 +29,21 @@ export function conversationDbPath(): string {
         new URL("../.data-wipe-me/conversations.db", import.meta.url),
       );
 }
+
+/**
+ * Capture JSON lives beside the Flue sqlite file, named by Flue instance id.
+ * The hermetic chat test sets `BRUNCH_CHAT_DB_PATH` (not `BRUNCH_DEV_DB_PATH`),
+ * so that directory wins when present.
+ */
+export function captureStorePath(instanceId: string): string {
+  if (instanceId.length === 0) {
+    throw new TypeError(
+      "A Flue instance id is required for the capture store path.",
+    );
+  }
+  const chatDb = process.env.BRUNCH_CHAT_DB_PATH;
+  const directory = dirname(
+    chatDb ? conversationDbFileFrom(chatDb) : conversationDbPath(),
+  );
+  return join(directory, `${instanceId}.json`);
+}

apps/brunch-agent/test/db-path.test.ts

@@ -16,18 +16,21 @@ import { fileURLToPath } from "node:url";
 
 import { afterEach, describe, expect, test } from "vitest";
 
-import { conversationDbPath } from "../src/db-path";
+import { conversationDbPath, captureStorePath } from "../src/db-path";
 
 const appDir = fileURLToPath(new URL("..", import.meta.url));
 
 describe("the conversation store path", () => {
   const originalCwd = process.cwd();
   const originalOverride = process.env.BRUNCH_DEV_DB_PATH;
+  const originalChatDb = process.env.BRUNCH_CHAT_DB_PATH;
 
   afterEach(() => {
     process.chdir(originalCwd);
     if (originalOverride === undefined) delete process.env.BRUNCH_DEV_DB_PATH;
     else process.env.BRUNCH_DEV_DB_PATH = originalOverride;
+    if (originalChatDb === undefined) delete process.env.BRUNCH_CHAT_DB_PATH;
+    else process.env.BRUNCH_CHAT_DB_PATH = originalChatDb;
   });
 
   test("is anchored to the package, wherever the process was launched from", () => {
@@ -55,3 +58,30 @@ describe("the conversation store path", () => {
     );
   });
 });
+
+describe("the capture store path", () => {
+  const originalChatDb = process.env.BRUNCH_CHAT_DB_PATH;
+  const originalOverride = process.env.BRUNCH_DEV_DB_PATH;
+
+  afterEach(() => {
+    if (originalChatDb === undefined) delete process.env.BRUNCH_CHAT_DB_PATH;
+    else process.env.BRUNCH_CHAT_DB_PATH = originalChatDb;
+    if (originalOverride === undefined) delete process.env.BRUNCH_DEV_DB_PATH;
+    else process.env.BRUNCH_DEV_DB_PATH = originalOverride;
+  });
+
+  test("sits beside the conversation database, named by Flue instance id", () => {
+    delete process.env.BRUNCH_CHAT_DB_PATH;
+    delete process.env.BRUNCH_DEV_DB_PATH;
+    expect(captureStorePath("flue-instance-1")).toBe(
+      join(appDir, ".data-wipe-me", "flue-instance-1.json"),
+    );
+  });
+
+  test("follows the hermetic chat database directory", () => {
+    process.env.BRUNCH_CHAT_DB_PATH = join(tmpdir(), "conversations.db");
+    expect(captureStorePath("flue-instance-1")).toBe(
+      join(tmpdir(), "flue-instance-1.json"),
+    );
+  });
+});

apps/brunch-agent/test/petrinaut-chat-result.ts

@@ -28,6 +28,17 @@ export interface PetrinautChatResult {
   readonly transcript: string;
   readonly instanceId: string;
   readonly dbPath: string;
+  readonly activateSkillCall: Extract<
+    UIMessageChunk,
+    { type: "tool-input-available" }
+  > | null;
+  readonly interviewerToolNames: readonly string[];
+  readonly captureUserText: string;
+  readonly captureIds: readonly string[];
+  readonly recaptureIds: readonly string[];
+  readonly skippedDedupKeys: readonly string[];
+  readonly capturePayloads: readonly unknown[];
+  readonly captureExcerpts: readonly string[];
 }
 
 export interface PetrinautResumeResult {

apps/brunch-agent/test/petrinaut-chat.integration.ts

@@ -13,7 +13,13 @@ import {
 import { sqlite, start } from "@flue/runtime/node";
 import { createFlueClient, FlueApiError } from "@flue/sdk";
 
-import { CHAT_MODEL_ID, ChatAgent } from "../src/agents/chat-agent.ts";
+import {
+  ACTIVATE_SKILL_TOOL_NAME,
+  CHAT_MODEL_ID,
+  ChatAgent,
+  STUB_SKILL_NAME,
+} from "../src/agents/chat-agent.ts";
+import { applyCaptureSweep } from "../src/capture-sweep.ts";
 import {
   agentOwnershipHeaders,
   flueConversationIdFrom,
@@ -108,6 +114,17 @@ try {
     process.stdout.write(`PETRINAUT_RESUME_RESULT ${JSON.stringify(result)}\n`);
   } else {
     faux.setResponses([
+      fauxAssistantMessage(
+        [
+          fauxThinking("Load the mount confirmation skill."),
+          fauxToolCall(
+            ACTIVATE_SKILL_TOOL_NAME,
+            { name: STUB_SKILL_NAME },
+            { id: "tool-skill-1" },
+          ),
+        ],
+        { stopReason: "toolUse" },
+      ),
       fauxAssistantMessage(
         [
           fauxThinking("Confirm the server path, then read the guide."),
@@ -177,6 +194,14 @@ try {
           chunk.type === "tool-input-available" &&
           chunk.toolName === PING_TOOL_NAME,
       ) ?? null;
+    const activateSkillCall =
+      initialChunks.find(
+        (
+          chunk,
+        ): chunk is Extract<UIMessageChunk, { type: "tool-input-available" }> =>
+          chunk.type === "tool-input-available" &&
+          chunk.toolName === ACTIVATE_SKILL_TOOL_NAME,
+      ) ?? null;
     const pingOutputChunk = initialChunks.find(
       (chunk) =>
         chunk.type === "tool-output-available" &&
@@ -225,6 +250,22 @@ try {
     );
     const resumedChunks = chunksFrom(await resumeResponse.text());
     const snapshot = await historyClient.history();
+    const userEntryIds = snapshot.messages
+      .filter(
+        (message) => message.role === "user" && message.purpose === "user",
+      )
+      .map((message) => message.id);
+    const firstSweep = await applyCaptureSweep(identity, userEntryIds);
+    const secondSweep = await applyCaptureSweep(identity, userEntryIds);
+    const interviewerToolNames = [
+      ...new Set(
+        snapshot.messages.flatMap((message) =>
+          message.parts
+            .filter((part) => part.type === "dynamic-tool")
+            .map((part) => part.toolName),
+        ),
+      ),
+    ];
     let unauthenticatedHistoryStatus = 0;
     try {
       await createFlueClient({
@@ -320,6 +361,19 @@ try {
       transcript: formatFlueTranscript(snapshot),
       instanceId,
       dbPath: dbFile,
+      activateSkillCall,
+      interviewerToolNames,
+      captureUserText: userTextFromHistory(
+        snapshot.messages.map((message) => ({
+          role: message.role,
+          parts: message.parts,
+        })),
+      ),
+      captureIds: firstSweep.captures.map((capture) => capture.id),
+      recaptureIds: secondSweep.captures.map((capture) => capture.id),
+      skippedDedupKeys: secondSweep.skippedDedupKeys,
+      capturePayloads: firstSweep.captures.map((capture) => capture.payload),
+      captureExcerpts: firstSweep.captures.map((capture) => capture.excerpt),
     };
     process.stdout.write(`PETRINAUT_CHAT_RESULT ${JSON.stringify(result)}\n`);
   }