feat(approval-gate): wire gateDestructive into live destructive handlers (Phase 4b)

Activates the merged 4a approval-gate infra. Scope is centralized in gateDestructive so each handler is a one-liner, and the policy is uniform + testable: - default-OFF rollout flag MOLECULE_PLATFORM_APPROVAL_GATE (mirrors 3a/3c default-off; protects existing org-token automation until the platform agent ships); - ONLY org-token callers are gated. The platform agent runs with MOLECULE_API_KEY=<org-admin token> (auth middleware sets org_token_id); ordinary workspace-token agents and human CP-session operators are NOT gated, so normal operation is byte-identical. Realises the file-header trust boundary without gating everyone. Wired: WorkspaceHandler.Delete (delete_workspace) + SecretsHandler.Set (secret_write). requireApproval/gateDestructive now take the events.EventEmitter interface and tolerate a nil emitter (SecretsHandler is broadcaster-less) — the pending approval row is still persisted, only the live canvas push is skipped. 3 new unit tests (flag default-off, org-token detection, scope short-circuits) + existing approval/secret/delete suites green; golangci-lint clean. Deferred (documented follow-on): org_token_mint (OrgTokenHandler is org-scoped, no workspace_id to key the approval) + deprovision_workspace (no workspace-server handler — lives CP-side). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Merge PR #1964 via Gitea merge queue
2026-06-06 16:18:09 -07:00 · 2026-06-06 22:58:23 +00:00 · 2026-06-06 22:52:28 +00:00 · 2026-06-06 22:39:17 +00:00 · 2026-06-06 22:33:24 +00:00 · 2026-06-06 22:18:51 +00:00
73 changed files with 4049 additions and 299 deletions
@@ -54,32 +54,57 @@ API="https://${GITEA_HOST}/api/v1"
 AUTH="Authorization: token ${GITEA_TOKEN}"

 # 1. Fetch the PR. If not merged, no-op.
