Extraction Links Maintenance System

Status: Active
Last Updated: 2026-08-14
Scope: Python backend, Neo4j graph database
Components: Maintenance.py, dataset_helpers.py, OntologyService

Overview

The Extraction Links Maintenance System automatically cleans and maintains relationships between entities and their source documents within the Neo4j knowledge graph. It runs as a background maintenance thread and performs four critical tasks:

Normalize existing :SOURCE relationships — Ensure all Entity→Document :SOURCE edges target :Document nodes only (not :Chapter or :Paragraph)
Extract entity reference targets — Parse entity attributes for source references (bare fragment IDs and canonical URIs)
Create :SOURCE relationships — Link entities to their referenced source nodes with proper validation
Remove malformed entities — Delete non-Innovator Entity nodes that have no :SOURCE relationship in either direction, even when they have other relationships

Architecture

Maintenance Thread (`Maintenance.py`)

The main entry point is the maintenance orchestration function:

knowledge_graph_extraction_links_maintenance_thread() -> bool

Workflow: 1. Sanitize existing :SOURCE relationships (one-time migration: Chapter edges → Document edges) 2. Fetch all Entity node IDs (excluding Innovator-labeled entities) 3. For each entity (with 100ms throttling between processing): - Fetch raw attributes - Extract reference targets - Check node existence - Prune obsolete references - Create SOURCE relationships for existing targets 4. After all entities are processed successfully, remove every non-Innovator Entity node with no SOURCE relationship in either direction. Synchronized ontology targets are preserved by their direct target-to-source :SOURCE link, even when they have no attributes.

Shutdown Handling: - Shutdown signaling via threading.Event with 100ms pauses - Safe interruption at entity boundaries (no partial operations) - Proper logging of completion status

Reference Resolution Utilities (`dataset_helpers.py`)

Centralized utility module for parsing and resolving reference URIs. Provides three core functions:

`parse_reference_uri(uri: str) -> Optional[Tuple[str, str]]`

Parse canonical source URIs into (node_label, node_id) tuples.

Supported schemes: - doc://docid-106-c0-p2 → ("Paragraph", "docid-106-c0-p2") - doc://docid-106-c0 → ("Chapter", "docid-106-c0") - doc://docid-106 → ("Document", "docid-106") - kg://charly → ("Entity", "charly") - inno://inno-42 → ("Entity", "inno-42") (Innovator entity)

