orchestrata_ultimate.factory.detectors.engine

Reference › orchestrata_ultimate › factory › detectors › engine

engine runs the C3 detector registry over a document (SPEC-033 row B22, requirement REQ-033-C3). run_detectors builds a rule context from the text with build_context and runs every rule in ALL_RULES from rules, minus any rule ids in disabled. When a Blacklist is passed, findings whose fingerprint was dismissed are moved to the report's dismissed list instead of findings. It returns a DetectorReport from models. tier1_gate returns the undismissed BLOCKING findings of a report; this is what the decision phase and the detector CLI's gate use. Blacklist is a persistent, append-only set of dismissed fingerprints stored as JSONL under an fcntl lock; dismiss is idempotent, contains tests one fingerprint and fingerprints reads the whole set in one locked read. The locking pattern is copied from the assumption ledger rather than imported, because that class is not public. Tests are in tests/factory/test_detector_engine.py.

At a glance

Class diagram

Dependency diagram

API

Classes

Blacklist

Decorators: traces(spec='033', reqs=['REQ-033-C3']) · Line 36

Persistent, append-only, lock-guarded set of dismissed fingerprints.

Functions

Source

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

"""Detector engine: run the rule registry, gate on Tier 1, dismiss by fingerprint.

:class:`Blacklist` is a persistent, JSONL, fcntl-locked dismissed-fingerprint
store -- the locking pattern mirrors
:class:`orchestrata_ultimate.factory.assumptions.ledger._LockedJsonlFile`
(that class is not exported publicly, so it is reimplemented here rather than
imported, per the task brief).
"""

from __future__ import annotations

import fcntl
import json
import os
from collections.abc import Set as AbstractSet
from pathlib import Path
from typing import IO

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.detectors.models import (
    DetectorReport,
    Finding,
    RuleMeta,
    Tier,
)
from orchestrata_ultimate.factory.detectors.rules import (
    ALL_RULES,
    RuleFn,
    build_context,
)

_LockFile = IO[str]


@traces(spec="033", reqs=["REQ-033-C3"])
class Blacklist:
    """Persistent, append-only, lock-guarded set of dismissed fingerprints.

    Backed by a JSONL file (one ``{"fingerprint": "..."}`` record per line);
    concurrent appends from multiple threads/processes serialize on an
    exclusive ``fcntl.flock`` of a sidecar ``.lock`` file, matching the
    pattern used by the B13 assumption ledger.
    """

    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.lock_path = self.path.with_name(self.path.name + ".lock")

    def _acquire(self) -> _LockFile:
        fp = open(self.lock_path, "a")
        fcntl.flock(fp.fileno(), fcntl.LOCK_EX)
        return fp

    @staticmethod
    def _release(fp: _LockFile) -> None:
        try:
            fcntl.flock(fp.fileno(), fcntl.LOCK_UN)
        finally:
            fp.close()

    def dismiss(self, fingerprint: str) -> None:
        """Append ``fingerprint`` to the blacklist (idempotent -- may repeat)."""
        lock = self._acquire()
        try:
            with open(self.path, "a", encoding="utf-8") as f:
                f.write(json.dumps({"fingerprint": fingerprint}) + "\n")
                f.flush()
                os.fsync(f.fileno())
        finally:
            self._release(lock)

    def contains(self, fingerprint: str) -> bool:
        """True if ``fingerprint`` has been dismissed."""
        return fingerprint in self._read_all()

    def fingerprints(self) -> frozenset[str]:
        """One locked read of the full dismissed set.

        Scans call this once and test membership against the snapshot —
        per-finding ``contains`` would re-lock and re-parse the whole file
        N times per run.
        """
        return frozenset(self._read_all())

    def _read_all(self) -> set[str]:
        lock = self._acquire()
        try:
            if not self.path.exists():
                return set()
            fingerprints: set[str] = set()
            with open(self.path, encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if line:
                        fingerprints.add(json.loads(line)["fingerprint"])
            return fingerprints
        finally:
            self._release(lock)


@traces(spec="033", reqs=["REQ-033-C3"])
def run_detectors(
    text: str,
    source_name: str,
    *,
    rules: dict[str, tuple[RuleMeta, RuleFn]] = ALL_RULES,
    disabled: AbstractSet[str] = frozenset(),
    blacklist: Blacklist | None = None,
    repo_root: Path | None = None,
) -> DetectorReport:
    """Run every enabled rule in ``rules`` over ``text``.

    Dismissed findings (fingerprint present in ``blacklist``) land in
    ``report.dismissed`` -- never silently dropped. ``repo_root`` is an
    optional override for rules that resolve references against the repo
    (``incomplete-reference``); it defaults to the live project root.
    """
    ctx = build_context(text, source_name, repo_root=repo_root)
    dismissed_fps: frozenset[str] = (
        blacklist.fingerprints() if blacklist is not None else frozenset()
    )
    findings: list[Finding] = []
    dismissed: list[Finding] = []
    for rule_id, (_meta, rule_fn) in rules.items():
        if rule_id in disabled:
            continue
        for finding in rule_fn(ctx):
            if finding.fingerprint in dismissed_fps:
                dismissed.append(finding)
            else:
                findings.append(finding)
    return DetectorReport(
        findings=findings, dismissed=dismissed, files_scanned=[source_name]
    )


@traces(spec="033", reqs=["REQ-033-C3"])
def tier1_gate(report: DetectorReport) -> list[Finding]:
    """Return the undismissed BLOCKING findings in ``report``.

    Callers exit non-zero on a non-empty result -- this function only
    enumerates; it never prints "ready" (ADR-0011's completeness-verdict
    boundary applies to every layer, not just the detectors themselves).
    """
    return [f for f in report.findings if f.tier == Tier.BLOCKING]