chore: merge master into feat/issue-646-policy-guardrail-visibility (base sync)

feat(webui): read-only workflow policy & guardrail visibility (Closes #646 )
Phase 3 child of the Web Console epic #631. Operators can now see the active workflow policy/guardrail configuration from the console instead of reading the repo tree. - webui/policy_inventory.py (new): redacted, machine-readable guardrail inventory. One row per major guardrail (role separation/RBAC, lease rules, worktree binding, merge confirmation, redaction, contamination, allocator policy, audit logging, mutation gating) with source pointers (file/module/doc) and a compact active projection from the existing safe policy accessors. Fail-soft per entry; whole payload run through console_redaction before emit; diff vs documented default where feasible. - webui/policy_views.py (new): HTML cards with source pointers, active config, and the documented-default diff; read-only page copy, no forms. - webui/app.py: register GET /policy and GET /api/v1/policy (additive). - webui/layout.py: add Policy nav item. - tests/test_webui_policy_visibility.py (new): guardrail presence + source pointers (AC1), redaction incl. planted-secret masking and scan_for_secrets (AC2/AC3), read-only page + no-mutation routes (AC4), fail-soft rendering. - docs/webui-local-dev.md: route table + read-only policy-visibility section. Read-only throughout; no policy editing, no gate-weakening toggle, secrets redacted. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 17:42:49 -04:00 · 2026-07-23 03:20:56 -05:00
24 changed files with 818 additions and 4771 deletions
@@ -1,976 +0,0 @@
 """Sanctioned author issue worktree bootstrap for allocated issues (#850).
 Bootstraps an allocated author branch, canonical worktree under ``branches/``,
 worktree registration, issue lock, and lease/assignment binding without
 caller-side Git, Bash, or helper scripts.
 Features:
 1. Durable phase journal with read-after-write evidence for every phase.
 2. Idempotent replay handling via idempotency keys.
 3. Authoritative expected-base / concurrency-pin validation.
 4. Typed stale-pin refusal without silent rebasing or repointing.
 5. Canonical branches-root enforcement (path MUST be inside branches/).
 6. Preexisting work preservation (dirty tracked/untracked check, foreign ownership refusal).
 7. Compensating recovery limited strictly to artifacts created by this transition.
 8. Satisfiable exact_next_action for MCP scheduled workers.
 """
 from __future__ import annotations
 import json
 import os
 import shutil
 import subprocess
 from typing import Any, Mapping
 import author_mutation_worktree
 import issue_lock_store
 import issue_lock_worktree
 import lease_lifecycle
 from reviewer_worktree import parse_dirty_tracked_files
 import task_capability_map
 BOOTSTRAP_TASKS = frozenset(
    {
        "bootstrap_author_issue_worktree",
        "gitea_bootstrap_author_issue_worktree",
    }
 )
 PHASE_1_REQUEST_ACCEPTED = "1_request_accepted"
 PHASE_2_BRANCH_CONFIRMED = "2_branch_confirmed"
 PHASE_3_PATH_RESERVED = "3_path_reserved"
 PHASE_4_WORKTREE_CONFIRMED = "4_worktree_confirmed"
 PHASE_5_REGISTRATION_VERIFIED = "5_registration_verified"
 PHASE_6_STATE_ESTABLISHED = "6_state_established"
 PHASE_7_TRANSITION_COMPLETED = "7_transition_completed"
 PHASE_COMPENSATING_RECOVERY = "compensating_recovery"
 JOURNAL_DIR_NAME = "bootstrap-journals"
 def is_author_issue_bootstrap_task(task: str | None) -> bool:
    """True when *task* is the author issue worktree bootstrap task."""
    return (task or "").strip() in BOOTSTRAP_TASKS
 def get_journal_dir(override: str | None = None) -> str:
    """Return the root directory for durable bootstrap phase journals."""
    if override:
        path = override
    elif os.environ.get("GITEA_BOOTSTRAP_JOURNAL_DIR"):
        path = os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"]
    else:
        cache_dir = os.path.expanduser("~/.cache/gitea-tools")
        path = os.path.join(cache_dir, JOURNAL_DIR_NAME)
    os.makedirs(path, exist_ok=True)
    return path
 def _journal_file_path(idempotency_key: str, journal_dir: str | None = None) -> str:
    safe_key = "".join(
        c if c.isalnum() or c in ("-", "_", ".") else "_"
        for c in idempotency_key
    )
    return os.path.join(get_journal_dir(journal_dir), f"{safe_key}.json")
 def load_phase_journal(
    idempotency_key: str, journal_dir: str | None = None
 ) -> dict[str, Any] | None:
    """Load a durable phase journal if it exists."""
    path = _journal_file_path(idempotency_key, journal_dir=journal_dir)
    if not os.path.isfile(path):
        return None
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None
 def save_phase_journal(
    journal: dict[str, Any], journal_dir: str | None = None
 ) -> None:
    """Persist a durable phase journal with write-through file sync."""
    key = journal["idempotency_key"]
    path = _journal_file_path(key, journal_dir=journal_dir)
    tmp_path = f"{path}.tmp.{os.getpid()}"
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(journal, f, indent=2, sort_keys=True)
    os.replace(tmp_path, path)
 def derive_default_idempotency_key(
    remote: str,
    org: str | None,
    repo: str | None,
    issue_number: int,
    assignment_id: str | None = None,
    lease_id: str | None = None,
 ) -> str:
    parts = [
        "bootstrap",
        (remote or "prgs").strip(),
        (org or "Scaled-Tech-Consulting").strip(),
        (repo or "Gitea-Tools").strip(),
        f"issue-{issue_number}",
    ]
    if assignment_id:
        parts.append(assignment_id.strip())
    if lease_id:
        parts.append(lease_id.strip())
    return ":".join(parts)
 def run_compensating_recovery(
    journal: dict[str, Any],
    canonical_repo_root: str,
    journal_dir: str | None = None,
 ) -> dict[str, Any]:
    """Execute compensating recovery for artifacts created by this transition only."""
    artifacts = journal.get("artifacts_created") or {}
    pending = journal.get("pending_creations") or {}
    rolled_back: list[str] = []
    worktree_path = journal.get("worktree_path")
    branch_name = journal.get("branch_name")
    # Roll back issue lock if created
    if artifacts.get("lock_created") or pending.get("lock"):
        issue_num = journal.get("issue_number")
        session_id = journal.get("owner_session")
        if issue_num and session_id:
            try:
                issue_lock_store.release_session_lock(
                    issue_number=issue_num,
                    session=session_id,
                    lock_dir=journal_dir,
                )
                rolled_back.append(f"lock:issue-{issue_num}")
            except Exception:
                pass
        artifacts["lock_created"] = False
    worktree_created = (
        artifacts.get("worktree_registered")
        or artifacts.get("worktree_dir_created")
        or (pending.get("worktree_path") == worktree_path and worktree_path)
    )
    if worktree_created and worktree_path:
        if os.path.exists(worktree_path):
            # Re-verify cleanliness before destructive removal (F-4)
            porc_res = subprocess.run(
                ["git", "-C", worktree_path, "status", "--porcelain"],
                capture_output=True,
                text=True,
                check=False,
            )
            is_dirty = porc_res.returncode == 0 and bool(porc_res.stdout.strip())
            if is_dirty:
                rolled_back.append(f"worktree_path_preserved_dirty:{worktree_path}")
            else:
                try:
                    subprocess.run(
                        [
                            "git",
                            "-C",
                            canonical_repo_root,
                            "worktree",
                            "remove",
                            "--force",
                            worktree_path,
                        ],
                        capture_output=True,
                        text=True,
                        check=False,
                    )
                except Exception:
                    pass
                if os.path.exists(worktree_path):
                    shutil.rmtree(worktree_path, ignore_errors=True)
                try:
                    subprocess.run(
                        ["git", "-C", canonical_repo_root, "worktree", "prune"],
                        capture_output=True,
                        text=True,
                        check=False,
                    )
                except Exception:
                    pass
                rolled_back.append(f"worktree_path:{worktree_path}")
        else:
            rolled_back.append(f"worktree_path:{worktree_path}")
    branch_created = (
        artifacts.get("branch_created")
        or (pending.get("branch_name") == branch_name and branch_name)
    )
    if branch_created and branch_name:
        try:
            res = subprocess.run(
                [
                    "git",
                    "-C",
                    canonical_repo_root,
                    "rev-parse",
                    "--verify",
                    branch_name,
                ],
                capture_output=True,
                text=True,
                check=False,
            )
            if res.returncode == 0:
                # Check for author commits on branch before branch deletion (F-4)
                resolved_base = journal.get("resolved_base_sha") or "master"
                rev_list_res = subprocess.run(
                    [
                        "git",
                        "-C",
                        canonical_repo_root,
                        "rev-list",
                        f"{resolved_base}..{branch_name}",
                    ],
                    capture_output=True,
                    text=True,
                    check=False,
                )
                has_commits = rev_list_res.returncode == 0 and bool(rev_list_res.stdout.strip())
                if has_commits:
                    rolled_back.append(f"branch_preserved_commits:{branch_name}")
                else:
                    subprocess.run(
                        [
                            "git",
                            "-C",
                            canonical_repo_root,
                            "branch",
                            "-D",
                            branch_name,
                        ],
                        capture_output=True,
                        text=True,
                        check=False,
                    )
                    rolled_back.append(f"branch:{branch_name}")
        except Exception:
            pass
    recovery_info = {
        "executed": True,
        "rolled_back": rolled_back,
        "reason": journal.get("failure_reason"),
    }
    journal["compensating_recovery"] = recovery_info
    journal["current_phase"] = PHASE_COMPENSATING_RECOVERY
    save_phase_journal(journal, journal_dir=journal_dir)
    return recovery_info
 def assess_author_issue_bootstrap(
    *,
    workspace_path: str,
    canonical_repo_root: str,
    current_branch: str | None = None,
    head_sha: str | None = None,
    porcelain_status: str = "",
    remote_master_sha: str | None = None,
    remote_master_sha_error: str | None = None,
    task: str | None = None,
 ) -> dict[str, Any]:
    """Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
    root = os.path.realpath(canonical_repo_root or "")
    workspace = os.path.realpath(workspace_path or root or ".")
    branch = (current_branch or "").strip()
    dirty = parse_dirty_tracked_files(porcelain_status or "")
    under_branches = (
        author_mutation_worktree.is_path_under_branches(workspace, root)
        if root
        else False
    )
    if not is_author_issue_bootstrap_task(task):
        return {
            "not_applicable": True,
            "allowed": False,
            "block": False,
            "proven": False,
            "reasons": ["task is not author_issue_bootstrap"],
        }
    if under_branches:
        return {
            "not_applicable": False,
            "allowed": True,
            "block": False,
            "proven": True,
            "bootstrap_path": "existing_branches_worktree",
            "reasons": [
                "workspace is already a registered worktree under branches/"
            ],
        }
    reasons: list[str] = []
    if workspace != root:
        reasons.append(
            "bootstrap requires workspace to be canonical control checkout or branches/ worktree"
        )
    if branch not in author_mutation_worktree.BASE_BRANCHES:
        reasons.append(
            f"control checkout branch '{branch}' is not an accepted base branch "
            f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
        )
    if dirty:
        reasons.append(
            f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
        )
    if remote_master_sha_error:
        reasons.append(
            f"could not verify live master tip: {remote_master_sha_error}"
        )
    elif remote_master_sha and head_sha:
        h = head_sha.strip().lower()
        rm = remote_master_sha.strip().lower()
        if h != rm:
            reasons.append(
                f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
            )
    if reasons:
        return {
            "not_applicable": False,
            "allowed": False,
            "block": True,
            "proven": False,
            "reasons": reasons,
        }
    return {
        "not_applicable": False,
        "allowed": True,
        "block": False,
        "proven": True,
        "bootstrap_path": "clean_canonical_control_checkout",
        "reasons": [
            "control checkout is clean on accepted base branch matching live master"
        ],
    }
 import fcntl
 import stat
 class BootstrapTransitionLock:
    """Inter-process file lock scoped to the transition identity / idempotency key."""
    def __init__(self, idempotency_key: str, journal_dir: str | None = None):
        safe_key = "".join(
            c if c.isalnum() or c in ("-", "_", ".") else "_"
            for c in idempotency_key
        )
        lock_dir = os.path.realpath(get_journal_dir(journal_dir))
        if not os.path.isdir(lock_dir):
            raise RuntimeError(f"Lock directory '{lock_dir}' does not exist or is not a directory")
        raw_lock_path = os.path.abspath(os.path.join(lock_dir, f"{safe_key}.lock"))
        try:
            common = os.path.commonpath([lock_dir, os.path.dirname(raw_lock_path)])
        except Exception:
            common = None
        if common != lock_dir:
            raise RuntimeError(f"Lock path '{raw_lock_path}' escapes canonical lock directory '{lock_dir}'")
        self.lock_path = raw_lock_path
        self.fd = None
    def __enter__(self):
        if os.path.islink(self.lock_path):
            raise RuntimeError(f"Refusing lock acquisition: lock path '{self.lock_path}' is a symlink")
        flags = os.O_RDWR | os.O_CREAT
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        if hasattr(os, "O_CLOEXEC"):
            flags |= os.O_CLOEXEC
        try:
            fd = os.open(self.lock_path, flags, 0o600)
        except OSError as exc:
            raise RuntimeError(f"Failed to open lock file safely '{self.lock_path}': {exc}") from exc
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode):
            os.close(fd)
            raise RuntimeError(f"Lock target '{self.lock_path}' is not a regular file")
        self.fd = fd
        fcntl.flock(self.fd, fcntl.LOCK_EX)
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.fd is not None:
            try:
                fcntl.flock(self.fd, fcntl.LOCK_UN)
            except Exception:
                pass
            try:
                os.close(self.fd)
            except Exception:
                pass
            self.fd = None
 def bootstrap_author_issue_worktree(
    *,
    issue_number: int,
    canonical_repo_root: str,
    assignment_id: str | None = None,
    lease_id: str | None = None,
    expected_base_sha: str | None = None,
    branch_name: str | None = None,
    worktree_path: str | None = None,
    idempotency_key: str | None = None,
    remote: str = "prgs",
    host: str | None = None,
    org: str | None = "Scaled-Tech-Consulting",
    repo: str | None = "Gitea-Tools",
    active_identity: str | None = "jcwalker3",
    active_profile: str | None = "prgs-author",
    owner_session: str | None = None,
    lock_dir: str | None = None,
    dry_run: bool = False,
 ) -> dict[str, Any]:
    """Execute the sanctioned author issue worktree bootstrap transition."""
    root = os.path.realpath(canonical_repo_root)
    session = (owner_session or "").strip()
    if not session:
        return {
            "success": False,
            "reason_code": "missing_owner_session",
            "message": "Missing required owner_session parameter (fail closed). Session identifier cannot be fabricated or defaulted.",
            "exact_next_action": (
                "Pass explicit owner_session resolved from gitea_whoami or session context."
            ),
        }
    if active_identity is None or not str(active_identity).strip():
        return {
            "success": False,
            "reason_code": "missing_active_identity",
            "message": "Missing required active_identity parameter (fail closed). Identity cannot be fabricated or defaulted.",
            "exact_next_action": "Pass explicit active_identity resolved from gitea_whoami.",
        }
    identity = str(active_identity).strip()
    if active_profile is None or not str(active_profile).strip():
        return {
            "success": False,
            "reason_code": "missing_active_profile",
            "message": "Missing required active_profile parameter (fail closed). Profile cannot be fabricated or defaulted.",
            "exact_next_action": "Pass explicit active_profile resolved from gitea_whoami.",
        }
    profile = str(active_profile).strip()
    # Derive standard inputs
    expected_pattern = f"issue-{issue_number}"
    target_branch = (branch_name or "").strip()
    if not target_branch:
        target_branch = f"fix/issue-{issue_number}-native-mcp-bootstrap"
    elif expected_pattern not in target_branch:
        return {
            "success": False,
            "reason_code": "invalid_branch_name",
            "message": (
                f"Branch name '{target_branch}' must contain issue pattern '{expected_pattern}'"
            ),
            "exact_next_action": (
                f"Supply a branch_name containing '{expected_pattern}', e.g., 'fix/issue-{issue_number}-...'"
            ),
        }
    worktree_name = target_branch.replace("/", "-")
    target_worktree = (worktree_path or "").strip()
    if not target_worktree:
        target_worktree = os.path.join(root, "branches", worktree_name)
    target_worktree = os.path.realpath(os.path.abspath(target_worktree))
    key = (idempotency_key or "").strip()
    if not key:
        key = derive_default_idempotency_key(
            remote=remote,
            org=org,
            repo=repo,
            issue_number=issue_number,
            assignment_id=assignment_id,
            lease_id=lease_id,
        )
    # Acquire cross-process file lock scoped to the idempotency key / transition identity
    with BootstrapTransitionLock(key, journal_dir=lock_dir):
        # Idempotency check
        existing = load_phase_journal(key, journal_dir=lock_dir)
        if existing and existing.get("completed"):
            if (
                existing.get("issue_number") == issue_number
                and existing.get("branch_name") == target_branch
                and os.path.realpath(existing.get("worktree_path", ""))
                == target_worktree
            ):
                return {
                    "success": True,
                    "replayed": True,
                    "message": (
                        f"Idempotent replay: worktree for issue #{issue_number} already bootstrapped at {target_worktree}"
                    ),
                    "issue_number": issue_number,
                    "branch_name": target_branch,
                    "worktree_path": target_worktree,
                    "base_sha": existing.get("resolved_base_sha"),
                    "lease_id": existing.get("lease_id"),
                    "assignment_id": existing.get("assignment_id"),
                    "idempotency_key": key,
                    "phase_journal": existing,
                    "exact_next_action": (
                        "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
                        "and proceed with author implementation in the bootstrapped worktree."
                    ),
                }
            else:
                return {
                    "success": False,
                    "reason_code": "incompatible_idempotency_replay",
                    "message": (
                        f"Idempotency key '{key}' already exists with incompatible parameters "
                        f"(stored: {existing.get('branch_name')}, {existing.get('worktree_path')}; "
                        f"requested: {target_branch}, {target_worktree})"
                    ),
                    "exact_next_action": (
                        "Supply a unique idempotency_key or pass compatible parameters."
                    ),
                }
        # Initialize or resume Phase Journal
        if existing:
            journal = existing
            artifacts = journal.setdefault("artifacts_created", {})
            artifacts.setdefault("branch_created", False)
            artifacts.setdefault("worktree_dir_created", False)
            artifacts.setdefault("worktree_registered", False)
            artifacts.setdefault("lock_created", False)
        else:
            journal = {
                "idempotency_key": key,
                "issue_number": issue_number,
                "assignment_id": assignment_id,
                "lease_id": lease_id,
                "expected_base_sha": expected_base_sha,
                "resolved_base_sha": None,
                "branch_name": target_branch,
                "worktree_path": target_worktree,
                "active_identity": identity,
                "active_profile": profile,
                "owner_session": session,
                "remote": remote,
                "org": org,
                "repo": repo,
                "phases": {},
                "artifacts_created": {
                    "branch_created": False,
                    "worktree_dir_created": False,
                    "worktree_registered": False,
                    "lock_created": False,
                },
                "current_phase": PHASE_1_REQUEST_ACCEPTED,
                "completed": False,
            }
        # Fetch current live master SHA
        try:
            rev_res = subprocess.run(
                ["git", "-C", root, "rev-parse", "HEAD"],
                capture_output=True,
                text=True,
                check=True,
            )
            live_master_sha = rev_res.stdout.strip()
        except Exception as exc:
            return {
                "success": False,
                "reason_code": "git_rev_parse_failed",
                "message": f"Could not determine repository HEAD: {exc}",
                "exact_next_action": "Verify repository git state and retry.",
            }
        # Phase 1: REQUEST_ACCEPTED & Concurrency Pin Check
        if expected_base_sha:
            exp_norm = expected_base_sha.strip().lower()
            live_norm = live_master_sha.lower()
            if exp_norm != live_norm:
                journal["failure_reason"] = (
                    f"stale concurrency pin: expected {exp_norm[:12]} != live {live_norm[:12]}"
                )
                save_phase_journal(journal, journal_dir=lock_dir)
                return {
                    "success": False,
                    "reason_code": "stale_concurrency_pin",
                    "message": (
                        f"Expected base SHA {exp_norm[:12]} does not match live master SHA {live_norm[:12]} (fail closed)."
                    ),
                    "expected_base_sha": expected_base_sha,
                    "live_master_sha": live_master_sha,
                    "exact_next_action": (
                        "Re-evaluate assignment against current live master SHA and retry with updated expected_base_sha."
                    ),
                }
        journal["resolved_base_sha"] = live_master_sha
        journal["phases"][PHASE_1_REQUEST_ACCEPTED] = {
            "status": "completed",
            "live_master_sha": live_master_sha,
            "expected_base_sha": expected_base_sha,
        }
        journal["current_phase"] = PHASE_2_BRANCH_CONFIRMED
        save_phase_journal(journal, journal_dir=lock_dir)
        if dry_run:
            return {
                "success": True,
                "dry_run": True,
                "message": f"Dry-run: validated bootstrap intent for issue #{issue_number}",
                "issue_number": issue_number,
                "branch_name": target_branch,
                "worktree_path": target_worktree,
                "base_sha": live_master_sha,
                "phase_journal": journal,
                "exact_next_action": "Run without dry_run=True to execute bootstrap.",
            }
        # Phase 2: BRANCH_CONFIRMED
        was_branch_created_previously = journal["artifacts_created"].get("branch_created", False)
        pending_branch = (journal.get("pending_creations") or {}).get("branch_name")
        branch_check = subprocess.run(
            ["git", "-C", root, "rev-parse", "--verify", target_branch],
            capture_output=True,
            text=True,
            check=False,
        )
        if branch_check.returncode == 0:
            branch_head = branch_check.stdout.strip()
            # Verify branch head descends from base
            anc_check = subprocess.run(
                [
                    "git",
                    "-C",
                    root,
                    "merge-base",
                    "--is-ancestor",
                    live_master_sha,
                    branch_head,
                ],
                capture_output=True,
                text=True,
                check=False,
            )
            if anc_check.returncode != 0 and branch_head.lower() != live_master_sha.lower():
                # F-8: Check if branch shares a common ancestor with live master
                mb_check = subprocess.run(
                    ["git", "-C", root, "merge-base", live_master_sha, branch_head],
                    capture_output=True,
                    text=True,
                    check=False,
                )
                if mb_check.returncode != 0 or not mb_check.stdout.strip():
                    journal["failure_reason"] = (
                        f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})"
                    )
                    save_phase_journal(journal, journal_dir=lock_dir)
                    return {
                        "success": False,
                        "reason_code": "incompatible_existing_branch",
                        "message": (
                            f"Existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})."
                        ),
                        "exact_next_action": (
                            "Inspect or sync the existing branch with master before bootstrapping."
                        ),
                    }
            # Preserve creation provenance monotonically across interruption and replay
            if was_branch_created_previously or pending_branch == target_branch:
                journal["artifacts_created"]["branch_created"] = True
            else:
                journal["artifacts_created"]["branch_created"] = False
        else:
            # Persist creation intent/provenance to disk BEFORE executing external mutation
            journal.setdefault("pending_creations", {})["branch_name"] = target_branch
            journal["artifacts_created"]["branch_created"] = True
            save_phase_journal(journal, journal_dir=lock_dir)
            # Create branch
            create_res = subprocess.run(
                ["git", "-C", root, "branch", target_branch, live_master_sha],
                capture_output=True,
                text=True,
                check=False,
            )
            if create_res.returncode != 0:
                journal["artifacts_created"]["branch_created"] = False
                journal.get("pending_creations", {}).pop("branch_name", None)
                journal["failure_reason"] = (
                    f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}"
                )
                save_phase_journal(journal, journal_dir=lock_dir)
                return {
                    "success": False,
                    "reason_code": "branch_creation_failed",
                    "message": f"Failed to create git branch '{target_branch}': {create_res.stderr.strip()}",
                    "exact_next_action": "Verify branch availability and retry.",
                }
        journal["phases"][PHASE_2_BRANCH_CONFIRMED] = {
            "status": "completed",
            "branch_name": target_branch,
            "created": journal["artifacts_created"]["branch_created"],
        }
        journal["current_phase"] = PHASE_3_PATH_RESERVED
        save_phase_journal(journal, journal_dir=lock_dir)
        # Phase 3: PATH_RESERVED & Phase 4: WORKTREE_CONFIRMED
        if not author_mutation_worktree.is_path_under_branches(
            target_worktree, root
        ):
            journal["failure_reason"] = (
                f"target_worktree '{target_worktree}' is outside canonical branches/ root"
            )
            run_compensating_recovery(journal, root, journal_dir=lock_dir)
            return {
                "success": False,
                "reason_code": "path_outside_canonical_branches_root",
                "message": (
                    f"Worktree path '{target_worktree}' is outside canonical branches/ root (fail closed)."
                ),
                "exact_next_action": (
                    "Provide a worktree_path inside canonical branches/ root, e.g., 'branches/issue-...'"
                ),
            }
        was_dir_created_previously = journal["artifacts_created"].get("worktree_dir_created", False)
        was_registered_previously = journal["artifacts_created"].get("worktree_registered", False)
        pending_wt = (journal.get("pending_creations") or {}).get("worktree_path")
        dir_exists = os.path.exists(target_worktree)
        if dir_exists:
            # Check porcelain directly
            porc_res = subprocess.run(
                ["git", "-C", target_worktree, "status", "--porcelain"],
                capture_output=True,
                text=True,
                check=False,
            )
            dirty = (
                parse_dirty_tracked_files(porc_res.stdout)
                if porc_res.returncode == 0
                else []
            )
            if dirty or (porc_res.returncode == 0 and porc_res.stdout.strip()):
                journal["failure_reason"] = (
                    f"target worktree '{target_worktree}' contains dirty tracked/untracked files"
                )
                run_compensating_recovery(journal, root, journal_dir=lock_dir)
                return {
                    "success": False,
                    "reason_code": "preexisting_dirty_worktree",
                    "message": (
                        f"Preexisting worktree '{target_worktree}' has dirty tracked/untracked files (fail closed)."
                    ),
                    "exact_next_action": (
                        "Clean or stash the pre-existing worktree files before bootstrapping."
                    ),
                }
            # Check registered branch
            wt_state = issue_lock_worktree.read_worktree_git_state(target_worktree)
            wt_branch = (wt_state.get("current_branch") or "").strip()
            if wt_branch and wt_branch != target_branch:
                journal["failure_reason"] = (
                    f"existing worktree '{target_worktree}' is on branch '{wt_branch}' != expected '{target_branch}'"
                )
                run_compensating_recovery(journal, root, journal_dir=lock_dir)
                return {
                    "success": False,
                    "reason_code": "incompatible_existing_directory",
                    "message": (
                        f"Existing worktree '{target_worktree}' is registered to branch '{wt_branch}' instead of '{target_branch}'."
                    ),
                    "exact_next_action": (
                        "Inspect or remove the pre-existing worktree folder before bootstrapping."
                    ),
                }
            if was_dir_created_previously or pending_wt == target_worktree:
                journal["artifacts_created"]["worktree_dir_created"] = True
                journal["artifacts_created"]["worktree_registered"] = True
            else:
                journal["artifacts_created"]["worktree_dir_created"] = False
                journal["artifacts_created"]["worktree_registered"] = False
        else:
            # Persist creation intent/provenance to disk BEFORE executing external worktree add mutation
            journal.setdefault("pending_creations", {})["worktree_path"] = target_worktree
            journal["artifacts_created"]["worktree_dir_created"] = True
            journal["artifacts_created"]["worktree_registered"] = True
            save_phase_journal(journal, journal_dir=lock_dir)
            wt_add_res = subprocess.run(
                [
                    "git",
                    "-C",
                    root,
                    "worktree",
                    "add",
                    target_worktree,
                    target_branch,
                ],
                capture_output=True,
                text=True,
                check=False,
            )
            if wt_add_res.returncode != 0:
                journal["artifacts_created"]["worktree_dir_created"] = False
                journal["artifacts_created"]["worktree_registered"] = False
                journal.get("pending_creations", {}).pop("worktree_path", None)
                journal["failure_reason"] = (
                    f"git worktree add failed: {wt_add_res.stderr.strip()}"
                )
                run_compensating_recovery(journal, root, journal_dir=lock_dir)
                return {
                    "success": False,
                    "reason_code": "worktree_add_failed",
                    "message": f"Failed to execute git worktree add: {wt_add_res.stderr.strip()}",
                    "exact_next_action": "Verify git worktree capabilities and retry.",
                }
        journal["phases"][PHASE_3_PATH_RESERVED] = {
            "status": "completed",
            "worktree_path": target_worktree,
            "preexisting_dir": dir_exists,
        }
        journal["current_phase"] = PHASE_4_WORKTREE_CONFIRMED
        save_phase_journal(journal, journal_dir=lock_dir)
        # Phase 5: REGISTRATION_VERIFIED
        wt_list_res = subprocess.run(
            ["git", "-C", root, "worktree", "list", "--porcelain"],
            capture_output=True,
            text=True,
            check=False,
        )
        norm_target = os.path.realpath(target_worktree)
        found_registration = False
        if wt_list_res.returncode == 0:
            for block in wt_list_res.stdout.split("\n\n"):
                lines = block.strip().splitlines()
                worktree_line = next(
                    (l[9:].strip() for l in lines if l.startswith("worktree ")),
                    None,
                )
                if worktree_line and os.path.realpath(worktree_line) == norm_target:
                    found_registration = True
                    break
        if not found_registration:
            journal["failure_reason"] = (
                f"worktree registration for '{target_worktree}' not found in git worktree list"
            )
            run_compensating_recovery(journal, root, journal_dir=lock_dir)
            return {
                "success": False,
                "reason_code": "worktree_registration_verification_failed",
                "message": f"Worktree '{target_worktree}' registration verification failed.",
                "exact_next_action": "Check git worktree list integrity and retry.",
            }
        journal["phases"][PHASE_4_WORKTREE_CONFIRMED] = {
            "status": "completed",
            "worktree_path": target_worktree,
        }
        journal["phases"][PHASE_5_REGISTRATION_VERIFIED] = {
            "status": "completed",
            "registered": True,
        }
        journal["current_phase"] = PHASE_6_STATE_ESTABLISHED
        save_phase_journal(journal, journal_dir=lock_dir)
        # Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition
        from datetime import datetime, timezone
        try:
            lock_data = {
                "remote": remote,
                "org": org or "Scaled-Tech-Consulting",
                "repo": repo or "Gitea-Tools",
                "issue_number": issue_number,
                "branch": target_branch,
                "branch_name": target_branch,
                "worktree_path": target_worktree,
                "owner_session": session,
                "claimant": {
                    "username": identity,
                    "profile": profile,
                },
                "assignment_id": assignment_id,
                "lease_id": lease_id,
                "expected_base_sha": live_master_sha,
                "created_at": datetime.now(timezone.utc).isoformat(),
            }
            journal.setdefault("pending_creations", {})["lock"] = True
            journal["artifacts_created"]["lock_created"] = True
            save_phase_journal(journal, journal_dir=lock_dir)
            lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir)
        except Exception as exc:
            journal["artifacts_created"]["lock_created"] = False
            journal.get("pending_creations", {}).pop("lock", None)
            journal["failure_reason"] = f"issue lock binding failed: {exc}"
            run_compensating_recovery(journal, root, journal_dir=lock_dir)
            return {
                "success": False,
                "reason_code": "issue_lock_acquisition_failed",
                "message": f"Could not bind canonical issue lock for issue #{issue_number}: {exc}",
                "exact_next_action": "Verify lease/assignment state and retry.",
            }
        journal["phases"][PHASE_6_STATE_ESTABLISHED] = {
            "status": "completed",
            "lock": lock_res,
        }
        journal["phases"][PHASE_7_TRANSITION_COMPLETED] = {
            "status": "completed",
        }
        journal["current_phase"] = PHASE_7_TRANSITION_COMPLETED
        journal["completed"] = True
        save_phase_journal(journal, journal_dir=lock_dir)
        return {
            "success": True,
            "replayed": False,
            "message": (
                f"Successfully bootstrapped author issue worktree for issue #{issue_number} "
                f"at branch '{target_branch}' and worktree '{target_worktree}'."
            ),
            "issue_number": issue_number,
            "branch_name": target_branch,
            "worktree_path": target_worktree,
            "base_sha": live_master_sha,
            "lease_id": lease_id,
            "assignment_id": assignment_id,
            "idempotency_key": key,
            "lock_state": lock_res,
            "phase_journal": journal,
            "exact_next_action": (
                "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
                "and proceed with author implementation in the bootstrapped worktree."
            ),
        }
@@ -40,66 +40,22 @@ def _normalize_path(path: str) -> str:
    return (path or "").replace("\\", "/").rstrip("/")
 def get_canonical_branches_root(project_root: str | None = None) -> str:
    """Return the absolute path of the canonical branches directory for *project_root*."""
    root = os.path.realpath(project_root) if project_root else os.path.realpath(os.getcwd())
    canonical_repo_root = resolve_canonical_repo_root(root, root)
    return os.path.realpath(os.path.join(canonical_repo_root, "branches"))
 def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
-    """True when *path* resolves inside a canonical ``branches/`` directory."""
+    """True when *path* resolves inside ``<project_root>/branches/``."""
-    if not path or not str(path).strip():
+    normalized = _normalize_path(path)
    if not normalized:
        return False
-    try:
+    if "/branches/" in f"{normalized}/":
-        real_path = os.path.realpath(os.path.abspath(str(path).strip()))
+        return True
-    except Exception:
+    if normalized.endswith("/branches"):
-        return False
+        return True
-
+    if project_root:
-    branches_root = get_canonical_branches_root(project_root or real_path)
+        root = _normalize_path(os.path.realpath(project_root))
-    try:
+        real = _normalize_path(os.path.realpath(path))
-        common = os.path.commonpath([branches_root, real_path])
+        if real.startswith(f"{root}/"):
-    except Exception:
+            rel = real[len(root) + 1 :]
-        return False
+            return rel == "branches" or rel.startswith("branches/")
-
+    return False
    if common != branches_root:
        return False
    rel = os.path.relpath(real_path, branches_root)
    return rel != "." and not rel.startswith("..")
 def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
    """Return the stable repository root for *workspace_path* via git metadata (#460)."""
    p = (workspace_path or "").strip()
    if p:
        try:
            res = subprocess.run(
                ["git", "-C", p, "rev-parse", "--git-common-dir"],
                capture_output=True,
                text=True,
                check=True,
            )
            common = _realpath_git_common_dir(p, res.stdout)
            if common.endswith(f"{os.sep}.git") or os.path.basename(common) == ".git":
                candidate_root = os.path.dirname(common)
                real_p = os.path.realpath(p)
                try:
                    if os.path.commonpath([candidate_root, real_p]) == candidate_root:
                        return candidate_root
                except Exception:
                    pass
        except Exception:
            pass
    fallback = os.path.realpath(fallback_project_root or workspace_path or ".")
    norm = fallback.replace("\\", "/")
    if "/branches/" in norm:
        return os.path.realpath(norm.split("/branches/")[0])
    elif norm.endswith("/branches"):
        return os.path.realpath(os.path.dirname(fallback))
    return fallback
 def resolve_mutation_workspace(
@@ -131,6 +87,29 @@ def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
    return os.path.realpath(os.path.join(workspace_path, raw))
 def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
    """Return the stable repository root for *workspace_path* via git metadata (#460)."""
    path = (workspace_path or "").strip()
    fallback = os.path.realpath(fallback_project_root)
    if not path:
        return fallback
    try:
        res = subprocess.run(
            ["git", "-C", path, "rev-parse", "--git-common-dir"],
            capture_output=True,
            text=True,
            check=True,
        )
        common = _realpath_git_common_dir(path, res.stdout)
    except Exception:
        return fallback
    if common.endswith(f"{os.sep}.git"):
        return os.path.dirname(common)
    if os.path.basename(common) == ".git":
        return os.path.dirname(common)
    return fallback
 def resolve_author_mutation_context(
    worktree_path: str | None,
    process_project_root: str,
@@ -46,17 +46,6 @@ _FIELD_RE = re.compile(
 )
 def is_known_cth_type(value: str | None) -> bool:
    """True when *value* is a declared member of the :data:`CTH_TYPES` contract.
    ``CTH_TYPES`` is the single authority for what a CTH type may be. The
    heading a comment carries is free text, so a *read* path that turns a parsed
    type into something durable — a serialized field, a routing decision — must
    check membership here rather than trust the parse or keep a list of its own.
    """
    return (value or "").strip() in CTH_TYPES
 def format_cth_body(
    *,
    cth_type: str,
@@ -71,7 +60,7 @@ def format_cth_body(
 ) -> str:
    """Render a canonical CTH comment body."""
    normalized_type = (cth_type or "").strip()
-    if not is_known_cth_type(normalized_type):
+    if normalized_type not in CTH_TYPES:
        raise ValueError(
            f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}"
        )
@@ -112,12 +101,6 @@ def parse_cth_comment(body: str) -> dict[str, Any] | None:
        fields[key] = match.group(2).strip()
    return {
        "cth_type": cth_type,
        # The heading capture is unconstrained free text, so the parse states
        # whether it satisfies the CTH_TYPES contract instead of leaving every
        # reader to decide (or forget). Parsing stays total — an unknown type is
        # still parsed and reported, never raised on — but a reader that turns
        # the type into a durable value can now tell the two apart.
        "cth_type_known": is_known_cth_type(cth_type),
        "fields": fields,
        "raw_body": text,
    }
@@ -136,7 +119,7 @@ def assess_cth_comment(body: str) -> dict[str, Any]:
        }
    cth_type = parsed.get("cth_type") or ""
-    if not is_known_cth_type(cth_type):
+    if cth_type not in CTH_TYPES:
        reasons.append(
            f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}"
        )
@@ -247,8 +247,7 @@ def bootstrap_permits_control_checkout(
    """
    if not isinstance(assessment, dict):
        return False
-    import author_issue_bootstrap
+    if not is_create_issue_task(task):
    if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task):
        return False
    # Positive proof: the assessment must affirmatively allow, with no
@@ -69,7 +69,6 @@ that gates each call, not which tools exist.
 - `gitea_audit_worktree_cleanup`
 - `gitea_authorize_reconciliation_cleanup_phase`
 - `gitea_authorize_review_correction`
 - `gitea_bootstrap_author_issue_worktree`
 - `gitea_capability_stop_terminal_report`
 - `gitea_capture_branches_worktree_snapshot`
 - `gitea_check_pr_eligibility`
@@ -65,6 +65,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/prompts` | JSON prompt export with workflow hashes |
 | `/runtime` | MCP runtime health and stale detection (#430) |
 | `/api/runtime` | JSON runtime health export |
 | `/policy` | Workflow policy and guardrail configuration visibility (#646) |
 | `/api/v1/policy` | Versioned JSON guardrail inventory (redacted, read-only) |
 | `/audit` | Report audit paste + validator preview (#431) |
 | `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
 | `/worktrees` | Worktree hygiene dashboard (#432) |
@@ -74,11 +76,6 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
 | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
 | `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
 | `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
 | `/timeline` | Phase 1 shell stub — workflow event timeline |
 | `/policy` | Phase 1 shell stub — capability/role policy placeholder |
 | `/insights` | Phase 1 shell stub — operational insights placeholder |
 Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
@@ -238,25 +235,18 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
-## Application shell — Phase 1 (#638)
+## Policy & guardrail visibility (#646)
-The console shell (`webui/layout.py`) renders a grouped navigation driven by a
+`/policy` (HTML) and `/api/v1/policy` (JSON) surface a **read-only** projection
-single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
+of the major workflow guardrails — role separation/RBAC, lease lifecycle,
-Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
+author worktree binding, merge confirmation, secret redaction, contamination
-Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
+containment, allocator policy, audit logging, and mutation gating. Each entry
-(placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
+carries source pointers to the file/module/doc that owns it, a compact active
-place so the layout and the route table cannot drift.
+value derived from the existing safe policy accessors, and — where a documented
-
+default is declared — a diff of active vs documented. The whole payload is run
-The header carries two read-only status badges — an **environment** badge
+through the console redaction pass before it is emitted, so a planted or
-(`local` for loopback binds, `remote` otherwise, derived from `WEBUI_HOST`) and
+accidental secret degrades to the placeholder rather than reaching a client.
-a **mode: read-only** badge — plus a **Docs** link to this document. No
+The view never edits policy and exposes no gate-weakening toggle.
 privileged action controls are present in the Phase 1 shell.
 Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
 `/policy`, `/insights`) resolve to graceful read-only stub pages instead of
 404s; their backing views land in later child issues of #631 (the inventory
 surfaces are backed by #636). Mutating methods on stub routes still fail closed
 with `read-only-mvp`.
 ## Deployment boundary (#435)
@@ -317,108 +307,6 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
 ## Workflow-event timeline (#637)
 `GET /api/v1/timeline` is a read-only, versioned aggregation of workflow
 events from every available source into one normalised, filterable stream. It
 is the model layer for the Phase 1 timeline console view (a later child issue
 of #631); this issue ships the schema, adapters, and read API only.
 ### Schema (versioned)
 `webui/timeline.py` declares `TIMELINE_SCHEMA_VERSION` (currently `1`) and the
 frozen `WorkflowEvent` record. Every response carries `schema_version` so a
 consumer can branch on shape. One event:
 ```json
 {
  "source": "control_plane",
  "event_type": "lease.renew",
  "event_key": "cp:1421",
  "timestamp": "2026-07-23T02:00:00Z",
  "actor": null,
  "role": null,
  "issue_number": 637,
  "pr_number": null,
  "session_id": null,
  "tool_name": null,
  "decision": null,
  "message": "lease renewed",
  "correlation_id": "issue#637",
  "evidence_refs": [],
  "sensitive": true
 }
 ```
 `event_key` is stable and unique per source (`cp:<event_id>`,
 `cth:<kind>:<number>:<comment_id>`), so pagination and dedup are deterministic.
 ### Sources and field authority
 | Source | Adapter | Authority |
 |---|---|---|
 | Control-plane `events` ⋈ `work_items` | `adapt_cp_events` | `event_type`, `message`, `timestamp`, issue/PR scope come from the CP database, read through a `mode=ro` URI (never creates the DB or runs migrations) |
 | Gitea Canonical Thread Handoff comments | `adapt_cth_comments` | `actor`, `role` (next owner), `decision`, `evidence_refs`, `timestamp` come from the parsed CTH comment body (`canonical_thread_handoff`) |
 Handoff comments are thread-scoped: they are only read when the request filters
 by a single `issue` or `pr`. Otherwise the handoff source reports `not run`
 with a reason — it is never rendered as empty-and-healthy. Each source degrades
 independently: an unavailable control-plane DB or a failed comment fetch is a
 `sources[]` entry with `ok:false` and a `reason`, never a dropped timeline.
 ### Query parameters
 `issue`, `pr`, `session` (conjunctive filters); `limit` (default 50, max 500)
 and `offset` for pagination; `remote`, `org`, `repo` to override the default
 registry-project scope. Events sort ascending by
 `(timestamp, source_rank, event_key)`; missing timestamps sort last.
 ### Filter authority, and refusing what cannot be answered
 A filter dimension is only meaningful for a source whose records carry it.
 Each source declares its own support in `_SOURCE_FILTER_SUPPORT` and reports it
 per response as `supported_filters` / `unsupported_filters`:
 | Source | issue | pr | session |
 |---|---|---|---|
 | `control_plane` | yes | yes | **no** — the `events` table is `(event_id, work_item_id, event_type, message, created_at)` and records no session |
 | `gitea_handoff` | yes | yes | yes — a CTH comment declares its own `Session:` field |
 `session_id` is read only from that declared CTH field. It is never inferred
 from a work item, an actor, or message text, and a value that is
 redaction-altering or bare-secret-shaped is dropped rather than emitted.
 When **no source that ran** can carry a requested dimension, the request is
 refused rather than answered: the response is `422` with `ok:false` and a
 structured `error` naming `unsupported_filters` and the per-source reason. A
 `200` with zero events would tell an operator that no such activity exists,
 which is a stronger — and false — claim than "this cannot be answered here".
 A source that *can* answer the dimension and simply matched nothing still
 returns `200` with `ok:true` and an empty page.
 ### Redaction
 Every free-text field (event messages, decision/proof text, roles, actors) is
 passed through the console redaction policy (`webui.console_redaction`, backed
 by `gitea_audit.redact`) before it leaves the module, failing closed to the
 placeholder. No unredacted tool arguments or secrets are ever emitted, and a
 generation error never drops raw data to a caller or a log.
 Redaction also runs *before* any structured value is derived from free text.
 `evidence_refs` are extracted from already-redacted proof/decision text, and a
 commit reference is recognised only where the text declares one (`commit`,
 `head`, `base`, `sha`, …). An undeclared 40-character hex run has the exact
 shape of a Gitea access token, so it is never lifted out of prose into a
 structured field. Every reference is then independently revalidated against an
 allowed shape and a second redaction pass immediately before serialization;
 anything unproven is dropped and the event is flagged `sensitive`.
 ### Tests
 ```bash
 pytest tests/test_webui_timeline.py -q
 ```
 ## Tests
 ```bash
@@ -958,32 +958,6 @@ def _create_issue_bootstrap_assessment(
    """
    import create_issue_bootstrap as _cib
    import author_issue_bootstrap as _aib
    if _aib.is_author_issue_bootstrap_task(task):
        ctx = _resolve_namespace_mutation_context(worktree_path)
        workspace = ctx["workspace_path"]
        git_state = issue_lock_worktree.read_worktree_git_state(workspace)
        remote_master_sha_error: str | None = None
        try:
            remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
                ctx["canonical_repo_root"]
            )
        except Exception as exc:
            remote_master_sha = None
            remote_master_sha_error = (
                f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
            )
        return _aib.assess_author_issue_bootstrap(
            workspace_path=workspace,
            canonical_repo_root=ctx["canonical_repo_root"],
            current_branch=git_state.get("current_branch"),
            head_sha=git_state.get("head_sha"),
            porcelain_status=git_state.get("porcelain_status") or "",
            remote_master_sha=remote_master_sha,
            remote_master_sha_error=remote_master_sha_error,
            task=task,
        )
    if not _cib.is_create_issue_task(task):
        return None
@@ -1476,7 +1450,7 @@ def verify_preflight_purity(
            dirty_files = sorted(
                _parse_porcelain_entries(_get_workspace_porcelain(workspace))
            )
-            if dirty_files and task != "commit_files":
+            if dirty_files:
                raise RuntimeError(
                    nwb.format_namespace_workspace_binding_error(
                        role_kind=role,
@@ -9036,7 +9010,6 @@ def gitea_commit_files(
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    worktree_path: str | None = None,
 ) -> dict:
    """Commit changes to multiple files in a Gitea repository in a single atomic commit.
@@ -9049,46 +9022,10 @@ def gitea_commit_files(
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.
        worktree_path: Optional worktree path for author mutation context.
    Returns:
        dict with success status and commit/branch information.
    """
    if worktree_path is None:
        lock_data = issue_lock_store.read_session_issue_lock() or {}
        worktree_path = lock_data.get("worktree_path")
        if not worktree_path:
            try:
                prof = get_profile()
                uname = prof.get("username") or prof.get("profile_name")
                for path in issue_lock_store.iter_lock_files():
                    lk = issue_lock_store.read_lock_file(path) or {}
                    claimant = lk.get("claimant") or {}
                    if lk.get("remote") == remote and (claimant.get("username") == uname or lk.get("profile") == prof.get("profile_name")):
                        issue_lock_store.bind_session_lock(lk, renewal_sanctioned=True)
                        worktree_path = lk.get("worktree_path")
                        break
            except Exception:
                pass
    if worktree_path is None and files:
        for f in files:
            p = f.get("workspace_path") or f.get("local_path") or ""
            if p and os.path.isabs(p):
                real_p = os.path.realpath(p)
                real_root = os.path.realpath(PROJECT_ROOT)
                branches_dir = os.path.join(real_root, "branches")
                if real_p.startswith(branches_dir + os.sep):
                    rel_sub = os.path.relpath(real_p, branches_dir)
                    wt_folder = rel_sub.split(os.sep)[0]
                    if wt_folder and wt_folder != "..":
                        worktree_path = os.path.join(branches_dir, wt_folder)
                        break
    if worktree_path:
        os.environ["GITEA_AUTHOR_WORKTREE"] = worktree_path
        os.environ["GITEA_ACTIVE_WORKTREE"] = worktree_path
    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
        "commit_files"
    )
@@ -9101,7 +9038,7 @@ def gitea_commit_files(
            "reasons": block_reasons,
        }
    blocked = _namespace_mutation_block(
-        "commit_files", commit="", branch="", remote=remote, worktree_path=worktree_path
+        "commit_files", commit="", branch="", remote=remote
    )
    if blocked:
        return blocked
@@ -9129,7 +9066,7 @@ def gitea_commit_files(
        )
    # #735: forward explicit org/repo into shared anti-stomp preflight.
-    verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
+    verify_preflight_purity(remote, task="commit_files", org=org, repo=repo)
    processed_files, source_proofs = _prepare_commit_payload_files(files)
    h, o, r = _resolve(remote, host, org, repo)
@@ -9406,96 +9343,6 @@ def gitea_publish_unpublished_issue_branch(
    }
@mcp.tool()
 def gitea_bootstrap_author_issue_worktree(
    issue_number: int,
    assignment_id: str | None = None,
    lease_id: str | None = None,
    expected_base_sha: str | None = None,
    branch_name: str | None = None,
    worktree_path: str | None = None,
    idempotency_key: str | None = None,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    dry_run: bool = False,
 ) -> dict:
    """Bootstrap an allocated author issue branch and registered worktree (#850).
    Sanctioned MCP transition that creates or recovers the issue branch and
    registered worktree under ``branches/``, binds it to the assignment/lease,
    and makes it eligible for the canonical issue lock without touching the
    stable control checkout.
    Args:
        issue_number: Allocated issue number to bootstrap.
        assignment_id: Optional allocation assignment ID.
        lease_id: Optional workflow lease ID.
        expected_base_sha: Authoritative expected base SHA / concurrency pin.
        branch_name: Optional custom branch name (must match issue-<N> pattern).
        worktree_path: Optional custom worktree path under branches/.
        idempotency_key: Optional key for idempotent replay/resume.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override Gitea host.
        org: Override Org.
        repo: Override Repo.
        dry_run: Report planned transition without mutating repository.
    """
    task = "bootstrap_author_issue_worktree"
    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
        task
    )
    if not ok:
        return _author_mutation_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)
    canonical_root = _canonical_local_git_root()
    import author_issue_bootstrap
    return author_issue_bootstrap.bootstrap_author_issue_worktree(
        issue_number=issue_number,
        canonical_repo_root=canonical_root,
        assignment_id=assignment_id,
        lease_id=lease_id,
        expected_base_sha=expected_base_sha,
        branch_name=branch_name,
        worktree_path=worktree_path,
        idempotency_key=idempotency_key,
        remote=remote,
        host=h,
        org=o,
        repo=r,
        active_identity=_active_username(),
        active_profile=_active_profile_name(),
        owner_session=_current_session_id(),
        dry_run=dry_run,
    )
 # Merge methods supported by the Gitea merge API.
 _MERGE_METHODS = ("merge", "squash", "rebase")
@@ -19356,12 +19203,6 @@ def gitea_resolve_task_capability(
    task: str,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    issue_number: int | None = None,
    worktree_path: str | None = None,
    pr_number: int | None = None,
    **kwargs: Any,
 ) -> dict:
    """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
@@ -19378,14 +19219,13 @@ def gitea_resolve_task_capability(
        remote: Known remote instance name.
        host: Optional override for the Gitea host.
    """
    task_key = task_capability_map._canonical_preflight_task(task)
    TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
    # Every fresh attempt invalidates the previous task/role stamp before any
    # fallible resolver work. Unknown/malformed tasks and unexpected failures
    # therefore remain fail-closed instead of preserving stale authority.
    _clear_resolved_capability_stamp()
-    if task_key not in TASK_MAP:
+    if task not in TASK_MAP:
        # #723: structured fail-closed unknown_task (never raise into internal_error).
        profile = get_profile()
        h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
@@ -19431,8 +19271,8 @@ def gitea_resolve_task_capability(
            result["cleared_stale_denial"] = True
        return result
-    required_permission = task_capability_map.required_permission(task_key)
+    required_permission = task_capability_map.required_permission(task)
-    required_role = task_capability_map.required_role(task_key)
+    required_role = task_capability_map.required_role(task)
    role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
    infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -19618,10 +19458,7 @@ def gitea_resolve_task_capability(
    available_in_session = allowed_in_current_session
    runtime_stale_blocker = False
-    if (
+    if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
        "PYTEST_CURRENT_TEST" not in os.environ
        or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ
    ) and os.environ.get("GITEA_ALLOW_STALE_RUNTIME") != "1":
        runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
        if runtime_reasons:
            restart_required = True
@@ -566,10 +566,6 @@ def build_pr_cleanup_entry(
        worktree_state=worktree_state,
        active_lock=active_lock,
    )
    planned = plan_cleanup_execution_order(
        remote_assessment=remote,
        local_assessment=local,
    )
    return {
        "pr_number": pr_number,
        "issue_number": issue_number,
@@ -580,63 +576,9 @@ def build_pr_cleanup_entry(
        "merged": merged,
        "remote_branch": remote,
        "local_worktree": local,
        # #851: dry-run and execute share the same lifecycle order description.
        "planned_execution_order": planned,
    }
 def plan_cleanup_execution_order(
    *,
    remote_assessment: dict[str, Any] | None,
    local_assessment: dict[str, Any] | None,
 ) -> list[dict[str, Any]]:
    """Describe independent worktree-then-reassess-then-remote cleanup order (#851).
    Remote ownership protection remains fail-closed at execute time. A worktree
    that is independently safe to remove is never skipped merely because remote
    deletion may be blocked by that same ``worktree_binding``.
    """
    remote = remote_assessment or {}
    local = local_assessment or {}
    steps: list[dict[str, Any]] = []
    worktree_safe = bool(local.get("safe_to_remove_worktree"))
    remote_safe = bool(remote.get("safe_to_delete_remote"))
    if worktree_safe:
        steps.append(
            {
                "action": "remove_local_worktree",
                "reason": "independently_safe_to_remove",
                "phase": 1,
            }
        )
    if remote_safe:
        if worktree_safe:
            steps.append(
                {
                    "action": "reassess_branch_ownership",
                    "reason": "after_worktree_removal_clear_worktree_binding",
                    "phase": 2,
                }
            )
            steps.append(
                {
                    "action": "delete_remote_branch",
                    "reason": "only_if_independently_safe_after_reassessment",
                    "phase": 3,
                }
            )
        else:
            steps.append(
                {
                    "action": "delete_remote_branch",
                    "reason": "safe_to_delete_and_no_independent_worktree_removal",
                    "phase": 1,
                }
            )
    return steps
 def build_reconciliation_report(
    *,
    project_root: str,
@@ -73,8 +73,6 @@ AUTHOR_TASKS = frozenset({
    "claim_issue",
    "create_branch",
    "push_branch",
    "bootstrap_author_issue_worktree",
    "gitea_bootstrap_author_issue_worktree",
    "create_pr",
    "comment_pr",
    "address_pr_change_requests",
@@ -58,14 +58,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.branch.create",
        "role": "author",
    },
    "bootstrap_author_issue_worktree": {
        "permission": "gitea.branch.create",
        "role": "author",
    },
    "gitea_bootstrap_author_issue_worktree": {
        "permission": "gitea.branch.create",
        "role": "author",
    },
    "push_branch": {
        "permission": "gitea.branch.push",
        "role": "author",
@@ -485,8 +477,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
 # merger lease (#763).
 _PREFLIGHT_TASK_TRANSITIONS = frozenset({
    ("review_pr", "acquire_reviewer_pr_lease"),
    ("work_issue", "bootstrap_author_issue_worktree"),
    ("bootstrap_author_issue_worktree", "lock_issue"),
 })
@@ -533,8 +523,6 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
        "gitea_release_merger_pr_lease",
        "create_branch",
        "push_branch",
        "bootstrap_author_issue_worktree",
        "gitea_bootstrap_author_issue_worktree",
        "publish_unpublished_branch",
        "create_pr",
        "commit_files",
@@ -560,7 +548,6 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
    "gitea_set_issue_labels": "set_issue_labels",
    "gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
    "gitea_create_label": "create_label",
    "gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree",
    "gitea_commit_files": "commit_files",
 }
@@ -1,483 +0,0 @@
 """Regression test suite for native author issue worktree bootstrap (#850)."""
 from __future__ import annotations
 import json
 import os
 import shutil
 import subprocess
 import tempfile
 import unittest
 from unittest import mock
 import author_issue_bootstrap
 import task_capability_map
 def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict:
    repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args
    os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir
    return author_issue_bootstrap.bootstrap_author_issue_worktree(
        issue_number=issue_num,
        canonical_repo_root=repo_dir,
        expected_base_sha=master_sha,
        idempotency_key=key,
        lock_dir=lock_dir,
        owner_session="session-concurrent-test",
        active_identity="jcwalker3",
        active_profile="prgs-author",
    )
 class TestAuthorIssueBootstrap(unittest.TestCase):
    """Test suite covering AC1-AC10 and comment #14959 specification."""
    def setUp(self):
        self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_")
        self.repo_dir = os.path.join(self.tmp_dir, "repo")
        os.makedirs(self.repo_dir)
        # Initialize synthetic git repo
        subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True)
        subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True)
        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo_dir, check=True)
        readme = os.path.join(self.repo_dir, "README.md")
        with open(readme, "w", encoding="utf-8") as f:
            f.write("# Test Repo\n")
        subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True)
        subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True)
        rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True)
        self.master_sha = rev_res.stdout.strip()
        self.branches_dir = os.path.join(self.repo_dir, "branches")
        os.makedirs(self.branches_dir, exist_ok=True)
        self.lock_dir = os.path.join(self.tmp_dir, "locks")
        os.makedirs(self.lock_dir, exist_ok=True)
        self.journal_dir = os.path.join(self.tmp_dir, "journals")
        os.makedirs(self.journal_dir, exist_ok=True)
        os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir
    def tearDown(self):
        os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None)
        shutil.rmtree(self.tmp_dir, ignore_errors=True)
    def test_bootstrap_success_path(self):
        """AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof."""
        key = "test_key_success_1"
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            assignment_id="asn-12345",
            lease_id="lease-67890",
            expected_base_sha=self.master_sha,
            idempotency_key=key,
            remote="prgs",
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res.get("success"), f"Bootstrap failed: {res}")
        self.assertFalse(res.get("replayed"))
        self.assertEqual(res.get("issue_number"), 850)
        self.assertEqual(res.get("base_sha"), self.master_sha)
        self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path"))
        # Verify worktree directory exists and is registered
        worktree_path = res["worktree_path"]
        self.assertTrue(os.path.isdir(worktree_path))
        wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True)
        self.assertIn(worktree_path, wt_list.stdout)
        # Verify phase journal written
        journal = author_issue_bootstrap.load_phase_journal(key, journal_dir=self.lock_dir)
        self.assertIsNotNone(journal)
        self.assertTrue(journal.get("completed"))
        self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED)
    def test_idempotent_replay(self):
        """Item 2: Replaying with identical key returns cached transition without duplicate creation."""
        key = "test_key_idempotent_1"
        res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res1["success"], f"res1 failed: {res1}")
        self.assertFalse(res1.get("replayed"))
        # Second call
        res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res2["success"], f"res2 failed: {res2}")
        self.assertTrue(res2.get("replayed"))
        self.assertEqual(res1["worktree_path"], res2["worktree_path"])
    def test_stale_concurrency_pin_refusal(self):
        """Item 3: Mismatched expected base SHA fails closed without silent rebasing."""
        stale_sha = "0000000000000000000000000000000000000000"
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            expected_base_sha=stale_sha,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
        self.assertIn("exact_next_action", res)
    def test_path_outside_branches_root_refusal(self):
        """Item 6: Worktree path outside branches/ root is refused."""
        outside_path = os.path.join(self.tmp_dir, "outside_worktree")
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            worktree_path=outside_path,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root")
    def test_preexisting_dirty_worktree_preservation(self):
        """Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned."""
        branch = "fix/issue-850-dirty-test"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
        # Create dirty untracked file
        dirty_file = os.path.join(wt_path, "dirty.txt")
        with open(dirty_file, "w") as f:
            f.write("dirty edits\n")
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name=branch,
            worktree_path=wt_path,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree")
        # Prove dirty file is preserved byte-for-byte
        self.assertTrue(os.path.exists(dirty_file))
        with open(dirty_file, "r") as f:
            self.assertEqual(f.read(), "dirty edits\n")
    def test_compensating_recovery_on_failed_phase(self):
        """AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts."""
        key = "test_key_recovery_1"
        # Simulate partial progress in journal
        journal = {
            "idempotency_key": key,
            "issue_number": 850,
            "branch_name": "fix/issue-850-recovery-test",
            "worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"),
            "artifacts_created": {
                "branch_created": True,
                "worktree_dir_created": True,
                "worktree_registered": True,
                "lock_created": False,
            },
            "failure_reason": "simulated lock failure",
            "current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED,
            "completed": False,
        }
        # Create the branch and worktree manually to simulate partial state
        subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True)
        # Run compensating recovery
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir)
        self.assertTrue(rec["executed"])
        self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"])
        self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"])
        # Prove worktree directory and branch were rolled back
        self.assertFalse(os.path.exists(journal["worktree_path"]))
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False)
        self.assertNotEqual(branch_check.returncode, 0)
    def test_cross_process_concurrency(self):
        """Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption."""
        import concurrent.futures
        key = "test_concurrent_key_850"
        args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha)
        with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
            fut1 = executor.submit(_concurrent_bootstrap_worker, args)
            fut2 = executor.submit(_concurrent_bootstrap_worker, args)
            res1 = fut1.result(timeout=10)
            res2 = fut2.result(timeout=10)
        self.assertTrue(res1["success"], f"res1 failed: {res1}")
        self.assertTrue(res2["success"], f"res2 failed: {res2}")
        # One process performs creation, the other process receives idempotent replay
        replayed_count = sum(1 for r in (res1, res2) if r.get("replayed"))
        created_count = sum(1 for r in (res1, res2) if not r.get("replayed"))
        self.assertEqual(replayed_count, 1)
        self.assertEqual(created_count, 1)
        self.assertEqual(res1["worktree_path"], res2["worktree_path"])
    def test_interrupted_replay_preserves_artifacts_created_provenance(self):
        """Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically."""
        key = "test_key_interrupted_replay_1"
        branch = "fix/issue-850-interrupted-replay"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay")
        # Simulate Phase 2/3 completion where branch and worktree directory were created by this transition
        journal = {
            "idempotency_key": key,
            "issue_number": 850,
            "branch_name": branch,
            "worktree_path": wt_path,
            "active_identity": "jcwalker3",
            "active_profile": "prgs-author",
            "remote": "prgs",
            "org": "Scaled-Tech-Consulting",
            "repo": "Gitea-Tools",
            "phases": {
                author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"},
                author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True},
            },
            "artifacts_created": {
                "branch_created": True,
                "worktree_dir_created": True,
                "worktree_registered": True,
                "lock_created": False,
            },
            "current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED,
            "completed": False,
        }
        # Pre-create the branch and worktree on disk to simulate partial state after crash
        subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
        author_issue_bootstrap.save_phase_journal(journal, journal_dir=self.lock_dir)
        # Now resume/replay the transition but simulate lock binding failure during Phase 6
        with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")):
            res = author_issue_bootstrap.bootstrap_author_issue_worktree(
                issue_number=850,
                canonical_repo_root=self.repo_dir,
                branch_name=branch,
                worktree_path=wt_path,
                idempotency_key=key,
                lock_dir=self.lock_dir,
                owner_session="session-test-1234",
            )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed")
        # Verify that compensating recovery correctly deleted transition-created branch & worktree
        # because creation provenance was preserved across replay (NOT downgraded to False!)
        self.assertFalse(os.path.exists(wt_path))
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
        self.assertNotEqual(branch_check.returncode, 0)
    def test_transition_created_only_compensation(self):
        """Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created."""
        key = "test_key_preexisting_branch_compensation"
        preexisting_branch = "fix/issue-850-preexisting"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting")
        # Create branch BEFORE bootstrap (preexisting branch)
        subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True)
        # Call bootstrap with simulated failure during Phase 6 (lock binding)
        with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")):
            res = author_issue_bootstrap.bootstrap_author_issue_worktree(
                issue_number=850,
                canonical_repo_root=self.repo_dir,
                branch_name=preexisting_branch,
                worktree_path=wt_path,
                idempotency_key=key,
                lock_dir=self.lock_dir,
                owner_session="session-test-1234",
            )
        self.assertFalse(res["success"])
        # Worktree dir was created by transition -> removed by compensation
        self.assertFalse(os.path.exists(wt_path))
        # Preexisting branch was NOT created by transition -> MUST BE PRESERVED!
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False)
        self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!")
    def test_incompatible_idempotency_replay_refusal(self):
        """Review #525 Finding 4: Replaying key with incompatible parameters returns refusal."""
        key = "test_key_incompatible_replay"
        res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name="fix/issue-850-param-a",
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertTrue(res1["success"])
        # Second call with different branch_name
        res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            branch_name="fix/issue-850-param-b",
            idempotency_key=key,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        self.assertFalse(res2["success"])
        self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay")
    def test_exact_next_action_satisfiable_via_mcp(self):
        """Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands."""
        key = "test_key_next_action_mcp"
        stale_sha = "0000000000000000000000000000000000000000"
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            expected_base_sha=stale_sha,
            lock_dir=self.lock_dir,
            owner_session="session-test-1234",
        )
        next_action = res.get("exact_next_action", "")
        self.assertNotIn("scripts/worktree-start", next_action)
        self.assertNotIn("git worktree add", next_action)
        self.assertNotIn("bash", next_action.lower())
    def test_missing_owner_session_refusal(self):
        """Finding D: Missing owner_session context fails closed with typed refusal and zero mutation."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session=None,
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_owner_session")
        self.assertIn("exact_next_action", res)
    def test_symlink_lock_file_refusal(self):
        """Finding C: BootstrapTransitionLock refuses to follow symlinks."""
        key = "test_symlink_lock_key"
        safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key)
        lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock")
        target_file = os.path.join(self.tmp_dir, "fake_target")
        with open(target_file, "w") as f:
            f.write("target")
        os.symlink(target_file, lock_path)
        with self.assertRaises(RuntimeError) as ctx:
            with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir):
                pass
        self.assertIn("symlink", str(ctx.exception).lower())
    def test_lock_directory_escape_refusal(self):
        """Finding C: BootstrapTransitionLock refuses keys that escape lock directory."""
        with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"):
            with self.assertRaises(RuntimeError) as ctx:
                author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir)
        self.assertIn("escapes", str(ctx.exception).lower())
    def test_missing_active_identity_refusal(self):
        """F-5: Missing active_identity parameter fails closed."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session="session-test-1234",
            active_identity=None,
            active_profile="prgs-author",
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_active_identity")
    def test_missing_active_profile_refusal(self):
        """F-5: Missing active_profile parameter fails closed."""
        res = author_issue_bootstrap.bootstrap_author_issue_worktree(
            issue_number=850,
            canonical_repo_root=self.repo_dir,
            owner_session="session-test-1234",
            active_identity="jcwalker3",
            active_profile=None,
            lock_dir=self.lock_dir,
        )
        self.assertFalse(res["success"])
        self.assertEqual(res.get("reason_code"), "missing_active_profile")
    def test_dirty_worktree_preserved_during_recovery(self):
        """F-4: Compensating recovery does not delete dirty worktree."""
        branch = "fix/issue-850-rec-dirty"
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
        dirty_file = os.path.join(wt_path, "dirty.txt")
        with open(dirty_file, "w") as f:
            f.write("uncommitted work")
        journal = {
            "idempotency_key": "test_dirty_rec",
            "issue_number": 850,
            "branch_name": branch,
            "worktree_path": wt_path,
            "artifacts_created": {
                "worktree_dir_created": True,
                "worktree_registered": True,
            },
            "failure_reason": "test dirty recovery",
        }
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
        self.assertTrue(os.path.exists(wt_path))
        self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"])
    def test_branch_with_commits_preserved_during_recovery(self):
        """F-4: Compensating recovery does not delete branch with author commits."""
        branch = "fix/issue-850-rec-commits"
        subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
        # Add a commit on the branch
        wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits")
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
        cfile = os.path.join(wt_path, "commit.txt")
        with open(cfile, "w") as f:
            f.write("author commit")
        subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True)
        subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True)
        subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True)
        journal = {
            "idempotency_key": "test_commits_rec",
            "issue_number": 850,
            "branch_name": branch,
            "resolved_base_sha": self.master_sha,
            "artifacts_created": {
                "branch_created": True,
            },
            "failure_reason": "test commit branch recovery",
        }
        rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
        branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
        self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!")
        self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"])
    def test_task_capability_map_integration(self):
        """Verify task_capability_map has bootstrap_author_issue_worktree configured correctly."""
        self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author")
        self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create")
        self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree"))
        self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue"))
 if __name__ == "__main__":
    unittest.main()