Returns: None for unsupported schemes (http://, https://, skill://, ds://) or malformed URIs.

`build_reference_uri(node_label: str, node_id: str, qualifier: Optional[str] = None) -> Optional[str]`

Inverse of parse_reference_uri(). Build canonical URIs from Neo4j node label and ID. The optional qualifier appends document paragraph metadata (?p=<qualifier> for Document/Chapter/ Paragraph) or a knowledge-graph sub-path (/<qualifier> for Entity).

Examples: - ("Document", "docid-42") → "doc://docid-42" - ("Entity", "researcher") → "kg://researcher"

Returns: None for labels without canonical URI schemes.

`resolve_source_reference(fragment_id_or_uri: str) -> Optional[Tuple[str, str]]`

Key unifying function — Handle both bare fragment IDs and canonical URIs with transparent resolution.

Handles: - Bare fragment IDs (e.g., "docid-19-c7-p3") — Normalize Chapter/Paragraph to root Document - Canonical URIs (doc://, kg://, inno://) — Parse and normalize Chapter/Paragraph to root Document - Chapter/Paragraph normalization — Always resolve Chapter and Paragraph fragments to their root Document ID via DocumentsInterface.get_document_id_from_fragment_id()

Returns: (node_label, node_id) or None if unresolvable.

Example:

resolve_source_reference("docid-19-c7-p3")           # → ("Document", "docid-19")
resolve_source_reference("doc://docid-19-c7-p3")    # → ("Document", "docid-19")
resolve_source_reference("kg://entity_name")        # → ("Entity", "entity_name")

`EMBEDDED_REFERENCE_URI_RE`

Compiled regex for extracting canonical URIs embedded in free-text attribute values:

EMBEDDED_REFERENCE_URI_RE = re.compile(r'\b(?:doc|kg|inno)://[^\s<>"\')]+')

Matches: doc://..., kg://..., inno://... anywhere in text, stopping at whitespace or HTML delimiters.

OntologyService Helper Methods

Four new methods in OntologyService provide Neo4j integration for maintenance logic:

`get_entity_raw_attributes(element_id: str) -> Dict[str, str]`

Fetch non-reserved, unconsolidated JSON attributes from an entity node.

Query pattern: Returns attributes with keys starting with "[" (non-reserved namespace).

Returns: Dictionary mapping attribute name → JSON string value.

`nodes_exist(node_references: List[Tuple[str, str]]) -> Dict[Tuple[str, str], Optional[bool]]`

Batch existence check for multiple (node_label, node_id) references.

Optimization: Groups references by node_label to minimize Neo4j queries (e.g., one MATCH per label type).

Returns: Dictionary mapping (node_label, node_id) → True (exists), False (confirmed missing), or None (existence query failed).

Maintenance prunes obsolete references only when every existence check completed successfully. A failed query is treated as unknown and cannot cause source attributes to be deleted.

`remove_obsolete_references_from_entity_attributes(element_id: str, obsolete_references: Set[Tuple[str, str]]) -> bool`

Critical pruning operation — Remove all obsolete references from entity attributes:

Source ID field pruning — Parse JSON arrays of {value, source_id} objects
For each source_id entry, resolve via resolve_source_reference()
If resolved reference is in obsolete_references, filter it out
Delete attribute if all entries are filtered; otherwise update with remaining entries
Embedded URI pruning — Scan attribute value text for embedded URIs
Extract URIs via EMBEDDED_REFERENCE_URI_RE
For each URI, resolve via resolve_source_reference()
If resolved reference is in obsolete_references, prune URI from value text

Returns: bool indicating success/failure.

`create_source_relationship(entity_element_id: str, target_label: str, target_id: str) -> bool`

Create or update a relationship from entity to target node:

MERGE (e:Entity {elementId: $entity_id})-[:SOURCE]->(target)
WHERE target:$target_label AND target.id = $target_id

Returns: bool indicating success/failure.

Final malformed-entity cleanup

After extraction-link processing completes without shutdown interruption, the maintenance thread removes every :Entity node that:

does not have the :Innovator label; and
has no :SOURCE relationship in either direction.

The node is removed with DETACH DELETE, so other relationship types do not prevent deletion. This intentionally treats non-Innovator entities without source provenance as malformed, even when they remain connected to other nodes.

Ontology data-change notification

The consolidation maintenance loop tracks whether any consolidation or hint work was completed during an iteration. When work was done, Python posts an authenticated POST /v1/api/ontology/data-changed request to Blazor using PLATFORM_API_TOKEN. The endpoint calls IOntologyService.NotifyDataChanged(), which raises OnDataChanged for subscribed Blazor components so they reload their ontology data. Notification failures are logged and do not stop maintenance.

Entity Reference Sources

Entities can reference source materials through multiple formats, which are all normalized by the maintenance system:

1. Source ID Fields

Found in unconsolidated entity attributes under [source_id] key. Format:

{
  "[source_id]": [
    {"value": "Fragment A", "source_id": "doc://docid-20?p=p-17"},
    {"value": "Fragment B", "source_id": "doc://docid-20?p=p-5"},
    {"value": "Fragment C", "source_id": "doc://docid-22"}
  ]
}

Structure: canonical doc://<document_id>[?p=<paragraph_id>] tokens (comma-separated multiple references).

2. Embedded URIs in Attribute Text

References embedded in the value field of attributes using canonical URI schemes:

{
  "[description]": [
    {"value": "Related doc://docid-25 and entity kg://researcher_X", "source_id": "doc://docid-20"}
  ]
}

Schemes: - doc://docid-NNN[−c−p] — Document/Chapter/Paragraph fragments - kg://entity_name — Entity references - inno://inno-NNN — Innovator entities (dual :Entity:Innovator label)

3. Document Hierarchy Normalization

Key normalization rule: All Chapter and Paragraph fragment references (whether bare IDs or URIs) are automatically resolved to their root Document ID via DocumentsInterface.get_document_id_from_fragment_id().

Rationale: Entity :SOURCE relationships only target Document or Entity nodes in the Neo4j schema; Chapter/Paragraph are internal hierarchy detail.

Data Flow Example

Entity with Mixed References

Entity node in Neo4j:

(:Entity {elementId: "entity_42", name: "Legal Precedent A"})

Raw attributes:

{
  "[citations]": [
    {"value": "Judgment doc://docid-15-c0-p2", "source_id": "doc://docid-15?p=p-2"},
    {"value": "Policy doc://docid-18", "source_id": "doc://docid-18"}
  ],
  "[related_entities]": [
    {"value": "See kg://precedent_B for comparison", "source_id": null}
  ]
}

Maintenance Thread Processing

Extract targets:
From [citations][0]: doc://docid-15?p=p-2 → resolve → ("Document", "docid-15") + embedded URI doc://docid-15-c0-p2 → resolve → ("Document", "docid-15")
From [citations][1]: doc://docid-18 → resolve → ("Document", "docid-18") + embedded URI doc://docid-18 → resolve → ("Document", "docid-18")
From [related_entities][0]: embedded URI kg://precedent_B → resolve → ("Entity", "precedent_B")
Deduplicate: Resulting targets: {("Document", "docid-15"), ("Document", "docid-18"), ("Entity", "precedent_B")}
Check existence: Batch query verifies all three exist
Create relationships:
(entity_42)-[:SOURCE]->(docid-15:Document)
(entity_42)-[:SOURCE]->(docid-18:Document)
(entity_42)-[:SOURCE]->(precedent_B:Entity)
Pruning (if any target missing): Automatically remove from attributes and skip :SOURCE creation

Configuration & Deployment

Threading

Located in main_utility.py. Maintenance thread is instantiated with:

from src.core.base.Maintenance import knowledge_graph_extraction_links_maintenance_thread

# Spawn thread with shutdown signaling
shutdown_event = threading.Event()
maintenance_thread = threading.Thread(
    target=knowledge_graph_extraction_links_maintenance_thread,
    args=(shutdown_event,),
    daemon=False
)
maintenance_thread.start()

Throttling

Entity processing uses 100ms pauses between operations:

shutdown_event.wait(0.1)  # 100ms pause, interruptible

This prevents database overload during large entity batches.

Logging

Module-level logging via log_start_file(__file__) / log_end_file(__file__):

from src.utils.log import log, INFO, WARNING, ERROR
log(INFO, "Starting extraction links maintenance...")

All errors logged at ERROR or WARNING level with context.

Implementation Notes

Single Return Pattern

All maintenance functions follow project code style: single return statement at function end with strategic blank lines before return.

def example_function():
    result: bool = False

    try:
        # main logic
        result = True
    except Exception as e:
        log(ERROR, f"Error: {e}")

    return result

Type Hints

Comprehensive type annotations on all functions:

from typing import Optional, List, Dict, Set, Tuple

def resolve_source_reference(fragment_id_or_uri: str) -> Optional[Tuple[str, str]]:
    ...

def nodes_exist(node_references: List[Tuple[str, str]]) -> Dict[Tuple[str, str], bool]:
    ...

Parameterized Cypher Queries

All Neo4j queries use parameterized values (via dict) to prevent injection:

query = "MATCH (e:Entity {elementId: $element_id}) RETURN e"
execute_neo4j_query(query, {"element_id": element_id})

Backward Compatibility

dataset_interface.py re-exports helpers from dataset_helpers.py for backward compatibility:

from src.utils.dataset_helpers import parse_reference_uri, build_reference_uri, resolve_source_reference, EMBEDDED_REFERENCE_URI_RE

Old code importing from the interface continues to work.

Testing

Comprehensive unit tests cover:

Reference parsing — Bare fragment IDs, canonical URIs, malformed inputs
Reference resolution — Chapter/Paragraph normalization, all URI schemes
Attribute extraction — JSON parsing, embedded URI detection, deduplication
Node existence checking — Batch queries, mixed label types
Pruning logic — source_id field updates, embedded URI removal, attribute deletion
Relationship creation — MERGE idempotency, error handling
Shutdown handling — Early termination, partial entity skipping

Test data files in tests/: - test_extraction_links_maintenance.py — Full integration test suite - test_dataset_helpers.py — Unit tests for URI parsing/resolution

Migration Notes

Moving from DatasetInterface to dataset_helpers

Prior to 2026-08-14, static helpers were defined in DatasetInterface (an ABC). Refactored to separate utility module for clarity:

Before:

from src.interfaces.dataset_interface import DatasetInterface
uri_parsed = DatasetInterface.parse_reference_uri(uri)

After:

from src.utils.dataset_helpers import parse_reference_uri
uri_parsed = parse_reference_uri(uri)

Backward compatibility: dataset_interface.py still re-exports, so old imports continue to work.

Troubleshooting

Orphaned References

If maintenance encounters references to non-existent nodes:

Logged at WARNING level — Each stale reference recorded
Automatically pruned — Removed from entity attributes
Skipped from :SOURCE — No relationship created

Check logs for ERROR messages if systematic pruning failures occur.

Performance

For large knowledge graphs (>100K entities): - 100ms throttling may cause maintenance to run for several hours - Consider adjusting throttle interval via code config (default: 0.1 seconds) - Batch operations in nodes_exist() minimize database load

Shutdown Cleanup

If thread is forcibly killed: - Partially processed entity is skipped - Next maintenance run picks up from entity list again (no corruption) - Idempotent MERGE operations ensure no duplicate relationships