Oikos Week 3: scheduler, drift detectors, signals, classifier, approvals

New kernel modules, all wired into `homelab` CLI + tested against live
production where reachable:

- oikos/scheduler.py — Observe stage: HTTP health probes for every
  service, disk-usage probes on hubris/strong, writes oikos/state.json
  (gitignored — regenerates every run). `homelab service <name> health`
  is now cache-first; `--live` forces a fresh probe. Deploys via
  oikos/systemd/oikos-scheduler.{timer,service} on LXC 105.

- oikos/drift.py — SOPS-recipient-vs-inventory and lifecycle-consistency
  detectors (fully local, no SSH) plus pct-list and Caddy-backend
  detectors (best-effort SSH, degrade to an info finding when
  unreachable rather than a false drift alarm). Found real, currently-
  true drift on first run: republic-laptop's age key granted on every
  secret but missing from inventory.yaml, grimmory missing from
  hello.yaml's recipients, and an undocumented pve_id 131 on hubris —
  recorded in OIKOS.md for the operator, not auto-fixed (each is a
  config_mutation/destructive decision).

- oikos/signal.py — the attention layer: raised -> acknowledged ->
  acting -> resolved|muted lifecycle, severity-based routing, dedup via
  open_signal_for(). `homelab signal list|raise|ack|resolve|mute`.

- oikos/decide.py — the Decide-stage classifier: risk class x blast
  radius x ledger-history confidence -> auto-act/escalate. Adds an
  action-alias layer (oikos/policy.py ACTION_ALIASES) and auto-infers
  service_name from the entity for per-service policy overrides.
  `homelab decide <action> <entity>`.

- oikos/approve.py — the escalate route. No dedicated Matrix bot exists
  in this homelab, so this is the repo-side half only: request/reply/
  grant lifecycle with short-TTL HMAC-signed tokens (new secret
  secrets/oikos-approval-hmac.yaml, recipients apps+hubris). Matrix
  delivery is Hermes's existing @dtoro:avispero send path (documented
  integration contract in the module docstring), not a new bot.
  `homelab restart` now mechanically refuses config_mutation/destructive
  services without a valid --approval-id, regardless of -y/interactivity.

- oikos/report.py — daily brief + weekly report from signal/approval/
  ledger state (no Prometheus yet, so point-in-time counts only).

- plans/2026-07-05-oikos-prometheus-lxc.md — Prometheus is `planned`,
  not provisioned: no pve_id is guessed here since Proxmox assigns real
  IDs at creation time, and drift already found an unclaimed ID (131) to
  investigate first.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.gitignore

@@ -1,3 +1,8 @@
 .DS_Store
 __pycache__/
-*.pyc
+*.pyc
+
+# Regenerated every scheduler run (every 10 min); no audit value in the
+# diff. Signals (signals/*.jsonl) ARE tracked — this is just the ephemeral
+# health-probe cache. See oikos/scheduler.py.
+oikos/state.json

.sops.yaml

@@ -115,4 +115,12 @@ creation_rules:
       age1vf8h7s8mqsn2q5eadgpdupsj4mwn8zguc77d85ws3xj40sl9rgksx2rxw6,
       age1z62ff2ak9zj5ctcvaxwyyhedwjvlwgm2dkn9nk3wrwk8fkavcpmsqwc2vs,
       age1s07zs83ehtlg8jtwvr75ltc3c4cdlemfwjuxrwjtwkqxkl9tpggsyrzn2h
+
+  - path_regex: ^secrets/oikos-approval-hmac\.yaml$
+    # HMAC signing key for Oikos approval-grant tokens (oikos/approve.py).
+    # Recipients: apps (105, runs the approval engine alongside homelab-mcp)
+    # and hubris (admin/debug decrypt). See OIKOS.md "Approval engine".
+    age: >-
+      age1xkklkvnk5z0fsnh6cfgv70hy9ksfy8rdprwerzw4yk3p4p7cxcqs2yvpz6,
+      age1duyl8mkpgu80uv934dy8q7enqjms6yvdz264hme8uryuxmvvqesq6rusq0
 # webhook noop 2026-05-20T18:16:57+02:00

OIKOS.md

@@ -95,13 +95,46 @@ Generated views: [infrastructure/topology.md](infrastructure/topology.md)
 
 ## Build status (30-day roadmap, started 2026-07-05)
 
-- **Week 1 (this)**: policy, ontology, service contract, archaeology,
+- **Week 1**: policy, ontology, service contract, archaeology, topology
-  topology generator, this brief.
+  generator, this brief. Shipped.
 - **Week 2**: context cards, `homelab service <name> …`, change ledger,
-  `node relations`, runbooks.
+  `node relations`, runbooks. Shipped.
-- **Week 3**: Prometheus + node_exporter, ops scheduler + state cache,
+- **Week 3**: ops scheduler + state cache (`homelab service <name> health`
-  drift detectors, signal engine, decision classifier, approval engine
+  is cache-first, `--live` forces a probe), drift detectors, signal engine
-  (Matrix ✅/❌), daily brief.
+  (`homelab signal …`), decision classifier (`homelab decide …`), approval
+  engine (`homelab approval …` — shared-HMAC grants; Matrix delivery is
+  Hermes's existing `@dtoro:avispero` send path, not a new bot, see
+  `oikos/approve.py`), daily brief + weekly report (`oikos/report.py`).
+  Shipped, except: Prometheus is still `planned` (see
+  [plans/2026-07-05-oikos-prometheus-lxc.md](plans/2026-07-05-oikos-prometheus-lxc.md)) —
+  trend signals (disk-full prediction, temp creep) wait on that LXC; the
+  scheduler's disk check today is point-in-time only, and CPU/NVMe
+  temperature isn't probed at all yet (no confirmed sensor path on
+  hubris/strong). DNS-vs-inventory and generic tracked-config-cleanliness
+  drift checks are also deferred (see `oikos/drift.py` docstring).
 - **Week 4**: Oikos Console (oikos.hubris.network, behind Authentik with
   step-up re-auth on approvals), per-agent age-key-signed approval
-  requests, docs pass, 60/90-day backlog.
+  requests (upgrading from Week 3's shared-HMAC), docs pass, 60/90-day
+  backlog.
+
+### Real drift found while building Week 3 (unresolved, needs operator action)
+
+The drift detectors surfaced genuine, currently-true findings on first
+run against production — recorded here rather than silently fixed, since
+each is a `config_mutation`/`destructive`-class decision:
+
+- `republic-laptop` has no `age_pubkey:` in `inventory.yaml`, but its real
+  age key is granted on nearly every shared secret in `.sops.yaml`
+  (`age1vf8h7...`) — the enrollment write-back to inventory never
+  happened. Fix: `homelab client add republic-laptop --finalize-pubkey
+  age1vf8h7s8mqsn2q5eadgpdupsj4mwn8zguc77d85ws3xj40sl9rgksx2rxw6`.
+- `grimmory` has an `age_pubkey` in inventory but is missing from
+  `secrets/hello.yaml`'s recipient list — incomplete enrollment the
+  other direction. Fix: re-run `homelab client add grimmory
+  --finalize-pubkey <its key>`.
+- `pve_id 131` exists live on hubris (`pct list`) with no inventory entry
+  — investigate before assuming it's a stale ID (see the Prometheus LXC
+  plan doc above, which flags this explicitly).
+- Three `lifecycle-pve-id-reuse` info findings (100, 106, 107 each shared
+  between an active host and an archaeology entry) — expected/benign ID
+  reuse after destroy, no action needed.

bin/homelab

@@ -42,11 +42,15 @@ AGE_KEY = Path(os.environ.get("SOPS_AGE_KEY_FILE", "/etc/age/key.txt"))
 # "feature unavailable" instead of crashing every subcommand.
 sys.path.insert(0, str(CONTEXT))
 try:
+    from oikos import approve as oikos_approve
+    from oikos import decide as oikos_decide
     from oikos import ledger as oikos_ledger
     from oikos import policy as oikos_policy
     from oikos import relations as oikos_relations
+    from oikos import signal as oikos_signal
 except ImportError:
-    oikos_ledger = oikos_policy = oikos_relations = None
+    oikos_approve = oikos_decide = oikos_ledger = oikos_policy = None
+    oikos_relations = oikos_signal = None
 
 
 # ---------- helpers ----------
@@ -184,12 +188,14 @@ def service_backend_host(name: str) -> str:
 
 
 def _record_change(entity: str, action: str, risk: str, *,
-                    result: str | None = None, verification: str | None = None) -> None:
+                    result: str | None = None, verification: str | None = None,
+                    approval_ref: str | None = None) -> None:
     """Append a ledger entry and push it standalone (used by mutations that
     don't already go through push_inventory, e.g. restart)."""
     if oikos_ledger is None:
         return
-    oikos_ledger.append(entity, action, risk, result=result, verification=verification)
+    oikos_ledger.append(entity, action, risk, result=result, verification=verification,
+                        approval_ref=approval_ref)
     try:
         subprocess.run(["git", "add", "ledger/"], check=True, cwd=CONTEXT)
         if subprocess.run(["git", "diff", "--cached", "--quiet"], cwd=CONTEXT).returncode != 0:
@@ -598,15 +604,30 @@ def cmd_restart(args: argparse.Namespace) -> int:
     svc = args.service
     host_name = service_backend_host(svc)
     unit = service(svc).get("systemd_unit", svc)
+    risk = (oikos_policy.classify_action("service-restart", svc)
+            if oikos_policy else "reversible_low") or "reversible_low"
+    approval = oikos_policy.approval_for(risk) if oikos_policy else "none"
+
+    # Mechanical gate: config_mutation/destructive risk classes require a
+    # live grant regardless of -y/interactivity — an agent (or a human
+    # bypassing the confirm() prompt with -y) cannot mutate a gated service
+    # without a real oikos/approve.py approval. See oikos/policy.yaml.
+    if approval != "none":
+        if not args.approval_id:
+            die(f"restarting '{svc}' is risk class '{risk}' (approval: {approval}) — "
+                f"pass --approval-id <id> from an approved 'homelab approval request'")
+        ok, reason = oikos_approve.check_grant(args.approval_id, f"service:{svc}", "service-restart")
+        if not ok:
+            die(f"approval {args.approval_id} not valid for this action: {reason}")
+
     if not args.yes:
         if not confirm(f"restart systemd unit '{unit}' on {host_name}?"):
             return 1
     base = ssh_base(host_name)
     rc = subprocess.call(base + ["--", "systemctl", "restart", unit])
-    risk = (oikos_policy.classify_action("service-restart", svc)
-            if oikos_policy else "reversible_low") or "reversible_low"
     _record_change(f"service:{svc}", "restart", risk,
-                   result=("ok" if rc == 0 else f"failed rc={rc}"))
+                   result=("ok" if rc == 0 else f"failed rc={rc}"),
+                   approval_ref=args.approval_id)
     return rc
 
 
