Stage-5 entity-resolution — TECH (ID-53.3)

Stage-5 entity-resolution — TECH (ID-53.3)

Spec slug: stage-5-entity-resolution Subtask: ID-53.3 (TECH, succeeds {53.2 PRODUCT}, precedes {53.4 PLAN} per Q-PLANNER-2 fresh-per-Subtask discipline) Parent Task: ID-53 — Canonical pipeline Stage-5 entity-resolution — spec rescope + op_id migration (S273 deferred from ID-49.5) Critical-path position: Implements the second half of the canonical-pipeline §1 critical-path risk (“unresolved entities”). Stage-4 embedding LANDED via ID-49.2; until Stage-5 lands the corpus serves vector search but cannot serve deduplicated entity queries. Gates Mempalace T12 KG integration (UNPROMOTED per PRODUCT Inv-19) and conditions the v1 product behaviour for cross-document entity dedup. Author: task-planner (opus-4-7, thinking: max, isolation=none) on cmux-worker-subo-id-53-f63aba0a from parent tip 80fbbd5e. FRESH INSTANCE — NOT the {53.1} RESEARCH author and NOT the {53.2} PRODUCT author (Q-PLANNER-2 fresh-per-Subtask discipline).

---

§1. Context

This TECH ratifies and implements Option B as ratified by PRODUCT Inv-1: Stage-5 entity-resolution runs as a flow-scope post-fan-out resolution stage inside app_main, between await handle.ready() and the flow-end webhook emit. The S265 → S273 → ID-53 lineage and the empirical mismatch between cocoindex 1.0.3’s collection-level resolve_entities API and the per-item mount_each topology are documented in full at docs/specs/id-53-stage-5-entity-resolution/RESEARCH.md §R0–§R7; this TECH does not restate them.

This TECH commits the 14 Proposed changes (§P-1 through §P-14) that turn the 21 ratified PRODUCT invariants into a concrete implementation plan. The substrate landed by ID-49.2/49.4/49.6 (Stage-4 embedding, bind_stage_counter, _FlowStageCounter, _empty_stage_counts() with the "entity_resolution": 0 slot, the _PIPELINE_ERROR_CLASSES tuple including "entity_resolution_failed", _emit_stage_error_log for structured-log emission, the _emit_pipeline_run_webhook rollup, and the FLOW_META_CTX / bind_flow_meta / current_flow_meta substrate) survives unchanged; this TECH consumes it from the per-item phase (canonicalise_entity_name, extract_entity_context, entity_mentions declare_row) and from the new Stage-5 post-pass (_run_stage_5_resolution).

Lineage references (read in full to validate Proposed changes):

• docs/specs/id-53-stage-5-entity-resolution/PRODUCT.md — ratified {53.2} (commit 80fbbd5e; 21 invariants; Inv-1 Option B; Inv-11 per-row delta semantics — PRODUCT elevation; 3 P-OQs deferred to this TECH).
• docs/specs/id-53-stage-5-entity-resolution/RESEARCH.md — ratified {53.1} (commit 369e77be; §R1 capability probes; §R3.2 Option B architecture sketch; §R4 platform implications; §R5 faiss pin assessment; §R6.2 op_id migration SQL verbatim; §R-RECOMMENDATION 8-item mitigation stack).
• docs/specs/id-28-cocoindex-flow-scaffolding/TECH.md — peer TECH (T8); §P-4.M1 op_id pattern this TECH mirrors verbatim for entity_mentions; §P-2 reactive mount_each shape; §P-5 _emit_upsert_log substrate semantic (declared ≠ applied).
• scripts/cocoindex_pipeline/flow.py HEAD 80fbbd5e — line 864 extract_entity_mentions discard site; lines 1060–1069 mount_each + handle.ready() attach point; lines 1083–1084 Stage-5 deferred comment range; line 1135 _emit_pipeline_run_webhook; line 162 _PIPELINE_ERROR_CLASSES; line 168 "entity_resolution_failed" literal; line 278 _emit_stage_error_log; lines 382–395 _empty_stage_counts() with "entity_resolution": 0 slot; line 915 stage_counter.increment("embedding") per-row platform pattern.
• scripts/cocoindex_pipeline/extraction.py — line 180 EntityMentionExtraction class (entity_type / entity_name / canonical_name / source_span_start / source_span_end / mention_confidence); line 551 extract_entity_mentions @coco.fn(memo=True).
• scripts/cocoindex_pipeline/flow_context.py — FLOW_META_CTX / bind_flow_meta / current_flow_meta / bind_retry_counter / bind_stage_counter substrate (RESEARCH §R2.7 verified reusable).
• lib/entities/entity-context.ts:19 — extractEntityContext(text, entityName): string | null TypeScript source TECH ports to Python (§P-5).
• lib/ai/classify.ts:1543–1546 (delete-before-insert); lib/ai/classify.ts:1611 (extractEntityContext call site); lib/ai/classify.ts:1751 (INSERT path) — coexistence reference.
• lib/entities/entity-aliases.ts:50–95 — legacy entity_aliases map shape (Inv-10 preload reference; alias is keyed by alias field, returns canonical).
• supabase/migrations/20260521203414_t8_op_id_propagation.sql — peer migration this TECH mirrors verbatim for entity_mentions.
• supabase/types/database.types.ts lines 1106–1131 (entity_aliases row) + 1133–1182 (entity_mentions row).
• CLAUDE.md — DDL via CLI only; SET search_path = public, extensions on new PL/pgSQL fns; IF NOT EXISTS on every ALTER + CREATE; UK English; no barrel re-exports; semantic tokens (not applicable — server-side); classifyContent userId UUID gotcha.

What this TECH does:

• Maps every PRODUCT invariant Inv-1..Inv-21 to one or more Proposed changes §P-1..§P-14 with file paths, function signatures, schema diffs, migration shape, and acceptance test shape.
• Resolves the 3 P-OQs left open by PRODUCT (§4): P-OQ1 contract-break permanence — RATIFIED PERMANENT; P-OQ2 narrow race window — DOCUMENTED + ACCEPTED; P-OQ3 PairResolver cache table shape — RATIFIED via new entity_pair_resolutions table.
• Commits the entity_mentions op_id migration SQL from RESEARCH §R6.2 verbatim + the new entity_pair_resolutions cache table in a single CLI migration.
• Commits the requirements.txt faiss-cpu==1.14.2 pin per PRODUCT Inv-18.
• Documents the layered implementation plan so {53.4} PLAN can decompose into TM-shape implementation Subtasks {53.5+}.

What this TECH does NOT do (per PRODUCT §3 out-of-scope):

• Implement Mempalace T12 KG integration — UNPROMOTED per PRODUCT Inv-19; hard-gated on Stage-5 stable + tier-1 observation window.
• Author the canonical-pipeline-sequencing doc amendment that documents Option B as the stable architectural pattern — that amendment lands at implementation close (a {53.4+} Subtask), not during TECH authoring.
• Backfill historic entity_mentions rows with NULL op_id — forward-only migration per PRODUCT §3.
• Introduce cross-workspace entity merging — explicit v1 scope decision per Inv-21.
• Touch any other Task in task-list.json (peer-Task changes are out of scope per dispatch brief).

---

§2. Proposed changes (§P-1 through §P-14)

Each Proposed change maps one or more PRODUCT invariants to a concrete implementation slice. Coverage matrix at §3 confirms 21/21 invariants covered. Each §P-N includes: covers list, implementation summary, files touched / created, migration shape (where applicable), acceptance test shape, effort estimate.

§P-1. Stage-5 post-pass attach at app_main

Covers: Inv-1, Inv-2.

Implementation summary. Introduce a new async function _run_stage_5_resolution(meta: FlowRunMeta, db_pool: asyncpg.Pool, flow_stage_counter: _FlowStageCounter) -> int invoked from app_main immediately after await handle.ready() returns at scripts/cocoindex_pipeline/flow.py:1069 and before _emit_pipeline_run_webhook at flow.py:1135. The current Stage-5 deferred comment at flow.py:1084 (line 1083 is blank) is replaced with the actual Stage-5 call. The function returns the count of entity_mentions rows whose canonical_name it materially changed; that count feeds the per-row stage_counts["entity_resolution"] delta per Inv-11 (§P-6).

The Stage-5 call is wrapped in a try/except (§P-10) so a Stage-5 failure routes through the existing _classify_stage_exception substrate (flow.py:182) — the "entity_resolution_failed" literal at flow.py:168 is already declared in _PIPELINE_ERROR_CLASSES. The per-item write phase (which already committed all entity_mentions rows by the time handle.ready() returns) is non-destructively preserved per Inv-12; the failure surface adds no row deletion.

The managed_by=USER row-only contract is deliberately broken by this UPDATE-pass write — PRODUCT Inv-2 names this as the architectural choice (P-OQ1 PERMANENT, §4 below). cocoindex itself stays row-only via the per-item declare_row writes inside ingest_file; the Stage-5 UPDATE phase is KH-owned and reads as “second-phase canonicalisation”. The post-pass is op_id-scoped (Inv-5) so it does not collide with concurrent app-side writes on rows from other runs.

Files touched / created:

• scripts/cocoindex_pipeline/flow.py — EDIT. Replace line 1084 (the Stage-5 deferred comment; line 1083 is blank) with the Stage-5 invocation. The call sits inside the outer try block (~line 1006) so any exception routes through the existing except clause at line 1089 unchanged. New imports at the top of flow.py: from scripts.cocoindex_pipeline.stage_5 import _run_stage_5_resolution.

  Insertion shape (positioned between current flow.py:1083 “Stage 5: entity resolution — deferred” and flow.py:1086 “Inv-13 per-row upsert logging — deferred” comments):

  # ── Stage 5: entity resolution — flow-scope post-fan-out (ID-53) ──
  # Op-scope cross-document canonicalisation. Reads the run's
  # entity_mentions rows (per-doc canonicals written by the per-item phase
  # via the §P-3 declare_row site), invokes
  # cocoindex.ops.entity_resolution.resolve_entities over them with the KH
  # entity embedder (§P-7) + KH PairResolver (§P-8), and UPDATEs rows
  # whose canonical_name resolves to a different cross-document value.
  # Strictly op_id-scoped per Inv-5 — the post-pass NEVER touches rows
  # from prior runs or NULL-op_id rows (app-side writes).
  #
  # PRODUCT.md §2 Area A (Inv-1, Inv-2) — Option B ratification.
  # Read FlowRunMeta from the active FLOW_META_CTX (bound by bind_flow_meta
  # earlier in app_main per the existing flow_context.py substrate); avoids
  # constructing a detached FlowRunMeta from raw run_op_id.
  resolved_count = await _run_stage_5_resolution(
      meta=current_flow_meta(),
      db_pool=_resolve_db_pool(),
      flow_stage_counter=flow_stage_counter,
  )

• scripts/cocoindex_pipeline/stage_5.py — NEW (§P-6 owns the body). Houses _run_stage_5_resolution plus its private helpers (_preload_entity_aliases, _select_run_entity_mentions, _update_resolved_canonicals).

Migration shape: None.

Acceptance test shape:

