feat(governed): absorb control-plane integrity hardening into authority source line

Turn historical AIOS failure modes (dual-plane divergence, fake execution/
PASS-without-evidence, queued-without-claimant, exact-scope mismatch,
hash-only/placeholder artifacts, OpenAPI/backend asymmetry, capability
singular/plural asymmetry, terminal success without commit/readback) into
executable read-only invariants absorbed on the authoritative source line
(beabd05 = V2R2C2 authority main). This is a successor line; it does NOT
revive the old p0-chatgpt-bridge-v2 control-plane-convergence task line.

Adds:
- aios_control_plane_consistency: read-only derived SystemTruth (derive_truth)
  + artifact reality gate + OpenAPI/backend parity + capability alias checks.
- aios_control_plane_snapshot.attach_control_plane_truth: injects derived truth
  (state + findings + evidence_flow_confirmed) into the existing explain/status
  surface without mutating any persistence plane; base snapshot contract intact.
- config/gitops/system_truth_contract.json: state-ownership + completion/queue
  requirements + forbidden equivalences.
- tests: 13 original hardening tests + 5 snapshot-truth integration tests
  (18 passed locally).
- ADR + remediation work package docs; fault-injection + audit scripts.

Derivation is read-only; ADMITTED/QUEUED/service-active/drift_status=PASS are
not treated as evidence-complete. Evidence-required transitions and the final
E2E chain (through external projection readback / SCENARIO_ACCEPTED) remain the
release gate and are NOT claimed by this commit.

Signed-off-by: AIOS Governed Continuation <aios-governed-continuation@localhost>

active-runtime/app/aios_control_plane_consistency.py

