fix(eve): address review — async-write status, reasoning filter, turn-bounded capture, best-effort dedup
Review by @kartik-mem0. Root cause across all four: Mem0's hosted add with
infer:true is async and returns a PENDING event before the memory is visible.
- store.ts/tools.ts: add() returns AddResult{status,eventId} via parseAddResult;
the v3 add path returns {status:"PENDING", event_id}, so remember reports
{status:"queued", eventId} for async writes and {status:"saved"} only when
resolved. FAILED/unknown status maps to completed, not queued.
- messages.ts: drop typed non-text parts (reasoning no longer leaks into memory);
bound the captured assistant reply to the current turn and collect all of its
text segments (text->tool->text) instead of a stale prior answer.
- capture.ts + README + eve.mdx: document operationId dedup as best-effort
(lookup+add is not atomic; pending/concurrent windows can double-write);
removed the "restart replays cannot write twice" claim.
Tests: 68 pass (+15). Added reasoning, tool-only/cross-turn, multi-segment,
pending/concurrent capture replay, and PENDING/FAILED add-result cases.
Typecheck + build clean.
This commit is contained in:
@@ -10,6 +10,12 @@ export async function captureCompletedTurn(input: {
|
||||
messages: readonly ConversationMessage[];
|
||||
turnInput: readonly ConversationMessage[];
|
||||
}): Promise<boolean> {
|
||||
// Best-effort deduplication, not a guarantee. This lookup + add is not atomic,
|
||||
// and with infer:true the prior write can still be a PENDING event whose
|
||||
// memory is not yet visible here. So a restart during that window, or two
|
||||
// workers that both clear this check before either write, can capture the
|
||||
// same operation twice. The in-process gate in provider.ts narrows the common
|
||||
// case; eliminating it would need a backend-supported atomic idempotency key.
|
||||
const existing = await input.store.listByMetadata({
|
||||
userId: input.scopeKey,
|
||||
metadata: { operation_id: input.operationId },
|
||||
|
||||
@@ -9,9 +9,16 @@ function textFromPart(part: unknown): string {
|
||||
if (typeof part === "string") {
|
||||
return part;
|
||||
}
|
||||
if (part && typeof part === "object" && "text" in part) {
|
||||
const text = (part as { text: unknown }).text;
|
||||
return typeof text === "string" ? text : "";
|
||||
if (part && typeof part === "object") {
|
||||
const record = part as { type?: unknown; text?: unknown };
|
||||
// Typed parts must be explicit text. A `reasoning` part also carries a
|
||||
// `text` field, so accepting any object with `text` would leak the model's
|
||||
// private reasoning into long-term memory. Untyped `{ text }` (no `type`
|
||||
// key) stays supported for callers that pass a bare text object.
|
||||
if ("type" in record && record.type !== "text") {
|
||||
return "";
|
||||
}
|
||||
return typeof record.text === "string" ? record.text : "";
|
||||
}
|
||||
return "";
|
||||
}
|
||||
@@ -72,11 +79,42 @@ export function completedTurnMessages(input: {
|
||||
return [];
|
||||
}
|
||||
|
||||
// Pair this turn's users with the latest assistant in the settled history.
|
||||
const history = conversationMessages(input.messages);
|
||||
const lastAssistant = [...history]
|
||||
.reverse()
|
||||
.find((message) => message.role === "assistant");
|
||||
// Pair this turn's users with the assistant reply produced *in this turn*
|
||||
// only. The current turn begins at the last user message in the settled
|
||||
// history; any assistant text after it belongs to this turn. Walking the
|
||||
// whole history instead would attach an older answer to the new user input
|
||||
// when this turn is tool-only (no assistant text of its own).
|
||||
let lastUserIndex = -1;
|
||||
for (let index = input.messages.length - 1; index >= 0; index -= 1) {
|
||||
if (input.messages[index]?.role === "user") {
|
||||
lastUserIndex = index;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return lastAssistant ? [...users, lastAssistant] : users;
|
||||
// Collect every assistant text segment produced in this turn, in order, so a
|
||||
// text -> tool-call -> text answer is captured whole rather than losing the
|
||||
// pre-tool content.
|
||||
const assistantSegments: string[] = [];
|
||||
if (lastUserIndex >= 0) {
|
||||
for (
|
||||
let index = lastUserIndex + 1;
|
||||
index < input.messages.length;
|
||||
index += 1
|
||||
) {
|
||||
const message = input.messages[index];
|
||||
if (message?.role !== "assistant") {
|
||||
continue;
|
||||
}
|
||||
const text = extractText(message.content);
|
||||
if (text.length > 0) {
|
||||
assistantSegments.push(text);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const assistantText = assistantSegments.join("\n");
|
||||
return assistantText
|
||||
? [...users, { role: "assistant", content: assistantText }]
|
||||
: users;
|
||||
}
|
||||
|
||||
@@ -27,6 +27,15 @@ export interface AddMemoryInput {
|
||||
readonly metadata: StoreMetadata;
|
||||
}
|
||||
|
||||
// A hosted-platform write with `infer: true` is asynchronous: Mem0 accepts the
|
||||
// request and returns an `event_id` with a PENDING status before extraction has
|
||||
// produced (or rejected) any memory. `queued` reflects that accepted-not-yet-
|
||||
// stored state; `completed` means the write resolved synchronously.
|
||||
export interface AddResult {
|
||||
readonly status: "queued" | "completed";
|
||||
readonly eventId?: string;
|
||||
}
|
||||
|
||||
export interface SearchMemoryInput {
|
||||
readonly userId: string;
|
||||
readonly topK: number;
|
||||
@@ -35,7 +44,10 @@ export interface SearchMemoryInput {
|
||||
}
|
||||
|
||||
export interface MemoryStore {
|
||||
add(messages: readonly MemoryMessage[], input: AddMemoryInput): Promise<void>;
|
||||
add(
|
||||
messages: readonly MemoryMessage[],
|
||||
input: AddMemoryInput,
|
||||
): Promise<AddResult>;
|
||||
search(query: string, input: SearchMemoryInput): Promise<{ results: readonly SearchHit[] }>;
|
||||
get(memoryId: string): Promise<MemoryRecord | null>;
|
||||
listByMetadata(input: {
|
||||
@@ -70,6 +82,35 @@ export function parseSearchHit(item: unknown): SearchHit | null {
|
||||
return { id, memory };
|
||||
}
|
||||
|
||||
// Event statuses that mean "accepted, extraction not finished". Only these map
|
||||
// to `queued`; a terminal status (SUCCEEDED/FAILED) or an unknown/absent status
|
||||
// falls through to `completed`, so a FAILED write is never reported as still
|
||||
// pending. (The hosted add path returns PENDING synchronously; FAILED only
|
||||
// appears later via event polling.)
|
||||
const PENDING_STATUSES = new Set(["PENDING", "RUNNING", "PROCESSING", "QUEUED"]);
|
||||
|
||||
export function parseAddResult(response: unknown): AddResult {
|
||||
if (response && typeof response === "object" && !Array.isArray(response)) {
|
||||
const record = response as Record<string, unknown>;
|
||||
const eventIdValue = record.event_id ?? record.eventId;
|
||||
const eventId =
|
||||
typeof eventIdValue === "string" && eventIdValue.length > 0
|
||||
? eventIdValue
|
||||
: undefined;
|
||||
const status =
|
||||
typeof record.status === "string" ? record.status.toUpperCase() : "";
|
||||
if (eventId && PENDING_STATUSES.has(status)) {
|
||||
return { status: "queued", eventId };
|
||||
}
|
||||
if (eventId) {
|
||||
return { status: "completed", eventId };
|
||||
}
|
||||
}
|
||||
// A memory-results array (or any non-event payload) means the write already
|
||||
// resolved; there is nothing left pending to report.
|
||||
return { status: "completed" };
|
||||
}
|
||||
|
||||
export function parseMemoryRecord(item: unknown): MemoryRecord | null {
|
||||
if (!item || typeof item !== "object") {
|
||||
return null;
|
||||
@@ -138,11 +179,12 @@ export async function createMem0Store(input: {
|
||||
|
||||
const store: MemoryStore = {
|
||||
async add(messages, options) {
|
||||
await client.add([...messages], {
|
||||
const response = await client.add([...messages], {
|
||||
userId: options.userId,
|
||||
infer: options.infer,
|
||||
metadata: { ...options.metadata },
|
||||
});
|
||||
return parseAddResult(response);
|
||||
},
|
||||
async search(query, options) {
|
||||
const response = await client.search(query, {
|
||||
|
||||
@@ -37,17 +37,29 @@ export function createMem0Tools(
|
||||
}),
|
||||
remember: defineTool({
|
||||
description:
|
||||
"Save one durable fact or preference about the current caller.",
|
||||
"Save one durable fact or preference about the current caller. On the " +
|
||||
"hosted platform the write is queued for extraction and may take a " +
|
||||
"moment to become searchable; a `queued` status means accepted, not yet " +
|
||||
"stored. Do not claim the fact is saved when the status is queued.",
|
||||
inputSchema: z.object({
|
||||
text: z.string().min(1).max(4000),
|
||||
}),
|
||||
async execute({ text }) {
|
||||
await store.add([{ role: "user", content: text }], {
|
||||
const result = await store.add([{ role: "user", content: text }], {
|
||||
userId: scopeKey,
|
||||
infer: options.infer,
|
||||
metadata: { source: "eve-tool" },
|
||||
});
|
||||
return { saved: true };
|
||||
// Report the true write state. With infer:true the platform returns a
|
||||
// PENDING event, so promising "saved" would let the model tell the user
|
||||
// a fact is stored when only the request was queued. Surface the eventId
|
||||
// so the write can be correlated/confirmed rather than left a dead end.
|
||||
return result.status === "queued"
|
||||
? {
|
||||
status: "queued" as const,
|
||||
...(result.eventId ? { eventId: result.eventId } : {}),
|
||||
}
|
||||
: { status: "saved" as const };
|
||||
},
|
||||
}),
|
||||
forget: defineTool({
|
||||
|
||||
Reference in New Issue
Block a user