Gitea-Tools/author_lock_contract.py

"""One canonical author issue-lock contract shared by every writer (#953).

Before this module, ``gitea_lock_issue`` and
``gitea_bootstrap_author_issue_worktree`` each wrote their own lock record.
``gitea_lock_issue`` wrote the canonical shape — ``work_lease`` carrying the
claimant plus a sanctioned ``lock_provenance`` — while bootstrap wrote a thinner
record with the claimant at the lock top level, ``lease_id: null``, and no
``work_lease``, ``lock_provenance``, or expiry at all.

Every downstream reader was written against the canonical shape, so a lock that
bootstrap reported as successfully created was simultaneously:

* un-heartbeatable — the ownership check read the claimant only from
  ``work_lease.claimant``;
* un-renewable — expiry is read only from ``work_lease.expires_at``, so a
  missing lease read as "never expires", and #760 exact-owner renewal only ever
  assesses an *expired* lease;
* un-re-lockable — the branch had by then advanced past its base;
* and rejected by the #447 create-PR provenance guard.

Each of those gates is individually correct. The defect was that two writers
disagreed about what a lock *is*. This module is the single definition, and both
writers now build through it.

Nothing here weakens a guard. ``build_sanctioned_lock_provenance`` remains the
only provenance source, provenance is never accepted from a caller, and the
#447 guard is untouched — this module simply makes bootstrap satisfy it.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any, Mapping

import issue_lock_provenance
import issue_lock_store
import lease_policy

# Bootstrap writes through the same sanctioned source as gitea_lock_issue: the
# lock it produces *is* a canonical lock, not a second dialect that readers must
# learn. Adding a distinct source would have required widening
# SANCTIONED_LOCK_SOURCES, which is exactly the #447 weakening this issue's
# safety requirements forbid.
SOURCE_BOOTSTRAP = issue_lock_provenance.SOURCE_LOCK_ISSUE

# Recovery of an incomplete bootstrap lock (#953 AC8-AC11) deliberately writes
# through SOURCE_LOCK_ISSUE too, and records its distinctness in
# ``lock_provenance.written_by_tool`` plus the ``bootstrap_lock_recovery``
# transition block instead. There is no distinct recovery *source* constant, for
# the same reason bootstrap has none: minting one would require widening
# SANCTIONED_LOCK_SOURCES, which the #447 safety requirements forbid.

#: Top-level keys every canonical author issue lock must carry.
REQUIRED_LOCK_FIELDS: tuple[str, ...] = (
    "remote",
    "org",
    "repo",
    "issue_number",
    "branch_name",
    "worktree_path",
    "work_lease",
    "lock_provenance",
)

#: Keys every canonical ``work_lease`` must carry.
REQUIRED_WORK_LEASE_FIELDS: tuple[str, ...] = (
    "operation_type",
    "issue_number",
    "branch",
    "worktree_path",
    "claimant",
    "created_at",
    "expires_at",
    "last_heartbeat_at",
    "task_session_id",
    "lifecycle_version",
)

# ── Explicit expiration states (AC12) ──
# The bug this replaces: a lock with no recorded expiry produced
# ``is_lease_expired() -> False``, which reads as "not yet expired" and made the
# lock permanently non-expiring *and* permanently ineligible for the renewal
# path, which only ever assesses an expired lease. "Absent" and "in the future"
# are different facts and are now named differently.
EXPIRATION_RECORDED = "recorded"
EXPIRATION_MISSING = "missing"
EXPIRATION_UNPARSEABLE = "unparseable"

#: Structural verdicts returned by :func:`assess_lock_contract`.
CONTRACT_CANONICAL = "canonical"
CONTRACT_INCOMPLETE = "incomplete"
CONTRACT_LEGACY = "legacy"
CONTRACT_ABSENT = "absent"


def _text(value: Any) -> str:
    return str(value or "").strip()


def now_utc() -> datetime:
    return datetime.now(timezone.utc)


def format_timestamp(value: datetime) -> str:
    """Serialize in the durable ``...Z`` form already used on disk."""
    return (
        value.astimezone(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )


def lock_claimant(lock: Mapping[str, Any] | None) -> dict[str, str]:
    """Read the claimant from either canonical or legacy placement.

    ``work_lease.claimant`` is canonical and is preferred. A top-level
    ``claimant`` is the legacy/bootstrap placement and is accepted as a
    fallback (AC14) — three separate readers already disagreed about this
    (``issue_lock_store``, ``issue_lock_renewal``, ``issue_lock_recovery``),
    which is why it now lives in one place.

    Reading a legacy placement is *not* a widening: every caller still compares
    the values it returns against server-resolved identity and profile. This
    only decides where to look, never whether ownership is proven.

    Delegates to ``issue_lock_store.lock_claimant`` rather than reimplementing
    the rule. A second copy here would be a fourth reader that could drift from
    the other three, which is the exact failure #953 exists to end. It lives in
    the store because ``author_lock_contract`` imports the store, so defining it
    here would make that import circular.
    """
    recorded = issue_lock_store.lock_claimant(dict(lock) if isinstance(lock, Mapping) else None)
    return {
        "username": _text(recorded.get("username")),
        "profile": _text(recorded.get("profile")),
    }


def claimant_placement(lock: Mapping[str, Any] | None) -> str:
    """Where the claimant was found: ``work_lease``, ``top_level``, or ``absent``."""
    if not isinstance(lock, Mapping):
        return "absent"
    lease = lock.get("work_lease")
    if isinstance(lease, Mapping) and isinstance(lease.get("claimant"), Mapping):
        return "work_lease"
    if isinstance(lock.get("claimant"), Mapping):
        return "top_level"
    return "absent"


def build_claimant(*, username: str | None, profile: str | None) -> dict[str, str]:
    """Build the canonical claimant pair from server-resolved values."""
    return {"username": _text(username), "profile": _text(profile)}


def build_author_issue_work_lease(
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    claimant: Mapping[str, Any],
    task_session_id: str | None = None,
    created: datetime | None = None,
) -> dict[str, Any]:
    """Build the canonical author ``work_lease``.

    The single definition behind both writers. The TTL comes from the central
    policy rather than a literal, and the window slides from the last valid
    heartbeat (#790), so an abandoned task releases its claim within one TTL.
    """
    started = created or now_utc()
    policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
    expires = started + timedelta(minutes=policy.initial_ttl_minutes)
    session_id = _text(task_session_id) or issue_lock_store.mint_task_session_id(
        issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
    )
    return {
        "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
        "issue_number": int(issue_number),
        "pr_number": None,
        "branch": branch_name,
        "worktree_path": worktree_path,
        "claimant": dict(claimant),
        "created_at": format_timestamp(started),
        "expires_at": format_timestamp(expires),
        "last_heartbeat_at": format_timestamp(started),
        # #790 AC-N1: the ownership key for this task, distinct from the
        # recorded PID, which is the shared daemon and identifies no task.
        "task_session_id": session_id,
        # #790 AC-N8: the explicit lifecycle marker. Its absence — never a
        # timestamp comparison — is what makes a lock legacy.
        "lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1,
        "heartbeat_count": 1,
    }


def build_canonical_issue_lock(
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    remote: str,
    org: str,
    repo: str,
    identity: str | None,
    profile: str | None,
    tool: str,
    source: str = issue_lock_provenance.SOURCE_LOCK_ISSUE,
    owner_session: str | None = None,
    assignment_id: str | None = None,
    lease_id: str | None = None,
    expected_base_sha: str | None = None,
    task_session_id: str | None = None,
    created: datetime | None = None,
) -> dict[str, Any]:
    """Build a complete canonical lock record.

    ``tool`` and ``source`` are server-supplied. There is deliberately no
    parameter through which a caller could inject provenance: the #953 safety
    requirements forbid caller-manufactured provenance, so provenance is always
    minted here from ``build_sanctioned_lock_provenance``.
    """
    claimant = build_claimant(username=identity, profile=profile)
    work_lease = build_author_issue_work_lease(
        issue_number=issue_number,
        branch_name=branch_name,
        worktree_path=worktree_path,
        claimant=claimant,
        task_session_id=task_session_id,
        created=created,
    )
    record: dict[str, Any] = {
        "remote": remote,
        "org": org,
        "repo": repo,
        "issue_number": int(issue_number),
        "branch": branch_name,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool=tool,
            source=source,
            claimant=claimant,
        ),
    }
    if owner_session is not None:
        record["owner_session"] = owner_session
    if assignment_id is not None:
        record["assignment_id"] = assignment_id
    # #953 AC6: a null lease id is recorded only when no workflow lease was
    # allocated for this bootstrap. The task-session identifier in the
    # work_lease is what downstream ownership checks fence on, and it is never
    # null on a canonical lock.
    if lease_id is not None:
        record["lease_id"] = lease_id
    if expected_base_sha is not None:
        record["expected_base_sha"] = expected_base_sha
    return record


def expiration_state(lock: Mapping[str, Any] | None) -> dict[str, Any]:
    """Classify a lock's recorded expiry explicitly (AC12).

    Distinguishes "no expiry was ever recorded" from "an expiry was recorded
    and is still in the future". Collapsing those two into a single ``False``
    from ``is_lease_expired`` is what let a malformed lock be treated as
    permanently live and simultaneously never renewable.
    """
    if not isinstance(lock, Mapping):
        return {"state": EXPIRATION_MISSING, "expires_at": None, "expired": None}
    lease = lock.get("work_lease")
    raw = lease.get("expires_at") if isinstance(lease, Mapping) else None
    text = _text(raw)
    if not text:
        return {"state": EXPIRATION_MISSING, "expires_at": None, "expired": None}
    try:
        parsed = datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(
            timezone.utc
        )
    except ValueError:
        return {"state": EXPIRATION_UNPARSEABLE, "expires_at": text, "expired": None}
    return {
        "state": EXPIRATION_RECORDED,
        "expires_at": text,
        "expired": parsed <= now_utc(),
    }


def missing_contract_fields(lock: Mapping[str, Any] | None) -> list[str]:
    """Name every canonical field a lock does not carry (AC7)."""
    if not isinstance(lock, Mapping):
        return ["<no lock record>"]
    missing: list[str] = []
    for field in REQUIRED_LOCK_FIELDS:
        value = lock.get(field)
        if value is None or (isinstance(value, str) and not value.strip()):
            missing.append(field)
    lease = lock.get("work_lease")
    if not isinstance(lease, Mapping):
        if "work_lease" not in missing:
            missing.append("work_lease")
    else:
        for field in REQUIRED_WORK_LEASE_FIELDS:
            value = lease.get(field)
            if value is None or (isinstance(value, str) and not value.strip()):
                missing.append(f"work_lease.{field}")
    provenance = lock.get("lock_provenance")
    if isinstance(provenance, Mapping):
        if (
            _text(provenance.get("source"))
            not in issue_lock_provenance.SANCTIONED_LOCK_SOURCES
        ):
            missing.append("lock_provenance.source (not sanctioned)")
        if not _text(provenance.get("written_by_tool")):
            missing.append("lock_provenance.written_by_tool")
    claimant = lock_claimant(lock)
    if not claimant["username"]:
        missing.append("claimant.username")
    if not claimant["profile"]:
        missing.append("claimant.profile")
    return missing


def assess_lock_contract(lock: Mapping[str, Any] | None) -> dict[str, Any]:
    """Structural, read-only verdict on a durable lock record (AC7, AC16).

    Pure inspection: it reads the record it is handed and mutates nothing —
    no lock, lease, branch, worktree, issue, or PR. Callers use it both to
    verify a lock they just wrote and to report on one they found.
    """
    if not isinstance(lock, Mapping) or not lock:
        return {
            "contract": CONTRACT_ABSENT,
            "canonical": False,
            "missing_fields": ["<no lock record>"],
            "claimant": {"username": "", "profile": ""},
            "claimant_placement": "absent",
            "expiration": {
                "state": EXPIRATION_MISSING,
                "expires_at": None,
                "expired": None,
            },
            "heartbeatable": False,
            "create_pr_eligible": False,
            "lock_generation": None,
            "task_session_id": None,
            "reasons": ["no durable lock record"],
        }

    missing = missing_contract_fields(lock)
    claimant = lock_claimant(lock)
    placement = claimant_placement(lock)
    expiration = expiration_state(lock)
    provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(dict(lock))

    # Canonical means: every required field present, the claimant in the
    # canonical placement, an expiry actually recorded, and the untouched #447
    # guard satisfied.
    canonical = (
        not missing
        and placement == "work_lease"
        and expiration["state"] == EXPIRATION_RECORDED
        and bool(provenance_check.get("proven"))
    )
    if canonical:
        contract = CONTRACT_CANONICAL
    elif placement == "top_level" and claimant["username"] and claimant["profile"]:
        contract = CONTRACT_LEGACY
    else:
        contract = CONTRACT_INCOMPLETE

    reasons: list[str] = []
    if missing:
        reasons.append("missing canonical fields: " + ", ".join(missing))
    if placement == "top_level":
        reasons.append(
            "claimant recorded at the lock top level rather than in work_lease "
            "(legacy/bootstrap placement)"
        )
    if expiration["state"] == EXPIRATION_MISSING:
        reasons.append(
            "no expiration recorded; the lock is neither expirable nor renewable "
            "until it is upgraded"
        )
    elif expiration["state"] == EXPIRATION_UNPARSEABLE:
        reasons.append(f"unparseable expires_at '{expiration['expires_at']}'")
    if provenance_check.get("block"):
        reasons.extend(provenance_check.get("reasons") or [])

    # Heartbeat needs the claimant pair (from either placement, post-fix) plus a
    # task-session identifier to fence on.
    lease = lock.get("work_lease")
    task_session_id = (
        _text(lease.get("task_session_id")) if isinstance(lease, Mapping) else ""
    )
    heartbeatable = bool(
        claimant["username"] and claimant["profile"] and task_session_id
    )

    return {
        "contract": contract,
        "canonical": canonical,
        "missing_fields": missing,
        "claimant": claimant,
        "claimant_placement": placement,
        "expiration": expiration,
        "heartbeatable": heartbeatable,
        "create_pr_eligible": bool(provenance_check.get("proven")),
        "lock_generation": lock.get("lock_generation"),
        "task_session_id": task_session_id or None,
        "reasons": reasons,
    }


def format_contract_refusal(assessment: Mapping[str, Any]) -> str:
    """Human-readable refusal naming exactly what the lock is missing."""
    missing = ", ".join(assessment.get("missing_fields") or []) or "unknown fields"
    return (
        "Issue lock contract incomplete (#953): "
        f"{missing}. The lock cannot be heartbeated, renewed, or accepted by "
        "gitea_create_pr in this state (fail closed)"
    )


def recommended_action(assessment: Mapping[str, Any]) -> str:
    """The one executable next step for a lock in this state (AC5, AC15)."""
    contract = assessment.get("contract")
    if contract == CONTRACT_CANONICAL:
        return (
            "Lock is canonical. Call gitea_whoami, then "
            "gitea_resolve_task_capability(task='work_issue'), then proceed with "
            "author implementation in the bootstrapped worktree."
        )
    if contract == CONTRACT_ABSENT:
        return (
            "No durable lock exists. Call gitea_lock_issue for this issue and "
            "branch before writing any implementation bytes."
        )
    return (
        "Do not begin implementation. Call "
        "gitea_recover_incomplete_bootstrap_lock for this exact issue, branch, "
        "and worktree to upgrade the lock to the canonical contract, or "
        "gitea_lock_issue while the worktree is still base-equivalent."
    )


# ── Post-compensation recovery guidance (#953 AC5/AC15, review 632 F2) ──
#
# ``recommended_action`` above answers "what can be done about a lock in this
# shape". That is the wrong question on the bootstrap AC7 refusal path, because
# ``run_compensating_recovery`` has already run by the time the answer is
# reported: it releases the lock, removes the worktree when clean — which it
# always is there, no implementation bytes having been written — and deletes the
# created branch. Recommending incomplete-lock recovery for those artifacts
# hands the author two refusals in a row (``no_durable_lock``, then
# ``worktree_invalid``) for a state that a plain bootstrap retry would fix. The
# advice must describe the state that actually *remains*.

#: Compensation removed every artifact this transition created.
CLEANUP_COMPLETE = "complete"
#: Compensation removed some artifacts; others survive and are still actionable.
CLEANUP_PARTIAL = "partial"
#: Compensation itself failed or could not be observed; nothing is provable.
CLEANUP_FAILED = "failed"


def assess_post_compensation_state(
    recovery: Mapping[str, Any] | None,
    *,
    lock_present: bool,
    worktree_present: bool,
    branch_present: bool,
) -> dict[str, Any]:
    """Classify what survived compensation, from observed durable state.

    Pure. The caller observes the filesystem and git; this decides. Observation
    is authoritative over the journal's ``rolled_back`` list, which records what
    compensation *attempted*: ``run_compensating_recovery`` swallows a failed
    lock release and appends nothing, so an absent marker proves nothing either
    way. The list is still carried through as corroborating evidence.

    The three states are distinct facts, not degrees of the same one:

    * ``CLEANUP_COMPLETE`` — compensation ran and nothing it created remains.
    * ``CLEANUP_PARTIAL`` — compensation ran and artifacts survive, whether by
      design (a worktree dirty at rollback time, a branch carrying commits) or
      because a rollback step errored. Either way the surviving set was observed
      directly, so it is known and actionable; ``failed_rollback_steps`` records
      which cause applies.
    * ``CLEANUP_FAILED`` — compensation never ran to completion, so nothing it
      would have removed can be assumed removed.
    """
    rolled_back = list((recovery or {}).get("rolled_back") or [])
    executed = bool((recovery or {}).get("executed"))
    failed_steps = [entry for entry in rolled_back if "_failed" in entry]

    surviving: list[str] = []
    if lock_present:
        surviving.append("lock")
    if worktree_present:
        surviving.append("worktree")
    if branch_present:
        surviving.append("branch")

    if not executed:
        state = CLEANUP_FAILED
    elif surviving:
        state = CLEANUP_PARTIAL
    else:
        state = CLEANUP_COMPLETE

    return {
        "cleanup_state": state,
        "compensation_executed": executed,
        "lock_present": bool(lock_present),
        "worktree_present": bool(worktree_present),
        "branch_present": bool(branch_present),
        "surviving_artifacts": surviving,
        "removed_artifacts": [
            name
            for name, present in (
                ("lock", lock_present),
                ("worktree", worktree_present),
                ("branch", branch_present),
            )
            if not present
        ],
        "failed_rollback_steps": failed_steps,
        "rolled_back": rolled_back,
    }


def post_compensation_action(
    state: Mapping[str, Any],
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    missing_fields: list[str] | None = None,
) -> str:
    """The one executable next step for the state compensation actually left.

    Every branch names only artifacts the classification says still exist, so no
    recommendation can point at something the rollback deleted.
    """
    missing = ", ".join(missing_fields or []) or "the reported missing fields"
    cleanup_state = state.get("cleanup_state")
    lock_present = bool(state.get("lock_present"))
    worktree_present = bool(state.get("worktree_present"))
    branch_present = bool(state.get("branch_present"))

    if not state.get("compensation_executed"):
        # Compensation never ran, so nothing was rolled back and nothing about
        # the remaining state was decided. The read-only surface is the only
        # action executable under any state.
        return (
            "Compensating rollback did not complete, so the remaining state is "
            f"not proven. Call gitea_inspect_issue_lock_contract for issue "
            f"#{issue_number} (read-only) to establish what survives before any "
            "further action. Do not retry bootstrap until it is known."
        )

    prefix = ""
    failed_steps = state.get("failed_rollback_steps") or []
    if failed_steps:
        prefix = (
            "Compensating rollback reported a failed step "
            f"({', '.join(failed_steps)}); what survives was observed directly "
            "and the action below is scoped to exactly that. "
        )

    if cleanup_state == CLEANUP_COMPLETE:
        return (
            "Compensating rollback removed the malformed lock, the branch, and "
            f"the worktree, so nothing from this attempt remains. Resolve "
            f"{missing} and re-run gitea_bootstrap_author_issue_worktree for "
            f"issue #{issue_number} from the clean pre-bootstrap state. Do not "
            "call gitea_recover_incomplete_bootstrap_lock: there is no lock, "
            "branch, or worktree left for it to act on."
        )

    if lock_present and worktree_present and branch_present:
        return prefix + (
            "The lock, branch, and worktree all survive. Call "
            "gitea_recover_incomplete_bootstrap_lock for issue "
            f"#{issue_number}, branch '{branch_name}', and worktree "
            f"'{worktree_path}', passing the worktree's current head as "
            "expected_head, to upgrade the lock to the canonical contract."
        )

    if not lock_present and worktree_present and branch_present:
        return prefix + (
            "The malformed lock was released but the branch and worktree "
            "survive. No implementation bytes were written, so the worktree is "
            f"still base-equivalent: call gitea_lock_issue for issue "
            f"#{issue_number} on branch '{branch_name}' from worktree "
            f"'{worktree_path}' to acquire a canonical lock."
        )

    if lock_present and not worktree_present:
        return prefix + (
            f"The worktree for issue #{issue_number} is gone but the durable "
            "lock survived, so neither gitea_recover_incomplete_bootstrap_lock "
            "(it would refuse worktree_invalid) nor gitea_lock_issue (it has no "
            "worktree to bind) is executable. Call "
            "gitea_inspect_issue_lock_contract for issue "
            f"#{issue_number} (read-only) to confirm the surviving lock; it "
            "must be released by its recorded owner before bootstrap is "
            "retried."
        )

    # Lock gone, worktree gone, some git artifact left (a branch with commits,
    # or a branch this transition did not create).
    return prefix + (
        "Compensating rollback removed the lock and worktree; branch "
        f"'{branch_name}' survives and was not deleted. Call "
        f"gitea_inspect_issue_lock_contract for issue #{issue_number} "
        "(read-only) to confirm no durable lock remains, then re-run "
        "gitea_bootstrap_author_issue_worktree, which will adopt the existing "
        "branch rather than recreating it."
    )