orchestrata_ultimate.factory.promotion.store

Reference › orchestrata_ultimate › factory › promotion › store

store is the graph-backed store of namespaced facts, with promotion and garbage collection (SPEC-033 section 4, row B4a, ADR-0004 and ADR-0019). ADR-0004 keys facts by (worktree, commit) and treats a merge as fact promotion; ADR-0019 requires gate reads to be causally consistent through a snapshot and a watermark rather than replicas. PromotionStore is built over the shared graph driver factory in core.graph_factory, adding no second driver path, and takes a governed-write EventLedger. It is label-agnostic: it matches fact nodes only on their namespace property, so it works with any writer that tags nodes that way, including the knowledge-graph syncer.

• write_fact creates a fact node under a namespace; facts_in returns the properties of the facts in a namespace.
• promote flips the matching facts' namespace property from the source key to the trunk key inside one transaction, then records a governed event in the ledger. A flip is itself the removal of the source namespace. It takes a gate_token and the author and promoter identities and applies the no_self_merge check. It returns a PromotionRecord with a watermark.
• await_watermark polls a session factory until the watermark is visible in the read view, with a backoff, or raises WatermarkTimeout.
• gc deletes eligible namespaces for worktrees that no longer exist, following a RetentionPolicy (see models); list_namespaces lists the control nodes.

Under the root-connectivity ruling, every FactNamespace control node is linked to the root Project node with HAS_FACT_NAMESPACE and every fact hangs off its control node with HAS_FACT. The merge queue, the scheduler and the scripts.factory_pipeline and scripts.factory_promote CLIs import it; tests are in tests/factory/test_promotion_store.py.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

WatermarkTimeout

Bases: Exception · Line 144

Raised by :func:await_watermark when the watermark is not honored in time.

_WatermarkNotVisible

Bases: RuntimeError · Line 154

Internal park signal — the watermark is not in the read view yet.

PromotionStore

Line 237

Graph-backed namespaced-fact store: writes, promotion, and GC (B4a).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/promotion/store.py

"""PromotionStore -- namespaced facts, fact-promotion, and GC (B4a).

SPEC-033 §4, ADR-0004 ("facts keyed by `(worktree, commit)`; merge is fact
promotion") and ADR-0019 ("gate reads require causal consistency: snapshot +
watermark, not replicas"). Built over the EXISTING shared graph driver
factory (:mod:`orchestrata_ultimate.core.graph_factory`) -- no second driver
path.

This store is **label-agnostic**: it matches fact nodes purely on their
``namespace`` property, so it composes with any writer that tags nodes that
way -- including :class:`orchestrata_ultimate.knowledge_graph.sync.GraphSyncer`,
whose ``full_sync``/``incremental_sync`` gained an additive, optional
``namespace=`` parameter for exactly this purpose (see that module's
``_tag_namespace``). :meth:`PromotionStore.write_fact` is the generic
namespaced-write seam used directly by this package's own tests and any
caller that is not the AST syncer.

``promote()`` is a **flip**, not a copy-then-delete: it rewrites the
matching facts' ``namespace`` property in place from the source key to the
trunk key, inside one transaction, then records a governed event in the
B1c ledger (:mod:`orchestrata_ultimate.factory.enforcement.ledger`). A flip
IS the deletion of the source namespace -- nothing remains tagged with the
old key once it commits.

Root-connectivity (session-254 ruling): EVERY node in the operational KG
must be reachable from the root ``:Project`` node
(:mod:`scripts.check_kg_connectivity`). Every ``:FactNamespace`` control
node this store MERGEs is hub-linked to ``:Project``
(``HAS_FACT_NAMESPACE``), and every fact node it owns hangs off its control
node (``HAS_FACT``) -- see :meth:`write_fact` and :meth:`promote`.
"""

from __future__ import annotations

import os
import re
import time
import uuid
from collections.abc import Callable, Iterable, Mapping
from contextlib import AbstractContextManager
from datetime import UTC, datetime
from typing import Any

from neo4j import Driver

from orchestrata_ultimate.core.graph_factory import (
    create_graph,
    resolve_kg_database,
)
from orchestrata_ultimate.factory.dispatch.gates import no_self_merge
from orchestrata_ultimate.factory.enforcement.ledger import EventLedger
from orchestrata_ultimate.factory.enforcement.models import WriterIdentity
from orchestrata_ultimate.factory.promotion.models import (
    FactNamespace,
    PromotionRecord,
    RetentionPolicy,
)
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    LABEL_FACT_NAMESPACE,
    LABEL_PROJECT,
    REL_HAS_FACT,
    REL_HAS_FACT_NAMESPACE,
)

