feat(chat): per-user inbox anchor for listing kind=value threads

EA2 doesn't expose a generic flow-by-kind list endpoint; flow/processes
is definition-only. Migrated chat to flow.kind=value, so direct listing
needed a new strategy.

- Each user has a deterministic inbox flow.kind=value doc keyed by
  email slug. ensureInbox() creates it idempotently on first listThreads.
- createThread links the new thread into BOTH participants' inboxes via
  defines edges with role='presentation' (each link is one apply-batch).
- listThreads(userEmail) reads defines edges from the user's inbox key.
  No more reliance on source_context regex filtering through the
  definition catalogue.

Updated all 3 call sites: Chat.tsx (mount + after-create refresh) and
agentTools.send_chat. 31/31 tests green.
This commit is contained in:
2026-06-14 13:51:53 +04:00
parent 8f35f7b88b
commit 7aaba5fb47
3 changed files with 85 additions and 17 deletions
+1 -1
View File
@@ -107,7 +107,7 @@ const TOOLS: ToolDef[] = [
}; };
const recipient = Object.entries(directory).find(([k]) => target.includes(k))?.[1]; const recipient = Object.entries(directory).find(([k]) => target.includes(k))?.[1];
if (!recipient) return { ok: false, display: `I don't know who "${target}" is. Try Mariana, Aisha, or Rohan.` }; if (!recipient) return { ok: false, display: `I don't know who "${target}" is. Try Mariana, Aisha, or Rohan.` };
const threads = await chatApi.listThreads(); const threads = await chatApi.listThreads(ctx.userEmail);
const existing = threads.find((t) => t.participants.includes(recipient) && t.participants.includes(ctx.userEmail)); const existing = threads.find((t) => t.participants.includes(recipient) && t.participants.includes(ctx.userEmail));
const threadKey = existing?._key || (await chatApi.createThread(`${ctx.userEmail}${recipient}`, [ctx.userEmail, recipient])); const threadKey = existing?._key || (await chatApi.createThread(`${ctx.userEmail}${recipient}`, [ctx.userEmail, recipient]));
await chatApi.sendMessage(threadKey, body, ctx.userEmail); await chatApi.sendMessage(threadKey, body, ctx.userEmail);
+82 -14
View File
@@ -31,6 +31,49 @@ function newRequestId(): string {
return `chat-${Date.now()}-${Math.random().toString(36).slice(2, 12)}`; return `chat-${Date.now()}-${Math.random().toString(36).slice(2, 12)}`;
} }
function inboxKeyFor(email: string): string {
const slug = email.toLowerCase().replace(/[^a-z0-9]+/g, "_").slice(0, 60);
return `inbox_${slug}`;
}
async function ensureInbox(inboxKey: string, email: string, signal?: AbortSignal): Promise<boolean> {
try {
const head = await fetch(`${api.config.baseUrl}/api/ea2/flow/${inboxKey}`, { headers: authHeaders(), signal });
if (head.ok) return true;
} catch { /* fall through */ }
const created = await fetch(`${api.config.baseUrl}/api/ea2/flow`, {
method: "POST",
headers: authHeaders(),
body: JSON.stringify({
_key: inboxKey,
kind: "value",
status: "published",
name: inboxKey,
display_name: `Inbox · ${email}`,
description: `Chat inbox anchor for ${email}`,
source_context: "CANVAS_CHAT_INBOX",
config: { chat: { owner_email: email } },
}),
signal,
});
return created.ok || created.status === 409;
}
async function linkThreadToInbox(inboxKey: string, threadKey: string, signal?: AbortSignal): Promise<void> {
const body = {
request_id: newRequestId(),
ops: [
{ op: "create_edge", edge_coll: "defines", from: `flow/${inboxKey}`, to: `flow/${threadKey}`, role: "presentation" },
],
};
await fetch(`${api.config.baseUrl}/api/ea2/apply-batch`, {
method: "POST",
headers: authHeaders(),
body: JSON.stringify(body),
signal,
});
}
function authHeaders(): Record<string, string> { function authHeaders(): Record<string, string> {
return { return {
Authorization: `Bearer ${sessionStorage.getItem("fm.mc.token.v1")}`, Authorization: `Bearer ${sessionStorage.getItem("fm.mc.token.v1")}`,
@@ -48,19 +91,36 @@ async function jsonFetch(path: string, init?: RequestInit) {
} }
export const chatApi = { export const chatApi = {
/** List existing chat-thread flow docs (kind=value, source_context=CANVAS_CHAT_THREAD). */ /**
async listThreads(signal?: AbortSignal): Promise<ChatThread[]> { * List chat threads via the per-user inbox anchor and defines edges. Per-tenant
const res = await jsonFetch(`/api/ea2/flow/processes?limit=500&kind=value`, { signal }); * tenant_id of the inbox is implicit from the Bearer token.
const items = (res?.items || []) as any[]; */
return items async listThreads(userEmail: string, signal?: AbortSignal): Promise<ChatThread[]> {
.filter((it) => it?.source_context === "CANVAS_CHAT_THREAD") const inboxKey = inboxKeyFor(userEmail);
.map((it) => ({ const inbox = await ensureInbox(inboxKey, userEmail, signal);
_key: it._key, if (!inbox) return [];
display_name: it.display_name || "Chat", const edges = await jsonFetch(`/api/ea2/edges/defines?from=flow/${inboxKey}&limit=200`, { signal });
description: it.description || "", const threadKeys = ((edges?.items || []) as any[])
created_at: it.created_at, .filter((e) => e?.role === "presentation")
participants: it.config?.chat?.participants || [], .map((e) => (e._to || "").split("/").pop())
})); .filter(Boolean) as string[];
const threads = await Promise.all(
threadKeys.map(async (k) => {
try {
const doc = await jsonFetch(`/api/ea2/flow/${k}`, { signal });
return {
_key: doc._key,
display_name: doc.display_name || "Chat",
description: doc.description || "",
created_at: doc.created_at,
participants: doc.config?.chat?.participants || [],
} as ChatThread;
} catch {
return null;
}
})
);
return threads.filter((t): t is ChatThread => !!t).sort((a, b) => (b.created_at || "").localeCompare(a.created_at || ""));
}, },
/** /**
@@ -85,7 +145,15 @@ export const chatApi = {
}); });
if (!res.ok) throw new Error(`createThread ${res.status}`); if (!res.ok) throw new Error(`createThread ${res.status}`);
const body = await res.json(); const body = await res.json();
return body._key; const threadKey = body._key as string;
await Promise.all(
participants.map(async (email) => {
const inboxKey = inboxKeyFor(email);
await ensureInbox(inboxKey, email, signal);
await linkThreadToInbox(inboxKey, threadKey, signal);
})
);
return threadKey;
}, },
/** Read every message under a thread via the edges endpoint. */ /** Read every message under a thread via the edges endpoint. */
+2 -2
View File
@@ -34,7 +34,7 @@ export default function Chat() {
let cancelled = false; let cancelled = false;
setLoadingThreads(true); setLoadingThreads(true);
chatApi chatApi
.listThreads() .listThreads(me)
.then((rows) => { .then((rows) => {
if (cancelled) return; if (cancelled) return;
setThreads(rows); setThreads(rows);
@@ -73,7 +73,7 @@ export default function Chat() {
try { try {
const otherLabel = personaLabel(recipient); const otherLabel = personaLabel(recipient);
const key = await chatApi.createThread(`${personaLabel(me)}${otherLabel}`, [me, recipient]); const key = await chatApi.createThread(`${personaLabel(me)}${otherLabel}`, [me, recipient]);
const next = await chatApi.listThreads(); const next = await chatApi.listThreads(me);
setThreads(next); setThreads(next);
setActiveKey(key); setActiveKey(key);
pushToast("ok", `Thread opened with ${otherLabel}`); pushToast("ok", `Thread opened with ${otherLabel}`);