admin/ComfyUI-OpenClaw
1
0
Fork 0
mirror of https://github.com/rookiestar28/ComfyUI-OpenClaw.git synced 2026-08-14 00:48:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement R90/R95 governance and query contracts

Browse Source
This commit is contained in:
rookiestar28
2026-02-26 12:48:03 +08:00
parent 28126613e2
commit f6b8b2e0e9
17 changed files with 1545 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+14
View File
@@ -50,6 +50,20 @@ Deployment profiles and hardening checklists:
<details>
<summary><strong>Recent hardening and reliability improvements: runtime guardrails, crypto drills, compatibility governance, and safer management queries</strong></summary>
- Completed a focused reliability + operations hardening batch with full SOP verification:
- consolidated shared frontend/backed helper paths to reduce duplicated cancellation, JSON parsing, and import-fallback logic
- added runtime guardrails diagnostics/contract enforcement so runtime-only safety limits stay visible and cannot be persisted back into config
- added cryptographic lifecycle drill automation with machine-readable evidence for rotation, revoke, key-loss recovery, and token-compromise scenarios
- added compatibility matrix governance metadata plus a refresh workflow script and operator-doctor freshness/drift warnings
- hardened management query pagination behavior with deterministic malformed-input handling, bounded scans, and clearer cursor diagnostics for admin/event list paths
- completed full verification gate pass (detect-secrets, pre-commit, backend unit suites, and frontend Playwright E2E)
</details>
<details>
<summary><strong>Latest completion: automation composer endpoint, safer payload drafting, and full verification pass</strong></summary>
- Completed the automation payload composer flow for safe draft generation:
api/approvals.py
+32 -6
View File
@@ -13,6 +13,7 @@ try:
from ..services.approvals.models import ApprovalStatus
from ..services.approvals.service import get_approval_service
from ..services.audit import emit_audit_event
from ..services.management_query import bounded_scan_collect, normalize_limit_offset
from ..services.webhook_auth import AuthError
except ImportError:
# Fallback for ComfyUI's non-package loader or ad-hoc imports.
@@ -20,6 +21,10 @@ except ImportError:
from services.approvals.models import ApprovalStatus
from services.approvals.service import get_approval_service
from services.audit import emit_audit_event # type: ignore
from services.management_query import ( # type: ignore
bounded_scan_collect,
normalize_limit_offset,
)
from services.webhook_auth import AuthError
logger = logging.getLogger("ComfyUI-OpenClaw.api.approvals")
@@ -106,8 +111,13 @@ class ApprovalHandlers:
# Parse query params
status_filter = request.query.get("status")
limit = int(request.query.get("limit", "100"))
offset = int(request.query.get("offset", "0"))
page = normalize_limit_offset(
request.query,
default_limit=100,
max_limit=500,
default_offset=0,
max_offset=5000,
)
# Validate and convert status
status = None
@@ -120,17 +130,33 @@ class ApprovalHandlers:
)
# Get approvals
# R95: bounded scan window protects API serialization path and keeps
# malformed-record behavior deterministic without swallowing service errors.
scan_cap = max(page.offset + page.limit + 200, page.limit * 10)
approvals = self._service.list_all(
status=status,
limit=min(limit, 500),
offset=offset,
limit=min(scan_cap, 5000),
offset=0,
)
page_result = bounded_scan_collect(
approvals,
skip=page.offset,
take=page.limit,
scan_cap=min(scan_cap, 5000),
serializer=lambda a: a.to_dict(),
)
return web.json_response(
{
"approvals": [a.to_dict() for a in approvals],
"count": len(approvals),
"approvals": page_result.items,
"count": len(page_result.items),
"pending_count": self._service.count_pending(),
"pagination": {
"limit": page.limit,
"offset": page.offset,
"warnings": page.warnings,
},
"scan": page_result.to_dict(),
}
)
api/events.py
+66 -17
View File
@@ -24,11 +24,13 @@ except ModuleNotFoundError: # pragma: no cover
if __package__ and "." in __package__:
from ..services.access_control import require_observability_access
from ..services.job_events import get_job_event_store
from ..services.management_query import normalize_cursor_limit
from ..services.metrics import metrics
from ..services.rate_limit import check_rate_limit
else: # pragma: no cover
from services.access_control import require_observability_access # type: ignore
from services.job_events import get_job_event_store # type: ignore
from services.management_query import normalize_cursor_limit # type: ignore
from services.metrics import metrics # type: ignore
from services.rate_limit import check_rate_limit # type: ignore
@@ -193,29 +195,76 @@ async def events_poll_handler(request: web.Request) -> web.Response:
store = get_job_event_store()
# Parse query params
try:
since = int(request.query.get("since", "0"))
except ValueError:
since = 0
# R95: deterministic pagination normalization + bounded scan diagnostics
prompt_id = request.query.get("prompt_id")
try:
limit = max(1, min(int(request.query.get("limit", "50")), 200))
except ValueError:
limit = 50
events = store.events_since(
last_seq=since,
limit=limit,
prompt_id=prompt_id,
page = normalize_cursor_limit(
request.query,
cursor_key="since",
default_cursor=0,
min_cursor=0,
default_limit=50,
max_limit=200,
)
since_requested = int(page.cursor or 0)
latest_seq = store.latest_seq()
cursor_status = "ok"
since_effective = since_requested
if since_requested > latest_seq:
cursor_status = "future_cursor_reset"
since_effective = latest_seq
page.warnings.append(
{
"code": "R95_STALE_CURSOR_FUTURE",
"field": "since",
"raw": str(since_requested),
"normalized": since_effective,
}
)
scan_cap = max(page.limit * 10, 500)
events, scan = store.events_since_bounded(
last_seq=since_effective,
limit=page.limit,
prompt_id=prompt_id,
scan_cap=scan_cap,
)
earliest_retained = scan.get("earliest_retained_seq")
if (
isinstance(earliest_retained, int)
and since_effective != 0
and since_effective < (earliest_retained - 1)
):
cursor_status = "stale_cursor_reset"
since_effective = max(0, earliest_retained - 1)
page.warnings.append(
{
"code": "R95_STALE_CURSOR_RESET",
"field": "since",
"raw": str(since_requested),
"normalized": since_effective,
}
)
events, scan = store.events_since_bounded(
last_seq=since_effective,
limit=page.limit,
prompt_id=prompt_id,
scan_cap=scan_cap,
)
return web.json_response(
{
"ok": True,
"events": [e.to_dict() for e in events],
"latest_seq": store.latest_seq(),
"latest_seq": latest_seq,
"pagination": {
"limit": page.limit,
"since_requested": since_requested,
"since_effective": since_effective,
"cursor_status": cursor_status,
"warnings": page.warnings,
},
"scan": scan,
}
)
docs/release/compatibility_matrix.md
+22
View File
@@ -1,5 +1,27 @@
# Compatibility Matrix
```openclaw-compat-matrix-meta
{
"anchors": {
"comfyui": "unknown",
"comfyui_frontend": "unknown",
"desktop": "unknown"
},
"evidence": {
"evidence_id": "compat-matrix-20260225",
"updated_at": "2026-02-25T00:00:00+00:00",
"updated_by": "manual"
},
"last_validated_date": "2026-02-25",
"matrix_version": "v0.2.1",
"policy": {
"max_age_days": 45,
"warn_age_days": 30
},
"schema_version": 1
}
```
This document outlines the validated environments for ComfyUI-OpenClaw `v0.2.1` (M1 Release).
## Core Dependencies
scripts/compatibility_matrix_refresh.py
+124
View File
@@ -0,0 +1,124 @@
"""
R90 compatibility matrix refresh workflow.
Implements a repeatable `collect -> diff -> validate -> publish` flow and emits
machine-readable evidence for date-stamped refresh operations.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
def _repo_root() -> Path:
return Path(__file__).resolve().parents[1]
def _ensure_repo_on_path() -> None:
root = str(_repo_root())
if root not in sys.path:
sys.path.insert(0, root)
def main() -> int:
_ensure_repo_on_path()
from services.compatibility_matrix_governance import (
normalize_observed_anchors,
run_refresh_workflow,
)
parser = argparse.ArgumentParser(
description=(
"Run compatibility matrix refresh workflow (collect/diff/validate/publish) "
"and emit machine-readable evidence."
)
)
parser.add_argument(
"--matrix-path",
default=str(_repo_root() / "docs" / "release" / "compatibility_matrix.md"),
help="Path to compatibility matrix markdown file",
)
parser.add_argument(
"--anchor-comfyui",
default=None,
help="Observed ComfyUI anchor/version (optional)",
)
parser.add_argument(
"--anchor-frontend",
default=None,
help="Observed ComfyUI frontend anchor/version (optional)",
)
parser.add_argument(
"--anchor-desktop",
default=None,
help="Observed ComfyUI Desktop anchor/version (optional)",
)
parser.add_argument(
"--updated-by",
default="script",
help="Evidence updated_by marker",
)
parser.add_argument(
"--apply",
action="store_true",
help="Apply/publish metadata updates to the matrix file (default: dry-run)",
)
parser.add_argument(
"--strict",
action="store_true",
help="Exit non-zero when validation fails or drift is detected",
)
parser.add_argument(
"--output",
default=None,
help="Optional path to write JSON evidence bundle",
)
parser.add_argument(
"--pretty",
action="store_true",
help="Pretty-print JSON output",
)
args = parser.parse_args()
observed = normalize_observed_anchors(
comfyui=args.anchor_comfyui,
comfyui_frontend=args.anchor_frontend,
desktop=args.anchor_desktop,
)
result = run_refresh_workflow(
matrix_path=args.matrix_path,
observed_anchors=observed,
apply=args.apply,
updated_by=args.updated_by,
)
payload = result.to_dict()
if args.output:
Path(args.output).parent.mkdir(parents=True, exist_ok=True)
Path(args.output).write_text(
json.dumps(payload, indent=2 if args.pretty else None, ensure_ascii=False),
encoding="utf-8",
)
if args.pretty:
print(json.dumps(payload, indent=2, ensure_ascii=False))
else:
print(json.dumps(payload, separators=(",", ":"), ensure_ascii=False))
if not args.strict:
return 0
validate_after = payload["stages"]["validate"]["after"]
drift_before = payload["stages"]["diff"]["drift"]
if (not validate_after.get("ok")) or (not drift_before.get("ok")):
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/run_crypto_lifecycle_drills.py
+4 -1
View File
@@ -29,7 +29,10 @@ def _parse_scenarios(raw: str) -> list[str]:
def main() -> int:
_ensure_repo_on_path()
from services.crypto_lifecycle_drills import DEFAULT_SCENARIOS, run_crypto_lifecycle_drills
from services.crypto_lifecycle_drills import (
DEFAULT_SCENARIOS,
run_crypto_lifecycle_drills,
)
parser = argparse.ArgumentParser(
description="Run R119 crypto lifecycle drills and emit machine-readable evidence."
services/compatibility_matrix_governance.py
+406
View File
@@ -0,0 +1,406 @@
"""
R90: Compatibility matrix governance helpers.
Machine-readable metadata and refresh workflow primitives for
`docs/release/compatibility_matrix.md`.
"""
from __future__ import annotations
import copy
import hashlib
import json
import re
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
META_BLOCK_TAG = "openclaw-compat-matrix-meta"
DEFAULT_WARN_AGE_DAYS = 30
DEFAULT_MAX_AGE_DAYS = 45
ANCHOR_KEYS = ("comfyui", "comfyui_frontend", "desktop")
META_BLOCK_RE = re.compile(
r"```" + re.escape(META_BLOCK_TAG) + r"\s*\n(?P<body>.*?)\n```",
re.DOTALL,
)
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _today_iso() -> str:
return _utc_now().date().isoformat()
def _parse_date(value: str) -> Optional[date]:
try:
return date.fromisoformat(value)
except Exception:
return None
def _json_hash(payload: Any) -> str:
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def _default_metadata() -> Dict[str, Any]:
today = _today_iso()
return {
"schema_version": 1,
"matrix_version": "v0.2.1",
"last_validated_date": today,
"policy": {
"warn_age_days": DEFAULT_WARN_AGE_DAYS,
"max_age_days": DEFAULT_MAX_AGE_DAYS,
},
"anchors": {key: "unknown" for key in ANCHOR_KEYS},
"evidence": {
"evidence_id": f"compat-matrix-{today.replace('-', '')}",
"updated_at": _utc_now().isoformat(),
"updated_by": "manual",
},
}
def format_metadata_block(metadata: Dict[str, Any]) -> str:
return (
f"```{META_BLOCK_TAG}\n"
+ json.dumps(metadata, indent=2, sort_keys=True)
+ "\n```\n"
)
def extract_metadata_block(text: str) -> Tuple[Optional[Dict[str, Any]], List[str], Optional[str]]:
"""
Extract JSON metadata block.
Returns: (metadata, issues, raw_json_text)
"""
match = META_BLOCK_RE.search(text)
if not match:
return None, ["R90_META_BLOCK_MISSING"], None
raw = match.group("body").strip()
try:
parsed = json.loads(raw)
except json.JSONDecodeError:
return None, ["R90_META_BLOCK_INVALID_JSON"], raw
if not isinstance(parsed, dict):
return None, ["R90_META_BLOCK_NOT_OBJECT"], raw
return parsed, [], raw
def replace_metadata_block(text: str, metadata: Dict[str, Any]) -> str:
block = format_metadata_block(metadata)
if META_BLOCK_RE.search(text):
return META_BLOCK_RE.sub(lambda _m: block.rstrip("\n"), text, count=1)
# Insert after first heading if present; otherwise prepend.
lines = text.splitlines(keepends=True)
for idx, line in enumerate(lines):
if line.lstrip().startswith("# "):
return "".join(lines[: idx + 1] + ["\n", block] + lines[idx + 1 :])
return block + text
def _body_without_meta(text: str) -> str:
return META_BLOCK_RE.sub("", text).strip()
def read_matrix_document(path: Path | str) -> Dict[str, Any]:
p = Path(path)
text = p.read_text(encoding="utf-8")
metadata, issues, raw = extract_metadata_block(text)
return {
"path": str(p),
"text": text,
"metadata": metadata,
"issues": issues,
"raw_metadata": raw,
"body_sha256": hashlib.sha256(_body_without_meta(text).encode("utf-8")).hexdigest(),
"has_meta": metadata is not None,
}
def validate_metadata(metadata: Optional[Dict[str, Any]], *, today: Optional[date] = None) -> Dict[str, Any]:
today = today or _utc_now().date()
violations: List[Dict[str, Any]] = []
if not isinstance(metadata, dict):
return {
"ok": False,
"status": "invalid",
"code": "R90_META_INVALID",
"age_days": None,
"violations": [{"code": "R90_META_MISSING", "message": "Metadata missing"}],
}
schema_version = metadata.get("schema_version")
if schema_version != 1:
violations.append(
{
"code": "R90_META_SCHEMA_VERSION",
"message": f"Unsupported schema_version: {schema_version!r}",
}
)
last_validated = metadata.get("last_validated_date")
parsed_last = _parse_date(last_validated) if isinstance(last_validated, str) else None
if parsed_last is None:
violations.append(
{
"code": "R90_META_LAST_VALIDATED_DATE",
"message": "Missing/invalid last_validated_date (YYYY-MM-DD)",
}
)
policy = metadata.get("policy")
if not isinstance(policy, dict):
policy = {}
violations.append({"code": "R90_META_POLICY", "message": "Missing policy object"})
try:
warn_age_days = int(policy.get("warn_age_days", DEFAULT_WARN_AGE_DAYS))
except Exception:
warn_age_days = DEFAULT_WARN_AGE_DAYS
violations.append(
{"code": "R90_META_WARN_AGE", "message": "Invalid policy.warn_age_days"}
)
try:
max_age_days = int(policy.get("max_age_days", DEFAULT_MAX_AGE_DAYS))
except Exception:
max_age_days = DEFAULT_MAX_AGE_DAYS
violations.append(
{"code": "R90_META_MAX_AGE", "message": "Invalid policy.max_age_days"}
)
if warn_age_days < 0 or max_age_days < 0 or warn_age_days > max_age_days:
violations.append(
{
"code": "R90_META_AGE_POLICY_ORDER",
"message": "Age policy must satisfy 0 <= warn_age_days <= max_age_days",
}
)
anchors = metadata.get("anchors")
if not isinstance(anchors, dict):
anchors = {}
violations.append({"code": "R90_META_ANCHORS", "message": "Missing anchors object"})
else:
for key in ANCHOR_KEYS:
if key not in anchors:
violations.append(
{"code": "R90_META_ANCHOR_MISSING", "message": f"Missing anchors.{key}"}
)
age_days: Optional[int] = None
if parsed_last is not None:
age_days = (today - parsed_last).days
if age_days < 0:
violations.append(
{
"code": "R90_META_FUTURE_DATE",
"message": f"last_validated_date is in the future: {last_validated}",
}
)
if violations:
status = "invalid"
code = "R90_META_INVALID"
else:
assert age_days is not None
if age_days > max_age_days:
status = "stale"
code = "R90_MATRIX_STALE"
elif age_days > warn_age_days:
status = "warning"
code = "R90_MATRIX_AGING"
else:
status = "fresh"
code = "R90_MATRIX_FRESH"
return {
"ok": len(violations) == 0,
"status": status,
"code": code,
"age_days": age_days,
"warn_age_days": warn_age_days,
"max_age_days": max_age_days,
"violations": violations,
}
def normalize_observed_anchors(
*,
comfyui: Optional[str] = None,
comfyui_frontend: Optional[str] = None,
desktop: Optional[str] = None,
) -> Dict[str, str]:
return {
"comfyui": (comfyui or "").strip() or "unknown",
"comfyui_frontend": (comfyui_frontend or "").strip() or "unknown",
"desktop": (desktop or "").strip() or "unknown",
}
def detect_anchor_drift(
published_anchors: Optional[Dict[str, Any]],
observed_anchors: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
drift: List[Dict[str, str]] = []
pub = published_anchors or {}
obs = observed_anchors or {}
for key in ANCHOR_KEYS:
published = str(pub.get(key, "unknown"))
observed = str(obs.get(key, "unknown"))
if observed == "unknown":
continue
if published == "unknown":
drift.append(
{
"anchor": key,
"status": "untracked",
"published": published,
"observed": observed,
}
)
continue
if published != observed:
drift.append(
{
"anchor": key,
"status": "drift",
"published": published,
"observed": observed,
}
)
return {
"ok": len(drift) == 0,
"code": "R90_ANCHORS_IN_SYNC" if not drift else "R90_ANCHOR_DRIFT",
"drift": drift,
}
@dataclass
class RefreshWorkflowResult:
ok: bool
matrix_path: str
run_date: str
stages: Dict[str, Any]
decision_codes: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
"ok": self.ok,
"matrix_path": self.matrix_path,
"run_date": self.run_date,
"stages": copy.deepcopy(self.stages),
"decision_codes": list(self.decision_codes),
}
def run_refresh_workflow(
*,
matrix_path: Path | str,
observed_anchors: Optional[Dict[str, str]] = None,
apply: bool = False,
updated_by: str = "script",
today: Optional[date] = None,
) -> RefreshWorkflowResult:
p = Path(matrix_path)
today = today or _utc_now().date()
observed = dict(observed_anchors or normalize_observed_anchors())
doc = read_matrix_document(p)
metadata = copy.deepcopy(doc["metadata"]) if isinstance(doc["metadata"], dict) else None
if metadata is None:
metadata = _default_metadata()
# Preserve compatibility for first adoption while making missing metadata visible.
bootstrap_mode = True
else:
bootstrap_mode = False
validate_before = validate_metadata(doc["metadata"], today=today)
drift_before = detect_anchor_drift(metadata.get("anchors"), observed)
collect_stage = {
"matrix_exists": p.exists(),
"metadata_present": doc["has_meta"],
"body_sha256": doc["body_sha256"],
"observed_anchors": observed,
"doc_issues": list(doc["issues"]),
}
diff_stage = {
"metadata_hash_before": _json_hash(doc["metadata"]) if doc["metadata"] is not None else None,
"drift": drift_before,
"bootstrap_metadata": bootstrap_mode,
}
validate_stage = {
"before": validate_before,
}
publish_stage: Dict[str, Any] = {"mode": "dry-run", "updated": False}
updated_text = doc["text"]
metadata_after = copy.deepcopy(metadata)
metadata_after.setdefault("policy", {})
metadata_after.setdefault("anchors", {})
metadata_after.setdefault("evidence", {})
metadata_after["last_validated_date"] = today.isoformat()
for key in ANCHOR_KEYS:
metadata_after["anchors"][key] = observed.get(key, "unknown")
metadata_after["evidence"]["updated_by"] = updated_by
metadata_after["evidence"]["updated_at"] = _utc_now().isoformat()
metadata_after["evidence"]["evidence_id"] = (
f"compat-matrix-refresh-{today.strftime('%Y%m%d')}"
)
validate_after = validate_metadata(metadata_after, today=today)
drift_after = detect_anchor_drift(metadata_after.get("anchors"), observed)
validate_stage["after"] = validate_after
if apply:
updated_text = replace_metadata_block(doc["text"], metadata_after)
p.write_text(updated_text, encoding="utf-8")
publish_stage = {
"mode": "apply",
"updated": True,
"metadata_hash_after": _json_hash(metadata_after),
"drift_after": drift_after,
"body_sha256_after": hashlib.sha256(
_body_without_meta(updated_text).encode("utf-8")
).hexdigest(),
}
else:
publish_stage = {
"mode": "dry-run",
"updated": False,
"metadata_preview_hash": _json_hash(metadata_after),
"drift_after": drift_after,
}
decision_codes: List[str] = []
decision_codes.append(validate_after["code"])
decision_codes.append(drift_before["code"])
if bootstrap_mode:
decision_codes.append("R90_BOOTSTRAP_METADATA")
if apply:
decision_codes.append("R90_PUBLISH_APPLY")
else:
decision_codes.append("R90_PUBLISH_DRY_RUN")
ok = bool(validate_after["ok"])
stages = {
"collect": collect_stage,
"diff": diff_stage,
"validate": validate_stage,
"publish": publish_stage,
}
return RefreshWorkflowResult(
ok=ok,
matrix_path=str(p),
run_date=today.isoformat(),
stages=stages,
decision_codes=decision_codes,
)
services/crypto_lifecycle_drills.py
+50 -18
View File
@@ -25,7 +25,9 @@ try:
from .sidecar.bridge_contract import BridgeScope
except ImportError:
from services.bridge_token_lifecycle import BridgeTokenStore # type: ignore
from services.registry_quarantine import _HAS_CRYPTO as _HAS_REGISTRY_CRYPTO # type: ignore
from services.registry_quarantine import (
_HAS_CRYPTO as _HAS_REGISTRY_CRYPTO, # type: ignore
)
from services.registry_quarantine import TrustRoot, TrustRootStore # type: ignore
from services.sidecar.bridge_contract import BridgeScope # type: ignore
@@ -58,7 +60,9 @@ class DrillEvidence:
artifacts: List[Dict[str, Any]] = field(default_factory=list)
decision_codes: List[str] = field(default_factory=list)
fail_closed_assertions: List[Dict[str, Any]] = field(default_factory=list)
generated_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
generated_at: str = field(
default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
)
schema_version: int = 1
def to_dict(self) -> Dict[str, Any]:
@@ -117,7 +121,9 @@ class CryptoLifecycleDrillRunner:
def run_many(self, scenarios: Iterable[str]) -> List[Dict[str, Any]]:
return [self.run(s) for s in scenarios]
def write_evidence(self, evidence: List[Dict[str, Any]], output_path: str | Path) -> Path:
def write_evidence(
self, evidence: List[Dict[str, Any]], output_path: str | Path
) -> Path:
path = Path(output_path)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
@@ -135,7 +141,9 @@ class CryptoLifecycleDrillRunner:
f.write("\n")
return path
def _artifact(self, path: Path, *, kind: str, exists: Optional[bool] = None) -> Dict[str, Any]:
def _artifact(
self, path: Path, *, kind: str, exists: Optional[bool] = None
) -> Dict[str, Any]:
return {
"kind": kind,
"path": str(path),
@@ -188,7 +196,9 @@ class CryptoLifecycleDrillRunner:
"active_before_count": len(pre_active),
},
result={
"status": "pass" if all(c["passed"] for c in fail_closed_checks) else "fail",
"status": (
"pass" if all(c["passed"] for c in fail_closed_checks) else "fail"
),
"active_before": pre_active,
"active_after": post_active,
"rotation_overlap_supported": True,
@@ -231,7 +241,9 @@ class CryptoLifecycleDrillRunner:
}
)
decision_codes.append(
"R119_TRUST_ROOT_EMERGENCY_REVOKE_BLOCKED" if passed else "R119_TRUST_ROOT_REVOKE_UNEXPECTED"
"R119_TRUST_ROOT_EMERGENCY_REVOKE_BLOCKED"
if passed
else "R119_TRUST_ROOT_REVOKE_UNEXPECTED"
)
return DrillEvidence(
@@ -313,14 +325,18 @@ class CryptoLifecycleDrillRunner:
scenario=SCENARIO_KEY_LOSS_RECOVERY,
precheck={
"state_dir": str(state_dir),
"crypto_available": bool(getattr(_secrets_encryption, "_HAS_CRYPTO", False)),
"crypto_available": bool(
getattr(_secrets_encryption, "_HAS_CRYPTO", False)
),
"encrypted_store_exists": enc_path.exists(),
"key_file_exists_before_loss": backup_path.exists(),
},
result={
"status": "pass"
if all(c["passed"] for c in fail_closed_checks) and recovered_ok
else "fail",
"status": (
"pass"
if all(c["passed"] for c in fail_closed_checks) and recovered_ok
else "fail"
),
"key_loss_blocked": blocked,
"recovery_loaded": recovered_ok,
"scope_widened": False,
@@ -335,8 +351,16 @@ class CryptoLifecycleDrillRunner:
self._artifact(backup_path, kind="secret_key_backup"),
],
decision_codes=[
"R119_SECRETS_KEY_LOSS_FAIL_CLOSED" if blocked else "R119_SECRETS_KEY_LOSS_NOT_BLOCKED",
"R119_SECRETS_KEY_RECOVERY_OK" if recovered_ok else "R119_SECRETS_KEY_RECOVERY_FAILED",
(
"R119_SECRETS_KEY_LOSS_FAIL_CLOSED"
if blocked
else "R119_SECRETS_KEY_LOSS_NOT_BLOCKED"
),
(
"R119_SECRETS_KEY_RECOVERY_OK"
if recovered_ok
else "R119_SECRETS_KEY_RECOVERY_FAILED"
),
],
fail_closed_assertions=fail_closed_checks,
)
@@ -352,7 +376,9 @@ class CryptoLifecycleDrillRunner:
s.value if hasattr(s, "value") else str(s) for s in token.scopes
}
store.revoke_token(token.token_id, reason="drill_compromise")
validation = store.validate_token(token.device_token, required_scope="job:submit")
validation = store.validate_token(
token.device_token, required_scope="job:submit"
)
fail_closed_checks = [
{
@@ -374,7 +400,9 @@ class CryptoLifecycleDrillRunner:
"issued_scopes": sorted(before_scope_values),
},
result={
"status": "pass" if all(c["passed"] for c in fail_closed_checks) else "fail",
"status": (
"pass" if all(c["passed"] for c in fail_closed_checks) else "fail"
),
"validation_ok": validation.ok,
"reject_reason": validation.reject_reason,
"scope_widened": scope_widened,
@@ -386,13 +414,17 @@ class CryptoLifecycleDrillRunner:
"detail": "Compromised token remains revoked after drill by design.",
},
artifacts=[
self._artifact(state_dir / "bridge_tokens.json", kind="bridge_token_store")
self._artifact(
state_dir / "bridge_tokens.json", kind="bridge_token_store"
)
],
decision_codes=[
"R119_TOKEN_COMPROMISE_REVOKED",
"R119_FAIL_CLOSED_TOKEN_REVOKED"
if fail_closed_checks[0]["passed"]
else "R119_FAIL_CLOSED_UNEXPECTED",
(
"R119_FAIL_CLOSED_TOKEN_REVOKED"
if fail_closed_checks[0]["passed"]
else "R119_FAIL_CLOSED_UNEXPECTED"
),
],
fail_closed_assertions=fail_closed_checks,
)
services/job_events.py
+67
View File
@@ -182,6 +182,73 @@ class JobEventStore:
break
return results
def events_since_bounded(
self,
*,
last_seq: int = 0,
limit: int = 100,
prompt_id: Optional[str] = None,
scan_cap: int = 2000,
) -> tuple[List[JobEvent], Dict[str, Any]]:
"""
R95: Bounded scan variant of events_since() for management endpoints.
Prevents full-buffer traversal when many entries are skipped due to TTL,
prompt filter, or stale cursor ranges. Returns diagnostics so the API can
surface deterministic pagination behavior.
"""
now = time.time()
all_events = self._queue.get_all()
if scan_cap < 1:
scan_cap = 1
results: List[JobEvent] = []
scanned = 0
earliest_retained_seq: Optional[int] = None
latest_retained_seq: Optional[int] = None
for evt in all_events:
if scanned >= scan_cap:
break
scanned += 1
if evt.seq <= last_seq:
continue
if now - evt.timestamp > EVENT_TTL_SEC:
continue
if prompt_id and evt.prompt_id != prompt_id:
continue
if earliest_retained_seq is None:
earliest_retained_seq = evt.seq
latest_retained_seq = evt.seq
results.append(evt)
if len(results) >= limit:
break
# If we didn't collect any matching events, still compute retained bounds
# from a second cheap pass to support stale cursor diagnostics.
if earliest_retained_seq is None or latest_retained_seq is None:
for evt in all_events:
if evt.seq <= 0:
continue
if now - evt.timestamp > EVENT_TTL_SEC:
continue
if prompt_id and evt.prompt_id != prompt_id:
continue
if earliest_retained_seq is None:
earliest_retained_seq = evt.seq
latest_retained_seq = evt.seq
return results, {
"scanned": scanned,
"scan_cap": scan_cap,
"truncated": scanned >= scan_cap and len(results) < limit,
"earliest_retained_seq": earliest_retained_seq,
"latest_retained_seq": latest_retained_seq,
}
def latest_seq(self) -> int:
"""Return the latest sequence number."""
with self._lock:
services/management_query.py
+179
View File
@@ -0,0 +1,179 @@
"""
R95: Management query pagination + bounded-scan helpers.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Iterable, List, Optional
@dataclass
class PaginationMeta:
limit: int
offset: int = 0
cursor: Optional[int] = None
warnings: List[Dict[str, Any]] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"limit": self.limit,
"offset": self.offset,
"warnings": list(self.warnings),
}
if self.cursor is not None:
payload["cursor"] = self.cursor
return payload
@dataclass
class BoundedScanResult:
items: List[Any]
scanned: int
skipped_malformed: int
truncated: bool
def to_dict(self) -> Dict[str, Any]:
return {
"scanned": self.scanned,
"skipped_malformed": self.skipped_malformed,
"truncated": self.truncated,
}
def _warn(code: str, field: str, raw: Any, normalized: Any) -> Dict[str, Any]:
return {
"code": code,
"field": field,
"raw": "" if raw is None else str(raw),
"normalized": normalized,
}
def _parse_int(raw: Any) -> Optional[int]:
if raw is None:
return None
try:
return int(str(raw).strip())
except Exception:
return None
def normalize_limit_offset(
query: Dict[str, Any],
*,
default_limit: int,
max_limit: int,
default_offset: int = 0,
max_offset: Optional[int] = None,
) -> PaginationMeta:
warnings: List[Dict[str, Any]] = []
raw_limit = query.get("limit")
limit = _parse_int(raw_limit)
if limit is None:
if raw_limit is not None:
warnings.append(_warn("R95_INVALID_LIMIT", "limit", raw_limit, default_limit))
limit = default_limit
if limit < 1:
warnings.append(_warn("R95_LIMIT_BELOW_MIN", "limit", raw_limit, 1))
limit = 1
if limit > max_limit:
warnings.append(_warn("R95_LIMIT_CLAMPED", "limit", raw_limit, max_limit))
limit = max_limit
raw_offset = query.get("offset")
offset = _parse_int(raw_offset)
if offset is None:
if raw_offset is not None:
warnings.append(
_warn("R95_INVALID_OFFSET", "offset", raw_offset, default_offset)
)
offset = default_offset
if offset < 0:
warnings.append(_warn("R95_OFFSET_BELOW_MIN", "offset", raw_offset, 0))
offset = 0
if max_offset is not None and offset > max_offset:
warnings.append(_warn("R95_OFFSET_CLAMPED", "offset", raw_offset, max_offset))
offset = max_offset
return PaginationMeta(limit=limit, offset=offset, warnings=warnings)
def normalize_cursor_limit(
query: Dict[str, Any],
*,
cursor_key: str = "since",
default_cursor: int = 0,
min_cursor: int = 0,
default_limit: int,
max_limit: int,
) -> PaginationMeta:
page = normalize_limit_offset(
query,
default_limit=default_limit,
max_limit=max_limit,
default_offset=0,
)
raw_cursor = query.get(cursor_key)
cursor = _parse_int(raw_cursor)
if cursor is None:
if raw_cursor is not None:
page.warnings.append(
_warn(
"R95_INVALID_CURSOR",
cursor_key,
raw_cursor,
default_cursor,
)
)
cursor = default_cursor
if cursor < min_cursor:
page.warnings.append(_warn("R95_CURSOR_BELOW_MIN", cursor_key, raw_cursor, min_cursor))
cursor = min_cursor
page.cursor = cursor
return page
def bounded_scan_collect(
records: Iterable[Any],
*,
skip: int,
take: int,
scan_cap: int,
serializer: Callable[[Any], Any],
) -> BoundedScanResult:
if scan_cap < 1:
scan_cap = 1
scanned = 0
skipped_valid = 0
skipped_malformed = 0
items: List[Any] = []
for rec in records:
if scanned >= scan_cap:
break
scanned += 1
try:
payload = serializer(rec)
except (AttributeError, TypeError, ValueError):
skipped_malformed += 1
continue
if skipped_valid < skip:
skipped_valid += 1
continue
items.append(payload)
if len(items) >= take:
break
truncated = scanned >= scan_cap and len(items) < take
return BoundedScanResult(
items=items,
scanned=scanned,
skipped_malformed=skipped_malformed,
truncated=truncated,
)
services/operator_doctor.py
+126
View File
@@ -278,6 +278,131 @@ def check_contract_files(report: DoctorReport, pack_root: Path) -> None:
)
def check_compatibility_matrix_governance(
report: DoctorReport, pack_root: Path
) -> None:
"""
R90: Compatibility matrix freshness + anchor drift visibility.
Read-only local check. Uses optional env overrides for observed anchors:
- OPENCLAW_COMPAT_ANCHOR_COMFYUI
- OPENCLAW_COMPAT_ANCHOR_COMFYUI_FRONTEND
- OPENCLAW_COMPAT_ANCHOR_DESKTOP
"""
matrix_path = pack_root / "docs" / "release" / "compatibility_matrix.md"
if not matrix_path.exists():
report.add(
CheckResult(
name="compatibility_matrix_governance",
severity=Severity.SKIP.value,
message="Compatibility matrix file missing",
)
)
return
try:
from .compatibility_matrix_governance import (
detect_anchor_drift,
normalize_observed_anchors,
read_matrix_document,
validate_metadata,
)
except Exception as e:
report.add(
CheckResult(
name="compatibility_matrix_governance",
severity=Severity.WARN.value,
message="Compatibility matrix governance helpers unavailable",
detail=str(e),
)
)
return
doc = read_matrix_document(matrix_path)
validation = validate_metadata(doc.get("metadata"))
observed = normalize_observed_anchors(
comfyui=os.environ.get("OPENCLAW_COMPAT_ANCHOR_COMFYUI"),
comfyui_frontend=os.environ.get("OPENCLAW_COMPAT_ANCHOR_COMFYUI_FRONTEND"),
desktop=os.environ.get("OPENCLAW_COMPAT_ANCHOR_DESKTOP"),
)
drift = detect_anchor_drift((doc.get("metadata") or {}).get("anchors"), observed)
report.environment["compat_matrix_validation_code"] = str(
validation.get("code", "")
)
if validation.get("age_days") is not None:
report.environment["compat_matrix_age_days"] = str(validation["age_days"])
report.environment["compat_matrix_drift_code"] = str(drift.get("code", ""))
if not validation.get("ok"):
report.add(
CheckResult(
name="compatibility_matrix_governance",
severity=Severity.WARN.value,
message="Compatibility matrix metadata invalid",
detail=json.dumps(
{
"doc_issues": doc.get("issues", []),
"violations": validation.get("violations", []),
},
ensure_ascii=False,
),
remediation=(
"Repair the metadata block in docs/release/compatibility_matrix.md "
"or run scripts/compatibility_matrix_refresh.py --apply."
),
)
)
return
age_days = validation.get("age_days")
status = validation.get("status")
if status in ("warning", "stale"):
report.add(
CheckResult(
name="compatibility_matrix_governance",
severity=Severity.WARN.value,
message=(
"Compatibility matrix age exceeds warning policy"
if status == "warning"
else "Compatibility matrix is stale"
),
detail=json.dumps(
{
"age_days": age_days,
"warn_age_days": validation.get("warn_age_days"),
"max_age_days": validation.get("max_age_days"),
}
),
remediation=(
"Refresh and validate the compatibility matrix before release tagging "
"(scripts/compatibility_matrix_refresh.py)."
),
)
)
else:
report.add(
CheckResult(
name="compatibility_matrix_governance",
severity=Severity.PASS.value,
message=f"Compatibility matrix metadata fresh (age={age_days}d)",
)
)
if not drift.get("ok"):
report.add(
CheckResult(
name="compatibility_matrix_anchor_drift",
severity=Severity.WARN.value,
message="Compatibility matrix anchor drift detected",
detail=json.dumps(drift.get("drift", []), ensure_ascii=False),
remediation=(
"Refresh matrix anchors and publish updated evidence before release."
),
)
)
def check_state_dir(report: DoctorReport) -> None:
"""Check state directory accessibility."""
state_dir = os.environ.get("MOLTBOT_STATE_DIR") or os.environ.get(
@@ -727,6 +852,7 @@ def run_doctor(pack_root: Optional[Path] = None) -> DoctorReport:
check_state_dir(report)
check_token_posture(report)
check_contract_files(report, pack_root)
check_compatibility_matrix_governance(report, pack_root) # R90
check_core_imports(report)
report.build_summary()
services/runtime_guardrails.py
+13 -7
View File
@@ -241,7 +241,10 @@ def _emit_degraded_guardrails_audit(snapshot: Dict[str, Any]) -> None:
from .audit_events import build_audit_event, emit_audit_event
except ImportError:
try:
from services.audit_events import build_audit_event, emit_audit_event # type: ignore
from services.audit_events import ( # type: ignore
build_audit_event,
emit_audit_event,
)
except ImportError:
return
@@ -266,7 +269,9 @@ def reset_runtime_guardrails_audit_cache() -> None:
_AUDIT_FINGERPRINTS_EMITTED.clear()
def strip_runtime_only_config_fields(config_blob: Dict[str, Any]) -> Tuple[Dict[str, Any], List[dict]]:
def strip_runtime_only_config_fields(
config_blob: Dict[str, Any]
) -> Tuple[Dict[str, Any], List[dict]]:
"""
Remove runtime-only guardrail fields from config blobs before persistence/use.
@@ -283,7 +288,9 @@ def strip_runtime_only_config_fields(config_blob: Dict[str, Any]) -> Tuple[Dict[
"code": CODE_RUNTIME_ONLY_STRIPPED,
"path": key,
"reason": "runtime_only_guardrails_not_persisted",
"removed_type": type(removed).__name__ if removed is not None else "none",
"removed_type": (
type(removed).__name__ if removed is not None else "none"
),
}
)
@@ -297,9 +304,9 @@ def strip_runtime_only_config_fields(config_blob: Dict[str, Any]) -> Tuple[Dict[
"code": CODE_RUNTIME_ONLY_STRIPPED,
"path": f"llm.{key}",
"reason": "runtime_only_guardrails_not_persisted",
"removed_type": type(removed).__name__
if removed is not None
else "none",
"removed_type": (
type(removed).__name__ if removed is not None else "none"
),
}
)
@@ -314,4 +321,3 @@ def payload_contains_runtime_guardrails(payload: Dict[str, Any]) -> bool:
return True
llm = payload.get("llm")
return isinstance(llm, dict) and any(key in llm for key in RUNTIME_ONLY_CONFIG_KEYS)
tests/test_r119_crypto_lifecycle_drills.py
+7 -2
View File
@@ -44,9 +44,14 @@ class TestR119CryptoLifecycleDrills(unittest.TestCase):
# Emergency revoke / token compromise must fail-closed.
self.assertEqual(drills["emergency_revoke"]["result"]["status"], "pass")
self.assertTrue(
any(a["passed"] for a in drills["emergency_revoke"]["fail_closed_assertions"])
any(
a["passed"]
for a in drills["emergency_revoke"]["fail_closed_assertions"]
)
)
self.assertEqual(
drills["token_compromise"]["result"]["reject_reason"], "token_revoked"
)
self.assertEqual(drills["token_compromise"]["result"]["reject_reason"], "token_revoked")
self.assertTrue(
drills["token_compromise"]["result"]["scope_widened"] is False,
"Drill flow must not widen privileges",
tests/test_r90_compatibility_matrix_governance.py
+173
View File
@@ -0,0 +1,173 @@
"""
R90 compatibility matrix governance tests.
"""
from __future__ import annotations
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from services.compatibility_matrix_governance import (
detect_anchor_drift,
read_matrix_document,
run_refresh_workflow,
validate_metadata,
)
from services.operator_doctor import DoctorReport, check_compatibility_matrix_governance
REPO_ROOT = Path(__file__).resolve().parents[1]
class TestR90CompatMatrixGovernance(unittest.TestCase):
def test_repo_matrix_has_valid_metadata(self):
doc = read_matrix_document(REPO_ROOT / "docs" / "release" / "compatibility_matrix.md")
self.assertTrue(doc["has_meta"], msg=doc["issues"])
validation = validate_metadata(doc["metadata"])
self.assertTrue(validation["ok"], msg=validation)
self.assertIn(validation["status"], ("fresh", "warning", "stale"))
def test_detect_anchor_drift(self):
published = {
"comfyui": "a",
"comfyui_frontend": "b",
"desktop": "c",
}
observed = {
"comfyui": "a",
"comfyui_frontend": "b2",
"desktop": "unknown",
}
drift = detect_anchor_drift(published, observed)
self.assertFalse(drift["ok"])
self.assertEqual(drift["code"], "R90_ANCHOR_DRIFT")
self.assertEqual(drift["drift"][0]["anchor"], "comfyui_frontend")
def test_validate_stale_metadata(self):
metadata = {
"schema_version": 1,
"last_validated_date": "2020-01-01",
"policy": {"warn_age_days": 1, "max_age_days": 2},
"anchors": {
"comfyui": "unknown",
"comfyui_frontend": "unknown",
"desktop": "unknown",
},
}
validation = validate_metadata(metadata)
self.assertTrue(validation["ok"])
self.assertEqual(validation["status"], "stale")
self.assertEqual(validation["code"], "R90_MATRIX_STALE")
def test_refresh_workflow_dry_run_and_apply(self):
src = REPO_ROOT / "docs" / "release" / "compatibility_matrix.md"
with tempfile.TemporaryDirectory() as td:
matrix = Path(td) / "compatibility_matrix.md"
matrix.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
dry = run_refresh_workflow(
matrix_path=matrix,
observed_anchors={
"comfyui": "core-1",
"comfyui_frontend": "fe-1",
"desktop": "desktop-1",
},
apply=False,
updated_by="test",
)
dry_payload = dry.to_dict()
self.assertIn("collect", dry_payload["stages"])
self.assertEqual(dry_payload["stages"]["publish"]["mode"], "dry-run")
self.assertFalse(dry_payload["stages"]["publish"]["updated"])
applied = run_refresh_workflow(
matrix_path=matrix,
observed_anchors={
"comfyui": "core-2",
"comfyui_frontend": "fe-2",
"desktop": "desktop-2",
},
apply=True,
updated_by="test",
)
self.assertTrue(applied.ok)
doc = read_matrix_document(matrix)
self.assertEqual(doc["metadata"]["anchors"]["comfyui"], "core-2")
self.assertEqual(doc["metadata"]["evidence"]["updated_by"], "test")
def test_operator_doctor_warns_when_matrix_stale(self):
with tempfile.TemporaryDirectory() as td:
pack_root = Path(td)
matrix_path = pack_root / "docs" / "release"
matrix_path.mkdir(parents=True, exist_ok=True)
matrix_path.joinpath("compatibility_matrix.md").write_text(
(
"# Compatibility Matrix\n\n"
"```openclaw-compat-matrix-meta\n"
+ json.dumps(
{
"schema_version": 1,
"last_validated_date": "2020-01-01",
"policy": {"warn_age_days": 1, "max_age_days": 2},
"anchors": {
"comfyui": "unknown",
"comfyui_frontend": "unknown",
"desktop": "unknown",
},
}
)
+ "\n```\n\nbody\n"
),
encoding="utf-8",
)
report = DoctorReport()
check_compatibility_matrix_governance(report, pack_root)
checks = {c.name: c for c in report.checks}
self.assertIn("compatibility_matrix_governance", checks)
self.assertEqual(checks["compatibility_matrix_governance"].severity, "warn")
self.assertEqual(report.environment["compat_matrix_validation_code"], "R90_MATRIX_STALE")
def test_script_smoke_emits_evidence(self):
with tempfile.TemporaryDirectory() as td:
matrix = Path(td) / "compatibility_matrix.md"
matrix.write_text(
(REPO_ROOT / "docs" / "release" / "compatibility_matrix.md").read_text(
encoding="utf-8"
),
encoding="utf-8",
)
out = Path(td) / "evidence.json"
proc = subprocess.run(
[
sys.executable,
str(REPO_ROOT / "scripts" / "compatibility_matrix_refresh.py"),
"--matrix-path",
str(matrix),
"--anchor-comfyui",
"core-x",
"--anchor-frontend",
"fe-x",
"--anchor-desktop",
"desk-x",
"--output",
str(out),
"--pretty",
],
capture_output=True,
text=True,
check=False,
)
self.assertEqual(proc.returncode, 0, msg=proc.stderr or proc.stdout)
payload = json.loads(out.read_text(encoding="utf-8"))
self.assertIn("stages", payload)
self.assertIn("collect", payload["stages"])
self.assertIn("R90_PUBLISH_DRY_RUN", payload["decision_codes"])
if __name__ == "__main__":
unittest.main()
tests/test_r95_management_query_contracts.py
+256
View File
@@ -0,0 +1,256 @@
"""
R95 management query pagination and bounded-scan contract tests.
"""
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import api.approvals
import api.events
from services.management_query import (
bounded_scan_collect,
normalize_cursor_limit,
normalize_limit_offset,
)
class _DummyApproval:
def __init__(self, approval_id: str):
self.approval_id = approval_id
def to_dict(self):
return {"approval_id": self.approval_id}
class _BadApprovalNoSerializer:
pass
class _ExplodingApproval:
def to_dict(self):
raise RuntimeError("backend explode")
class _DummyEvent:
def __init__(self, seq: int):
self.seq = seq
def to_dict(self):
return {"seq": self.seq}
class TestR95PaginationHelpers(unittest.TestCase):
def test_normalize_limit_offset_clamps_and_warns(self):
page = normalize_limit_offset(
{"limit": "9999", "offset": "-10"},
default_limit=100,
max_limit=500,
max_offset=5000,
)
self.assertEqual(page.limit, 500)
self.assertEqual(page.offset, 0)
codes = {w["code"] for w in page.warnings}
self.assertIn("R95_LIMIT_CLAMPED", codes)
self.assertIn("R95_OFFSET_BELOW_MIN", codes)
def test_normalize_cursor_limit_invalid_cursor_defaults(self):
page = normalize_cursor_limit(
{"since": "bad", "limit": "0"},
cursor_key="since",
default_cursor=0,
min_cursor=0,
default_limit=50,
max_limit=200,
)
self.assertEqual(page.cursor, 0)
self.assertEqual(page.limit, 1)
codes = {w["code"] for w in page.warnings}
self.assertIn("R95_INVALID_CURSOR", codes)
self.assertIn("R95_LIMIT_BELOW_MIN", codes)
def test_bounded_scan_collect_skips_malformed_and_not_swallow_runtime_errors(self):
result = bounded_scan_collect(
[_DummyApproval("a"), _BadApprovalNoSerializer(), _DummyApproval("b")],
skip=0,
take=10,
scan_cap=10,
serializer=lambda x: x.to_dict(),
)
self.assertEqual([i["approval_id"] for i in result.items], ["a", "b"])
self.assertEqual(result.skipped_malformed, 1)
with self.assertRaises(RuntimeError):
bounded_scan_collect(
[_ExplodingApproval()],
skip=0,
take=1,
scan_cap=10,
serializer=lambda x: x.to_dict(),
)
class TestR95EventsApi(unittest.IsolatedAsyncioTestCase):
async def test_events_poll_normalizes_and_resets_stale_cursor(self):
req = MagicMock()
req.query = {"since": "3", "limit": "2"}
req.headers = {}
class StubStore:
def __init__(self):
self.calls = []
def latest_seq(self):
return 100
def events_since_bounded(self, **kwargs):
self.calls.append(kwargs)
if len(self.calls) == 1:
return [], {
"scanned": 5,
"scan_cap": kwargs["scan_cap"],
"truncated": False,
"earliest_retained_seq": 50,
"latest_retained_seq": 100,
}
return [_DummyEvent(50), _DummyEvent(51)], {
"scanned": 2,
"scan_cap": kwargs["scan_cap"],
"truncated": False,
"earliest_retained_seq": 50,
"latest_retained_seq": 100,
}
store = StubStore()
fake_web = SimpleNamespace(json_response=MagicMock(return_value=SimpleNamespace(status=200)))
with (
patch.object(api.events, "web", fake_web),
patch.object(api.events, "check_rate_limit", return_value=True),
patch.object(api.events, "require_observability_access", return_value=(True, None)),
patch.object(api.events, "get_job_event_store", return_value=store),
):
resp = await api.events.events_poll_handler(req)
self.assertEqual(resp.status, 200)
payload = fake_web.json_response.call_args.args[0]
self.assertTrue(payload["ok"])
self.assertEqual(payload["pagination"]["cursor_status"], "stale_cursor_reset")
self.assertEqual(payload["pagination"]["since_requested"], 3)
self.assertEqual(payload["pagination"]["since_effective"], 49)
self.assertEqual([e["seq"] for e in payload["events"]], [50, 51])
self.assertEqual(len(store.calls), 2)
async def test_events_poll_future_cursor_and_invalid_limit(self):
req = MagicMock()
req.query = {"since": "999", "limit": "bad"}
req.headers = {}
class StubStore:
def latest_seq(self):
return 10
def events_since_bounded(self, **kwargs):
return [], {
"scanned": 0,
"scan_cap": kwargs["scan_cap"],
"truncated": False,
"earliest_retained_seq": None,
"latest_retained_seq": 10,
}
fake_web = SimpleNamespace(json_response=MagicMock(return_value=SimpleNamespace(status=200)))
with (
patch.object(api.events, "web", fake_web),
patch.object(api.events, "check_rate_limit", return_value=True),
patch.object(api.events, "require_observability_access", return_value=(True, None)),
patch.object(api.events, "get_job_event_store", return_value=StubStore()),
):
await api.events.events_poll_handler(req)
payload = fake_web.json_response.call_args.args[0]
self.assertEqual(payload["pagination"]["cursor_status"], "future_cursor_reset")
self.assertEqual(payload["pagination"]["since_effective"], 10)
codes = {w["code"] for w in payload["pagination"]["warnings"]}
self.assertIn("R95_INVALID_LIMIT", codes)
self.assertIn("R95_STALE_CURSOR_FUTURE", codes)
async def test_events_poll_does_not_swallow_backend_errors(self):
req = MagicMock()
req.query = {}
req.headers = {}
class StubStore:
def latest_seq(self):
return 1
def events_since_bounded(self, **kwargs):
raise RuntimeError("store failure")
with (
patch.object(api.events, "check_rate_limit", return_value=True),
patch.object(api.events, "require_observability_access", return_value=(True, None)),
patch.object(api.events, "get_job_event_store", return_value=StubStore()),
):
with self.assertRaises(RuntimeError):
await api.events.events_poll_handler(req)
class TestR95ApprovalsApi(unittest.IsolatedAsyncioTestCase):
async def test_approvals_list_normalizes_pagination_and_skips_malformed(self):
req = MagicMock()
req.query = {"limit": "9999", "offset": "-7"}
handler = api.approvals.ApprovalHandlers(require_admin_token_fn=lambda _r: (True, None))
handler._service = MagicMock()
handler._service.list_all.return_value = [
_DummyApproval("a1"),
_BadApprovalNoSerializer(),
_DummyApproval("a2"),
]
handler._service.count_pending.return_value = 1
fake_web = SimpleNamespace(json_response=MagicMock(return_value=SimpleNamespace(status=200)))
with patch.object(api.approvals, "web", fake_web):
resp = await handler.list_approvals(req)
self.assertEqual(resp.status, 200)
payload = fake_web.json_response.call_args.args[0]
self.assertEqual(payload["count"], 2)
self.assertEqual([a["approval_id"] for a in payload["approvals"]], ["a1", "a2"])
self.assertEqual(payload["pagination"]["limit"], 500)
self.assertEqual(payload["pagination"]["offset"], 0)
self.assertEqual(payload["scan"]["skipped_malformed"], 1)
warn_codes = {w["code"] for w in payload["pagination"]["warnings"]}
self.assertIn("R95_LIMIT_CLAMPED", warn_codes)
self.assertIn("R95_OFFSET_BELOW_MIN", warn_codes)
async def test_approvals_list_invalid_status_still_400(self):
req = MagicMock()
req.query = {"status": "not-a-status"}
handler = api.approvals.ApprovalHandlers(require_admin_token_fn=lambda _r: (True, None))
fake_web = SimpleNamespace(
json_response=MagicMock(return_value=SimpleNamespace(status=400))
)
with patch.object(api.approvals, "web", fake_web):
resp = await handler.list_approvals(req)
self.assertEqual(resp.status, 400)
async def test_approvals_list_does_not_swallow_backend_errors(self):
req = MagicMock()
req.query = {}
handler = api.approvals.ApprovalHandlers(require_admin_token_fn=lambda _r: (True, None))
handler._service = MagicMock()
handler._service.list_all.side_effect = RuntimeError("db failed")
with self.assertRaises(RuntimeError):
await handler.list_approvals(req)
if __name__ == "__main__":
unittest.main()
tests/test_s66_api_config_guardrails.py
+3 -1
View File
@@ -35,7 +35,9 @@ class TestS66ConfigApiGuardrails(unittest.TestCase):
mock_guardrails.return_value = {
"status": "degraded",
"code": "S66_GUARDRAILS_DEGRADED",
"violations": [{"code": "S66_INVALID_INT", "path": "timeout_retry.llm_timeout_cap_sec"}],
"violations": [
{"code": "S66_INVALID_INT", "path": "timeout_retry.llm_timeout_cap_sec"}
],
"values": {},
"sources": {},
"runtime_only": True,
tests/test_s66_runtime_guardrails.py
+3 -1
View File
@@ -48,7 +48,9 @@ class TestS66RuntimeGuardrails(unittest.TestCase):
self.assertEqual(snap["code"], CODE_OK)
self.assertTrue(snap["runtime_only"])
self.assertEqual(snap["values"]["timeout_retry"]["llm_timeout_cap_sec"], 300)
self.assertEqual(snap["values"]["bounded_queues"]["max_inflight_submits_total"], 2)
self.assertEqual(
snap["values"]["bounded_queues"]["max_inflight_submits_total"], 2
)
def test_invalid_and_clamped_envs_degrade_with_machine_codes(self):
from services.runtime_guardrails import (