# Node labels are not parameterizable in Cypher -- this bounds what
# `write_fact`'s caller-supplied `label` may contain before it is
# interpolated into a query string (test-scoped labels only, per project
# test-isolation convention: e.g. "TestFact_<test_id>").
_LABEL_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")

# The default project slug hub-linked writes MERGE onto when no explicit
# project is resolvable -- the same default the kernel's own KG writers use
# (e.g. kernel/pipeline_kg.py's sync_volks_to_kg). Read at call time via
# `_project_slug()`, never cached at import time (RUL-003: no bare
# `os.environ[...]`, no import-time constant).
_DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"

# The writer identity every promotion event is attributed to. Distinct from
# `factory.enforcement.bootstrap.HUMAN_BOOTSTRAP_WRITER`: a promotion event
# is machine-issued (gated by `gate_token`, see `promote()`), not a human
# bootstrap act, so it is never intended to pass `WriterRegistry.registered`
# -- B1a/B1c file-diff enforcement and B4a graph-fact promotion are separate
# governed-event streams that happen to share one ledger shape (ADR-0030).
_PROMOTION_WRITER = WriterIdentity(
    name="factory:promotion", binary_hash="promotion", config_hash="promotion"
)

# B4a.2 (ADR-0004): the exact node labels `knowledge_graph.sync.GraphSyncer.
# _tag_namespace` tags -- and therefore the only labels whose `id` can ever
# carry an `"ns::{namespace}::"` qualifier (see that method's docstring).
# Fixed/known set: Cypher cannot parameterize a MATCH's label, so `promote()`
# loops over this literal tuple rather than reading labels off live data.
_SYNCER_NODE_LABELS: tuple[str, ...] = (
    "Module",
    "Class",
    "Function",
    "Method",
    "Parameter",
)

# Edge specs among the syncer's namespace-scoped subgraph that `promote()`
# must recreate between TRUNK nodes (DETACH DELETE-ing the qualified
# originals would otherwise leave them dangling): (rel_type, from_label(s),
# to_label(s), from_is_also_qualified). `from_is_also_qualified=False` only
# for CONTAINS, whose source (Package) is never namespace-tagged (B4a's
# "Project/Package stay shared and unqualified" scoping decision) -- every
# other edge here connects two `_SYNCER_NODE_LABELS` members, which
# `_tag_namespace` always tags together (same module's whole owned subtree,
# one namespace per call), so requiring both sides on those is safe.
_SYNCER_EDGE_SPECS: tuple[tuple[str, str, str, bool], ...] = (
    ("CONTAINS", "Package", "Module", False),
    ("DEFINES", "Module", "Class", True),
    ("DEFINES", "Module", "Function", True),
    ("DEFINES", "Class", "Method", True),
    ("HAS_PARAMETER", "Function|Method", "Parameter", True),
    ("INHERITS", "Class", "Class", True),
    ("IMPORTS", "Module", "Module", True),
    ("CALLS", "Method|Function", "Method|Function", True),
)


def _utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


def _validate_label(label: str) -> str:
    if not _LABEL_RE.match(label):
        raise ValueError(
            f"invalid node label {label!r} -- must match {_LABEL_RE.pattern}"
        )
    return label


def _project_slug() -> str:
    """Resolve the root ``:Project`` slug hub-linked control nodes MERGE onto.

    Read at call time (RUL-003) so a caller/test that sets
    ``ORCHESTRATA_PROJECT_ID`` per-process is honored -- never cached at
    import time.
    """
    return os.environ.get("ORCHESTRATA_PROJECT_ID", _DEFAULT_PROJECT_SLUG)


class WatermarkTimeout(Exception):
    """Raised by :func:`await_watermark` when the watermark is not honored in time.

    ADR-0019 is fail-closed by construction: a caller catching this must
    PARK/retry the gate evaluation, never fall back to reading a possibly
    stale view. There is deliberately no "return the stale answer instead"
    path anywhere in this module.
    """


