<USER_REQUEST>
Meridian evidence init · PY

"""Phase 11.1 — Evidence Engine.

 

CapabilityClaim (Phase 10) becomes Evidence: one immutable, hash-

chained record per claim, with a blended confidence and a verifiable

tamper-detection chain.

 

Confidence is exactly two real, independent factors multiplied

together — not three. A Python detector's rule_confidence and a YAML

rule's base_confidence are the SAME number in this architecture

(CapabilityClaim.rule_confidence) — multiplying it in twice would

double-count one signal, not add a real third one. See models.py for

the actual computation.

 

evidence_id is derived purely from claim_id (itself derived from

rule_id + capability_id + site_id, all stable across re-scans of an

unchanged repository per Phase 10's design) — so evidence_id is stable

across repeated scans, by construction, not by coincidence. The hash

chain's hash/previous_hash fields are NOT required to match across

separate scan runs (each scan is its own audit event with its own

timestamp) — only evidence_id is the stable cross-scan identity.

"""

Meridian evidence models · PY

"""Phase 11.1 — Evidence model.

 

Schema approved by Karthik with two additions (site_id, rule_id) over

the original sketch, and IRLocation used in place of separate

file_path/start_line/end_line fields — the same type Fact and

CapabilityClaim already use everywhere else in this codebase, with

the same audit-independence properties.

"""

from __future__ import annotations

 

import hashlib

from dataclasses import dataclass

 

from meridian.models.ir import IRLocation

 

GENESIS_HASH = "0" * 64

 

 

def stable_evidence_id(claim_id: str) -> str:

    """Derived purely from claim_id — itself a stable hash of

    (rule_id, capability_id, site_id), none of which change across

    repeated scans of an unchanged repository. Re-hashed with a

    distinguishing prefix so evidence-space IDs and claim-space IDs

    never collide even though they're derived from the same input."""

    return hashlib.sha256(f"evidence|{claim_id}".encode("utf-8")).hexdigest()[:24]

 

 

def compute_record_hash(

    evidence_id: str, capability_id: str, site_id: str | None, rule_id: str,

    supporting_fact_ids: tuple[str, ...], confidence: float, location: IRLocation,

    created_at: str, previous_hash: str,

) -> str:

    """SHA-256 over this record's full content plus the prior record's

    hash — the chain link. created_at is included deliberately: each

    scan is its own audit event, so the resulting hash legitimately

    differs between separate runs of an otherwise-unchanged repository.

    That's expected and correct — evidence_id, not this hash, is the

    stable cross-scan identity (see stable_evidence_id)."""

    payload = "|".join([

        evidence_id, capability_id, site_id or "", rule_id,

        ",".join(supporting_fact_ids), f"{confidence:.6f}", str(location),

        created_at, previous_hash,

    ])

    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

 

 

@dataclass(frozen=True, slots=True)

class Evidence:

    evidence_id:           str

    capability_id:         str

    site_id:               str | None

    rule_id:                str

    supporting_fact_ids:   tuple[str, ...]

    confidence:             float

    location:               IRLocation

    created_at:             str

    hash:                   str

    previous_hash:          str

 
Meridian evidence engine · PY

"""Phase 11.1 — Evidence Engine.

 

build_evidence_ledger() is the one function this module exists for:

list[CapabilityClaim] + FactStore -> list[Evidence], one Evidence per

claim, chained in a deterministic order so the chain's *structure* is

reproducible across runs even though hash *values* legitimately differ

(created_at is real wall-clock time — see models.py).

"""

from __future__ import annotations

 

from datetime import datetime, timezone

 

from meridian.detectors.models import CapabilityClaim

from meridian.evidence.models import GENESIS_HASH, Evidence, compute_record_hash, stable_evidence_id

from meridian.facts.models import FactStore

 

 

def build_evidence_ledger(claims: list[CapabilityClaim], facts: FactStore) -> list[Evidence]:

    fact_by_id = {f.fact_id: f for f in facts.all()}

 

    # Deterministic chain order — same set of claims always produces

    # the same chain structure (which record links to which) regardless

    # of the order claims happened to be generated in. claim_id is

    # itself stable (Phase 10), so sorting by it is reproducible.

    sorted_claims = sorted(claims, key=lambda c: c.claim_id)

 

    ledger: list[Evidence] = []

    previous_hash = GENESIS_HASH

 

    for claim in sorted_claims:

        supporting_facts = [fact_by_id[fid] for fid in claim.supporting_fact_ids if fid in fact_by_id]

        mean_fact_confidence = (

            sum(f.confidence for f in supporting_facts) / len(supporting_facts)

            if supporting_facts else 0.0

        )

        confidence = mean_fact_confidence * claim.rule_confidence

 

        evidence_id = stable_evidence_id(claim.claim_id)

        created_at = datetime.now(timezone.utc).isoformat()

        record_hash = compute_record_hash(

            evidence_id, claim.capability_id, claim.site_id, claim.rule_id,

            claim.supporting_fact_ids, confidence, claim.location, created_at, previous_hash,

        )

 

        ledger.append(Evidence(

            evidence_id=evidence_id, capability_id=claim.capability_id, site_id=claim.site_id,

            rule_id=claim.rule_id, supporting_fact_ids=claim.supporting_fact_ids,

            confidence=confidence, location=claim.location, created_at=created_at,

            hash=record_hash, previous_hash=previous_hash,

        ))

        previous_hash = record_hash

 

    return ledger

 

 

