feat(mcp): publish an unpublished local commit on a registered issue worktree (Closes #812)

Entry point B of #812 is the state where an author's work has already advanced to a local commit: the worktree is registered, clean, on the issue branch, and carries the only copy of the implementation, but the branch has never been published. Two individually correct predicates close a cycle around it: exact-owner lease renewal refuses without an observable remote head, and every publication path is lock-derived under #618, so nothing can create that remote head without first holding the lock renewal would grant.

This adds the missing operation. gitea_publish_unpublished_issue_branch publishes an already-committed local head to its remote branch, so exact-owner renewal has the evidence its model requires. Publication is the whole of its authority: it renews, reclaims, rebinds, and clears nothing.

Why this is not a lock bypass: the operation can only publish a branch whose durable issue-lock record already names the caller as claimant. Ownership is read from the lock file, never asserted by the caller. Guard strictness is unchanged (AC15): a dirty tree, an untracked file the commit does not carry, an unregistered worktree, a non-issue or stable branch, a changed local HEAD, a remote head that is not an ancestor of the commit, a competing open PR for the same issue on another branch, and any declared-hash mismatch each fail closed. The refspec names the commit SHA explicitly and never forces.

Record separation (AC23): the durable issue-lock file and the control-plane workflow lease are distinct records. This reads the former as ownership evidence and writes neither.

Truthful process evidence (AC24): the recorded owner pid's liveness is never consulted or asserted. A regression pins the recorded pid to a live process, proves publication still succeeds, and proves expired-lock reclaim still refuses for that same pid.

Scope is AC20 only. AC21 cannot unblock on its own, so the two are separable and only the smaller one is implemented here.

Tests: 36 new cases against synthetic fixtures only, using a real git repository with a real local bare remote so publication and read-after-write verification are genuinely executed rather than mocked. AC17 is honoured and a regression asserts the protected worktree is never referenced.

Full suite: 11 failed, 4354 passed, 6 skipped, 533 subtests passed. The 11 failures are the documented pre-existing drift baseline at 9eb0f29, unchanged in count and identity.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

anti_stomp_preflight.py

@@ -87,6 +87,7 @@ MUTATION_TASKS = frozenset({
     "edit_pr",
     "commit_files",
     "gitea_commit_files",
+    "publish_unpublished_branch",
     "delete_branch",
     "cleanup_merged_pr_branch",
     "cleanup_stale_claims",

branch_publish.py

@@ -0,0 +1,591 @@
+"""Publish an unpublished local commit on a registered issue worktree (#812 AC20).
+
+Entry point B of #812 is the state where an author's work has already advanced
+to a local commit: the worktree is registered, clean, on the issue branch, and
+carries the only copy of the implementation, but the branch has never been
+published. That state deadlocks, because two individually correct predicates
+close a cycle:
+
+* ``issue_lock_renewal.assess_exact_owner_lease_renewal`` refuses to renew an
+  expired lease without an observable remote head — an unpublished branch has
+  none.
+* Every publication path (``gitea_commit_files``, ``gitea_create_pr``) derives
+  its workspace from the author issue lock under #618, so nothing can create
+  that remote head without first holding the lock.
+
+This module supplies the missing operation: it publishes an *already committed*
+local head to the remote branch, so exact-owner renewal has the evidence it
+requires. It deliberately does **not** renew, reclaim, rebind, or clear any
+lock. Publication is the whole of its authority.
+
+Why this is not a lock bypass
+-----------------------------
+The operation can only publish a branch whose **durable issue-lock record
+already names the caller as claimant**. Ownership is read from the lock file on
+disk (``issue_lock_store``), never from a caller-supplied flag, so the tool
+cannot manufacture a claim it does not already hold. Nothing here weakens the
+#510/#618/#713 guards: a dirty tree, an unregistered worktree, a foreign
+claimant, a changed HEAD, or a divergent remote head each refuse, exactly as
+they do today. The only thing this adds is the ability to make an existing,
+owned, committed, clean branch observable on the remote.
+
+Separation of records (#812 AC23)
+---------------------------------
+The durable **issue-lock file** and the control-plane **workflow lease** are
+distinct records. This module reads the former as ownership evidence and writes
+neither. Publishing changes remote git state only; no lock is renewed,
+abandoned, reclaimed, or generation-bumped here.
+
+Process evidence (#812 AC24)
+----------------------------
+Liveness of the lock's recorded pid is **not consulted**. That is deliberate:
+the recorded pid routinely belongs to the long-running MCP daemon rather than to
+an active author client, and the existing reclaim predicate
+(``assess_expired_lock_reclaim``) can never be satisfied while that daemon runs.
+Publication does not require the recording process to be dead, so this module
+never asserts, infers, or depends on a process being dead. Ownership is proven
+by identity and profile match against the recorded claimant instead.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import subprocess
+
+from reviewer_worktree import parse_dirty_tracked_files
+from stable_branch_push_guard import is_stable_ref, redact_command
+
+# Assessment outcomes.
+PUBLISH_SANCTIONED = "publish_sanctioned"
+ALREADY_PUBLISHED = "already_published"
+REFUSED = "refused"
+
+#: Implementation branches must stay traceable to their issue (#713 lineage).
+ISSUE_BRANCH_RE = re.compile(r"^(fix|feat|docs|chore)/issue-(\d+)-.+$")
+
+_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+
+
+def _text(value: object) -> str:
+    return value.strip() if isinstance(value, str) else ""
+
+
+def _realpath(value: str | None) -> str | None:
+    path = _text(value)
+    return os.path.realpath(path) if path else None
+
+
+def parse_untracked_files(porcelain_status: str) -> list[str]:
+    """Return untracked paths from ``git status --porcelain`` output.
+
+    ``reviewer_worktree.parse_dirty_tracked_files`` deliberately skips ``??``
+    entries. Publication needs both halves: an untracked file in the worktree is
+    unpublished content that the commit does not carry, so publishing would
+    silently leave it behind.
+    """
+    untracked: list[str] = []
+    for line in (porcelain_status or "").splitlines():
+        if not line.startswith("??"):
+            continue
+        path = line[2:].strip()
+        if path:
+            untracked.append(path)
+    return untracked
+
+
+def hash_worktree_files(worktree_path: str, paths) -> dict[str, str | None]:
+    """SHA-256 each path under *worktree_path*; ``None`` when unreadable."""
+    root = _text(worktree_path)
+    hashes: dict[str, str | None] = {}
+    for rel in paths or ():
+        rel_text = _text(rel)
+        if not rel_text:
+            continue
+        full = os.path.join(root, rel_text)
+        try:
+            with open(full, "rb") as handle:
+                digest = hashlib.sha256()
+                for chunk in iter(lambda: handle.read(65536), b""):
+                    digest.update(chunk)
+            hashes[rel_text] = digest.hexdigest()
+        except OSError:
+            hashes[rel_text] = None
+    return hashes
+
+
+def read_remote_branch_head(
+    worktree_path: str, remote_name: str, branch_name: str
+) -> dict:
+    """Observe the remote head for *branch_name*, read-only.
+
+    ``probe_ok`` False means git could not answer at all. That is kept distinct
+    from "the branch does not exist": an unobservable remote must fail closed
+    rather than be mistaken for an absent branch, because the two lead to
+    opposite dispositions.
+    """
+    path = _text(worktree_path)
+    remote = _text(remote_name)
+    branch = _text(branch_name)
+    result: dict = {
+        "probe_ok": False,
+        "remote_branch_exists": False,
+        "remote_head_sha": None,
+        "reasons": [],
+    }
+    if not (path and remote and branch):
+        result["reasons"].append(
+            "remote head probe requires a worktree path, remote name, and branch"
+        )
+        return result
+    try:
+        res = subprocess.run(
+            ["git", "-C", path, "ls-remote", remote, f"refs/heads/{branch}"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError as exc:  # git unavailable — fail closed, never assume absent
+        result["reasons"].append(f"remote head probe could not run: {exc}")
+        return result
+    if res.returncode != 0:
+        result["reasons"].append(
+            f"remote head probe failed for '{branch}' on remote '{remote}'"
+        )
+        return result
+
+    result["probe_ok"] = True
+    for line in (res.stdout or "").splitlines():
+        parts = line.split()
+        if len(parts) >= 2 and parts[1] == f"refs/heads/{branch}":
+            result["remote_branch_exists"] = True
+            result["remote_head_sha"] = parts[0].strip()
+            break
+    return result
+
+
+def read_is_ancestor(
+    worktree_path: str, ancestor_sha: str, descendant_sha: str
+) -> dict:
+    """Observe whether *ancestor_sha* is an ancestor of *descendant_sha*."""
+    path = _text(worktree_path)
+    ancestor = _text(ancestor_sha)
+    descendant = _text(descendant_sha)
+    result: dict = {"probe_ok": False, "is_ancestor": False, "reasons": []}
+    if not (path and ancestor and descendant):
+        result["reasons"].append(
+            "ancestry probe requires a worktree path and both commit SHAs"
+        )
+        return result
+    try:
+        present = subprocess.run(
+            ["git", "-C", path, "rev-parse", "--verify", "--quiet",
+             f"{ancestor}^{{commit}}"],
+            capture_output=True, text=True, check=False,
+        )
+        if present.returncode != 0:
+            result["reasons"].append(
+                f"remote head {ancestor} is not present locally, so it cannot be "
+                "proven an ancestor of the commit being published"
+            )
+            return result
+        res = subprocess.run(
+            ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
+            capture_output=True, text=True, check=False,
+        )
+    except OSError as exc:
+        result["reasons"].append(f"ancestry probe could not run: {exc}")
+        return result
+    result["probe_ok"] = res.returncode in (0, 1)
+    result["is_ancestor"] = res.returncode == 0
+    return result
+
+
+def assess_unpublished_commit_publication(
+    existing_lock,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    expected_head: str,
+    remote: str,
+    org: str,
+    repo: str,
+    identity: str | None,
+    profile: str | None,
+    worktree_state,
+    worktree_registered: bool | None = None,
+    remote_probe=None,
+    ancestry=None,
+    competing_open_prs=(),
+    expected_file_hashes=None,
+    observed_file_hashes=None,
+) -> dict:
+    """Decide whether an unpublished local commit may be published (#812 AC20).
+
+    Pure predicate. Every input is either a caller-declared expectation that
+    must be *matched* against observation, or a server-side observation. No
+    caller-supplied boolean is accepted as proof of ownership, liveness, or
+    eligibility: ``existing_lock`` comes from the durable lock file and the
+    git/PR state is observed by the server.
+
+    The single mutating disposition it can return is "publish this exact commit
+    to this exact branch". It never sanctions renewal, reclamation, force
+    updates, history rewriting, or publication of uncommitted content.
+    """
+    reasons: list[str] = []
+    branch = _text(branch_name)
+    head = _text(expected_head).lower()
+    workspace = _realpath(worktree_path)
+    state = worktree_state if isinstance(worktree_state, dict) else {}
+    lock = existing_lock if isinstance(existing_lock, dict) else None
+
+    evidence: dict = {
+        "issue_number": issue_number,
+        "branch_name": branch or None,
+        "worktree_path": workspace,
+        "expected_head": head or None,
+        "remote": _text(remote) or None,
+        "org": _text(org) or None,
+        "repo": _text(repo) or None,
+        "identity": _text(identity) or None,
+        "profile": _text(profile) or None,
+        "lock_record_present": lock is not None,
+        "recorded_claimant": None,
+        "recorded_branch": None,
+        "recorded_worktree": None,
+        "lock_generation": None,
+        "local_head_sha": _text(state.get("head_sha")) or None,
+        "current_branch": _text(state.get("current_branch")) or None,
+        "dirty_tracked_files": [],
+        "untracked_files": [],
+        "worktree_registered": worktree_registered,
+        "remote_branch_exists": None,
+        "remote_head_sha": None,
+        "fast_forward_from_remote": None,
+        "competing_open_prs": [],
+        "file_hashes_verified": None,
+        "hash_mismatches": [],
+        # Recorded explicitly so no reader mistakes silence for a liveness
+        # claim, and so the audit shows which records were left alone (AC23/AC24).
+        "owner_pid_liveness_consulted": False,
+        "workflow_lease_touched": False,
+        "issue_lock_record_mutated": False,
+    }
+
+    # ── declared shape ────────────────────────────────────────────────────
+    if not branch:
+        reasons.append("branch name not declared; fail closed")
+    if not head:
+        reasons.append(
+            "expected_head not declared; publication must name the exact commit"
+        )
+    elif not _SHA_RE.match(head):
+        reasons.append(
+            f"expected_head '{head}' is not a full 40-character commit SHA; "
+            "abbreviated or symbolic revisions are refused"
+        )
+    if not workspace:
+        reasons.append("worktree path not declared; fail closed")
+
+    if branch:
+        match = ISSUE_BRANCH_RE.match(branch)
+        if not match:
+            reasons.append(
+                f"branch '{branch}' is not an issue-linked implementation branch "
+                "((fix|feat|docs|chore)/issue-<number>-<description>); fail closed"
+            )
+        elif int(match.group(2)) != int(issue_number):
+            reasons.append(
+                f"branch '{branch}' does not carry issue number {issue_number}; "
+                "fail closed"
+            )
+        if is_stable_ref(branch):
+            reasons.append(
+                f"refusing to publish stable branch '{branch}'; this operation "
+                "publishes issue branches only"
+            )
+
+    # ── ownership: durable issue-lock record only (AC8, AC20, AC24) ───────
+    if lock is None:
+        reasons.append(
+            "no durable issue-lock record for this issue; publication requires an "
+            "existing recorded claim naming the caller, so this operation cannot "
+            "be used to bypass the author lock"
+        )
+    else:
+        lease = lock.get("work_lease")
+        lease = lease if isinstance(lease, dict) else {}
+        claimant = lease.get("claimant")
+        claimant = claimant if isinstance(claimant, dict) else {}
+        recorded_user = _text(claimant.get("username"))
+        recorded_profile = _text(claimant.get("profile"))
+        recorded_branch = _text(lock.get("branch_name")) or _text(lease.get("branch"))
+        recorded_worktree = _realpath(
+            _text(lock.get("worktree_path")) or _text(lease.get("worktree_path"))
+        )
+        evidence["recorded_claimant"] = {
+            "username": recorded_user or None,
+            "profile": recorded_profile or None,
+        }
+        evidence["recorded_branch"] = recorded_branch or None
+        evidence["recorded_worktree"] = recorded_worktree
+        try:
+            evidence["lock_generation"] = int(lock.get("lock_generation") or 0)
+        except (TypeError, ValueError):
+            evidence["lock_generation"] = 0
+
+        try:
+            recorded_issue = int(lock.get("issue_number") or 0)
+        except (TypeError, ValueError):
+            recorded_issue = 0
+        if recorded_issue != int(issue_number):
+            reasons.append(
+                f"durable lock records issue {lock.get('issue_number')}, not "
+                f"{issue_number}; ambiguous ownership, fail closed"
+            )
+        for field, declared in (
+            ("remote", _text(remote)),
+            ("org", _text(org)),
+            ("repo", _text(repo)),
+        ):
+            recorded = _text(lock.get(field))
+            if recorded and declared and recorded != declared:
+                reasons.append(
+                    f"durable lock records {field} '{recorded}' but the request "
+                    f"declares '{declared}'; repository mismatch, fail closed"
+                )
+        if recorded_branch and branch and recorded_branch != branch:
+            reasons.append(
+                f"durable lock records branch '{recorded_branch}' but the request "
+                f"declares '{branch}'; fail closed"
+            )
+        if recorded_worktree and workspace and recorded_worktree != workspace:
+            reasons.append(
+                f"durable lock records worktree '{recorded_worktree}' but the "
+                f"request declares '{workspace}'; fail closed"
+            )
+        if not recorded_user or not recorded_profile:
+            reasons.append(
+                "durable lock does not record a claimant username and profile; "
+                "ownership cannot be proven, fail closed"
+            )
+        else:
+            if recorded_user != _text(identity):
+                reasons.append(
+                    f"durable lock claimant '{recorded_user}' is not the acting "
+                    f"identity '{_text(identity) or '(unknown)'}'; foreign claim, "
+                    "fail closed"
+                )
+            if recorded_profile != _text(profile):
+                reasons.append(
+                    f"durable lock claimant profile '{recorded_profile}' is not "
+                    f"the active profile '{_text(profile) or '(unknown)'}'; "
+                    "fail closed"
+                )
+
+    # ── worktree: registered, on-branch, clean, at the expected commit ────
+    if worktree_registered is False:
+        reasons.append(
+            f"worktree '{workspace}' is not listed in git worktree list; #713 "
+            "requires a genuinely registered worktree, fail closed"
+        )
+
+    current_branch = _text(state.get("current_branch"))
+    if not current_branch:
+        reasons.append("worktree branch could not be observed; fail closed")
+    elif branch and current_branch != branch:
+        reasons.append(
+            f"worktree is on branch '{current_branch}', not '{branch}'; fail closed"
+        )
+
+    porcelain = state.get("porcelain_status") or ""
+    dirty_tracked = parse_dirty_tracked_files(porcelain)
+    untracked = parse_untracked_files(porcelain)
+    evidence["dirty_tracked_files"] = dirty_tracked
+    evidence["untracked_files"] = untracked
+    if dirty_tracked:
+        reasons.append(
+            "worktree has dirty tracked files, so the commit is not the whole of "
+            f"the work: {', '.join(dirty_tracked)}. This operation publishes an "
+            "existing clean commit only; uncommitted content is out of scope"
+        )
+    if untracked:
+        reasons.append(
+            "worktree has untracked files that the commit does not carry: "
+            f"{', '.join(untracked)}. Publishing would silently leave them "
+            "behind; fail closed"
+        )
+
+    local_head = _text(state.get("head_sha")).lower()
+    if not local_head:
+        reasons.append("local HEAD could not be observed; fail closed")
+    elif head and local_head != head:
+        reasons.append(
+            f"worktree HEAD is {local_head} but the request declares {head}; the "
+            "local commit changed since it was recorded, fail closed"
+        )
+
+    # ── remote state ──────────────────────────────────────────────────────
+    probe = remote_probe if isinstance(remote_probe, dict) else {}
+    already_published = False
+    if not probe.get("probe_ok"):
+        reasons.append(
+            "remote branch head could not be observed; publication must not "
+            "proceed against an unknown remote state, fail closed"
+        )
+        reasons.extend(probe.get("reasons") or [])
+    else:
+        remote_exists = bool(probe.get("remote_branch_exists"))
+        remote_head = _text(probe.get("remote_head_sha")).lower() or None
+        evidence["remote_branch_exists"] = remote_exists
+        evidence["remote_head_sha"] = remote_head
+        if remote_exists and remote_head and head:
+            if remote_head == head:
+                already_published = True
+                evidence["fast_forward_from_remote"] = True
+            else:
+                anc = ancestry if isinstance(ancestry, dict) else {}
+                is_anc = bool(anc.get("probe_ok")) and bool(anc.get("is_ancestor"))
+                evidence["fast_forward_from_remote"] = is_anc
+                if not is_anc:
+                    reasons.append(
+                        f"remote branch '{branch}' already exists at {remote_head}, "
+                        f"which is not an ancestor of {head}; publishing would "
+                        "discard or rewrite published history, fail closed"
+                    )
+                    reasons.extend(anc.get("reasons") or [])
+        elif remote_exists and not remote_head:
+            reasons.append(
+                f"remote branch '{branch}' exists but its head could not be read; "
+                "fail closed"
+            )
+
+    # ── competing claims ──────────────────────────────────────────────────
+    competing = [p for p in (competing_open_prs or ()) if p]
+    evidence["competing_open_prs"] = list(competing)
+    if competing:
+        reasons.append(
+            f"open pull request(s) {competing} already claim issue {issue_number} "
+            "or this branch; ambiguous ownership, fail closed"
+        )
+
+    # ── content verification before publication ───────────────────────────
+    if expected_file_hashes:
+        observed = (
+            observed_file_hashes if isinstance(observed_file_hashes, dict) else {}
+        )
+        mismatches: list[str] = []
+        for path, expected_digest in dict(expected_file_hashes).items():
+            actual = observed.get(path)
+            if actual is None:
+                mismatches.append(f"{path}: missing or unreadable in the worktree")
+            elif _text(actual).lower() != _text(expected_digest).lower():
+                mismatches.append(
+                    f"{path}: expected {expected_digest}, observed {actual}"
+                )
+        evidence["hash_mismatches"] = mismatches
+        evidence["file_hashes_verified"] = not mismatches
+        if mismatches:
+            reasons.append(
+                "declared content hashes do not match the worktree: "
+                + "; ".join(mismatches)
+                + ". Refusing to publish content that is not what was recorded"
+            )
+
+    if reasons:
+        return {
+            "outcome": REFUSED,
+            "publish_sanctioned": False,
+            "already_published": False,
+            "reasons": reasons,
+            "evidence": evidence,
+        }
+    return {
+        "outcome": ALREADY_PUBLISHED if already_published else PUBLISH_SANCTIONED,
+        # Idempotent retry: a remote head that already equals the assessed commit
+        # needs no second push, so the caller verifies instead of acting.
+        "publish_sanctioned": not already_published,
+        "already_published": already_published,
+        "reasons": [],
+        "evidence": evidence,
+    }
+
+
+def publish_commit_to_remote_branch(
+    *,
+    worktree_path: str,
+    remote_name: str,
+    branch_name: str,
+    expected_head: str,
+) -> dict:
+    """Send exactly *expected_head* to ``refs/heads/<branch_name>``.
+
+    The refspec names the commit SHA explicitly rather than ``HEAD`` or the
+    local branch, so what lands is the commit that was assessed and nothing
+    else. No force, no lease, no ``+`` prefix: a non-fast-forward is rejected by
+    git itself, the last of several independent guards against overwriting
+    published history.
+    """
+    path = _text(worktree_path)
+    remote = _text(remote_name)
+    branch = _text(branch_name)
+    head = _text(expected_head)
+    result: dict = {
+        "success": False,
+        "pushed_ref": f"refs/heads/{branch}" if branch else None,
+        "pushed_sha": head or None,
+        "stderr": None,
+        "reasons": [],
+    }
+    if not (path and remote and branch and head):
+        result["reasons"].append(
+            "publication requires a worktree path, remote, branch, and commit SHA"
+        )
+        return result
+
+    refspec = f"{head}:refs/heads/{branch}"
+    try:
+        res = subprocess.run(
+            ["git", "-C", path, "push", remote, refspec],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError as exc:
+        result["reasons"].append(f"publication could not run: {exc}")
+        return result
+
+    if res.returncode != 0:
+        # Redact before surfacing: failures can echo credentialed remote URLs.
+        result["stderr"] = redact_command(res.stderr or "")
+        result["reasons"].append(
+            f"publication of {head} to '{branch}' on remote '{remote}' failed"
+        )
+        return result
+
+    result["success"] = True
+    return result
+
+
+def verify_published_head(
+    *, worktree_path: str, remote_name: str, branch_name: str, expected_head: str
+) -> dict:
+    """Read-after-write: confirm the remote head equals *expected_head* (AC20)."""
+    probe = read_remote_branch_head(worktree_path, remote_name, branch_name)
+    head = _text(expected_head).lower()
+    observed = _text(probe.get("remote_head_sha")).lower() or None
+    verified = bool(head) and bool(probe.get("probe_ok")) and observed == head
+    reasons: list[str] = list(probe.get("reasons") or [])
+    if probe.get("probe_ok") and not verified:
+        reasons.append(
+            "read-after-write verification failed: remote head is "
+            f"{observed or '(absent)'}, expected {head}"
+        )
+    return {
+        "verified": verified,
+        "remote_head_sha": observed,
+        "expected_head": head or None,
+        "reasons": reasons,
+    }

docs/mcp-tool-inventory.md

@@ -120,6 +120,7 @@ that gates each call, not which tools exist.
 - `gitea_observability_list_projects`
 - `gitea_observability_reconcile_incident`
 - `gitea_post_heartbeat`
+- `gitea_publish_unpublished_issue_branch`
 - `gitea_quarantine_contaminated_review`
 - `gitea_reclaim_expired_workflow_lease`
 - `gitea_reconcile_already_landed_pr`

docs/webui-authz-audit.md

@@ -1,295 +0,0 @@
-# Web console authorization, RBAC, redaction, and audit model (#633)
-
-**Phase 1. Read-only. This document defines the model that future console
-writes must pass through; it enables none of them.**
-
-The MVP deployment boundary ([`webui-deployment.md`](webui-deployment.md), #435)
-documents internal-only serving and states plainly that MVP authentication is
-*none* — protection comes from network placement. That is adequate while every
-route is a GET, and inadequate the moment a gated write ships. This document
-and the three modules it describes land **before** any write exists, so no
-Phase 2 action can be added without an authority to check it against.
-
-| Concern | Module |
-|---------|--------|
-| Identity, roles, authorization decision | `webui/console_authz.py` |
-| Secret redaction for every surface | `webui/console_redaction.py` |
-| Audit event schema, retention, sink | `webui/console_audit.py` |
-| Machine-readable publication | `GET /api/console/security-model` |
-
-Two invariants hold everywhere and are non-negotiable for every child of #631:
-
-1. **No secrets reach the browser.** Credentials are resolved server-side and
-   redacted before any payload, page, log line, or audit record leaves.
-2. **No ungated mutations.** Authorization is necessary but never sufficient;
-   execution stays disabled until the Phase 2 framework ships.
-
-## Identity sources
-
-The console performs *authorization*. Authentication is delegated, because a
-console that mints its own sessions is a credential store, and this one must
-not be.
-
-| Source | Mode value | Authenticated | Shared host | Phase |
-|--------|-----------|---------------|-------------|-------|
-| None | `none` (default) | No — anonymous, capped at `viewer` | No | 1 |
-| Local dev | `local-dev` / `local_dev` | Yes, **asserted not verified** | No | 1 |
-| Access proxy | `access-proxy` / `access_proxy` | Yes, asserted by trusted proxy | Yes | 2 |
-
-Selected by `WEBUI_AUTH_MODE`. An unrecognised value falls back to `none`
-rather than erroring open.
-
-**Access-proxy mode** reads the subject from the
-`Cf-Access-Authenticated-User-Email` header, set by Cloudflare Access, WARP, or
-an equivalent org portal that terminates authentication in front of the
-console. If the header is absent the request did not traverse the proxy, so the
-principal degrades to anonymous — it is never trusted by default.
-
-The **role is always server-side configuration**, never a client assertion. It
-comes from `WEBUI_ROLE_MAP`, a JSON object of subject → role:
-
-```json
-{"[email protected]": "operator", "[email protected]": "controller"}
-```
-
-An unmapped subject gets `viewer`. Malformed JSON yields an empty map, so
-everyone gets `viewer` — a parse failure loses authority rather than granting
-it.
-
-Full SSO is explicitly a non-goal of this issue.
-
-## Role matrix
-
-Four roles, ordered least to most authority. Each role inherits every lower
-role's actions; the table states the *minimum* rank required.
-
-| Role | Authority |
-|------|-----------|
-| `viewer` | Read every console view. No write, ever, in any phase. |
-| `operator` | Viewer, plus author-class work: claim, comment, open a PR. |
-| `controller` | Operator, plus reviewer/merger-class decisions on a PR. |
-| `admin` | Controller, plus destructive and policy-editing actions. |
-
-`viewer` holds the empty write set by construction, and a test asserts it stays
-empty.
-
-## Privileged actions
-
-Every console action maps to a `task_key` in `task_capability_map.py`, the same
-single source of truth `gitea_resolve_task_capability` and the MCP tool gates
-use. The console therefore cannot invent an authority the MCP layer does not
-already define, and a regression test asserts each mapping matches.
-
-| Action | Minimum role | Class | MCP permission | Confirm | Dual control | Break-glass | Phase |
-|--------|--------------|-------|----------------|---------|--------------|-------------|-------|
-| `claim_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
-| `comment_issue` | operator | gated_write | `gitea.issue.comment` | Yes | No | No | 2 |
-| `create_issue` | operator | gated_write | `gitea.issue.create` | Yes | No | No | 2 |
-| `comment_pr` | operator | gated_write | `gitea.pr.comment` | Yes | No | No | 2 |
-| `create_pr` | operator | gated_write | `gitea.pr.create` | Yes | No | No | 2 |
-| `review_pr` | controller | privileged | `gitea.pr.review` | Yes | No | No | 3 |
-| `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 |
-| `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 |
-| `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 |
-
-**Dual control** means the acting principal may not be the sole authority: a
-second distinct principal must confirm. **Break-glass** means the action is
-expected to be unavailable in normal operation and its use is retained for two
-years. Both are declared here and enforced by the Phase 2 framework; Phase 1
-records the requirement on every decision so the framework cannot ship without
-honouring it.
-
-`delete_branch` is admin-only rather than controller because it is the one
-irreversible action in the set.
-
-### Authorization decision
-
-`authorize(action_id, principal, for_execution=False)` returns a decision
-record and **denies by default**. The deny reasons are closed and enumerated:
-
-| Reason code | Meaning |
-|-------------|---------|
-| `unknown_action` | No such console action is registered. |
-| `unauthenticated` | The principal is anonymous. |
-| `unknown_role` | The role is not in the matrix. |
-| `insufficient_role` | The role ranks below the action's minimum. |
-| `phase_not_active` | Execution requested for an action whose phase is not open. |
-| `allowed_preview_only` | Authorized — preview only, execution still disabled. |
-
-There is no implicit allow branch. Even the allow result reports
-`execution_enabled: false` while the console is in Phase 1, so no caller can
-read an allow as permission to mutate.
-
-## Secret redaction
-
-One pass applies to **API payloads, rendered HTML, server logs, and audit
-records** — the four surfaces where a credential could escape.
-
-Redaction reuses `gitea_audit.redact` rather than forking it: that remains the
-authority for secret-looking dict keys, `Authorization` material, and raw URLs.
-The console layer then applies its own patterns:
-
-Each rule below matches an *assignment form*: the named key, followed by `=` or
-`:`, followed by the value. The keys are listed bare rather than spelled out as
-complete assignments, because this document is itself scanned by
-`scan_for_secrets` — writing the examples in full assignment form would make the
-documentation trip the very detectors it documents.
-
-| Rule | Catches (as an assignment) |
-|------|----------------------------|
-| `credential_assignment` | `token`, `password`, `passwd`, `secret`, `api_key`, `access_key`, `client_secret`, `private_key` |
-| `credential_env_assignment` | `GITEA_TOKEN`, `GITEA_PASS`, `GITEA_PASSWORD` and suffixed variants |
-| `keychain_reference` | `keychain:` entry references |
-| `keychain_command` | macOS `security` keychain lookups (`find-generic-password`, `find-internet-password`) |
-| `private_key_block` | PEM `BEGIN ... PRIVATE KEY` blocks |
-| `json_web_token` | Three-segment `eyJ...` JWTs |
-| `bearer_credential` | `Bearer` / `Basic` credentials |
-
-Assignments keep the key and replace only the value, so an operator can still
-see *what* was removed. Two behaviours are deliberate:
-
-- **Fail closed.** A value that cannot be redacted becomes `[REDACTED]`
-  outright rather than being emitted raw. Redaction never raises.
-- **Redact before persist.** `console_audit.build_event` redacts before
-  serialization, and `write_event` re-scans and **drops** any record that still
-  trips a detector. An unredacted record is never durable.
-
-`scan_for_secrets` is the assertion helper: it returns the detector names still
-matching a payload, and already-redacted hits are not findings. Tests use it to
-prove the published policy, the security-model endpoint, and this document
-itself carry no secret material.
-
-## Audit event schema
-
-`gitea_audit` records MCP-side *mutations* — which profile and Gitea user
-performed which tool call. It has no console actor, no identity source, no
-correlation identifier, and no retention class, and an authorization **denial**
-is not a mutation, so it would never appear there at all. The console record is
-additive, not a replacement: a Phase 2 action emits both, joined on
-`correlation.request_id`.
-
-Required fields, all asserted by tests so an edit cannot quietly drop one:
-
-| Field | Content |
-|-------|---------|
-| `schema_version` | Currently `1`. |
-| `event_id` | Unique per record. |
-| `timestamp` | Timezone-aware ISO-8601, UTC. |
-| `actor` | `subject`, `role`, `identity_source`, `authenticated`. |
-| `action` | Console action id. |
-| `action_class` | `gated_write`, `privileged`, `destructive`, or `unknown`. |
-| `target` | `{kind, ref}`, e.g. `{"kind": "pr", "ref": "#123"}`. |
-| `result` | `allowed`, `denied`, `previewed`, `failed`, `succeeded`. |
-| `reason_code` | The authorization reason code above. |
-| `correlation` | `request_id`, `session_id`, `mcp_task`, `mcp_permission`. |
-| `retention` | `class`, `days`, `expires_at`. |
-| `redacted` | Always `true`; records are redacted at build time. |
-
-An unrecognised `result` degrades to `failed` rather than being stored
-verbatim.
-
-The sink is an append-only JSON Lines file named by
-`WEBUI_CONSOLE_AUDIT_LOG`. It is **off by default**: with the variable unset,
-events are still built — so callers and tests exercise the schema — but nothing
-is written. Auditing never raises; a failed write returns `False` rather than
-breaking the request it describes.
-
-## Retention
-
-| Class | Applies to | Default |
-|-------|-----------|---------|
-| `standard` | Routine gated writes | 90 days |
-| `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days |
-| `break_glass` | `merge_pr`, `delete_branch` | 730 days |
-
-Each record carries its own class, day count, and computed `expires_at`, so
-retention is auditable per record rather than inferred from file age. An
-**unknown action is retained as privileged, not standard** — for a safety
-control the conservative direction is to keep the record longer.
-
-Nothing in this module updates or deletes. Expiry is enforced by an
-operator-run policy against `expires_at`, never by the console silently
-rewriting its own history.
-
-## Phase 2 integration
-
-Phase 2 opens gated writes. It must reuse this model rather than introduce a
-second one. The integration points are already wired and observable:
-
-- **`GET /api/actions/{action_id}/preview`** attaches an `authorization` block
-  to the existing preview payload and records a `previewed` audit event.
-- **`POST /api/actions/{action_id}/attempt`** attaches the same block and
-  records a `denied` event. The terminal outcome is unchanged — the MVP
-  registry in `webui/gated_actions.py` still fails closed for every action — so
-  Phase 1 cannot loosen anything. Phase 2 enforces on this same decision
-  instead of adding a parallel check.
-- **`GET /api/console/security-model`** publishes the RBAC matrix, redaction
-  policy, and audit policy as JSON for operators and tests.
-
-To open Phase 2, a child issue must: raise `ACTIVE_PHASE`, implement the
-confirmation and dual-control flow the matrix already declares, emit a
-`succeeded` or `failed` record alongside the `gitea_audit` mutation record, and
-keep `viewer` unable to reach any of it. Turning on execution without the
-confirmation flow contradicts a declared requirement and is a review failure,
-not a shortcut.
-
-## Local-dev mode
-
-`WEBUI_AUTH_MODE=local-dev` reads the principal straight from the environment:
-
-| Variable | Purpose |
-|----------|---------|
-| `WEBUI_DEV_SUBJECT` | Subject string; absent ⇒ anonymous |
-| `WEBUI_DEV_ROLE` | One of `viewer`, `operator`, `controller`, `admin`; unrecognised ⇒ `viewer` |
-
-**INSECURE — this mode is for loopback development only.** The subject and role
-are *asserted by the developer running the process and verified by nothing*.
-Anyone able to set an environment variable on the host is an `admin`, and
-anyone able to reach the port inherits that principal. It provides no
-authentication whatsoever; it exists so Phase 2 authorization paths can be
-exercised without standing up a proxy.
-
-Never enable local-dev mode on a non-loopback bind. Combining it with
-`WEBUI_ALLOW_PUBLIC_BIND=1` or `WEBUI_ALLOW_REMOTE_BIND=1` publishes an
-unauthenticated admin console.
-
-For anything beyond a laptop use `access-proxy` mode behind Cloudflare Access,
-WARP, or a VPN, as [`webui-deployment.md`](webui-deployment.md) requires.
-
-### Probe authentication
-
-`WEBUI_REQUIRE_PROBE_AUTH=1` declares that non-public probes should require an
-authenticated principal. It is **opt-in**: the default is off so the MVP
-`/health` contract is unchanged.
-
-**This flag is declarative in Phase 1 and enforces nothing today.**
-`console_authz.probe_auth_required()` reports the operator's intent, and no
-route consults it — setting the variable does not currently change the
-behaviour of `/health` or any other endpoint. It is published here so the Phase
-2 action framework has a declared policy to honour rather than inventing a
-second one, exactly as `ACTIVE_PHASE` gates execution while the matrix is
-already declared. A regression test pins this "declared, not enforced" status,
-so wiring it later is a deliberate change rather than a silent one.
-
-Until Phase 2 wires it, probe protection rests on network placement alone, as
-[`webui-deployment.md`](webui-deployment.md) (#435) states.
-
-## Environment variables
-
-| Variable | Default | Purpose |
-|----------|---------|---------|
-| `WEBUI_AUTH_MODE` | `none` | Identity source selection |
-| `WEBUI_DEV_SUBJECT` | unset | Local-dev subject (insecure) |
-| `WEBUI_DEV_ROLE` | `viewer` | Local-dev role (insecure) |
-| `WEBUI_ROLE_MAP` | unset | JSON subject → role map |
-| `WEBUI_REQUIRE_PROBE_AUTH` | unset | Require auth for non-public probes |
-| `WEBUI_CONSOLE_AUDIT_LOG` | unset | Append-only audit sink path |
-
-All are read server-side only. None is ever rendered into a page or returned by
-an API.
-
-## Non-goals
-
-- No full SSO product; authentication stays delegated to the proxy.
-- No browser-initiated merges or approvals in any phase covered here.
-- No tokens in the frontend, in browser storage, or in committed config.

docs/webui-deployment.md

@@ -7,10 +7,7 @@ only.
 ## MVP deployment model
 
 - **Default bind:** `127.0.0.1:8765` (`WEBUI_HOST` / `WEBUI_PORT`)
-- **Authentication:** none in MVP — protection comes from network placement.
+- **Authentication:** none in MVP — protection comes from network placement
-  The authorization, RBAC, redaction, and audit model that future gated writes
-  must pass through is defined in
-  [`webui-authz-audit.md`](webui-authz-audit.md) (#633).
 - **Mutations:** read-only routes; gated write actions remain disabled (#434)
 - **Secrets:** resolved server-side via `gitea_auth` / `GITEA_MCP_CONFIG`; never
   embedded in HTML, JavaScript, or browser storage

gitea_mcp_server.py

@@ -2024,6 +2024,7 @@ import review_quarantine  # noqa: E402  # #695 contaminated formal-review quaran
 import mcp_daemon_guard  # noqa: E402  # #695 native transport provenance
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
+import branch_publish  # noqa: E402  # #812 AC20 unpublished-commit publication
 import root_checkout_guard  # noqa: E402
 import workflow_scope_guard  # noqa: E402  # #683 production scope / force-on guards
 import stable_branch_push_guard  # noqa: E402
@@ -9082,6 +9083,237 @@ def gitea_commit_files(
     }
 
 
+def _publication_block(reasons: list[str], **extra) -> dict:
+    """Uniform fail-closed shape for publication refusals (#812 AC20)."""
+    payload = {
+        "success": False,
+        "performed": False,
+        "published": False,
+        "verified": False,
+        "outcome": branch_publish.REFUSED,
+        "reasons": reasons,
+        # State every record this operation left alone, so a refusal can never
+        # be misread as a lock mutation (#812 AC23).
+        "issue_lock_record_mutated": False,
+        "workflow_lease_touched": False,
+    }
+    payload.update(extra)
+    return payload
+
+
+@mcp.tool()
+def gitea_publish_unpublished_issue_branch(
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    expected_head: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    git_remote_name: str | None = None,
+    expected_file_hashes: dict | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Publish an already-committed, unpublished issue branch (#812 AC20).
+
+    Creates the remote head for a branch whose work is *already* a local commit
+    on a registered, clean worktree, so exact-owner lease renewal
+    (``issue_lock_renewal``) has the published head its evidence model requires.
+    This is the one step of the entry point B deadlock that no existing tool can
+    perform: publication is otherwise lock-derived under #618, and the lock
+    itself is withheld until a remote head exists.
+
+    Not a lock bypass. The branch's **durable issue-lock record must already
+    name the caller as claimant** — ownership is read from the lock file, never
+    asserted by the caller — so this can only publish work the caller already
+    owns. It refuses a dirty or untracked-carrying worktree, an unregistered
+    worktree, a changed local HEAD, a remote head that is not an ancestor of the
+    commit, a competing open PR on another branch for the same issue, and any
+    declared-hash mismatch. It renews, reclaims, and clears nothing: the durable
+    issue-lock file and the control-plane workflow lease are both left untouched
+    (#812 AC23), and the recorded owner pid's liveness is never consulted or
+    asserted (#812 AC24).
+
+    Args:
+        issue_number: The issue whose recorded claim authorizes publication.
+        branch_name: Issue branch to publish, ``(fix|feat|docs|chore)/issue-N-…``.
+        worktree_path: Registered worktree holding the commit.
+        expected_head: Full 40-character SHA of the commit to publish. Required:
+            publication names the exact commit, and a mismatch fails closed.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+        git_remote_name: Git remote to publish to; defaults to *remote*.
+        expected_file_hashes: Optional ``{path: sha256}`` verified against the
+            worktree before publication. Any mismatch or missing file refuses.
+        dry_run: Report the decision and evidence, mutate nothing.
+
+    Returns:
+        dict with 'success', 'performed', 'published', 'verified', 'outcome',
+        'remote_head_sha', 'reasons', and 'evidence'.
+    """
+    task = "publish_unpublished_branch"
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        task
+    )
+    if not ok:
+        return _publication_block(block_reasons)
+
+    blocked = _namespace_mutation_block(task, remote=remote)
+    if blocked:
+        return blocked
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission(task),
+        remote=remote, host=host, org=org, repo=repo,
+        org_explicit=org is not None,
+        repo_explicit=repo is not None,
+    )
+    if blocked:
+        return blocked
+
+    verify_preflight_purity(remote, task=task, org=org, repo=repo)
+
+    h, o, r = _resolve(remote, host, org, repo)
+    git_remote = (git_remote_name or remote or "").strip()
+    workspace = os.path.realpath(os.path.abspath((worktree_path or "").strip() or "."))
+
+    existing_lock = issue_lock_store.load_issue_lock(
+        remote=remote, org=o, repo=r, issue_number=int(issue_number)
+    )
+    git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+    registered = author_mutation_worktree.path_in_git_worktree_list(
+        workspace, PROJECT_ROOT
+    )
+
+    remote_probe = branch_publish.read_remote_branch_head(
+        workspace, git_remote, branch_name
+    )
+    ancestry = None
+    probe_head = (remote_probe.get("remote_head_sha") or "").strip()
+    if probe_head and probe_head.lower() != (expected_head or "").strip().lower():
+        ancestry = branch_publish.read_is_ancestor(workspace, probe_head, expected_head)
+
+    # A PR on this very branch is this work's own PR, not a rival claim. Only an
+    # open PR for the same issue on a *different* branch is a competing claim.
+    competing: list = []
+    for pull in _list_open_pulls(h, o, r, _auth(h)):
+        ref = str((pull.get("head") or {}).get("ref") or "")
+        if not ref or ref == branch_name:
+            continue
+        if issue_lock_adoption.branch_carries_issue_marker(ref, int(issue_number)):
+            competing.append(pull.get("number"))
+
+    observed_hashes = None
+    if expected_file_hashes:
+        observed_hashes = branch_publish.hash_worktree_files(
+            workspace, list(expected_file_hashes.keys())
+        )
+
+    claimant = _work_lease_claimant(h)
+    assessment = branch_publish.assess_unpublished_commit_publication(
+        existing_lock,
+        issue_number=int(issue_number),
+        branch_name=branch_name,
+        worktree_path=workspace,
+        expected_head=expected_head,
+        remote=remote,
+        org=o,
+        repo=r,
+        identity=claimant.get("username"),
+        profile=claimant.get("profile"),
+        worktree_state=git_state,
+        worktree_registered=registered,
+        remote_probe=remote_probe,
+        ancestry=ancestry,
+        competing_open_prs=competing,
+        expected_file_hashes=expected_file_hashes,
+        observed_file_hashes=observed_hashes,
+    )
+
+    if assessment["outcome"] == branch_publish.REFUSED:
+        return _publication_block(
+            assessment["reasons"], evidence=assessment["evidence"]
+        )
+
+    if dry_run:
+        return {
+            "success": True,
+            "performed": False,
+            "published": False,
+            "verified": False,
+            "dry_run": True,
+            "outcome": assessment["outcome"],
+            "would_publish": assessment["publish_sanctioned"],
+            "remote_head_sha": assessment["evidence"].get("remote_head_sha"),
+            "reasons": [],
+            "evidence": assessment["evidence"],
+            "issue_lock_record_mutated": False,
+            "workflow_lease_touched": False,
+        }
+
+    performed = False
+    if assessment["publish_sanctioned"]:
+        with _audited(
+            task, host=h, remote=remote, org=o, repo=r,
+            target_branch=branch_name,
+            request_metadata={
+                "issue_number": int(issue_number),
+                "expected_head": expected_head,
+                "git_remote": git_remote,
+            },
+        ):
+            push = branch_publish.publish_commit_to_remote_branch(
+                worktree_path=workspace,
+                remote_name=git_remote,
+                branch_name=branch_name,
+                expected_head=expected_head,
+            )
+        if not push.get("success"):
+            return _publication_block(
+                push.get("reasons") or ["publication failed"],
+                evidence=assessment["evidence"],
+                stderr=push.get("stderr"),
+            )
+        performed = True
+
+    verification = branch_publish.verify_published_head(
+        worktree_path=workspace,
+        remote_name=git_remote,
+        branch_name=branch_name,
+        expected_head=expected_head,
+    )
+    if not verification.get("verified"):
+        return _publication_block(
+            verification.get("reasons") or ["read-after-write verification failed"],
+            evidence=assessment["evidence"],
+            performed=performed,
+            published=performed,
+            remote_head_sha=verification.get("remote_head_sha"),
+        )
+
+    return {
+        "success": True,
+        "performed": performed,
+        "published": True,
+        "verified": True,
+        "outcome": assessment["outcome"],
+        "remote_head_sha": verification.get("remote_head_sha"),
+        "reasons": [],
+        "evidence": assessment["evidence"],
+        # Publication is the whole of this operation's authority (#812 AC23).
+        "issue_lock_record_mutated": False,
+        "workflow_lease_touched": False,
+        "exact_next_action": (
+            f"Remote head for '{branch_name}' is now observable. Call "
+            f"gitea_lock_issue(issue_number={int(issue_number)}, "
+            f"branch_name='{branch_name}', worktree_path='{workspace}') to renew "
+            "the exact-owner lease, then gitea_create_pr."
+        ),
+    }
+
+
 # Merge methods supported by the Gitea merge API.
 _MERGE_METHODS = ("merge", "squash", "rebase")
 

task_capability_map.py

@@ -62,6 +62,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
+    # #812 AC20: publish an already-committed, unpublished local head so
+    # exact-owner lease renewal has an observable remote head to reason about.
+    # Same authority as any other author push — deliberately not a new
+    # operation name, so it cannot widen an already-configured author profile.
+    "publish_unpublished_branch": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
     "create_pr": {
         "permission": "gitea.pr.create",
         "role": "author",
@@ -500,6 +508,7 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
         "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
+        "publish_unpublished_branch",
         "create_pr",
         "commit_files",
         "gitea_commit_files",

tests/test_issue_812_publish_unpublished_commit.py

@@ -0,0 +1,650 @@
+"""Publication of an unpublished local commit (#812 AC20).
+
+Entry point B of #812: a registered worktree, clean, on its issue branch,
+holding a local commit that has never been published. Exact-owner lease renewal
+refuses such a claim for want of an observable remote head, and every existing
+publication path is lock-derived, so the two predicates close a cycle around
+work that is otherwise complete.
+
+These tests exercise the disposition through its *evidence*, never through any
+particular issue number: every case uses an arbitrary issue number against a
+synthetic repository, and the same assertions hold for any other. Nothing here
+reads, writes, or references the live protected worktree named in #812 AC17 —
+that content is preserved evidence for the duration of this work, so the
+fixtures below build their own repositories from scratch.
+
+The remote is a local bare repository, so publication and read-after-write
+verification are genuinely executed rather than mocked.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import branch_publish  # noqa: E402
+import issue_lock_provenance  # noqa: E402
+import issue_lock_renewal  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+
+ISSUE = 9812
+BRANCH = f"feat/issue-{ISSUE}-publish-fixture"
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+GIT_REMOTE = "prgs"
+
+
+def _ts(hours: int) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+class _PublishBase(unittest.TestCase):
+    """Real git repo + real bare remote + durable lock naming the caller.
+
+    The recorded owner pid is deliberately **this live process**. That mirrors
+    the production shape #812 documents, where the pid belongs to a long-running
+    MCP daemon rather than to a dead author client, and it proves publication
+    never depends on a dead process (#812 AC24).
+    """
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.origin = tempfile.mkdtemp(prefix="issue812-origin-")
+        self.repo = tempfile.mkdtemp(prefix="issue812-work-")
+        for path in (self.origin, self.repo):
+            self.addCleanup(
+                lambda p=path: subprocess.run(["rm", "-rf", p], check=False)
+            )
+        self._init_repos()
+        self.remotes = patch.dict(
+            mcp_server.REMOTES,
+            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
+        )
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    # ── fixture construction ─────────────────────────────────────────────
+    def _git(self, *args, cwd=None):
+        return subprocess.run(
+            ["git", "-C", cwd or self.repo, *args],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+
+    def _init_repos(self):
+        subprocess.run(
+            ["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True
+        )
+        self._git("init", "-q", "-b", "master")
+        self._git("config", "user.email", "[email protected]")
+        self._git("config", "user.name", "Test")
+        self._git("remote", "add", GIT_REMOTE, self.origin)
+
+        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
+            fh.write("seed\n")
+        self._git("add", "seed.txt")
+        self._git("commit", "-q", "-m", "seed")
+        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        self._git("push", "-q", GIT_REMOTE, "master")
+
+        self._git("checkout", "-q", "-b", BRANCH)
+        with open(os.path.join(self.repo, "work.txt"), "w") as fh:
+            fh.write("unpublished implementation\n")
+        self._git("add", "work.txt")
+        self._git("commit", "-q", "-m", "unpublished implementation")
+        self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        self.worktree = os.path.realpath(self.repo)
+
+    def lock_path(self):
+        return issue_lock_store.lock_file_path(
+            remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+
+    def write_lock(self, **overrides):
+        path = self.lock_path()
+        claimant = overrides.pop(
+            "claimant", {"username": IDENTITY, "profile": PROFILE}
+        )
+        pid = overrides.pop("session_pid", os.getpid())
+        lease = {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "pr_number": None,
+            "branch": overrides.get("branch_name", BRANCH),
+            "worktree_path": overrides.get("worktree_path", self.worktree),
+            "claimant": claimant,
+            "created_at": _ts(-2),
+            "last_heartbeat_at": _ts(-2),
+            # Expired: entry point B's lease has lapsed, which is precisely why
+            # renewal — and therefore a published head — is needed.
+            "expires_at": _ts(-1),
+        }
+        lease.update(overrides.pop("work_lease", {}))
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "lock_generation": 1,
+            "work_lease": lease,
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue", claimant=claimant
+            ),
+        }
+        data.update(overrides)
+        data["lock_file_path"] = path
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    def _tool_env(self):
+        env = shared_mutation_env(
+            PROFILE, include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+        # These tests repoint PROJECT_ROOT at a synthetic repository so the
+        # registered-worktree proof runs for real. Pin the parity gate to the
+        # server's own startup head so that repointing does not read as a stale
+        # daemon; the gate itself stays live and enforced.
+        startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or ""
+        env["GITEA_TEST_CURRENT_HEAD"] = startup_head
+        env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head
+        return env
+
+    # ── tool driver ──────────────────────────────────────────────────────
+    def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
+        """Drive the public publication tool against the synthetic fixture."""
+        env = self._tool_env()
+        with patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs or [])
+        ), patch(
+            "mcp_server._auth", return_value="token x"
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": IDENTITY, "profile": PROFILE},
+        ), patch.object(
+            mcp_server, "PROJECT_ROOT", self.repo
+        ), patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            return mcp_server.gitea_publish_unpublished_issue_branch(
+                issue_number=kwargs.pop("issue_number", ISSUE),
+                branch_name=kwargs.pop("branch_name", BRANCH),
+                worktree_path=kwargs.pop("worktree_path", self.worktree),
+                expected_head=expected_head or self.head_sha,
+                remote="prgs",
+                git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE),
+                **kwargs,
+            )
+
+    def remote_head(self, branch=BRANCH):
+        res = subprocess.run(
+            ["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch],
+            capture_output=True, text=True, check=False,
+        )
+        return (res.stdout or "").strip() or None
+
+
+class TestSuccessfulPublication(_PublishBase):
+    """AC20 — the branch becomes observable and is verified after the write."""
+
+    def test_publishes_clean_unpublished_commit(self):
+        self.write_lock()
+        self.assertIsNone(self.remote_head(), "fixture must start unpublished")
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["performed"])
+        self.assertTrue(result["published"])
+        self.assertTrue(result["verified"], "read-after-write must be proven")
+        self.assertEqual(result["remote_head_sha"], self.head_sha)
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+    def test_publication_does_not_rewrite_the_commit(self):
+        self.write_lock()
+        self.run_publish()
+        # The published object is the same commit, not a copy or a rewrite.
+        self.assertEqual(self.remote_head(), self.head_sha)
+        self.assertEqual(
+            self._git("rev-parse", "HEAD").stdout.strip(), self.head_sha
+        )
+
+    def test_exact_next_action_names_the_lock_call(self):
+        self.write_lock()
+        result = self.run_publish()
+        self.assertIn("gitea_lock_issue", result["exact_next_action"])
+
+
+class TestFailsClosed(_PublishBase):
+    """AC20/AC9 — each refusal reason, exercised independently."""
+
+    def test_changed_local_head_refuses(self):
+        self.write_lock()
+        stale = self.base_sha  # a real commit, but not the declared head
+        result = self.run_publish(expected_head=stale)
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("local commit changed" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIsNone(self.remote_head(), "refusal must not publish")
+
+    def test_abbreviated_sha_refuses(self):
+        self.write_lock()
+        result = self.run_publish(expected_head=self.head_sha[:8])
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("40-character" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_dirty_tracked_worktree_refuses(self):
+        self.write_lock()
+        with open(os.path.join(self.repo, "work.txt"), "a") as fh:
+            fh.write("uncommitted edit\n")
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("dirty tracked files" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIn("work.txt", result["evidence"]["dirty_tracked_files"])
+        self.assertIsNone(self.remote_head())
+
+    def test_untracked_file_refuses(self):
+        self.write_lock()
+        with open(os.path.join(self.repo, "stray.txt"), "w") as fh:
+            fh.write("not committed\n")
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("untracked files" in r for r in result["reasons"]), result["reasons"]
+        )
+        self.assertIn("stray.txt", result["evidence"]["untracked_files"])
+        self.assertIsNone(self.remote_head())
+
+    def test_unexpected_remote_head_refuses(self):
+        """A remote head that is not an ancestor must never be overwritten."""
+        self.write_lock()
+        # Publish a divergent commit to the branch from a separate line.
+        self._git("checkout", "-q", "-b", "divergent", self.base_sha)
+        with open(os.path.join(self.repo, "other.txt"), "w") as fh:
+            fh.write("someone else's work\n")
+        self._git("add", "other.txt")
+        self._git("commit", "-q", "-m", "divergent")
+        divergent = self._git("rev-parse", "HEAD").stdout.strip()
+        self._git("push", "-q", GIT_REMOTE, f"{divergent}:refs/heads/{BRANCH}")
+        self._git("checkout", "-q", BRANCH)
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("not an ancestor" in r for r in result["reasons"]), result["reasons"]
+        )
+        self.assertEqual(
+            self.remote_head(), divergent, "the other head must survive intact"
+        )
+
+    def test_fast_forward_remote_head_is_allowed(self):
+        """An ancestor head is an honest fast-forward, not a conflict."""
+        self.write_lock()
+        self._git("push", "-q", GIT_REMOTE, f"{self.base_sha}:refs/heads/{BRANCH}")
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["evidence"]["fast_forward_from_remote"])
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+    def test_content_hash_mismatch_refuses(self):
+        self.write_lock()
+        wrong = {"work.txt": "0" * 64}
+
+        result = self.run_publish(expected_file_hashes=wrong)
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("declared content hashes" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertFalse(result["evidence"]["file_hashes_verified"])
+        self.assertIsNone(self.remote_head())
+
+    def test_matching_content_hashes_publish(self):
+        self.write_lock()
+        digests = branch_publish.hash_worktree_files(self.worktree, ["work.txt"])
+
+        result = self.run_publish(expected_file_hashes=digests)
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["evidence"]["file_hashes_verified"])
+
+    def test_missing_declared_file_refuses(self):
+        self.write_lock()
+        result = self.run_publish(expected_file_hashes={"absent.txt": "0" * 64})
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("missing or unreadable" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_foreign_claimant_refuses(self):
+        """Ownership comes from the durable record, not from the caller."""
+        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
+
+        result = self.run_publish()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("foreign claim" in r for r in result["reasons"]), result["reasons"]
+        )
+        self.assertIsNone(self.remote_head())
+
+    def test_foreign_profile_refuses(self):
+        self.write_lock(
+            claimant={"username": IDENTITY, "profile": "test-reviewer-prgs"}
+        )
+        result = self.run_publish()
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("claimant profile" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_absent_lock_record_refuses(self):
+        """No recorded claim means this cannot be used to bypass the lock."""
+        result = self.run_publish()  # no write_lock()
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("no durable issue-lock record" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIsNone(self.remote_head())
+
+    def test_branch_mismatch_against_lock_refuses(self):
+        self.write_lock(branch_name=f"feat/issue-{ISSUE}-different")
+        result = self.run_publish()
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("records branch" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_worktree_mismatch_against_lock_refuses(self):
+        self.write_lock(worktree_path="/tmp/some/other/worktree")
+        result = self.run_publish()
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("records worktree" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_competing_open_pr_on_another_branch_refuses(self):
+        self.write_lock()
+        competing = [{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
+
+        result = self.run_publish(open_prs=competing)
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("already claim issue" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+        self.assertIsNone(self.remote_head())
+
+    def test_open_pr_on_the_same_branch_is_not_competing(self):
+        """This branch's own PR is not a rival claim against itself."""
+        self.write_lock()
+        own = [{"number": 77, "head": {"ref": BRANCH}}]
+
+        result = self.run_publish(open_prs=own)
+
+        self.assertTrue(result["success"], result.get("reasons"))
+
+
+class TestGuardStrictnessPreserved(_PublishBase):
+    """AC15 — publication is an operation, never a weakening of the guards."""
+
+    def test_non_issue_branch_refuses(self):
+        self._git("checkout", "-q", "-b", "scratch/not-issue-linked")
+        self.write_lock(branch_name="scratch/not-issue-linked")
+
+        result = self.run_publish(branch_name="scratch/not-issue-linked")
+
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("issue-linked" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_stable_branch_refuses(self):
+        self.write_lock(branch_name="master")
+        result = self.run_publish(branch_name="master")
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("issue-linked" in r or "stable branch" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_branch_number_must_match_the_issue(self):
+        other = "feat/issue-7777-mismatched"
+        self._git("checkout", "-q", "-b", other)
+        self.write_lock(branch_name=other)
+        result = self.run_publish(branch_name=other)
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("does not carry issue number" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_unregistered_worktree_refuses(self):
+        """#713 — an improvised directory is not a registered worktree."""
+        path = self.write_lock()
+        assessment = branch_publish.assess_unpublished_commit_publication(
+            issue_lock_store.read_lock_file(path),
+            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
+            expected_head=self.head_sha, remote="prgs", org=ORG, repo=REPO,
+            identity=IDENTITY, profile=PROFILE,
+            worktree_state={
+                "current_branch": BRANCH, "porcelain_status": "",
+                "head_sha": self.head_sha,
+            },
+            worktree_registered=False,
+            remote_probe={"probe_ok": True, "remote_branch_exists": False},
+        )
+        self.assertEqual(assessment["outcome"], branch_publish.REFUSED)
+        self.assertTrue(
+            any("not listed in git worktree list" in r
+                for r in assessment["reasons"]),
+            assessment["reasons"],
+        )
+
+    def test_unobservable_remote_refuses(self):
+        """An unknown remote state must not be mistaken for an absent branch."""
+        self.write_lock()
+        result = self.run_publish(git_remote_name="no-such-remote")
+        self.assertFalse(result["success"])
+        self.assertTrue(
+            any("could not be observed" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+
+class TestRecordSeparation(_PublishBase):
+    """AC23 — the durable issue lock and the workflow lease are distinct."""
+
+    def test_publication_leaves_the_issue_lock_byte_identical(self):
+        path = self.write_lock()
+        with open(path, "rb") as fh:
+            before = fh.read()
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        with open(path, "rb") as fh:
+            after = fh.read()
+        self.assertEqual(before, after, "publication must not mutate the lock record")
+        self.assertFalse(result["issue_lock_record_mutated"])
+        self.assertFalse(result["workflow_lease_touched"])
+
+    def test_refusal_also_reports_untouched_records(self):
+        result = self.run_publish()  # refuses: no lock record
+        self.assertFalse(result["issue_lock_record_mutated"])
+        self.assertFalse(result["workflow_lease_touched"])
+
+    def test_lock_generation_is_not_advanced(self):
+        path = self.write_lock()
+        self.run_publish()
+        lock = issue_lock_store.read_lock_file(path)
+        self.assertEqual(lock["lock_generation"], 1)
+
+
+class TestTruthfulProcessEvidence(_PublishBase):
+    """AC24 — a live daemon pid is never represented as a dead process."""
+
+    def test_live_recorded_pid_does_not_block_publication(self):
+        # The recorded pid is this live process, standing in for the live MCP
+        # daemon. Reclaim would refuse here; publication legitimately does not.
+        path = self.write_lock(session_pid=os.getpid())
+        lock = issue_lock_store.read_lock_file(path)
+        self.assertEqual(lock["pid"], os.getpid())
+
+        result = self.run_publish()
+
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+    def test_liveness_is_not_consulted_as_evidence(self):
+        self.write_lock(session_pid=os.getpid())
+        result = self.run_publish()
+        self.assertFalse(result["evidence"]["owner_pid_liveness_consulted"])
+
+    def test_reclaim_still_refuses_for_the_same_live_pid(self):
+        """Publication does not soften the reclaim predicate it routes around."""
+        path = self.write_lock(session_pid=os.getpid())
+        lock = issue_lock_store.read_lock_file(path)
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
+        self.assertFalse(reclaim["reclaim_allowed"])
+
+
+class TestIdempotentRetry(_PublishBase):
+    """AC20 — retry is safe and read-after-write is proven every time."""
+
+    def test_second_publication_reports_already_published(self):
+        self.write_lock()
+        first = self.run_publish()
+        self.assertTrue(first["performed"])
+
+        second = self.run_publish()
+
+        self.assertTrue(second["success"], second.get("reasons"))
+        self.assertFalse(second["performed"], "no second push is needed")
+        self.assertTrue(second["published"])
+        self.assertTrue(second["verified"])
+        self.assertEqual(second["outcome"], branch_publish.ALREADY_PUBLISHED)
+        self.assertEqual(self.remote_head(), self.head_sha)
+
+
+class TestDryRun(_PublishBase):
+    """AC12 — dry run reports the decision and mutates nothing."""
+
+    def test_dry_run_reports_intent_without_publishing(self):
+        self.write_lock()
+
+        result = self.run_publish(dry_run=True)
+
+        self.assertTrue(result["success"])
+        self.assertTrue(result["dry_run"])
+        self.assertTrue(result["would_publish"])
+        self.assertFalse(result["performed"])
+        self.assertIsNone(self.remote_head(), "dry run must not publish")
+
+    def test_dry_run_and_apply_agree_on_a_refusal(self):
+        """AC11 — the reported decision does not depend on which mode ran."""
+        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
+
+        dry = self.run_publish(dry_run=True)
+        applied = self.run_publish()
+
+        self.assertFalse(dry["success"])
+        self.assertFalse(applied["success"])
+        self.assertEqual(dry["reasons"], applied["reasons"])
+
+
+class TestRenewalUnblocked(_PublishBase):
+    """AC20/AC21 — renewal is permitted only after verified publication."""
+
+    def _renewal(self, remote_head):
+        return issue_lock_renewal.assess_exact_owner_lease_renewal(
+            issue_lock_store.read_lock_file(self.lock_path()),
+            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
+            remote="prgs", org=ORG, repo=REPO,
+            identity=IDENTITY, profile=PROFILE,
+            current_branch=BRANCH, porcelain_status="", worktree_exists=True,
+            head_sha=self.head_sha, remote_head_sha=remote_head,
+        )
+
+    def test_renewal_refuses_before_publication(self):
+        self.write_lock()
+        decision = self._renewal(None)
+        self.assertFalse(decision["renewal_sanctioned"])
+        self.assertTrue(
+            any("unpublished branch" in r for r in decision["reasons"]),
+            decision["reasons"],
+        )
+
+    def test_renewal_is_sanctioned_after_publication(self):
+        self.write_lock()
+        result = self.run_publish()
+        self.assertTrue(result["verified"], result.get("reasons"))
+
+        decision = self._renewal(self.remote_head())
+
+        self.assertTrue(decision["renewal_sanctioned"], decision["reasons"])
+
+
+class TestProtectedAssetUntouched(unittest.TestCase):
+    """AC17 — no test or fixture may reference the protected worktree."""
+
+    def test_no_reference_to_the_protected_worktree(self):
+        here = os.path.dirname(os.path.abspath(__file__))
+        root = os.path.dirname(here)
+        needle = "issue-635-project-registry" + "-api"
+        for path in (
+            os.path.join(here, "test_issue_812_publish_unpublished_commit.py"),
+            os.path.join(root, "branch_publish.py"),
+        ):
+            with open(path, "r", encoding="utf-8") as fh:
+                body = fh.read()
+            self.assertNotIn(needle, body)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/test_task_capability_role_invariants.py

@@ -139,6 +139,9 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
         "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
+        # #812 AC20: publishing an unpublished local head is author-only for the
+        # same reason every other push is — it writes a branch to the remote.
+        "publish_unpublished_branch",
         "create_pr",
         "commit_files",
         "gitea_commit_files",

tests/test_webui_console_authz_audit.py

@@ -1,703 +0,0 @@
-"""Console authorization, redaction, and audit model tests (#633).
-
-Covers each acceptance criterion and each required test named in the issue:
-
-* AC1 — RBAC matrix and privileged-action list.
-* AC2 — redaction rules, unit-tested against sample payloads.
-* AC3 — audit event schema with required fields and retention defaults.
-* AC4 — Phase 2 integration points.
-* AC5 — local-dev mode with explicit insecurity warnings.
-
-Required tests: redaction units (token, keychain, password patterns),
-default-deny for unauthenticated write stubs, and audit record creation for a
-simulated privileged preview.
-"""
-
-from __future__ import annotations
-
-import datetime
-import json
-import os
-import pathlib
-import sys
-import tempfile
-import unittest
-
-from starlette.testclient import TestClient
-
-sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
-
-from task_capability_map import TASK_CAPABILITY_MAP  # noqa: E402
-from webui import console_audit, console_authz  # noqa: E402
-from webui.app import create_app  # noqa: E402
-from webui.console_redaction import (  # noqa: E402
-    REDACTED,
-    redact_payload,
-    redact_text,
-    redaction_policy,
-    scan_for_secrets,
-)
-
-DOCS = pathlib.Path(__file__).resolve().parents[1] / "docs"
-AUTHZ_DOC = DOCS / "webui-authz-audit.md"
-
-
-def _principal(role: str) -> console_authz.Principal:
-    return console_authz.Principal(
-        subject=f"{role}@example.com",
-        role=role,
-        identity_source=console_authz.IDENTITY_ACCESS_PROXY,
-        authenticated=True,
-    )
-
-
-class TestRoleMatrix(unittest.TestCase):
-    """AC1 — the written RBAC matrix and privileged-action list."""
-
-    def test_roles_are_ordered_least_to_most_authority(self):
-        self.assertEqual(
-            console_authz.ROLE_ORDER,
-            ("viewer", "operator", "controller", "admin"),
-        )
-
-    def test_every_role_has_a_description(self):
-        for role in console_authz.ROLE_ORDER:
-            with self.subTest(role=role):
-                self.assertTrue(console_authz.ROLE_DESCRIPTIONS[role].strip())
-
-    def test_higher_roles_inherit_lower_role_actions(self):
-        matrix = {
-            entry["role"]: set(entry["permitted_actions"])
-            for entry in console_authz.rbac_matrix()["roles"]
-        }
-        for lower, higher in zip(
-            console_authz.ROLE_ORDER, console_authz.ROLE_ORDER[1:]
-        ):
-            with self.subTest(lower=lower, higher=higher):
-                self.assertTrue(matrix[lower].issubset(matrix[higher]))
-
-    def test_viewer_holds_no_write_action(self):
-        matrix = {
-            entry["role"]: set(entry["permitted_actions"])
-            for entry in console_authz.rbac_matrix()["roles"]
-        }
-        self.assertEqual(matrix["viewer"], set())
-
-    def test_privileged_action_list_is_non_empty_and_classified(self):
-        privileged = console_authz.privileged_actions()
-        self.assertTrue(privileged)
-        ids = {action.action_id for action in privileged}
-        # Merge and branch deletion are the canonical privileged pair.
-        self.assertIn("merge_pr", ids)
-        self.assertIn("delete_branch", ids)
-
-    def test_merge_and_delete_require_dual_control_and_break_glass(self):
-        for action_id in ("merge_pr", "delete_branch"):
-            with self.subTest(action=action_id):
-                action = console_authz.get_action(action_id)
-                self.assertTrue(action.dual_control)
-                self.assertTrue(action.break_glass)
-                self.assertTrue(action.requires_confirmation)
-
-    def test_every_write_action_requires_confirmation(self):
-        for action in console_authz.ACTIONS.values():
-            with self.subTest(action=action.action_id):
-                self.assertTrue(action.requires_confirmation)
-
-    def test_delete_branch_is_admin_only(self):
-        self.assertEqual(
-            console_authz.get_action("delete_branch").minimum_role,
-            console_authz.ADMIN,
-        )
-
-    def test_actions_map_to_real_mcp_capability_vocabulary(self):
-        """The console must not invent an authority the MCP layer lacks."""
-        for action in console_authz.ACTIONS.values():
-            with self.subTest(action=action.action_id):
-                self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
-                self.assertEqual(
-                    action.mcp_permission,
-                    TASK_CAPABILITY_MAP[action.task_key]["permission"],
-                )
-                self.assertEqual(
-                    action.mcp_role,
-                    TASK_CAPABILITY_MAP[action.task_key]["role"],
-                )
-
-    def test_matrix_declares_deny_by_default_and_execution_disabled(self):
-        matrix = console_authz.rbac_matrix()
-        self.assertEqual(matrix["default_decision"], "deny")
-        self.assertFalse(matrix["execution_enabled"])
-
-
-class TestAuthorizeDefaultDeny(unittest.TestCase):
-    """Fail-closed behaviour of the authorization decision."""
-
-    def test_anonymous_is_denied_every_action(self):
-        for action_id in console_authz.ACTIONS:
-            with self.subTest(action=action_id):
-                decision = console_authz.authorize(action_id)
-                self.assertFalse(decision.allowed)
-                self.assertEqual(
-                    decision.reason_code, console_authz.DENY_UNAUTHENTICATED
-                )
-
-    def test_unknown_action_is_denied(self):
-        decision = console_authz.authorize(
-            "not_a_real_action", _principal("admin")
-        )
-        self.assertFalse(decision.allowed)
-        self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ACTION)
-
-    def test_unknown_role_is_denied(self):
-        rogue = console_authz.Principal(
-            subject="[email protected]",
-            role="superuser",
-            identity_source=console_authz.IDENTITY_ACCESS_PROXY,
-            authenticated=True,
-        )
-        decision = console_authz.authorize("comment_issue", rogue)
-        self.assertFalse(decision.allowed)
-        self.assertEqual(decision.reason_code, console_authz.DENY_UNKNOWN_ROLE)
-
-    def test_insufficient_role_is_denied(self):
-        decision = console_authz.authorize("merge_pr", _principal("operator"))
-        self.assertFalse(decision.allowed)
-        self.assertEqual(
-            decision.reason_code, console_authz.DENY_INSUFFICIENT_ROLE
-        )
-
-    def test_sufficient_role_allows_preview_only(self):
-        decision = console_authz.authorize("merge_pr", _principal("controller"))
-        self.assertTrue(decision.allowed)
-        self.assertFalse(decision.execution_enabled)
-
-    def test_execution_is_refused_while_phase_is_not_active(self):
-        decision = console_authz.authorize(
-            "merge_pr", _principal("controller"), for_execution=True
-        )
-        self.assertFalse(decision.allowed)
-        self.assertEqual(
-            decision.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE
-        )
-
-    def test_allowed_decision_never_reports_execution_enabled(self):
-        for action_id in console_authz.ACTIONS:
-            with self.subTest(action=action_id):
-                decision = console_authz.authorize(
-                    action_id, _principal("admin")
-                )
-                self.assertFalse(decision.execution_enabled)
-
-
-class TestIdentityResolution(unittest.TestCase):
-    """AC5 — identity sources, including the insecure local-dev mode."""
-
-    def test_no_auth_mode_yields_anonymous_viewer(self):
-        principal = console_authz.resolve_principal(env={})
-        self.assertFalse(principal.authenticated)
-        self.assertEqual(principal.role, console_authz.VIEWER)
-        self.assertEqual(principal.identity_source, console_authz.IDENTITY_NONE)
-
-    def test_local_dev_mode_warns_that_identity_is_unverified(self):
-        principal = console_authz.resolve_principal(
-            env={
-                console_authz.AUTH_MODE_ENV: "local-dev",
-                console_authz.DEV_SUBJECT_ENV: "[email protected]",
-                console_authz.DEV_ROLE_ENV: "admin",
-            }
-        )
-        self.assertTrue(principal.authenticated)
-        self.assertEqual(principal.role, "admin")
-        self.assertTrue(principal.warnings)
-        self.assertIn("asserted", " ".join(principal.warnings).lower())
-
-    def test_local_dev_without_subject_falls_back_to_anonymous(self):
-        principal = console_authz.resolve_principal(
-            env={console_authz.AUTH_MODE_ENV: "local-dev"}
-        )
-        self.assertFalse(principal.authenticated)
-
-    def test_local_dev_unknown_role_degrades_to_viewer(self):
-        principal = console_authz.resolve_principal(
-            env={
-                console_authz.AUTH_MODE_ENV: "local_dev",
-                console_authz.DEV_SUBJECT_ENV: "[email protected]",
-                console_authz.DEV_ROLE_ENV: "root",
-            }
-        )
-        self.assertEqual(principal.role, console_authz.VIEWER)
-
-    def test_access_proxy_without_header_fails_closed(self):
-        """A proxy-mode request that did not traverse the proxy is anonymous."""
-        principal = console_authz.resolve_principal(
-            headers={},
-            env={console_authz.AUTH_MODE_ENV: "access_proxy"},
-        )
-        self.assertFalse(principal.authenticated)
-
-    def test_access_proxy_role_comes_from_server_config_not_client(self):
-        env = {
-            console_authz.AUTH_MODE_ENV: "access_proxy",
-            console_authz.ROLE_MAP_ENV: json.dumps(
-                {"[email protected]": "controller"}
-            ),
-        }
-        principal = console_authz.resolve_principal(
-            headers={
-                console_authz.ACCESS_SUBJECT_HEADER: "[email protected]",
-                "x-role": "admin",  # client-supplied role must be ignored
-            },
-            env=env,
-        )
-        self.assertEqual(principal.role, "controller")
-
-    def test_access_proxy_unmapped_subject_defaults_to_viewer(self):
-        principal = console_authz.resolve_principal(
-            headers={
-                console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"
-            },
-            env={console_authz.AUTH_MODE_ENV: "access_proxy"},
-        )
-        self.assertEqual(principal.role, console_authz.VIEWER)
-
-    def test_malformed_role_map_does_not_raise_and_denies(self):
-        principal = console_authz.resolve_principal(
-            headers={console_authz.ACCESS_SUBJECT_HEADER: "[email protected]"},
-            env={
-                console_authz.AUTH_MODE_ENV: "access_proxy",
-                console_authz.ROLE_MAP_ENV: "{not json",
-            },
-        )
-        self.assertEqual(principal.role, console_authz.VIEWER)
-
-    def test_probe_auth_is_opt_in(self):
-        self.assertFalse(console_authz.probe_auth_required(env={}))
-        self.assertTrue(
-            console_authz.probe_auth_required(
-                env={console_authz.REQUIRE_PROBE_AUTH_ENV: "1"}
-            )
-        )
-
-    def test_probe_auth_is_declared_but_not_yet_enforced(self):
-        """Phase 1 declares the probe-auth policy; no route enforces it yet.
-
-        The flag exists so the Phase 2 action framework has a declared policy
-        to honour instead of inventing a second one. Pinning the current
-        not-enforced status here means wiring it later is a deliberate change
-        that updates this test and the documentation together, rather than a
-        silent behaviour shift. The documentation must say so plainly, because
-        an operator who sets the variable believing it protects a probe is
-        worse off than one who knows it does not.
-        """
-        import inspect
-
-        from webui import app as webui_app
-
-        source = inspect.getsource(webui_app)
-        self.assertNotIn(
-            "probe_auth_required",
-            source,
-            msg=(
-                "webui.app now consults probe_auth_required, so probe auth is "
-                "no longer merely declared. Update the 'Probe authentication' "
-                "section of docs/webui-authz-audit.md, which states it "
-                "enforces nothing, and replace this test with real "
-                "enforcement coverage."
-            ),
-        )
-        self.assertIn(
-            "enforces nothing today",
-            AUTHZ_DOC.read_text(encoding="utf-8"),
-        )
-
-
-class TestRedaction(unittest.TestCase):
-    """AC2 — required redaction units: token, keychain, password patterns."""
-
-    def test_token_assignment_is_redacted(self):
-        out = redact_text("GITEA_TOKEN=abcd1234efgh5678ijkl")
-        self.assertIn(REDACTED, out)
-        self.assertNotIn("abcd1234efgh5678ijkl", out)
-
-    def test_password_assignment_is_redacted(self):
-        out = redact_text("password: hunter2supersecret")
-        self.assertIn(REDACTED, out)
-        self.assertNotIn("hunter2supersecret", out)
-
-    def test_keychain_reference_is_redacted(self):
-        out = redact_text("keychain:gitea-prgs-token")
-        self.assertIn(REDACTED, out)
-        self.assertNotIn("gitea-prgs-token", out)
-
-    def test_keychain_command_is_redacted(self):
-        out = redact_text("security find-generic-password -s gitea -w")
-        self.assertIn(REDACTED, out)
-        self.assertNotIn("find-generic-password -s gitea", out)
-
-    def test_bearer_credential_is_redacted(self):
-        out = redact_text("Authorization: Bearer abcdef1234567890abcdef")
-        self.assertNotIn("abcdef1234567890abcdef", out)
-
-    def test_jwt_is_redacted(self):
-        token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.abcdefghijklmnop"
-        out = redact_text(f"session={token}")
-        self.assertNotIn(token, out)
-
-    def test_private_key_block_is_redacted(self):
-        pem = (
-            "-----BEGIN RSA PRIVATE KEY-----\n"
-            "MIIEowIBAAKCAQEAsecretmaterial\n"
-            "-----END RSA PRIVATE KEY-----"
-        )
-        out = redact_text(pem)
-        self.assertNotIn("MIIEowIBAAKCAQEAsecretmaterial", out)
-
-    def test_api_key_assignment_is_redacted(self):
-        out = redact_text('api_key = "sk-live-9f8e7d6c5b4a3210"')
-        self.assertNotIn("sk-live-9f8e7d6c5b4a3210", out)
-
-    def test_nested_payload_is_redacted_recursively(self):
-        payload = {
-            "token": "abc123456789",
-            "nested": {"note": "password=letmein12345"},
-            "list": ["keychain:some-entry"],
-            "safe": "plain text",
-        }
-        out = redact_payload(payload)
-        self.assertEqual(out["token"], REDACTED)
-        self.assertNotIn("letmein12345", json.dumps(out))
-        self.assertNotIn("some-entry", json.dumps(out))
-        self.assertEqual(out["safe"], "plain text")
-
-    def test_scan_reports_findings_before_and_none_after(self):
-        dirty = "password: hunter2supersecret"
-        self.assertTrue(scan_for_secrets(dirty))
-        self.assertEqual(scan_for_secrets(redact_text(dirty)), [])
-
-    def test_non_strings_pass_through_untouched(self):
-        self.assertEqual(redact_text(42), 42)
-        self.assertEqual(
-            redact_payload({"n": 1, "b": True}), {"n": 1, "b": True}
-        )
-
-    def test_policy_is_documented_and_declares_redact_before_persist(self):
-        policy = redaction_policy()
-        self.assertTrue(policy["redact_before_persist"])
-        self.assertIn("audit_records", policy["applies_to"])
-        self.assertTrue(policy["console_rules"])
-
-    def test_policy_statement_contains_no_secret_material(self):
-        self.assertEqual(scan_for_secrets(redaction_policy()), [])
-
-
-class TestAuditSchema(unittest.TestCase):
-    """AC3 — audit event schema, required fields, and retention defaults."""
-
-    def _event(self, action_id="merge_pr", **kwargs):
-        return console_audit.build_event(
-            action_id=action_id,
-            result=console_audit.RESULT_DENIED,
-            decision=console_authz.authorize(action_id, _principal("operator")),
-            target={"kind": "pr", "ref": "#123"},
-            request_id="req-test",
-            **kwargs,
-        )
-
-    def test_every_required_field_is_present(self):
-        event = self._event()
-        for field in console_audit.REQUIRED_FIELDS:
-            with self.subTest(field=field):
-                self.assertIn(field, event)
-
-    def test_actor_carries_who_and_how_they_were_identified(self):
-        event = self._event()
-        for field in console_audit.REQUIRED_ACTOR_FIELDS:
-            with self.subTest(field=field):
-                self.assertIn(field, event["actor"])
-
-    def test_correlation_ids_are_present(self):
-        event = self._event()
-        for field in console_audit.REQUIRED_CORRELATION_FIELDS:
-            with self.subTest(field=field):
-                self.assertIn(field, event["correlation"])
-        self.assertEqual(event["correlation"]["request_id"], "req-test")
-        self.assertEqual(event["correlation"]["mcp_task"], "merge_pr")
-
-    def test_timestamp_is_timezone_aware_utc_iso8601(self):
-        now = datetime.datetime(
-            2026, 7, 22, 10, 16, 42, tzinfo=datetime.timezone.utc
-        )
-        event = self._event(now=now)
-        self.assertEqual(event["timestamp"], "2026-07-22T10:16:42+00:00")
-        parsed = datetime.datetime.fromisoformat(event["timestamp"])
-        self.assertIsNotNone(parsed.tzinfo)
-
-    def test_retention_defaults_by_class(self):
-        self.assertEqual(
-            console_audit.RETENTION_DAYS[console_audit.RETENTION_STANDARD], 90
-        )
-        self.assertEqual(
-            console_audit.RETENTION_DAYS[console_audit.RETENTION_PRIVILEGED],
-            365,
-        )
-        self.assertEqual(
-            console_audit.RETENTION_DAYS[console_audit.RETENTION_BREAK_GLASS],
-            730,
-        )
-
-    def test_break_glass_action_retains_longest(self):
-        event = self._event("merge_pr")
-        self.assertEqual(
-            event["retention"]["class"], console_audit.RETENTION_BREAK_GLASS
-        )
-
-    def test_routine_write_uses_standard_retention(self):
-        event = self._event("comment_issue")
-        self.assertEqual(
-            event["retention"]["class"], console_audit.RETENTION_STANDARD
-        )
-
-    def test_unknown_action_retains_as_privileged_not_standard(self):
-        """Conservative direction: keep an unclassifiable record longer."""
-        self.assertEqual(
-            console_audit.retention_class_for(None),
-            console_audit.RETENTION_PRIVILEGED,
-        )
-
-    def test_retention_expiry_matches_declared_days(self):
-        now = datetime.datetime(2026, 7, 22, tzinfo=datetime.timezone.utc)
-        event = self._event("comment_issue", now=now)
-        expires = datetime.datetime.fromisoformat(
-            event["retention"]["expires_at"]
-        )
-        self.assertEqual((expires - now).days, 90)
-
-    def test_invalid_result_degrades_to_failed(self):
-        event = console_audit.build_event(action_id="merge_pr", result="banana")
-        self.assertEqual(event["result"], console_audit.RESULT_FAILED)
-
-    def test_denied_result_is_representable(self):
-        """An authorization denial has no MCP-side mutation record."""
-        self.assertIn(console_audit.RESULT_DENIED, console_audit.RESULTS)
-
-    def test_event_is_redacted_before_it_is_returned(self):
-        event = console_audit.build_event(
-            action_id="merge_pr",
-            result=console_audit.RESULT_DENIED,
-            detail="failed with token=abcdef1234567890",
-            metadata={"password": "hunter2supersecret"},
-        )
-        serialized = json.dumps(event)
-        self.assertNotIn("abcdef1234567890", serialized)
-        self.assertNotIn("hunter2supersecret", serialized)
-        self.assertTrue(event["redacted"])
-
-    def test_audit_policy_reports_schema_and_retention(self):
-        policy = console_audit.audit_policy()
-        self.assertTrue(policy["append_only"])
-        self.assertTrue(policy["redact_before_persist"])
-        self.assertEqual(
-            policy["retention_defaults_days"], console_audit.RETENTION_DAYS
-        )
-
-
-class TestAuditSink(unittest.TestCase):
-    """Append-only persistence behaviour."""
-
-    def test_write_is_a_noop_when_sink_is_unconfigured(self):
-        saved = os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
-        try:
-            self.assertFalse(console_audit.audit_enabled())
-            self.assertFalse(console_audit.write_event({"schema_version": 1}))
-        finally:
-            if saved is not None:
-                os.environ[console_audit.AUDIT_LOG_ENV] = saved
-
-    def test_records_append_one_json_line_each(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            sink = os.path.join(tmp, "console-audit.jsonl")
-            for _ in range(3):
-                event = console_audit.build_event(
-                    action_id="merge_pr", result=console_audit.RESULT_DENIED
-                )
-                self.assertTrue(console_audit.write_event(event, path=sink))
-            with open(sink, encoding="utf-8") as handle:
-                lines = [json.loads(line) for line in handle if line.strip()]
-            self.assertEqual(len(lines), 3)
-            self.assertEqual(len({line["event_id"] for line in lines}), 3)
-
-    def test_a_record_that_still_carries_a_secret_is_not_persisted(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            sink = os.path.join(tmp, "console-audit.jsonl")
-            leaky = {
-                "schema_version": 1,
-                "detail": "password: hunter2supersecret",
-            }
-            self.assertFalse(console_audit.write_event(leaky, path=sink))
-            self.assertFalse(os.path.exists(sink))
-
-    def test_write_never_raises_on_a_bad_path(self):
-        self.assertFalse(
-            console_audit.write_event(
-                {"schema_version": 1}, path="/nonexistent-dir/audit.jsonl"
-            )
-        )
-
-    def test_simulated_privileged_preview_creates_an_audit_record(self):
-        """Required test: audit record creation for a privileged preview."""
-        with tempfile.TemporaryDirectory() as tmp:
-            sink = os.path.join(tmp, "console-audit.jsonl")
-            os.environ[console_audit.AUDIT_LOG_ENV] = sink
-            try:
-                decision = console_authz.authorize(
-                    "merge_pr", _principal("controller")
-                )
-                outcome = console_audit.record_event(
-                    action_id="merge_pr",
-                    result=console_audit.RESULT_PREVIEWED,
-                    decision=decision,
-                    target={"kind": "pr", "ref": "#123"},
-                    request_id="req-preview",
-                )
-            finally:
-                os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
-            self.assertTrue(outcome["written"])
-            with open(sink, encoding="utf-8") as handle:
-                record = json.loads(handle.read().strip())
-            self.assertEqual(record["action"], "merge_pr")
-            self.assertEqual(record["result"], console_audit.RESULT_PREVIEWED)
-            self.assertEqual(record["action_class"], "privileged")
-            self.assertTrue(record["decision"]["allowed"])
-            self.assertFalse(record["decision"]["execution_enabled"])
-            self.assertEqual(record["actor"]["role"], "controller")
-
-    def test_decision_block_survives_redaction(self):
-        """Regression: naming it 'authorization' collided with a secret hint.
-
-        ``gitea_audit._SECRET_KEY_HINTS`` contains "authorization" (for the
-        HTTP header), so a block under that key was replaced wholesale by the
-        placeholder and the record lost its decision entirely.
-        """
-        event = console_audit.build_event(
-            action_id="merge_pr",
-            result=console_audit.RESULT_DENIED,
-            decision=console_authz.authorize("merge_pr", _principal("admin")),
-        )
-        self.assertIsInstance(event["decision"], dict)
-        self.assertIn("allowed", event["decision"])
-
-
-class TestConsoleRoutes(unittest.TestCase):
-    """AC4 — the wired Phase 2 integration points, still fail-closed."""
-
-    def setUp(self):
-        self.client = TestClient(create_app(bind_host="127.0.0.1"))
-
-    def test_unauthenticated_write_stub_is_denied(self):
-        """Required test: default-deny for unauthenticated write stubs."""
-        response = self.client.post(
-            "/api/actions/merge_pr/attempt", json={"pr_number": 99}
-        )
-        self.assertEqual(response.status_code, 403)
-        body = response.json()
-        self.assertFalse(body["success"])
-        authorization = body["authorization"]
-        self.assertFalse(authorization["allowed"])
-        self.assertEqual(
-            authorization["reason_code"], console_authz.DENY_UNAUTHENTICATED
-        )
-        self.assertFalse(authorization["execution_enabled"])
-
-    def test_preview_reports_an_authorization_decision(self):
-        response = self.client.get("/api/actions/merge_pr/preview?pr_number=7")
-        self.assertEqual(response.status_code, 200)
-        authorization = response.json()["authorization"]
-        self.assertFalse(authorization["allowed"])
-        self.assertTrue(authorization["dual_control"])
-        self.assertEqual(authorization["required_role"], "controller")
-
-    def test_unknown_action_preview_still_404s(self):
-        response = self.client.get("/api/actions/no_such_action/preview")
-        self.assertEqual(response.status_code, 404)
-
-    def test_security_model_endpoint_publishes_all_three_policies(self):
-        response = self.client.get("/api/console/security-model")
-        self.assertEqual(response.status_code, 200)
-        body = response.json()
-        self.assertIn("rbac", body)
-        self.assertIn("redaction", body)
-        self.assertIn("audit", body)
-        self.assertEqual(body["rbac"]["default_decision"], "deny")
-
-    def test_security_model_endpoint_leaks_no_secrets(self):
-        response = self.client.get("/api/console/security-model")
-        self.assertEqual(scan_for_secrets(response.json()), [])
-
-    def test_security_model_rejects_writes(self):
-        response = self.client.post("/api/console/security-model", json={})
-        self.assertEqual(response.status_code, 405)
-
-    def test_existing_read_routes_are_unaffected(self):
-        for path in ("/", "/health", "/actions", "/api/actions"):
-            with self.subTest(path=path):
-                self.assertEqual(self.client.get(path).status_code, 200)
-
-
-class TestAuthzAuditDoc(unittest.TestCase):
-    """The model must be written down, not only coded."""
-
-    @classmethod
-    def setUpClass(cls):
-        cls.text = (
-            AUTHZ_DOC.read_text(encoding="utf-8") if AUTHZ_DOC.exists() else ""
-        )
-
-    def test_doc_exists(self):
-        self.assertTrue(AUTHZ_DOC.exists(), f"missing {AUTHZ_DOC}")
-
-    def test_doc_covers_each_required_section(self):
-        for heading in (
-            "Identity sources",
-            "Role matrix",
-            "Privileged actions",
-            "Secret redaction",
-            "Audit event schema",
-            "Retention",
-            "Phase 2 integration",
-            "Local-dev mode",
-        ):
-            with self.subTest(heading=heading):
-                self.assertIn(heading, self.text)
-
-    def test_doc_names_every_role(self):
-        for role in console_authz.ROLE_ORDER:
-            with self.subTest(role=role):
-                self.assertIn(role, self.text)
-
-    def test_doc_names_every_console_action(self):
-        for action_id in console_authz.ACTIONS:
-            with self.subTest(action=action_id):
-                self.assertIn(action_id, self.text)
-
-    def test_doc_states_retention_defaults(self):
-        for days in console_audit.RETENTION_DAYS.values():
-            with self.subTest(days=days):
-                self.assertIn(str(days), self.text)
-
-    def test_doc_warns_local_dev_is_insecure(self):
-        self.assertIn("INSECURE", self.text.upper())
-
-    def test_doc_states_default_deny(self):
-        self.assertIn("deny", self.text.lower())
-
-    def test_doc_contains_no_secret_material(self):
-        self.assertEqual(scan_for_secrets(self.text), [])
-
-    def test_deployment_doc_links_to_the_model(self):
-        deployment = (DOCS / "webui-deployment.md").read_text(encoding="utf-8")
-        self.assertIn("webui-authz-audit", deployment)
-
-
-if __name__ == "__main__":  # pragma: no cover
-    unittest.main()

webui/app.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import uuid
 from datetime import datetime, timezone
 
 from starlette.applications import Starlette
@@ -20,9 +19,6 @@ from final_report_validator import FINAL_REPORT_TASK_KINDS
 
 from webui.gated_actions import attempt_action, load_action_registry, preview_action
 from webui.gated_action_views import render_actions_page
-from webui import console_audit
-from webui.console_authz import authorize, rbac_matrix, resolve_principal
-from webui.console_redaction import redaction_policy
 from webui.audit_validator import audit_report, audit_to_dict
 from webui.audit_views import render_audit_page
 from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
@@ -219,49 +215,6 @@ async def api_actions(_request: Request) -> JSONResponse:
     return JSONResponse(load_action_registry().to_dict())
 
 
-def _request_id() -> str:
-    return f"req-{uuid.uuid4().hex}"
-
-
-def _audit_target(action_id: str, params: dict[str, object]) -> dict[str, object]:
-    """Describe the action target for the audit record (never secrets)."""
-    if "pr_number" in params:
-        return {"kind": "pr", "ref": f"#{params['pr_number']}"}
-    if "issue_number" in params:
-        return {"kind": "issue", "ref": f"#{params['issue_number']}"}
-    if "branch_name" in params:
-        return {"kind": "branch", "ref": str(params["branch_name"])}
-    return {"kind": "unspecified", "ref": action_id}
-
-
-def _authorize_request(
-    request: Request,
-    action_id: str,
-    params: dict[str, object],
-    *,
-    for_execution: bool,
-    result: str,
-) -> dict[str, object]:
-    """Resolve principal, decide, and audit. Returns the decision payload.
-
-    Phase 1 records the decision rather than enforcing it as the terminal
-    outcome: ``webui.gated_actions`` already fails closed for every action, so
-    this layer cannot loosen anything. Phase 2 enforces on this same decision.
-    """
-    principal = resolve_principal(headers=dict(request.headers))
-    decision = authorize(action_id, principal, for_execution=for_execution)
-    console_audit.record_event(
-        action_id=action_id,
-        result=result,
-        decision=decision,
-        principal=principal,
-        target=_audit_target(action_id, params),
-        request_id=_request_id(),
-        detail=decision.detail,
-    )
-    return decision.to_dict()
-
-
 async def api_action_preview(request: Request) -> JSONResponse:
     action_id = request.path_params["action_id"]
     params = dict(request.query_params)
@@ -271,13 +224,6 @@ async def api_action_preview(request: Request) -> JSONResponse:
     result = preview_action(action_id, **params)
     if "error" in result:
         return JSONResponse(result, status_code=404)
-    result["authorization"] = _authorize_request(
-        request,
-        action_id,
-        params,
-        for_execution=False,
-        result=console_audit.RESULT_PREVIEWED,
-    )
     return JSONResponse(result)
 
 
@@ -291,31 +237,10 @@ async def api_action_attempt(request: Request) -> JSONResponse:
     if not isinstance(body, dict):
         body = {}
     result = attempt_action(action_id, **body)
-    authorization = _authorize_request(
-        request,
-        action_id,
-        body,
-        for_execution=True,
-        result=(
-            console_audit.RESULT_DENIED
-            if not result.get("success")
-            else console_audit.RESULT_ALLOWED
-        ),
-    )
-    result["authorization"] = authorization
     status = 403 if not result.get("success") else 200
     return JSONResponse(result, status_code=status)
 
 
-async def api_console_security_model(_request: Request) -> JSONResponse:
-    """Read-only publication of the #633 authorization/redaction/audit model."""
-    return JSONResponse({
-        "rbac": rbac_matrix(),
-        "redaction": redaction_policy(),
-        "audit": console_audit.audit_policy(),
-    })
-
-
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -366,11 +291,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 methods=["POST"],
             ),
             Route("/api/leases", api_leases, methods=["GET"]),
-            Route(
-                "/api/console/security-model",
-                api_console_security_model,
-                methods=["GET"],
-            ),
         ],
         exception_handlers={405: method_not_allowed},
     )

webui/console_audit.py

@@ -1,281 +0,0 @@
-"""Console audit event schema, retention, and append-only sink (#633).
-
-``gitea_audit`` records MCP-side *mutations*: which profile and Gitea user
-performed which tool call. It carries no console actor, no identity source, no
-correlation identifier, and no retention class, so it cannot answer the
-question #633 exists to answer — *who sat at the console, what did they
-attempt, and was it authorized?* An authorization denial is not a mutation and
-would never appear there at all.
-
-This module adds the console-side record. It does not replace ``gitea_audit``:
-when a Phase 2 action eventually reaches MCP, both fire, correlated by
-``correlation.request_id``.
-
-Design constraints:
-
-- **Redact before persist.** Every record passes through
-  ``webui.console_redaction.redact_payload`` before serialization, so an
-  unredacted field is never durable.
-- **Append-only.** Records are appended as JSON lines. Nothing here updates or
-  deletes; retention is metadata on each record, enforced by an operator-run
-  policy, never by silent rewriting.
-- **Never raises.** Auditing must not break the request it describes. A failed
-  write returns ``False``.
-- **Off by default.** With ``WEBUI_CONSOLE_AUDIT_LOG`` unset, events are still
-  *built* (so callers and tests see the schema) but nothing is written.
-
-A record looks like this (synthetic values):
-
-    {"schema_version": 1, "event_id": "evt-0001",
-     "timestamp": "2026-07-22T10:16:42+00:00",
-     "actor": {"subject": "[email protected]", "role": "operator",
-               "identity_source": "access_proxy", "authenticated": true},
-     "action": "merge_pr", "action_class": "privileged",
-     "target": {"kind": "pr", "ref": "#123"},
-     "result": "denied", "reason_code": "insufficient_role",
-     "correlation": {"request_id": "req-abc", "session_id": null,
-                     "mcp_task": "merge_pr", "mcp_permission": "gitea.pr.merge"},
-     "retention": {"class": "privileged", "days": 365,
-                   "expires_at": "2027-07-22T10:16:42+00:00"},
-     "redacted": true}
-
-Timestamps are timezone-aware ISO-8601 in UTC.
-"""
-
-from __future__ import annotations
-
-import datetime
-import json
-import os
-import uuid
-from typing import Any
-
-from webui import console_authz
-from webui.console_redaction import redact_payload, scan_for_secrets
-
-SCHEMA_VERSION = 1
-
-AUDIT_LOG_ENV = "WEBUI_CONSOLE_AUDIT_LOG"
-
-# Result vocabulary. ``denied`` is the one ``gitea_audit`` has no equivalent
-# for: an authorization refusal never reaches the MCP layer.
-RESULT_ALLOWED = "allowed"
-RESULT_DENIED = "denied"
-RESULT_PREVIEWED = "previewed"
-RESULT_FAILED = "failed"
-RESULT_SUCCEEDED = "succeeded"
-
-RESULTS = frozenset(
-    {
-        RESULT_ALLOWED,
-        RESULT_DENIED,
-        RESULT_PREVIEWED,
-        RESULT_FAILED,
-        RESULT_SUCCEEDED,
-    }
-)
-
-# Retention classes and default lifetimes in days. Privileged and break-glass
-# records outlive routine ones because they are what an incident review needs.
-RETENTION_STANDARD = "standard"
-RETENTION_PRIVILEGED = "privileged"
-RETENTION_BREAK_GLASS = "break_glass"
-
-RETENTION_DAYS: dict[str, int] = {
-    RETENTION_STANDARD: 90,
-    RETENTION_PRIVILEGED: 365,
-    RETENTION_BREAK_GLASS: 730,
-}
-
-# Fields every record must carry. Asserted by the test suite so a future edit
-# cannot quietly drop one.
-REQUIRED_FIELDS: tuple[str, ...] = (
-    "schema_version",
-    "event_id",
-    "timestamp",
-    "actor",
-    "action",
-    "action_class",
-    "target",
-    "result",
-    "reason_code",
-    "correlation",
-    "retention",
-    "redacted",
-)
-
-REQUIRED_ACTOR_FIELDS: tuple[str, ...] = (
-    "subject",
-    "role",
-    "identity_source",
-    "authenticated",
-)
-
-REQUIRED_CORRELATION_FIELDS: tuple[str, ...] = (
-    "request_id",
-    "session_id",
-    "mcp_task",
-    "mcp_permission",
-)
-
-
-def audit_log_path() -> str | None:
-    """Configured sink path, or ``None`` when console auditing is off."""
-    return (os.environ.get(AUDIT_LOG_ENV) or "").strip() or None
-
-
-def audit_enabled() -> bool:
-    return audit_log_path() is not None
-
-
-def retention_class_for(action: console_authz.ConsoleAction | None) -> str:
-    """Classify retention from the action, defaulting to the longest-lived.
-
-    An unknown action is treated as privileged rather than standard: for a
-    safety control the conservative direction is to keep the record longer.
-    """
-    if action is None:
-        return RETENTION_PRIVILEGED
-    if action.break_glass:
-        return RETENTION_BREAK_GLASS
-    if action.privileged:
-        return RETENTION_PRIVILEGED
-    return RETENTION_STANDARD
-
-
-def _retention_block(
-    retention_class: str, now: datetime.datetime
-) -> dict[str, Any]:
-    days = RETENTION_DAYS.get(
-        retention_class, RETENTION_DAYS[RETENTION_PRIVILEGED]
-    )
-    return {
-        "class": retention_class,
-        "days": days,
-        "expires_at": (now + datetime.timedelta(days=days)).isoformat(),
-    }
-
-
-def build_event(
-    *,
-    action_id: str,
-    result: str,
-    decision: console_authz.AuthorizationDecision | None = None,
-    principal: console_authz.Principal | None = None,
-    target: dict[str, Any] | None = None,
-    reason_code: str | None = None,
-    request_id: str | None = None,
-    session_id: str | None = None,
-    detail: str | None = None,
-    metadata: dict[str, Any] | None = None,
-    now: datetime.datetime | None = None,
-    event_id: str | None = None,
-) -> dict[str, Any]:
-    """Build one redacted, JSON-able console audit record.
-
-    Redaction runs here rather than at write time so an in-memory record handed
-    to a template or an API response is already clean.
-    """
-    ts = now or datetime.datetime.now(datetime.timezone.utc)
-    action = console_authz.get_action(action_id)
-    who = principal or (
-        decision.principal if decision else console_authz.ANONYMOUS
-    )
-    resolved_result = result if result in RESULTS else RESULT_FAILED
-    resolved_reason = reason_code or (
-        decision.reason_code if decision else "unspecified"
-    )
-    retention_class = retention_class_for(action)
-
-    event: dict[str, Any] = {
-        "schema_version": SCHEMA_VERSION,
-        "event_id": event_id or f"evt-{uuid.uuid4().hex}",
-        "timestamp": ts.isoformat(),
-        "actor": who.to_dict(),
-        "action": action_id,
-        "action_class": action.action_class if action else "unknown",
-        "target": dict(target or {}),
-        "result": resolved_result,
-        "reason_code": resolved_reason,
-        "correlation": {
-            "request_id": request_id,
-            "session_id": session_id,
-            "mcp_task": action.task_key if action else None,
-            "mcp_permission": action.mcp_permission if action else None,
-        },
-        "retention": _retention_block(retention_class, ts),
-        "redacted": True,
-        "detail": detail,
-        "metadata": dict(metadata or {}),
-    }
-    if decision is not None:
-        # Deliberately *not* named "authorization": ``gitea_audit`` treats that
-        # substring as a secret key hint (it matches the HTTP Authorization
-        # header) and would replace this whole block with the placeholder.
-        event["decision"] = {
-            "allowed": decision.allowed,
-            "required_role": decision.required_role,
-            "requires_confirmation": decision.requires_confirmation,
-            "dual_control": decision.dual_control,
-            "break_glass": decision.break_glass,
-            "execution_enabled": decision.execution_enabled,
-        }
-
-    redacted = redact_payload(event)
-    if not isinstance(redacted, dict):  # pragma: no cover - defensive
-        return {"schema_version": SCHEMA_VERSION, "redacted": True}
-    return redacted
-
-
-def write_event(event: dict[str, Any], path: str | None = None) -> bool:
-    """Append *event* as one JSON line. Never raises.
-
-    Returns ``True`` when a line was written, ``False`` when auditing is off or
-    the write failed. A record that still trips a secret detector is dropped
-    rather than persisted.
-    """
-    sink = path or audit_log_path()
-    if not sink:
-        return False
-    try:
-        if scan_for_secrets(event):
-            return False
-        line = json.dumps(event, default=str, sort_keys=True)
-        with open(sink, "a", encoding="utf-8") as handle:
-            handle.write(line + "\n")
-        return True
-    except Exception:
-        return False
-
-
-def record_event(**kwargs: Any) -> dict[str, Any]:
-    """Build and persist one record; return the record either way.
-
-    Callers get the record back so it can be surfaced in a response or a test
-    regardless of whether a sink is configured.
-    """
-    event = build_event(**kwargs)
-    written = write_event(event)
-    return {"event": event, "written": written}
-
-
-def audit_policy() -> dict[str, Any]:
-    """Machine-readable audit schema and retention defaults (never secrets)."""
-    return {
-        "schema_version": SCHEMA_VERSION,
-        "required_fields": list(REQUIRED_FIELDS),
-        "required_actor_fields": list(REQUIRED_ACTOR_FIELDS),
-        "required_correlation_fields": list(REQUIRED_CORRELATION_FIELDS),
-        "results": sorted(RESULTS),
-        "retention_defaults_days": dict(RETENTION_DAYS),
-        "sink_env": AUDIT_LOG_ENV,
-        "enabled": audit_enabled(),
-        "append_only": True,
-        "redact_before_persist": True,
-        "timestamp_format": "ISO-8601, timezone-aware, UTC",
-        "relationship_to_mcp_audit": (
-            "webui.console_audit records console intent and authorization "
-            "outcomes; gitea_audit records MCP mutations. A Phase 2 action "
-            "emits both, correlated by correlation.request_id."
-        ),
-    }

webui/console_authz.py

@@ -1,537 +0,0 @@
-"""Console authorization and RBAC model (#633, Phase 1).
-
-The read-only MVP (#426–#436) ships with no authentication: protection comes
-from network placement alone (#435). That is adequate while every route is a
-GET, and inadequate the moment Phase 2 wires a gated write. This module is the
-authorization model those writes must go through, landed *before* any of them
-exists so no write can be added without an authority to check against.
-
-Phase 1 scope is the model itself: identity resolution, the role matrix, the
-privileged-action list, and a fail-closed :func:`authorize`. It deliberately
-does **not** enable any write. ``webui.gated_actions`` stays globally disabled,
-so an allow decision here is necessary but never sufficient.
-
-Two invariants hold for every caller:
-
-- **Default deny.** An unrecognised action, an unknown role, or an absent
-  principal denies. There is no implicit allow branch and no "unless" clause.
-- **Authorization is not execution.** :func:`authorize` returns a decision
-  record. It never calls MCP, never mutates, and never consults credentials.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-from dataclasses import asdict, dataclass, field
-from typing import Any
-
-from task_capability_map import required_permission, required_role
-
-# --- Roles ------------------------------------------------------------------
-# Ordered least to most authority. Higher ranks inherit every lower rank's
-# permitted actions; the matrix below is expressed as a minimum required rank.
-VIEWER = "viewer"
-OPERATOR = "operator"
-CONTROLLER = "controller"
-ADMIN = "admin"
-
-ROLE_ORDER: tuple[str, ...] = (VIEWER, OPERATOR, CONTROLLER, ADMIN)
-_ROLE_RANK: dict[str, int] = {role: idx for idx, role in enumerate(ROLE_ORDER)}
-
-ROLE_DESCRIPTIONS: dict[str, str] = {
-    VIEWER: "Read every console view. No write, ever, in any phase.",
-    OPERATOR: "Viewer, plus author-class work: claim, comment, open a PR.",
-    CONTROLLER: "Operator, plus reviewer/merger-class decisions on a PR.",
-    ADMIN: "Controller, plus destructive and policy-editing actions.",
-}
-
-# --- Identity sources -------------------------------------------------------
-IDENTITY_NONE = "none"
-IDENTITY_LOCAL_DEV = "local_dev"
-IDENTITY_ACCESS_PROXY = "access_proxy"
-
-IDENTITY_SOURCES: dict[str, dict[str, Any]] = {
-    IDENTITY_NONE: {
-        "description": (
-            "No authentication configured. Every request is anonymous and "
-            "capped at viewer. This is the MVP default and the only mode "
-            "whose safety rests entirely on network placement (#435)."
-        ),
-        "authenticated": False,
-        "safe_for_shared_host": False,
-        "phase_available": 1,
-    },
-    IDENTITY_LOCAL_DEV: {
-        "description": (
-            "Developer-supplied principal read from the environment. INSECURE: "
-            "the subject and role are asserted, never verified. Loopback only."
-        ),
-        "authenticated": True,
-        "safe_for_shared_host": False,
-        "phase_available": 1,
-    },
-    IDENTITY_ACCESS_PROXY: {
-        "description": (
-            "Subject asserted by a trusted access proxy (Cloudflare Access, "
-            "WARP, or an org VPN portal) via a verified request header. The "
-            "proxy performs authentication; the console performs authorization."
-        ),
-        "authenticated": True,
-        "safe_for_shared_host": True,
-        "phase_available": 2,
-    },
-}
-
-# Environment configuration. All are read server-side and never rendered.
-AUTH_MODE_ENV = "WEBUI_AUTH_MODE"
-DEV_SUBJECT_ENV = "WEBUI_DEV_SUBJECT"
-DEV_ROLE_ENV = "WEBUI_DEV_ROLE"
-ROLE_MAP_ENV = "WEBUI_ROLE_MAP"
-REQUIRE_PROBE_AUTH_ENV = "WEBUI_REQUIRE_PROBE_AUTH"
-ACCESS_SUBJECT_HEADER = "cf-access-authenticated-user-email"
-
-# --- Action classes ---------------------------------------------------------
-CLASS_READ = "read"
-CLASS_WRITE = "gated_write"
-CLASS_PRIVILEGED = "privileged"
-CLASS_DESTRUCTIVE = "destructive"
-
-# --- Privileged action list -------------------------------------------------
-# ``task_key`` ties each console action back to ``task_capability_map``, so the
-# console cannot invent an authority the MCP layer does not already define.
-
-
-@dataclass(frozen=True)
-class ConsoleAction:
-    """One console action and the authority required to invoke it."""
-
-    action_id: str
-    task_key: str
-    action_class: str
-    minimum_role: str
-    requires_confirmation: bool
-    dual_control: bool
-    break_glass: bool
-    phase: int
-    summary: str
-
-    @property
-    def mcp_permission(self) -> str:
-        return required_permission(self.task_key)
-
-    @property
-    def mcp_role(self) -> str:
-        return required_role(self.task_key)
-
-    @property
-    def privileged(self) -> bool:
-        return self.action_class in {CLASS_PRIVILEGED, CLASS_DESTRUCTIVE}
-
-    def to_dict(self) -> dict[str, Any]:
-        data = asdict(self)
-        data["mcp_permission"] = self.mcp_permission
-        data["mcp_role"] = self.mcp_role
-        data["privileged"] = self.privileged
-        return data
-
-
-_ACTION_SPECS: tuple[ConsoleAction, ...] = (
-    ConsoleAction(
-        action_id="claim_issue",
-        task_key="claim_issue",
-        action_class=CLASS_WRITE,
-        minimum_role=OPERATOR,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=2,
-        summary="Apply status:in-progress to an issue.",
-    ),
-    ConsoleAction(
-        action_id="comment_issue",
-        task_key="comment_issue",
-        action_class=CLASS_WRITE,
-        minimum_role=OPERATOR,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=2,
-        summary="Post an issue comment.",
-    ),
-    ConsoleAction(
-        action_id="create_issue",
-        task_key="create_issue",
-        action_class=CLASS_WRITE,
-        minimum_role=OPERATOR,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=2,
-        summary="Open a new tracking issue.",
-    ),
-    ConsoleAction(
-        action_id="comment_pr",
-        task_key="comment_pr",
-        action_class=CLASS_WRITE,
-        minimum_role=OPERATOR,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=2,
-        summary="Post a PR thread comment.",
-    ),
-    ConsoleAction(
-        action_id="create_pr",
-        task_key="create_pr",
-        action_class=CLASS_WRITE,
-        minimum_role=OPERATOR,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=2,
-        summary="Open a PR from a locked feature branch.",
-    ),
-    ConsoleAction(
-        action_id="review_pr",
-        task_key="review_pr",
-        action_class=CLASS_PRIVILEGED,
-        minimum_role=CONTROLLER,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=3,
-        summary="Submit an approve / request-changes verdict.",
-    ),
-    ConsoleAction(
-        action_id="close_pr",
-        task_key="close_pr",
-        action_class=CLASS_PRIVILEGED,
-        minimum_role=CONTROLLER,
-        requires_confirmation=True,
-        dual_control=False,
-        break_glass=False,
-        phase=3,
-        summary="Close a pull request without merging.",
-    ),
-    ConsoleAction(
-        action_id="merge_pr",
-        task_key="merge_pr",
-        action_class=CLASS_PRIVILEGED,
-        minimum_role=CONTROLLER,
-        requires_confirmation=True,
-        dual_control=True,
-        break_glass=True,
-        phase=3,
-        summary="Merge an approved pull request.",
-    ),
-    ConsoleAction(
-        action_id="delete_branch",
-        task_key="delete_branch",
-        action_class=CLASS_DESTRUCTIVE,
-        minimum_role=ADMIN,
-        requires_confirmation=True,
-        dual_control=True,
-        break_glass=True,
-        phase=3,
-        summary="Remove a remote feature branch.",
-    ),
-)
-
-ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
-
-
-def privileged_actions() -> tuple[ConsoleAction, ...]:
-    """Actions requiring dual control, break-glass, or controller+ authority."""
-    return tuple(a for a in _ACTION_SPECS if a.privileged)
-
-
-def get_action(action_id: str) -> ConsoleAction | None:
-    return ACTIONS.get(action_id)
-
-
-# --- Principals -------------------------------------------------------------
-
-
-@dataclass(frozen=True)
-class Principal:
-    """Who is making a request, and how strongly that is known."""
-
-    subject: str
-    role: str
-    identity_source: str
-    authenticated: bool
-    warnings: tuple[str, ...] = field(default_factory=tuple)
-
-    @property
-    def rank(self) -> int:
-        return _ROLE_RANK.get(self.role, -1)
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "subject": self.subject,
-            "role": self.role,
-            "identity_source": self.identity_source,
-            "authenticated": self.authenticated,
-            "warnings": list(self.warnings),
-        }
-
-
-ANONYMOUS = Principal(
-    subject="anonymous",
-    role=VIEWER,
-    identity_source=IDENTITY_NONE,
-    authenticated=False,
-    warnings=("No authentication configured; capped at viewer.",),
-)
-
-
-def auth_mode(env: dict[str, str] | None = None) -> str:
-    """Resolve the configured identity source, defaulting to ``none``."""
-    source = env if env is not None else os.environ
-    raw = (source.get(AUTH_MODE_ENV) or "").strip().lower().replace("-", "_")
-    if raw in IDENTITY_SOURCES:
-        return raw
-    return IDENTITY_NONE
-
-
-def _role_map(env: dict[str, str]) -> dict[str, str]:
-    """Parse ``WEBUI_ROLE_MAP`` (JSON subject→role). Invalid config yields {}."""
-    raw = (env.get(ROLE_MAP_ENV) or "").strip()
-    if not raw:
-        return {}
-    try:
-        parsed = json.loads(raw)
-    except Exception:
-        return {}
-    if not isinstance(parsed, dict):
-        return {}
-    return {
-        str(k): str(v).strip().lower()
-        for k, v in parsed.items()
-        if str(v).strip().lower() in _ROLE_RANK
-    }
-
-
-def resolve_principal(
-    headers: dict[str, str] | None = None,
-    env: dict[str, str] | None = None,
-) -> Principal:
-    """Resolve the requesting principal. Unknown or unconfigured → anonymous.
-
-    Never raises and never trusts a client-supplied role: the role always comes
-    from server-side configuration keyed by the resolved subject.
-    """
-    source_env = dict(env) if env is not None else dict(os.environ)
-    lowered = {str(k).lower(): str(v) for k, v in (headers or {}).items()}
-    mode = auth_mode(source_env)
-
-    if mode == IDENTITY_LOCAL_DEV:
-        subject = (source_env.get(DEV_SUBJECT_ENV) or "").strip()
-        if not subject:
-            return ANONYMOUS
-        role = (source_env.get(DEV_ROLE_ENV) or VIEWER).strip().lower()
-        if role not in _ROLE_RANK:
-            role = VIEWER
-        return Principal(
-            subject=subject,
-            role=role,
-            identity_source=IDENTITY_LOCAL_DEV,
-            authenticated=True,
-            warnings=(
-                "local-dev identity is asserted, not verified; never use "
-                "outside loopback.",
-            ),
-        )
-
-    if mode == IDENTITY_ACCESS_PROXY:
-        subject = (lowered.get(ACCESS_SUBJECT_HEADER) or "").strip()
-        if not subject:
-            # Proxy mode with no proxy header means the request did not
-            # traverse the proxy. Fail closed rather than trust it.
-            return ANONYMOUS
-        role = _role_map(source_env).get(subject, VIEWER)
-        return Principal(
-            subject=subject,
-            role=role,
-            identity_source=IDENTITY_ACCESS_PROXY,
-            authenticated=True,
-        )
-
-    return ANONYMOUS
-
-
-def probe_auth_required(env: dict[str, str] | None = None) -> bool:
-    """Whether non-public probes must be authenticated. Default False.
-
-    #633 requires the console to *fail closed on missing auth for non-public
-    health probes if configured*. The default stays off so the MVP ``/health``
-    contract is unchanged; an operator opts in explicitly.
-    """
-    source = env if env is not None else os.environ
-    return (source.get(REQUIRE_PROBE_AUTH_ENV) or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-    }
-
-
-# --- Authorization ----------------------------------------------------------
-
-DENY_UNKNOWN_ACTION = "unknown_action"
-DENY_UNAUTHENTICATED = "unauthenticated"
-DENY_INSUFFICIENT_ROLE = "insufficient_role"
-DENY_UNKNOWN_ROLE = "unknown_role"
-DENY_PHASE_NOT_ACTIVE = "phase_not_active"
-ALLOW_PREVIEW = "allowed_preview_only"
-
-# Phase 1 is the only active console phase. Phase 2 opens gated writes and is
-# gated on this model landing; nothing here enables it.
-ACTIVE_PHASE = 1
-
-
-@dataclass(frozen=True)
-class AuthorizationDecision:
-    """Result of an authorization check. Never an execution grant."""
-
-    allowed: bool
-    reason_code: str
-    detail: str
-    action_id: str
-    principal: Principal
-    required_role: str | None = None
-    action_class: str | None = None
-    requires_confirmation: bool = False
-    dual_control: bool = False
-    break_glass: bool = False
-    execution_enabled: bool = False
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "allowed": self.allowed,
-            "reason_code": self.reason_code,
-            "detail": self.detail,
-            "action_id": self.action_id,
-            "principal": self.principal.to_dict(),
-            "required_role": self.required_role,
-            "action_class": self.action_class,
-            "requires_confirmation": self.requires_confirmation,
-            "dual_control": self.dual_control,
-            "break_glass": self.break_glass,
-            "execution_enabled": self.execution_enabled,
-            "active_phase": ACTIVE_PHASE,
-        }
-
-
-def authorize(
-    action_id: str,
-    principal: Principal | None = None,
-    *,
-    for_execution: bool = False,
-) -> AuthorizationDecision:
-    """Decide whether *principal* may invoke *action_id*. Deny by default.
-
-    ``for_execution`` distinguishes a read-only preview from a real invocation.
-    Even an allowed decision reports ``execution_enabled=False`` while the
-    console is in Phase 1, so no caller can read an allow as permission to
-    mutate.
-    """
-    who = principal if principal is not None else ANONYMOUS
-    action = get_action(action_id)
-
-    if action is None:
-        return AuthorizationDecision(
-            allowed=False,
-            reason_code=DENY_UNKNOWN_ACTION,
-            detail=f"No console action registered as {action_id!r}.",
-            action_id=action_id,
-            principal=who,
-        )
-
-    base: dict[str, Any] = {
-        "action_id": action_id,
-        "principal": who,
-        "required_role": action.minimum_role,
-        "action_class": action.action_class,
-        "requires_confirmation": action.requires_confirmation,
-        "dual_control": action.dual_control,
-        "break_glass": action.break_glass,
-        "execution_enabled": False,
-    }
-
-    if not who.authenticated:
-        return AuthorizationDecision(
-            allowed=False,
-            reason_code=DENY_UNAUTHENTICATED,
-            detail=(
-                "Write actions require an authenticated principal; this "
-                "request is anonymous."
-            ),
-            **base,
-        )
-
-    if who.rank < 0:
-        return AuthorizationDecision(
-            allowed=False,
-            reason_code=DENY_UNKNOWN_ROLE,
-            detail=f"Role {who.role!r} is not in the console role matrix.",
-            **base,
-        )
-
-    if who.rank < _ROLE_RANK[action.minimum_role]:
-        return AuthorizationDecision(
-            allowed=False,
-            reason_code=DENY_INSUFFICIENT_ROLE,
-            detail=(
-                f"Action {action_id!r} requires {action.minimum_role!r}; "
-                f"principal holds {who.role!r}."
-            ),
-            **base,
-        )
-
-    if for_execution and action.phase > ACTIVE_PHASE:
-        return AuthorizationDecision(
-            allowed=False,
-            reason_code=DENY_PHASE_NOT_ACTIVE,
-            detail=(
-                f"Action {action_id!r} belongs to phase {action.phase}; the "
-                f"console is in phase {ACTIVE_PHASE}. Execution is not wired."
-            ),
-            **base,
-        )
-
-    return AuthorizationDecision(
-        allowed=True,
-        reason_code=ALLOW_PREVIEW,
-        detail=(
-            "Principal holds the required role. Preview only — execution "
-            "remains disabled until the Phase 2 action framework ships."
-        ),
-        **base,
-    )
-
-
-def rbac_matrix() -> dict[str, Any]:
-    """Machine-readable RBAC matrix and privileged-action list."""
-    return {
-        "model_version": 1,
-        "active_phase": ACTIVE_PHASE,
-        "roles": [
-            {
-                "role": role,
-                "rank": _ROLE_RANK[role],
-                "description": ROLE_DESCRIPTIONS[role],
-                "permitted_actions": sorted(
-                    a.action_id
-                    for a in _ACTION_SPECS
-                    if _ROLE_RANK[role] >= _ROLE_RANK[a.minimum_role]
-                ),
-            }
-            for role in ROLE_ORDER
-        ],
-        "identity_sources": IDENTITY_SOURCES,
-        "actions": [a.to_dict() for a in _ACTION_SPECS],
-        "privileged_actions": [a.action_id for a in privileged_actions()],
-        "default_decision": "deny",
-        "execution_enabled": False,
-    }

webui/console_redaction.py

@@ -1,169 +0,0 @@
-"""Secret redaction policy for every console surface (#633).
-
-The MVP already redacts MCP-side mutation records through ``gitea_audit``.
-This module is the console-facing policy: one redaction pass applied to API
-payloads, rendered HTML, log lines, and audit records *before* they leave the
-server or reach persistent storage.
-
-Design constraints:
-
-- **Reuse, never fork.** ``gitea_audit.redact`` remains the authority for
-  secret-looking dict keys, ``Authorization`` material, and raw URLs. This
-  module runs that pass first and then applies console-specific patterns for
-  keychain references, key/value assignments, private-key blocks, and JWTs.
-- **Never raises.** Redaction is a safety control; a malformed payload must
-  degrade to a redacted placeholder rather than propagate an exception.
-- **Redact before persist.** ``webui.console_audit`` calls this module before
-  writing, so an unredacted record is never durable.
-"""
-
-from __future__ import annotations
-
-import json
-import re
-from typing import Any
-
-import gitea_audit
-
-REDACTED = gitea_audit.REDACTED
-
-# Console-specific patterns applied after the shared ``gitea_audit`` pass.
-# Each keeps the identifying key so an operator can still tell *what* was
-# removed, and replaces only the secret run itself.
-_KEYCHAIN_REF = re.compile(r"(?i)\bkeychain:[\w.\-/@]+")
-_KEYCHAIN_CMD = re.compile(
-    r"(?i)\bsecurity\s+find-(?:generic|internet)-password\b[^\n]*"
-)
-_ASSIGNMENT = re.compile(
-    r"(?i)\b(token|password|passwd|secret|api[_-]?key|access[_-]?key|"
-    r"client[_-]?secret|private[_-]?key)\b(\s*[:=]\s*)"
-    r"(\"[^\"]*\"|'[^']*'|\S+)"
-)
-_ENV_ASSIGNMENT = re.compile(
-    r"(?i)\b(GITEA_(?:TOKEN|PASS|PASSWORD)[A-Z0-9_]*)(\s*=\s*)"
-    r"(\"[^\"]*\"|'[^']*'|\S+)"
-)
-_PRIVATE_KEY_BLOCK = re.compile(
-    r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
-    re.S,
-)
-_JWT = re.compile(
-    r"\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\b"
-)
-
-# Shapes that mean a payload still carries a secret. ``scan_for_secrets`` uses
-# these to assert a surface is clean.
-_DETECTORS: tuple[tuple[str, re.Pattern[str]], ...] = (
-    ("keychain_reference", _KEYCHAIN_REF),
-    ("keychain_command", _KEYCHAIN_CMD),
-    ("credential_assignment", _ASSIGNMENT),
-    ("credential_env_assignment", _ENV_ASSIGNMENT),
-    ("private_key_block", _PRIVATE_KEY_BLOCK),
-    ("json_web_token", _JWT),
-    ("bearer_credential", re.compile(r"(?i)\b(?:bearer|basic)\s+\S{8,}")),
-)
-
-
-def _mask_assignment(match: re.Match[str]) -> str:
-    """Keep the key and separator, replace the value."""
-    return f"{match.group(1)}{match.group(2)}{REDACTED}"
-
-
-def redact_text(text: Any) -> Any:
-    """Redact secret material from a single string.
-
-    Non-strings are returned unchanged so this is safe to map over mixed
-    payloads. Runs the shared ``gitea_audit`` pass first, then the
-    console-specific patterns.
-    """
-    if not isinstance(text, str) or not text:
-        return text
-    try:
-        out = gitea_audit.redact(text)
-        if not isinstance(out, str):  # defensive; redact() returns str for str
-            return REDACTED
-        out = _PRIVATE_KEY_BLOCK.sub(f"{REDACTED}_PRIVATE_KEY", out)
-        out = _ENV_ASSIGNMENT.sub(_mask_assignment, out)
-        out = _ASSIGNMENT.sub(_mask_assignment, out)
-        out = _KEYCHAIN_CMD.sub(f"{REDACTED}_KEYCHAIN_COMMAND", out)
-        out = _KEYCHAIN_REF.sub(f"{REDACTED}_KEYCHAIN_REF", out)
-        out = _JWT.sub(f"{REDACTED}_JWT", out)
-        return out
-    except Exception:
-        # Fail closed: an unredactable string is dropped rather than emitted raw.
-        return REDACTED
-
-
-def redact_payload(value: Any) -> Any:
-    """Recursively redact a JSON-able payload for any console surface.
-
-    Secret-looking dict keys are replaced wholesale by the shared
-    ``gitea_audit`` policy; every remaining string is run through
-    :func:`redact_text`.
-    """
-    try:
-        shared = gitea_audit.redact(value)
-    except Exception:
-        return REDACTED
-    return _walk(shared)
-
-
-def _walk(value: Any) -> Any:
-    if isinstance(value, dict):
-        return {k: _walk(v) for k, v in value.items()}
-    if isinstance(value, (list, tuple)):
-        return [_walk(v) for v in value]
-    if isinstance(value, str):
-        return redact_text(value)
-    return value
-
-
-def scan_for_secrets(value: Any) -> list[str]:
-    """Return detector names that still match *value* after serialization.
-
-    Used to assert an outbound payload or rendered page is clean. An empty
-    list means no known secret shape was found. Already-redacted hits are not
-    findings.
-    """
-    if isinstance(value, str):
-        text = value
-    else:
-        try:
-            text = json.dumps(value, default=str)
-        except Exception:
-            text = str(value)
-    findings: list[str] = []
-    for name, pattern in _DETECTORS:
-        for match in pattern.finditer(text):
-            if REDACTED in match.group(0):
-                continue
-            findings.append(name)
-            break
-    return findings
-
-
-def redaction_policy() -> dict[str, Any]:
-    """Machine-readable statement of the redaction rules (never secrets)."""
-    return {
-        "policy_version": 1,
-        "applies_to": [
-            "json_api_responses",
-            "rendered_html",
-            "server_logs",
-            "audit_records",
-        ],
-        "ordering": "shared gitea_audit pass, then console patterns",
-        "redact_before_persist": True,
-        "shared_rules": {
-            "source": "gitea_audit.redact",
-            "secret_key_hints": list(gitea_audit._SECRET_KEY_HINTS),
-            "secret_value_prefixes": list(gitea_audit._SECRET_VALUE_PREFIXES),
-            "urls": "credentials, secret query parameters, and real hosts redacted",
-        },
-        "console_rules": [
-            {"name": name, "pattern": pattern.pattern}
-            for name, pattern in _DETECTORS
-        ],
-        "placeholder": REDACTED,
-        "failure_mode": "fail closed — unredactable values become the placeholder",
-    }