cortex/state.py

373 lines · python

"""Darwin Cortex - LIVE-STATE awareness.
 
Per Darwin's source-of-truth hierarchy, LIVE INFRA is the HIGHEST authority
(live infra > docs). This module builds a compact, structured snapshot of the
organism's REAL current state - pulled live, not from RAG documents - so that
status/state questions ("is white-pc down?", "what's stale?", "what are my
agents doing?", "how many live sessions?") get answered from ground truth.
 
Sources (all reachable from the Cortex host):
  1. Cortex /health (SELF)        - the 10 nodes: 5 engines + 5 data-sources
                                     with per-source freshness. Called in-process
                                     (build_health_snapshot) - no self-HTTP.
  2. Deck   :9893/health + /tree  - active agents, by host/model, what they are.
  3. darwin-term :9894/sessions   - live tmux sessions across ct215/241/247.
 
Every source is wrapped in try/except + a short timeout, so a dead source
degrades the snapshot gracefully and NEVER breaks /ask. The whole snapshot is
cached for a few seconds so repeated /ask calls don't re-probe every time.
 
The snapshot is exposed two ways:
  - as_context_block(): a compact top-authority text block fed to the synthesizer
  - state-intent detection: is_state_query() routes status questions to lead
    with this live snapshot instead of RAG.
"""
from __future__ import annotations
 
import os
import re
import time
from datetime import datetime, timezone
from typing import Any, Optional
 
import httpx
 
DECK_URL = os.environ.get("DECK_URL", "http://REDACTED-IP:9893")
DARWIN_TERM_URL = os.environ.get("DARWIN_TERM_URL", "http://REDACTED-IP:9894")
DARWIN_TERM_TOKEN = os.environ.get("DARWIN_TERM_TOKEN", "")
RAG_BRIDGE_URL = os.environ.get("RAG_BRIDGE_URL", "http://REDACTED-IP:9875")
 
CACHE_TTL_S = float(os.environ.get("DARWIN_STATE_CACHE_TTL", "8"))
_PROBE_TIMEOUT = 3.0
 
STALE_THRESHOLD_H = float(os.environ.get("SOURCE_STALE_THRESHOLD_H", "12"))
 
_cache: dict[str, Any] = {"snapshot": None, "ts": 0.0}
 
 
_STATE_PATTERNS = [
    re.compile(r"\b(is|are|was|were)\b.*\b(up|down|live|stale|running|active|asleep|alive|dead|online|offline|reachable)\b", re.I),
    re.compile(r"\bwhat('?s| is| are)\b.*\b(stale|status|happening|going on|running|active|live|down|up)\b", re.I),
    re.compile(r"\bwhat\b.*\b(agents?|sessions?|engines?|sources?|nodes?|containers?|cts?)\b.*\b(doing|running|active|up|down|live|stale)\b", re.I),
    re.compile(r"\bhow many\b.*\b(sessions?|agents?|engines?|sources?|nodes?|containers?|cts?)\b", re.I),
    re.compile(r"\b(status|health|state)\b.*\b(infra|infrastructure|ingestion|ingest|organism|system|darwin|engines?|sources?|agents?|sessions?)\b", re.I),
    re.compile(r"\bwhat are my\b.*\b(agents?|sessions?|engines?|sources?)\b", re.I),
    re.compile(r"\b(anything|what'?s?)\s+(stale|down|broken|offline|asleep)\b", re.I),
]
 
 
def is_state_query(query: str) -> bool:
    """True when the query is asking about live state/status, not knowledge."""
    if not query:
        return False
    q = query.strip()
    return any(p.search(q) for p in _STATE_PATTERNS)
 
 
async def _build_health_snapshot() -> dict | None:
    """Call the Cortex /health builder in-process (no self-HTTP) so the live
    organism state (engines + data-source freshness, incl. white-pc DOWN and
    ct243-media stale) is part of the snapshot."""
    try:
        from main import health as _health
        resp = await _health()
        engines = []
        for s in resp.engines:
            engines.append({
                "name": s.engine,
                "status": s.status,
                "staleness_h": s.staleness_h,
                "last_ingest": s.last_ingest.isoformat() if s.last_ingest else None,
                "ms": s.ms,
            })
        return {"ok": resp.ok, "engines": engines}
    except Exception:
        return None
 
 
