Contract Enforcement End-to-End — Implementation Roadmap

Last updated: Feb 28, 2026 18:45 UTC Revision: R2 — post peer review (ChatGPT audit). Tutti i finding critici incorporati.

Executive Summary

L'analisi del codebase Vitruvyan Core rivela che la catena di garanzia contrattuale copre circa il 60% del flusso dati. I contratti Pydantic sono forti ai bordi (servizi Babel Gardens / Pattern Weavers, output GraphResponseMin) ma l'intero pipeline LangGraph interno (19+ nodi) opera senza alcuna validazione runtime. Questa roadmap descrive 8 fasi (espanse dopo peer review — 7 buchi identificati, registry allineato al codice reale, copertura estesa a nodi dominio e build_minimal_graph()) per raggiungere il 100% di enforcement.

Stato Attuale — Mappa dei Contratti

Tipi di contratto esistenti

Mappa visuale del flusso

INGESTION                   PIPELINE                           OUTPUT
─────────                   ────────                           ──────

ComprehendReq ──► Babel ──► [Pydantic ✅]
  (Pydantic ✅)   Service    ComprehensionResult
                    │
                    ▼ RAW DICT ─── contratto perso ❌
              ┌─────────────────────────────────────────┐
              │          LangGraph Pipeline              │
              │                                         │
              │  parse → intent → weaver → entity →     │
              │  babel → semantic → params → decide →   │
              │  [route] → exec/qdrant/llm →            │
              │  normalizer → orthodoxy → vault →       │
              │  compose → can → advisor → END          │
              │                                         │
              │  TUTTI I NODI: Dict[str, Any] ❌        │
              │  State = TypedDict(total=False) ❌      │
              │  NodeContract.required_fields ❌ UNUSED │
              │                                         │
              │  ⚠️ codex_hunters → END (skip ortho)   │
              └─────────────────────────────────────────┘
                    │
                    ▼ RAW DICT
              graph_adapter.py → GraphResponseMin ✅
                                  SessionMin ✅
                                  OrthodoxyStatus ✅
                                       │
                                       ▼
EVENT BUS                         HTTP Response
StreamBus.emit()                  al client
  payload: Any ❌
  no schema ❌

RAG / Qdrant
  upsert: warn-only ⚠️

I 7 Buchi Critici

BUCO 1 — Pipeline LangGraph senza validazione (CRITICO)

• Dove: tra tutti i 20 nodi in graph_flow.py (linee 272-500)
• Causa: BaseGraphState è TypedDict(total=False) — tutti i campi opzionali, nessun enforcement runtime. TypedDict è un costrutto esclusivamente per type checker statici (mypy/pyright), Python non lo valida mai a runtime.
• Conseguenza: un nodo può scrivere state["intent"] = 42 (tipo sbagliato) o omettere input_text, e nulla lo intercetta.
• Evidenza: NodeContract.required_fields e produced_fields sono dichiarati nella dataclass (graph_engine.py:35-45) ma GraphEngine.get_all_nodes() (graph_engine.py:257) non li verifica mai. Inoltre, graph_flow.py NON usa GraphEngine per il pipeline principale — assembla il grafo direttamente.

BUCO 2 — Boundary servizio→grafo perde Pydantic (ALTO)

• Dove: pw_compile_node.py:89, pattern_weavers_node.py, emotion_detector.py
• Causa: i nodi fanno state["ontology_payload"] = result.get("payload", {}) — raw dict. OntologyPayload.model_validate() non viene mai chiamato nonostante il modello Pydantic con extra="forbid" esista.
• Conseguenza: la validazione forte del servizio Pattern Weavers (via FastAPI) viene persa quando il dato entra nel grafo.

BUCO 3 — Orthodoxy saltata su codex_hunters (ALTO)

