Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 2 additions & 13 deletions plugins/memory/mem0/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ The plugin authenticates with `X-API-Key` and uses the server's `/search` and `/

> Setting `host` routes to the self-hosted server automatically. Don't set `mode: oss` — OSS takes precedence and ignores `host`.

## Capture router and corrections
## Capture router

The optional `mem0_capture_router` block runs two dedicated extraction passes after a turn, then
partitions candidates deterministically by class:
Expand All @@ -76,18 +76,7 @@ partitions candidates deterministically by class:
gbrain inbox. That inbox is staging for `inbox-filer.py`; it is not a flat gbrain import root. The
filer is the single door that routes captures into entity pages before those pages are ingested.

**Corrections are highest-value and bypass the salience threshold.** Before `client.add`, the router
checks two deterministic signals: a configured correction marker (defaults include `no,`, `actually`,
`wrong`, and explicit correction phrasing), or a conservative contradiction between an extracted
world fact and a retrieved existing fact with the same subject but a different exact IP, port, or
enabled/disabled value. When either fires, the drain worker omits the normal salience prompt and stamps
`capture_correction` plus `capture_correction_signal` metadata; the ordinary class partition still
decides the destination.

This detector is deliberately conservative. If a marker is missed, existing-fact lookup is disabled or
fails, or the contradiction is semantic rather than one of the supported exact values, the turn follows
the normal salience path. That is the documented false-negative mode: degraded to ordinary capture, not
a second correction-specific drop path.
Every candidate goes through the same salience path; there is no correction-specific bypass.

## OSS (Self-Hosted) Mode

Expand Down
36 changes: 1 addition & 35 deletions plugins/memory/mem0/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1946,41 +1946,7 @@ def _build_capture_router(self):
except Exception:
return None
try:
router_cfg = ((self._config or {}).get("mem0_capture_router") or {})

def _existing_fact_lookup(query: str):
"""Best-effort existing-fact recall for RC1 contradiction correction detection.
Any failure degrades to [] so false negatives fall back to the normal salience gate."""
out = []
try:
top_k = int(router_cfg.get("contradiction_lookup_k", self._dedup_candidate_k()))
except Exception:
top_k = 5
try:
resp = self._get_client().search(
query=query,
filters=self._read_filters(),
top_k=top_k,
keyword_search=self._keyword_search,
)
for row in self._drop_forgotten(self._unwrap_results(resp)):
mem = row.get("memory") or row.get("text") or row.get("content")
if mem:
out.append(str(mem))
except Exception as e:
logger.debug("mem0 capture-router contradiction mem0 lookup degraded: %s", e)
try:
if self._gbrain_search_enabled:
for hit in self._gbrain_pointers(query, limit=top_k, deadline_s=1.5):
text = " ".join(str(hit.get(k) or "") for k in ("file", "title"))
if text.strip():
out.append(text.strip())
except Exception as e:
logger.debug("mem0 capture-router contradiction gbrain lookup degraded: %s", e)
return out

return capture_router.build_router_from_config(
self._config or {}, existing_fact_lookup_fn=_existing_fact_lookup)
return capture_router.build_router_from_config(self._config or {})
except Exception as e:
logger.warning("mem0 capture-router build failed (router disabled): %s", e)
return None
Expand Down
75 changes: 15 additions & 60 deletions plugins/memory/mem0/capture_drain.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,7 @@
store over HTTP, so the queue + worker live here, not in the mem0 container). A single background
loop that pulls queued turns and does the slow/failable work off the critical path:

lease a due row -> router preflight (optional correction detector; no staging yet)
-> client.add(messages, prompt=GATE, model=MODEL) [server-side extraction+gate]
(correction signals omit prompt=GATE so below-threshold corrections capture)
lease a due row -> client.add(messages, prompt=GATE, model=MODEL) [server-side extraction+gate]
-> record model_verdict (D-10)
-> reconcile by capture_idem: if rows already exist, mark done (exactly-once, D-8)
-> else scrub the just-written facts (INV-4 defense-in-depth) — A-lite has no
Expand Down Expand Up @@ -143,9 +141,9 @@ def __init__(
self._breaker_open = breaker_open_fn or (lambda: False)
self._alert = alert_fn or (lambda m: logger.error("mem0 capture alert: %s", m))
# Arm-B two-pass capture router (Phase 2.5), flag-gated: None => today's behavior EXACTLY
# (byte-identical). When injected, it runs ADDITIVELY: preflight computes correction signals
# before add() so corrections can bypass the salience prompt; world/event facts are staged
# only after the mem0 write+scrub boundary is clean. It NEVER touches mem0 and is fail-soft.
# (byte-identical). When injected, it runs ADDITIVELY after the unchanged mem0 add() path:
# two dedicated extraction passes -> deterministic class router -> world/event facts staged
# to disk. It NEVER touches mem0 and is fully fail-soft (a router error never fails a turn).
self._router = router
# Scrub failures never dead-letter (a secret must not be abandoned): requeue indefinitely
# with a capped backoff, and escalate once at this attempt threshold.
Expand Down Expand Up @@ -340,21 +338,12 @@ def drain_once(self) -> bool:
logger.warning("capture idem pre-check failed; requeued (fail-closed, no re-add): %s", e)
return True

# Router preflight (flag ON only): compute deterministic correction signals before add() so
# correction turns can bypass the salience prompt. World/event staging is deferred until the
# mem0 write+scrub boundary below is proven clean.
route_result = self._precompute_route(key, payload)
correction_signal = self._correction_signal_from_route(route_result)

# SERVER-SIDE extraction + gate. Stamp capture_idem so the reconcile can find the rows.
kwargs = dict(self._write_filters)
md = dict(kwargs.get("metadata") or {})
md["capture_idem"] = key
if correction_signal:
md["capture_correction"] = True
md["capture_correction_signal"] = str(correction_signal.get("type") or "unknown")[:80]
kwargs["metadata"] = md
if self._gate and not correction_signal:
if self._gate:
kwargs["prompt"] = self._gate
if self._model:
kwargs["model"] = self._model
Expand Down Expand Up @@ -419,62 +408,28 @@ def drain_once(self) -> bool:
if self._scrub_written_or_requeue(key, row, require_rows=bool(added_count)):
return True

# ARM-B ROUTER (Phase 2.5), flag-gated + ADDITIVE + fail-soft. The preflight above already
# computed world/event facts and correction signals. Here — only after the mem0 write+scrub
# boundary is clean — we stage world/event facts before mark_done. Router staging failure must
# NEVER requeue or fail the turn: the mem0 write is already durable and complete.
self._maybe_route(key, payload, messages, route_result=route_result)
# ARM-B ROUTER (Phase 2.5), flag-gated + ADDITIVE + fail-soft. The mem0 add() above already
# handled the preference/ops_state facts (unchanged path). Here — only when a router is wired
# (flag ON) — the two dedicated passes extract world/event facts and stage them to disk. This
# runs AFTER the row's mem0 write+scrub is proven clean and BEFORE mark_done, but a router
# failure must NEVER requeue or fail the turn: the mem0 write is already durable and complete.
self._maybe_route(key, payload, messages)

self._q.mark_done(key)
self.stats["drained"] += 1
return True

def _precompute_route(self, key, payload):
"""Run the capture router without staging so correction signals are known before add().
Fully fail-soft: failures return an error-shaped result and do not affect the mem0 path."""
if self._router is None:
return None
try:
session = payload.get("session_id") or "default"
return self._router.route_turn(
payload.get("user", ""), payload.get("assistant", ""),
turn_id=key[:16], session=session, ts=payload.get("ts"), stage=False)
except Exception as e:
logger.warning("capture-router: preflight failed for turn %s (mem0 write unaffected): %s",
key[:16], e)
return {"error": str(e), "world_facts": [], "prefs_facts": [], "correction_signals": []}

@staticmethod
def _correction_signal_from_route(route_result):
if not isinstance(route_result, dict) or not route_result.get("correction_detected"):
return None
signals = route_result.get("correction_signals") or []
for sig in signals:
if isinstance(sig, dict) and sig.get("type") == "marker":
return sig
for sig in signals:
if isinstance(sig, dict):
return sig
return {"type": "unknown"}

def _maybe_route(self, key, payload, messages, *, route_result=None) -> None:
def _maybe_route(self, key, payload, messages) -> None:
"""Invoke the Arm-B router for this turn if one is wired (flag ON). Fully fail-soft: any
error is swallowed (the mem0 write path is already complete and durable). No-op when the
router is None -> byte-identical to today."""
if self._router is None:
return
try:
session = payload.get("session_id") or "default"
if isinstance(route_result, dict):
res = route_result
if hasattr(self._router, "stage_route_result"):
res = self._router.stage_route_result(res)
else:
res = self._router.route_turn(
payload.get("user", ""), payload.get("assistant", ""),
turn_id=key[:16], session=session, ts=payload.get("ts"), stage=False)
if hasattr(self._router, "stage_route_result"):
res = self._router.stage_route_result(res)
res = self._router.route_turn(
payload.get("user", ""), payload.get("assistant", ""),
turn_id=key[:16], session=session, ts=payload.get("ts"))
if res.get("error"):
logger.warning("capture-router: turn %s routed with extract error: %s",
key[:16], res["error"])
Expand Down
Loading
Loading