async def _probe_deck() -> dict | None:
    try:
        async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
            h = await client.get(f"{DECK_URL}/health")
            h.raise_for_status()
            health = h.json()
            agents: list[dict] = []
            try:
                t = await client.get(f"{DECK_URL}/tree")
                t.raise_for_status()
                nodes = t.json().get("nodes", [])
                for n in nodes:
                    if not isinstance(n, dict):
                        continue
                    if n.get("status") != "running":
                        continue
                    model = n.get("model", "")
                    if model in ("system", "orchestrator"):
                        continue
                    agents.append({
                        "host": n.get("host"),
                        "model": model,
                        "label": (n.get("label") or "")[:90],
                        "tool": n.get("current_tool"),
                    })
            except Exception:
                pass
            return {
                "active_agents": int(health.get("active_agents", 0)),
                "total_agents": int(health.get("total_agents", 0)),
                "error_agents": int(health.get("error_agents", 0)),
                "running": agents,
            }
    except Exception:
        return None
 
 
async def _probe_ingestion() -> dict | None:
    """Read ALL homelab-rag ingestion sources directly from the bridge /sources.
 
    The Cortex /health organism only surfaces 5 mapped nodes, so a stale source
    that isn't a mapped organism node (e.g. ct243-media, days stale) would never
    appear. This probe captures the FULL ingestion picture so 'what's stale'
    names every stale/down source by its real source_id.
    """
    try:
        async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
            r = await client.get(f"{RAG_BRIDGE_URL}/sources")
            r.raise_for_status()
            payload = r.json()
        sources = payload.get("sources", []) if isinstance(payload, dict) else []
        total = 0
        stale: list[dict] = []
        freshest = None
        for s in sources:
            if not isinstance(s, dict):
                continue
            total += 1
            sh = s.get("staleness_h")
            if sh is not None:
                if freshest is None or sh < freshest:
                    freshest = sh
                if sh > STALE_THRESHOLD_H:
                    stale.append({"id": s.get("source_id"), "staleness_h": sh})
        stale.sort(key=lambda x: -(x["staleness_h"] or 0))
        return {"total": total, "stale": stale, "freshest_h": freshest}
    except Exception:
        return None
 
 
async def _probe_term() -> dict | None:
    if not DARWIN_TERM_TOKEN:
        return None
    try:
        async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
            r = await client.get(
                f"{DARWIN_TERM_URL}/sessions",
                params={"token": DARWIN_TERM_TOKEN},
            )
            r.raise_for_status()
            sessions = r.json().get("sessions", [])
            by_host: dict[str, int] = {}
            by_agent: dict[str, int] = {}
            for s in sessions:
                if not isinstance(s, dict):
                    continue
                by_host[s.get("target", "?")] = by_host.get(s.get("target", "?"), 0) + 1
                by_agent[s.get("agent", "?")] = by_agent.get(s.get("agent", "?"), 0) + 1
            return {
                "count": len(sessions),
                "by_host": by_host,
                "by_agent": by_agent,
            }
    except Exception:
        return None
 
 
async def get_snapshot() -> dict:
    """Return the live-state snapshot, cached for CACHE_TTL_S seconds."""
    now = time.monotonic()
    snap = _cache.get("snapshot")
    if snap is not None and (now - _cache["ts"]) < CACHE_TTL_S:
        return snap
 
    import asyncio
    health, deck, term, ingestion = await asyncio.gather(
        _build_health_snapshot(), _probe_deck(), _probe_term(), _probe_ingestion(),
        return_exceptions=True,
    )
    snap = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "health": health if isinstance(health, dict) else None,
        "deck": deck if isinstance(deck, dict) else None,
        "term": term if isinstance(term, dict) else None,
        "ingestion": ingestion if isinstance(ingestion, dict) else None,
    }
    _cache["snapshot"] = snap
    _cache["ts"] = now
    return snap
 
 