class _WatermarkNotVisible(RuntimeError):
    """Internal park signal — the watermark is not in the read view yet.

    Raised (and swallowed) inside :func:`await_watermark`'s poll loop so a
    not-yet-visible token takes exactly the same backoff path as a transient
    connection failure, and surfaces as :class:`WatermarkTimeout` if the
    deadline passes.
    """


def _new_watermark() -> str:
    """Mint an opaque, collision-free causal-consistency token (ADR-0019)."""
    return f"wm-{uuid.uuid4().hex}"


def await_watermark(
    session_factory: Callable[[str], AbstractContextManager[Any]],
    watermark: str,
    timeout_s: float,
    *,
    poll_interval_s: float = 0.05,
    max_poll_interval_s: float = 2.0,
) -> None:
    """Park (poll+backoff) until a read view actually SEES ``watermark``.

    ADR-0019: gate reads must be causally consistent with the commit under
    evaluation -- the read view's watermark must be >= the commit's
    fact-write. The ADR sanctions two mechanisms, "Neo4j bookmarks, or an
    explicit commit watermark over a read-through snapshot"; this is the
    second, and it is the one that holds on BOTH engines.

    The first was implemented as "open a bookmarked session and run
    ``RETURN 1``" — which is only a freshness check on an engine that
    enforces bookmarks. The orchgraph cluster returns no bookmarks at all
    (``last_bookmarks()`` is empty) and silently IGNORES any passed in, so
    after the 2026-07-26 cutover that probe returned immediately for every
    token, including an unreachable one: the fail-closed gate had become
    fail-open. Probing for the token's visibility instead is engine-
    independent — an unknown watermark is never visible, so it can only
    ever time out.

    Visibility is membership in the control node's ``watermarks`` LIST, not
    equality against a scalar: several promotions share one trunk control
    node per commit, and a scalar would let each new token erase the one an
    earlier caller is still parked on.

    ``session_factory(watermark)`` returns a context manager for a session;
    on an engine with bookmarks it may additionally bind them, which is
    strictly stronger and costs nothing here.

    Raises:
        WatermarkTimeout: The watermark was not visible within ``timeout_s``.
            Callers must treat this as a park/retry condition, not license
            to read a stale view (fail-closed, per ADR-0019).
    """
    deadline = time.monotonic() + timeout_s
    interval = poll_interval_s
    last_exc: Exception | None = None
    while True:
        try:
            with session_factory(watermark) as session:
                record = session.run(
                    f"MATCH (fn:{LABEL_FACT_NAMESPACE}) "
                    "WHERE $w IN coalesce(fn.watermarks, []) "
                    "RETURN count(fn) AS c",
                    {"w": watermark},
                ).single()
                seen = record is not None and record["c"] > 0
            if seen:
                return
            raise _WatermarkNotVisible(watermark)
        except Exception as exc:  # noqa: BLE001 -- any failure means "not yet"
            last_exc = exc
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise WatermarkTimeout(
                    f"watermark {watermark!r} not visible within {timeout_s}s: "
                    f"{last_exc!r}"
                ) from last_exc
            time.sleep(min(interval, remaining))
            interval = min(interval * 2, max_poll_interval_s)


