diff --git a/backend-compliance/compliance/api/__init__.py b/backend-compliance/compliance/api/__init__.py index 8faf5866..3c47c08d 100644 --- a/backend-compliance/compliance/api/__init__.py +++ b/backend-compliance/compliance/api/__init__.py @@ -78,6 +78,7 @@ _ROUTER_MODULES = [ "template_rule_routes", "specialist_agent_routes", "reasoning_routes", + "onboarding_routes", ] _loaded_count = 0 diff --git a/backend-compliance/compliance/api/onboarding_routes.py b/backend-compliance/compliance/api/onboarding_routes.py new file mode 100644 index 00000000..f8ef92a8 --- /dev/null +++ b/backend-compliance/compliance/api/onboarding_routes.py @@ -0,0 +1,72 @@ +"""Onboarding Advisor endpoint — exposes the existing Smart Onboarding Advisor at runtime. + +This adds NO new reasoning logic. It exposes the already-built, tested orchestration (Signal Producers +-> Normalizer -> Silent Knowledge Pass -> Advisor) through one runtime endpoint. No DB, no persistence. + + POST /onboarding/advisor-start — (company + certs + target + scanner findings) -> advisory payload + GET /onboarding/targets — the supported target ids +""" + +import logging +from typing import List, Optional + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from compliance.onboarding import ( + AdvisorMeasure, + AdvisorQuestion, + InferredAssumption, + ProducedSignal, + RejectedAssumption, +) +from compliance.services.onboarding_service import run_advisor, supported_targets + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/onboarding", tags=["onboarding"]) + + +class OnboardingAdvisorRequest(BaseModel): + company: str = "" + industry: Optional[str] = None + products: List[str] = Field(default_factory=list) + markets: List[str] = Field(default_factory=list) + certifications: List[str] = Field(default_factory=list) + known_evidence: List[str] = Field(default_factory=list) + target: str = "CRA" + scanner_findings: List[ProducedSignal] = Field(default_factory=list) # adapters upstream produced these + + +class AdvisorResponse(BaseModel): + silent_intake_summary: str = "" + headline: str = "" + auto_detected: List[str] = Field(default_factory=list) + inferred_assumptions: List[InferredAssumption] = Field(default_factory=list) + rejected_assumptions: List[RejectedAssumption] = Field(default_factory=list) + top_5_questions: List[AdvisorQuestion] = Field(default_factory=list) + capability_delta: List[str] = Field(default_factory=list) + top_measures: List[AdvisorMeasure] = Field(default_factory=list) + evidence_requests: List[str] = Field(default_factory=list) + unsupported_domains: List[str] = Field(default_factory=list) + completeness_summary: str = "" + + +@router.get("/targets") +def list_targets() -> dict: + return {"targets": supported_targets()} + + +@router.post("/advisor-start", response_model=AdvisorResponse) +def advisor_start_endpoint(req: OnboardingAdvisorRequest) -> AdvisorResponse: + if req.target not in supported_targets(): + raise HTTPException(status_code=404, detail="unsupported target '%s'; supported: %s" % (req.target, supported_targets())) + result, si_summary = run_advisor( + company=req.company, certifications=req.certifications, target=req.target, + signals=req.scanner_findings, known_evidence=req.known_evidence, + products=req.products, markets=req.markets, industry=req.industry or "") + return AdvisorResponse( + silent_intake_summary=si_summary, headline=result.headline, auto_detected=result.auto_detected, + inferred_assumptions=result.inferred_assumptions, rejected_assumptions=result.rejected_assumptions, + top_5_questions=result.next_best_questions, capability_delta=result.capability_delta, + top_measures=result.top_measures, evidence_requests=result.evidence_requests, + unsupported_domains=result.unsupported_domains, completeness_summary=result.completeness_summary) diff --git a/backend-compliance/compliance/services/onboarding_service.py b/backend-compliance/compliance/services/onboarding_service.py new file mode 100644 index 00000000..c57c4f0e --- /dev/null +++ b/backend-compliance/compliance/services/onboarding_service.py @@ -0,0 +1,80 @@ +"""Onboarding Advisor service — the app-caller that loads knowledge and runs the pure orchestration. + +This is the SERVICE layer that makes the Smart Onboarding Advisor runtime-usable: it loads the curated +knowledge (certification hypotheses, signal vocabulary + map, the target's required capabilities) once +and calls the already-built, pure orchestration (normalize_signals -> silent_intake -> advisor_start). +It adds NO new reasoning logic — it only exposes what exists. No DB, no persistence (by scope). +""" + +from __future__ import annotations + +import os +from typing import Any, Dict, List, Sequence, Tuple + +import yaml + +from compliance.onboarding import ( + AdvisorResult, + CapabilityHypothesis, + OnboardingInput, + ProducedSignal, + SignalMapping, + SignalVocabularyEntry, + advisor_start, + normalize_signals, + resolve_for_certifications, + silent_intake, +) +from compliance.transition_reasoning import TargetRequirement + +_K = os.path.join(os.path.dirname(__file__), "..", "..", "knowledge") + + +def _load(*parts: str) -> Any: + return yaml.safe_load(open(os.path.join(_K, *parts), encoding="utf-8")) + + +_HYP_LIB = [CapabilityHypothesis(**h) for h in _load("certification_hypotheses", "hypotheses.yaml")["hypotheses"]] +_VOCAB = [SignalVocabularyEntry(**v) for v in _load("onboarding", "signal_vocabulary.yaml")["signals"]] +_SIGNAL_MAP = [SignalMapping(**m) for m in _load("onboarding", "intake_signal_map.yaml")["mappings"]] + +# target id -> transition pattern that defines its required capabilities (curated registry) +_TARGET_PATTERNS = { + "CRA": "transition_pattern_iso27001_to_cra_maschinenvo_v1.yaml", + "TISAX": "transition_pattern_isms_to_tisax_v1.yaml", + "MDR": "transition_pattern_iso13485_to_medical_v1.yaml", + "Environmental": "transition_pattern_iso14001_to_environmental_v1.yaml", +} + + +def supported_targets() -> List[str]: + return sorted(_TARGET_PATTERNS) + + +def _target(target_id: str) -> Tuple[List[TargetRequirement], Dict[str, List[str]]]: + pat = _load("transition_patterns", _TARGET_PATTERNS[target_id]) + reqs = [TargetRequirement(capability_id=a["capability"]) for a in pat["likely_covered"]] + reqs += [TargetRequirement(capability_id=d["capability"], question_intent=d.get("needed_information", "verify_existence"), + expected_evidence=d.get("expected_evidence", [])) for d in pat["delta_requirements"]] + covers = {d["capability"]: d.get("covers_targets", []) for d in pat["delta_requirements"]} + return reqs, covers + + +def run_advisor( + company: str, certifications: Sequence[str], target: str, + signals: Sequence[ProducedSignal], known_evidence: Sequence[str], + products: Sequence[str], markets: Sequence[str], industry: str = "", +) -> Tuple[AdvisorResult, str]: + """Producers (ProducedSignal) -> Normalizer -> Silent Pass -> Advisor. Returns an AdvisorResult. + + `target` must be a supported target id. Raises KeyError otherwise (the handler maps it to 400/404). + """ + reqs, covers = _target(target) + si = silent_intake(normalize_signals(signals, _VOCAB), _SIGNAL_MAP) + inp = OnboardingInput(company=company, industry=industry or None, products=list(products), + markets=list(markets), certifications=list(certifications), + known_evidence=list(known_evidence), target=[target]) + result = advisor_start( + inp, resolve_for_certifications(certifications, _HYP_LIB), reqs, target_id=target, + covers_targets=covers, corpus_status={target: "validated"}, detected_capabilities=si.capability_ids()) + return result, si.summary diff --git a/backend-compliance/tests/test_onboarding_endpoint.py b/backend-compliance/tests/test_onboarding_endpoint.py new file mode 100644 index 00000000..9029415a --- /dev/null +++ b/backend-compliance/tests/test_onboarding_endpoint.py @@ -0,0 +1,53 @@ +"""POST /onboarding/advisor-start — the runtime endpoint that exposes the existing Advisor. + +Exercises the router in isolation (no DB, no full app): scanner findings (ProducedSignal) -> normalize +-> Silent Pass -> Advisor -> the advisory payload. No new reasoning logic — just the wiring. +""" + +from __future__ import annotations + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from compliance.api.onboarding_routes import router + +_app = FastAPI() +_app.include_router(router) +_client = TestClient(_app) + +_BODY = { + "company": "synthetic", "industry": "machine_builder", "products": ["parking payment system"], + "markets": ["EU"], "certifications": ["ISO27001", "ISO9001"], "known_evidence": ["CE process"], + "target": "CRA", + "scanner_findings": [ + {"signal_id": "cyclonedx_found", "source_type": "repository", "evidence": "sbom", "provenance": "sbom.cdx.json"}, + {"signal_id": "vdp_found", "source_type": "website", "provenance": "/.well-known/security.txt"}, + {"signal_id": "risk_assessment_pdf", "source_type": "document", "provenance": "risk.pdf"}, + {"signal_id": "cloud_hosted", "source_type": "product"}, + ], +} + + +def test_targets_endpoint_lists_supported(): + r = _client.get("/onboarding/targets") + assert r.status_code == 200 + assert "CRA" in r.json()["targets"] + + +def test_advisor_start_returns_full_payload(): + r = _client.post("/onboarding/advisor-start", json=_BODY) + assert r.status_code == 200, r.text + d = r.json() + for field in ["silent_intake_summary", "inferred_assumptions", "rejected_assumptions", + "top_5_questions", "capability_delta", "top_measures", "evidence_requests", + "completeness_summary"]: + assert field in d + assert len(d["top_5_questions"]) <= 5 + assert d["auto_detected"] # Silent Pass recognised things from the scanners + assert "sbom_creation" not in {q["capability_id"] for q in d["top_5_questions"]} # detected -> not asked + + +def test_unknown_target_is_404(): + body = dict(_BODY, target="NOPE") + r = _client.post("/onboarding/advisor-start", json=body) + assert r.status_code == 404