Benjamin Admin
a4123ace71
This exposes the existing Smart Onboarding Advisor through a runtime endpoint; it does not add new
reasoning logic. Tightly scoped: adapter boundary + endpoint, no big frontend, no persistence, no
empirical learning, no new scanners, no LLM.
POST /onboarding/advisor-start : (company + certifications + target + scanner_findings[ProducedSignal])
-> Normalizer -> Silent Knowledge Pass -> Advisor -> { silent_intake_summary, inferred_assumptions,
rejected_assumptions, top_5_questions, capability_delta, top_measures, evidence_requests,
completeness_summary, auto_detected, headline }
GET /onboarding/targets : the supported target ids (CRA, TISAX, MDR, Environmental)
compliance/services/onboarding_service.py is the app-caller: it loads the curated knowledge (hypothesis
library, signal vocabulary + map, the target's required capabilities) once and calls the pure, tested
orchestration (normalize_signals -> silent_intake -> advisor_start). The scanner ADAPTER boundary is the
ProducedSignal format the request carries — existing scanners emit it, no new scanners. Thin handler
(<30 LOC), registered in the auto-load list. No DB. Additive to the OpenAPI contract (contract test is
additive-friendly; baseline regenerates on CI/py3.12). First deployable runtime feature -> dev deploy +
smoke. mypy --strict clean, 22 onboarding tests pass, check-loc 0.
81 lines
3.5 KiB
Python
81 lines
3.5 KiB
Python
"""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
|