@@ -1241,12 +1262,24 @@ def cmd_service(args: argparse.Namespace) -> int:
         url = svc.get("url") or svc.get("endpoint")
         if not url:
             die(f"service {name} has no url/endpoint in inventory")
+        if not args.live:
+            try:
+                from oikos import scheduler as oikos_scheduler
+                cached = oikos_scheduler.cached_service_health(name)
+            except ImportError:
+                cached = None
+            if cached is not None and cached.get("checked"):
+                status = "ok" if cached.get("ok") else "unhealthy"
+                print(f"{name}: {cached.get('checked_url', url)} -> "
+                      f"{cached.get('http_code') or 'no response'} ({status}, "
+                      f"as of {cached['as_of']} — pass --live to force a fresh probe)")
+                return 0 if cached.get("ok") else 1
         proc = subprocess.run(
             ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "5", url],
             capture_output=True, text=True,
         )
         code = proc.stdout.strip() or "no response"
-        print(f"{name}: {url} -> {code}  (live probe — cache-first reads land Week 3)")
+        print(f"{name}: {url} -> {code}  (live probe)")
         return 0 if code.startswith(("2", "3")) else 1
 
     if args.action == "docs":
@@ -1325,6 +1358,93 @@ def cmd_node_relations(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_decide(args: argparse.Namespace) -> int:
+    """Route a proposed action: auto-act or escalate. See oikos/decide.py."""
+    _require_oikos()
+    result = oikos_decide.classify(args.action, args.entity, service_name=args.service_name,
+                                   record=not args.no_record)
+    print(json.dumps(result, indent=2))
+    return 0 if result["route"] == "auto-act" else 1
+
+
+def cmd_approval_request(args: argparse.Namespace) -> int:
+    _require_oikos()
+    entry = oikos_approve.request(
+        args.entity, args.action, args.risk, args.evidence,
+        verification=args.verification, requires_phrase=args.requires_phrase,
+        ttl_hours=args.ttl_hours,
+    )
+    print(json.dumps({k: v for k, v in entry.items() if k != "matrix_message"}, indent=2))
+    print()
+    print("--- post this to Matrix ---")
+    print(entry["matrix_message"])
+    return 0
+
+
+def cmd_approval_list(args: argparse.Namespace) -> int:
+    _require_oikos()
+    for e in oikos_approve.list_approvals(state=args.state):
+        print(json.dumps(e))
+    return 0
+
+
+def cmd_approval_reply(args: argparse.Namespace) -> int:
+    _require_oikos()
+    try:
+        entry = oikos_approve.reply(args.id, args.decision, phrase=args.phrase,
+                                    decided_by=args.decided_by)
+    except (ValueError, RuntimeError) as e:
+        die(str(e))
+    print(json.dumps(entry, indent=2))
+    return 0
+
+
+def cmd_approval_check(args: argparse.Namespace) -> int:
+    _require_oikos()
+    ok, reason = oikos_approve.check_grant(args.id, args.entity, args.action)
+    print(f"{'GRANTED' if ok else 'DENIED'}: {reason}")
+    return 0 if ok else 1
+
+
+def cmd_signal_raise(args: argparse.Namespace) -> int:
+    _require_oikos()
+    action = None
+    if args.action_runbook or args.action_risk:
+        action = {"runbook": args.action_runbook, "risk": args.action_risk}
+    entry = oikos_signal.raise_signal(args.kind, args.severity, args.entity, args.evidence,
+                                      likely_cause=args.likely_cause,
+                                      recommended_action=action,
+                                      verification=args.verification)
+    print(json.dumps(entry, indent=2))
+    return 0
+
+
+def cmd_signal_list(args: argparse.Namespace) -> int:
+    _require_oikos()
+    for e in oikos_signal.list_signals(state=args.state, entity=args.entity,
+                                       severity=args.severity, kind=args.kind):
+        print(json.dumps(e))
+    return 0
+
+
+def cmd_signal_ack(args: argparse.Namespace) -> int:
+    _require_oikos()
+    print(json.dumps(oikos_signal.acknowledge(args.id, args.note), indent=2))
+    return 0
+
+
+def cmd_signal_resolve(args: argparse.Namespace) -> int:
+    _require_oikos()
+    print(json.dumps(oikos_signal.resolve(args.id, args.note), indent=2))
+    return 0
+
+
+def cmd_signal_mute(args: argparse.Namespace) -> int:
+    _require_oikos()
+    print(json.dumps(oikos_signal.mute(args.id, args.ttl_hours, args.note), indent=2))
+    return 0
+
+
 def cmd_nuke(args: argparse.Namespace) -> int:
     name = args.name
     if not args.yes:
@@ -1642,6 +1762,9 @@ def main() -> int:
     sp = sub.add_parser("restart", help="restart a service")
     sp.add_argument("service")
     sp.add_argument("--yes", "-y", action="store_true")
+    sp.add_argument("--approval-id", default=None,
+                    help="required if the service's risk class needs approval "
+                         "(see 'homelab approval request')")
     sp.set_defaults(func=cmd_restart)
 
     sp = sub.add_parser("open", help="open a service's URL in browser")
@@ -1702,6 +1825,8 @@ def main() -> int:
     sp.add_argument("action", choices=["explain", "health", "docs", "log", "actions", "history"])
     sp.add_argument("--lines", "-n", type=int, default=200, help="for 'log'")
     sp.add_argument("--limit", type=int, default=20, help="for 'history'")
+    sp.add_argument("--live", action="store_true",
+                    help="for 'health': force a fresh probe instead of the scheduler's cache")
     sp.set_defaults(func=cmd_service)
 
     change = sub.add_parser("change", help="change/mutation workflow")
@@ -1715,6 +1840,81 @@ def main() -> int:
     sp.add_argument("action", choices=["relations"])
     sp.set_defaults(func=cmd_node_relations)
 
+    sp = sub.add_parser("decide", help="classify a proposed action: auto-act or escalate")
+    sp.add_argument("action")
+    sp.add_argument("entity")
+    sp.add_argument("--service-name", default=None)
+    sp.add_argument("--no-record", action="store_true",
+                    help="skip writing this classification to the change ledger")
+    sp.set_defaults(func=cmd_decide)
+
+    approval = sub.add_parser("approval", help="approval-engine requests (escalate route)")
+    apsub = approval.add_subparsers(dest="action", required=True)
+
+    ap_req = apsub.add_parser("request")
+    ap_req.add_argument("entity")
+    ap_req.add_argument("action")
+    ap_req.add_argument("risk")
+    ap_req.add_argument("evidence")
+    ap_req.add_argument("--verification")
+    ap_req.add_argument("--requires-phrase", action="store_true")
+    ap_req.add_argument("--ttl-hours", type=int, default=24)
+    ap_req.set_defaults(func=cmd_approval_request)
+
+    ap_list = apsub.add_parser("list")
+    ap_list.add_argument("--state", choices=["pending", "approved", "denied", "expired"])
+    ap_list.set_defaults(func=cmd_approval_list)
+
+    ap_reply = apsub.add_parser("reply")
+    ap_reply.add_argument("id")
+    ap_reply.add_argument("decision", choices=["approve", "deny"])
+    ap_reply.add_argument("--phrase")
+    ap_reply.add_argument("--decided-by")
+    ap_reply.set_defaults(func=cmd_approval_reply)
+
+    ap_check = apsub.add_parser("check")
+    ap_check.add_argument("id")
+    ap_check.add_argument("entity")
+    ap_check.add_argument("action")
+    ap_check.set_defaults(func=cmd_approval_check)
+
+    signal = sub.add_parser("signal", help="the attention layer (oikos/signal.py)")
+    sigsub = signal.add_subparsers(dest="action", required=True)
+
+    sig_raise = sigsub.add_parser("raise")
+    sig_raise.add_argument("kind")
+    sig_raise.add_argument("severity", choices=["info", "warning", "critical"])
+    sig_raise.add_argument("entity")
+    sig_raise.add_argument("evidence")
+    sig_raise.add_argument("--likely-cause")
+    sig_raise.add_argument("--action-runbook")
+    sig_raise.add_argument("--action-risk")
+    sig_raise.add_argument("--verification")
+    sig_raise.set_defaults(func=cmd_signal_raise)
+
+    sig_list = sigsub.add_parser("list")
+    sig_list.add_argument("--state", choices=["raised", "acknowledged", "acting", "resolved", "muted"])
+    sig_list.add_argument("--entity")
+    sig_list.add_argument("--severity", choices=["info", "warning", "critical"])
+    sig_list.add_argument("--kind")
+    sig_list.set_defaults(func=cmd_signal_list)
+
+    sig_ack = sigsub.add_parser("ack")
+    sig_ack.add_argument("id")
+    sig_ack.add_argument("--note")
+    sig_ack.set_defaults(func=cmd_signal_ack)
+
+    sig_resolve = sigsub.add_parser("resolve")
+    sig_resolve.add_argument("id")
+    sig_resolve.add_argument("--note")
+    sig_resolve.set_defaults(func=cmd_signal_resolve)
+
+    sig_mute = sigsub.add_parser("mute")
+    sig_mute.add_argument("id")
+    sig_mute.add_argument("--ttl-hours", type=int, default=24)
+    sig_mute.add_argument("--note")
+    sig_mute.set_defaults(func=cmd_signal_mute)
+
     sp = sub.add_parser("nuke", help="shred /etc/age/key.txt + /opt/homelab-context on a host")
     sp.add_argument("name")
     sp.add_argument("--yes", "-y", action="store_true")

mcp/server.py

@@ -283,6 +283,21 @@ def get_relations(entity: str) -> list[dict]:
     return oikos_relations.relations_for_name(entity)
 
 
+@mcp.tool()
+def get_state_snapshot() -> dict:
+    """The Week-3 scheduler's last Observe-pass snapshot (service health,
+    host disk usage, drift-finding count, generated_at timestamp). This is
+    what makes cache-first reads work from ANY client, not just the one the
+    scheduler runs on: the CLI's local-file cache only helps on that host;
+    agents elsewhere should call this tool instead of assuming a local
+    oikos/state.json exists."""
+    from oikos import scheduler as oikos_scheduler
+    state = oikos_scheduler.read_state()
+    if state is None:
+        raise ValueError("no scheduler snapshot yet — has oikos-scheduler.timer run?")
+    return state
+
+
 @mcp.tool()
 def get_change_history(entity: str, limit: int = 20) -> list[dict]:
     """Ledger entries for `entity` (e.g. "service:jellyfin", "host:strong"),

oikos/approve.py

@@ -0,0 +1,300 @@
+#!/usr/bin/env python3
+"""oikos/approve.py — the approval engine (escalate route of the OODA loop).
+
+Repo-side half of the Week-3 approval flow. This module owns the request/
+grant lifecycle and the HMAC signing; it does NOT talk to Matrix directly.
+There is no dedicated Matrix bot in this homelab — alerts already go out
+as the operator's own Hermes agent posting to @dtoro:avispero (see
+infrastructure/monitoring.md's homelab-health-watchdog). The integration
+contract is:
+
+  1. An agent or the Week-3 scheduler calls `request()` (or the CLI
+     `request` subcommand) to open an approval. This prints the Matrix-
+     ready message text (evidence, options, reply format).
+  2. Hermes posts that text to Matrix using its existing send capability
+     (the same one homelab-health-watchdog already uses) and later reads
+     the operator's reply/reaction.
+  3. Hermes calls back `reply()` (or `oikos/approve.py reply <id>
+     approve|deny`) with the operator's decision. This module verifies
+     `requires_phrase` for destructive actions, then issues a short-TTL
+     HMAC-signed grant.
+  4. Mutating `homelab` commands call `check_grant()` before executing a
+     config_mutation/destructive action.
+
+Storage: approvals/<YYYY-MM>.jsonl, same append-only-JSONL-with-latest-
+state-wins convention as oikos/signal.py.
+
+Grant signing key: secrets/oikos-approval-hmac.yaml (SOPS, recipients:
+apps + hubris — see .sops.yaml). Decrypted on demand; never cached to
+disk in plaintext.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import subprocess
+import sys
+from datetime import datetime, timedelta, timezone
+from functools import lru_cache
+from pathlib import Path
+
+import yaml
+
+REPO = Path(__file__).resolve().parent.parent
+APPROVALS_DIR = REPO / "approvals"
+HMAC_SECRET = REPO / "secrets" / "oikos-approval-hmac.yaml"
+AGE_KEY = Path(os.environ.get("SOPS_AGE_KEY_FILE", "/etc/age/key.txt"))
+
+VALID_STATES = ("pending", "approved", "denied", "expired")
+REQUEST_TTL_HOURS = 24
+GRANT_TTL_MINUTES = 15
+
+
+def _month_path(dt: datetime | None = None) -> Path:
+    dt = dt or datetime.now(timezone.utc)
+    return APPROVALS_DIR / f"{dt.strftime('%Y-%m')}.jsonl"
+
+
+def _all_entries() -> list[dict]:
+    if not APPROVALS_DIR.exists():
+        return []
+    out = []
+    for path in sorted(APPROVALS_DIR.glob("*.jsonl")):
+        for line in path.read_text().splitlines():
+            if not line.strip():
+                continue
+            try:
+                out.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return out
+
+
+def _next_id(dt: datetime | None = None) -> str:
+    dt = dt or datetime.now(timezone.utc)
+    prefix = f"appr-{dt.strftime('%Y-%m-%d')}-"
+    existing = [e["id"] for e in _all_entries() if e.get("id", "").startswith(prefix)]
+    n = 1
+    while f"{prefix}{n:04d}" in existing:
+        n += 1
+    return f"{prefix}{n:04d}"
+
+
+def _append(entry: dict) -> dict:
+    APPROVALS_DIR.mkdir(exist_ok=True)
+    entry = {k: v for k, v in entry.items() if v is not None}
+    with _month_path().open("a") as f:
+        f.write(json.dumps(entry, sort_keys=False) + "\n")
+    return entry
+
+
+def current(request_id: str) -> dict | None:
+    matches = [e for e in _all_entries() if e.get("id") == request_id]
+    if not matches:
+        return None
+    matches.sort(key=lambda e: e.get("ts", ""))
+    return matches[-1]
+
+
+def list_approvals(state: str | None = None) -> list[dict]:
+    latest: dict[str, dict] = {}
+    for e in sorted(_all_entries(), key=lambda e: e.get("ts", "")):
+        rid = e.get("id")
+        if rid:
+            latest[rid] = e
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    out = []
+    for e in latest.values():
+        if e.get("state") == "pending" and e.get("expires_at") and e["expires_at"] < now:
+            e = {**e, "state": "expired"}
+        if state and e.get("state") != state:
+            continue
+        out.append(e)
+    out.sort(key=lambda e: e.get("ts", ""), reverse=True)
+    return out
+
+
+def _matrix_message(entry: dict) -> str:
+    lines = [
+        f"[Oikos approval {entry['id']}] {entry['entity']} — {entry['action']}",
+        f"risk: {entry['risk']}",
+        f"evidence: {entry['evidence']}",
+    ]
+    if entry.get("verification"):
+        lines.append(f"verification: {entry['verification']}")
+    if entry.get("requires_phrase"):
+        lines.append(f'reply: "approve {entry["id"]} {entry["confirmation_phrase"]}" or "deny {entry["id"]}"')
+    else:
+        lines.append(f'reply: ✅ to approve, ❌ to deny (or "approve {entry["id"]}" / "deny {entry["id"]}")')
+    return "\n".join(lines)
+
+
+def request(entity: str, action: str, risk: str, evidence: str, *,
+            verification: str | None = None, signal_id: str | None = None,
+            requires_phrase: bool = False, requested_by: str | None = None,
+            ttl_hours: int = REQUEST_TTL_HOURS) -> dict:
+    """Open a new approval request. `requires_phrase=True` (use for
+    `destructive`-class actions per oikos/policy.yaml) generates a
+    confirmation phrase the operator must echo back — a reaction alone
+    can't authorize it."""
+    rid = _next_id()
+    now = datetime.now(timezone.utc)
+    phrase = f"{action} {entity}" if requires_phrase else None
+    entry = {
+        "id": rid,
+        "ts": now.isoformat(timespec="seconds"),
+        "entity": entity,
+        "action": action,
+        "risk": risk,
+        "evidence": evidence,
+        "verification": verification,
+        "signal_id": signal_id,
+        "requires_phrase": requires_phrase,
+        "confirmation_phrase": phrase,
+        "requested_by": requested_by or os.environ.get("HOMELAB_AGENT_ID"),
+        "state": "pending",
+        "expires_at": (now + timedelta(hours=ttl_hours)).isoformat(timespec="seconds"),
+    }
+    recorded = _append(entry)
+    recorded["matrix_message"] = _matrix_message(recorded)
+    return recorded
+
+
+@lru_cache(maxsize=1)
+def _hmac_key() -> str:
+    if not HMAC_SECRET.exists():
+        raise RuntimeError(f"no secret at {HMAC_SECRET} — has it been provisioned?")
+    env = {**os.environ}
+    if AGE_KEY.exists():
+        env["SOPS_AGE_KEY_FILE"] = str(AGE_KEY)
+    proc = subprocess.run(["sops", "-d", str(HMAC_SECRET)],
+                          capture_output=True, text=True, env=env)
+    if proc.returncode != 0:
+        raise RuntimeError(
+            f"sops decrypt of oikos-approval-hmac failed (is this host a recipient?): "
+            f"{proc.stderr.strip()}")
+    return yaml.safe_load(proc.stdout)["hmac_key"]
+
+
+def _sign(request_id: str, entity: str, action: str, expires_at: str) -> str:
+    key = _hmac_key().encode()
+    msg = f"{request_id}:{entity}:{action}:{expires_at}".encode()
+    return hmac.new(key, msg, hashlib.sha256).hexdigest()
+
+
+def reply(request_id: str, decision: str, *, phrase: str | None = None,
+          decided_by: str | None = None) -> dict:
+    """Record the operator's decision. On approval, issues a short-TTL
+    HMAC-signed grant token. Raises ValueError if a required confirmation
+    phrase is missing or wrong, or if the request already expired."""
+    if decision not in ("approve", "deny"):
+        raise ValueError("decision must be 'approve' or 'deny'")
+    entry = current(request_id)
+    if entry is None:
+        raise ValueError(f"unknown approval id: {request_id}")
+    if entry.get("state") != "pending":
+        raise ValueError(f"{request_id} is not pending (state={entry.get('state')})")
+    now_s = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    if entry.get("expires_at") and entry["expires_at"] < now_s:
+        _append({**entry, "ts": now_s, "state": "expired"})
+        raise ValueError(f"{request_id} expired at {entry['expires_at']}")
+
+    if decision == "deny":
+        return _append({**entry, "ts": now_s, "state": "denied", "decided_by": decided_by})
+
+    if entry.get("requires_phrase"):
+        if phrase != entry.get("confirmation_phrase"):
+            raise ValueError(
+                "confirmation phrase missing or incorrect — a reaction alone "
+                "cannot approve a destructive action")
+
+    grant_expires = (datetime.now(timezone.utc) + timedelta(minutes=GRANT_TTL_MINUTES)) \
+        .isoformat(timespec="seconds")
+    grant_token = _sign(request_id, entry["entity"], entry["action"], grant_expires)
+    return _append({**entry, "ts": now_s, "state": "approved", "decided_by": decided_by,
+                    "grant_token": grant_token, "grant_expires": grant_expires})
+
+
+def check_grant(request_id: str, entity: str, action: str) -> tuple[bool, str]:
+    """Verify a request carries a valid, unexpired, matching grant. Returns
+    (ok, reason). Callers (homelab CLI mutating commands) must call this
+    immediately before executing — grants are single-use in spirit (the
+    60/90-day backlog adds exact command+target binding + replay
+    prevention; today re-checking the same still-valid grant twice is
+    possible, so keep grant TTLs short)."""
+    entry = current(request_id)
+    if entry is None:
+        return False, "unknown approval id"
+    if entry.get("state") != "approved":
+        return False, f"not approved (state={entry.get('state')})"
+    if entry.get("entity") != entity or entry.get("action") != action:
+        return False, "grant does not match entity/action"
+    now_s = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    if not entry.get("grant_expires") or entry["grant_expires"] < now_s:
+        return False, "grant expired"
+    expected = _sign(request_id, entity, action, entry["grant_expires"])
+    if not hmac.compare_digest(expected, entry.get("grant_token", "")):
+        return False, "grant signature invalid"
+    return True, "ok"
+
+
+def main() -> int:
+    import argparse
+    p = argparse.ArgumentParser(description="oikos approval engine")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    r = sub.add_parser("request")
+    r.add_argument("entity")
+    r.add_argument("action")
+    r.add_argument("risk")
+    r.add_argument("evidence")
+    r.add_argument("--verification")
+    r.add_argument("--signal-id")
+    r.add_argument("--requires-phrase", action="store_true")
+    r.add_argument("--ttl-hours", type=int, default=REQUEST_TTL_HOURS)
+
+    ls = sub.add_parser("list")
+    ls.add_argument("--state", choices=VALID_STATES)
+
+    rp = sub.add_parser("reply")
+    rp.add_argument("id")
+    rp.add_argument("decision", choices=["approve", "deny"])
+    rp.add_argument("--phrase")
+    rp.add_argument("--decided-by")
+
+    ck = sub.add_parser("check")
+    ck.add_argument("id")
+    ck.add_argument("entity")
+    ck.add_argument("action")
+
+    args = p.parse_args()
+    if args.cmd == "request":
+        entry = request(args.entity, args.action, args.risk, args.evidence,
+                        verification=args.verification, signal_id=args.signal_id,
+                        requires_phrase=args.requires_phrase, ttl_hours=args.ttl_hours)
+        print(json.dumps({k: v for k, v in entry.items() if k != "matrix_message"}, indent=2))
+        print()
+        print("--- post this to Matrix ---")
+        print(entry["matrix_message"])
+    elif args.cmd == "list":
+        for e in list_approvals(state=args.state):
+            print(json.dumps(e))
+    elif args.cmd == "reply":
+        try:
+            entry = reply(args.id, args.decision, phrase=args.phrase, decided_by=args.decided_by)
+        except (ValueError, RuntimeError) as e:
+            print(f"error: {e}", file=sys.stderr)
+            return 1
+        print(json.dumps(entry, indent=2))
+    elif args.cmd == "check":
+        ok, reason = check_grant(args.id, args.entity, args.action)
+        print(f"{'GRANTED' if ok else 'DENIED'}: {reason}")
+        return 0 if ok else 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

oikos/decide.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+"""oikos/decide.py — the Decide stage of the OODA loop (decision classifier).
+
+Scores a proposed action against three inputs and routes it:
+
+  1. risk class       — oikos/policy.yaml (read_only / reversible_low /
+                         config_mutation / destructive)
+  2. blast radius      — oikos/relations.py transitive impact graph
+  3. confidence        — prior ledger history of this exact action on this
+                         exact entity, falling back to "is this a
+                         well-known mechanical action" when there's no
+                         history yet
+
+Routes: **auto-act** (execute — risk is within unattended policy, blast
+radius is contained, confidence isn't low) or **escalate** (request
+operator approval via oikos/approve.py — anything else). The classifier
+can only make an action MORE cautious than policy says, never less: if
+policy already requires approval, escalate always wins regardless of
+confidence or radius.
+
+Note on the third OODA route, "queue": that's the Signal engine's
+info-severity routing (oikos/signal.py SEVERITY_ROUTE — informational
+findings that need no action land in the console/weekly report, not
+here). This module only classifies things that might actually need to
+run, so it only ever returns auto-act or escalate.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO))
+from oikos import ledger as oikos_ledger  # noqa: E402
+from oikos import policy as oikos_policy  # noqa: E402
+from oikos import relations as oikos_relations  # noqa: E402
+
+# Actions we consider "well-known mechanical" absent any ledger history —
+# mirrors oikos/policy.py's safe_actions_for_service() baseline.
+_ROUTINE_ACTIONS = {"restart", "service-restart", "health-check", "view-logs",
+                    "view-docs", "sync", "cache-clear"}
+
+# A blast radius at or below this size counts as "contained" for auto-act
+# purposes. Anything wider always escalates regardless of risk/confidence.
+_CONTAINED_RADIUS = 1
+
+
+def _confidence(action: str, entity_id: str) -> tuple[str, str]:
+    hist = oikos_ledger.history(entity_id, limit=50)
+    successes = [h for h in hist if h.get("action") == action and h.get("result") == "ok"]
+    failures = [h for h in hist
+               if h.get("action") == action and str(h.get("result", "")).startswith("failed")]
+    if failures and not successes:
+        return "low", f"{len(failures)} prior failed attempt(s) of '{action}' on this entity"
+    if successes:
+        return "high", f"{len(successes)} prior successful run(s) of '{action}' on this entity"
+    if action in _ROUTINE_ACTIONS:
+        return "medium", "no history yet, but this is a well-known mechanical action"
+    return "low", "no ledger history and not a recognized routine action"
+
+
+def classify(action: str, entity: str, *, service_name: str | None = None,
+             record: bool = False) -> dict:
+    """Classify one proposed action against one entity. `entity` may be a
+    bare name (resolved via oikos/relations.py) or an already-namespaced
+    id ("service:x" / "host:x"). If bare and ambiguous (matches both a
+    host and a service), the first resolved id is used — pass a
+    namespaced id explicitly to disambiguate.
+    """
+    entity_ids = oikos_relations.resolve(entity)
+    entity_id = entity_ids[0]
+
+    action = oikos_policy.canonical_action(action)
+    # Per-service policy overrides are keyed by service name (e.g. "caddy").
+    # Most single-purpose nodes are named identically to the service they
+    # run, so infer it from the entity's bare name when not given explicitly.
+    if service_name is None:
+        service_name = entity_id.split(":", 1)[1] if ":" in entity_id else entity_id
+
+    risk = (oikos_policy.classify_action(action, service_name)
+            or oikos_policy.classify_command(action))
+    if risk is None:
+        risk = "config_mutation"  # unknown action: default to the cautious class
+
+    approval = oikos_policy.approval_for(risk)
+    radius = oikos_relations.blast_radius(entity_id)
+    contained = len(radius) <= _CONTAINED_RADIUS
+    confidence, why = _confidence(action, entity_id)
+
+    if approval != "none":
+        route = "escalate"
+        reasoning = f"risk class '{risk}' requires approval ({approval}) per oikos/policy.yaml"
+    elif not contained:
+        route = "escalate"
+        reasoning = f"blast radius not contained ({len(radius)} entities: {', '.join(radius)})"
+    elif confidence == "low":
+        route = "escalate"
+        reasoning = f"low confidence — {why}"
+    else:
+        route = "auto-act"
+        reasoning = (f"risk '{risk}' is unattended-safe, blast radius contained "
+                    f"({radius or 'none'}), confidence {confidence} — {why}")
+
+    result = {
+        "entity": entity_id,
+        "action": action,
+        "risk": risk,
+        "approval": approval,
+        "blast_radius": radius,
+        "contained": contained,
+        "confidence": confidence,
+        "confidence_reason": why,
+        "route": route,
+        "reasoning": reasoning,
+    }
+    if record:
+        oikos_ledger.append(entity_id, f"decide:{action}", risk,
+                            result=route, notes=reasoning)
+    return result
+
+
+def main() -> int:
+    import argparse
+    p = argparse.ArgumentParser(description="oikos decision classifier")
+    p.add_argument("action")
+    p.add_argument("entity")
+    p.add_argument("--service-name", help="disambiguate policy overrides for a service action")
+    p.add_argument("--record", action="store_true",
+                  help="append this classification to the change ledger")
+    args = p.parse_args()
+    result = classify(args.action, args.entity, service_name=args.service_name,
+                      record=args.record)
+    print(json.dumps(result, indent=2))
+    return 0 if result["route"] == "auto-act" else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

oikos/drift.py

@@ -0,0 +1,302 @@
+#!/usr/bin/env python3
+"""oikos/drift.py — drift detectors (Week 3 reliability layer).
+
+Each detector returns a list of finding dicts:
+    {kind, severity, entity, evidence, likely_cause, recommended_action,
+     verification}
+matching the Signal schema in oikos/signal.py. `run_all()` runs every
+detector and raises a Signal for each finding, skipping ones that already
+have an open Signal for the same (entity, kind) — see
+oikos/signal.py open_signal_for(). SSH-based detectors degrade to a
+"probe-unreachable" info finding instead of a false drift signal when the
+target can't be reached (no ssh, no network) — drift means "state
+disagrees," not "couldn't check."
+
+Two detectors are fully local (no SSH, always safe to run from anywhere
+with the repo checked out): SOPS-recipient-vs-inventory and lifecycle
+consistency. Two need live access to hubris (pct list, caddy backend
+config) and gracefully no-op when unreachable.
+
+Not yet implemented — needs data this repo doesn't structure yet:
+  - DNS records vs inventory services (no Technitium API wiring here)
+  - Generic tracked-config-repo cleanliness for every service (needs the
+    `mutation_path`/local-checkout-path field per service; only caddy's
+    path is documented today, so that's the one config-cleanliness check
+    implemented below)
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO))
+from oikos import relations as oikos_relations  # noqa: E402
+from oikos import signal as oikos_signal  # noqa: E402
+
+import yaml  # noqa: E402
+
+SOPS_FILE = REPO / ".sops.yaml"
+INVENTORY = REPO / "inventory.yaml"
+
+_AGE_RE = re.compile(r"age1[a-z0-9]+")
+
+
+def _load_inventory() -> dict:
+    return yaml.safe_load(INVENTORY.read_text())
+
+
+def _sops_rules() -> list[dict]:
+    """Parse .sops.yaml's creation_rules: [{path_regex, recipients: [...]}]."""
+    doc = yaml.safe_load(SOPS_FILE.read_text())
+    return [
+        {"path_regex": r.get("path_regex", ""),
+         "recipients": _AGE_RE.findall(r.get("age", "") or "")}
+        for r in doc.get("creation_rules", [])
+    ]
+
+
+def check_sops_recipients(inv: dict | None = None) -> list[dict]:
+    """Two directions of drift:
+    1. A recipient pubkey in .sops.yaml that matches no host's age_pubkey
+       in inventory — likely a removed/revoked host whose secrets weren't
+       fully cleaned up (or a stale rule).
+    2. An enrolled host (non-empty age_pubkey) missing from the
+       secrets/hello.yaml rule — every enrolled client should be able to
+       decrypt the bootstrap canary; absence usually means enrollment
+       stalled partway.
+    """
+    inv = inv or _load_inventory()
+    hosts = inv.get("hosts", {})
+    known_pubkeys = {e["age_pubkey"]: name for name, e in hosts.items()
+                     if e.get("age_pubkey")}
+    rules = _sops_rules()
+    findings = []
+
+    all_recipients = {pk for r in rules for pk in r["recipients"]}
+    for pk in all_recipients:
+        if pk not in known_pubkeys:
+            findings.append({
+                "kind": "sops-orphan-recipient", "severity": "warning",
+                "entity": "repo:Homelab-Docs",
+                "evidence": f"age recipient {pk[:20]}... appears in .sops.yaml but "
+                            f"matches no host's age_pubkey in inventory.yaml",
+                "likely_cause": "host removed without full secret revocation, or a stale rule",
+                "recommended_action": {"runbook": "lifecycle-destroy-node", "risk": "destructive"},
+                "verification": "grep <pubkey> .sops.yaml; check inventory.yaml + archaeology",
+            })
+
+    hello_rule = next((r for r in rules if "hello" in r["path_regex"]), None)
+    if hello_rule:
+        for pubkey, name in known_pubkeys.items():
+            if pubkey not in hello_rule["recipients"]:
+                findings.append({
+                    "kind": "sops-missing-recipient", "severity": "warning",
+                    "entity": f"host:{name}",
+                    "evidence": f"{name}'s age_pubkey is not a recipient of secrets/hello.yaml",
+                    "likely_cause": "client-add --finalize-pubkey ran incompletely",
+                    "recommended_action": {"runbook": "client-enrollment", "risk": "config_mutation"},
+                    "verification": f"homelab client add {name} --finalize-pubkey <key>",
+                })
+    return findings
+
+
+def check_lifecycle_consistency(inv: dict | None = None) -> list[dict]:
+    """state: vs reality, using the ontology graph (oikos/relations.py) —
+    no SSH needed, this only reasons over inventory.yaml itself."""
+    inv = inv or _load_inventory()
+    hosts = inv.get("hosts", {})
+    archaeology = inv.get("archaeology", {})
+    findings = []
+
+    for name, e in hosts.items():
+        state = e.get("state", "active")
+        if state == "deprecated":
+            rel = oikos_relations.relations(f"host:{name}", inv)
+            if rel["affected_by"]:
+                findings.append({
+                    "kind": "lifecycle-inconsistent", "severity": "warning",
+                    "entity": f"host:{name}",
+                    "evidence": f"deprecated but still has inbound edges: {', '.join(rel['affected_by'])}",
+                    "likely_cause": "deprecation started before dependents were migrated off",
+                    "recommended_action": {"runbook": "lifecycle-deprecate-node", "risk": "config_mutation"},
+                    "verification": f"homelab node {name} relations",
+                })
+        elif state == "destroyed":
+            findings.append({
+                "kind": "lifecycle-inconsistent", "severity": "critical",
+                "entity": f"host:{name}",
+                "evidence": "state is 'destroyed' but the host still has a live hosts.<name>: entry "
+                            "(destroyed nodes belong in the archaeology: section, not hosts:)",
+                "likely_cause": "lifecycle-destroy-node runbook step 6 (move to archaeology) was skipped",
+                "recommended_action": {"runbook": "lifecycle-destroy-node", "risk": "destructive"},
+                "verification": "grep -A3 '^  " + name + ":' inventory.yaml",
+            })
+
+    # A pve_id collision between an active host and an archaeology entry on
+    # the same Proxmox node usually means the ID was reused without the old
+    # entry's data being fully accounted for.
+    active_by_id = {(e.get("host"), e.get("pve_id")): name
+                    for name, e in hosts.items() if e.get("pve_id")}
+    for aname, ae in archaeology.items():
+        key = (None, ae.get("pve_id"))  # archaeology doesn't record which node
+        for (host, pve_id), name in active_by_id.items():
+            if pve_id == ae.get("pve_id") and name != aname:
+                findings.append({
+                    "kind": "lifecycle-pve-id-reuse", "severity": "info",
+                    "entity": f"host:{name}",
+                    "evidence": f"pve_id {pve_id} is both active ({name}) and in archaeology ({aname}, "
+                                f"destroyed {ae.get('destroyed', '?')})",
+                    "likely_cause": "normal ID reuse after destroy — informational only",
+                    "recommended_action": None,
+                    "verification": None,
+                })
+    return findings
+
+
+def _ssh_hubris(cmd: list[str]) -> subprocess.CompletedProcess | None:
+    """Best-effort ssh to hubris. Returns None (not a CompletedProcess with
+    nonzero rc) if ssh itself can't run/connect at all, so callers can
+    distinguish "unreachable" from "reachable but the command failed."""
+    try:
+        return subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
+                               "root@192.168.8.77", " ".join(cmd)],
+                              capture_output=True, text=True, timeout=15)
+    except (subprocess.TimeoutExpired, OSError):
+        return None
+
+
+def check_pct_list(inv: dict | None = None) -> list[dict]:
+    """Inventory LXCs/VMs on hubris vs live `pct list` + `qm list`."""
+    inv = inv or _load_inventory()
+    hosts = inv.get("hosts", {})
+    proc = _ssh_hubris(["pct", "list"])
+    if proc is None or proc.returncode != 0:
+        return [{
+            "kind": "probe-unreachable", "severity": "info", "entity": "host:hubris",
+            "evidence": "could not reach hubris to run `pct list`",
+            "likely_cause": "no network path from this client to hubris, or ssh key not authorized",
+            "recommended_action": None, "verification": "homelab ssh hubris -- pct list",
+        }]
+    live_ids = set()
+    for line in proc.stdout.splitlines()[1:]:
+        parts = line.split()
+        if parts and parts[0].isdigit():
+            live_ids.add(int(parts[0]))
+
+    inventory_ids = {e["pve_id"] for name, e in hosts.items()
+                    if e.get("kind") == "lxc" and e.get("host") == "hubris" and e.get("pve_id")}
+
+    findings = []
+    for missing in sorted(inventory_ids - live_ids):
+        name = next(n for n, e in hosts.items() if e.get("pve_id") == missing)
+        findings.append({
+            "kind": "inventory-vs-live", "severity": "critical", "entity": f"host:{name}",
+            "evidence": f"pve_id {missing} ({name}) is in inventory.yaml but absent from "
+                        f"`pct list` on hubris",
+            "likely_cause": "destroyed outside the lifecycle-destroy-node runbook, or migrated "
+                            "without updating inventory",
+            "recommended_action": {"runbook": "lifecycle-destroy-node", "risk": "destructive"},
+            "verification": "homelab ssh hubris -- pct list",
+        })
+    for extra in sorted(live_ids - inventory_ids):
+        findings.append({
+            "kind": "inventory-vs-live", "severity": "warning", "entity": "host:hubris",
+            "evidence": f"pve_id {extra} exists on hubris (`pct list`) but has no inventory.yaml entry",
+            "likely_cause": "created outside the provision-node runbook",
+            "recommended_action": {"runbook": "lifecycle-provision-node", "risk": "config_mutation"},
+            "verification": "homelab ssh hubris -- pct config " + str(extra),
+        })
+    return findings
+
+
+def check_caddy_backends(inv: dict | None = None) -> list[dict]:
+    """Caddy's /etc/caddy (a git checkout of dtoro/caddy-conf, per
+    containers/121-caddy.md) vs inventory service backend IPs. Best-effort
+    grep for reverse_proxy targets; skips services whose Caddyfile snippet
+    doesn't use a bare IP (e.g. references a Caddy snippet/import)."""
+    inv = inv or _load_inventory()
+    hosts = inv.get("hosts", {})
+    services = inv.get("services", {})
+    caddy_ip = hosts.get("caddy", {}).get("lan_ip")
+    if not caddy_ip:
+        return []
+    try:
+        proc = subprocess.run(
+            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", f"root@{caddy_ip}",
+             "grep -rhoE 'reverse_proxy[^{]*' /etc/caddy/ 2>/dev/null"],
+            capture_output=True, text=True, timeout=15,
+        )
+    except (subprocess.TimeoutExpired, OSError):
+        proc = None
+    if proc is None or proc.returncode != 0:
+        return [{
+            "kind": "probe-unreachable", "severity": "info", "entity": "host:caddy",
+            "evidence": "could not reach caddy to inspect /etc/caddy",
+            "likely_cause": "no network path from this client to caddy, or ssh key not authorized",
+            "recommended_action": None, "verification": "homelab ssh caddy -- grep -r reverse_proxy /etc/caddy",
+        }]
+
+    live_ips = set(re.findall(r"\d+\.\d+\.\d+\.\d+", proc.stdout))
+    findings = []
+    for svc, e in services.items():
+        if not isinstance(e, dict):
+            continue
+        backend = e.get("backend")
+        if not backend or not (e.get("url") or e.get("public_host")):
+            continue
+        backend_ip = hosts.get(backend, {}).get("lan_ip")
+        if backend_ip and live_ips and backend_ip not in live_ips:
+            findings.append({
+                "kind": "caddy-backend-mismatch", "severity": "warning", "entity": f"service:{svc}",
+                "evidence": f"inventory backend IP {backend_ip} ({backend}) not found in any "
+                            f"reverse_proxy directive on caddy",
+                "likely_cause": "backend migrated (lan_ip changed) without updating the Caddyfile, "
+                                "or the service uses a snippet/import this grep can't resolve",
+                "recommended_action": {"runbook": "config-change-deploy", "risk": "config_mutation"},
+                "verification": f"homelab service {svc} health",
+            })
+    return findings
+
+
+DETECTORS = [check_sops_recipients, check_lifecycle_consistency,
+             check_pct_list, check_caddy_backends]
+
+
+def run_all(inv: dict | None = None, *, raise_signals: bool = False) -> list[dict]:
+    inv = inv or _load_inventory()
+    findings: list[dict] = []
+    for detector in DETECTORS:
+        findings.extend(detector(inv))
+    if raise_signals:
+        for f in findings:
+            if oikos_signal.open_signal_for(f["entity"], f["kind"]) is not None:
+                continue  # already open, don't duplicate
+            oikos_signal.raise_signal(
+                f["kind"], f["severity"], f["entity"], f["evidence"],
+                likely_cause=f.get("likely_cause"),
+                recommended_action=f.get("recommended_action"),
+                verification=f.get("verification"),
+            )
+    return findings
+
+
+def main() -> int:
+    import argparse
+    import json
+    p = argparse.ArgumentParser(description="oikos drift detectors")
+    p.add_argument("--raise-signals", action="store_true",
+                   help="raise a Signal for each new finding (default: print only)")
+    args = p.parse_args()
+    findings = run_all(raise_signals=args.raise_signals)
+    for f in findings:
+        print(json.dumps(f))
+    print(f"\n{len(findings)} finding(s)", file=sys.stderr)
+    return 1 if any(f["severity"] == "critical" for f in findings) else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