• Inv-1: __tests__/integration/cocoindex/stage-5-attach-point.integration.test.ts — ingest a corpus with at least two documents sharing a near-match entity (“ISO 27001” / “ISO27001”). Assert: (a) all entity_mentions rows for the run are committed BEFORE any Stage-5 UPDATE fires (verifiable via a tap inside _run_stage_5_resolution that logs mentions_loaded count); (b) pipeline_runs.result.stage_counts["entity_resolution"] >= 0; (c) the post-pass executes entirely between handle.ready() and _emit_pipeline_run_webhook.
• Inv-2: Documented in §4 P-OQ1; verification is via the canonical-pipeline-sequencing.md amendment that lands at impl close. No integration test; the architecture documentation IS the proof.

Effort estimate: ~1.5 h (0.5 h flow.py edit + 1 h skeleton stage_5.py with thin body; §P-6 fills out the body).

---

§P-2. Per-document deterministic canonicalisation function

Covers: Inv-4.

Implementation summary. Introduce a pure Python function canonicalise_entity_name(name: str, entity_type: str) -> str in a new module scripts/cocoindex_pipeline/canonicalisation.py. The function mirrors the semantics of scripts/kb_pipeline/classify.py canonicalise() (the old KB pipeline reference TECH already cites — see RESEARCH §R3.1). Algorithm:

1. Trim whitespace.
2. Lowercase + strip + ASCII-fold.
3. Apply entity_type-aware normalisation rules:
   • certification: collapse “ISO27001” → “ISO 27001”; strip version suffix “ISO 27001:2022” → “ISO 27001”; handle “ISO/IEC 27001” → “ISO 27001”.
   • technology / product: strip the trailing “(.com)” / “(.io)” common suffixes.
   • regulation, framework, standard, methodology, capability, organisation, person, project, sector: pass through after step (2) — entity-type-specific normalisation is bounded to the four types where deterministic patterns hold.
4. Return the normalised value.

The function is deterministic (no randomness, no external calls, no time-dependence) and idempotent (canonicalise_entity_name(canonicalise_entity_name(x, t), t) == canonicalise_entity_name(x, t)). It is wrapped with @coco.fn(memo=True) ONLY when consumed inside ingest_file (so cocoindex memoises the per-row result on content-hash); the pure function is also importable directly for unit tests.

Files touched / created:

• scripts/cocoindex_pipeline/canonicalisation.py — NEW.

  """Per-document deterministic canonicalisation for entity names.
  
  PRODUCT.md Inv-4: the per-item phase writes a deterministic per-doc
  canonical_name via this function BEFORE Stage-5 runs. The Stage-5
  post-pass (§P-6) UPDATEs the value when cross-document resolution
  maps to a different canonical; Stage-5 NEVER inserts rows.
  
  Algorithm mirrors scripts/kb_pipeline/classify.py canonicalise() so
  pipeline-produced canonicals match the legacy KB-pipeline contract.
  """
  
  from __future__ import annotations
  
  import re
  import unicodedata
  
  _ISO_SLASH_RE = re.compile(r"\biso\s*/\s*iec\s+", re.IGNORECASE)
  _ISO_TIGHT_RE = re.compile(r"\biso\s*(\d{4,5})\b", re.IGNORECASE)
  _ISO_VERSION_RE = re.compile(r":(\d{4})\b")
  
  def canonicalise_entity_name(name: str, entity_type: str) -> str:
      """Return the per-document canonical for an entity name.
  
      Args:
          name: The raw entity name extracted by the LLM.
          entity_type: One of the 12 canonical entity_type values
              (database.types.ts:1141 enum).
  
      Returns:
          The lowercase + ASCII-folded + entity_type-aware-normalised
          canonical_name. Deterministic + idempotent.
      """
      if not name:
          return ""
      # Step 1: trim
      result = name.strip()
      # Step 2: ASCII-fold then lowercase
      result = unicodedata.normalize("NFKD", result)
      result = "".join(c for c in result if not unicodedata.combining(c))
      result = result.lower()
      # Step 3: entity_type-aware normalisation
      if entity_type == "certification":
          result = _ISO_SLASH_RE.sub("iso ", result)
          result = _ISO_TIGHT_RE.sub(lambda m: f"iso {m.group(1)}", result)
          result = _ISO_VERSION_RE.sub("", result)
      # Steps for technology/product trailing-suffix strip omitted for brevity
      # — TECH commits the v1 surface; richer rules surfaced in {53.4} PLAN.
      return result.strip()

• scripts/tests/test_canonicalise_entity_name.py — NEW. Unit tests: idempotent under repeated application; deterministic under identical input; “ISO27001” + “ISO 27001” + “iso/iec 27001” + “ISO 27001:2022” all collapse to “iso 27001”.

Migration shape: None.

Acceptance test shape:

• Inv-4: __tests__/integration/cocoindex/per-doc-canonicalisation.integration.test.ts — ingest a single document containing “ISO 27001”; assert entity_mentions.canonical_name = "iso 27001" for the resulting row. Re-ingest the same document; assert the row is byte-identical (memo + idempotent canonicaliser).

Effort estimate: ~1.5 h (0.5 h function + 0.5 h unit tests + 0.5 h integration test wiring).

---

§P-3. entity_mentions declare_row in ingest_file

Covers: Inv-4, Inv-7, Inv-15, Inv-16, Inv-17.

Implementation summary. Replace the discard pattern at scripts/cocoindex_pipeline/flow.py:864 (currently await extract_entity_mentions(content_text) with no assignment) with a meaningful assignment + a row-construction loop that writes entity_mentions rows via a new em_target TableTarget handle (§P-4).

For each EntityMentionExtraction Pydantic instance the LLM returned (scripts/cocoindex_pipeline/extraction.py:551 extract_entity_mentions @coco.fn(memo=True)), the per-item phase:

1. Computes the per-document canonical via canonicalise_entity_name(mention.entity_name, mention.entity_type) (§P-2).
2. Computes the context snippet via the Python port extract_entity_context(content_text, mention.entity_name) (§P-5).
3. Constructs an entity_mentions row dict mapping the Pydantic fields to DB columns:
   • Pydantic entity_name → DB entity_name (passthrough).
   • Pydantic entity_type → DB entity_type (passthrough; 12-CHECK enum matches per RESEARCH §R2.3).
   • Pydantic mention_confidence: float → DB confidence: numeric(3,2) (Inv-15: KH preserves the Pydantic field name mention_confidence; the mapping happens at row-construction time, NOT at the Pydantic schema level — preserves the LLM-output contract).
   • Pydantic source_span_start / source_span_end → DB metadata jsonb (Inv-16: stash under metadata.source_span_start / metadata.source_span_end rather than dedicated columns; no DDL).
   • Computed canonical_name (per-doc default) → DB canonical_name (Inv-4: written BEFORE Stage-5 runs; Stage-5’s UPDATE may rewrite this value if cross-document resolution maps to a different canonical).
   • Computed context_snippet → DB context_snippet (Inv-17: NOT NULL — populated by the Python port).
   • op_id from current_flow_meta().op_id → DB op_id (Inv-6 + Inv-7: stamped per-flow; memo-respecting because ingest_file is @coco.fn(memo=True), so re-ingest of unchanged source bytes does NOT re-stamp op_id).
   • Stable per-mention PK: uuid.uuid5(_KH_PIPELINE_DOC_NS, f"em:{rel_path}:{idx}") — mirrors the existing pattern at flow.py:872 (sd_id) and flow.py:873 (ci_id). Re-ingesting the same document mints the same PK → declare_row UPSERTs the existing row + re-stamps any changed fields per Inv-7.
4. Issues em_target.declare_row(row=<dict>).

This replaces the discard at line 864:

# OLD (flow.py:864): await extract_entity_mentions(content_text)
# NEW:
entity_mentions = await extract_entity_mentions(content_text)
for idx, mention in enumerate(entity_mentions):
    # canonicalise_entity_name is SYNC (-> str); do NOT await it.
    per_doc_canonical = canonicalise_entity_name(
        mention.entity_name, mention.entity_type,
    )
    context_snippet = extract_entity_context(content_text, mention.entity_name)
    em_target.declare_row(
        row={
            "id": uuid.uuid5(_KH_PIPELINE_DOC_NS, f"em:{rel_path}:{idx}"),
            "content_item_id": content_item_id,
            "entity_type": mention.entity_type,
            "entity_name": mention.entity_name,
            "canonical_name": per_doc_canonical,
            "confidence": mention.mention_confidence,  # Inv-15 map-at-declare-row
            "context_snippet": context_snippet,
            "metadata": {
                "source_span_start": mention.source_span_start,
                "source_span_end": mention.source_span_end,
            },
            "op_id": op_id,
        }
    )

The em_target handle is mounted in app_main (§P-4); it is passed positionally to ingest_file as a fourth target argument (mirroring ci_target, qa_target, sd_target).

Files touched / created:

• scripts/cocoindex_pipeline/flow.py — EDIT. Add em_target: Any to the ingest_file signature; replace the discard at line 864 with the assignment + row-construction loop above; reference §P-4 for the em_target mount_table_target call in app_main.

Migration shape: None.

Acceptance test shape:

• Inv-4: Same test as §P-2; the per-doc canonical_name lands on the row.
• Inv-7: __tests__/integration/cocoindex/stage-5-op-id-memo.integration.test.ts — ingest a file at op_id A; assert entity_mentions.op_id = A for the produced rows. Re-ingest unchanged at op_id B; assert rows still show op_id A (memo SKIP — ingest_file is @coco.fn(memo=True)). Trigger full_reprocess=True at op_id C; assert rows show op_id C.
• Inv-15: Same Inv-4 integration test extended — assert entity_mentions.confidence equals mention.mention_confidence for each row.
• Inv-16: Same test extended — assert entity_mentions.metadata->>'source_span_start' and entity_mentions.metadata->>'source_span_end' match the LLM-extracted spans.
• Inv-17: __tests__/integration/cocoindex/context-snippet-populated.integration.test.ts — ingest a file containing a known entity; assert entity_mentions.context_snippet IS NOT NULL for every produced row AND matches the Python-port output for the same (content_text, entity_name) input (§P-5 parity gate).

Effort estimate: ~1.5 h (0.5 h ingest_file signature change + row-construction loop; 1 h 4 integration tests).

---

§P-4. entity_mentions TableTarget mount in app_main

Covers: Inv-6.

Implementation summary. Mount a new em_target TableTarget alongside the existing three (ci_target, qa_target, sd_target) at scripts/cocoindex_pipeline/flow.py:1012–1029. The TableSchema declares the subset of entity_mentions columns that the per-item phase writes (other columns — created_at, entity_type_override, normalisation_version — are PG-defaulted and OMITTED per the existing content_text_hash GENERATED ALWAYS convention).

Files touched / created:

