diff --git a/.claude/rules/capability-registry.md b/.claude/rules/capability-registry.md index 6088e3e5..0143de9b 100644 --- a/.claude/rules/capability-registry.md +++ b/.claude/rules/capability-registry.md @@ -28,3 +28,11 @@ Before building any of these, check if it already exists. Use the existing imple | Pareto scoring | engine/scoring/pareto.py | multi-objective optimization | | Convergence loop | engine/feedback/convergence.py | `ConvergenceLoop` | | Label sanitization | engine/utils/security.py | `sanitize_label()` | +| CEG → Gate → EIE egress (the ONLY peer egress) | engine/gate_egress.py | `request_enrichment()`, `emit_graph_inference_result()` | +| Inference outputs → EIE wire shape | engine/gate_egress.py | `build_inference_outputs()` (0.55 floor, matches EIE) | +| Domain database provisioning | engine/graph/driver.py | `GraphDriver.ensure_database()` (flag: `auto_create_domain_database`) | +| Database-name validation for DDL | engine/graph/driver.py | `_DATABASE_NAME_RE` — `sanitize_label()` forbids the dashes domain ids use | +| Inference rule execution | engine/inference_rule_registry.py | `execute_rule()`, `list_registered_rules()` | + +Do not add a second outbound client. `engine/gate_egress.py` is the only +CEG → peer egress; Gate resolves the destination from the action. diff --git a/.claude/rules/feature-flags.md b/.claude/rules/feature-flags.md index ce168d68..169ffa55 100644 --- a/.claude/rules/feature-flags.md +++ b/.claude/rules/feature-flags.md @@ -40,6 +40,15 @@ All in Settings class, controllable via env vars. Contract 21: every behavioral | strict_tenant_database | False | Require explicit `database=` on GraphDriver calls; no implicit 'neo4j' fallback (W7-01) | | require_sdk_chassis_in_prod | False | Fail startup if `L9_CHASSIS != sdk` when `l9_env == prod` (W7-02) | +## Constellation Seam (EIE ↔ Gate ↔ CEG) +| Flag | Default | Purpose | +|------|---------|---------| +| auto_enrich_via_gate | False | Dispatch `enrich` to EIE through Gate (spends EIE budget) | +| graph_inference_feedback_enabled | False | Emit `graph-inference-result` to EIE through Gate (EIE-008 / CEG-006) | +| auto_create_domain_database | False | Provision the tenant domain database on first use (CEG-008; Enterprise-only CREATE DATABASE) | +| health_api_enabled | False | Expose engine/health/api.py via admin health_* subactions (CEG-006) | +| unvalidated_domain_packs_enabled | False | Serve the 5 migrated packs whose gates do not compile to executable Cypher (CEG-009) | + ## Entity Resolution | Flag | Default | Purpose | |------|---------|---------| diff --git a/.claude/rules/subsystems.md b/.claude/rules/subsystems.md index ec6ed096..9a0491a9 100644 --- a/.claude/rules/subsystems.md +++ b/.claude/rules/subsystems.md @@ -28,6 +28,15 @@ All handlers: `async def handle_*(tenant: str, payload: dict) -> dict` | calibration_run | Score calibration vs expected ranges | | score_feedback | Compute weight adjustment proposal | | apply_weight_proposal | Apply proposed weight change | +| health_assess | AI-readiness assessment for one entity (`engine/health/api.py`) | +| health_batch_assess | Incremental batch readiness scan with cost ceilings | +| health_report | Health report for an entity (Seed tier) | +| emit_inference_feedback | Run inference rules and send `graph-inference-result` to EIE via Gate (flag: `graph_inference_feedback_enabled`) | + +CEG-006: the three `health_*` subactions are what makes `engine/health/api.py` +reachable. It was imported by nothing, so `trigger_reenrichment_v2` — and with +it the whole CEG → Gate → EIE enrichment request — had no trigger any inbound +packet could reach. ## Dependency Map ``` diff --git a/.env.template b/.env.template index 67f80c57..0d163756 100644 --- a/.env.template +++ b/.env.template @@ -21,9 +21,6 @@ NEO4J_USERNAME=neo4j NEO4J_PASSWORD=l9-dev-password NEO4J_POOL_SIZE=20 -── Redis Cache ───────────────────────────────────────────── -REDIS_URL=redis://redis:6379/0 - ── API Configuration ─────────────────────────────────────── API_KEY=dev-key-sha256-not-for-production LOG_LEVEL=debug @@ -47,7 +44,6 @@ LOCAL DEVELOPMENT ACCESS (from host machine, not Docker) API: http://localhost:8000/v1/health Neo4j Browser: http://localhost:7475 Neo4j Bolt: bolt://localhost:7688 -Redis: localhost:6379 Neo4j Credentials: User: neo4j Password: l9-dev-password @@ -68,11 +64,16 @@ KGE_EMBEDDING_DIM=300 GATE_URL=http://gate:9000 GATE_ADMIN_TOKEN=your-gate-admin-token-here L9_NODE_NAME=graph -L9_NODE_SPEC_PATH=engine/spec.yaml GATE_REGISTRATION_ENABLED=true GATE_REGISTER_OVERWRITE=true GATE_REGISTER_RETRIES=3 GATE_CLIENT_TIMEOUT_SECONDS=30.0 +# CEG-003: this is the name the SDK reads (gate/config.py: +# spec_path=os.getenv("GATE_NODE_SPEC_PATH", "engine/spec.yaml")). An +# L9_NODE_SPEC_PATH line sat above it with no reader at all, so an operator +# relocating the spec edited the variable that does nothing and registration +# kept using the default until that path stopped existing — at which point +# register_with_gate swallows FileNotFoundError and returns False. GATE_NODE_SPEC_PATH=engine/spec.yaml # ── SDK chassis (L9_CHASSIS=sdk) ───────────────────────────── @@ -104,7 +105,14 @@ L9_SIGNING_ALGORITHM=hmac-sha256 L9_SIGNING_KEY=change-me-shared-hmac-secret L9_SIGNING_KEY_ID=graph-engine-v1 # ed25519 alternative: L9_SIGNING_ALGORITHM=ed25519 + L9_SIGNING_PRIVATE_KEY -# L9_VERIFYING_KEYS_JSON={"gate-v1":""} +# +# CEG-002: REQUIRED whenever L9_REQUIRE_SIGNATURE=true, not optional. The same +# flag turns on verification of the responses Gate signs, and Gate signs with +# *its* key id, which CEG cannot resolve from its own signing key. Leave this +# unset and every signed Gate response is rejected with "no verifying key +# available for transport signature verification". Under hmac-sha256 the value +# is the shared secret; under ed25519 it is the peer's public key. +L9_VERIFYING_KEYS_JSON={"gate-v1":"change-me-shared-hmac-secret"} # Allowed actions — engine.handlers.ACTION_HANDLERS minus `enrich`: Gate owns # the `enrich` name for EIE, so CEG must never accept or advertise it. diff --git a/.github/workflows/ci-quality.yml b/.github/workflows/ci-quality.yml index e5a4ab6f..3651146a 100644 --- a/.github/workflows/ci-quality.yml +++ b/.github/workflows/ci-quality.yml @@ -182,15 +182,7 @@ jobs: --health-timeout 5s --health-retries 5 - redis: - image: redis:7-alpine - ports: - - 6379:6379 - options: >- - --health-cmd "redis-cli ping" - --health-interval 10s - --health-timeout 5s - --health-retries 5 + # CEG-007: redis service container removed with the dependency. steps: - name: Checkout code @@ -215,7 +207,6 @@ jobs: - name: Run tests with coverage env: DATABASE_URL: postgresql://${{ vars.TEST_DB_USER || 'test' }}:${{ vars.TEST_DB_PASSWORD || 'test' }}@localhost:5432/${{ vars.TEST_DB_NAME || 'test_db' }} - REDIS_URL: redis://localhost:6379 TESTING: "true" run: PYTHONPATH=. pytest tests/ -v --cov=engine --cov-report=xml --cov-report=term --ignore=tests/e2e diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fa28a0f4..1a8a3d7e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -258,15 +258,9 @@ jobs: --health-retries 5 # Service only runs if POSTGRES_ENABLED=true (GitHub ignores services with falsy conditions) - redis: - image: ${{ vars.REDIS_IMAGE || 'redis:7-alpine' }} - ports: - - 6379:6379 - options: >- - --health-cmd "redis-cli ping" - --health-interval 10s - --health-timeout 5s - --health-retries 5 + # CEG-007: the redis service container is gone with the dependency. No + # code under engine/ or chassis/ imports redis, so it started a container + # every run for nothing. steps: - name: Checkout Repository @@ -291,7 +285,6 @@ jobs: - name: Run Tests with Coverage env: DATABASE_URL: postgresql://${{ vars.POSTGRES_USER || 'test_user' }}:test_password@localhost:5432/${{ vars.POSTGRES_DB || 'test_db' }} - REDIS_URL: redis://localhost:6379/0 run: | echo "Running test suite..." PYTHONPATH=. pytest ${{ env.TEST_DIR }} \ diff --git a/AGENTS.md b/AGENTS.md index 8152aa6e..305eda9f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,7 +22,7 @@ Cross-tool agent instructions for the CEG repository. Read by Claude Code, Codex ```bash make setup # Install deps, pre-commit hooks, verify Neo4j -make dev # docker-compose (app + Neo4j + Redis + Prometheus + Grafana) +make dev # docker-compose (app + Neo4j + Postgres) make test # Full pytest suite (unit + integration + compliance) make test-unit # Gate compilation, scoring math, parameter resolution make test-integration # testcontainers-neo4j full pipeline diff --git a/Makefile b/Makefile index 06f64a9c..1042e563 100644 --- a/Makefile +++ b/Makefile @@ -45,8 +45,6 @@ health:Check all service health @curl -sf http://localhost:8000/v1/health | python -m json.tool || echo "API: DOWN" @echo "── Neo4j ──" @docker exec l9-graph-neo4j cypher-shell -u neo4j -p l9-dev-password "RETURN 'ok'" 2>/dev/null || echo "Neo4j: DOWN" - @echo "── Redis ──" - @docker exec l9-graph-redis redis-cli ping || echo "Redis: DOWN" # ── Testing ──────────────────────────────────────────────── @@ -72,13 +70,10 @@ shell:Python shell inside API container neo4j-shell:Cypher shell into Neo4j docker exec -it l9-graph-neo4j cypher-shell -u neo4j -p l9-dev-password -redis-shell:Redis CLI - docker exec -it l9-graph-redis redis-cli - # ── Local Dev (API outside Docker, DBs in Docker) ───────── -local-dbs:Start only Neo4j + Redis - docker compose up -d neo4j redis +local-dbs:Start only Neo4j + Postgres + docker compose up -d neo4j postgres local-api:Run API locally against Dockerized DBs (SDK chassis; alias of local-api-sdk) $(MAKE) local-api-sdk @@ -86,7 +81,6 @@ local-api:Run API locally against Dockerized DBs (SDK chassis; alias of local-ap local-api-legacy:Run the legacy dict chassis locally (permitted in L9_ENV=dev|local|test) PLASTICOS_NEO4J_URI=bolt://localhost:7687 \ PLASTICOS_NEO4J_PASSWORD=l9-dev-password \ - PLASTICOS_REDIS_URL=redis://localhost:6379/0 \ PLASTICOS_LOG_LEVEL=debug \ L9_LIFECYCLE_HOOK=engine.boot:GraphLifecycle \ L9_CHASSIS=legacy \ @@ -95,7 +89,6 @@ local-api-legacy:Run the legacy dict chassis locally (permitted in L9_ENV=dev|lo local-api-sdk:Run API locally on the SDK chassis (L9_CHASSIS=sdk) PLASTICOS_NEO4J_URI=bolt://localhost:7687 \ PLASTICOS_NEO4J_PASSWORD=l9-dev-password \ - PLASTICOS_REDIS_URL=redis://localhost:6379/0 \ PLASTICOS_LOG_LEVEL=debug \ L9_LIFECYCLE_HOOK=engine.boot:GraphLifecycle \ L9_CHASSIS=sdk \ @@ -199,8 +192,6 @@ deploy-health: guard-vps-host ## Remote healthcheck over SSH (VPS ports may be f @ssh $(SSH_OPTS) $(SSH_TARGET) "curl -sf http://localhost:8000/v1/health && echo" || echo "API: DOWN" @echo "── Neo4j ──" @ssh $(SSH_OPTS) $(SSH_TARGET) "curl -sf http://localhost:7474 >/dev/null" && echo "Neo4j: UP" || echo "Neo4j: DOWN" - @echo "── Redis ──" - @ssh $(SSH_OPTS) $(SSH_TARGET) "docker exec l9-redis-prod redis-cli ping" || echo "Redis: DOWN" # ── Cleanup ──────────────────────────────────────────────── diff --git a/README.md b/README.md index aa412e84..0083fa47 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ One engine. Any vertical. No custom code per domain. git clone && cd l9-engine ./scripts/setup.sh -# 2. Start local stack (Neo4j + Redis + API) +# 2. Start local stack (Neo4j + Postgres + API) ./scripts/dev.sh # 3. Seed sample data @@ -81,8 +81,8 @@ All intelligence features are disabled by default and activated per-domain via Y │ Domain Spec Loader │ │ YAML → Pydantic → Compiled Cypher │ ├──────────────────────────────────────────────────────┤ -│ Neo4j (multi-database) │ Redis (cache/scheduler) │ -└──────────────────────────┴───────────────────────────┘ +│ Neo4j (multi-database) │ +└──────────────────────────────────────────────────────┘ ``` ### Core Concepts @@ -121,8 +121,7 @@ l9-engine/ │ │ ├── pii.py # PII hash/encrypt/redact │ │ └── audit.py # Audit logging │ └── db/ # Database layer -│ ├── neo4j.py # Async Neo4j driver pool -│ └── redis.py # Redis connection +│ └── neo4j.py # Async Neo4j driver pool ├── domains/ # Domain specification packs │ ├── plasticos/ │ ├── mortgage-brokerage/ @@ -153,7 +152,6 @@ l9-engine/ | Component | Version | Required Plugins | |-----------|---------|------------------| | Neo4j | 5.15+ Enterprise | APOC, Graph Data Science (GDS) | -| Redis | 7.x | — | | Python | 3.12+ | — | **Neo4j Plugins:** @@ -174,7 +172,6 @@ Consistent across all L9 repos. Set in `.env` (local) or SSM Parameter Store (pr | `NEO4J_USERNAME` | `neo4j` | Neo4j username | | `NEO4J_PASSWORD` | — | Neo4j password (SSM in prod) | | `NEO4J_DATABASE` | `neo4j` | Default database | -| `REDIS_URL` | `redis://localhost:6379/0` | Redis connection | | `API_PORT` | `8000` | API listen port | | `API_WORKERS` | `4` | Uvicorn workers | | `DOMAINS_ROOT` | `./domains` | Path to domain specs | diff --git a/Readme-Requirements.md b/Readme-Requirements.md index e512527c..35207a32 100644 --- a/Readme-Requirements.md +++ b/Readme-Requirements.md @@ -31,7 +31,6 @@ status: active - `neo4j_uri`, `neo4j_username`, `neo4j_password` → consumed by `dependencies.py` to init `GraphDriver`[^1] - `domains_root` → consumed by `dependencies.py` to init `DomainPackLoader`[^2] -- `redis_url` → lazy-init Redis client - Scoring weights (`w_structural`, `w_geo`, etc.) → match the spec's config reference[^3] - Decay half-lives → match the temporal decay system in `engine-core-modules.py`[^1] @@ -39,7 +38,6 @@ status: active - `get_graph_driver()` → returns the shared async Neo4j driver - `get_domain_loader()` → returns the cached domain pack loader -- `get_redis()` → lazy Redis with graceful degradation - `startup()` / `shutdown()` → called from `create_app()` lifespan **`requirements.txt`** pins every dep from `pyproject.toml` with `>=X,"}. + L9_VERIFYING_KEYS_JSON: ${L9_VERIFYING_KEYS_JSON:?L9_VERIFYING_KEYS_JSON must be set when L9_REQUIRE_SIGNATURE is true} L9_ALLOWED_ACTIONS: match,sync,admin,outcomes,resolve,health,healthcheck L9_MAX_ATTACHMENTS: "0" L9_MAX_ATTACHMENT_SIZE_BYTES: "0" @@ -50,7 +59,6 @@ services: NEO4J_USERNAME: ${NEO4J_USERNAME:-neo4j} NEO4J_PASSWORD: ${NEO4J_PASSWORD:?NEO4J_PASSWORD must be set for production} NEO4J_POOL_SIZE: "50" - REDIS_URL: redis://redis:6379/0 API_SECRET_KEY: ${API_SECRET_KEY:?API_SECRET_KEY must be set for production} LOG_LEVEL: info CORS_ORIGINS: ${CORS_ORIGINS:-[]} @@ -59,11 +67,11 @@ services: KGE_CONFIDENCE_THRESHOLD: "0.3" KGE_EMBEDDING_DIM: "300" PARETO_ENABLED: "true" + # CEG-007: redis was a depends_on: service_healthy here for a dependency + # no code imports. depends_on: neo4j: condition: service_healthy - redis: - condition: service_healthy healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/v1/health')"] interval: 30s @@ -106,28 +114,9 @@ services: retries: 10 restart: always - # ── Redis 7.x ─────────────────────────────────────────── - redis: - image: redis:7-alpine - container_name: l9-redis-prod - ports: - - "6379:6379" - command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru - volumes: - - redis-data:/data - networks: - - l9-net - healthcheck: - test: ["CMD", "redis-cli", "ping"] - interval: 10s - timeout: 3s - retries: 5 - restart: always - volumes: neo4j-data: neo4j-logs: - redis-data: networks: l9-net: diff --git a/docker-compose.yml b/docker-compose.yml index 8f12b4b8..b6d1e80c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -12,7 +12,6 @@ # Usage: docker compose up -d # API: http://localhost:8000/v1/health # Neo4j: http://localhost:7474 (browser), bolt://localhost:7687 -# Redis: localhost:6379 # Postgres: localhost:5432 (PacketStore + ComplianceEngine audit-flush pool) services: @@ -39,7 +38,13 @@ services: GATE_ADMIN_TOKEN: ${GATE_ADMIN_TOKEN:-dev-gate-admin-token-not-for-production} GATE_REGISTRATION_ENABLED: "true" L9_NODE_NAME: graph - L9_NODE_SPEC_PATH: engine/spec.yaml + # CEG-003: the SDK reads GATE_NODE_SPEC_PATH (gate/config.py). This was + # L9_NODE_SPEC_PATH, which nothing consumes — registration worked only + # because the SDK's default happens to be engine/spec.yaml and WORKDIR is + # /app. Relocating the spec would have silently had no effect until the + # default path stopped existing, at which point register_with_gate + # swallows FileNotFoundError and returns False. + GATE_NODE_SPEC_PATH: engine/spec.yaml L9_ENVIRONMENT: local L9_SERVICE_NAME: graph-engine L9_SERVICE_VERSION: 1.1.0 @@ -50,6 +55,16 @@ services: L9_SIGNING_ALGORITHM: hmac-sha256 L9_SIGNING_KEY: dev-shared-hmac-secret-not-for-production L9_SIGNING_KEY_ID: graph-engine-v1 + # CEG-002: L9_REQUIRE_SIGNATURE also turns on verification of the + # responses Gate signs, and Gate signs with *its* key id — not CEG's. With + # no verifying-key map this file shipped a node that rejected every signed + # Gate response with "no verifying key available for transport signature + # verification". The E2E only passed because its harness supplied the map. + # HMAC verification uses the same shared secret as signing; under ed25519 + # these are Gate's and CEG's public keys instead. + L9_VERIFYING_KEYS_JSON: >- + {"gate-v1": "dev-shared-hmac-secret-not-for-production", + "graph-engine-v1": "dev-shared-hmac-secret-not-for-production"} # `enrich` deliberately absent: Gate owns that action name for EIE. L9_ALLOWED_ACTIONS: match,sync,admin,outcomes,resolve,health,healthcheck L9_MAX_ATTACHMENTS: "0" @@ -61,8 +76,6 @@ services: NEO4J_USERNAME: neo4j NEO4J_PASSWORD: l9-dev-password NEO4J_POOL_SIZE: "20" - # Redis - REDIS_URL: redis://redis:6379/0 # API API_KEY: dev-key-sha256-not-for-production LOG_LEVEL: debug @@ -88,11 +101,11 @@ services: # Hot-reload: mount source code so changes reflect without rebuild - ./engine:/app/engine:ro - ./domains:/app/domains:ro + # CEG-007: redis was a depends_on: service_healthy here, so the api service + # could not start without a Redis nothing imports. depends_on: neo4j: condition: service_healthy - redis: - condition: service_healthy postgres: condition: service_healthy healthcheck: @@ -135,24 +148,6 @@ services: networks: - l9-graph - # ── Redis 7.x ─────────────────────────────────────────── - redis: - image: redis:7-alpine - container_name: l9-graph-redis - ports: - - "6379:6379" - command: redis-server --maxmemory 128mb --maxmemory-policy allkeys-lru - volumes: - - redis_data:/data - healthcheck: - test: ["CMD", "redis-cli", "ping"] - interval: 5s - timeout: 3s - retries: 5 - restart: unless-stopped - networks: - - l9-graph - # ── PostgreSQL 16 (PacketStore + ComplianceEngine audit) ── postgres: image: postgres:16-alpine @@ -183,7 +178,6 @@ volumes: neo4j_data: neo4j_logs: neo4j_plugins: - redis_data: postgres_data: networks: diff --git a/docs/CI_PIPELINE.md b/docs/CI_PIPELINE.md index c0f37ce1..02b75b39 100644 --- a/docs/CI_PIPELINE.md +++ b/docs/CI_PIPELINE.md @@ -46,7 +46,6 @@ status: active test instead of stopping at the first one, while still exiting non-zero (failing the job) if any test fails - PostgreSQL service (postgres:16) -- Redis service (redis:7-alpine) ### Phase 4: Security Scanning - Gitleaks (secret detection) diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index c0f76dd2..f92cd909 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -39,7 +39,6 @@ The Graph Cognitive Engine is deployed on Hetzner Cloud. | **Health Check** | `http://178.104.43.11:8000/v1/health` | 8000 | | **Neo4j Browser** | `http://178.104.43.11:7474` | 7474 | | **Neo4j Bolt** | `bolt://178.104.43.11:7687` | 7687 | -| **Redis** | `redis://178.104.43.11:6379` | 6379 | ### SSH Access @@ -65,7 +64,6 @@ L9_LIFECYCLE_HOOK=engine.boot:GraphLifecycle NEO4J_URI=bolt://neo4j:7687 NEO4J_USERNAME=neo4j NEO4J_PASSWORD= -REDIS_URL=redis://redis:6379/0 API_PORT=8000 LOG_LEVEL=info CORS_ORIGINS=[] @@ -113,9 +111,9 @@ ssh root@178.104.43.11 "cd /opt/ceg && docker compose -f docker-compose.prod.yml │ l9-ceg (178.104.43.11) │ ├─────────────────────────────────────────────────────────────┤ │ │ -│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ -│ │ API │ │ Neo4j │ │ Redis │ │ -│ │ :8000 │ │ :7474/7687 │ │ :6379 │ │ +│ ┌─────────────┐ ┌─────────────┐ │ +│ │ API │ │ Neo4j │ │ +│ │ :8000 │ │ :7474/7687 │ │ │ │ │ │ │ │ │ │ │ │ FastAPI + │ │ Graph DB + │ │ Cache + │ │ │ │ Uvicorn │ │ GDS Plugin │ │ Sessions │ │ diff --git a/docs/FEATURE_GATES.md b/docs/FEATURE_GATES.md index 75effa8d..9adbf918 100644 --- a/docs/FEATURE_GATES.md +++ b/docs/FEATURE_GATES.md @@ -62,6 +62,10 @@ independently of the code default. | Capability Auth (domain-spec model) | `CAPABILITY_AUTH_ENABLED` | `True` | `True` | active | | PostgreSQL Audit Pool | `POSTGRES_DSN` | unset (`None`) | set | active (opt-in, soft dependency — see §7) | | Idea Portfolio Graph | `IDEA_PORTFOLIO_ENABLED` (`idea_portfolio_enabled`) | `False` | unset | dormant; opt-in IdeaOS portfolio reads/hydration | +| Graph Inference Feedback | `GRAPH_INFERENCE_FEEDBACK_ENABLED` (`graph_inference_feedback_enabled`) | `False` | unset | dormant; CEG → Gate → EIE `graph-inference-result` — see §13 | +| Domain Database Provisioning | `AUTO_CREATE_DOMAIN_DATABASE` (`auto_create_domain_database`) | `False` | unset | dormant; create the tenant domain database on first use — see §14 | +| Health API (admin health_* subactions) | `HEALTH_API_ENABLED` (`health_api_enabled`) | `False` | unset | dormant; AI-readiness assess/report surface — see §15 | +| Unvalidated Domain Packs | `UNVALIDATED_DOMAIN_PACKS_ENABLED` (`unvalidated_domain_packs_enabled`) | `False` | unset | dormant; five packs whose gates do not compile to executable Cypher — see §16 | | Constellation Orchestration | — | — | — | accepted architectural gap — see §9 | --- @@ -325,6 +329,152 @@ intact. --- +## 13. Graph Inference Feedback (EIE-008 / CEG-006) + +**State**: Dormant +**Flag**: `GRAPH_INFERENCE_FEEDBACK_ENABLED=False` (default off) + +Emits `graph-inference-result` to Enrichment.Inference.Engine through Gate. +EIE advertises that action to Gate and implements the whole consumer side — +packet validation, per-tenant queues, target extraction, a 0.55 confidence +floor, injection into the convergence loop — and nothing in CEG ever produced +the packet, so the loop had a consumer and no producer. Both sides were even +built to the same confidence floor. + +### Prerequisites + +- `GATE_URL` configured and Gate reachable; `graph-inference-result` is owned by + `eie` in Gate's `CANONICAL_ACTION_OWNERS`, so Gate resolves the destination. +- EIE registered with Gate advertising `graph-inference-result`. + +### Activation Steps + +1. Set `GRAPH_INFERENCE_FEEDBACK_ENABLED=true`. +2. Drive the `admin` subaction `emit_inference_feedback` with an `entity`, + an `entity_id`, and optionally a `rules` list (defaults to every registered + inference rule). + +### Validation + +The dispatch result carries `sent_outputs`: how many findings cleared the 0.55 +floor and were actually sent. `status: "skipped"` with +`no_outputs_above_confidence_floor` means nothing qualified and no packet was +sent — an empty `inference_outputs` list is valid to EIE and would cost a Gate +round trip to queue nothing. + +### Rollback + +Set the flag back to `False`. Each emission queues re-enrichment targets in EIE +and therefore spends EIE budget, which is why it ships off — the same reason as +`AUTO_ENRICH_VIA_GATE`. + +--- + +## 14. Domain Database Provisioning (CEG-008) + +**State**: Dormant +**Flag**: `AUTO_CREATE_DOMAIN_DATABASE=False` (default off) + +`match` and `sync` route queries to a Neo4j database named after the domain id. +Neo4j does not create databases implicitly, so on a fresh instance every sync +and match failed with an `ExecutionError` until an operator ran +`CREATE DATABASE` by hand. With this flag on, `GraphDriver` provisions the +database on first use — once per database per process. + +### Prerequisites + +- **Neo4j Enterprise Edition.** `CREATE DATABASE` is an Enterprise + administrative command; Community Edition rejects it. +- Credentials with database administration privileges. + +### Activation Steps + +1. Set `AUTO_CREATE_DOMAIN_DATABASE=true`. +2. No restart of Neo4j is required; the next query against an unprovisioned + domain creates it. + +### Validation + +`Ensured Neo4j database '' exists` is logged on the provisioning call. +A refusal (Community Edition, or missing privilege) is logged as a warning and +does **not** raise: a deployment whose database already exists is never blocked +by a CREATE it is not allowed to run. + +### Rollback + +Set the flag back to `False`. Provisioning stops; a query against an absent +database then raises `DatabaseNotProvisionedError`, which names the missing +database and the exact `CREATE DATABASE` command. That message is present +whether or not the flag is on. + +--- + +## 15. Health API (CEG-006) + +**State**: Dormant +**Flag**: `HEALTH_API_ENABLED=False` (default off) + +Exposes `engine/health/api.py` through the `admin` subactions `health_assess`, +`health_batch_assess` and `health_report`. That module implemented three +handlers and was imported by nothing, so `request_enrichment` — the whole +CEG → Gate → EIE direction — had no trigger an inbound packet could reach. + +`AUTO_ENRICH_VIA_GATE` is not a substitute: it gates only the eventual outbound +Gate request, not assessment, reporting, or conversion-event tracking. The +surface needs a gate of its own, which is what this is. + +### Validation + +With the flag off the subactions return `{"status": "disabled"}` rather than +404-ing, so an operator can tell "switched off" from "not a subaction". + +### Rollback + +Set back to `False`. Note that assess/report append to the conversion-event +store in `engine/health/health_report.py`, which is bounded +(`CONVERSION_EVENT_MAX`) and discards oldest-first. + +--- + +## 16. Unvalidated Domain Packs (CEG-009) + +**State**: Dormant +**Flag**: `UNVALIDATED_DOMAIN_PACKS_ENABLED=False` (default off) + +CEG-009 moved nine domain packs out of a flat `_domain_spec.yaml` shape +the loader never reads into `domains//spec.yaml`. Making them readable +exposed that five do not compile to executable Cypher: + +| Pack | Defect | +|---|---| +| `executive-assistant` | `RELATES_TO` fallback + `$85.0` | +| `repo-as-agent` | `RELATES_TO` fallback + `$5` | +| `roofing-company` | `RELATES_TO` fallback + `$1` | +| `aios-god-agent` | `RELATES_TO` fallback | +| `healthcare-referral` | `$1` | + +Two root causes, both in the spec-to-Cypher contract rather than in the files: + +1. `type: traversal` gates written with `pattern` and `condition`, which + `GateCompiler` does not consume. With no `edgetype` the compiler falls back + to `RELATES_TO`, an edge no ontology here declares, so the gate is a hard + filter that matches nothing. +2. A scalar `queryparam` (`85.0`, `5`, `1`). `GateSpec.coerce_queryparam_to_str` + turns it into a string and the compiler emits it as a parameter *name*, so + Cypher receives `$85.0`. + +**Do not turn this on to "see if they work".** They do not: the first serves +zero candidates, the second fails at execution. Fixing them means either +compiler support for `pattern`/`condition` and literal operands, or a declared +query-schema parameter per constant — a schema decision, not a file move. + +`tests/unit/test_domain_pack_shape.py` holds both ends: a discoverable pack +must compile to executable Cypher, and a gated pack that starts compiling +cleanly must be removed from `_DOMAIN_FEATURE_FLAGS`, so this list can only +shrink. + +--- + ## Querying Feature Status Use the `feature_status` admin subaction to get current state of all gates: diff --git a/docs/contracts/config/env-contract.yaml b/docs/contracts/config/env-contract.yaml index 477d775c..20d2581a 100644 --- a/docs/contracts/config/env-contract.yaml +++ b/docs/contracts/config/env-contract.yaml @@ -94,17 +94,12 @@ variables: source: env sensitive: false - # ── Redis ───────────────────────────────────────────────── - - name: REDIS_URL - type: url - required: true - default: "redis://localhost:6379/0" - description: | - Redis connection URL. In Docker Compose: redis://redis:6379/0. - Used for scoring result caching and GDS output caching. - source: env - sensitive: false - example: "redis://redis:6379/0" + # CEG-007: REDIS_URL was declared here as required: true, described as backing + # "scoring result caching and GDS output caching". No such caching exists — + # nothing under engine/ or chassis/ imports redis at all. The variable, the + # dependency, the compose services and docs/contracts/dependencies/redis.yaml + # are removed together rather than leaving a contract that describes behaviour + # the code never had. # ── API Surface ─────────────────────────────────────────── - name: API_KEY diff --git a/docs/contracts/dependencies/redis.yaml b/docs/contracts/dependencies/redis.yaml deleted file mode 100644 index 31b25750..00000000 --- a/docs/contracts/dependencies/redis.yaml +++ /dev/null @@ -1,38 +0,0 @@ -# ═══════════════════════════════════════════════════════════════ -# Contract: Redis 7 Dependency -# Source: docker-compose.yml (redis service) -# engine/config/settings.py:redis_url -# Version: 1.0.0 -# Updated: 2026-04-26 -# ═══════════════════════════════════════════════════════════════ - -service_name: redis -type: external -protocol: redis -version: "7-alpine" - -connection: - base_url_env: REDIS_URL - default_url: "redis://redis:6379/0" - auth_method: none - -usage: - - purpose: Scoring result caching - description: GDS and scoring dimension results are cached to avoid recomputation. - - purpose: Domain pack cache - description: Async domain spec cache with TTL (W4-03). TTL=DOMAIN_CACHE_TTL_SECONDS. - -docker_compose: - image: redis:7-alpine - port: "6379:6379" - command: "redis-server --maxmemory 128mb --maxmemory-policy allkeys-lru" - eviction_policy: allkeys-lru - max_memory: 128mb - -health_check: - command: "redis-cli ping" - -fallback: | - No explicit Redis fallback/circuit breaker implemented. - Engine will degrade if Redis is unavailable — scoring will proceed - without cache. diff --git a/domains/aios_god_agent_domain_spec.yaml b/domains/aios-god-agent/spec.yaml similarity index 100% rename from domains/aios_god_agent_domain_spec.yaml rename to domains/aios-god-agent/spec.yaml diff --git a/domains/executive_assistant_domain_spec.yaml b/domains/executive-assistant/spec.yaml similarity index 88% rename from domains/executive_assistant_domain_spec.yaml rename to domains/executive-assistant/spec.yaml index 3135e784..f68b5441 100644 --- a/domains/executive_assistant_domain_spec.yaml +++ b/domains/executive-assistant/spec.yaml @@ -53,6 +53,18 @@ ontology: - {name: category, type: enum} - {name: steps, type: string} + # CEG-009: HAS_SKILL below targets Skill, which was never declared, so this + # spec failed the ontology cross-reference validator and could not load at + # all. It went unnoticed because the file was also in the flat + # _domain_spec.yaml shape the loader never reads. + - label: Skill + managedby: static + auxiliary: true + properties: + - {name: skillid, type: string, required: true} + - {name: name, type: string} + - {name: category, type: enum} + edges: - {type: HAS_SKILL, from: Expert, to: Skill, direction: DIRECTED, category: capability, managedby: sync} - {type: COMPLETED_TASK, from: Expert, to: Task, direction: DIRECTED, category: transaction, managedby: api} diff --git a/domains/freight_matching_domain_spec.yaml b/domains/freight-matching/spec.yaml similarity index 100% rename from domains/freight_matching_domain_spec.yaml rename to domains/freight-matching/spec.yaml diff --git a/domains/healthcare_referral_domain_spec.yaml b/domains/healthcare-referral/spec.yaml similarity index 100% rename from domains/healthcare_referral_domain_spec.yaml rename to domains/healthcare-referral/spec.yaml diff --git a/domains/legal_discovery_domain_spec.yaml b/domains/legal-discovery/spec.yaml similarity index 100% rename from domains/legal_discovery_domain_spec.yaml rename to domains/legal-discovery/spec.yaml diff --git a/domains/mortgage_brokerage_domain_spec.yaml b/domains/mortgage-brokerage/spec.yaml similarity index 100% rename from domains/mortgage_brokerage_domain_spec.yaml rename to domains/mortgage-brokerage/spec.yaml diff --git a/domains/repo_as_agent_domain_spec.yaml b/domains/repo-as-agent/spec.yaml similarity index 100% rename from domains/repo_as_agent_domain_spec.yaml rename to domains/repo-as-agent/spec.yaml diff --git a/domains/research_agent_domain_spec.yaml b/domains/research-agent/spec.yaml similarity index 100% rename from domains/research_agent_domain_spec.yaml rename to domains/research-agent/spec.yaml diff --git a/domains/roofing_company_domain_spec.yaml b/domains/roofing-company/spec.yaml similarity index 87% rename from domains/roofing_company_domain_spec.yaml rename to domains/roofing-company/spec.yaml index 0f970460..35f75eb9 100644 --- a/domains/roofing_company_domain_spec.yaml +++ b/domains/roofing-company/spec.yaml @@ -57,6 +57,19 @@ ontology: - {name: priceperunit, type: float, unit: currency} - {name: warrantyears, type: int} + # CEG-009: SERVICES_ZIP below targets ZipCode, which was never declared, so + # this spec failed the ontology cross-reference validator and could not load + # at all. Hidden by the flat _domain_spec.yaml shape the loader never + # reads. Contractor.servicezips carries the same coverage as a raw string; + # the node is what the edge needs to resolve against. + - label: ZipCode + managedby: static + auxiliary: true + properties: + - {name: zipcode, type: string, required: true} + - {name: city, type: string} + - {name: state, type: string} + edges: - {type: SERVICES_ZIP, from: Contractor, to: ZipCode, direction: DIRECTED, category: capability, managedby: sync} - {type: USES_MATERIAL, from: Contractor, to: Material, direction: DIRECTED, category: capability, managedby: sync} diff --git a/engine/config/loader.py b/engine/config/loader.py index cec63d27..dd063778 100644 --- a/engine/config/loader.py +++ b/engine/config/loader.py @@ -32,7 +32,30 @@ MAX_SPEC_BYTES = 5 * 1024 * 1024 SPEC_FILENAME = "spec.yaml" -_DOMAIN_FEATURE_FLAGS = {"idea-portfolio": "idea_portfolio_enabled"} +_DOMAIN_FEATURE_FLAGS = { + "idea-portfolio": "idea_portfolio_enabled", + # CEG-009 moved nine packs into the shape this loader reads. Five of them + # carry gates that compile to Cypher that cannot execute, which was + # invisible while the loader never opened them: + # + # * `type: traversal` gates written with `pattern` + `condition`, which + # GateCompiler does not consume. With no `edgetype` they fall back to + # `RELATES_TO`, an edge no ontology here declares, so the gate rejects + # every candidate. + # * a scalar `queryparam` (85.0, 5, 1) — GateSpec coerces it to a string + # and the compiler emits it as a parameter NAME, producing `$85.0`. + # + # Making a pack readable is not the same as making it correct. Reaching + # these needs either compiler support for pattern/condition and literal + # operands, or a query-schema parameter per constant — a schema decision, + # not a file move. Dormant until then, guarded by + # tests/unit/test_domain_pack_shape.py so the set cannot grow silently. + "executive-assistant": "unvalidated_domain_packs_enabled", + "aios-god-agent": "unvalidated_domain_packs_enabled", + "repo-as-agent": "unvalidated_domain_packs_enabled", + "roofing-company": "unvalidated_domain_packs_enabled", + "healthcare-referral": "unvalidated_domain_packs_enabled", +} class DomainNotFoundError(Exception): diff --git a/engine/config/settings.py b/engine/config/settings.py index efb9cd16..46664855 100644 --- a/engine/config/settings.py +++ b/engine/config/settings.py @@ -56,8 +56,12 @@ class Settings(BaseSettings): neo4j_max_connection_lifetime: int = 3600 neo4j_connection_acquisition_timeout: int = 60 - # --- Redis --- - redis_url: str = "redis://localhost:6379/0" + # CEG-007: there was a `redis_url` here, and a redis service in both compose + # files, and a redis dependency in requirements — and no `import redis` + # anywhere under engine/ or chassis/. The declared contract claimed scoring + # and domain-pack caching that was never built. Operators could not start the + # api service without Redis running, for a dependency nothing used. All of it + # is removed; if shared state is wanted later, add it with the code that uses it. # --- API --- api_port: int = 8000 @@ -144,6 +148,29 @@ class Settings(BaseSettings): # Distinct from PACKET_STORE_DSN (engine/packet/packet_store.py), which manages # its own lazy pool — both may point at the same Postgres instance. + # CEG-008: `match` and `sync` route queries to a Neo4j database named after + # the domain id, and nothing ever created it — Neo4j does not create + # databases implicitly, so on a fresh instance every sync and match failed + # with an ExecutionError until an operator ran CREATE DATABASE by hand. + # With this on, GraphDriver provisions the database on first use (once per + # database per process). Off by default: CREATE DATABASE is an Enterprise + # Edition administrative command requiring privileges a read-only + # deployment is not expected to hold. Either way the failure now names the + # missing database and the command that provides it. + auto_create_domain_database: bool = False + # CEG-006: the `health_*` admin subactions make engine/health/api.py + # reachable. auto_enrich_via_gate gates only the eventual outbound Gate + # request, not assessment, reporting or conversion tracking, so the surface + # itself needs its own gate. Mechanism ships dormant; operator activates. + health_api_enabled: bool = False + # CEG-009: five migrated domain packs compile to Cypher that cannot execute + # (a `RELATES_TO` fallback for gates written with `pattern`/`condition`, and + # scalar `queryparam` values emitted as parameter names such as `$85.0`). + # They are readable now but not correct, so they stay undiscoverable until + # the compiler or the query schema grows the support they assume. Turning + # this on serves packs whose gates are known not to execute. + unvalidated_domain_packs_enabled: bool = False + # --- Wave 7: Explicit Tenant Database Binding --- strict_tenant_database: bool = ( False # W7-01: require explicit database= on GraphDriver calls; no implicit 'neo4j' fallback @@ -155,6 +182,12 @@ class Settings(BaseSettings): # Seam audit / PR remediation: paid-tier enrich_now Gate dispatch is opt-in. # Default off so deploy does not immediately spend EIE budget until enabled. auto_enrich_via_gate: bool = False + # EIE-008 / CEG-006: emit `graph-inference-result` to EIE through Gate. EIE + # advertises and implements the consumer side end to end; nothing here ever + # produced the packet, so the feedback loop had no producer. Off by default + # for the same reason as auto_enrich_via_gate — each emission queues + # re-enrichment targets and so spends EIE budget. + graph_inference_feedback_enabled: bool = False # IdeaOS portfolio graph is a new behavioral surface. Keep both corpus writes # and portfolio-context reads dormant until explicitly activated by an operator. idea_portfolio_enabled: bool = False diff --git a/engine/gate_egress.py b/engine/gate_egress.py index 7d7cbbc9..586b9d07 100644 --- a/engine/gate_egress.py +++ b/engine/gate_egress.py @@ -11,10 +11,23 @@ engine/gate_egress.py — the only CEG -> peer egress: CEG -> Gate -> EIE. -CEG never addresses the enrichment node. It asks Gate to run the `enrich` -action (owned by Enrichment.Inference.Engine in Gate's ownership map) and -receives Gate's response packet. The SDK owns packet construction, signing, -the single HTTP attempt, and the deadline derived from ``timeout_ms``. +CEG never addresses the enrichment node. It asks Gate to run an EIE-owned +action (``enrich``, ``graph-inference-result``; see Gate's +CANONICAL_ACTION_OWNERS) and receives Gate's response packet. The SDK owns +packet construction, signing, the single HTTP attempt, and the deadline derived +from ``timeout_ms``. + +Two directions live here: + +* ``request_enrichment`` — CEG asks EIE to research fields it is missing. +* ``emit_graph_inference_result`` — CEG hands EIE values it derived from the + graph, which EIE's ``GraphReturnChannel`` feeds into the convergence loop. + +The second closes a loop that was open in one direction only (EIE-008 / +CEG-006): EIE advertised ``graph-inference-result`` to Gate and implemented the +whole consumer side — validation, per-tenant queues, target extraction — while +no code in CEG ever constructed such a packet. The two halves were even built +to the same 0.55 confidence floor and never connected. Fail-closed rules (seam audit 2026-09-02): * no GATE_URL -> ``gate_not_configured``; there is no direct fallback; @@ -38,9 +51,18 @@ logger = logging.getLogger(__name__) ENRICH_ACTION = "enrich" +GRAPH_INFERENCE_ACTION = "graph-inference-result" DEFAULT_ENRICH_TIMEOUT_MS = 25_000 +DEFAULT_INFERENCE_TIMEOUT_MS = 10_000 _SEAM_TAGS: tuple[str, ...] = ("INTER_NODE",) +# EIE drops any inference output below this confidence +# (app/services/graph_return_channel.py CONFIDENCE_FLOOR). CEG's own rule +# registry suppresses below the same number (InferenceContext.confidence_floor). +# Filtering here too means a packet never carries outputs the receiver will +# silently discard, so "sent 6, queued 2" is visible at the sender. +INFERENCE_CONFIDENCE_FLOOR = 0.55 + def build_enrichment_request( *, @@ -112,6 +134,9 @@ async def request_enrichment( ) try: + # See emit_graph_inference_result: get_gate_client() raises ValueError on + # missing or invalid SDK environment material, which is a configuration + # failure this module reports as a typed result, never one it propagates. client = get_gate_client() response = await client.execute( action=ENRICH_ACTION, @@ -122,7 +147,7 @@ async def request_enrichment( correlation_id=correlation_id, compliance_tags=_SEAM_TAGS, ) - except GateClientError as exc: + except (GateClientError, ValueError) as exc: logger.warning("gate_egress: %s for entity=%s tenant=%s: %s", type(exc).__name__, entity_id, tenant, exc) return { "status": "failed", @@ -144,10 +169,158 @@ async def request_enrichment( } +def inference_idempotency_key(tenant: str, entity_id: str, outputs: Sequence[dict[str, Any]]) -> str: + """Stable over the same findings, so a retry cannot double-queue targets.""" + parts = sorted(f"{o.get('field')}={o.get('value')!r}@{o.get('confidence')}" for o in outputs) + digest = hashlib.sha256("|".join([tenant, entity_id, *parts]).encode("utf-8")).hexdigest() + return f"ceg:graph-inference:{tenant}:{entity_id}:{digest[:16]}" + + +def build_inference_outputs( + entity_id: str, + results: Sequence[Any], +) -> list[dict[str, Any]]: + """Shape CEG inference results into EIE's ``inference_outputs`` elements. + + Accepts ``InferenceResult`` (engine.inference_rule_registry) or a mapping + already in that shape. EIE requires ``entity_id``, ``field``, ``value``, + ``confidence`` and ``rule`` per element; ``InferenceResult.to_dict()`` + already produces all but ``entity_id``. + + Outputs below ``INFERENCE_CONFIDENCE_FLOOR`` are dropped here rather than + sent to be dropped there. + """ + outputs: list[dict[str, Any]] = [] + for result in results: + raw = result.to_dict() if hasattr(result, "to_dict") else dict(result) + field_name = raw.get("field") + if not field_name: + logger.warning("gate_egress: inference output without a field name — dropped") + continue + try: + confidence = float(raw.get("confidence", 0.0)) + except (TypeError, ValueError): + logger.warning("gate_egress: non-numeric confidence for field=%s — dropped", field_name) + continue + if confidence < INFERENCE_CONFIDENCE_FLOOR: + logger.debug( + "gate_egress: confidence %.3f below floor %.3f for field=%s — not sent", + confidence, + INFERENCE_CONFIDENCE_FLOOR, + field_name, + ) + continue + outputs.append( + { + "entity_id": entity_id, + "field": str(field_name), + "value": raw.get("value"), + "confidence": confidence, + "rule": str(raw.get("rule", "unknown")), + "provenance": str(raw.get("provenance", "inference")), + "rationale": str(raw.get("rationale", "")), + } + ) + return outputs + + +async def emit_graph_inference_result( + *, + tenant: str, + entity_id: str, + results: Sequence[Any], + timeout_ms: int = DEFAULT_INFERENCE_TIMEOUT_MS, + correlation_id: str | None = None, +) -> dict[str, Any]: + """Hand graph-derived field values to EIE via Gate. One attempt, fail closed. + + EIE's ``GraphReturnChannel`` converts each output into an ``EnrichmentTarget`` + and feeds the convergence loop, so this is the producer for the action EIE + has advertised and implemented all along (EIE-008). + + Returns a result dict shaped like ``request_enrichment``'s, with + ``sent_outputs`` so a caller can see how many findings survived the floor. + A call with nothing above the floor is reported as ``skipped`` and sends no + packet — an empty ``inference_outputs`` list is valid to EIE and would cost + a Gate round trip to queue nothing. + """ + outputs = build_inference_outputs(entity_id, results) + key = inference_idempotency_key(tenant, entity_id, outputs) + + if not outputs: + return { + "status": "skipped", + "error": "no_outputs_above_confidence_floor", + "action": GRAPH_INFERENCE_ACTION, + "idempotency_key": key, + "sent_outputs": 0, + } + + if not os.environ.get("GATE_URL", "").strip(): + logger.warning("gate_egress: GATE_URL unset — inference result for %s not sent", entity_id) + return { + "status": "failed", + "error": "gate_not_configured", + "action": GRAPH_INFERENCE_ACTION, + "idempotency_key": key, + "sent_outputs": 0, + } + + try: + # ValueError as well as GateClientError: get_gate_client() builds the SDK + # config from the environment and raises ValueError on missing or invalid + # material. Letting that escape would crash the admin subaction instead of + # returning the typed failure every other path in this module returns. + client = get_gate_client() + response = await client.execute( + action=GRAPH_INFERENCE_ACTION, + payload={"inference_outputs": outputs}, + tenant=tenant, + idempotency_key=key, + timeout_ms=timeout_ms, + correlation_id=correlation_id, + compliance_tags=_SEAM_TAGS, + ) + except (GateClientError, ValueError) as exc: + logger.warning( + "gate_egress: %s for inference entity=%s tenant=%s: %s", + type(exc).__name__, + entity_id, + tenant, + exc, + ) + return { + "status": "failed", + "error": type(exc).__name__, + "detail": str(exc), + "action": GRAPH_INFERENCE_ACTION, + "idempotency_key": key, + "sent_outputs": 0, + } + + failed = response.header.packet_type == "failure" + return { + "status": "failed" if failed else "ok", + "action": GRAPH_INFERENCE_ACTION, + "idempotency_key": key, + "sent_outputs": len(outputs), + "packet_id": str(response.header.packet_id), + "packet_type": response.header.packet_type, + "correlation_id": str(response.header.correlation_id) if response.header.correlation_id else None, + "payload": dict(response.payload), + } + + __all__ = [ "DEFAULT_ENRICH_TIMEOUT_MS", + "DEFAULT_INFERENCE_TIMEOUT_MS", "ENRICH_ACTION", + "GRAPH_INFERENCE_ACTION", + "INFERENCE_CONFIDENCE_FLOOR", "build_enrichment_request", + "build_inference_outputs", + "emit_graph_inference_result", "enrichment_idempotency_key", + "inference_idempotency_key", "request_enrichment", ] diff --git a/engine/graph/driver.py b/engine/graph/driver.py index 973c5b78..d93e8fa6 100644 --- a/engine/graph/driver.py +++ b/engine/graph/driver.py @@ -16,6 +16,7 @@ import asyncio import logging import os +import re from typing import Any from neo4j import AsyncDriver, AsyncGraphDatabase @@ -24,6 +25,42 @@ logger = logging.getLogger(__name__) +# Databases the DBMS always provides; never candidates for provisioning. +_BUILTIN_DATABASES = frozenset({"neo4j", "system"}) + +# Neo4j database naming rules: begins with an ASCII letter, then letters, +# digits, dots, dashes or underscores, 3-63 characters. CREATE DATABASE cannot +# take the name as a query parameter (it is an administrative command, not a +# read/write query), so the name is quoted into the statement — and therefore +# must be validated first. Domain ids legitimately contain dashes +# ("healthcare-referral"), which is why engine.utils.security.sanitize_label +# does not apply here: its label grammar forbids them. +_DATABASE_NAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._-]{2,62}$") + +# Substrings Neo4j uses when the target database is absent. Matched case +# insensitively against the driver's message. +_DATABASE_ABSENT_MARKERS = ( + "database does not exist", + "databasenotfound", + "unable to get a routing table for database", +) + + +class DatabaseNotProvisionedError(RuntimeError): + """A query named a database the DBMS does not have. + + CEG-008: `match` and `sync` route to a database named after the domain id, + and nothing created it. Neo4j does not create databases implicitly, so on a + fresh instance every sync and match failed with the driver's own message + until an operator happened to run CREATE DATABASE by hand. This names the + missing database and the command that provides it. + """ + + +def _looks_like_absent_database(exc: Exception) -> bool: + message = str(exc).casefold() + return any(marker in message for marker in _DATABASE_ABSENT_MARKERS) + class GraphDriver: """Neo4j async driver manager.""" @@ -48,6 +85,9 @@ def __init__( self._driver: AsyncDriver | None = None self._lock = asyncio.Lock() + # Databases this process has already provisioned or confirmed, so the + # CREATE is attempted at most once per database per process. + self._ensured_databases: set[str] = set() # W4-02: Circuit breaker — configured via settings, defaults provided from engine.config.settings import settings @@ -131,7 +171,78 @@ async def execute_query( ) raise ValueError(msg) database = "neo4j" - return await self._circuit_breaker.call(self._raw_execute_query, cypher, parameters, database) + + # CEG-008: a tenant domain database has to exist before it can be + # queried. Under the flag we provision it on first use; without it we at + # least say what is missing instead of surfacing the driver's message. + from engine.config.settings import settings as _db_settings + + if _db_settings.auto_create_domain_database: + await self.ensure_database(database) + + try: + return await self._circuit_breaker.call(self._raw_execute_query, cypher, parameters, database) + except Exception as exc: + if _looks_like_absent_database(exc) and database not in _BUILTIN_DATABASES: + msg = ( + f"Neo4j database {database!r} does not exist. Domain queries route to a " + f"database named after the domain id, and Neo4j does not create one " + f"implicitly. Run this against the system database (Enterprise " + f"Edition): CREATE DATABASE `{database}` IF NOT EXISTS WAIT " + f"-- or set AUTO_CREATE_DOMAIN_DATABASE=true to have the engine " + f"create it on first use." + ) + raise DatabaseNotProvisionedError(msg) from exc + raise + + async def ensure_database(self, name: str) -> bool: + """Create the domain database if it is absent. Idempotent per process. + + Returns True when the database is known to exist afterwards, False when + provisioning was not possible (Community Edition, or insufficient + privileges) — in which case the query that follows fails with the + message above rather than here, so a read-only deployment that has the + database already is not blocked by a CREATE it is not allowed to run. + """ + if name in _BUILTIN_DATABASES or name in self._ensured_databases: + return True + if not _DATABASE_NAME_RE.fullmatch(name): + msg = ( + f"refusing to provision Neo4j database {name!r}: a database name must begin " + f"with a letter and contain only letters, digits, dots, dashes or underscores " + f"(3-63 characters). CREATE DATABASE takes no query parameter, so an " + f"unvalidated name would be interpolated into an administrative command." + ) + raise ValueError(msg) + + # Claim the name BEFORE the await, not after. `_raw_execute_query` is a + # suspension point: with the add afterwards, every request that arrived + # while the first CREATE was in flight passed the membership check above + # and issued its own administrative command — "at most once per database + # per process" held only when calls did not overlap, which is exactly + # when it does not matter. The claim is released on failure so a later + # attempt (different privileges, Enterprise now licensed) can retry. + self._ensured_databases.add(name) + + # Administrative commands must run against `system`, and the name is + # back-quoted because a dash is legal in a database name but not in a + # bare identifier. The regex above is what makes that quoting safe. + cypher = f"CREATE DATABASE `{name}` IF NOT EXISTS WAIT" + try: + await self._raw_execute_query(cypher, None, "system") + except Exception as exc: + self._ensured_databases.discard(name) + logger.warning( + "Could not provision Neo4j database %r (%s: %s). " + "CREATE DATABASE is Enterprise Edition only and requires admin privileges.", + name, + type(exc).__name__, + exc, + ) + return False + + logger.info("Ensured Neo4j database %r exists", name) + return True async def _raw_execute_write( self, diff --git a/engine/handlers.py b/engine/handlers.py index fb1b426d..3a1842a8 100644 --- a/engine/handlers.py +++ b/engine/handlers.py @@ -1197,6 +1197,12 @@ async def handle_admin(tenant: str, payload: dict[str, Any]) -> dict[str, Any]: "gdpr_erasure_enabled": _fs.gdpr_erasure_enabled, "gdpr_dry_run": _fs.gdpr_dry_run, "gds_max_staleness_hours": _fs.gds_max_staleness_hours, + # --- Constellation seam (EIE <-> Gate <-> CEG) --- + "auto_enrich_via_gate": _fs.auto_enrich_via_gate, + "graph_inference_feedback_enabled": _fs.graph_inference_feedback_enabled, + "health_api_enabled": _fs.health_api_enabled, + "auto_create_domain_database": _fs.auto_create_domain_database, + "idea_portfolio_enabled": _fs.idea_portfolio_enabled, }, } @@ -1424,6 +1430,99 @@ async def handle_admin(tenant: str, payload: dict[str, Any]) -> dict[str, Any]: "outcome_history_size": len(outcome_history), } + # ── CEG-006: AI-readiness health, reachable from an inbound packet ─────── + # engine/health/api.py implemented three handlers and was imported by + # nothing, so `request_enrichment` — the whole CEG -> Gate -> EIE direction + # — had no trigger any packet could reach. It hangs off `admin` rather than + # a new advertised action for the same reason `trigger_gds` does: these are + # operator surfaces, not collaboration routes Gate load-balances. + if subaction in {"health_assess", "health_batch_assess", "health_report"}: + from engine.config.settings import settings as _health_settings + + # Mechanism ships dormant, operator activates — the same contract every + # other behavioural surface here follows. auto_enrich_via_gate gates only + # the eventual outbound request, not assessment, reporting or conversion + # tracking, so making these reachable needs a gate of their own. + if not _health_settings.health_api_enabled: + return { + "status": "disabled", + "subaction": subaction, + "message": "Health API is disabled. Set HEALTH_API_ENABLED=True.", + } + + from engine.health import api as health_api + + health_handlers = { + "health_assess": health_api.handle_health_assess, + "health_batch_assess": health_api.handle_health_batch_assess, + "health_report": health_api.handle_health_report, + } + return await health_handlers[subaction](tenant, payload) + + # ── EIE-008: hand graph-derived field values back to EIE through Gate ──── + # EIE advertises and implements `graph-inference-result` end to end and no + # code here ever produced one, so the feedback loop the architecture implies + # had a consumer and no producer. Flag-gated: it spends EIE budget. + if subaction == "emit_inference_feedback": + from engine.config.settings import settings as _inf_settings + + if not _inf_settings.graph_inference_feedback_enabled: + return { + "status": "disabled", + "message": ("Graph inference feedback is disabled. Set GRAPH_INFERENCE_FEEDBACK_ENABLED=True."), + } + + from engine.gate_egress import emit_graph_inference_result + from engine.inference_rule_registry import ( + InferenceContext, + execute_rule, + list_registered_rules, + ) + + entity = _require_key(payload, "entity", "admin", tenant) + entity_id = _require_key(payload, "entity_id", "admin", tenant) + domain_id = payload.get("domain_id", tenant) + + # `rules` absent means "every registered rule"; `rules: []` means "none" + # and must not silently become "all". A non-list is rejected rather than + # iterated — a bare string would otherwise be walked character by + # character and every char looked up as a rule name. + if "rules" in payload: + requested = payload["rules"] + if not isinstance(requested, list) or not all(isinstance(r, str) for r in requested): + raise ValidationError( + "admin.rules must be a list of rule-name strings", + action="admin", + tenant=tenant, + ) + else: + requested = list_registered_rules() + + context = InferenceContext( + tenant_id=tenant, + domain_id=domain_id, + pass_number=int(payload.get("pass_number", 1)), + known_fields=dict(entity), + ) + # Named `inferred`, not `result`: `result` is already bound to a + # dict[str, Any] earlier in this function, and reusing it here is a type + # error mypy catches (and did). + results = [ + inferred for rule_name in requested if (inferred := execute_rule(rule_name, entity, context)) is not None + ] + dispatch = await emit_graph_inference_result( + tenant=tenant, + entity_id=str(entity_id), + results=results, + ) + return { + "status": "inference_feedback_emitted", + "entity_id": entity_id, + "rules_run": list(requested), + "results_produced": len(results), + "dispatch": dispatch, + } + raise ValidationError(f"Unknown admin subaction: {subaction!r}", action="admin", tenant=tenant) diff --git a/engine/health/health_report.py b/engine/health/health_report.py index 94e4f3e0..cdcde7c3 100644 --- a/engine/health/health_report.py +++ b/engine/health/health_report.py @@ -15,6 +15,7 @@ from __future__ import annotations import logging +from collections import deque from datetime import UTC, datetime from typing import Any @@ -29,8 +30,17 @@ logger = logging.getLogger(__name__) -# In-memory conversion event store (replace with persistent store in production) -_conversion_events: list[ConversionEvent] = [] +# In-memory conversion event store (replace with persistent store in production). +# +# Bounded, deliberately. CEG-006 made the `health_*` admin subactions reachable, +# and every Seed-tier assess/report call appends a tenant- and entity-bearing +# record here. An unbounded list in a long-lived process is a memory leak that +# grows with traffic, and CLAUDE.md's "never create unbounded caches" covers +# exactly this. A deque discards oldest-first at the ceiling, so funnel analysis +# keeps a recent window rather than the whole history — the same trade the +# "replace with persistent store" note above already anticipates. +CONVERSION_EVENT_MAX = 10_000 +_conversion_events: deque[ConversionEvent] = deque(maxlen=CONVERSION_EVENT_MAX) def generate_health_report( @@ -166,7 +176,7 @@ def analyze_conversion_funnel( tenant: str | None = None, ) -> ConversionFunnelMetrics: """Analyze conversion funnel metrics, optionally filtered by tenant.""" - events = _conversion_events + events: list[ConversionEvent] = list(_conversion_events) if tenant: events = [e for e in events if e.tenant == tenant] diff --git a/engine/spec.yaml b/engine/spec.yaml index d58d5897..676eb1cf 100644 --- a/engine/spec.yaml +++ b/engine/spec.yaml @@ -15,6 +15,14 @@ node: id: graph version: "1.1.0" + # CEG-005: declare the semantic owner instead of letting Gate infer it. + # The SDK flattens node.owner into metadata.owner, and Gate's registration + # ownership assertion is fail-closed for canonical actions. With no owner + # here, registration succeeded only because Gate's _OWNER_ALIASES maps the + # node name "graph" to "ceg" — correct today, and a name-based alias rather + # than a declaration. Renaming the node would have broken registration for + # every canonical action at once. EIE declares owner=eie for the same reason. + owner: ceg type: engine internal_url: "http://graph:8000" health_endpoint: /v1/health diff --git a/poetry.lock b/poetry.lock index d606da32..9c697805 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.3.2 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.3.3 and should not be changed by hand. [[package]] name = "annotated-doc" @@ -2036,26 +2036,6 @@ files = [ {file = "pyyaml-6.0.3.tar.gz", hash = "sha256:d76623373421df22fb4cf8817020cbb7ef15c725b9d5e45f17e189bfc384190f"}, ] -[[package]] -name = "redis" -version = "8.1.0" -description = "Python client for Redis database and key-value store" -optional = false -python-versions = ">=3.10" -groups = ["main"] -files = [ - {file = "redis-8.1.0-py3-none-any.whl", hash = "sha256:a4fe1aac3d3b3cc791d4b3d5931c5a956045dc951ee74d1c913ee3ac4d2ee9fb"}, - {file = "redis-8.1.0.tar.gz", hash = "sha256:6e1a19beef9225c83efd689c7e6b7da2d5215b1f42cd13b7fc3714d0a09c7b25"}, -] - -[package.extras] -circuit-breaker = ["pybreaker (>=1.4.0)"] -hiredis = ["hiredis (>=3.2.0)"] -jwt = ["pyjwt (>=2.13.0)"] -ocsp = ["cryptography (>=36.0.1)", "pyopenssl (>=20.0.1)", "requests (>=2.31.0)"] -otel = ["opentelemetry-api (>=1.39.1)", "opentelemetry-exporter-otlp-proto-http (>=1.39.1)", "opentelemetry-sdk (>=1.39.1)"] -xxhash = ["xxhash (>=3.6.0,<3.7.0)"] - [[package]] name = "requests" version = "2.33.0" @@ -2715,4 +2695,4 @@ dev = ["pytest", "setuptools"] [metadata] lock-version = "2.1" python-versions = "^3.12" -content-hash = "1e726460d8433f929a08ddca9a953e4b2af22a87b594fc075e64a48c378147be" +content-hash = "8dd164d3ea9e0833dbca78643fefd7e161ef7bcdbea0625392c3a7bab26e51a0" diff --git a/pyproject.toml b/pyproject.toml index 58e285f1..42153b31 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,6 @@ pydantic = "^2.10.0" pydantic-settings = "^2.6.0" pyyaml = "^6.0" python-multipart = ">=0.0.9,<0.0.32" -redis = ">=7.4,<9.0" apscheduler = "^3.10.0" httpx = "^0.28.0" structlog = ">=25.5,<27.0" diff --git a/requirements.txt b/requirements.txt index 8deaa8c5..dec56400 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,7 +10,6 @@ pydantic-settings>=2.6.0,<3.0.0 pyyaml>=6.0,<7.0 types-PyYAML>=6.0 python-multipart>=0.0.9,<1.0.0 -redis>=7.4.0,<8.0.0 apscheduler>=3.10.0,<4.0.0 httpx>=0.28.0,<0.29.0 structlog>=25.5.0,<27.0.0 diff --git a/scripts/validate_sdk_pin.py b/scripts/validate_sdk_pin.py index a7627888..7993e7d1 100644 --- a/scripts/validate_sdk_pin.py +++ b/scripts/validate_sdk_pin.py @@ -4,6 +4,21 @@ Policy (CEG#266 / PR #263): constellation-node-sdk tracks ``Quantum-L9/Gate_SDK@v1``. Manifests name the major tag. poetry.lock records that tag as ``reference`` and the resolved object as ``resolved_reference``. + +CEG-001 asked for an immutable SHA in the manifests instead. The release set +kept the moving tag and closed the finding the other way: the **lock** is the +identity every deployed image installs, so the lock is what this script +verifies, and the resolved commit is printed on PASS so a build log records +which SDK object was actually taken. ``Enrichment.Inference.Engine/scripts/ +validate_sdk_pin.py`` enforces the same contract over that repo's +``requirements.lock``. + +Accepted trade-off, stated rather than hidden: a build that resolves the +manifest live (``pip install -r requirements.txt``, and CI installs generally) +takes whatever ``v1`` points at that minute, while a lock-driven build takes +``resolved_reference``. They agree today. Moving the ``v1`` tag without +refreshing the lock is what would separate them, and that is a deliberate act +with a diff, not silent drift between two files in this repository. """ from __future__ import annotations @@ -57,13 +72,29 @@ def check_tree(root: Path) -> list[str]: return errors +def resolved_commit(root: Path) -> str | None: + """The commit poetry.lock records for the major tag, for the PASS line.""" + path = root / "poetry.lock" + if not path.exists(): + return None + match = LOCK_REFERENCE_RE.search(path.read_text(encoding="utf-8")) + return None if match is None else match.group(2) + + def main() -> int: errors = check_tree(ROOT) if errors: print("FAIL") print("\n".join(errors)) return 1 - print(f"PASS CEG pin {CANONICAL_REPO}@{MAJOR_TAG}") + resolved = resolved_commit(ROOT) + if resolved is None: + # check_tree() already accepts a tree with no poetry.lock, so this is + # reachable. Printing "PASS ... -> None" in a build log reads as a pin + # that resolved to nothing; say what is actually true instead. + print(f"PASS CEG pin {CANONICAL_REPO}@{MAJOR_TAG} (no poetry.lock; no resolved commit recorded)") + return 0 + print(f"PASS CEG pin {CANONICAL_REPO}@{MAJOR_TAG} -> {resolved}") return 0 diff --git a/templates/.env.recommended.template b/templates/.env.recommended.template index 70f134e1..75e24555 100755 --- a/templates/.env.recommended.template +++ b/templates/.env.recommended.template @@ -2,5 +2,4 @@ POSTGRES_USER POSTGRES_DB POSTGRES_PASSWORD -REDIS_URL LOG_LEVEL diff --git a/tests/contracts/_constants.py b/tests/contracts/_constants.py index 722941ea..df298d78 100644 --- a/tests/contracts/_constants.py +++ b/tests/contracts/_constants.py @@ -15,7 +15,8 @@ OUTCOME_VALUES = {"success", "failure", "partial"} -REQUIRED_ENV_VARS = ["NEO4J_URI", "NEO4J_USERNAME", "NEO4J_PASSWORD", "REDIS_URL", "API_KEY"] +# CEG-007: REDIS_URL removed — no code under engine/ or chassis/ imports redis. +REQUIRED_ENV_VARS = ["NEO4J_URI", "NEO4J_USERNAME", "NEO4J_PASSWORD", "API_KEY"] FORBIDDEN_PROD_SECRETS: dict[str, list[str]] = { "NEO4J_PASSWORD": ["password", "change-me-in-production"], diff --git a/tests/contracts/conftest.py b/tests/contracts/conftest.py index 25f5e8a6..9f55c86d 100644 --- a/tests/contracts/conftest.py +++ b/tests/contracts/conftest.py @@ -69,14 +69,6 @@ def neo4j_dep() -> dict: return _load_yaml(path) -@pytest.fixture -def redis_dep() -> dict: - path = CONTRACTS_ROOT / "dependencies" / "redis.yaml" - if not path.exists(): - pytest.skip("redis.yaml not present") - return _load_yaml(path) - - @pytest.fixture def tool_index() -> dict: path = CONTRACTS_ROOT / "agents" / "tool-schemas" / "_index.yaml" diff --git a/tests/contracts/test_dependency_contracts.py b/tests/contracts/test_dependency_contracts.py index e098d817..2b11daa7 100644 --- a/tests/contracts/test_dependency_contracts.py +++ b/tests/contracts/test_dependency_contracts.py @@ -1,5 +1,5 @@ """ -External dependency contract tests for Neo4j and Redis. +External dependency contract tests for Neo4j. Sources: engine/graph/driver.py:GraphDriver, CircuitBreaker @@ -62,24 +62,9 @@ def test_neo4j_dep_direct_access_forbidden(neo4j_dep): assert direct == "forbidden" -# ── Redis dependency contract ──────────────────────────────────────────────── - - -def test_redis_dep_service_name(redis_dep): - assert redis_dep.get("service_name") == "redis" - - -def test_redis_dep_version_is_7(redis_dep): - version = str(redis_dep.get("version", "")) - assert "7" in version - - -def test_redis_dep_uses_env_var(redis_dep): - conn = redis_dep.get("connection", {}) - assert conn.get("base_url_env") == "REDIS_URL" - - -def test_redis_dep_usage_documents_scoring_cache(redis_dep): - usages = [u.get("purpose", "").lower() for u in redis_dep.get("usage", [])] - scoring_cached = any("scor" in u or "cache" in u or "gds" in u for u in usages) - assert scoring_cached +# CEG-007: the Redis dependency-contract tests are gone with the contract they +# asserted. redis.yaml declared "Scoring result caching" and "Domain pack cache" +# as its usages; neither was ever implemented, and nothing under engine/ or +# chassis/ imports redis. Four tests passed continuously against a description +# of behaviour that did not exist — which is what made the unused dependency +# look load-bearing. diff --git a/tests/unit/test_domain_database_provisioning.py b/tests/unit/test_domain_database_provisioning.py new file mode 100644 index 00000000..032317db --- /dev/null +++ b/tests/unit/test_domain_database_provisioning.py @@ -0,0 +1,172 @@ +"""CEG-008 — a domain database that does not exist must say so, or be created. + +`match` and `sync` route queries to a Neo4j database named after the domain id +(`engine/handlers.py`, `database=domain_spec.domain.id`). Neo4j does not create +databases implicitly, so on a fresh instance every sync and match failed with +the driver's own message until an operator ran CREATE DATABASE by hand — the +Constellation E2E hit exactly this and had to add the step to its boot sequence. + +Two halves, both tested here: with `auto_create_domain_database` on the engine +provisions on first use; with it off the failure names the missing database and +the exact command that provides it. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from engine.graph.driver import ( + DatabaseNotProvisionedError, + GraphDriver, + _looks_like_absent_database, +) + + +class _RecordingDriver(GraphDriver): + """GraphDriver with the Neo4j round trip replaced by a recorder.""" + + def __init__(self, *, fail_with: Exception | None = None) -> None: + super().__init__(uri="bolt://unused", username="u", password="p") + self.calls: list[tuple[str, str]] = [] + self._fail_with = fail_with + + async def _raw_execute_query( # type: ignore[override] + self, + cypher: str, + parameters: dict[str, Any] | None = None, + database: str = "neo4j", + ) -> list[Any]: + self.calls.append((cypher, database)) + if self._fail_with is not None and database != "system": + raise self._fail_with + return [] + + +@pytest.fixture +def auto_create(monkeypatch): + from engine.config import settings as settings_module + + monkeypatch.setattr(settings_module.settings, "auto_create_domain_database", True) + + +@pytest.fixture +def no_auto_create(monkeypatch): + from engine.config import settings as settings_module + + monkeypatch.setattr(settings_module.settings, "auto_create_domain_database", False) + + +# ── Off: the failure has to be actionable ─────────────────────────────────── + + +def test_absent_database_is_recognised() -> None: + assert _looks_like_absent_database(Exception("Database does not exist: plasticos")) + assert _looks_like_absent_database(Exception("Unable to get a routing table for database")) + assert not _looks_like_absent_database(Exception("SyntaxError: bad cypher")) + + +@pytest.mark.asyncio +async def test_missing_database_names_itself_and_the_fix(no_auto_create) -> None: + driver = _RecordingDriver(fail_with=Exception("Database does not exist: plasticos")) + with pytest.raises(DatabaseNotProvisionedError) as excinfo: + await driver.execute_query("MATCH (n) RETURN n", database="plasticos") + + message = str(excinfo.value) + assert "plasticos" in message + assert "CREATE DATABASE" in message + assert "AUTO_CREATE_DOMAIN_DATABASE" in message + + +@pytest.mark.asyncio +async def test_unrelated_errors_are_not_reinterpreted(no_auto_create) -> None: + boom = ValueError("SyntaxError: unexpected token") + driver = _RecordingDriver(fail_with=boom) + with pytest.raises(ValueError, match="unexpected token"): + await driver.execute_query("MATCH (n RETURN n", database="plasticos") + + +@pytest.mark.asyncio +async def test_nothing_is_provisioned_while_the_flag_is_off(no_auto_create) -> None: + driver = _RecordingDriver() + await driver.execute_query("MATCH (n) RETURN n", database="plasticos") + assert not any("CREATE DATABASE" in cypher for cypher, _ in driver.calls) + + +# ── On: provision once, on first use ──────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_database_is_created_on_first_use(auto_create) -> None: + driver = _RecordingDriver() + await driver.execute_query("MATCH (n) RETURN n", database="plasticos") + + creates = [(c, db) for c, db in driver.calls if "CREATE DATABASE" in c] + assert creates, "expected the domain database to be provisioned" + cypher, target = creates[0] + assert cypher == "CREATE DATABASE `plasticos` IF NOT EXISTS WAIT" + assert target == "system", "administrative commands must run against `system`" + + +@pytest.mark.asyncio +async def test_creation_is_attempted_only_once_per_process(auto_create) -> None: + driver = _RecordingDriver() + for _ in range(3): + await driver.execute_query("MATCH (n) RETURN n", database="plasticos") + assert sum("CREATE DATABASE" in c for c, _ in driver.calls) == 1 + + +@pytest.mark.asyncio +async def test_builtin_databases_are_never_provisioned(auto_create) -> None: + driver = _RecordingDriver() + await driver.execute_query("RETURN 1", database="neo4j") + await driver.execute_query("SHOW DATABASES", database="system") + assert not any("CREATE DATABASE" in c for c, _ in driver.calls) + + +@pytest.mark.asyncio +async def test_a_hyphenated_domain_id_is_accepted() -> None: + """Domain ids legitimately contain dashes, so sanitize_label cannot be used.""" + driver = _RecordingDriver() + assert await driver.ensure_database("healthcare-referral") is True + assert ("CREATE DATABASE `healthcare-referral` IF NOT EXISTS WAIT", "system") in driver.calls + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "name", + [ + "plasticos`; DROP DATABASE neo4j; --", # back-quote escape + "1leading-digit", + "ab", # under the 3-character minimum + "has space", + "", + ], +) +async def test_an_unsafe_database_name_is_refused(name: str) -> None: + """CREATE DATABASE takes no parameter, so the name is interpolated — and + therefore validated before it ever reaches the statement.""" + driver = _RecordingDriver() + with pytest.raises(ValueError, match="refusing to provision"): + await driver.ensure_database(name) + assert driver.calls == [] + + +@pytest.mark.asyncio +async def test_provisioning_failure_is_reported_not_raised(auto_create) -> None: + """Community Edition rejects CREATE DATABASE. A deployment whose database + already exists must not be blocked by a CREATE it is not allowed to run.""" + + class _RefusingDriver(_RecordingDriver): + async def _raw_execute_query(self, cypher, parameters=None, database="neo4j"): + self.calls.append((cypher, database)) + if "CREATE DATABASE" in cypher: + msg = "Unsupported administration command in Community Edition" + raise RuntimeError(msg) + return [] + + driver = _RefusingDriver() + assert await driver.ensure_database("plasticos") is False + # The ordinary query still runs; it is the query that reports the truth. + assert await driver.execute_query("MATCH (n) RETURN n", database="plasticos") == [] diff --git a/tests/unit/test_domain_pack_shape.py b/tests/unit/test_domain_pack_shape.py new file mode 100644 index 00000000..417c5d14 --- /dev/null +++ b/tests/unit/test_domain_pack_shape.py @@ -0,0 +1,153 @@ +"""CEG-009 — every domain under domains/ must be in the shape the loader reads. + +``DomainPackLoader`` resolves ``domains//spec.yaml`` +(``loader.SPEC_FILENAME``). Nine specs used a flat +``_domain_spec.yaml`` convention instead and were therefore unreachable: +they looked like available verticals in a directory listing, but a tenant id +matching one of them failed to resolve. Only ``plasticos`` loaded by default. + +Two of those nine were additionally invalid — ``executive-assistant`` declared +an edge to an undeclared ``Skill`` node, ``roofing-company`` one to ``ZipCode`` +— and nothing caught it, because a file the loader never opens is never +validated either. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +import pytest +import yaml + +from engine.config.loader import SPEC_FILENAME, DomainPackLoader +from engine.config.schema import DomainSpec + +REPO_ROOT = Path(__file__).resolve().parents[2] +DOMAINS_ROOT = REPO_ROOT / "domains" + + +def _domain_dirs() -> list[Path]: + return sorted(p for p in DOMAINS_ROOT.iterdir() if p.is_dir()) + + +def test_no_flat_domain_spec_files_remain() -> None: + stragglers = sorted(p.name for p in DOMAINS_ROOT.glob("*_domain_spec.yaml")) + assert stragglers == [], ( + f"{stragglers} use the flat shape the loader never reads. Move each to " + f"domains//{SPEC_FILENAME}, or delete it if superseded." + ) + + +def test_every_domain_directory_holds_a_spec() -> None: + missing = [d.name for d in _domain_dirs() if not (d / SPEC_FILENAME).is_file()] + assert missing == [], f"domain directories without {SPEC_FILENAME}: {missing}" + + +@pytest.mark.parametrize("domain_dir", _domain_dirs(), ids=lambda p: p.name) +def test_domain_spec_validates(domain_dir: Path) -> None: + """A spec the loader can find is a spec that must survive validation.""" + raw = yaml.safe_load((domain_dir / SPEC_FILENAME).read_text(encoding="utf-8")) + DomainSpec.model_validate(raw) + + +@pytest.mark.parametrize("domain_dir", _domain_dirs(), ids=lambda p: p.name) +def test_directory_name_matches_declared_domain_id(domain_dir: Path) -> None: + """The loader resolves by domain id, so the directory name IS the lookup key.""" + raw = yaml.safe_load((domain_dir / SPEC_FILENAME).read_text(encoding="utf-8")) + assert raw["domain"]["id"] == domain_dir.name + + +def test_loader_discovers_every_unflagged_domain() -> None: + from engine.config.loader import _DOMAIN_FEATURE_FLAGS + + loader = DomainPackLoader(str(DOMAINS_ROOT)) + discovered = set(loader.list_domains()) + on_disk = {d.name for d in _domain_dirs()} + assert discovered <= on_disk + # Everything withheld is withheld deliberately, by a declared flag. + assert on_disk - discovered <= set(_DOMAIN_FEATURE_FLAGS) + + +def test_every_discovered_domain_actually_loads() -> None: + loader = DomainPackLoader(str(DOMAINS_ROOT)) + for domain_id in sorted(loader.list_domains()): + assert loader.load_domain(domain_id).domain.id == domain_id + + +# -------------------------------------------------------------------------- +# Readable is not the same as correct. +# -------------------------------------------------------------------------- + +_IDENTIFIER = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") + + +def _cypher_defects(spec) -> list[str]: + """Gate compilations that would fail or silently reject every candidate. + + Two shapes, both found only after CEG-009 made these packs readable: + + * ``RELATES_TO`` — the compiler's fallback when a ``type: traversal`` gate + declares no ``edgetype``. Packs written with ``pattern``/``condition`` + hit it, and no ontology here declares ``RELATES_TO``, so the gate is a + hard filter that matches nothing. + * ``$`` — a scalar ``queryparam`` (``85.0``, ``5``, ``1``). + GateSpec coerces it to a string and the compiler emits it as a parameter + *name*, so Cypher receives ``$85.0``. + """ + from engine.gates.compiler import GateCompiler + + compiler = GateCompiler(spec) + defects: list[str] = [] + for gate in spec.gates: + cypher = compiler.compile(gate) + if "RELATES_TO" in cypher and not gate.edgetype: + defects.append(f"{gate.name}: RELATES_TO fallback — {cypher}") + for param in re.findall(r"\$([^\s)]+)", cypher): + if not _IDENTIFIER.fullmatch(param): + defects.append(f"{gate.name}: '${param}' is not a parameter name — {cypher}") + return defects + + +@pytest.mark.parametrize("domain_dir", _domain_dirs(), ids=lambda p: p.name) +def test_discoverable_domain_gates_compile_to_executable_cypher(domain_dir: Path) -> None: + """A pack the loader will serve must produce Cypher that can run. + + The gap this closes: validating a spec against DomainSpec proves it parses, + not that its gates execute. Five packs passed Pydantic validation and still + compiled to `exists((candidate)-[:RELATES_TO]->(t))` or `$85.0`. + + A pack behind a feature flag is exempt — that is what the flag records. + """ + from engine.config.loader import _DOMAIN_FEATURE_FLAGS + + if domain_dir.name in _DOMAIN_FEATURE_FLAGS: + pytest.skip(f"{domain_dir.name} is flag-gated: {_DOMAIN_FEATURE_FLAGS[domain_dir.name]}") + + raw = yaml.safe_load((domain_dir / SPEC_FILENAME).read_text(encoding="utf-8")) + defects = _cypher_defects(DomainSpec.model_validate(raw)) + assert defects == [], f"{domain_dir.name} would serve unexecutable Cypher:\n " + "\n ".join(defects) + + +def test_flag_gated_packs_are_gated_for_a_reason_that_still_holds() -> None: + """Shrink-only: when a dormant pack starts compiling, un-gate it. + + Without this, a pack fixed later stays invisible and the flag becomes a + place defects go to be forgotten. + """ + from engine.config.loader import _DOMAIN_FEATURE_FLAGS + + still_broken, now_clean = [], [] + for name, flag in _DOMAIN_FEATURE_FLAGS.items(): + if flag != "unvalidated_domain_packs_enabled": + continue + path = DOMAINS_ROOT / name / SPEC_FILENAME + if not path.is_file(): + continue + spec = DomainSpec.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) + (still_broken if _cypher_defects(spec) else now_clean).append(name) + + assert now_clean == [], ( + f"{now_clean} now compile cleanly — remove them from _DOMAIN_FEATURE_FLAGS so the loader serves them again." + ) + assert still_broken, "no pack is gated as unvalidated; drop the flag and this test" diff --git a/tests/unit/test_graph_inference_egress.py b/tests/unit/test_graph_inference_egress.py new file mode 100644 index 00000000..2be3afe6 --- /dev/null +++ b/tests/unit/test_graph_inference_egress.py @@ -0,0 +1,145 @@ +"""EIE-008 / CEG-006 — the graph-inference feedback loop needs a producer. + +EIE advertises `graph-inference-result` to Gate and implements the entire +consumer side: packet validation, per-tenant queues, target extraction, a 0.55 +confidence floor, and injection into the convergence loop. No code in CEG ever +constructed such a packet, so the loop the architecture implies had a consumer +and nothing on the other end — and the two halves were built to the same +confidence floor without ever being connected. + +The mirror finding: CEG's own outbound `request_enrichment` was reachable only +through `engine/health/api.py`, which nothing imported, so no inbound packet +could cause CEG to call EIE either. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from engine.gate_egress import ( + GRAPH_INFERENCE_ACTION, + INFERENCE_CONFIDENCE_FLOOR, + build_inference_outputs, + emit_graph_inference_result, + inference_idempotency_key, +) +from engine.inference_rule_registry import InferenceResult + + +def _result(field: str, value: Any, confidence: float, rule: str = "r") -> InferenceResult: + return InferenceResult(field_name=field, value=value, confidence=confidence, rule_name=rule) + + +# ── Output shaping: EIE's contract is the spec ────────────────────────────── + + +def test_outputs_carry_every_key_eie_requires() -> None: + """EIE's extract_targets_from_packet reads entity_id, field, value, + confidence and rule off each element.""" + outputs = build_inference_outputs("e-1", [_result("tier", "mid_market", 0.85, "size_rule")]) + assert len(outputs) == 1 + assert outputs[0]["entity_id"] == "e-1" + assert outputs[0]["field"] == "tier" + assert outputs[0]["value"] == "mid_market" + assert outputs[0]["confidence"] == 0.85 + assert outputs[0]["rule"] == "size_rule" + + +def test_the_floor_matches_the_receiver() -> None: + """EIE drops below 0.55 (graph_return_channel.CONFIDENCE_FLOOR); so do we, + so a packet never carries outputs the receiver will silently discard.""" + assert INFERENCE_CONFIDENCE_FLOOR == 0.55 + outputs = build_inference_outputs( + "e-1", + [_result("kept", "a", 0.56), _result("dropped", "b", 0.54)], + ) + assert [o["field"] for o in outputs] == ["kept"] + + +def test_plain_mappings_are_accepted_too() -> None: + outputs = build_inference_outputs("e-1", [{"field": "f", "value": 1, "confidence": 0.9}]) + assert outputs[0]["rule"] == "unknown" + + +@pytest.mark.parametrize( + "bad", + [{"value": 1, "confidence": 0.9}, {"field": "f", "value": 1, "confidence": "high"}], +) +def test_malformed_outputs_are_dropped_not_sent(bad: dict) -> None: + assert build_inference_outputs("e-1", [bad]) == [] + + +def test_idempotency_key_is_stable_over_the_same_findings() -> None: + a = build_inference_outputs("e-1", [_result("f", "v", 0.9)]) + b = build_inference_outputs("e-1", [_result("f", "v", 0.9)]) + assert inference_idempotency_key("t", "e-1", a) == inference_idempotency_key("t", "e-1", b) + + +def test_idempotency_key_changes_when_the_finding_changes() -> None: + a = build_inference_outputs("e-1", [_result("f", "v1", 0.9)]) + b = build_inference_outputs("e-1", [_result("f", "v2", 0.9)]) + assert inference_idempotency_key("t", "e-1", a) != inference_idempotency_key("t", "e-1", b) + + +# ── Dispatch ──────────────────────────────────────────────────────────────── + + +class _FakeHeader: + packet_id = "pkt-1" + packet_type = "response" + correlation_id = None + + +class _FakeResponse: + header = _FakeHeader() + payload: dict[str, Any] = {"queued": 1} + + +@pytest.mark.asyncio +async def test_nothing_above_the_floor_sends_no_packet(monkeypatch) -> None: + """An empty inference_outputs list is valid to EIE and would cost a Gate + round trip to queue nothing.""" + + def _no_client(): + msg = "the Gate client must not be constructed" + raise AssertionError(msg) + + monkeypatch.setenv("GATE_URL", "http://gate.test") + monkeypatch.setattr("engine.gate_egress.get_gate_client", _no_client) + + result = await emit_graph_inference_result(tenant="t", entity_id="e-1", results=[_result("f", "v", 0.1)]) + assert result["status"] == "skipped" + assert result["sent_outputs"] == 0 + + +@pytest.mark.asyncio +async def test_unconfigured_gate_fails_closed(monkeypatch) -> None: + monkeypatch.delenv("GATE_URL", raising=False) + result = await emit_graph_inference_result(tenant="t", entity_id="e-1", results=[_result("f", "v", 0.9)]) + assert result["status"] == "failed" + assert result["error"] == "gate_not_configured" + + +@pytest.mark.asyncio +async def test_a_successful_emission_addresses_gate_with_eies_action(monkeypatch) -> None: + sent: dict[str, Any] = {} + + class _Client: + async def execute(self, **kwargs: Any) -> _FakeResponse: + sent.update(kwargs) + return _FakeResponse() + + monkeypatch.setenv("GATE_URL", "http://gate.test") + monkeypatch.setattr("engine.gate_egress.get_gate_client", _Client) + + result = await emit_graph_inference_result(tenant="acme", entity_id="e-1", results=[_result("tier", "small", 0.9)]) + + assert result["status"] == "ok" + assert result["sent_outputs"] == 1 + assert sent["action"] == GRAPH_INFERENCE_ACTION + assert sent["tenant"] == "acme" + assert sent["payload"]["inference_outputs"][0]["field"] == "tier" + # CEG never addresses EIE: the destination is Gate, resolved by action. + assert "destination_node" not in sent