def verify_ledger(ledger: list[Evidence]) -> list[str]:

    """Recomputes every record's hash and checks chain linkage. Returns

    a list of human-readable problems — empty list means the ledger is

    intact. Tampering with any field of any record, or reordering

    records, breaks the chain from that point forward, which is the

    entire point of chaining rather than hashing each record alone."""

    problems: list[str] = []

    expected_previous = GENESIS_HASH

 

    for i, record in enumerate(ledger):

        if record.previous_hash != expected_previous:

            problems.append(f"record {i} ({record.evidence_id}): previous_hash mismatch — chain broken")

 

        recomputed = compute_record_hash(

            record.evidence_id, record.capability_id, record.site_id, record.rule_id,

            record.supporting_fact_ids, record.confidence, record.location,

            record.created_at, record.previous_hash,

        )

        if recomputed != record.hash:

            problems.append(f"record {i} ({record.evidence_id}): hash does not match content — tampered")

 

        expected_previous = record.hash

 

    return problems

 
Meridian inventory init · PY

"""Phase 11.2 — Inventory Aggregation.

 

Evidence (Phase 11.1) rolls up into a RepositoryInventory: exactly 49

CapabilityState rows — every real capability_id from

ComplianceGraphResolver.all_capability_ids(), seeded up front, so a

capability with zero evidence appears as an explicit NOT_DETECTED row,

never silently absent. This is the whole reason CapabilityStatus is a

4-value enum and not a boolean — "True/False" can't distinguish

"checked and absent" from "never checked," and can't represent partial

coverage at all.

 

The coverage denominator problem, named honestly: by the time evidence

reaches this layer, the connection to "which SiteKinds this

capability's rule considered relevant" is gone — only Evidence

(capability_id, site_id, rule_id) remains. This module infers the

relevant SiteKind(s) from which kinds of sites actually produced

evidence, then uses ALL sites of those same kinds in the repository as

the coverage denominator. Reasonable and derivable from what's

available, not a perfect proxy for what a rule's own site_kinds list

would have said directly — see aggregator.py's docstring for the

specific case (a rule spanning multiple site_kinds) where this

approximation is coarsest.

"""
Meridian inventory models · PY

"""Phase 11.2 — CapabilityStatus / CapabilityState / RepositoryInventory.

 

Schema exactly as approved: 4-value status enum (no boolean), no 5th

status added beyond what was specified.

"""

from __future__ import annotations

 

from dataclasses import dataclass

from enum import Enum

 

 

class CapabilityStatus(str, Enum):

    DETECTED            = "detected"

    PARTIALLY_DETECTED  = "partially_detected"

    NOT_DETECTED        = "not_detected"

    INCONCLUSIVE        = "inconclusive"

 

 

@dataclass(frozen=True, slots=True)

class CapabilityState:

    capability_id:   str

    status:           CapabilityStatus

    confidence:       float             # mean confidence across this capability's Evidence; 0.0 if none

    evidence_count:   int

    site_count:       int               # distinct sites with evidence; 0 for repo-wide-only capabilities

    coverage:         float             # see aggregator.py for the denominator logic and its honest limits

    evidence_ids:     tuple[str, ...]

 

 

@dataclass(frozen=True, slots=True)

class RepositoryInventory:

    repository_path:    str

    generated_at:        str

    capability_states:  tuple[CapabilityState, ...]   # exactly len(resolver.all_capability_ids())

 

    def get(self, capability_id: str) -> CapabilityState | None:

        """Linear scan — fine at this scale. capability_states is a

        fixed, small set (currently 49); this isn't FactStore/SiteStore,

        which index because they handle thousands of items per scan."""

        for state in self.capability_states:

            if state.capability_id == capability_id:

                return state

        return None

 
Meridian inventory aggregator · PY

"""Phase 11.2 — InventoryAggregator.

 

aggregate_inventory() is the one function this module exists for:

list[Evidence] + SiteStore + ComplianceGraphResolver -> RepositoryInventory.

 

Status decision logic, the part with real judgment calls in it:

 

    evidence_count == 0           -> NOT_DETECTED

    mean_confidence < threshold   -> INCONCLUSIVE  (evidence exists, too weak to trust)

    coverage >= 1.0                -> DETECTED

    0 < coverage < 1.0             -> PARTIALLY_DETECTED

 

`threshold` is configurable via AggregationConfig, per Karthik's

requirement — never hardcoded inline.

 

Coverage denominator, stated precisely: for a capability whose evidence

is entirely repo-wide (every Evidence.site_id is None — e.g.

Consent_Management, Key_Management, which have no site_kinds in their

rules), coverage is binary: 1.0 if any evidence exists, 0.0 otherwise

— approved as the correct behavior for repository-scoped capabilities.

 

For a capability with at least one site-scoped Evidence, coverage is

computed by INFERRING the relevant SiteKind(s) from which kinds of

sites actually produced evidence, then dividing distinct evidenced

sites by the total count of sites of those same kinds in the

repository. This is reasonable and derivable from what's available at

this layer (Evidence no longer carries the originating rule's

site_kinds list — only Phase 10's Rule object had that, and it's gone

by the time Evidence exists) — but it has one named, honest rough edge:

a rule spanning MULTIPLE site_kinds (e.g. authentication.yaml lists

both HTTP_ROUTE_SITE and AUTHENTICATION_SITE) sums total sites across

ALL inferred kinds as the denominator, even if evidence only actually

touches one of them in practice. This can understate coverage for such

capabilities. Not fixed here because doing so would require Evidence

to carry the rule's declared site_kinds forward — a real schema change

warranting its own decision, not something to slip in silently.

"""