• scripts/cocoindex_pipeline/flow.py — EDIT. Add a new ENTITY_MENTIONS_SCHEMA TableSchema declaration adjacent to the existing three (~line 685–706). Add em_target = await mount_table_target(DB_CTX, "entity_mentions", ENTITY_MENTIONS_SCHEMA, managed_by=ManagedBy.USER) adjacent to the existing three target mounts (~line 1025–1029). Pass em_target as a fourth positional argument to coco.mount_each(ingest_file, source.items(), ci_target, qa_target, sd_target, em_target) at line 1060–1066. Update the ingest_file signature accordingly (§P-3).

  ENTITY_MENTIONS_SCHEMA shape (read off supabase/types/database.types.ts:1133–1182 + the op_id column added by §P-9):

  ENTITY_MENTIONS_SCHEMA = TableSchema(
      columns={
          "id": ColumnDef(type="uuid", nullable=False),
          "content_item_id": ColumnDef(type="uuid", nullable=False),
          "entity_type": ColumnDef(type="text", nullable=False),
          "entity_name": ColumnDef(type="text", nullable=False),
          "canonical_name": ColumnDef(type="text", nullable=False),
          "confidence": ColumnDef(type="numeric", nullable=True),
          "context_snippet": ColumnDef(type="text", nullable=True),
          "metadata": ColumnDef(type="jsonb", nullable=True),
          "op_id": ColumnDef(type="uuid", nullable=True),  # §P-9 migration
      },
      primary_key=("id",),
  )

Migration shape: None (the migration that adds op_id to entity_mentions is §P-9).

Acceptance test shape:

• Inv-6: __tests__/integration/cocoindex/op-id-round-trip.integration.test.ts — ingest a file; query an entity_mentions row the run produced; read its op_id; SELECT * FROM pipeline_runs WHERE op_id = <value> returns exactly one row. Mirrors peer T8 PRODUCT.md Inv-12 verification.

Effort estimate: ~1 h (0.5 h schema declaration + mount edit; 0.5 h integration test).

---

§P-5. Python port of extractEntityContext

Covers: Inv-17.

Implementation summary. Port lib/entities/entity-context.ts:19 extractEntityContext(text, entityName): string | null to Python at a new module scripts/cocoindex_pipeline/entity_context.py. The TypeScript source is short (47 lines including JSDoc); the Python port reproduces the algorithm verbatim:

1. If text or entity_name is empty, return "".
2. Lowercase both text and entity_name; find the first index of entity_name.lower() in text.lower().
3. If not found, return "" (PRODUCT Inv-17 reads “the column is NOT left NULL”; the Python port returns empty-string rather than None to match entity_mentions.context_snippet text NULL semantics — empty-string and NULL are both treated as “no usable snippet” by downstream consumers and empty-string is slightly more diagnostic).
4. Compute start = max(0, idx - 80) and end = min(len(text), idx + len(entity_name) + 80).
5. Slice text[start:end].strip(); prepend "..." if start > 0; append "..." if end < len(text).
6. Return the snippet.

Determinism tests verify identical input → identical output and that the Python port’s output matches the TypeScript output byte-for-byte (parity-gated via a small fixture set).

Files touched / created:

• scripts/cocoindex_pipeline/entity_context.py — NEW.

  """Per-mention context snippet extraction.
  
  Python port of lib/entities/entity-context.ts:19 extractEntityContext.
  PRODUCT.md Inv-17: every Stage-5-produced entity_mentions row carries
  a non-NULL context_snippet — the per-item phase calls this function
  inside ingest_file (§P-3) before the declare_row site.
  """
  
  from __future__ import annotations
  
  CONTEXT_RADIUS = 80
  
  
  def extract_entity_context(text: str, entity_name: str) -> str:
      """Return a context snippet showing where entity_name appears in text.
  
      Mirrors lib/entities/entity-context.ts:19. Case-insensitive search;
      returns the first occurrence with ±80 chars of surrounding context.
      Adds "..." ellipsis markers where the snippet was truncated.
  
      Args:
          text:        Full plain text to search in.
          entity_name: Entity name to locate (case-insensitive).
  
      Returns:
          Context snippet string; empty string if entity not found.
      """
      if not text or not entity_name:
          return ""
      lower_text = text.lower()
      lower_entity = entity_name.lower()
      idx = lower_text.find(lower_entity)
      if idx == -1:
          return ""
      start = max(0, idx - CONTEXT_RADIUS)
      end = min(len(text), idx + len(entity_name) + CONTEXT_RADIUS)
      snippet = text[start:end].strip()
      if start > 0:
          snippet = f"...{snippet}"
      if end < len(text):
          snippet = f"{snippet}..."
      return snippet

• scripts/tests/test_entity_context.py — NEW. Unit tests: empty text / empty entity_name; entity-not-found; entity at start of text (no leading ellipsis); entity at end (no trailing ellipsis); entity mid-text (both ellipses); case-insensitive match (entity “ISO 27001” in text “iso 27001”); idempotent under repeated invocation.

Migration shape: None.

Acceptance test shape:

• Inv-17: Same as §P-3 Inv-17 — __tests__/integration/cocoindex/context-snippet-populated.integration.test.ts verifies entity_mentions.context_snippet matches extract_entity_context(content_text, entity_name) for the same inputs.

Effort estimate: ~1 h (0.5 h port + 0.5 h unit + parity tests).

---

§P-6. Stage-5 resolution pass core (_run_stage_5_resolution)

Covers: Inv-3, Inv-5, Inv-10, Inv-11.

Implementation summary. Implement the Stage-5 post-pass body inside scripts/cocoindex_pipeline/stage_5.py (the new module created by §P-1). The function reads the run’s entity_mentions rows (op_id-scoped per Inv-5), applies the legacy entity_aliases map (Inv-10), invokes cocoindex.ops.entity_resolution.resolve_entities with the KH entity embedder (§P-7) + KH PairResolver (§P-8), walks the resolution mapping, and issues op_id-scoped UPDATEs for rows whose canonical_name resolves to a different value. The function returns the count of rows it materially changed; the caller (§P-1) does NOT need this return value at the call-site (Inv-11 is satisfied via the per-row counter bump inside the function — see step 7 below), but the return supports verification + future observability.

Per-row counter bump per PRODUCT Inv-11 (the elevation): flow_stage_counter.increment("entity_resolution") fires ONCE PER UPDATE-eligible row inside the bulk-update loop — NOT once per pass. This matches the established platform pattern at scripts/cocoindex_pipeline/flow.py:915 (the stage_counter.increment("embedding") per-row bump). The flow-end fold-back at flow.py:1132 (stage_counts["embedding"] = flow_stage_counter.get("embedding")) extends naturally — TECH adds the symmetric stage_counts["entity_resolution"] = flow_stage_counter.get("entity_resolution") line at flow.py:1132+ (§P-1 EDIT range).

Function body (the canonical shape — full implementation lands in {53.4+} dispatch):

async def _run_stage_5_resolution(
    *,
    meta: FlowRunMeta,
    db_pool: asyncpg.Pool,
    flow_stage_counter: _FlowStageCounter,
) -> int:
    """Stage-5 post-pass: cross-document canonical resolution.

    PRODUCT.md §2 Area B + Area C + Area D: applies cocoindex
    resolve_entities over the run's entity_mentions, preloading the
    legacy entity_aliases map (Inv-10), and issuing op_id-scoped
    UPDATEs (Inv-5) for rows whose canonical_name resolves to a
    different cross-document value.

    Args:
      meta:                FlowRunMeta carrying the run's op_id (Inv-5 scope).
      db_pool:             asyncpg pool (resolved env-scope via DB_CTX).
      flow_stage_counter:  Per-flow stage counter (bumped per UPDATE per Inv-11).

    Returns:
      Count of entity_mentions rows whose canonical_name Stage-5 changed.
    """
    # Step 1: preload the legacy entity_aliases map (Inv-10).
    alias_map: dict[str, str] = await _preload_entity_aliases(db_pool)

    # Step 2: read the run's entity_mentions rows (op_id-scoped — Inv-5).
    rows = await _select_run_entity_mentions(db_pool, meta.op_id)
    if not rows:
        return 0

    # Step 3: apply alias map to the per-doc canonicals BEFORE resolve_entities.
    # Inv-10: outputs are consistent with legacy entity_aliases reads.
    # Carry the native uuid.UUID for row["id"] — asyncpg is strict on uuid
    # binds, so a str() cast risks a bind-type error on the Step-6 UPDATE.
    name_pairs: list[tuple[UUID, str, str]] = [
        (
            row["id"],
            alias_map.get(row["canonical_name"], row["canonical_name"]),
            row["entity_type"],
        )
        for row in rows
    ]

    # Step 4: invoke cocoindex resolve_entities — ONE CALL PER entity_type
    # group. resolve_entities is collection-level (faiss IP over Iterable[str]),
    # and the KhPairResolver cache key is (name_a, name_b, entity_type) per
    # P-OQ3 — so each entity_type batch gets a fresh resolver instance carrying
    # its entity_type. This also keeps 'Cisco' as organisation independent of
    # 'Cisco' as technology in the cache (the explicit Inv-14 / P-OQ3 rationale).
    # Inv-3: cross-doc canonicalisation; Inv-21 v1 single-workspace scope is
    # naturally honoured because the run's rows come from one workspace.
    from collections import defaultdict
    from cocoindex.ops.entity_resolution import resolve_entities
    from scripts.cocoindex_pipeline.entity_embedder import KhEntityEmbedder
    from scripts.cocoindex_pipeline.pair_resolver import KhPairResolver

    # Group (alias_applied_canonical) by entity_type for per-type batches.
    names_by_type: dict[str, set[str]] = defaultdict(set)
    for _row_id, alias_applied_canonical, entity_type in name_pairs:
        names_by_type[entity_type].add(alias_applied_canonical)

    # resolved_by_type maps entity_type -> ResolvedEntities for that batch.
    resolved_by_type: dict[str, "ResolvedEntities"] = {}
    for entity_type, names in names_by_type.items():
        resolved_by_type[entity_type] = await resolve_entities(
            sorted(names),
            embedder=KhEntityEmbedder(),
            resolve_pair=KhPairResolver(
                db_pool=db_pool, op_id=meta.op_id, entity_type=entity_type,
            ),
        )

    # Step 5: walk ResolvedEntities.canonical_of() — find UPDATE-eligible rows.
    #
    # Inv-20 ("an unresolved mention retains its per-document canonical") is
    # realised by the `new_canonical == alias_applied_canonical` equality
    # branch below, NOT by the `is None` guard. Under cocoindex 1.0.3,
    # `ResolvedEntities.canonical_of(name)` returns a `str` (the name itself
    # when already-canonical) and RAISES KeyError for an unknown name — it
    # NEVER returns None. The `is None` guard is retained purely as
    # forward-compatibility defence against a future cocoindex contract change;
    # a re-implementer MUST NOT treat it as the load-bearing Inv-20 path.
    updates: list[tuple[UUID, str]] = []  # (row_id, new_canonical)
    for row_id, alias_applied_canonical, entity_type in name_pairs:
        resolved = resolved_by_type[entity_type]
        new_canonical = resolved.canonical_of(alias_applied_canonical)
        if new_canonical is None:
            continue  # forward-compat defence only — see note above; not the Inv-20 path.
        if new_canonical == alias_applied_canonical:
            continue  # Inv-20: unresolved retains its per-document canonical — no UPDATE.
        updates.append((row_id, new_canonical))

    # Step 6: op_id-scoped UPDATE batch (Inv-5).
    if updates:
        async with db_pool.acquire() as conn:
            async with conn.transaction():
                for row_id, new_canonical in updates:
                    # WHERE id = $1 AND op_id = $2 — op_id scope is the
                    # forcing function (Inv-5); the row id is naturally
                    # scoped to this run because Step 2 selected by op_id.
                    await conn.execute(
                        "UPDATE public.entity_mentions "
                        "SET canonical_name = $1 "
                        "WHERE id = $2 AND op_id = $3",
                        new_canonical, row_id, meta.op_id,
                    )
                    # Step 7: per-row counter bump (Inv-11 PRODUCT elevation).
                    flow_stage_counter.increment("entity_resolution")

    return len(updates)