def _render_health(health: dict | None) -> list[str]:
    if not health:
        return ["Organism /health: UNAVAILABLE (could not read live health)."]
    lines: list[str] = []
    engines = health.get("engines", [])
    up = [e for e in engines if e.get("status") == "up"]
    down = [e for e in engines if e.get("status") != "up"]
    stale = [
        e for e in engines
        if e.get("staleness_h") is not None and e["staleness_h"] > STALE_THRESHOLD_H
    ]
    lines.append(
        f"Organism: {len(up)}/{len(engines)} nodes up. "
        f"Down: {', '.join(e['name'] for e in down) if down else 'none'}."
    )
    if stale:
        stale_desc = ", ".join(
            f"{e['name']} (~{e['staleness_h']:.0f}h stale)" for e in stale
        )
        lines.append(f"Stale data-sources (>{STALE_THRESHOLD_H:.0f}h): {stale_desc}.")
    else:
        lines.append("No data-source is stale beyond threshold.")
    detail = []
    for e in engines:
        bit = e["name"] + "=" + e["status"]
        if e.get("staleness_h") is not None:
            bit += f"({e['staleness_h']:.1f}h)"
        detail.append(bit)
    lines.append("Nodes: " + ", ".join(detail) + ".")
    return lines
 
 
def _render_ingestion(ing: dict | None) -> list[str]:
    if not ing:
        return ["Ingestion sources: UNAVAILABLE."]
    stale = ing.get("stale") or []
    total = ing.get("total", 0)
    if not stale:
        return [f"Ingestion: all {total} sources fresh (freshest ~{ing.get('freshest_h')}h)."]
    desc = ", ".join(
        f"{s['id']} (~{s['staleness_h']:.0f}h stale)" for s in stale
    )
    return [f"Ingestion: {len(stale)}/{total} sources STALE (>{STALE_THRESHOLD_H:.0f}h): {desc}."]
 
 
def _render_deck(deck: dict | None) -> list[str]:
    if not deck:
        return ["Agents (Deck): UNAVAILABLE."]
    lines = [
        f"Active agents right now: {deck.get('active_agents', 0)} "
        f"(total ever: {deck.get('total_agents', 0)}, errored: {deck.get('error_agents', 0)})."
    ]
    running = deck.get("running") or []
    if running:
        for a in running[:6]:
            label = a.get("label") or "(no label)"
            tool = f", using {a['tool']}" if a.get("tool") else ""
            lines.append(
                f"  - {a.get('host', '?')}/{a.get('model', '?')}: {label}{tool}"
            )
    return lines
 
 
def _render_term(term: dict | None) -> list[str]:
    if not term:
        return ["Live tmux sessions (darwin-term): UNAVAILABLE."]
    by_host = ", ".join(f"{h}:{n}" for h, n in (term.get("by_host") or {}).items())
    by_agent = ", ".join(f"{a}:{n}" for a, n in (term.get("by_agent") or {}).items())
    return [
        f"Live tmux sessions: {term.get('count', 0)} total"
        + (f" (by host: {by_host})" if by_host else "")
        + (f" (by type: {by_agent})" if by_agent else "")
        + "."
    ]
 
 
def render_block(snap: dict) -> str:
    """Compact human/LLM-readable live-state block (top authority)."""
    lines = ["LIVE STATE (authoritative - live infra outranks docs):"]
    lines += _render_health(snap.get("health"))
    lines += _render_ingestion(snap.get("ingestion"))
    lines += _render_deck(snap.get("deck"))
    lines += _render_term(snap.get("term"))
    return "\n".join(lines)
 
 
async def as_context_block() -> str:
    """Convenience: snapshot -> rendered block."""
    return render_block(await get_snapshot())
 
 