from __future__ import annotations

 

from collections import Counter, defaultdict

from dataclasses import dataclass

from datetime import datetime, timezone

 

from meridian.compliance.resolver import ComplianceGraphResolver

from meridian.evidence.models import Evidence

from meridian.inventory.models import CapabilityState, CapabilityStatus, RepositoryInventory

from meridian.sites.models import SiteKind, SiteStore

 

 

@dataclass(frozen=True)

class AggregationConfig:

    inconclusive_confidence_threshold: float = 0.5

 

 

def aggregate_inventory(

    evidence_list: list[Evidence],

    sites: SiteStore,

    resolver: ComplianceGraphResolver,

    repository_path: str,

    config: AggregationConfig | None = None,

) -> RepositoryInventory:

    config = config or AggregationConfig()

 

    site_kind_by_id = {s.site_id: s.site_kind for s in sites.all()}

    total_sites_by_kind: Counter[SiteKind] = Counter(s.site_kind for s in sites.all())

 

    evidence_by_capability: dict[str, list[Evidence]] = defaultdict(list)

    for e in evidence_list:

        evidence_by_capability[e.capability_id].append(e)

 

    states = []

    for capability_id in resolver.all_capability_ids():

        cap_evidence = evidence_by_capability.get(capability_id, [])

        states.append(_build_state(capability_id, cap_evidence, site_kind_by_id, total_sites_by_kind, config))

 

    return RepositoryInventory(

        repository_path=repository_path,

        generated_at=datetime.now(timezone.utc).isoformat(),

        capability_states=tuple(states),

    )

 

 

def _build_state(

    capability_id: str,

    cap_evidence: list[Evidence],

    site_kind_by_id: dict[str, SiteKind],

    total_sites_by_kind: Counter,

    config: AggregationConfig,

) -> CapabilityState:

    if not cap_evidence:

        return CapabilityState(

            capability_id=capability_id, status=CapabilityStatus.NOT_DETECTED,

            confidence=0.0, evidence_count=0, site_count=0, coverage=0.0, evidence_ids=(),

        )

 

    mean_confidence = sum(e.confidence for e in cap_evidence) / len(cap_evidence)

    evidence_ids = tuple(sorted(e.evidence_id for e in cap_evidence))

 

    site_scoped = [e for e in cap_evidence if e.site_id is not None]

    if not site_scoped:

        site_count = 0

        coverage = 1.0   # repo-wide capability, binary coverage — approved

    else:

        evidenced_site_ids = {e.site_id for e in site_scoped}

        relevant_kinds = {site_kind_by_id[sid] for sid in evidenced_site_ids if sid in site_kind_by_id}

        total_applicable = sum(total_sites_by_kind.get(k, 0) for k in relevant_kinds)

        site_count = len(evidenced_site_ids)

        coverage = (site_count / total_applicable) if total_applicable > 0 else 1.0

 

    if mean_confidence < config.inconclusive_confidence_threshold:

        status = CapabilityStatus.INCONCLUSIVE

    elif coverage >= 1.0:

        status = CapabilityStatus.DETECTED

    elif coverage > 0.0:

        status = CapabilityStatus.PARTIALLY_DETECTED

    else:

        status = CapabilityStatus.NOT_DETECTED  # defensive; cap_evidence non-empty makes this unreachable today

 

    return CapabilityState(

        capability_id=capability_id, status=status, confidence=mean_confidence,

        evidence_count=len(cap_evidence), site_count=site_count, coverage=coverage,

        evidence_ids=evidence_ids,

    )

Tests test evidence engine · PY

"""Phase 11.1 tests — Evidence model + engine, built against real IR/

Facts/Sites/Claims, since confidence blending and the hash chain

depend on real data shapes.

"""

from __future__ import annotations

 

from pathlib import Path

 

from meridian.compliance.resolver import ComplianceGraphResolver

from meridian.detectors.detector import detect_all_capabilities

from meridian.evidence.engine import build_evidence_ledger, verify_ledger

from meridian.evidence.models import GENESIS_HASH, compute_record_hash, stable_evidence_id

from meridian.facts.extractor import extract_all_facts

from meridian.graph.graph_builder import build_repository_graph

from meridian.ir_builder import build_ir_repository

from meridian.models.language import Language

from meridian.parsers.parser_factory import parse_file

from meridian.sites.finder import discover_all_sites

 

import dataclasses

import pytest

 

DB_PATH = Path("/home/claude/meridian.db")

 

 

def _pipeline(source: bytes, rel_path: str = "a.py"):

    parsed = parse_file(Path(f"/tmp/{rel_path}"), rel_path, Language.PYTHON, source)

    repo = build_ir_repository("/tmp", [parsed])

    graph = build_repository_graph(repo, repository_path="/tmp")

    facts = extract_all_facts(repo, graph)

    sites = discover_all_sites(repo, graph, facts)

    return repo, graph, facts, sites

 

 

@pytest.fixture

def resolver():

    if not DB_PATH.exists():

        pytest.skip("meridian.db not present in this sandbox session")

    return ComplianceGraphResolver(DB_PATH)

 

 

# ── stable_evidence_id ────────────────────────────────────────────────────────

 

def test_stable_evidence_id_deterministic() -> None:

    assert stable_evidence_id("claim123") == stable_evidence_id("claim123")

 

 

def test_stable_evidence_id_differs_by_claim() -> None:

    assert stable_evidence_id("claim123") != stable_evidence_id("claim456")

 

 

