---
title: 'TEXTIQ-382 — Per-node scoped domain term dictionary (v2.0)'
type: 'feature'
created: '2026-04-17'
status: 'done'
context:
  - '_planning/docs/services/textiq-doc-extraction/specs/tech-spec-dynamic-domain-term-dictionary.md'
jira_id: TEXTIQ-382
baseline_commit: 6fbfc45e9df3523e54557f73953d9b7fcea9e9a1
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## Intent

**Problem:** The `domain_term_dictionary` enforces a **global** UNIQUE on `term_original`. Once a term (e.g., "revenue") is ingested by any node in the hierarchy, no other node can ever own that term — cross-tenant/cross-node imports silently drop. Additionally, `DomainDictionaryService.get_existing_terms_set()` scans the whole table without node filtering, so the in-memory dedup set returns terms from unrelated nodes and blocks legitimate inserts before hitting the DB.

**Approach:** Replace the global UNIQUE with a composite UNIQUE `(term_original, node_id)` via Alembic migration + ORM update, then scope the service's dedup/query methods by `node_id`. `node_id` already exists on the model (via `TenantIsolationMixin`) and the Airflow task already passes it down — this change only flips dedup from global to per-node. Also fix an unrelated NameError in `_llm_extract` where a logger call references an undefined `node` variable.

## Boundaries & Constraints

**Always:**
- Preserve existing tenant RLS. Node scoping is an additional application-level filter on top of RLS — **not** a replacement.
- New constraint name: `uq_term_original_node` on `(term_original, node_id)`.
- Migration must be reversible (`downgrade()` restores the global UNIQUE).
- `get_existing_terms_set(node_id)` — `node_id` is required (no default). Callers must provide it.
- `add_terms_batch` `ON CONFLICT` target must match the new unique constraint `(term_original, node_id)`.

**Ask First:**
- None. Target DB state verified: `domain_term_dictionary_term_original_key` is the exact constraint to drop; all rows have `tenant_id` + `node_id` populated; tenant RLS is active and stays untouched.

**Never:**
- Do **not** change the Airflow task signature — `node_id` is already wired.
- Do **not** add cross-node inheritance, template dictionaries, or "copy from node" features (deferred per tech-spec §Notes).
- Do **not** add API endpoints — this is a schema + service layer change only.
- Do **not** remove `tenant_id` scoping — tenant RLS remains the outer boundary.

## I/O & Edge-Case Matrix

| Scenario | Input / State | Expected Output / Behavior | Error Handling |
|----------|---------------|---------------------------|----------------|
| Same term in different nodes | "revenue" exists in node A; node B extracts "revenue" | Row inserted into node B | N/A |
| Same term in same node | "revenue" exists in node A; node A re-extracts "revenue" | No new row; `get_existing_terms_set(A)` sees it and skips | N/A |
| `get_existing_terms_set(node_id)` | Table has ("revenue", A), ("revenue", B), ("cost", A) | Returns `{"revenue", "cost"}` when called with A; `{"revenue"}` when called with B | N/A |
| Batch insert with mixed nodes | Entries for nodes A and B, one duplicate in A | A's duplicate skipped via ON CONFLICT; B's row inserted | Existing try/rollback unchanged |
| Migration downgrade | `alembic downgrade -1` | Composite constraint dropped, global UNIQUE restored | Fail loudly if global uniqueness would be violated by current data |
| `_llm_extract` LLM empty response | `_llm_extract(content="...")` returns empty `result_text` | Logs `llm_empty_after_think_strip` warning without `node_id` kwarg; returns `[]` | Was raising NameError; now logs cleanly |

</frozen-after-approval>

## Code Map