oikos/policy.py

@@ -23,8 +23,22 @@ def classify_command(cmd: str) -> str | None:
     return load().get("commands", {}).get(cmd)
 
 
+# Synonyms for the canonical action keys in oikos/policy.yaml `actions:`.
+# `homelab restart <service>` is the real CLI verb; "restart" is what an
+# agent proposing the action is most likely to say. Keep this list in sync
+# with anything oikos/decide.py or the CLI classifies by name.
+ACTION_ALIASES = {
+    "restart": "service-restart",
+}
+
+
+def canonical_action(action: str) -> str:
+    return ACTION_ALIASES.get(action, action)
+
+
 def classify_action(action: str, service: str | None = None) -> str | None:
     """Risk class for a generic action, honoring per-service overrides."""
+    action = canonical_action(action)
     pol = load()
     if service:
         override = pol.get("service_overrides", {}).get(service, {}).get(action)

oikos/report.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""oikos/report.py — daily morning brief + weekly operator report.
+
+Both are generated text, not sent directly to Matrix — same integration
+contract as oikos/approve.py: there's no dedicated Matrix bot in this
+homelab, so Hermes (already posting alerts as @dtoro:avispero, see
+infrastructure/monitoring.md) is the one that actually delivers this text.
+The daily brief is meant to run once a day (e.g. chained after an early
+oikos-scheduler.service run, or its own systemd timer); the weekly report
+is a deeper markdown review.
+
+Both are grounded only in what's actually implemented: signal state,
+pending approvals, and change-ledger history. There's no Prometheus yet
+(Week-3 backlog item pending LXC provisioning), so there are no trend
+lines here — only point-in-time counts.
+
+CLI:
+    python3 oikos/report.py daily
+    python3 oikos/report.py weekly
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO))
+from oikos import approve as oikos_approve  # noqa: E402
+from oikos import ledger as oikos_ledger  # noqa: E402
+from oikos import scheduler as oikos_scheduler  # noqa: E402
+from oikos import signal as oikos_signal  # noqa: E402
+
+
+def _recent_ledger_entries(since: datetime) -> list[dict]:
+    since_s = since.isoformat(timespec="seconds")
+    out = []
+    if not oikos_ledger.LEDGER_DIR.exists():
+        return out
+    for path in sorted(oikos_ledger.LEDGER_DIR.glob("*.jsonl")):
+        for line in path.read_text().splitlines():
+            if not line.strip():
+                continue
+            import json
+            try:
+                e = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if e.get("ts", "") >= since_s:
+                out.append(e)
+    out.sort(key=lambda e: e.get("ts", ""))
+    return out
+
+
+def daily_brief() -> str:
+    now = datetime.now(timezone.utc)
+    since = now - timedelta(hours=24)
+
+    state = oikos_scheduler.read_state()
+    if state is None:
+        health_line = "no scheduler snapshot yet — has oikos-scheduler.timer run?"
+    else:
+        svcs = state.get("services", {})
+        checked = [s for s in svcs.values() if s.get("checked")]
+        healthy = sum(1 for s in checked if s.get("ok"))
+        health_line = f"{healthy}/{len(checked)} services healthy (as of {state['generated_at']})"
+
+    open_signals = oikos_signal.list_signals()
+    open_signals = [s for s in open_signals if s.get("state") in ("raised", "acknowledged", "acting")]
+    by_sev = {"critical": 0, "warning": 0, "info": 0}
+    for s in open_signals:
+        by_sev[s.get("severity", "info")] = by_sev.get(s.get("severity", "info"), 0) + 1
+
+    pending = oikos_approve.list_approvals(state="pending")
+    changes = _recent_ledger_entries(since)
+
+    lines = [
+        f"Oikos daily brief — {now.strftime('%Y-%m-%d')}",
+        f"health: {health_line}",
+        f"signals: {by_sev['critical']} critical, {by_sev['warning']} warning, "
+        f"{by_sev['info']} info ({len(open_signals)} open)",
+        f"pending approvals: {len(pending)}",
+        f"changes in last 24h: {len(changes)}",
+    ]
+    if by_sev["critical"]:
+        lines.append("")
+        lines.append("Critical signals:")
+        for s in open_signals:
+            if s.get("severity") == "critical":
+                lines.append(f"  - {s['id']} {s['entity']}: {s['evidence']}")
+    if pending:
+        lines.append("")
+        lines.append("Pending approvals:")
+        for a in pending:
+            lines.append(f"  - {a['id']} {a['entity']} {a['action']} ({a['risk']})")
+    return "\n".join(lines)
+
+
+def weekly_report() -> str:
+    now = datetime.now(timezone.utc)
+    since = now - timedelta(days=7)
+    changes = _recent_ledger_entries(since)
+    decisions = [c for c in changes if c.get("action", "").startswith("decide:")]
+    auto_act = sum(1 for d in decisions if d.get("result") == "auto-act")
+    escalate = sum(1 for d in decisions if d.get("result") == "escalate")
+    mutations = [c for c in changes if not c.get("action", "").startswith("decide:")]
+
+    all_signals = oikos_signal.list_signals()
+    recent_signals = [s for s in all_signals if s.get("ts", "") >= since.isoformat(timespec="seconds")]
+    by_kind: dict[str, int] = {}
+    for s in recent_signals:
+        by_kind[s.get("kind", "?")] = by_kind.get(s.get("kind", "?"), 0) + 1
+
+    lines = [
+        f"# Oikos weekly report — {since.strftime('%Y-%m-%d')} to {now.strftime('%Y-%m-%d')}",
+        "",
+        "## Changes",
+        f"- {len(mutations)} mutation(s) recorded in the change ledger",
+        f"- {len(decisions)} classifier decision(s): {auto_act} auto-act, {escalate} escalate",
+        "",
+        "## Signals raised this week",
+    ]
+    if by_kind:
+        for kind, count in sorted(by_kind.items(), key=lambda kv: -kv[1]):
+            lines.append(f"- {kind}: {count}")
+    else:
+        lines.append("- none")
+    lines += [
+        "",
+        "## Open at time of report",
+        f"- {len([s for s in all_signals if s.get('state') in ('raised', 'acknowledged', 'acting')])} signal(s)",
+        f"- {len(oikos_approve.list_approvals(state='pending'))} pending approval(s)",
+        "",
+        "## Not yet available",
+        "- Trend lines (disk-full prediction, temp creep) — pending Prometheus provisioning",
+        "- Stale-doc detection — not yet implemented (60/90-day backlog)",
+    ]
+    return "\n".join(lines)
+
+
+def main() -> int:
+    import argparse
+    p = argparse.ArgumentParser(description="oikos daily brief / weekly report")
+    p.add_argument("kind", choices=["daily", "weekly"])
+    args = p.parse_args()
+    print(daily_brief() if args.kind == "daily" else weekly_report())
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