-PR=$(curl -sS -H "$AUTH" "${API}/repos/${OWNER}/${NAME}/pulls/${PR_NUMBER}")
-MERGED=$(echo "$PR" | jq -r '.merged // false')
+# Fail-closed: verify HTTP 200 before parsing. A 401/403/404 means the token
+# is invalid or the PR is inaccessible — we must NOT silently treat that as
+# "not merged" and skip the audit.
+PR_TMP=$(mktemp)
+PR_HTTP=$(curl -sS -o "$PR_TMP" -w '%{http_code}' -H "$AUTH" \
+  "${API}/repos/${OWNER}/${NAME}/pulls/${PR_NUMBER}")
+PR=$(cat "$PR_TMP")
+rm -f "$PR_TMP"
+if [ "$PR_HTTP" != "200" ]; then
+  echo "::error::GET /pulls/${PR_NUMBER} returned HTTP ${PR_HTTP} — cannot evaluate merge state."
+  exit 1
+fi
+# FAIL-CLOSED: a 200 response with a missing/malformed `merged` field must
+# NOT be treated as "not merged" (that would silently skip the audit).
+# We verify both presence AND correct type for every field we consume.
+PR_SCHEMA_OK=$(echo "$PR" | jq -r '
+  (.merged | type == "boolean") and
+  (.merge_commit_sha | type == "string") and
+  (.merged_by | type == "object") and (.merged_by.login | type == "string") and
+  (.base | type == "object") and (.base.ref | type == "string") and
+  (.head | type == "object") and (.head.sha | type == "string")
+')
+if [ "$PR_SCHEMA_OK" != "true" ]; then
+  echo "::error::GET /pulls/${PR_NUMBER} returned HTTP 200 but one or more required fields are missing, null, or of wrong type — cannot evaluate force-merge."
+  exit 1
+fi
+MERGED=$(echo "$PR" | jq -r '.merged')
 if [ "$MERGED" != "true" ]; then
  echo "::notice::PR #${PR_NUMBER} closed without merge — no audit emission."
  exit 0
 fi

-# NOTE: no || true — with set -euo pipefail, jq parse failures (e.g. field
-# missing from API response) propagate as hard errors. Use jq's // operator
-# for graceful defaults instead of bash || true guards. This was re-added by
-# 8c343e3a ("fix(gitea): add || true guards to jq pipelines") — reverted
-# here because the guards mask silent failures that hide malformed API responses.
-MERGE_SHA=$(echo "$PR" | jq -r '.merge_commit_sha // empty')
-MERGED_BY=$(echo "$PR" | jq -r '.merged_by.login // "unknown"')
+MERGE_SHA=$(echo "$PR" | jq -r '.merge_commit_sha')
+MERGED_BY=$(echo "$PR" | jq -r '.merged_by.login')
 TITLE=$(echo "$PR" | jq -r '.title // ""')
-BASE_BRANCH=$(echo "$PR" | jq -r '.base.ref // "main"')
-HEAD_SHA=$(echo "$PR" | jq -r '.head.sha // empty')
-
-if [ -z "$MERGE_SHA" ]; then
-  echo "::warning::PR #${PR_NUMBER} merged=true but no merge_commit_sha — cannot evaluate force-merge."
-  exit 0
-fi
+BASE_BRANCH=$(echo "$PR" | jq -r '.base.ref')
+HEAD_SHA=$(echo "$PR" | jq -r '.head.sha')

 # 2. Required status checks — branch-aware JSON dict takes precedence.
 if [ -n "${REQUIRED_CHECKS_JSON:-}" ]; then
-  REQUIRED=$(echo "$REQUIRED_CHECKS_JSON" | jq -r --arg branch "$BASE_BRANCH" '.[$branch] // [] | .[]')
+  # FAIL-CLOSED: if REQUIRED_CHECKS_JSON is set, the branch entry must exist
+  # and be an array. A missing branch or non-array value means the config is
+  # malformed or drifted — we must NOT silently treat it as "no checks".
+  _RC_JSON_OK=$(echo "$REQUIRED_CHECKS_JSON" | jq -r --arg branch "$BASE_BRANCH" '
+    has($branch) and (.[$branch] | type == "array")
+  ')
+  if [ "$_RC_JSON_OK" != "true" ]; then
+    echo "::error::REQUIRED_CHECKS_JSON missing or non-array entry for branch '$BASE_BRANCH' — cannot evaluate required checks."
+    exit 1
+  fi
+  REQUIRED=$(echo "$REQUIRED_CHECKS_JSON" | jq -r --arg branch "$BASE_BRANCH" '.[$branch] | .[]')
 else
  REQUIRED="$REQUIRED_CHECKS"
 fi
@@ -91,12 +116,28 @@ fi
 # 3. Status-check state at the PR HEAD (where checks ran). The merge
 #    commit doesn't get its own checks; we evaluate the PR's last
 #    commit, which is what branch protection compared against.
-STATUS=$(curl -sS -H "$AUTH" \
+# Fail-closed: verify HTTP 200. A 401/403/404 means the status is
+# unreadable — we must NOT treat that as "no statuses" and skip checks.
+STATUS_TMP=$(mktemp)
+STATUS_HTTP=$(curl -sS -o "$STATUS_TMP" -w '%{http_code}' -H "$AUTH" \
  "${API}/repos/${OWNER}/${NAME}/commits/${HEAD_SHA}/status")
+STATUS=$(cat "$STATUS_TMP")
+rm -f "$STATUS_TMP"
+if [ "$STATUS_HTTP" != "200" ]; then
+  echo "::error::GET /commits/${HEAD_SHA}/status returned HTTP ${STATUS_HTTP} — cannot evaluate required checks."
+  exit 1
+fi
+# FAIL-CLOSED: a 200 status response missing the 'statuses' array, or with
+# 'statuses' set to a non-array type (null/string/object), must NOT be treated
+# as "no checks" — that would silently declare all checks green.
+if ! echo "$STATUS" | jq -e '(.statuses | type) == "array"' >/dev/null; then
+  echo "::error::GET /commits/${HEAD_SHA}/status returned HTTP 200 but 'statuses' is missing or not an array — cannot evaluate required checks."
+  exit 1
+fi
 declare -A CHECK_STATE
 while IFS=$'\t' read -r ctx state; do
  [ -n "$ctx" ] && CHECK_STATE[$ctx]="$state"
-done < <(echo "$STATUS" | jq -r '.statuses // [] | .[] | "\(.context)\t\(.status)"')
+done < <(echo "$STATUS" | jq -r '.statuses | .[] | "\(.context)\t\(.status)"')

 # 4. For each required check, was it green at merge? YAML block scalars
 #    (`|`) leave a trailing newline; skip blank/whitespace-only lines.
@@ -552,23 +552,34 @@ def find_open_issue(title: str) -> dict | None:
    hourly; failing one cycle loudly is strictly better than silently
    duplicating.

-    Gitea issue search returns at most page=50 per page; one page is
-    enough as long as `[ci-drift]` issues are a tiny minority. (See
-    follow-up issue for Link-header pagination.)
+    Paginates through all open issues (limit=50 per page) until the
+    title is found or the result set is exhausted. Previously only one
+    page was fetched, causing duplicate [ci-drift] issues when the
+    existing tracking issue fell beyond page 1.
    """
-    _, results = api(
-        "GET",
-        f"/repos/{OWNER}/{NAME}/issues",
-        query={"state": "open", "type": "issues", "limit": "50"},
-    )
-    if not isinstance(results, list):
-        raise ApiError(
-            f"issue search returned non-list body (got {type(results).__name__})"
+    page = 1
+    while True:
+        _, results = api(
+            "GET",
+            f"/repos/{OWNER}/{NAME}/issues",
+            query={
+                "state": "open",
+                "type": "issues",
+                "limit": "50",
+                "page": str(page),
+            },
        )
-    for issue in results:
-        if issue.get("title") == title:
-            return issue
-    return None
+        if not isinstance(results, list):
+            raise ApiError(
+                f"issue search returned non-list body (got {type(results).__name__})"
+            )
+        for issue in results:
+            if issue.get("title") == title:
+                return issue
+        # Fewer than limit results means last page reached.
+        if len(results) < 50:
+            return None
+        page += 1


 def render_body(branch: str, findings: list[str], debug: dict) -> str:
@@ -9,17 +9,33 @@ queue. This script provides the missing serialized policy in user space:
   candidate (REQUEST_CHANGES, mergeable!=True, insufficient genuine approvals,
   or red required CI) is SKIPPED so it cannot head-of-line block newer ready
   PRs; the scan continues to the next candidate.
-2. Refuse to act unless main's BP-required contexts are green.
+2. Refuse to act unless main's BP-required contexts are green. This is also
+   the serialized backstop for direct-merge (see below): after a direct merge,
+   main re-runs push CI and this gate PAUSES the queue if main goes red, so no
+   merge piles onto an unverified/red main (issue #2358).
 3. Refuse fork PRs; the queue may only mutate same-repo branches.
-4. If the PR branch does not contain current main, call Gitea's
-   /pulls/{n}/update endpoint and stop. CI must rerun on the updated head.
+4. DIRECT-MERGE when conflict-free (issue #2358). When Gitea reports the PR
+   conflict-free (mergeable is True) and the merge bar below is met, MERGE IT
+   DIRECTLY — even if its head does not contain current main. We do NOT call
+   /pulls/{n}/update first: branch protection does not require strict
+   up-to-date, so behind-main conflict-free PRs merge cleanly, and calling
+   /update would trigger Gitea dismiss_stale_approvals (dismissing the genuine
+   approvals and forcing a re-review every tick — the rebase-churn bottleneck).
+   The /update path is used ONLY when the PR is DEFINITIVELY not mergeable
+   (mergeable is literal False) AND its head lacks current main — refreshing the
+   branch may resolve a behind-main non-conflict; a real conflict returns HTTP
+   409 and the PR is HELD per #2352. mergeable=None/missing (Gitea STILL
+   COMPUTING conflict state) is a distinct fail-closed WAIT: never merged AND
+   never /update'd — calling /update during the compute window would dismiss the
+   PR's genuine approvals (dismiss_stale_approvals) and re-introduce the exact
+   rebase-churn this queue eliminates. None is re-checked next tick.
 5. Merge ONLY when, on the PR's CURRENT head sha:
     - >= REQUIRED_APPROVALS distinct GENUINE official APPROVED reviews from
       the recognised reviewer set (not stale, not dismissed, commit_id ==
       current head), AND
     - no open official REQUEST_CHANGES on the current head, AND
     - every BP-required status context is green, AND
-     - the PR is mergeable.
+     - the PR is mergeable (Gitea reports it conflict-free).

 Authoritative gates (fail-closed):
  - The REQUIRED status contexts come from BRANCH PROTECTION
@@ -268,6 +284,34 @@ def api(
        return status, {"_raw": raw.decode("utf-8", errors="replace")}


+def api_paginated(
+    method: str,
+    path: str,
+    *,
+    query: dict[str, str] | None = None,
+    page_size: int = 50,
+) -> list[dict]:
+    """Fetch all pages of a paginated Gitea list endpoint.
+
+    Gitea paginates with `page` (1-indexed) and `limit`. We loop until a
+    page returns fewer than `page_size` items, indicating the end.
+    """
+    results: list[dict] = []
+    page = 1
+    while True:
+        page_query = dict(query or {})
+        page_query["page"] = str(page)
+        page_query["limit"] = str(page_size)
+        _, body = api(method, path, query=page_query)
+        if not isinstance(body, list):
+            raise ApiError(f"{path} paginated response not list")
+        results.extend(body)
+        if len(body) < page_size:
+            break
+        page += 1
+    return results
+
+
 def required_contexts(raw: str) -> list[str]:
    return [part.strip() for part in raw.split(",") if part.strip()]

@@ -300,17 +344,18 @@ def _is_tier_low_pending_ok(
 ) -> bool:
    """Return True if tier:low PR can tolerate sop-checklist pending state.

-    Per sop-checklist-config.yaml tier_failure_mode, tier:low uses soft-fail:
-    sop-checklist posts state=pending when acks are satisfied (missing
-    manager/ceo acks are informational only). The queue should accept
-    pending instead of waiting for success.
+    GENERIC PENDING-AS-GREEN REMOVED (Researcher + CR2 RC on #2368):
+    The prior soft-fail accepted ANY pending sop-checklist for tier:low,
+    which allowed required checks to pass without genuine verification.
+    Pending required sop-checklist must now always HOLD and appear in
+    missing_or_bad. This function is retained as a policy hook but
+    currently always returns False so pending never counts green.
+
+    If a positively identifiable genuine soft-fail state is defined in
+    future (e.g., a specific check-run conclusion), implement it here
+    with strict positive identification — never default to pass.
    """
-    if "tier:low" not in pr_labels:
-        return False
-    if "sop-checklist" not in context:
-        return False
-    status = latest_statuses.get(context) or {}
-    return status_state(status) == "pending"
+    return False


 def required_contexts_green(
@@ -593,29 +638,32 @@ def evaluate_merge_readiness(
    approvers: set[str],
    request_changes: list[str],
    pr_has_current_base: bool,
-    mergeable: bool,
+    mergeable: bool | None,
    pr_labels: set[str] | None = None,
 ) -> MergeDecision:
    # 1) Main's push-required contexts must be green. Combined state can be
    #    "failure" due to non-blocking jobs (continue-on-error: true) that do
    #    not gate merges, so check the explicit required set, not combined.
+    #
+    #    This main-green gate is ALSO the serialized backstop that makes the
+    #    direct-merge (no update) path safe (issue #2358): after a direct merge
+    #    of a behind-main PR, main re-runs its push CI; if a semantic main-break
+    #    slips through (PR green standalone but broken when combined with newer
+    #    main), main's required contexts go red and this gate PAUSES the queue —
+    #    no further merge piles onto an unverified/red main until it is green.
    main_latest = latest_statuses_by_context(main_status.get("statuses") or [])
    main_ok, main_bad = required_contexts_green(main_latest, push_required_contexts())
    if not main_ok:
        return MergeDecision(False, "pause", "main required contexts not green: " + ", ".join(main_bad))

-    # 2) PR head must contain current main.
-    if not pr_has_current_base:
-        return MergeDecision(False, "update", "PR head does not contain current main")
-
-    # 3) No open official REQUEST_CHANGES on the current head.
+    # 2) No open official REQUEST_CHANGES on the current head.
    if request_changes:
        return MergeDecision(
            False, "wait",
            "open REQUEST_CHANGES on current head from: " + ", ".join(sorted(request_changes)),
        )

-    # 4) Enough distinct genuine official approvals on the current head.
+    # 3) Enough distinct genuine official approvals on the current head.
    if len(approvers) < required_approvals:
        return MergeDecision(
            False, "wait",
@@ -624,7 +672,7 @@ def evaluate_merge_readiness(
            f"need {required_approvals}",
        )

-    # 5) Every BRANCH-PROTECTION-REQUIRED status context must be green. This is
+    # 4) Every BRANCH-PROTECTION-REQUIRED status context must be green. This is
    #    the authoritative status gate — NON-required reds (qa-review,
    #    security-review, sop-tier/sop-checklist when not BP-required, E2E Chat,
    #    Staging SaaS, ci-arm64-advisory, continue-on-error jobs) are NOT
@@ -634,16 +682,53 @@ def evaluate_merge_readiness(
    if not ok:
        return MergeDecision(False, "wait", "required contexts not green: " + ", ".join(missing_or_bad))

-    # 6) Gitea must consider the PR mergeable (no conflicts).
-    if not mergeable:
-        return MergeDecision(False, "wait", "PR is not mergeable (conflicts)")
+    # 5) DIRECT-MERGE when conflict-free (issue #2358 — throughput fix).
+    #    If Gitea reports the PR conflict-free (mergeable is True), MERGE IT
+    #    DIRECTLY even if its head does not yet contain current main. Branch
+    #    protection does NOT require strict up-to-date, so a behind-main but
+    #    conflict-free PR merges cleanly. We deliberately do NOT call
+    #    /pulls/{n}/update first: update triggers Gitea dismiss_stale_approvals,
+    #    which would dismiss the PR's genuine approvals and force a full
+    #    re-review every tick — the rebase-churn bottleneck that collapsed
+    #    throughput to ~0/hr with dozens of mergeable PRs open.
+    #
+    #    The merge bar is UNCHANGED: we only reach here with main green +
+    #    >= required genuine approvals on the current head + no open
+    #    REQUEST_CHANGES + every BP-required context green. The trade-off is
+    #    that the PR's CI ran on a possibly-behind base, so a SEMANTIC main-break
+    #    is caught by POST-merge main CI (step 1's pause backstop) rather than
+    #    pre-merge. force_merge is used ONLY for missing-but-non-required
+    #    governance reds (required are green + approvals genuine), never to
+    #    bypass a failing required context or an approval shortfall.
+    if mergeable is True:
+        force = _non_required_red_present(latest, required_contexts)
+        return MergeDecision(True, "merge", "ready", force=force)

-    # Ready. Use force_merge ONLY if the merge would otherwise be blocked by
-    # missing-but-non-required governance contexts. Required are green and
-    # approvals are genuine, so force only bypasses non-required reds — never a
-    # failing required context or missing approval.
-    force = _non_required_red_present(latest, required_contexts)
-    return MergeDecision(True, "merge", "ready", force=force)
+    # 6) NOT (yet) mergeable. TRI-STATE, fail-closed — never merge on an unknown.
+    #    We MUST distinguish "still computing" (None/missing) from a "definitive
+    #    conflict" (False); collapsing them would route a behind-main but
+    #    STILL-COMPUTING PR into the /update path, whose dismiss_stale_approvals
+    #    is the rebase-churn this change eliminates.
+    #
+    #    mergeable is None  → Gitea has NOT finished computing conflict state.
+    #    WAIT: do nothing this tick — never /update (would dismiss genuine
+    #    approvals during the compute window → churn), never merge. Re-check next
+    #    tick once Gitea reports a decisive True/False.
+    if mergeable is None:
+        return MergeDecision(
+            False, "wait",
+            "PR mergeability is still being computed (mergeable=None) — waiting",
+        )
+
+    # mergeable is False → DEFINITIVE not-mergeable. If the head also does not
+    #    contain current main, try the /update path to refresh the branch (this
+    #    may resolve a behind-main non-conflict; a real conflict returns HTTP 409
+    #    and process_once HOLDs the PR per #2352). If the head already contains
+    #    current main yet Gitea still reports not-mergeable, there is nothing the
+    #    queue can do (genuine conflict against current main) — WAIT.
+    if not pr_has_current_base:
+        return MergeDecision(False, "update", "PR not mergeable and head does not contain current main")
+    return MergeDecision(False, "wait", "PR is not mergeable (conflicts)")


 def get_branch_head(branch: str) -> str:
@@ -659,32 +744,23 @@ def get_combined_status(sha: str) -> dict:
    """Combined status + all individual statuses for `sha`.

    The /status endpoint caps the `statuses` array at 30 entries (Gitea
-    default page size), so we fetch the full list via /statuses with a
-    higher limit. The combined `state` still comes from /status.
+    default page size), so we fetch the full list via /statuses. The combined
+    `state` still comes from /status.

-    Fail-closed: the PRIMARY /status fetch must succeed. If it raises, the
-    error propagates so the caller skips this PR this tick (we never treat a
-    failed status fetch as green — dev-sop "no fail-open"). Only the SECONDARY
-    /statuses enrichment (which merely extends the per-context list beyond the
-    30-entry cap) is best-effort; if it fails we still have the combined set.
+    Fail-closed: BOTH the PRIMARY /status fetch AND the SECONDARY /statuses
+    enrichment must succeed. If either raises, the error propagates so the
+    caller skips this PR this tick (we never treat a failed status fetch as
+    green — dev-sop "no fail-open"). A paginated /statuses error must NOT
+    silently degrade to an incomplete status set.
    """
    _, combined = api("GET", f"/repos/{OWNER}/{NAME}/commits/{sha}/status")
    if not isinstance(combined, dict):
        raise ApiError(f"status for {sha} response not object")
    combined_statuses: list[dict] = combined.get("statuses") or []
-    try:
-        _, all_statuses_raw = api(
-            "GET",
-            f"/repos/{OWNER}/{NAME}/commits/{sha}/statuses",
-            query={"limit": "50"},
-        )
-        if isinstance(all_statuses_raw, list):
-            all_statuses: list[dict] = list(all_statuses_raw)
-        else:
-            all_statuses = []
-    except (ApiError, urllib.error.URLError, TimeoutError, OSError) as exc:
-        sys.stderr.write(f"::warning::could not fetch full statuses list for {sha[:8]}: {exc}\n")
-        all_statuses = []
+    all_statuses = api_paginated(
+        "GET",
+        f"/repos/{OWNER}/{NAME}/commits/{sha}/statuses",
+    )
    # Build latest per context: process combined (ascending→reverse=newest
    # first), then fill gaps from all_statuses (already newest-first).
    latest: dict[str, dict] = {}
@@ -701,19 +777,15 @@ def get_combined_status(sha: str) -> dict:


 def list_queued_issues() -> list[dict]:
-    _, body = api(
+    return api_paginated(
        "GET",
        f"/repos/{OWNER}/{NAME}/issues",
        query={
            "state": "open",
            "type": "pulls",
            "labels": QUEUE_LABEL,
-            "limit": "50",
        },
    )
-    if not isinstance(body, list):
-        raise ApiError("queued issues response not list")
-    return body


 def list_candidate_issues(*, auto_discover: bool) -> list[dict]:
@@ -727,18 +799,14 @@ def list_candidate_issues(*, auto_discover: bool) -> list[dict]:
    """
    if not auto_discover:
        return list_queued_issues()
-    _, body = api(
+    return api_paginated(
        "GET",
        f"/repos/{OWNER}/{NAME}/issues",
        query={
            "state": "open",
            "type": "pulls",
-            "limit": "50",
        },
    )
-    if not isinstance(body, list):
-        raise ApiError("candidate issues response not list")
-    return body


 def get_pull(pr_number: int) -> dict:
@@ -1064,12 +1132,20 @@ def _evaluate_candidate(
    # never treated as green).
    pr_status = get_combined_status(head_sha)
    pr_labels = label_names(pr)
-    # FAIL-CLOSED: Gitea returns mergeable=None (or omits the field) while it is
-    # still COMPUTING conflict state. Only the literal True is decisive proof the
-    # PR is conflict-free; None and False both mean "not (yet) mergeable". We must
-    # NOT autonomously merge on an unknown — treat anything but True as not-yet-
-    # mergeable so evaluate_merge_readiness returns a "wait" decision.
-    mergeable = pr.get("mergeable") is True
+    # FAIL-CLOSED, TRI-STATE: Gitea returns mergeable=None (or omits the field)
+    # while it is still COMPUTING conflict state, mergeable=False for a definitive
+    # conflict, and mergeable=True only when it has proven the PR conflict-free.
+    # We preserve all THREE states (do NOT collapse None/missing into False):
+    #   - True            → direct-merge eligible (step 5).
+    #   - None / missing  → still computing → WAIT (never merge, never update,
+    #                       never dismiss approvals); re-check next tick.
+    #   - False           → definitive conflict → the update/hold path (step 6).
+    # Collapsing None→False would route a behind-main but STILL-COMPUTING PR into
+    # the /update path, which triggers dismiss_stale_approvals — the exact
+    # rebase-churn this change eliminates. Normalize only to the literal True /
+    # False / None set (some Gitea versions omit the key entirely → None).
+    raw_mergeable = pr.get("mergeable")
+    mergeable: bool | None = raw_mergeable if isinstance(raw_mergeable, bool) else None

    reviews = get_pull_reviews(pr_number)
    approvers, request_changes = genuine_approvals(
@@ -1098,18 +1174,18 @@ def main() -> int:
    try:
        return process_once(dry_run=args.dry_run)
    except ApiError as exc:
-        # API errors (401/403/404/500) are transient for a queue tick —
-        # log and exit 0 so the workflow is not marked failed and the next
-        # tick can retry. Returning non-zero would permanently fail the
-        # workflow run, blocking future ticks.
+        # FAIL-CLOSED: API errors are not "transient success" — they mean
+        # the queue could not evaluate merge state. Returning 0 hides
+        # persistent infra issues (auth drift, endpoint outages) from
+        # operators. Return 1 so the cron job surfaces red and paging fires.
        sys.stderr.write(f"::error::queue API error: {exc}\n")
-        return 0
+        return 1
    except urllib.error.URLError as exc:
        sys.stderr.write(f"::error::queue network error: {exc}\n")
-        return 0
+        return 1
    except TimeoutError as exc:
        sys.stderr.write(f"::error::queue timeout: {exc}\n")
-        return 0
+        return 1


 if __name__ == "__main__":
@@ -305,9 +305,9 @@ def validate_tracker(
    if status == "error":
        sys.stderr.write(
            f"::error::issue {slug}#{num} fetch errored — treating as "
-            f"unverified, skipping this check.\n"
+            f"unverified, FAILING CLOSED (do not skip on outage).\n"
        )
-        return (True, "fetch-error — skipped")
+        return (False, f"{slug}#{num} fetch errored — cannot verify tracker")

    assert payload is not None
    state = payload.get("state", "")
@@ -546,16 +546,24 @@ def verify_flip(flip: dict, branch: str, n: int) -> dict:

    shas = recent_commits_on_branch(branch, n)
    if not shas:
-        result["warnings"].append(
-            f"no recent commits on {branch} (cannot verify flip)"
-        )
+        result["masked_runs"].append({
+            "sha": "",
+            "status": "unverified",
+            "target_url": "",
+            "samples": [f"no recent commits on {branch} — cannot verify flip"],
+        })
        return result

    for sha in shas:
        try:
            status_doc = combined_status(sha)
        except ApiError as e:
-            result["warnings"].append(f"combined-status for {sha}: {e}")
+            result["masked_runs"].append({
+                "sha": sha,
+                "status": "error",
+                "target_url": "",
+                "samples": [f"combined-status API error: {e}"],
+            })
            continue
        statuses = status_doc.get("statuses") or []
        # First entry matching the context name. Newest SHAs come
@@ -582,6 +590,17 @@ def verify_flip(flip: dict, branch: str, n: int) -> dict:
                        "target_url": target_url,
                        "samples": ["[log unavailable; status itself is " + state + "]"],
                    })
+                elif state == "success":
+                    # Fail-closed: unreadable log on a success status is a
+                    # potential Quirk #10 mask (continue-on-error hiding real
+                    # failures). We cannot verify it's clean, so treat as
+                    # masked rather than allowing the flip.
+                    result["masked_runs"].append({
+                        "sha": sha,
+                        "status": state,
+                        "target_url": target_url,
+                        "samples": ["[log unavailable; cannot verify status is genuine — treat as masked]"],
+                    })
                break
            samples = grep_fail_markers(log_text)
            if state in ("failure", "error"):
@@ -605,10 +624,12 @@ def verify_flip(flip: dict, branch: str, n: int) -> dict:
            break

    if result["checked_commits"] == 0:
-        result["warnings"].append(
-            f"no runs of {target_context!r} found in the last {n} commits on "
-            f"{branch} — cannot verify; allowing flip with warning"
-        )
+        result["masked_runs"].append({
+            "sha": "",
+            "status": "unverified",
+            "target_url": "",
+            "samples": [f"no runs of {target_context!r} found in the last {n} commits on {branch} — cannot verify flip"],
+        })
    return result


@@ -144,18 +144,14 @@ debug "tier=$TIER"
 # as unachievable (would always fail) — operators notice the clear error
 # and create the missing team.
 #
-# Current Gitea teams: ceo, engineers, managers
-# Future teams (create before removing "???" fallback): qa, security, security-audit
+# Current Gitea teams: ceo, engineers, managers, qa, security
 declare -A TIER_EXPR=(
  # tier:low — same as previous OR gate: any engineer, manager, or ceo.
  ["tier:low"]="engineers,managers,ceo"

-  # tier:medium — AND of (managers) AND (engineers) AND (qa???,security???)
-  # The qa+security clause requires both teams to exist; when not yet
-  # created, the PR author is responsible for adding them before requesting
-  # approval on a tier:medium PR. Ops: create qa + security Gitea teams
-  # and update this map to remove the "???" markers (internal#189 follow-up).
-  ["tier:medium"]="managers AND engineers AND qa???,security???"
+  # tier:medium — AND of (managers) AND (engineers) AND (qa,security)
+  # ≥1 approver from managers AND ≥1 from engineers AND ≥1 from qa OR security.
+  ["tier:medium"]="managers AND engineers AND qa,security"

  # tier:high — ceo only. The AND-composition adds no value for a
  # single-team gate, but the framework is wired for consistency.
@@ -689,8 +689,8 @@ def reap_branch(
        shas = list_recent_commit_shas(branch, limit)
    except ApiError as e:
        print(
-            "::warning::status-reaper skipped this tick because the "
-            f"commit list could not be read after retries: {e}"
+            "::error::status-reaper cannot run: commit-list API failed "
+            f"after retries: {e}"
        )
        return {
            "scanned_shas": 0,
@@ -704,6 +704,7 @@ def reap_branch(
            "compensated_cancelled_push": 0,
            "preserved_pr_without_push_success": 0,
            "compensated_per_sha": {},
+            "sha_api_errors": 0,
            "skipped": True,
            "skip_reason": "commit-list-api-error",
        }
@@ -720,6 +721,7 @@ def reap_branch(
        "compensated_cancelled_push": 0,
        "preserved_pr_without_push_success": 0,
        "compensated_per_sha": {},
+        "sha_api_errors": 0,
    }

    for sha in shas:
@@ -731,8 +733,9 @@ def reap_branch(
        try:
            combined = get_combined_status(sha)
        except ApiError as e:
+            aggregate["sha_api_errors"] += 1
            print(
-                f"::warning::get_combined_status({sha[:10]}) failed; "
+                f"::error::get_combined_status({sha[:10]}) failed; "
                f"skipping this SHA: {e}"
            )
            continue
@@ -819,6 +822,14 @@ def main() -> int:
            sort_keys=True,
        )
    )
+    # Observability: infra-failure → red. If the commit list could not be
+    # read or any per-SHA status fetch failed, the tick is incomplete and
+    # must be observable as a failure (non-zero exit) so the cron bot or
+    # runner surface alerts.
+    if counters.get("skipped"):
+        return 1
+    if counters.get("sha_api_errors", 0) > 0:
+        return 1
    return 0


@@ -0,0 +1,119 @@
+#!/usr/bin/env bash
+# test_audit_force_merge.sh — regression lock for audit-force-merge fail-closed
+# behavior. Verifies every schema validation path via direct jq filter tests.
+#
+# Usage: bash test_audit_force_merge.sh
+
+set -euo pipefail
+
+fail() { echo "FAIL: $*" >&2; exit 1; }
+pass() { echo "PASS: $*"; }
+
+[ -x "$(command -v jq)" ] || { echo "SKIP: jq not on PATH"; exit 0; }
+
+HEAD_SHA="deadbeef00000000000000000000000000000000"
+
+# The schema validation jq expression from audit-force-merge.sh.
+validate_pr_schema() {
+  jq -r '
+    (.merged | type == "boolean") and
+    (.merge_commit_sha | type == "string") and
+    (.merged_by | type == "object") and (.merged_by.login | type == "string") and
+    (.base | type == "object") and (.base.ref | type == "string") and
+    (.head | type == "object") and (.head.sha | type == "string")
+  '
+}
+
+validate_statuses_type() {
+  jq -r '(.statuses | type) == "array"'
+}
+
+# T1 — valid PR payload → true
+T1=$(echo '{"merged":true,"merge_commit_sha":"abc","merged_by":{"login":"u"},"base":{"ref":"main"},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T1" = "true" ] || fail "T1: valid payload should pass schema"
+pass "T1: valid payload passes schema"
+
+# T2 — merged=false (valid types) → true (schema is about types, not values)
+T2=$(echo '{"merged":false,"merge_commit_sha":"abc","merged_by":{"login":"u"},"base":{"ref":"main"},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T2" = "true" ] || fail "T2: merged=false with valid types should pass schema"
+pass "T2: merged=false with valid types passes schema"
+
+# T3 — missing merged field → false
+T3=$(echo '{"merge_commit_sha":"abc","merged_by":{"login":"u"},"base":{"ref":"main"},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T3" = "false" ] || fail "T3: missing merged should fail schema"
+pass "T3: missing merged fails schema"
+
+# T4 — merged is string "true" instead of boolean → false
+T4=$(echo '{"merged":"true","merge_commit_sha":"abc","merged_by":{"login":"u"},"base":{"ref":"main"},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T4" = "false" ] || fail "T4: merged as string should fail schema"
+pass "T4: merged as string fails schema"
+
+# T5 — merge_commit_sha is null → false
+T5=$(echo '{"merged":true,"merge_commit_sha":null,"merged_by":{"login":"u"},"base":{"ref":"main"},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T5" = "false" ] || fail "T5: null merge_commit_sha should fail schema"
+pass "T5: null merge_commit_sha fails schema"
+
+# T6 — merged_by is null → false
+T6=$(echo '{"merged":true,"merge_commit_sha":"abc","merged_by":null,"base":{"ref":"main"},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T6" = "false" ] || fail "T6: null merged_by should fail schema"
+pass "T6: null merged_by fails schema"
+
+# T7 — base.ref is number → false
+T7=$(echo '{"merged":true,"merge_commit_sha":"abc","merged_by":{"login":"u"},"base":{"ref":123},"head":{"sha":"def"}}' | validate_pr_schema)
+[ "$T7" = "false" ] || fail "T7: numeric base.ref should fail schema"
+pass "T7: numeric base.ref fails schema"
+
+# T8 — head is missing → false
+T8=$(echo '{"merged":true,"merge_commit_sha":"abc","merged_by":{"login":"u"},"base":{"ref":"main"}}' | validate_pr_schema)
+[ "$T8" = "false" ] || fail "T8: missing head should fail schema"
+pass "T8: missing head fails schema"
+
+# T9 — statuses missing → false
+T9=$(echo '{}' | validate_statuses_type)
+[ "$T9" = "false" ] || fail "T9: missing statuses should fail type check"
+pass "T9: missing statuses fails type check"
+
+# T10 — statuses is string → false
+T10=$(echo '{"statuses":"unexpected"}' | validate_statuses_type)
+[ "$T10" = "false" ] || fail "T10: string statuses should fail type check"
+pass "T10: string statuses fails type check"
+
+# T11 — statuses is null → false
+T11=$(echo '{"statuses":null}' | validate_statuses_type)
+[ "$T11" = "false" ] || fail "T11: null statuses should fail type check"
+pass "T11: null statuses fails type check"
+
+# T12 — statuses is array → true
+T12=$(echo '{"statuses":[{"context":"c1","status":"success"}]}' | validate_statuses_type)
+[ "$T12" = "true" ] || fail "T12: array statuses should pass type check"
+pass "T12: array statuses passes type check"
+
+# T13 — empty array statuses → true
+T13=$(echo '{"statuses":[]}' | validate_statuses_type)
+[ "$T13" = "true" ] || fail "T13: empty array statuses should pass type check"
+pass "T13: empty array statuses passes type check"
+
+# T14-T16: REQUIRED_CHECKS_JSON branch entry validation
+validate_required_checks_json() {
+  local branch="$1"
+  local json="$2"
+  echo "$json" | jq -r --arg branch "$branch" 'has($branch) and (.[$branch] | type == "array")'
+}
+
+# T14 — branch exists and is array → true
+T14=$(validate_required_checks_json "main" '{"main":["CI / all-required"]}')
+[ "$T14" = "true" ] || fail "T14: existing array branch should pass"
+pass "T14: existing array branch passes"
+
+# T15 — branch missing → false
+T15=$(validate_required_checks_json "staging" '{"main":["CI / all-required"]}')
+[ "$T15" = "false" ] || fail "T15: missing branch should fail"
+pass "T15: missing branch fails"
+
+# T16 — branch entry is string instead of array → false
+T16=$(validate_required_checks_json "main" '{"main":"CI / all-required"}')
+[ "$T16" = "false" ] || fail "T16: string branch entry should fail"
+pass "T16: string branch entry fails"
+
+echo
+echo "ALL AUDIT-FORCE-MERGE CHECKS PASSED"
@@ -2,6 +2,8 @@ import importlib.util
 import sys
 from pathlib import Path

+import pytest
+
 SCRIPT = Path(__file__).resolve().parents[1] / "gitea-merge-queue.py"
 spec = importlib.util.spec_from_file_location("gitea_merge_queue", SCRIPT)
 mq = importlib.util.module_from_spec(spec)
@@ -44,6 +46,35 @@ def test_required_contexts_green_rejects_missing_and_pending():
    ]


+def test_required_contexts_green_rejects_volume_skipped_even_for_tier_low():
+    """volume-skipped pending is a partial view, not a genuine soft-fail.
+
+    Per sop-checklist.py:1179-1187, volume_skipped posts pending with a
+    '[volume-skipped]' prefix. The merge queue must NOT treat this as an
+    acceptable soft-fail for tier:low — the gate did not finish evaluating.
+    """
+    latest = mq.latest_statuses_by_context([
+        {"context": "CI / all-required (pull_request)", "status": "success"},
+        {
+            "context": "sop-checklist / all-items-acked (pull_request)",
+            "status": "pending",
+            "description": "[volume-skipped] comment-cap=1000 hit; please file ...",
+        },
+    ])
+
+    ok, missing_or_bad = mq.required_contexts_green(
+        latest,
+        [
+            "CI / all-required (pull_request)",
+            "sop-checklist / all-items-acked (pull_request)",
+        ],
+        pr_labels={"tier:low"},
+    )
+
+    assert ok is False
+    assert "sop-checklist / all-items-acked (pull_request)=pending" in missing_or_bad
+
+
 def test_choose_next_pr_sorts_by_queue_label_timestamp_then_number():
    issues = [
        {
@@ -112,13 +143,72 @@ def test_merge_decision_requires_main_green_pr_green_and_current_base():
    assert decision.force is False  # no non-required reds present


-def test_merge_decision_updates_stale_pr_before_merge():
-    decision = mq.evaluate_merge_readiness(**_ready_kwargs(pr_has_current_base=False))
+def test_behind_main_but_mergeable_pr_merges_directly():
+    """§SOP-22 (#2358): a behind-main but CONFLICT-FREE PR (mergeable is True)
+    merges DIRECTLY — no update step. Branch protection does not require strict
+    up-to-date, and calling /update would dismiss the genuine approvals
+    (dismiss_stale_approvals), forcing re-review every tick (the throughput
+    bottleneck). This replaces the old update-before-merge behavior."""
+    decision = mq.evaluate_merge_readiness(
+        **_ready_kwargs(pr_has_current_base=False, mergeable=True)
+    )
+
+    assert decision.ready is True
+    assert decision.action == "merge"
+
+
+def test_behind_main_and_not_mergeable_pr_updates():
+    """The /update path is reached ONLY when the PR is NOT mergeable AND its head
+    lacks current main — refreshing the branch may resolve a behind-main
+    non-conflict; a real conflict 409s and is held (#2352)."""
+    decision = mq.evaluate_merge_readiness(
+        **_ready_kwargs(pr_has_current_base=False, mergeable=False)
+    )

    assert decision.ready is False
    assert decision.action == "update"


+def test_current_base_but_not_mergeable_pr_waits():
+    """Up-to-date with main yet Gitea reports not-mergeable → genuine conflict
+    against current main (or still computing). The queue cannot act: WAIT,
+    never update (update would not help) and never merge (fail-closed)."""
+    decision = mq.evaluate_merge_readiness(
+        **_ready_kwargs(pr_has_current_base=True, mergeable=False)
+    )
+
+    assert decision.ready is False
+    assert decision.action == "wait"
+    assert "not mergeable" in decision.reason
+
+
+def test_behind_main_and_mergeable_none_waits_not_update():
+    """§SOP-22 (CR2 #2374) — the churn-residual fix. A BEHIND-MAIN PR whose
+    mergeability Gitea is STILL COMPUTING (mergeable is None) must WAIT, NOT take
+    the /update path. The old code collapsed None→False, so a behind-main +
+    None PR returned action="update" → /pulls/{n}/update → dismiss_stale_approvals
+    → the exact rebase-churn this change eliminates, fired during the compute
+    window. None and False are now DISTINCT: None waits, False updates."""
+    decision = mq.evaluate_merge_readiness(
+        **_ready_kwargs(pr_has_current_base=False, mergeable=None)
+    )
+
+    assert decision.ready is False
+    assert decision.action == "wait"  # NOT "update" — no churn during compute
+    assert "computed" in decision.reason
+
+
+def test_current_base_and_mergeable_none_waits():
+    """Up-to-date with main + mergeable None (still computing) → WAIT (unchanged
+    fail-closed; just confirming None is never merged regardless of base)."""
+    decision = mq.evaluate_merge_readiness(
+        **_ready_kwargs(pr_has_current_base=True, mergeable=None)
+    )
+
+    assert decision.ready is False
+    assert decision.action == "wait"
+
+
 def test_MergePermissionError_inherits_from_ApiError():
    assert issubclass(mq.MergePermissionError, mq.ApiError)

@@ -475,6 +565,131 @@ def test_process_once_merges_when_mergeable_is_true(monkeypatch):
    assert calls["hold_label"] is None


+def test_process_once_behind_main_mergeable_none_waits_no_update(monkeypatch):
+    """§SOP-22 (CR2 #2374) — end-to-end churn-residual regression. A BEHIND-MAIN
+    PR (commits do NOT contain main_sha) whose mergeability Gitea is STILL
+    COMPUTING (mergeable=None) must WAIT: process_once returns 0 and NEVER calls
+    update_pull (which dismisses genuine approvals via dismiss_stale_approvals)
+    NOR merge_pull NOR hold. The old None→False collapse routed this exact case
+    into the /update path → approval-dismissing rebase churn during the compute
+    window. This proves the durable churn elimination: no update, approvals
+    preserved, re-checked next tick."""
+    calls = {"merge_attempts": 0, "hold_label": None, "updated": False}
+    _fully_ready_process_once_monkeypatch(monkeypatch, mergeable=None, calls=calls)
+    # Make the head BEHIND main: commits do NOT contain main_sha. This is the
+    # case the bug missed (the prior None test had current base, masking it).
+    behind_head = "a" * 40
+    monkeypatch.setattr(mq, "get_pull_commits", lambda n: [{"sha": behind_head}])
+
+    rc = mq.process_once(dry_run=False)
+
+    assert rc == 0
+    assert calls["updated"] is False  # NO /update → approvals NOT dismissed
+    assert calls["merge_attempts"] == 0  # never merge on an unknown
+    assert calls["hold_label"] is None  # transient → not held, retried next tick
+
+
+# --------------------------------------------------------------------------
+# §SOP-22: DIRECT-MERGE throughput fix (#2358). A conflict-free 2-genuine PR
+# merges WITHOUT a pre-merge /update call, so its approvals are NOT dismissed by
+# dismiss_stale_approvals. The merge bar (2-genuine-on-current-head +
+# BP-required green + mergeable + no RC + opt-out) is UNCHANGED; only the
+# unnecessary update-before-merge churn is removed. The /update path survives
+# for the genuine case it is needed (not-mergeable + behind-main), where a real
+# conflict 409s and is held per #2352. mergeable=None stays fail-closed.
+# --------------------------------------------------------------------------
+
+
+def test_process_once_merges_conflict_free_pr_without_update(monkeypatch):
+    """§SOP-22(a) — the core throughput fix. A conflict-free, fully-approved PR
+    merges WITHOUT update_pull ever being called. The old behavior called
+    /update first whenever the head lacked current main, which dismissed the 2
+    genuine approvals (dismiss_stale_approvals) and forced re-review every tick.
+    Assert update_pull is NOT invoked and merge_pull IS invoked."""
+    calls = {"merge_attempts": 0, "hold_label": None, "updated": False}
+    _fully_ready_process_once_monkeypatch(monkeypatch, mergeable=True, calls=calls)
+    # Make the head BEHIND main: commits do NOT contain main_sha. Under the old
+    # logic this alone forced an update_pull; under the fix it merges directly.
+    head_sha = "a" * 40
+    monkeypatch.setattr(mq, "get_pull_commits", lambda n: [{"sha": head_sha}])
+
+    rc = mq.process_once(dry_run=False)
+
+    assert rc == 0
+    assert calls["merge_attempts"] == 1  # merged directly
+    assert calls["updated"] is False  # NO update_pull → approvals NOT dismissed
+    assert calls["hold_label"] is None
+
+
+def test_process_once_behind_main_conflict_free_merges_directly(monkeypatch):
+    """§SOP-22(b) — explicit behind-main + conflict-free case: it still merges
+    directly (branch protection does not require strict up-to-date)."""
+    calls = {"merge_attempts": 0, "hold_label": None, "updated": False}
+    _fully_ready_process_once_monkeypatch(monkeypatch, mergeable=True, calls=calls)
+    behind_head = "a" * 40
+    monkeypatch.setattr(mq, "get_pull_commits", lambda n: [{"sha": behind_head}])
+
+    rc = mq.process_once(dry_run=False)
+
+    assert rc == 0
+    assert calls["merge_attempts"] == 1
+    assert calls["updated"] is False
+
+
+def test_process_once_pauses_when_main_not_green_no_direct_merge(monkeypatch):
+    """§SOP-22 backstop — the serialized safety that makes direct-merge safe:
+    when main's required push contexts are NOT green (e.g. a prior direct merge
+    introduced a semantic main-break caught by post-merge main CI), the queue
+    PAUSES — it does NOT merge the next PR onto an unverified/red main."""
+    calls = {"merge_attempts": 0, "hold_label": None, "updated": False}
+    _fully_ready_process_once_monkeypatch(monkeypatch, mergeable=True, calls=calls)
+    main_sha = "b" * 40
+
+    def red_main_combined(sha):
+        if sha == main_sha:
+            return {"state": "failure",
+                    "statuses": [{"context": "CI / all-required (push)", "status": "failure"}]}
+        return {"state": "success",
+                "statuses": [{"context": "CI / all-required (pull_request)", "status": "success"}]}
+    monkeypatch.setattr(mq, "get_combined_status", red_main_combined)
+
+    rc = mq.process_once(dry_run=False)
+
+    assert rc == 0
+    assert calls["merge_attempts"] == 0  # paused — no merge onto red main
+    assert calls["updated"] is False
+
+
+def test_direct_merge_bar_unchanged_behind_main(monkeypatch):
+    """§SOP-22(d) — the merge bar is UNCHANGED on the new direct-merge path. A
+    behind-main + conflict-free PR is still rejected (no merge) when ANY gate
+    fails: insufficient genuine approvals, red required context, open
+    REQUEST_CHANGES, or opt-out label. Direct-merge removes the update churn, it
+    does NOT weaken the bar — fail-closed on every gate."""
+    head_sha = "a" * 40
+    behind_main = dict(pr_has_current_base=False, mergeable=True)
+
+    # <2 genuine approvals → wait, not merge.
+    d = mq.evaluate_merge_readiness(
+        **_ready_kwargs(approvers={"agent-researcher"}, **behind_main)
+    )
+    assert d.action == "wait" and d.ready is False
+
+    # Red required context → wait, not merge.
+    red_required = {"state": "failure", "statuses": [
+        {"context": "CI / all-required (pull_request)", "status": "failure"}]}
+    d = mq.evaluate_merge_readiness(
+        **_ready_kwargs(pr_status=red_required, **behind_main)
+    )
+    assert d.action == "wait" and d.ready is False
+
+    # Open REQUEST_CHANGES on current head → wait, not merge.
+    d = mq.evaluate_merge_readiness(
+        **_ready_kwargs(request_changes=["agent-reviewer-cr2"], **behind_main)
+    )
+    assert d.action == "wait" and d.ready is False
+
+
 # --------------------------------------------------------------------------
 # Fix 3: status fetch is fail-closed (failed fetch != green)
 # --------------------------------------------------------------------------
@@ -532,6 +747,61 @@ def test_status_fetch_failure_is_fail_closed(monkeypatch):
    assert merged["called"] is False


+# --------------------------------------------------------------------------
+# Pagination: api_paginated loops pages and is fail-closed on page errors
+# --------------------------------------------------------------------------
+
+def test_api_paginated_loops_pages_until_partial(monkeypatch):
+    """api_paginated fetches all pages and stops when a page is < page_size."""
+    calls = []
+
+    def fake_api(method, path, *, query=None, **kw):
+        page = int((query or {}).get("page", "1"))
+        limit = int((query or {}).get("limit", "50"))
+        calls.append((page, limit))
+        if page == 1:
+            return 200, [{"number": 1}, {"number": 2}]
+        if page == 2:
+            return 200, [{"number": 3}]
+        return 200, []
+
+    monkeypatch.setattr(mq, "api", fake_api)
+    results = mq.api_paginated("GET", "/repos/o/r/issues", page_size=2)
+    assert len(results) == 3
+    assert results[0]["number"] == 1
+    assert results[1]["number"] == 2
+    assert results[2]["number"] == 3
+    assert calls == [(1, 2), (2, 2)]
+
+
+def test_api_paginated_raises_on_non_list(monkeypatch):
+    """A page that returns a dict instead of list is an error."""
+    def fake_api(method, path, *, query=None, **kw):
+        return 200, {"message": "not found"}
+
+    monkeypatch.setattr(mq, "api", fake_api)
+    with pytest.raises(mq.ApiError):
+        mq.api_paginated("GET", "/repos/o/r/issues")
+
+
+def test_get_combined_status_propagates_paginated_statuses_error(monkeypatch):
+    """If the paginated /statuses enrichment raises, the error propagates
+    (fail-closed — we do NOT silently fall back to an incomplete status set)."""
+    monkeypatch.setattr(mq, "OWNER", "o")
+    monkeypatch.setattr(mq, "NAME", "r")
+
+    def fake_api(method, path, *, query=None, **kw):
+        if path.endswith("/status"):
+            return 200, {"state": "success", "statuses": [{"context": "c1", "status": "success", "id": 1}]}
+        if path.endswith("/statuses"):
+            raise mq.ApiError("GET /statuses -> HTTP 502")
+        raise mq.ApiError(f"unexpected {path}")
+
+    monkeypatch.setattr(mq, "api", fake_api)
+    with pytest.raises(mq.ApiError, match="GET /statuses"):
+        mq.get_combined_status("a" * 40)
+
+
 def test_process_once_holds_tick_when_branch_protection_unavailable(monkeypatch):
    """BP enumeration failure → HOLD the whole tick (no merge, rc 0)."""
    merged = {"called": False}
@@ -621,13 +891,17 @@ def _stale_pr_update_409_monkeypatch(monkeypatch, queued_issues, calls):
    # Scan-loop process_once enumerates candidates via list_candidate_issues.
    monkeypatch.setattr(mq, "list_candidate_issues", lambda *, auto_discover: queued_issues)
    monkeypatch.setattr(mq, "get_pull", lambda n: {
-        "state": "open", "number": n, "mergeable": True,
+        "state": "open", "number": n, "mergeable": False,
        "base": {"ref": "main", "repo_id": 1},
        "head": {"sha": head_sha, "repo_id": 1},
        "labels": [{"name": "merge-queue"}],
    })
-    # NOTE: commits do NOT contain main_sha → pr_has_current_base is False →
-    # decision.action == "update".
+    # NOTE: mergeable is False (real conflict) AND commits do NOT contain
+    # main_sha → pr_has_current_base is False → decision.action == "update".
+    # Under the #2358 direct-merge fix the update path is reached ONLY when the
+    # PR is NOT mergeable; a mergeable=True behind-main PR would merge directly,
+    # so this fixture sets mergeable=False to exercise the #2352 409-on-update
+    # hold path.
    monkeypatch.setattr(mq, "get_pull_commits", lambda n: [{"sha": head_sha}])
    monkeypatch.setattr(mq, "get_pull_reviews", lambda n: [
        {"state": "APPROVED", "user": {"login": "agent-researcher"},
@@ -320,10 +320,10 @@ class TestVerifyFlip(unittest.TestCase):
        self.assertEqual(len(verdict["fail_runs"]), 1)
        self.assertEqual(verdict["fail_runs"][0]["status"], "failure")

-    def test_unreadable_log_warns_not_blocks(self):
-        # Acceptance test #5: log fetch 404 (None) → warn, not block.
-        # Status is `success`, log is None — we can't tell, so we warn
-        # and allow.
+    def test_unreadable_log_on_success_blocks(self):
+        # Fail-closed: log fetch 404 (None) on a success status is a
+        # potential Quirk #10 mask — we cannot verify it's genuine, so
+        # we block the flip rather than allowing it.
        with mock.patch.object(lpfc, "recent_commits_on_branch", return_value=["sha1"]):
            with mock.patch.object(
                lpfc, "combined_status",
@@ -332,7 +332,8 @@ class TestVerifyFlip(unittest.TestCase):
                with mock.patch.object(lpfc, "fetch_log", return_value=None):
                    verdict = lpfc.verify_flip(FLIP_FIXTURE, "main", 5)
        self.assertEqual(verdict["fail_runs"], [])
-        self.assertEqual(verdict["masked_runs"], [])
+        self.assertEqual(len(verdict["masked_runs"]), 1)
+        self.assertIn("log unavailable", verdict["masked_runs"][0]["samples"][0])
        self.assertTrue(any("log unavailable" in w for w in verdict["warnings"]))

    def test_unreadable_log_with_failure_status_still_blocks(self):
@@ -349,9 +350,9 @@ class TestVerifyFlip(unittest.TestCase):
        self.assertEqual(len(verdict["fail_runs"]), 1)
        self.assertIn("log unavailable", verdict["fail_runs"][0]["samples"][0])

-    def test_zero_runs_history_warns_allows(self):
-        # No commits with a matching context — newly added workflow.
-        # Allow with warning.
+    def test_zero_runs_history_blocks(self):
+        # No commits with a matching context — cannot verify the flip.
+        # Fail-closed: treat as masked rather than allowing.
        with mock.patch.object(lpfc, "recent_commits_on_branch", return_value=["sha1", "sha2"]):
            with mock.patch.object(
                lpfc, "combined_status",
@@ -360,17 +361,32 @@ class TestVerifyFlip(unittest.TestCase):
                verdict = lpfc.verify_flip(FLIP_FIXTURE, "main", 5)
        self.assertEqual(verdict["checked_commits"], 0)
        self.assertEqual(verdict["fail_runs"], [])
-        self.assertEqual(verdict["masked_runs"], [])
-        self.assertTrue(any("no runs of" in w for w in verdict["warnings"]))
+        self.assertEqual(len(verdict["masked_runs"]), 1)
+        self.assertIn("cannot verify flip", verdict["masked_runs"][0]["samples"][0])

-    def test_zero_commits_warns_allows(self):
-        # Empty branch (newly created repo, e.g.). Allow with warning.
+    def test_zero_commits_blocks(self):
+        # Empty branch (newly created repo, e.g.). Fail-closed: block.
        with mock.patch.object(lpfc, "recent_commits_on_branch", return_value=[]):
            verdict = lpfc.verify_flip(FLIP_FIXTURE, "main", 5)
        self.assertEqual(verdict["checked_commits"], 0)
        self.assertEqual(verdict["fail_runs"], [])
-        self.assertEqual(verdict["masked_runs"], [])
-        self.assertTrue(any("no recent commits" in w for w in verdict["warnings"]))
+        self.assertEqual(len(verdict["masked_runs"]), 1)
+        self.assertIn("cannot verify flip", verdict["masked_runs"][0]["samples"][0])
+
+    def test_combined_status_api_error_blocks(self):
+        # Fail-closed: combined_status ApiError means the check history is
+        # unreadable — we cannot verify the flip, so block as masked.
+        with mock.patch.object(lpfc, "recent_commits_on_branch", return_value=["sha1"]):
+            with mock.patch.object(
+                lpfc, "combined_status",
+                side_effect=lpfc.ApiError("GET /statuses/sha → HTTP 500"),
+            ):
+                verdict = lpfc.verify_flip(FLIP_FIXTURE, "main", 5)
+        self.assertEqual(verdict["checked_commits"], 0)
+        self.assertEqual(verdict["fail_runs"], [])
+        # One masked_run from the ApiError, one from zero checked_commits.
+        self.assertEqual(len(verdict["masked_runs"]), 2)
+        self.assertIn("API error", verdict["masked_runs"][0]["samples"][0])


 # --------------------------------------------------------------------------
@@ -57,12 +57,12 @@ echo "test: tier:low OR-clause splits to 3 tokens"
 assert_eq "tier:low" "engineers|managers|ceo" "$(split_clause "engineers,managers,ceo")"

 echo "test: tier:medium AND-expression — bash word-split on \$EXPR yields 5 tokens"
-EXPR="managers AND engineers AND qa???,security???"
+EXPR="managers AND engineers AND qa,security"
 out=""
 for _raw in $EXPR; do
  out="${out}${out:+ ; }$(split_clause "$_raw")"
 done
-assert_eq "tier:medium" "managers ; AND ; engineers ; AND ; qa???|security???" "$out"
+assert_eq "tier:medium" "managers ; AND ; engineers ; AND ; qa|security" "$out"

 echo "test: tier:high single-team OR-clause"
 assert_eq "tier:high" "ceo" "$(split_clause "ceo")"
@@ -0,0 +1,474 @@
+import importlib.util
+import json
+import pathlib
+import urllib.error
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "umbrella-reaper.py"
+
+
+def load_reaper():
+    spec = importlib.util.spec_from_file_location("umbrella_reaper", SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(mod)
+    mod.API = "https://git.example.test/api/v1"
+    mod.GITEA_TOKEN = "fixture-token"
+    mod.GITEA_HOST = "git.example.test"
+    mod.REPO = "owner/repo"
+    return mod
+
+
+class FakeResponse:
+    status = 200
+
+    def __init__(self, payload):
+        self.payload = payload
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self):
+        return json.dumps(self.payload).encode("utf-8")
+
+
+def _pr_fixture(number: int, sha: str) -> dict:
+    return {"number": number, "head": {"sha": sha}}
+
+
+def _status_entry(context: str, state: str) -> dict:
+    return {"context": context, "status": state}
+
+
+def test_process_pr_compensates_when_all_sub_jobs_success(monkeypatch):
+    mod = load_reaper()
+    posted = []
+
+    def fake_post_status(sha, context, description):
+        posted.append((sha, context, description))
+
+    monkeypatch.setattr(mod, "post_status", fake_post_status)
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+
+    pr = _pr_fixture(1, "abc123")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    ok = mod.process_pr(pr)
+    assert ok is True
+    assert len(posted) == 1
+    assert posted[0][0] == "abc123"
+    assert posted[0][1] == "CI / all-required (pull_request)"
+    assert "Compensating status" in posted[0][2]
+
+
+def test_process_pr_skips_when_umbrella_missing(monkeypatch):
+    mod = load_reaper()
+    posted = []
+    monkeypatch.setattr(mod, "post_status", lambda *a, **k: posted.append(a))
+    monkeypatch.setattr(mod, "REQUIRED_SUB_JOBS", ["CI / Platform (Go) (pull_request)"])
+
+    pr = _pr_fixture(2, "def456")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / Platform (Go) (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    ok = mod.process_pr(pr)
+    assert ok is True
+    assert posted == []
+
+
+def test_process_pr_skips_when_sub_job_pending(monkeypatch):
+    mod = load_reaper()
+    posted = []
+    monkeypatch.setattr(mod, "post_status", lambda *a, **k: posted.append(a))
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+
+    pr = _pr_fixture(3, "ghi789")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "pending"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    ok = mod.process_pr(pr)
+    assert ok is True
+    assert posted == []
+
+
+def test_process_pr_skips_when_sub_job_failure(monkeypatch):
+    mod = load_reaper()
+    posted = []
+    monkeypatch.setattr(mod, "post_status", lambda *a, **k: posted.append(a))
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+
+    pr = _pr_fixture(4, "jkl012")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "failure"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    ok = mod.process_pr(pr)
+    assert ok is True
+    assert posted == []
+
+
+def test_process_pr_returns_false_on_post_failure(monkeypatch):
+    mod = load_reaper()
+
+    def fake_post_status(sha, context, description):
+        raise mod.ApiError("POST /statuses/abc123 -> HTTP 500: simulated failure")
+
+    monkeypatch.setattr(mod, "post_status", fake_post_status)
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+
+    pr = _pr_fixture(5, "abc123")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    ok = mod.process_pr(pr)
+    assert ok is False
+
+
+def test_main_exits_nonzero_when_any_post_fails(monkeypatch):
+    mod = load_reaper()
+
+    monkeypatch.setenv("GITEA_TOKEN", "fixture-token")
+    monkeypatch.setenv("GITEA_HOST", "git.example.test")
+    monkeypatch.setenv("REPO", "owner/repo")
+
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+    monkeypatch.setattr(
+        mod,
+        "list_open_prs",
+        lambda limit: [
+            _pr_fixture(1, "abc123"),
+            _pr_fixture(2, "def456"),
+        ],
+    )
+
+    calls = {"n": 0}
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    def fake_post_status(sha, context, description):
+        calls["n"] += 1
+        if calls["n"] == 2:
+            raise mod.ApiError("simulated failure")
+
+    monkeypatch.setattr(mod, "post_status", fake_post_status)
+
+    exit_code = mod.main()
+    assert exit_code == 1
+
+
+def test_main_exits_zero_when_all_posts_succeed(monkeypatch):
+    mod = load_reaper()
+
+    monkeypatch.setenv("GITEA_TOKEN", "fixture-token")
+    monkeypatch.setenv("GITEA_HOST", "git.example.test")
+    monkeypatch.setenv("REPO", "owner/repo")
+
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+    monkeypatch.setattr(
+        mod,
+        "list_open_prs",
+        lambda limit: [_pr_fixture(1, "abc123")],
+    )
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+    monkeypatch.setattr(mod, "post_status", lambda *a, **k: None)
+
+    exit_code = mod.main()
+    assert exit_code == 0
+
+
+def test_dry_run_does_not_post(monkeypatch):
+    mod = load_reaper()
+    api_calls = []
+
+    def fake_api(method, path, *, body=None, query=None, expect_json=True):
+        api_calls.append((method, path, body))
+        return 200, {"ok": True}
+
+    monkeypatch.setattr(mod, "api", fake_api)
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+
+    pr = _pr_fixture(6, "mno345")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+                _status_entry("CI / Platform (Go) (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+    monkeypatch.setattr(mod, "DRY_RUN", True)
+
+    ok = mod.process_pr(pr)
+    assert ok is True
+    # DRY_RUN should prevent the POST /statuses call
+    assert not any(
+        method == "POST" and "/statuses/" in path for method, path, _ in api_calls
+    )
+
+
+def test_duplicate_contexts_use_latest_state(monkeypatch):
+    mod = load_reaper()
+    posted = []
+    monkeypatch.setattr(mod, "post_status", lambda *a, **k: posted.append(a))
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+        ],
+    )
+
+    pr = _pr_fixture(7, "pqr678")
+
+    def fake_combined_status(sha):
+        return {
+            "statuses": [
+                _status_entry("CI / all-required (pull_request)", "failure"),
+                # duplicate: first pending, then success — the loop overwrites
+                _status_entry("CI / Detect changes (pull_request)", "pending"),
+                _status_entry("CI / Detect changes (pull_request)", "success"),
+            ]
+        }
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+
+    ok = mod.process_pr(pr)
+    assert ok is True
+    assert len(posted) == 1
+
+
+def test_load_required_sub_jobs_from_ci_yml_pull_request_event():
+    mod = load_reaper()
+    # UMBRELLA_CONTEXT defaults to pull_request, so derivation should yield
+    # the pull_request suffix.
+    jobs = mod._load_required_sub_jobs_from_ci_yml(".gitea/workflows")
+    assert all(j.endswith(" (pull_request)") for j in jobs)
+    assert "CI / Detect changes (pull_request)" in jobs
+    assert "CI / Python Lint & Test (pull_request)" in jobs
+
+
+def test_load_required_sub_jobs_from_ci_yml_push_event(monkeypatch):
+    mod = load_reaper()
+    monkeypatch.setattr(mod, "UMBRELLA_CONTEXT", "CI / all-required (push)")
+    jobs = mod._load_required_sub_jobs_from_ci_yml(".gitea/workflows")
+    assert all(j.endswith(" (push)") for j in jobs)
+    assert "CI / Detect changes (push)" in jobs
+
+
+def test_list_open_prs_paginates(monkeypatch):
+    mod = load_reaper()
+    calls = []
+
+    def fake_api(method, path, *, body=None, query=None, expect_json=True):
+        calls.append(query)
+        page = int(query.get("page", 1))
+        limit = int(query.get("limit", 50))
+        if page == 1:
+            return 200, [{"number": 1}, {"number": 2}]
+        if page == 2:
+            return 200, [{"number": 3}]
+        return 200, []
+
+    monkeypatch.setattr(mod, "api", fake_api)
+    prs = mod.list_open_prs(limit=2)
+    assert len(prs) == 3
+    assert prs[0]["number"] == 1
+    assert prs[2]["number"] == 3
+    assert calls[0]["page"] == "1"
+    assert calls[1]["page"] == "2"
+
+
+def test_process_pr_returns_false_on_status_fetch_failure(monkeypatch):
+    mod = load_reaper()
+
+    def fake_get_combined_status(sha):
+        raise mod.ApiError("GET /statuses/abc123 -> HTTP 500: simulated outage")
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_get_combined_status)
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        ["CI / Detect changes (pull_request)"],
+    )
+
+    pr = _pr_fixture(8, "abc123")
+    ok = mod.process_pr(pr)
+    assert ok is False
+
+
+def test_process_pr_returns_false_on_missing_statuses_array(monkeypatch):
+    mod = load_reaper()
+
+    def fake_get_combined_status(sha):
+        return {"state": "success"}  # missing 'statuses' array
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_get_combined_status)
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        ["CI / Detect changes (pull_request)"],
+    )
+
+    pr = _pr_fixture(9, "def456")
+    ok = mod.process_pr(pr)
+    assert ok is False
+
+
+def test_main_exits_nonzero_when_any_status_read_fails(monkeypatch):
+    mod = load_reaper()
+
+    monkeypatch.setenv("GITEA_TOKEN", "fixture-token")
+    monkeypatch.setenv("GITEA_HOST", "git.example.test")
+    monkeypatch.setenv("REPO", "owner/repo")
+
+    monkeypatch.setattr(
+        mod,
+        "REQUIRED_SUB_JOBS",
+        [
+            "CI / Detect changes (pull_request)",
+            "CI / Platform (Go) (pull_request)",
+        ],
+    )
+    monkeypatch.setattr(
+        mod,
+        "list_open_prs",
+        lambda limit: [
+            _pr_fixture(1, "abc123"),
+            _pr_fixture(2, "def456"),
+        ],
+    )
+
+    def fake_combined_status(sha):
+        if sha == "abc123":
+            return {
+                "statuses": [
+                    _status_entry("CI / all-required (pull_request)", "failure"),
+                    _status_entry("CI / Detect changes (pull_request)", "success"),
+                    _status_entry("CI / Platform (Go) (pull_request)", "success"),
+                ]
+            }
+        raise mod.ApiError("simulated status fetch failure")
+
+    monkeypatch.setattr(mod, "get_combined_status", fake_combined_status)
+    monkeypatch.setattr(mod, "post_status", lambda *a, **k: None)
+
+    exit_code = mod.main()
+    assert exit_code == 1
@@ -0,0 +1,360 @@
+#!/usr/bin/env python3
+"""umbrella-reaper — auto-recovery for stale CI umbrella statuses on PRs.
+
+Tracking: molecule-core#1780.
+
+Sibling to status-reaper.py (default-branch push-suffix compensation),
+but scoped to pull_request umbrellas instead of main-branch contexts.
+
+What this script does, per `.gitea/workflows/umbrella-reaper.yml` invocation:
+
+  1. List open PRs via GET /repos/{o}/{r}/pulls?state=open&limit={N}.
+  2. For EACH PR:
+     - GET combined commit status for PR head SHA.
+     - Look for the umbrella context (default: "CI / all-required (pull_request)").
+     - If umbrella state is "failure":
+         - Verify ALL required sub-job contexts are "success".
+         - If yes → POST compensating success to /statuses/{sha} with the
+           same umbrella context and an honest description.
+         - If any required sub-job is NOT success → skip (umbrella correctly
+           reflects reality; do NOT lie).
+     - If umbrella state is "success" or "pending" → skip.
+  3. Exit 0. Re-running is idempotent — Gitea de-dups by context.
+
+What it does NOT do:
+  - Touch non-umbrella contexts.
+  - Compensate when ANY required sub-job is missing, pending, failure, or
+    cancelled. Only the "all sub-jobs green, umbrella stale" race.
+  - Merge PRs. It only posts a status; branch protection still requires
+    human approval.
+  - Run on closed PRs.
+
+Halt conditions:
+  - Missing required env vars → exit 1 with ::error:: message.
+  - API 5xx on PR list → fail-loud (can't assess state).
+  - API 5xx on an individual PR's status → ::warning:: + continue to next PR.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+def _load_required_sub_jobs_from_ci_yml(workflows_dir: str) -> list[str]:
+    """Parse ci.yml and extract the all-required sentinel's sub-job contexts.
+
+    Supports two shapes of the all-required job run block:
+      1. Legacy Python f-string list (pre-2026-06-01):
+         f"CI / Detect changes ({event})"
+      2. Current shell-script shape (post-2026-06-01 scheduler fix):
+         check "Detect changes"        "$CHANGES_RESULT"
+
+    Raises RuntimeError if ci.yml is missing, has no all-required job, or the
+    run block cannot be parsed.
+    """
+    ci_path = Path(workflows_dir) / "ci.yml"
+    if not ci_path.exists():
+        raise RuntimeError(f"ci.yml not found at {ci_path}")
+
+    # PyYAML is installed by the workflow (same as status-reaper.py).
+    import yaml
+
+    with ci_path.open() as f:
+        doc = yaml.safe_load(f)
+
+    jobs = doc.get("jobs", {})
+    all_required = jobs.get("all-required")
+    if not isinstance(all_required, dict):
+        raise RuntimeError("ci.yml missing 'all-required' job")
+
+    steps = all_required.get("steps", [])
+    run_block = ""
+    for step in steps:
+        if isinstance(step, dict):
+            run_text = step.get("run", "")
+            if run_text:
+                run_block = run_text
+                break
+
+    if not run_block:
+        raise RuntimeError("all-required job missing run block")
+
+    # Determine event suffix from the umbrella context we are watching.
+    if UMBRELLA_CONTEXT.endswith(" (pull_request)"):
+        suffix = "(pull_request)"
+    elif UMBRELLA_CONTEXT.endswith(" (push)"):
+        suffix = "(push)"
+    else:
+        m = re.search(r' \(([^)]+)\)$', UMBRELLA_CONTEXT)
+        suffix = m.group(1) if m else "pull_request"
+
+    # Try legacy f-string format first.
+    if "({event})" in run_block:
+        matches = re.findall(r'f["\'](.*?\(\{event\}\))["\']', run_block)
+        if matches:
+            return [m.replace("({event})", suffix) for m in matches]
+
+    # Try current shell-script format: check "Name" "$RESULT"
+    matches = re.findall(r'check\s+"([^"]+)"', run_block)
+    if matches:
+        return [f"CI / {name} {suffix}" for name in matches]
+
+    raise RuntimeError("unable to derive required sub-jobs from all-required run block")
+
+
+# --------------------------------------------------------------------------
+# Environment
+# --------------------------------------------------------------------------
+def _env(key: str, *, default: str = "") -> str:
+    return os.environ.get(key, default)
+
+
+GITEA_TOKEN = _env("GITEA_TOKEN")
+GITEA_HOST = _env("GITEA_HOST")
+REPO = _env("REPO")
+DRY_RUN = _env("DRY_RUN", default="").lower() in ("1", "true", "yes")
+
+# The umbrella context to watch. Must match the branch-protection name
+# exactly (Gitea de-dups by context string).
+UMBRELLA_CONTEXT = _env("UMBRELLA_CONTEXT", default="CI / all-required (pull_request)")
+
+# Required sub-job contexts. The umbrella is only compensated when ALL of
+# these are "success" on the same SHA. Order does not matter.
+#
+# Derive from ci.yml at runtime to prevent drift (CR2 blocker #1).
+# The env var REQUIRED_SUB_JOBS overrides derivation for emergency
+# tuning or local testing.
+_REQUIRED_SUB_JOBS_OVERRIDE = _env("REQUIRED_SUB_JOBS")
+if _REQUIRED_SUB_JOBS_OVERRIDE:
+    REQUIRED_SUB_JOBS = [
+        ctx.strip()
+        for ctx in _REQUIRED_SUB_JOBS_OVERRIDE.split(";")
+        if ctx.strip()
+    ]
+else:
+    try:
+        REQUIRED_SUB_JOBS = _load_required_sub_jobs_from_ci_yml(".gitea/workflows")
+    except Exception as exc:
+        sys.stderr.write(
+            f"::error::Failed to derive REQUIRED_SUB_JOBS from ci.yml: {exc}\n"
+        )
+        sys.exit(1)
+
+OWNER, NAME = (REPO.split("/", 1) + [""])[:2] if REPO else ("", "")
+API = f"https://{GITEA_HOST}/api/v1" if GITEA_HOST else ""
+PR_LIMIT = int(_env("PR_LIMIT", default="50"))
+
+
+def _require_runtime_env() -> None:
+    for key in ("GITEA_TOKEN", "GITEA_HOST", "REPO"):
+        if not os.environ.get(key):
+            sys.stderr.write(f"::error::missing required env var: {key}\n")
+            sys.exit(1)
+
+
+# --------------------------------------------------------------------------
+# Tiny HTTP helper
+# --------------------------------------------------------------------------
+class ApiError(RuntimeError):
+    pass
+
+
+def api(
+    method: str,
+    path: str,
+    *,
+    body: dict | None = None,
+    query: dict[str, str] | None = None,
+    expect_json: bool = True,
+) -> tuple[int, Any]:
+    url = f"{API}{path}"
+    if query:
+        url = f"{url}?{urllib.parse.urlencode(query)}"
+    data = None
+    headers = {
+        "Authorization": f"token {GITEA_TOKEN}",
+        "Accept": "application/json",
+    }
+    if body is not None:
+        data = json.dumps(body).encode("utf-8")
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, method=method, data=data, headers=headers)
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            raw = resp.read()
+            status = resp.status
+    except urllib.error.HTTPError as e:
+        raw = e.read()
+        status = e.code
+
+    if not (200 <= status < 300):
+        snippet = raw[:500].decode("utf-8", errors="replace") if raw else ""
+        raise ApiError(f"{method} {path} -> HTTP {status}: {snippet}")
+
+    if not raw:
+        return status, None
+    try:
+        return status, json.loads(raw)
+    except json.JSONDecodeError as e:
+        if expect_json:
+            raise ApiError(
+                f"{method} {path} -> HTTP {status} but body is not JSON: {e}"
+            ) from e
+        return status, {"_raw": raw.decode("utf-8", errors="replace")}
+
+
+# --------------------------------------------------------------------------
+# Gitea reads / writes
+# --------------------------------------------------------------------------
+def list_open_prs(limit: int = 50) -> list[dict]:
+    """Paginate through all open PR pages. Fail closed on non-list responses."""
+    all_prs: list[dict] = []
+    page = 1
+    while True:
+        _, body = api(
+            "GET",
+            f"/repos/{OWNER}/{NAME}/pulls",
+            query={"state": "open", "limit": str(limit), "page": str(page)},
+        )
+        if not isinstance(body, list):
+            raise ApiError(f"PR list page {page} response is not a JSON array")
+        if not body:
+            break
+        all_prs.extend(body)
+        if len(body) < limit:
+            break
+        page += 1
+    return all_prs
+
+
+def get_combined_status(sha: str) -> dict:
+    _, body = api("GET", f"/repos/{OWNER}/{NAME}/commits/{sha}/status")
+    if not isinstance(body, dict):
+        raise ApiError(f"status for {sha} response is not a JSON object")
+    return body
+
+
+def post_status(sha: str, context: str, description: str) -> None:
+    payload = {
+        "context": context,
+        "state": "success",
+        "description": description,
+    }
+    if DRY_RUN:
+        print(f"[DRY-RUN] Would POST /statuses/{sha}: {json.dumps(payload)}")
+        return
+    api("POST", f"/repos/{OWNER}/{NAME}/statuses/{sha}", body=payload)
+
+
+# --------------------------------------------------------------------------
+# Core logic
+# --------------------------------------------------------------------------
+def _entry_state(s: dict) -> str:
+    return s.get("status") or s.get("state") or ""
+
+
+def process_pr(pr: dict) -> bool:
+    """Process a single PR. Returns True if the tick succeeded for this PR
+    (including no-op skips), False if a compensating POST failed.
+    """
+    num = pr.get("number")
+    sha = pr.get("head", {}).get("sha")
+    if not sha:
+        print(f"::warning::PR #{num}: missing head.sha; skipping")
+        return True
+
+    try:
+        status = get_combined_status(sha)
+    except ApiError as e:
+        print(f"::error::PR #{num}: status fetch failed: {e}")
+        return False
+
+    statuses = status.get("statuses")
+    if not isinstance(statuses, list):
+        print(f"::error::PR #{num}: combined status missing 'statuses' array")
+        return False
+    umbrella_entry = None
+    subjob_states: dict[str, str] = {}
+
+    for s in statuses:
+        if not isinstance(s, dict):
+            continue
+        ctx = s.get("context", "")
+        state = _entry_state(s)
+        if ctx == UMBRELLA_CONTEXT:
+            umbrella_entry = s
+        if ctx in REQUIRED_SUB_JOBS:
+            subjob_states[ctx] = state
+
+    if umbrella_entry is None:
+        print(f"::notice::PR #{num}: no umbrella context '{UMBRELLA_CONTEXT}'; skipping")
+        return True
+
+    umbrella_state = _entry_state(umbrella_entry)
+    if umbrella_state != "failure":
+        print(f"::notice::PR #{num}: umbrella is '{umbrella_state}'; skipping")
+        return True
+
+    # Verify ALL required sub-jobs are present and success
+    missing = [ctx for ctx in REQUIRED_SUB_JOBS if ctx not in subjob_states]
+    if missing:
+        print(
+            f"::notice::PR #{num}: umbrella=failure, but missing sub-jobs: {missing}; "
+            "skipping (sub-jobs may still be running)"
+        )
+        return True
+
+    not_success = [ctx for ctx in REQUIRED_SUB_JOBS if subjob_states[ctx] != "success"]
+    if not_success:
+        print(
+            f"::notice::PR #{num}: umbrella=failure, but sub-jobs not all success: "
+            f"{[(ctx, subjob_states[ctx]) for ctx in not_success]}; skipping"
+        )
+        return True
+
+    # All checks pass — post compensating status
+    desc = (
+        "Compensating status: all required sub-jobs verified success; "
+        "umbrella stale due to commit-status propagation race. "
+        f"Auto-posted by umbrella-reaper for PR #{num}."
+    )
+    try:
+        post_status(sha, UMBRELLA_CONTEXT, desc)
+        print(f"::notice::PR #{num}: posted compensating success for {UMBRELLA_CONTEXT}")
+        return True
+    except ApiError as e:
+        print(f"::error::PR #{num}: failed to post compensating status: {e}")
+        return False
+
+
+def main() -> int:
+    _require_runtime_env()
+
+    # Drift guard: ci.yml derivation already happened at module load, but
+    # we sanity-check it is non-empty so the loop below doesn't trivially
+    # no-op because of a parse bug.
+    if not REQUIRED_SUB_JOBS:
+        sys.stderr.write("::error::REQUIRED_SUB_JOBS is empty; bailing out\n")
+        return 1
+
+    prs = list_open_prs(limit=PR_LIMIT)
+    print(f"::notice::Scanning {len(prs)} open PRs for stale umbrella statuses")
+    compensated = 0
+    failed = 0
+    for pr in prs:
+        ok = process_pr(pr)
+        if not ok:
+            failed += 1
+    print(f"::notice::umbrella-reaper complete (failed POSTs={failed})")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -34,6 +34,8 @@ jobs:
  check:
    name: Block forbidden paths
    runs-on: ubuntu-latest
+    # Hard gate — detected internal-path leaks fail the workflow.
+    # continue-on-error removed per directive (fail-open → fail-closed).
    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
        with:
@@ -0,0 +1,165 @@
+name: boot-to-registration-e2e (advisory)
+
+# cp#455 — Minimal-cell boot-to-registration e2e.
+# CTO directive 14eb4f07: "build the minimal claude-code+kimi cell,
+# it should now go GREEN since the fix is live."
+#
+# Stage 1 of 5-stage rollout. Reuses the dispatch-only EC2
+# provisioning path from test_staging_full_saas.sh but reduced to
+# the minimum boot-to-registration surface:
+#
+#   1. Provision request accepted; workspace transitions to booting/running
+#   2. Controlplane receives /registry/register for that workspace_id
+#   3. JSON-RPC/completion route returns successful minimal response
+#   4. Teardown terminates workspace even on failure (trap)
+#
+# Advisory (non-blocking) per Researcher Stage 2 design — RED on
+# current main is expected pre-cp#469-cluster. After cp#477 deploy
+# (888efceb) + PR #2167 merge, cell should turn GREEN. THAT green
+# is the cluster-proof signal.
+#
+# Cost controls (mandatory):
+#   - SPOT instances (tagged run_id/workspace_id for cost attribution)
+#   - Fast teardown (~3-5 min wall-clock) even on assertion failure
+#   - Structured per-cell results JSON (runtime/provider/model/
+#     billing_mode/workspace_id/register_status/completion_status/
+#     teardown_status/elapsed_seconds)
+#
+# Inputs:
+#   runtime        : default claude-code
+#   billing_mode   : default platform_managed (the cp#469-cluster path)
+#   provider       : default platform (vs direct-to-provider)
+#   model          : default moonshot/kimi-k2.6 (CTO-specified)
+#
+# PR target: molecule-core (this file). Companion harness extension
+# (test_minimal_boot_cell.sh) lives in tests/e2e/ alongside
+# test_staging_full_saas.sh — same repo, same branch.
+#
+# Note: cp#455 was originally spec'd to live in molecule-controlplane
+# (`.gitea/workflows/` path), but molecule-core's CI is the home for
+# tenant-boot e2e tests in this stage. Stage 2 may move the path.
+
+on:
+  workflow_dispatch:
+  # Note: Gitea 1.22.6 does not support workflow_dispatch.inputs
+  # (feedback_gitea_workflow_dispatch_inputs_unsupported). Defaults
+  # are hardcoded in the job env below. Stage 2 can add matrix/
+  # param support once the Gitea version supports it.
+
+# Advisory: no cron schedule, manual dispatch only. Branch protection
+# doesn't require this — RED on main is expected pre-cp#469-cluster
+# deploy, GREEN signals the cluster is live.
+permissions:
+  contents: read
+  # No issue-write; failures surface as red runs in workflow history.
+
+concurrency:
+  group: boot-to-registration-e2e
+  cancel-in-progress: false
+
+jobs:
+  # bp-exempt: advisory e2e — non-gating, manual dispatch only (cp#455 Stage 1)
+  minimal-cell:
+    name: Minimal cell (claude-code + platform + moonshot/kimi-k2.6)
+    runs-on: ubuntu-latest
+    # Bounded at 12 min. Wall-clock budget breakdown:
+    #   - cold EC2 provision: ~3-4 min (SPOT)
+    #   - /registry/register wait: ~30s
+    #   - completion call: ~10s
+    #   - teardown: ~30-60s
+    #   - tail headroom: ~6-7 min
+    timeout-minutes: 12
+    env:
+      # Hardcoded defaults — Gitea 1.22.6 does not support workflow_dispatch.inputs
+      # (feedback_gitea_workflow_dispatch_inputs_unsupported). Stage 2 can add
+      # matrix/param support once the Gitea version supports it.
+      E2E_RUNTIME: claude-code
+      E2E_BILLING_MODE: platform_managed
+      E2E_PROVIDER: platform
+      E2E_MODEL: moonshot/kimi-k2.6
+      E2E_RUN_ID: cp455-${{ github.run_id }}
+      E2E_PROVISION_TIMEOUT_SECS: '300' # 5 min — fast teardown budget
+      MOLECULE_CP_URL: ${{ vars.STAGING_CP_URL || 'https://staging-api.moleculesai.app' }}
+      MOLECULE_ADMIN_TOKEN: ${{ secrets.CP_STAGING_ADMIN_API_TOKEN }}
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Verify required secrets present
+        run: |
+          if [ -z "${MOLECULE_ADMIN_TOKEN:-}" ]; then
+            echo "::error::CP_STAGING_ADMIN_API_TOKEN secret missing — minimal-cell e2e cannot run"
+            echo "::error::Set it at Settings → Secrets and Variables → Actions; pull from staging-CP's CP_ADMIN_API_TOKEN env in Railway."
+            exit 1
+          fi
+
+      - name: Install required tools
+        run: |
+          for cmd in jq curl python3; do
+            command -v "$cmd" >/dev/null 2>&1 || {
+              echo "::error::required tool '$cmd' not on PATH — runner image regression?"
+              exit 1
+            }
+          done
+
+      - name: Run minimal-cell boot-to-registration harness
+        # The harness script handles its own teardown via EXIT trap;
+        # even on assertion failure (provision timeout, register
+        # timeout, completion failure), the workspace is deprovisioned
+        # and a leak is reported. Exit code propagates from the script.
+        # Structured per-cell results are emitted to ${GITHUB_STEP_SUMMARY}
+        # so operators see pass/fail per assertion without scrolling.
+        run: |
+          bash tests/e2e/test_minimal_boot_cell.sh
+
+      - name: Emit structured per-cell results
+        if: always()
+        # Always run (even on failure) so the structured results are
+        # visible in the workflow summary. The script writes a JSON
+        # file at /tmp/cell-result.json; this step renders it as a
+        # job summary.
+        run: |
+          if [ -f /tmp/cell-result.json ]; then
+            echo "## Minimal-cell results" >> "$GITHUB_STEP_SUMMARY"
+            echo "" >> "$GITHUB_STEP_SUMMARY"
+            echo '```json' >> "$GITHUB_STEP_SUMMARY"
+            cat /tmp/cell-result.json >> "$GITHUB_STEP_SUMMARY"
+            echo "" >> "$GITHUB_STEP_SUMMARY"
+            echo '```' >> "$GITHUB_STEP_SUMMARY"
+          else
+            echo "## Minimal-cell results: NO_RESULT_FILE" >> "$GITHUB_STEP_SUMMARY"
+            echo "" >> "$GITHUB_STEP_SUMMARY"
+            echo "Harness did not produce /tmp/cell-result.json — likely crashed before trap fired." >> "$GITHUB_STEP_SUMMARY"
+          fi
+
+      - name: Failure summary
+        if: failure()
+        run: |
+          {
+            echo "## cp#455 minimal-cell FAILED"
+            echo ""
+            echo "**Run ID:** ${{ github.run_id }}"
+            echo "**Runtime:** ${E2E_RUNTIME}"
+            echo "**Billing mode:** ${E2E_BILLING_MODE}"
+            echo "**Provider:** ${E2E_PROVIDER}"
+            echo "**Model:** ${E2E_MODEL}"
+            echo "**Slug:** ${E2E_RUN_ID}"
+            echo ""
+            echo "### What this means"
+            echo ""
+            echo "The minimal claude-code+kimi cell did not pass all 4 assertions:"
+            echo "1. Provision request accepted; workspace transitions to booting/running"
+            echo "2. Controlplane receives /registry/register for that workspace_id"
+            echo "3. JSON-RPC/completion route returns successful minimal response"
+            echo "4. Teardown terminates workspace even on failure (trap)"
+            echo ""
+            echo "RED is expected pre-cp#469-cluster. After cp#477 deploy (888efceb) + PR #2167 merge,"
+            echo "this should turn GREEN. Persistent RED after both merge = cluster bug, not e2e bug."
+            echo ""
+            echo "### Next steps"
+            echo ""
+            echo "1. Check the harness output above for the assertion that failed"
+            echo "2. If assertion 1 fails: provision path broken — check CP admin API + EC2 quota"
+            echo "3. If assertion 2 fails: /registry/register path broken — check workspace-server boot"
+            echo "4. If assertion 3 fails: LLM proxy / completion path broken — check cp#469 cluster"
+            echo "5. If assertion 4 fails: teardown trap broken — leak risk, fix immediately"
+          } >> "$GITHUB_STEP_SUMMARY"
@@ -418,10 +418,9 @@ jobs:
    # a manual action that determinism made obsolete.
    name: Canvas Deploy Status
    runs-on: docker-host
-    # Job-level `if:` so ci-required-drift.py's ci_job_names() detects this as
-    # github.ref-gated and skips it from the required-context F1 set (mc#1982).
+    # Per-step no-op (not job-level `if:`) so the job reaches SUCCESS on PRs
+    # instead of skipped — skipped poisons the PR combined status (internal#817).
    # Step-level exit 0 handles the "not a canvas main push" case.
-    if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/staging' }}
    needs: [changes, canvas-build]
    steps:
      - name: Record canvas ordered-deploy status
@@ -533,9 +532,8 @@ jobs:
    # The `needs:` list MUST stay in lockstep with ci-required-drift.py's
    # F1 check (`ci_job_names()` = every job MINUS the sentinel MINUS jobs
    # whose `if:` gates on github.event_name/github.ref). canvas-deploy-
-    # reminder is event-gated (`if: github.ref == refs/heads/{main,staging}`)
-    # so it is intentionally EXCLUDED — it skips on PRs and a `needs:` on a
-    # skipped job would never let the sentinel run. If a new always-running
+    # status is per-step-gated (not job-level `if:`) so it reaches SUCCESS
+    # on PRs and is included here — internal#817. If a new always-running
    # CI job is added, add it here too or ci-required-drift F1 will flag it.
    #
    # Stays on the dedicated `ci-meta` lane (no docker work, so the
@@ -549,6 +547,7 @@ jobs:
      - canvas-build
      - shellcheck
      - python-lint
+      - canvas-deploy-status
    continue-on-error: false
    runs-on: ci-meta
    timeout-minutes: 5
@@ -567,6 +566,7 @@ jobs:
          CANVAS_RESULT: ${{ needs.canvas-build.result }}
          SHELLCHECK_RESULT: ${{ needs.shellcheck.result }}
          PYTHON_LINT_RESULT: ${{ needs.python-lint.result }}
+          CANVAS_DEPLOY_RESULT: ${{ needs.canvas-deploy-status.result }}
        run: |
          set -euo pipefail
          fail=0
@@ -588,6 +588,7 @@ jobs:
          check "Canvas (Next.js)"      "$CANVAS_RESULT"
          check "Shellcheck (E2E scripts)" "$SHELLCHECK_RESULT"
          check "Python Lint & Test"    "$PYTHON_LINT_RESULT"
+          check "Canvas Deploy Status"  "$CANVAS_DEPLOY_RESULT"
          if [ "$fail" -ne 0 ]; then
            echo "::error::all-required: one or more aggregated CI jobs did not succeed"
            exit 1
@@ -131,9 +131,9 @@ jobs:
      # on the per-runtime default ("sonnet" → routes to direct
      # Anthropic, defeats the cost saving). Operators can override
      # via workflow_dispatch by setting a different E2E_MODEL_SLUG
-      # input if they need to exercise a specific model. MiniMax-M2 is the
-      # stable staging MiniMax path used by the full-SaaS smoke.
-      E2E_MODEL_SLUG: ${{ github.event.inputs.model_slug || 'MiniMax-M2' }}
+      # input if they need to exercise a specific model. MiniMax-M2.7 is the
+      # stable staging MiniMax path used by the full-SaaS smoke (#1997).
+      E2E_MODEL_SLUG: ${{ github.event.inputs.model_slug || 'MiniMax-M2.7' }}
      # Bound to 10 min so a stuck provision fails the run instead of
      # holding up the next cron firing. 15-min default in the script
      # is for the on-PR full lifecycle where we have more headroom.
@@ -250,6 +250,20 @@ jobs:
          echo "CANVAS_PORT=${CANVAS_PORT}" >> "$GITHUB_ENV"
          echo "Canvas host port: ${CANVAS_PORT}"

+      - name: Set deterministic admin token
+        if: needs.detect-changes.outputs.chat == 'true'
+        run: |
+          # PR #2291 made auth fail-closed everywhere (no dev-mode escape).
+          # The platform server requires ADMIN_TOKEN; the canvas requires the
+          # matching NEXT_PUBLIC_ADMIN_TOKEN or every API call 401s.
+          # We set a deterministic per-run value so the ephemeral platform
+          # and canvas are paired correctly.
+          E2E_ADMIN_TOKEN="e2e-chat-admin-${{ github.run_id }}-${{ github.run_attempt }}"
+          echo "ADMIN_TOKEN=${E2E_ADMIN_TOKEN}" >> "$GITHUB_ENV"
+          echo "MOLECULE_ADMIN_TOKEN=${E2E_ADMIN_TOKEN}" >> "$GITHUB_ENV"
+          echo "NEXT_PUBLIC_ADMIN_TOKEN=${E2E_ADMIN_TOKEN}" >> "$GITHUB_ENV"
+          echo "Admin token configured for e2e-chat platform + canvas."
+
      - name: Start platform (background)
        if: needs.detect-changes.outputs.chat == 'true'
        working-directory: workspace-server
@@ -61,11 +61,9 @@ name: Lint pre-flip continue-on-error
 # feedback_no_shared_persona_token_use.
 #
 # Phase contract (RFC internal#219 §1 ladder):
-#   - This workflow lands at `continue-on-error: true` (Phase 3 —
-#     surface defects without blocking). Follow-up PR flips it to
-#     `false` ONLY after this workflow's own recent runs on `main`
-#     are confirmed clean — exactly the discipline the workflow
-#     itself enforces. Eat your own dogfood.
+#   - Flipped to `continue-on-error: false` after Researcher live-verified
+#     clean runs. The script's own 35 pytest tests pass and recent PR
+#     history shows no masked regressions — the gate is now enforcing.

 on:
  pull_request:
@@ -97,10 +95,9 @@ jobs:
    name: Verify continue-on-error flips have run-log proof
    runs-on: ubuntu-latest
    timeout-minutes: 8
-    # Phase 3 (RFC internal#219 §1): surface broken flips without blocking
-    # the PR yet. Follow-up flips this to `false` once the workflow itself
-    # has clean recent runs on main. mc#1982 interim — remove when CoE→false.
-    continue-on-error: true  # mc#1982
+    # Fail-closed: the lint script is verified clean (35/35 tests pass,
+    # Researcher live-check confirmed). Masking removed per mc#1982 close-out.
+    continue-on-error: false
    steps:
      - name: Check out PR head (full history for base-SHA access)
        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
@@ -40,6 +40,7 @@ env:
  GITHUB_SERVER_URL: https://git.moleculesai.app

 jobs:
+  # bp-exempt: informational lint enforcing docker-host/publish pin convention (internal#512), not a merge gate
  lint-docker-host-pin:
    name: Lint docker-host pin on docker-touching workflows
    runs-on: docker-host
@@ -112,9 +112,9 @@ jobs:
      E2E_RUNTIME: claude-code
      # Pin the smoke to a specific MiniMax model rather than relying
      # on the per-runtime default (which could resolve to "sonnet" →
-      # direct Anthropic and defeat the cost saving). MiniMax-M2 is the
-      # stable staging MiniMax path used by the full-SaaS smoke.
-      E2E_MODEL_SLUG: MiniMax-M2
+      # direct Anthropic and defeat the cost saving). MiniMax-M2.7 is the
+      # stable staging MiniMax path used by the full-SaaS smoke (#1997).
+      E2E_MODEL_SLUG: MiniMax-M2.7
      E2E_RUN_ID: "smoke-${{ github.run_id }}"
      # Debug-only: when an operator dispatches with keep_on_failure=true,
      # the smoke script's E2E_KEEP_ORG=1 path skips teardown so the
@@ -34,8 +34,10 @@ name: Sweep stale Cloudflare DNS records
 # scripts/ops/test_sweep_cf_decide.py (#2027) cover the rule
 # classifier.
 #
-# Secrets: CF_API_TOKEN, CF_ZONE_ID, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
-# are confirmed existing per issue #425 §425 audit. CP_ADMIN_API_TOKEN and
+# Secrets: CF_API_TOKEN (preferred CI-scoped name) or CLOUDFLARE_API_TOKEN
+# (operator-host canonical name) are accepted — the workflow falls back
+# automatically. Same for CF_ZONE_ID / CLOUDFLARE_ZONE_ID. Confirmed
+# existing per issue #425 §425 audit. CP_ADMIN_API_TOKEN and
 # CP_STAGING_ADMIN_API_TOKEN are unconfirmed — if missing, the verify step
 # (schedule → hard-fail, dispatch → soft-skip) surfaces it clearly.

@@ -79,8 +81,8 @@ jobs:
    # each individually capped at 10s by the script's curl -m flag.
    timeout-minutes: 3
    env:
-      CF_API_TOKEN: ${{ secrets.CF_API_TOKEN }}
-      CF_ZONE_ID: ${{ secrets.CF_ZONE_ID }}
+      CF_API_TOKEN: ${{ secrets.CF_API_TOKEN || secrets.CLOUDFLARE_API_TOKEN }}
+      CF_ZONE_ID: ${{ secrets.CF_ZONE_ID || secrets.CLOUDFLARE_ZONE_ID }}
      CP_ADMIN_API_TOKEN: ${{ secrets.CP_ADMIN_API_TOKEN }}
      CP_STAGING_ADMIN_API_TOKEN: ${{ secrets.CP_STAGING_ADMIN_API_TOKEN }}
      AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
@@ -129,6 +131,7 @@ jobs:
            fi
            echo "::error::sweep cannot run — required secrets missing: ${missing[*]}"
            echo "::error::set them at Settings → Secrets and Variables → Actions, or disable this workflow."
+            echo "::error::Cloudflare secrets accept either the CI-scoped name (CF_API_TOKEN / CF_ZONE_ID) or the operator-host canonical name (CLOUDFLARE_API_TOKEN / CLOUDFLARE_ZONE_ID)."
            echo "::error::a silent skip masked an active CF DNS leak (152/200 zone records) caught only by a manual audit on 2026-04-28; this gate exists to make the gap visible."
            exit 1
          fi
@@ -29,10 +29,12 @@ name: Sweep stale Cloudflare Tunnels
 # the DNS sweep's 50% because tenant-shaped tunnels are mostly
 # orphans by design) refuses to nuke past the threshold.
 #
-# Secrets: CF_API_TOKEN, CF_ACCOUNT_ID are confirmed existing per
-# issue #425 §425 audit. CP_ADMIN_API_TOKEN and CP_STAGING_ADMIN_API_TOKEN
-# are unconfirmed — if missing, the verify step (schedule → hard-fail,
-# dispatch → soft-skip) surfaces it clearly.
+# Secrets: CF_API_TOKEN (preferred CI-scoped name) or CLOUDFLARE_API_TOKEN
+# (operator-host canonical name) are accepted — the workflow falls back
+# automatically. Same for CF_ACCOUNT_ID / CLOUDFLARE_ACCOUNT_ID. Confirmed
+# existing per issue #425 §425 audit. CP_ADMIN_API_TOKEN and
+# CP_STAGING_ADMIN_API_TOKEN are unconfirmed — if missing, the verify step
+# (schedule → hard-fail, dispatch → soft-skip) surfaces it clearly.

 on:
  schedule:
@@ -74,8 +76,8 @@ jobs:
    # the sweep-cf-orphans companion job).
    timeout-minutes: 30
    env:
-      CF_API_TOKEN: ${{ secrets.CF_API_TOKEN }}
-      CF_ACCOUNT_ID: ${{ secrets.CF_ACCOUNT_ID }}
+      CF_API_TOKEN: ${{ secrets.CF_API_TOKEN || secrets.CLOUDFLARE_API_TOKEN }}
+      CF_ACCOUNT_ID: ${{ secrets.CF_ACCOUNT_ID || secrets.CLOUDFLARE_ACCOUNT_ID }}
      CP_ADMIN_API_TOKEN: ${{ secrets.CP_ADMIN_API_TOKEN }}
      CP_STAGING_ADMIN_API_TOKEN: ${{ secrets.CP_STAGING_ADMIN_API_TOKEN }}
      MAX_DELETE_PCT: ${{ github.event.inputs.max_delete_pct || '90' }}
@@ -0,0 +1,67 @@
+# umbrella-reaper — auto-recovery for stale CI umbrella statuses on open PRs.
+#
+# Tracking: molecule-core#1780.
+#
+# Problem: when `CI / all-required (pull_request)` reports failure due to
+# a propagation/timing race despite all required sub-jobs being success,
+# branch protection blocks the merge. Operators currently recover manually
+# per docs/runbooks/ci-umbrella-stale-compensating-status.md.
+#
+# This workflow automates that recovery: it scans open PRs and posts a
+# compensating success status when the umbrella is stale but all sub-jobs
+# are verified green.
+#
+# Trust boundary: the script only reads PR lists + statuses and POSTs to
+# /statuses/{sha}. It never checks out PR HEAD code. The Gitea token has
+# write:repository scope for statuses only.
+#
+# Sibling: .gitea/workflows/status-reaper.yml (default-branch push-suffix
+# compensation). Same persona provisioning model.
+
+name: umbrella-reaper
+
+# IMPORTANT — Schedule moved to operator-config:
+#   /etc/cron.d/molecule-core-umbrella-reaper ->
+#   /usr/local/bin/molecule-core-cron-bot.sh umbrella-reaper
+#
+# This keeps the compensation cadence but stops a maintenance bot from
+# consuming Gitea Actions runner slots during PR merge waves.
+# Gitea 1.22.6 parser quirk per
+# `feedback_gitea_workflow_dispatch_inputs_unsupported`: do NOT add an
+# `inputs:` block here. Gitea 1.22.6 rejects the whole workflow as
+# "unknown on type" when `workflow_dispatch.inputs.X` is present.
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+# NOTE: NO `concurrency:` block is intentional — same reasoning as
+# status-reaper.yml. Gitea 1.22.6 doesn't honor cancel-in-progress for
+# queued ticks; the POST is idempotent so concurrent ticks are safe.
+
+jobs:
+  reap:
+    runs-on: ubuntu-latest
+    timeout-minutes: 8
+    steps:
+      - name: Check out repo at default-branch HEAD
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
+        with:
+          python-version: '3.12'
+
+      - name: Install PyYAML
+        run: python -m pip install --quiet 'PyYAML==6.0.2'
+
+      - name: Compensate stale PR umbrella statuses
+        env:
+          GITEA_TOKEN: ${{ secrets.UMBRELLA_REAPER_TOKEN }}
+          GITEA_HOST: git.moleculesai.app
+          REPO: ${{ github.repository }}
+          PR_LIMIT: "50"
+        run: python3 .gitea/scripts/umbrella-reaper.py
@@ -27,9 +27,13 @@ export async function seedWorkspace(echoURL: string): Promise<SeededWorkspace> {
  // 1. Create external workspace pointing at the in-process echo runtime.
  const runId = Math.random().toString(36).slice(2, 8);
  const wsName = `Chat E2E Agent ${runId}`;
+  const adminToken = process.env.E2E_ADMIN_TOKEN ?? process.env.ADMIN_TOKEN;
  const createRes = await fetch(`${PLATFORM_URL}/workspaces`, {
    method: "POST",
-    headers: { "Content-Type": "application/json" },
+    headers: {
+      "Content-Type": "application/json",
+      ...(adminToken ? { Authorization: `Bearer ${adminToken}` } : {}),
+    },
    body: JSON.stringify({
      name: wsName,
      tier: 1,
@@ -234,30 +234,44 @@ export default async function globalSetup(_config: FullConfig): Promise<void> {
    "Authorization": `Bearer ${tenantToken}`,
    "X-Molecule-Org-Id": orgID,
  };
-  const ws = await jsonFetch(`${tenantURL}/workspaces`, {
-    method: "POST",
-    headers: tenantAuth,
-    body: JSON.stringify({
-      name: "E2E Canvas Test",
-      runtime: "hermes",
-      tier: 2,
-      // Provider-registry SSOT (internal#718) registers ONLY Kimi models for
-      // the hermes runtime — `moonshot/kimi-k2.6` is the platform-managed
-      // entry (workspace-server/internal/providers/providers.yaml, hermes ->
-      // platform). The old `gpt-4o` was never a registered hermes model and
-      // now 422s UNREGISTERED_MODEL_FOR_RUNTIME (core#2225). This workspace
-      // defaults closed to platform_managed (see the boot-shape note below),
-      // so a platform-namespaced model id is the registry-correct choice.
-      model: "moonshot/kimi-k2.6",
-    }),
-  });
-  if (ws.status >= 400 || !ws.body?.id) {
-    throw new Error(`Workspace create ${ws.status}: ${JSON.stringify(ws.body)}`);
+  // Retry workspace creation on transient 5xx / timeout — staging CP can
+  // return 502/503/504 under load and a single-shot failure kills the
+  // entire E2E run. 3 attempts with 3s exponential backoff (3s, 6s, 12s)
+  // gives ~21s total budget, well inside the 20-min provision envelope.
+  let workspaceId = "";
+  for (let attempt = 1; attempt <= 3; attempt++) {
+    const ws = await jsonFetch(`${tenantURL}/workspaces`, {
+      method: "POST",
+      headers: tenantAuth,
+      body: JSON.stringify({
+        name: "E2E Canvas Test",
+        runtime: "hermes",
+        tier: 2,
+        // Provider-registry SSOT (internal#718) registers ONLY Kimi models for
+        // the hermes runtime — `moonshot/kimi-k2.6` is the platform-managed
+        // entry (workspace-server/internal/providers/providers.yaml, hermes ->
+        // platform). The old `gpt-4o` was never a registered hermes model and
+        // now 422s UNREGISTERED_MODEL_FOR_RUNTIME (core#2225). This workspace
+        // defaults closed to platform_managed (see the boot-shape note below),
+        // so a platform-namespaced model id is the registry-correct choice.
+        model: "moonshot/kimi-k2.6",
+      }),
+    });
+    if (ws.status >= 200 && ws.status < 300 && ws.body?.id) {
+      workspaceId = ws.body.id as string;
+      break;
+    }
+    const isTransient = ws.status >= 500 || ws.status === 0;
+    if (!isTransient || attempt === 3) {
+      throw new Error(`Workspace create ${ws.status} (attempt ${attempt}): ${JSON.stringify(ws.body)}`);
+    }
+    const backoff = 3000 * Math.pow(2, attempt - 1);
+    console.log(`[staging-setup] Workspace create transient ${ws.status}, retrying in ${backoff}ms...`);
+    await new Promise((r) => setTimeout(r, backoff));
  }
-  const workspaceId = ws.body.id as string;
  console.log(`[staging-setup] Workspace created: ${workspaceId}`);

-  // 6. Wait for workspace RENDERABLE.
+  // 6. Wait for workspace online
  //
  // This harness exists to verify the canvas *tab UI* renders (staging-
  // tabs.spec.ts: open each of the 13 workspace-panel tabs, assert no hard
@@ -266,6 +280,16 @@ export default async function globalSetup(_config: FullConfig): Promise<void> {
  // it needs is a workspace ROW that the canvas lists so the node renders
  // and the side-panel tabs open. A fully-`online` agent is NOT required.
  //
+  // Hermes cold-boot takes 10-13 min on slow apt days (apt + uv + hermes
+  // install + npm browser-tools). The controlplane bootstrap-watcher
+  // deadline fires at 5 min and sets status=failed prematurely; heartbeat
+  // then transitions failed → online after install.sh finishes. The ONLY
+  // failed shape we tolerate is the pre-start credential-abort
+  // (uptime_seconds=0, no last_sample_error) — the agent never ran. Real
+  // boot regressions (image pull error, panic, PYTHONPATH, etc.) still
+  // hard-throw immediately so triage gets detail without waiting for a
+  // polling timeout. See test_staging_full_saas.sh step 7/11 and issue #2632.
+  //
  // That distinction became load-bearing on 2026-06-03: workspace-server
  // #2162 (fix(provision): platform-managed workspace must fail-closed when
  // CP proxy env absent) made a platform_managed workspace ABORT AT BOOT
@@ -287,8 +311,10 @@ export default async function globalSetup(_config: FullConfig): Promise<void> {
  // the node + tabs render, proceed. We do NOT mask a real boot regression:
  // any `failed` carrying a last_sample_error, OR a non-zero uptime (the
  // agent started then crashed — image pull, panic, PYTHONPATH, etc.),
-  // still hard-throws. Genuine *infra* provision failure is already caught
-  // loud one step earlier at the org level (instance_status === "failed").
+  // still hard-throws immediately so triage gets boot_stage / last_error /
+  // image fields without waiting for a polling timeout.
+  // Genuine *infra* provision failure is already caught loud one step
+  // earlier at the org level (instance_status === "failed").
  await waitFor<boolean>(
    async () => {
      const r = await jsonFetch(`${tenantURL}/workspaces/${workspaceId}`, {
@@ -315,13 +341,7 @@ export default async function globalSetup(_config: FullConfig): Promise<void> {
          );
          return true;
        }
-        // last_sample_error is often empty when the failure happens before
-        // the agent emits a sample (e.g. boot crash, image pull error,
-        // missing PYTHONPATH, OpenAI quota at startup). Dumping the full
-        // body gives triage the boot_stage / last_error / image fields it
-        // needs without a second probe. Otherwise this propagates as a
-        // bare "Workspace failed: " — the exact useless message that
-        // sent #2632 to the issue tracker.
+        // Real boot regression — hard-throw immediately with full detail.
        const detail = sampleErr
          ? sampleErr
          : `(no last_sample_error) full body: ${JSON.stringify(r.body)}`;
@@ -333,7 +353,7 @@ export default async function globalSetup(_config: FullConfig): Promise<void> {
    10_000,
    "workspace online",
  );
-  console.log(`[staging-setup] Workspace renderable`);
+  console.log(`[staging-setup] Workspace online`);

  // 7. Hand state off to tests + teardown — overwrite the slug-only
  // bootstrap state with the full state spec tests need.
@@ -370,7 +370,7 @@ test.describe("staging canvas tabs", () => {

    // The tablist appears once the side panel mounts. Condition-based
    // wait — no fixed delay.
-    const tablist = page.locator('[role="tablist"]');
+    const tablist = page.getByRole("tablist", { name: "Workspace panel tabs" });
    await expect(
      tablist,
      "side panel tablist never appeared after clicking the workspace node",
@@ -172,7 +172,7 @@ export function ContextMenu() {
    const nodeId = contextMenu.nodeId;
    closeContextMenu();
    try {
-      await api.post(`/workspaces/${nodeId}/pause`, {});
+      await api.post(`/workspaces/${nodeId}/pause?cascade=true`, {});
      updateNodeData(nodeId, { status: "paused" });
    } catch (e) {
      showToast("Pause failed", "error");
@@ -184,7 +184,7 @@ export function ContextMenu() {
    const nodeId = contextMenu.nodeId;
    closeContextMenu();
    try {
-      await api.post(`/workspaces/${nodeId}/resume`, {});
+      await api.post(`/workspaces/${nodeId}/resume?cascade=true`, {});
      updateNodeData(nodeId, { status: "provisioning" });
    } catch (e) {
      showToast("Resume failed", "error");
@@ -385,7 +385,7 @@ describe("ContextMenu — item actions", () => {
    render(<ContextMenu />);
    fireEvent.click(screen.getByRole("menuitem", { name: /pause/i }));
    await act(async () => { /* flush */ });
-    expect(mockPost).toHaveBeenCalledWith("/workspaces/n1/pause", {});
+    expect(mockPost).toHaveBeenCalledWith("/workspaces/n1/pause?cascade=true", {});
    expect(mockStoreState.updateNodeData).toHaveBeenCalledWith("n1", { status: "paused" });
  });

@@ -395,7 +395,7 @@ describe("ContextMenu — item actions", () => {
    render(<ContextMenu />);
    fireEvent.click(screen.getByRole("menuitem", { name: /resume/i }));
    await act(async () => { /* flush */ });
-    expect(mockPost).toHaveBeenCalledWith("/workspaces/n1/resume", {});
+    expect(mockPost).toHaveBeenCalledWith("/workspaces/n1/resume?cascade=true", {});
  });
 });

@@ -324,7 +324,7 @@ export const useCanvasStore = create<CanvasState>((set, get) => ({
  batchPause: async () => {
    const ids = Array.from(get().selectedNodeIds);
    const results = await Promise.allSettled(
-      ids.map((id) => api.post(`/workspaces/${id}/pause`))
+      ids.map((id) => api.post(`/workspaces/${id}/pause?cascade=true`))
    );
    const failed: string[] = [];
    results.forEach((r, i) => {
@@ -2,7 +2,7 @@

 **Status:** living document — update when you ship a feature that touches one backend.
 **Owner:** workspace-server + controlplane teams.
-**Last audit:** 2026-05-07 (plugin install/uninstall closed for EC2 backend via EIC SSH push to the bind-mounted `/configs/plugins/<name>/`, mirroring the Files API PR #1702 pattern).
+**Last audit:** 2026-05-31 (Claude agent — drift risk #6 verified resolved; nil guards present, contract tests run without Skip).

 ## Why this exists

@@ -93,12 +93,12 @@ For "do we have any backend?", use `HasProvisioner()`, never bare `h.provisioner
 3. **Restart divergence on runtime changes.** Docker re-reads `/configs/config.yaml` from the container before stop, so a changed `runtime:` survives a restart even if the DB isn't synced. EC2 trusts the DB only. If you change the runtime via the Config tab and the handler races the restart, Docker will land on the new runtime, EC2 will land on the old one. **Fix path:** make the Config-tab save explicitly flush to DB before kicking off a restart, not deferred.
 4. **Console-output asymmetry.** Users debugging a stuck workspace on Docker see `docker logs`; on EC2 they see `GetConsoleOutput`. The two outputs look nothing alike. **Fix path:** expose a unified `GET /workspaces/:id/boot-log` that proxies to whichever backend serves the data. Already partly there via `cp_provisioner.Console`.
 5. **Template script drift.** `install.sh` and `start.sh` in each template repo do the same high-level work (install hermes-agent, write .env, write config.yaml, start gateway) but must be kept byte-level consistent on the provider-key forwarding block. Easy to forget. Enforced now by `tools/check-template-parity.sh` (see below) — run it in each template repo's CI.
-6. **Both backends panic when underlying client is nil.** ✅ **Resolved** (`fix/provisioner-nil-guards-1813`). `Provisioner.{Stop,IsRunning}` and `CPProvisioner.{Stop,IsRunning}` now guard against nil clients with `ErrNoBackend`, so the contract-test runner executes scenarios against zero-valued backends without panic.
+6. ~~**Both backends panic when underlying client is nil.**~~ **RESOLVED** — nil guards landed in `Provisioner` (`Start`, `Stop`, `IsRunning`, `ExecRead`, `RemoveVolume`, `VolumeHasFile`, `WriteAuthTokenToVolume`) and `CPProvisioner` (`Stop`, `IsRunning`), all returning `ErrNoBackend`. Contract tests (`TestDockerBackend_Contract`, `TestCPProvisionerBackend_Contract`, `TestZeroValuedBackends_NoPanic`) run in CI without `t.Skip`.

 ## Enforcement

 - **`tools/check-template-parity.sh`** (this repo) — ensures `install.sh` and `start.sh` in a template repo forward identical sets of provider keys. Wire into each template repo's CI as `bash $MONOREPO/tools/check-template-parity.sh install.sh start.sh`.
- **Contract tests** — `workspace-server/internal/provisioner/backend_contract_test.go` defines the behaviors every `provisioner.Provisioner` implementation must satisfy. Fails compile when a method drifts between `Docker` and `CPProvisioner`. Scenario-level runs execute against zero-valued backends since drift risk #6 was resolved (`fix/provisioner-nil-guards-1813`).
+- **Contract tests** — `workspace-server/internal/provisioner/backend_contract_test.go` defines the behaviors every `provisioner.Provisioner` implementation must satisfy. Fails compile when a method drifts between `Docker` and `CPProvisioner`. Scenario-level runs (`TestDockerBackend_Contract`, `TestCPProvisionerBackend_Contract`, `TestZeroValuedBackends_NoPanic`) execute in CI — drift risk #6 resolved.
 - **Source-level dispatcher pins** — `workspace_provision_auto_test.go` enforces the SoT pattern documented above:
  - `TestNoCallSiteCallsDirectProvisionerExceptAuto` — no handler calls `.provisionWorkspace(` or `.provisionWorkspaceCP(` directly outside the dispatcher's allowlist.
  - `TestNoCallSiteCallsBareStop` — no handler calls `.provisioner.Stop(` or `.cpProv.Stop(` directly outside the dispatcher's allowlist (strips Go comments before substring match so archaeology in code comments doesn't trip the gate).
@@ -0,0 +1,293 @@
+# RFC: Org-level Platform Agent — a tenant-resident concierge
+
+**Perspective:** CTO + Backend Engineer + DevOps
+**Status:** Draft — pre-implementation, **CTO sign-off required before any implementation PR**
+**Scope:** `molecule-core` (workspace-server), `molecule-controlplane`, workspace runtime, `molecule-app`
+**This document is the single source of truth (SSOT) for the feature.** Code, OpenAPI, the platform
+MCP, and end-user docs reconcile to this RFC — not to each other.
+
+---
+
+## 1. Summary
+
+Today a Molecule tenant is a control/router box: one EC2 runs the `workspace-server`
+(`molecule-tenant` container) + Postgres + Redis, and **each workspace is its own separate EC2**
+running a runtime image that joins the tenant's A2A mesh. A2A has exactly two participant kinds:
+**workspaces** (agents) and the **user** (the canvas, modeled implicitly as `activity_logs.source_id
+IS NULL`). A user who wants to *do* anything must drive individual workspaces directly — create them,
+assign agents, wire channels/schedules/secrets — i.e. they must carry a lot of platform knowledge.
+
+This RFC introduces a **platform agent**: an always-on org-level agent that
+
+1. runs as a **container on the tenant EC2** itself (beside `molecule-tenant`),
+2. natively holds the **platform-management MCP** (the org-admin tool surface) so it can do anything
+   in the org,
+3. joins A2A as a **first-class third participant** (`kind='platform'`) that sits at the org root, and
+4. becomes the **user's default chat target** — a concierge the user talks to like a chatbot, which
+   then orchestrates the org on their behalf.
+
+Destructive actions the concierge triggers are **human-approved** through the existing approvals
+subsystem.
+
+## 2. Motivation
+
+- **Lower the knowledge floor.** "Spin up an SEO team and have them publish weekly" should be a
+  sentence, not a sequence of workspace/agent/schedule/secret operations.
+- **One front door.** A single conversational entry point that *is* the org, instead of N per-workspace
+  chats the user has to coordinate.
+- **Reuse, don't rebuild.** The agent runtime, A2A mesh, the 87-tool platform MCP, and the approvals
+  subsystem already exist. This feature is mostly *composition* plus one honest new participant kind.
+
+## 3. Goals / Non-Goals
+
+**Goals**
+- A per-tenant platform agent, provisioned automatically, that controls the org via the platform MCP.
+- A first-class `platform` participant in A2A with correct routing and tenant isolation.
+- Server-side approval gating for destructive org operations.
+- Parity with normal workspaces for runtime/model/provider/billing (no special-casing).
+
+**Non-Goals (this RFC)**
+- Replacing the canvas. The canvas remains the advanced/power-user surface.
+- Multi-concierge / per-team concierges. Exactly **one** platform agent per org.
+- A new scoped-down token system for the MCP (tracked separately; see §10 Open Questions).
+
+## 4. Current-state ground truth (verified, with references)
+
+- **Topology.** Tenant EC2 runs `molecule-tenant` (workspace-server) + Postgres + Redis;
+  `controlplane/internal/provisioner/ec2.go:buildTenantUserDataSM()` `docker run`s it with
+  `--network host`, `PORT=8080`. Each **workspace is its own EC2** (`ec2.go:ProvisionWorkspace`).
+- **No `org_id` column.** An "org" is the `parent_id IS NULL` subtree root;
+  `workspace-server/internal/handlers/org_scope.go` resolves it with a recursive CTE (`orgRootID`) and
+  `sameOrg()` compares two workspaces' resolved roots for tenant isolation (#1953/OFFSEC-015).
+- **A2A authorization is hierarchy-based.** `workspace-server/internal/registry/access.go:CanCommunicate`
+  permits self / siblings / ancestor↔descendant. Root-level rows are "siblings" but every routing path
+  is additionally gated by `sameOrg()`.
+- **No participant-kind discriminator.** `workspaces.role` is a free-form string; the user is implicit
+  (`activity_logs.source_id IS NULL`). `migrations/001_workspaces.sql`.
+- **Runtime injects MCP servers** in the claude-code executor's `mcp_servers` dict — today exactly one
+  entry, `"a2a"` (`molecule-ai-workspace-template-claude-code/claude_sdk_executor.py`,
+  `molecule_runtime/claude_sdk_executor.py`). The agent self-registers via `POST /registry/register`
+  (`molecule_runtime/main.py`) and is identified by `WORKSPACE_ID` + `X-Molecule-Org-Id`.
+- **Platform MCP** (`molecule-mcp-server`, stdio Node) authenticates purely from env
+  (`MOLECULE_API_KEY` = org-admin token, `MOLECULE_API_URL`, `MOLECULE_ORG_ID`; `src/api.ts`), is a
+  thin proxy over the tenant REST/A2A API (`chat_with_agent` → `POST /workspaces/:id/a2a`,
+  `async_delegate` → `/delegate`), and has **zero embeddability blockers**.
+- **Billing** is a per-workspace resolver — `ResolveLLMBillingModeDerived`
+  (`workspace-server/internal/handlers/workspace_provision.go`, `llm_billing_mode.go`), defaulting
+  closed to `platform_managed`; `byok` runs on the tenant's own provider key (see
+  `docs/architecture/byok-fail-closed-billing.md`).
+- **Approvals** exist: `migrations/007_approvals.sql`, `internal/handlers/approvals.go`,
+  `EventApprovalRequested`, decide route `POST /workspaces/:id/approvals/:approvalId/decide`.
+
+## 5. Design
+
+### 5.1 The platform agent IS the org root
+
+Because `sameOrg()` resolves each workspace to its topmost `parent_id IS NULL` root, a platform agent
+added as a *second* root would resolve to a *different* root than the existing team and be **blocked**
+by `sameOrg`. Therefore the platform agent **becomes the single org root**, and the org's existing
+root is **re-parented under it**. Consequences:
+
+- `orgRootID(any workspace) == platform-agent-id`; `sameOrg(platform, any in-org ws) == true`.
+- The platform agent reaches every workspace (and is reachable) via the **existing**
+  ancestor↔descendant rules — **no `CanCommunicate` change**, and tenant isolation is unchanged.
+
+This is the honest realization of "a third participant above workspace and user": the concierge is
+literally the org.
+
+### 5.2 `kind` discriminator (the only new marker)
+
+Add a single column `workspaces.kind TEXT NOT NULL DEFAULT 'workspace'`, constrained to
+`('workspace','platform')`. It is the **only** marker of the platform agent — we do **not** also
+encode identity in `role`/`tier` (those stay descriptive). The enum is defined once: the migration
+`CHECK` and the Go constants `KindWorkspace`/`KindPlatform` (+ one `IsValidKind`) are kept in lockstep.
+
+Invariants (handler-enforced, since there is no `org_id` for a pure-SQL unique):
+- `kind='platform' ⇒ parent_id IS NULL`.
+- A row may be `kind='platform'` only if it is its own org root (`orgRootID(self) == self`), giving
+  "exactly one platform agent per org". Guard the check+write in a tx with `FOR UPDATE` on the root.
+
+### 5.3 Identity & registration
+
+- **ID** = derived `uuidv5(org-namespace, "platform-agent")` — reproducible, no stored-vs-derived
+  drift, lowercase so it satisfies the runtime's `WORKSPACE_ID` validator.
+- CP **pre-seeds** the `workspaces` row (`kind='platform'`, `parent_id=NULL`, `tier=0`) before the
+  agent boots; the agent self-registers (`POST /registry/register`) into that row. `Register` accepts
+  an optional `kind` and reconciles it, enforcing the §5.2 invariants.
+
+### 5.4 Default-target resolver
+
+New `GET /registry/platform-agent` (handler `internal/handlers/platform_agent.go`): resolve the
+caller's `orgRootID()` and return it iff `kind='platform'`. This is the server hook the dashboard
+targets by default; no change to `ProxyA2A`. **Authored in the OpenAPI SSOT first**; MCP/CLI/docs
+derive from it.
+
+### 5.5 Runtime: two MCPs, config-driven
+
+Make the runtime's `mcp_servers` **config-driven** rather than hardcoded:
+- `molecule_runtime/config.py`: add `extra_mcp_servers: list[dict]` to `WorkspaceConfig`, read
+  `raw.get("mcp_servers", [])`.
+- Both executors merge `extra_mcp_servers` into the `mcp_servers` dict after the always-on `"a2a"`
+  entry (the template `claude_sdk_executor.py` is the live one; the runtime-package copy is the
+  fallback).
+
+The platform agent's `config.yaml` then declares:
+
+```yaml
+runtime: claude-code
+model: sonnet            # default; user-switchable model AND provider via providers.yaml
+a2a:
+  port: 8090             # avoid the workspace default 8000 under host networking
+mcp_servers:
+  - name: platform
+    command: node
+    args: ["/opt/molecule-mcp-server/dist/index.js"]
+```
+
+The `platform` MCP reads `MOLECULE_API_KEY`/`MOLECULE_API_URL`/`MOLECULE_ORG_ID` from the container
+env (passed through to the stdio child) — no per-server `env` block needed.
+
+### 5.6 Hosting & provisioning (tenant EC2 container)
+
+In `ec2.go:buildTenantUserDataSM()` add a `start_platform_agent` stage **after** `wait_platform_health`
+(the agent registers against `localhost:8080` on boot):
+
+```bash
+docker run -d --restart=always --name molecule-platform-agent --network host \
+  -v /data/platform-agent/configs:/configs \
+  -e WORKSPACE_ID=<platform-uuid> -e WORKSPACE_CONFIG_PATH=/configs \
+  -e PLATFORM_URL=http://localhost:8080 \
+  -e MOLECULE_API_URL=http://localhost:8080 -e MOLECULE_API_KEY=$ADMIN_TOKEN -e MOLECULE_ORG_ID=<orgID> \
+  -e ANTHROPIC_AUTH_TOKEN=$ADMIN_TOKEN -e MOLECULE_LLM_ANTHROPIC_BASE_URL=$MOLECULE_LLM_ANTHROPIC_BASE_URL \
+  <platform-agent-image>
+```
+
+- The org `admin_token` is already on the box (Secrets Manager `molecule/tenant/{orgID}`).
+- `--restart=always` provides Docker-level supervision (matches `molecule-tenant`).
+- Mirror the block into the redeploy path (`buildRedeployScript`) so existing tenants backfill it.
+
+### 5.7 Image
+
+A **dedicated `molecule-platform-agent` image**: `FROM workspace-template-claude-code`, `COPY` the
+prebuilt `molecule-mcp-server/dist` + `node_modules` into `/opt/molecule-mcp-server`, and **pin Node
+20** (the slim base ships Node 18; the MCP expects ≥20). A dedicated image keeps the org-admin MCP
+**out of** ordinary workspace images (security hygiene) and lets us set concierge defaults without
+touching the workspace template. `molecule-ci` publishes it.
+
+### 5.8 Approval gate (server-side trust boundary)
+
+The MCP is a *client* of the tenant handlers, so enforcement lives in the **handlers**, not the MCP.
+
+- `internal/approvals/policy.go` (new): one auditable map of gated actions —
+  `delete_workspace`, `deprovision`, `secret_write`, `org_token_mint`.
+- `requireApproval(ctx, workspaceID, action, contextHash)` reuses the existing approvals
+  INSERT/broadcast/escalate. If an `approved`+unconsumed row matches → consume it → proceed. Else
+  create a `pending` row, broadcast `EventApprovalRequested`, and return **HTTP 202
+  `{approval_id, status:"pending"}`** instead of executing. The human decides via the existing decide
+  route; the agent retries and the gate now passes.
+- Add `approval_requests.consumed_at` (single-use) and optional `request_hash` (dedupe identical
+  pending requests).
+- **Escalation:** the platform agent's `parent_id` is NULL, so platform-originated approvals escalate
+  to the **user** (canvas notify), not a parent.
+- The 202 response shape is authored in the **OpenAPI SSOT**.
+
+### 5.9 Billing & model/provider parity
+
+The platform agent is a `workspaces` row, so it inherits the one billing resolver and the
+`providers.yaml` runtime matrix unchanged:
+- **Default `platform_managed`** (metered CP proxy, billed to org credits) — the env wiring in §5.6.
+- **`byok`** = flip `/admin/workspaces/:id/llm-billing-mode` + supply the org's `ANTHROPIC_API_KEY`
+  secret (workspace or global). Exposed as a provisioning flag so a tenant can choose at create time.
+- Model **and provider** are switchable (Claude, Kimi-for-coding, …) via the same dashboard
+  model-switcher any workspace uses.
+
+### 5.10 UX (summary; detailed in app RFC / Phase 5)
+
+The **dashboard** (`molecule-app`) becomes the primary entry: a concierge chat (default-targeting the
+§5.4 resolver) plus a live org overview, with pending approvals surfaced inline. The **canvas** stays
+for advanced users. First UI version is produced in Claude Design and iterated before build.
+
+## 6. SSOT mapping (derive, don't fork)
+
+| Concern | Single source of truth | This RFC's rule |
+|---|---|---|
+| "The org" | `orgRootID()`/`sameOrg()` (`org_scope.go`) | platform agent *becomes* the root; no `org_id` column |
+| Platform marker | `workspaces.kind` | `kind` only; never also `role`/`tier` |
+| Model/provider | `providers.yaml` runtime matrix | concierge switches via the same registry |
+| LLM billing | `ResolveLLMBillingModeDerived` | inherits the one resolver; no new path |
+| Config/secrets delivery | tenant Secrets Manager bundle (`seedWorkspaceConfigSecret`) | no new S3 prefix / second store |
+| Management API | OpenAPI spec | new endpoints authored there first; MCP/CLI/docs derive |
+| Gated actions | `internal/approvals/policy.go` | one map |
+| Platform-agent id | `uuidv5(org, "platform-agent")` | derived, never stored separately |
+
+## 7. Security & blast radius
+
+The concierge holds the org **admin token** (full tenant-root, self-minting) and is driven by
+end-user chat. Mitigations:
+- **Approval gate (§5.8)** must ship *with* the agent going user-facing, not after. Until then the
+  agent is operator-only.
+- **Tenant isolation** is unchanged — every reach path still passes `sameOrg()`.
+- **MCP not in workspace images** (dedicated image, §5.7); the admin token lives only in the
+  platform-agent container env on the tenant box.
+- **Token rotation:** the MCP reads env once at spawn → rotation = `docker restart
+  molecule-platform-agent` (runbook item).
+- Future: a scoped-down org token (no delete/billing/member) — see §10.
+
+## 8. Migration & rollout
+
+Phase ordering is the rollout contract:
+- **Phase 0** (schema) ships and bakes before anything writes `kind`. Backward-compatible: every
+  existing row defaults to `kind='workspace'`; the `CHECK` is added `NOT VALID` then validated.
+- **Phase 1 re-parenting backfill** is the one real watch-item. **Before** running it, audit whether
+  any org-scoped table keys off the *root workspace id* (e.g. `org_api_tokens`, `org_plugin_allowlist`)
+  versus the CP org UUID. If they reference the root workspace id, re-parenting changes "the root" and
+  those refs must migrate too. The backfill is per-org, idempotent, and reversible.
+- New orgs get the platform agent from first boot; existing orgs backfill via `/admin/tenants
+  redeploy` + a one-time re-parent migration.
+
+## 9. Implementation phases
+
+0. **Schema + model** (`molecule-core`): `kind` column + `approval_requests.consumed_at`; model field +
+   constants; `Register` accepts/validates `kind` with invariants.
+1. **Platform-as-root + resolver** (`molecule-core` + CP): CP pre-seeds the platform row and creates
+   teams under it; per-org re-parent backfill (after the §8 audit); `GET /registry/platform-agent`.
+2. **Config-driven two-MCP runtime** (runtime + claude-code template).
+3. **Image + tenant provisioning** (CP + image + `molecule-ci`): dedicated image; `start_platform_agent`
+   in user-data + redeploy; config via the tenant Secrets Manager bundle; billing knob.
+4. **Approval gate** (`molecule-core`): policy map + `requireApproval` at destructive handlers; OpenAPI
+   202 shape.
+5. **Dashboard concierge UX** (`molecule-app`): design-first, then build against the resolver.
+6. **Cleanup**: exclude the platform agent from billable counts; canvas visibility; rotation runbook.
+
+## 10. Open questions
+
+- **Scoped-down token.** Should the concierge hold a reduced-scope token (no delete/billing/member)
+  instead of full admin + an approval gate? The token-scope system does not exist yet (`orgtoken`
+  TODO). Recommendation: ship admin-token + approval gate now; add scope-down as a follow-up.
+- **Re-parenting vs. wrapper.** If product later wants a platform agent that is *not* the topological
+  root, a `CanCommunicateWithKind` wrapper (guarded by `sameOrg`) is the alternative. Deferred —
+  platform-as-root is lower-risk and needs zero access-control change.
+- **Canvas visibility** of the root concierge node (hide vs. show as the org anchor).
+
+## 11. Verification (end-to-end on a staging tenant)
+
+1. **Schema:** Phase-0 migrations applied; existing workspaces report `kind='workspace'`; `go test
+   ./...` + `-tags=integration` green.
+2. **Provision:** redeploy a staging tenant; `docker ps` shows `molecule-platform-agent` healthy; its
+   logs show a successful `/registry/register`.
+3. **Identity:** the platform row is `kind='platform'`, `parent_id IS NULL`; the former root now has
+   `parent_id = <platform id>`; `GET /registry/platform-agent` returns it.
+4. **Reach:** chat the platform agent → it `list_workspaces` then `create_workspace` via the platform
+   MCP and reports back via `send_message_to_user`.
+5. **Isolation:** it reaches every workspace in its org and **cannot** reach another tenant's
+   workspace.
+6. **Approval gate:** `delete_workspace` → HTTP 202 pending + approval event; decide-approve →
+   completes; a second delete with the same approval is rejected (consumed).
+7. Drive a real concierge flow ("spin up a PM + engineer to build X") and watch the delegation/activity
+   ledger.
+
+---
+
+*Derived from a read-only multi-agent source audit of `molecule-core`, `molecule-controlplane`,
+`molecule-ai-workspace-runtime`, `molecule-ai-workspace-template-claude-code`, and
+`molecule-mcp-server`. No secret values recorded.*
@@ -19,7 +19,10 @@
 #
 # Env vars required:
 #   CF_API_TOKEN        — Cloudflare token with zone:dns:edit
+#     (falls back to CLOUDFLARE_API_TOKEN if CF_API_TOKEN is unset;
+#      the workflow YAML maps both secret names into CF_API_TOKEN)
 #   CF_ZONE_ID          — the zone (moleculesai.app)
+#     (falls back to CLOUDFLARE_ZONE_ID if CF_ZONE_ID is unset)
 #   CP_ADMIN_API_TOKEN — CP admin bearer for api.moleculesai.app
 #   CP_STAGING_ADMIN_API_TOKEN — CP admin bearer for staging-api.moleculesai.app
 #   AWS_*               — standard AWS creds (default region us-east-2)
@@ -56,6 +59,12 @@ need() {
    exit 1
  fi
 }
+# Fallback: operator-host canonical names → CI-scoped names.
+# The workflow YAML already maps both, but direct script invocation
+# (e.g. local ops) may only have the canonical names set.
+CF_API_TOKEN="${CF_API_TOKEN:-${CLOUDFLARE_API_TOKEN:-}}"
+CF_ZONE_ID="${CF_ZONE_ID:-${CLOUDFLARE_ZONE_ID:-}}"
+
 need CF_API_TOKEN
 need CF_ZONE_ID
 need CP_ADMIN_API_TOKEN
@@ -121,7 +130,7 @@ if not payload.get("success", False) or not isinstance(payload.get("result"), li
    print(f"ERROR: Cloudflare DNS list failed: {detail}", file=sys.stderr)
    raise SystemExit(1)
 '; then
-  log "Cloudflare DNS list failed; verify CF_API_TOKEN has Zone:DNS:Edit and CF_ZONE_ID is the moleculesai.app zone."
+  log "Cloudflare DNS list failed; verify CF_API_TOKEN (or CLOUDFLARE_API_TOKEN) has Zone:DNS:Edit and CF_ZONE_ID (or CLOUDFLARE_ZONE_ID) is the moleculesai.app zone."
  exit 1
 fi
 TOTAL_CF=$(echo "$CF_JSON" | python3 -c "import json,sys; print(len(json.load(sys.stdin)['result']))")
@@ -29,8 +29,11 @@
 #                          account:cloudflare_tunnel:edit scope.
 #                          (Same secret as sweep-cf-orphans, but the
 #                          token must include the tunnel scope.)
+#     (falls back to CLOUDFLARE_API_TOKEN if CF_API_TOKEN is unset;
+#      the workflow YAML maps both secret names into CF_API_TOKEN)
 #   CF_ACCOUNT_ID       — the account that owns the tunnels (visible
 #                          in dash.cloudflare.com URL path)
+#     (falls back to CLOUDFLARE_ACCOUNT_ID if CF_ACCOUNT_ID is unset)
 #   CP_ADMIN_API_TOKEN — CP admin bearer for api.moleculesai.app
 #   CP_STAGING_ADMIN_API_TOKEN — CP admin bearer for staging-api.moleculesai.app
 #
@@ -70,6 +73,12 @@ need() {
    exit 1
  fi
 }
+# Fallback: operator-host canonical names → CI-scoped names.
+# The workflow YAML already maps both, but direct script invocation
+# (e.g. local ops) may only have the canonical names set.
+CF_API_TOKEN="${CF_API_TOKEN:-${CLOUDFLARE_API_TOKEN:-}}"
+CF_ACCOUNT_ID="${CF_ACCOUNT_ID:-${CLOUDFLARE_ACCOUNT_ID:-}}"
+
 need CF_API_TOKEN
 need CF_ACCOUNT_ID
 need CP_ADMIN_API_TOKEN
@@ -0,0 +1,299 @@
+#!/usr/bin/env bash
+# cp#455 — Minimal-cell boot-to-registration harness.
+# CTO directive 14eb4f07: "build the minimal claude-code+kimi cell,
+# it should now go GREEN since the fix is live."
+#
+# Stage 1 of 5-stage rollout. Reduced to the minimum boot-to-
+# registration surface so each cell run is ~3-5 min wall-clock.
+#
+# Four assertions (per Researcher Task #79 spec):
+#   1. Provision request accepted; workspace transitions to booting/running
+#   2. Controlplane receives /registry/register for that workspace_id
+#   3. JSON-RPC/completion route returns successful minimal response
+#   4. Teardown terminates workspace even on failure (trap)
+#
+# Cost controls (mandatory):
+#   - SPOT instances (via the dispatch-only EC2 provisioning path;
+#     we don't set instance type — that's the provisioner's call)
+#   - Fast teardown ~3-5 min wall-clock
+#   - Structured per-cell results JSON output
+#
+# Auth model (mirrors test_staging_full_saas.sh):
+#   Single MOLECULE_ADMIN_TOKEN drives everything.
+#     - POST /cp/admin/orgs to provision
+#     - GET  /cp/admin/orgs/:slug/admin-token for per-tenant token
+#     - DELETE /cp/admin/tenants/:slug for teardown
+#   Per-tenant admin token drives tenant API calls (workspaces,
+#   /registry/register, JSON-RPC completion).
+#
+# Required env:
+#   MOLECULE_CP_URL        default: https://staging-api.moleculesai.app
+#   MOLECULE_ADMIN_TOKEN   CP admin bearer
+#
+# Optional env (passed from workflow_dispatch inputs):
+#   E2E_RUNTIME            default claude-code
+#   E2E_BILLING_MODE       default platform_managed
+#   E2E_PROVIDER           default platform
+#   E2E_MODEL              default moonshot/kimi-k2.6
+#   E2E_RUN_ID             Slug suffix; CI: cp455-${GITHUB_RUN_ID}
+#   E2E_PROVISION_TIMEOUT_SECS  default 300 (5 min — fast teardown budget)
+#   E2E_KEEP_ORG           1 → skip teardown (debugging only)
+#
+# Exit codes:
+#   0  happy path
+#   1  generic failure
+#   2  missing required env
+#   3  provisioning timed out (assertion 1)
+#   4  register timeout (assertion 2)
+#   5  completion failure (assertion 3)
+#   6  teardown left orphan (assertion 4)
+
+set -uo pipefail
+
+CP_URL="${MOLECULE_CP_URL:-https://staging-api.moleculesai.app}"
+ADMIN_TOKEN="${MOLECULE_ADMIN_TOKEN:?MOLECULE_ADMIN_TOKEN required — Railway staging CP_ADMIN_API_TOKEN}"
+RUNTIME="${E2E_RUNTIME:-claude-code}"
+BILLING_MODE="${E2E_BILLING_MODE:-platform_managed}"
+PROVIDER="${E2E_PROVIDER:-platform}"
+MODEL="${E2E_MODEL:-moonshot/kimi-k2.6}"
+PROVISION_TIMEOUT_SECS="${E2E_PROVISION_TIMEOUT_SECS:-300}"
+KEEP_ORG="${E2E_KEEP_ORG:-}"
+RUN_ID_SUFFIX="${E2E_RUN_ID:-$(date +%H%M%S)-$$}"
+SLUG="cp455-${RUNTIME}-${RUN_ID_SUFFIX}"
+WORKSPACE_ID=""
+TENANT_TOKEN=""
+RESULT_JSON="/tmp/cell-result.json"
+PROVISION_START_EPOCH=""
+PROVISION_END_EPOCH=""
+REGISTER_STATUS="not_attempted"
+COMPLETION_STATUS="not_attempted"
+TEARDOWN_STATUS="not_attempted"
+EXIT_CODE=0
+
+# Structured per-cell results writer. Emits JSON with all 4
+# assertion statuses + elapsed timing. Called from EXIT trap so
+# results are captured even on early failure.
+write_result() {
+  local elapsed="${1:-0}"
+  cat > "${RESULT_JSON}" <<JSON
+{
+  "runtime": "${RUNTIME}",
+  "billing_mode": "${BILLING_MODE}",
+  "provider": "${PROVIDER}",
+  "model": "${MODEL}",
+  "workspace_id": "${WORKSPACE_ID}",
+  "register_status": "${REGISTER_STATUS}",
+  "completion_status": "${COMPLETION_STATUS}",
+  "teardown_status": "${TEARDOWN_STATUS}",
+  "elapsed_seconds": ${elapsed},
+  "exit_code": ${EXIT_CODE},
+  "ts": "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+}
+JSON
+}
+
+# EXIT trap — ALWAYS run. Writes structured results, tears down
+# workspace if we have one, never lets the script exit without
+# emitting /tmp/cell-result.json.
+on_exit() {
+  local exit_code=$?
+  EXIT_CODE=${exit_code}
+  local now
+  now=$(date +%s)
+  local elapsed=0
+  if [ -n "${PROVISION_START_EPOCH:-}" ] && [ "${PROVISION_START_EPOCH}" -gt 0 ] 2>/dev/null; then
+    elapsed=$(( now - PROVISION_START_EPOCH ))
+  fi
+
+  # Assertion 4: teardown terminates workspace even on failure.
+  if [ -z "${KEEP_ORG}" ] && [ -n "${SLUG:-}" ]; then
+    if [ -n "${WORKSPACE_ID:-}" ] || [ -n "${SLUG:-}" ]; then
+      echo "::group::Teardown (trap)"
+      echo "DELETE ${CP_URL}/cp/admin/tenants/${SLUG}"
+      local teardown_http_code
+      teardown_http_code=$(curl -sS -o /dev/null -w '%{http_code}' \
+        -X DELETE \
+        -H "Authorization: Bearer ${ADMIN_TOKEN}" \
+        --max-time 60 \
+        "${CP_URL}/cp/admin/tenants/${SLUG}" || echo "000")
+      if [ "${teardown_http_code}" = "200" ] || [ "${teardown_http_code}" = "204" ] || [ "${teardown_http_code}" = "404" ]; then
+        TEARDOWN_STATUS="ok"
+        echo "Teardown OK (HTTP ${teardown_http_code})"
+      else
+        TEARDOWN_STATUS="leak_risk_http_${teardown_http_code}"
+        echo "::error::Teardown returned HTTP ${teardown_http_code} — orphan risk"
+        # Bump exit code to 6 if teardown is the failure source.
+        if [ "${EXIT_CODE}" -eq 0 ]; then
+          EXIT_CODE=6
+        fi
+      fi
+      echo "::endgroup::"
+    fi
+  else
+    TEARDOWN_STATUS="skipped_keep_org"
+  fi
+
+  write_result "${elapsed}"
+  echo "Structured results written to ${RESULT_JSON}"
+  cat "${RESULT_JSON}"
+  exit "${EXIT_CODE}"
+}
+trap on_exit EXIT
+trap 'echo "::error::Script aborted on signal"; exit 130' INT TERM
+
+PROVISION_START_EPOCH=$(date +%s)
+
+# Assertion 1: Provision request accepted; workspace transitions to
+# booting/running.
+echo "::group::Assertion 1: Provision"
+echo "POST ${CP_URL}/cp/admin/orgs  slug=${SLUG}  runtime=${RUNTIME}  billing_mode=${BILLING_MODE}  provider=${PROVIDER}  model=${MODEL}"
+PROVISION_HTTP_CODE=$(curl -sS -o /tmp/provision-resp.json -w '%{http_code}' \
+  -X POST \
+  -H "Authorization: Bearer ${ADMIN_TOKEN}" \
+  -H "Content-Type: application/json" \
+  --max-time 30 \
+  -d "$(cat <<JSON
+{
+  "slug": "${SLUG}",
+  "runtime": "${RUNTIME}",
+  "billing_mode": "${BILLING_MODE}",
+  "provider": "${PROVIDER}",
+  "model": "${MODEL}",
+  "tier": "spot",
+  "tags": {
+    "cp455_minimal_cell": "1",
+    "run_id": "${RUN_ID_SUFFIX}"
+  }
+}
+JSON
+)" \
+  "${CP_URL}/cp/admin/orgs" || echo "000")
+echo "HTTP ${PROVISION_HTTP_CODE}"
+if [ "${PROVISION_HTTP_CODE}" != "202" ] && [ "${PROVISION_HTTP_CODE}" != "200" ]; then
+  echo "::error::Provision failed (HTTP ${PROVISION_HTTP_CODE})"
+  cat /tmp/provision-resp.json 2>/dev/null || true
+  EXIT_CODE=1
+  exit "${EXIT_CODE}"
+fi
+echo "::endgroup::"
+
+# Wait for org to reach running + retrieve per-tenant token. Bounded
+# at PROVISION_TIMEOUT_SECS. We poll the admin token endpoint; once
+# the org is up, the endpoint returns 200 with the token, and the
+# workspace_id is in the same response or in a follow-up /orgs/:slug
+# call.
+echo "::group::Wait for org to be ready (max ${PROVISION_TIMEOUT_SECS}s)"
+WAIT_START=$(date +%s)
+WAIT_DEADLINE=$(( WAIT_START + PROVISION_TIMEOUT_SECS ))
+TENANT_TOKEN=""
+while [ "$(date +%s)" -lt "${WAIT_DEADLINE}" ]; do
+  TOKEN_HTTP_CODE=$(curl -sS -o /tmp/token-resp.json -w '%{http_code}' \
+    -H "Authorization: Bearer ${ADMIN_TOKEN}" \
+    --max-time 10 \
+    "${CP_URL}/cp/admin/orgs/${SLUG}/admin-token" || echo "000")
+  if [ "${TOKEN_HTTP_CODE}" = "200" ]; then
+    TENANT_TOKEN=$(jq -r '.admin_token // .token // empty' /tmp/token-resp.json 2>/dev/null || echo "")
+    if [ -n "${TENANT_TOKEN}" ]; then
+      WORKSPACE_ID=$(jq -r '.workspace_id // .default_workspace_id // empty' /tmp/token-resp.json 2>/dev/null || echo "")
+      if [ -z "${WORKSPACE_ID}" ]; then
+        # Fallback: list orgs and find by slug
+        WORKSPACE_ID=$(curl -sS -H "Authorization: Bearer ${ADMIN_TOKEN}" \
+          "${CP_URL}/cp/admin/orgs/${SLUG}" | jq -r '.workspace_id // .default_workspace_id // empty' 2>/dev/null || echo "")
+      fi
+      if [ -n "${WORKSPACE_ID}" ]; then
+        PROVISION_END_EPOCH=$(date +%s)
+        echo "Org ready in $(( PROVISION_END_EPOCH - WAIT_START ))s — workspace_id=${WORKSPACE_ID}"
+        break
+      fi
+    fi
+  fi
+  sleep 5
+done
+if [ -z "${TENANT_TOKEN}" ] || [ -z "${WORKSPACE_ID}" ]; then
+  echo "::error::Provision timed out (org never reached running within ${PROVISION_TIMEOUT_SECS}s)"
+  EXIT_CODE=3
+  exit "${EXIT_CODE}"
+fi
+echo "::endgroup::"
+
+# Assertion 2: Controlplane receives /registry/register for that
+# workspace_id. The harness doesn't POST to /registry/register
+# directly — that's the workspace-server's own job on boot. We
+# verify the registration was received by polling the registry
+# endpoint (or by checking that a /workspaces/:id call returns
+# the expected fields).
+echo "::group::Assertion 2: /registry/register for workspace_id=${WORKSPACE_ID}"
+REGISTER_DEADLINE=$(( $(date +%s) + 60 ))
+while [ "$(date +%s)" -lt "${REGISTER_DEADLINE}" ]; do
+  REG_HTTP_CODE=$(curl -sS -o /tmp/reg-resp.json -w '%{http_code}' \
+    -H "Authorization: Bearer ${TENANT_TOKEN}" \
+    --max-time 10 \
+    "${CP_URL}/cp/registry/workspaces/${WORKSPACE_ID}" || echo "000")
+  if [ "${REG_HTTP_CODE}" = "200" ]; then
+    REGISTERED=$(jq -r '.registered // .workspace_id // empty' /tmp/reg-resp.json 2>/dev/null || echo "")
+    if [ -n "${REGISTERED}" ]; then
+      REGISTER_STATUS="ok"
+      echo "Registry confirms workspace_id=${WORKSPACE_ID} registered"
+      break
+    fi
+  fi
+  sleep 3
+done
+if [ "${REGISTER_STATUS}" != "ok" ]; then
+  echo "::error::Registry did not confirm registration within 60s"
+  cat /tmp/reg-resp.json 2>/dev/null || true
+  EXIT_CODE=4
+  exit "${EXIT_CODE}"
+fi
+echo "::endgroup::"
+
+# Assertion 3: JSON-RPC/completion route returns successful minimal
+# response. One minimal completion call — keep payload small.
+echo "::group::Assertion 3: JSON-RPC completion"
+COMPLETION_HTTP_CODE=$(curl -sS -o /tmp/completion-resp.json -w '%{http_code}' \
+  -X POST \
+  -H "Authorization: Bearer ${TENANT_TOKEN}" \
+  -H "Content-Type: application/json" \
+  --max-time 30 \
+  -d "$(cat <<JSON
+{
+  "jsonrpc": "2.0",
+  "id": 1,
+  "method": "completion",
+  "params": {
+    "workspace_id": "${WORKSPACE_ID}",
+    "model": "${MODEL}",
+    "messages": [{"role": "user", "content": "ping"}],
+    "max_tokens": 1
+  }
+}
+JSON
+)" \
+  "${CP_URL}/cp/rpc" || echo "000")
+echo "HTTP ${COMPLETION_HTTP_CODE}"
+if [ "${COMPLETION_HTTP_CODE}" != "200" ]; then
+  echo "::error::Completion failed (HTTP ${COMPLETION_HTTP_CODE})"
+  cat /tmp/completion-resp.json 2>/dev/null || true
+  EXIT_CODE=5
+  exit "${EXIT_CODE}"
+fi
+# Verify JSON-RPC 2.0 success envelope
+RPC_ERROR=$(jq -r '.error // empty' /tmp/completion-resp.json 2>/dev/null || echo "")
+if [ -n "${RPC_ERROR}" ]; then
+  echo "::error::Completion returned JSON-RPC error: ${RPC_ERROR}"
+  cat /tmp/completion-resp.json 2>/dev/null || true
+  EXIT_CODE=5
+  exit "${EXIT_CODE}"
+fi
+RPC_RESULT=$(jq -r '.result // empty' /tmp/completion-resp.json 2>/dev/null || echo "")
+if [ -z "${RPC_RESULT}" ] || [ "${RPC_RESULT}" = "null" ]; then
+  echo "::error::Completion response missing result field"
+  cat /tmp/completion-resp.json 2>/dev/null || true
+  EXIT_CODE=5
+  exit "${EXIT_CODE}"
+fi
+COMPLETION_STATUS="ok"
+echo "Completion OK"
+echo "::endgroup::"
+
+echo "All 4 assertions passed for ${SLUG} (workspace_id=${WORKSPACE_ID})"
@@ -53,7 +53,9 @@
 #   PV_RUNTIMES            space list; default "hermes openclaw claude-code"
 #   E2E_PROVISION_TIMEOUT_SECS  default 1800 (hermes/openclaw cold EC2 budget)
 #   E2E_MINIMAX_API_KEY / E2E_ANTHROPIC_API_KEY / E2E_OPENAI_API_KEY
-#                          LLM provider key injected so the runtime can boot
+#                          DEPRECATED for this script — platform-managed models
+#                          use the CP LLM proxy; direct vendor keys are blocked
+#                          by PR #2291. Kept in workflow env for other E2Es.
 #   PV_TOKEN_DIAGNOSTIC_ONLY
 #                          1 -> stop after create/token acquisition. Useful
 #                          to classify Hermes-only vs shared auth-route issues.
@@ -222,17 +224,14 @@ else
 fi

 # ─── 4. Provision the parent + one sibling per runtime under test ──────
-# Inject the LLM provider key so each runtime can authenticate at boot.
-# Priority: MiniMax → direct-Anthropic → OpenAI (mirrors
-# test_staging_full_saas.sh's secrets-injection chain).
+# Platform-managed models: Molecule owns billing via the CP LLM proxy, so
+# the workspace needs NO tenant key. PR #2291 blocks direct vendor key writes
+# (ANTHROPIC_API_KEY, ANTHROPIC_AUTH_TOKEN, MINIMAX_API_KEY, etc.) for
+# platform-managed workspaces. We intentionally keep SECRETS_JSON empty so a
+# stray E2E_*_API_KEY in the runner env cannot silently convert this into a
+# BYOK run and mask the platform-managed path (mirrors
+# test_staging_full_saas.sh's E2E_LLM_PATH=platform branch).
 SECRETS_JSON='{}'
-if [ -n "${E2E_MINIMAX_API_KEY:-}" ]; then
-  SECRETS_JSON=$(python3 -c "import json,os;k=os.environ['E2E_MINIMAX_API_KEY'];print(json.dumps({'ANTHROPIC_BASE_URL':'https://api.minimax.io/anthropic','ANTHROPIC_AUTH_TOKEN':k,'MINIMAX_API_KEY':k}))")
-elif [ -n "${E2E_ANTHROPIC_API_KEY:-}" ]; then
-  SECRETS_JSON=$(python3 -c "import json,os;k=os.environ['E2E_ANTHROPIC_API_KEY'];print(json.dumps({'ANTHROPIC_API_KEY':k}))")
-elif [ -n "${E2E_OPENAI_API_KEY:-}" ]; then
-  SECRETS_JSON=$(python3 -c "import json,os;k=os.environ['E2E_OPENAI_API_KEY'];print(json.dumps({'OPENAI_API_KEY':k,'OPENAI_BASE_URL':'https://api.openai.com/v1','MODEL_PROVIDER':'openai:gpt-4o','HERMES_INFERENCE_PROVIDER':'custom','HERMES_CUSTOM_BASE_URL':'https://api.openai.com/v1','HERMES_CUSTOM_API_KEY':k,'HERMES_CUSTOM_API_MODE':'chat_completions'}))")
-fi

 # Workspace-create now enforces the MODEL_REQUIRED contract: there is NO
 # platform-side default model for a runtime (feedback_workspace_model_required_
@@ -584,6 +584,54 @@ def test_find_open_issue_raises_on_transient_error(drift_module, monkeypatch):
        drift_module.find_open_issue("[ci-drift] foo")


+# --------------------------------------------------------------------------
+# Pagination: search beyond page 1 so an existing issue on any page is found
+# --------------------------------------------------------------------------
+def test_find_open_issue_paginates_to_page_2(drift_module, monkeypatch):
+    """Issue exists on page 2 → paginate and find it."""
+    target = {"number": 99, "title": "[ci-drift] foo"}
+    filler = [{"number": i, "title": f"other-{i}"} for i in range(1, 51)]
+
+    class PaginatedStub:
+        def __init__(self):
+            self.calls = []
+
+        def __call__(self, method, path, *, body=None, query=None, expect_json=True):
+            self.calls.append((method, path, body, query))
+            page = int((query or {}).get("page", "1"))
+            if page == 1:
+                return 200, filler
+            if page == 2:
+                return 200, [target]
+            return 200, []
+
+    stub = PaginatedStub()
+    monkeypatch.setattr(drift_module, "api", stub)
+    assert drift_module.find_open_issue("[ci-drift] foo") == target
+    assert len(stub.calls) == 2
+
+
+def test_find_open_issue_stops_at_last_page(drift_module, monkeypatch):
+    """No match across pages → stop when a page has <50 results."""
+    filler = [{"number": i, "title": f"other-{i}"} for i in range(1, 51)]
+
+    class PaginatedStub:
+        def __init__(self):
+            self.calls = []
+
+        def __call__(self, method, path, *, body=None, query=None, expect_json=True):
+            self.calls.append((method, path, body, query))
+            page = int((query or {}).get("page", "1"))
+            if page == 1:
+                return 200, filler
+            return 200, []
+
+    stub = PaginatedStub()
+    monkeypatch.setattr(drift_module, "api", stub)
+    assert drift_module.find_open_issue("[ci-drift] foo") is None
+    assert len(stub.calls) == 2
+
+
 # --------------------------------------------------------------------------
 # Idempotent path: existing issue is PATCHed, NOT duplicated
 # --------------------------------------------------------------------------
@@ -1050,12 +1050,13 @@ def test_reap_continues_on_per_sha_apierror(sr_module, monkeypatch, capsys):


 def test_main_soft_skips_when_commit_listing_times_out(sr_module, monkeypatch, capsys):
-    """A transient outage while listing recent commits should not paint main red.
+    """A transient outage while listing recent commits fails the tick visibly.

    Per-SHA status read failures are already isolated inside `reap_branch`.
    The real 2026-05-14 failure was earlier: `/commits?sha=main&limit=30`
    timed out after all retries, aborting the tick. The next 5-minute tick can
-    retry safely, so `main()` should emit an observable warning and return 0.
+    retry safely, but the tick itself must be observable as red (exit 1 + error
+    annotation) so the cron bot alerts on persistent infra issues.
    """

    monkeypatch.setattr(sr_module, "scan_workflows", lambda _: {"workflow-without-push": False})
@@ -1068,9 +1069,9 @@ def test_main_soft_skips_when_commit_listing_times_out(sr_module, monkeypatch, c
    monkeypatch.setattr(sr_module, "list_recent_commit_shas", fake_list_recent_commit_shas)
    monkeypatch.setattr(sys, "argv", ["status-reaper.py"])

-    assert sr_module.main() == 0
+    assert sr_module.main() == 1
    captured = capsys.readouterr()
-    assert "::warning::status-reaper skipped this tick" in captured.out
+    assert "::error::status-reaper cannot run" in captured.out
    assert '"skipped": true' in captured.out
    assert '"skip_reason": "commit-list-api-error"' in captured.out

@@ -351,8 +351,17 @@ func main() {
 	// (true, err) on any transient error, so a CP blip never flips a healthy
 	// workspace.
 	if cpProv != nil {
+		// Guard against double-reprovision thrash (internal#544): the restart
+		// debounce window must cover the reconciler interval so a workspace
+		// flipped offline by one reconcile tick isn't immediately reprovisioned
+		// again by the next tick before the debounce drops it. If the interval
+		// ever shrinks below the debounce window, the coupling silently breaks.
+		reconcileInterval := 60 * time.Second
+		if handlers.RestartDebounceWindow < reconcileInterval {
+			log.Fatalf("RestartDebounceWindow (%s) must be >= CP instance reconciler interval (%s) to prevent double-reprovision thrash (internal#544)", handlers.RestartDebounceWindow, reconcileInterval)
+		}
 		go supervised.RunWithRecover(ctx, "cp-instance-reconciler", func(c context.Context) {
-			registry.StartCPInstanceReconciler(c, cpProv, onWorkspaceOffline, 60*time.Second)
+			registry.StartCPInstanceReconciler(c, cpProv, onWorkspaceOffline, reconcileInterval)
 		})
 	}

@@ -0,0 +1,39 @@
+// Package approvals holds the single source of truth for which destructive
+// org operations require a human approval before they execute.
+//
+// (RFC docs/design/rfc-platform-agent.md — Phase 4)
+//
+// The org-level platform agent is driven by end-user chat and holds an org-admin
+// token, so destructive/irreversible operations it can trigger are gated: the
+// handler creates a pending approval and returns it instead of executing, and a
+// human decides via the existing approvals subsystem. Keeping the gated-action
+// list in ONE map makes the blast-radius boundary auditable in a single place —
+// a handler not listed here behaves exactly as before.
+package approvals
+
+// Action is the canonical identifier of a gated destructive operation. The same
+// string is stored in approval_requests.action so the gate can match a pending/
+// approved request to the operation being retried.
+type Action string
+
+const (
+	ActionDeleteWorkspace Action = "delete_workspace"
+	ActionDeprovision     Action = "deprovision_workspace"
+	ActionSecretWrite     Action = "secret_write"
+	ActionOrgTokenMint    Action = "org_token_mint"
+)
+
+// gated is the set of actions that require a human approval. Add an entry here
+// (and gate the corresponding handler with requireApproval) to expand the
+// boundary; remove one to drop a gate. This is the only place the policy lives.
+var gated = map[Action]bool{
+	ActionDeleteWorkspace: true,
+	ActionDeprovision:     true,
+	ActionSecretWrite:     true,
+	ActionOrgTokenMint:    true,
+}
+
+// IsGated reports whether action requires a human approval before executing.
+func IsGated(action Action) bool {
+	return gated[action]
+}
@@ -271,6 +271,11 @@ func (m *Manager) Reload(ctx context.Context) {
 		ch.Config["_channel_id"] = ch.ID

 		go func(a ChannelAdapter, c ChannelRow, pCtx context.Context) {
+			defer func() {
+				if r := recover(); r != nil {
+					log.Printf("PANIC recovered in channel polling goroutine: %v", r)
+				}
+			}()
 			if err := a.StartPolling(pCtx, c.Config, m.onInboundMessage); err != nil {
 				log.Printf("Channels: polling error for %s/%s: %v", c.ChannelType, truncID(c.ID), err)
 			}
@@ -354,6 +359,11 @@ func (m *Manager) HandleInbound(ctx context.Context, ch ChannelRow, msg *Inbound
 			typingCtx, typingCancel := context.WithCancel(fireCtx)
 			defer typingCancel()
 			go func() {
+				defer func() {
+					if r := recover(); r != nil {
+						log.Printf("PANIC recovered in typing indicator goroutine: %v", r)
+					}
+				}()
 				typer.SendTyping(ch.Config, msg.ChatID)
 				ticker := time.NewTicker(4 * time.Second)
 				defer ticker.Stop()
@@ -142,7 +142,7 @@ func ghcrAuthHeader() string {
 		log.Printf("workspace-images: failed to marshal GHCR auth: %v", err)
 		return ""
 	}
-	return base64.URLEncoding.EncodeToString(js)
+	return base64.StdEncoding.EncodeToString(js)
 }

 // Refresh pulls the requested runtimes' template images from GHCR and (if
@@ -47,9 +47,9 @@ func TestGHCRAuthHeader_EncodesDockerEnginePayload(t *testing.T) {
 	if got == "" {
 		t.Fatal("expected non-empty auth header")
 	}
-	raw, err := base64.URLEncoding.DecodeString(got)
+	raw, err := base64.StdEncoding.DecodeString(got)
 	if err != nil {
-		t.Fatalf("auth header is not valid base64-url: %v", err)
+		t.Fatalf("auth header is not valid base64: %v", err)
 	}
 	var payload map[string]string
 	if err := json.Unmarshal(raw, &payload); err != nil {
@@ -80,9 +80,9 @@ func TestGHCRAuthHeader_RespectsRegistryEnv(t *testing.T) {
 	if got == "" {
 		t.Fatal("expected non-empty auth header")
 	}
-	raw, err := base64.URLEncoding.DecodeString(got)
+	raw, err := base64.StdEncoding.DecodeString(got)
 	if err != nil {
-		t.Fatalf("auth header is not valid base64-url: %v", err)
+		t.Fatalf("auth header is not valid base64: %v", err)
 	}
 	var payload map[string]string
 	if err := json.Unmarshal(raw, &payload); err != nil {
@@ -220,7 +220,7 @@ func TestGHCRAuthHeader_TrimsWhitespace(t *testing.T) {
 	t.Setenv("GHCR_USER", "  alice  ")
 	t.Setenv("GHCR_TOKEN", "\tfake-tok-value\n")
 	got := ghcrAuthHeader()
-	raw, _ := base64.URLEncoding.DecodeString(got)
+	raw, _ := base64.StdEncoding.DecodeString(got)
 	var payload map[string]string
 	_ = json.Unmarshal(raw, &payload)
 	if payload["username"] != "alice" {
@@ -0,0 +1,196 @@
+package handlers
+
+// approval_gate.go — server-side gate for destructive org operations.
+// (RFC docs/design/rfc-platform-agent.md — Phase 4)
+//
+// requireApproval is the choke point a destructive handler calls before
+// executing. It is the trust boundary: the platform-management MCP is a CLIENT
+// of these handlers, so enforcing here (not in the MCP) means anything holding
+// an org-admin token still goes through the gate. The flow:
+//
+//   - if a matching APPROVED + unconsumed approval exists, consume it (single-
+//     use) and let the operation proceed;
+//   - otherwise create (or reuse) a PENDING approval, broadcast it to the canvas
+//     (and escalate to the parent if any), and the handler returns HTTP 202 so a
+//     human can decide. The agent retries after approval and the gate passes.
+//
+// Matching is by (workspace_id, action, request_hash) where request_hash is a
+// stable digest of the operation + its context, so a retried op reuses its own
+// request instead of flooding the table, and an approval for "delete ws A"
+// cannot be replayed to "delete ws B".
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/approvals"
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/db"
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/events"
+	"github.com/gin-gonic/gin"
+)
+
+// approvalRequestHash is a stable digest of the gated operation. Go's
+// json.Marshal sorts map keys, so the same context always hashes the same.
+func approvalRequestHash(workspaceID, action string, contextMap map[string]interface{}) string {
+	cj, err := json.Marshal(contextMap)
+	if err != nil || cj == nil {
+		cj = []byte("{}")
+	}
+	sum := sha256.Sum256([]byte(workspaceID + "\x00" + action + "\x00" + string(cj)))
+	return hex.EncodeToString(sum[:])
+}
+
+// requireApproval returns (approved=true, consumedID) when a matching approval
+// exists and was just consumed; otherwise it creates/reuses a pending approval
+// and returns (false, pendingID). A non-nil error is a server error.
+func requireApproval(ctx context.Context, b events.EventEmitter, workspaceID string, action approvals.Action, reason string, contextMap map[string]interface{}) (bool, string, error) {
+	hash := approvalRequestHash(workspaceID, string(action), contextMap)
+
+	// 1. Atomically consume an approved + unconsumed request, if one exists.
+	//    The conditional UPDATE ... RETURNING makes consumption race-safe: two
+	//    concurrent destructive calls cannot both consume the same approval.
+	var consumedID string
+	err := db.DB.QueryRowContext(ctx, `
+		UPDATE approval_requests SET consumed_at = now()
+		WHERE id = (
+			SELECT id FROM approval_requests
+			WHERE workspace_id = $1 AND action = $2 AND request_hash = $3
+			  AND status = 'approved' AND consumed_at IS NULL
+			ORDER BY decided_at DESC NULLS LAST
+			LIMIT 1
+			FOR UPDATE SKIP LOCKED
+		)
+		RETURNING id
+	`, workspaceID, string(action), hash).Scan(&consumedID)
+	if err == nil {
+		return true, consumedID, nil
+	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		return false, "", fmt.Errorf("consume approval: %w", err)
+	}
+
+	// 2. No usable approval — create a pending one, or reuse an existing pending
+	//    request for the same operation so retries don't flood the table.
+	cj, mErr := json.Marshal(contextMap)
+	if mErr != nil || cj == nil {
+		cj = []byte("{}")
+	}
+	var approvalID string
+	err = db.DB.QueryRowContext(ctx, `
+		WITH existing AS (
+			SELECT id FROM approval_requests
+			WHERE workspace_id = $1 AND action = $2 AND request_hash = $3 AND status = 'pending'
+			LIMIT 1
+		), ins AS (
+			INSERT INTO approval_requests (workspace_id, action, reason, context, request_hash)
+			SELECT $1, $2, $4, $5::jsonb, $3
+			WHERE NOT EXISTS (SELECT 1 FROM existing)
+			RETURNING id
+		)
+		SELECT id FROM ins UNION ALL SELECT id FROM existing LIMIT 1
+	`, workspaceID, string(action), hash, reason, string(cj)).Scan(&approvalID)
+	if err != nil {
+		return false, "", fmt.Errorf("create approval: %w", err)
+	}
+
+	// Broadcast to the canvas (the user-facing signal). For a platform agent the
+	// parent_id is NULL, so the requested-event on its own workspace IS the user
+	// prompt; ordinary workspaces also escalate to their parent.
+	//
+	// b may be nil: stateless handlers (e.g. org-token mint — OrgTokenHandler is
+	// an empty struct with no broadcaster) still gate; they just can't push a
+	// live canvas event. The pending approval row is persisted regardless, so
+	// the request is never lost — only the notification is skipped.
+	if b != nil {
+		if bErr := b.RecordAndBroadcast(ctx, string(events.EventApprovalRequested), workspaceID, map[string]interface{}{
+			"approval_id": approvalID,
+			"action":      string(action),
+			"reason":      reason,
+		}); bErr != nil {
+			log.Printf("approval_gate: broadcast requested failed (ws=%s): %v", workspaceID, bErr)
+		}
+	}
+	var parentID *string
+	if pErr := db.DB.QueryRowContext(ctx, `SELECT parent_id FROM workspaces WHERE id = $1`, workspaceID).Scan(&parentID); pErr != nil {
+		log.Printf("approval_gate: parent lookup failed (ws=%s): %v", workspaceID, pErr)
+	}
+	if parentID != nil && b != nil {
+		if bErr := b.RecordAndBroadcast(ctx, string(events.EventApprovalEscalated), *parentID, map[string]interface{}{
+			"approval_id":       approvalID,
+			"from_workspace_id": workspaceID,
+			"action":            string(action),
+			"reason":            reason,
+		}); bErr != nil {
+			log.Printf("approval_gate: broadcast escalated failed (ws=%s): %v", workspaceID, bErr)
+		}
+	}
+	return false, approvalID, nil
+}
+
+// gateDestructive runs requireApproval for a gated action and, when approval is
+// still pending, writes the 202 response and returns false (caller must stop).
+// Returns true when the caller may proceed (action consumed an approval).
+func gateDestructive(c *gin.Context, b events.EventEmitter, workspaceID string, action approvals.Action, reason string, contextMap map[string]interface{}) bool {
+	if !approvals.IsGated(action) {
+		return true
+	}
+	// Scope (RFC platform-agent Phase 4b). Wiring is a one-liner in each
+	// destructive handler; the activation policy lives here, centrally, so it is
+	// uniform and testable:
+	//   - default-OFF rollout flag, so the wiring is inert until an operator
+	//     enables it (mirrors the 3a/3c default-off design and protects existing
+	//     org-token automation from a surprise async-approval behaviour change);
+	//   - only callers holding an ORG token are gated. The platform agent runs
+	//     with MOLECULE_API_KEY=<org-admin token>, so the auth middleware sets
+	//     org_token_id. Ordinary workspace-token agents and human CP-session
+	//     operators (cp_session_actor — the approvers themselves) are NOT gated,
+	//     so normal operation is byte-identical. This realises the file-header
+	//     trust boundary ("anything holding an org-admin token still goes
+	//     through the gate") without gating everyone.
+	if !destructiveGateEnabled() || !callerHoldsOrgToken(c) {
+		return true
+	}
+	approved, approvalID, err := requireApproval(c.Request.Context(), b, workspaceID, action, reason, contextMap)
+	if err != nil {
+		log.Printf("gateDestructive: %v (ws=%s action=%s)", err, workspaceID, action)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "approval gate failed"})
+		return false
+	}
+	if !approved {
+		c.JSON(http.StatusAccepted, gin.H{
+			"status":      "pending_approval",
+			"approval_id": approvalID,
+			"action":      string(action),
+			"reason":      reason,
+		})
+		return false
+	}
+	return true
+}
+
+// destructiveGateEnabled is the default-off rollout flag for the org-level
+// destructive-op approval gate. Inert until an operator sets
+// MOLECULE_PLATFORM_APPROVAL_GATE=1 (or "true") — typically when the platform
+// agent is deployed to the org. Keeps 4b's wiring shipped-but-dormant, matching
+// the platform-agent feature's default-off posture (3a/3c).
+func destructiveGateEnabled() bool {
+	v := os.Getenv("MOLECULE_PLATFORM_APPROVAL_GATE")
+	return v == "1" || v == "true"
+}
+
+// callerHoldsOrgToken reports whether the request authenticated with an org
+// token (the auth middleware sets org_token_id, see middleware/wsauth_middleware.go).
+// The platform agent uses an org-admin token; ordinary workspace-token agents
+// and human CP sessions do not, so they bypass the gate entirely.
+func callerHoldsOrgToken(c *gin.Context) bool {
+	_, ok := c.Get("org_token_id")
+	return ok
+}
@@ -0,0 +1,137 @@
+//go:build integration
+// +build integration
+
+// approval_gate_integration_test.go — REAL Postgres gate for requireApproval.
+//
+// Run with:
+//
+//	INTEGRATION_DB_URL="postgres://postgres:test@localhost:55432/molecule?sslmode=disable" \
+//	  go test -tags=integration ./internal/handlers/ -run Integration_RequireApproval -v
+//
+// Why this is NOT a sqlmock test
+// ------------------------------
+// The whole gate is about row state across calls: a pending request is created
+// once and reused (dedup), an approval is consumed exactly once (single-use via
+// the conditional UPDATE ... RETURNING), and a different operation context hashes
+// to a different request. sqlmock returns whatever the stub says; only a real
+// Postgres proves the consume-once semantics and the partial-index lookup.
+
+package handlers
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/approvals"
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/db"
+	"github.com/google/uuid"
+	_ "github.com/lib/pq"
+)
+
+func TestIntegration_RequireApproval_GateCycle(t *testing.T) {
+	url := requireIntegrationDBURL(t)
+	conn, err := sql.Open("postgres", url)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if err := conn.Ping(); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+
+	// requireApproval + the broadcaster's structure_events write use the db.DB
+	// global; point it at the integration DB and restore afterwards.
+	prev := db.DB
+	db.DB = conn
+	t.Cleanup(func() { db.DB = prev })
+	setupTestRedis(t) // broadcaster publishes to db.RDB; miniredis backs it
+
+	ctx := context.Background()
+	b := newTestBroadcaster()
+
+	wsID := uuid.New().String()
+	t.Cleanup(func() {
+		_, _ = conn.ExecContext(ctx, `DELETE FROM approval_requests WHERE workspace_id = $1`, wsID)
+		_, _ = conn.ExecContext(ctx, `DELETE FROM workspaces WHERE id = $1`, wsID)
+	})
+	// A root workspace (parent_id NULL) — like the platform agent, it has no
+	// parent, so the gate's escalation target is the user/canvas. (This branch
+	// is off main and has no kind column; the gate is kind-agnostic.)
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO workspaces (id, name, tier, status, runtime, parent_id)
+		VALUES ($1, 'Org Concierge', 0, 'online', 'claude-code', NULL)`, wsID); err != nil {
+		t.Fatalf("seed root workspace: %v", err)
+	}
+
+	action := approvals.ActionDeleteWorkspace
+	ctxA := map[string]interface{}{"target": "ws-A"}
+
+	// 1. First call → no approval yet → pending created.
+	ok, id1, err := requireApproval(ctx, b, wsID, action, "delete ws-A", ctxA)
+	if err != nil {
+		t.Fatalf("call 1: %v", err)
+	}
+	if ok {
+		t.Fatal("call 1: approved=true, want false (no approval exists yet)")
+	}
+
+	// 2. Same operation again → must REUSE the same pending row (dedup), not flood.
+	ok, id2, err := requireApproval(ctx, b, wsID, action, "delete ws-A", ctxA)
+	if err != nil {
+		t.Fatalf("call 2: %v", err)
+	}
+	if ok || id2 != id1 {
+		t.Fatalf("call 2: ok=%v id2=%s, want false and id2==id1(%s) (dedup)", ok, id2, id1)
+	}
+	var nPending int
+	if err := conn.QueryRowContext(ctx,
+		`SELECT count(*) FROM approval_requests WHERE workspace_id=$1 AND status='pending'`, wsID).Scan(&nPending); err != nil {
+		t.Fatalf("count pending: %v", err)
+	}
+	if nPending != 1 {
+		t.Fatalf("pending rows = %d, want 1 (dedup must not flood)", nPending)
+	}
+
+	// 3. A human approves it (simulating the Decide handler).
+	if _, err := conn.ExecContext(ctx,
+		`UPDATE approval_requests SET status='approved', decided_by='human', decided_at=now() WHERE id=$1`, id1); err != nil {
+		t.Fatalf("approve: %v", err)
+	}
+
+	// 4. Now the gate consumes the approval and lets the op proceed.
+	ok, consumedID, err := requireApproval(ctx, b, wsID, action, "delete ws-A", ctxA)
+	if err != nil {
+		t.Fatalf("call 4: %v", err)
+	}
+	if !ok || consumedID != id1 {
+		t.Fatalf("call 4: ok=%v consumedID=%s, want true and id1(%s)", ok, consumedID, id1)
+	}
+
+	// 5. Single-use: the SAME approval cannot be replayed — the next call is
+	//    pending again (a fresh request), not approved.
+	ok, id5, err := requireApproval(ctx, b, wsID, action, "delete ws-A", ctxA)
+	if err != nil {
+		t.Fatalf("call 5: %v", err)
+	}
+	if ok {
+		t.Fatal("call 5: approved=true — a consumed approval was replayed")
+	}
+	if id5 == id1 {
+		t.Fatal("call 5: reused the consumed request id; want a new pending request")
+	}
+
+	// 6. Context isolation: an approval for ws-A must not authorize ws-B.
+	//    Approve the ws-A request, then a ws-B op must still be pending.
+	if _, err := conn.ExecContext(ctx,
+		`UPDATE approval_requests SET status='approved', decided_at=now() WHERE id=$1`, id5); err != nil {
+		t.Fatalf("approve id5: %v", err)
+	}
+	ok, _, err = requireApproval(ctx, b, wsID, action, "delete ws-B", map[string]interface{}{"target": "ws-B"})
+	if err != nil {
+		t.Fatalf("call 6: %v", err)
+	}
+	if ok {
+		t.Fatal("call 6: ws-B proceeded on a ws-A approval — context isolation broken")
+	}
+}
@@ -0,0 +1,76 @@
+package handlers
+
+// Phase 4b — unit coverage for the gate's activation SCOPE: the default-off
+// rollout flag + org-token-only targeting. These exercise the short-circuit
+// paths that return "proceed" BEFORE requireApproval, so they need no DB. The
+// full flag-on + org-token + gated → 202 path is covered by the real-Postgres
+// approval_gate_integration_test.go.
+
+import (
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/approvals"
+	"github.com/gin-gonic/gin"
+)
+
+func TestDestructiveGateEnabled_DefaultOff(t *testing.T) {
+	os.Unsetenv("MOLECULE_PLATFORM_APPROVAL_GATE")
+	if destructiveGateEnabled() {
+		t.Fatal("gate must be OFF by default (no env)")
+	}
+	for _, v := range []string{"1", "true"} {
+		t.Setenv("MOLECULE_PLATFORM_APPROVAL_GATE", v)
+		if !destructiveGateEnabled() {
+			t.Errorf("%q must enable the gate", v)
+		}
+	}
+	t.Setenv("MOLECULE_PLATFORM_APPROVAL_GATE", "0")
+	if destructiveGateEnabled() {
+		t.Error(`"0" must keep the gate off`)
+	}
+}
+
+func TestCallerHoldsOrgToken(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	c, _ := gin.CreateTestContext(httptest.NewRecorder())
+	if callerHoldsOrgToken(c) {
+		t.Error("no org_token_id in context → must be false (workspace/CP caller)")
+	}
+	c.Set("org_token_id", "tok-abc")
+	if !callerHoldsOrgToken(c) {
+		t.Error("org_token_id set → must be true (platform-agent / org-admin caller)")
+	}
+}
+
+// gateDestructive must return true (proceed, no 202, no DB touch) whenever the
+// scope excludes the call: non-gated action, flag off, or non-org-token caller.
+func TestGateDestructive_ScopeShortCircuits(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	newCtx := func(orgToken bool) *gin.Context {
+		c, _ := gin.CreateTestContext(httptest.NewRecorder())
+		c.Request = httptest.NewRequest("DELETE", "/x", nil)
+		if orgToken {
+			c.Set("org_token_id", "tok")
+		}
+		return c
+	}
+
+	// flag OFF (default) + org-token + gated action → proceed.
+	os.Unsetenv("MOLECULE_PLATFORM_APPROVAL_GATE")
+	if !gateDestructive(newCtx(true), nil, "ws", approvals.ActionDeleteWorkspace, "r", nil) {
+		t.Error("flag off must proceed (gate dormant)")
+	}
+
+	// flag ON + NO org token (workspace agent / human CP session) → proceed.
+	t.Setenv("MOLECULE_PLATFORM_APPROVAL_GATE", "1")
+	if !gateDestructive(newCtx(false), nil, "ws", approvals.ActionDeleteWorkspace, "r", nil) {
+		t.Error("non-org-token caller must proceed (normal operation unchanged)")
+	}
+
+	// flag ON + org token + NON-gated action → proceed (IsGated short-circuit).
+	if !gateDestructive(newCtx(true), nil, "ws", approvals.Action("not_a_gated_action"), "r", nil) {
+		t.Error("non-gated action must proceed")
+	}
+}
@@ -0,0 +1,46 @@
+package handlers
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/approvals"
+	"github.com/gin-gonic/gin"
+)
+
+// TestGateDestructive_NonGatedPassesThrough verifies a non-gated action skips
+// the gate entirely (no DB access, no 202) so handlers whose action isn't in the
+// policy map behave exactly as before.
+func TestGateDestructive_NonGatedPassesThrough(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("POST", "/x", nil)
+
+	proceed := gateDestructive(c, newTestBroadcaster(), "ws-1",
+		approvals.Action("not_a_gated_action"), "noop", nil)
+
+	if !proceed {
+		t.Fatalf("non-gated action must proceed, got proceed=false (status %d)", w.Code)
+	}
+	if w.Code != http.StatusOK { // CreateTestContext default; nothing written
+		t.Errorf("non-gated action wrote a response (status %d), want none", w.Code)
+	}
+}
+
+// TestApprovalRequestHash_StableAndContextSensitive pins the two properties the
+// gate relies on: the same operation hashes identically across calls, and a
+// different context yields a different hash (so an approval can't be replayed
+// onto a different target).
+func TestApprovalRequestHash_StableAndContextSensitive(t *testing.T) {
+	a := approvalRequestHash("ws", "delete_workspace", map[string]interface{}{"target": "A", "n": 1})
+	aAgain := approvalRequestHash("ws", "delete_workspace", map[string]interface{}{"n": 1, "target": "A"})
+	b := approvalRequestHash("ws", "delete_workspace", map[string]interface{}{"target": "B", "n": 1})
+	if a != aAgain {
+		t.Errorf("hash not stable across equal contexts: %s vs %s", a, aAgain)
+	}
+	if a == b {
+		t.Errorf("hash not context-sensitive: target A and B collided (%s)", a)
+	}
+}
@@ -173,20 +173,8 @@ func (h *DelegationHandler) Delegate(c *gin.Context) {
 	// check_task_status returned status='queued' forever even after a
 	// real reply landed). messageId mirrors delegation_id so the
 	// platform's idempotency-key extraction also keys off the same id.
-	a2aBody, marshalErr := json.Marshal(map[string]interface{}{
-		"method": "message/send",
-		"params": map[string]interface{}{
-			"message": map[string]interface{}{
-				"role":      "user",
-				"messageId": delegationID,
-				// A2A v0.3 Part discriminator is `kind`, NOT `type` (#2251) —
-				// a `type`-keyed Part is dropped by the receiver's v0.3
-				// validator, silently losing the delegated task.
-				"parts":    []map[string]interface{}{{"kind": "text", "text": body.Task}},
-				"metadata": map[string]interface{}{"delegation_id": delegationID},
-			},
-		},
-	})
+	// Build A2A payload via helper so contract tests can assert the envelope shape.
+	a2aBody, marshalErr := buildDelegateA2ABody(delegationID, body.Task)
 	if marshalErr != nil {
 		log.Printf("Delegation %s: json.Marshal a2aBody failed: %v", delegationID, marshalErr)
 	}
@@ -374,6 +362,27 @@ func insertDelegationRow(ctx context.Context, c *gin.Context, sourceID string, b
 	return insertTrackingUnavailable
 }

+// buildDelegateA2ABody constructs the A2A JSON-RPC envelope for a delegation.
+// The returned shape is a schema-valid SendMessageRequest with role="user",
+// messageId, parts, and delegation metadata. Extracted to a pure function so
+// unit tests can assert the envelope contract without standing up HTTP or DB.
+func buildDelegateA2ABody(delegationID, task string) ([]byte, error) {
+	return json.Marshal(map[string]interface{}{
+		"method": "message/send",
+		"params": map[string]interface{}{
+			"message": map[string]interface{}{
+				"role":      "user",
+				"messageId": delegationID,
+				// A2A v0.3 Part discriminator is `kind`, NOT `type` (#2251) —
+				// a `type`-keyed Part is dropped by the receiver's v0.3
+				// validator, silently losing the delegated task.
+				"parts":    []map[string]interface{}{{"kind": "text", "text": task}},
+				"metadata": map[string]interface{}{"delegation_id": delegationID},
+			},
+		},
+	})
+}
+
 // executeDelegation runs in a goroutine — sends A2A and stores the result.
 // Updates delegation status through: pending → dispatched → received → completed/failed
 // delegationRetryDelay is the pause between the first failed proxy attempt
@@ -1762,3 +1762,74 @@ func TestListDelegations_LedgerFailedIncludesErrorDetail(t *testing.T) {
 		t.Errorf("unmet sqlmock expectations: %v", err)
 	}
 }
+
+// ---------- buildDelegateA2ABody: schema-valid SendMessageRequest ----------
+
+// TestBuildDelegateA2ABody_SchemaValidSendMessageRequest pins the contract
+// requested by issue #2251: delegate_task must produce a schema-valid A2A
+// SendMessageRequest with role="user", messageId, parts, and metadata.
+func TestBuildDelegateA2ABody_SchemaValidSendMessageRequest(t *testing.T) {
+	delegationID := "del-2251-test"
+	task := "write a contract test"
+
+	body, err := buildDelegateA2ABody(delegationID, task)
+	if err != nil {
+		t.Fatalf("buildDelegateA2ABody failed: %v", err)
+	}
+
+	var envelope map[string]interface{}
+	if err := json.Unmarshal(body, &envelope); err != nil {
+		t.Fatalf("body is not valid JSON: %v", err)
+	}
+
+	// Top-level envelope shape
+	if envelope["method"] != "message/send" {
+		t.Errorf("method = %v, want message/send", envelope["method"])
+	}
+
+	params, ok := envelope["params"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("params missing or not a map: %T", envelope["params"])
+	}
+
+	msg, ok := params["message"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("message missing or not a map: %T", params["message"])
+	}
+
+	// Issue #2251: role is required
+	if msg["role"] != "user" {
+		t.Errorf("message.role = %v, want \"user\"", msg["role"])
+	}
+
+	// messageId must be present and match delegationID
+	if msg["messageId"] != delegationID {
+		t.Errorf("message.messageId = %v, want %s", msg["messageId"], delegationID)
+	}
+
+	// parts must be a non-empty list with a text part
+	parts, ok := msg["parts"].([]interface{})
+	if !ok || len(parts) == 0 {
+		t.Fatalf("message.parts missing or empty: %T", msg["parts"])
+	}
+	firstPart, ok := parts[0].(map[string]interface{})
+	if !ok {
+		t.Fatalf("first part is not a map: %T", parts[0])
+	}
+	// A2A v0.3 Part discriminator is `kind`, NOT `type` (#2251)
+	if firstPart["kind"] != "text" {
+		t.Errorf("first part kind = %v, want text", firstPart["kind"])
+	}
+	if firstPart["text"] != task {
+		t.Errorf("first part text = %v, want %q", firstPart["text"], task)
+	}
+
+	// metadata.delegation_id must match
+	meta, ok := msg["metadata"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("metadata missing or not a map: %T", msg["metadata"])
+	}
+	if meta["delegation_id"] != delegationID {
+		t.Errorf("metadata.delegation_id = %v, want %s", meta["delegation_id"], delegationID)
+	}
+}
@@ -337,7 +337,7 @@ func TestRegister_ProvisionerURLPreserved(t *testing.T) {
 		WillReturnError(sql.ErrNoRows)

 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs("ws-prov", "ws-prov", "http://localhost:8000", `{"name":"agent"}`, "push").
+		WithArgs("ws-prov", "ws-prov", "http://localhost:8000", `{"name":"agent"}`, "push", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))

 	// DB returns provisioner URL (127.0.0.1) — should take precedence over agent-reported URL
@@ -180,7 +180,7 @@ func TestRegisterHandler(t *testing.T) {

 	// Expect the upsert INSERT ... ON CONFLICT
 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs("ws-123", "ws-123", "http://localhost:8000", `{"name":"test"}`, "push").
+		WithArgs("ws-123", "ws-123", "http://localhost:8000", `{"name":"test"}`, "push", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))

 	// Expect the SELECT url query (for cache URL logic)
@@ -0,0 +1,122 @@
+//go:build integration
+// +build integration
+
+// kind_platform_root_integration_test.go — REAL Postgres gate for the
+// platform-agent participant kind (RFC docs/design/rfc-platform-agent.md).
+//
+// Run with:
+//
+//	INTEGRATION_DB_URL="postgres://postgres:test@localhost:55432/molecule?sslmode=disable" \
+//	  go test -tags=integration ./internal/handlers/ -run Integration_PlatformKind -v
+//
+// CI: piggybacks on the handlers-postgres-integration workflow (path filter
+// includes workspace-server/internal/handlers/** and migrations/**).
+//
+// Why this is NOT a sqlmock test
+// ------------------------------
+// The invariant "a platform agent must be the org root (parent_id IS NULL),
+// which structurally also means at most one platform agent per org" is enforced
+// by the workspaces_platform_root_check CHECK constraint in migration
+// 20260606000000_workspaces_kind. sqlmock cannot execute DDL or evaluate a CHECK
+// constraint, so only a real Postgres can prove the constraint actually rejects
+// a non-root platform agent and accepts a root one. The Register handler's
+// isPlatformRootViolation()/409 path depends on this constraint firing.
+
+package handlers
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	_ "github.com/lib/pq"
+)
+
+func integrationDB_PlatformKind(t *testing.T) *sql.DB {
+	t.Helper()
+	url := requireIntegrationDBURL(t)
+	conn, err := sql.Open("postgres", url)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if err := conn.Ping(); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return conn
+}
+
+// TestIntegration_PlatformKind_RootAllowed_NonRootRejected proves the three
+// guarantees of the kind column against a real Postgres:
+//
+//  1. a fresh workspace defaults to kind='workspace';
+//  2. a root row (parent_id IS NULL) may be kind='platform';
+//  3. a non-root row (parent_id set) may NOT be kind='platform' — the
+//     workspaces_platform_root_check constraint rejects it (23514).
+func TestIntegration_PlatformKind_RootAllowed_NonRootRejected(t *testing.T) {
+	conn := integrationDB_PlatformKind(t)
+	ctx := context.Background()
+
+	prefix := fmt.Sprintf("itest-kind-%s", uuid.New().String()[:8])
+	cleanup := func() {
+		if _, err := conn.ExecContext(ctx,
+			`DELETE FROM workspaces WHERE name LIKE $1`, prefix+"%"); err != nil {
+			t.Logf("cleanup (non-fatal): %v", err)
+		}
+	}
+	t.Cleanup(cleanup)
+	cleanup() // pre-test hygiene in the shared integration DB
+
+	rootID := uuid.New().String()
+	childID := uuid.New().String()
+
+	// 1. Default kind is 'workspace' when the column is omitted on INSERT.
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO workspaces (id, name, tier, runtime, status, parent_id)
+		VALUES ($1, $2, 2, 'claude-code', 'online', NULL)
+	`, rootID, prefix+"-root"); err != nil {
+		t.Fatalf("seed root: %v", err)
+	}
+	var gotKind string
+	if err := conn.QueryRowContext(ctx,
+		`SELECT kind FROM workspaces WHERE id = $1`, rootID).Scan(&gotKind); err != nil {
+		t.Fatalf("read kind: %v", err)
+	}
+	if gotKind != "workspace" {
+		t.Fatalf("default kind = %q, want \"workspace\"", gotKind)
+	}
+
+	// 2. The root row may become a platform agent.
+	if _, err := conn.ExecContext(ctx,
+		`UPDATE workspaces SET kind = 'platform' WHERE id = $1`, rootID); err != nil {
+		t.Fatalf("promote root to platform: unexpected error: %v", err)
+	}
+
+	// A child of the platform root (an ordinary workspace) inserts fine.
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO workspaces (id, name, tier, runtime, status, parent_id)
+		VALUES ($1, $2, 2, 'claude-code', 'online', $3)
+	`, childID, prefix+"-child", rootID); err != nil {
+		t.Fatalf("seed child: %v", err)
+	}
+
+	// 3. The non-root child may NOT be a platform agent — the CHECK rejects it.
+	_, err := conn.ExecContext(ctx,
+		`UPDATE workspaces SET kind = 'platform' WHERE id = $1`, childID)
+	if err == nil {
+		t.Fatalf("non-root child accepted kind='platform' — constraint did not fire")
+	}
+	if !strings.Contains(err.Error(), "workspaces_platform_root_check") {
+		t.Fatalf("non-root platform rejection wanted workspaces_platform_root_check, got: %v", err)
+	}
+
+	// And the unknown-kind value is rejected by workspaces_kind_check.
+	_, err = conn.ExecContext(ctx,
+		`UPDATE workspaces SET kind = 'bogus' WHERE id = $1`, rootID)
+	if err == nil || !strings.Contains(err.Error(), "workspaces_kind_check") {
+		t.Fatalf("unknown kind wanted workspaces_kind_check rejection, got: %v", err)
+	}
+}
@@ -54,6 +54,55 @@ func mcpPost(t *testing.T, h *MCPHandler, workspaceID string, body interface{})
 	return w
 }

+// assertA2ASendMessageSchema validates that body is a schema-valid A2A
+// SendMessageRequest with role="user", messageId, and non-empty parts.
+// Issue #2251 contract test: delegate_task must always produce this shape.
+func assertA2ASendMessageSchema(t *testing.T, body []byte, wantTask string) {
+	t.Helper()
+	var envelope map[string]interface{}
+	if err := json.Unmarshal(body, &envelope); err != nil {
+		t.Fatalf("A2A body is not valid JSON: %v", err)
+	}
+	if envelope["jsonrpc"] != "2.0" {
+		t.Errorf("jsonrpc = %v, want 2.0", envelope["jsonrpc"])
+	}
+	if envelope["method"] != "message/send" {
+		t.Errorf("method = %v, want message/send", envelope["method"])
+	}
+
+	params, ok := envelope["params"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("params missing or not a map: %T", envelope["params"])
+	}
+	msg, ok := params["message"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("message missing or not a map: %T", params["message"])
+	}
+
+	if msg["role"] != "user" {
+		t.Errorf("message.role = %v, want \"user\"", msg["role"])
+	}
+	if msg["messageId"] == "" {
+		t.Error("message.messageId is empty")
+	}
+
+	parts, ok := msg["parts"].([]interface{})
+	if !ok || len(parts) == 0 {
+		t.Fatalf("message.parts missing or empty: %T", msg["parts"])
+	}
+	firstPart, ok := parts[0].(map[string]interface{})
+	if !ok {
+		t.Fatalf("first part is not a map: %T", parts[0])
+	}
+	// A2A v0.3 Part discriminator is `kind`, NOT `type` (#2251)
+	if firstPart["kind"] != "text" {
+		t.Errorf("first part kind = %v, want text", firstPart["kind"])
+	}
+	if firstPart["text"] != wantTask {
+		t.Errorf("first part text = %v, want %q", firstPart["text"], wantTask)
+	}
+}
+
 func expectCanCommunicateSiblings(mock sqlmock.Sqlmock, callerID, targetID, parentID string) {
 	mock.ExpectQuery(`SELECT id, parent_id FROM workspaces WHERE id = \$1`).
 		WithArgs(callerID).
@@ -209,9 +258,7 @@ func TestMCPHandler_DelegateTask_RoutesThroughPlatformA2AProxy(t *testing.T) {
 		if !logActivity {
 			t.Fatal("delegate_task should log through platform A2A proxy")
 		}
-		if !strings.Contains(string(body), "do work") {
-			t.Fatalf("A2A body missing task text: %s", string(body))
-		}
+		assertA2ASendMessageSchema(t, body, "do work")
 		return 200, []byte(`{"result":{"message":{"parts":[{"text":"done"}]}}}`), nil
 	}

@@ -252,9 +299,7 @@ func TestMCPHandler_DelegateTaskAsync_RoutesThroughPlatformA2AProxy(t *testing.T
 		if workspaceID != targetID || proxyCallerID != callerID {
 			t.Fatalf("unexpected proxy route target=%q caller=%q", workspaceID, proxyCallerID)
 		}
-		if !strings.Contains(string(body), "async work") {
-			t.Fatalf("A2A body missing task text: %s", string(body))
-		}
+		assertA2ASendMessageSchema(t, body, "async work")
 		called <- struct{}{}
 		return 200, []byte(`{"result":{"message":{"parts":[{"text":"accepted"}]}}}`), nil
 	}
@@ -304,10 +349,8 @@ func TestMCPHandler_DelegateTask_WithAttachments(t *testing.T) {
 		if workspaceID != targetID || proxyCallerID != callerID {
 			t.Fatalf("unexpected proxy route target=%q caller=%q", workspaceID, proxyCallerID)
 		}
+		assertA2ASendMessageSchema(t, body, "review this video")
 		bodyStr := string(body)
-		if !strings.Contains(bodyStr, `"text":"review this video"`) {
-			t.Fatalf("A2A body missing task text: %s", bodyStr)
-		}
 		if !strings.Contains(bodyStr, `"kind":"video"`) {
 			t.Fatalf("A2A body missing video attachment kind: %s", bodyStr)
 		}
@@ -386,6 +429,7 @@ func TestMCPHandler_DelegateTaskAsync_WithAttachments(t *testing.T) {
 	waitGlobalAsyncForTest()
 	select {
 	case body := <-called:
+		assertA2ASendMessageSchema(t, body, "async work with image")
 		bodyStr := string(body)
 		if !strings.Contains(bodyStr, `"kind":"image"`) {
 			t.Fatalf("A2A body missing image attachment kind: %s", bodyStr)
@@ -177,7 +177,7 @@ func isEnvIdentPart(c byte) bool {
 	return isEnvIdentStart(c) || (c >= '0' && c <= '9')
 }

-// loadWorkspaceEnv reads the org root .env and the workspace-specific .env
+// loadWorkspaceEnv reads the org root .env and the workspace-specific .env files.
 // (workspace overrides org root). Used by both secret injection and channel
 // config expansion.
 //
@@ -0,0 +1,135 @@
+package handlers
+
+// platform_agent.go — installs the org-level platform agent as the org root.
+// (RFC docs/design/rfc-platform-agent.md)
+//
+// The platform agent IS the org root: an org is the subtree under the single
+// parent_id IS NULL row (org_scope.go), so making the concierge the user's
+// universal A2A peer means making it that root. Installing it therefore:
+//
+//	1. upserts the platform-agent row (kind='platform', parent_id NULL);
+//	2. re-parents the org's existing root(s) under it;
+//	3. moves the org-anchor references — org_api_tokens.org_id and
+//	   org_plugin_allowlist.org_id, both of which key off the root workspace id
+//	   (see migrations 035/036 + 026) — from each old root to the platform agent.
+//
+// All of that happens in ONE transaction so a tenant's auth tokens and plugin
+// allowlist never point at a stale anchor. The operation is idempotent: a second
+// call finds the platform agent already the sole root and does nothing.
+//
+// Routing (CanCommunicate/sameOrg in registry/access.go + org_scope.go) needs NO
+// change — once the platform agent is the root, the existing ancestor/descendant
+// rules already give it universal in-org reach and keep tenant isolation intact.
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log"
+	"net/http"
+
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/db"
+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/models"
+	"github.com/gin-gonic/gin"
+)
+
+type installPlatformAgentPayload struct {
+	// ID is the platform agent's workspace id (a deterministic uuidv5 the
+	// control plane derives per org). Required.
+	ID string `json:"id" binding:"required"`
+	// Name is the display name; defaults to "Org Concierge" when omitted.
+	Name string `json:"name"`
+}
+
+// InstallPlatformAgent handles POST /admin/org/platform-agent (AdminAuth).
+//
+// Idempotently installs the platform agent as the org root for THIS tenant. The
+// control plane calls it at org-provision time (new orgs) and during the
+// existing-org backfill rollout. Safe to call repeatedly.
+func InstallPlatformAgent(c *gin.Context) {
+	var p installPlatformAgentPayload
+	if err := c.ShouldBindJSON(&p); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"})
+		return
+	}
+	name := p.Name
+	if name == "" {
+		name = "Org Concierge"
+	}
+	if err := installPlatformAgent(c.Request.Context(), db.DB, p.ID, name); err != nil {
+		log.Printf("InstallPlatformAgent: %v (id=%s)", err, p.ID)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "install failed"})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{
+		"status":            "installed",
+		"platform_agent_id": p.ID,
+		"kind":              models.KindPlatform,
+	})
+}
+
+// installPlatformAgent performs the idempotent, transactional install described
+// in the file header. Separated from the gin handler so integration tests can
+// exercise it directly against a real Postgres (the org-anchor migration cannot
+// be proven with sqlmock).
+func installPlatformAgent(ctx context.Context, database *sql.DB, platformID, name string) error {
+	tx, err := database.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }() // no-op after Commit
+
+	// 1. Ensure the platform-agent row exists as a kind='platform' root.
+	//    ON CONFLICT keeps it a platform root if it was pre-seeded; the row is
+	//    tier 0 and never billed/provisioned as an ordinary workspace EC2.
+	if _, err := tx.ExecContext(ctx, `
+		INSERT INTO workspaces (id, name, kind, tier, status, runtime, parent_id)
+		VALUES ($1, $2, 'platform', 0, 'online', 'claude-code', NULL)
+		ON CONFLICT (id) DO UPDATE SET kind = 'platform', parent_id = NULL
+	`, platformID, name); err != nil {
+		return fmt.Errorf("upsert platform agent: %w", err)
+	}
+
+	// 2. Capture the org's other current roots (everything at parent_id IS NULL
+	//    except the platform agent itself). In a one-org tenant DB this is the
+	//    single team root; the query tolerates 0 (already installed) or N.
+	rows, err := tx.QueryContext(ctx,
+		`SELECT id FROM workspaces WHERE parent_id IS NULL AND id <> $1`, platformID)
+	if err != nil {
+		return fmt.Errorf("select old roots: %w", err)
+	}
+	var oldRoots []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			rows.Close()
+			return fmt.Errorf("scan old root: %w", err)
+		}
+		oldRoots = append(oldRoots, id)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("iterate old roots: %w", err)
+	}
+
+	// 3 + 4. Re-parent each old root under the platform agent and move its
+	//        org-anchor references in the same transaction. A non-root old root
+	//        is kind='workspace', so it does not trip workspaces_platform_root_check.
+	for _, root := range oldRoots {
+		if _, err := tx.ExecContext(ctx,
+			`UPDATE workspaces SET parent_id = $1, updated_at = now() WHERE id = $2`,
+			platformID, root); err != nil {
+			return fmt.Errorf("re-parent %s: %w", root, err)
+		}
+		if _, err := tx.ExecContext(ctx,
+			`UPDATE org_api_tokens SET org_id = $1 WHERE org_id = $2`, platformID, root); err != nil {
+			return fmt.Errorf("migrate org_api_tokens for %s: %w", root, err)
+		}
+		if _, err := tx.ExecContext(ctx,
+			`UPDATE org_plugin_allowlist SET org_id = $1 WHERE org_id = $2`, platformID, root); err != nil {
+			return fmt.Errorf("migrate org_plugin_allowlist for %s: %w", root, err)
+		}
+	}
+
+	return tx.Commit()
+}
@@ -0,0 +1,188 @@
+//go:build integration
+// +build integration
+
+// platform_agent_integration_test.go — REAL Postgres gate for installPlatformAgent.
+//
+// Run with:
+//
+//	INTEGRATION_DB_URL="postgres://postgres:test@localhost:55432/molecule?sslmode=disable" \
+//	  go test -tags=integration ./internal/handlers/ -run Integration_PlatformAgentInstall -v
+//
+// CI: handlers-postgres-integration workflow (handlers + migrations path filter).
+//
+// Why this is NOT a sqlmock test
+// ------------------------------
+// The install re-parents the org's existing root under the platform agent AND
+// moves the org-anchor references (org_api_tokens.org_id, org_plugin_allowlist.
+// org_id) from old root to platform agent, atomically. The whole point is the
+// post-transaction row state: orgRootID() must resolve every node to the platform
+// agent, sameOrg() must still hold, and the auth/allowlist anchors must point at
+// the new root. Only a real Postgres can prove that; sqlmock cannot.
+
+package handlers
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"testing"
+
+	"github.com/google/uuid"
+	_ "github.com/lib/pq"
+)
+
+func integrationDB_PlatformAgentInstall(t *testing.T) *sql.DB {
+	t.Helper()
+	url := requireIntegrationDBURL(t)
+	conn, err := sql.Open("postgres", url)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if err := conn.Ping(); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return conn
+}
+
+// TestIntegration_PlatformAgentInstall_ReparentsRootAndMovesAnchors builds a
+// real org in Postgres:
+//
+//	root (parent_id NULL, kind=workspace)
+//	└── child
+//	+ an org_api_token  anchored to root
+//	+ an org_plugin_allowlist entry anchored to root
+//
+// then installs the platform agent and asserts:
+//   - the platform agent is the new sole root (kind=platform, parent_id NULL);
+//   - the old root is re-parented under it; the child is untouched;
+//   - both org-anchor references now point at the platform agent;
+//   - a second install is a no-op (idempotent).
+func TestIntegration_PlatformAgentInstall_ReparentsRootAndMovesAnchors(t *testing.T) {
+	conn := integrationDB_PlatformAgentInstall(t)
+	ctx := context.Background()
+
+	tag := uuid.New().String()[:8]
+	prefix := fmt.Sprintf("itest-pinstall-%s", tag)
+	rootID := uuid.New().String()
+	childID := uuid.New().String()
+	platformID := uuid.New().String()
+
+	cleanup := func() {
+		_, _ = conn.ExecContext(ctx, `DELETE FROM org_plugin_allowlist WHERE plugin_name = $1`, prefix+"-plugin")
+		_, _ = conn.ExecContext(ctx, `DELETE FROM org_api_tokens WHERE prefix = $1`, tag)
+		// child + old root (prefixed names) first, then the platform agent by id
+		// (root.parent_id references it, so it must go last).
+		_, _ = conn.ExecContext(ctx, `DELETE FROM workspaces WHERE name LIKE $1`, prefix+"%")
+		_, _ = conn.ExecContext(ctx, `DELETE FROM workspaces WHERE id = $1`, platformID)
+	}
+	t.Cleanup(cleanup)
+	cleanup()
+
+	// Seed org tree.
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO workspaces (id, name, tier, runtime, status, parent_id)
+		VALUES ($1, $2, 2, 'claude-code', 'online', NULL)`, rootID, prefix+"-root"); err != nil {
+		t.Fatalf("seed root: %v", err)
+	}
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO workspaces (id, name, tier, runtime, status, parent_id)
+		VALUES ($1, $2, 2, 'claude-code', 'online', $3)`, childID, prefix+"-child", rootID); err != nil {
+		t.Fatalf("seed child: %v", err)
+	}
+	// Org-anchor rows keyed to the OLD root.
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO org_api_tokens (token_hash, prefix, name, org_id)
+		VALUES ($1, $2, $3, $4)`,
+		[]byte("hash-"+tag), tag, prefix+"-tok", rootID); err != nil {
+		t.Fatalf("seed org_api_token: %v", err)
+	}
+	if _, err := conn.ExecContext(ctx, `
+		INSERT INTO org_plugin_allowlist (org_id, plugin_name, enabled_by)
+		VALUES ($1, $2, $3)`, rootID, prefix+"-plugin", childID); err != nil {
+		t.Fatalf("seed allowlist: %v", err)
+	}
+
+	// Install.
+	if err := installPlatformAgent(ctx, conn, platformID, "Org Concierge"); err != nil {
+		t.Fatalf("install: %v", err)
+	}
+
+	assertState := func(stage string) {
+		// platform agent is a kind=platform root.
+		var kind string
+		var parent sql.NullString
+		if err := conn.QueryRowContext(ctx,
+			`SELECT kind, parent_id FROM workspaces WHERE id = $1`, platformID).Scan(&kind, &parent); err != nil {
+			t.Fatalf("[%s] read platform agent: %v", stage, err)
+		}
+		if kind != "platform" || parent.Valid {
+			t.Fatalf("[%s] platform agent kind=%q parent=%v, want platform/NULL", stage, kind, parent)
+		}
+		// old root re-parented under the platform agent.
+		var rootParent sql.NullString
+		if err := conn.QueryRowContext(ctx,
+			`SELECT parent_id FROM workspaces WHERE id = $1`, rootID).Scan(&rootParent); err != nil {
+			t.Fatalf("[%s] read old root: %v", stage, err)
+		}
+		if !rootParent.Valid || rootParent.String != platformID {
+			t.Fatalf("[%s] old root parent=%v, want %s", stage, rootParent, platformID)
+		}
+		// child untouched.
+		var childParent sql.NullString
+		if err := conn.QueryRowContext(ctx,
+			`SELECT parent_id FROM workspaces WHERE id = $1`, childID).Scan(&childParent); err != nil {
+			t.Fatalf("[%s] read child: %v", stage, err)
+		}
+		if !childParent.Valid || childParent.String != rootID {
+			t.Fatalf("[%s] child parent=%v, want %s (unchanged)", stage, childParent, rootID)
+		}
+		// org-anchor references moved to the platform agent.
+		var tokOrg, alOrg string
+		if err := conn.QueryRowContext(ctx,
+			`SELECT org_id FROM org_api_tokens WHERE prefix = $1`, tag).Scan(&tokOrg); err != nil {
+			t.Fatalf("[%s] read token org_id: %v", stage, err)
+		}
+		if tokOrg != platformID {
+			t.Fatalf("[%s] org_api_tokens.org_id=%s, want %s", stage, tokOrg, platformID)
+		}
+		if err := conn.QueryRowContext(ctx,
+			`SELECT org_id FROM org_plugin_allowlist WHERE plugin_name = $1`, prefix+"-plugin").Scan(&alOrg); err != nil {
+			t.Fatalf("[%s] read allowlist org_id: %v", stage, err)
+		}
+		if alOrg != platformID {
+			t.Fatalf("[%s] org_plugin_allowlist.org_id=%s, want %s", stage, alOrg, platformID)
+		}
+		// orgRootID + sameOrg now resolve everything to the platform agent.
+		got, err := orgRootID(ctx, conn, childID)
+		if err != nil {
+			t.Fatalf("[%s] orgRootID(child): %v", stage, err)
+		}
+		if got != platformID {
+			t.Fatalf("[%s] orgRootID(child)=%s, want %s", stage, got, platformID)
+		}
+		same, err := sameOrg(ctx, conn, childID, platformID)
+		if err != nil || !same {
+			t.Fatalf("[%s] sameOrg(child, platform)=%v err=%v, want true", stage, same, err)
+		}
+	}
+
+	assertState("first install")
+
+	// Idempotent: second install must not error or change state.
+	if err := installPlatformAgent(ctx, conn, platformID, "Org Concierge"); err != nil {
+		t.Fatalf("second install (idempotent): %v", err)
+	}
+	assertState("second install")
+
+	// Neither seeded team node is a root any more — the platform agent is.
+	var nRoots int
+	if err := conn.QueryRowContext(ctx,
+		`SELECT count(*) FROM workspaces WHERE parent_id IS NULL AND id IN ($1, $2)`,
+		rootID, childID).Scan(&nRoots); err != nil {
+		t.Fatalf("count roots: %v", err)
+	}
+	if nRoots != 0 {
+		t.Fatalf("team roots after install = %d, want 0 (old root re-parented under platform agent)", nRoots)
+	}
+}
@@ -0,0 +1,27 @@
+package handlers
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+)
+
+// TestInstallPlatformAgent_BadJSON rejects a payload missing the required id
+// before touching the DB (binding:"required" on ID).
+func TestInstallPlatformAgent_BadJSON(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("POST", "/admin/org/platform-agent",
+		bytes.NewBufferString(`{"name":"Org Concierge"}`)) // no id
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	InstallPlatformAgent(c)
+
+	if w.Code != http.StatusBadRequest {
+		t.Errorf("missing id: expected 400, got %d: %s", w.Code, w.Body.String())
+	}
+}
@@ -164,6 +164,20 @@ func (h *RegistryHandler) resolveDeliveryMode(ctx context.Context, workspaceID,
 	return models.DeliveryModePush, nil
 }

+// errPlatformNotRoot is the client-facing message when a register call tried to
+// mark a non-root workspace as a platform agent.
+const errPlatformNotRoot = "a platform agent must be the org root (parent_id must be null)"
+
+// isPlatformRootViolation reports whether err is the DB rejecting a register
+// that tried to mark a non-root workspace as a platform agent (the
+// workspaces_platform_root_check CHECK constraint). The handler maps it to a
+// friendly HTTP 409 instead of a raw 500. The invariant — platform == org root,
+// which structurally also guarantees one platform agent per org — is enforced
+// race-proof at the DB level; this is just the friendly surface.
+func isPlatformRootViolation(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "workspaces_platform_root_check")
+}
+
 // Returns a non-nil error suitable for including in a 400 Bad Request response.
 func validateAgentURL(rawURL string) error {
 	if rawURL == "" {
@@ -277,6 +291,14 @@ func (h *RegistryHandler) Register(c *gin.Context) {
 		return
 	}

+	// Validate explicit kind if the agent declared one; empty is allowed and
+	// resolves to the row's existing value (or "workspace" default) in
+	// resolveKind below. Only the platform-agent container declares 'platform'.
+	if payload.Kind != "" && !models.IsValidKind(payload.Kind) {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "kind must be 'workspace' or 'platform'"})
+		return
+	}
+
 	ctx := c.Request.Context()

 	// C18: prevent workspace URL hijacking on re-registration.
@@ -390,9 +412,15 @@ func (h *RegistryHandler) Register(c *gin.Context) {
 	// the row. Without this guard, bulk deletes left tier-3 stragglers because
 	// the last pre-teardown heartbeat flipped status back to 'online' after
 	// Delete's UPDATE.
+	// kind ($6) is the raw payload value (validated above; "" = unspecified).
+	// COALESCE(NULLIF($6,''), …) means: an explicit kind wins; an unspecified
+	// kind defaults to 'workspace' for a NEW row and KEEPS the existing kind on
+	// re-register (so a platform agent re-registering without kind is never
+	// downgraded). A non-root row asking for 'platform' is rejected by the
+	// workspaces_platform_root_check constraint → friendly 409 below.
 	_, err = db.DB.ExecContext(ctx, `
-		INSERT INTO workspaces (id, name, url, agent_card, status, last_heartbeat_at, delivery_mode)
-		VALUES ($1, $2, $3, $4::jsonb, 'online', now(), $5)
+		INSERT INTO workspaces (id, name, url, agent_card, status, last_heartbeat_at, delivery_mode, kind)
+		VALUES ($1, $2, $3, $4::jsonb, 'online', now(), $5, COALESCE(NULLIF($6, ''), 'workspace'))
 		ON CONFLICT (id) DO UPDATE SET
 			url = CASE
 				WHEN workspaces.url LIKE 'http://127.0.0.1%' THEN workspaces.url
@@ -402,10 +430,15 @@ func (h *RegistryHandler) Register(c *gin.Context) {
 			status = 'online',
 			last_heartbeat_at = now(),
 			delivery_mode = EXCLUDED.delivery_mode,
+			kind = COALESCE(NULLIF($6, ''), workspaces.kind),
 			updated_at = now()
 		WHERE workspaces.status IS DISTINCT FROM 'removed'
-	`, payload.ID, payload.ID, urlForUpsert, agentCardStr, modeForUpsert)
+	`, payload.ID, payload.ID, urlForUpsert, agentCardStr, modeForUpsert, payload.Kind)
 	if err != nil {
+		if isPlatformRootViolation(err) {
+			c.JSON(http.StatusConflict, gin.H{"error": errPlatformNotRoot})
+			return
+		}
 		log.Printf("Registry register error: %v (id=%s)", err, payload.ID)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "registration failed"})
 		return
@@ -72,7 +72,7 @@ func TestRegister_DBError(t *testing.T) {

 	// DB insert fails
 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs("ws-fail", "ws-fail", "http://localhost:8000", `{"name":"test"}`, "push").
+		WithArgs("ws-fail", "ws-fail", "http://localhost:8000", `{"name":"test"}`, "push", "").
 		WillReturnError(sql.ErrConnDone)

 	w := httptest.NewRecorder()
@@ -647,7 +647,7 @@ func TestRegister_GuardAgainstResurrectingRemovedRow(t *testing.T) {
 	// This regex-ish match requires the guard. If the handler ever drops
 	// the clause the test fails because the emitted SQL won't match.
 	mock.ExpectExec("ON CONFLICT.*WHERE workspaces.status IS DISTINCT FROM 'removed'").
-		WithArgs("ws-resurrect", "ws-resurrect", "http://localhost:8000", `{"name":"x"}`, "push").
+		WithArgs("ws-resurrect", "ws-resurrect", "http://localhost:8000", `{"name":"x"}`, "push", "").
 		WillReturnResult(sqlmock.NewResult(0, 0)) // 0 rows affected = correctly guarded
 	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
 		WithArgs("ws-resurrect").
@@ -917,7 +917,7 @@ func TestRegister_C18_BootstrapAllowedNoTokens(t *testing.T) {

 	// Workspace upsert proceeds normally.
 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs("ws-new", "ws-new", "http://localhost:9100", `{"name":"new-agent"}`, "push").
+		WithArgs("ws-new", "ws-new", "http://localhost:9100", `{"name":"new-agent"}`, "push", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))

 	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
@@ -1228,7 +1228,7 @@ func TestRegister_DBErrorResponseIsOpaque(t *testing.T) {

 	// DB upsert fails with a descriptive internal error.
 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs("ws-errtest", "ws-errtest", "http://localhost:9200", `{"name":"err-agent"}`, "push").
+		WithArgs("ws-errtest", "ws-errtest", "http://localhost:9200", `{"name":"err-agent"}`, "push", "").
 		WillReturnError(sql.ErrConnDone)

 	w := httptest.NewRecorder()
@@ -1476,7 +1476,7 @@ func TestRegister_PollMode_AcceptsEmptyURL(t *testing.T) {

 	// Upsert MUST run with empty URL (sql.NullString) and delivery_mode=poll.
 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"poll-agent"}`, "poll").
+		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"poll-agent"}`, "poll", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))

 	// SELECT url for cache: returns NULL/empty for poll-mode rows. The
@@ -1591,6 +1591,89 @@ func TestRegister_InvalidDeliveryMode(t *testing.T) {
 	}
 }

+// TestRegister_InvalidKind rejects payloads that declare an unrecognised kind —
+// only 'workspace' and 'platform' are valid. Mirrors the delivery_mode guard;
+// the rejection happens before any DB access.
+func TestRegister_InvalidKind(t *testing.T) {
+	setupTestDB(t)
+	setupTestRedis(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewRegistryHandler(broadcaster)
+
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("POST", "/registry/register",
+		bytes.NewBufferString(`{"id":"ws-x","url":"http://localhost:8000","agent_card":{"name":"a"},"kind":"bogus"}`))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	handler.Register(c)
+
+	if w.Code != http.StatusBadRequest {
+		t.Errorf("invalid kind: expected 400, got %d: %s", w.Code, w.Body.String())
+	}
+	if !strings.Contains(w.Body.String(), "kind") {
+		t.Errorf("expected error body to mention kind, got: %s", w.Body.String())
+	}
+}
+
+// TestRegister_PlatformKind_PersistsKind verifies that a workspace registering
+// with kind="platform" has that value written through the upsert (the platform
+// agent self-registers as the org root). The platform==root invariant itself is
+// enforced by the workspaces_platform_root_check DB constraint and exercised by
+// the integration test, which sqlmock cannot enforce.
+func TestRegister_PlatformKind_PersistsKind(t *testing.T) {
+	mock := setupTestDB(t)
+	setupTestRedis(t)
+	broadcaster := newTestBroadcaster()
+	handler := NewRegistryHandler(broadcaster)
+
+	const wsID = "ws-platform-agent"
+
+	// Bootstrap path — no live tokens.
+	mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM workspace_auth_tokens").
+		WithArgs(wsID).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
+
+	// delivery_mode="push" is set explicitly, so resolveDeliveryMode
+	// short-circuits (no SELECT delivery_mode lookup). The upsert MUST carry
+	// kind="platform" as the 6th arg.
+	mock.ExpectExec("INSERT INTO workspaces").
+		WithArgs(wsID, wsID, "http://localhost:9100", `{"name":"concierge"}`, "push", "platform").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
+		WithArgs(wsID).
+		WillReturnRows(sqlmock.NewRows([]string{"url"}).AddRow("http://localhost:9100"))
+
+	mock.ExpectExec("INSERT INTO structure_events").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	// Token issuance — first-register path.
+	mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM workspace_auth_tokens").
+		WithArgs(wsID).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
+	mock.ExpectExec("INSERT INTO workspace_auth_tokens").
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectQuery(`SELECT platform_inbound_secret FROM workspaces WHERE id = \$1`).
+		WithArgs(wsID).
+		WillReturnRows(sqlmock.NewRows([]string{"platform_inbound_secret"}).AddRow(nil))
+
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("POST", "/registry/register",
+		bytes.NewBufferString(`{"id":"`+wsID+`","url":"http://localhost:9100","delivery_mode":"push","kind":"platform","agent_card":{"name":"concierge"}}`))
+	c.Request.Header.Set("Content-Type", "application/json")
+
+	handler.Register(c)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("platform register: expected 200, got %d: %s", w.Code, w.Body.String())
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("unmet expectations: %v", err)
+	}
+}
+
 // TestRegister_PollMode_PreservesExistingValue: when the row already
 // has delivery_mode=poll and the payload doesn't set it, the resolved
 // mode should be poll — i.e. "absent payload mode" must NOT silently
@@ -1616,7 +1699,7 @@ func TestRegister_PollMode_PreservesExistingValue(t *testing.T) {
 	// Upsert carries the resolved poll mode forward — even though
 	// payload didn't restate it. URL still empty (poll-mode shape).
 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"a"}`, "poll").
+		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"a"}`, "poll", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))
 	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
 		WithArgs(wsID).
@@ -1685,7 +1768,7 @@ func TestRegister_ExternalRuntime_DefaultsToPoll(t *testing.T) {
 			AddRow(sql.NullString{}, "external"))

 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"a"}`, "poll").
+		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"a"}`, "poll", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))
 	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
 		WithArgs(wsID).
@@ -1744,7 +1827,7 @@ func TestRegister_KimiRuntime_DefaultsToPoll(t *testing.T) {
 			AddRow(sql.NullString{}, "kimi-cli"))

 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"a"}`, "poll").
+		WithArgs(wsID, wsID, sql.NullString{}, `{"name":"a"}`, "poll", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))
 	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
 		WithArgs(wsID).
@@ -1804,7 +1887,7 @@ func TestRegister_NonExternalRuntime_StillDefaultsToPush(t *testing.T) {
 			AddRow(sql.NullString{}, "claude-code"))

 	mock.ExpectExec("INSERT INTO workspaces").
-		WithArgs(wsID, wsID, "http://localhost:8000", `{"name":"a"}`, "push").
+		WithArgs(wsID, wsID, "http://localhost:8000", `{"name":"a"}`, "push", "").
 		WillReturnResult(sqlmock.NewResult(0, 1))
 	mock.ExpectQuery("SELECT url FROM workspaces WHERE id").
 		WithArgs(wsID).
@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strings"

+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/approvals"
 	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/audit"
 	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/crypto"
 	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/db"
@@ -320,6 +321,18 @@ func (h *SecretsHandler) Set(c *gin.Context) {
 		return
 	}

+	// RFC platform-agent Phase 4b: gate org-token (platform-agent) secret writes
+	// behind human approval. The context includes the key so an approval for one
+	// secret cannot authorise writing another. No-op for ordinary callers and
+	// when the rollout flag is off (scoping lives in gateDestructive).
+	// SecretsHandler has no broadcaster, so pass nil — requireApproval persists
+	// the pending row regardless; only the live canvas push is skipped.
+	if !gateDestructive(c, nil, workspaceID, approvals.ActionSecretWrite,
+		"write secret "+body.Key,
+		map[string]interface{}{"workspace_id": workspaceID, "key": body.Key}) {
+		return
+	}
+
 	// Encrypt the value (AES-256-GCM if SECRETS_ENCRYPTION_KEY is set, plaintext otherwise)
 	encrypted, err := crypto.Encrypt([]byte(body.Value))
 	if err != nil {
@@ -14,6 +14,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"time"
@@ -113,6 +114,11 @@ func (h *WorkspaceHandler) goAsync(fn func()) {
 	h.asyncWG.Add(1)
 	go func() {
 		defer h.asyncWG.Done()
+		defer func() {
+			if r := recover(); r != nil {
+				log.Printf("PANIC recovered in goAsync goroutine: %v\n%s", r, debug.Stack())
+			}
+		}()
 		fn()
 	}()
 }
@@ -151,6 +157,11 @@ func globalGoAsync(fn func()) {
 	globalAsync.Add(1)
 	go func() {
 		defer globalAsync.Done()
+		defer func() {
+			if r := recover(); r != nil {
+				log.Printf("PANIC recovered in globalGoAsync goroutine: %v\n%s", r, debug.Stack())
+			}
+		}()
 		fn()
 	}()
 }
@@ -16,6 +16,7 @@ import (
 	"strings"
 	"time"

+	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/approvals"
 	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/db"
 	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/events"
 	"git.moleculesai.app/molecule-ai/molecule-core/workspace-server/internal/models"
@@ -356,6 +357,16 @@ func (h *WorkspaceHandler) Delete(c *gin.Context) {
 		return
 	}

+	// RFC platform-agent Phase 4b: gate org-token (platform-agent) deletes behind
+	// human approval. No-op for ordinary workspace/CP-session callers and when
+	// the rollout flag is off (scoping lives in gateDestructive). Placed after
+	// the synchronous X-Confirm-Name guard, before any destruction.
+	if !gateDestructive(c, h.broadcaster, id, approvals.ActionDeleteWorkspace,
+		"delete workspace "+workspaceName,
+		map[string]interface{}{"workspace_id": id, "name": workspaceName}) {
+		return
+	}
+
 	// Check for children
 	rows, err := db.DB.QueryContext(ctx,
 		`SELECT id, name FROM workspaces WHERE parent_id = $1 AND status != 'removed'`, id)
@@ -45,7 +45,7 @@ type restartState struct {
 	// flipped running=true. Used by the self-fire debounce (internal#544,
 	// the ws-server self-fire restart feedback loop seen in prod-Reviewer/
 	// Researcher 2026-05-19 ~00:05Z 4x reprov thrash): any RestartByID
-	// arriving within restartDebounceWindow of this timestamp is silently
+	// arriving within RestartDebounceWindow of this timestamp is silently
 	// dropped so a probe firing during the EC2-pending window can't
 	// re-trigger a fresh full cycle on the just-launched instance.
 	restartStartedAt time.Time
@@ -55,13 +55,19 @@ type restartState struct {
 // its own entry so unrelated workspaces don't serialize on each other.
 var restartStates sync.Map // map[workspaceID]*restartState

-// restartDebounceWindow is the silent-drop window for successive RestartByID
+// RestartDebounceWindow is the silent-drop window for successive RestartByID
 // calls. Sized to cover the typical EC2 pending → online interval (20-30s)
 // with a margin so a probe firing during the just-after-online but still-
 // flaky heartbeat window also gets dropped. Bigger than that would block
 // legitimate "Restart failed, retry" recoveries; smaller would let the
 // 4x thrash class through. Package-level so tests can shrink it.
-var restartDebounceWindow = 60 * time.Second
+//
+// COUPLING: this window MUST be >= the CP instance reconciler interval
+// (cmd/server/main.go:355, currently 60s). If the interval ever shrinks
+// below this window, a workspace flipped offline by one reconcile tick can
+// be reprovisioned again by the next tick before the debounce drops it,
+// reopening the double-reprovision thrash class (internal#544).
+var RestartDebounceWindow = 60 * time.Second

 // restartByIDDropCounter is incremented every time RestartByID drops a call
 // inside the debounce window. Exposed as a package-level atomic counter so
@@ -536,7 +542,7 @@ func (h *WorkspaceHandler) RestartByID(workspaceID string) {
 		return
 	}
 	// Self-fire debounce: drop (not coalesce) successive RestartByID calls
-	// within restartDebounceWindow of the most recent cycle's start. This
+	// within RestartDebounceWindow of the most recent cycle's start. This
 	// is the load-bearing protection against the 4x reprov thrash class —
 	// coalesceRestart's pending-flag would otherwise drain by running
 	// ANOTHER full cycle of stop+provision on the just-launched EC2 (still
@@ -550,14 +556,14 @@ func (h *WorkspaceHandler) RestartByID(workspaceID string) {
 	if shouldDebounceRestart(workspaceID) {
 		restartByIDDropCounter.Add(1)
 		log.Printf("RestartByID: %s — dropped (within %s self-fire debounce window; total dropped=%d)",
-			workspaceID, restartDebounceWindow, restartByIDDropCounter.Load())
+			workspaceID, RestartDebounceWindow, restartByIDDropCounter.Load())
 		return
 	}
 	coalesceRestart(workspaceID, func() { h.runRestartCycle(workspaceID) })
 }

 // shouldDebounceRestart reports whether the most recent cycle for this
-// workspace started within restartDebounceWindow. Read-only on
+// workspace started within RestartDebounceWindow. Read-only on
 // restartState; the actual restartStartedAt stamp is written in
 // coalesceRestart when running flips false→true.
 func shouldDebounceRestart(workspaceID string) bool {
@@ -571,7 +577,7 @@ func shouldDebounceRestart(workspaceID string) bool {
 	if state.restartStartedAt.IsZero() {
 		return false
 	}
-	return time.Since(state.restartStartedAt) < restartDebounceWindow
+	return time.Since(state.restartStartedAt) < RestartDebounceWindow
 }

 // coalesceRestart implements the pending-flag gate around an arbitrary cycle
@@ -594,7 +600,7 @@ func coalesceRestart(workspaceID string, cycle func()) {
 	}
 	state.running = true
 	// Stamp the start time so the RestartByID debounce can drop any
-	// self-fire probe that hits within restartDebounceWindow. Only the
+	// self-fire probe that hits within RestartDebounceWindow. Only the
 	// false→true edge stamps; the drain-loop's inner cycles re-use the
 	// same start (they're effectively one "restart event" from the
 	// debounce's POV).
@@ -203,9 +203,9 @@ func TestRestartByID_DebounceExpiresAfterWindow(t *testing.T) {
 	const wsID = "self-fire-ws-debounce-release"
 	resetSelfFireState(wsID)

-	orig := restartDebounceWindow
-	restartDebounceWindow = 5 * time.Millisecond
-	defer func() { restartDebounceWindow = orig }()
+	orig := RestartDebounceWindow
+	RestartDebounceWindow = 5 * time.Millisecond
+	defer func() { RestartDebounceWindow = orig }()

 	// Stamp inside the window.
 	sv, _ := restartStates.LoadOrStore(wsID, &restartState{})
@@ -56,6 +56,11 @@ func TestINSERTworkspacesAllowlist(t *testing.T) {
 		// workspace; UUID is server-generated. Caller intent IS to
 		// create, so no idempotency check is needed.
 		"workspace.go:Create": "single-workspace POST, server-generated UUID",
+		// platform_agent.installPlatformAgent: single platform-agent row,
+		// caller-supplied deterministic UUID; INSERT is idempotent via
+		// ON CONFLICT (id) DO UPDATE and runs inside the install transaction.
+		// Pinned by TestIntegration_PlatformAgentInstall_ReparentsRootAndMovesAnchors.
+		"platform_agent.go:installPlatformAgent": "ON CONFLICT (id) DO UPDATE, single row in tx",
 	}

 	actual := map[string]string{}
@@ -13,11 +13,16 @@ import (
 const DefaultMaxConcurrentTasks = 1

 type Workspace struct {
-	ID                 string          `json:"id" db:"id"`
-	Name               string          `json:"name" db:"name"`
-	Role               sql.NullString  `json:"role" db:"role"`
-	Tier               int             `json:"tier" db:"tier"`
-	Status             string          `json:"status" db:"status"`
+	ID     string         `json:"id" db:"id"`
+	Name   string         `json:"name" db:"name"`
+	Role   sql.NullString `json:"role" db:"role"`
+	Tier   int            `json:"tier" db:"tier"`
+	Status string         `json:"status" db:"status"`
+	// Kind: "workspace" (default) or "platform". A "platform" workspace is the
+	// org-level concierge (the platform agent) that sits at the org root and is
+	// the user's default A2A target. See migration
+	// 20260606000000_workspaces_kind + RFC docs/design/rfc-platform-agent.md.
+	Kind               string          `json:"kind" db:"kind"`
 	SourceBundleID     sql.NullString  `json:"source_bundle_id" db:"source_bundle_id"`
 	AgentCard          json.RawMessage `json:"agent_card" db:"agent_card"`
 	URL                sql.NullString  `json:"url" db:"url"`
@@ -63,6 +68,21 @@ func IsValidDeliveryMode(s string) bool {
 	return s == DeliveryModePush || s == DeliveryModePoll
 }

+// Workspace kind constants. Matches the CHECK constraint in migration
+// 20260606000000_workspaces_kind. KindPlatform marks the org-level concierge
+// (the platform agent) which sits at the org root; see
+// docs/design/rfc-platform-agent.md.
+const (
+	KindWorkspace = "workspace"
+	KindPlatform  = "platform"
+)
+
+// IsValidKind reports whether s is a recognised workspace kind. Empty string is
+// NOT valid here — callers resolve the default (KindWorkspace) before calling.
+func IsValidKind(s string) bool {
+	return s == KindWorkspace || s == KindPlatform
+}
+
 type RegisterPayload struct {
 	ID string `json:"id" binding:"required"`
 	// URL is required for push-mode workspaces; optional / unused for
@@ -76,6 +96,12 @@ type RegisterPayload struct {
 	// value on the workspace row, or default to push for new rows".
 	// When set, must be one of DeliveryModePush / DeliveryModePoll.
 	DeliveryMode string `json:"delivery_mode,omitempty"`
+	// Kind is optional. Empty string means "keep the existing value on the
+	// workspace row, or default to KindWorkspace for new rows". When set, must
+	// be one of KindWorkspace / KindPlatform. KindPlatform additionally requires
+	// the row to be its own org root (parent_id IS NULL) and to be the only
+	// platform agent in the org — enforced by the Register handler.
+	Kind string `json:"kind,omitempty"`
 }

 type HeartbeatPayload struct {
@@ -34,6 +34,35 @@ func TestIsValidDeliveryMode_Invalid(t *testing.T) {
 	}
 }

+// ==================== IsValidKind ====================
+
+func TestIsValidKind_Valid(t *testing.T) {
+	for _, k := range []string{KindWorkspace, KindPlatform} {
+		if !IsValidKind(k) {
+			t.Errorf("IsValidKind(%q) = false, want true", k)
+		}
+	}
+}
+
+func TestIsValidKind_Invalid(t *testing.T) {
+	cases := []struct {
+		val  string
+		want bool
+	}{
+		{"", false},          // empty is not valid — callers resolve the default
+		{"platforms", false}, // typo
+		{"Platform", false},  // case-sensitive
+		{"platform ", false}, // trailing space
+		{"root", false},      // not a kind
+		{"user", false},      // the user is implicit, not a workspace kind
+	}
+	for _, tc := range cases {
+		if got := IsValidKind(tc.val); got != tc.want {
+			t.Errorf("IsValidKind(%q) = %v, want %v", tc.val, got, tc.want)
+		}
+	}
+}
+
 // ==================== WorkspaceStatus ====================

 func TestWorkspaceStatus_String(t *testing.T) {
@@ -426,6 +426,12 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
 		adminTokH := handlers.NewAdminWorkspaceTokenHandler()
 		r.POST("/admin/workspaces/:id/tokens", middleware.AdminAuth(db.DB), adminTokH.Create)

+		// Platform agent install — idempotently makes the org-level concierge
+		// the org root (re-parents the existing root + moves org-anchor refs).
+		// Called by the control plane at org provision + existing-org backfill.
+		// (RFC docs/design/rfc-platform-agent.md)
+		r.POST("/admin/org/platform-agent", middleware.AdminAuth(db.DB), handlers.InstallPlatformAgent)
+
 		// Memory
 		memh := handlers.NewMemoryHandler()
 		wsAuth.GET("/memory", memh.List)
@@ -199,6 +199,11 @@ func (s *Scheduler) Start(ctx context.Context) {
 	// entry/exit — those are kept as redundant signals but this pulse is the
 	// one that guarantees liveness freshness regardless of tick state.
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				log.Printf("PANIC recovered in scheduler heartbeat goroutine: %v", r)
+			}
+		}()
 		pulseTicker := time.NewTicker(10 * time.Second)
 		defer pulseTicker.Stop()
 		for {
@@ -638,6 +643,11 @@ func (s *Scheduler) fireSchedule(ctx context.Context, sched scheduleRow) {
 		summary := s.extractResponseSummary(respBody)
 		if summary != "" {
 			go func(wsID, text string) {
+				defer func() {
+					if r := recover(); r != nil {
+						log.Printf("PANIC recovered in broadcast summary goroutine: %v", r)
+					}
+				}()
 				postCtx, postCancel := context.WithTimeout(context.Background(), 30*time.Second)
 				defer postCancel()
 				s.channels.BroadcastToWorkspaceChannels(postCtx, wsID, text)
@@ -0,0 +1,7 @@
+-- Reverse the participant-kind discriminator.
+-- Non-destructive: dropping the column makes every workspace an ordinary
+-- workspace again (the platform agent loses its marker but its row survives).
+DROP INDEX IF EXISTS idx_workspaces_kind;
+ALTER TABLE workspaces DROP CONSTRAINT IF EXISTS workspaces_platform_root_check;
+ALTER TABLE workspaces DROP CONSTRAINT IF EXISTS workspaces_kind_check;
+ALTER TABLE workspaces DROP COLUMN IF EXISTS kind;
@@ -0,0 +1,45 @@
+-- Participant-kind discriminator for the org-level platform agent.
+-- (RFC: docs/design/rfc-platform-agent.md)
+--
+-- 'workspace' (default) = an ordinary workspace / agent.
+-- 'platform'            = the org-level concierge (the "platform agent"). It is
+--                         the single org root (parent_id IS NULL) and the user's
+--                         default A2A chat target. Exactly one per org.
+--
+-- There is no org_id column — an "org" is the parent_id-chain root resolved by
+-- org_scope.go (orgRootID/sameOrg). "platform == org root" and "one platform
+-- agent per org" are therefore enforced in the Register/create handlers, not in
+-- pure SQL. This column is only the discriminator (default-target / billing
+-- exclusion / UX), defined once here and mirrored by the Go constants
+-- models.KindWorkspace / models.KindPlatform.
+--
+-- Backward-compatible: every existing row defaults to 'workspace'. The CHECK is
+-- added NOT VALID then validated so the ALTER can never fail on legacy data.
+ALTER TABLE workspaces
+  ADD COLUMN IF NOT EXISTS kind TEXT NOT NULL DEFAULT 'workspace';
+
+DO $$
+BEGIN
+  IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'workspaces_kind_check') THEN
+    ALTER TABLE workspaces
+      ADD CONSTRAINT workspaces_kind_check CHECK (kind IN ('workspace', 'platform')) NOT VALID;
+    ALTER TABLE workspaces VALIDATE CONSTRAINT workspaces_kind_check;
+  END IF;
+END $$;
+
+-- platform == org root, enforced at the DB level (race-proof). A platform agent
+-- MUST have parent_id IS NULL. Because an org is the subtree under a single
+-- parent_id IS NULL root (org_scope.go) and only a root may be 'platform', this
+-- also structurally guarantees at most ONE platform agent per org. The handler
+-- additionally pre-checks this to return a friendly 409 instead of a raw 23514.
+DO $$
+BEGIN
+  IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'workspaces_platform_root_check') THEN
+    ALTER TABLE workspaces
+      ADD CONSTRAINT workspaces_platform_root_check
+      CHECK (kind <> 'platform' OR parent_id IS NULL) NOT VALID;
+    ALTER TABLE workspaces VALIDATE CONSTRAINT workspaces_platform_root_check;
+  END IF;
+END $$;
+
+CREATE INDEX IF NOT EXISTS idx_workspaces_kind ON workspaces(kind);
@@ -0,0 +1,5 @@
+-- Reverse the approval-gate single-use/dedup columns.
+DROP INDEX IF EXISTS approval_requests_gate_idx;
+ALTER TABLE approval_requests
+  DROP COLUMN IF EXISTS request_hash,
+  DROP COLUMN IF EXISTS consumed_at;
@@ -0,0 +1,18 @@
+-- Single-use + dedup support for the destructive-op approval gate.
+-- (RFC docs/design/rfc-platform-agent.md — Phase 4)
+--
+-- consumed_at: an approval is single-use. Once a destructive op consumes an
+--   approved request, consumed_at is stamped so the same approval can't be
+--   replayed for a second destructive call.
+-- request_hash: a stable hash of (workspace_id, action, context) so a repeated
+--   destructive attempt matches its own pending/approved request instead of
+--   flooding the table with duplicates.
+ALTER TABLE approval_requests
+  ADD COLUMN IF NOT EXISTS consumed_at  TIMESTAMPTZ,
+  ADD COLUMN IF NOT EXISTS request_hash TEXT;
+
+-- Hot path: the gate looks up an approved + unconsumed row matching
+-- (workspace_id, action, request_hash). Partial index keeps that O(log live).
+CREATE INDEX IF NOT EXISTS approval_requests_gate_idx
+  ON approval_requests (workspace_id, action, request_hash)
+  WHERE status = 'approved' AND consumed_at IS NULL;