The helpers _preload_entity_aliases and _select_run_entity_mentions are private to stage_5.py:

• _preload_entity_aliases(db_pool) -> dict[str, str] — SELECT alias, canonical FROM public.entity_aliases WHERE is_active = true. Returns dict mapping alias → canonical. Cached per-call (no module-level cache — each run gets a fresh snapshot).
• _select_run_entity_mentions(db_pool, op_id) -> list[Record] — SELECT id, canonical_name, entity_type FROM public.entity_mentions WHERE op_id = $1. Returns the rows the per-item phase wrote in this run.

Files touched / created:

• scripts/cocoindex_pipeline/stage_5.py — NEW (skeleton landed by §P-1). Fills out _run_stage_5_resolution + the two helpers.
• scripts/cocoindex_pipeline/flow.py — EDIT. Extend the finally block at flow.py:1132 to fold stage_counts["entity_resolution"] = flow_stage_counter.get("entity_resolution") (mirror of the existing line for "embedding").

Migration shape: None for §P-6 itself; §P-9 lands the op_id migration.

Acceptance test shape:

• Inv-3: __tests__/integration/cocoindex/cross-document-dedup.integration.test.ts — ingest a 2-document corpus where doc A contains “ISO 27001” and doc B contains “ISO27001”; after pipeline_runs.status='completed', both entity_mentions rows show the same canonical_name. The specific canonical (whichever resolve_entities picked) is implementation-determined; the assertion is that they MATCH each other.
• Inv-5: __tests__/integration/cocoindex/op-id-scoping.integration.test.ts — ingest run 1 (op_id A); ingest run 2 (op_id B) with new documents. Assert all run-1 rows retain their run-1 canonical regardless of what run-2’s resolve_entities produced.
• Inv-10: __tests__/integration/cocoindex/legacy-alias-preload.integration.test.ts — seed entity_aliases with ("iso 27001 (uk)", "ISO 27001 (UK)", is_active=true); ingest a corpus that resolves to "iso 27001 (uk)"; assert the Stage-5-produced canonical is "ISO 27001 (UK)".
• Inv-11: __tests__/integration/cocoindex/stage-5-row-counter.integration.test.ts — ingest a corpus where Stage-5 UPDATEs M rows; assert pipeline_runs.result.stage_counts["entity_resolution"] == M. The test seeds the corpus such that M is known (e.g. 5 documents, 3 of which contain near-matches of the same entity → 2 UPDATEs expected after the first row sets the canonical).

Effort estimate: ~3 h (1.5 h core function + helpers; 1.5 h 4 integration tests).

---

§P-7. KH-owned entity embedder (KhEntityEmbedder)

Covers: Inv-3, Inv-14.

Implementation summary. Introduce a KH-owned entity embedder at scripts/cocoindex_pipeline/entity_embedder.py implementing the cocoindex _Embedder Protocol verified empirically in RESEARCH §R1.2 (async def embed(self, text: str) -> NDArray[float32]). The embedder is distinct from the Stage-4 LiteLLMEmbedder (which embeds content_text — long-form document text) because Stage-5 embeds entity-NAME strings — typically short (1–4 tokens) and domain-specific (organisation names, certification labels, regulation IDs).

The TECH ratifies the concrete embedder choice: reuse the same OpenAI text-embedding-3-large model (dimensions=1024 to match the existing Stage-4 vector(1024) dimension; per flow.py:602–603) for consistency with the existing pipeline embedding model + LiteLLM config. The reason is twofold:

1. No new dep: the OpenAI client and LiteLLM are already in requirements.txt; introducing a second embedder model (e.g. a sentence-transformer) would add a model download cost at sidecar cold-start that PRODUCT does not motivate.
2. Bounded cost: entity-name strings are short → embedding cost per call is small even on text-embedding-3-large; cocoindex’s resolve_entities deduplicates via sorted(set(entities)) before embedding (RESEARCH §R1.2 source-read), so the embedding cost is proportional to distinct entity names per run, NOT total mentions.

The embedder is a class implementing the _Embedder Protocol (no inheritance needed because the protocol is @runtime_checkable):

class KhEntityEmbedder:
    """Entity-name embedder for Stage-5 cocoindex.resolve_entities.

    Distinct from Stage-4 content_text LiteLLMEmbedder — embeds entity-name
    strings (short, often domain-specific) rather than document content.
    Uses the same OpenAI text-embedding-3-large model + dimensions=1024
    for consistency; the embedding is per-name not per-document.

    PRODUCT.md Inv-3 (cross-doc canonicalisation) + Inv-14 (PairResolver
    consumer). The cocoindex resolve_entities collection-level mechanic
    deduplicates inputs via sorted(set(entities)) so the embedding budget
    is proportional to distinct entity names per run.
    """

    def __init__(self) -> None:
        from cocoindex.ops.litellm import LiteLLMEmbedder
        self._embedder = LiteLLMEmbedder(
            "text-embedding-3-large",
            dimensions=1024,
        )

    async def embed(self, text: str) -> "NDArray[np.float32]":
        """Embed a single entity-name string (cocoindex _Embedder protocol)."""
        return await self._embedder.embed(text)

Files touched / created:

• scripts/cocoindex_pipeline/entity_embedder.py — NEW (shape above).
• scripts/tests/test_entity_embedder.py — NEW. Unit test: smoke-call against a mock LiteLLMEmbedder; assert the returned vector has length 1024.

Migration shape: None.

Acceptance test shape:

• Inv-3: Covered by §P-6 cross-document-dedup integration test (the embedder is consumed by resolve_entities inside _run_stage_5_resolution).
• Inv-14: Covered by §P-8 (the embedder is one of two collaborators in the cocoindex resolution call; the PairResolver is the other).

Effort estimate: ~1 h (0.5 h class + 0.5 h unit test).

---

§P-8. PairResolver implementation + determinism cache (KhPairResolver)

Covers: Inv-14.

Implementation summary. Introduce a KH-owned PairResolver callable at scripts/cocoindex_pipeline/pair_resolver.py implementing the cocoindex PairResolver Protocol verified empirically:

async def __call__(self, entity: str, candidates: list[str]) -> PairDecision

The resolver makes an Anthropic SDK call (mirroring the existing extract_* patterns in extraction.py) to classify whether entity matches any of the supplied candidates. Each invocation hits the new entity_pair_resolutions table (§P-9) FIRST to check for a cached decision; on cache miss, invokes the LLM and writes the decision back. Cache keying is (name_a, name_b, entity_type) — name_a is the lexicographically smaller of (entity, candidate) and name_b is the larger, so (a, b) and (b, a) resolve to the same cache row. The cocoindex protocol passes a LIST of candidates per call; the resolver iterates the candidates and returns either a PairDecision(matched=<candidate>, canonical=<canonical>) for the first match OR PairDecision(matched=None, canonical=None) if no candidate matches. The canonical field is typed CanonicalSide (StrEnum members NEW='new' and MATCHED='matched'), not a free string; cocoindex’s _new_wins decision consumer evaluates decision.canonical == CanonicalSide.NEW to decide which side of the pair wins. Passing a raw string silently fails the comparison and demotes entity unconditionally, breaking Inv-3 at Stage-5 runtime — see §7 OQ-3 verification row for the corrected type annotation.

Implementation:

class KhPairResolver:
    """LLM-backed PairResolver with persistent determinism cache.

    PRODUCT.md Inv-14: decisions persist across runs via
    entity_pair_resolutions (§P-9 cache table). Re-running Stage-5 on the
    same input produces the same canonical mapping byte-for-byte (cache
    hits replay prior LLM decisions without re-invoking the model).
    """

    def __init__(self, *, db_pool: asyncpg.Pool, op_id: UUID, entity_type: str) -> None:
        # Per Inv-14 + P-OQ3 cache-keying rationale: one KhPairResolver instance
        # per (run × entity_type) batch. Stage-5 instantiates a fresh resolver
        # for each entity_type group it submits to resolve_entities, so the
        # entity_type passed to the cache lookup / INSERT below corresponds to
        # the batch this resolver was constructed for (e.g. 'Cisco' as
        # organisation vs technology resolve to independent cache rows).
        self._pool = db_pool
        self._op_id = op_id
        self._entity_type = entity_type

    async def __call__(
        self,
        entity: str,
        candidates: list[str],
    ) -> "PairDecision":
        from cocoindex.ops.entity_resolution import PairDecision, CanonicalSide
        for candidate in candidates:
            decision = await self._resolve_one_pair(entity, candidate)
            if decision == "same":
                # Canonical preference: the longer / more-disambiguating name wins.
                # `PairDecision.canonical` is CanonicalSide (StrEnum: NEW='new',
                # MATCHED='matched'); cocoindex's `_new_wins` consumer evaluates
                # `decision.canonical == CanonicalSide.NEW` — passing a raw string
                # silently demotes the entity. Map longer-name-wins to the enum:
                # entity is "NEW" (positional first arg); candidate is "MATCHED".
                canonical = (
                    CanonicalSide.NEW
                    if len(entity) >= len(candidate)
                    else CanonicalSide.MATCHED
                )
                return PairDecision(matched=candidate, canonical=canonical)
        return PairDecision(matched=None, canonical=None)

    async def _resolve_one_pair(self, name_a: str, name_b: str) -> str:
        # Cache lookup: lexicographic ordering ensures (a,b) and (b,a) hit
        # the same row. entity_type from __init__ scopes the key.
        key_a, key_b = sorted((name_a, name_b))
        async with self._pool.acquire() as conn:
            cached = await conn.fetchval(
                "SELECT decision FROM public.entity_pair_resolutions "
                "WHERE name_a = $1 AND name_b = $2 AND entity_type = $3",
                key_a, key_b, self._entity_type,
            )
            if cached is not None:
                return cached
        # Cache miss: invoke the LLM.
        decision = await self._invoke_llm(key_a, key_b)
        # Write back to cache (UPSERT for race-safety across concurrent runs).
        async with self._pool.acquire() as conn:
            await conn.execute(
                "INSERT INTO public.entity_pair_resolutions "
                "(name_a, name_b, entity_type, decision, op_id) "
                "VALUES ($1, $2, $3, $4, $5) "
                "ON CONFLICT (name_a, name_b, entity_type) DO NOTHING",
                key_a, key_b, self._entity_type, decision, self._op_id,
            )
        return decision

    async def _invoke_llm(self, name_a: str, name_b: str) -> str:
        # Returns "same" or "different" — LLM call shape mirrors
        # extraction.py's extract_classification (Anthropic SDK,
        # _anthropic_retry tenacity wrapper, ANTHROPIC_MODEL).
        # Prompt: "Are these two entity names referring to the same
        # real-world thing? Reply with exactly 'same' or 'different'.
        # Name A: {name_a}\nName B: {name_b}"
        # Full body owned by {53.4+} dispatch.
        ...

The determinism guarantee (Inv-14) comes from two layers: (a) (name_a, name_b) cache lookups make the second run’s PairResolver invocations a cache-hit replay; (b) the cache row carries op_id of the FIRST run that wrote the decision, so audit-forensics can trace which run originated each pair decision.