• Dove: graph_flow.py:440 — conditional edge codex_hunters → END (success path)
• Causa: il path di successo bypassa orthodoxy_node, vault_node, compose_node, can_node.
• Conseguenza: nessun orthodoxy_status settato. Il contratto GraphResponseMin lo richiede come campo obbligatorio — graph_adapter.py:183 mappa qualsiasi valore sconosciuto a "blessed" (default silenzioso).
• Confronto: early_exit_node.py:85-93 setta correttamente tutti i campi orthodoxy.

BUCO 4 — Event Bus accetta qualsiasi payload (MEDIO)

• Dove: streams.py:209-260 — metodo StreamBus.emit()
• Causa: emit() accetta payload: Dict[str, Any] e fa json.dumps(payload) + XADD. TransportEvent e CognitiveEvent esistono come dataclass ma non vengono mai costruiti da emit().
• Conseguenza: payload arbitrari circolano nel bus senza contratto.

BUCO 5 — RAG warn-only di default (MEDIO)

• Dove: qdrant_agent.py:62 — _check_collection_registered() e _check_payload_contract() (linea 77)
• Causa: RAG_ENFORCE_REGISTRY default = "warn". Collezioni non dichiarate e payload senza source → warning nei log, dato che passa.
• Conseguenza: violazioni contrattuali invisibili in produzione.
• Aggravante (trovata in peer review): _check_payload_contract() fa SOLO logger.warning() anche quando _RAG_ENFORCE == "strict". La modalità strict alza ValueError solo in _check_collection_contract() (collezioni), NON per payload mancante di source. Quindi anche con strict abilitato, payload invalidi passano.

BUCO 6 — graph_adapter fallback silenzioso a "blessed" (ALTO)

• Dove: services/api_graph/adapters/graph_adapter.py:205 — _CANONICAL_MAP.get(raw_status, "blessed")
• Causa: se orthodoxy_status e orthodoxy_verdict sono entrambi assenti o vuoti nello state, raw_status è "". _CANONICAL_MAP.get("", "blessed") restituisce il default "blessed" — senza log, senza warning.
• Conseguenza: qualsiasi path che non setta orthodoxy (es. codex_hunters, errori, future regressioni) viene automaticamente marcato come "blessed" nell'output API. Maschera problemi reali.
• Evidenza: _CANONICAL_MAP a linea 197 non contiene "" come chiave, quindi .get("", "blessed") usa sempre il default.

BUCO 7 — Nodi dominio e build_minimal_graph() non coperti (ALTO)

• Dove: graph_flow.py:323-355 (nodi dominio), graph_flow.py:560-580 (build_minimal_graph())
• Causa: I nodi dominio vengono caricati dinamicamente a runtime da domains.<GRAPH_DOMAIN>.graph_nodes.registry via _load_domain_graph_extension(). Vengono aggiunti al grafo con g.add_node() ma senza alcun wrapping @enforced. Separatamente, build_minimal_graph() costruisce un grafo a 4 nodi (parse → intent → decide → compose → END) senza wrapping.
• Conseguenza: due path runtime senza enforcement contrattuale:
  1. Nodi domain-specific aggiunti via plugin non vengono validati
  2. Il grafo minimale (usato quando ENABLE_MINIMAL_GRAPH=true) bypassa completamente il sistema di enforcement

Decisione Architetturale: Decorator @enforced (Non Agent)

Opzione scartata: ContractAgent

Un ContractAgent centralizzato sarebbe un anti-pattern perché:

1. Gli Agent in Vitruvyan wrappano I/O esterno (PostgreSQL/Qdrant/OpenAI). La validazione contrattuale è logica interna pura.
2. Richiederebbe chiamate esplicite contract_agent.validate() in ogni nodo — opt-in, dimenticabile.
3. Creerebbe un god object con accoppiamento universale.

Opzione scelta: decorator @enforced + wrapping centralizzato in build_graph()

• Cross-cutting concern implementato come middleware/decorator
• Applicato una volta sola nel build_graph(), non in ogni file nodo
• Non bypassabile — il wrapping avviene all'assemblaggio del grafo
• Configurabile — ENV var CONTRACT_ENFORCE_MODE (warn/strict/off)