def deterministic_state_answer(query: str, snap: dict) -> str:
    """A direct, LLM-free answer for state questions, built from the snapshot.
 
    Used (a) as the deterministic fallback when the LLM is cold, and (b) to LEAD
    the answer for state-intent queries. Tries to answer the specific question;
    always appends the compact state block as evidence.
    """
    q = query.lower()
    health = snap.get("health") or {}
    engines = {e["name"]: e for e in (health.get("engines") or [])}
    lead: Optional[str] = None
 
    alias = {
        "white pc": "white-pc", "white-pc": "white-pc", "whitepc": "white-pc",
        "3060": "3060", "rtx": "3060", "gpu": "3060",
        "proxmox": "proxmox", "pve": "proxmox",
        "browser": "browser-history", "browser history": "browser-history",
    }
    target = None
    for k, v in alias.items():
        if k in q:
            target = v
            break
    if target and target in engines:
        e = engines[target]
        if e["status"] != "up":
            lead = f"Yes - {target} is DOWN (not ingesting / no live data)."
        else:
            sh = e.get("staleness_h")
            fresh = f"last ingest ~{sh:.1f}h ago" if sh is not None else "live"
            lead = f"No - {target} is UP ({fresh})."
 
    if lead is None and ("stale" in q or "out of date" in q or "outdated" in q):
        ing = snap.get("ingestion") or {}
        stale_src = ing.get("stale") or []
        down = [e for e in (health.get("engines") or []) if e.get("status") != "up"]
        parts = []
        if stale_src:
            parts.append(
                "stale ingestion: " + ", ".join(
                    f"{s['id']} (~{s['staleness_h']:.0f}h)" for s in stale_src
                )
            )
        if down:
            parts.append("down/not-ingesting: " + ", ".join(e["name"] for e in down))
        lead = ("What's stale right now - " + "; ".join(parts) + ".") if parts \
            else "Nothing is stale right now; all data-sources are fresh."
 
    if lead is None and "session" in q:
        term = snap.get("term")
        if term:
            by_host = ", ".join(f"{h}:{n}" for h, n in (term.get("by_host") or {}).items())
            lead = (
                f"You have {term.get('count', 0)} live tmux sessions"
                + (f" ({by_host})" if by_host else "") + "."
            )
 
    if lead is None and "agent" in q:
        deck = snap.get("deck")
        if deck:
            running = deck.get("running") or []
            if running:
                summ = "; ".join(
                    f"{a.get('host', '?')}/{a.get('model', '?')} - {(a.get('label') or '')[:60]}"
                    for a in running[:4]
                )
                lead = f"{deck.get('active_agents', 0)} agents active right now: {summ}."
            else:
                lead = f"{deck.get('active_agents', 0)} agents active right now."
 
    block = render_block(snap)
    if lead:
        return f"{lead}\n\n{block}"
    return block