- `src/db/models.py:299-327` — `DomainTermDictionary` ORM model. Remove `unique=True` on line 307; add `UniqueConstraint` to `__table_args__`.
- `src/knowledge/domain_dictionary_service.py` — add `node_id` to `__init__`; filter `get_all_terms` and `get_existing_terms_set` by `node_id`; update `add_terms_batch` ON CONFLICT target.
- `src/extraction_v2/domain_term_extractor.py:521` — bug: `self.logger.warning("llm_empty_after_think_strip", node_id=str(node.id))` — `node` is undefined in `_llm_extract(self, content: str)`. Drop the kwarg.
- `src/extraction_v2/domain_term_extractor.py:~830-860` — `persist_new_terms` already accepts `node_id`. Pass it into `DomainDictionaryService(session, tenant_id, node_id)` and `get_existing_terms_set(node_id)`.
- `src/db/migrations/versions/` — new migration. `down_revision = "3c4d5e6f7a8b"` (current head: `20260413_100000_add_document_approval_fields.py`).
- `tests/knowledge/test_domain_dictionary_service.py` — update stale `test_add_terms_batch` assert; add per-node dedup tests.
- `tests/knowledge/conftest.py` — add `node_id` fixture.
- `airflow/dags/tasks/processing_tasks.py:733,1047` — **no changes**; already passes `node_id` to `persist_new_terms`.

## Tasks & Acceptance

**Execution:**
- [ ] `src/db/migrations/versions/20260417_XXXXXX_scope_domain_term_unique_per_node.py` -- new migration: `op.drop_constraint("domain_term_dictionary_term_original_key", "domain_term_dictionary", type_="unique")` then `op.create_unique_constraint("uq_term_original_node", "domain_term_dictionary", ["term_original", "node_id"])`; inverse in `downgrade()`. Rationale: flip DB-level uniqueness from global to per-node.
- [ ] `src/db/models.py` -- remove `unique=True` from `term_original` column (line 307); add `UniqueConstraint("term_original", "node_id", name="uq_term_original_node")` as first entry in `__table_args__`. Rationale: keep ORM in sync with schema.
- [ ] `src/knowledge/domain_dictionary_service.py` -- add `node_id: str | UUID | None = None` to `__init__`; add `.filter(DomainTermDictionary.node_id == self.node_id)` to `get_all_terms` and `get_existing_terms_set`; change `get_existing_terms_set` signature to `(self, node_id: UUID | None = None)` allowing override (fall back to `self.node_id`); change `add_terms_batch` `on_conflict_do_nothing(index_elements=["term_original", "node_id"])`. Rationale: node-scoped dedup + constraint alignment.
- [ ] `src/extraction_v2/domain_term_extractor.py:521` -- remove the `node_id=str(node.id)` kwarg from the `self.logger.warning("llm_empty_after_think_strip", ...)` call. Rationale: `node` is out of scope — NameError when this branch fires.
- [ ] `src/extraction_v2/domain_term_extractor.py` (`persist_new_terms`) -- instantiate `DomainDictionaryService(session, tenant_id=tenant_id, node_id=node_id)`; call `service.get_existing_terms_set(node_id=node_id)`. Rationale: dedup must be scoped to the owning node.
- [ ] `tests/knowledge/conftest.py` -- add `sample_node_id` fixture (UUID). Rationale: reusable across service tests.
- [ ] `tests/knowledge/test_domain_dictionary_service.py` -- fix `test_add_terms_batch` to assert against `session.execute` (not `session.add_all`); add `test_get_existing_terms_set_filters_by_node`, `test_same_term_different_nodes_allowed`, `test_same_term_same_node_deduped`. Rationale: lock in per-node behavior; retire stale assert.

**Acceptance Criteria:**
- Given the migration has run, when I inspect the DB, then constraint `uq_term_original_node` exists on `(term_original, node_id)` and no constraint enforces global uniqueness of `term_original`.
- Given "revenue" exists for node A, when I insert ("revenue", node B) via `add_terms_batch`, then the row is persisted.
- Given "revenue" exists for node A, when I insert ("revenue", node A) via `add_terms_batch`, then `rowcount == 0` and the original row is unchanged.
- Given a `DomainDictionaryService(session, node_id=A)`, when I call `get_existing_terms_set()`, then only terms with `node_id == A` are returned.
- Given `_llm_extract` is called with content producing an empty post-think-strip result, when the warning branch fires, then the method returns `[]` without raising `NameError`.
- Given `alembic downgrade -1` is run, when I inspect the DB, then the global UNIQUE on `term_original` is restored and `uq_term_original_node` is gone.

## Design Notes