oikos/scheduler.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+"""oikos/scheduler.py — the Observe stage: periodic probes -> Signals + state cache.
+
+Intended to run under a systemd timer (see scripts/sync/linux/ for the
+existing timer pattern this should mirror — deploy target: LXC 105
+alongside the MCP server) every 5-15 minutes. Each run:
+
+  1. Probes every service's HTTP health and writes oikos/state.json — the
+     snapshot `homelab service <name> health` serves by default (cache-
+     first reads); pass --live on that command to force a fresh probe
+     instead of trusting the cache.
+  2. Probes disk usage on hubris + strong (best-effort SSH; degrades to a
+     "probe-unreachable" info Signal rather than a false disk-threshold
+     alarm when unreachable).
+  3. Runs the Week-3 drift detectors (oikos/drift.py).
+  4. Raises Signals for anything over threshold or drifted — skipping
+     duplicates via oikos_signal.open_signal_for() — and auto-resolves
+     any open Signal whose condition has since cleared.
+
+Temperature probing is NOT implemented here: LXCs don't expose host
+sensors, and hubris/strong's actual sensor path (lm-sensors vs vendor
+tool) hasn't been confirmed on either box yet — a real check needs that
+groundwork first rather than a guessed command. Backup-freshness is
+likewise deferred: `backs-up-to` isn't populated in inventory yet (see
+oikos/ontology.yaml — it's a documented thin field), so there's nothing
+structured to check freshness against.
+
+CLI:
+    python3 oikos/scheduler.py run             # probe, raise signals, write state.json
+    python3 oikos/scheduler.py run --dry-run   # probe and print only, no side effects
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+import yaml
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO))
+from oikos import drift as oikos_drift  # noqa: E402
+from oikos import signal as oikos_signal  # noqa: E402
+
+INVENTORY = REPO / "inventory.yaml"
+STATE_FILE = REPO / "oikos" / "state.json"
+
+DISK_WARN_PCT = 85
+DISK_CRIT_PCT = 95
+HEALTH_TIMEOUT_S = 5
+
+
+def _load_inventory() -> dict:
+    return yaml.safe_load(INVENTORY.read_text())
+
+
+# Services whose public `url`/`endpoint` isn't a plain-GET health check —
+# mirrors the special-casing bin/homelab's `doctor` command already does.
+# A generic per-service `health:` inventory field (checked URL/command
+# distinct from the public url) would remove the need for this; until
+# that's backfilled, these are the two known exceptions.
+_HEALTH_OVERRIDES = {
+    # FastMCP's streamable-http endpoint expects a JSON-RPC POST handshake,
+    # not a bare GET — a plain curl gets 4xx even when the server is fully
+    # healthy (confirmed against the live server: consistently 400, never
+    # 2xx/3xx, for a GET). Any HTTP response at all (vs. no response/
+    # connection refused) proves the process is up and answering, which is
+    # what "service-down" alerting actually cares about here.
+    "homelab_mcp": {"extra_curl_args": ["-H", "Accept: text/event-stream"],
+                    "any_response_ok": True},
+    "secrets_issuance": {
+        "url_transform": lambda url: url.rstrip("/").rsplit("/", 1)[0] + "/health",
+    },
+    # Token-gated at the app level (401 without a token is correct, not
+    # down) — see containers/128-trmnl.md, which documents a dedicated
+    # /health endpoint returning 200 unauthenticated.
+    "trmnl": {"url_transform": lambda url: url.rstrip("/") + "/health"},
+}
+
+
+def probe_service_health(name: str, entry: dict) -> dict:
+    url = entry.get("url") or entry.get("endpoint")
+    if not url:
+        return {"service": name, "checked": False}
+    override = _HEALTH_OVERRIDES.get(name, {})
+    check_url = override.get("url_transform", lambda u: u)(url)
+    curl_cmd = ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}",
+               "--max-time", str(HEALTH_TIMEOUT_S), *override.get("extra_curl_args", []), check_url]
+    try:
+        proc = subprocess.run(curl_cmd, capture_output=True, text=True)
+        code = proc.stdout.strip()
+    except OSError:
+        code = ""
+    ok = bool(code) if override.get("any_response_ok") else code.startswith(("2", "3"))
+    return {"service": name, "checked": True, "url": url, "checked_url": check_url,
+           "http_code": code or None, "ok": ok}
+
+
+def _ssh(host_ip: str, remote_cmd: str, timeout: int = 10) -> str | None:
+    try:
+        proc = subprocess.run(
+            ["ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={min(timeout, 5)}",
+             f"root@{host_ip}", remote_cmd],
+            capture_output=True, text=True, timeout=timeout,
+        )
+    except (subprocess.TimeoutExpired, OSError):
+        return None
+    return proc.stdout.strip() if proc.returncode == 0 else None
+
+
+def probe_disk(name: str, entry: dict) -> dict | None:
+    """`df -P /` on a proxmox-host's own root fs. LXC-level disk usage is
+    already covered by MCP's get_lxc_state; this probe is host-level."""
+    if entry.get("kind") != "proxmox-host" or not entry.get("lan_ip"):
+        return None
+    out = _ssh(entry["lan_ip"], "df -P / | tail -1")
+    if out is None:
+        return {"host": name, "checked": False}
+    parts = out.split()
+    if len(parts) < 5 or not parts[4].rstrip("%").isdigit():
+        return {"host": name, "checked": False}
+    return {"host": name, "checked": True, "used_pct": int(parts[4].rstrip("%"))}
+
+
+def run(dry_run: bool = False) -> dict:
+    inv = _load_inventory()
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+    services_state = {}
+    for name, entry in inv.get("services", {}).items():
+        if not isinstance(entry, dict):
+            continue
+        result = probe_service_health(name, entry)
+        services_state[name] = result
+        if not result.get("checked"):
+            continue
+        sig_kind = "service-down"
+        if result["ok"]:
+            open_sig = oikos_signal.open_signal_for(f"service:{name}", sig_kind)
+            if open_sig and not dry_run:
+                oikos_signal.resolve(open_sig["id"], note="health probe recovered")
+        else:
+            if not dry_run and oikos_signal.open_signal_for(f"service:{name}", sig_kind) is None:
+                oikos_signal.raise_signal(
+                    sig_kind, "critical", f"service:{name}",
+                    f"{result['url']} -> {result.get('http_code') or 'no response'}",
+                    likely_cause="backend down, crashed, or ingress misconfigured",
+                    recommended_action={"runbook": "service-health-check", "risk": "reversible_low"},
+                    verification=f"homelab service {name} health --live",
+                )
+
+    hosts_state = {}
+    for name, entry in inv.get("hosts", {}).items():
+        disk = probe_disk(name, entry)
+        if disk is None:
+            continue
+        hosts_state[name] = disk
+        if not disk.get("checked"):
+            continue
+        pct = disk["used_pct"]
+        sig_kind = "disk-threshold"
+        if pct >= DISK_CRIT_PCT:
+            severity = "critical"
+        elif pct >= DISK_WARN_PCT:
+            severity = "warning"
+        else:
+            severity = None
+        open_sig = oikos_signal.open_signal_for(f"host:{name}", sig_kind)
+        if severity is None:
+            if open_sig and not dry_run:
+                oikos_signal.resolve(open_sig["id"], note=f"disk usage back to {pct}%")
+        elif open_sig is None and not dry_run:
+            oikos_signal.raise_signal(
+                sig_kind, severity, f"host:{name}", f"root filesystem {pct}% used",
+                likely_cause="data growth or a runaway log",
+                recommended_action={"runbook": "config-change-deploy", "risk": "config_mutation"},
+                verification=f"homelab ssh {name} -- df -h /",
+            )
+
+    drift_findings = oikos_drift.run_all(inv, raise_signals=not dry_run)
+
+    snapshot = {
+        "generated_at": now,
+        "services": services_state,
+        "hosts": hosts_state,
+        "drift_findings": len(drift_findings),
+    }
+    if not dry_run:
+        STATE_FILE.write_text(json.dumps(snapshot, indent=2) + "\n")
+    return snapshot
+
+
+def read_state() -> dict | None:
+    """Load the last scheduler snapshot, or None if it's never run."""
+    if not STATE_FILE.exists():
+        return None
+    try:
+        return json.loads(STATE_FILE.read_text())
+    except json.JSONDecodeError:
+        return None
+
+
+def cached_service_health(name: str) -> dict | None:
+    state = read_state()
+    if state is None:
+        return None
+    entry = state.get("services", {}).get(name)
+    if entry is None:
+        return None
+    return {**entry, "as_of": state.get("generated_at")}
+
+
+def main() -> int:
+    import argparse
+    p = argparse.ArgumentParser(description="oikos scheduler (Observe stage)")
+    sub = p.add_subparsers(dest="cmd", required=True)
+    r = sub.add_parser("run")
+    r.add_argument("--dry-run", action="store_true")
+    args = p.parse_args()
+    if args.cmd == "run":
+        snapshot = run(dry_run=args.dry_run)
+        print(json.dumps(snapshot, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

oikos/signal.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""oikos/signal.py — the Signal engine (attention layer).
+
+A Signal is anything the lab notices that needs attention and possibly
+action: pending updates, high temperature, low disk, a service down, a
+cert expiring, a stale backup, inventory drift, a node stuck mid-
+lifecycle-transition. See OIKOS.md / oikos/ontology.yaml "Signals — the
+attention layer".
+
+Storage: signals/<YYYY-MM>.jsonl, same append-only-JSONL convention as
+oikos/ledger.py, but a signal's *state* changes over its lifecycle
+(raised -> acknowledged -> acting -> resolved | muted). Each state change
+is a NEW line with the same id — never edit a line in place, so git
+history stays a legible append-only audit trail. `current()` /
+`list_signals()` reduce the log to each id's latest state.
+
+CLI:
+    python3 oikos/signal.py raise <kind> <severity> <entity> <evidence> [--likely-cause] [--action-runbook] [--action-risk] [--verification]
+    python3 oikos/signal.py list [--state raised] [--entity strong] [--severity warning]
+    python3 oikos/signal.py ack <id>
+    python3 oikos/signal.py resolve <id> [--note ...]
+    python3 oikos/signal.py mute <id> [--ttl-hours 24]
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+SIGNALS_DIR = REPO / "signals"
+
+VALID_SEVERITIES = ("info", "warning", "critical")
+VALID_STATES = ("raised", "acknowledged", "acting", "resolved", "muted")
+# Routing per oikos/ontology.yaml "Signals" section.
+SEVERITY_ROUTE = {
+    "info": "console+report",
+    "warning": "matrix-digest",
+    "critical": "matrix-immediate",
+}
+
+
+def _month_path(dt: datetime | None = None) -> Path:
+    dt = dt or datetime.now(timezone.utc)
+    return SIGNALS_DIR / f"{dt.strftime('%Y-%m')}.jsonl"
+
+
+def _all_entries() -> list[dict]:
+    if not SIGNALS_DIR.exists():
+        return []
+    out = []
+    for path in sorted(SIGNALS_DIR.glob("*.jsonl")):
+        for line in path.read_text().splitlines():
+            if not line.strip():
+                continue
+            try:
+                out.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return out
+
+
+def _next_id(dt: datetime | None = None) -> str:
+    dt = dt or datetime.now(timezone.utc)
+    prefix = f"sig-{dt.strftime('%Y-%m-%d')}-"
+    existing = [e["id"] for e in _all_entries() if e.get("id", "").startswith(prefix)]
+    n = 1
+    while f"{prefix}{n:04d}" in existing:
+        n += 1
+    return f"{prefix}{n:04d}"
+
+
+def _append(entry: dict) -> dict:
+    SIGNALS_DIR.mkdir(exist_ok=True)
+    entry = {k: v for k, v in entry.items() if v is not None}
+    with _month_path().open("a") as f:
+        f.write(json.dumps(entry, sort_keys=False) + "\n")
+    return entry
+
+
+def raise_signal(kind: str, severity: str, entity: str, evidence: str, *,
+                  likely_cause: str | None = None,
+                  recommended_action: dict | None = None,
+                  verification: str | None = None,
+                  signal_id: str | None = None) -> dict:
+    """Raise a new Signal, or (if `signal_id` names an existing one and it's
+    not currently open) re-raise it. Returns the recorded entry."""
+    if severity not in VALID_SEVERITIES:
+        raise ValueError(f"severity must be one of {VALID_SEVERITIES}")
+    sid = signal_id or _next_id()
+    entry = {
+        "id": sid,
+        "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "kind": kind,
+        "severity": severity,
+        "entity": entity,
+        "evidence": evidence,
+        "likely_cause": likely_cause,
+        "recommended_action": recommended_action,
+        "verification": verification,
+        "state": "raised",
+        "route": SEVERITY_ROUTE.get(severity, "console+report"),
+    }
+    return _append(entry)
+
+
+def _transition(signal_id: str, state: str, *, note: str | None = None,
+                mute_until: str | None = None) -> dict:
+    if state not in VALID_STATES:
+        raise ValueError(f"state must be one of {VALID_STATES}")
+    prior = current(signal_id)
+    if prior is None:
+        raise ValueError(f"unknown signal id: {signal_id}")
+    entry = {**prior, "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+             "state": state, "note": note, "mute_until": mute_until}
+    return _append(entry)
+
+
+def acknowledge(signal_id: str, note: str | None = None) -> dict:
+    return _transition(signal_id, "acknowledged", note=note)
+
+
+def start_acting(signal_id: str, note: str | None = None) -> dict:
+    return _transition(signal_id, "acting", note=note)
+
+
+def resolve(signal_id: str, note: str | None = None) -> dict:
+    return _transition(signal_id, "resolved", note=note)
+
+
+def mute(signal_id: str, ttl_hours: int = 24, note: str | None = None) -> dict:
+    until = (datetime.now(timezone.utc) + timedelta(hours=ttl_hours)).isoformat(timespec="seconds")
+    return _transition(signal_id, "muted", note=note, mute_until=until)
+
+
+def current(signal_id: str) -> dict | None:
+    """Latest recorded state for one signal id, or None if unknown."""
+    matches = [e for e in _all_entries() if e.get("id") == signal_id]
+    if not matches:
+        return None
+    matches.sort(key=lambda e: e.get("ts", ""))
+    return matches[-1]
+
+
+def list_signals(state: str | None = None, entity: str | None = None,
+                  severity: str | None = None, kind: str | None = None) -> list[dict]:
+    """Every signal's latest state, optionally filtered. Auto-expires mute
+    (a signal muted past its mute_until is treated as raised again)."""
+    latest: dict[str, dict] = {}
+    for e in sorted(_all_entries(), key=lambda e: e.get("ts", "")):
+        sid = e.get("id")
+        if sid:
+            latest[sid] = e
+
+    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    out = []
+    for e in latest.values():
+        if e.get("state") == "muted" and e.get("mute_until") and e["mute_until"] < now:
+            e = {**e, "state": "raised", "note": "mute expired"}
+        if state and e.get("state") != state:
+            continue
+        if entity and e.get("entity") != entity:
+            continue
+        if severity and e.get("severity") != severity:
+            continue
+        if kind and e.get("kind") != kind:
+            continue
+        out.append(e)
+    out.sort(key=lambda e: e.get("ts", ""), reverse=True)
+    return out
+
+
+def open_signal_for(entity: str, kind: str) -> dict | None:
+    """The currently-open (raised/acknowledged/acting) signal for this
+    entity+kind, if any. Used by probes to avoid raising duplicates and to
+    auto-resolve when a condition clears."""
+    for e in list_signals(entity=entity, kind=kind):
+        if e.get("state") in ("raised", "acknowledged", "acting"):
+            return e
+    return None
+
+
+def main() -> int:
+    import argparse
+    p = argparse.ArgumentParser(description="oikos signal engine")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    r = sub.add_parser("raise")
+    r.add_argument("kind")
+    r.add_argument("severity", choices=VALID_SEVERITIES)
+    r.add_argument("entity")
+    r.add_argument("evidence")
+    r.add_argument("--likely-cause")
+    r.add_argument("--action-runbook")
+    r.add_argument("--action-risk")
+    r.add_argument("--verification")
+
+    ls = sub.add_parser("list")
+    ls.add_argument("--state", choices=VALID_STATES)
+    ls.add_argument("--entity")
+    ls.add_argument("--severity", choices=VALID_SEVERITIES)
+    ls.add_argument("--kind")
+
+    for name in ("ack", "resolve"):
+        sp = sub.add_parser(name)
+        sp.add_argument("id")
+        sp.add_argument("--note")
+
+    mt = sub.add_parser("mute")
+    mt.add_argument("id")
+    mt.add_argument("--ttl-hours", type=int, default=24)
+    mt.add_argument("--note")
+
+    args = p.parse_args()
+    if args.cmd == "raise":
+        action = None
+        if args.action_runbook or args.action_risk:
+            action = {"runbook": args.action_runbook, "risk": args.action_risk}
+        entry = raise_signal(args.kind, args.severity, args.entity, args.evidence,
+                             likely_cause=args.likely_cause, recommended_action=action,
+                             verification=args.verification)
+        print(json.dumps(entry, indent=2))
+    elif args.cmd == "list":
+        for e in list_signals(state=args.state, entity=args.entity,
+                              severity=args.severity, kind=args.kind):
+            print(json.dumps(e))
+    elif args.cmd == "ack":
+        print(json.dumps(acknowledge(args.id, args.note), indent=2))
+    elif args.cmd == "resolve":
+        print(json.dumps(resolve(args.id, args.note), indent=2))
+    elif args.cmd == "mute":
+        print(json.dumps(mute(args.id, args.ttl_hours, args.note), indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

oikos/systemd/oikos-scheduler.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=Oikos observe pass: health probes, drift detectors, signal engine
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+WorkingDirectory=/opt/homelab-context
+ExecStart=/usr/bin/env bash /opt/homelab-context/oikos/systemd/run-scheduler.sh
+TimeoutStartSec=120
+Nice=10
+
+[Install]
+WantedBy=multi-user.target

oikos/systemd/oikos-scheduler.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=Periodic Oikos observe pass (health, drift, signals)
+
+[Timer]
+OnBootSec=5min
+OnUnitActiveSec=10min
+AccuracySec=1min
+Unit=oikos-scheduler.service
+
+[Install]
+WantedBy=timers.target

oikos/systemd/run-scheduler.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+# Wrapper for oikos-scheduler.service: run one Observe pass, then commit +
+# push signals/*.jsonl if the run raised/transitioned anything. state.json
+# is intentionally NOT committed (see .gitignore) — it's regenerated every
+# run with no audit value in the diff; signals/ is the tracked ledger.
+set -euo pipefail
+cd /opt/homelab-context
+
+python3 oikos/scheduler.py run
+
+# `git status --porcelain` (not `git diff`) so a brand-new signals/*.jsonl
+# file (untracked) is caught too, not just modifications to tracked ones.
+if [ -d signals ] && [ -n "$(git status --porcelain -- signals/)" ]; then
+    git add signals/
+    if ! git diff --cached --quiet; then
+        git commit -m "signals: oikos-scheduler observe pass $(date -u +%Y-%m-%dT%H:%M:%SZ)"
+        git push
+    fi
+fi

plans/2026-07-05-oikos-prometheus-lxc.md

@@ -0,0 +1,69 @@
+# Oikos metrics stack — Prometheus LXC (planned)
+
+Lifecycle state: **planned** (see [oikos/ontology.yaml](../oikos/ontology.yaml)
+lifecycle). No LXC exists yet — this is the plan doc that state requires
+before provisioning starts. Do not add a `hosts:` entry with a guessed
+`pve_id` until the LXC is actually created; Proxmox assigns the real ID
+at `pct create` time.
+
+## Why
+
+Week-3 reliability layer (see [OIKOS.md](../OIKOS.md)) wants trend
+signals — "disk full in ~9 days at current rate", temperature creep —
+which need a real time-series store. The scheduler
+([oikos/scheduler.py](../oikos/scheduler.py)) currently does point-in-time
+threshold checks only; Prometheus is the one new piece of infrastructure
+the 30-day roadmap calls for.
+
+## Note: undocumented LXC 131 on hubris
+
+Oikos's drift detector ([oikos/drift.py](../oikos/drift.py)) found
+`pve_id 131` live on hubris (via `pct list`) with no `inventory.yaml`
+entry — created outside the provision-node runbook, identity unknown
+from this repo. **Investigate what 131 is before assuming any pve_id is
+free**; don't let Proxmox auto-assign into a range you haven't confirmed
+is actually unused end-to-end.
+
+## Plan
+
+- **Host:** hubris (per the earlier decision: new LXC, not co-located).
+- **Role:** `metrics` (Prometheus + local TSDB retention; no Grafana yet —
+  the Week-4 Oikos Console renders its own sparklines from the Prometheus
+  HTTP API, per the plan's Week-3 scope decision).
+- **Networking:** LAN + mesh-gated only, no public ingress (matches
+  `homelab_mcp`/`secrets_issuance`'s `MESH_SUBNETS` pattern) — Prometheus
+  exposes host/service metadata that shouldn't be public.
+- **Scrape targets:** node_exporter on hubris and strong (Proxmox hosts)
+  + any LXC the scheduler needs disk/temp trend data from beyond what
+  `pct`/`df` already gives (start with hubris + strong only; expand only
+  if a specific signal needs it).
+- **Storage:** local LXC rootfs is sufficient (metrics-only workload,
+  short retention — no `/mnt/library` mount needed).
+
+## Provisioning steps (once pve_id is assigned)
+
+Follow [runbooks/lifecycle-provision-node.md](../runbooks/lifecycle-provision-node.md):
+
+1. `pct create <new-id> ...` on hubris — confirm the assigned ID doesn't
+   collide with 131 or anything else live.
+2. `homelab client add metrics` (or the chosen name) with `state:
+   provisioning`, stub `containers/<id>-metrics.md`.
+3. Install Prometheus + node_exporter (Debian package or binary release —
+   decide at implementation time; no strong preference recorded here).
+4. Point node_exporter at hubris + strong (either install locally on each,
+   or scrape via SSH-tunneled metrics — install locally is simpler and is
+   the standard approach).
+5. Follow [runbooks/lifecycle-activate-node.md](../runbooks/lifecycle-activate-node.md)
+   to flip to `active`, complete the doc page, regenerate
+   `hosts/*.yaml` + `infrastructure/topology.md`.
+6. Extend `oikos/scheduler.py`'s disk/temp probes to query Prometheus
+   rate() instead of (or alongside) the current live SSH `df` probe, so
+   drift/threshold Signals gain trend evidence ("full in ~9 days") instead
+   of only a point-in-time percentage.
+
+## Open question for the operator
+
+Package choice (apt `prometheus` vs upstream binary release) and exact
+scrape interval aren't decided here — pick at implementation time based
+on what's easiest to keep patched via the existing `homelab apt-audit`/
+`apt-upgrade` fleet tooling if using the Debian package.

secrets/oikos-approval-hmac.yaml

@@ -0,0 +1,25 @@
+hmac_key: ENC[AES256_GCM,data:KEvRp+N/NYHlTJZoehf8cehYMpShnP43s+KWX3OR6p9udA4d1B2lMqXFhrXHxEuWDzLdfOP97AG19DDYgjemPw==,iv:SH8cyI9HP974Af54v0ecwN3ICnX+eBudP6XwB/9nwMU=,tag:VzyLoHy5LRa/5hjL43O2eQ==,type:str]
+sops:
+    age:
+        - enc: |
+            -----BEGIN AGE ENCRYPTED FILE-----
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSAyQ2s4QVJQY1VSZG5JN1NL
+            Vll1MGZNSGIxQ0NJZEFpYUJ4OWxweXFOUmtRCitsODBCTVhMempzYVFpNHpmNWds
+            Qk5SL2pOUGVkbzBtanRmM3BVT1JwYmcKLS0tIGpTV3k5OVF2aXZFRk1JZDVNNk1C
+            VkY1SXpQS2YyM2g3TGpDWG9HdEZqTDAKYhb7yKywJ+jj7ChWCDRmhHsYCASMxBgd
+            hLZV9pmIYCC4jse8q4wJK+tbT+Za5mklXt+j8nZoqm+sA0+MNlnV6g==
+            -----END AGE ENCRYPTED FILE-----
+          recipient: age1xkklkvnk5z0fsnh6cfgv70hy9ksfy8rdprwerzw4yk3p4p7cxcqs2yvpz6
+        - enc: |
+            -----BEGIN AGE ENCRYPTED FILE-----
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBHV0F6RzFiVHF0bS9pNER3
+            elQ1YUorYUsvVC9zWXVBeXYzcThOU3JBbEVZCkxMQlVnMVNtNElibGhUaHhJM0hO
+            cXhFUU5KYzAzQUREMVNJTjhmYlpXekkKLS0tIGhxb3Q3blQ2a29YM3VLdWY2aFgw
+            Qm9HUEl4ZEd0WDVOZG96WXdORjdGVmMKNDNwkB4IF1xcFqG79dnUpHamOzlwQZJt
+            3qfpdXquDeOOomnk7FLlnF/IqyHnhXNGqg+W1XBbZpwvqcAGtTFxQA==
+            -----END AGE ENCRYPTED FILE-----
+          recipient: age1duyl8mkpgu80uv934dy8q7enqjms6yvdz264hme8uryuxmvvqesq6rusq0
+    lastmodified: "2026-07-05T21:11:20Z"
+    mac: ENC[AES256_GCM,data:yz52Q1+49ntOFMESD3QHb5yGE7LB9SfSnHIWeMmYllaYTqR/rcNSXAb/b04ZWvLHqESGLLfZif7gNQYx3Fes0AFnAccowec7IWk/wNY0QoHLA8tZOw2Y1J1Hm32uVOWLl+G6MQybHGt9vHPlz33CLDeB/V07j7WjORqtYjVl8A4=,iv:zvx/bV3WX48OzN3Hi+moL+cELOHaw1SzW/eSCjJ25bg=,tag:DiIZfLyljK+xB3NjZsrwlA==,type:str]
+    unencrypted_suffix: _unencrypted
+    version: 3.13.1