ci(governance): harden aggregate against plan tampering (A1-A8)

.github/workflows/work-lab-gate.yml

@@ -264,11 +264,14 @@ jobs:
           GATE_PLAN_JSON: ${{ needs.plan.outputs.plan_json }}
           EXPECTED_PLAN_DIGEST: ${{ needs.plan.outputs.plan_digest }}
           EXPECTED_HEAD_SHA: ${{ github.sha }}
+          EXPECTED_REPOSITORY: ${{ github.repository }}
           WORKFLOW_RESULT: ${{ needs.workflow.result }}
           OBSERVER_RESULT: ${{ needs.observer.result }}
           TOKEN_MONITOR_RESULT: ${{ needs.token-monitor.result }}
           SUPPLY_CHAIN_SECURITY_RESULT: ${{ needs.supply-chain-security.result }}
           INTEGRATION_RESULT: ${{ needs.integration.result }}
         run: |
           test "$PLAN_RESULT" = success
-          python -c 'import json, os, subprocess, sys; payload=json.dumps({"gate_plan": json.loads(os.environ["GATE_PLAN_JSON"]), "expected_plan_digest": os.environ["EXPECTED_PLAN_DIGEST"], "expected_head_sha": os.environ["EXPECTED_HEAD_SHA"], "jobs": {"workflow": os.environ["WORKFLOW_RESULT"], "observer": os.environ["OBSERVER_RESULT"], "token-monitor": os.environ["TOKEN_MONITOR_RESULT"], "supply-chain-security": os.environ["SUPPLY_CHAIN_SECURITY_RESULT"], "integration": os.environ["INTEGRATION_RESULT"]}}); result=subprocess.run([sys.executable, "scripts/ci/aggregate_gate.py"], input=payload, text=True); raise SystemExit(result.returncode)'
+          EXPECTED_HEAD_TREE="$(git rev-parse 'HEAD^{tree}')"
+          export EXPECTED_HEAD_TREE
+          python -c 'import json, os, subprocess, sys; payload=json.dumps({"gate_plan": json.loads(os.environ["GATE_PLAN_JSON"]), "expected_plan_digest": os.environ["EXPECTED_PLAN_DIGEST"], "expected_head_sha": os.environ["EXPECTED_HEAD_SHA"], "expected_head_tree": os.environ["EXPECTED_HEAD_TREE"], "expected_repository": os.environ["EXPECTED_REPOSITORY"], "jobs": {"workflow": os.environ["WORKFLOW_RESULT"], "observer": os.environ["OBSERVER_RESULT"], "token-monitor": os.environ["TOKEN_MONITOR_RESULT"], "supply-chain-security": os.environ["SUPPLY_CHAIN_SECURITY_RESULT"], "integration": os.environ["INTEGRATION_RESULT"]}}); result=subprocess.run([sys.executable, "scripts/ci/aggregate_gate.py"], input=payload, text=True); raise SystemExit(result.returncode)'

00-governance/generated/CURRENT_STATE.json

@@ -1,7 +1,7 @@
 {
   "schema_version": "work-lab-current-state/v1",
-  "generated_at": "2026-08-15T12:16:15.112025Z",
-  "source_digest": "da6d0999f0a53746994745b950dfc8259d7b7c1340d35719c268d05ce161e2fa",
+  "generated_at": "2026-08-15T12:44:43.562105Z",
+  "source_digest": "cd6355e8621818b973d4277c6a4773f7c14e5aaf00f6fa548dc68415652b3608",
   "checkout_attestation": {
     "status": "RUNTIME_REQUIRED",
     "tracked_projection": "NO_HEAD_OR_BRANCH_CLAIM",
@@ -219,5 +219,5 @@
     "real_device_validation",
     "commercial_release"
   ],
-  "content_digest": "3d68e0203302fb14d3eb660b1c762d1bb318b9d29ab2b8fcd5cd58055eb5019d"
+  "content_digest": "e3857b619b3cd604bbae67fb37f080a06cc3ba54d31ed122f74056db44370ac7"
 }

00-governance/generated/CURRENT_STATE.md

@@ -1,8 +1,8 @@
 # WORK-LAB current state
 
