scripts.backfill_thread_links

Reference › scripts › backfill_thread_links

A one-shot, idempotent backfill for the knowledge graph. It creates Thread correlation-anchor nodes and the IN_THREAD, HAS_THREAD, BY_VOLK and TO_VOLK edges for records already in the graph (session-254, option A). A human runs it; no module or test imports it. It uses the Bolt repository from core.graph and the @traces decorator from traces.

Invocation: main takes --uri, --user, a password option, --project (default orchestrata-ultimate), --batch-size (default 500) and --apply. By default it is a dry run: the plan reads the graph and the write count stays 0. --apply merges the Thread ontology. This is the repository's convention for every tool that mutates the knowledge graph. Missing connection settings are filled from the repository's dotenv file without overriding the ambient environment.

The flow is _driver, then verify_connectivity, then run_backfill. check_hub first confirms the project hub node exists and aborts with a reason otherwise. plan_backfill reads the source labels and builds a _BackfillPlan; _resolve_volk_alias_map maps Volk aliases to canonical names. Writes go in batches of batch_size (_batched) as MERGE operations, so re-running adds nothing. BackfillStats counts what was done, including the total of IN_THREAD edges.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

BackfillStats (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-008']), dataclass(frozen=True) · Line 113

Plan/execution counts — the --dry-run report and the real result.

_BackfillPlan (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-008']), dataclass(frozen=True) · Line 130

Pure-read plan — the whole dry-run, and the input to the writer.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/backfill_thread_links.py

#!/usr/bin/env python3
"""One-shot, idempotent backfill: Thread correlation-anchor nodes +
``IN_THREAD``/``HAS_THREAD``/``BY_VOLK``/``TO_VOLK`` edges (session-254
Option A, ``plans/thread-ontology-design.md``).

25,225 CXPPacket / 556 Message / 244 ToolCall / 82 Artifact nodes on the
live operational KG carry a producer ``thread_id`` but are unreachable from
the ``(:Project)`` hub — 7,766 of them are fully isolated, and the rest
("connected") only join each other via a packet<->packet ``ROUTED_TO`` web
that never touches the hub. Every one of those nodes already carries
``thread_id``, so the backfill is driven by **``thread_id`` presence**, not
by an isolation predicate (simpler, idempotent, and repairs all 25,225
packets rather than just the isolated 6,885 — see design §0.3).

For each source label (``CXPPacket``/``Message``/``ToolCall``/``Artifact``):
  1. MERGE one ``:Thread`` node per distinct ``thread_id``, keyed on the
     project-qualified :func:`canonical_schema.thread_entity_id`, plus a
     ``(:Project)-[:HAS_THREAD]->(:Thread)`` hub edge.
  2. MERGE ``(record)-[:IN_THREAD]->(:Thread)`` for every record carrying a
     non-empty ``thread_id``.
  3. Where the label carries agent fields (``CXPPacket``: ``from_agent``/
     ``to_agent``; ``Message``: ``sender``/``recipient``), resolve each
     against the live ``:Volk`` identifier pool (``alias`` -> ``volk_id`` ->
     ``name``, alias always wins — the same lesson
     ``kernel/audit/writers.py:311`` already learned) and MERGE
     ``-[:BY_VOLK]->``/``-[:TO_VOLK]->`` accordingly. An agent string with no
     match (``HUMAN``, test-pollution tokens, the real-but-Volk-less
     ``U:Do``) gets no volk edge — the record still gets its ``IN_THREAD``
     edge, so it is never left isolated.

All vocabulary (label, relationship names, the id scheme) is imported from
``knowledge_graph.canonical_schema`` — never a hard-coded graph string
(constitutional rule: ontology before graph import). Structurally modelled
on ``scripts/canonical_conformance.py``: ``plan_*`` pure-read functions,
a ``run_*`` writer, dry-run by default, ``--uri``/``--user``/``--password``, and
batched writes (``_batched(seq, size=500)``). Deliberately NOT folded into
``canonical_conformance.py``: that script's job is keyless-node
purge/backfill and is documented as already-executed; mixing an edge
backfill into it would break its "re-running is a no-op" contract review.

Idempotent: MERGE only — zero ``CREATE``, zero ``DELETE``, zero
``DETACH``, zero property removal. Re-running produces byte-identical
counts (node MERGE on the unique ``id``, edge MERGE on an existing pair).
The default invocation (no ``--apply``) performs the plan reads only and
writes nothing — the house convention for every KG-mutating tool here. A bad run is
repaired by editing the vocabulary and re-running; a full rollback (not
implemented as a flag — nothing here is ever destructive by default) is a
single ``MATCH (t:Thread) DETACH DELETE t``.

Connection defaults to ``NEO4J_URI``/``NEO4J_USERNAME``/``NEO4J_PASSWORD``;
``--uri``/``--user``/``--password`` override so a test-fleet clone can be
targeted without touching the shared default instance.

Usage::

    python scripts/backfill_thread_links.py                    # dry run (default)
    python scripts/backfill_thread_links.py --apply            # actually MERGE
    python scripts/backfill_thread_links.py --uri bolt://10.53.1.13:7687
"""

from __future__ import annotations

import argparse
import os
import sys
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from pathlib import Path

_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from neo4j import GraphDatabase  # noqa: E402

from orchestrata_ultimate.core.base.traces import traces  # noqa: E402
from orchestrata_ultimate.core.graph import BoltSyncRepository  # noqa: E402
from orchestrata_ultimate.core.graph_factory import resolve_uri  # noqa: E402
from orchestrata_ultimate.knowledge_graph.canonical_schema import (  # noqa: E402
    LABEL_PROJECT,
    LABEL_THREAD,
    LABEL_VOLK,
    REL_BY_VOLK,
    REL_HAS_THREAD,
    REL_IN_THREAD,
    REL_TO_VOLK,
    apply,
    get_all_ddl,
    thread_entity_id,
)

# Source-label table (the only per-label knowledge in this script): each
# label maps to its (sender_property, recipient_property) pair, or
# (None, None) when the label carries no agent fields at all (ToolCall,
# Artifact). Overridable (``source_labels=``) only as a test seam, scoped to
# an isolated test-only label — production always uses this default.
_SOURCE_LABELS: dict[str, tuple[str | None, str | None]] = {
    "CXPPacket": ("from_agent", "to_agent"),
    "Message": ("sender", "recipient"),
    "ToolCall": (None, None),
    "Artifact": (None, None),
    # Hub-link wave (plans/hub-link-plan.md §8, session-254 ruling): Intent
    # and PipelineRun are Thread-family too -- 187 of each carry a
    # thread_id but link only to each other via TRIGGERED. Neither carries
    # agent fields.
    "Intent": (None, None),
    "PipelineRun": (None, None),
}


@traces(spec="106", reqs=["REQ-106-008"])
@dataclass(frozen=True)
class BackfillStats:
    """Plan/execution counts — the ``--dry-run`` report and the real result."""

    threads: int = 0
    has_thread_edges: int = 0
    in_thread_by_label: dict[str, int] = field(default_factory=dict)
    by_volk_edges: int = 0
    to_volk_edges: int = 0
    unmatched_agents: dict[str, int] = field(default_factory=dict)

    @property
    def in_thread_edges(self) -> int:
        return sum(self.in_thread_by_label.values())


@traces(spec="106", reqs=["REQ-106-008"])
@dataclass(frozen=True)
class _BackfillPlan:
    """Pure-read plan — the whole dry-run, and the input to the writer."""

    # thread_id -> canonical Thread.id
    threads: dict[str, str]
    # label -> [elementId, ...] of every record with a non-empty thread_id
    in_thread_eids: dict[str, list[str]]
    # [{"eid": ..., "alias": ...}, ...] for the sender/BY_VOLK side
    by_volk_rows: list[dict[str, str]]
    # [{"eid": ..., "alias": ...}, ...] for the recipient/TO_VOLK side
    to_volk_rows: list[dict[str, str]]
    # unresolved agent string -> occurrence count (sender + recipient sides)
    unmatched_agents: dict[str, int]


@traces(spec="106", reqs=["REQ-106-003"])
def _resolve_volk_alias_map(
    volk_rows: Iterable[Mapping[str, object]],
) -> dict[str, str]:
    """Build ``identifier -> alias`` from the Volk pool.

    Precedence order — ``alias`` -> itself, then ``volk_id`` -> its node's
    alias, then ``name`` -> its node's alias, each added only if the key is
    not already present. ``alias`` therefore always wins: it is the
    canonical merge key (REQ-106-003), and ``kernel/audit/writers.py:311``
    already learned that MERGEing on ``volk_id`` created divergent nodes.
    """
    rows = list(volk_rows)
    alias_map: dict[str, str] = {}
    for row in rows:
        alias = row.get("alias")
        if alias and alias not in alias_map:
            alias_map[str(alias)] = str(alias)
    for row in rows:
        alias = row.get("alias")
        volk_id = row.get("volk_id")
        if alias and volk_id and volk_id not in alias_map:
            alias_map[str(volk_id)] = str(alias)
    for row in rows:
        alias = row.get("alias")
        name = row.get("name")
        if alias and name and name not in alias_map:
            alias_map[str(name)] = str(alias)
    return alias_map


@traces(spec="106", reqs=["REQ-106-008"])
def plan_backfill(
    repo: BoltSyncRepository,
    *,
    project: str,
    source_labels: Mapping[str, tuple[str | None, str | None]] | None = None,
    volk_label: str = LABEL_VOLK,
) -> _BackfillPlan:
    """Compute the full backfill plan. Pure read — never mutates.

    ``source_labels``/``volk_label`` are test seams (mirroring
    ``canonical_conformance.plan_migration``'s ``merge_keys`` param):
    production always uses the real :data:`_SOURCE_LABELS` /
    :data:`~orchestrata_ultimate.knowledge_graph.canonical_schema.LABEL_VOLK`.
    """
    labels = source_labels if source_labels is not None else _SOURCE_LABELS

    volk_rows = repo.execute_read(
        f"MATCH (v:`{volk_label}`) WHERE v.alias IS NOT NULL "
        f"RETURN v.alias AS alias, v.volk_id AS volk_id, v.name AS name"
    )
    alias_map = _resolve_volk_alias_map(volk_rows)

    threads: dict[str, str] = {}
    in_thread_eids: dict[str, list[str]] = {}
    by_volk_rows: list[dict[str, str]] = []
    to_volk_rows: list[dict[str, str]] = []
    unmatched: dict[str, int] = {}

    for label, (sender_prop, recipient_prop) in labels.items():
        thread_rows = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.thread_id IS NOT NULL "
            f"AND n.thread_id <> '' RETURN DISTINCT n.thread_id AS thread_id"
        )
        for row in thread_rows:
            tid = str(row["thread_id"])
            threads.setdefault(tid, thread_entity_id(project, tid))

        eid_rows = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.thread_id IS NOT NULL "
            f"AND n.thread_id <> '' RETURN elementId(n) AS eid"
        )
        in_thread_eids[label] = [str(row["eid"]) for row in eid_rows]

        if sender_prop is None and recipient_prop is None:
            continue

        agent_rows = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.thread_id IS NOT NULL "
            f"AND n.thread_id <> '' "
            f"RETURN elementId(n) AS eid, n.`{sender_prop}` AS sender, "
            f"n.`{recipient_prop}` AS recipient"
        )
        for row in agent_rows:
            eid = str(row["eid"])
            sender = row.get("sender")
            recipient = row.get("recipient")
            if sender:
                alias = alias_map.get(str(sender))
                if alias:
                    by_volk_rows.append({"eid": eid, "alias": alias})
                else:
                    unmatched[str(sender)] = unmatched.get(str(sender), 0) + 1
            if recipient:
                alias = alias_map.get(str(recipient))
                if alias:
                    to_volk_rows.append({"eid": eid, "alias": alias})
                else:
                    unmatched[str(recipient)] = unmatched.get(str(recipient), 0) + 1

    return _BackfillPlan(
        threads=threads,
        in_thread_eids=in_thread_eids,
        by_volk_rows=by_volk_rows,
        to_volk_rows=to_volk_rows,
        unmatched_agents=unmatched,
    )


