feat(server): add guarded prompt retrieval adapter

server/index.ts

@@ -82,6 +82,10 @@ import {
 import * as tts from "./tts/index.ts";
 import { narrateTool, toUtterances } from "./tts/speech-text.ts";
 import { buildTurnContext, engineIsFresh } from "./turn-context.ts";
+import {
+  appendPromptRetrievalContext,
+  retrievePromptContext,
+} from "./prompt-retrieval.ts";
 import { TurnWatchdog } from "./turn-watchdog.ts";
 import {
   ensureWorkspace,
@@ -1436,6 +1440,11 @@ async function startTurn(
 
   void (async () => {
     try {
+      const retrievalContext = await retrievePromptContext(text, threadId);
+      const providerTurnText = appendPromptRetrievalContext(
+        turnText,
+        retrievalContext,
+      );
       const integrations: NonNullable<Parameters<typeof instance.adapter.sendTurn>[0]["integrations"]> = {};
       const selectedSkills = selectBundledSkills(
         text,
@@ -1659,7 +1668,7 @@ async function startTurn(
       watchdog.watch(threadId, bot.id);
       await instance.adapter.sendTurn({
         threadId,
-        text: turnText,
+        text: providerTurnText,
         model,
         effort,
         // a rewound thread never resumes the abandoned branch's session
@@ -1818,6 +1827,13 @@ function serializeRoomContext(threadId: string, userName: string): string {
     .join("\n");
 }
 
+function latestRoomUserPrompt(threadId: string): string {
+  return [...store.messagesFor(threadId)]
+    .reverse()
+    .find((message) => message.role === "user" && message.kind === "text" && message.text)
+    ?.text ?? "";
+}
+
 
 // comms bus: passed into the visibility helpers in comms-visibility.ts so
 // they can mirror messages + chips without re-deriving SSE plumbing. Same
@@ -1932,6 +1948,11 @@ async function runGroupMemberTurn(
   const text = `${serializeRoomContext(group.threadId, userName)}\n\n(Reply to the conversation above as ${bot.name}.)${
     connectorContinuation ? `\n\n${connectorContinuation}` : ""
   }`;
+  const retrievalContext = await retrievePromptContext(
+    latestRoomUserPrompt(group.threadId),
+    group.threadId,
+  );
+  const providerTurnText = appendPromptRetrievalContext(text, retrievalContext);
 
   // same workspace + memory as a 1:1 turn — the room is a different
   // conversation, not a different bot
@@ -1985,7 +2006,7 @@ async function runGroupMemberTurn(
     instance.adapter
       .sendTurn({
         threadId: group.threadId,
-        text,
+        text: providerTurnText,
         system: roomSystem,
         cwd,
         integrations,

server/prompt-retrieval.test.ts

@@ -0,0 +1,121 @@
+import { describe, expect, it, vi } from "vitest";
+
+import {
+  appendPromptRetrievalContext,
+  retrievePromptContext,
+} from "./prompt-retrieval.ts";
+
+const CONTEXT = [
+  '<fleet-retrieval-evidence trust="untrusted" instruction-authority="false">',
+  '{"hits":[]}',
+  "</fleet-retrieval-evidence>",
+].join("\n");
+
+interface ResponseOverrides {
+  schema?: string;
+  status?: string;
+  surface?: string;
+  interface?: string;
+  context?: string;
+  content_trust?: string;
+  instruction_authority?: boolean;
+  tool_authority?: boolean;
+  write_authority?: boolean;
+  selector_authority?: boolean;
+  promotion_authority?: boolean;
+  prompt_or_content_recorded_by_adapter?: boolean;
+}
+
+function response(overrides: ResponseOverrides = {}): Response {
+  return new Response(JSON.stringify({
+    schema: "aos.openmausbot-retrieval-adapter.v1",
+    status: "context_ready",
+    surface: "openmausbot",
+    interface: "loopback",
+    context: CONTEXT,
+    content_trust: "untrusted_retrieval_evidence",
+    instruction_authority: false,
+    tool_authority: false,
+    write_authority: false,
+    selector_authority: false,
+    promotion_authority: false,
+    prompt_or_content_recorded_by_adapter: false,
+    ...overrides,
+  }), { status: 200, headers: { "content-type": "application/json" } });
+}
+
+describe("guarded OpenMaus prompt retrieval", () => {
+  it("stays disabled without an explicit loopback endpoint", async () => {
+    const fetchImpl = vi.fn<typeof fetch>();
+    expect(await retrievePromptContext("find source", "thread-1", {
+      endpoint: "",
+      fetchImpl,
+    })).toBeNull();
+    expect(fetchImpl).not.toHaveBeenCalled();
+  });
+
+  it.each([
+    "https://127.0.0.1:8798",
+    "http://192.168.1.2:8798",
+    "http://user@127.0.0.1:8798",
+    "http://127.0.0.1:8798/other",
+  ])("rejects a non-loopback or ambiguous endpoint: %s", async (endpoint) => {
+    const fetchImpl = vi.fn<typeof fetch>();
+    expect(await retrievePromptContext("find source", "thread-1", {
+      endpoint,
+      fetchImpl,
+    })).toBeNull();
+    expect(fetchImpl).not.toHaveBeenCalled();
+  });
+
+  it("passes only the prompt and native session id and returns bounded context", async () => {
+    const fetchImpl = vi.fn<typeof fetch>().mockResolvedValue(response());
+    const context = await retrievePromptContext(
+      "Find the exact implementation",
+      "openmaus-thread-1",
+      { endpoint: "http://127.0.0.1:8798", fetchImpl },
+    );
+
+    expect(context).toBe(CONTEXT);
+    expect(fetchImpl).toHaveBeenCalledOnce();
+    const [url, init] = fetchImpl.mock.calls[0]!;
+    expect(String(url)).toBe("http://127.0.0.1:8798/v1/retrieve");
+    expect(JSON.parse(String(init?.body))).toEqual({
+      prompt: "Find the exact implementation",
+      session_id: "openmaus-thread-1",
+    });
+    expect(init?.cache).toBe("no-store");
+    expect(appendPromptRetrievalContext("user text", context)).toBe(
+      `user text\n\n${CONTEXT}`,
+    );
+  });
+
+  it.each([
+    { instruction_authority: true },
+    { tool_authority: true },
+    { write_authority: true },
+    { selector_authority: true },
+    { promotion_authority: true },
+    { prompt_or_content_recorded_by_adapter: true },
+    { context: "unwrapped" },
+  ])("drops an unsafe adapter response: %j", async (unsafe) => {
+    const fetchImpl = vi.fn<typeof fetch>().mockResolvedValue(response(unsafe));
+    expect(await retrievePromptContext("find source", "thread-1", {
+      endpoint: "http://localhost:8798/v1/retrieve",
+      fetchImpl,
+    })).toBeNull();
+  });
+
+  it("fails open on transport errors and oversize prompts", async () => {
+    const fetchImpl = vi.fn<typeof fetch>().mockRejectedValue(new Error("offline"));
+    expect(await retrievePromptContext("find source", "thread-1", {
+      endpoint: "http://127.0.0.1:8798",
+      fetchImpl,
+    })).toBeNull();
+    expect(await retrievePromptContext("x".repeat(8_193), "thread-1", {
+      endpoint: "http://127.0.0.1:8798",
+      fetchImpl,
+    })).toBeNull();
+    expect(fetchImpl).toHaveBeenCalledOnce();
+  });
+});

server/prompt-retrieval.ts

@@ -0,0 +1,118 @@
+import { z } from "zod";
+
+const RESPONSE_SCHEMA = "aos.openmausbot-retrieval-adapter.v1";
+const CONTENT_TRUST = "untrusted_retrieval_evidence";
+const DEFAULT_TIMEOUT_MS = 2_500;
+const MAX_PROMPT_BYTES = 8_192;
+const MAX_CONTEXT_BYTES = 2_048;
+const LOOPBACK_HOSTS = new Set(["127.0.0.1", "localhost", "[::1]"]);
+
+export interface PromptRetrievalOptions {
+  endpoint?: string;
+  fetchImpl?: typeof fetch;
+  timeoutMs?: number;
+}
+
+const AdapterResponseSchema = z.object({
+  schema: z.literal(RESPONSE_SCHEMA),
+  status: z.literal("context_ready"),
+  surface: z.literal("openmausbot"),
+  interface: z.literal("loopback"),
+  context: z.string(),
+  content_trust: z.literal(CONTENT_TRUST),
+  instruction_authority: z.literal(false),
+  tool_authority: z.literal(false),
+  write_authority: z.literal(false),
+  selector_authority: z.literal(false),
+  promotion_authority: z.literal(false),
+  prompt_or_content_recorded_by_adapter: z.literal(false),
+});
+
+type AdapterResponse = z.infer<typeof AdapterResponseSchema>;
+
+function adapterUrl(raw: string | undefined): URL | null {
+  if (!raw?.trim()) return null;
+  try {
+    const endpoint = new URL(raw.trim());
+    if (
+      endpoint.protocol !== "http:" ||
+      !LOOPBACK_HOSTS.has(endpoint.hostname) ||
+      endpoint.username ||
+      endpoint.password ||
+      endpoint.search ||
+      endpoint.hash ||
+      !["", "/", "/v1/retrieve"].includes(endpoint.pathname)
+    ) {
+      return null;
+    }
+    endpoint.pathname = "/v1/retrieve";
+    return endpoint;
+  } catch {
+    return null;
+  }
+}
+
+function acceptedContext(value: AdapterResponse): string | null {
+  if (
+    Buffer.byteLength(value.context, "utf8") > MAX_CONTEXT_BYTES ||
+    !value.context.startsWith(
+      '<fleet-retrieval-evidence trust="untrusted" instruction-authority="false">',
+    ) ||
+    !value.context.endsWith("</fleet-retrieval-evidence>")
+  ) {
+    return null;
+  }
+  return value.context;
+}
+
+/**
+ * Fetch one bounded, non-authoritative retrieval block for an OpenMaus turn.
+ *
+ * The adapter is optional and loopback-only. Every configuration, transport,
+ * timeout, or response-contract failure returns null without logging or
+ * retaining the prompt. The caller appends accepted context only to the
+ * provider-bound turn text, never to OpenMaus's durable transcript.
+ */
+export async function retrievePromptContext(
+  prompt: string,
+  sessionId: string,
+  options: PromptRetrievalOptions = {},
+): Promise<string | null> {
+  const endpoint = adapterUrl(
+    options.endpoint ?? process.env.OMB_PROMPT_RETRIEVAL_URL,
+  );
+  if (
+    !endpoint ||
+    !prompt.trim() ||
+    Buffer.byteLength(prompt, "utf8") > MAX_PROMPT_BYTES ||
+    !sessionId.trim()
+  ) {
+    return null;
+  }
+  const timeoutMs = Math.max(
+    100,
+    Math.min(options.timeoutMs ?? DEFAULT_TIMEOUT_MS, DEFAULT_TIMEOUT_MS),
+  );
+  try {
+    const response = await (options.fetchImpl ?? fetch)(endpoint, {
+      method: "POST",
+      headers: { "content-type": "application/json", accept: "application/json" },
+      body: JSON.stringify({ prompt, session_id: sessionId }),
+      cache: "no-store",
+      signal: AbortSignal.timeout(timeoutMs),
+    });
+    if (!response.ok) return null;
+    const value = AdapterResponseSchema.safeParse(await response.json());
+    if (!value.success) return null;
+    return acceptedContext(value.data);
+  } catch {
+    return null;
+  }
+}
+
+export function appendPromptRetrievalContext(
+  turnText: string,
+  context: string | null,
+): string {
+  return context ? `${turnText}\n\n${context}` : turnText;
+}