Fasi di Implementazione

FASE 1 — Decorator @enforced + ContractViolation (2 file nuovi)

Obiettivo: creare il meccanismo di enforcement senza toccare nessun nodo esistente.

File da creare:

1. vitruvyan_core/core/orchestration/contract_enforcement.py (~120 righe)
2. vitruvyan_core/core/orchestration/tests/test_contract_enforcement.py (~80 righe)

Specifiche del decorator:

@enforced(
    requires=["input_text"],
    produces=["intent", "needs_clarification"],
    validate_types={                          # opzionale
        "input_text": str,
        "intent": str,
    }
)
def intent_detection_node(state: dict) -> dict:
    ...

Comportamento:

• PRE: verifica che ogni campo in requires esista in state e non sia None
• POST: verifica che ogni campo in produces esista nel dict restituito
• TYPE (opzionale): verifica isinstance() sui campi dichiarati
• Modalità controllata da env CONTRACT_ENFORCE_MODE (letta UNA VOLTA a import-time per zero overhead):
  • warn (default): log WARNING + incremento contatore interno (nome metric string only)
  • strict: raise ContractViolationError (per test/staging)
  • off: restituisce la funzione originale non wrappata — zero overhead reale (nessun wrapper)

LIVELLO 1 compliance: il file espone SOLO costanti con i nomi delle metriche (es. METRIC_CONTRACT_VIOLATIONS = "contract_violations_total"). Nessun from prometheus_client import .... L'istanziazione delle metriche Prometheus avviene in LIVELLO 2 (servizio), seguendo il pattern di vault_keepers/monitoring/metrics.py e codex_hunters/monitoring/__init__.py.

Performance off mode:

_MODE = os.getenv("CONTRACT_ENFORCE_MODE", "warn")  # letto UNA VOLTA
 
def enforced(requires, produces):
    def decorator(fn):
        if _MODE == "off":
            return fn  # restituisce funzione originale — zero wrapping
        @functools.wraps(fn)
        def wrapper(state):
            ...
        return wrapper
    return decorator

Dipendenze: nessuna esterna (pure Python + logging). NO prometheus_client in LIVELLO 1.

Verifica:

python3 -c "from core.orchestration.contract_enforcement import enforced; print('✅')"
pytest vitruvyan_core/core/orchestration/tests/test_contract_enforcement.py -v

FASE 2 — Registry contratti per tutti i 20 nodi (1 file nuovo)

Obiettivo: definire requires/produces per ogni nodo in un registro centralizzato.

File da creare: vitruvyan_core/core/orchestration/node_contracts_registry.py (~150 righe)

Contenuto — dizionario NODE_CONTRACTS:

⚠️ R2: Tabella completamente riscritta dopo audit del codice reale (peer review). Ogni campo è stato verificato contro il return statement effettivo del nodo.

Differenze rispetto alla versione R1 (pre-peer review):

• parse: language → domain_params, route (parse non produce language, lo fa intent_detection)
• intent_detection: solo intent → aggiunto language, language_detected, babel_status, route
• weaver: solo weaver_context → aggiunto weave_result, weave_confidence
• entity_resolver: entity_ids → [] (delega a registry, output variabile)
• cached_llm: result → llm_response (campo effettivo diverso)
• output_normalizer: requires result, produces response → requires [], produces result (normalizza in-place)
• orthodoxy: requires response → requires narrative + produce 5 campi, non 1
• vault: requires response, produces vault_status → requires [], produces vault_blessing, route
• compose: requires response → requires [], produce narrative + action
• can: final_response → narrative, follow_ups, route (campo final_response non esiste)
• exec: result → [] (delega a registry)
• codex_hunters: aggiunto codex_success, response
• early_exit: rimosso final_response (non esiste)

Dipendenze: FASE 1 completata.

Verifica:

python3 -c "from core.orchestration.node_contracts_registry import NODE_CONTRACTS; print(f'✅ {len(NODE_CONTRACTS)} contratti')"

FASE 3 — Applicazione a build_graph() (1 file modificato)

Obiettivo: wrappare ogni nodo al momento della registrazione nel grafo.

File da modificare: vitruvyan_core/core/orchestration/langgraph/graph_flow.py

Cambiamento (~25 righe aggiunte, zero modifiche ai file dei singoli nodi):

from core.orchestration.contract_enforcement import enforced
from core.orchestration.node_contracts_registry import NODE_CONTRACTS
 
def _wrap(name, fn):
    """Wrap node function with contract enforcement if registered."""
    c = NODE_CONTRACTS.get(name)
    if c:
        return enforced(requires=c.requires, produces=c.produces)(fn)
    return fn
 
# Nel build_graph():
g.add_node("parse", _wrap("parse", parse_node))
g.add_node("intent_detection", _wrap("intent_detection", intent_detection_node))
g.add_node("weaver", _wrap("weaver", weaver_node))
# ... per tutti i 20 nodi

Impatto: il wrapping è centralizzato in un solo punto. I nodi non sanno di essere wrappati.

⚠️ R2 — Copertura nodi dominio e build_minimal_graph() (BUCO 7 fix):

Il wrapping _wrap() DEVE coprire anche:

1. Nodi dominio caricati dinamicamente (linee 323-355 di graph_flow.py):

# Dopo il caricamento dei nodi dominio:
for ext_name, ext_fn in _graph_nodes_ext.items():
    if ext_name not in registered_nodes:
        g.add_node(ext_name, _wrap(ext_name, ext_fn))  # ← wrapping!
        registered_nodes.add(ext_name)

I nodi dominio NON avranno entry nel NODE_CONTRACTS registry (sono sconosciuti a compile-time). _wrap() restituirà la funzione invariata (NODE_CONTRACTS.get(name) → None → return fn). Per aggiungere contratti ai nodi dominio, il domain plugin deve esportare anche i contratti via get_<domain>_node_contracts() → Dict[str, NodeContract] che viene unito a NODE_CONTRACTS a runtime.

2. build_minimal_graph() (linee 560-580):

def build_minimal_graph():
    g = StateGraph(GraphState)
    g.add_node("parse", _wrap("parse", parse_node))       # ← wrapping!
    g.add_node("intent", _wrap("intent_detection", intent_detection_node))
    g.add_node("decide", _wrap("decide", route_node))
    g.add_node("compose", _wrap("compose", compose_node))
    ...

Dipendenze: FASE 1 + FASE 2 completate.

Verifica:

# Compilazione del grafo con enforcement attivo
CONTRACT_ENFORCE_MODE=warn python3 -c "
from core.orchestration.langgraph.graph_flow import build_graph
g = build_graph()
print('✅ Graph compiled with contract enforcement')
"
 
# Pipeline end-to-end con enforcement
CONTRACT_ENFORCE_MODE=warn python3 -c "
from core.orchestration.langgraph.graph_runner import run_graph_once
result = run_graph_once('hello', user_id='test_contracts')
print(f'✅ Pipeline completa, route={result.get(\"route\")}')
"
 
# Verifica build_minimal_graph coperto
CONTRACT_ENFORCE_MODE=warn ENABLE_MINIMAL_GRAPH=true python3 -c "
from core.orchestration.langgraph.graph_flow import build_minimal_graph
g = build_minimal_graph()
print('✅ Minimal graph compiled with contract enforcement')
"

FASE 4 — Fix codex_hunters → END senza orthodoxy (1 file modificato)

Obiettivo: garantire che orthodoxy_status sia SEMPRE presente in output.

File da modificare: vitruvyan_core/core/orchestration/langgraph/node/codex_hunters_node.py

Cambiamento: nel branch di successo (prima del return), aggiungere:

from contracts.graph_response import ORTHODOXY_BLESSED  # costante canonica
 
state["orthodoxy_status"] = ORTHODOXY_BLESSED
state["orthodoxy_verdict"] = ORTHODOXY_BLESSED
state["orthodoxy_confidence"] = 0.99
state["orthodoxy_findings"] = 0
state["orthodoxy_message"] = "Codex expedition: maintenance system response"
state["orthodoxy_timestamp"] = datetime.utcnow().isoformat()

Pattern identico a early_exit_node.py:85-93.

Prerequisito (FASE 4a): aggiungere costanti canoniche in contracts/graph_response.py:

# Canonical orthodoxy status constants (avoid hardcoded strings)
ORTHODOXY_BLESSED = "blessed"
ORTHODOXY_PURIFIED = "purified"
ORTHODOXY_HERETICAL = "heretical"
ORTHODOXY_NON_LIQUET = "non_liquet"
ORTHODOXY_CLARIFICATION_NEEDED = "clarification_needed"

Tutti i nodi che settano orthodoxy_status (early_exit, codex_hunters, orthodoxy_node, graph_adapter) devono importare queste costanti anziché usare stringhe letterali.

Alternativa (modifica topologica): cambiare l'edge in graph_flow.py per far passare anche Codex per output_normalizer → orthodoxy → vault → compose → can → END.

Pro: enforcement strutturale. Contro: aggiunge latenza al path di manutenzione.

Dipendenze: nessuna (parallelizzabile con FASE 1-3).

Verifica:

python3 -c "
from contracts.graph_response import GraphResponseMin, SessionMin
from datetime import datetime, timezone
# Simula output codex senza orthodoxy → deve fallire
try:
    GraphResponseMin(
        human='test', follow_ups=[], orthodoxy_status=None,
        route_taken='codex', correlation_id='x',
        as_of=datetime.now(timezone.utc),
        session_min=SessionMin(user_id='a', session_id='a', turn_id='a')
    )
    print('❌ Doveva fallire')
except Exception as e:
    print(f'✅ Pydantic ha bloccato input invalido')
"

FASE 4b — Eliminare fallback silenzioso in graph_adapter.py (BUCO 6 fix)

Obiettivo: il default "blessed" silenzioso in _CANONICAL_MAP.get(raw_status, "blessed") maschera bug. Qualsiasi status vuoto o sconosciuto deve essere visibile, non mascherato.

File da modificare: services/api_graph/adapters/graph_adapter.py (~linea 205)

Cambiamento:

# Da (silenzioso):
orthodoxy_status: OrthodoxyStatusType = _CANONICAL_MAP.get(raw_status, "blessed")
 
# A (esplicito — log + default difensivo ma tracciabile):
if raw_status not in _CANONICAL_MAP:
    logger.warning(
        f"[GRAPH_ADAPTER] orthodoxy_status '{raw_status}' non canonico — "
        f"fallback a 'non_liquet'. Controllare il path del grafo."
    )
orthodoxy_status: OrthodoxyStatusType = _CANONICAL_MAP.get(raw_status, "non_liquet")

Razionale: il default difensivo cambia da "blessed" a "non_liquet" ("non determinato"). Un output non validato NON dovrebbe essere considerato "benedetto" — deve essere segnalato come incerto. Il warning nel log rende visibile il problema.

Dipendenze: FASE 4a (costanti canoniche).

FASE 5 — Validazione payload lato emittente (5-6 file servizio modificati, 0 file bus)

Obiettivo: validazione dei payload emessi sul bus PRIMA della chiamata a emit(), mantenendo il bus 100% payload-blind.

Principio architetturale: il sacro invariante dice:

"Sacred invariant: the bus is payload-blind (no semantic routing/correlation/synthesis in transport)"

Aggiungere validazione dentro StreamBus.emit() violerebbe questo principio — il trasporto "guarderebbe" il payload. La soluzione corretta è validare lato emittente, prima di chiamare emit().