class PromotionStore:
    """Graph-backed namespaced-fact store: writes, promotion, and GC (B4a).

    Constructed with an :class:`EventLedger` (B1c) -- every :meth:`promote`
    call appends a governed event to it. The graph connection is resolved
    exactly like every other real-driver caller in this codebase: pass
    ``driver=`` to inject one (tests, shared-driver reuse), or ``uri``/
    ``username``/``password`` to override env, or nothing to fall back to
    ``core.graph_factory.create_graph``'s own env resolution
    (``KG_ROUTING_URI``/``KG_URI``).
    """

    def __init__(
        self,
        ledger: EventLedger,
        *,
        driver: Driver | None = None,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
    ) -> None:
        self._ledger = ledger
        if driver is not None:
            self._driver = driver
            self._owns_driver = False
        else:
            auth = (
                (username, password)
                if username is not None and password is not None
                else None
            )
            self._driver = create_graph(uri, auth=auth)
            self._owns_driver = True

    @property
    def driver(self) -> Driver:
        """Raw ``neo4j.Driver`` escape hatch (e.g. to build a bookmarked
        ``session_factory`` for :func:`await_watermark`)."""
        return self._driver

    def session(self) -> Any:
        """Open a session bound to the operational KG database.

        Since the 2026-07-26 cutover the operational KG and the shared
        semantic/RAG graph are two DATABASES on one orchgraph cluster, so the
        URI no longer picks between them — ``database=`` does. A bare
        ``driver.session()`` here silently wrote promoted facts into the
        cluster default (the semantic graph) instead of ``orchkg``. Resolved
        at CALL time so per-test-worker ``KG_DATABASE`` values apply.

        Public because it is also the only correct way for a caller to VERIFY
        what this store wrote: a bare ``store.driver.session()`` reads the
        cluster default, which is a different graph.
        """
        return self._driver.session(database=resolve_kg_database())

    def close(self) -> None:
        """Close the underlying driver, but only if this store created it."""
        if self._owns_driver:
            self._driver.close()

    # ── Namespaced writes ────────────────────────────────────────────────

    def write_fact(
        self,
        ns: FactNamespace,
        label: str,
        properties: Mapping[str, Any],
    ) -> str | None:
        """Create one fact node labeled ``label`` under ``ns``'s namespace.

        Generic, label-agnostic write seam: real AST/trace facts instead
        reach namespacing via
        :meth:`orchestrata_ultimate.knowledge_graph.sync.GraphSyncer.full_sync`'s
        additive ``namespace=`` parameter. This method also MERGEs a
        ``:FactNamespace`` control node recording ``ns``'s first-write
        timestamp -- :meth:`gc` and :meth:`list_namespaces` read that
        control node (not the (potentially huge) fact-node population) to
        answer "how old is this namespace" / "what namespaces exist".

        Root-connectivity (session-254 ruling): in the SAME statement, the
        control node is hub-linked to the root ``:Project`` node
        (``HAS_FACT_NAMESPACE``) and the new fact node is hung off its
        control node (``HAS_FACT``) -- both must be reachable from
        ``:Project``, never a floating island (this was a real bug: these
        nodes previously had no edge at all, and leaked into the live graph
        more than once before being swept by hand).

        Returns:
            The ADR-0019 causal-consistency watermark for this write — an
            opaque token stamped onto the ``:FactNamespace`` control node in
            the SAME transaction as the fact, so a reader that can see the
            token is guaranteed to see the fact. Pass it to
            :func:`await_watermark`.
        """
        _validate_label(label)
        props = dict(properties)
        props["namespace"] = ns.key
        watermark = _new_watermark()

        with self.session() as session:

            def _txn(tx: Any) -> None:
                tx.run(
                    f"CREATE (n:{label} $props) "  # label validated above
                    "WITH n "
                    # MERGE keys on `key` -- the ONLY property the
                    # FactNamespace uniqueness constraint backs. MERGEing on
                    # {worktree, commit} (unconstrained/unindexed) instead
                    # is not atomic: two concurrent first-writes to the same
                    # namespace can both pass the "does not exist" check,
                    # both CREATE, and the loser hits
                    # Neo.ClientError.Schema.ConstraintValidationFailed on
                    # `key` instead of the MERGE just converging (reproduced
                    # 2026-07-26 under concurrent CLI invocations sharing a
                    # commit).
                    f"MERGE (fn:{LABEL_FACT_NAMESPACE} {{key: $key}}) "
                    "ON CREATE SET fn.worktree = $w, fn.commit = $c, fn.created_at = $now "
                    # APPEND, never overwrite: one control node accumulates a
                    # watermark per write/promotion, and a scalar property
                    # would let the newest one erase a token another caller is
                    # still parked on (see `await_watermark`).
                    "SET fn.watermarks = coalesce(fn.watermarks, []) + $watermark "
                    f"MERGE (p:{LABEL_PROJECT} {{slug: $project}}) "
                    f"MERGE (p)-[:{REL_HAS_FACT_NAMESPACE}]->(fn) "
                    f"MERGE (fn)-[:{REL_HAS_FACT}]->(n)",
                    {
                        "props": props,
                        "w": ns.worktree,
                        "c": ns.commit,
                        "key": ns.key,
                        "now": time.time(),
                        "project": _project_slug(),
                        "watermark": watermark,
                    },
                )

            session.execute_write(_txn)

        return watermark

    def facts_in(self, ns: FactNamespace) -> list[dict[str, Any]]:
        """Return every fact node's properties currently tagged with ``ns.key``.

        Used to prove isolation: a fact written under one namespace must be
        invisible when reading a different namespace (trunk or a sibling
        worktree's).
        """
        with self.session() as session:

            def _txn(tx: Any) -> list[dict[str, Any]]:
                result = tx.run(
                    "MATCH (n) WHERE n.namespace = $ns RETURN n AS node",
                    {"ns": ns.key},
                )
                return [dict(record["node"]) for record in result]

            facts: list[dict[str, Any]] = session.execute_read(_txn)
            return facts

    # ── Promotion (ADR-0004: merge = fact promotion) ───────────────────

    def promote(
        self,
        ns: FactNamespace,
        *,
        gate_token: str,
        author_identity: str | None = None,
        promoter_identity: str | None = None,
    ) -> PromotionRecord:
        """Promote ``ns``'s facts into the trunk namespace for its commit.

        Flips every fact node's ``namespace`` property from ``ns.key`` to
        ``FactNamespace.trunk(ns.commit).key`` in ONE transaction (a flip is
        itself the deletion of the source namespace -- nothing remains
        tagged with the old key once it commits), deletes ``ns``'s
        ``:FactNamespace`` control node, and appends a governed event to the
        B1c ledger.

        B4a.2 (ADR-0004): facts written by
        :meth:`orchestrata_ultimate.knowledge_graph.sync.GraphSyncer.
        full_sync`'s namespace-qualification path (``_SYNCER_NODE_LABELS``,
        whose ``id`` carries an ``"ns::{ns.key}::"`` prefix) are NOT a
        simple in-place flip -- their qualified ``id`` must be stripped back
        to the legacy (trunk) id, and that id is MERGEd (not created fresh):
        a prior promotion, or a trunk sync, may already own that id, and the
        newly-promoted node's properties win (last-promoted-wins, matching
        this method's existing flip semantics for generic facts). Every
        relationship among the promoted subgraph is recreated between the
        resulting TRUNK nodes before the qualified originals are deleted, so
        DEFINES/HAS_PARAMETER/INHERITS/IMPORTS/CALLS/CONTAINS edges are never
        left dangling. B4a's generic :meth:`write_fact` facts (``CREATE``,
        never namespace-id-qualified) are unaffected and keep the original
        flip path -- this all happens in the SAME transaction as that flip.

        Root-connectivity (session-254 ruling): a TRUNK ``:FactNamespace``
        control node is MERGEd and hub-linked to the root ``:Project`` node
        FIRST (idempotent -- a second promotion sharing the same commit
        reuses it), and every promoted fact's ``HAS_FACT`` edge -- both
        de-qualified syncer nodes and generic ``write_fact`` facts -- is
        re-pointed at it BEFORE the source control node is deleted, so
        nothing this method promotes is ever left unreachable from
        ``:Project``. The source control node's now-stale edges (including
        any HAS_FACT pointing at nodes that have since been flipped or
        deleted) vanish with it via DETACH DELETE.

        Args:
            ns: The source namespace to promote. Must not be a trunk
                namespace itself.
            gate_token: Opaque proof that the integration re-gate passed.
                B4a only requires this be non-empty and threads it into the
                ledger event's byproducts -- enforcing that it was actually
                ISSUED by a passing re-gate is B9/B21's job, not this
                store's (seam, not a TODO: no code here should be added to
                validate token provenance until that lane lands).
            author_identity: Optional B21 (ADR-0033) actor who authored the
                change being promoted.
            promoter_identity: Optional B21 (ADR-0033) actor performing this
                promotion. When BOTH ``author_identity`` and
                ``promoter_identity`` are given,
                :func:`orchestrata_ultimate.factory.dispatch.gates.no_self_merge`
                is checked BEFORE any write, and both identities are
                threaded into the ledger event's byproducts. When either is
                absent, behavior is byte-identical to promotion without
                identities -- B9 makes both mandatory at the scheduler
                (seam, not a TODO: no code here should be added to enforce
                that mandate until that lane lands).

        Raises:
            ValueError: ``gate_token`` is empty, ``ns`` is itself a trunk
                namespace, or ``ns`` has no facts (unknown/empty namespace)
                -- promotion never silently no-ops.
            SelfMergeRefused: Both identities are given and equal
                (ADR-0033 §1).
            IdentityUnknown: Both identities are given and either is empty
                (ADR-0032: undetermined is refused, never allowed).
        """
        if not gate_token:
            raise ValueError("promote() requires a non-empty gate_token")
        if ns.worktree == "trunk":
            raise ValueError(f"cannot promote a trunk namespace: {ns.key!r}")
        if author_identity is not None and promoter_identity is not None:
            no_self_merge(author_identity, promoter_identity)

        trunk = FactNamespace.trunk(ns.commit)
        qprefix = f"ns::{ns.key}::"
        watermark = _new_watermark()

        with self.session() as session:

            def _txn(tx: Any) -> tuple[int, int]:
                count_row = tx.run(
                    "MATCH (n) WHERE n.namespace = $src RETURN count(n) AS c",
                    {"src": ns.key},
                ).single()
                count = count_row["c"] if count_row else 0
                if count == 0:
                    raise ValueError(
                        f"cannot promote unknown/empty namespace {ns.key!r} "
                        "-- no facts found"
                    )

                # Root-connectivity (session-254): MERGE the TRUNK control
                # node and hub-link it to the Project BEFORE any fact is
                # re-pointed onto it -- every step below just MATCHes it
                # back by its (worktree, commit) identity. Idempotent: a
                # second promotion sharing this commit reuses the same node.
                # MERGE keys on `key` (the constrained/indexed property, not
                # the unconstrained (worktree, commit) pair) so concurrent
                # first-promotions of the same commit converge atomically
                # instead of racing into
                # Neo.ClientError.Schema.ConstraintValidationFailed on `key`
                # (reproduced 2026-07-26 under concurrent CLI invocations).
                tx.run(
                    f"MERGE (fn:{LABEL_FACT_NAMESPACE} {{key: $trunk_key}}) "
                    "ON CREATE SET fn.worktree = $trunk_wt, fn.commit = $commit, "
                    "fn.created_at = $now "
                    f"MERGE (p:{LABEL_PROJECT} {{slug: $project}}) "
                    f"MERGE (p)-[:{REL_HAS_FACT_NAMESPACE}]->(fn)",
                    {
                        "trunk_wt": trunk.worktree,
                        "commit": ns.commit,
                        "trunk_key": trunk.key,
                        "now": time.time(),
                        "project": _project_slug(),
                    },
                )

                # B4a.2: de-qualify + MERGE syncer-namespace facts onto their
                # trunk id FIRST (last-promoted-wins). The `id`/`namespace`
                # override MUST happen inside the SAME map-projection
                # expression (`n{.*, id: tid, namespace: $trunk}`), not via a
                # separate `SET t = properties(n)` followed by a fix-up SET:
                # the latter would transiently assign `n`'s still-QUALIFIED
                # id onto `t` (a second node with that id, since `n` is not
                # deleted yet), tripping the `Module`/`Class`/... `id`
                # uniqueness constraint mid-transaction. Each de-qualified
                # trunk node also gets its HAS_FACT edge from the TRUNK
                # control node MERGEd above (root-connectivity).
                dequalified_count = 0
                for label in _SYNCER_NODE_LABELS:
                    result = tx.run(
                        f"MATCH (n:{label}) "
                        "WHERE n.namespace = $src AND n.id STARTS WITH $qprefix "
                        "WITH n, substring(n.id, size($qprefix)) AS tid "
                        f"MERGE (t:{label} {{id: tid}}) "
                        "SET t = n{.*, id: tid, namespace: $trunk} "
                        "WITH n, t "
                        f"MATCH (fn:{LABEL_FACT_NAMESPACE} "
                        "{worktree: $trunk_wt, commit: $commit}) "
                        f"MERGE (fn)-[:{REL_HAS_FACT}]->(t) "
                        "RETURN count(n) AS c",
                        {
                            "src": ns.key,
                            "qprefix": qprefix,
                            "trunk": trunk.key,
                            "trunk_wt": trunk.worktree,
                            "commit": ns.commit,
                        },
                    ).single()
                    dequalified_count += result["c"] if result else 0

                # Recreate relationships between the newly-MERGEd trunk
                # nodes -- the DETACH DELETE below would otherwise leave
                # them dangling (ADR-0004 promotion contract).
                for (
                    rel_type,
                    from_label,
                    to_label,
                    from_qualified,
                ) in _SYNCER_EDGE_SPECS:
                    if from_qualified:
                        tx.run(
                            f"MATCH (a:{from_label})-[:{rel_type}]->(b:{to_label}) "
                            "WHERE a.namespace = $src AND a.id STARTS WITH $qprefix "
                            "AND b.namespace = $src AND b.id STARTS WITH $qprefix "
                            "WITH substring(a.id, size($qprefix)) AS atid, "
                            "     substring(b.id, size($qprefix)) AS btid "
                            f"MATCH (ta:{from_label} {{id: atid}}) "
                            f"MATCH (tb:{to_label} {{id: btid}}) "
                            f"MERGE (ta)-[:{rel_type}]->(tb)",
                            {"src": ns.key, "qprefix": qprefix},
                        )
                    else:
                        tx.run(
                            f"MATCH (a:{from_label})-[:{rel_type}]->(b:{to_label}) "
                            "WHERE b.namespace = $src AND b.id STARTS WITH $qprefix "
                            "WITH a, substring(b.id, size($qprefix)) AS btid "
                            f"MATCH (tb:{to_label} {{id: btid}}) "
                            f"MERGE (a)-[:{rel_type}]->(tb)",
                            {"src": ns.key, "qprefix": qprefix},
                        )

                # Delete the namespace-qualified originals now that their
                # content and relationships live on the trunk nodes.
                for label in _SYNCER_NODE_LABELS:
                    tx.run(
                        f"MATCH (n:{label}) "
                        "WHERE n.namespace = $src AND n.id STARTS WITH $qprefix "
                        "DETACH DELETE n",
                        {"src": ns.key, "qprefix": qprefix},
                    )

                # Generic (B4a `write_fact`) facts: the original in-place
                # flip, EXCLUDING the syncer-namespace facts already handled
                # (and deleted) above -- a no-op exclusion for real
                # `write_fact` facts (they never carry a "ns::"-qualified
                # id), but guards against ever double-processing a syncer
                # fact here. Each flipped fact also gets its HAS_FACT edge
                # re-pointed at the TRUNK control node -- the source control
                # node's stale HAS_FACT edge (and everything else touching
                # it) is removed below by DETACH DELETE.
                tx.run(
                    "MATCH (n) WHERE n.namespace = $src "
                    "AND (n.id IS NULL OR NOT n.id STARTS WITH $qprefix) "
                    "SET n.namespace = $trunk "
                    "WITH n "
                    f"MATCH (fn:{LABEL_FACT_NAMESPACE} "
                    "{worktree: $trunk_wt, commit: $commit}) "
                    f"MERGE (fn)-[:{REL_HAS_FACT}]->(n)",
                    {
                        "src": ns.key,
                        "trunk": trunk.key,
                        "qprefix": qprefix,
                        "trunk_wt": trunk.worktree,
                        "commit": ns.commit,
                    },
                )

                tx.run(
                    f"MATCH (fn:{LABEL_FACT_NAMESPACE} {{worktree: $w, commit: $c}}) "
                    "DETACH DELETE fn",
                    {"w": ns.worktree, "c": ns.commit},
                )
                # ADR-0019 watermark: APPENDED to the SURVIVING trunk control
                # node inside this same transaction, so a reader that sees
                # the token is guaranteed to see every fact this promotion
                # moved. See `await_watermark` for why a bookmark alone is
                # not a freshness check on the orchgraph engine.
                #
                # A LIST, not a scalar: every promotion of the same commit
                # lands on the SAME trunk control node, so a scalar
                # `fn.watermark = $watermark` meant promotion N+1 erased the
                # token promotion N's caller was still parked on. In the
                # scheduler that is the normal case — a DAG level promotes
                # several tasks against one base commit and then awaits each
                # token in turn — so every await but the last timed out after
                # 30s and failed the pipeline (reproduced by
                # test_pipeline_e2e_diamond_stale_duplicate_and_gc).
                tx.run(
                    f"MATCH (fn:{LABEL_FACT_NAMESPACE} "
                    "{worktree: $trunk_wt, commit: $commit}) "
                    "SET fn.watermarks = coalesce(fn.watermarks, []) + $watermark",
                    {
                        "trunk_wt": trunk.worktree,
                        "commit": ns.commit,
                        "watermark": watermark,
                    },
                )
                return count, dequalified_count

            fact_count, dequalified_count = session.execute_write(_txn)

        # `dequalified_count` is only ADDED when non-zero: a promotion of
        # purely-generic `write_fact` facts (no syncer-namespace ids
        # involved) must keep the pre-B4a.2 byproducts shape byte-identical
        # (regression-pinned: `test_promote_omitted_identities_is_legacy_
        # behavior` asserts `{"gate_token", "fact_count"}` exactly).
        byproducts = {"gate_token": gate_token, "fact_count": str(fact_count)}
        if dequalified_count:
            byproducts["dequalified_count"] = str(dequalified_count)
        if author_identity is not None and promoter_identity is not None:
            byproducts["author_identity"] = author_identity
            byproducts["promoter_identity"] = promoter_identity

        record = self._ledger.record(
            writer=_PROMOTION_WRITER,
            materials={"namespace": ns.key},
            products={"namespace": trunk.key},
            worktree=ns.worktree,
            commit=ns.commit,
            byproducts=byproducts,
        )

        return PromotionRecord(
            namespace=trunk,
            promoted_at=_utc_now_iso(),
            commit=ns.commit,
            fact_count=fact_count,
            ledger_event_id=record.record_id,
            watermark=watermark,
        )

    # ── GC (ADR-0004: abandoned-worktree retention) ────────────────────

    def gc(
        self, policy: RetentionPolicy, live_worktrees: Iterable[str]
    ) -> list[FactNamespace]:
        """Delete GC-eligible namespaces for worktrees no longer present.

        A namespace is only ever a GC *candidate* if its worktree is not in
        ``live_worktrees`` (the caller's live `git worktree list` -- this
        store makes no git calls itself). Among candidates, the newest
        ``policy.keep_last_n`` always survive regardless of age; the rest
        are deleted only if ALSO older than ``policy.max_age_seconds`` --
        both bounds must be violated. The trunk namespace is never a
        candidate at all, regardless of ``live_worktrees`` content.

        Deployment contract (ADR-0004, ruled 2026-07-24): ONE factory per
        graph database. This sweep judges every non-trunk namespace in the
        graph against the caller's worktree list, so a second factory
        sharing the database would have its in-flight namespaces deleted.
        Isolation is by database, not by key -- gc is deliberately not
        repo-scoped.

        Returns:
            The namespaces actually deleted.
        """
        live = set(live_worktrees)
        now = time.time()

        with self.session() as session:

            def _read(tx: Any) -> list[dict[str, Any]]:
                result = tx.run(
                    "MATCH (fn:FactNamespace) WHERE fn.worktree <> 'trunk' "
                    "RETURN fn.worktree AS worktree, fn.commit AS commit, "
                    "fn.created_at AS created_at "
                    "ORDER BY fn.created_at DESC"
                )
                return [dict(record) for record in result]

            rows = session.execute_read(_read)

        dead = [r for r in rows if r["worktree"] not in live]
        keep_keys = {(r["worktree"], r["commit"]) for r in dead[: policy.keep_last_n]}

        deleted: list[FactNamespace] = []
        for row in dead:
            key_tuple = (row["worktree"], row["commit"])
            if key_tuple in keep_keys:
                continue
            age = now - row["created_at"]
            if age > policy.max_age_seconds:
                candidate = FactNamespace(
                    worktree=row["worktree"], commit=row["commit"]
                )
                self._delete_namespace(candidate)
                deleted.append(candidate)

        return deleted

    def _delete_namespace(self, ns: FactNamespace) -> None:
        """Delete every fact node under ``ns`` plus its control node."""
        with self.session() as session:

            def _txn(tx: Any) -> None:
                tx.run(
                    "MATCH (n) WHERE n.namespace = $ns DETACH DELETE n",
                    {"ns": ns.key},
                )
                tx.run(
                    "MATCH (fn:FactNamespace {worktree: $w, commit: $c}) "
                    "DETACH DELETE fn",
                    {"w": ns.worktree, "c": ns.commit},
                )

            session.execute_write(_txn)

    # ── Introspection ────────────────────────────────────────────────────

    def list_namespaces(self) -> list[dict[str, Any]]:
        """Return every currently-known namespace's control-node properties."""
        with self.session() as session:

            def _txn(tx: Any) -> list[dict[str, Any]]:
                result = tx.run(
                    "MATCH (fn:FactNamespace) "
                    "RETURN fn.worktree AS worktree, fn.commit AS commit, "
                    "fn.key AS key, fn.created_at AS created_at "
                    "ORDER BY fn.created_at DESC"
                )
                return [dict(record) for record in result]

            namespaces: list[dict[str, Any]] = session.execute_read(_txn)
            return namespaces