The existing column-level `unique=True` in SQLAlchemy generates a constraint named `<table>_<column>_key` in PostgreSQL (i.e. `domain_term_dictionary_term_original_key`). If autogenerate was never used on this column, the name is deterministic. The migration should drop by that name explicitly rather than relying on `op.drop_constraint(None, ...)`.

`get_existing_terms_set` keeps the lowercase normalization (`row[0].lower()`) — existing behavior. Node filter is added before `.all()`:
```python
def get_existing_terms_set(self, node_id: UUID | None = None) -> set[str]:
    self._set_rls_context()
    node_filter = node_id if node_id is not None else self.node_id
    query = self.session.query(DomainTermDictionary.term_original).filter(
        DomainTermDictionary.node_id == node_filter
    )
    return {row[0].lower() for row in query.all()}
```

## Verification

**Commands:**
- `cd /home/nguyen/code/textiq-doc-extraction && uv run alembic -c alembic.ini upgrade head` -- expected: migration applies without error.
- `cd /home/nguyen/code/textiq-doc-extraction && uv run alembic -c alembic.ini downgrade -1 && uv run alembic -c alembic.ini upgrade head` -- expected: reversible.
- `cd /home/nguyen/code/textiq-doc-extraction && uv run pytest tests/knowledge/test_domain_dictionary_service.py -v` -- expected: all pass (including new per-node tests).
- `cd /home/nguyen/code/textiq-doc-extraction && uv run pytest tests/extraction_v2/ -v` -- expected: no regressions.

**Manual checks:**
- `psql -c "\d domain_term_dictionary"` — expect `uq_term_original_node` unique constraint; no standalone unique on `term_original`.

## Suggested Review Order

**Schema — flip uniqueness from global to per-node**

- New composite UNIQUE + its drop of the legacy column-level key; this is the root of the change.
  [`20260417_100000_scope_domain_term_unique_per_node.py:22`](../../src/db/migrations/versions/20260417_100000_scope_domain_term_unique_per_node.py#L22)

- ORM aligned with the new DB state — `unique=True` removed; composite `UniqueConstraint` added to `__table_args__`.
  [`models.py:308`](../../src/db/models.py#L308)
  [`models.py:324`](../../src/db/models.py#L324)

**Service — node-scoped dedup**

- New `node_id` parameter wired through the constructor; `tenant_id` stays untouched.
  [`domain_dictionary_service.py:12`](../../src/knowledge/domain_dictionary_service.py#L12)

- Dedup query now filters by `node_id`; hard-fails if neither arg nor service-level node_id is set (intentional — silent cross-node scans would defeat the feature).
  [`domain_dictionary_service.py:52`](../../src/knowledge/domain_dictionary_service.py#L52)

- ON CONFLICT target switched to composite `(term_original, node_id)` — must match the new unique constraint for inference to work.
  [`domain_dictionary_service.py:99`](../../src/knowledge/domain_dictionary_service.py#L99)

**Extractor — wire node context + defensive guard**

- `persist_new_terms` now short-circuits when `node_id` is missing, surfacing a warning instead of silently raising inside the service (review-finding patch).
  [`domain_term_extractor.py:857`](../../src/extraction_v2/domain_term_extractor.py#L857)

- Service constructed with node_id passthrough; this is the one line in `persist_new_terms` that enables per-node dedup.
  [`domain_term_extractor.py:882`](../../src/extraction_v2/domain_term_extractor.py#L882)

- Unrelated NameError fix: `node` was undefined in `_llm_extract`; the warning now fires without crashing.
  [`domain_term_extractor.py:521`](../../src/extraction_v2/domain_term_extractor.py#L521)

**Tests — lock in the new contract**

- Composite ON CONFLICT target verified by compiling the statement and asserting both column names appear.
  [`test_domain_dictionary_service.py:119`](../../tests/knowledge/test_domain_dictionary_service.py#L119)

- Per-node isolation: same term in two different nodes does not collide.
  [`test_domain_dictionary_service.py:168`](../../tests/knowledge/test_domain_dictionary_service.py#L168)

- ValueError guard prevents accidental global scans.
  [`test_domain_dictionary_service.py:91`](../../tests/knowledge/test_domain_dictionary_service.py#L91)