-Generated at: `2026-08-15T12:16:15.112025Z`  \
-Source digest: `da6d0999f0a53746994745b950dfc8259d7b7c1340d35719c268d05ce161e2fa`  \
-Content digest: `3d68e0203302fb14d3eb660b1c762d1bb318b9d29ab2b8fcd5cd58055eb5019d`
+Generated at: `2026-08-15T12:44:43.562105Z`  \
+Source digest: `cd6355e8621818b973d4277c6a4773f7c14e5aaf00f6fa548dc68415652b3608`  \
+Content digest: `e3857b619b3cd604bbae67fb37f080a06cc3ba54d31ed122f74056db44370ac7`
 
 ## Git and CI attestation
 

scripts/ci/aggregate_gate.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
+import fnmatch
 import hashlib
 import json
 import sys
+from datetime import datetime
+from pathlib import Path
+
+import yaml
 
 REQUIRED = {"workflow", "observer", "token-monitor", "supply-chain-security", "integration"}
 PLAN_GATES = {"workflow", "observer", "token-monitor", "supply-chain-security", "integration"}
@@ -13,6 +18,11 @@
     "supply-chain-security": "supply-chain-security",
     "integration": "integration",
 }
+EXPECTED_PLAN_ID = "work-lab-gate"
+ROOT = Path(__file__).resolve().parents[2]
+PROFILE = ROOT / "00-governance" / "work-lab.project-profile.yaml"
+RISK_VALUES = {"low", "medium", "high", "critical"}
+DELIVERY_VALUES = {"none", "commit", "push", "pull_request", "merge", "release"}
 
 
 def _plan_digest(plan: dict[str, object]) -> str:
@@ -21,18 +31,53 @@ def _plan_digest(plan: dict[str, object]) -> str:
     return hashlib.sha256(canonical).hexdigest()
 
 