The TECH does NOT spec the exact LLM prompt + temperature — that’s a {53.4+} implementation slice. PRODUCT Inv-14 only requires (a) the resolver exists, (b) decisions are cached, (c) re-runs replay from cache. Temperature should be 0 for further determinism (LLMs at temperature=0 are still not strictly deterministic across model versions, but they minimise stochasticity within a fixed version).

Files touched / created:

• scripts/cocoindex_pipeline/pair_resolver.py — NEW (shape above).
• scripts/tests/test_pair_resolver.py — NEW. Unit tests: cache-hit path (no LLM invocation); cache-miss path (LLM invoked once, then written back); concurrent runs racing on the same (name_a, name_b) are race-safe via ON CONFLICT DO NOTHING.

Migration shape: §P-9 commits the entity_pair_resolutions table in the same migration as the entity_mentions op_id ALTER (single CLI migration covering both).

Acceptance test shape:

• Inv-14: __tests__/integration/cocoindex/pair-resolver-determinism.integration.test.ts — ingest a corpus triggering PairResolver decisions in run 1; trigger full_reprocess=True for run 2; assert the resulting entity_mentions.canonical_name values match run 1 byte-for-byte AND entity_pair_resolutions row count is unchanged between runs (cache hits on run 2; no new rows).

Effort estimate: ~2 h (1 h resolver class + 1 h unit + integration tests).

---

§P-9. op_id migration on entity_mentions + entity_pair_resolutions cache table

Covers: Inv-6, Inv-7 (op_id substrate), Inv-14 (cache table).

Implementation summary. One CLI migration adds: (a) op_id uuid NULL + B-tree partial index on entity_mentions per RESEARCH §R6.2 verbatim; (b) the new entity_pair_resolutions cache table per P-OQ3 resolution (§4). Both ALTER + CREATE statements are IF NOT EXISTS-guarded for idempotent re-apply. DDL via Supabase CLI (supabase migration new + db push) per CLAUDE.md gotcha — NEVER via MCP execute_sql / mcp__supabase__apply_migration. Pre-push: cat supabase/.temp/project-ref to verify env; relink via supabase link --project-ref <correct> if drift detected.

Files touched / created:

• supabase/migrations/<timestamp>_id53_entity_mentions_op_id_and_pair_cache.sql — NEW. Combined ADD COLUMN + CREATE TABLE migration. Timestamp filled in at impl time via supabase migration new id53_entity_mentions_op_id_and_pair_cache.

  -- ID-53 — Stage-5 entity-resolution: op_id on entity_mentions + PairResolver cache.
  -- Spec: docs/specs/id-53-stage-5-entity-resolution/TECH.md §P-9.
  -- Ratification:
  --   - PRODUCT.md Inv-6 / Inv-7 — op_id round-trip + memo-respecting semantic.
  --   - PRODUCT.md Inv-14 + §5 P-OQ3 — entity_pair_resolutions cache table shape.
  --   - RESEARCH.md §R6.2 — op_id migration SQL (verbatim).
  --   - S273 OQ-2 — entity_mentions op_id gap (folded into ID-53 per Liam ruling).
  -- Mirrors T8 pattern at supabase/migrations/20260521203414_t8_op_id_propagation.sql.
  -- Idempotency: IF NOT EXISTS guards so re-apply is no-op.
  -- DDL via Supabase CLI ONLY (supabase migration new + db push); NEVER MCP
  -- execute_sql / apply_migration (CLAUDE.md gotcha "DDL via CLI only").
  
  SET search_path = public, extensions;
  
  -- 1. entity_mentions.op_id (S273 OQ-2 ruling; ID-53 spec rescope).
  --    NULL-default; existing rows stay NULL until backfilled OR re-ingested.
  --    Stage-5 post-pass (§P-6) UPDATEs are op_id-scoped — NULL-op_id rows
  --    are READ for resolution input but NEVER UPDATED (Inv-5).
  ALTER TABLE public.entity_mentions
    ADD COLUMN IF NOT EXISTS op_id uuid NULL;
  CREATE INDEX IF NOT EXISTS idx_entity_mentions_op_id
    ON public.entity_mentions (op_id) WHERE op_id IS NOT NULL;
  
  COMMENT ON COLUMN public.entity_mentions.op_id IS
    'KH-generated per-run op_id, written as a declare_row field at UPSERT '
    'time per N7 hybrid (02-data-flow.md §5). Round-trip: pipeline_runs.op_id. '
    'Required for Stage-5 op_id-scoped UPDATEs per PRODUCT.md Inv-5. '
    'Mirrors T8 pattern at 20260521203414_t8_op_id_propagation.sql.';
  
  -- 2. entity_pair_resolutions — PairResolver determinism cache (Inv-14).
  --    P-OQ3 resolution: new table (NOT extending entity_aliases, NOT in-memory).
  --    name_a / name_b lexicographically ordered by the resolver at insert
  --    time so (a,b) and (b,a) hit the same row. UNIQUE constraint on the
  --    pair triple is the cache key; op_id audits which run originated the
  --    decision.
  CREATE TABLE IF NOT EXISTS public.entity_pair_resolutions (
      id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
      name_a        text NOT NULL,
      name_b        text NOT NULL,
      entity_type   text NOT NULL,
      decision      text NOT NULL CHECK (decision IN ('same', 'different')),
      resolved_at   timestamptz NOT NULL DEFAULT now(),
      op_id         uuid NULL,
      CONSTRAINT entity_pair_resolutions_pair_unique UNIQUE (name_a, name_b, entity_type)
  );
  
  CREATE INDEX IF NOT EXISTS idx_entity_pair_resolutions_op_id
    ON public.entity_pair_resolutions (op_id) WHERE op_id IS NOT NULL;
  
  COMMENT ON TABLE public.entity_pair_resolutions IS
    'PairResolver determinism cache for Stage-5 cocoindex.resolve_entities. '
    'PRODUCT.md Inv-14 + §5 P-OQ3. Lexicographic ordering of (name_a, name_b) '
    'at insert time ensures cache-key stability; UNIQUE constraint backs the '
    'cache lookup. op_id records the originating run for audit-forensics.';
  
  COMMENT ON COLUMN public.entity_pair_resolutions.name_a IS
    'Lexicographically smaller of the (entity, candidate) pair at insert time.';
  COMMENT ON COLUMN public.entity_pair_resolutions.name_b IS
    'Lexicographically larger of the (entity, candidate) pair at insert time.';
  COMMENT ON COLUMN public.entity_pair_resolutions.decision IS
    'Resolver decision: same | different. Checked at LOAD time by KhPairResolver.';
  COMMENT ON COLUMN public.entity_pair_resolutions.op_id IS
    'op_id of the run that originated this decision (NULL if backfilled).';

No new PL/pgSQL functions in this migration — no SET search_path = public, extensions body, no anon-EXECUTE REVOKE pattern. If a future v1.1 follow-up adds e.g. an entity_pair_resolution_emit() helper, that migration MUST apply the canonical REVOKE FROM PUBLIC + REVOKE FROM anon + GRANT TO authenticated, service_role pattern per CLAUDE.md.

Migration apply procedure:

1. Pre-push: cat supabase/.temp/project-ref against expected (turayklvaunphgbgscat staging / rovrymhhffssilaftdwd prod).
2. Apply via /opt/homebrew/bin/supabase db push with dangerouslyDisableSandbox: true.
3. Post-apply: type regen via /opt/homebrew/bin/supabase gen types typescript --project-id rovrymhhffssilaftdwd --schema public > supabase/types/database.types.ts and commit.
4. Re-run the column query SELECT column_name FROM information_schema.columns WHERE table_schema='public' AND column_name='op_id' AND table_name='entity_mentions' to verify.

Migration shape: Single CLI migration. ADD COLUMN-only on entity_mentions (no rewrite cost). CREATE TABLE for entity_pair_resolutions (new table; no existing rows). Idempotent (IF NOT EXISTS on every ALTER + CREATE).

Acceptance test shape:

• Inv-6: __tests__/integration/cocoindex/op-id-round-trip.integration.test.ts — given any pipeline-produced entity_mentions.op_id value, SELECT * FROM pipeline_runs WHERE op_id = <value> returns exactly one row (mirrors peer T8 PRODUCT.md Inv-12 verification).
• Inv-7: Covered by §P-3 Inv-7 integration test.
• Inv-14: Covered by §P-8 Inv-14 integration test (cache hit on second run).

Effort estimate: ~1 h (0.5 h migration authoring + apply + 0.5 h type regen + post-apply verification).

---

§P-10. Failure mode wiring

Covers: Inv-12, Inv-13.

Implementation summary. The _run_stage_5_resolution body (§P-6) is wrapped at the call-site (flow.py:1083+ per §P-1) inside the existing outer try block at flow.py:1006. On any exception raised inside the Stage-5 post-pass, control flows to the existing except Exception as exc clause at flow.py:1089, which:

1. Sets flow_status = "failed".

2. Sets flow_error_message = str(exc).

3. Calls _classify_stage_exception(exc) at flow.py:1097 — TECH adds NO new branch to this function; the existing classifier returns None for unmapped types and routes through the unclassified fallback at flow.py:1102 (logs cocoindex.stage_error.unclassified with the exception type+module). The intent of Inv-12 is non-destructiveness, NOT precise stage-classification of the exception class — that’s a separate enhancement.

   Note for {53.4+} implementation: the _classify_stage_exception body at flow.py:200–204 already documents the future-branch reservation (“Embedding-stage + entity-resolution-stage classes … are NOT auto-classified at v1: their upstream exception types … are not yet wired in flow.py”). The TECH does not commit a new branch because the actual exception types resolve_entities raises (and that the KhPairResolver raises) are not yet known empirically. A {53.4+} dispatch can add an entity_resolution_failed branch keyed on the resolver / embedder module prefix once the impl reveals the surface.

4. Calls _emit_stage_error_log(stage="flow", error_class=flow_error_class, ...) at flow.py:1116.

5. Re-raises via raise at flow.py:1123 (the outer try’s re-raise contract).

6. The finally block at flow.py:1124 runs unconditionally; per §P-6 the stage_counts["entity_resolution"] fold-back at flow.py:1132+ lands the per-row counter value (zero if Stage-5 never bumped it, the actual count if Stage-5 partially completed before raising).

7. The flow-end _emit_pipeline_run_webhook call at flow.py:1135 carries status="failed", errorClass=flow_error_class, populated stageCounts including "entity_resolution".

Inv-12 (non-destructiveness): the per-item phase rows have ALREADY been committed by handle.ready() returning at flow.py:1069. Stage-5’s UPDATE statements are op_id-scoped to the in-flight run; a Stage-5 failure mid-pass leaves any UPDATEs that did fire in place (the per-mention canonical_name landed) and any UPDATEs that did NOT fire still carry the per-document default canonical_name from the per-item phase. NO row deletion, NO row insertion-rollback; the per-item rows survive intact.

Inv-13 (webhook surfaces Stage-5 outcome): the existing payload at flow.py:510–516 already carries stage_counts as a top-level dict; once §P-6 folds entity_resolution back at line 1132+, the existing _emit_pipeline_run_webhook call surfaces the value automatically. No new payload field. The error_class for a Stage-5 failure routes through the existing errorClass field; the structured log emission via _emit_stage_error_log is already wired.