@traces(spec="106", reqs=["REQ-106-008"])
def check_hub(
    repo: BoltSyncRepository,
    *,
    project: str,
    hub_label: str = LABEL_PROJECT,
) -> str | None:
    """Step 0 — read-only hub guard.

    Returns an abort reason (and the caller must make zero writes) unless
    exactly one hub node exists for ``project``. The backfill never creates
    a Project — a missing/duplicated hub is a real signal, not something to
    paper over (the ``run_component_id_migration`` abort precedent).
    """
    rows = repo.execute_read(
        f"MATCH (p:`{hub_label}` {{slug: $project}}) RETURN count(p) AS hubs",
        {"project": project},
    )
    hubs = rows[0]["hubs"] if rows else 0
    if hubs != 1:
        return (
            f"expected exactly one {hub_label} hub for slug={project!r}, "
            f"found {hubs}"
        )
    return None


@traces(spec="106", reqs=["REQ-106-008"])
def _batched(seq: list, size: int = 500):
    for i in range(0, len(seq), size):
        yield seq[i : i + size]


@traces(spec="106", reqs=["REQ-106-008"])
def run_backfill(
    repo: BoltSyncRepository,
    *,
    project: str,
    dry_run: bool,
    batch_size: int = 500,
    source_labels: Mapping[str, tuple[str | None, str | None]] | None = None,
    volk_label: str = LABEL_VOLK,
    hub_label: str = LABEL_PROJECT,
) -> tuple[BackfillStats, str | None]:
    """Execute (or, when ``dry_run``, only report) the Thread backfill.

    Returns ``(stats, abort_reason)``. ``abort_reason`` is set (and
    ``stats`` is empty, and nothing is written) only when the Step-0 hub
    guard fails. MERGE-only, batched, idempotent — see the module docstring.
    """
    abort_reason = check_hub(repo, project=project, hub_label=hub_label)
    if abort_reason is not None:
        return BackfillStats(), abort_reason

    plan = plan_backfill(
        repo, project=project, source_labels=source_labels, volk_label=volk_label
    )
    stats = BackfillStats(
        threads=len(plan.threads),
        has_thread_edges=len(plan.threads),
        in_thread_by_label={
            label: len(eids) for label, eids in plan.in_thread_eids.items()
        },
        by_volk_edges=len(plan.by_volk_rows),
        to_volk_edges=len(plan.to_volk_rows),
        unmatched_agents=dict(plan.unmatched_agents),
    )
    if dry_run:
        return stats, None

    # Step 2 — apply the Thread constraint before any MERGE (scoped to the
    # Thread statement only; the DDL for every other canonical label is
    # applied separately by kg_schema.py). Safe to run standalone on a
    # fresh instance without a prior `kg_schema.py apply`.
    apply(repo, [stmt for stmt in get_all_ddl() if f":{LABEL_THREAD})" in stmt])

    # Step 3 — MERGE Threads + hub edge.
    thread_rows = [
        {"id": canonical_id, "thread_id": tid}
        for tid, canonical_id in plan.threads.items()
    ]
    for batch in _batched(thread_rows, batch_size):
        repo.execute_write(
            f"MATCH (p:`{hub_label}` {{slug: $project}}) "
            f"UNWIND $rows AS row "
            f"MERGE (t:`{LABEL_THREAD}` {{id: row.id}}) "
            f"SET   t.thread_id = row.thread_id, t.project = $project "
            f"MERGE (p)-[:{REL_HAS_THREAD}]->(t)",
            {"project": project, "rows": batch},
        )

    # Step 4 — MERGE IN_THREAD, per label, over elementId batches. Runs
    # after Step 3: it MATCHes (never MERGEs) the Thread, so a missing
    # Thread silently drops the row rather than creating an unanchored one.
    for eids in plan.in_thread_eids.values():
        for batch in _batched(eids, batch_size):
            repo.execute_write(
                f"MATCH (n) WHERE elementId(n) IN $eids "
                f"MATCH (t:`{LABEL_THREAD}` {{id: $project + ':' + n.thread_id}}) "
                f"MERGE (n)-[:{REL_IN_THREAD}]->(t)",
                {"eids": batch, "project": project},
            )

    # Step 5 — MERGE BY_VOLK / TO_VOLK. Rows for unresolved agents are
    # simply absent — never a fabricated Volk.
    for batch in _batched(plan.by_volk_rows, batch_size):
        repo.execute_write(
            f"UNWIND $rows AS row "
            f"MATCH (n) WHERE elementId(n) = row.eid "
            f"MATCH (v:`{volk_label}` {{alias: row.alias}}) "
            f"MERGE (n)-[:{REL_BY_VOLK}]->(v)",
            {"rows": batch},
        )
    for batch in _batched(plan.to_volk_rows, batch_size):
        repo.execute_write(
            f"UNWIND $rows AS row "
            f"MATCH (n) WHERE elementId(n) = row.eid "
            f"MATCH (v:`{volk_label}` {{alias: row.alias}}) "
            f"MERGE (n)-[:{REL_TO_VOLK}]->(v)",
            {"rows": batch},
        )

    return stats, None