-def _resolve_plan(data: dict[str, object]) -> tuple[set[str], str | None, str | None]:
+def _matches(path: str, patterns: list[str]) -> bool:
+    normalized = path.replace("\\", "/")
+    return any(
+        fnmatch.fnmatch(normalized, pattern) or normalized.startswith(pattern.rstrip("*").rstrip("/"))
+        for pattern in patterns
+    )
+
+
+def _load_critical_prefixes() -> list[str]:
+    if not PROFILE.is_file():
+        raise ValueError(f"project profile missing: {PROFILE}; cannot re-derive critical risk")
+    data = yaml.safe_load(PROFILE.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError("project profile must be an object")
+    prefixes = (data.get("risk_zones") or {}).get("critical")
+    if not isinstance(prefixes, list) or not prefixes or not all(isinstance(p, str) and p for p in prefixes):
+        raise ValueError("project profile risk_zones.critical must be a non-empty string list")
+    return prefixes
+
+
+def _is_critical(changed_paths: list[str]) -> bool:
+    """A6: re-derive critical risk from the profile's risk_zones, never trust
+    the plan's own risk field, so a candidate cannot evade full gates by
+    editing the planner."""
+    prefixes = _load_critical_prefixes()
+    return any(_matches(path, prefixes) for path in changed_paths)
+
+
+def _resolve_plan(data: dict[str, object]) -> tuple[set[str], str, str, str, str, list[str]]:
     raw_plan = data.get("gate_plan")
     if raw_plan is None:
-        return set(REQUIRED), None, None
+        raise ValueError("gate_plan is required")
     if not isinstance(raw_plan, dict):
         raise ValueError("gate_plan must be an object")
     if raw_plan.get("schema_version") != "workflow/gate-plan/v1":
         raise ValueError("gate_plan schema_version must be workflow/gate-plan/v1")
+    if raw_plan.get("plan_id") != EXPECTED_PLAN_ID:
+        raise ValueError(f"gate_plan plan_id must be {EXPECTED_PLAN_ID}")
     required = raw_plan.get("required_gates")
     changed_paths = raw_plan.get("changed_paths")
     digest = raw_plan.get("plan_digest")
     source_identity = raw_plan.get("source_identity")
+    skipped = raw_plan.get("skipped_gates")
+    risk = raw_plan.get("risk")
+    delivery_effect = raw_plan.get("delivery_effect")
+    platform_scope = raw_plan.get("platform_scope")
+    generated_at = raw_plan.get("generated_at")
     if not isinstance(required, list) or not all(isinstance(item, str) for item in required):
         raise ValueError("gate_plan.required_gates must be a string list")
     if len(required) != len(set(required)):
@@ -56,47 +101,103 @@ def _resolve_plan(data: dict[str, object]) -> tuple[set[str], str | None, str |
         raise ValueError("gate_plan digest mismatch")
     if not isinstance(source_identity, dict):
         raise ValueError("gate_plan.source_identity must be an object")
+    repository = source_identity.get("repository")
     commit_identity = source_identity.get("commit")
     tree_identity = source_identity.get("tree")
+    if not isinstance(repository, str) or not repository:
+        raise ValueError("gate_plan source repository is required")
     if not isinstance(commit_identity, dict) or not isinstance(commit_identity.get("oid"), str) or not commit_identity["oid"]:
         raise ValueError("gate_plan source commit oid is required")
     if not isinstance(tree_identity, dict) or not isinstance(tree_identity.get("oid"), str) or not tree_identity["oid"]:
         raise ValueError("gate_plan source tree oid is required")
     commit = commit_identity["oid"]
-    return normalized_required, digest["value"], commit
+    tree = tree_identity["oid"]
+    if not isinstance(skipped, list) or not all(isinstance(item, dict) for item in skipped):
+        raise ValueError("gate_plan.skipped_gates must be an object list")
+    skipped_ids: set[str] = set()
+    for item in skipped:
+        gate_id = item.get("gate_id")
+        reason = item.get("reason")
+        if not isinstance(gate_id, str) or not gate_id:
+            raise ValueError("gate_plan skipped_gates[].gate_id is required")
+        if not isinstance(reason, str) or not reason:
+            raise ValueError("gate_plan skipped_gates[].reason is required")
+        skipped_ids.add(gate_id)
+    if skipped_ids != PLAN_GATES - normalized_required:
+        missing = sorted((PLAN_GATES - normalized_required) - skipped_ids)
+        raise ValueError(f"skipped_gates must cover exactly the non-required gates; missing: {missing}")
+    if risk not in RISK_VALUES:
+        raise ValueError(f"gate_plan.risk must be one of {sorted(RISK_VALUES)}")
+    if delivery_effect not in DELIVERY_VALUES:
+        raise ValueError(f"gate_plan.delivery_effect must be one of {sorted(DELIVERY_VALUES)}")
+    if (
+        not isinstance(platform_scope, list)
+        or not platform_scope
+        or len(platform_scope) != len(set(platform_scope))
+        or not all(isinstance(item, str) and item for item in platform_scope)
+    ):
+        raise ValueError("gate_plan.platform_scope must be a non-empty unique string list")
+    if not isinstance(generated_at, str) or not generated_at:
+        raise ValueError("gate_plan.generated_at is required")
+    try:
+        datetime.fromisoformat(generated_at.replace("Z", "+00:00"))
+    except ValueError:
+        raise ValueError("gate_plan.generated_at must be ISO8601")
+    if _is_critical(changed_paths) and normalized_required != PLAN_GATES:
+        raise ValueError("critical changed paths require ALL gates; plan under-selects")
+    return normalized_required, digest["value"], commit, tree, repository, sorted(changed_paths)
+
+
+def _fail(reason: str, digest: str | None) -> int:
+    result: dict[str, object] = {"status": "FAIL", "reason": reason}
+    if digest is not None:
+        result["plan_digest"] = digest
+    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
+    return 1
 
 
 def main(payload: str) -> int:
     try:
         data = json.loads(payload)
         if not isinstance(data, dict):
             raise ValueError("aggregate payload must be an object")
-        required, digest, plan_commit = _resolve_plan(data)
+        required, digest, plan_commit, plan_tree, plan_repository, _ = _resolve_plan(data)
         expected_digest = data.get("expected_plan_digest")
-        if digest is not None:
-            if not isinstance(expected_digest, str) or not expected_digest:
-                raise ValueError("aggregate expected plan digest is required")
-            if expected_digest != digest:
-                raise ValueError("aggregate expected plan digest does not match plan")
+        if not isinstance(expected_digest, str) or not expected_digest:
+            raise ValueError("aggregate expected plan digest is required")
+        if expected_digest != digest:
+            raise ValueError("aggregate expected plan digest does not match plan")
         expected_commit = data.get("expected_head_sha")
-        if plan_commit is not None:
-            if not isinstance(expected_commit, str) or not expected_commit:
-                raise ValueError("aggregate expected head SHA is required")
-            if expected_commit != plan_commit:
-                raise ValueError("aggregate plan commit does not match head SHA")
+        if not isinstance(expected_commit, str) or not expected_commit:
+            raise ValueError("aggregate expected head SHA is required")
+        if expected_commit != plan_commit:
+            raise ValueError("aggregate plan commit does not match head SHA")
+        expected_tree = data.get("expected_head_tree")
+        if not isinstance(expected_tree, str) or not expected_tree:
+            raise ValueError("aggregate expected head tree is required")
+        if expected_tree != plan_tree:
+            raise ValueError("aggregate plan tree does not match head tree")
+        expected_repository = data.get("expected_repository")
+        if not isinstance(expected_repository, str) or not expected_repository:
+            raise ValueError("aggregate expected repository is required")
+        if expected_repository != plan_repository:
+            raise ValueError("aggregate plan repository does not match expected repository")
         jobs = data.get("jobs", {})
         if not isinstance(jobs, dict):
             raise ValueError("aggregate jobs must be an object")
         bad = [JOB_NAMES[name] for name in required if jobs.get(JOB_NAMES[name]) not in {"success", "passed"}]
         if bad:
-            result = {"status": "FAIL", "missing_or_failed": sorted(bad)}
-            if digest is not None:
-                result["plan_digest"] = digest
-            print(json.dumps(result, ensure_ascii=False, sort_keys=True))
-            return 1
-        result = {"status": "PASS", "required": sorted(JOB_NAMES[name] for name in required)}
-        if digest is not None:
-            result["plan_digest"] = digest
+            return _fail(f"required job(s) missing or failed: {sorted(bad)}", digest)
+        not_required = PLAN_GATES - required
+        not_skipped = [JOB_NAMES[name] for name in not_required if jobs.get(JOB_NAMES[name]) != "skipped"]
+        if not_skipped:
+            return _fail(f"non-selected gate(s) not explicitly skipped: {sorted(not_skipped)}", digest)
+        result: dict[str, object] = {
+            "status": "PASS",
+            "required": sorted(JOB_NAMES[name] for name in required),
+            "skipped": sorted(JOB_NAMES[name] for name in not_required),
+        }
+        result["plan_digest"] = digest
         print(json.dumps(result, ensure_ascii=False, sort_keys=True))
         return 0
     except (ValueError, TypeError, json.JSONDecodeError) as exc:

scripts/ci/verify_capability_conformance.py

@@ -30,11 +30,19 @@ def verify_document(document: dict[str, Any], root: Path = ROOT) -> dict[str, in
             total += 1
             if FORBIDDEN.intersection(entry["permissions"]):
                 raise ValueError(f"{protocol_name}/{entry['id']}: write/execute/network permissions require explicit approval")
-            if any(token in json.dumps(entry).lower() for token in ("open-design", "opendesign-assistance")):
-                raise ValueError(f"{protocol_name}/{entry['id']}: retired Open Design capability")
-    adapter_ids = {entry["id"] for entry in json.loads((root / ADAPTERS.relative_to(ROOT)).read_text(encoding="utf-8"))["entries"]}
-    if "open-design" in adapter_ids:
-        raise ValueError("adapter registry still contains retired Open Design adapter")
+            if any(token in json.dumps(entry).lower() for token in ("opendesign-assistance", "open-design-assistance")):
+                raise ValueError(f"{protocol_name}/{entry['id']}: retired Open Design migration alias capability")
+    adapter_entries = json.loads((root / ADAPTERS.relative_to(ROOT)).read_text(encoding="utf-8"))["entries"]
+    adapter_by_id = {entry["id"]: entry for entry in adapter_entries}
+    for client_id in ("open-design", "openhuman"):
+        entry = adapter_by_id.get(client_id)
+        if entry is None:
+            raise ValueError(f"adapter registry missing client adapter: {client_id}")
+        if entry.get("support_level") != "experimental":
+            raise ValueError(f"{client_id} adapter must be experimental until a reviewed official interface exists")
+        operations = set(entry.get("operations", []))
+        if not operations.issubset({"detect", "capabilities", "observe"}):
+            raise ValueError(f"{client_id} adapter must be read-only (detect/capabilities/observe)")
     if not (root / SKILLS.relative_to(ROOT)).is_file():
         raise ValueError("skill provenance manifest missing")
     return {"protocols": 3, "entries": total, "mcp_unverified": 1}