@@ -0,0 +1,396 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict, field
+from enum import Enum
+from pathlib import Path
+from typing import Any, Iterable, Mapping, Sequence
+import hashlib
+import json
+import re
+import sqlite3
+import subprocess
+import time
+
+PLACEHOLDER_PATTERNS = (
+    re.compile(r"(?im)^\s*(?:todo|fixme|tbd)\b"),
+    re.compile(r"(?im)^\s*pass\s*(?:#.*)?$"),
+    re.compile(r"(?im)\bnot\s+implemented\b"),
+    re.compile(r"(?im)\bplaceholder\b"),
+    re.compile(r"(?im)^\s*\.\.\.\s*$"),
+)
+
+class Severity(str, Enum):
+    BLOCKER = "BLOCKER"
+    MAJOR = "MAJOR"
+    MINOR = "MINOR"
+    INFO = "INFO"
+
+class TruthState(str, Enum):
+    UNKNOWN = "UNKNOWN"
+    ADMITTED_NOT_PROJECTED = "ADMITTED_NOT_PROJECTED"
+    WAITING_CAPACITY = "WAITING_CAPACITY"
+    QUEUED = "QUEUED"
+    CLAIMED = "CLAIMED"
+    RUNNING = "RUNNING"
+    TERMINAL_UNPROVEN = "TERMINAL_UNPROVEN"
+    COMPLETE = "COMPLETE"
+    EXPIRED = "EXPIRED"
+
+@dataclass(frozen=True)
+class Finding:
+    code: str
+    severity: Severity
+    message: str
+    evidence: dict[str, Any] = field(default_factory=dict)
+    remediation: str | None = None
+
+@dataclass
+class AdmissionSnapshot:
+    task_id: str
+    envelope_id: str | None = None
+    admission_status: str | None = None
+    grant_id: str | None = None
+    grant_expires_at: int | None = None
+    one_time_use: bool | None = None
+    authority_base_oid: str | None = None
+    execution_digest: str | None = None
+    requested_capabilities: tuple[str, ...] = ()
+    file_allowlist: tuple[str, ...] = ()
+    admitted_at: int | None = None
+
+@dataclass
+class TaskSnapshot:
+    task_id: str
+    status: str | None = None
+    claimed_by: str | None = None
+    lease_expires_at: int | None = None
+    attempt: int | None = None
+    revision: int | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+    deadline: str | None = None
+    required_capabilities: tuple[str, ...] = ()
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+@dataclass
+class WorkerSnapshot:
+    worker_id: str
+    status: str | None = None
+    last_seen_epoch: int | None = None
+    capabilities: tuple[str, ...] = ()
+    scope_task_id: str | None = None
+    scope_job_id: str | None = None
+    scope_branch: str | None = None
+    scope_execution_digest: str | None = None
+
+@dataclass
+class LeaseSnapshot:
+    task_id: str
+    holder: str | None = None
+    expires_at: int | None = None
+    state: str | None = None
+
+@dataclass
+class GitSnapshot:
+    branch: str
+    current_oid: str | None = None
+    expected_oid: str | None = None
+    commit_oid: str | None = None
+    post_oid: str | None = None
+
+@dataclass
+class EvidenceSnapshot:
+    commit_receipt: dict[str, Any] | None = None
+    runtime_readback: dict[str, Any] | None = None
+    external_projection_readback: dict[str, Any] | None = None
+    output_hashes: dict[str, str] = field(default_factory=dict)
+
+def parse_iso_epoch(value: str | None) -> int | None:
+    if not value:
+        return None
+    from datetime import datetime
+    try:
+        return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp())
+    except Exception:
+        return None
+
+def worker_is_healthy(worker: WorkerSnapshot, now: int, heartbeat_ttl: int = 180) -> bool:
+    return bool(
+        worker.status == "ACTIVE"
+        and worker.last_seen_epoch is not None
+        and now - worker.last_seen_epoch <= heartbeat_ttl
+    )
+
+def worker_scope_matches(worker: WorkerSnapshot, task: TaskSnapshot) -> bool:
+    meta = task.metadata or {}
+    plan = meta.get("execution_plan") or meta.get("plan") or meta
+    job_id = plan.get("external_job_id") if isinstance(plan, dict) else None
+    branch = plan.get("target_branch") if isinstance(plan, dict) else None
+    digest = plan.get("execution_digest") if isinstance(plan, dict) else None
+    if worker.scope_task_id and worker.scope_task_id != task.task_id:
+        return False
+    if worker.scope_job_id and job_id and worker.scope_job_id != job_id:
+        return False
+    if worker.scope_branch and branch and worker.scope_branch != branch:
+        return False
+    if worker.scope_execution_digest and digest and worker.scope_execution_digest != digest:
+        return False
+    return True
+
+def worker_has_capabilities(worker: WorkerSnapshot, required: Sequence[str]) -> bool:
+    have = set(worker.capabilities)
+    return set(required).issubset(have)
+
+def eligible_workers(task: TaskSnapshot, workers: Sequence[WorkerSnapshot], now: int, heartbeat_ttl: int = 180) -> list[WorkerSnapshot]:
+    return [
+        w for w in workers
+        if worker_is_healthy(w, now, heartbeat_ttl)
+        and worker_has_capabilities(w, task.required_capabilities)
+        and worker_scope_matches(w, task)
+    ]
+
+def derive_truth(
+    admission: AdmissionSnapshot | None,
+    task: TaskSnapshot | None,
+    workers: Sequence[WorkerSnapshot],
+    lease: LeaseSnapshot | None,
+    git: GitSnapshot | None,
+    evidence: EvidenceSnapshot | None,
+    now: int | None = None,
+    queue_claim_threshold: int = 600,
+    heartbeat_ttl: int = 180,
+) -> tuple[TruthState, list[Finding]]:
+    now = int(time.time()) if now is None else now
+    findings: list[Finding] = []
+
+    if admission and admission.admission_status == "ADMITTED" and not task:
+        findings.append(Finding(
+            "DUAL_PLANE_ADMITTED_TASK_MISSING",
+            Severity.BLOCKER,
+            "Admission plane says ADMITTED but TaskPlane has no task.",
+            {"task_id": admission.task_id, "envelope_id": admission.envelope_id},
+            "Use durable projection/outbox + exact readback; Admission must not imply execution.",
+        ))
+        return TruthState.ADMITTED_NOT_PROJECTED, findings
+
+    if not task:
+        return TruthState.UNKNOWN, findings
+
+    if admission and admission.grant_expires_at and admission.grant_expires_at <= now and task.status in {"QUEUED", "CLAIMED", "RUNNING"}:
+        findings.append(Finding(
+            "GRANT_EXPIRED_BEFORE_EXECUTION",
+            Severity.BLOCKER,
+            "Execution state depends on an expired grant.",
+            {"expires_at": admission.grant_expires_at, "now": now, "task_state": task.status},
+            "Mint runtime TaskPrincipal just-in-time at claim; do not burn grant TTL while waiting in queue.",
+        ))
+
+    if admission and admission.admission_status == "ADMITTED" and task.status in {"EXPIRED", "FAILED", "CANCELLED"}:
+        findings.append(Finding(
+            "DUAL_PLANE_TERMINAL_DRIFT",
+            Severity.MAJOR,
+            "Admission remains ADMITTED while TaskPlane is terminal.",
+            {"admission": admission.admission_status, "task": task.status},
+            "Expose admission_state and execution_state separately; never collapse both into one PASS/complete status.",
+        ))
+
+    if task.status == "QUEUED":
+        elig = eligible_workers(task, workers, now, heartbeat_ttl)
+        if not elig:
+            findings.append(Finding(
+                "QUEUED_WITHOUT_ELIGIBLE_WORKER",
+                Severity.BLOCKER,
+                "Task is QUEUED but no healthy, capability- and scope-matching worker exists.",
+                {
+                    "required_capabilities": list(task.required_capabilities),
+                    "healthy_workers": [w.worker_id for w in workers if worker_is_healthy(w, now, heartbeat_ttl)],
+                },
+                "Gate TaskPlane projection on a SchedulabilityProof; keep Admission ADMITTED/WAITING_CAPACITY until claimable.",
+            ))
+            return TruthState.WAITING_CAPACITY, findings
+        created = parse_iso_epoch(task.created_at)
+        if created is not None and now - created > queue_claim_threshold and not task.claimed_by:
+            findings.append(Finding(
+                "UNCLAIMED_TIMEOUT",
+                Severity.BLOCKER,
+                "QUEUED task exceeded claim SLA without a claim holder.",
+                {"queued_seconds": now - created, "threshold": queue_claim_threshold},
+                "Trigger scheduler diagnosis/takeover contract; do not extend TTL blindly.",
+            ))
+        return TruthState.QUEUED, findings
+
+    if task.status == "CLAIMED":
+        if not task.claimed_by:
+            findings.append(Finding(
+                "FAKE_CLAIM_STATE",
+                Severity.BLOCKER,
+                "Task says CLAIMED but claimed_by is empty.",
+                {"task_id": task.task_id},
+                "Reject state transition unless claim holder + fencing generation are persisted atomically.",
+            ))
+        if lease is None or lease.holder != task.claimed_by or (lease.expires_at and lease.expires_at <= now):
+            findings.append(Finding(
+                "CLAIM_WITHOUT_VALID_LEASE",
+                Severity.BLOCKER,
+                "CLAIMED task has no valid matching lease.",
+                {"claimed_by": task.claimed_by, "lease": asdict(lease) if lease else None},
+                "Require claim+lease readback before exposing CLAIMED.",
+            ))
+        return TruthState.CLAIMED, findings
+
+    if task.status == "RUNNING":
+        if not task.claimed_by:
+            findings.append(Finding("FAKE_RUNNING_NO_CLAIM", Severity.BLOCKER, "RUNNING without claim holder."))
+        claimed_worker = next((w for w in workers if w.worker_id == task.claimed_by), None)
+        if not claimed_worker or not worker_is_healthy(claimed_worker, now, heartbeat_ttl):
+            findings.append(Finding(
+                "FAKE_RUNNING_NO_HEARTBEAT",
+                Severity.BLOCKER,
+                "RUNNING task has no fresh worker heartbeat.",
+                {"claimed_by": task.claimed_by},
+                "Transition to recovery only through lease/fencing rules; never report RUNNING from a stale projection.",
+            ))
+        if lease is None or (lease.expires_at and lease.expires_at <= now):
+            findings.append(Finding("RUNNING_WITHOUT_VALID_LEASE", Severity.BLOCKER, "RUNNING without valid lease."))
+        return TruthState.RUNNING, findings
+
+    if task.status in {"SUCCEEDED", "COMPLETE"}:
+        ev = evidence or EvidenceSnapshot()
+        commit = ev.commit_receipt or {}
+        before = git.expected_oid if git else None
+        post = (git.post_oid if git else None) or commit.get("post_oid")
+        commit_oid = (git.commit_oid if git else None) or commit.get("commit_oid")
+        if not commit_oid or not post or (before and post == before):
+            findings.append(Finding(
+                "TERMINAL_SUCCESS_UNPROVEN",
+                Severity.BLOCKER,
+                "Task reports success without a proven branch mutation receipt.",
+                {"commit_oid": commit_oid, "post_oid": post, "pre_oid": before},
+                "SUCCEEDED requires lease id, commit receipt, pre/post OID change, output hashes, and runtime/projection readback.",
+            ))
+            return TruthState.TERMINAL_UNPROVEN, findings
+        if not ev.runtime_readback:
+            findings.append(Finding("SUCCESS_WITHOUT_RUNTIME_READBACK", Severity.MAJOR, "Commit exists but runtime readback is absent."))
+        return TruthState.COMPLETE if not findings else TruthState.TERMINAL_UNPROVEN, findings
+
+    if task.status == "EXPIRED":
+        if task.attempt in (None, 0) and not task.claimed_by:
+            findings.append(Finding(
+                "EXPIRED_WITHOUT_ANY_CLAIM",
+                Severity.MAJOR,
+                "Task expired without a single execution attempt.",
+                {"attempt": task.attempt, "claimed_by": task.claimed_by},
+                "Treat as schedulability failure, not execution failure; fix worker/principal supply before recovery.",
+            ))
+        return TruthState.EXPIRED, findings
+
+    return TruthState.UNKNOWN, findings
+
+def inspect_artifact(path: Path, expected_sha256: str | None = None, min_bytes: int = 8) -> list[Finding]:
+    findings: list[Finding] = []
+    if not path.is_file():
+        return [Finding("ARTIFACT_MISSING", Severity.BLOCKER, f"Artifact missing: {path}")]
+    data = path.read_bytes()
+    if len(data) < min_bytes:
+        findings.append(Finding("ARTIFACT_TOO_SMALL", Severity.BLOCKER, f"Artifact is too small: {path}", {"bytes": len(data)}))
+    digest = hashlib.sha256(data).hexdigest()
+    if expected_sha256 and digest != expected_sha256:
+        findings.append(Finding("ARTIFACT_HASH_MISMATCH", Severity.BLOCKER, f"Artifact hash mismatch: {path}", {"expected": expected_sha256, "actual": digest}))
+    try:
+        text = data.decode("utf-8")
+    except UnicodeDecodeError:
+        text = ""
+    if text and any(p.search(text) for p in PLACEHOLDER_PATTERNS):
+        findings.append(Finding(
+            "PLACEHOLDER_ARTIFACT",
+            Severity.BLOCKER,
+            f"Artifact contains placeholder/stub markers: {path}",
+            remediation="Reject plan admission until actual bytes exist and match the declared digest.",
+        ))
+    return findings
+
+def operation_ids_from_openapi(path: Path) -> dict[str, str]:
+    try:
+        import yaml  # type: ignore
+    except ImportError as exc:
+        raise RuntimeError("PyYAML is required for OpenAPI parity checks") from exc
+    doc = yaml.safe_load(path.read_text(encoding="utf-8-sig"))
+    result: dict[str, str] = {}
+    for route, methods in (doc.get("paths") or {}).items():
+        if not isinstance(methods, dict):
+            continue
+        for _, spec in methods.items():
+            if isinstance(spec, dict) and spec.get("operationId"):
+                result[str(spec["operationId"])] = str(route)
+    return result
+
+ROUTE_PAIR_RE = re.compile(r'["\'](?P<route>/v\d+/[^"\']+)["\']\s*:\s*["\'](?P<op>[A-Za-z0-9_]+)["\']')
+
+def operation_ids_from_server_source(path: Path) -> dict[str, str]:
+    text = path.read_text(encoding="utf-8")
+    out: dict[str, str] = {}
+    for m in ROUTE_PAIR_RE.finditer(text):
+        out[m.group("op")] = m.group("route")
+    return out
+
+def check_openapi_backend_parity(openapi_path: Path, server_path: Path, required_ops: Iterable[str] = ()) -> list[Finding]:
+    oa = operation_ids_from_openapi(openapi_path)
+    be = operation_ids_from_server_source(server_path)
+    findings: list[Finding] = []
+    for op in sorted(set(required_ops) | set(oa)):
+        if op in oa and op not in be:
+            findings.append(Finding(
+                "OPENAPI_DEAD_ENDPOINT",
+                Severity.BLOCKER,
+                f"OpenAPI advertises {op} ({oa[op]}) but backend route mapping is absent.",
+                remediation="Deploy projection and backend from the same source commit; reject schema-only rollout.",
+            ))
+    for op in sorted(set(required_ops) | set(be)):
+        if op in be and op not in oa:
+            findings.append(Finding(
+                "BACKEND_UNPROJECTED_OPERATION",
+                Severity.MAJOR,
+                f"Backend implements {op} ({be[op]}) but OpenAPI projection is absent.",
+                remediation="Generate/validate external projection from the canonical backend route manifest.",
+            ))
+    return findings
+
+def check_capability_aliases(required: Sequence[str], workers: Sequence[WorkerSnapshot]) -> list[Finding]:
+    findings: list[Finding] = []
+    req = set(required)
+    if "governed.write.files" in req:
+        for w in workers:
+            caps = set(w.capabilities)
+            if "governed.write.file" in caps and "governed.write.files" not in caps:
+                findings.append(Finding(
+                    "CAPABILITY_SINGULAR_PLURAL_ASYMMETRY",
+                    Severity.MAJOR,
+                    f"Worker {w.worker_id} exposes governed.write.file but task requires governed.write.files.",
+                    {"worker_capabilities": sorted(caps)},
+                    "Do not treat names as aliases implicitly; define one canonical capability or an explicit server-side compatibility mapping with tests.",
+                ))
+    return findings
+
+def git_ref(git_dir: Path, ref: str) -> str | None:
+    cp = subprocess.run(
+        ["git", f"--git-dir={git_dir}", "rev-parse", ref],
+        text=True, capture_output=True
+    )
+    if cp.returncode != 0:
+        return None
+    value = cp.stdout.strip()
+    return value if re.fullmatch(r"[0-9a-f]{40}", value) else None
+
+def sqlite_rows(path: Path, query: str, params: Sequence[Any] = ()) -> list[sqlite3.Row]:
+    uri = f"file:{path}?mode=ro&immutable=1"
+    con = sqlite3.connect(uri, uri=True)
+    con.row_factory = sqlite3.Row
+    try:
+        return list(con.execute(query, params))
+    finally:
+        con.close()
+
+def dump_report(path: Path, report: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(report, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8")

active-runtime/app/aios_control_plane_snapshot.py

@@ -1,7 +1,27 @@
-"""Non-sensitive control-plane snapshot assembly."""
+"""Non-sensitive control-plane snapshot assembly.
+
+The base snapshot is a pure field-copy of each plane's reported state. It does
+NOT itself assert whether that state is backed by evidence — a drift_status=PASS
+or an ADMITTED status is a *claim*, not a proof alert/status change.
+
+`attach_control_plane_truth` (see below) injects the read-only derived `SystemTruth`
+(states + findings) from `aios_control_plane_consistency.derive_truth` into the
+snapshot so that explain/status surfaces can expose *derived* truth alongside
+the reported state, without ever mutating the persistence planes.
+"""
 from __future__ import annotations
 
-from typing import Any, Mapping
+from typing import Any, Mapping, Sequence
+
+from aios_control_plane_consistency import (
+    AdmissionSnapshot,
+    EvidenceSnapshot,
+    GitSnapshot,
+    LeaseSnapshot,
+    TaskSnapshot,
+    WorkerSnapshot,
+    derive_truth,
+)
 
 
 def _copy(value: Mapping[str, Any] | None, keys: tuple[str, ...]) -> dict[str, Any]:
@@ -28,3 +48,48 @@ def build_control_plane_snapshot(*, target_version: str, runtime: Mapping[str, A
 
 def getControlPlaneSnapshot(**kwargs: Any) -> dict[str, Any]:
     return build_control_plane_snapshot(**kwargs)
+
+
+def attach_control_plane_truth(
+    snapshot: Mapping[str, Any],
+    *,
+    admission: AdmissionSnapshot | None = None,
+    task: TaskSnapshot | None = None,
+    workers: Sequence[WorkerSnapshot] = (),
+    lease: LeaseSnapshot | None = None,
+    git: GitSnapshot | None = None,
+    evidence: EvidenceSnapshot | None = None,
+    now: int | None = None,
+    queue_claim_threshold: int = 600,
+    heartbeat_ttl: int = 180,
+) -> dict[str, Any]:
+    """Attach read-only derived SystemTruth to a control-plane snapshot.
+
+    This never writes to any persistence plane. It derives a ``system_truth``
+    block from the already-loaded plane snapshots and adds it under
+    ``snapshot["derived"]``, leaving all existing keys intact.
+
+    The block includes the derived truth state, any findings, and a compact
+    "equivalence" guard (see config/gitops/system_truth_contract.json
+    ``forbidden_equivalences``) so surfaces cannot silently present a local
+    plane claim as end-to-end evidence.
+    """
+    from dataclasses import asdict
+    truth_state, findings = derive_truth(
+        admission, task, workers, lease, git, evidence, now,
+        queue_claim_threshold=queue_claim_threshold,
+        heartbeat_ttl=heartbeat_ttl,
+    )
+    out = dict(snapshot)
+    if "derived" not in out:
+        out["derived"] = {}
+    blockers = [f for f in findings if f.severity.value == "BLOCKER"]
+    out["derived"]["system_truth"] = {
+        "truth_state": truth_state.value,
+        "findings": [asdict(f) for f in findings],
+        "blocker_count": sum(1 for f in findings if f.severity.value == "BLOCKER"),
+        "major_count": sum(1 for f in findings if f.severity.value == "MAJOR"),
+        # Derivation is read-only; it never confirms completion on its own.
+        "evidence_flow_confirmed": truth_state.value == "COMPLETE" and not blockers,
+    }
+    return out

config/gitops/system_truth_contract.json

@@ -0,0 +1,76 @@
+{
+  "schema": "aios.system-truth-contract.v1",
+  "state_ownership": {
+    "TaskAdmission": {
+      "owns": [
+        "envelope",
+        "authorization",
+        "grant",
+        "admission_receipt"
+      ],
+      "must_not_own": [
+        "claim",
+        "heartbeat",
+        "execution_status",
+        "commit_result"
+      ]
+    },
+    "TaskPlane": {
+      "owns": [
+        "task_state",
+        "claim",
+        "lease",
+        "heartbeat",
+        "attempt",
+        "deadline"
+      ],
+      "must_not_own": [
+        "owner_signature",
+        "git_commit_truth"
+      ]
+    },
+    "GitAuthority": {
+      "owns": [
+        "branch_oid",
+        "commit_oid",
+        "file_bytes"
+      ],
+      "must_not_own": [
+        "task_state"
+      ]
+    },
+    "RuntimeProjection": {
+      "owns": [
+        "deployed_source_oid",
+        "served_routes",
+        "external_action_readback"
+      ],
+      "must_not_own": [
+        "source_commit_status"
+      ]
+    }
+  },
+  "completion_requires": [
+    "valid_claim_holder",
+    "fresh_worker_heartbeat",
+    "valid_writer_lease",
+    "commit_receipt",
+    "post_oid_differs_from_pre_oid",
+    "output_sha256_matches_plan",
+    "runtime_readback",
+    "external_projection_readback"
+  ],
+  "queue_requires": [
+    "fresh_schedulability_proof",
+    "at_least_one_healthy_scope_matching_worker",
+    "grant_or_jit_principal_can_cover_claim_window"
+  ],
+  "forbidden_equivalences": [
+    "ADMITTED == EXECUTED",
+    "QUEUED == CLAIMED",
+    "service_active == scenario_pass",
+    "drift_status_PASS == evidence_complete",
+    "hash_declared == artifact_exists",
+    "OpenAPI_operation == backend_route_exists"
+  ]
+}

docs/architecture/adr-control-plane-system-truth.md

@@ -0,0 +1,64 @@
+# ADR: AIOS Control-Plane System Truth and Execution Integrity
+
+## Status
+Proposed implementation baseline for ANNEX-2.3 hardening.
+
+## Problem
+Historical failures show that multiple planes can each report locally valid state while the end-to-end goal is false: Admission can be ADMITTED while TaskPlane has no claim; a task can be QUEUED until its grant expires; a worker can have matching capability names but an exact scope for another job; OpenAPI can advertise a route absent from the deployed backend; and PASS labels can be emitted without production evidence.
+
+## Decision
+Do not create a third persistence plane. Keep one authority per state dimension and derive `SystemTruth` read-only.
+
+| State dimension | Authority |
+|---|---|
+| envelope / grant / admission receipt | Task Admission |
+| task lifecycle / claim / attempt / heartbeat / deadline | TaskPlane |
+| writer lease | governed writer / lease authority |
+| source bytes / branch and commit OIDs | Git authority |
+| deployed source / served routes | runtime |
+| external ChatGPT Action operations | external projection readback |
+
+### 1. Admission is not execution
+`ADMITTED` means authorization has been accepted. It must never be presented as running, claimed, committed, or complete.
+
+### 2. Schedulability before QUEUED
+Before TaskPlane projection to `QUEUED`, produce a fresh `SchedulabilityProof` containing:
+- required capabilities
+- exact task/job/branch/execution-digest scope
+- at least one healthy eligible worker
+- current grant/principal viability
+- dependencies/freeze state
+
+If no eligible worker exists, keep execution projection in `WAITING_CAPACITY` (Admission may remain ADMITTED). Do not burn a short one-time grant while the task is waiting for capacity.
+
+### 3. Mint runtime authority just-in-time
+Separate long-lived Owner authorization from short-lived runtime TaskPrincipal. Mint the execution principal at claim time or immediately before claim, not at admission time. This prevents 15-minute grants expiring in a queue with no eligible worker.
+
+### 4. Exact-scope is mandatory
+Capabilities alone never establish eligibility. A worker is eligible only when `task_id`, `external_job_id`, `target_branch`, `generation`, and `execution_digest` are within its signed scope.
+
+### 5. No fake execution
+Transitions require evidence:
+- CLAIMED: claim holder + fencing generation + lease readback
+- RUNNING: CLAIMED + fresh heartbeat + valid lease
+- SUCCEEDED: commit receipt + pre/post OID change + planned file hashes + runtime readback
+- SCENARIO_ACCEPTED: SUCCEEDED + external projection readback
+
+`service active`, `drift_status=PASS`, focused tests, or an Admission receipt are not substitutes.
+
+### 6. Real bytes, not placeholders
+Execution plans must reference content-addressed bytes that exist before admission. Hash-only plans with missing bytes, zero-length files, TODO/pass placeholders, or mismatched hashes are rejected.
+
+### 7. One source for route contracts
+Backend route registry and external OpenAPI must derive from one canonical route manifest, or be parity-checked in CI. Schema-only deployment is blocked if the matching backend route is absent; backend-only deployment is blocked if the external projection is absent when required.
+
+### 8. Dual-plane reconciliation
+Projection is idempotent and read-back verified. Admission writes an append-only projection intent/outbox in the same transaction as the grant; a projector writes TaskPlane, reads back exact task identity/revision, and records a receipt. Divergence is a first-class failure, never silently masked.
+
+## Required release gates
+- no BLOCKER from control-plane consistency audit
+- C4 regression: legacy task creation path remains TaskPrincipal-gated
+- admission -> CLAIMED -> RUNNING -> WriterLease -> governed commit -> readback integration test
+- real activation + real SSH signature host test
+- external projection readback test
+- no placeholder artifact findings

docs/architecture/remediation-control-plane-workpackages.md

@@ -0,0 +1,33 @@
+# Remediation Work Packages
+
+## P0-A — Single System Truth (no new DB)
+Implement the read-only SystemTruth derivation in the runtime explain/status surface. Remove or rename PASS labels that are configuration-only.
+
+## P0-B — Schedulability Gate
+Admission may authorize a task, but TaskPlane projection to QUEUED requires a fresh SchedulabilityProof. If no eligible worker exists, expose WAITING_CAPACITY and do not start the short runtime grant TTL.
+
+## P0-C — Just-in-time TaskPrincipal
+Keep Owner authorization durable; mint short-lived TaskPrincipal only when an eligible worker is ready to claim. Bind task/job/branch/generation/execution_digest exactly.
+
+## P0-D — Evidence-required state transitions
+Guard CLAIMED/RUNNING/SUCCEEDED with claim holder, heartbeat, lease, commit receipt, OID change, file hashes, runtime readback, external projection readback.
+
+## P0-E — Artifact reality gate
+Before admission, verify every planned file byte exists, is non-placeholder, and matches SHA256. Reject hash-only or placeholder plans.
+
+## P0-F — Route/projection single source
+Generate or verify ChatGPT Actions OpenAPI from the backend route manifest. Block dead endpoints and unprojected required operations.
+
+## P0-G — Dual-plane durable projection
+Use append-only projection intent + idempotent TaskPlane write + exact readback receipt. Admission status and execution status remain separate.
+
+## P0-H — Worker scope/schedulability diagnostics
+Eligibility = capability AND heartbeat AND exact signed scope. Never select a worker only because capability names match.
+
+## P0-I — QA gaps
+Add:
+1. C4 legacy createGovernedTask/projectOnlineAgentExecution remains TaskPrincipal-gated.
+2. `_verified_task_principal` body injection negative test.
+3. bearer-only no-delegation negative test.
+4. real CLAIMED/RUNNING/WriterLease/commit readback path.
+5. real activation/SSH signature host test.

scripts/control_plane/aios_control_plane_audit.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+from pathlib import Path
+import argparse, json, sys, time, re
+from aios_control_plane_consistency import *
+
+def jload(value):
+    try:
+        return json.loads(value or "{}")
+    except Exception:
+        return {}
+
+def rows(path, query, params=()):
+    try:
+        return sqlite_rows(path, query, params)
+    except Exception as exc:
+        return [{"_error": str(exc)}]
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="AIOS read-only control-plane consistency auditor")
+    ap.add_argument("--task-id", required=True)
+    ap.add_argument("--admission-db", type=Path, default=Path("/var/lib/aios-gitops/bridge-v2-staging/task-admission.sqlite3"))
+    ap.add_argument("--task-db", type=Path, default=Path("/var/lib/aios-agent-task/state.sqlite3"))
+    ap.add_argument("--git-dir", type=Path, default=Path("/data/aios-gitops/repo.git"))
+    ap.add_argument("--branch")
+    ap.add_argument("--openapi", type=Path)
+    ap.add_argument("--server-source", type=Path)
+    ap.add_argument("--artifact-root", type=Path)
+    ap.add_argument("--now", type=int)
+    ap.add_argument("--out", type=Path, default=Path("aios-control-plane-audit.json"))
+    ns = ap.parse_args()
+    now = ns.now or int(time.time())
+    findings = []
+    admission = None
+    task = None
+    workers = []
+    lease = None
+    git = None
+    evidence = EvidenceSnapshot()
+
+    # Admission plane.
+    try:
+        receipt = sqlite_rows(ns.admission_db,
+            "SELECT * FROM task_admission_receipts WHERE task_id=? ORDER BY rowid DESC LIMIT 1", (ns.task_id,))
+        if receipt:
+            r = dict(receipt[0])
+            envelope_id = r.get("envelope_id")
+            env = sqlite_rows(ns.admission_db,
+                "SELECT * FROM task_admission_envelopes WHERE envelope_id=? LIMIT 1", (envelope_id,))
+            grant = sqlite_rows(ns.admission_db,
+                "SELECT * FROM task_admission_grants WHERE envelope_id=? ORDER BY rowid DESC LIMIT 1", (envelope_id,))
+            ed = dict(env[0]) if env else {}
+            gd = dict(grant[0]) if grant else {}
+            gj = jload(gd.get("grant_json"))
+            ej = jload(ed.get("envelope_json"))
+            admission = AdmissionSnapshot(
+                task_id=ns.task_id,
+                envelope_id=envelope_id,
+                admission_status=ed.get("status"),
+                grant_id=gd.get("grant_id"),
+                grant_expires_at=gj.get("expires_at"),
+                one_time_use=gj.get("one_time_use"),
+                authority_base_oid=r.get("authority_base_oid") or ej.get("authority_base_oid"),
+                execution_digest=r.get("execution_digest") or ej.get("execution_digest"),
+                requested_capabilities=tuple(ej.get("requested_capabilities") or ()),
+                file_allowlist=tuple(ej.get("file_allowlist") or ()),
+                admitted_at=r.get("created_at") or r.get("admitted_at"),
+            )
+    except Exception as exc:
+        findings.append(Finding("ADMISSION_PLANE_UNREADABLE", Severity.BLOCKER, str(exc)))
+
+    # TaskPlane.
+    try:
+        trows = sqlite_rows(ns.task_db, "SELECT * FROM tasks WHERE task_id=? LIMIT 1", (ns.task_id,))
+        if trows:
+            td = dict(trows[0])
+            meta = jload(td.get("metadata_json"))
+            required = jload(td.get("required_capabilities_json"))
+            if isinstance(required, dict):
+                required = required.get("capabilities") or []
+            task = TaskSnapshot(
+                task_id=ns.task_id,
+                status=td.get("status"),
+                claimed_by=td.get("claimed_by"),
+                lease_expires_at=td.get("lease_expires_at"),
+                attempt=td.get("attempt"),
+                revision=td.get("revision"),
+                created_at=td.get("created_at"),
+                updated_at=td.get("updated_at"),
+                deadline=td.get("deadline"),
+                required_capabilities=tuple(required or ()),
+                metadata=meta if isinstance(meta, dict) else {},
+            )
+        arows = sqlite_rows(ns.task_db, "SELECT * FROM agents")
+        for row in arows:
+            d = dict(row)
+            caps = jload(d.get("capabilities_json"))
+            if isinstance(caps, dict):
+                caps = caps.get("capabilities") or []
+            last = parse_iso_epoch(d.get("last_seen_at"))
+            workers.append(WorkerSnapshot(
+                worker_id=d.get("agent_id") or d.get("worker_id") or "unknown",
+                status=d.get("status"),
+                last_seen_epoch=last,
+                capabilities=tuple(caps or ()),
+            ))
+    except Exception as exc:
+        findings.append(Finding("TASK_PLANE_UNREADABLE", Severity.BLOCKER, str(exc)))
+
+    branch = ns.branch
+    if not branch and task and isinstance(task.metadata, dict):
+        plan = task.metadata.get("execution_plan") or task.metadata.get("plan") or task.metadata
+        if isinstance(plan, dict):
+            branch = plan.get("target_branch")
+    if branch:
+        current = git_ref(ns.git_dir, f"refs/heads/{branch}")
+        git = GitSnapshot(branch=branch, current_oid=current, expected_oid=admission.authority_base_oid if admission else None)
+        if admission and current and admission.authority_base_oid and current != admission.authority_base_oid:
+            findings.append(Finding(
+                "BRANCH_BINDING_DRIFT",
+                Severity.BLOCKER,
+                "Current branch OID differs from Admission authority_base_oid.",
+                {"current": current, "bound": admission.authority_base_oid},
+                "Replan/rebind formally; do not reuse old execution_digest.",
+            ))
+
+    truth, derived = derive_truth(admission, task, workers, lease, git, evidence, now)
+    findings.extend(derived)
+    if task:
+        findings.extend(check_capability_aliases(task.required_capabilities, workers))
+
+    # Plan/artifact integrity if the immutable plan points to files.
+    if task and ns.artifact_root:
+        plan = task.metadata.get("execution_plan") or task.metadata.get("plan") or task.metadata
+        if isinstance(plan, dict):
+            hashes = plan.get("file_sha256") or {}
+            for rel in plan.get("file_allowlist") or []:
+                findings.extend(inspect_artifact(ns.artifact_root / rel, hashes.get(rel)))
+
+    if ns.openapi and ns.server_source:
+        findings.extend(check_openapi_backend_parity(
+            ns.openapi, ns.server_source,
+            required_ops=("getTaskAdmissionEnvelope","submitTaskAdmissionAuthorization","getTaskAdmissionStatus"),
+        ))
+
+    report = {
+        "schema": "aios.control-plane-consistency-audit.v1",
+        "observed_at": now,
+        "task_id": ns.task_id,
+        "truth_state": truth.value,
+        "admission": asdict(admission) if admission else None,
+        "task": asdict(task) if task else None,
+        "eligible_workers": [asdict(w) for w in eligible_workers(task, workers, now)] if task else [],
+        "git": asdict(git) if git else None,
+        "findings": [asdict(f) for f in findings],
+        "summary": {
+            "blockers": sum(f.severity == Severity.BLOCKER for f in findings),
+            "majors": sum(f.severity == Severity.MAJOR for f in findings),
+            "minors": sum(f.severity == Severity.MINOR for f in findings),
+        },
+    }
+    dump_report(ns.out, report)
+    print(json.dumps(report["summary"], sort_keys=True))
+    return 2 if report["summary"]["blockers"] else 1 if report["summary"]["majors"] else 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/control_plane/controlled_fault_injection.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+from aios_control_plane_consistency import *
+NOW=2_000_000_000
+
+scenarios = {}
+
+a=AdmissionSnapshot(task_id="T",admission_status="ADMITTED",grant_expires_at=NOW+60)
+scenarios["dual_plane_missing_task"]=derive_truth(a,None,[],None,None,None,NOW)
+
+t=TaskSnapshot(task_id="T",status="QUEUED",required_capabilities=("writer.lease",),created_at="2033-05-18T03:23:20+00:00")
+scenarios["queued_no_worker"]=derive_truth(a,t,[],None,None,None,NOW)
+
+t2=TaskSnapshot(task_id="T",status="RUNNING",claimed_by="ghost")
+scenarios["fake_running"]=derive_truth(a,t2,[],None,None,None,NOW)
+
+t3=TaskSnapshot(task_id="T",status="SUCCEEDED")
+scenarios["fake_success"]=derive_truth(a,t3,[],None,GitSnapshot("agent/x",expected_oid="a"*40),EvidenceSnapshot(),NOW)
+
+for name,(state,findings) in scenarios.items():
+    print(name,state.value,[f.code for f in findings])

tests/task_plane/test_control_plane_consistency.py

@@ -0,0 +1,76 @@
+from pathlib import Path
+import time
+from aios_control_plane_consistency import *
+
+NOW=2_000_000_000
+
+def task(status="QUEUED", caps=("governed.write.files","task.principal.issue","writer.lease"), **kw):
+    return TaskSnapshot(
+        task_id="T", status=status, required_capabilities=tuple(caps),
+        created_at="2033-05-18T03:23:20+00:00", metadata={
+            "external_job_id":"job-a",
+            "target_branch":"agent/x",
+            "execution_digest":"d"*64,
+        }, **kw
+    )
+
+def admission(**kw):
+    base=dict(task_id="T", admission_status="ADMITTED", envelope_id="E", grant_expires_at=NOW+600)
+    base.update(kw)
+    return AdmissionSnapshot(**base)
+
+def worker(worker_id="w", caps=("governed.write.files","task.principal.issue","writer.lease"), last_seen=NOW-10, **kw):
+    return WorkerSnapshot(worker_id=worker_id,status="ACTIVE",last_seen_epoch=last_seen,capabilities=tuple(caps),**kw)
+
+def test_admitted_without_taskplane_is_not_execution():
+    state, fs=derive_truth(admission(),None,[],None,None,None,NOW)
+    assert state==TruthState.ADMITTED_NOT_PROJECTED
+    assert any(f.code=="DUAL_PLANE_ADMITTED_TASK_MISSING" for f in fs)
+
+def test_queued_without_eligible_worker_is_waiting_capacity_not_real_execution():
+    state, fs=derive_truth(admission(),task(),[],None,None,None,NOW)
+    assert state==TruthState.WAITING_CAPACITY
+    assert any(f.code=="QUEUED_WITHOUT_ELIGIBLE_WORKER" for f in fs)
+
+def test_scope_mismatched_worker_is_not_eligible():
+    w=worker(scope_task_id="OTHER")
+    assert eligible_workers(task(),[w],NOW)==[]
+
+def test_stale_worker_is_not_eligible():
+    w=worker(last_seen=NOW-1000)
+    assert eligible_workers(task(),[w],NOW)==[]
+
+def test_capability_singular_plural_is_not_silent_alias():
+    fs=check_capability_aliases(("governed.write.files",),[
+        worker(caps=("governed.write.file","task.principal.issue","writer.lease"))
+    ])
+    assert any(f.code=="CAPABILITY_SINGULAR_PLURAL_ASYMMETRY" for f in fs)
+
+def test_claimed_without_claim_holder_is_fake():
+    state, fs=derive_truth(admission(),task(status="CLAIMED"),[],None,None,None,NOW)
+    assert state==TruthState.CLAIMED
+    assert any(f.code=="FAKE_CLAIM_STATE" for f in fs)
+
+def test_running_without_heartbeat_is_fake_execution():
+    t=task(status="RUNNING",claimed_by="w")
+    state, fs=derive_truth(admission(),t,[worker(last_seen=NOW-999)],LeaseSnapshot("T","w",NOW+60,"ACTIVE"),None,None,NOW)
+    assert any(f.code=="FAKE_RUNNING_NO_HEARTBEAT" for f in fs)
+
+def test_succeeded_without_commit_receipt_is_unproven():
+    state, fs=derive_truth(admission(),task(status="SUCCEEDED"),[],None,GitSnapshot("agent/x",expected_oid="a"*40),EvidenceSnapshot(),NOW)
+    assert state==TruthState.TERMINAL_UNPROVEN
+    assert any(f.code=="TERMINAL_SUCCESS_UNPROVEN" for f in fs)
+
+def test_expired_without_claim_is_schedulability_failure():
+    state, fs=derive_truth(admission(grant_expires_at=NOW-1),task(status="EXPIRED",attempt=0),[],None,None,None,NOW)
+    assert state==TruthState.EXPIRED
+    assert any(f.code=="EXPIRED_WITHOUT_ANY_CLAIM" for f in fs)
+
+def test_placeholder_artifact_is_rejected(tmp_path):
+    p=tmp_path/"x.py"; p.write_text("TODO: real content\n",encoding="utf-8")
+    fs=inspect_artifact(p)
+    assert any(f.code=="PLACEHOLDER_ARTIFACT" for f in fs)
+
+def test_hash_only_does_not_make_missing_artifact_real(tmp_path):
+    fs=inspect_artifact(tmp_path/"missing.py","0"*64)
+    assert any(f.code=="ARTIFACT_MISSING" for f in fs)

tests/task_plane/test_control_plane_snapshot_truth.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from aios_control_plane_consistency import (
+    AdmissionSnapshot,
+    EvidenceSnapshot,
+    GitSnapshot,
+    LeaseSnapshot,
+    TaskSnapshot,
+    WorkerSnapshot,
+)
+from aios_control_plane_snapshot import (
+    attach_control_plane_truth,
+    build_control_plane_snapshot,
+)
+
+NOW = 2_000_000_000
+
+
+def _base():
+    return build_control_plane_snapshot(
+        target_version="V2R2C2",
+        runtime={"drift_status": "PASS"},
+        task={"task_id": "T", "reconciliation_state": "ok"},
+    )
+
+
+def test_snapshot_base_has_no_derived_block_by_default():
+    assert "derived" not in _base()
+
+
+def test_truth_attach_preserves_existing_keys():
+    out = attach_control_plane_truth(_base(), now=NOW)
+    assert set(_base().keys()) <= set(out.keys())
+    assert "derived" in out
+
+
+def test_admitted_without_taskplane_is_admitted_not_projected():
+    out = attach_control_plane_truth(
+        _base(),
+        admission=AdmissionSnapshot(task_id="T", admission_status="ADMITTED", envelope_id="E"),
+        now=NOW,
+    )
+    truth = out["derived"]["system_truth"]
+    assert truth["truth_state"] == "ADMITTED_NOT_PROJECTED"
+    assert truth["blocker_count"] == 1
+    assert any(f["code"] == "DUAL_PLANE_ADMITTED_TASK_MISSING" for f in truth["findings"])
+    # A drift_status=PASS surface must NOT be reported as evidence-complete.
+    assert truth["evidence_flow_confirmed"] is False
+
+
+def test_fake_success_is_terminal_unproven_even_under_drift_pass():
+    out = attach_control_plane_truth(
+        _base(),
+        admission=AdmissionSnapshot(task_id="T"),
+        task=TaskSnapshot(task_id="T", status="SUCCEEDED"),
+        git=GitSnapshot("agent/x", expected_oid="a" * 40),
+        evidence=EvidenceSnapshot(),
+        now=NOW,
+    )
+    truth = out["derived"]["system_truth"]
+    assert truth["truth_state"] == "TERMINAL_UNPROVEN"
+    assert truth["blocker_count"] == 1
+    assert truth["evidence_flow_confirmed"] is False
+
+
+def test_complete_requires_full_evidence_chain():
+    worker = WorkerSnapshot(
+        worker_id="w", status="ACTIVE", last_seen_epoch=NOW - 10,
+        capabilities=("governed.write.files",),
+    )
+    task = TaskSnapshot(
+        task_id="T", status="SUCCEEDED", claimed_by="w",
+        required_capabilities=("governed.write.files",),
+        metadata={"external_job_id": "job-a", "target_branch": "agent/x",
+                  "execution_digest": "d" * 64},
+    )
+    git = GitSnapshot("agent/x", expected_oid="a" * 40, commit_oid="b" * 40, post_oid="c" * 40)
+    evidence = EvidenceSnapshot(runtime_readback={"ok": True})
+    out = attach_control_plane_truth(
+        _base(),
+        admission=AdmissionSnapshot(task_id="T"),
+        task=task, workers=[worker],
+        lease=LeaseSnapshot("T", "w", NOW + 60, "ACTIVE"),
+        git=git, evidence=evidence, now=NOW,
+    )
+    truth = out["derived"]["system_truth"]
+    assert truth["truth_state"] == "COMPLETE"
+    assert truth["blocker_count"] == 0
+    assert truth["evidence_flow_confirmed"] is True

tests/task_plane/test_projection_parity.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+from aios_control_plane_consistency import check_openapi_backend_parity
+
+def test_openapi_dead_endpoint_is_blocker(tmp_path):
+    oa=tmp_path/"o.yaml"
+    oa.write_text("""openapi: 3.1.0
+info: {title: t, version: '1'}
+paths:
+  /v2/task-admission/envelope:
+    post:
+      operationId: getTaskAdmissionEnvelope
+""",encoding="utf-8")
+    srv=tmp_path/"s.py"
+    srv.write_text('ROUTES = {"/v2/other": "other"}\n',encoding="utf-8")
+    fs=check_openapi_backend_parity(oa,srv,("getTaskAdmissionEnvelope",))
+    assert any(f.code=="OPENAPI_DEAD_ENDPOINT" for f in fs)
+
+def test_backend_unprojected_operation_is_reported(tmp_path):
+    oa=tmp_path/"o.yaml"
+    oa.write_text("""openapi: 3.1.0
+info: {title: t, version: '1'}
+paths: {}
+""",encoding="utf-8")
+    srv=tmp_path/"s.py"
+    srv.write_text('ROUTES = {"/v2/task-admission/status": "getTaskAdmissionStatus"}\n',encoding="utf-8")
+    fs=check_openapi_backend_parity(oa,srv,("getTaskAdmissionStatus",))
+    assert any(f.code=="BACKEND_UNPROJECTED_OPERATION" for f in fs)