def _driver(args: argparse.Namespace, env: Mapping[str, str] | None = None):
    env = env if env is not None else os.environ
    uri = args.uri or resolve_uri(env=env)
    user = args.user or ""
    password = args.password or ""
    return GraphDatabase.driver(uri, auth=(user, password))


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="backfill_thread_links",
        description=(
            "Backfill Thread correlation-anchor nodes + IN_THREAD/HAS_THREAD/"
            "BY_VOLK/TO_VOLK edges (session-254 Option A)."
        ),
    )
    parser.add_argument("--uri", default=None)
    parser.add_argument("--user", default=None)
    parser.add_argument("--password", default=None)
    parser.add_argument("--project", default="orchestrata-ultimate")
    parser.add_argument("--batch-size", type=int, default=500)
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Actually MERGE the Thread ontology. Default is a dry run "
        "(plan reads only, write count 0) — house convention for every "
        "KG-mutating tool in this repo.",
    )
    args = parser.parse_args(argv)

    # Standalone usability: fill NEO4J_* from the repo .env when the
    # caller's environment lacks them (override=False — ambient env always
    # wins), mirroring canonical_conformance.main / tests/conftest.py.
    from dotenv import load_dotenv

    load_dotenv(Path(__file__).resolve().parent.parent / ".env", override=False)

    driver = _driver(args)
    try:
        driver.verify_connectivity()
        repo = BoltSyncRepository(driver=driver)
        stats, abort_reason = run_backfill(
            repo,
            project=args.project,
            dry_run=not args.apply,
            batch_size=args.batch_size,
        )
    finally:
        driver.close()

    if abort_reason is not None:
        print(f"[ABORT] {abort_reason}")
        return 1

    verb = "DID" if args.apply else "WOULD"
    print(f"[{verb}] MERGE Thread nodes     : {stats.threads}")
    print(f"[{verb}] MERGE HAS_THREAD edges  : {stats.has_thread_edges}")
    print(f"[{verb}] MERGE IN_THREAD edges   : {stats.in_thread_edges}")
    for label, count in sorted(stats.in_thread_by_label.items()):
        print(f"           - {label}: {count}")
    print(f"[{verb}] MERGE BY_VOLK edges     : {stats.by_volk_edges}")
    print(f"[{verb}] MERGE TO_VOLK edges     : {stats.to_volk_edges}")
    if stats.unmatched_agents:
        total_unmatched = sum(stats.unmatched_agents.values())
        print(
            f"[{verb}] unmatched agent strings : {len(stats.unmatched_agents)} "
            f"distinct ({total_unmatched} occurrences)"
        )
        for agent, count in sorted(
            stats.unmatched_agents.items(), key=lambda kv: (-kv[1], kv[0])
        ):
            print(f"           - {agent!r}: {count}")
    return 0


if __name__ == "__main__":
    sys.exit(main())