diff --git a/loopx/capabilities/decision_context/README.md b/loopx/capabilities/decision_context/README.md index 03c079352..95fe5cf85 100644 --- a/loopx/capabilities/decision_context/README.md +++ b/loopx/capabilities/decision_context/README.md @@ -422,12 +422,38 @@ available instead of waiting an additional scan interval. `capture-status` separates active `pending_batch_count`, unresolved `held_batch_count`, per-source `acquisition_held` and `semantic_review_completion=not_inferred_from_capture`. `last_checked_at` is -the last attempt, not necessarily a successful scan; host service liveness and -successful-scan timestamps remain separate. No status-only call proves historical -replay or complete decision coverage. Disable capture using the existing profile +the last attempt, not necessarily a successful scan. No status-only call proves +historical replay or complete decision coverage. Disable capture using the existing profile switch; stop the scheduler before downgrading, since older runtimes do not honor recovery holds. Retain the spool/receipts rather than treating downgrade as rollback. +#### Source freshness contract + +An enabled profile, a healthy `loopx doctor` or a settled projection never +implies fresh sources. Every `prepare-evidence` / `prepare-review` assembly and +every `capture` / `capture-status` result carries `source_freshness` +(`decision_source_freshness_v0`): one row per enabled source with +`last_read_at` (last *successful* read), `staleness_seconds`, the source +`freshness_seconds` window, `status` (`fresh`, `stale`, `never_read`, +`not_scanned`), `failure_streak` and `alert_reasons`. Enabled sources outside +the current scan (for example on-demand sources) appear as `not_scanned` +instead of disappearing. Markdown output marks every alerted row with 🔴. +Consumers must disclose alerted sources before presenting a conclusion as current. + +A failed provider attempt updates `last_checked_at` and increments +`failure_streak`, but never advances `last_read_at`. Existing spools migrate in +place; a legacy row whose last attempt succeeded uses that attempt as its last read. + +`loopx decision-context capture --execute` records a local host health file +under `/decision-context/capture-hosts/`. Private hosts calling +`capture_profile_sources` should pass `health_runtime_root` for the same effect. +`loopx doctor` reports the optional `decision_context_capture_hosts_healthy` +check without opening private spools. It alerts when a registered host has not +ticked within `max(2 × interval, interval + 600s)` (for example a scheduler still +pointing at a deleted checkout), when the last tick failed, when the spool is +gone, or when a recorded source is stale or failing. Remove the record of a +deliberately retired host. + ## Relationship To Other Capabilities | Capability | Primary question | Relationship | diff --git a/loopx/capabilities/decision_context/README.zh-CN.md b/loopx/capabilities/decision_context/README.zh-CN.md index 8de9c26b6..de85e1756 100644 --- a/loopx/capabilities/decision_context/README.zh-CN.md +++ b/loopx/capabilities/decision_context/README.zh-CN.md @@ -361,11 +361,33 @@ python3 -m pytest -q tests/capabilities/test_decision_context_capture.py `capture-status` 分开报告 active pending、held 历史、每来源 acquisition hold， 并明确 `semantic_review_completion=not_inferred_from_capture`。`last_checked_at` -是尝试时间，不保证成功；服务存活和最近成功扫描时间仍由 host 独立报告。 +是尝试时间，不保证成功。 仅看 status 不能证明历史可重放或决策覆盖完整。停用仍使用原 profile 开关； 降级旧版本前必须停止调度器，因为旧运行时不认识 recovery hold。 保留 spool 与回执，不能把软件降级当成状态回滚。 +#### 来源新鲜度合同 + +profile 已启用、`loopx doctor` 健康或已有结算投影，都不代表来源是新鲜的。 +每个 `prepare-evidence` / `prepare-review` 组装结果，以及每次 `capture` / +`capture-status` 输出，都携带 `source_freshness`（`decision_source_freshness_v0`）： +每个已启用来源一行，包含 `last_read_at`（最近一次**成功**读取）、`staleness_seconds`、 +来源的 `freshness_seconds` 窗口、`status`（`fresh`、`stale`、`never_read`、 +`not_scanned`）、`failure_streak` 和 `alert_reasons`。不在本次扫描范围内的已启用来源 +（例如按需来源）以 `not_scanned` 出现，不会被静默省略。Markdown 输出对所有告警行标 🔴。 +消费方在把结论当作"当前情况"之前，必须先披露告警来源。 + +provider 读取失败会更新 `last_checked_at` 并累加 `failure_streak`，但绝不推进 +`last_read_at`。已有 spool 原地迁移；旧记录若最后一次尝试成功，则以该次尝试作为最近读取。 + +`loopx decision-context capture --execute` 会在 +`/decision-context/capture-hosts/` 下写入本机 host 健康记录。 +直接调用 `capture_profile_sources` 的私有 host 应传入 `health_runtime_root` 获得同样效果。 +`loopx doctor` 不打开私有 spool，以可选检查 `decision_context_capture_hosts_healthy` +报告：已登记 host 超过 `max(2 × interval, interval + 600s)` 未 tick（例如调度器仍指向 +已删除的 checkout）、最后一次 tick 失败、spool 丢失，或记录中的来源陈旧／持续失败时告警。 +主动退役的 host 需删除其记录。 + ## 与其他能力的关系 | 能力 | 核心问题 | 与 Decision Context 的关系 | diff --git a/loopx/capabilities/decision_context/assembler.py b/loopx/capabilities/decision_context/assembler.py index 8f7f7ab8b..3a39e81c0 100644 --- a/loopx/capabilities/decision_context/assembler.py +++ b/loopx/capabilities/decision_context/assembler.py @@ -16,6 +16,7 @@ canonical_context_matches, opaque_provider_ref, ) +from .freshness import build_source_freshness_report, source_freshness_row from .packets import build_decision_evidence_packet from .sources import ( DecisionSourceExactRead, @@ -451,6 +452,47 @@ def _verified_changed_facts( return verified +def _assembly_source_freshness( + *, + collected_sources: Sequence[_CollectedSource], + coverage_sources: Sequence[DecisionSourceSpec], + assembly_time: datetime, +) -> dict[str, Any]: + collected_by_id = { + collected.source.source_id: collected for collected in collected_sources + } + rows = [] + for source in coverage_sources: + collected = collected_by_id.get(source.source_id) + if collected is None: + rows.append( + source_freshness_row( + source=source, + observed_at=assembly_time, + last_read_at=None, + last_attempt_status=None, + scanned=False, + ) + ) + continue + attempt = ( + "exact_read_failed" if collected.exact_read_failed else collected.scan.status + ) + rows.append( + source_freshness_row( + source=source, + observed_at=assembly_time, + last_read_at=( + assembly_time.isoformat() + if attempt in {"completed", "no_change"} + else None + ), + last_attempt_status=attempt, + ) + ) + return build_source_freshness_report(observed_at=assembly_time, rows=rows) + + def _accounted_authority( evidence: Mapping[str, Any], ) -> tuple[set[tuple[str, str]], set[str]]: @@ -558,12 +600,16 @@ def assemble_decision_evidence( recall_query_summary: str = "current decision evidence", recall_limit: int = 5, timeout_seconds: float = 10.0, + coverage_sources: Sequence[DecisionSourceSpec] | None = None, ) -> DecisionContextAssembly: """Collect, rebase, and assemble one public-safe decision evidence packet. Cursor values are returned only as private proposals. A caller may persist them after a reviewed proposal or explicit semantic no-change result has been written back and validated. Later outcome observation is separate. + ``coverage_sources`` lists every enabled source the freshness projection + must account for; enabled sources outside this scan are reported as + ``not_scanned`` instead of being silently omitted. """ assembly_time = _timestamp(observed_at, field_name="observed_at") @@ -812,6 +858,18 @@ def assemble_decision_evidence( "source_manifest": manifest, "source_scan_receipts": source_scan_receipts, "source_coverage": _source_coverage(collected_sources), + "source_freshness": _assembly_source_freshness( + collected_sources=collected_sources, + coverage_sources=sorted( + { + source.source_id: source + for source in (*(coverage_sources or ()), *enabled_sources) + if source.enabled + }.values(), + key=lambda source: source.source_id, + ), + assembly_time=assembly_time, + ), "context_retrieval_receipt": retrieval_receipt, "evidence_packet": evidence, "semantic_rebase": { diff --git a/loopx/capabilities/decision_context/capture.py b/loopx/capabilities/decision_context/capture.py index 5994517f7..a525192eb 100644 --- a/loopx/capabilities/decision_context/capture.py +++ b/loopx/capabilities/decision_context/capture.py @@ -22,11 +22,19 @@ DecisionEvidenceRebaser, assemble_decision_evidence, ) +from .freshness import ( + build_source_freshness_report, + source_freshness_row, + write_capture_host_health, +) from .private_state import load_private_decision_cursors, private_file_digest from .profile import DecisionContextProfile, resolve_decision_context_activation from .runtime import _build_source_providers from .sources import DecisionSourceProvider, DecisionSourceSpec +_READ_SUCCESS_STATUSES = frozenset({"completed", "no_change"}) +_READ_FAILURE_STATUSES = frozenset({"provider_failed", "failed", "unavailable"}) + class CaptureReplayError(ValueError): """Typed recovery diagnosis; never classify provider exception prose.""" @@ -61,6 +69,19 @@ def _open_spool(path: Path, *, goal_id: str, agent_id: str) -> sqlite3.Connectio source_id TEXT PRIMARY KEY, cursor TEXT); """) db.execute("BEGIN IMMEDIATE") + columns = _source_columns(db) + if "last_success_at" not in columns: + # Legacy spools only knew the last attempt; a successful last + # attempt is the best available evidence of the last read. + db.execute("ALTER TABLE sources ADD COLUMN last_success_at TEXT") + db.execute( + "UPDATE sources SET last_success_at=checked_at " + "WHERE status IN ('completed','no_change')" + ) + if "failure_streak" not in columns: + db.execute( + "ALTER TABLE sources ADD COLUMN failure_streak INTEGER NOT NULL DEFAULT 0" + ) identity = db.execute("SELECT goal, agent FROM identity").fetchone() if identity is None: db.execute("INSERT INTO identity VALUES (?, ?)", (goal_id, agent_id)) @@ -83,23 +104,62 @@ def _binding_digest(profile: DecisionContextProfile, source: DecisionSourceSpec) return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest() -def _status(db: sqlite3.Connection, source_ids: tuple[str, ...]) -> dict[str, Any]: +def _source_columns(db: sqlite3.Connection) -> set[str]: + return {str(row[1]) for row in db.execute("PRAGMA table_info(sources)")} + + +def _status( + db: sqlite3.Connection, + sources: tuple[DecisionSourceSpec, ...], + *, + now: datetime, +) -> dict[str, Any]: has_recovery = ( db.execute("SELECT 1 FROM sqlite_master WHERE name='capture_holds'").fetchone() is not None ) + columns = _source_columns(db) rows = [] - for source_id in source_ids: + freshness_rows = [] + for spec in sources: + source_id = spec.source_id source = db.execute( - "SELECT checked_at, status FROM sources WHERE source_id=?", (source_id,) + "SELECT * FROM sources WHERE source_id=?", (source_id,) ).fetchone() pending = db.execute( "SELECT count(*), min(id) FROM batches WHERE source_id=?", (source_id,) ).fetchone() + if source is None: + last_read_at = None + elif "last_success_at" in columns: + last_read_at = source["last_success_at"] + else: + last_read_at = ( + source["checked_at"] + if source["status"] in _READ_SUCCESS_STATUSES + else None + ) + failure_streak = ( + int(source["failure_streak"] or 0) + if source is not None and "failure_streak" in columns + else 0 + ) + freshness = source_freshness_row( + source=spec, + observed_at=now, + last_read_at=last_read_at, + last_attempt_status=source["status"] if source else None, + failure_streak=failure_streak, + ) + freshness_rows.append(freshness) rows.append( { "source_id": source_id, "last_checked_at": source["checked_at"] if source else None, + "last_read_at": freshness["last_read_at"], + "staleness_seconds": freshness["staleness_seconds"], + "freshness": freshness["status"], + "failure_streak": failure_streak, "status": source["status"] if source else "never_checked", "pending_batch_count": pending[0], "next_batch_id": pending[1], @@ -120,6 +180,9 @@ def _status(db: sqlite3.Connection, source_ids: tuple[str, ...]) -> dict[str, An return { "schema_version": "decision_context_capture_status_v0", "sources": rows, + "source_freshness": build_source_freshness_report( + observed_at=now, rows=freshness_rows + ), "pending_batch_count": db.execute("SELECT count(*) FROM batches").fetchone()[0], "held_batch_count": db.execute("SELECT count(*) FROM held_batches").fetchone()[ 0 @@ -144,6 +207,7 @@ def capture_profile_sources( cursor_path: Path | None = None, execute: bool = False, timeout_seconds: float = 20.0, + health_runtime_root: Path | None = None, ) -> dict[str, Any]: """Run a bounded tick. Disabled profiles and preview never create a spool. @@ -151,6 +215,9 @@ def capture_profile_sources( A newly observed review transition may retire only the oldest matching batch. Ambiguous or unobserved transitions retain batches; capture never writes review. Provider deadlines are cooperative, so hosts must also bound process runtime. + With ``health_runtime_root``, every executed tick (including a failed one) + refreshes the host health record that ``loopx doctor`` inspects; a host that + stops ticking is then reported by heartbeat age rather than going silent. """ if isinstance(timeout_seconds, bool) or not 0 < timeout_seconds <= 60: raise ValueError("capture timeout must be between 0 and 60 seconds") @@ -184,6 +251,12 @@ def capture_profile_sources( ) if not execute and not spool_path.exists(): return {"activation": activation, "status": "not_started", "executed": False} + now = datetime.now(timezone.utc) + sources = tuple( + source + for source in profile.sources + if source.source_id in profile.capture_source_ids + ) if not execute: # Read-only diagnostics do not create tables, change permissions or retire rows. with closing( @@ -196,15 +269,61 @@ def capture_profile_sources( return { "activation": activation, "executed": False, - **_status(db, profile.capture_source_ids), + **_status(db, sources, now=now), } - now = datetime.now(timezone.utc) + + def record_health(tick_status: str, freshness: Mapping[str, Any] | None) -> None: + if health_runtime_root is None: + return + write_capture_host_health( + runtime_root=health_runtime_root, + spool_path=spool_path, + goal_id=goal_id, + agent_id=agent_id, + interval_seconds=profile.capture_interval_seconds, + tick_status=tick_status, + observed_at=now, + freshness=freshness, + ) + + try: + result = _execute_capture_tick( + activation=activation, + profile=profile, + profile_path=profile_path, + digest_before=digest_before, + spool_path=spool_path, + cursor_path=cursor_path, + goal_id=goal_id, + agent_id=agent_id, + sources=sources, + overrides=overrides, + timeout_seconds=timeout_seconds, + now=now, + ) + except BaseException: + record_health("failed", None) + raise + record_health("completed", result["source_freshness"]) + return result + + +def _execute_capture_tick( + *, + activation: Mapping[str, Any], + profile: DecisionContextProfile, + profile_path: Path, + digest_before: str | None, + spool_path: Path, + cursor_path: Path | None, + goal_id: str, + agent_id: str, + sources: tuple[DecisionSourceSpec, ...], + overrides: Mapping[str, DecisionSourceProvider], + timeout_seconds: float, + now: datetime, +) -> dict[str, Any]: observed_at = now.isoformat() - sources = tuple( - source - for source in profile.sources - if source.source_id in profile.capture_source_ids - ) providers = _build_source_providers( profile, sources=sources, source_provider_overrides=overrides ) @@ -311,16 +430,37 @@ def capture_profile_sources( ), ) cursor = scan.cursor_after + # checked_at records the attempt; only a successful read may + # advance last_success_at, so repeated failures cannot look fresh. + last_success_at = row["last_success_at"] if row else None + failure_streak = int(row["failure_streak"] or 0) if row else 0 + if status in _READ_SUCCESS_STATUSES: + last_success_at, failure_streak = observed_at, 0 + elif status in _READ_FAILURE_STATUSES: + failure_streak += 1 db.execute( - "INSERT INTO sources VALUES(?,?,?,?,?) ON CONFLICT(source_id) DO UPDATE SET cursor=excluded.cursor, checked_at=excluded.checked_at, status=excluded.status", - (source.source_id, binding, cursor, observed_at, status), + "INSERT INTO sources(source_id,binding_digest,cursor,checked_at,status," + "last_success_at,failure_streak) VALUES(?,?,?,?,?,?,?) " + "ON CONFLICT(source_id) DO UPDATE SET cursor=excluded.cursor, " + "checked_at=excluded.checked_at, status=excluded.status, " + "last_success_at=excluded.last_success_at, " + "failure_streak=excluded.failure_streak", + ( + source.source_id, + binding, + cursor, + observed_at, + status, + last_success_at, + failure_streak, + ), ) if private_file_digest(profile_path) != digest_before: raise ValueError("capture profile changed during tick") result = { "activation": activation, "executed": True, - **_status(db, profile.capture_source_ids), + **_status(db, sources, now=now), } db.commit() return result diff --git a/loopx/capabilities/decision_context/cli.py b/loopx/capabilities/decision_context/cli.py index fa4e5f620..30652b6ce 100644 --- a/loopx/capabilities/decision_context/cli.py +++ b/loopx/capabilities/decision_context/cli.py @@ -8,6 +8,7 @@ from .assembler import DecisionEvidenceRecords from .architecture import build_decision_context_architecture_packet +from .freshness import render_source_freshness_markdown from .profile import resolve_decision_context_activation from .private_state import ( load_private_pending_decision_settlement, @@ -52,6 +53,10 @@ def _render(payload: dict[str, object]) -> str: if isinstance(capability, dict) else "decision_context" ) + assembly = payload.get("assembly") + freshness = payload.get("source_freshness") or ( + assembly.get("source_freshness") if isinstance(assembly, dict) else None + ) return "\n".join( [ "# Decision Context", @@ -62,6 +67,11 @@ def _render(payload: dict[str, object]) -> str: f"- source_schemas: `{_collection_size(payload.get('source_schemas'))}`", f"- source_count: `{payload.get('source_count', 0)}`", "", + *( + render_source_freshness_markdown(freshness) + if isinstance(freshness, dict) + else [] + ), ] ) @@ -324,7 +334,9 @@ def handle_decision_context_command( } else: payload = capture_profile_sources( - **capture_args, execute=bool(getattr(args, "execute", False)) + **capture_args, + execute=bool(getattr(args, "execute", False)), + health_runtime_root=runtime_root, ) elif args.decision_context_command == "architecture": payload = build_decision_context_architecture_packet() diff --git a/loopx/capabilities/decision_context/freshness.py b/loopx/capabilities/decision_context/freshness.py new file mode 100644 index 000000000..a5057db46 --- /dev/null +++ b/loopx/capabilities/decision_context/freshness.py @@ -0,0 +1,318 @@ +"""Per-source freshness contract and capture host health projection. + +Every Decision Context projection carries one row per enabled source with the +last successful read time and its staleness against the source window. A +healthy control plane or an enabled profile never implies fresh sources. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +from collections.abc import Iterable, Mapping, Sequence +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from .sources import DecisionSourceSpec + +DECISION_SOURCE_FRESHNESS_SCHEMA_VERSION = "decision_source_freshness_v0" +DECISION_CAPTURE_HOST_HEALTH_SCHEMA_VERSION = "decision_capture_host_health_v0" +DECISION_CAPTURE_HOSTS_DIAGNOSTICS_SCHEMA_VERSION = ( + "decision_capture_hosts_diagnostics_v0" +) +CAPTURE_HOST_HEALTH_DIRNAME = Path("decision-context") / "capture-hosts" +FRESHNESS_ALERT_MARKER = "🔴" +_SUCCESS_STATUSES = frozenset({"completed", "no_change"}) + + +def _parse_time(value: object) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.strip()) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None + + +def source_freshness_row( + *, + source: DecisionSourceSpec, + observed_at: datetime, + last_read_at: str | None, + last_attempt_status: str | None, + failure_streak: int = 0, + scanned: bool = True, +) -> dict[str, Any]: + """Project one source; a missing or old successful read is always alerted.""" + + read_time = _parse_time(last_read_at) + staleness_seconds = ( + int(max(0.0, (observed_at - read_time).total_seconds())) + if read_time is not None + else None + ) + if not scanned and read_time is None: + status = "not_scanned" + elif read_time is None: + status = "never_read" + elif staleness_seconds is not None and staleness_seconds > source.freshness_seconds: + status = "stale" + else: + status = "fresh" + alert_reasons = [] if status == "fresh" else [status] + if last_attempt_status is not None and last_attempt_status not in _SUCCESS_STATUSES: + alert_reasons.append(f"last_attempt_{last_attempt_status}") + if failure_streak > 0: + alert_reasons.append("consecutive_failures") + return { + "source_id": source.source_id, + "priority": source.priority, + "scan_mode": source.scan_mode, + "freshness_seconds": source.freshness_seconds, + "last_read_at": read_time.isoformat() if read_time is not None else None, + "staleness_seconds": staleness_seconds, + "status": status, + "last_attempt_status": last_attempt_status, + "failure_streak": failure_streak, + "alert": bool(alert_reasons), + "alert_reasons": alert_reasons, + } + + +def build_source_freshness_report( + *, + observed_at: datetime, + rows: Iterable[Mapping[str, Any]], +) -> dict[str, Any]: + ordered = sorted((dict(row) for row in rows), key=lambda row: row["source_id"]) + alerted = [row["source_id"] for row in ordered if row["alert"]] + return { + "schema_version": DECISION_SOURCE_FRESHNESS_SCHEMA_VERSION, + "observed_at": observed_at.isoformat(), + "window_policy": "per_source_freshness_seconds", + "source_count": len(ordered), + "fresh_count": sum(1 for row in ordered if row["status"] == "fresh"), + "alert_source_ids": alerted, + "stale_source_ids": [ + row["source_id"] for row in ordered if row["status"] == "stale" + ], + "all_fresh": not alerted, + "enabled_state_implies_freshness": False, + "sources": ordered, + } + + +def render_source_freshness_markdown(report: Mapping[str, Any]) -> list[str]: + rows = report.get("sources") + if not isinstance(rows, Sequence): + return [] + lines = [ + "## Source Freshness", + "", + f"- all_fresh: `{report.get('all_fresh')}`", + f"- alert_sources: `{len(report.get('alert_source_ids') or [])}`" + f" / `{report.get('source_count')}`", + "", + ] + for row in rows: + if not isinstance(row, Mapping): + continue + marker = f"{FRESHNESS_ALERT_MARKER} " if row.get("alert") else "" + reasons = ",".join(row.get("alert_reasons") or []) or "-" + lines.append( + f"- {marker}`{row.get('source_id')}` ({row.get('priority')}): " + f"`{row.get('status')}` last_read_at=`{row.get('last_read_at')}` " + f"staleness_seconds=`{row.get('staleness_seconds')}` " + f"window=`{row.get('freshness_seconds')}` alerts=`{reasons}`" + ) + lines.append("") + return lines + + +def _host_record_path(runtime_root: Path, spool_path: Path) -> Path: + digest = hashlib.sha256(str(spool_path.resolve()).encode("utf-8")).hexdigest() + return runtime_root / CAPTURE_HOST_HEALTH_DIRNAME / f"{digest[:24]}.json" + + +def write_capture_host_health( + *, + runtime_root: Path, + spool_path: Path, + goal_id: str, + agent_id: str, + interval_seconds: int, + tick_status: str, + observed_at: datetime, + freshness: Mapping[str, Any] | None, +) -> Path: + """Record one capture tick so ``loopx doctor`` can notice a silent host.""" + + path = _host_record_path(runtime_root, spool_path) + previous: Mapping[str, Any] = {} + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + if isinstance(loaded, Mapping): + previous = loaded + except (OSError, json.JSONDecodeError): + previous = {} + succeeded = tick_status == "completed" + previous_failures = previous.get("consecutive_tick_failures") + record = { + "schema_version": DECISION_CAPTURE_HOST_HEALTH_SCHEMA_VERSION, + "goal_id": goal_id, + "agent_id": agent_id, + "spool_path": str(spool_path.resolve()), + "interval_seconds": interval_seconds, + "last_tick_at": observed_at.isoformat(), + "last_tick_status": tick_status, + "last_successful_tick_at": ( + observed_at.isoformat() + if succeeded + else previous.get("last_successful_tick_at") + ), + "consecutive_tick_failures": ( + 0 + if succeeded + else (previous_failures if isinstance(previous_failures, int) else 0) + 1 + ), + "source_freshness": ( + dict(freshness) if freshness is not None else previous.get("source_freshness") + ), + } + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent + ) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(record, handle, ensure_ascii=True, sort_keys=True) + handle.write("\n") + os.replace(temporary_name, path) + finally: + Path(temporary_name).unlink(missing_ok=True) + return path + + +def _host_diagnosis( + record: Mapping[str, Any], + *, + now: datetime, + record_path: Path, +) -> dict[str, Any]: + interval = record.get("interval_seconds") + interval = interval if isinstance(interval, int) and interval > 0 else 900 + grace_seconds = max(2 * interval, interval + 600) + last_tick = _parse_time(record.get("last_tick_at")) + tick_age = ( + int(max(0.0, (now - last_tick).total_seconds())) if last_tick else None + ) + reasons: list[str] = [] + if tick_age is None or tick_age > grace_seconds: + reasons.append("host_heartbeat_stale") + if record.get("last_tick_status") != "completed": + reasons.append("last_tick_failed") + spool_path = record.get("spool_path") + if not isinstance(spool_path, str) or not Path(spool_path).exists(): + reasons.append("spool_missing") + + alert_source_ids: list[str] = [] + freshness = record.get("source_freshness") + rows = freshness.get("sources") if isinstance(freshness, Mapping) else None + for row in rows if isinstance(rows, Sequence) else (): + if not isinstance(row, Mapping): + continue + read_time = _parse_time(row.get("last_read_at")) + window = row.get("freshness_seconds") + stale_now = ( + read_time is None + or not isinstance(window, int) + or (now - read_time).total_seconds() > window + ) + if stale_now or row.get("failure_streak"): + alert_source_ids.append(str(row.get("source_id"))) + if alert_source_ids: + reasons.append("source_freshness_alert") + return { + "goal_id": record.get("goal_id"), + "agent_id": record.get("agent_id"), + "record": str(record_path), + "interval_seconds": interval, + "heartbeat_grace_seconds": grace_seconds, + "last_tick_at": record.get("last_tick_at"), + "last_tick_age_seconds": tick_age, + "last_tick_status": record.get("last_tick_status"), + "last_successful_tick_at": record.get("last_successful_tick_at"), + "consecutive_tick_failures": record.get("consecutive_tick_failures", 0), + "alert_source_ids": sorted(alert_source_ids), + "healthy": not reasons, + "alert_reasons": reasons, + } + + +def collect_capture_host_diagnostics( + runtime_root: Path, + *, + now: datetime | None = None, +) -> dict[str, Any]: + """Read registered capture host records without opening private spools.""" + + current = now or datetime.now(timezone.utc) + directory = runtime_root / CAPTURE_HOST_HEALTH_DIRNAME + hosts: list[dict[str, Any]] = [] + for record_path in sorted(directory.glob("*.json")) if directory.is_dir() else (): + try: + record = json.loads(record_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + record = None + if ( + not isinstance(record, Mapping) + or record.get("schema_version") != DECISION_CAPTURE_HOST_HEALTH_SCHEMA_VERSION + ): + hosts.append( + { + "record": str(record_path), + "healthy": False, + "alert_reasons": ["record_invalid"], + } + ) + continue + hosts.append(_host_diagnosis(record, now=current, record_path=record_path)) + unhealthy = [host for host in hosts if not host["healthy"]] + return { + "schema_version": DECISION_CAPTURE_HOSTS_DIAGNOSTICS_SCHEMA_VERSION, + "registry": str(directory), + "observed_at": current.isoformat(), + "host_count": len(hosts), + "unhealthy_count": len(unhealthy), + "healthy": not unhealthy, + "hosts": hosts, + "private_spools_opened": False, + } + + +def capture_host_diagnostics_detail(diagnostics: Mapping[str, Any]) -> str: + hosts = diagnostics.get("hosts") or [] + if not hosts: + return "no Decision Context capture hosts registered" + unhealthy = [host for host in hosts if not host.get("healthy")] + if not unhealthy: + return f"{len(hosts)} capture host(s) healthy" + parts = [ + f"{host.get('goal_id')}/{host.get('agent_id')}: " + + ",".join(host.get("alert_reasons") or []) + + ( + f" sources={','.join(host['alert_source_ids'])}" + if host.get("alert_source_ids") + else "" + ) + for host in unhealthy + ] + return ( + f"{FRESHNESS_ALERT_MARKER} {len(unhealthy)}/{len(hosts)} capture host(s) " + "unhealthy; " + "; ".join(parts) + + ". Repair the host binding or remove the retired record." + ) diff --git a/loopx/capabilities/decision_context/runtime.py b/loopx/capabilities/decision_context/runtime.py index 81df6138f..347450494 100644 --- a/loopx/capabilities/decision_context/runtime.py +++ b/loopx/capabilities/decision_context/runtime.py @@ -302,6 +302,7 @@ def assemble_profile_decision_evidence( recall_query_summary="current decision evidence", recall_limit=int(context_config.get("max_results", 5)), timeout_seconds=effective_timeout, + coverage_sources=profile.sources, ) if profile_path is None or profile_digest_before is None: raise ValueError("decision-context profile became unavailable") diff --git a/loopx/cli.py b/loopx/cli.py index 21bd947bc..b344eb46a 100644 --- a/loopx/cli.py +++ b/loopx/cli.py @@ -646,7 +646,7 @@ def main(argv: list[str] | None = None) -> int: effective_runtime_root(registry_path, args.runtime_root) if args.command == "decision-context" and args.decision_context_command - in {"recall-context", "prepare-evidence", "prepare-review"} + in {"recall-context", "prepare-evidence", "prepare-review", "capture"} else None ), output_format=output_format, diff --git a/loopx/doctor.py b/loopx/doctor.py index ab5923544..11e33ef17 100644 --- a/loopx/doctor.py +++ b/loopx/doctor.py @@ -908,6 +908,12 @@ def collect_doctor( "items": [], } ) + from .capabilities.decision_context.freshness import ( + capture_host_diagnostics_detail, + collect_capture_host_diagnostics, + ) + + decision_context_capture = collect_capture_host_diagnostics(DEFAULT_RUNTIME_ROOT) typescript_control_plane = collect_effect_runtime_readiness(deep=deep) typescript_runtime_required = True deep_validation = None @@ -1054,6 +1060,12 @@ def collect_doctor( sort_keys=True, ), }, + { + "id": "decision_context_capture_hosts_healthy", + "required": False, + "ok": bool(decision_context_capture["healthy"]), + "detail": capture_host_diagnostics_detail(decision_context_capture), + }, { "id": "typescript_effect_runtime_ready", "required": typescript_runtime_required, @@ -1113,6 +1125,7 @@ def collect_doctor( "release_provenance": release_provenance, "global_registry_writability": global_registry_writability, "runtime_projection_routes": runtime_projection_routes, + "decision_context_capture": decision_context_capture, "typescript_control_plane": typescript_control_plane, "install_freshness": install_freshness, "upgrade_hint": install_freshness, @@ -1200,6 +1213,9 @@ def render_doctor_markdown(payload: dict[str, Any]) -> str: f" (registry=`{(payload.get('runtime_projection_routes') or {}).get('registry')}`," f" goals=`{(payload.get('runtime_projection_routes') or {}).get('goal_count')}`," f" counts=`{json.dumps((payload.get('runtime_projection_routes') or {}).get('counts') or {}, sort_keys=True)}`)", + f"- decision_context_capture_hosts_healthy: `{(payload.get('decision_context_capture') or {}).get('healthy')}`" + f" (hosts=`{(payload.get('decision_context_capture') or {}).get('host_count')}`," + f" unhealthy=`{(payload.get('decision_context_capture') or {}).get('unhealthy_count')}`)", f"- user_local_bin_on_path: `{(payload.get('path') or {}).get('user_local_bin_on_path')}`", f"- python: `{(payload.get('python') or {}).get('executable')}`", f"- typescript_control_plane: `{typescript_control_plane.get('status')}`", diff --git a/tests/capabilities/test_decision_context_capture.py b/tests/capabilities/test_decision_context_capture.py index bde1b60fb..792d97dda 100644 --- a/tests/capabilities/test_decision_context_capture.py +++ b/tests/capabilities/test_decision_context_capture.py @@ -220,10 +220,23 @@ def test_capture_cli_and_status_are_public_safe(setup, capsys): "--spool", str(args["spool_path"]), ] + runtime_root = args["spool_path"].parent / "runtime" assert ( - main(["--format", "json", "decision-context", "capture", *common, "--execute"]) + main( + [ + "--format", + "json", + "--runtime-root", + str(runtime_root), + "decision-context", + "capture", + *common, + "--execute", + ] + ) == 0 ) + assert len(list((runtime_root / "decision-context" / "capture-hosts").glob("*.json"))) == 1 captured = capsys.readouterr().out assert "private-body" not in captured assert json.loads(captured)["pending_batch_count"] == 1 diff --git a/tests/capabilities/test_decision_context_freshness.py b/tests/capabilities/test_decision_context_freshness.py new file mode 100644 index 000000000..ca4a3bc2a --- /dev/null +++ b/tests/capabilities/test_decision_context_freshness.py @@ -0,0 +1,235 @@ +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime, timedelta, timezone + +import pytest + +from loopx.capabilities.decision_context import DecisionEvidenceRecords +from loopx.capabilities.decision_context.capture import capture_profile_sources +from loopx.capabilities.decision_context.cli import _render +from loopx.capabilities.decision_context.freshness import ( + FRESHNESS_ALERT_MARKER, + capture_host_diagnostics_detail, + collect_capture_host_diagnostics, +) +from loopx.capabilities.decision_context.providers import ( + LocalFileDecisionSourceProvider, +) +from loopx.capabilities.decision_context.runtime import ( + assemble_profile_decision_evidence, +) +from test_decision_context_profile import profile_payload + +BASELINE = "source:authority:baseline" +ON_DEMAND = "source:authority:on-demand" + + +class FailingProvider(LocalFileDecisionSourceProvider): + def scan(self, **kwargs): + raise RuntimeError("private-secret-must-not-leak") + + +def _failing(): + return { + "local-authority": FailingProvider( + provider_id="local-authority", max_bytes=4096 + ) + } + + +@pytest.fixture +def setup(tmp_path): + authority = tmp_path / "authority.txt" + authority.write_text("private-body-not-for-output") + payload = profile_payload(authority) + on_demand = dict(payload["sources"][0]) + on_demand.update(source_id=ON_DEMAND, scan_mode="on_demand", priority="p1") + payload["sources"].append(on_demand) + payload["automation"].update(automatic_capture=True, source_ids=[BASELINE]) + profile = tmp_path / "profile.json" + profile.write_text(json.dumps(payload)) + args = dict( + goal_id=payload["goal_id"], + agent_id="example-agent", + profile_path=profile, + spool_path=tmp_path / "spool.sqlite", + cursor_path=tmp_path / "reviewed.json", + ) + return args, payload, tmp_path / "runtime" + + +def _row(report, source_id): + return next(row for row in report["sources"] if row["source_id"] == source_id) + + +def test_assembly_reports_every_enabled_source_and_alerts_unscanned(setup): + args, _, _ = setup + now = datetime.now(timezone.utc).isoformat() + _, assembly = assemble_profile_decision_evidence( + goal_id=args["goal_id"], + agent_id=args["agent_id"], + profile_path=args["profile_path"], + decision_id="decision:freshness", + observed_at=now, + before=now, + rebase=lambda _collection: DecisionEvidenceRecords(), + ) + report = assembly.public_packet()["source_freshness"] + assert report["schema_version"] == "decision_source_freshness_v0" + assert report["enabled_state_implies_freshness"] is False + assert _row(report, BASELINE)["status"] == "fresh" + assert _row(report, BASELINE)["last_read_at"] == now + unscanned = _row(report, ON_DEMAND) + assert unscanned["status"] == "not_scanned" + assert unscanned["alert"] is True + assert report["alert_source_ids"] == [ON_DEMAND] + assert report["all_fresh"] is False + rendered = _render({"status": "available", "assembly": assembly.public_packet()}) + assert f"{FRESHNESS_ALERT_MARKER} `{ON_DEMAND}`" in rendered + assert f"{FRESHNESS_ALERT_MARKER} `{BASELINE}`" not in rendered + + +def test_assembly_failed_source_is_never_reported_fresh(setup): + args, _, _ = setup + now = datetime.now(timezone.utc).isoformat() + _, assembly = assemble_profile_decision_evidence( + goal_id=args["goal_id"], + agent_id=args["agent_id"], + profile_path=args["profile_path"], + decision_id="decision:freshness", + observed_at=now, + before=now, + source_ids=[BASELINE, ON_DEMAND], + rebase=lambda _collection: DecisionEvidenceRecords(), + source_provider_overrides=_failing(), + ) + report = assembly.public_packet()["source_freshness"] + row = _row(report, BASELINE) + assert row["status"] == "never_read" + assert row["last_read_at"] is None + assert any(reason.startswith("last_attempt_") for reason in row["alert_reasons"]) + assert "private-secret" not in json.dumps(report) + + +def test_capture_failure_cannot_refresh_last_read(setup): + args, _, _ = setup + failed = capture_profile_sources( + **args, execute=True, source_provider_overrides=_failing() + ) + row = failed["sources"][0] + assert row["status"] == "provider_failed" + assert row["last_checked_at"] is not None + assert row["last_read_at"] is None + assert row["freshness"] == "never_read" + assert failed["source_freshness"]["alert_source_ids"] == [BASELINE] + + with sqlite3.connect(args["spool_path"]) as db: + db.execute("UPDATE sources SET checked_at=NULL") + ok = capture_profile_sources(**args, execute=True) + read_at = ok["sources"][0]["last_read_at"] + assert ok["sources"][0]["freshness"] == "fresh" + assert ok["sources"][0]["failure_streak"] == 0 + assert ok["source_freshness"]["all_fresh"] is True + + for expected_streak in (1, 2): + with sqlite3.connect(args["spool_path"]) as db: + db.execute("UPDATE sources SET checked_at=NULL") + again = capture_profile_sources( + **args, execute=True, source_provider_overrides=_failing() + ) + row = again["sources"][0] + assert row["last_read_at"] == read_at + assert row["failure_streak"] == expected_streak + assert "consecutive_failures" in _row( + again["source_freshness"], BASELINE + )["alert_reasons"] + + old = (datetime.now(timezone.utc) - timedelta(days=4)).isoformat() + with sqlite3.connect(args["spool_path"]) as db: + db.execute("UPDATE sources SET last_success_at=?", (old,)) + status = capture_profile_sources(**args) + assert status["executed"] is False + assert status["sources"][0]["freshness"] == "stale" + assert status["source_freshness"]["stale_source_ids"] == [BASELINE] + assert f"{FRESHNESS_ALERT_MARKER} `{BASELINE}`" in _render(status) + + +def test_legacy_spool_is_readable_and_migrated(setup): + args, _, _ = setup + checked = datetime.now(timezone.utc).isoformat() + with sqlite3.connect(args["spool_path"]) as db: + db.executescript(""" + CREATE TABLE identity (goal TEXT, agent TEXT); + CREATE TABLE sources ( + source_id TEXT PRIMARY KEY, binding_digest TEXT NOT NULL, + cursor TEXT, checked_at TEXT, status TEXT NOT NULL); + CREATE TABLE batches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, source_id TEXT NOT NULL, + cursor_before TEXT, cursor_after TEXT NOT NULL, + before_time TEXT NOT NULL, receipt TEXT NOT NULL); + """) + db.execute( + "INSERT INTO identity VALUES (?, ?)", (args["goal_id"], args["agent_id"]) + ) + db.execute( + "INSERT INTO sources VALUES (?, 'legacy', NULL, ?, 'completed')", + (BASELINE, checked), + ) + preview = capture_profile_sources(**args) + assert preview["sources"][0]["last_read_at"] == checked + assert preview["sources"][0]["freshness"] == "fresh" + + capture_profile_sources(**args, execute=True) + with sqlite3.connect(args["spool_path"]) as db: + columns = {row[1] for row in db.execute("PRAGMA table_info(sources)")} + assert {"last_success_at", "failure_streak"} <= columns + assert db.execute("SELECT last_success_at FROM sources").fetchone()[0] == checked + + +def test_doctor_host_diagnostics_detect_silent_and_failing_hosts(setup): + args, payload, runtime_root = setup + assert collect_capture_host_diagnostics(runtime_root)["host_count"] == 0 + + capture_profile_sources(**args, execute=True, health_runtime_root=runtime_root) + now = datetime.now(timezone.utc) + healthy = collect_capture_host_diagnostics(runtime_root, now=now) + assert healthy["healthy"] is True + assert healthy["private_spools_opened"] is False + assert str(args["spool_path"]) not in capture_host_diagnostics_detail(healthy) + + interval = payload["automation"].get("interval_seconds", 900) + silent = collect_capture_host_diagnostics( + runtime_root, now=now + timedelta(seconds=3 * interval) + ) + host = silent["hosts"][0] + assert "host_heartbeat_stale" in host["alert_reasons"] + assert FRESHNESS_ALERT_MARKER in capture_host_diagnostics_detail(silent) + + much_later = collect_capture_host_diagnostics( + runtime_root, now=now + timedelta(days=4) + ) + assert much_later["hosts"][0]["alert_source_ids"] == [BASELINE] + + payload["enabled_agents"].append("other-agent") + args["profile_path"].write_text(json.dumps(payload)) + for _ in range(2): + with pytest.raises(ValueError, match="goal/agent mismatch"): + capture_profile_sources( + **(args | {"agent_id": "other-agent"}), + execute=True, + health_runtime_root=runtime_root, + ) + records = sorted( + (runtime_root / "decision-context" / "capture-hosts").glob("*.json") + ) + assert len(records) == 1 + failing = collect_capture_host_diagnostics(runtime_root)["hosts"][0] + assert failing["last_tick_status"] == "failed" + assert failing["consecutive_tick_failures"] == 2 + assert "last_tick_failed" in failing["alert_reasons"] + + args["spool_path"].unlink() + missing = collect_capture_host_diagnostics(runtime_root)["hosts"][0] + assert "spool_missing" in missing["alert_reasons"]