@@ -28,21 +28,6 @@ class TestPathUnderBranches(unittest.TestCase):
            amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
        )
    def test_unrelated_branches_dir_fails(self):
        self.assertFalse(
            amw.is_path_under_branches("/tmp/branches/evil", self.ROOT)
        )
    def test_prefix_confusion_fails(self):
        self.assertFalse(
            amw.is_path_under_branches(f"{self.ROOT}/branches-other/foo", self.ROOT)
        )
    def test_traversal_fails(self):
        self.assertFalse(
            amw.is_path_under_branches(f"{self.ROOT}/branches/../evil", self.ROOT)
        )
 class TestAssessAuthorMutationWorktree(unittest.TestCase):
    ROOT = "/repo/Gitea-Tools"
@@ -1266,378 +1266,6 @@ class TestSecondRemediationIntegration(unittest.TestCase):
        self.assertIn("delete_acknowledged", delete_actions[0])
        self.assertTrue(delete_actions[0].get("verified_absent"))
    def test_issue_851_worktree_removed_when_remote_blocked_only_by_worktree_binding(self):
        """#851: remote blocked by worktree_binding must not skip safe worktree removal.
        Lifecycle: remove clean owned worktree → reassess ownership → delete
        remote only if independently safe. Unrelated entries stay untouched.
        """
        from mcp_server import gitea_reconcile_merged_cleanups
        target_branch = "fix/issue-844-exclude-epic-containers"
        foreign_branch = "fix/issue-999-unrelated-active"
        worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
        ownership_calls = []
        remove_calls = []
        delete_api_calls = []
        def fake_collect(**kwargs):
            ownership_calls.append(dict(kwargs))
            # Ownership is reassessed *after* independent worktree removal (#851).
            # Target worktree is already gone → no worktree_binding remains.
            # Foreign branch keeps an active author lease → remote delete blocked.
            if kwargs.get("branch") == foreign_branch:
                # Match session-bound org/repo + host used by the tool resolve path.
                return {
                    "records": [
                        {
                            "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
                            "status": "active",
                            "remote": kwargs.get("remote") or "prgs",
                            "host": kwargs.get("host") or "gitea.example.com",
                            "org": kwargs.get("org") or "Scaled-Tech-Consulting",
                            "repo": kwargs.get("repo") or "Gitea-Tools",
                            "branch": foreign_branch,
                            "reclaim_allowed": False,
                        }
                    ],
                    "inventory_error": False,
                }
            return {"records": [], "inventory_error": False}
        def fake_remove(project_root, branch, worktree_path=None):
            remove_calls.append(
                {"branch": branch, "worktree_path": worktree_path}
            )
            return {
                "success": True,
                "performed": True,
                "message": f"removed worktree {worktree_path}",
                "worktree_path": worktree_path,
            }
        def fake_probe(h, o, r, auth, br):
            return guard.classify_branch_readback_http_status(
                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
            )
        def fake_api(method, url, auth, **kwargs):
            if method == "DELETE":
                delete_api_calls.append(url)
            return {}
        report = {
            "entries": [
                {
                    "pr_number": 848,
                    "head_branch": target_branch,
                    "remote_branch": {"safe_to_delete_remote": True},
                    "local_worktree": {
                        "safe_to_remove_worktree": True,
                        "worktree_path": worktree_path,
                    },
                },
                {
                    "pr_number": 999,
                    "head_branch": foreign_branch,
                    "remote_branch": {"safe_to_delete_remote": True},
                    "local_worktree": {
                        "safe_to_remove_worktree": False,
                        "worktree_path": None,
                    },
                },
            ],
            "reviewer_scratch_entries": [],
        }
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "prgs-reconciler",
                "role": "reconciler",
                "allowed_operations": [
                    "gitea.read",
                    "gitea.branch.delete",
                    "gitea.pr.close",
                ],
                "forbidden_operations": [],
            },
        ).start()
        patch("mcp_server.api_get_all", return_value=[]).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            return_value=report,
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
            return_value=[],
        ).start()
        patch(
            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
            return_value=(True, []),
        ).start()
        patch("mcp_server.verify_preflight_purity", return_value=None).start()
        patch(
            "mcp_server._collect_branch_ownership_records",
            side_effect=fake_collect,
        ).start()
        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
            side_effect=fake_remove,
        ).start()
        self.mock_api.side_effect = fake_api
        res = gitea_reconcile_merged_cleanups(
            dry_run=False,
            execute_confirmed=True,
            remote="prgs",
        )
        self.assertTrue(res.get("performed") or res.get("executed"))
        actions = res.get("actions") or []
        remove_actions = [
            a for a in actions if a.get("action") == "remove_local_worktree"
        ]
        self.assertEqual(len(remove_actions), 1, actions)
        self.assertTrue(remove_actions[0].get("success"))
        self.assertEqual(remove_calls[0]["branch"], target_branch)
        self.assertEqual(remove_calls[0]["worktree_path"], worktree_path)
        # Target remote delete succeeds after worktree removal + reassessment.
        target_deletes = [
            a
            for a in actions
            if a.get("action") == "delete_remote_branch"
            and a.get("branch") == target_branch
        ]
        self.assertEqual(len(target_deletes), 1, actions)
        self.assertTrue(target_deletes[0].get("success"))
        self.assertTrue(target_deletes[0].get("after_worktree_removal"))
        self.assertTrue(target_deletes[0].get("ownership_reassessed"))
        self.assertTrue(target_deletes[0].get("verified_absent"))
        # Foreign branch remains protected (author lease) and is not deleted.
        foreign_deletes = [
            a
            for a in actions
            if a.get("action") == "delete_remote_branch"
            and a.get("branch") == foreign_branch
        ]
        self.assertEqual(len(foreign_deletes), 1, actions)
        self.assertFalse(foreign_deletes[0].get("success"))
        self.assertEqual(
            foreign_deletes[0].get("blocker_kind"), "active_branch_ownership"
        )
        self.assertIn(
            guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
            foreign_deletes[0].get("blocking_categories") or [],
        )
        # Only the target branch should hit the DELETE API.
        self.assertEqual(len(delete_api_calls), 1)
        # Ownership collected for target (post-removal) and foreign; worktree
        # removal happened before target remote delete in the action log.
        target_idx = next(
            i
            for i, a in enumerate(actions)
            if a.get("action") == "remove_local_worktree"
        )
        delete_idx = next(
            i
            for i, a in enumerate(actions)
            if a.get("action") == "delete_remote_branch"
            and a.get("branch") == target_branch
            and a.get("success")
        )
        self.assertLess(target_idx, delete_idx)
    def test_issue_851_dirty_worktree_not_removed_and_remote_stays_protected(self):
        """#851: dirty/foreign worktrees remain protected; no unsafe cleanup."""
        from mcp_server import gitea_reconcile_merged_cleanups
        branch = "fix/issue-851-dirty"
        remove_calls = []
        def fake_collect(**kwargs):
            return {
                "records": [
                    {
                        "category": guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
                        "status": "active",
                        "remote": kwargs.get("remote") or "prgs",
                        "host": kwargs.get("host") or "gitea.example.com",
                        "org": kwargs.get("org") or "Scaled-Tech-Consulting",
                        "repo": kwargs.get("repo") or "Gitea-Tools",
                        "branch": branch,
                        "reclaim_allowed": False,
                    }
                ],
                "inventory_error": False,
            }
        report = {
            "entries": [
                {
                    "pr_number": 851,
                    "head_branch": branch,
                    "remote_branch": {"safe_to_delete_remote": True},
                    "local_worktree": {
                        "safe_to_remove_worktree": False,
                        "worktree_path": "/tmp/dirty-wt",
                    },
                }
            ],
            "reviewer_scratch_entries": [],
        }
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "prgs-reconciler",
                "role": "reconciler",
                "allowed_operations": [
                    "gitea.read",
                    "gitea.branch.delete",
                ],
                "forbidden_operations": [],
            },
        ).start()
        patch("mcp_server.api_get_all", return_value=[]).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            return_value=report,
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
            return_value=[],
        ).start()
        patch(
            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
            return_value=(True, []),
        ).start()
        patch("mcp_server.verify_preflight_purity", return_value=None).start()
        patch(
            "mcp_server._collect_branch_ownership_records",
            side_effect=fake_collect,
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
            side_effect=lambda *a, **k: remove_calls.append(k) or {
                "success": True,
                "performed": True,
            },
        ).start()
        self.mock_api.side_effect = lambda *a, **k: {}
        res = gitea_reconcile_merged_cleanups(
            dry_run=False,
            execute_confirmed=True,
            remote="prgs",
        )
        actions = res.get("actions") or []
        self.assertEqual(remove_calls, [])
        self.assertFalse(
            any(a.get("action") == "remove_local_worktree" for a in actions)
        )
        deletes = [
            a for a in actions if a.get("action") == "delete_remote_branch"
        ]
        self.assertEqual(len(deletes), 1)
        self.assertFalse(deletes[0].get("success"))
        self.assertEqual(deletes[0].get("blocker_kind"), "active_branch_ownership")
        self.assertIn(
            guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
            deletes[0].get("blocking_categories") or [],
        )
    def test_issue_851_idempotent_resume_when_worktree_already_absent(self):
        """#851: partial failures remain resumable and idempotent."""
        from mcp_server import gitea_reconcile_merged_cleanups
        branch = "fix/issue-851-resume"
        ownership_calls = []
        def fake_collect(**kwargs):
            ownership_calls.append(kwargs)
            return {"records": [], "inventory_error": False}
        def fake_remove(project_root, branch, worktree_path=None):
            return {
                "success": False,
                "performed": False,
                "message": f"worktree not found: {worktree_path}",
            }
        def fake_probe(h, o, r, auth, br):
            return guard.classify_branch_readback_http_status(
                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
            )
        report = {
            "entries": [
                {
                    "pr_number": 851,
                    "head_branch": branch,
                    "remote_branch": {"safe_to_delete_remote": True},
                    "local_worktree": {
                        "safe_to_remove_worktree": True,
                        "worktree_path": "/tmp/already-gone",
                    },
                }
            ],
            "reviewer_scratch_entries": [],
        }
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "prgs-reconciler",
                "role": "reconciler",
                "allowed_operations": [
                    "gitea.read",
                    "gitea.branch.delete",
                ],
                "forbidden_operations": [],
            },
        ).start()
        patch("mcp_server.api_get_all", return_value=[]).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            return_value=report,
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
            return_value=[],
        ).start()
        patch(
            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
            return_value=(True, []),
        ).start()
        patch("mcp_server.verify_preflight_purity", return_value=None).start()
        patch(
            "mcp_server._collect_branch_ownership_records",
            side_effect=fake_collect,
        ).start()
        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.remove_local_worktree",
            side_effect=fake_remove,
        ).start()
        self.mock_api.side_effect = lambda *a, **k: {}
        res = gitea_reconcile_merged_cleanups(
            dry_run=False,
            execute_confirmed=True,
            remote="prgs",
        )
        actions = res.get("actions") or []
        removes = [a for a in actions if a.get("action") == "remove_local_worktree"]
        deletes = [a for a in actions if a.get("action") == "delete_remote_branch"]
        self.assertEqual(len(removes), 1)
        self.assertFalse(removes[0].get("success"))
        self.assertEqual(len(deletes), 1)
        self.assertTrue(deletes[0].get("success"))
        self.assertTrue(deletes[0].get("after_worktree_removal"))
        self.assertTrue(ownership_calls)
 if __name__ == "__main__":
