Gitea-Tools/create_issue_bootstrap.py

"""Sanctioned pre-issue bootstrap for ``create_issue`` (#749).

``gitea_create_issue`` is a pure remote mutation: it creates a tracking issue
and writes nothing to the local working tree. The issue-first gate forbids
creating ``branches/issue-<N>-*`` before the issue number exists, while the
#274 branches-only guard previously demanded that worktree first — a deadlock.

This module defines a **narrow, phase-scoped** exemption:

* Only tasks in :data:`CREATE_ISSUE_TASKS` may use it.
* Only the **canonical control checkout** may be used (never an arbitrary
  directory, unrelated worktree, or foreign clone).
* The control checkout must be clean, on an accepted base branch, and
  base-equivalent to live master when a remote tip is known.
* Every post-creation author mutation keeps the ordinary ``branches/`` rule.

The exemption cannot widen: unknown tasks, dirty roots, drifted HEADs, non-base
branches, and non-control workspaces fall through to the existing fail-closed
guards.
"""

from __future__ import annotations

import os
from typing import Any

from author_mutation_worktree import BASE_BRANCHES, is_path_under_branches
from reviewer_worktree import parse_dirty_tracked_files

CREATE_ISSUE_TASKS = frozenset({"create_issue", "gitea_create_issue"})

# Satisfiable before an issue number exists — never names issue-<N>.
EXACT_NEXT_ACTION_BOOTSTRAP = (
    "Restore the canonical control checkout to a clean accepted base branch "
    "(master/main/dev) that matches live master, with no tracked local edits "
    "and no detached HEAD. Re-resolve the exact create_issue task, then re-run "
    "gitea_create_issue from that clean control checkout. Do not create "
    "branches/issue-<N>-* worktrees, dummy directories, or borrow unrelated "
    "worktrees before the issue exists."
)

EXACT_NEXT_ACTION_POST_CREATE = (
    "After the issue exists: create a registered worktree under "
    "branches/issue-<N>-* from clean master, claim/lock the issue, set "
    "GITEA_AUTHOR_WORKTREE / worktree_path to that path, then continue author "
    "mutations from the issue-backed worktree only."
)


def is_create_issue_task(task: str | None) -> bool:
    """True when *task* is the create_issue mutation (or tool alias)."""
    return (task or "").strip() in CREATE_ISSUE_TASKS


def normalize_sha(value: str | None) -> str | None:
    """Normalize a Git object id for comparison, or ``None`` when unknown.

    Whitespace and case are the only permitted variation between two spellings
    of the same commit; anything else is a different commit. Empty and
    whitespace-only values normalize to ``None`` so an unknown tip can never
    compare equal to another unknown tip.
    """
    normalized = (value or "").strip().lower()
    return normalized or None


