orchestrata_ultimate.factory.detectors.c1_bridge

Reference › orchestrata_ultimate › factory › detectors › c1_bridge

c1_bridge connects the Tier-3 detectors to the assumption ledger. Tier 3 findings are questions, never verdicts (ADR-0029). questions_to_assumptions takes a DetectorReport from models, a design node and an AssumptionLedger, and turns each undismissed finding's question into a PENDING Assumption through the public ledger API. From there the ordinary cascade and the DoR check take over; the module duplicates none of that machinery. The scripts.factory_detectors CLI imports it, and tests/factory/test_detector_c1_bridge.py covers it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Bridge Tier-3 detector questions into the SPEC-033 C1 assumption ledger.

Tier 3 never produces a verdict (ADR-0029) -- its findings are questions, and
the only thing this module does with them is turn each undismissed one into a
``PENDING`` :class:`~orchestrata_ultimate.factory.assumptions.models.Assumption`
via the existing B13 public API. From there the ordinary C1 cascade (stage 0
KB lookup, stage 1 prefilter, stage 2 adjudication, stage 3 human) and
:func:`~orchestrata_ultimate.factory.assumptions.dor.check_dor` handle it --
this module does not duplicate any of that machinery.
"""

from __future__ import annotations

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.assumptions.ledger import AssumptionLedger
from orchestrata_ultimate.factory.assumptions.models import Assumption
from orchestrata_ultimate.factory.detectors.models import DetectorReport, Tier


@traces(spec="033", reqs=["REQ-033-C3"])
def questions_to_assumptions(
    report: DetectorReport, design_node: str, ledger: AssumptionLedger
) -> list[Assumption]:
    """Turn each undismissed Tier-3 finding's question into a PENDING assumption.

    Idempotent: an :class:`Assumption`'s id is deterministic
    (``fingerprint(text) + design_node``, per :meth:`Assumption.new`), so a
    finding whose question text already produced an assumption at this
    design node resolves to the same id -- it is skipped rather than raising
    on the ledger's duplicate guard, making a re-run over unchanged text add
    nothing.
    """
    existing_ids = {a.id for a in ledger.assumptions()}
    added: list[Assumption] = []
    for finding in report.findings:
        if finding.tier != Tier.QUESTION or finding.question is None:
            continue
        assumption = Assumption.new(
            text=finding.question,
            design_node=design_node,
            author_identity=f"detector:{finding.rule_id}",
        )
        if assumption.id in existing_ids:
            continue
        ledger.add(assumption)
        existing_ids.add(assumption.id)
        added.append(assumption)
    return added