feat: Silent Knowledge Pass — recognise before asking (Phase 0, before the endpoint)
Not the endpoint yet — the bigger knowledge lever first. The Advisor can say "I need 5 answers" but does not yet decide what it can find out by ITSELF. The Silent Knowledge Pass runs in front of the Advisor and, from signals existing scanners/parsers already produce (website, repository, documents, product data), deterministically derives capabilities the company demonstrably HAS + product facts that drive scope — so every recognised item shrinks the delta and removes a question. compliance/onboarding/silent_intake.py: silent_intake(signals, signal_map) -> detected_capabilities (+ evidence already in hand) + product_facts. The signal->conclusion map is curated DATA (knowledge/onboarding/intake_signal_map.yaml), signals are injected (scanners are upstream). Pure, deterministic, no LLM. advisor_start gains detected_capabilities (folded into the profile at HIGH confidence -> covered, not asked) and an auto_detected result + headline. The experience flips from a question wall to "we already recognised 4 capabilities, 2 product facts and have 4 pieces of evidence in hand — only these few remain". Order now: Silent Pass -> #58 endpoint/frontend -> #59 empirical loop. NOT new architecture, just an orchestration step in front. Non-runtime (no app caller) -> no deploy. 15 onboarding tests pass, mypy --strict clean, check-loc 0.
This commit is contained in:
@@ -0,0 +1,79 @@
|
||||
"""Silent Knowledge Pass — recognise before asking (Phase 0).
|
||||
|
||||
Pins the deterministic signal->capability/product-fact mapping and the product effect that matters: when
|
||||
the Silent Pass feeds detected capabilities into the Advisor, the delta shrinks and the number of
|
||||
next-best questions DROPS — "we already recognised X, only these few remain" instead of a question wall.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
import yaml
|
||||
|
||||
from compliance.onboarding import (
|
||||
IntakeSignal,
|
||||
OnboardingInput,
|
||||
SignalMapping,
|
||||
advisor_start,
|
||||
resolve_for_certifications,
|
||||
silent_intake,
|
||||
)
|
||||
from compliance.onboarding import CapabilityHypothesis
|
||||
from compliance.transition_reasoning import TargetRequirement
|
||||
|
||||
_DIR = os.path.dirname(__file__)
|
||||
_MAP = [SignalMapping(**m) for m in yaml.safe_load(
|
||||
open(os.path.join(_DIR, "..", "knowledge", "onboarding", "intake_signal_map.yaml"), encoding="utf-8"))["mappings"]]
|
||||
_LIB = [CapabilityHypothesis(**h) for h in yaml.safe_load(
|
||||
open(os.path.join(_DIR, "..", "knowledge", "certification_hypotheses", "hypotheses.yaml"), encoding="utf-8"))["hypotheses"]]
|
||||
_CRA = yaml.safe_load(open(os.path.join(_DIR, "..", "knowledge", "transition_patterns",
|
||||
"transition_pattern_iso27001_to_cra_maschinenvo_v1.yaml"), encoding="utf-8"))
|
||||
_REQ = [TargetRequirement(capability_id=a["capability"]) for a in _CRA["likely_covered"]]
|
||||
_REQ += [TargetRequirement(capability_id=d["capability"], expected_evidence=d.get("expected_evidence", []))
|
||||
for d in _CRA["delta_requirements"]]
|
||||
|
||||
# scanner findings (injected): a machine builder with a public CVD policy, an SBOM + signed releases in
|
||||
# the repo, a product risk-assessment doc, and a cloud-connected PLC product.
|
||||
_SIGNALS = [
|
||||
IntakeSignal(source="website", signal="security_txt_or_cvd_policy", detail="/.well-known/security.txt"),
|
||||
IntakeSignal(source="repository", signal="sbom_file_found", detail="sbom.cdx.json"),
|
||||
IntakeSignal(source="repository", signal="signed_releases"),
|
||||
IntakeSignal(source="document", signal="product_risk_assessment_doc"),
|
||||
IntakeSignal(source="product", signal="cloud_connectivity"),
|
||||
IntakeSignal(source="product", signal="plc_sps"),
|
||||
]
|
||||
|
||||
|
||||
def test_silent_intake_is_deterministic_signal_mapping():
|
||||
res = silent_intake(_SIGNALS, _MAP)
|
||||
caps = set(res.capability_ids())
|
||||
assert {"coordinated_vulnerability_disclosure", "sbom_creation", "secure_signed_update_distribution",
|
||||
"product_cyber_risk_assessment"} <= caps
|
||||
assert "sbom" in res.evidence_found # evidence already in hand -> no upload needed
|
||||
facts = {f.key for f in res.product_facts}
|
||||
assert "connected_to_internet" in facts and "is_machine" in facts
|
||||
|
||||
|
||||
def test_silent_pass_reduces_the_questions():
|
||||
inp = OnboardingInput(company="x", certifications=["ISO27001", "ISO9001"], target=["CRA"])
|
||||
hyp = resolve_for_certifications(inp.certifications, _LIB)
|
||||
without = advisor_start(inp, hyp, _REQ, target_id="CRA", corpus_status={"CRA": "validated"})
|
||||
detected = silent_intake(_SIGNALS, _MAP).capability_ids()
|
||||
with_pass = advisor_start(inp, hyp, _REQ, target_id="CRA", corpus_status={"CRA": "validated"},
|
||||
detected_capabilities=detected)
|
||||
# the whole point: recognising things automatically leaves FEWER open questions
|
||||
assert len(with_pass.capability_delta) < len(without.capability_delta)
|
||||
assert len(with_pass.next_best_questions) <= len(without.next_best_questions)
|
||||
assert with_pass.auto_detected # recognised without asking
|
||||
assert "automatisch erkannt (Intake)" in with_pass.headline
|
||||
|
||||
|
||||
def test_detected_capabilities_are_not_asked_again():
|
||||
inp = OnboardingInput(company="x", certifications=["ISO27001"], target=["CRA"])
|
||||
hyp = resolve_for_certifications(inp.certifications, _LIB)
|
||||
detected = silent_intake(_SIGNALS, _MAP).capability_ids()
|
||||
res = advisor_start(inp, hyp, _REQ, target_id="CRA", corpus_status={"CRA": "validated"},
|
||||
detected_capabilities=detected)
|
||||
asked = {q.capability_id for q in res.next_best_questions}
|
||||
assert "sbom_creation" not in asked and "sbom_creation" not in res.capability_delta
|
||||
Reference in New Issue
Block a user