Files touched / created:

• scripts/cocoindex_pipeline/flow.py — EDIT. The §P-1 + §P-6 edits cover this; no additional change here. (This §P-10 documents the failure-mode behaviour the existing substrate already provides; no new code.)

Migration shape: None.

Acceptance test shape:

• Inv-12: __tests__/integration/cocoindex/stage-5-failure-non-destructive.integration.test.ts — inject a deterministic failure into _run_stage_5_resolution (mock resolve_entities to raise); after the run terminates as status='failed', query entity_mentions for the rows the per-item phase wrote — they retain their per-document canonical_name and their op_id value (Inv-7). The number of rows is exactly the per-item write count; no rows deleted by the failure.
• Inv-13: Same integration test extended — assert the captured flow-end webhook payload carries errorClass = "entity_resolution_failed" (or whichever class the _classify_stage_exception fallback yields for the injected exception class) AND stageCounts.entity_resolution is present (zero if the failure was at preload step; positive if partial updates fired before the raise).

Effort estimate: ~1.5 h (1 h failure-injection harness + 0.5 h 2 integration tests).

---

§P-11. App-side coexistence guarantees (Inv-8, Inv-9)

Covers: Inv-8, Inv-9.

Implementation summary. No application-side code changes are required to satisfy Inv-8 and Inv-9 — they fall out as a natural consequence of Inv-5’s op_id scoping (§P-6 step 6). The TECH documents the coexistence guarantee with cross-references to the five app-side write paths so future maintenance has the load-bearing call sites in one place.

Five app-side write paths (from RESEARCH §R4.1):

Path	File:line	Operation	Stage-5 collision behaviour
classifyContent re-classify	lib/ai/classify.ts:1543–1546	DELETE FROM entity_mentions WHERE content_item_id = $1	INSERTs at lib/ai/classify.ts:1751 carry NULL op_id (no in-flight pipeline run); Stage-5 NEVER UPDATEs NULL-op_id rows (Inv-5). Race-free.
classifyContent INSERT	lib/ai/classify.ts:1751	INSERT INTO entity_mentions (...) (no op_id field — column is nullable per §P-9 migration)	NULL op_id; never UPDATEd by Stage-5.
Admin merge	app/api/entities/merge/route.ts:48–52	atomic merge_entities RPC → UPDATE on canonical_name + entity_type_override	Operates on rows from ARBITRARY HISTORICAL runs; Stage-5 only UPDATEs rows from THE IN-FLIGHT run’s op_id (Inv-5). Admin merges on prior runs are NEVER reverted.
Admin split	app/api/entities/split/route.ts:51,78	.update({canonical_name: newCanonical})	Same as merge — operates on historical-run rows; op_id-scope protects them.
Admin type-override	app/api/entities/[canonical_name]/type/route.ts:48	.update({entity_type_override})	Same — Stage-5 NEVER touches entity_type_override; only canonical_name. Admin type-override is independent.
Admin metadata-edit	app/api/entities/[canonical_name]/metadata/route.ts:57,74,94	.update({metadata: ...})	Same — Stage-5 NEVER touches metadata directly; only canonical_name. (The per-item phase WRITES metadata.source_span_start / metadata.source_span_end at row insert time per §P-3 Inv-16; Stage-5 does NOT modify metadata post-insert.)
Governance publish-from-draft	lib/mcp/tools/governance.ts:472–499	calls classifyContent (delete-before-insert)	Same NULL-op_id semantics as classifyContent direct invocation.

P-OQ2 (narrow race window): the narrow race window is between an admin RPC operating on rows whose op_id matches the IN-FLIGHT pipeline run’s op_id (i.e. rows the pipeline JUST wrote and Stage-5 is about to UPDATE) and Stage-5’s UPDATE pass on those same rows. TECH RATIFIES the same-window-as-per-item-declare_row semantic from RESEARCH §R4.1: this race window is identical in shape to what declare_row already has against admin RPCs, so the post-pass does NOT expand the failure surface. No new locking, no advisory-lock infrastructure, no transactional wrapping — acceptable v1 risk per PRODUCT §5 P-OQ2 recommended default.

Files touched / created: None (this Proposed change documents existing substrate; no code modification).

Migration shape: None.

Acceptance test shape:

• Inv-8: __tests__/integration/cocoindex/classify-content-coexistence.integration.test.ts — while a pipeline run is in flight (or just after handle.ready() but before Stage-5 fires), invoke classifyContent on a content_item that is NOT one of the pipeline’s documents; assert classifyContent completes successfully AND the resulting entity_mentions rows have NULL op_id AND Stage-5’s UPDATE pass does NOT touch them.
• Inv-9: __tests__/integration/cocoindex/admin-merge-coexistence.integration.test.ts — complete a pipeline run with op_id A; admin-merge two rows from op_id A’s output (“ISO 27001” + “iso 27001” → “iso 27001”); ingest a second corpus at op_id B (new documents only); after run B completes, assert the run-A merged rows retain their admin-merged canonical_name (Stage-5 of run B did NOT revert them because their op_id != B).

Effort estimate: ~1.5 h (no code; 1.5 h 2 integration tests + race-window harness).

---

§P-12. faiss-cpu pin in requirements.txt

Covers: Inv-18.

Implementation summary. EDIT requirements.txt to append faiss-cpu==1.14.2 (verified ABSENT at HEAD 80fbbd5e per RESEARCH §R5.4 — confirmed via grep -E "^faiss" requirements.txt returning zero matches; the cocoindex / anthropic / asyncpg / aiohttp / pydantic pins are already present and unchanged). The exact-pin discipline matches the cocoindex / anthropic / pydantic pins already in requirements.txt; range-pinning (>=1.14.2,<2.0) is REJECTED to avoid silent upgrades that could break the cocoindex IndexFlatIP shape used inside resolve_entities.

Cloud Run sidecar pre-warm. PRODUCT Inv-18 second clause says “the Cloud Run sidecar build installs faiss-cpu in the pre-warm step so the cold-start budget is not blown by a runtime pip-install”. TECH flags this for the cloudrun/cloudbuild-cocoindex.yaml build config (which lives under T8’s §P-1 surface, NOT this TECH’s primary scope). The pin in requirements.txt (the primary source of truth) is sufficient for the build to pull faiss-cpu into the layer at image build time IF the buildpack reads requirements.txt (which the Python buildpack does by default per cloudrun/cloudbuild-cocoindex.yaml). A {53.4+} Subtask MAY also verify the pre-warm at deploy by running a smoke probe python3 -c "import faiss; print(faiss.__version__)" against the deployed sidecar; this is a verification step, not an additional code change.

Files touched / created:

• requirements.txt — EDIT. Append one line: faiss-cpu==1.14.2. Position alphabetically; cocoindex line at line 38 per current shape; add faiss-cpu==1.14.2 immediately after.

Migration shape: None.

Acceptance test shape:

• Inv-18: __tests__/integration/cocoindex/faiss-pin.integration.test.ts — python3 -c "import faiss; assert faiss.__version__ == '1.14.2'; print(faiss.IndexFlatIP)" invocation against the deployed sidecar; assert exit code 0 + correct version + IndexFlatIP accessible. (For local-dev gating: grep -E "^faiss-cpu==1.14.2$" requirements.txt returns one match.)

Effort estimate: ~0.5 h (edit + smoke-test the install).

---

§P-13. T12 hard-gate documentation

Covers: Inv-19.

Implementation summary. TECH does NOT touch T12 (UNPROMOTED per PRODUCT Inv-19); the gate is documentation-only. At {53.4+} implementation close (i.e. when ID-53 reaches done status AND a tier-1 observation window has passed without Stage-5 regressions), the Workflow Curator (NOT this Planner) promotes T12 and amends docs/themes/canonical-pipeline/reference/canonical-pipeline-sequencing.md §3 to reflect the promotion. TECH commits a one-line spec change to the same doc at impl close:

• docs/themes/canonical-pipeline/reference/canonical-pipeline-sequencing.md §3 T12 row — append “Hard-gated on ID-53 done + tier-1 observation window (PRODUCT.md Inv-19, ratified S274)” to the existing T12 row’s status column.

The actual amendment lands at a {53.4+} ledger-close Subtask (not authored by this TECH’s primary set), so this §P-13 is a placeholder that the PLAN.md decomposition can route to a specific Subtask.

Files touched / created:

• docs/themes/canonical-pipeline/reference/canonical-pipeline-sequencing.md — EDIT (at impl close, NOT during {53.3} TECH authoring). One line append.

Migration shape: None.

Acceptance test shape:

• Inv-19: grep -E "Hard-gated on ID-53" docs/themes/canonical-pipeline/reference/canonical-pipeline-sequencing.md returns one match after impl close. (No automated integration test; the assertion IS the spec amendment landing.)

Effort estimate: ~0.25 h (one-line documentation edit at impl close).

---

§P-14. Corner cases (Inv-20, Inv-21)

Covers: Inv-20, Inv-21.

Implementation summary. Both corner cases are naturally handled by §P-6’s resolution-walk loop (step 5) without additional code:

• Inv-20 (unresolved mentions retain per-doc canonical): §P-6 step 5 explicitly skips rows where resolved.canonical_of(name) returns None (if new_canonical is None: continue). For those rows, no UPDATE is issued; the per-document default canonical (Inv-4) stays in place; the flow_stage_counter is NOT bumped for that row. This matches PRODUCT Inv-20 verbatim.

• Inv-21 (single-workspace scoping): entity_mentions does NOT carry a workspace_id column (verified via grep workspace_id supabase/types/database.types.ts; the FK chain runs entity_mentions.content_item_id → content_items.id → content_item_workspaces.workspace_id, and source_documents.workspace_id is the canonical workspace anchor for the document the mention came from). The Stage-1 source walk (localfs.walk_dir) operates on a single COCOINDEX_SOURCE_PATH per pipeline invocation, which by convention contains documents belonging to one workspace. Therefore the run’s entity_mentions rows (selected by op_id in §P-6 step 2) are naturally workspace-scoped — Stage-5 NEVER reads rows from a different workspace because no rows from a different workspace share the in-flight run’s op_id. PRODUCT Inv-21 single-workspace scoping holds without additional filtering.

  Note for future cross-workspace work (out of v1 scope): if a future spec lifts the single-workspace constraint, the Stage-5 query would need to add a join entity_mentions → content_items → source_documents.workspace_id and either scope-by-workspace OR pass a workspace_id parameter. Out of v1 scope per PRODUCT §3.

Files touched / created: None.

Migration shape: None.

Acceptance test shape:

• Inv-20: __tests__/integration/cocoindex/unresolved-mention-retains-canonical.integration.test.ts — ingest a corpus containing a unique entity name with no near-matches; assert the row’s canonical_name equals the per-document default (the §P-2 output) AND pipeline_runs.result.stage_counts["entity_resolution"] does NOT include this row in its delta.
• Inv-21: __tests__/integration/cocoindex/cross-workspace-isolation.integration.test.ts — at v1 the test seeds two workspaces with overlapping entity names (W1: “ISO 27001”; W2: “ISO27001”); runs the pipeline twice (once with W1’s source path, once with W2’s); asserts the resulting entity_mentions rows in W1 do NOT share canonical_name with W2’s rows (Stage-5 of each run operates on its own entity_mentions subset).

