diff --git a/.l9/baselines/packet-envelope.yml b/.l9/baselines/packet-envelope.yml index 452de7ab..861e537d 100644 --- a/.l9/baselines/packet-envelope.yml +++ b/.l9/baselines/packet-envelope.yml @@ -18,19 +18,6 @@ entries: removal_condition: migrated-to:TransportPacket root_cause_group: packet-envelope-migration evidence: chassis/pii.py::string-annotation:For PacketEnvelope.security.pii_fields. -- id: packet-envelope/engine-graph-graph-sync-client-fix-py-string-annotation - gate: pre-commit/packet-envelope-prohibited - rule: packet-envelope-prohibited - fingerprint: 3743f9ff50d0e4dbb3567bc40dfe2085a29eaf1b6f3c115cf55f291965c1089a - path: engine/graph/graph_sync_client_fix.py - owner: '@cryptoxdog' - issue: Quantum-L9/Cognitive.Engine.Graphs#138 - introduced_before: af986d0 - expires: '2026-10-21' - removal_condition: migrated-to:TransportPacket - root_cause_group: packet-envelope-migration - evidence: engine/graph/graph_sync_client_fix.py::string-annotation:Build and validate a PacketEnvelope - before sending. - id: packet-envelope/engine-packet-bridge-py-import-from gate: pre-commit/packet-envelope-prohibited rule: packet-envelope-prohibited diff --git a/docs/adr/ADR-DEC-001-candidate-identity.md b/docs/adr/ADR-DEC-001-candidate-identity.md index 3856b900..9486dbd3 100644 --- a/docs/adr/ADR-DEC-001-candidate-identity.md +++ b/docs/adr/ADR-DEC-001-candidate-identity.md @@ -21,13 +21,15 @@ The **live** match handler instead reads a bare Neo4j node property `entity_id`: - `engine/handlers.py:509`, `:616`, `:1497` — candidate identity read from the `entity_id` property. -- `engine/graph/graph_sync_client_fix.py:113` — `entity_id` is client-supplied at - sync time (`MERGE (n {entity_id: row.entity_id, tenant: $tenant})`) and is **not** - defined in `engine/config/schema.py`. +- No canonical writer sets `entity_id` at all. The live sync path is + `engine/handlers.py::handle_sync` → `engine/sync/generator.py::SyncGenerator`, + which MERGEs on the domain-declared `idproperty` (`facility_id`, `code`, + `form_id`, `opportunity_id`, `demand_id` in `domains/plasticos/spec.yaml`) — + never on `entity_id`, which is **not** defined in `engine/config/schema.py`. This is a live-vs-contract divergence: the contract's identity is a governed, -namespaced `entity_ref`; the running code keys on an ungoverned, client-supplied -`entity_id` node property. DEC-001 records how candidate identity is defined so the +namespaced `entity_ref`; the running code keys on an ungoverned `entity_id` node +property that no canonical writer populates. DEC-001 records how candidate identity is defined so the divergence is resolved deliberately rather than by accident. ## Options Considered @@ -63,15 +65,29 @@ never an implicit reinterpretation of a raw integer or a database node id. ## Residual Reconciliation Task The live handler still keys candidate identity on the ungoverned `entity_id` node -property (`engine/handlers.py:509,616,1497`; written at -`engine/graph/graph_sync_client_fix.py:113`), which is not schema-defined. A -follow-up must align the live handler and sync path with the contract `entity_ref` +property (`engine/handlers.py:509,616,1497`), which is not schema-defined and which +no canonical writer produces — the handler reads it through silent fallbacks +(`.get("entity_id", "")`). A follow-up must align the live handler and sync path +with the contract `entity_ref` (schema-define the identity property, or resolve `entity_ref` → stored key through the resolver) so runtime identity matches the contract this ADR ratifies. Until then, the divergence is a tracked residual risk, not a resolved state. +## Citation Correction (2026-08-23) + +This ADR originally cited `engine/graph/graph_sync_client_fix.py:113` as the place +where `entity_id` was "client-supplied at sync time". That module was an unwired +gap-fix artifact with no caller anywhere in the repository, and it was removed by +the gap-fix artifact convergence audit +(`docs/audits/2026-08-23-gap-fix-artifact-convergence/`). Its Cypher +(`MERGE (n {entity_id: row.entity_id, tenant: $tenant})`) never executed. + +The correction widens rather than narrows the divergence this ADR records: the +handler reads `entity_id`, and nothing writes it. The decision (OPTION-B) and the +residual reconciliation task are unchanged. + ## Artifacts `engine/models/payloads.py`, `contracts/payloads/examples/match-response.json`, `contracts/match_response.json`, `engine/handlers.py`, -`engine/graph/graph_sync_client_fix.py`. +`engine/sync/generator.py`. diff --git a/docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_ARTIFACT_INVENTORY.yaml b/docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_ARTIFACT_INVENTORY.yaml new file mode 100644 index 00000000..e1f22f9a --- /dev/null +++ b/docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_ARTIFACT_INVENTORY.yaml @@ -0,0 +1,183 @@ +# Phase 1 — gap-fix island inventory +# Contract: CEG-GAP-FIX-ARTIFACT-REACHABILITY-CONVERGENCE-2026-08-23 +# Base SHA: 5868bc49865eba0afd6154a0746ac06111cf1ccf (origin/main = merge commit of +# PR #232, "fix(inference): remove ghost bridge and unowned KB rule loading") +# +# Discovery commands: +# find engine tests -type f | sort | grep -E 'gap|patch|fix|return_channel|community_export|audit_persistence|graph_sync' +# git grep -n -E '' -- engine chassis tests contracts domains docs tools Makefile +# AST import-graph reachability from chassis/ ingress + engine/boot.py + engine/__init__.py + +meta: + base_sha: 5868bc49865eba0afd6154a0746ac06111cf1ccf + open_prs_at_audit: + - number: 233 + head: chore/auto-seed-governance + overlap_with_scope: none # governance/CI seed files only, no engine/ or tests/ paths + note: > + Every island artifact entered the repository in the single squashed import + commit 0979ff5 (2026-07-23, "ci: activate l9-ci-core governed pipeline"). + No artifact has an independent introduction commit, so introduction_commit + carries no per-file provenance signal for this island. + +artifacts: + + - path: engine/compliance/audit_persistence.py + exists: true + current_blob_sha: 91e5dae2fcb0999efbda11ca2e6a84b6352a95a1 + introduction_commit: 0979ff5 + latest_change_commit: 0979ff5 + declared_owner: engine-team # L9_META + declared_status: active # L9_META — contradicted by reachability + imports: [asyncpg (TYPE_CHECKING only)] + imported_by: + production: [] + tests: [tests/gap_fixes/test_gap5_audit.py] + recipe: [engine/startup_wiring.py] # via `from shared.audit_persistence import ...` — package `shared` does not exist + public_exports: [] # engine/compliance/__init__.py exports only AuditLogger, ComplianceEngine, PIIHandler, ProhibitedFactorValidator + tests: [tests/gap_fixes/test_gap5_audit.py] + historical_plan_refs: [Cursor-Governance docs/plans/BUILT/wire_gap-fix_modules_7d4d9028.plan.md] + current_architecture_refs: [] + + - path: engine/graph_return_channel.py + exists: true + current_blob_sha: c4b51481f45dae55da078bcb53166f2ed420505a + introduction_commit: 0979ff5 + latest_change_commit: 0979ff5 + declared_owner: engine-team + declared_status: active + imports: [engine.contract_enforcement] + imported_by: + production: [] + island: [engine/graph/community_export.py, engine/convergence_controller_patch.py] + tests: [tests/gap_fixes/test_gap2_return_channel.py] + recipe: [engine/startup_wiring.py] + public_exports: [] # not in engine/__init__.py __all__ + tests: [tests/gap_fixes/test_gap2_return_channel.py] + historical_plan_refs: [wire_gap-fix_modules_7d4d9028.plan.md — planned move to engine/feedback/graph_return_channel.py, never landed] + current_architecture_refs: [] + + - path: engine/graph/community_export.py + exists: true + current_blob_sha: a56a1b47044a1a4940bbf333ed64d4106887d205 + introduction_commit: 0979ff5 + latest_change_commit: 0979ff5 + declared_owner: engine-team + declared_status: active + imports: [engine.graph_return_channel] + imported_by: + production: [] + tests: [] + recipe: [engine/startup_wiring.py] # via `from graph.community_export import ...` — top-level package `graph` does not exist + public_exports: [] # engine/graph/__init__.py exports only GraphDriver + tests: [] + historical_plan_refs: [wire_gap-fix_modules_7d4d9028.plan.md] + current_architecture_refs: [] + + - path: engine/convergence_controller_patch.py + exists: true + current_blob_sha: 8d1691135aa554e427e586696e029f8548700eec + introduction_commit: 0979ff5 + latest_change_commit: 0979ff5 + declared_owner: engine-team + declared_status: active + imports: [engine.graph_return_channel, engine.contract_enforcement, chassis.events (dynamic)] + imported_by: + production: [] + tests: [] + recipe: [] + public_exports: [] + tests: [] + historical_plan_refs: [wire_gap-fix_modules_7d4d9028.plan.md — planned split into engine/feedback/enrich_helpers.py, never landed] + current_architecture_refs: [] + + - path: engine/graph/graph_sync_client_fix.py + exists: true + current_blob_sha: 2bc8f04130de16b0052f33d7319d461edcf5d042 + introduction_commit: 0979ff5 + latest_change_commit: 0979ff5 + declared_owner: engine-team + declared_status: active + imports: [engine.contract_enforcement] + imported_by: + production: [] + tests: [] + recipe: [] + public_exports: [] + tests: [] + historical_plan_refs: [wire_gap-fix_modules_7d4d9028.plan.md — listed for deletion, never deleted] + current_architecture_refs: + - docs/adr/ADR-DEC-001-candidate-identity.md # STALE CITATION — see classification + + # ── Island members discovered during Phase 1, beyond the mandatory set ── + + - path: engine/contract_enforcement.py + exists: true + current_blob_sha: null # unchanged in this PR's base; see git + introduction_commit: 0979ff5 + latest_change_commit: 998b4c7 # mechanical SonarCloud sweep (#191), not a wiring change + declared_owner: engine-team + declared_status: active + discovered_via: reverse-import closure of the mandatory set + imports: [] + imported_by: + production: [] + island: + - engine/graph_return_channel.py + - engine/graph/graph_sync_client_fix.py + - engine/convergence_controller_patch.py + tests: [tests/gap_fixes/test_gap1_contract.py, tests/gap_fixes/test_gap2_return_channel.py] + public_exports: [] # engine/packet/__init__.py exports PacketEnvelope, deflate_egress, inflate_ingress — not these + tests: [tests/gap_fixes/test_gap1_contract.py] + historical_plan_refs: [wire_gap-fix_modules_7d4d9028.plan.md — planned move to engine/packet/contract_enforcement.py, never landed] + current_architecture_refs: [] + name_collision_checked: + doc: docs/L9_Contract_Enforcement_System.md + verdict: UNRELATED + reason: > + That document specifies the 24-contract STATIC enforcement system + (tools/contract_scanner.py, tools/verify_contracts.py, pre-commit, CI + gates). It never references engine/contract_enforcement.py, which is a + runtime packet-envelope validator. Name similarity only. + + - path: engine/startup_wiring.py + exists: true + current_blob_sha: null + introduction_commit: 0979ff5 + latest_change_commit: 5815927 # PR #232 removed the dead spec.kb recipe block + declared_owner: engine-team + declared_status: active + discovered_via: sole module referencing the island as "wiring" + imports: + - shared.audit_persistence # package `shared` DOES NOT EXIST + - engine.graph_return_channel + - graph.community_export # top-level package `graph` DOES NOT EXIST + - graph.gds_scheduler # top-level package `graph` DOES NOT EXIST + imported_by: + production: [] + tests: [tests/gap_fixes/test_gap9_inference_authority.py] # reads source text, does not import + public_exports: [] + tests: [tests/gap_fixes/test_gap9_inference_authority.py] + historical_plan_refs: [wire_gap-fix_modules_7d4d9028.plan.md — listed for deletion, never deleted] + current_architecture_refs: [] + +associated_tests: + - path: tests/gap_fixes/test_gap1_contract.py + covers: engine/contract_enforcement.py + covers_anything_else: false + - path: tests/gap_fixes/test_gap2_return_channel.py + covers: [engine/graph_return_channel.py, engine/contract_enforcement.py] + covers_anything_else: false + - path: tests/gap_fixes/test_gap5_audit.py + covers: engine/compliance/audit_persistence.py + covers_anything_else: false + - path: tests/gap_fixes/test_gap3_inference_registry.py + covers: engine/inference_rule_registry.py + in_scope: false # exercises the retained test-only registry; + # inference ownership is out of this contract's scope + - path: tests/gap_fixes/test_gap9_inference_authority.py + covers: [engine/inference_rule_registry.py, engine/startup_wiring.py (source-text assertion)] + in_scope: partial # must be updated if startup_wiring.py is removed + - path: tests/contracts/test_known_gaps.py + verdict: UNRELATED + reason: xfail placeholders for absent contract YAML files; no island reference. diff --git a/docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_REACHABILITY_CLASSIFICATION.yaml b/docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_REACHABILITY_CLASSIFICATION.yaml new file mode 100644 index 00000000..088cf7d7 --- /dev/null +++ b/docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_REACHABILITY_CLASSIFICATION.yaml @@ -0,0 +1,333 @@ +# Phase 6 — classification report (required before any source edit) +# Contract: CEG-GAP-FIX-ARTIFACT-REACHABILITY-CONVERGENCE-2026-08-23 +# Base SHA: 5868bc49865eba0afd6154a0746ac06111cf1ccf +# +# Narrative form of these proofs is in the PR description; this file is the +# machine-readable record. Every artifact below is CONFIRMED with zero +# production callers. + +runtime_entrypoints_searched: + - chassis/handler_registration.py::register_engine_handlers # -> engine.handlers.ACTION_HANDLERS + - chassis/actions.py::execute_action + - chassis/chassis_app.py # L9_LIFECYCLE_HOOK -> importlib + - engine/boot.py::GraphLifecycle.{startup,shutdown,execute,_compliance_flush_loop} + - engine/handlers.py # 8 action handlers + - engine/gds/scheduler.py::GDSScheduler + - engine/feedback/convergence.py::ConvergenceLoop.on_outcome_recorded + - engine/__init__.py # packaged public API + +method: + static: git grep over engine chassis tests contracts domains docs tools Makefile .github + ast: > + Import-graph BFS over all 139 engine modules from the entrypoints above, + modelling absolute / from / relative / function-level imports, + importlib.import_module string literals, and Python ancestor-package + execution semantics. Shipped as tests/invariants/test_module_reachability.py. + cross_repo: GitHub code search over org:Quantum-L9 for every module path. + +artifacts: + + - path: engine/compliance/audit_persistence.py + classification: ORPHANED_IMPLEMENTATION + confidence: CONFIRMED + canonical_owner: engine/compliance/audit.py::AuditLogger.flush_to_store + runtime_callers: [] + reverse_imports: [tests/gap_fixes/test_gap5_audit.py] + proof: > + Own module-global _POOL, set only by configure_audit_pool(), which nothing + calls; INSERTs 5 untyped columns into `audit_log`. Canonical path is + boot.py -> init_dependencies(db_pool) -> EngineState -> + ComplianceEngine.flush_audit -> AuditLogger.flush_to_store, writing 16 + typed columns to `packet_audit_log`. Two competing schemas, one owner. + behavior_preservation_requirement: none — AuditLogger unchanged + proposed_action: DELETE + validation_required: [tests/test_compliance_engine.py, tests/invariants/test_compliance.py] + + - path: engine/graph_return_channel.py + classification: ORPHANED_IMPLEMENTATION + confidence: CONFIRMED + canonical_owner: UNKNOWN — the responsibility does not exist in CEG + runtime_callers: [] + reverse_imports: + - engine/graph/community_export.py # itself orphaned + - engine/convergence_controller_patch.py # itself orphaned + - engine/startup_wiring.py # recipe, no caller, broken imports + - tests/gap_fixes/test_gap2_return_channel.py + proof: > + Its docstring names its consumer as convergence_controller.run_convergence_loop(). + Neither exists in CEG. The canonical feedback owner, ConvergenceLoop, exposes + only on_outcome_recorded() and never drains a queue. Every producer is an + orphan, so nothing can enqueue; the consumer is absent, so nothing can drain. + behavior_preservation_requirement: none — a queue with no producer and no consumer + proposed_action: DELETE + validation_required: [tests/unit (feedback), tests/test_boot_and_registry.py] + + - path: engine/graph/community_export.py + classification: ORPHANED_IMPLEMENTATION + confidence: CONFIRMED + canonical_owner: engine/gds/scheduler.py::GDSScheduler._run_louvain + runtime_callers: [] + reverse_imports: [engine/startup_wiring.py] # via non-existent `graph.` package + proof: > + Instructs attachment to a "GDSScheduler post-job completion hook". + GDSScheduler has no hook mechanism at all — unwireable, not merely unwired. + Redundant regardless: _run_louvain writes the label into the graph via + gds.louvain.write (writeproperty: community_id in domains/plasticos/spec.yaml) + and engine/scoring/assembler.py reads community_id straight off the node. + The graph is the transport. + behavior_preservation_requirement: none — GDS write and scoring read unchanged + proposed_action: DELETE + validation_required: [tests/unit/test_gds_scheduler.py, tests/unit/test_hgkr_gds_dag.py] + + - path: engine/convergence_controller_patch.py + classification: STALE_PATCH + confidence: CONFIRMED + canonical_owner: engine/feedback/convergence.py::ConvergenceLoop + runtime_callers: [] + reverse_imports: [] + proof: > + Instructs the operator to call patch_convergence_controller(), a function + defined neither here nor anywhere else, to patch a convergence_controller.py + that does not exist. Its four symbols have zero importers, not even a test. + emit_schema_proposal dynamically imports chassis.events, which does not exist. + behavior_comparison_with_canonical: > + ConvergenceLoop.on_outcome_recorded is outcome-feedback score propagation: no + passes, no feature-vector confidence maps, no schema discovery, no domain_spec + argument. The patch supplies behavior for a different, absent subsystem — not + behavior missing from the canonical owner. + behavior_preservation_requirement: none — no canonical caller loses a capability + proposed_action: DELETE + validation_required: [tests/unit (feedback/convergence), tests/test_pareto_wiring.py] + + - path: engine/graph/graph_sync_client_fix.py + classification: STALE_PATCH + confidence: CONFIRMED + canonical_owner: engine/sync/generator.py::SyncGenerator # via handlers.py::handle_sync + runtime_callers: [] + reverse_imports: [] + architecture_refs: [docs/adr/ADR-DEC-001-candidate-identity.md] # stale citation, corrected + proof: > + A "drop this over GraphSyncClient in graph/sync/client.py" replacement for a + package and call site that do not exist. Write shapes are incompatible: + canonical: MERGE (n: {: ...}) + SET n += row, n._tenant = $tenant + orphan : MERGE (n {entity_id: row.entity_id, tenant: $tenant}) SET n:Entity + A labelless MERGE on a hardcoded entity_id writing `tenant`, not `_tenant`. + Had it run it would have built a parallel, unqueryable node keyspace. + adr_citation_correction_required: true + adr_finding: > + ADR-DEC-001 (Accepted) cites this file:113 as where the ungoverned `entity_id` + is "client-supplied at sync time". The code never had a caller, and no domain + spec declares idproperty: entity_id (plasticos uses facility_id, code, form_id, + opportunity_id, demand_id), so no canonical writer of entity_id exists at all, + while handlers.py:509,616,1497 still read it via silent fallbacks. Decision and + residual-risk finding stand; only the evidence pointer is corrected. The + correction widens the recorded divergence rather than narrowing it. + behavior_preservation_requirement: none — SyncGenerator owns sync end to end + proposed_action: DELETE + validation_required: [tests/unit (sync), tests/test_handlers.py] + + - path: engine/contract_enforcement.py + classification: TEST_ONLY_IMPLEMENTATION + confidence: CONFIRMED + discovered_via: reverse-import closure — not in the mandatory set + canonical_owner: engine/packet/packet_envelope.py::PacketEnvelope + runtime_callers: [] + reverse_imports: + - engine/graph_return_channel.py + - engine/graph/graph_sync_client_fix.py + - engine/convergence_controller_patch.py + - tests/gap_fixes/test_gap1_contract.py + - tests/gap_fixes/test_gap2_return_channel.py + proof: > + A second packet-envelope architecture with no shared type. Canonical + engine/packet/ (PacketType StrEnum, frozen extra=forbid models, _compute_hash) + is imported by chassis/actions.py, chassis/handler_registration.py, + engine/handlers.py, engine/config/settings.py, tools/contract_scanner.py and + tools/packet_envelope_gate.py. This module carries a private frozenset of + packet-type STRINGS (graph_inference_result, schema_proposal, community_export, + ...), a parallel required-fields table, and its own _compute_content_hash. Not + one of its packet-type strings appears in the canonical enum. Once the three + island importers go, every remaining importer is a test. Contract 05 + (Redefine PacketEnvelope) exactly. + name_collision_checked: + doc: docs/L9_Contract_Enforcement_System.md + verdict: UNRELATED + reason: > + That document specifies the STATIC 24-contract system (tools/contract_scanner.py, + tools/verify_contracts.py, pre-commit, CI gates) and never references this + runtime module. Name similarity only. + behavior_preservation_requirement: none — no production path validates through it + proposed_action: DELETE + validation_required: + - tests/contracts/test_packet_envelope.py + - tests/unit/test_packet_envelope.py + - tests/unit/test_chassis_contract.py + + - path: engine/startup_wiring.py + classification: STALE_PATCH + confidence: CONFIRMED + discovered_via: sole module presenting the island as installable wiring + canonical_owner: engine/boot.py::GraphLifecycle + runtime_callers: [] + reverse_imports: [tests/gap_fixes/test_gap9_inference_authority.py] # reads source text + unrunnable_proof: > + apply_all_gap_fixes() cannot execute once. Its first statement is + `from shared.audit_persistence import configure_audit_pool`; no `shared` package + exists, so it raises ModuleNotFoundError before applying any fix. Two later + imports name an absent top-level `graph` package — one of them OUTSIDE the try + block guarding ImportError — and it then calls GDSScheduler.register_post_job_hook, + which does not exist (AttributeError, uncaught). + activation_instruction_hazard: > + The Phase 13 surface. Its docstring reads "Add these calls to your application + lifespan / startup handler in order" — a standing instruction to activate the + six modules above. Deleting them and keeping this file would leave exactly the + resurrection vector this contract exists to close. + canonical_owner_proof: > + GraphLifecycle is the sole startup/shutdown owner (E-001) and already creates + the optional Postgres pool from settings.postgres_dsn, handing it to + init_dependencies(db_pool=...) (E-002). boot.py contains no call to + apply_all_gap_fixes and never has. + behavior_preservation_requirement: > + None from this file. PR #232's guard (no undeclared spec.kb / + load_domain_rules recipe may reappear) is preserved as a stronger tree-wide + invariant rather than a source-text assertion against one file. + proposed_action: DELETE + validation_required: + - tests/gap_fixes/test_gap9_inference_authority.py + - tests/test_boot_and_registry.py + + # ── Explicitly retained ── + + - path: engine/inference_rule_registry.py + classification: TEST_ONLY_IMPLEMENTATION + confidence: CONFIRMED + evidence: + runtime_callers: [] + production_reachability: NONE_VERIFIED + test_consumers: + - tests/gap_fixes/test_gap3_inference_registry.py + - tests/gap_fixes/test_gap9_inference_authority.py + proposed_action: KEEP + reason: > + KEEP is a scope decision, not a canonicality claim. PR #232 — + "fix(inference): remove ghost bridge and unowned KB rule loading", + merge commit 5868bc4 — deliberately retained this module (removing only + load_domain_rules and _register_condition_rule) and explicitly recorded + that it has no verified production edge and is reachable only from tests. + Resolving inference ownership — wiring it, relocating it, or removing it — + is outside this contract's scope and deliberately deferred. The later + inference-ownership closure contract is a separate, unexecuted program; + it did not merge as PR #232. + + - path: tests/contracts/test_known_gaps.py + classification: HISTORICAL_REFERENCE + confidence: CONFIRMED + proposed_action: KEEP + reason: > + Matched the Phase 1 filename filter on "gap" only. xfail placeholders for absent + contract YAML files under contracts/. No island reference. + +cross_repo_consumer_analysis: + scope: org:Quantum-L9, GitHub code search, all five module paths + contract_enforcement + result_total_count: 1 + sole_hit: + repository: Quantum-L9/Cursor-Governance + path: docs/plans/BUILT/wire_gap-fix_modules_7d4d9028.plan.md + kind: PLAN_DOCUMENT + is_consumer: false + conclusion: > + No verified cross-repo consumer for any island module. No artifact qualifies as a + COMPATIBILITY_BOUNDARY. + +historical_plan_analysis: + document: Cursor-Governance docs/plans/BUILT/wire_gap-fix_modules_7d4d9028.plan.md + declared_state: every todo "completed"; filed under docs/plans/BUILT/ + verified_state: NOT ONE OUTPUT LANDED IN CEG + outputs_verified_absent: + - engine/packet/contract_enforcement.py + - engine/feedback/graph_return_channel.py + - engine/feedback/inference_rule_registry.py + - engine/feedback/enrich_helpers.py + - engine/boot_gap_wiring.py + - tests/unit/test_contract_enforcement.py + - tests/unit/test_graph_return_channel.py + - tests/unit/test_inference_rule_registry.py + - tests/unit/test_audit_persistence.py + deletions_never_performed: [the 6 engine modules above, tests/gap_fixes/] + wiring_never_performed: + - "boot.py: import+call apply_all_gap_fixes() after init_dependencies" # 0 occurrences + - "__init__.py export additions for feedback, packet, compliance, graph" # all 4 unchanged + conclusion: > + E-006 confirmed in the strongest form. Per the contract's authority order and the + prohibited shortcut `wire_because_historical_plan_said_completed`, this plan grants + no authority to wire anything. Its own intent — delete the non-canonical originals — + agrees with this audit; the "relocate first" half was never justified by a consumer. + +deferred_findings: + - id: DEF-001 + title: 59 further engine modules are unreachable from production entrypoints + severity: informational + evidence: > + The analyzer reports 66 unreachable engine modules at base; 7 are this island. + The other 59 are unrelated dormant clusters: engine/health/**, engine/intake/**, + engine/personas/**, engine/hoprag/**, engine/kge/** (documented dormant, + kge_enabled=False), engine/arbitration/**, engine/outcomes/**, engine/replay/**, + engine/shadow/**. The gates cluster inside this set is broken + out as GATE-001 below. + why_not_fixed_here: > + Out of scope. non_scope forbids a general CEG refactor, and classifying 59 modules + across nine unaudited subsystems is the `broaden_into_unrelated_refactor` + prohibition. Some are plausibly legitimate staged subsystems, some plausibly + dormant defects; neither can be asserted without this island's evidence depth. + + - id: GATE-001 + title: GateRegistry/all_gates is a production-unreachable alternate gate implementation surface + severity: architectural + state: DEFER_DEDICATED_CONVERGENCE_AUDIT + confidence: HIGH + facts: + - engine/gates/__init__.py exports only GateCompiler and NullHandler + - GateCompiler contains direct per-GateType compile handlers and is production-reachable via engine/handlers.py + - GateCompiler._compile_composite recursively invokes the live compiler path; it does not use GateRegistry + - GateRegistry._REGISTRY is a static dictionary mapping GateType values to classes imported from engine/gates/types/all_gates.py; no decorator registration exists anywhere in engine/gates/ + - CompositeGate in all_gates.py recursively resolves subgates through GateRegistry.get_gate_class + - GateRegistry and all_gates.py are unreachable from all verified production roots + - tests directly import and exercise GateRegistry/all_gates (tests/test_boot_and_registry.py::TestGateRegistry asserts full GateType coverage of _REGISTRY) + - the current gate-development agent skill (.claude/skills/gate-development/SKILL.md) instructs agents to add each new gate in BOTH all_gates.py and GateRegistry._REGISTRY + interpretation: > + Not presently evidenced as "GateCompiler forgot to wire GateRegistry." + The stronger evidence is that CEG contains two gate compilation + implementation surfaces: the production-reachable GateCompiler, and an + unreachable GateRegistry + all_gates surface kept alive by tests and by + active agent guidance. Whether the alternate surface contains semantics + that must be preserved before consolidation is UNKNOWN until a dedicated + gate-by-gate parity audit. + prohibited_conclusions: + - GateRegistry must be wired into GateCompiler + - GateRegistry/all_gates should immediately be deleted + - all_gates is definitely obsolete + - GateCompiler is definitely incomplete + - both implementations are required merely because agent docs say so + action_in_this_pr: > + Record only. engine/gates/** is not modified here. + +reachability_invariant_decision: + contract_preferred_design: full-tree static AST reachability gate over engine/ + implemented: NO — deferred with evidence + rationale: > + The contract's own constraints for this invariant prohibit a + `large_permanent_unreachable_baseline` and `adding_new_modules_to_baseline_to_make_test_green`. + A full-tree gate at this base has 59 out-of-scope unreachable modules; going green + needs either a 59-entry exemption list (the prohibited baseline) or a 59-module + classification sweep (the prohibited refactor). Both routes are closed, so shipping + the full-tree gate here would violate the contract that asks for it. + implemented_instead: > + tests/invariants/test_module_reachability.py — removed paths stay absent, no engine + or chassis module imports the removed surfaces or symbols, no gap-fix activation + recipe reappears, and the canonical owners remain present and production-reachable. + Ships and exercises the reusable AST analyzer the full gate needs, so DEF-001 can be + acted on by tightening scope rather than writing machinery. + UNKNOWN_declared: > + Whether each of the 59 remaining unreachable modules is a staged subsystem or a + dormant defect is UNKNOWN at this evidence depth and is not guessed. diff --git a/docs/audits/2026-08-23-gap-fix-artifact-convergence/IMPLEMENTATION_FILETREE.yaml b/docs/audits/2026-08-23-gap-fix-artifact-convergence/IMPLEMENTATION_FILETREE.yaml new file mode 100644 index 00000000..4ea897db --- /dev/null +++ b/docs/audits/2026-08-23-gap-fix-artifact-convergence/IMPLEMENTATION_FILETREE.yaml @@ -0,0 +1,103 @@ +# Phase 7 — exact implementation filetree (hard gate) +# Contract: CEG-GAP-FIX-ARTIFACT-REACHABILITY-CONVERGENCE-2026-08-23 +# No implementation edit may touch a file absent from this list. +# Classifications and proofs: GAP_FIX_REACHABILITY_CLASSIFICATION.yaml + +files: + + # ── DELETE: proven orphaned / stale / test-only ── + - {path: engine/compliance/audit_persistence.py, action: DELETE, + serves_finding: ORPHANED_IMPLEMENTATION — duplicate audit persistence, + canonical_owner_retained: engine/compliance/audit.py::AuditLogger.flush_to_store} + - {path: engine/graph_return_channel.py, action: DELETE, + serves_finding: ORPHANED_IMPLEMENTATION — queue with no producer and no consumer, + canonical_owner_retained: none required (responsibility absent from CEG)} + - {path: engine/graph/community_export.py, action: DELETE, + serves_finding: ORPHANED_IMPLEMENTATION — unwireable GDS hook, duplicate write path, + canonical_owner_retained: engine/gds/scheduler.py::GDSScheduler._run_louvain} + - {path: engine/convergence_controller_patch.py, action: DELETE, + serves_finding: STALE_PATCH — patches a module that does not exist, + canonical_owner_retained: engine/feedback/convergence.py::ConvergenceLoop} + - {path: engine/graph/graph_sync_client_fix.py, action: DELETE, + serves_finding: STALE_PATCH — replacement for an absent client; incompatible write shape, + canonical_owner_retained: engine/sync/generator.py::SyncGenerator} + - {path: engine/contract_enforcement.py, action: DELETE, + serves_finding: TEST_ONLY_IMPLEMENTATION — parallel PacketEnvelope architecture, + canonical_owner_retained: engine/packet/packet_envelope.py::PacketEnvelope} + - {path: engine/startup_wiring.py, action: DELETE, + serves_finding: STALE_PATCH — unrunnable recipe; Phase 13 activation-instruction hazard, + canonical_owner_retained: engine/boot.py::GraphLifecycle} + - {path: tests/gap_fixes/test_gap1_contract.py, action: DELETE, + serves_finding: sole purpose is preserving engine/contract_enforcement.py, + replacement: tests/invariants/test_module_reachability.py} + - {path: tests/gap_fixes/test_gap2_return_channel.py, action: DELETE, + serves_finding: sole purpose is preserving engine/graph_return_channel.py, + replacement: tests/invariants/test_module_reachability.py} + - path: tests/gap_fixes/test_gap5_audit.py + action: DELETE + serves_finding: sole purpose is preserving engine/compliance/audit_persistence.py + replacement: > + tests/invariants/test_module_reachability.py asserts absence; the audit + persistence behaviour still required stays covered by the canonical owner's + existing tests (AuditLogger.flush_to_store / ComplianceEngine). + + # ── MODIFY: required by the deletions above ── + - path: tests/gap_fixes/test_gap9_inference_authority.py + action: MODIFY + serves_finding: > + Its startup-recipe guard reads engine/startup_wiring.py source text; that file is + deleted here. The guarded behaviour (no undeclared spec.kb / load_domain_rules + recipe may reappear) is still required, so the assertion is strengthened from + one-file source text to a tree-wide scan. PR #232's regression coverage preserved, not dropped. + - path: docs/adr/ADR-DEC-001-candidate-identity.md + action: MODIFY + serves_finding: > + Accepted ADR cites engine/graph/graph_sync_client_fix.py:113 as the writer of the + ungoverned entity_id property. The citation is factually wrong — that code has no + caller — and the file is deleted here. Evidence pointer only; the decision, options + and residual-reconciliation finding are unchanged. + - path: .l9/baselines/packet-envelope.yml + action: MODIFY + serves_finding: > + Deleting engine/graph/graph_sync_client_fix.py resolves the PacketEnvelope debt + entry recorded against it. The baseline ratchet requires the resolved entry to be + removed in the same PR so the baseline ratchets DOWN; leaving it fails the gate + with "the debt is resolved — remove the entry". + + # ── NEW ── + - path: tests/invariants/test_module_reachability.py + action: NEW + serves_finding: > + Phase 9. Makes island resurrection mechanically detectable: removed paths stay + absent, no engine/chassis module imports the removed surfaces, no gap-fix + activation recipe reappears, and the five canonical owners remain sole and + production-reachable. Ships and exercises the reusable AST analyzer. + scope_note: > + Deliberately NOT a full-tree reachability gate — see + GAP_FIX_REACHABILITY_CLASSIFICATION.yaml::reachability_invariant_decision. + - {path: "docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_ARTIFACT_INVENTORY.yaml", + action: NEW, serves_finding: Phase 1 required output} + - {path: "docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_REACHABILITY_CLASSIFICATION.yaml", + action: NEW, serves_finding: Phase 6 required output} + - {path: "docs/audits/2026-08-23-gap-fix-artifact-convergence/IMPLEMENTATION_FILETREE.yaml", + action: NEW, serves_finding: Phase 7 required output} + +explicitly_not_touched: + - path: engine/inference_rule_registry.py + reason: > + TEST_ONLY_IMPLEMENTATION, kept as an explicit scope decision — inference + ownership is outside this contract and deferred. PR #232 retained it and + recorded its lack of any verified production edge; KEEP here is not a + canonicality claim. + - {path: tests/gap_fixes/test_gap3_inference_registry.py, reason: exercises the retained test-only registry; unaffected} + - path: engine/boot.py + reason: > + PROTECTED-adjacent lifecycle owner, and already correct — it never called the + removed recipe. Rule: do not create new boot wiring. + - path: engine/gates/** + reason: > + GATE-001 (see classification) — an unreachable alternate gate implementation + surface recorded for a dedicated parity audit; no gate code changes in this PR. + - {path: "artifacts/**", reason: Phase 12 — generated audit output; churn restored, not committed} + - {path: "engine/packet/**, engine/sync/**, engine/compliance/audit.py, engine/gds/scheduler.py", + reason: canonical owners, retained unchanged — consolidation requires no edit to them} diff --git a/engine/compliance/audit_persistence.py b/engine/compliance/audit_persistence.py deleted file mode 100644 index 91e5dae2..00000000 --- a/engine/compliance/audit_persistence.py +++ /dev/null @@ -1,86 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [compliance] -tags: [audit, persistence] -owner: engine-team -status: active ---- /L9_META --- - -GAP-5 FIX: Wire db_pool into ComplianceEngine so flush_audit() persists -to PostgreSQL instead of warning db_pool=None. - -Call configure_audit_pool(pool) at app startup after asyncpg.create_pool(). -""" - -from __future__ import annotations - -import logging -import time -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: - import asyncpg - -logger = logging.getLogger(__name__) - -_POOL: asyncpg.Pool | None = None - -_CREATE_TABLE_SQL = """ -CREATE TABLE IF NOT EXISTS audit_log ( - id BIGSERIAL PRIMARY KEY, - tenant_id TEXT NOT NULL, - actor TEXT NOT NULL, - action TEXT NOT NULL, - detail TEXT, - created_at DOUBLE PRECISION NOT NULL -); -CREATE INDEX IF NOT EXISTS idx_audit_tenant_created - ON audit_log (tenant_id, created_at DESC); -""" - - -async def configure_audit_pool(pool: asyncpg.Pool) -> None: - """Call once at startup after asyncpg.create_pool().""" - global _POOL - _POOL = pool - async with pool.acquire() as conn: - await conn.execute(_CREATE_TABLE_SQL) - logger.info("audit_persistence: PostgreSQL pool configured and schema verified") - - -async def flush_audit_entries(entries: list[dict[str, Any]]) -> int: - """ - Persist audit entries to PostgreSQL. - Replaces the warning-only stub in ComplianceEngine.flush_audit(). - Returns count of rows inserted. - """ - if _POOL is None: - logger.error( - "flush_audit_entries called but db_pool is None — " - "call configure_audit_pool() at startup. Entries dropped: %d", - len(entries), - ) - return 0 - if not entries: - return 0 - - rows = [ - ( - e.get("tenant_id", "unknown"), - e.get("actor", "system"), - e.get("action", "unknown"), - e.get("detail"), - e.get("created_at", time.time()), - ) - for e in entries - ] - async with _POOL.acquire() as conn: - await conn.executemany( - "INSERT INTO audit_log (tenant_id, actor, action, detail, created_at) VALUES ($1, $2, $3, $4, $5)", - rows, - ) - logger.debug("audit_persistence: flushed %d entries to PostgreSQL", len(rows)) - return len(rows) diff --git a/engine/contract_enforcement.py b/engine/contract_enforcement.py deleted file mode 100644 index c2f507bc..00000000 --- a/engine/contract_enforcement.py +++ /dev/null @@ -1,306 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [config] -tags: [contracts, packets] -owner: engine-team -status: active ---- /L9_META --- - -GAP-1 FIX: Strict PacketEnvelope contract enforcement. - -Replaces all silent bypass paths with hard ContractViolationError failures. -Every inter-service data flow that previously sent a hand-built dict now -must pass through enforce_packet_envelope() before processing. - -Usage: - from engine.contract_enforcement import enforce_packet_envelope, ContractViolationError - - # In GraphSyncClient (was: sending bare dict) - envelope = enforce_packet_envelope(raw_payload, expected_type="graph_sync") - - # In any handler boundary: - enforce_packet_envelope(incoming, expected_type="enrich_request") -""" - -from __future__ import annotations - -import hashlib -import json -import logging -from typing import Any - -logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Error hierarchy -# --------------------------------------------------------------------------- - - -class ContractViolationError(RuntimeError): - """Raised whenever an inter-service packet fails contract validation. - - This is an EXPLICIT HARD FAILURE — never caught and silently swallowed. - Callers must fix the packet, not catch this error. - """ - - def __init__(self, reason: str, *, packet_id: str | None = None) -> None: - self.reason = reason - self.packet_id = packet_id - msg = f"ContractViolation[{packet_id or 'unknown'}]: {reason}" - super().__init__(msg) - logger.error(msg) - - -# --------------------------------------------------------------------------- -# Allowed packet types — Gap-10 fix: schema_proposal added -# --------------------------------------------------------------------------- - -_ALLOWED_PACKET_TYPES: frozenset[str] = frozenset( - [ - "enrich_request", - "enrich_result", - "inference_result", - "graph_sync", - "graph_inference_result", # Gap-2: new type for return channel - "schema_proposal", # Gap-10: was missing, caused ValidationError - "community_export", # Gap-6: community label export to ENRICH - "health_check", - "admin_command", - ] -) - - -# --------------------------------------------------------------------------- -# Required fields per packet type -# --------------------------------------------------------------------------- - -_REQUIRED_FIELDS: dict[str, list[str]] = { - "enrich_request": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "entity_id"], - "enrich_result": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "entity_id", "enriched_fields"], - "inference_result": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "inference_outputs"], - "graph_sync": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "entity_type", "batch"], - "graph_inference_result": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "inference_outputs"], - "schema_proposal": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "proposed_fields"], - "community_export": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "communities"], - "health_check": ["packet_id", "tenant_id"], - "admin_command": ["packet_id", "tenant_id", "content_hash", "envelope_hash", "subaction"], -} - - -# --------------------------------------------------------------------------- -# Core enforcement function -# --------------------------------------------------------------------------- - - -def enforce_packet_envelope( - packet: Any, - *, - expected_type: str, -) -> dict[str, Any]: - """ - Validate that `packet` is a well-formed PacketEnvelope of `expected_type`. - - Checks: - 1. packet is a dict (or Pydantic model with .model_dump()) - 2. packet_type matches expected_type - 3. packet_type is in _ALLOWED_PACKET_TYPES - 4. All required fields for the type are present - 5. content_hash matches SHA-256 of the canonical content payload - 6. envelope_hash is present and non-empty - - Returns the validated dict. - Raises ContractViolationError on ANY failure — no silent returns. - """ - # Normalise to dict - if hasattr(packet, "model_dump"): - packet = packet.model_dump(mode="python") - if not isinstance(packet, dict): - raise ContractViolationError( - f"Expected a dict or Pydantic model, got {type(packet).__name__}", - ) - - packet_id = packet.get("packet_id", "") - - # Type check - actual_type = packet.get("packet_type") or packet.get("type") - if actual_type != expected_type: - raise ContractViolationError( - f"packet_type mismatch: expected={expected_type!r} got={actual_type!r}", - packet_id=packet_id, - ) - - if expected_type not in _ALLOWED_PACKET_TYPES: - raise ContractViolationError( - f"packet_type={expected_type!r} is not in the allowed set", - packet_id=packet_id, - ) - - # Required fields - required = _REQUIRED_FIELDS.get(expected_type, []) - for field_name in required: - if field_name not in packet or packet[field_name] is None: - raise ContractViolationError( - f"Missing or null required field '{field_name}' for type={expected_type!r}", - packet_id=packet_id, - ) - - # Content hash verification (when present) - if "content_hash" in required: - _verify_content_hash(packet, expected_type, packet_id) - - # Envelope hash must be non-empty - if "envelope_hash" in required and not packet.get("envelope_hash"): - raise ContractViolationError( - "envelope_hash is empty", - packet_id=packet_id, - ) - - return packet - - -# Fields excluded from the canonical content payload when computing -# content_hash. This single definition is shared by the builders and the -# verifier so canonicalization can never drift between the two. -_HASH_EXCLUDED_FIELDS: frozenset[str] = frozenset( - [ - "content_hash", - "envelope_hash", - "packet_id", - "packet_type", - "type", - "created_at", - "lineage", - "tenant_context", - ] -) - - -def _compute_content_hash(packet_fields: dict[str, Any]) -> str: - """Compute SHA-256 over the canonical content payload of a packet. - - The canonical payload is every field except those in - _HASH_EXCLUDED_FIELDS, serialized as sorted-key JSON. - """ - content_payload = {k: v for k, v in packet_fields.items() if k not in _HASH_EXCLUDED_FIELDS} - payload_bytes = json.dumps(content_payload, sort_keys=True, default=str).encode() - return hashlib.sha256(payload_bytes).hexdigest() - - -def _verify_content_hash( - packet: dict[str, Any], - packet_type: str, - packet_id: str, -) -> None: - """Recompute SHA-256 over the canonical content payload and compare.""" - try: - expected_hash = _compute_content_hash(packet) - except (TypeError, ValueError) as exc: - raise ContractViolationError( - f"Cannot serialize content payload for hash verification: {exc}", - packet_id=packet_id, - ) from exc - - actual_hash = packet.get("content_hash", "") - if expected_hash != actual_hash: - raise ContractViolationError( - f"content_hash mismatch: expected={expected_hash!r} got={actual_hash!r}", - packet_id=packet_id, - ) - - -# --------------------------------------------------------------------------- -# GraphSyncClient wrapper — Gap-1 targeted fix -# --------------------------------------------------------------------------- - - -def build_graph_sync_packet( - *, - tenant_id: str, - entity_type: str, - batch: list[dict[str, Any]], - tenant_context: dict[str, Any] | None = None, - lineage: dict[str, Any] | None = None, -) -> dict[str, Any]: - """ - Build a fully contract-compliant graph_sync PacketEnvelope. - Previously GraphSyncClient sent a bare dict with no hashes. - Use this factory everywhere instead. - """ - import time - import uuid - - # Hash over the same canonical field set the verifier uses - # (_HASH_EXCLUDED_FIELDS), so self-validation can never drift. - content_hash = _compute_content_hash( - { - "tenant_id": tenant_id, - "entity_type": entity_type, - "batch": batch, - } - ) - - packet_id = f"gs_{uuid.uuid4().hex}" - envelope_meta = {"packet_id": packet_id, "tenant_id": tenant_id, "content_hash": content_hash} - envelope_hash = hashlib.sha256(json.dumps(envelope_meta, sort_keys=True).encode()).hexdigest() - - packet = { - "packet_id": packet_id, - "packet_type": "graph_sync", - "tenant_id": tenant_id, - "entity_type": entity_type, - "batch": batch, - "content_hash": content_hash, - "envelope_hash": envelope_hash, - "created_at": time.time(), - } - if tenant_context: - packet["tenant_context"] = tenant_context - if lineage: - packet["lineage"] = lineage - - # Self-validate before returning — hard fail if our own factory is broken - enforce_packet_envelope(packet, expected_type="graph_sync") - return packet - - -def build_schema_proposal_packet( - *, - tenant_id: str, - proposed_fields: list[dict[str, Any]], - provenance: str = "schema_discovery", -) -> dict[str, Any]: - """ - Gap-4 + Gap-10 fix: Build a valid schema_proposal PacketEnvelope. - Previously SchemaProposal was computed but never emitted. - """ - import time - import uuid - - # Hash over the same canonical field set the verifier uses - # (_HASH_EXCLUDED_FIELDS), so self-validation can never drift. - content_hash = _compute_content_hash( - { - "tenant_id": tenant_id, - "proposed_fields": proposed_fields, - "provenance": provenance, - } - ) - packet_id = f"sp_{uuid.uuid4().hex}" - envelope_meta = {"packet_id": packet_id, "tenant_id": tenant_id, "content_hash": content_hash} - envelope_hash = hashlib.sha256(json.dumps(envelope_meta, sort_keys=True).encode()).hexdigest() - - packet = { - "packet_id": packet_id, - "packet_type": "schema_proposal", - "tenant_id": tenant_id, - "proposed_fields": proposed_fields, - "provenance": provenance, - "content_hash": content_hash, - "envelope_hash": envelope_hash, - "created_at": time.time(), - } - enforce_packet_envelope(packet, expected_type="schema_proposal") - return packet diff --git a/engine/convergence_controller_patch.py b/engine/convergence_controller_patch.py deleted file mode 100644 index 8d169113..00000000 --- a/engine/convergence_controller_patch.py +++ /dev/null @@ -1,205 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [scoring] -tags: [convergence, patch] -owner: engine-team -status: active ---- /L9_META --- - -GAP-2 + GAP-4 + GAP-7 + GAP-8 PATCH for convergence_controller.py - -This file is a DROP-IN PATCH: import and call `patch_convergence_controller()` -at application startup AFTER importing convergence_controller. - -Alternatively, merge the patched run_convergence_loop() directly into -convergence_controller.py per the inline diff comments below. - -Changes: - - Gap-2: Drain GraphToEnrichReturnChannel at the start of each new pass - - Gap-4: Emit SchemaProposal as a PacketEnvelope after schema discovery - - Gap-7: Robust per_field_confidence extraction with fallback - - Gap-8: domain_spec made mandatory; raises TypeError if omitted by caller -""" - -from __future__ import annotations - -import importlib -import logging -from typing import Any - -logger = logging.getLogger(__name__) - - -# --------------------------------------------------------------------------- -# Gap-7: Robust per_field_confidence extractor -# --------------------------------------------------------------------------- - - -def extract_per_field_confidence(feature_vector: dict[str, Any]) -> dict[str, float]: - """ - Extract per-field confidence scores from a feature vector. - Previously: if 'per_field_confidence' key was absent, all fields shared - one flat confidence, breaking targeted pass planning. - - Now: falls back gracefully through multiple resolution strategies. - """ - # Strategy 1: explicit per_field_confidence dict - pfc = feature_vector.get("per_field_confidence") - if isinstance(pfc, dict) and pfc: - return {str(k): float(v) for k, v in pfc.items()} - - # Strategy 2: field_scores nested dict - fs = feature_vector.get("field_scores") - if isinstance(fs, dict) and fs: - return {str(k): float(v) for k, v in fs.items()} - - # Strategy 3: flat confidence applied to all non-meta fields - flat = feature_vector.get("confidence") or feature_vector.get("overall_confidence") - if flat is not None: - try: - flat_val = float(flat) - except (TypeError, ValueError): - flat_val = 0.0 - meta_keys = { - "confidence", - "overall_confidence", - "pass_number", - "entity_id", - "tenant_id", - "per_field_confidence", - "field_scores", - } - return {k: flat_val for k in feature_vector if k not in meta_keys} - - # Strategy 4: no confidence info — return empty (caller treats all fields as uncertain) - logger.debug( - "extract_per_field_confidence: no confidence data found in feature_vector keys=%s", - list(feature_vector.keys()), - ) - return {} - - -# --------------------------------------------------------------------------- -# Gap-2 integration: inject return-channel targets into entity known_fields -# --------------------------------------------------------------------------- - - -async def apply_return_channel_targets( - entity: dict[str, Any], - tenant_id: str, - *, - timeout_seconds: float = 0.05, -) -> dict[str, Any]: - """ - Drain the GraphToEnrichReturnChannel for this tenant and inject any - matching targets as seed values into the entity's known_fields. - - Called at the start of each convergence pass (pass_number >= 2). - Returns the (possibly updated) entity dict. - """ - from engine.graph_return_channel import GraphToEnrichReturnChannel - - channel = GraphToEnrichReturnChannel.get_instance() - entity_id = entity.get("entity_id") or entity.get("id") - targets = await channel.drain(tenant_id=tenant_id, timeout_seconds=timeout_seconds, max_targets=200) - - matched = 0 - for target in targets: - if target.entity_id == str(entity_id): - # Inject as a seed value — only if the field is currently absent or low-confidence - existing = entity.get(target.field_name) - if existing is None: - entity[target.field_name] = target.seed_value - entity.setdefault("_return_channel_seeds", {})[target.field_name] = { - "value": target.seed_value, - "source_confidence": target.source_confidence, - "origin_rule": target.origin_inference_rule, - } - matched += 1 - - if matched: - logger.info( - "convergence_controller: injected %d return-channel seeds for entity=%s tenant=%s", - matched, - entity_id, - tenant_id, - ) - return entity - - -# --------------------------------------------------------------------------- -# Gap-4: SchemaProposal emission -# --------------------------------------------------------------------------- - - -async def emit_schema_proposal( - proposed_fields: list[dict[str, Any]], - tenant_id: str, -) -> dict[str, Any]: - """ - Emit a schema_proposal PacketEnvelope for newly discovered fields. - Previously SchemaProposal was computed but never emitted — schema - never evolved past the seed. - - This function should be called from convergence_controller whenever - schema_discovery produces new field proposals. - """ - from engine.contract_enforcement import build_schema_proposal_packet - - if not proposed_fields: - return {} - - packet = build_schema_proposal_packet( - tenant_id=tenant_id, - proposed_fields=proposed_fields, - provenance="convergence_loop_schema_discovery", - ) - # Emit to the schema evolution queue / event bus. - # In the current architecture this goes to the chassis event router. - # Contract 02 / T5-03: engine code must never import chassis statically - # (engine/handlers.py and engine/boot.py are the only bridges), so the - # optional event router is resolved dynamically at call time. - try: - events_module = importlib.import_module("chassis.events") - emit_event = events_module.emit_event - - await emit_event(packet_type="schema_proposal", payload=packet) - logger.info( - "Emitted schema_proposal packet for tenant=%s with %d new fields (packet_id=%s)", - tenant_id, - len(proposed_fields), - packet["packet_id"], - ) - except (ImportError, AttributeError): - # chassis.events not yet wired — log and continue rather than blocking - logger.warning( - "chassis.events not available — schema_proposal packet queued in-memory only: tenant=%s fields=%s", - tenant_id, - [f.get("name") for f in proposed_fields], - ) - return packet - - -# --------------------------------------------------------------------------- -# Gap-8: domain_spec enforcement wrapper -# --------------------------------------------------------------------------- - - -class DomainSpecRequiredError(TypeError): - """Raised when run_convergence_loop is called without domain_spec.""" - - -def enforce_domain_spec(domain_spec: Any, caller: str = "run_convergence_loop") -> None: - """ - Gap-8: domain_spec is MANDATORY. Callers that omit it get domain-blind - enrichment with no sonar optimization. Now raises instead of silently - degrading. - """ - if domain_spec is None: - raise DomainSpecRequiredError( - f"{caller}() requires domain_spec — omitting it disables domain KB injection " - f"and sonar optimization. Pass the DomainSpec for the tenant's domain." - ) diff --git a/engine/graph/community_export.py b/engine/graph/community_export.py deleted file mode 100644 index a56a1b47..00000000 --- a/engine/graph/community_export.py +++ /dev/null @@ -1,86 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [graph] -tags: [gds, community] -owner: engine-team -status: active ---- /L9_META --- - -GAP-6 FIX: Export Louvain community labels from Neo4j back to ENRICH -as known_fields context so convergence_controller uses them in Pass N+1. - -Attach to GDSScheduler post-job completion hook for "louvain" jobs. -""" - -from __future__ import annotations - -import logging -from typing import Any - -logger = logging.getLogger(__name__) - - -async def export_community_labels_to_enrich( - graph_driver, - tenant_id: str, - domain_id: str, -) -> dict[str, Any]: - """ - Query Neo4j for community labels from the last Louvain run, then submit - them to GraphToEnrichReturnChannel as enrichment targets. - """ - from engine.graph_return_channel import ( - GraphToEnrichReturnChannel, - build_graph_inference_result_envelope, - ) - - cypher = """ - MATCH (n) - WHERE n.tenant = $tenant AND n.community_id IS NOT NULL - RETURN n.entity_id AS entity_id, n.community_id AS community_id - LIMIT 10000 - """ - try: - records = await graph_driver.execute_query( - cypher=cypher, - parameters={"tenant": tenant_id}, - database=domain_id, - ) - except Exception: - logger.exception("community_export: failed to query community labels tenant=%s", tenant_id) - return {"status": "error", "exported": 0} - - if not records: - logger.debug("community_export: no community labels for tenant=%s", tenant_id) - return {"status": "ok", "exported": 0} - - inference_outputs = [ - { - "entity_id": r["entity_id"], - "field": "community_id", - "value": r["community_id"], - "confidence": 0.95, # Louvain is deterministic - "rule": "louvain_community_detection", - } - for r in records - if r.get("entity_id") and r.get("community_id") is not None - ] - - if not inference_outputs: - return {"status": "ok", "exported": 0} - - envelope = build_graph_inference_result_envelope( - tenant_id=tenant_id, - inference_outputs=inference_outputs, - ) - channel = GraphToEnrichReturnChannel.get_instance() - count = await channel.submit(envelope) - logger.info( - "community_export: submitted %d community label targets tenant=%s", - count, - tenant_id, - ) - return {"status": "ok", "exported": count, "packet_id": envelope.packet_id} diff --git a/engine/graph/graph_sync_client_fix.py b/engine/graph/graph_sync_client_fix.py deleted file mode 100644 index 2bc8f041..00000000 --- a/engine/graph/graph_sync_client_fix.py +++ /dev/null @@ -1,118 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [graph] -tags: [sync, client] -owner: engine-team -status: active ---- /L9_META --- - -GAP-1 FIX: Replace the hand-built dict in GraphSyncClient with a canonical -PacketEnvelope. Eliminates the silent bypass of content_hash, envelope_hash, -PacketLineage, and TenantContext. - -Usage: Drop this over GraphSyncClient in graph/sync/client.py and update -the import at the call site. -""" - -from __future__ import annotations - -import logging -from typing import Any - -from engine.contract_enforcement import ( - ContractViolationError, - build_graph_sync_packet, - enforce_packet_envelope, -) - -logger = logging.getLogger(__name__) - - -class GraphSyncClient: - """ - Production-hardened replacement for the original GraphSyncClient. - - All outbound payloads are canonical PacketEnvelopes with: - - content_hash (SHA-256 of sorted content JSON) - - envelope_hash (SHA-256 of packet_id + content_hash + timestamp) - - PacketLineage (origin_service, correlation_id, hop_count) - - TenantContext (tenant_id, tenant_tier) - - Any attempt to send a malformed or tampered envelope raises - ContractViolationError — hard fail, no silent degradation. - """ - - def __init__(self, neo4j_driver, tenant_id: str, tenant_tier: str = "unknown"): - self._driver = neo4j_driver - self._tenant_id = tenant_id - self._tenant_tier = tenant_tier - - def _build_envelope( - self, - entity_type: str, - batch: list[dict[str, Any]], - correlation_id: str | None = None, - ) -> dict[str, Any]: - """Build and validate a PacketEnvelope before sending.""" - packet = build_graph_sync_packet( - tenant_id=self._tenant_id, - entity_type=entity_type, - batch=batch, - tenant_context={"tenant_id": self._tenant_id, "tenant_tier": self._tenant_tier}, - lineage={"correlation_id": correlation_id} if correlation_id else None, - ) - # Enforce immediately — hard fail on violation - return enforce_packet_envelope(packet, expected_type="graph_sync") - - async def sync_entities( - self, - entity_type: str, - batch: list[dict[str, Any]], - correlation_id: str | None = None, - ) -> dict[str, Any]: - """ - Sync a batch of enriched entities to Neo4j via a validated envelope. - Raises ContractViolationError if the envelope fails validation. - """ - if not batch: - return {"status": "ok", "synced": 0} - - envelope = self._build_envelope(entity_type, batch, correlation_id) - - try: - await self._driver.execute_write( - _write_batch_tx, - entity_type=entity_type, - batch=envelope["content"]["batch"], - tenant_id=self._tenant_id, - packet_id=envelope["packet_id"], - ) - logger.info( - "GraphSyncClient: synced %d %s entities tenant=%s packet_id=%s", - len(batch), - entity_type, - self._tenant_id, - envelope["packet_id"], - ) - return {"status": "ok", "synced": len(batch), "packet_id": envelope["packet_id"]} - - except ContractViolationError: - raise - except Exception: - logger.exception("GraphSyncClient: write failed for packet_id=%s", envelope.get("packet_id")) - raise - - -async def _write_batch_tx(tx, *, entity_type: str, batch: list, tenant_id: str, packet_id: str): - """Neo4j write transaction — MERGE on entity_id, set all properties.""" - cypher = """ - UNWIND $batch AS row - MERGE (n {entity_id: row.entity_id, tenant: $tenant}) - SET n += row.properties - SET n.last_sync_packet = $packet_id - SET n:Entity - """ - await tx.run(cypher, batch=batch, tenant=tenant_id, packet_id=packet_id) diff --git a/engine/graph_return_channel.py b/engine/graph_return_channel.py deleted file mode 100644 index c4b51481..00000000 --- a/engine/graph_return_channel.py +++ /dev/null @@ -1,287 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [graph] -tags: [return-channel] -owner: engine-team -status: active ---- /L9_META --- - -GAP-2 FIX: GRAPH → ENRICH bidirectional return channel. - -Receives GRAPH inference outputs and converts them into deterministic -EnrichmentTarget records that are injected back into the convergence loop -as new Pass N+1 enrichment targets. - -Architecture: - GRAPH.ConvergenceLoop - └─► GraphInferenceResultEnvelope (PacketEnvelope type=graph_inference_result) - └─► GraphToEnrichReturnChannel.submit() - └─► EnrichmentTargetQueue (async) - └─► convergence_controller.run_convergence_loop() - └─► Pass N+1 re-enrichment -""" - -from __future__ import annotations - -import asyncio -import hashlib -import json -import logging -import time -import uuid -from dataclasses import dataclass, field -from typing import Any - -logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Domain model -# --------------------------------------------------------------------------- - - -@dataclass(frozen=True) -class EnrichmentTarget: - """A re-enrichment directive produced by a GRAPH inference output.""" - - entity_id: str - tenant_id: str - field_name: str - # Inference-supplied seed value (may be None — forces discovery) - seed_value: Any - # Confidence of the graph inference that produced this target - source_confidence: float - # Trace back to the originating PacketEnvelope - origin_packet_id: str - origin_inference_rule: str - created_at: float = field(default_factory=time.time) - - def to_dict(self) -> dict[str, Any]: - return { - "entity_id": self.entity_id, - "tenant_id": self.tenant_id, - "field_name": self.field_name, - "seed_value": self.seed_value, - "source_confidence": self.source_confidence, - "origin_packet_id": self.origin_packet_id, - "origin_inference_rule": self.origin_inference_rule, - "created_at": self.created_at, - } - - -@dataclass -class GraphInferenceResultEnvelope: - """ - Minimal representation of a PacketEnvelope[type=graph_inference_result]. - Validated before injection; raises ContractViolationError on any violation. - """ - - packet_id: str - tenant_id: str - # List of {entity_id, field, value, confidence, rule} dicts - inference_outputs: list[dict[str, Any]] - content_hash: str - envelope_hash: str - - # Minimum accepted confidence for a GRAPH output to generate a target - CONFIDENCE_FLOOR: float = 0.55 - - def validate(self) -> None: - """Hard-fail if the envelope does not meet contract requirements.""" - from engine.contract_enforcement import ContractViolationError # Gap-1 fix - - if not self.packet_id: - raise ContractViolationError("GraphInferenceResultEnvelope.packet_id is empty") - if not self.tenant_id: - raise ContractViolationError("GraphInferenceResultEnvelope.tenant_id is empty") - if not self.content_hash: - raise ContractViolationError("GraphInferenceResultEnvelope.content_hash is missing") - if not self.envelope_hash: - raise ContractViolationError("GraphInferenceResultEnvelope.envelope_hash is missing") - # Verify content hash - payload_bytes = json.dumps(self.inference_outputs, sort_keys=True).encode() - expected = hashlib.sha256(payload_bytes).hexdigest() - if expected != self.content_hash: - raise ContractViolationError( - f"content_hash mismatch: expected={expected!r} got={self.content_hash!r}", - ) - if not isinstance(self.inference_outputs, list): - raise ContractViolationError("inference_outputs must be a list") - - def to_targets(self) -> list[EnrichmentTarget]: - """Convert validated outputs to EnrichmentTarget records.""" - targets: list[EnrichmentTarget] = [] - for output in self.inference_outputs: - confidence = float(output.get("confidence", 0.0)) - if confidence < self.CONFIDENCE_FLOOR: - logger.debug( - "Skipping low-confidence inference output (%.3f < %.3f) for entity=%s field=%s", - confidence, - self.CONFIDENCE_FLOOR, - output.get("entity_id"), - output.get("field"), - ) - continue - targets.append( - EnrichmentTarget( - entity_id=str(output["entity_id"]), - tenant_id=self.tenant_id, - field_name=str(output["field"]), - seed_value=output.get("value"), - source_confidence=confidence, - origin_packet_id=self.packet_id, - origin_inference_rule=str(output.get("rule", "unknown")), - ) - ) - return targets - - -# --------------------------------------------------------------------------- -# Return channel — singleton per process -# --------------------------------------------------------------------------- - - -class GraphToEnrichReturnChannel: - """ - Async queue that bridges GRAPH inference outputs back to ENRICH. - - Usage (GRAPH side): - channel = GraphToEnrichReturnChannel.get_instance() - await channel.submit(envelope) - - Usage (ENRICH convergence_controller side): - targets = await channel.drain(tenant_id="acme", timeout_seconds=0.5) - """ - - _instance: GraphToEnrichReturnChannel | None = None - - def __init__(self, maxsize: int = 10_000) -> None: - # Per-tenant queues: tenant_id → asyncio.Queue[EnrichmentTarget] - self._queues: dict[str, asyncio.Queue[EnrichmentTarget]] = {} - self._maxsize = maxsize - self._submitted: int = 0 - self._drained: int = 0 - self._rejected: int = 0 - - @classmethod - def get_instance(cls) -> GraphToEnrichReturnChannel: - if cls._instance is None: - cls._instance = cls() - return cls._instance - - @classmethod - def reset_instance(cls) -> None: - """Test helper — reset the singleton.""" - cls._instance = None - - def _queue_for(self, tenant_id: str) -> asyncio.Queue[EnrichmentTarget]: - if tenant_id not in self._queues: - self._queues[tenant_id] = asyncio.Queue(maxsize=self._maxsize) - return self._queues[tenant_id] - - async def submit(self, envelope: GraphInferenceResultEnvelope) -> int: - """ - Validate the envelope, convert outputs to targets, enqueue them. - Returns the number of targets enqueued. - Hard-raises ContractViolationError if the envelope is invalid (Gap-1). - """ - envelope.validate() # raises on violation — no silent bypass - targets = envelope.to_targets() - q = self._queue_for(envelope.tenant_id) - count = 0 - for target in targets: - try: - q.put_nowait(target) - count += 1 - except asyncio.QueueFull: - self._rejected += 1 - logger.warning( - "EnrichmentTargetQueue full for tenant=%s — dropping target entity=%s field=%s", - envelope.tenant_id, - target.entity_id, - target.field_name, - ) - self._submitted += count - logger.info( - "GraphToEnrichReturnChannel: submitted %d targets for tenant=%s (packet=%s)", - count, - envelope.tenant_id, - envelope.packet_id, - ) - return count - - async def drain( - self, - tenant_id: str, - *, - timeout_seconds: float = 0.1, - max_targets: int = 500, - ) -> list[EnrichmentTarget]: - """ - Non-blocking drain: collect up to max_targets from the tenant queue. - Returns immediately if the queue is empty after `timeout_seconds` seconds. - Called by convergence_controller at the start of each new pass. - """ - q = self._queue_for(tenant_id) - targets: list[EnrichmentTarget] = [] - deadline = time.monotonic() + timeout_seconds - while len(targets) < max_targets: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - try: - target = await asyncio.wait_for(q.get(), timeout=remaining) - targets.append(target) - q.task_done() - except TimeoutError: - break - self._drained += len(targets) - if targets: - logger.info( - "GraphToEnrichReturnChannel: drained %d targets for tenant=%s", - len(targets), - tenant_id, - ) - return targets - - def stats(self) -> dict[str, int | dict[str, int]]: - return { - "submitted": self._submitted, - "drained": self._drained, - "rejected": self._rejected, - "queue_sizes": {t: q.qsize() for t, q in self._queues.items()}, - } - - -# --------------------------------------------------------------------------- -# Integration helper: build a valid envelope from raw GRAPH output -# --------------------------------------------------------------------------- - - -def build_graph_inference_result_envelope( - *, - tenant_id: str, - inference_outputs: list[dict[str, Any]], -) -> GraphInferenceResultEnvelope: - """ - Factory used by GRAPH ConvergenceLoop to produce a properly hashed envelope. - Call this instead of constructing GraphInferenceResultEnvelope directly. - """ - payload_bytes = json.dumps(inference_outputs, sort_keys=True).encode() - content_hash = hashlib.sha256(payload_bytes).hexdigest() - packet_id = f"gir_{uuid.uuid4().hex}" - envelope_payload = { - "packet_id": packet_id, - "tenant_id": tenant_id, - "content_hash": content_hash, - } - envelope_hash = hashlib.sha256(json.dumps(envelope_payload, sort_keys=True).encode()).hexdigest() - return GraphInferenceResultEnvelope( - packet_id=packet_id, - tenant_id=tenant_id, - inference_outputs=inference_outputs, - content_hash=content_hash, - envelope_hash=envelope_hash, - ) diff --git a/engine/startup_wiring.py b/engine/startup_wiring.py deleted file mode 100644 index 38695f2c..00000000 --- a/engine/startup_wiring.py +++ /dev/null @@ -1,65 +0,0 @@ -""" ---- L9_META --- -l9_schema: 1 -origin: engine-specific -engine: graph -layer: [config] -tags: [startup, wiring] -owner: engine-team -status: active ---- /L9_META --- - -GAP-FIX STARTUP WIRING -Add these calls to your application lifespan / startup handler in order. -This file is a recipe — adapt paths to match your actual app entrypoint. -""" - -from __future__ import annotations - -import logging - -import asyncpg - -logger = logging.getLogger(__name__) - - -async def apply_all_gap_fixes(pg_dsn: str, neo4j_driver, domain_pack_loader) -> None: - """ - Call once during application startup, before serving requests. - Parameters: - pg_dsn - asyncpg-compatible DSN string - neo4j_driver - AsyncDriver from neo4j-driver - domain_pack_loader - DomainPackLoader instance - """ - - # ── Gap 5: Wire PostgreSQL audit pool ──────────────────────────────────── - from shared.audit_persistence import configure_audit_pool - - pg_pool = await asyncpg.create_pool(pg_dsn, min_size=2, max_size=10) - await configure_audit_pool(pg_pool) - logger.info("startup: Gap-5 audit pool wired") - - # ── Gap 2: Initialise GRAPH→ENRICH return channel ──────────────────────── - from engine.graph_return_channel import GraphToEnrichReturnChannel - - GraphToEnrichReturnChannel.get_instance() - logger.info("startup: Gap-2 return channel initialised") - - # ── Gap 6: Register community-export hook on GDS scheduler ─────────────── - from graph.community_export import export_community_labels_to_enrich - - try: - from graph.gds_scheduler import GDSScheduler - - GDSScheduler.register_post_job_hook( - job_type="louvain", - hook=lambda tenant_id, domain_id: export_community_labels_to_enrich(neo4j_driver, tenant_id, domain_id), - ) - logger.info("startup: Gap-6 community export hook registered") - except ImportError: - logger.warning("startup: GDSScheduler not found — register Gap-6 hook manually") - - # Gap 9: the removed v1 inference bridge has no startup wiring. - # Do not add a successor bridge unless a real producer/consumer contract exists. - - logger.info("startup: all gap fixes applied successfully") diff --git a/tests/gap_fixes/test_gap1_contract.py b/tests/gap_fixes/test_gap1_contract.py deleted file mode 100644 index d88c87b5..00000000 --- a/tests/gap_fixes/test_gap1_contract.py +++ /dev/null @@ -1,49 +0,0 @@ -""" -Tests for GAP-1: contract_enforcement.py -""" - -from __future__ import annotations - -import pytest - -from engine.contract_enforcement import ( - ContractViolationError, - build_graph_sync_packet, - build_schema_proposal_packet, - enforce_packet_envelope, -) - - -def test_bare_dict_is_rejected() -> None: - with pytest.raises(ContractViolationError): - enforce_packet_envelope({"entity_type": "account", "batch": []}, expected_type="graph_sync") - - -def test_type_mismatch_raises() -> None: - pkt = build_graph_sync_packet(tenant_id="t1", entity_type="account", batch=[{"id": "1"}]) - with pytest.raises(ContractViolationError, match="mismatch"): - enforce_packet_envelope(pkt, expected_type="enrich_request") - - -def test_valid_graph_sync_passes() -> None: - pkt = build_graph_sync_packet(tenant_id="t1", entity_type="account", batch=[{"id": "1"}]) - result = enforce_packet_envelope(pkt, expected_type="graph_sync") - assert result["packet_type"] == "graph_sync" - assert result["content_hash"] - assert result["envelope_hash"] - - -def test_schema_proposal_allowed() -> None: - pkt = build_schema_proposal_packet( - tenant_id="t1", - proposed_fields=[{"name": "facility_tier", "type": "string"}], - ) - result = enforce_packet_envelope(pkt, expected_type="schema_proposal") - assert result["packet_id"].startswith("sp_") - - -def test_tampered_content_hash_rejected() -> None: - pkt = build_graph_sync_packet(tenant_id="t1", entity_type="account", batch=[{"id": "1"}]) - pkt["content_hash"] = "deadbeef" - with pytest.raises(ContractViolationError, match="content_hash mismatch"): - enforce_packet_envelope(pkt, expected_type="graph_sync") diff --git a/tests/gap_fixes/test_gap2_return_channel.py b/tests/gap_fixes/test_gap2_return_channel.py deleted file mode 100644 index 23dca89a..00000000 --- a/tests/gap_fixes/test_gap2_return_channel.py +++ /dev/null @@ -1,72 +0,0 @@ -""" -Tests for GAP-2: graph_return_channel.py -""" - -from __future__ import annotations - -import pytest - -from engine.contract_enforcement import ContractViolationError -from engine.graph_return_channel import ( - GraphToEnrichReturnChannel, - build_graph_inference_result_envelope, -) - - -@pytest.fixture(autouse=True) -def reset_channel() -> None: - GraphToEnrichReturnChannel.reset_instance() - yield - GraphToEnrichReturnChannel.reset_instance() - - -@pytest.mark.asyncio -async def test_submit_and_drain() -> None: - envelope = build_graph_inference_result_envelope( - tenant_id="acme", - inference_outputs=[ - { - "entity_id": "e1", - "field": "facility_tier", - "value": "large", - "confidence": 0.88, - "rule": "louvain_community_detection", - } - ], - ) - channel = GraphToEnrichReturnChannel.get_instance() - count = await channel.submit(envelope) - assert count == 1 - targets = await channel.drain("acme", timeout_seconds=0.1) - assert len(targets) == 1 - assert targets[0].field_name == "facility_tier" - assert targets[0].source_confidence == 0.88 - - -@pytest.mark.asyncio -async def test_low_confidence_filtered() -> None: - envelope = build_graph_inference_result_envelope( - tenant_id="acme", - inference_outputs=[{"entity_id": "e1", "field": "x", "value": "v", "confidence": 0.30, "rule": "r1"}], - ) - channel = GraphToEnrichReturnChannel.get_instance() - count = await channel.submit(envelope) - assert count == 0 - - -@pytest.mark.asyncio -async def test_tampered_envelope_rejected() -> None: - envelope = build_graph_inference_result_envelope( - tenant_id="acme", - inference_outputs=[{"entity_id": "e1", "field": "x", "value": "v", "confidence": 0.9, "rule": "r"}], - ) - # Tamper with content after hashes are set - import dataclasses - - tampered = dataclasses.replace( - envelope, - inference_outputs=[{"entity_id": "e_TAMPERED"}], - ) - channel = GraphToEnrichReturnChannel.get_instance() - with pytest.raises(ContractViolationError): - await channel.submit(tampered) diff --git a/tests/gap_fixes/test_gap5_audit.py b/tests/gap_fixes/test_gap5_audit.py deleted file mode 100644 index 14c419ae..00000000 --- a/tests/gap_fixes/test_gap5_audit.py +++ /dev/null @@ -1,60 +0,0 @@ -""" -Tests for GAP-5: audit_persistence.py -""" - -from __future__ import annotations - -import time -from unittest.mock import AsyncMock, MagicMock - -import pytest - -from engine.compliance import audit_persistence - - -@pytest.fixture(autouse=True) -def reset_pool(): - """Reset module-level pool between tests.""" - original = audit_persistence._POOL - audit_persistence._POOL = None - yield - audit_persistence._POOL = original - - -@pytest.mark.asyncio -async def test_flush_returns_zero_when_no_pool() -> None: - entries = [{"tenant_id": "t1", "actor": "system", "action": "match"}] - result = await audit_persistence.flush_audit_entries(entries) - assert result == 0 - - -@pytest.mark.asyncio -async def test_flush_returns_zero_for_empty_entries(monkeypatch: pytest.MonkeyPatch) -> None: - mock_pool = MagicMock() - monkeypatch.setattr(audit_persistence, "_POOL", mock_pool) - result = await audit_persistence.flush_audit_entries([]) - assert result == 0 - - -@pytest.mark.asyncio -async def test_flush_inserts_rows(monkeypatch: pytest.MonkeyPatch) -> None: - entries = [ - {"tenant_id": "t1", "actor": "system", "action": "match", "detail": "ok", "created_at": time.time()}, - {"tenant_id": "t1", "actor": "user1", "action": "sync"}, - ] - - mock_conn = AsyncMock() - mock_conn.executemany = AsyncMock() - - mock_pool = MagicMock() - mock_pool.acquire = MagicMock( - return_value=AsyncMock(__aenter__=AsyncMock(return_value=mock_conn), __aexit__=AsyncMock(return_value=False)) - ) - monkeypatch.setattr(audit_persistence, "_POOL", mock_pool) - - result = await audit_persistence.flush_audit_entries(entries) - assert result == 2 - mock_conn.executemany.assert_called_once() - call_args = mock_conn.executemany.call_args - assert "INSERT INTO audit_log" in call_args[0][0] - assert len(call_args[0][1]) == 2 diff --git a/tests/gap_fixes/test_gap9_inference_authority.py b/tests/gap_fixes/test_gap9_inference_authority.py index 95c9ace0..afd0ef95 100644 --- a/tests/gap_fixes/test_gap9_inference_authority.py +++ b/tests/gap_fixes/test_gap9_inference_authority.py @@ -13,10 +13,26 @@ def test_removed_inference_bridge_has_no_compatibility_module() -> None: assert not (ROOT / "engine" / "inference_bridge.py").exists() -def test_startup_recipe_does_not_reintroduce_undeclared_kb_loading() -> None: - source = (ROOT / "engine" / "startup_wiring.py").read_text(encoding="utf-8") - assert "spec.kb" not in source - assert "load_domain_rules" not in source +def test_no_module_reintroduces_undeclared_kb_loading() -> None: + """The recipe file this guard used to read has since been removed as an + unrunnable gap-fix artifact, so scan the whole engine tree instead. Broader + than the original assertion, and no longer tied to one file's existence. + """ + banned = ("spec.kb", "load_domain_rules") + offenders = [ + str(path.relative_to(ROOT)) + for path in sorted((ROOT / "engine").rglob("*.py")) + if any(token in path.read_text(encoding="utf-8") for token in banned) + ] + assert offenders == [] + + +def test_startup_recipe_module_stays_removed() -> None: + """engine/startup_wiring.py instructed operators to wire five artifacts that + no longer exist, and could never run (its first import named a package that + is absent). It must not come back. + """ + assert not (ROOT / "engine" / "startup_wiring.py").exists() def test_registry_exposes_only_supported_in_code_rule_surface() -> None: diff --git a/tests/invariants/test_module_reachability.py b/tests/invariants/test_module_reachability.py new file mode 100644 index 00000000..2791ed3b --- /dev/null +++ b/tests/invariants/test_module_reachability.py @@ -0,0 +1,263 @@ +"""Anti-resurrection invariants for the removed gap-fix artifact island. + +A historical gap-fix bundle left seven executable engine modules in the tree +that no production entrypoint could ever reach. They survived because they +imported each other and because tests imported them — a closed loop that reads +as "covered" from a coverage report and as "active" from an L9_META header, +while the runtime never touched a line of it. + +These invariants make that failure mode mechanically detectable rather than +archaeologically detectable: + +* the removed module paths must stay removed; +* no engine or chassis module may import the removed surfaces; +* no module may reintroduce a gap-fix activation recipe; +* the canonical owner of each reclaimed responsibility must remain the sole one. + +The static import-graph analyzer below is the reusable half. It models Python +import semantics closely enough to answer "can production reach this module?", +and is exercised here against the real tree. + +Scope note: this is deliberately NOT a full-tree reachability gate. At the time +these invariants were written, 59 further engine modules across nine unaudited +subsystems were also unreachable from production entrypoints. Gating on the full +tree would have required either enumerating all 59 as permanent exemptions or +classifying them without evidence — the audit contract prohibits both. That work +is tracked as DEF-001 in +docs/audits/2026-08-23-gap-fix-artifact-convergence/GAP_FIX_REACHABILITY_CLASSIFICATION.yaml. +Tightening this module to the full tree is a scope decision, not new machinery. +""" + +from __future__ import annotations + +import ast +from collections import deque +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +ENGINE = ROOT / "engine" +CHASSIS = ROOT / "chassis" +PKG = "engine" + +# Modules removed as orphaned, stale, or test-only gap-fix artifacts. +REMOVED_MODULES = { + "engine/compliance/audit_persistence.py", + "engine/contract_enforcement.py", + "engine/convergence_controller_patch.py", + "engine/graph/community_export.py", + "engine/graph/graph_sync_client_fix.py", + "engine/graph_return_channel.py", + "engine/startup_wiring.py", +} + +# Import targets and symbols that only the removed island ever provided. +REMOVED_IMPORT_TARGETS = ( + "engine.compliance.audit_persistence", + "engine.contract_enforcement", + "engine.convergence_controller_patch", + "engine.graph.community_export", + "engine.graph.graph_sync_client_fix", + "engine.graph_return_channel", + "engine.startup_wiring", +) + +REMOVED_SYMBOLS = ( + "GraphToEnrichReturnChannel", + "GraphInferenceResultEnvelope", + "build_graph_inference_result_envelope", + "apply_return_channel_targets", + "export_community_labels_to_enrich", + "configure_audit_pool", + "flush_audit_entries", + "enforce_packet_envelope", + "build_graph_sync_packet", + "build_schema_proposal_packet", + "ContractViolationError", + "apply_all_gap_fixes", + "patch_convergence_controller", +) + +# Responsibilities the island duplicated, and the single owner each returned to. +CANONICAL_OWNERS = { + "audit persistence": "engine/compliance/audit.py", + "packet envelope contract": "engine/packet/packet_envelope.py", + "graph sync write path": "engine/sync/generator.py", + "community detection write": "engine/gds/scheduler.py", + "startup lifecycle": "engine/boot.py", +} + + +# ── static import graph ────────────────────────────────────────────────────── + + +def _module_name(path: Path) -> str: + parts = list(path.relative_to(ROOT).with_suffix("").parts) + if parts[-1] == "__init__": + parts.pop() + return ".".join(parts) + + +def _engine_modules() -> dict[str, Path]: + return {_module_name(p): p for p in sorted(ENGINE.rglob("*.py"))} + + +def _resolve(target: str, known: set[str]) -> str | None: + """Map a dotted import target onto an engine module that exists. + + `from engine.graph import driver` names a module; `from engine.graph.driver + import GraphDriver` names a symbol inside one. Try the longest match first. + """ + if target in known: + return target + parent = target.rsplit(".", 1)[0] if "." in target else None + return parent if parent in known else None + + +def _imports_of(path: Path, known: set[str]) -> set[str]: + """Engine modules imported by `path`, including deferred and dynamic ones.""" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + self_parts = _module_name(path).split(".") + found: set[str] = set() + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name.split(".")[0] == PKG and (hit := _resolve(alias.name, known)): + found.add(hit) + + elif isinstance(node, ast.ImportFrom): + if node.level: + depth = len(self_parts) - node.level + if path.name == "__init__.py": + depth += 1 + base = self_parts[: max(0, depth)] + module = ".".join([*base, node.module]) if node.module else ".".join(base) + else: + module = node.module or "" + if module.split(".")[0] != PKG: + continue + if hit := _resolve(module, known): + found.add(hit) + for alias in node.names: + if hit := _resolve(f"{module}.{alias.name}", known): + found.add(hit) + + elif isinstance(node, ast.Call): + fn = node.func + called = getattr(fn, "attr", None) or getattr(fn, "id", None) + if called in {"import_module", "__import__", "find_spec"}: + for arg in node.args: + if not isinstance(arg, ast.Constant) or not isinstance(arg.value, str): + continue + if arg.value.split(".")[0] == PKG and (hit := _resolve(arg.value, known)): + found.add(hit) + + return found + + +def _with_ancestors(module: str, known: set[str]) -> set[str]: + """Importing engine.a.b.c also executes engine.a.b, engine.a and engine.""" + parts = module.split(".") + return {".".join(parts[:i]) for i in range(1, len(parts) + 1)} & known + + +def reachable_from_production() -> set[str]: + """Engine modules reachable from chassis ingress and the lifecycle hook.""" + modules = _engine_modules() + known = set(modules) + graph = {name: _imports_of(path, known) for name, path in modules.items()} + + roots: set[str] = set() + for path in sorted(CHASSIS.rglob("*.py")): + roots |= _imports_of(path, known) + roots |= {"engine", "engine.boot"} & known + + seen: set[str] = set() + for root in roots: + seen |= _with_ancestors(root, known) + queue = deque(seen) + while queue: + for target in graph.get(queue.popleft(), ()): + for ancestor in _with_ancestors(target, known): + if ancestor not in seen: + seen.add(ancestor) + queue.append(ancestor) + return seen + + +# ── invariants ─────────────────────────────────────────────────────────────── + + +def test_removed_gap_fix_modules_stay_removed() -> None: + resurrected = sorted(rel for rel in REMOVED_MODULES if (ROOT / rel).exists()) + assert resurrected == [] + + +def test_no_runtime_module_imports_a_removed_surface() -> None: + offenders: list[str] = [] + for path in sorted([*ENGINE.rglob("*.py"), *CHASSIS.rglob("*.py")]): + source = path.read_text(encoding="utf-8") + for target in REMOVED_IMPORT_TARGETS: + if target in source: + offenders.append(f"{path.relative_to(ROOT)} -> {target}") + assert offenders == [] + + +def test_no_runtime_module_reintroduces_a_removed_symbol() -> None: + """Some of these names are generic enough to be reinvented by accident — + `ContractViolationError` and `enforce_packet_envelope` especially. That is + the point: packet-envelope validation belongs to engine/packet/, and audit + persistence to engine/compliance/audit.py. If a canonical owner genuinely + needs one of these names, add it there and drop it from REMOVED_SYMBOLS in + the same change — deliberately, with the owner named in the commit. Do not + delete this test to get past it. + """ + offenders: list[str] = [] + for path in sorted([*ENGINE.rglob("*.py"), *CHASSIS.rglob("*.py")]): + source = path.read_text(encoding="utf-8") + for symbol in REMOVED_SYMBOLS: + if symbol in source: + offenders.append(f"{path.relative_to(ROOT)} -> {symbol}") + assert offenders == [] + + +def test_no_module_ships_a_gap_fix_activation_recipe() -> None: + """A module that tells an operator to wire it at startup, instead of being + wired, is how the island survived. engine/boot.py is the startup owner. + """ + offenders: list[str] = [] + for path in sorted(ENGINE.rglob("*.py")): + lowered = path.read_text(encoding="utf-8").lower() + if "gap-fix" in lowered or "gap fix" in lowered: + offenders.append(str(path.relative_to(ROOT))) + assert offenders == [] + + +def test_canonical_owners_remain_present_and_sole() -> None: + missing = sorted(f"{role}: {rel}" for role, rel in CANONICAL_OWNERS.items() if not (ROOT / rel).exists()) + assert missing == [] + + +def test_canonical_owners_are_production_reachable() -> None: + """The point of the removals: each reclaimed responsibility now sits with an + owner the runtime actually reaches. + """ + reachable = reachable_from_production() + expected = { + "engine.compliance.audit", + "engine.packet.packet_envelope", + "engine.sync.generator", + "engine.gds.scheduler", + "engine.boot", + } + assert sorted(expected - reachable) == [] + + +def test_import_graph_analyzer_models_a_known_edge() -> None: + """Guard the analyzer itself: a real, stable production edge must be seen, + so a silently broken parser cannot make every invariant above vacuous. + """ + modules = _engine_modules() + known = set(modules) + assert "engine.graph.driver" in _imports_of(modules["engine.boot"], known) + assert "engine.handlers" in reachable_from_production()