Approccio (caller-side validation):

Ogni servizio che emette eventi valida il payload prima dell'invio:

# Nell'emittente (es. bus_adapter.py di un Sacred Order):
from contracts.comprehension import ComprehensionResult
 
result = ComprehensionResult.model_validate(payload)  # emittente valida
bus.emit("babel.comprehension.completed", result.model_dump())  # bus riceve dato già validato

File da modificare (5-6 servizi):

1. services/api_babel_gardens/adapters/bus_adapter.py — attualmente self.bus.emit(stream, payload) senza validazione
2. services/api_pattern_weavers/adapters/bus_adapter.py — emit_weave_result() e publish_semantic_search_results() senza contratto
3. services/api_orthodoxy_wardens/adapters/bus_adapter.py
4. services/api_vault_keepers/adapters/bus_adapter.py
5. services/api_codex_hunters/adapters/bus_adapter.py
6. services/api_memory_orders/adapters/bus_adapter.py (se presente)

File NON modificato: vitruvyan_core/core/synaptic_conclave/transport/streams.py — il bus resta intatto e payload-blind.

Trade-off: la validazione è opt-in per ogni emittente (stessa natura del decorator @enforced — l'enforcement è al confine del componente, non nel trasporto).

⚠️ R2 — Meccanismo di enforcement CI (per compensare il trade-off opt-in):

Per garantire che tutti gli emittenti validino, aggiungere un test architetturale CI:

# tests/architectural/test_bus_emit_validation.py
import ast, pathlib
 
def test_all_bus_emit_calls_have_validation():
    """Verifica che ogni bus.emit() sia preceduto da model_validate() nello stesso metodo."""
    services_dir = pathlib.Path("services")
    violations = []
    for adapter in services_dir.rglob("bus_adapter.py"):
        source = adapter.read_text()
        tree = ast.parse(source)
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                # Cerca chiamate a .emit() senza .model_validate() nel metodo parent
                ...
    assert not violations, f"bus.emit() senza validazione: {violations}"

Questo test CI trasforma l'opt-in in un gate bloccante: i servizi non possono fare merge senza validare i payload.

Dipendenze: nessuna (parallelizzabile).

Verifica:

python3 -c "
from pydantic import BaseModel
 
class AuditEvent(BaseModel):
    source: str
    entity_id: str
 
# Emittente valida PRIMA di emit()
try:
    AuditEvent.model_validate({'source': 'test', 'entity_id': 'E1'})
    print('✅ Payload valido — pronto per emit()')
except: pass
 
try:
    AuditEvent.model_validate({'wrong': 'data'})
    print('❌ Doveva fallire')
except Exception as e:
    print(f'✅ Validazione ha bloccato prima di emit(): {type(e).__name__}')
"

FASE 6 — RAG enforce strict mode (1 env var + audit preventivo)

Obiettivo: passare da warn-only a strict per QdrantAgent.

File da modificare: vitruvyan_core/core/agents/qdrant_agent.py

Prerequisito: eseguire audit preventivo:

python3 scripts/audit_rag_collections.py
# Deve restituire 0 collezioni non dichiarate

Cambiamento in due parti:

Parte A — Il default resta warn (zero breaking changes). La modalità strict viene attivata esplicitamente:

• CI/staging: RAG_ENFORCE_REGISTRY=strict nel docker-compose di staging
• Produzione: si attiva manualmente SOLO dopo audit a zero violazioni

# Il codice NON cambia default — resta warn:
self.enforce_mode = os.getenv("RAG_ENFORCE_REGISTRY", "warn")

Parte B (⚠️ R2 — BUCO 5 aggravante) — Fix _check_payload_contract() per strict mode:

Attualmente _check_payload_contract() (qdrant_agent.py:77) fa solo logger.warning() anche quando _RAG_ENFORCE == "strict". Il raise ValueError esiste solo in _check_collection_contract() per le collezioni non dichiarate, ma i payload senza source passano sempre silenziosamente.

# Da (warn-only sempre):
def _check_payload_contract(points, collection):
    if _RAG_ENFORCE == "off":
        return
    for i, p in enumerate(points):
        payload = p.get("payload", {})
        if not payload.get("source"):
            logger.warning(f"[RAG GUARD] Point {i} in '{collection}' missing payload.source")
            break
 
# A (strict-aware):
def _check_payload_contract(points, collection):
    if _RAG_ENFORCE == "off":
        return
    for i, p in enumerate(points):
        payload = p.get("payload", {})
        if not payload.get("source"):
            msg = f"[RAG GUARD] Point {i} in '{collection}' missing payload.source"
            if _RAG_ENFORCE == "strict":
                raise ValueError(msg)
            logger.warning(msg)
            break

Dipendenze: audit preventivo con audit_rag_collections.py per verificare che tutte le collezioni in Qdrant siano dichiarate in ALL_DECLARED_COLLECTIONS E che tutti i payload abbiano source. Solo dopo audit a zero violazioni si può abilitare strict.

Verifica:

# Step 1: Audit (prerequisito)
python3 scripts/audit_rag_collections.py
# Deve restituire 0 collezioni non-dichiarate
 
# Step 2: Test strict mode — collezioni
RAG_ENFORCE_REGISTRY=strict python3 -c "
from core.agents.qdrant_agent import QdrantAgent
print('✅ QdrantAgent in strict mode')
"
 
# Step 3: Test strict mode — payload.source
RAG_ENFORCE_REGISTRY=strict python3 -c "
from core.agents.qdrant_agent import _check_payload_contract
try:
    _check_payload_contract([{'payload': {}}], 'test_collection')
    print('❌ Doveva alzare ValueError')
except ValueError:
    print('✅ Strict mode blocca payload senza source')
"

FASE 7 — Pydantic re-validation ai boundary servizio→grafo (2-3 file modificati)

Obiettivo: quando un nodo LangGraph riceve JSON da servizi esterni, rivalidare con il contratto Pydantic.

File da modificare:

1. vitruvyan_core/core/orchestration/langgraph/node/pw_compile_node.py (~linea 89):

# Da:
state["ontology_payload"] = result.get("payload", {})
# A:
from contracts.pattern_weavers import OntologyPayload
raw = result.get("payload", {})
try:
    OntologyPayload.model_validate(raw)
except Exception as e:
    logger.warning(f"[PW_COMPILE] OntologyPayload validation failed: {e}")
state["ontology_payload"] = raw

2. vitruvyan_core/core/orchestration/langgraph/node/pattern_weavers_node.py: stesso pattern per weaver_context.

3. vitruvyan_core/core/orchestration/langgraph/node/emotion_detector.py: stesso pattern per i campi emotion.

Dipendenze: nessuna (parallelizzabile).

FASE 8 — Test E2E di conformità contrattuale (1 file nuovo)

Obiettivo: test automatizzato che verifica l'intera catena contrattuale.

File da creare: tests/architectural/test_pipeline_contract_enforcement.py (~100 righe)

Casi di test:

1. run_graph_once("hello") → risultato contiene orthodoxy_status (non None, non vuoto)
2. run_graph_once("hello") → risultato serializzabile in GraphResponseMin senza errori Pydantic
3. run_graph_once("analyze this data") → risultato contiene route, intent, narrative
4. Con CONTRACT_ENFORCE_MODE=strict, nessun ContractViolationError durante esecuzione
5. Path early_exit → tutti i campi orthodoxy presenti e con valori canonici
6. Path codex_hunters → orthodoxy_status presente
7. orthodoxy_status del risultato finale è tra i 5 valori canonici: blessed|purified|heretical|non_liquet|clarification_needed

Dipendenze: FASI 1-4 completate.

Verifica:

CONTRACT_ENFORCE_MODE=strict pytest tests/architectural/test_pipeline_contract_enforcement.py -v

Grafo delle Dipendenze

FASE 1 ─────────────────────┐
  (decorator @enforced)      │
                              ├──► FASE 3 (applicazione a graph_flow.py)
FASE 2 ─────────────────────┘              │
  (registry contratti)                      │
                                            ├──► FASE 8 (test E2E)
FASE 4 ────────────────────────────────────┘
  (codex orthodoxy fix)

FASE 5 ──── indipendente (StreamBus schema)
FASE 6 ──── indipendente (RAG strict, dopo audit)
FASE 7 ──── indipendente (Pydantic boundary)

Parallelizzazione possibile: FASI 4, 5, 6, 7 possono partire tutte in parallelo con FASE 1.

Stima Effort

Matrice di Copertura Pre/Post

Architettura Target

       @enforced decorator (LIVELLO 1 — pure Python, zero I/O)
              │
              ▼
┌──────────────────────────────────────────────────────────────┐
│                    LangGraph Pipeline                         │
│                                                              │
│  parse ──► intent ──► weaver ──► entity ──► babel ──►       │
│  semantic ──► params ──► decide ──► [route] ──►             │
│  exec/qdrant/llm ──► normalizer ──► orthodoxy ──►           │
│  vault ──► compose ──► can ──► advisor ──► END              │
│                                                              │
│  OGNI NODO: @enforced(requires=[...], produces=[...])        │
│  Validazione automatica PRE (input) e POST (output)          │
│  Violazione → log WARNING + Prometheus metric                │
│  CONTRACT_ENFORCE_MODE=strict → raise (staging/test)         │
│  CONTRACT_ENFORCE_MODE=off → no-op (escape hatch)            │
└──────────────────────────────────────────────────────────────┘
              │
              ▼
       GraphResponseMin (Pydantic — checkpoint finale)

Vincoli Non-Negoziabili

1. Zero breaking changes: modalità warn di default. Nessun nodo esistente smette di funzionare.
2. Zero modifiche ai file dei nodi PER IL WRAPPING ENFORCEMENT: il decorator @enforced viene applicato esclusivamente in build_graph() e build_minimal_graph(). I file dei nodi non importano né conoscono il decorator. Le FASI 4 e 7 richiedono modifiche puntuali ai nodi per bug fix strutturali (campi orthodoxy mancanti in codex_hunters, re-validazione Pydantic ai boundary) — queste NON sono modifiche di enforcement wrapping.
3. LIVELLO 1 compliance: contract_enforcement.py è pure Python — no I/O, no Redis, no Postgres, NO prometheus_client. Solo nomi metriche come stringhe costanti.
4. Bus payload-blind: StreamBus.emit() NON viene modificato. La validazione avviene lato emittente.
5. Nessun god object: no ContractAgent, no singleton globale. Il decorator è funzionale e stateless.
6. No hardcoded strings: i valori orthodoxy (blessed, purified, etc.) sono costanti importate da contracts/graph_response.py, non stringhe letterali sparse.
7. Core agnostic: nessuna logica domain-specific introdotta in vitruvyan_core/core/. I contratti sono domain-agnostic.
8. Performance off mode: quando CONTRACT_ENFORCE_MODE=off, il decorator restituisce la funzione originale — zero wrapping, zero overhead.
9. RAG backward-compat: il default RAG_ENFORCE_REGISTRY resta warn. strict si attiva via env var solo dopo audit.

Riferimenti al Codice

Compliance Audit — Verifica Principi Architetturali

Questa sezione documenta la verifica del roadmap contro tutti i principi architetturali di Vitruvyan Core.

Review R1 (self-audit Copilot)

Violazioni trovate e corrette (R1)

Review R2 (peer review — ChatGPT audit, Feb 28 2026)

Verdetto peer review: NO-GO temporaneo con 7 finding. Tutti incorporati:

Post-R2 status: tutti i finding critici incorporati. Roadmap eseguibile.