Effort estimate: ~1 h (no code; 2 integration tests).

---

§3. Coverage matrix (PRODUCT inv → Proposed change → acceptance test)

PRODUCT Inv	Area	Proposed change(s)	Acceptance test file
Inv-1	A — Architecture	§P-1	stage-5-attach-point.integration.test.ts
Inv-2	A — Architecture	§P-1 (P-OQ1 documentation)	Documented at impl close; no integration test (§P-13)
Inv-3	B — Freshness	§P-6, §P-7	cross-document-dedup.integration.test.ts
Inv-4	B — Freshness	§P-2, §P-3	per-doc-canonicalisation.integration.test.ts
Inv-5	C — op_id scope	§P-6 step 6	op-id-scoping.integration.test.ts
Inv-6	C — op_id scope	§P-4, §P-9 (migration)	op-id-round-trip.integration.test.ts
Inv-7	C — op_id scope	§P-3 (declare_row), §P-9 (migration)	stage-5-op-id-memo.integration.test.ts
Inv-8	D — Coexistence	§P-11 (no code; doc + test)	classify-content-coexistence.integration.test.ts
Inv-9	D — Coexistence	§P-11 (no code; doc + test)	admin-merge-coexistence.integration.test.ts
Inv-10	D — Coexistence	§P-6 step 1 + step 3 (alias preload)	legacy-alias-preload.integration.test.ts
Inv-11	E — Observability	§P-6 step 7 (per-row counter)	stage-5-row-counter.integration.test.ts
Inv-12	E — Observability	§P-10 (failure-mode wiring)	stage-5-failure-non-destructive.integration.test.ts
Inv-13	E — Observability	§P-10 (webhook payload via existing substrate)	Same as Inv-12 (extended assertions)
Inv-14	F — Determinism	§P-8 (resolver + cache), §P-9 (migration)	pair-resolver-determinism.integration.test.ts
Inv-15	G — Schema parity	§P-3 (map-at-declare-row)	Extended per-doc-canonicalisation test
Inv-16	G — Schema parity	§P-3 (metadata jsonb stash)	Extended per-doc-canonicalisation test
Inv-17	G — Schema parity	§P-3 + §P-5 (Python port)	context-snippet-populated.integration.test.ts
Inv-18	H — Substrate	§P-12 (requirements.txt)	faiss-pin.integration.test.ts
Inv-19	H — Substrate	§P-13 (doc only)	Grep guard at impl close
Inv-20	I — Corner cases	§P-6 step 5 (None-skip) + §P-14	unresolved-mention-retains-canonical.integration.test.ts
Inv-21	I — Corner cases	§P-14 (natural per-run workspace scope)	cross-workspace-isolation.integration.test.ts

Coverage: 21/21 PRODUCT invariants mapped to at least one Proposed change AND at least one acceptance test (or documented at impl close where the invariant is documentation-only per Inv-2 / Inv-19).

---

§4. P-OQ resolutions

PRODUCT.md {53.2} §5 left 3 P-OQs open for TECH ratification. TECH dispositions:

ID	OQ summary	TECH disposition
P-OQ1	Is the managed_by=USER row-only → row+update contract break (Inv-2) PERMANENT or TEMPORARY (revisit when cocoindex publishes a cross-doc reactive write API)?	RATIFIED PERMANENT. KH owns the row+update hybrid as the stable architectural pattern. The platform implications (RESEARCH §R4) are worked through; the mitigations (op_id-scoped UPDATEs in §P-6; legacy alias preload in §P-6 step 1; PairResolver determinism cache in §P-8; T12 sequencing gate in §P-13) are all in place. The §P-13 documentation amendment at impl close commits this stance to canonical-pipeline-sequencing.md §2. No re-evaluation TODO; if cocoindex eventually ships a cross-doc reactive write API (no roadmap as of 28/05/2026), revisit via spec amendment at that point.
P-OQ2	Narrow race window between admin RPCs and Stage-5 UPDATE pass on rows from the in-flight op_id.	DOCUMENTED + ACCEPTED. §P-11 documents the race window in detail and ratifies the same-window-as-per-item-declare_row semantic from RESEARCH §R4.1: the race window is identical in shape and surface area to what declare_row already has against admin RPCs, so the post-pass does NOT expand the failure surface. No new locking, no advisory-lock infrastructure, no transactional wrapping. §P-11 includes an integration test (admin-merge-coexistence) that exercises a deliberate concurrent-write scenario to ratify the existing race semantics for the post-pass surface. Acceptable v1 risk.
P-OQ3	Where does the PairResolver determinism cache (Inv-14) live?	RATIFIED via NEW entity_pair_resolutions table. §P-8 implements the cache as a new table per PRODUCT P-OQ3 recommended default (option (b)). §P-9 commits the table DDL in the same migration as the entity_mentions op_id ALTER. Schema: `(id uuid PK, name_a text, name_b text, entity_type text, decision text CHECK (‘same'

Deviation from PRODUCT P-OQ3 recommended schema (minor): PRODUCT §5 P-OQ3 recommended (name_a, name_b, decision, model_version, created_at, PRIMARY KEY (name_a, name_b)). TECH adopts (id uuid PK, name_a, name_b, entity_type, decision, resolved_at, op_id, UNIQUE (name_a, name_b, entity_type)) — four deltas:

1. Surrogate id PK + UNIQUE on the pair: standard KH-table convention (UUID PK for join compatibility); the UNIQUE constraint on (name_a, name_b) IS the cache key and is functionally equivalent to a composite PK.
2. entity_type field added: allows pair decisions to be entity_type-scoped (in case the same name string resolves differently across types, e.g. “Cisco” as organisation vs technology).
3. model_version dropped: model versioning at v1 is single (claude-opus-4-x); if future versions surface, add the column then. v1 cost reduction.
4. op_id added: audit-forensics — which run originated each decision.

These changes are non-breaking for the cache lookup (UNIQUE on (name_a, name_b) is the lookup key the resolver uses); they improve auditability without expanding the schema’s load surface.

---

§5. Open questions for Liam ratification before {53.4} PLAN dispatch

ID	OQ	Surfaced in	Recommended default for PLAN dispatch	Cost of deferral
T-OQ1	The _classify_stage_exception function at flow.py:182 does NOT yet have an entity_resolution_failed branch keyed on the specific exception types resolve_entities / KhPairResolver / KhEntityEmbedder raise. The TECH (§P-10) flags this as future enhancement; the flow.py:200–204 docstring already names this gap. At PLAN dispatch, do we (a) commit a {53.4+} Subtask to add the branches once impl reveals the exception types, OR (b) accept the unclassified fallback at v1 + revisit if observed in-prod?	§P-10.	Option (a) — add a {53.4+} Subtask. The fallback works at v1 (the structured-log line carries the exception type+module + the pipeline_runs.error_class records the Python class name), so v1 is shippable. But a deliberate branch reduces operator confusion when triaging entity_resolution failures in prod — without it, every entity_resolution failure surfaces as errorClass: <PythonClassName> rather than the canonical entity_resolution_failed. PLAN.md should include a small Subtask (~0.5 h) to add the branch once impl reveals the surface.	LOW — fallback works; the Subtask is small.
T-OQ2	KhPairResolver._invoke_llm body is not specified beyond “Anthropic SDK + prompt + temperature=0”. The exact prompt, the max_tokens, the JSON-or-text response shape, and the error-recovery path for malformed LLM responses are all impl decisions. At PLAN dispatch, does the prompt + response shape come from (a) a new T-PROMPT analogue to T8’s extraction.py prompts (full Pydantic + TypeAdapter shape), OR (b) a thin string-parser (“same"	"different” only, no Pydantic)?	§P-8.	Option (b) — thin string-parser at v1. The decision is binary; full Pydantic validation adds cost without value. The KhPairResolver implementation expects a 2-token response and falls back to “different” on any parse failure (defensive); this matches the binary-decision contract without adding a new Pydantic class. PLAN.md should commit one Subtask to author the prompt + parser; if future requirements surface (e.g. a confidence score), revisit.
T-OQ3	The entity_pair_resolutions cache’s pollution semantics over time. Over many runs, the cache will accumulate rows for every distinct name-pair Stage-5 has ever processed. Is there a v1.1 follow-up to add TTL eviction (e.g. rows older than 90 days that have not been hit in 30 days), OR is unbounded growth acceptable at v1 (the table is narrow — ~5 text fields per row — so even 100k rows is small)?	§P-8 + §P-9.	Accept unbounded growth at v1. The table is narrow; 100k pair-decisions across 1 year of operation at a moderate ingest rate (5k entities per month, ~5% pair-decision rate post-faiss-filter) is roughly 30k rows, well under any operational concern. A v1.1 follow-up can add TTL eviction if cache size becomes a load issue (unlikely). PLAN.md should NOT include a TTL-eviction Subtask at v1.	LOW — table size is bounded by entity-name diversity, not by run count.

---

§6. Effort estimate + PLAN decomposition flag

Proposed change	Effort
§P-1 Stage-5 post-pass attach at app_main	~1.5 h
§P-2 Per-document deterministic canonicalisation function	~1.5 h
§P-3 entity_mentions declare_row in ingest_file	~1.5 h
§P-4 entity_mentions TableTarget mount	~1 h
§P-5 Python port of extractEntityContext	~1 h
§P-6 Stage-5 resolution pass core	~3 h
§P-7 KH-owned entity embedder	~1 h
§P-8 PairResolver + determinism cache	~2 h
§P-9 op_id migration + entity_pair_resolutions cache table	~1 h
§P-10 Failure mode wiring	~1.5 h
§P-11 App-side coexistence (doc + tests)	~1.5 h
§P-12 faiss-cpu pin	~0.5 h
§P-13 T12 hard-gate documentation (at impl close)	~0.25 h
§P-14 Corner-case integration tests	~1 h
Total	~18.25 h

PLAN.md decomposition: REQUIRED. Total effort (~18.25 h) substantially exceeds the 2 h single-slice threshold; the work spans multiple migrations (one CLI migration covering both op_id + cache table), multiple new modules (canonicalisation.py, entity_context.py, stage_5.py, entity_embedder.py, pair_resolver.py), and chain-dependent slices (§P-3 depends on §P-2 + §P-4 + §P-5; §P-6 depends on §P-7 + §P-8 + §P-9). PLAN.md should decompose into ~9–12 TM-shape Subtasks well within the 25-Subtask soft ceiling per workflow-orchestration §3.4. Suggested decomposition grouping (PLAN.md owns the final shape):

1. {53.5} — §P-9 migration (op_id ALTER + cache table CREATE) — unblocks all downstream slices.
2. {53.6} — §P-12 faiss-cpu pin — small but discrete; PR-able independently.
3. {53.7} — §P-2 canonicalisation function + unit tests.
4. {53.8} — §P-5 entity_context port + unit tests.
5. {53.9} — §P-4 em_target schema + mount.
6. {53.10} — §P-3 entity_mentions declare_row in ingest_file (depends on {53.7} + {53.8} + {53.9}).
7. {53.11} — §P-7 KhEntityEmbedder.
8. {53.12} — §P-8 KhPairResolver + cache table integration (depends on {53.5}).
9. {53.13} — §P-6 + §P-1 Stage-5 post-pass core + attach (depends on {53.10} + {53.11} + {53.12}).
10. {53.14} — §P-10 + §P-11 + §P-14 integration tests + failure-mode harness (depends on {53.13}).
11. {53.15} — §P-13 documentation amendment at impl close.

All Subtasks are SIBLING-ONLY dependencies (within ID-53); no cross-Task Subtask dependencies needed — confirms the Task boundary is correct.

---

§7. Verification (OQ-3 pre-ratification empirical import-and-call check)

Per the OQ-3 forcing function, every external cocoindex / faiss-cpu symbol this TECH cites was re-checked against the installed pin via inspect.signature / hasattr / importlib.import_module, sandbox-disabled per the CLAUDE.md LMDB / faiss gotcha. The exhaustive log is RESEARCH.md §R1; this section records the subset load-bearing for the §P-6 / §P-7 / §P-8 designs.

• Date: 28/05/2026.
• Pinned version: cocoindex[postgres]==1.0.3 (requirements.txt line 38); faiss-cpu==1.14.2 (post-§P-12 pin; verified installable per RESEARCH §R5.1).

PRESENT (the cocoindex 1.0.3 entity-resolution API surface this TECH consumes):

Symbol	Result	Bears on
cocoindex.ops.entity_resolution.resolve_entities (async, collection-level)	PRESENT — signature (entities: Iterable[str], *, embedder: _Embedder, resolve_pair: PairResolver, is_existing_canonical=None, existing_policy=ExistingCanonicalPolicy.PINNED, on_resolution=None, max_distance=0.3, top_n=5) -> ResolvedEntities	§P-6 step 4
cocoindex.ops.entity_resolution.ResolvedEntities	PRESENT — methods `canonical_of(name) -> str	None, canonicals, groups, to_dict()`
cocoindex.ops.entity_resolution.PairResolver (Protocol, @runtime_checkable)	PRESENT — async def __call__(entity: str, candidates: list[str]) -> PairDecision	§P-8
cocoindex.ops.entity_resolution.PairDecision	PRESENT — members matched: Optional[str], canonical: CanonicalSide (StrEnum: NEW='new', MATCHED='matched'; default CanonicalSide.MATCHED). Empirically re-probed S276 — the Optional[str] annotation in earlier ratifications conflated both fields; canonical is a constrained-value StrEnum that cocoindex.ops.entity_resolution._new_wins consumer evaluates against CanonicalSide.NEW — a raw string for canonical silently demotes the entity at runtime.	§P-8
cocoindex.ops.entity_resolution._Embedder (Protocol, @runtime_checkable)	PRESENT — async def embed(text: str) -> NDArray[float32]	§P-7
cocoindex.ops.litellm.LiteLLMEmbedder("text-embedding-3-large", dimensions=1024)	PRESENT (reused from Stage-4) — empirically verified per RESEARCH §R3.2 + flow.py:625	§P-7
faiss.IndexFlatIP (consumed by resolve_entities internals)	PRESENT at faiss-cpu==1.14.2 per RESEARCH §R5.1	§P-6 indirectly
cocoindex.ops.entity_resolution.ExistingCanonicalPolicy (enum: PINNED)	PRESENT	§P-6 (default policy at v1; not overridden)

No SIGNATURE_DRIFT / ABSENT findings introduced beyond RESEARCH §R1’s empirical evidence. The Stage-5 attach point at mount_each.handle.ready() was verified PRESENT in RESEARCH §R1.5; no further re-verification needed for this TECH’s design. The faiss-cpu IndexFlatIP path is consumed internally by resolve_entities per RESEARCH §R1.2 source-read at lines 25–30; TECH does not call it directly.

Internal-symbol citations (NOT subject to OQ-3 — caught by ast-dataflow + gitnexus):

• scripts/cocoindex_pipeline/flow.py symbols (line numbers verified against HEAD 80fbbd5e): 162 _PIPELINE_ERROR_CLASSES, 168 "entity_resolution_failed", 182 _classify_stage_exception, 278 _emit_stage_error_log, 382 _empty_stage_counts, 864 extract_entity_mentions discard, 915 stage_counter.increment("embedding"), 1060 mount_each, 1069 handle.ready, 1083–1084 Stage-5 deferred comment, 1132 stage_counts fold-back, 1135 _emit_pipeline_run_webhook.
• scripts/cocoindex_pipeline/extraction.py:180 EntityMentionExtraction, :551 extract_entity_mentions.
• scripts/cocoindex_pipeline/flow_context.py FLOW_META_CTX, bind_flow_meta, current_flow_meta, bind_stage_counter — all verified resolvable.
• lib/entities/entity-context.ts:19 extractEntityContext — verified at HEAD via grep.

No escalation findings. All cited symbols are PRESENT at the cited line numbers + signatures; the Stage-5 implementation can proceed under the cocoindex 1.0.3 + faiss-cpu 1.14.2 substrate without further upstream changes.

---

§8. Cross-references

§8.1 Source-of-truth spec lineage

• Predecessor specs (this directory):
  • docs/specs/id-53-stage-5-entity-resolution/RESEARCH.md — ratified S274 (Checker PASS_WITH_NOTES at commit 369e77be).
  • docs/specs/id-53-stage-5-entity-resolution/PRODUCT.md — ratified S274 (Checker PASS_WITH_NOTES; commit 80fbbd5e; 21 invariants + Inv-11 elevation to per-row delta).
• Successor (fresh Planner per Q-PLANNER-2): docs/specs/id-53-stage-5-entity-resolution/PLAN.md {53.4} — decomposes ratified PRODUCT + TECH pair into TM-shape implementation Subtasks {53.5+}. §6 above proposes a decomposition grouping.
• Sibling specs (cross-linked, NOT restated):
  • docs/specs/id-28-cocoindex-flow-scaffolding/{PRODUCT,TECH}.md — T8 substrate this TECH inherits (bind_stage_counter, _PIPELINE_ERROR_CLASSES, _emit_stage_error_log, _emit_pipeline_run_webhook, op_id pattern from §P-4.M1).
  • docs/specs/id-36-cocoindex-extraction-contract/{PRODUCT,TECH}.md — Q-EX2 ExtractionOutput contract Stage-5 inherits (EntityMentionExtraction Pydantic shape).
  • docs/themes/canonical-pipeline/reference/canonical-pipeline-sequencing.md §1 + §2 + §3 — critical-path framing + T12 sequencing.

§8.2 Implementation-relevant source files

• scripts/cocoindex_pipeline/flow.py HEAD 80fbbd5e — primary edit target; §P-1, §P-3, §P-4, §P-6, §P-10.
• scripts/cocoindex_pipeline/extraction.py — EntityMentionExtraction Pydantic class; consumed by §P-3 row construction.
• scripts/cocoindex_pipeline/flow_context.py — substrate; consumed by §P-6.
• scripts/cocoindex_pipeline/canonicalisation.py — NEW; §P-2.
• scripts/cocoindex_pipeline/entity_context.py — NEW; §P-5.
• scripts/cocoindex_pipeline/stage_5.py — NEW; §P-1 + §P-6.
• scripts/cocoindex_pipeline/entity_embedder.py — NEW; §P-7.
• scripts/cocoindex_pipeline/pair_resolver.py — NEW; §P-8.
• supabase/migrations/<timestamp>_id53_entity_mentions_op_id_and_pair_cache.sql — NEW; §P-9.
• supabase/types/database.types.ts — regenerated post-§P-9 migration apply.
• requirements.txt — EDIT; §P-12.

§8.3 CLAUDE.md gotchas applied (per Proposed change)

• §P-6, §P-9: DDL via Supabase CLI ONLY (supabase migration new + db push); NEVER MCP execute_sql / apply_migration. Pre-push cat supabase/.temp/project-ref verification per §P-9 apply procedure.
• §P-9: IF NOT EXISTS guards on every ALTER + CREATE for idempotent re-apply. No new PL/pgSQL functions, so no SET search_path = public, extensions body needed (migration top-level SET is sufficient); no anon-EXECUTE REVOKE pattern needed. Future v1.1 helpers (e.g. entity_pair_resolution_emit()) MUST apply the canonical REVOKE FROM PUBLIC + REVOKE FROM anon + GRANT TO authenticated, service_role pattern.
• §P-3, §P-6: Embedding vector serialisation NOT applicable to Stage-5 (the embedder writes to content_items.embedding at Stage 4, NOT to entity_mentions; Stage-5 reads canonical_name strings and writes canonical_name UPDATEs only — no vector columns).
• §P-6: getAuthorisedClient() discriminated union NOT applicable (Stage-5 is server-side Python pipeline; auth is via the Cloud Run Service’s per-tenant service-account + CRON_SECRET-mounted webhook).
• §P-6 step 6: sb() / tryQuery() from @/lib/supabase/safe NOT applicable (Python uses asyncpg directly, NOT the TypeScript Supabase client). Equivalent KH no-silent-failure ethos: Stage-5’s UPDATE batch runs inside async with conn.transaction() so any UPDATE failure aborts the batch + rolls back to a clean state.
• §P-3, §P-4: No barrel re-exports — all imports are direct file paths (e.g. from scripts.cocoindex_pipeline.canonicalisation import canonicalise_entity_name, NOT from scripts.cocoindex_pipeline import *).
• §P-8: classifyContent userId UUID gotcha — NOT applicable (Stage-5 does not invoke classifyContent; the resolver invokes the Anthropic SDK directly via the pipeline’s existing _anthropic_retry wrapper, which does not consume a userId field).
• §P-12: Python pipeline output via PYTHONUNBUFFERED=1 — already set in the Cloud Run sidecar manifest per cloudrun/services/*-cocoindex.yaml (T8 substrate).

§8.4 Test-philosophy alignment

Tests in §3 coverage matrix verify REAL BEHAVIOUR per docs/reference/test-philosophy.md:

• Real DB calls (not mocks) in integration tests — cross-document dedup happens; canonical_name values match; op_id round-trips; cache hits replay.
• Deterministic seeds for embedder + PairResolver tests — use fixed corpora that reliably trigger near-matches.
• Failure injection at the boundary (mock resolve_entities to raise) — verifies Inv-12 non-destructive behaviour without coupling to specific exception types.
• No implementation coupling — tests assert observable behaviour (op_id values; canonical_name string match; cache row count), NOT internal call counts or function-call sequence.

---

End of TECH.md. S274 ID-53.3 ratification gate: 14 Proposed changes covering 21 PRODUCT invariants 1-to-1; 3 P-OQs resolved (P-OQ1 PERMANENT; P-OQ2 DOCUMENTED + ACCEPTED; P-OQ3 RATIFIED via new entity_pair_resolutions table); 1 migration (CLI migration combining entity_mentions op_id ALTER + entity_pair_resolutions CREATE TABLE); 5 NEW files created + 2 EDITED files (flow.py, requirements.txt); 3 T-OQs surfaced for Liam ratification (T-OQ1 _classify_stage_exception branch addition; T-OQ2 PairResolver prompt + parser shape; T-OQ3 cache TTL eviction); total effort estimate ~18.25 h; PLAN.md decomposition REQUIRED. Re-grounded against parent tip 80fbbd5e; no ABSENT / SIGNATURE_DRIFT findings introduced. UK English throughout; no emoji; CLAUDE.md gotchas applied; DDL CLI-only discipline observed.