1	"""Darwin Cortex - LIVE-STATE awareness.
2
3	Per Darwin's source-of-truth hierarchy, LIVE INFRA is the HIGHEST authority
4	(live infra > docs). This module builds a compact, structured snapshot of the
5	organism's REAL current state - pulled live, not from RAG documents - so that
6	status/state questions ("is white-pc down?", "what's stale?", "what are my
7	agents doing?", "how many live sessions?") get answered from ground truth.
8
9	Sources (all reachable from the Cortex host):
10	1. Cortex /health (SELF) - the 10 nodes: 5 engines + 5 data-sources
11	with per-source freshness. Called in-process
12	(build_health_snapshot) - no self-HTTP.
13	2. Deck :9893/health + /tree - active agents, by host/model, what they are.
14	3. darwin-term :9894/sessions - live tmux sessions across ct215/241/247.
15
16	Every source is wrapped in try/except + a short timeout, so a dead source
17	degrades the snapshot gracefully and NEVER breaks /ask. The whole snapshot is
18	cached for a few seconds so repeated /ask calls don't re-probe every time.
19
20	The snapshot is exposed two ways:
21	- as_context_block(): a compact top-authority text block fed to the synthesizer
22	- state-intent detection: is_state_query() routes status questions to lead
23	with this live snapshot instead of RAG.
24	"""
25	from __future__ import annotations
26
27	import os
28	import re
29	import time
30	from datetime import datetime, timezone
31	from typing import Any, Optional
32
33	import httpx
34
35	DECK_URL = os.environ.get("DECK_URL", "http://REDACTED-IP:9893")
36	DARWIN_TERM_URL = os.environ.get("DARWIN_TERM_URL", "http://REDACTED-IP:9894")
37	DARWIN_TERM_TOKEN = os.environ.get("DARWIN_TERM_TOKEN", "")
38	RAG_BRIDGE_URL = os.environ.get("RAG_BRIDGE_URL", "http://REDACTED-IP:9875")
39
40	CACHE_TTL_S = float(os.environ.get("DARWIN_STATE_CACHE_TTL", "8"))
41	_PROBE_TIMEOUT = 3.0
42
43	STALE_THRESHOLD_H = float(os.environ.get("SOURCE_STALE_THRESHOLD_H", "12"))
44
45	_cache: dict[str, Any] = {"snapshot": None, "ts": 0.0}
46
47
48	_STATE_PATTERNS = [
49	re.compile(r"\b(is\|are\|was\|were)\b.*\b(up\|down\|live\|stale\|running\|active\|asleep\|alive\|dead\|online\|offline\|reachable)\b", re.I),
50	re.compile(r"\bwhat('?s\| is\| are)\b.*\b(stale\|status\|happening\|going on\|running\|active\|live\|down\|up)\b", re.I),
51	re.compile(r"\bwhat\b.\b(agents?\|sessions?\|engines?\|sources?\|nodes?\|containers?\|cts?)\b.\b(doing\|running\|active\|up\|down\|live\|stale)\b", re.I),
52	re.compile(r"\bhow many\b.*\b(sessions?\|agents?\|engines?\|sources?\|nodes?\|containers?\|cts?)\b", re.I),
53	re.compile(r"\b(status\|health\|state)\b.*\b(infra\|infrastructure\|ingestion\|ingest\|organism\|system\|darwin\|engines?\|sources?\|agents?\|sessions?)\b", re.I),
54	re.compile(r"\bwhat are my\b.*\b(agents?\|sessions?\|engines?\|sources?)\b", re.I),
55	re.compile(r"\b(anything\|what'?s?)\s+(stale\|down\|broken\|offline\|asleep)\b", re.I),
56	]
57
58
59	def is_state_query(query: str) -> bool:
60	"""True when the query is asking about live state/status, not knowledge."""
61	if not query:
62	return False
63	q = query.strip()
64	return any(p.search(q) for p in _STATE_PATTERNS)
65
66
67	async def _build_health_snapshot() -> dict \| None:
68	"""Call the Cortex /health builder in-process (no self-HTTP) so the live
69	organism state (engines + data-source freshness, incl. white-pc DOWN and
70	ct243-media stale) is part of the snapshot."""
71	try:
72	from main import health as _health
73	resp = await _health()
74	engines = []
75	for s in resp.engines:
76	engines.append({
77	"name": s.engine,
78	"status": s.status,
79	"staleness_h": s.staleness_h,
80	"last_ingest": s.last_ingest.isoformat() if s.last_ingest else None,
81	"ms": s.ms,
82	})
83	return {"ok": resp.ok, "engines": engines}
84	except Exception:
85	return None
86
87
88	async def _probe_deck() -> dict \| None:
89	try:
90	async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
91	h = await client.get(f"{DECK_URL}/health")
92	h.raise_for_status()
93	health = h.json()
94	agents: list[dict] = []
95	try:
96	t = await client.get(f"{DECK_URL}/tree")
97	t.raise_for_status()
98	nodes = t.json().get("nodes", [])
99	for n in nodes:
100	if not isinstance(n, dict):
101	continue
102	if n.get("status") != "running":
103	continue
104	model = n.get("model", "")
105	if model in ("system", "orchestrator"):
106	continue
107	agents.append({
108	"host": n.get("host"),
109	"model": model,
110	"label": (n.get("label") or "")[:90],
111	"tool": n.get("current_tool"),
112	})
113	except Exception:
114	pass
115	return {
116	"active_agents": int(health.get("active_agents", 0)),
117	"total_agents": int(health.get("total_agents", 0)),
118	"error_agents": int(health.get("error_agents", 0)),
119	"running": agents,
120	}
121	except Exception:
122	return None
123
124
125	async def _probe_ingestion() -> dict \| None:
126	"""Read ALL homelab-rag ingestion sources directly from the bridge /sources.
127
128	The Cortex /health organism only surfaces 5 mapped nodes, so a stale source
129	that isn't a mapped organism node (e.g. ct243-media, days stale) would never
130	appear. This probe captures the FULL ingestion picture so 'what's stale'
131	names every stale/down source by its real source_id.
132	"""
133	try:
134	async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
135	r = await client.get(f"{RAG_BRIDGE_URL}/sources")
136	r.raise_for_status()
137	payload = r.json()
138	sources = payload.get("sources", []) if isinstance(payload, dict) else []
139	total = 0
140	stale: list[dict] = []
141	freshest = None
142	for s in sources:
143	if not isinstance(s, dict):
144	continue
145	total += 1
146	sh = s.get("staleness_h")
147	if sh is not None:
148	if freshest is None or sh < freshest:
149	freshest = sh
150	if sh > STALE_THRESHOLD_H:
151	stale.append({"id": s.get("source_id"), "staleness_h": sh})
152	stale.sort(key=lambda x: -(x["staleness_h"] or 0))
153	return {"total": total, "stale": stale, "freshest_h": freshest}
154	except Exception:
155	return None
156
157
158	async def _probe_term() -> dict \| None:
159	if not DARWIN_TERM_TOKEN:
160	return None
161	try:
162	async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT) as client:
163	r = await client.get(
164	f"{DARWIN_TERM_URL}/sessions",
165	params={"token": DARWIN_TERM_TOKEN},
166	)
167	r.raise_for_status()
168	sessions = r.json().get("sessions", [])
169	by_host: dict[str, int] = {}
170	by_agent: dict[str, int] = {}
171	for s in sessions:
172	if not isinstance(s, dict):
173	continue
174	by_host[s.get("target", "?")] = by_host.get(s.get("target", "?"), 0) + 1
175	by_agent[s.get("agent", "?")] = by_agent.get(s.get("agent", "?"), 0) + 1
176	return {
177	"count": len(sessions),
178	"by_host": by_host,
179	"by_agent": by_agent,
180	}
181	except Exception:
182	return None
183
184
185	async def get_snapshot() -> dict:
186	"""Return the live-state snapshot, cached for CACHE_TTL_S seconds."""
187	now = time.monotonic()
188	snap = _cache.get("snapshot")
189	if snap is not None and (now - _cache["ts"]) < CACHE_TTL_S:
190	return snap
191
192	import asyncio
193	health, deck, term, ingestion = await asyncio.gather(
194	_build_health_snapshot(), _probe_deck(), _probe_term(), _probe_ingestion(),
195	return_exceptions=True,
196	)
197	snap = {
198	"ts": datetime.now(timezone.utc).isoformat(),
199	"health": health if isinstance(health, dict) else None,
200	"deck": deck if isinstance(deck, dict) else None,
201	"term": term if isinstance(term, dict) else None,
202	"ingestion": ingestion if isinstance(ingestion, dict) else None,
203	}
204	_cache["snapshot"] = snap
205	_cache["ts"] = now
206	return snap
207
208
209	def _render_health(health: dict \| None) -> list[str]:
210	if not health:
211	return ["Organism /health: UNAVAILABLE (could not read live health)."]
212	lines: list[str] = []
213	engines = health.get("engines", [])
214	up = [e for e in engines if e.get("status") == "up"]
215	down = [e for e in engines if e.get("status") != "up"]
216	stale = [
217	e for e in engines
218	if e.get("staleness_h") is not None and e["staleness_h"] > STALE_THRESHOLD_H
219	]
220	lines.append(
221	f"Organism: {len(up)}/{len(engines)} nodes up. "
222	f"Down: {', '.join(e['name'] for e in down) if down else 'none'}."
223	)
224	if stale:
225	stale_desc = ", ".join(
226	f"{e['name']} (~{e['staleness_h']:.0f}h stale)" for e in stale
227	)
228	lines.append(f"Stale data-sources (>{STALE_THRESHOLD_H:.0f}h): {stale_desc}.")
229	else:
230	lines.append("No data-source is stale beyond threshold.")
231	detail = []
232	for e in engines:
233	bit = e["name"] + "=" + e["status"]
234	if e.get("staleness_h") is not None:
235	bit += f"({e['staleness_h']:.1f}h)"
236	detail.append(bit)
237	lines.append("Nodes: " + ", ".join(detail) + ".")
238	return lines
239
240
241	def _render_ingestion(ing: dict \| None) -> list[str]:
242	if not ing:
243	return ["Ingestion sources: UNAVAILABLE."]
244	stale = ing.get("stale") or []
245	total = ing.get("total", 0)
246	if not stale:
247	return [f"Ingestion: all {total} sources fresh (freshest ~{ing.get('freshest_h')}h)."]
248	desc = ", ".join(
249	f"{s['id']} (~{s['staleness_h']:.0f}h stale)" for s in stale
250	)
251	return [f"Ingestion: {len(stale)}/{total} sources STALE (>{STALE_THRESHOLD_H:.0f}h): {desc}."]
252
253
254	def _render_deck(deck: dict \| None) -> list[str]:
255	if not deck:
256	return ["Agents (Deck): UNAVAILABLE."]
257	lines = [
258	f"Active agents right now: {deck.get('active_agents', 0)} "
259	f"(total ever: {deck.get('total_agents', 0)}, errored: {deck.get('error_agents', 0)})."
260	]
261	running = deck.get("running") or []
262	if running:
263	for a in running[:6]:
264	label = a.get("label") or "(no label)"
265	tool = f", using {a['tool']}" if a.get("tool") else ""
266	lines.append(
267	f" - {a.get('host', '?')}/{a.get('model', '?')}: {label}{tool}"
268	)
269	return lines
270
271
272	def _render_term(term: dict \| None) -> list[str]:
273	if not term:
274	return ["Live tmux sessions (darwin-term): UNAVAILABLE."]
275	by_host = ", ".join(f"{h}:{n}" for h, n in (term.get("by_host") or {}).items())
276	by_agent = ", ".join(f"{a}:{n}" for a, n in (term.get("by_agent") or {}).items())
277	return [
278	f"Live tmux sessions: {term.get('count', 0)} total"
279	+ (f" (by host: {by_host})" if by_host else "")
280	+ (f" (by type: {by_agent})" if by_agent else "")
281	+ "."
282	]
283
284
285	def render_block(snap: dict) -> str:
286	"""Compact human/LLM-readable live-state block (top authority)."""
287	lines = ["LIVE STATE (authoritative - live infra outranks docs):"]
288	lines += _render_health(snap.get("health"))
289	lines += _render_ingestion(snap.get("ingestion"))
290	lines += _render_deck(snap.get("deck"))
291	lines += _render_term(snap.get("term"))
292	return "\n".join(lines)
293
294
295	async def as_context_block() -> str:
296	"""Convenience: snapshot -> rendered block."""
297	return render_block(await get_snapshot())
298
299
300	def deterministic_state_answer(query: str, snap: dict) -> str:
301	"""A direct, LLM-free answer for state questions, built from the snapshot.
302
303	Used (a) as the deterministic fallback when the LLM is cold, and (b) to LEAD
304	the answer for state-intent queries. Tries to answer the specific question;
305	always appends the compact state block as evidence.
306	"""
307	q = query.lower()
308	health = snap.get("health") or {}
309	engines = {e["name"]: e for e in (health.get("engines") or [])}
310	lead: Optional[str] = None
311
312	alias = {
313	"white pc": "white-pc", "white-pc": "white-pc", "whitepc": "white-pc",
314	"3060": "3060", "rtx": "3060", "gpu": "3060",
315	"proxmox": "proxmox", "pve": "proxmox",
316	"browser": "browser-history", "browser history": "browser-history",
317	}
318	target = None
319	for k, v in alias.items():
320	if k in q:
321	target = v
322	break
323	if target and target in engines:
324	e = engines[target]
325	if e["status"] != "up":
326	lead = f"Yes - {target} is DOWN (not ingesting / no live data)."
327	else:
328	sh = e.get("staleness_h")
329	fresh = f"last ingest ~{sh:.1f}h ago" if sh is not None else "live"
330	lead = f"No - {target} is UP ({fresh})."
331
332	if lead is None and ("stale" in q or "out of date" in q or "outdated" in q):
333	ing = snap.get("ingestion") or {}
334	stale_src = ing.get("stale") or []
335	down = [e for e in (health.get("engines") or []) if e.get("status") != "up"]
336	parts = []
337	if stale_src:
338	parts.append(
339	"stale ingestion: " + ", ".join(
340	f"{s['id']} (~{s['staleness_h']:.0f}h)" for s in stale_src
341	)
342	)
343	if down:
344	parts.append("down/not-ingesting: " + ", ".join(e["name"] for e in down))
345	lead = ("What's stale right now - " + "; ".join(parts) + ".") if parts \
346	else "Nothing is stale right now; all data-sources are fresh."
347
348	if lead is None and "session" in q:
349	term = snap.get("term")
350	if term:
351	by_host = ", ".join(f"{h}:{n}" for h, n in (term.get("by_host") or {}).items())
352	lead = (
353	f"You have {term.get('count', 0)} live tmux sessions"
354	+ (f" ({by_host})" if by_host else "") + "."
355	)
356
357	if lead is None and "agent" in q:
358	deck = snap.get("deck")
359	if deck:
360	running = deck.get("running") or []
361	if running:
362	summ = "; ".join(
363	f"{a.get('host', '?')}/{a.get('model', '?')} - {(a.get('label') or '')[:60]}"
364	for a in running[:4]
365	)
366	lead = f"{deck.get('active_agents', 0)} agents active right now: {summ}."
367	else:
368	lead = f"{deck.get('active_agents', 0)} agents active right now."
369
370	block = render_block(snap)
371	if lead:
372	return f"{lead}\n\n{block}"
373	return block