Merge remote-tracking branch 'gitea/main' into codex/iwooos-wazuh-boundary-guard-20260624

# Conflicts:
#	docs/LOGBOOK.md
This commit is contained in:
ogt
2026-06-25 10:57:40 +08:00
20 changed files with 5435 additions and 12 deletions

View File

@@ -0,0 +1,68 @@
"""
AI Agent market radar readback.
Loads the committed read-only radar artifact. The radar is an operator
decision surface only; it does not approve SDK installs, paid API calls,
replay, shadow/canary, Telegram sends, host writes, or production routing.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from src.services.snapshot_paths import default_operations_dir
_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
_SNAPSHOT_NAME = "ai-agent-market-radar-readback.snapshot.json"
def load_latest_ai_agent_market_radar_readback(
operations_dir: Path | None = None,
) -> dict[str, Any]:
"""Load the committed AI Agent market radar readback snapshot."""
directory = operations_dir or _DEFAULT_OPERATIONS_DIR
snapshot_path = directory / _SNAPSHOT_NAME
with snapshot_path.open(encoding="utf-8") as handle:
payload = json.load(handle)
if not isinstance(payload, dict):
raise ValueError(f"{snapshot_path}: expected JSON object")
if payload.get("schema_version") != "ai_agent_market_radar_readback_v1":
raise ValueError(f"{snapshot_path}: unexpected schema_version")
policy = payload.get("policy") or {}
forbidden_true = [
key
for key in [
"sdk_installation_approved",
"paid_api_calls_approved",
"replay_candidate_approved",
"shadow_or_canary_approved",
"production_routing_approved",
"telegram_send_approved",
"host_write_approved",
"workflow_modification_approved",
"openclaw_replacement_approved",
]
if policy.get(key) is not False
]
if forbidden_true:
raise ValueError(f"{snapshot_path}: unsafe policy flags: {forbidden_true}")
serialized = json.dumps(payload, ensure_ascii=False)
forbidden_fragments = [
"/Users/",
".claude/projects",
".codex",
"192.168.",
"auth.json",
"conversations",
"sessions",
]
leaked = [fragment for fragment in forbidden_fragments if fragment in serialized]
if leaked:
raise ValueError(f"{snapshot_path}: forbidden local or raw-history fragment: {leaked}")
return payload