def assess_create_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 create_issue may proceed from the control checkout.

    Returns a structured assessment:

    * ``not_applicable`` — not a create_issue task, or workspace is already a
      ``branches/`` worktree (use ordinary guards).
    * ``allowed`` — create_issue bootstrap may proceed from this control root.
    * ``block`` — create_issue was attempted from control checkout but gates
      failed (dirty, wrong branch, base race, etc.).
    """
    reasons: list[str] = []
    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 = is_path_under_branches(workspace, root) if root else False

    if not is_create_issue_task(task):
        return _result(
            not_applicable=True,
            allowed=False,
            block=False,
            reasons=["task is not create_issue"],
            workspace=workspace,
            root=root,
            branch=branch,
            dirty=dirty,
            under_branches=under_branches,
        )

    # Registered branches/ worktrees keep the normal path (no bootstrap).
    if under_branches:
        return _result(
            not_applicable=True,
            allowed=False,
            block=False,
            reasons=["workspace is under branches/; ordinary #274 path applies"],
            workspace=workspace,
            root=root,
            branch=branch,
            dirty=dirty,
            under_branches=True,
        )

    # Only the exact canonical control checkout is eligible.
    if not root or workspace != root:
        reasons.append(
            "create_issue bootstrap requires the canonical control checkout; "
            f"workspace '{workspace}' is not the repository root '{root or '(unknown)'}'"
        )
        return _result(
            not_applicable=False,
            allowed=False,
            block=True,
            reasons=reasons,
            workspace=workspace,
            root=root,
            branch=branch,
            dirty=dirty,
            under_branches=False,
            exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
        )

    if dirty:
        reasons.append(
            "create_issue bootstrap blocked: control checkout has tracked local "
            f"edits (dirty files: {', '.join(dirty)})"
        )

    if not branch:
        reasons.append(
            "create_issue bootstrap blocked: control checkout is detached HEAD; "
            "expected an accepted base branch (master/main/dev)"
        )
    elif branch not in BASE_BRANCHES:
        reasons.append(
            f"create_issue bootstrap blocked: control checkout branch '{branch}' "
            f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
        )

    # #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An
    # unknown tip on either side is not evidence of agreement, so a missing
    # local HEAD, an unresolvable live master, or a resolver failure all block.
    remote_tip = normalize_sha(remote_master_sha)
    local_tip = normalize_sha(head_sha)
    resolver_error = (remote_master_sha_error or "").strip() or None

    if not local_tip:
        reasons.append(
            "create_issue bootstrap blocked: control checkout HEAD SHA is "
            "unknown; base equivalence to live master cannot be proven "
            "(fail closed)"
        )

    if resolver_error:
        reasons.append(
            "create_issue bootstrap blocked: live master tip could not be "
            f"resolved ({resolver_error}); base equivalence cannot be proven "
            "(fail closed)"
        )
    elif not remote_tip:
        reasons.append(
            "create_issue bootstrap blocked: live master tip is unknown; base "
            "equivalence to live master cannot be proven (fail closed)"
        )

    if remote_tip and local_tip and remote_tip != local_tip:
        reasons.append(
            "create_issue bootstrap blocked: control checkout HEAD does not match "
            f"live master (HEAD {local_tip[:12]}, live master {remote_tip[:12]})"
        )

    if reasons:
        return _result(
            not_applicable=False,
            allowed=False,
            block=True,
            reasons=reasons,
            workspace=workspace,
            root=root,
            branch=branch or None,
            dirty=dirty,
            under_branches=False,
            exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
            local_head_sha=local_tip,
            remote_master_sha=remote_tip,
        )

    return _result(
        not_applicable=False,
        allowed=True,
        block=False,
        reasons=[],
        workspace=workspace,
        root=root,
        branch=branch or None,
        dirty=dirty,
        under_branches=False,
        exact_next_action=EXACT_NEXT_ACTION_POST_CREATE,
        bootstrap_path="clean_canonical_control_checkout",
        local_head_sha=local_tip,
        remote_master_sha=remote_tip,
    )


def bootstrap_permits_control_checkout(
    assessment: Any,
    *,
    task: str | None,
    workspace_path: str | None,
    canonical_repo_root: str | None,
) -> bool:
    """Single interpretation of a bootstrap assessment (#757).

    Both author-mutation guards — the #274 branches-only enforcer and the #604
    anti-stomp preflight — route their "may this workspace mutate" decision
    through this predicate, so the two can never reach opposite conclusions
    about identical evidence.

    Fail-closed by construction. Every proof obligation must be present and
    affirmative in *assessment*, and the assessment must describe the very
    workspace and canonical root being guarded. A missing, malformed, refused,
    incomplete, or contradictory assessment returns ``False``, which leaves the
    caller's ordinary block in force.

    ``assessment`` is server-derived only: it is produced by
    :func:`assess_create_issue_bootstrap` from inspected repository state. It is
    never accepted from an MCP tool argument, so no caller can assert
    eligibility it has not proven.
    """
    if not isinstance(assessment, dict):
        return False
    if not is_create_issue_task(task):
        return False

    # Positive proof: the assessment must affirmatively allow, with no
    # competing refusal or not-applicable disposition recorded alongside it.
    if assessment.get("allowed") is not True:
        return False
    if assessment.get("proven") is not True:
        return False
    if assessment.get("block") is not False:
        return False
    if assessment.get("not_applicable") is not False:
        return False
    if assessment.get("reasons"):
        return False

    # Scope proof: only the create_issue bootstrap, only via the clean
    # canonical control checkout path.
    if assessment.get("task_scope") != "create_issue_only":
        return False
    if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
        return False

    # State proof: clean, and not a branches/ worktree (those keep #274).
    if assessment.get("dirty_files"):
        return False
    if assessment.get("under_branches") is not False:
        return False

    # Base-equivalence proof (#757 AC3/AC4): both tips must be recorded,
    # nonempty, and equal. Re-derived here rather than trusted from the
    # assessment's own flag, so a hand-built or truncated assessment cannot
    # assert agreement it never proved.
    if assessment.get("base_tips_verified") is not True:
        return False
    local_tip = normalize_sha(assessment.get("local_head_sha"))
    remote_tip = normalize_sha(assessment.get("remote_master_sha"))
    if not local_tip or not remote_tip or local_tip != remote_tip:
        return False

    # Binding proof: the assessment must describe *this* workspace and root,
    # and that workspace must be exactly the canonical control checkout.
    root = os.path.realpath(canonical_repo_root or "")
    workspace = os.path.realpath(workspace_path or root or ".")
    if not root or workspace != root:
        return False
    if assessment.get("canonical_repo_root") != root:
        return False
    if assessment.get("workspace_path") != workspace:
        return False

    return True


def format_create_issue_bootstrap_error(assessment: dict[str, Any]) -> str:
    """RuntimeError / typed-block message for a failed bootstrap assessment."""
    reasons = "; ".join(
        assessment.get("reasons") or ["create_issue bootstrap failed"]
    )
    next_action = (
        assessment.get("exact_next_action") or EXACT_NEXT_ACTION_BOOTSTRAP
    )
    root = assessment.get("canonical_repo_root") or "(unknown)"
    workspace = assessment.get("workspace_path") or "(unknown)"
    return (
        f"Create-issue bootstrap guard (#749): {reasons}. "
        f"canonical repository root: {root}; workspace: {workspace}. "
        f"exact_next_action: {next_action}"
    )


def _result(
    *,
    not_applicable: bool,
    allowed: bool,
    block: bool,
    reasons: list[str],
    workspace: str,
    root: str,
    branch: str | None,
    dirty: list[str],
    under_branches: bool,
    exact_next_action: str | None = None,
    bootstrap_path: str | None = None,
    local_head_sha: str | None = None,
    remote_master_sha: str | None = None,
) -> dict[str, Any]:
    # #757 AC3/AC4: equality is recorded only when BOTH tips are known, so a
    # consumer can never read agreement out of two missing values.
    base_tips_verified = bool(
        local_head_sha and remote_master_sha and local_head_sha == remote_master_sha
    )
    return {
        "not_applicable": not_applicable,
        "allowed": allowed,
        "block": block,
        "proven": allowed and not block,
        "reasons": list(reasons),
        "workspace_path": workspace,
        "canonical_repo_root": root,
        "current_branch": branch,
        "dirty_files": list(dirty),
        "under_branches": under_branches,
        "exact_next_action": exact_next_action,
        "bootstrap_path": bootstrap_path,
        "task_scope": "create_issue_only",
        "local_head_sha": local_head_sha,
        "remote_master_sha": remote_master_sha,
        "base_tips_verified": base_tips_verified,
    }