def test_evidence_id_namespace_distinct_from_claim_id() -> None:

    """evidence_id must never literally equal claim_id — re-hashed with

    a distinguishing prefix so the two ID spaces never collide."""

    assert stable_evidence_id("claim123") != "claim123"

 

 

# ── build_evidence_ledger: confidence blending ────────────────────────────────

 

def test_confidence_blends_fact_and_rule_confidence(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"""

class User:

    email: str

    def save(self):

        kms.encrypt(self.email)

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    encryption_evidence = next(e for e in ledger if e.capability_id == "Encryption_At_Rest")

    # fact confidence (1.0, SUCCESS parse) x rule base_confidence (0.9) = 0.9

    assert encryption_evidence.confidence == pytest.approx(0.9)

 

 

def test_confidence_is_mean_across_multiple_supporting_facts() -> None:

    """If minimum_matches>1 caused multiple facts to back one claim,

    confidence should be the MEAN of their source confidences, not

    just the first one's."""

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import Fact, FactKind, FactStore

    from meridian.models.ir import IRLocation

    from meridian.models.parse_status import ParseStatus

 

    loc = IRLocation("a.py", 1, 1, 0, 0)

    f1 = Fact.create(FactKind.ENCRYPTION_CALL, loc, "a.py::f", {}, 1.0, ParseStatus.SUCCESS)

    f2 = Fact.create(FactKind.AUDIT_LOG_CALL, loc, "a.py::f", {}, 0.6, ParseStatus.FALLBACK)

    store = FactStore([f1, f2])

    claim = CapabilityClaim.create("Audit_Logging", "test_rule", (f1.fact_id, f2.fact_id), 1.0, loc)

 

    ledger = build_evidence_ledger([claim], store)

    assert len(ledger) == 1

    assert ledger[0].confidence == pytest.approx((1.0 + 0.6) / 2 * 1.0)

 

 

def test_confidence_zero_when_no_resolvable_supporting_facts() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    loc = IRLocation("a.py", 1, 1, 0, 0)

    claim = CapabilityClaim.create("Access_Control", "test_rule", ("nonexistent_fact_id",), 0.9, loc)

    ledger = build_evidence_ledger([claim], FactStore([]))

    assert ledger[0].confidence == 0.0

 

 

# ── Evidence carries site_id and rule_id (approved additions) ────────────────

 

def test_evidence_carries_site_id_and_rule_id_from_claim(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"""

@app.route("/x")

@login_required

def view():

    pass

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    access_evidence = next(e for e in ledger if e.capability_id == "Access_Control")

    assert access_evidence.site_id is not None

    assert access_evidence.rule_id == "access_control"

 

 

# ── Stability across re-scans ─────────────────────────────────────────────────

 

def test_evidence_id_stable_across_repeated_scans_of_unchanged_repo(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"""

class User:

    email: str

    def save(self):

        kms.encrypt(self.email)

""")

    claims1 = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger1 = build_evidence_ledger(claims1, facts)

    claims2 = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger2 = build_evidence_ledger(claims2, facts)

    assert sorted(e.evidence_id for e in ledger1) == sorted(e.evidence_id for e in ledger2)

 

 

def test_chain_hash_legitimately_differs_across_separate_scans(resolver) -> None:

    """The hash/previous_hash fields are NOT required to be stable

    across scans — created_at is real wall-clock time, by design. Only

    evidence_id is the stable cross-scan identity."""

    import time

    repo, graph, facts, sites = _pipeline(b"""

class User:

    email: str

    def save(self):

        kms.encrypt(self.email)

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger1 = build_evidence_ledger(claims, facts)

    time.sleep(0.01)

    ledger2 = build_evidence_ledger(claims, facts)

    h1 = sorted(e.hash for e in ledger1)

    h2 = sorted(e.hash for e in ledger2)

    assert h1 != h2  # different created_at -> different hash, expected

 

 

# ── Hash chain structure and tamper detection ─────────────────────────────────

 

def test_genesis_record_links_to_genesis_hash() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    claim = CapabilityClaim.create("Access_Control", "test", (), 0.9, IRLocation("a.py", 1, 1, 0, 0))

    ledger = build_evidence_ledger([claim], FactStore([]))

    assert ledger[0].previous_hash == GENESIS_HASH

 

 

def test_chain_links_sequentially() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    loc = IRLocation("a.py", 1, 1, 0, 0)

    claims = [

        CapabilityClaim.create("Access_Control", "rule_a", (), 0.9, loc),

        CapabilityClaim.create("Audit_Logging", "rule_b", (), 0.8, loc),

    ]

    ledger = build_evidence_ledger(claims, FactStore([]))

    assert ledger[1].previous_hash == ledger[0].hash

 

 

def test_verify_ledger_clean_chain_has_no_problems() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    loc = IRLocation("a.py", 1, 1, 0, 0)

    claims = [CapabilityClaim.create("Access_Control", "rule_a", (), 0.9, loc)]

    ledger = build_evidence_ledger(claims, FactStore([]))

    assert verify_ledger(ledger) == []

 

 

def test_verify_ledger_detects_content_tampering() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    loc = IRLocation("a.py", 1, 1, 0, 0)

    claims = [CapabilityClaim.create("Access_Control", "rule_a", (), 0.9, loc)]

    ledger = build_evidence_ledger(claims, FactStore([]))

    tampered = [dataclasses.replace(ledger[0], confidence=0.999)]

    problems = verify_ledger(tampered)

    assert len(problems) == 1

    assert "tampered" in problems[0]

 

 

def test_verify_ledger_detects_reordering() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    loc = IRLocation("a.py", 1, 1, 0, 0)

    claims = [

        CapabilityClaim.create("Access_Control", "rule_a", (), 0.9, loc),

        CapabilityClaim.create("Audit_Logging", "rule_b", (), 0.8, loc),

    ]

    ledger = build_evidence_ledger(claims, FactStore([]))

    reordered = list(reversed(ledger))

    assert len(verify_ledger(reordered)) > 0

 

 

def test_evidence_is_frozen() -> None:

    from meridian.detectors.models import CapabilityClaim

    from meridian.facts.models import FactStore

    from meridian.models.ir import IRLocation

 

    claim = CapabilityClaim.create("Access_Control", "test", (), 0.9, IRLocation("a.py", 1, 1, 0, 0))

    ledger = build_evidence_ledger([claim], FactStore([]))

    with pytest.raises(dataclasses.FrozenInstanceError):

        ledger[0].confidence = 0.5  # type: ignore[misc]

Tests test inventory aggregator · PY

"""Phase 11.2 tests — InventoryAggregator."""

from __future__ import annotations

 

from pathlib import Path

 

import pytest

 

from meridian.compliance.resolver import ComplianceGraphResolver

from meridian.detectors.detector import detect_all_capabilities

from meridian.evidence.engine import build_evidence_ledger

from meridian.facts.extractor import extract_all_facts

from meridian.graph.graph_builder import build_repository_graph

from meridian.inventory.aggregator import AggregationConfig, aggregate_inventory

from meridian.inventory.models import CapabilityStatus

from meridian.ir_builder import build_ir_repository

from meridian.models.language import Language

from meridian.parsers.parser_factory import parse_file

from meridian.sites.finder import discover_all_sites

 

DB_PATH = Path("/home/claude/meridian.db")

 

 

@pytest.fixture

def resolver():

    if not DB_PATH.exists():

        pytest.skip("meridian.db not present in this sandbox session")

    return ComplianceGraphResolver(DB_PATH)

 

 

def _pipeline(source: bytes, rel_path: str = "a.py"):

    parsed = parse_file(Path(f"/tmp/{rel_path}"), rel_path, Language.PYTHON, source)

    repo = build_ir_repository("/tmp", [parsed])

    graph = build_repository_graph(repo, repository_path="/tmp")

    facts = extract_all_facts(repo, graph)

    sites = discover_all_sites(repo, graph, facts)

    return repo, graph, facts, sites

 

 

# ── Seeding requirement — the whole reason this design exists ────────────────

 

def test_inventory_seeds_exactly_all_real_capability_ids(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"def f():\n    pass\n")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

    assert len(inventory.capability_states) == len(resolver.all_capability_ids())

    assert {s.capability_id for s in inventory.capability_states} == set(resolver.all_capability_ids())

 

 

def test_capability_with_zero_evidence_is_explicit_not_detected_not_absent(resolver) -> None:

    """The entire point of seeding: a capability nobody's code ever

    evidences must still appear as a row, status NOT_DETECTED — never

    silently missing from the inventory."""

    repo, graph, facts, sites = _pipeline(b"def f():\n    pass\n")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

    state = inventory.get("Backup_And_Recovery")  # almost certainly undetected on a trivial repo

    assert state is not None

    assert state.status == CapabilityStatus.NOT_DETECTED

    assert state.evidence_count == 0

    assert state.confidence == 0.0

 

 

def test_inventory_get_returns_none_for_unknown_capability(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"def f():\n    pass\n")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

    assert inventory.get("Not_A_Real_Capability") is None

 

 

# ── Status decision logic ─────────────────────────────────────────────────────

 

def test_detected_when_full_coverage_repo_wide(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"""

def f():

    record_consent(user, purpose)

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

    state = inventory.get("Consent_Management")

    assert state.status == CapabilityStatus.DETECTED

    assert state.coverage == 1.0

    assert state.site_count == 0  # repo-wide, no site denominator

 

 

def test_partially_detected_when_some_but_not_all_sites_covered(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"""

@app.route("/a")

@login_required

def route_a():

    pass

 

@app.route("/b")

def route_b():

    pass

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

    state = inventory.get("Access_Control")

    assert state.status == CapabilityStatus.PARTIALLY_DETECTED

    assert state.site_count == 1

    assert 0.0 < state.coverage < 1.0

    assert state.coverage == pytest.approx(0.5)  # 1 of 2 HTTP_ROUTE_SITEs

 

 

def test_inconclusive_when_confidence_below_configured_threshold(resolver) -> None:

    """Karthik's explicit requirement: threshold must be configurable,

    never hardcoded. A high threshold should flip a normally-DETECTED

    result to INCONCLUSIVE."""

    repo, graph, facts, sites = _pipeline(b"""

def f():

    record_consent(user, purpose)

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

 

    lenient = aggregate_inventory(ledger, sites, resolver, "/tmp", AggregationConfig(inconclusive_confidence_threshold=0.1))

    assert lenient.get("Consent_Management").status == CapabilityStatus.DETECTED

 

    strict = aggregate_inventory(ledger, sites, resolver, "/tmp", AggregationConfig(inconclusive_confidence_threshold=0.95))

    assert strict.get("Consent_Management").status == CapabilityStatus.INCONCLUSIVE

 

 

def test_not_detected_when_no_evidence_even_with_lenient_threshold(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"def f():\n    pass\n")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp", AggregationConfig(inconclusive_confidence_threshold=0.0))

    state = inventory.get("Encryption_At_Rest")

    assert state.status == CapabilityStatus.NOT_DETECTED

 

 

# ── evidence_ids traceability ─────────────────────────────────────────────────

 

def test_capability_state_evidence_ids_resolve_to_real_ledger_entries(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"""

class User:

    email: str

    def save(self):

        kms.encrypt(self.email)

""")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

 

    ledger_ids = {e.evidence_id for e in ledger}

    state = inventory.get("Encryption_At_Rest")

    assert state.evidence_count == 1

    for eid in state.evidence_ids:

        assert eid in ledger_ids

 

 

def test_repository_inventory_generated_at_is_set(resolver) -> None:

    repo, graph, facts, sites = _pipeline(b"def f():\n    pass\n")

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, "/tmp")

    assert inventory.generated_at

    assert inventory.repository_path == "/tmp"

Tests test phase11 extraction e2e · PY

"""Phase 11 — end-to-end: Fact -> CapabilityClaim -> Evidence ->

CapabilityState -> RepositoryInventory, against the real Flask

repository and the real meridian.db. Locks in the exact numbers

inspected manually while building this phase.

"""

from __future__ import annotations

 

from pathlib import Path

 

import pytest

 

from meridian.compliance.resolver import ComplianceGraphResolver

from meridian.detectors.detector import detect_all_capabilities

from meridian.discovery import discover_repository

from meridian.evidence.engine import build_evidence_ledger, verify_ledger

from meridian.facts.extractor import extract_all_facts

from meridian.file_ranker import rank_files

from meridian.graph.graph_builder import build_repository_graph

from meridian.inventory.aggregator import aggregate_inventory

from meridian.inventory.models import CapabilityStatus

from meridian.ir_builder import build_ir_repository

from meridian.parsers.parser_factory import parse_file

from meridian.sites.finder import discover_all_sites

 

FLASK_REPO = Path("/tmp/flask-repo")

DB_PATH = Path("/home/claude/meridian.db")

 

 

@pytest.fixture(scope="module")

def flask_pipeline():

    if not FLASK_REPO.exists():

        pytest.skip("Flask repo not present at /tmp/flask-repo in this sandbox session")

    if not DB_PATH.exists():

        pytest.skip("meridian.db not present in this sandbox session")

 

    md = discover_repository(FLASK_REPO)

    ranked = rank_files(md)

    parsed = [

        parse_file(rf.file.path, rf.file.rel_path, rf.file.language, rf.file.path.read_bytes())

        for rf in ranked

    ]

    repo = build_ir_repository(str(FLASK_REPO), parsed)

    graph = build_repository_graph(repo, repository_path=str(FLASK_REPO))

    facts = extract_all_facts(repo, graph)

    sites = discover_all_sites(repo, graph, facts)

    resolver = ComplianceGraphResolver(DB_PATH)

    claims = detect_all_capabilities(repo, graph, facts, sites, resolver)

    ledger = build_evidence_ledger(claims, facts)

    inventory = aggregate_inventory(ledger, sites, resolver, str(FLASK_REPO))

    return facts, sites, resolver, claims, ledger, inventory

 

 

def test_ledger_verifies_clean_on_real_data(flask_pipeline) -> None:

    _, _, _, _, ledger, _ = flask_pipeline

    assert verify_ledger(ledger) == []

 

 

def test_inventory_has_all_49_real_capabilities(flask_pipeline) -> None:

    _, _, resolver, _, _, inventory = flask_pipeline

    assert len(inventory.capability_states) == len(resolver.all_capability_ids())

 

 

def test_access_control_state_matches_manually_verified_numbers(flask_pipeline) -> None:

    """Locks in the exact numbers inspected manually: 3 evidenced

    HTTP_ROUTE_SITEs (blog.py create/update/delete, all @login_required)

    out of 30 total HTTP_ROUTE_SITEs in Flask's real tutorial app."""

    _, sites, _, _, _, inventory = flask_pipeline

    from meridian.sites.models import SiteKind

    state = inventory.get("Access_Control")

    assert state.status == CapabilityStatus.PARTIALLY_DETECTED

    assert state.evidence_count == 3

    assert state.site_count == 3

    total_routes = len(sites.by_kind(SiteKind.HTTP_ROUTE_SITE))

    assert state.coverage == pytest.approx(3 / total_routes)

 

 

def test_authentication_state_matches_manually_verified_numbers(flask_pipeline) -> None:

    _, _, _, _, _, inventory = flask_pipeline

    state = inventory.get("Authentication")

    assert state.status == CapabilityStatus.PARTIALLY_DETECTED

    assert state.evidence_count == 2

    assert state.confidence == pytest.approx(0.85)

 

 

def test_capability_with_zero_real_evidence_is_explicit_row(flask_pipeline) -> None:

    """Flask itself has no PII fields (it's a framework, not an app

    with data models) — Encryption_At_Rest must still appear as an

    explicit NOT_DETECTED row, not be missing."""

    _, _, _, _, _, inventory = flask_pipeline

    state = inventory.get("Encryption_At_Rest")

    assert state is not None

    assert state.status == CapabilityStatus.NOT_DETECTED

    assert state.evidence_count == 0

 

 

def test_full_chain_traces_forward_to_real_dpdpa_control(flask_pipeline) -> None:

    """The complete demonstration: a real CapabilityState from real

    Flask source resolves forward through the real compliance graph to

    DPDPA's actual DPDP_SCH_001 control with its real audit_question."""

    _, _, resolver, _, _, inventory = flask_pipeline

    state = inventory.get("Access_Control")

    assert state.evidence_count > 0

 

    objectives = resolver.objectives_for_capability("Access_Control")

    assert len(objectives) > 0

 

    contexts = resolver.controls_for_capability("Access_Control")

    assert len(contexts) > 0

    assert all(cc.control.framework_id == cc.framework.id for cc in contexts)

 

 

def test_full_chain_traces_backward_to_real_source_line(flask_pipeline) -> None:

    """The other direction: a CapabilityState's evidence_ids resolve

    back through Evidence -> Fact -> a real, exact file:line."""

    facts, _, _, _, ledger, inventory = flask_pipeline

    state = inventory.get("Access_Control")

    evidence_by_id = {e.evidence_id: e for e in ledger}

    fact_by_id = {f.fact_id: f for f in facts.all()}

 

    for eid in state.evidence_ids:

        evidence = evidence_by_id[eid]

        assert evidence.location.rel_path.endswith("blog.py")

        for fid in evidence.supporting_fact_ids:

            fact = fact_by_id[fid]

            assert fact.attributes.get("decorator") == "@login_required"

 

 

def test_inventory_is_deterministic_across_repeated_runs(flask_pipeline) -> None:

    """Re-running the entire Phase 9-11 pipeline on the same unchanged

    repo must produce the same inventory numbers — no run-to-run

    randomness anywhere in facts, sites, claims, or aggregation."""

    facts, sites, resolver, _, _, inventory1 = flask_pipeline

 

    # Rebuild fully independently rather than reusing facts/sites, to

    # prove the WHOLE pipeline is deterministic, not just aggregation.

    md = discover_repository(FLASK_REPO)

    ranked = rank_files(md)

    parsed = [

        parse_file(rf.file.path, rf.file.rel_path, rf.file.language, rf.file.path.read_bytes())

        for rf in ranked

    ]

    repo2 = build_ir_repository(str(FLASK_REPO), parsed)

    graph2 = build_repository_graph(repo2, repository_path=str(FLASK_REPO))

    facts2 = extract_all_facts(repo2, graph2)

    sites2 = discover_all_sites(repo2, graph2, facts2)

    claims2 = detect_all_capabilities(repo2, graph2, facts2, sites2, resolver)

    ledger2 = build_evidence_ledger(claims2, facts2)

    inventory2 = aggregate_inventory(ledger2, sites2, resolver, str(FLASK_REPO))

 

    states1 = {s.capability_id: (s.status, s.evidence_count, s.coverage) for s in inventory1.capability_states}

    states2 = {s.capability_id: (s.status, s.evidence_count, s.coverage) for s in inventory2.capability_states}

    assert states1 == states2

Meridian verdicts init · PY

"""Phase 12 — Control Verdict Resolution. The actual finish line for M3.

 

RepositoryInventory (Phase 11.2) -> list[Finding]: one Finding per

control in meridian.db (all 1,445), with a Verdict derived from the

per-objective capability states that control depends on.

 

Hard rule, enforced structurally: Finding.rationale is control.audit_question

read verbatim from meridian.db. Nothing in this package generates

rationale text — no LLM call, no string templating beyond the literal

fallback for controls with a null audit_question. See

test_verdicts_no_generation.py for the AST-level proof.

 

Two distinct "coverage-shaped" numbers exist in this codebase and

mean different things on purpose:

  - CapabilityState.coverage (Phase 11.2): sites with evidence / total

    applicable sites, for ONE capability.

  - Finding.objective_coverage (this phase): objectives fully DETECTED

    / total objectives, for ONE control. Counts DETECTED only, not

    PARTIALLY_DETECTED — a control can have verdict=PARTIAL with

    objective_coverage=0.0, which is not a contradiction: coverage

    answers "how many objectives are fully met," verdict answers "is

    there any positive signal at all."

"""

 

Meridian verdicts models · PY

"""Phase 12 — Finding, Verdict, FrameworkSummary."""

from __future__ import annotations

 

import hashlib

from dataclasses import dataclass

from enum import Enum

 

from meridian.models.ir import IRLocation

 

 

class Verdict(str, Enum):

    SATISFIED    = "satisfied"

    VIOLATED     = "violated"

    PARTIAL      = "partial"

    INCONCLUSIVE = "inconclusive"

 

 

def _stable_finding_id(control_id: str, framework_id: str) -> str:

    """Stable across repeated scans — same (control, framework) pair

    always hashes the same, matching fact_id/claim_id/evidence_id's

    stability guarantee. Independent of verdict/confidence, which can

    legitimately change between scans as the repository changes —

    finding_id identifies WHICH control this Finding is about, not the

    specific verdict it carries this run."""

    return hashlib.sha256(f"finding|{control_id}|{framework_id}".encode("utf-8")).hexdigest()[:24]

 

 

@dataclass(frozen=True, slots=True)

class Finding:

    finding_id:          str

    framework_id:        str

    control_id:          str

    objective_ids:       tuple[str, ...]

    capability_ids:      tuple[str, ...]

    verdict:             Verdict

    objective_coverage:  float    # DETECTED objectives / total objectives — DETECTED only, see package docstring

    confidence:           float

    evidence_ids:         tuple[str, ...]

    rationale:             str      # control.audit_question, verbatim from meridian.db — never generated

    severity:              str | None

    location_samples:      tuple[IRLocation, ...]

 

    @staticmethod

    def create(

        framework_id: str, control_id: str, objective_ids: tuple[str, ...],

        capability_ids: tuple[str, ...], verdict: Verdict, objective_coverage: float,

        confidence: float, evidence_ids: tuple[str, ...], rationale: str,

        severity: str | None, location_samples: tuple[IRLocation, ...],

    ) -> "Finding":

        return Finding(

            finding_id=_stable_finding_id(control_id, framework_id),

            framework_id=framework_id, control_id=control_id, objective_ids=objective_ids,

            capability_ids=capability_ids, verdict=verdict, objective_coverage=objective_coverage,

            confidence=confidence, evidence_ids=evidence_ids, rationale=rationale,

            severity=severity, location_samples=location_samples,

        )

 

 

@dataclass(frozen=True, slots=True)

class FrameworkSummary:

    framework_id:           str

    framework_name:          str

    total_controls:          int

    satisfied_count:         int

    violated_count:          int

    partial_count:           int

    inconclusive_count:      int

    compliance_percentage:   float   # satisfied_count / total_controls x 100 — STRICT, no partial-credit blending

    finding_ids:             tuple[str, ...]
Meridian verdicts resolver · PY

"""Phase 12 — ControlVerdictResolver.

 

resolve_all() is the one function this module exists for:

RepositoryInventory + list[Evidence] + ComplianceGraphResolver ->

list[Finding], one per control in meridian.db (all 1,445).

 

No LLM call anywhere in this file. rationale is control.audit_question,

read directly from the resolver, with one literal fallback string for

the real (if rare) case where a control's audit_question is null in

the DB — that fallback is honest about its own limitation rather than

inventing plausible-sounding regulatory language.

"""

from __future__ import annotations

 

from meridian.compliance.resolver import ComplianceGraphResolver

from meridian.evidence.models import Evidence

from meridian.inventory.models import CapabilityState, CapabilityStatus, RepositoryInventory

from meridian.models.ir import IRLocation

from meridian.verdicts.models import Finding, Verdict

 

_STATUS_RANK = {

    CapabilityStatus.DETECTED: 3,

    CapabilityStatus.PARTIALLY_DETECTED: 2,

    CapabilityStatus.INCONCLUSIVE: 1,

    CapabilityStatus.NOT_DETECTED: 0,

}

 

_MAX_LOCATION_SAMPLES = 3

 

_NO_AUDIT_QUESTION_FALLBACK = (

    "No audit_question is recorded for this control in meridian.db — "

    "manual verification against the control's regulatory text is required."

)

 

 

class ControlVerdictResolver:

    def __init__(self, resolver: ComplianceGraphResolver) -> None:

        self._resolver = resolver

 

    def resolve_all(self, inventory: RepositoryInventory, evidence_ledger: list[Evidence]) -> list[Finding]:

        state_by_capability = {s.capability_id: s for s in inventory.capability_states}

        evidence_by_id = {e.evidence_id: e for e in evidence_ledger}

        return [

            self._resolve_one(control_id, state_by_capability, evidence_by_id)

            for control_id in self._resolver.all_control_ids()

        ]

 

    def _resolve_one(

        self, control_id: str, state_by_capability: dict[str, CapabilityState],

        evidence_by_id: dict[str, Evidence],

    ) -> Finding:

        control = self._resolver.control(control_id)

        objectives = self._resolver.objectives_for_control(control_id)

 

        objective_statuses: list[CapabilityStatus] = []

        contributing_capability_ids: set[str] = set()

        contributing_evidence_ids: set[str] = set()

        pooled_confidences: list[float] = []

 

        for objective in objectives:

            capabilities = self._resolver.capabilities_for_objective(objective.id)

            best_state = self._best_state(capabilities, state_by_capability)

            best_status = best_state.status if best_state is not None else CapabilityStatus.NOT_DETECTED

            objective_statuses.append(best_status)

 

            if best_state is not None and best_state.evidence_count > 0:

                contributing_capability_ids.add(best_state.capability_id)

                contributing_evidence_ids.update(best_state.evidence_ids)

                pooled_confidences.append(best_state.confidence)

 

        verdict = self._decide_verdict(objective_statuses)

        detected_count = sum(1 for s in objective_statuses if s == CapabilityStatus.DETECTED)

        objective_coverage = (detected_count / len(objective_statuses)) if objective_statuses else 0.0

        confidence = (sum(pooled_confidences) / len(pooled_confidences)) if pooled_confidences else 0.0

 

        location_samples = self._sample_locations(contributing_evidence_ids, evidence_by_id)

 

        return Finding.create(

            framework_id=control.framework_id, control_id=control.id,

            objective_ids=tuple(o.id for o in objectives),

            capability_ids=tuple(sorted(contributing_capability_ids)),

            verdict=verdict, objective_coverage=objective_coverage, confidence=confidence,

            evidence_ids=tuple(sorted(contributing_evidence_ids)),

            rationale=control.audit_question or _NO_AUDIT_QUESTION_FALLBACK,

            severity=control.severity, location_samples=location_samples,

        )

 

    def _best_state(self, capabilities, state_by_capability: dict[str, CapabilityState]) -> CapabilityState | None:

        best: CapabilityState | None = None

        for cap in capabilities:

            state = state_by_capability.get(cap.id)

            if state is None:

                continue  # shouldn't happen given Phase 11.2's exhaustive seeding; defensive only

            if best is None or _STATUS_RANK[state.status] > _STATUS_RANK[best.status]:

                best = state

        return best

 

    def _decide_verdict(self, objective_statuses: list[CapabilityStatus]) -> Verdict:

        if not objective_statuses:

            return Verdict.INCONCLUSIVE  # control with no mapped objectives — ca
<truncated 29130 bytes>

NOTE: The output was truncated because it was too long. Use a more targeted query or a smaller range to get the information you need.