diff --git a/api/domain_event_handlers.py b/api/domain_event_handlers.py index 4126cb44..8ca010b1 100644 --- a/api/domain_event_handlers.py +++ b/api/domain_event_handlers.py @@ -117,27 +117,109 @@ async def _handle_forget(conn, domain: str, rid: str, payload: Dict[str, Any]): logger.info(f"domain.forget domain={domain} rid={rid_val}") +# Entity-domain embedding columns. Entity events do NOT carry a vector today +# (see `_build_entity_federation_payload` in personal_ingest_api for why: entity +# events run ~22k/30d at ~241 bytes, so attaching a 3072-float vector would take +# that domain from ~5 MB to ~1.3 GB/month). This set exists so that an inbound +# event which *does* carry one is honoured rather than silently dropped. +_ENTITY_EMBEDDING_COLS = frozenset({"embedding", "embedding_3072"}) + + async def _apply_entity(conn, rid: str, event_type: str, payload: Dict[str, Any], source_node: str): - """UPSERT entity into entity_registry.""" + """UPSERT entity into entity_registry with NON-DESTRUCTIVE merge semantics. + + A peer's copy of an entity is frequently *thinner* than ours — until + 2026-08-03 both emit sites sent hardcoded ``aliases: []`` / ``metadata: {}``, + so an inbound event did not merely fail to teach us anything, it actively + ERASED locally-computed values via ``SET metadata = EXCLUDED.metadata``. + Measured blast radius at the time of the fix: 644 rows on the MacBook node + and 964 on the NUC had ``first_seen_rid IS NOT NULL`` (proof that + ``store_new_entity`` ran locally and wrote a populated metadata JSON) yet + held ``metadata = '{}'``. There is no audit table, so those are unrecoverable + and that count is a lower bound. + + Merge rules — every optional field is merge-or-keep, never replace-with-empty: + * ``metadata`` jsonb concat; incoming keys win, absent keys survive, + and an empty ``{}`` is a no-op. + * ``aliases`` set union; the array can never shrink. + * ``description`` keep local unless the incoming value is non-empty. + * ``phonetic_code`` fill-if-missing (drives Tier-1.x phonetic matching). + * embedding fill-if-missing; never overwrite a local vector. + + Identity fields (``entity_text``/``entity_type``/``normalized_text``) are + still replaced outright — unchanged from the original behaviour, so that a + legitimate rename or retype still converges. + """ + _assert_embedding_format(payload, rid) + fuseki_uri = payload.get("fuseki_uri", rid) entity_text = payload.get("entity_text", "") entity_type = payload.get("entity_type", "") normalized_text = payload.get("normalized_text", entity_text.lower().strip()) aliases = normalize_alias_list(payload.get("aliases", [])) + metadata = payload.get("metadata", {}) + if isinstance(metadata, str): + try: + metadata = json.loads(metadata) + except (ValueError, TypeError): + metadata = {} + if not isinstance(metadata, dict): + metadata = {} + + description = payload.get("description") or None + phonetic_code = payload.get("phonetic_code") or None + + # Optional inbound vector (not emitted today; honoured if present). + emb_col = payload.get("embedding_column") + emb_literal = None + if emb_col in _ENTITY_EMBEDDING_COLS: + emb_literal = _format_vector(payload.get("embedding_value")) + if emb_literal is None: + emb_col = None + + cols = [ + "fuseki_uri", "entity_text", "entity_type", "normalized_text", + "aliases", "metadata", "description", "phonetic_code", + ] + casts = ["", "", "", "", "", "::jsonb", "", ""] + args = [ + fuseki_uri, entity_text, entity_type, normalized_text, + aliases or [], json.dumps(metadata), description, phonetic_code, + ] + if emb_col: + cols.append(emb_col) + casts.append("::vector") + args.append(emb_literal) + + placeholders = ", ".join(f"${i + 1}{casts[i]}" for i in range(len(cols))) + + set_clauses = [ + "entity_text = EXCLUDED.entity_text", + "entity_type = EXCLUDED.entity_type", + "normalized_text = EXCLUDED.normalized_text", + # Set union — aliases can never shrink across a federation round-trip. + "aliases = (SELECT COALESCE(array_agg(DISTINCT a), '{}'::text[]) " + "FROM unnest(COALESCE(entity_registry.aliases, '{}'::text[]) " + "|| COALESCE(EXCLUDED.aliases, '{}'::text[])) AS a)", + # jsonb concat — an empty incoming object is a no-op, so a thin peer + # can no longer blank a rich local row. + "metadata = COALESCE(entity_registry.metadata, '{}'::jsonb) " + "|| COALESCE(EXCLUDED.metadata, '{}'::jsonb)", + "description = COALESCE(NULLIF(EXCLUDED.description, ''), entity_registry.description)", + "phonetic_code = COALESCE(entity_registry.phonetic_code, EXCLUDED.phonetic_code)", + "updated_at = NOW()", + ] + if emb_col: + set_clauses.append( + f"{emb_col} = COALESCE(entity_registry.{emb_col}, EXCLUDED.{emb_col})" + ) - await conn.execute(""" - INSERT INTO entity_registry (fuseki_uri, entity_text, entity_type, normalized_text, aliases, metadata) - VALUES ($1, $2, $3, $4, $5, $6::jsonb) - ON CONFLICT (fuseki_uri) DO UPDATE SET - entity_text = EXCLUDED.entity_text, - entity_type = EXCLUDED.entity_type, - normalized_text = EXCLUDED.normalized_text, - aliases = EXCLUDED.aliases, - metadata = EXCLUDED.metadata, - updated_at = NOW() - """, fuseki_uri, entity_text, entity_type, normalized_text, - aliases or [], json.dumps(metadata) if isinstance(metadata, dict) else metadata) + await conn.execute( + f"INSERT INTO entity_registry ({', '.join(cols)}) VALUES ({placeholders}) " + f"ON CONFLICT (fuseki_uri) DO UPDATE SET {', '.join(set_clauses)}", + *args, + ) # Upsert relationships if included relationships = payload.get("relationships", []) diff --git a/api/personal_ingest_api.py b/api/personal_ingest_api.py index ae22f735..fa62b23c 100644 --- a/api/personal_ingest_api.py +++ b/api/personal_ingest_api.py @@ -1517,6 +1517,90 @@ async def store_new_entity( }, rid=canonical.uri, source_rid=document_rid) +# ============================================================================= +# Entity federation payload +# ============================================================================= + +async def _build_entity_federation_payload( + conn, + uri: str, + *, + fallback_name: str = "", + fallback_type: str = "", +) -> Dict[str, Any]: + """Build an entity federation payload from the authoritative registry row. + + Both emit sites used to inline a 6-key literal that hardcoded + ``"aliases": []`` and ``"metadata": {}``. That was doubly wrong: the peer + learned nothing, and — because the subscriber's UPSERT did + ``SET metadata = EXCLUDED.metadata`` — the empty payload ERASED whatever the + peer had computed locally. See ``_apply_entity`` in domain_event_handlers.py + for the merge side of this fix. + + Deliberately NOT included: the embedding vector. Entity events run ~22k per + 30 days at ~241 bytes each; a 3072-float vector is ~60 kB, which would take + this domain from roughly 5 MB to 1.3 GB per month. (For contrast the fact + domain does ship vectors, but it is only ~620 events per 30 days.) Receiving + nodes fill ``embedding_3072`` via ``scripts/backfill_entity_embeddings.py`` + instead, which also composes a richer text (name + context + description) + than the register path's bare name. + + Falls back to the caller-supplied name/type if the row has gone (e.g. the + entity was merged away between write and emit), so the emit never crashes + the request path. + """ + row = None + try: + row = await conn.fetchrow( + """ + SELECT fuseki_uri, entity_text, entity_type, normalized_text, + aliases, metadata, description, phonetic_code + FROM entity_registry + WHERE fuseki_uri = $1 + """, + uri, + ) + except Exception as e: # never let a federation emit break the write path + logger.warning(f"entity federation payload lookup failed for {uri}: {e}") + + if row is None: + return { + "fuseki_uri": uri, + "entity_text": fallback_name, + "entity_type": fallback_type, + "normalized_text": (fallback_name or "").lower().strip(), + "aliases": [], + "metadata": {}, + } + + metadata = row["metadata"] + if isinstance(metadata, str): + # NB: this module aliases the stdlib json as `json_module_global` + # (line 22) and never binds bare `json` — using `json.loads` here + # raises NameError at runtime, which py_compile does not catch. + try: + metadata = json_module_global.loads(metadata) + except (ValueError, TypeError): + metadata = {} + if not isinstance(metadata, dict): + metadata = {} + + payload: Dict[str, Any] = { + "fuseki_uri": row["fuseki_uri"], + "entity_text": row["entity_text"] or fallback_name, + "entity_type": row["entity_type"] or fallback_type, + "normalized_text": row["normalized_text"] + or (row["entity_text"] or fallback_name or "").lower().strip(), + "aliases": list(row["aliases"] or []), + "metadata": metadata, + } + if row["description"]: + payload["description"] = row["description"] + if row["phonetic_code"]: + payload["phonetic_code"] = row["phonetic_code"] + return payload + + # ============================================================================= # TerminusDB Outbox Helpers # ============================================================================= @@ -2692,16 +2776,22 @@ async def ingest_extraction(request: IngestRequest): ) logger.info(f"Resolved to existing: {canonical.uri}") - # Emit federation event for entity replication + # Emit federation event for entity replication. + # Payload is read back from entity_registry so the peer + # gets the real aliases/metadata/description rather than + # hardcoded empties that would blank its own copy. from api.federation_events import emit_domain_event - await emit_domain_event("entity", "NEW" if is_new else "UPDATE", canonical.uri, { - "fuseki_uri": canonical.uri, - "entity_text": canonical.name, - "entity_type": entity.type, - "normalized_text": canonical.name.lower().strip(), - "aliases": [], - "metadata": {}, - }) + await emit_domain_event( + "entity", + "NEW" if is_new else "UPDATE", + canonical.uri, + await _build_entity_federation_payload( + conn, + canonical.uri, + fallback_name=canonical.name, + fallback_type=entity.type, + ), + ) # Link entity to document await conn.execute(""" @@ -4082,16 +4172,22 @@ async def register_vault_entity(request: RegisterEntityRequest): koi_rid=final_koi_rid if request.publication_scope == "federated" else None ) - # Emit federation event for entity replication + # Emit federation event for entity replication. See + # _build_entity_federation_payload — the previous hardcoded + # {"aliases": [], "metadata": {}} literal here is what blanked + # peers' locally-computed metadata on every round-trip. from api.federation_events import emit_domain_event - await emit_domain_event("entity", "NEW" if is_new else "UPDATE", canonical.uri, { - "fuseki_uri": canonical.uri, - "entity_text": request.name, - "entity_type": request.entity_type, - "normalized_text": canonical.name.lower().strip() if canonical.name else request.name.lower().strip(), - "aliases": [], - "metadata": {}, - }) + await emit_domain_event( + "entity", + "NEW" if is_new else "UPDATE", + canonical.uri, + await _build_entity_federation_payload( + conn, + canonical.uri, + fallback_name=request.name, + fallback_type=request.entity_type, + ), + ) return result diff --git a/tests/unit/test_entity_federation_merge.py b/tests/unit/test_entity_federation_merge.py new file mode 100644 index 00000000..8522b80c --- /dev/null +++ b/tests/unit/test_entity_federation_merge.py @@ -0,0 +1,286 @@ +"""Entity federation must never blank a peer's locally-computed data. + +Regression cover for the 2026-08-03 defect: both `/register-entity` emit sites +inlined a payload literal that hardcoded ``"aliases": []`` / ``"metadata": {}``, +and the subscriber's UPSERT did ``SET metadata = EXCLUDED.metadata``. The result +was not merely "federation teaches the peer nothing" — an inbound event actively +ERASED metadata the peer had computed locally. Measured at the time of the fix: +644 rows on the MacBook node and 964 on the NUC had ``first_seen_rid IS NOT NULL`` +(proof ``store_new_entity`` ran locally and wrote a populated metadata JSON) yet +held ``metadata = '{}'``. + +Style follows tests/test_alias_normalization.py: AST guards + a fake-conn unit +test, no live DB and no app import (importing personal_ingest_api triggers +FastAPI app side effects). The actual merge SEMANTICS are proven against a live +database by the deploy-time integration check, since expressing +``jsonb ||`` / ``array_agg(DISTINCT ...)`` behaviour in a fake is worthless. +""" +from __future__ import annotations + +import ast +import sys +from pathlib import Path + +import pytest + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parent.parent +sys.path.insert(0, str(ROOT)) + +INGEST_PY = ROOT / "api" / "personal_ingest_api.py" +HANDLERS_PY = ROOT / "api" / "domain_event_handlers.py" + + +# --- Fake conn -------------------------------------------------------------- + +class _RecordingConn: + """Minimal asyncpg-connection stand-in that records executed SQL.""" + + def __init__(self, row=None): + self.executed: list[tuple[str, tuple]] = [] + self._row = row + + async def execute(self, sql, *args): + self.executed.append((sql, args)) + + async def fetchrow(self, sql, *args): + return self._row + + @property + def upsert_sql(self) -> str: + for sql, _ in self.executed: + if "INSERT INTO entity_registry" in sql: + return " ".join(sql.split()) + raise AssertionError("no entity_registry INSERT was executed") + + @property + def upsert_args(self) -> tuple: + for sql, args in self.executed: + if "INSERT INTO entity_registry" in sql: + return args + raise AssertionError("no entity_registry INSERT was executed") + + +async def _apply(payload): + from api.domain_event_handlers import _apply_entity + + conn = _RecordingConn() + await _apply_entity(conn, payload.get("fuseki_uri", "rid:x"), "UPDATE", payload, "peer") + return conn + + +# --- The core invariant ----------------------------------------------------- + +@pytest.mark.asyncio +async def test_upsert_never_replaces_metadata_wholesale(): + """The exact clause that caused the data loss must not come back.""" + conn = await _apply({ + "fuseki_uri": "orn:personal-koi.entity:person-x-1", + "entity_text": "X", + "entity_type": "Person", + "metadata": {}, + "aliases": [], + }) + sql = conn.upsert_sql + assert "metadata = EXCLUDED.metadata" not in sql, ( + "regression: an inbound event can blank a peer's metadata again" + ) + assert "aliases = EXCLUDED.aliases" not in sql, ( + "regression: an inbound event can shrink a peer's alias list again" + ) + # jsonb concat, so an empty incoming object is a no-op. + assert "entity_registry.metadata" in sql and "|| COALESCE(EXCLUDED.metadata" in sql + + +@pytest.mark.asyncio +async def test_upsert_merges_the_optional_fields_non_destructively(): + conn = await _apply({ + "fuseki_uri": "orn:personal-koi.entity:person-x-1", + "entity_text": "X", + "entity_type": "Person", + }) + sql = conn.upsert_sql + # alias set-union: the array can never shrink + assert "array_agg(DISTINCT a)" in sql + # description: keep local unless the incoming value is non-empty + assert "description = COALESCE(NULLIF(EXCLUDED.description, ''), entity_registry.description)" in sql + # phonetic_code drives Tier-1.x phonetic matching -> fill-if-missing only + assert "phonetic_code = COALESCE(entity_registry.phonetic_code, EXCLUDED.phonetic_code)" in sql + + +@pytest.mark.asyncio +async def test_identity_fields_still_replace_so_renames_converge(): + conn = await _apply({"fuseki_uri": "u", "entity_text": "New", "entity_type": "Person"}) + sql = conn.upsert_sql + for field in ("entity_text", "entity_type", "normalized_text"): + assert f"{field} = EXCLUDED.{field}" in sql + + +# --- Embedding: not emitted, but honoured if a peer sends one --------------- + +@pytest.mark.asyncio +async def test_embedding_column_omitted_when_payload_has_no_vector(): + conn = await _apply({"fuseki_uri": "u", "entity_text": "X", "entity_type": "Person"}) + sql = conn.upsert_sql + assert "embedding" not in sql + assert len(conn.upsert_args) == 8 # the 8 non-vector columns + + +@pytest.mark.asyncio +async def test_inbound_vector_is_honoured_and_never_overwrites_a_local_one(): + conn = await _apply({ + "fuseki_uri": "u", + "entity_text": "X", + "entity_type": "Person", + "embedding_column": "embedding_3072", + "embedding_value": [0.1, 0.2, 0.3], + }) + sql = conn.upsert_sql + assert "embedding_3072" in sql and "::vector" in sql + assert "embedding_3072 = COALESCE(entity_registry.embedding_3072, EXCLUDED.embedding_3072)" in sql + assert conn.upsert_args[-1] == "[0.1,0.2,0.3]" + + +@pytest.mark.asyncio +async def test_unknown_embedding_column_is_ignored_not_injected(): + """Guard against SQL injection through the payload-supplied column name.""" + conn = await _apply({ + "fuseki_uri": "u", + "entity_text": "X", + "entity_type": "Person", + "embedding_column": "embedding_3072; DROP TABLE entity_registry --", + "embedding_value": [0.1], + }) + assert "DROP TABLE" not in conn.upsert_sql + + +@pytest.mark.asyncio +async def test_rejects_unsupported_embedding_format(): + from api.domain_event_handlers import _apply_entity + + with pytest.raises(ValueError): + await _apply_entity( + _RecordingConn(), "u", "UPDATE", + {"fuseki_uri": "u", "entity_text": "X", "embedding_format": "base64"}, + "peer", + ) + + +@pytest.mark.asyncio +async def test_string_metadata_is_parsed_not_stringified(): + conn = await _apply({ + "fuseki_uri": "u", "entity_text": "X", "entity_type": "Person", + "metadata": '{"context": "festival vendor"}', + }) + # arg 6 (0-indexed 5) is the metadata JSON + assert "festival vendor" in conn.upsert_args[5] + + +@pytest.mark.asyncio +async def test_malformed_metadata_degrades_to_empty_not_crash(): + conn = await _apply({ + "fuseki_uri": "u", "entity_text": "X", "entity_type": "Person", + "metadata": "not json at all", + }) + assert conn.upsert_args[5] == "{}" + + +# --- Producer side ---------------------------------------------------------- + +@pytest.mark.asyncio +async def test_payload_builder_emits_real_registry_values(): + from api.personal_ingest_api import _build_entity_federation_payload + + row = { + "fuseki_uri": "orn:personal-koi.entity:person-ash-946fb5407d72", + "entity_text": "Ash", + "entity_type": "Person", + "normalized_text": "ash", + "aliases": ["ash the coffee guy"], + "metadata": {"context": "Turkish coffee vendor", "confidence": 1.0}, + "description": "Serves Turkish coffee at festivals.", + "phonetic_code": "AX", + } + payload = await _build_entity_federation_payload(_RecordingConn(row), row["fuseki_uri"]) + + assert payload["aliases"] == ["ash the coffee guy"], "the empty-literal defect is back" + assert payload["metadata"]["context"] == "Turkish coffee vendor" + assert payload["description"] == "Serves Turkish coffee at festivals." + assert payload["phonetic_code"] == "AX" + # Vector deliberately excluded — see the helper's docstring for the + # 22k-events/30d * 60kB payload-size arithmetic. + assert "embedding_value" not in payload + + +@pytest.mark.asyncio +async def test_payload_builder_falls_back_when_row_vanished(): + """A merge between write and emit must not crash the request path.""" + from api.personal_ingest_api import _build_entity_federation_payload + + payload = await _build_entity_federation_payload( + _RecordingConn(None), "u", fallback_name="Ash", fallback_type="Person" + ) + assert payload["entity_text"] == "Ash" + assert payload["entity_type"] == "Person" + assert payload["normalized_text"] == "ash" + + +@pytest.mark.asyncio +async def test_payload_builder_survives_a_failing_lookup(): + from api.personal_ingest_api import _build_entity_federation_payload + + class _Boom(_RecordingConn): + async def fetchrow(self, sql, *args): + raise RuntimeError("db gone") + + payload = await _build_entity_federation_payload( + _Boom(), "u", fallback_name="Ash", fallback_type="Person" + ) + assert payload["entity_text"] == "Ash" + + +# --- AST guards: no app import --------------------------------------------- + +def _calls_in(path: Path, func_name: str): + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + if isinstance(node, ast.AsyncFunctionDef) and node.name == func_name: + return node + raise AssertionError(f"{func_name} not found in {path}") + + +def test_no_emit_site_hardcodes_an_empty_entity_payload(): + """The literal that caused the incident must not reappear at any emit site.""" + tree = ast.parse(INGEST_PY.read_text()) + offenders = [] + for node in ast.walk(tree): + if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Name)): + continue + if node.func.id != "emit_domain_event": + continue + if not node.args or not isinstance(node.args[0], ast.Constant): + continue + if node.args[0].value != "entity": + continue + # 4th positional arg is the payload; a dict literal here is the bug. + if len(node.args) >= 4 and isinstance(node.args[3], ast.Dict): + keys = [k.value for k in node.args[3].keys if isinstance(k, ast.Constant)] + offenders.append((node.lineno, keys)) + assert not offenders, ( + f"entity federation emitted from an inline dict literal at {offenders}; " + "use _build_entity_federation_payload so real aliases/metadata are sent" + ) + + +def test_payload_builder_uses_the_module_json_alias(): + """This module aliases stdlib json as `json_module_global` and never binds + bare `json`; `json.loads` here is a runtime NameError that py_compile and + import-free unit tests both miss.""" + src = ast.get_source_segment( + INGEST_PY.read_text(), + _calls_in(INGEST_PY, "_build_entity_federation_payload"), + ) + assert "json_module_global.loads" in src + assert "\n json.loads" not in src and " json.loads" not in src.replace( + "json_module_global.loads", "" + )