@@ -12,59 +12,6 @@ import merged_cleanup_reconcile as mcr  # noqa: E402
 class TestMergedCleanupAssessment(unittest.TestCase):
    def test_issue_851_plan_order_worktree_then_reassess_then_remote(self):
        """#851 dry-run plan: remove worktree, reassess ownership, then remote."""
        plan = mcr.plan_cleanup_execution_order(
            remote_assessment={"safe_to_delete_remote": True},
            local_assessment={"safe_to_remove_worktree": True},
        )
        actions = [s["action"] for s in plan]
        self.assertEqual(
            actions,
            [
                "remove_local_worktree",
                "reassess_branch_ownership",
                "delete_remote_branch",
            ],
        )
        self.assertEqual(plan[0]["phase"], 1)
        self.assertEqual(plan[-1]["phase"], 3)
        self.assertIn("independently_safe", plan[0]["reason"])
        self.assertIn("reassessment", plan[-1]["reason"])
    def test_issue_851_plan_remote_only_when_worktree_not_safe(self):
        plan = mcr.plan_cleanup_execution_order(
            remote_assessment={"safe_to_delete_remote": True},
            local_assessment={"safe_to_remove_worktree": False},
        )
        self.assertEqual([s["action"] for s in plan], ["delete_remote_branch"])
        self.assertNotIn("reassess_branch_ownership", [s["action"] for s in plan])
    def test_issue_851_plan_worktree_only_when_remote_not_safe(self):
        plan = mcr.plan_cleanup_execution_order(
            remote_assessment={"safe_to_delete_remote": False},
            local_assessment={"safe_to_remove_worktree": True},
        )
        self.assertEqual([s["action"] for s in plan], ["remove_local_worktree"])
    def test_issue_851_entry_includes_planned_execution_order(self):
        entry = mcr.build_pr_cleanup_entry(
            pr={
                "number": 848,
                "title": "Closes #844",
                "body": "",
                "merged_at": "2026-07-23T00:00:00Z",
                "head": {"ref": "fix/issue-844-x", "sha": "a" * 40},
            },
            project_root="/tmp/not-a-real-root",
            open_pr_heads=set(),
            remote_branch_exists=True,
            head_on_master=True,
            delete_capability_allowed=True,
        )
        self.assertIn("planned_execution_order", entry)
        self.assertIsInstance(entry["planned_execution_order"], list)
    def test_extract_linked_issue_from_closes(self):
        issue = mcr.extract_linked_issue(
            "feat: cleanup (Closes #269)",
@@ -139,8 +139,6 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
        "gitea_release_merger_pr_lease",
        "create_branch",
        "push_branch",
        "bootstrap_author_issue_worktree",
        "gitea_bootstrap_author_issue_worktree",
        # #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",
@@ -0,0 +1,222 @@
 """Tests for the read-only workflow policy/guardrail visibility view (#646).
 Covers issue #646 acceptance criteria:
 1. Console lists major guardrails with source pointers.
 2. Secrets redacted.
 3. Tests ensure sample secrets never appear.
 4. Docs explain read-only nature (asserted here for the page copy; the doc
   itself is covered by inspection).
 """
 import json
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from starlette.testclient import TestClient
 from webui import console_redaction
 from webui import policy_inventory
 from webui.app import create_app
 from webui.policy_inventory import (
    PolicyEntry,
    PolicyInventorySnapshot,
    SourcePointer,
    load_policy_inventory,
    snapshot_to_dict,
 )
 from webui.policy_views import render_policy_page
 def _entry(key, category, *, active=None, error=None):
    return PolicyEntry(
        key=key,
        title=key.replace("_", " ").title(),
        category=category,
        summary=f"summary for {key}",
        sources=(SourcePointer("src", f"{key}.py", "module"),),
        active=active,
        documented_default=None,
        diff=None,
        error=error,
    )
 def _snapshot(entries):
    return PolicyInventorySnapshot(
        schema_version=1,
        read_only=True,
        note="read-only projection",
        entries=tuple(entries),
        categories=tuple(dict.fromkeys(e.category for e in entries)),
        build_errors=(),
    )
 # The guardrail categories issue #646 names as in-scope.
 _EXPECTED_CATEGORIES = {
    "role_separation",
    "lease_rules",
    "worktree_rules",
    "merge_confirmation",
    "redaction",
    "contamination",
    "allocator_policy",
    "audit_logging",
    "mutation_gating",
 }
 class TestPolicyInventoryModel(unittest.TestCase):
    def test_major_guardrails_present(self):
        snapshot = load_policy_inventory()
        categories = {e.category for e in snapshot.entries}
        self.assertEqual(_EXPECTED_CATEGORIES, categories)
        self.assertGreaterEqual(len(snapshot.entries), len(_EXPECTED_CATEGORIES))
    def test_every_guardrail_has_source_pointers(self):
        # AC1: source attribution (file/module/doc) for every guardrail.
        snapshot = load_policy_inventory()
        for entry in snapshot.entries:
            with self.subTest(entry=entry.key):
                self.assertTrue(entry.sources, "guardrail must carry source pointers")
                for source in entry.sources:
                    self.assertTrue(source.path)
                    self.assertIn(source.kind, {"module", "doc", "script", "config"})
    def test_diff_reported_where_documented_default_declared(self):
        snapshot = load_policy_inventory()
        checked_any = False
        for entry in snapshot.entries:
            if entry.documented_default is None:
                self.assertIsNone(entry.diff)
                continue
            checked_any = True
            self.assertIsNotNone(entry.diff)
            self.assertEqual(
                entry.diff["status"],
                "matches_documented_default",
                f"{entry.key} drifted from its documented default: {entry.diff}",
            )
        self.assertTrue(checked_any, "at least one guardrail should declare a default")
    def test_live_projections_populate_active(self):
        snapshot = load_policy_inventory()
        by_key = {e.key: e for e in snapshot.entries}
        for key in ("role_separation", "redaction", "audit_logging"):
            self.assertIsNone(by_key[key].error, f"{key} projection failed")
            self.assertIsInstance(by_key[key].active, dict)
    def test_build_entry_is_fail_soft_on_projection_error(self):
        def _boom():
            raise RuntimeError("projection exploded")
        row = (
            "redaction",
            "Secret redaction",
            "redaction",
            "summary",
            (SourcePointer("x", "webui/console_redaction.py", "module"),),
            _boom,
            {"redact_before_persist": True},
        )
        entry = policy_inventory._build_entry(row)
        self.assertIsNone(entry.active)
        self.assertIsNotNone(entry.error)
        self.assertEqual(entry.diff["status"], "active_unavailable")
 class TestPolicyRedaction(unittest.TestCase):
    def test_real_snapshot_has_no_secret_shapes(self):
        # AC3: the real emitted payload never carries a known secret shape.
        payload = snapshot_to_dict(load_policy_inventory())
        self.assertEqual(console_redaction.scan_for_secrets(payload), [])
    def test_planted_keychain_secret_is_redacted(self):
        # AC2/AC3: a secret planted in an active projection is masked before emit.
        snapshot = _snapshot([
            _entry(
                "redaction",
                "redaction",
                active={"leaked": "keychain:prgs-author-super-secret", "roles": ["author"]},
            )
        ])
        payload = snapshot_to_dict(snapshot)
        blob = json.dumps(payload)
        self.assertNotIn("keychain:prgs-author-super-secret", blob)
        self.assertEqual(console_redaction.scan_for_secrets(payload), [])
    def test_planted_credential_assignment_is_redacted(self):
        snapshot = _snapshot([
            _entry(
                "audit_logging",
                "audit_logging",
                active={"leaked": "token=abcd1234efgh5678", "append_only": True},
            )
        ])
        payload = snapshot_to_dict(snapshot)
        blob = json.dumps(payload)
        self.assertNotIn("abcd1234efgh5678", blob)
        self.assertEqual(console_redaction.scan_for_secrets(payload), [])
 class TestPolicyRoutes(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(create_app())
    def test_policy_html_lists_guardrails_with_sources(self):
        response = self.client.get("/policy")
        self.assertEqual(response.status_code, 200)
        text = response.text
        self.assertIn("Workflow policy", text)
        self.assertIn("Role separation and RBAC", text)
        self.assertIn("Source pointers", text)
        self.assertIn("task_capability_map.py", text)
        self.assertIn("docs/safety-model.md", text)
    def test_policy_html_states_read_only(self):
        # AC4: the page explains its read-only nature.
        text = self.client.get("/policy").text
        self.assertIn("read-only", text.lower())
        self.assertNotIn("<form", text.lower())
    def test_policy_html_has_no_secret_shapes(self):
        text = self.client.get("/policy").text
        self.assertEqual(console_redaction.scan_for_secrets(text), [])
    def test_api_v1_policy_returns_inventory(self):
        response = self.client.get("/api/v1/policy")
        self.assertEqual(response.status_code, 200)
        data = response.json()
        self.assertEqual(data["schema_version"], policy_inventory.SCHEMA_VERSION)
        self.assertTrue(data["read_only"])
        self.assertEqual(data["entry_count"], len(data["entries"]))
        self.assertEqual(set(data["categories"]), _EXPECTED_CATEGORIES)
    def test_policy_is_read_only_no_post(self):
        # AC4 / non-goal: no mutation endpoint.
        response = self.client.post("/policy")
        self.assertIn(response.status_code, (404, 405))
    def test_nav_links_policy(self):
        text = self.client.get("/").text
        self.assertIn('href="/policy"', text)
 class TestPolicyViewFailSoft(unittest.TestCase):
    def test_page_renders_when_a_projection_errors(self):
        snapshot = _snapshot([
            _entry("role_separation", "role_separation", error="active projection unavailable: boom"),
            _entry("redaction", "redaction", active={"redact_before_persist": True}),
        ])
        page = render_policy_page(snapshot)
        # The errored guardrail surfaces its error; other guardrails still render.
        self.assertIn("Active value unavailable", page)
        self.assertIn("Redaction", page)
        self.assertIn("Workflow policy", page)
 if __name__ == "__main__":
    unittest.main()
@@ -1,135 +0,0 @@
 """Tests for the Phase 1 operator console application shell (#638)."""
 import sys
 import unittest
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from starlette.routing import Route
 from starlette.testclient import TestClient
 from webui import layout
 from webui.app import create_app
 from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
 class TestShellNav(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(create_app())
    def test_nav_group_labels_present(self):
        text = self.client.get("/").text
        for group in NAV_GROUPS:
            with self.subTest(group=group.label):
                self.assertIn(f">{group.label}<", text)
    def test_phase1_group_labels_cover_expected_ia(self):
        labels = {group.label for group in NAV_GROUPS}
        for expected in (
            "Health",
            "Traffic",
            "Runtime/Sessions",
            "Projects",
            "Inventory",
            "Timeline",
            "Policy",
            "Insights",
        ):
            with self.subTest(label=expected):
                self.assertIn(expected, labels)
    def test_every_nav_href_resolves_to_a_get_route(self):
        app = create_app()
        get_paths = {
            route.path
            for route in app.routes
            if isinstance(route, Route) and "GET" in route.methods
        }
        for href in nav_hrefs():
            with self.subTest(href=href):
                self.assertIn(href, get_paths, f"nav href {href} has no GET route")
    def test_legacy_hrefs_still_navigable(self):
        text = self.client.get("/").text
        for href in ("/queue", "/projects", "/prompts", "/runtime",
                     "/audit", "/worktrees", "/leases", "/actions"):
            with self.subTest(href=href):
                self.assertIn(f'href="{href}"', text)
 class TestShellBadges(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(create_app())
    def test_mode_badge_present(self):
        self.assertIn("mode: read-only", self.client.get("/").text)
    def test_environment_badge_present(self):
        self.assertIn("env:", self.client.get("/").text)
    def test_default_environment_is_local(self):
        self.assertEqual(layout.environment_label(), "local")
    def test_remote_bind_reports_remote_environment(self):
        import os
        prior = os.environ.get("WEBUI_HOST")
        os.environ["WEBUI_HOST"] = "10.0.0.5"
        try:
            self.assertEqual(layout.environment_label(), "remote")
        finally:
            if prior is None:
                os.environ.pop("WEBUI_HOST", None)
            else:
                os.environ["WEBUI_HOST"] = prior
    def test_docs_link_present(self):
        text = self.client.get("/").text
        self.assertIn(layout.DOCS_URL, text)
        self.assertIn(">Docs<", text)
 class TestShellStubs(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(create_app())
    def test_stub_routes_render_200(self):
        for path, (title, _desc) in STUB_PAGES.items():
            with self.subTest(path=path):
                response = self.client.get(path)
                self.assertEqual(response.status_code, 200, path)
                self.assertIn(title, response.text)
                self.assertIn("placeholder", response.text)
    def test_stub_routes_are_read_only(self):
        for path in STUB_PAGES:
            with self.subTest(path=path):
                response = self.client.post(path)
                self.assertEqual(response.status_code, 405)
                self.assertEqual(response.json()["error"], "read-only-mvp")
    def test_stub_pages_carry_nav_and_badges(self):
        response = self.client.get("/inventory")
        self.assertIn("mode: read-only", response.text)
        self.assertIn('href="/queue"', response.text)
 class TestShellHome(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(create_app())
    def test_home_summarizes_console(self):
        text = self.client.get("/").text
        self.assertIn("Operator console", text)
        self.assertIn("Phase 1", text)
    def test_home_links_legacy_pages(self):
        text = self.client.get("/").text
        self.assertIn("MVP legacy pages", text)
        for href in ("/queue", "/audit", "/leases"):
            with self.subTest(href=href):
                self.assertIn(f'href="{href}"', text)
 if __name__ == "__main__":
    unittest.main()
@@ -12,7 +12,6 @@ from starlette.routing import Route
 from webui.deployment_boundary import deployment_snapshot
 from webui.layout import render_page
 from webui.nav import NAV_GROUPS, STUB_PAGES
 from webui.project_registry import (
    ProjectRegistry,
    RegistryError,
@@ -46,7 +45,8 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
-from webui.timeline import load_timeline, snapshot_to_dict as timeline_snapshot_to_dict
+from webui.policy_inventory import load_policy_inventory, snapshot_to_dict as policy_snapshot_to_dict
 from webui.policy_views import render_policy_page
 from webui.system_health import (
    API_PATH as SYSTEM_HEALTH_API_PATH,
    load_system_health,
@@ -67,62 +67,25 @@ def _stub_page(title: str, description: str) -> HTMLResponse:
    return HTMLResponse(render_page(title=title, body_html=body))
 _LEGACY_PAGES = (
    ("/queue", "Queue", "live PR and issue dashboard (#429)"),
    ("/projects", "Projects", "registry and onboarding (#427)"),
    ("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
    ("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
    ("/audit", "Audit", "final-report paste and validator preview (#431)"),
    ("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
    ("/leases", "Leases", "collision and lease visibility (#433)"),
    ("/actions", "Actions", "gated write-action framework (#434)"),
 )
 def _render_home_nav_groups() -> str:
    groups = []
    for group in NAV_GROUPS:
        items = "".join(
            f'<li><a href="{item.href}">{item.label}</a>'
            + ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
            + "</li>"
            for item in group.items
        )
        groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
    return "".join(groups)
 async def home(_request: Request) -> HTMLResponse:
    legacy = "".join(
        f"<li><strong>{label}</strong> — {desc} "
        f'(<a href="{href}">{href}</a>)</li>'
        for href, label, desc in _LEGACY_PAGES
    )
    body = (
        "<h2>Operator console</h2>"
-        "<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
+        "<p>Local entry point for MCP Control Plane operational views.</p>"
-        "Gitea, MCP capability gates, and canonical workflows remain the source "
+        "<ul>"
-        "of truth; this console never mutates them.</p>"
+        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
-        "<h2>Phase 1 surfaces</h2>"
+        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
-        + _render_home_nav_groups()
+        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
-        + "<h2>MVP legacy pages</h2>"
+        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
-        "<ul>" + legacy + "</ul>"
+        "<li><strong>Policy</strong> — workflow guardrail configuration visibility (#646)</li>"
        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
        "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
        "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
        "</ul>"
    )
    return HTMLResponse(render_page(title="Home", body_html=body))
 async def phase_stub(request: Request) -> HTMLResponse:
    """Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
    title, description = STUB_PAGES[request.url.path]
    body = (
        f"<h2>{title}</h2>"
        f'<div class="stub"><p>{description}</p>'
        "<p>Phase 1 shell placeholder — no write actions. Tracked under "
        "epic #631.</p></div>"
    )
    return HTMLResponse(render_page(title=title, body_html=body))
 async def health(_request: Request) -> JSONResponse:
    """Liveness only — deliberately cheap, runs no dependency probe (#634).
@@ -283,6 +246,17 @@ async def api_runtime(_request: Request) -> JSONResponse:
    return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
 async def policy(_request: Request) -> HTMLResponse:
    snapshot = load_policy_inventory()
    return HTMLResponse(
        render_page(title="Policy", body_html=render_policy_page(snapshot))
    )
 async def api_v1_policy(_request: Request) -> JSONResponse:
    return JSONResponse(policy_snapshot_to_dict(load_policy_inventory()))
 async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
    if request.method == "GET":
        return "", None
@@ -450,104 +424,6 @@ async def api_console_security_model(_request: Request) -> JSONResponse:
    })
 def _query_int(request: Request, key: str) -> int | None:
    """Parse an optional integer query parameter; None when absent/invalid."""
    raw = request.query_params.get(key)
    if raw is None or not str(raw).strip():
        return None
    try:
        return int(str(raw).strip())
    except (TypeError, ValueError):
        return None
 def _derive_remote(host: str) -> str:
    """Map a Gitea host to its known short remote name (control-plane scope key)."""
    text = (host or "").lower()
    if "prgs" in text:
        return "prgs"
    if "dadeschools" in text:
        return "dadeschools"
    return text.split(".")[0] if text else ""
 def _timeline_comment_source(host: str, org: str, repo: str):
    """Build a fail-soft CTH-comment fetcher for one repo, or None when offline.
    Returns a callable ``(kind, number) -> list[comment]``. Credentials or
    network failures raise inside the callable so ``load_timeline`` degrades the
    handoff source rather than the whole timeline. Offline test mode yields no
    live source so the handoff section reports ``not run``.
    """
    import os
    from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
    offline = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {"1", "true", "yes"}
    if offline:
        return None
    auth = get_auth_header(host)
    if not auth:
        return None
    def _fetch(kind: str, number: int) -> list:
        segment = "pulls" if kind == "pr" else "issues"
        url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
        comments: list = []
        page = 1
        while page <= 20:
            raw, meta = api_fetch_page(url, auth, page=page, limit=50)
            comments.extend(raw)
            if bool(meta["is_final_page"]):
                break
            page += 1
        return comments
    return _fetch
 async def api_v1_timeline(request: Request) -> JSONResponse:
    """Read-only workflow-event timeline (#637). Filter by issue/PR/session."""
    from webui.queue_loader import _host_from_url  # host normalisation helper
    registry, error = _load_project_registry()
    if error is not None:
        return JSONResponse(error.to_dict(), status_code=500)
    project = registry.projects[0] if registry.projects else None
    org = request.query_params.get("org") or (project.gitea_owner if project else "")
    repo = request.query_params.get("repo") or (project.repo_name if project else "")
    host = _host_from_url(project.remote_host) if project else ""
    remote = request.query_params.get("remote") or _derive_remote(host)
    if not (remote and org and repo):
        return JSONResponse(
            {
                "error": "timeline_scope_unresolved",
                "detail": "no project in registry and no remote/org/repo query params provided",
            },
            status_code=400,
        )
    comment_source = _timeline_comment_source(host, org, repo) if (host and org and repo) else None
    snapshot = load_timeline(
        remote=remote,
        org=org,
        repo=repo,
        issue=_query_int(request, "issue"),
        pr=_query_int(request, "pr"),
        session=(request.query_params.get("session") or None),
        limit=_query_int(request, "limit"),
        offset=_query_int(request, "offset"),
        comment_source=comment_source,
    )
    # A filter no surviving source can carry is refused, not answered empty:
    # a 200 with zero events would tell the operator no such activity exists.
    status_code = 200 if snapshot.ok else 422
    return JSONResponse(timeline_snapshot_to_dict(snapshot), status_code=status_code)
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
    path = request.url.path
    if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -587,7 +463,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
            Route("/api/prompts", api_prompts, methods=["GET"]),
            Route("/runtime", runtime, methods=["GET"]),
            Route("/api/runtime", api_runtime, methods=["GET"]),
-            Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
+            Route("/policy", policy, methods=["GET"]),
            Route("/api/v1/policy", api_v1_policy, methods=["GET"]),
            Route("/audit", audit, methods=["GET", "POST"]),
            Route("/api/audit", api_audit, methods=["GET", "POST"]),
            Route("/worktrees", worktrees, methods=["GET"]),
@@ -611,10 +488,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                api_console_security_model,
                methods=["GET"],
            ),
            *[
                Route(path, phase_stub, methods=["GET"])
                for path in STUB_PAGES
            ],
        ],
        exception_handlers={405: method_not_allowed},
    )
@@ -2,66 +2,29 @@
 from __future__ import annotations
-import os
+NAV_ITEMS = (
-
+    ("/", "Home"),
-from webui.nav import NAV_GROUPS
+    ("/queue", "Queue"),
    ("/projects", "Projects"),
    ("/prompts", "Prompts"),
    ("/runtime", "Runtime"),
    ("/policy", "Policy"),
    ("/audit", "Audit"),
    ("/worktrees", "Worktrees"),
    ("/leases", "Leases"),
    ("/actions", "Actions"),
 )
 MVP_NOTICE = (
    "Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
    "source of truth. No mutation endpoints."
 )
 # Canonical docs entry point surfaced from the shell header (#638).
 DOCS_URL = (
    "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/src/branch/"
    "master/docs/webui-local-dev.md"
 )
 _LOCAL_HOSTS = frozenset({"", "127.0.0.1", "localhost", "::1"})
 def environment_label() -> str:
    """Classify the serving environment as ``local`` or ``remote`` (#638).
    Derived from the same ``WEBUI_HOST`` default the app binds to; loopback
    hosts are ``local``, anything else is ``remote``. Read-only signal only.
    """
    host = (os.environ.get("WEBUI_HOST", "127.0.0.1") or "").strip().lower()
    return "local" if host in _LOCAL_HOSTS else "remote"
 def _render_nav() -> str:
    groups_html = []
    for group in NAV_GROUPS:
        links = "".join(
            f'<a href="{item.href}"'
            + (' class="nav-stub"' if item.status == "stub" else "")
            + f'>{item.label}</a>'
            for item in group.items
        )
        groups_html.append(
            '<div class="nav-group">'
            f'<span class="nav-group-label">{group.label}</span>'
            f'<span class="nav-group-links">{links}</span>'
            "</div>"
        )
    return "".join(groups_html)
 def _render_badges() -> str:
    env = environment_label()
    return (
        '<div class="header-badges">'
        f'<span class="badge env-badge env-{env}">env: {env}</span>'
        '<span class="badge mode-badge">mode: read-only</span>'
        f'<a class="badge docs-link" href="{DOCS_URL}">Docs</a>'
        "</div>"
    )
 def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
-    nav_links = _render_nav()
+    nav_links = "".join(
-    header_badges = _render_badges()
+        f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
    )
    return f"""<!DOCTYPE html>
 <html lang="en">
 <head>
@@ -91,58 +54,21 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
      padding: 0.75rem 1.25rem;
    }}
    header h1 {{
-      margin: 0;
+      margin: 0 0 0.5rem;
      font-size: 1.1rem;
      font-weight: 600;
    }}
    .header-top {{
      display: flex;
      flex-wrap: wrap;
      align-items: center;
      justify-content: space-between;
      gap: 0.5rem 1rem;
      margin-bottom: 0.6rem;
    }}
    .header-badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }}
    .env-badge.env-local {{ color: #8fd19e; border-color: #3d6b4a; }}
    .env-badge.env-remote {{ color: #e0c27a; border-color: #6b5730; }}
    .mode-badge {{ color: #9ec8f0; border-color: #3d5f7a; }}
    a.docs-link {{
      color: var(--accent);
      border-color: var(--accent);
      text-decoration: none;
      text-transform: none;
    }}
    a.docs-link:hover {{ filter: brightness(1.12); }}
    nav {{
      display: flex;
      flex-wrap: wrap;
-      gap: 0.5rem 1.25rem;
+      gap: 0.75rem 1rem;
    }}
    .nav-group {{
      display: flex;
      flex-direction: column;
      gap: 0.15rem;
    }}
    .nav-group-label {{
      font-size: 0.68rem;
      text-transform: uppercase;
      letter-spacing: 0.04em;
      color: var(--muted);
    }}
    .nav-group-links {{ display: inline-flex; flex-wrap: wrap; gap: 0.6rem; }}
    nav a {{
      color: var(--accent);
      text-decoration: none;
      font-size: 0.9rem;
    }}
    nav a:hover {{ text-decoration: underline; }}
    nav a.nav-stub {{ color: var(--muted); }}
    nav a.nav-stub::after {{
      content: " ·stub";
      font-size: 0.7rem;
      color: var(--muted);
    }}
    main {{
      max-width: 52rem;
      margin: 0 auto;
@@ -241,10 +167,7 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
 </head>
 <body>
  <header>
-    <div class="header-top">
+    <h1>MCP Control Plane</h1>
      <h1>MCP Control Plane</h1>
      {header_badges}
    </div>
    <nav>{nav_links}</nav>
  </header>
  <main>
@@ -1,111 +0,0 @@
 """Navigation IA for the Phase 1 operator console shell (#638).
 Single source of truth for the console navigation so ``webui/layout.py`` and
 the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
 destination is a GET view or a Phase 1 placeholder. No mutation links.
 Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
 Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
 Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
 backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
 instead of a 404.
 """
 from __future__ import annotations
 from dataclasses import dataclass
@dataclass(frozen=True)
 class NavItem:
    """A single navigation destination.
    ``status`` is ``"live"`` for implemented views and ``"stub"`` for Phase 1
    placeholders whose backing view lands in a later child issue.
    """
    href: str
    label: str
    status: str = "live"
@dataclass(frozen=True)
 class NavGroup:
    label: str
    items: tuple[NavItem, ...]
 NAV_GROUPS: tuple[NavGroup, ...] = (
    NavGroup("Health", (
        NavItem("/health", "Liveness"),
    )),
    NavGroup("Traffic", (
        NavItem("/queue", "Queue"),
        NavItem("/leases", "Leases"),
        NavItem("/actions", "Actions"),
    )),
    NavGroup("Runtime/Sessions", (
        NavItem("/runtime", "Runtime health"),
        NavItem("/sessions", "Sessions", "stub"),
    )),
    NavGroup("Projects", (
        NavItem("/projects", "Projects"),
    )),
    NavGroup("Inventory", (
        NavItem("/inventory", "Inventory", "stub"),
        NavItem("/worktrees", "Worktrees"),
    )),
    NavGroup("Timeline", (
        NavItem("/timeline", "Timeline", "stub"),
    )),
    NavGroup("Policy", (
        NavItem("/policy", "Policy", "stub"),
        NavItem("/prompts", "Prompts"),
    )),
    NavGroup("Insights", (
        NavItem("/insights", "Insights", "stub"),
        NavItem("/audit", "Audit"),
    )),
 )
 # Phase 1 placeholder destinations whose backing views land in later child
 # issues of epic #631. Each maps a path to (title, description). Routes are
 # registered so nav links resolve to a graceful, read-only stub page.
 STUB_PAGES: dict[str, tuple[str, str]] = {
    "/sessions": (
        "Sessions",
        "Active session, capability, and role inventory. Backed by the unified "
        "inventory API (#636) once it lands.",
    ),
    "/inventory": (
        "Inventory",
        "Unified sessions, leases, locks, namespaces, and worktree inventory. "
        "Backed by the Phase 1 inventory API (#636).",
    ),
    "/timeline": (
        "Timeline",
        "Workflow event timeline across issues and PRs. A later Phase 1 surface.",
    ),
    "/policy": (
        "Policy",
        "Capability and role policy surface. Placeholder until a later phase.",
    ),
    "/insights": (
        "Insights",
        "Aggregate operational insights and trends. Placeholder until a later "
        "phase.",
    ),
 }
 def iter_nav_items():
    """Yield every ``NavItem`` across all groups in declared order."""
    for group in NAV_GROUPS:
        for item in group.items:
            yield item
 def nav_hrefs() -> tuple[str, ...]:
    """Return every navigation href in declared order."""
    return tuple(item.href for item in iter_nav_items())
@@ -0,0 +1,387 @@
 """Read-only workflow policy and guardrail inventory for the web UI (#646).
 Policy and guardrails live in code, profiles, docs, and skills. An operator
 cannot *see* the active workflow policy configuration from the console without
 reading the repository tree. This module projects the major guardrails into a
 redacted, machine-readable inventory with source attribution (file / module /
 doc), so the console can render them as HTML tables with source pointers.
 Design constraints (Phase 3, #646):
 - **Read-only projection.** Nothing here edits policy or exposes a toggle that
  could weaken a gate. It reports what is already enforced elsewhere.
 - **Source attribution without secrets.** Every guardrail carries pointers to
  the file/module/doc that owns it. Live values are compact summaries derived
  from the safe policy accessors that already exist (``rbac_matrix``,
  ``redaction_policy``, ``audit_policy``); raw regex, tokens, and endpoints are
  never embedded.
 - **Redact before emit.** ``snapshot_to_dict`` runs the whole payload through
  ``console_redaction.redact_payload`` so a planted or accidental secret in any
  projected value degrades to the placeholder rather than reaching a client.
 - **Fail soft.** A projection that raises is recorded as a per-entry error and
  never takes the page down; a guardrail is still listed with its sources.
 - **Diff vs documented defaults where feasible.** When a guardrail declares a
  documented invariant, the active projection is compared against it and the
  result is reported; otherwise the diff is explicitly ``None`` with a reason.
 """
 from __future__ import annotations
 from dataclasses import dataclass
 from typing import Any, Callable
 from webui import console_audit
 from webui import console_authz
 from webui import console_redaction
 SCHEMA_VERSION = 1
 READ_ONLY_NOTE = (
    "Read-only projection of guardrails enforced in code, profiles, docs, and "
    "skills. This view never edits policy and exposes no gate-weakening toggle."
 )
@dataclass(frozen=True)
 class SourcePointer:
    """Where a guardrail is defined. Attribution only — never a secret."""
    label: str
    path: str
    kind: str  # "module" | "doc" | "script" | "config"
    anchor: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return {
            "label": self.label,
            "path": self.path,
            "kind": self.kind,
            "anchor": self.anchor,
        }
@dataclass(frozen=True)
 class PolicyEntry:
    key: str
    title: str
    category: str
    summary: str
    sources: tuple[SourcePointer, ...]
    active: dict[str, Any] | None
    documented_default: dict[str, Any] | None
    diff: dict[str, Any] | None
    error: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "category": self.category,
            "summary": self.summary,
            "sources": [s.to_dict() for s in self.sources],
            "active": self.active,
            "documented_default": self.documented_default,
            "diff": self.diff,
            "error": self.error,
        }
@dataclass(frozen=True)
 class PolicyInventorySnapshot:
    schema_version: int
    read_only: bool
    note: str
    entries: tuple[PolicyEntry, ...]
    categories: tuple[str, ...]
    build_errors: tuple[str, ...]
 def _diff_active_vs_default(
    active: dict[str, Any] | None,
    documented_default: dict[str, Any] | None,
 ) -> dict[str, Any] | None:
    """Compare only the keys the documented default declares.
    Returns ``None`` when no documented default is declared (diff not feasible)
    or when the active projection is unavailable. Otherwise reports, per
    declared key, whether the active value matches the documented invariant.
    """
    if not documented_default:
        return None
    if not active:
        return {"status": "active_unavailable", "checked": {}}
    checked: dict[str, Any] = {}
    matches = True
    for key, expected in documented_default.items():
        observed = active.get(key)
        ok = observed == expected
        matches = matches and ok
        checked[key] = {"expected": expected, "observed": observed, "matches": ok}
    return {
        "status": "matches_documented_default" if matches else "drift_detected",
        "checked": checked,
    }
 # ── Live projections (compact, safe, fail-soft) ──────────────────────────────
 # Each returns a small dict of already-safe machine values. They are module
 # level so tests can substitute one to prove the redaction pass runs.
 def _project_role_separation() -> dict[str, Any]:
    matrix = console_authz.rbac_matrix()
    return {
        "model_version": matrix.get("model_version"),
        "active_phase": matrix.get("active_phase"),
        "roles": [r.get("role") for r in matrix.get("roles", [])],
        "privileged_action_count": len(matrix.get("privileged_actions", [])),
        "default_decision": matrix.get("default_decision"),
        "execution_enabled": matrix.get("execution_enabled"),
    }
 def _project_redaction() -> dict[str, Any]:
    policy = console_redaction.redaction_policy()
    return {
        "policy_version": policy.get("policy_version"),
        "placeholder": policy.get("placeholder"),
        "applies_to": policy.get("applies_to"),
        "console_detector_count": len(policy.get("console_rules", [])),
        "redact_before_persist": policy.get("redact_before_persist"),
        "failure_mode": policy.get("failure_mode"),
    }
 def _project_audit() -> dict[str, Any]:
    policy = console_audit.audit_policy()
    return {
        "schema_version": policy.get("schema_version"),
        "required_field_count": len(policy.get("required_fields", [])),
        "results": policy.get("results"),
        "retention_defaults_days": policy.get("retention_defaults_days"),
        "append_only": policy.get("append_only"),
        "redact_before_persist": policy.get("redact_before_persist"),
        "enabled": policy.get("enabled"),
    }
 def _static(value: dict[str, Any]) -> Callable[[], dict[str, Any]]:
    return lambda: dict(value)
 # ── Guardrail catalog ────────────────────────────────────────────────────────
 # One row per major guardrail. ``project`` yields the active value (may raise;
 # caught per entry). ``documented_default`` drives the feasible diff.
 _CatalogRow = tuple[
    str,
    str,
    str,
    str,
    tuple[SourcePointer, ...],
    Callable[[], dict[str, Any]] | None,
    dict[str, Any] | None,
 ]
 _CATALOG: tuple[_CatalogRow, ...] = (
    (
        "role_separation",
        "Role separation and RBAC",
        "role_separation",
        "Author, reviewer, merger, and reconciler capabilities are disjoint and "
        "role-exclusive; self-review and self-merge are always blocked. The "
        "console RBAC model defaults to deny.",
        (
            SourcePointer("task capability map", "task_capability_map.py", "module"),
            SourcePointer("role/namespace gate", "role_namespace_gate.py", "module"),
            SourcePointer("console RBAC", "webui/console_authz.py", "module"),
        ),
        _project_role_separation,
        {"default_decision": "deny", "execution_enabled": False},
    ),
    (
        "lease_rules",
        "Issue and PR lease lifecycle",
        "lease_rules",
        "Durable work is claimed through issue locks and control-plane leases "
        "with freshness, expiry, and dead-session recovery; abandoned or stale "
        "claims are reclaimed only through the sanctioned recovery path.",
        (
            SourcePointer("issue lock store", "issue_lock_store.py", "module"),
            SourcePointer("branch cleanup guard", "branch_cleanup_guard.py", "module"),
            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
        ),
        None,
        None,
    ),
    (
        "worktree_rules",
        "Author worktree binding",
        "worktree_rules",
        "Author mutations require a validated worktree under branches/ derived "
        "from the active issue lock; silent fallback to the stable control "
        "checkout or master is forbidden (#618).",
        (
            SourcePointer("author worktree gate", "author_mutation_worktree.py", "module"),
            SourcePointer("worktree bootstrap", "scripts/worktree-start", "script"),
            SourcePointer("workflow scope guard", "workflow_scope_guard.py", "module"),
        ),
        None,
        None,
    ),
    (
        "merge_confirmation",
        "Explicit merge confirmation",
        "merge_confirmation",
        "A merge fails closed unless the caller passes the exact confirmation "
        "phrase for that PR; reviewing never implies merging.",
        (
            SourcePointer("merge path", "merge_pr.py", "module"),
            SourcePointer("merge tool gate", "gitea_mcp_server.py", "module"),
        ),
        _static({"required_confirmation_format": "MERGE PR <n>", "auto_merge": False}),
        {"auto_merge": False},
    ),
    (
        "redaction",
        "Secret redaction",
        "redaction",
        "Every console surface runs the shared gitea_audit pass then console "
        "patterns before any payload, HTML, log line, or audit record leaves "
        "the server; unredactable values fail closed to the placeholder.",
        (
            SourcePointer("console redaction", "webui/console_redaction.py", "module"),
            SourcePointer("shared redaction", "gitea_audit.py", "module"),
            SourcePointer("safety model §3", "docs/safety-model.md", "doc", "3-secret-redaction"),
        ),
        _project_redaction,
        {"redact_before_persist": True},
    ),
    (
        "contamination",
        "Contamination containment",
        "contamination",
        "A session contaminated by a direct stable-branch push or a manual MCP "
        "daemon kill is blocked from review, merge, close, and completion "
        "mutations until cleared (reconciler-exempt).",
        (
            SourcePointer("contamination gates", "gitea_mcp_server.py", "module"),
            SourcePointer("stable-branch audit", "workflow_scope_guard.py", "module"),
        ),
        None,
        None,
    ),
    (
        "allocator_policy",
        "Work allocation policy",
        "allocator_policy",
        "Workers do not self-select exclusive work; the controller-owned "
        "allocator ranks the complete queue by priority then PRs-before-issues "
        "then ascending number, honoring dependency edges and foreign claims.",
        (
            SourcePointer("allocator", "gitea_mcp_server.py", "module"),
            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
        ),
        _static(
            {
                "self_select_exclusive_work": False,
                "ranking": "priority desc, PRs before issues, number asc",
                "respects_dependency_edges": True,
                "respects_foreign_claims": True,
            }
        ),
        {"self_select_exclusive_work": False},
    ),
    (
        "audit_logging",
        "Audit logging",
        "audit_logging",
        "Console intent and authorization outcomes are recorded to an "
        "append-only, redact-before-persist audit log; MCP mutations are "
        "recorded by gitea_audit and correlated by request id.",
        (
            SourcePointer("console audit", "webui/console_audit.py", "module"),
            SourcePointer("MCP audit", "gitea_audit.py", "module"),
            SourcePointer("safety model §1", "docs/safety-model.md", "doc", "1-audit-logging-and-confirmation"),
        ),
        _project_audit,
        {"append_only": True, "redact_before_persist": True},
    ),
    (
        "mutation_gating",
        "Mutation gating and master parity",
        "mutation_gating",
        "Mutations fail closed while the running server is stale relative to "
        "master, and every mutation is preceded by identity and capability "
        "resolution in a fixed pre-flight order.",
        (
            SourcePointer("mutation gate", "gitea_mcp_server.py", "module"),
            SourcePointer("safety model §5", "docs/safety-model.md", "doc", "5-mutation-gating"),
        ),
        _static(
            {
                "stale_runtime_blocks_mutations": True,
                "preflight_order": "whoami -> resolve_task_capability -> mutation",
            }
        ),
        {"stale_runtime_blocks_mutations": True},
    ),
 )
 def _build_entry(row: _CatalogRow) -> PolicyEntry:
    key, title, category, summary, sources, project, documented_default = row
    active: dict[str, Any] | None = None
    error: str | None = None
    if project is not None:
        try:
            active = project()
        except Exception as exc:  # noqa: BLE001 — fail soft; never take the page down
            active = None
            error = f"active projection unavailable: {exc}"
    diff = _diff_active_vs_default(active, documented_default)
    return PolicyEntry(
        key=key,
        title=title,
        category=category,
        summary=summary,
        sources=sources,
        active=active,
        documented_default=documented_default,
        diff=diff,
        error=error,
    )
 def load_policy_inventory() -> PolicyInventorySnapshot:
    """Build the read-only guardrail inventory. Never raises for one bad entry."""
    entries: list[PolicyEntry] = []
    build_errors: list[str] = []
    for row in _CATALOG:
        try:
            entries.append(_build_entry(row))
        except Exception as exc:  # noqa: BLE001 — one row must not break the rest
            build_errors.append(f"{row[0]}: {exc}")
    categories = tuple(dict.fromkeys(e.category for e in entries))
    return PolicyInventorySnapshot(
        schema_version=SCHEMA_VERSION,
        read_only=True,
        note=READ_ONLY_NOTE,
        entries=tuple(entries),
        categories=categories,
        build_errors=tuple(build_errors),
    )
 def snapshot_to_dict(snapshot: PolicyInventorySnapshot) -> dict[str, Any]:
    """Serialize the snapshot, redacting the entire payload before it is emitted."""
    payload = {
        "schema_version": snapshot.schema_version,
        "read_only": snapshot.read_only,
        "note": snapshot.note,
        "categories": list(snapshot.categories),
        "entry_count": len(snapshot.entries),
        "entries": [entry.to_dict() for entry in snapshot.entries],
        "build_errors": list(snapshot.build_errors),
    }
    return console_redaction.redact_payload(payload)
@@ -0,0 +1,104 @@
 """HTML views for the workflow policy and guardrail inventory (#646)."""
 from __future__ import annotations
 import html
 import json
 from webui.policy_inventory import PolicyEntry, PolicyInventorySnapshot
 def _source_pointer(source) -> str:
    path = source.path
    if source.anchor:
        path = f"{path}#{source.anchor}"
    return (
        f"<li>{html.escape(source.label)} — "
        f"<code>{html.escape(path)}</code> "
        f"<span class='muted'>({html.escape(source.kind)})</span></li>"
    )
 def _active_block(entry: PolicyEntry) -> str:
    if entry.error:
        return (
            "<p class='muted'><strong>Active value unavailable:</strong> "
            f"{html.escape(entry.error)}</p>"
        )
    if not entry.active:
        return "<p class='muted'>No live projection for this guardrail.</p>"
    pretty = json.dumps(entry.active, indent=2, sort_keys=True, default=str)
    return f"<pre class='prompt-text'>{html.escape(pretty)}</pre>"
 def _diff_block(entry: PolicyEntry) -> str:
    if entry.diff is None:
        if entry.documented_default is None:
            return "<p class='muted'>Diff vs documented default: not feasible (no declared default).</p>"
        return "<p class='muted'>Diff vs documented default: unavailable.</p>"
    status = entry.diff.get("status", "unknown")
    badge = "badge-claimed" if status == "matches_documented_default" else "badge-blocked"
    rows = []
    for key, cell in (entry.diff.get("checked") or {}).items():
        marker = "✓" if cell.get("matches") else "✗"
        rows.append(
            "<tr>"
            f"<td><code>{html.escape(str(key))}</code></td>"
            f"<td><code>{html.escape(str(cell.get('expected')))}</code></td>"
            f"<td><code>{html.escape(str(cell.get('observed')))}</code></td>"
            f"<td>{marker}</td>"
            "</tr>"
        )
    table = ""
    if rows:
        table = (
            "<table class='detail'><thead><tr>"
            "<th>Key</th><th>Documented</th><th>Active</th><th>Match</th>"
            "</tr></thead><tbody>"
            f"{''.join(rows)}</tbody></table>"
        )
    return (
        f"<p class='meta'>Diff vs documented default: "
        f"<span class='badge {badge}'>{html.escape(status)}</span></p>"
        f"{table}"
    )
 def _entry_card(entry: PolicyEntry) -> str:
    sources = "".join(_source_pointer(s) for s in entry.sources)
    return (
        "<div class='prompt-card'>"
        f"<h3>{html.escape(entry.title)} "
        f"<span class='badge'>{html.escape(entry.category)}</span></h3>"
        f"<p>{html.escape(entry.summary)}</p>"
        "<p class='meta'><strong>Source pointers</strong></p>"
        f"<ul>{sources}</ul>"
        "<p class='meta'><strong>Active configuration</strong></p>"
        f"{_active_block(entry)}"
        f"{_diff_block(entry)}"
        "</div>"
    )
 def render_policy_page(snapshot: PolicyInventorySnapshot) -> str:
    categories = ", ".join(html.escape(c) for c in snapshot.categories) or "none"
    cards = "".join(_entry_card(e) for e in snapshot.entries)
    build_errors = ""
    if snapshot.build_errors:
        items = "".join(
            f"<li>{html.escape(err)}</li>" for err in snapshot.build_errors
        )
        build_errors = (
            "<div class='stub'><p><strong>Some guardrails could not be built:"
            f"</strong></p><ul>{items}</ul></div>"
        )
    return (
        "<h2>Workflow policy &amp; guardrails</h2>"
        f"<p class='muted'>{html.escape(snapshot.note)}</p>"
        f"<p class='meta'>Schema v{snapshot.schema_version} · "
        f"{len(snapshot.entries)} guardrails · categories: {categories}</p>"
        f"{build_errors}"
        f"{cards}"
        "<p class='muted'>This page is read-only. It reports enforced policy "
        "and never edits or weakens a gate. Secret values are redacted.</p>"
    )
@@ -1,906 +0,0 @@
 """Workflow-event and conversation timeline model (#637, Phase 1).
 Operators cannot browse a unified timeline of workflow events, decisions,
 tool calls, and handoffs: the evidence is scattered across control-plane
 events, Gitea canonical handoff comments, and local logs. This module defines
 one durable, versioned event schema and per-source adapters that normalise
 those scattered records into a single ``WorkflowEvent`` stream, plus a
 read-only query layer (filter by issue / PR / session, stable ordering,
 pagination) that the ``/api/v1/timeline`` route serves.
 Design rules honoured here:
 - **Read-only.** Sources are read; nothing is mutated. The control-plane
  database is opened through a ``mode=ro`` URI so a missing or unwritable DB
  degrades to a reason instead of creating directories or running migrations.
 - **Fail-soft per source.** An unavailable source degrades to a status with a
  reason rather than raising, and a source that could not run is never
  rendered as an empty-and-healthy timeline.
 - **Answerable filters only.** Each source declares which filter dimensions it
  can actually answer. A filter dimension no source that ran can carry is
  refused with an explicit reason rather than silently matching nothing: an
  empty page from an unanswerable filter reads to an operator as "no such
  activity", which is a different — and false — statement.
 - **Redaction at the boundary, fail closed.** Every free-text field (event
  messages, redacted tool arguments, decision/proof text) is run through the
  console redaction policy before it leaves this module, and *before* any
  structured value is derived from it — evidence references are extracted from
  redacted text, then independently revalidated before serialization. An
  unredactable value becomes the placeholder, and a value that cannot be proven
  safe is dropped — an unredacted payload is never emitted, and a generation
  error never drops raw data to a caller or a log.
 - **Stable ordering.** Events sort by ``(timestamp, source_rank, event_key)``
  with a deterministic tiebreak, so pagination is stable across calls and
  events with equal or missing timestamps keep a fixed order.
 Non-goals (from the issue): no full chat replay, no mutation of historical
 events, no unredacted tool-argument storage.
 """
 from __future__ import annotations
 import re
 import sqlite3
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from typing import Any, Callable, Iterable
 import control_plane_db
 from webui import console_redaction
 # The schema is versioned so consumers can branch on shape. Bump on any
 # breaking change to WorkflowEvent's serialized form.
 TIMELINE_SCHEMA_VERSION = 1
 # Known event sources and their deterministic ordering rank. When two events
 # carry the same timestamp, the source rank breaks the tie before the
 # per-source event key, so a control-plane event and a handoff comment minted
 # in the same second always sort in a fixed order.
 SOURCE_CONTROL_PLANE = "control_plane"
 SOURCE_GITEA_HANDOFF = "gitea_handoff"
 _SOURCE_RANK = {
    SOURCE_CONTROL_PLANE: 0,
    SOURCE_GITEA_HANDOFF: 1,
 }
 # The filter dimensions the query layer accepts.
 FILTER_ISSUE = "issue"
 FILTER_PR = "pr"
 FILTER_SESSION = "session"
 # Which dimensions each source can actually answer. This is a property of the
 # underlying records, not of the query code: the control-plane ``events`` table
 # is (event_id, work_item_id, event_type, message, created_at) and carries no
 # session identity at all, so no control-plane event can ever match a session
 # filter. A CTH handoff comment can declare its session as a field, so the
 # handoff source answers all three. Filtering on a dimension the surviving
 # sources cannot carry is refused in ``load_timeline`` rather than answered
 # with an empty page.
 _SOURCE_FILTER_SUPPORT: dict[str, tuple[str, ...]] = {
    SOURCE_CONTROL_PLANE: (FILTER_ISSUE, FILTER_PR),
    SOURCE_GITEA_HANDOFF: (FILTER_ISSUE, FILTER_PR, FILTER_SESSION),
 }
 # Why a source cannot answer a dimension, for the refusal reason an operator reads.
 _SOURCE_FILTER_LIMITS: dict[tuple[str, str], str] = {
    (SOURCE_CONTROL_PLANE, FILTER_SESSION): (
        "control-plane events carry no session identity "
        "(the events table has no session column)"
    ),
 }
 # A timestamp far in the future so events with no parseable timestamp sort
 # last (after everything real) instead of first, without raising.
 _MISSING_TS_SORT = "9999-12-31T23:59:59Z"
 def _parse_ts(value: str | None) -> str | None:
    """Normalise a timestamp to ``...Z`` UTC, or None when unparseable."""
    if not value:
        return None
    text = str(value).strip()
    if not text:
        return None
    candidate = text[:-1] + "+00:00" if text.endswith("Z") else text
    try:
        parsed = datetime.fromisoformat(candidate)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
 def _redact(value: Any) -> Any:
    """Redact a single free-text field, failing closed to the placeholder."""
    if value is None:
        return None
    return console_redaction.redact_text(str(value))
@dataclass(frozen=True)
 class WorkflowEvent:
    """One normalised timeline event.
    Every field is optional except ``source``/``event_type``/``event_key``
    because sources carry different subsets. The class is frozen so an adapted
    event is an immutable record; a consumer that needs a variant builds a new
    one rather than mutating history.
    """
    source: str
    event_type: str
    event_key: str
    timestamp: str | None = None
    actor: str | None = None
    role: str | None = None
    issue_number: int | None = None
    pr_number: int | None = None
    session_id: str | None = None
    tool_name: str | None = None
    decision: str | None = None
    message: str | None = None
    correlation_id: str | None = None
    evidence_refs: tuple[str, ...] = ()
    sensitive: bool = False
    def sort_key(self) -> tuple[str, int, str]:
        return (
            self.timestamp or _MISSING_TS_SORT,
            _SOURCE_RANK.get(self.source, 99),
            self.event_key,
        )
    def to_dict(self) -> dict[str, Any]:
        return {
            "source": self.source,
            "event_type": self.event_type,
            "event_key": self.event_key,
            "timestamp": self.timestamp,
            "actor": self.actor,
            "role": self.role,
            "issue_number": self.issue_number,
            "pr_number": self.pr_number,
            "session_id": self.session_id,
            "tool_name": self.tool_name,
            "decision": self.decision,
            "message": self.message,
            "correlation_id": self.correlation_id,
            "evidence_refs": list(self.evidence_refs),
            "sensitive": self.sensitive,
        }
 # --------------------------------------------------------------------------- #
 # Adapters — pure functions from a source's raw records to WorkflowEvents.     #
 # Each is total: a malformed record is skipped, never raised on.               #
 # --------------------------------------------------------------------------- #
 # Event types whose payload is treated as sensitive and always redaction-hard
 # (they can carry lease/session provenance or tool arguments).
 _SENSITIVE_EVENT_HINTS = ("lease", "capability", "token", "auth", "secret")
 # Reference tokens (issue/PR/comment ids) and SHAs parsed out of proof text.
 _EVIDENCE_REF_RE = re.compile(r"(?:#|PR\s*#?|issue\s*#?|comment\s*#?)(\d+)", re.IGNORECASE)
 # A commit reference is only recognised when the text *declares* it as one.
 # A bare lowercase hex run is not evidence of anything: at 40 characters it is
 # exactly the shape of a Gitea personal access token, and at 7 it also matches
 # ordinary words such as "defaced". Requiring an anchoring keyword keeps real
 # references ("commit abc1234", "at head a209756...", "base caaae9b6") usable
 # while refusing to lift an undeclared secret-shaped run out of free text.
 _SHA_RE = re.compile(
    r"(?i:\b(?:commit|sha|head|base|parent|revision|rev|merge[- ]base)\b[\s:=@#]*)"
    r"([0-9a-f]{7,40})\b"
 )
 # Shapes a serialized evidence reference is allowed to take. Anything else is
 # dropped rather than emitted.
 _REF_ISSUE_SHAPE = re.compile(r"^#[0-9]{1,9}$")
 _REF_SHA_SHAPE = re.compile(r"^[0-9a-f]{7,40}$")
 # A long undelimited hex run with no declaring context is treated as credential
 # material wherever it appears, never as an identifier.
 _BARE_SECRET_SHAPE = re.compile(r"^[0-9a-f]{32,}$")
 # An event type reads like an identifier, but a stored one is externally
 # influenced: any producer that writes the control-plane ``events`` table
 # chooses the string. It reaches ``to_dict`` verbatim, so it is validated here
 # rather than trusted because of where it came from.
 _CP_EVENT_TYPE_SHAPE = re.compile(r"^[A-Za-z][A-Za-z0-9._:+-]{0,63}$")
 # Emitted in place of a value that cannot be proven safe. Deliberately not a
 # plausible workflow type: an unsafe value is refused, never quietly rewritten
 # into a different valid-looking one that would misdescribe the record.
 UNSAFE_EVENT_TYPE = "unsafe:redacted"
 # Emitted for a CTH heading that is not a declared member of ``CTH_TYPES``. The
 # contract is enforced on write (``format_cth_body``) and on assess; the read
 # path the timeline uses enforces it too rather than assuming it was.
 UNKNOWN_HANDOFF_EVENT_TYPE = "handoff:unrecognized"
 # A source record id is a plain integer in both sources it comes from: the
 # control-plane ``events`` primary key and a Gitea comment id. ``event_key`` is
 # serialized verbatim and is the pagination tiebreak, so anything else is
 # refused rather than interpolated into it.
 _RECORD_ID_SHAPE = re.compile(r"^[0-9]{1,19}$")
 def _kind_to_numbers(kind: str | None, number: int | None) -> tuple[int | None, int | None]:
    """Map a control-plane work-item (kind, number) to (issue_no, pr_no)."""
    if number is None:
        return (None, None)
    if kind == "pr":
        return (None, int(number))
    if kind == "issue":
        return (int(number), None)
    return (None, None)
 def _correlation_for(kind: str | None, number: int | None) -> str | None:
    if number is None or kind not in ("issue", "pr"):
        return None
    return f"{kind}#{number}"
 def _extract_evidence_refs(*texts: str | None) -> tuple[str, ...]:
    """Extract issue/PR and declared-commit references from **redacted** text.
    Callers must pass text that has already been through :func:`_redact`; this
    function derives a structured field from its input, so extracting ahead of
    redaction would republish whatever redaction was about to remove. Every
    reference is revalidated by :func:`_validated_evidence_refs` before it is
    serialized.
    """
    refs: list[str] = []
    for text in texts:
        if not text:
            continue
        for match in _EVIDENCE_REF_RE.finditer(text):
            token = f"#{match.group(1)}"
            if token not in refs:
                refs.append(token)
        for match in _SHA_RE.finditer(text):
            token = match.group(1)
            if token not in refs:
                refs.append(token)
    return tuple(refs)
 def _validated_evidence_refs(refs: Iterable[str]) -> tuple[tuple[str, ...], bool]:
    """Independently revalidate references immediately before serialization.
    Extraction is not trusted on its own. A reference survives only when it has
    a known reference shape and is unchanged by a second redaction pass — a
    value the redaction policy would alter is credential material that must not
    be emitted as a structured field. A full 40-character SHA stays usable
    because extraction only accepts a hex run the source text explicitly
    declared as a commit. Returns ``(safe_refs, dropped_any)``; ``dropped_any``
    marks the event sensitive so the drop is visible rather than silent.
    """
    safe: list[str] = []
    dropped = False
    for ref in refs or ():
        try:
            token = str(ref).strip()
            if not token:
                continue
            recognised = bool(_REF_ISSUE_SHAPE.match(token) or _REF_SHA_SHAPE.match(token))
            if not recognised:
                dropped = True
                continue
            if _redact(token) != token:
                dropped = True
                continue
            if token not in safe:
                safe.append(token)
        except Exception:
            # Fail closed: a reference that cannot be proven safe is dropped.
            dropped = True
            continue
    return (tuple(safe), dropped)
 def _safe_session_id(value: Any) -> str | None:
    """Return a session identifier only when it is safe to emit.
    The value is authoritative source data — a session the record names for
    itself — but it is still free text. It is dropped when redaction alters it
    or when it is a bare secret-shaped hex run, so a credential parked in a
    session field can never reach the payload or be echoed back by a filter.
    """
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None
    if _BARE_SECRET_SHAPE.match(text):
        return None
    return text if _redact(text) == text else None
 def _safe_record_id(value: Any) -> str | None:
    """Return a source record id only when it is a plain numeric identifier.
    ``event_key`` is serialized verbatim and is the deterministic pagination
    tiebreak, so an id is interpolated into it only when it has the shape both
    real sources actually produce. A record whose identity cannot be trusted is
    refused by the caller rather than keyed on.
    """
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, int):
        return str(value)
    text = str(value).strip()
    return text if _RECORD_ID_SHAPE.match(text) else None
 def _safe_cp_event_type(value: Any) -> tuple[str, bool]:
    """Validate a stored control-plane event type. Returns ``(type, unsafe)``.
    The stored value is externally influenced — whichever producer wrote the
    ``events`` row chose the string — and ``to_dict`` serializes it verbatim, so
    it passes a boundary of its own instead of relying on the one ``message``
    passes. A value survives only when it is an ordinary identifier, is not a
    bare secret-shaped hex run, and is unchanged by a redaction pass. Anything
    else fails closed to :data:`UNSAFE_EVENT_TYPE`: the record stays visible as
    an audit entry, but the value itself is never republished — not verbatim,
    not partially sanitized, and not rewritten into some other valid-looking
    type that would misdescribe what happened.
    """
    text = ("" if value is None else str(value)).strip()
    if not text:
        return ("", False)
    if _BARE_SECRET_SHAPE.match(text):
        return (UNSAFE_EVENT_TYPE, True)
    if not _CP_EVENT_TYPE_SHAPE.match(text):
        return (UNSAFE_EVENT_TYPE, True)
    if _redact(text) != text:
        return (UNSAFE_EVENT_TYPE, True)
    return (text, False)
 def _safe_echo(value: Any) -> Any:
    """Guard a scalar that is echoed back rather than derived from a record.
    Query scope and filter values are caller-supplied and are reflected in the
    response so an operator can see what was asked. Reflection is still
    emission: a value redaction would alter, or a bare secret-shaped hex run, is
    replaced by the placeholder instead of being echoed verbatim. Ordinary
    scope and filter values pass through untouched.
    """
    if value is None or isinstance(value, (int, bool)):
        return value
    text = str(value)
    if _BARE_SECRET_SHAPE.match(text.strip()):
        return console_redaction.REDACTED
    return _redact(text)
 def adapt_cp_events(rows: Iterable[dict[str, Any]]) -> list[WorkflowEvent]:
    """Adapt control-plane ``events`` rows (joined to work_items) into events.
    Each row is expected to carry ``event_id``, ``event_type``, ``message``,
    ``created_at`` and the joined work-item ``kind``/``number``. Rows missing
    an id or type are skipped so a partially written table never raises.
    """
    events: list[WorkflowEvent] = []
    for row in rows or []:
        try:
            event_id = _safe_record_id(row.get("event_id"))
            raw_event_type = (row.get("event_type") or "").strip()
            if event_id is None or not raw_event_type:
                continue
            # The stored type is source data, not a trusted constant: validate
            # it before it is serialized, exactly as `message` below is redacted
            # before it is serialized.
            event_type, event_type_unsafe = _safe_cp_event_type(raw_event_type)
            kind = row.get("kind")
            number = row.get("number")
            issue_no, pr_no = _kind_to_numbers(kind, number)
            sensitive = event_type_unsafe or any(
                hint in raw_event_type.lower() for hint in _SENSITIVE_EVENT_HINTS
            )
            events.append(
                WorkflowEvent(
                    source=SOURCE_CONTROL_PLANE,
                    event_type=event_type,
                    event_key=f"cp:{event_id}",
                    timestamp=_parse_ts(row.get("created_at")),
                    issue_number=issue_no,
                    pr_number=pr_no,
                    # No session_id: the control-plane events table is
                    # (event_id, work_item_id, event_type, message, created_at)
                    # and records no session. Inventing one from the work item
                    # or the message text would be a guess, so this source
                    # declares the session dimension unsupported instead
                    # (_SOURCE_FILTER_SUPPORT) and the query layer refuses a
                    # session filter it cannot honestly answer.
                    message=_redact(row.get("message")),
                    correlation_id=_correlation_for(kind, number),
                    sensitive=sensitive,
                )
            )
        except Exception:
            # A single malformed row must not sink the whole adaptation.
            continue
    return events
 def adapt_cth_comments(
    comments: Iterable[dict[str, Any]],
    *,
    kind: str,
    number: int,
 ) -> list[WorkflowEvent]:
    """Adapt Gitea Canonical Thread Handoff (CTH) comments into events.
    Only comments that parse as a CTH (``canonical_thread_handoff.parse_cth_comment``)
    become events; ordinary comments are ignored. ``kind``/``number`` scope the
    events to the issue or PR the comments belong to.
    """
    # Imported lazily so this module has no import-time dependency on the
    # handoff parser when only the control-plane adapter is used.
    from canonical_thread_handoff import is_known_cth_type, parse_cth_comment
    # ``kind``/``number`` are interpolated into event_key and correlation_id, so
    # they are normalised once here. A scope this adapter cannot express is
    # refused outright rather than serialized into an identifier.
    kind = (kind or "").strip().lower()
    if kind not in ("issue", "pr"):
        return []
    try:
        number = int(number)
    except (TypeError, ValueError):
        return []
    issue_no, pr_no = _kind_to_numbers(kind, number)
    correlation = _correlation_for(kind, number)
    events: list[WorkflowEvent] = []
    for comment in comments or []:
        try:
            body = comment.get("body") or ""
            parsed = parse_cth_comment(body)
            if not parsed:
                continue
            fields = parsed.get("fields") or {}
            cth_type = parsed.get("cth_type") or ""
            comment_id = _safe_record_id(comment.get("id"))
            if comment_id is None:
                continue
            # The CTH heading is free text: the parser accepts whatever follows
            # "## CTH:", and only the write and assess paths check it against
            # the contract. Check it here too — an unrecognised heading is
            # reported as such rather than serialized into event_type, so
            # arbitrary, malformed, or secret-shaped heading content has no way
            # through. Declared types are preserved exactly.
            cth_type_known = is_known_cth_type(cth_type)
            # Redaction runs first, and every derived value is taken from the
            # redacted text — deriving evidence refs from the raw proof would
            # re-emit exactly what redaction was about to remove.
            decision = _redact(fields.get("decision"))
            proof = _redact(fields.get("proof"))
            next_action = _redact(fields.get("next action"))
            refs, refs_dropped = _validated_evidence_refs(
                _extract_evidence_refs(proof, decision)
            )
            events.append(
                WorkflowEvent(
                    source=SOURCE_GITEA_HANDOFF,
                    event_type=(
                        f"handoff:{cth_type.strip()}"
                        if cth_type_known
                        else UNKNOWN_HANDOFF_EVENT_TYPE
                    ),
                    event_key=f"cth:{kind}:{number}:{comment_id}",
                    timestamp=_parse_ts(comment.get("created_at")),
                    actor=_redact((comment.get("user") or {}).get("login")),
                    role=_redact(fields.get("next owner")),
                    issue_number=issue_no,
                    pr_number=pr_no,
                    # A CTH names its own session when the producer records one;
                    # it is read from that declared field, never inferred from
                    # unrelated text.
                    session_id=_safe_session_id(fields.get("session")),
                    decision=decision,
                    message=next_action or _redact(fields.get("status")),
                    correlation_id=correlation,
                    evidence_refs=refs,
                    sensitive=refs_dropped or not cth_type_known,
                )
            )
        except Exception:
            continue
    return events
 # --------------------------------------------------------------------------- #
 # Read-only control-plane event source.                                       #
 # --------------------------------------------------------------------------- #
 _CP_EVENTS_QUERY = """
 SELECT e.event_id   AS event_id,
       e.event_type AS event_type,
       e.message    AS message,
       e.created_at AS created_at,
       w.kind       AS kind,
       w.number     AS number
 FROM events e
 JOIN work_items w ON e.work_item_id = w.work_item_id
 WHERE w.remote = ? AND w.org = ? AND w.repo = ?
 """
@dataclass(frozen=True)
 class SourceStatus:
    """Fail-soft status for one timeline source.
    ``supported_filters`` states which filter dimensions this source's records
    can carry; ``unsupported_filters`` names the requested dimensions it cannot,
    so an operator can see *why* a source contributed nothing rather than being
    left to read an empty list as an absence of activity.
    """
    name: str
    ok: bool
    reason: str | None = None
    count: int = 0
    supported_filters: tuple[str, ...] = ()
    unsupported_filters: tuple[str, ...] = ()
    def to_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "ok": self.ok,
            "reason": self.reason,
            "count": self.count,
            "supported_filters": list(self.supported_filters),
            "unsupported_filters": list(self.unsupported_filters),
        }
 def _cp_status(*, ok: bool, reason: str | None = None, count: int = 0) -> SourceStatus:
    return SourceStatus(
        SOURCE_CONTROL_PLANE,
        ok=ok,
        # A failure reason is serialized like any other field and is often an
        # exception string carrying a path or a transport error, so it crosses
        # the redaction boundary too. Static reasons pass through unchanged.
        reason=_redact(reason),
        count=count,
        supported_filters=_SOURCE_FILTER_SUPPORT[SOURCE_CONTROL_PLANE],
    )
 def _handoff_status(*, ok: bool, reason: str | None = None, count: int = 0) -> SourceStatus:
    return SourceStatus(
        SOURCE_GITEA_HANDOFF,
        ok=ok,
        # Same boundary as the control-plane status: this reason can quote an
        # error raised by a live authenticated fetch.
        reason=_redact(reason),
        count=count,
        supported_filters=_SOURCE_FILTER_SUPPORT[SOURCE_GITEA_HANDOFF],
    )
 def read_cp_events(
    *,
    remote: str,
    org: str,
    repo: str,
    db_path: str | None = None,
 ) -> tuple[list[WorkflowEvent], SourceStatus]:
    """Read scoped control-plane events read-only. Never creates the DB.
    Opens the SQLite file through a ``mode=ro`` URI: a health/timeline read
    must never create directories or run the schema migration that
    ``ControlPlaneDB()`` performs on construction. A missing or unreadable DB
    degrades to a status with a reason.
    """
    path = (db_path or control_plane_db.default_db_path()).strip()
    conn: sqlite3.Connection | None = None
    try:
        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
        conn.row_factory = sqlite3.Row
        cursor = conn.execute(_CP_EVENTS_QUERY, (remote, org, repo))
        rows = [dict(r) for r in cursor.fetchall()]
    except sqlite3.OperationalError as exc:
        return ([], _cp_status(ok=False, reason=f"control-plane DB unavailable: {exc}"))
    except sqlite3.Error as exc:
        return ([], _cp_status(ok=False, reason=f"control-plane read failed: {exc}"))
    finally:
        if conn is not None:
            conn.close()
    events = adapt_cp_events(rows)
    return (events, _cp_status(ok=True, count=len(events)))
 # --------------------------------------------------------------------------- #
 # Filter, sort, paginate.                                                     #
 # --------------------------------------------------------------------------- #
 def filter_events(
    events: Iterable[WorkflowEvent],
    *,
    issue: int | None = None,
    pr: int | None = None,
    session: str | None = None,
 ) -> list[WorkflowEvent]:
    """Filter events by issue number, PR number, and/or session id.
    Filters are conjunctive. A filter that names a dimension an event does not
    carry excludes that event (an issue filter excludes PR-only events).
    """
    out: list[WorkflowEvent] = []
    for ev in events:
        if issue is not None and ev.issue_number != issue:
            continue
        if pr is not None and ev.pr_number != pr:
            continue
        if session is not None and ev.session_id != session:
            continue
        out.append(ev)
    return out
 def sort_events(events: Iterable[WorkflowEvent]) -> list[WorkflowEvent]:
    """Return events in stable timeline order (ascending)."""
    return sorted(events, key=lambda ev: ev.sort_key())
@dataclass(frozen=True)
 class TimelinePage:
    """One page of the sorted, filtered timeline."""
    events: tuple[WorkflowEvent, ...]
    total: int
    limit: int
    offset: int
    @property
    def next_offset(self) -> int | None:
        nxt = self.offset + len(self.events)
        return nxt if nxt < self.total else None
    def to_dict(self) -> dict[str, Any]:
        return {
            "events": [ev.to_dict() for ev in self.events],
            "pagination": {
                "total": self.total,
                "limit": self.limit,
                "offset": self.offset,
                "returned": len(self.events),
                "next_offset": self.next_offset,
                "has_more": self.next_offset is not None,
            },
        }
 _MAX_LIMIT = 500
 _DEFAULT_LIMIT = 50
 def _coerce_bounds(limit: int | None, offset: int | None) -> tuple[int, int]:
    try:
        lim = int(limit) if limit is not None else _DEFAULT_LIMIT
    except (TypeError, ValueError):
        lim = _DEFAULT_LIMIT
    try:
        off = int(offset) if offset is not None else 0
    except (TypeError, ValueError):
        off = 0
    lim = max(1, min(lim, _MAX_LIMIT))
    off = max(0, off)
    return (lim, off)
 def paginate(events: list[WorkflowEvent], *, limit: int | None, offset: int | None) -> TimelinePage:
    lim, off = _coerce_bounds(limit, offset)
    window = events[off : off + lim]
    return TimelinePage(events=tuple(window), total=len(events), limit=lim, offset=off)
 # --------------------------------------------------------------------------- #
 # Composition — load_timeline aggregates all sources, fail-soft.              #
 # --------------------------------------------------------------------------- #
 # A comment source is a callable that, given (kind, number), returns the raw
 # Gitea comment list for that issue/PR. The route supplies a live fail-soft
 # fetcher; tests supply a fixture. When None, the handoff source is reported as
 # not-run (never silently empty-and-healthy).
 CommentSource = Callable[[str, int], list[dict[str, Any]]]
@dataclass(frozen=True)
 class TimelineSnapshot:
    """One answered timeline query.
    ``ok`` is False when the query could not be answered as asked — currently
    when a requested filter dimension no surviving source can carry was
    supplied. The page is then empty *and* the snapshot says so, because an
    ``ok`` empty page is a claim that no such activity exists.
    """
    schema_version: int
    remote: str
    org: str
    repo: str
    filters: dict[str, Any]
    page: TimelinePage
    sources: tuple[SourceStatus, ...]
    ok: bool = True
    error: dict[str, Any] | None = None
    def to_dict(self) -> dict[str, Any]:
        return {
            "ok": self.ok,
            "error": self.error,
            "schema_version": self.schema_version,
            # Scope and filters are echoed caller input, not derived record
            # data. Reflecting a value is still emitting it, so both cross the
            # same boundary; ordinary scope and filter values are unchanged.
            "scope": {
                "remote": _safe_echo(self.remote),
                "org": _safe_echo(self.org),
                "repo": _safe_echo(self.repo),
            },
            "filters": {key: _safe_echo(value) for key, value in self.filters.items()},
            "sources": [s.to_dict() for s in self.sources],
            **self.page.to_dict(),
        }
 def _unanswerable_reasons(
    statuses: Iterable[SourceStatus], unanswerable: Iterable[str]
 ) -> list[dict[str, str]]:
    """Explain, per source, why each unanswerable dimension went unanswered."""
    out: list[dict[str, str]] = []
    for status in statuses:
        for dim in unanswerable:
            if dim not in status.supported_filters:
                reason = _SOURCE_FILTER_LIMITS.get(
                    (status.name, dim), f"this source's records carry no {dim} identity"
                )
            elif not status.ok:
                reason = (
                    f"this source can carry {dim} but did not run: "
                    f"{status.reason or 'unavailable'}"
                )
            else:
                continue
            out.append({"source": status.name, "filter": dim, "reason": reason})
    return out
 def load_timeline(
    *,
    remote: str,
    org: str,
    repo: str,
    issue: int | None = None,
    pr: int | None = None,
    session: str | None = None,
    limit: int | None = None,
    offset: int | None = None,
    db_path: str | None = None,
    comment_source: CommentSource | None = None,
 ) -> TimelineSnapshot:
    """Aggregate every timeline source into one filtered, paginated snapshot.
    Sources are read independently and fail soft: an unavailable source
    contributes a ``SourceStatus`` with ``ok=False`` and a reason, and never
    collapses the whole timeline. The handoff source only runs when a specific
    issue or PR is requested (a handoff comment belongs to one thread) and a
    ``comment_source`` is available; otherwise it is reported as ``not run``
    rather than as an empty-and-healthy source.
    A filter dimension that no surviving source can carry — a ``session``
    filter when the only source that ran is the control plane, whose events
    record no session — is refused with ``ok=False`` and a structured error
    instead of being answered with an empty page.
    """
    all_events: list[WorkflowEvent] = []
    statuses: list[SourceStatus] = []
    cp_events, cp_status = read_cp_events(remote=remote, org=org, repo=repo, db_path=db_path)
    all_events.extend(cp_events)
    statuses.append(cp_status)
    # Gitea handoff comments are thread-scoped: only fetch when the caller
    # narrowed to one issue or PR, and only when a source was provided.
    handoff_target: tuple[str, int] | None = None
    if pr is not None:
        handoff_target = ("pr", pr)
    elif issue is not None:
        handoff_target = ("issue", issue)
    if handoff_target is None:
        statuses.append(
            _handoff_status(
                ok=False,
                reason="not run: handoff comments are thread-scoped; filter by issue or pr to include them",
            )
        )
    elif comment_source is None:
        statuses.append(
            _handoff_status(
                ok=False,
                reason="not run: no comment source configured for this timeline read",
            )
        )
    else:
        kind, number = handoff_target
        try:
            comments = comment_source(kind, number) or []
            handoff_events = adapt_cth_comments(comments, kind=kind, number=number)
            all_events.extend(handoff_events)
            statuses.append(_handoff_status(ok=True, count=len(handoff_events)))
        except Exception as exc:  # fail soft: a fetch/parse error degrades this source only
            statuses.append(_handoff_status(ok=False, reason=f"handoff source failed: {exc}"))
    requested = tuple(
        name
        for name, value in ((FILTER_ISSUE, issue), (FILTER_PR, pr), (FILTER_SESSION, session))
        if value is not None
    )
    statuses = [
        replace(
            status,
            unsupported_filters=tuple(
                dim for dim in requested if dim not in status.supported_filters
            ),
        )
        for status in statuses
    ]
    filters = {"issue": issue, "pr": pr, "session": session}
    # A dimension is answerable only if a source that actually ran can carry it.
    # If none can, refuse: an empty page would assert "no such activity", which
    # is a claim this timeline is not in a position to make.
    answerable: set[str] = set()
    for status in statuses:
        if status.ok:
            answerable.update(status.supported_filters)
    unanswerable = tuple(dim for dim in requested if dim not in answerable)
    if unanswerable:
        return TimelineSnapshot(
            schema_version=TIMELINE_SCHEMA_VERSION,
            remote=remote,
            org=org,
            repo=repo,
            filters=filters,
            page=paginate([], limit=limit, offset=offset),
            sources=tuple(statuses),
            ok=False,
            error={
                "code": "filter_not_supported",
                "unsupported_filters": list(unanswerable),
                "detail": (
                    "no timeline source that ran can answer "
                    + ", ".join(f"'{dim}'" for dim in unanswerable)
                    + "; the result is refused rather than returned empty"
                ),
                "sources": _unanswerable_reasons(statuses, unanswerable),
            },
        )
    filtered = filter_events(all_events, issue=issue, pr=pr, session=session)
    ordered = sort_events(filtered)
    page = paginate(ordered, limit=limit, offset=offset)
    return TimelineSnapshot(
        schema_version=TIMELINE_SCHEMA_VERSION,
        remote=remote,
        org=org,
        repo=repo,
        filters=filters,
        page=page,
        sources=tuple(statuses),
    )
 def snapshot_to_dict(snapshot: TimelineSnapshot) -> dict[str, Any]:
    return snapshot.to_dict()