feat(lease): make the author task heartbeat load-bearing (#790 Slice A)

Slice A of Issue #790, per the controller reassessment in comment 13958. Does not close the issue: terminal retirement (Slice B) and the read-side generation check plus the #760 renewal re-scope (Slice C) are deliberately not implemented. The defect. `issue_lock_store.assess_lock_freshness` parsed `last_heartbeat_at` and then never consulted it. Liveness was decided by an absolute four-hour `expires_at` and by PID liveness, and the recorded PID is the long-lived MCP daemon rather than the authoring task, so an abandoned claim stayed live for the full four hours. A tree-wide search found the field written in exactly one place and advanced by nothing. Issue #787 / PR #789 hit this; Issue #760 / PR #791 hit it again, blocking reconciliation for over five hours after its work had landed. A1 — central policy. New `lease_policy` declares every duration for every task class in one place: author initial/sliding TTL 10 minutes, heartbeat cadence 2, stale warning 5, missed-heartbeat grace 10, absolute cap 8 hours, recovery grace 10, terminal race-drain 2. It ships first so the first heartbeat and TTL behavior to run reads from it (AC-N7). The duplicated four-hour literal is gone from both `issue_lock_store` and `gitea_mcp_server`. Reviewer, merger, and conflict-fix classes are declared but not rewired — Slice C moves those call sites — and a test asserts the declaration still equals the constants #747 and `pr_work_lease` own, so the two cannot drift apart unnoticed. A2 — load-bearing freshness, with two deliberate asymmetries. An alive PID never establishes freshness anywhere (AC-N2); it is recorded as evidence and no branch returns live because of it. A dead PID still marks a lease stale, and that band still precedes every heartbeat evaluation, so #753 dead-session recovery keys on exactly the classification it always did. New bands `stale_missed_heartbeat` and `stale_absolute_cap` are classified in `branch_cleanup_guard` rather than falling through to unknown-status, and still block unless the ownership record proves `reclaim_allowed is True`. A heartbeat lease carrying no heartbeat is contradictory and fails closed. `assess_expired_lock_reclaim` accepts a lapsed heartbeat as reclaim grounds for heartbeat-lifecycle leases only: under this lifecycle the heartbeat is the liveness proof, and also requiring a dead PID would reinstate the original defect. A3/A4 — task-session identity and the writer. `mint_task_session_id` produces an ownership key containing no process identifier, since the daemon PID is reused by every task it serves and identifies none of them. `heartbeat_session_lock` writes inside the existing per-issue flock under the #772 generation compare-and-swap, verifying exact issue, branch, realpath-normalized worktree, claimant username, claimant profile, and recorded session identifier. It cannot acquire, take over, or revive: a lease past its grace is refused and must use the reclaim path, so a session that stopped proving liveness cannot restore ownership retroactively. New `gitea_heartbeat_issue_lock` gates on the same authority as `lock_issue`, being strictly narrower. A5 — legacy compatibility (AC-N8). The explicit `lifecycle_version` marker, never a timestamp comparison, discriminates legacy from heartbeat leases: a legacy lock has `last_heartbeat_at == created_at` forever precisely because nothing advanced it, and a freshly minted heartbeat lease has them equal too, so the equality carries no information in either direction. Legacy locks keep their recorded absolute expiry and are never evaluated against the short grace, so deployment cannot make an existing claim instantly reclaimable. They leave that state only by terminal retirement (Slice B) or by `rebind_legacy_lock`, which re-verifies the exact owner and mints a genuine identifier and first heartbeat while preserving the original claim under `legacy_origin`. Rebinding a lapsed legacy lease is refused; that belongs to #760 renewal or #601 reclaim. A6 — native coverage. Review #499 proved assessor-level tests miss discard points, so `tests/test_issue_790_heartbeat_mcp_path.py` drives the real tools against a real git repository and a real durable lock: lock creation and read-back, policy window, freshness, survival of `verify_lock_for_mutation`, invariance of the duplicate-work and linked-open-PR gates, CAS rejection, foreign-session and foreign-claimant refusal, alive-PID-only refusal, missed heartbeat, legacy protection on deployment, and legacy rebinding. Tests. New suites 55 passed. Lock and lease regression set (issue_lock_store, lease_lifecycle, #753, #755, #760 x2, #768, #772, lock registration, worktree, adoption, duplicate gate, branch cleanup guard, capability invariants, claim heartbeat, worktrees) 383 passed with 98 subtests. Full suite 4295 passed, 11 failed, 6 skipped, 499 subtests passed, against a clean master baseline worktree at 620ed6e9 that reports 11 failed and 4240 passed — the same eleven node IDs. The 55-test delta is exactly the new suites; no new failures. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_011u6GKSJwwrrYjguPjs1aK5
2026-07-22 04:19:48 -04:00
14 changed files with 2000 additions and 963 deletions
@@ -163,7 +163,19 @@ _TERMINAL_OWNERSHIP_STATUSES = frozenset(
    {"released", "abandoned", "done", "blocked", "terminal", "closed"}
 )
 _EXPIRED_STATUSES = frozenset({"expired"})
-_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"})
+_STALE_STATUSES = frozenset(
    {
        "stale",
        "stale_dead_process",
        "stale_missing_worktree",
        # #790 Slice A heartbeat-lifecycle bands. Listed here so they are
        # *classified* rather than falling through to the unknown-status branch;
        # they still block unless the ownership record proves
        # ``reclaim_allowed is True``, so the O2 fail-closed rule is unchanged.
        "stale_missed_heartbeat",
        "stale_absolute_cap",
    }
 )
 def _norm_str(value: Any) -> str:
@@ -1,201 +0,0 @@
 # ADR: MCP Control Plane Web Console architecture and information architecture
 - **Status:** Proposed (documentation only; blocks no code, gates every #631 child)
 - **Date:** 2026-07-22
 - **Tracking issue:** [#632](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/632) — architecture and information architecture (Phase 1)
 - **Parent epic:** [#631](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/631) — MCP Control Plane Web Console
 - **Foundation (closed, extend — do not recreate):** #425 tracker and children #426 skeleton, #427 projects, #428 prompts, #429 queue, #430 runtime, #431 audit paste, #432 worktrees, #433 leases, #434 gated actions, #435 auth/deployment boundary, #436 tests/CI
 - **Related:** `mcp-allocator-control-plane-observability-adr.md`, `mcp-stable-control-runtime-policy-adr.md`, `control-plane-db-substrate.md`, `../safety-model.md`, `../tool-boundaries.md`, `../credential-isolation.md`, `../webui-local-dev.md`, `../webui-deployment.md`
 ## 1. Context
 The MVP web UI shipped under `webui/` as a read-only Starlette application with ten operator routes and a JSON export beside most of them. It is a working foundation, not the console product described by epic #631, and it carries no durable architecture record: no layer contract, no authority boundary, no API versioning rule, no page map, and no statement of which phase may open a write path.
 Twenty children (#632–#651) hang off #631. Without one architecture document each implementer re-derives boundaries, and the most likely failure is not a bad view — it is a privileged action wired into the browser before the authorization and audit model of #633 exists.
 This ADR is the single retrievable design source for the console. It decides structure only. It implements no UI, no API, and no change to deployment topology.
 ## 2. Decision summary (core)
 | Layer | Owns | Must not |
 |-------|------|----------|
 | **Browser UI** | Rendering, navigation, operator affordances | Hold tokens, call Gitea/providers directly, or execute an action the server did not gate |
 | **HTTP route layer** (`webui/app.py`) | Versioned routing, authentication, authorization, redaction boundary, audit emission | Contain domain logic or reach past a loader to a raw credential |
 | **Domain loaders** (`webui/*_loader.py`, `*_scanner.py`, `runtime_health.py`, `project_registry.py`) | Assembling read models from authoritative sources | Mutate anything, or emit unredacted secrets across the boundary |
 | **Gitea** | Durable work record: issues, PRs, comments, reviews, labels, merges | Be the concurrency lock under multi-session load |
 | **Control-plane DB** | Sessions, assignment, leases, heartbeats, events | Replace Gitea history |
 | **MCP tools / capability gates** | Mutation authorization | Be re-implemented, mirrored, or bypassed by console code |
 | **External providers** (Sentry/GlitchTip, AI providers) | Incident and usage data | Assign work or mutate Gitea outside the #612 bridge |
 **One-liner:** **Gitea records. The DB coordinates. MCP tools authorize. The console projects state and executes only capability-checked, audited actions. Providers observe.**
 ## 3. Console surface today versus target
 `webui/app.py` currently registers these routes (see `../webui-local-dev.md` for the operator-facing table): `/`, `/health`, `/queue`, `/projects`, `/projects/{id}`, `/prompts`, `/runtime`, `/audit`, `/worktrees`, `/leases`, `/actions`, and the unversioned exports `/api/queue`, `/api/projects`, `/api/prompts`, `/api/runtime`, `/api/audit`, `/api/worktrees`, `/api/leases`, `/api/actions`, `/api/actions/{id}/preview`, `/api/actions/{id}/attempt`.
 Every one of these is **retained and evolved**. No child issue may recreate a route from scratch; each states in its PR which MVP surface it extends and what it changes.
 ## 4. Authority boundaries
 ### 4.1 Gitea (durable record)
 Authoritative for issue and PR identity and state, comments, reviews and verdicts, labels, merges, and branch refs. When the console and Gitea disagree about durable state, Gitea wins and the console view is refreshed — never the reverse.
 ### 4.2 Control-plane DB (coordination)
 Authoritative for live coordination: which session holds which assignment or lease, heartbeat freshness, expiry, and the allocation event log. The console reads it; only allocator and lease tools write it.
 ### 4.3 MCP capability gates (authorization)
 `task_capability_map.py` and `gitea_resolve_task_capability` remain the only authority that decides whether a mutation may run. The console asks; it never answers. A console action that cannot name the MCP tool it delegates to is not an action — it is a defect.
 ### 4.4 Filesystem and git (local state)
 Issue lock files, `branches/` worktrees, and registered git worktrees are read through existing scanners. The console never deletes, rebinds, or force-clears local state outside a Phase 2 gated action.
 ### 4.5 Providers (observe only)
 Sentry/GlitchTip and AI providers are read surfaces. The #612 incident bridge is the only path that turns an observation into Gitea work.
 ## 5. Request flow and the redaction boundary
 ```text
 browser ──HTTP──> route layer ──> domain loader ──> Gitea REST
                      │                        ├──> control-plane DB
                      │                        ├──> filesystem / git
                      │                        └──> providers
                      │
                 [redaction boundary]
                      │
                 audit event
 ```
 | Stage | May hold credentials | Emits |
 |-------|----------------------|-------|
 | Loader → route layer | yes (server-side, via `gitea_auth`) | domain objects |
 | Route layer → browser | **no** | redacted DTOs, HTML |
 Two invariants govern the boundary and are non-negotiable for every child:
 1. **No secrets to the browser.** Tokens, keychain identifiers, Authorization headers, raw provider endpoints, and credential-bearing URLs are redacted by default, consistent with `../safety-model.md` §3 and `../credential-isolation.md`. Serializers redact; templates do not sanitize after the fact.
 2. **No ungated mutations.** A write reaches an authoritative system only by delegating to an MCP tool that passed its own capability gate. HTML forms and JSON endpoints are transport, never authority.
 ## 6. API naming and versioning
 **Decision:** all console APIs added from Phase 1 onward are served under `/api/v1/...`.
 - Nouns are plural and hierarchical: `/api/v1/inventory/leases`, `/api/v1/system/health`.
 - Read endpoints are `GET` and side-effect free.
 - Phase 2 action endpoints are `POST /api/v1/actions/{action_id}/preview` and `POST /api/v1/actions/{action_id}/execute`; `preview` stays side-effect free and returns a mutation ledger.
 - The existing unversioned MVP exports remain as **compatibility aliases** for the whole of Phase 1 so the current operator flow never breaks. They may be retired no earlier than Phase 2, and only after the replacing `v1` route ships and `../webui-local-dev.md` records the swap.
 - A breaking change to a `v1` payload requires `/api/v2/...`, not an in-place edit.
 - Every JSON payload carries enough provenance for an auditor to tell where the data came from — at minimum the source system and whether the inventory was complete, matching the pagination-proof habit the MVP queue export already established.
 ## 7. Page map
 | Page | Purpose | Owning child | Evolves |
 |------|---------|--------------|---------|
 | `/` | Console shell, navigation, next-safe-action summary | #638 | MVP `/` (#426) |
 | `/system` | System-health dashboard | #639 | new, backed by #634 |
 | `/traffic` | Workflow traffic control, queues, blockers | #640 | MVP `/queue` (#429) |
 | `/runtime` | Runtime and session view | #641 | MVP `/runtime` (#430) |
 | `/projects`, `/projects/{id}` | Project registry and onboarding | #635 | MVP `/projects` (#427) |
 | `/inventory` | Sessions, leases, locks, worktrees in one surface | #636 | MVP `/leases` (#433) + `/worktrees` (#432) |
 | `/timeline` | Workflow events and conversation timeline | #637 | new |
 | `/actions` | Gated action registry, preview, execution | #642, #643, #644 | MVP `/actions` (#434) |
 | `/gitea` | Issue and PR linkage console | #645 | new |
 | `/policy` | Guardrail visibility, then versioned editing | #646, #647 | new |
 | `/notifications` | Human-attention routing | #648 | new |
 | `/observability` | Sentry/GlitchTip correlation and durable issue creation | #649 | new |
 | `/providers` | AI-provider connections and insights | #650 | new |
 | `/analytics` | Usage, token cost, latency, workflow performance | #651 | new |
 | `/audit` | Final-report validator preview and audit log | #431 foundation, extended by #633 | MVP `/audit` (#431) |
 | `/prompts`, `/prompts/{id}` | Canonical prompt library | #638 | MVP `/prompts` (#428) |
 | `/health` | Liveness and deployment metadata | #634 | MVP `/health` (#435) |
 ## 8. Component ownership for every epic child
 Each #631 child maps to at least one architectural component defined above.
 | Child | Capability area | Primary component | Phase |
 |-------|-----------------|-------------------|-------|
 | #632 | Architecture and information architecture | this ADR | 1 |
 | #633 | Authorization, RBAC, secret redaction, audit and retention | route layer + redaction boundary (§5) | 1 |
 | #634 | Read-only system-health API | `/api/v1/system/health` + health loader | 1 |
 | #635 | Project registry API evolution | `/api/v1/projects` + `project_registry.py` | 1 |
 | #636 | Session, lease, lock, worktree inventory API | `/api/v1/inventory/*` + `lease_loader.py`, `worktree_scanner.py` | 1 |
 | #637 | Workflow-event and conversation timeline model | `/api/v1/events` + control-plane DB event log | 1 |
 | #638 | Application shell evolution | browser UI layer + `layout.py` | 1 |
 | #639 | System-health dashboard | `/system` page over #634 | 1 |
 | #640 | Workflow traffic-control view | `/traffic` page over the queue loader | 1 |
 | #641 | Runtime and session view | `/runtime` page over `runtime_health.py` | 1 |
 | #642 | Sanctioned restart and graceful reload controls | gated action framework, restart class | 2 |
 | #643 | Requests, intent preview, authorization, workflow initiation | `/api/v1/actions/*` execute path | 2 |
 | #644 | Stale-runtime recovery, worktree rebinding, reconciliation controls | gated actions over filesystem/git authority | 2 |
 | #645 | Gitea issue and PR linkage console | `/gitea` page over Gitea authority | 3 |
 | #646 | Workflow policy and guardrail visibility | `/policy` read view over the capability map | 3 |
 | #647 | Versioned policy editing, validation, simulation, approval, rollback | `/policy` write path, gated | 3 |
 | #648 | Notifications and human-attention routing | notification component over the event model | 3 |
 | #649 | Sentry/GlitchTip connections, correlation, durable issue creation | provider layer + #612 incident bridge | 4 |
 | #650 | AI-provider connections and operational insights | provider layer | 4 |
 | #651 | Model usage, token cost, latency, workflow analytics | analytics component over the event model | 4 |
 Related but **outside** this epic: #667 (restart status, impact preview, and approval controls) belongs to the #655 restart-governance umbrella and must reuse the #642 action class rather than adding a second restart surface.
 ## 9. Phase gates
 | Phase | May ship | Entry condition |
 |-------|----------|-----------------|
 | **1 — read-only visibility** | `GET` pages and `GET /api/v1/...` | this ADR accepted |
 | **2 — controlled actions** | gated `POST` action execution | #633 authorization, RBAC, and audit model landed |
 | **3 — orchestration and policy** | linkage, policy visibility, versioned policy editing | Phase 1 inventory plus the Phase 2 action framework |
 | **4 — insights** | provider correlation, analytics | evidence-backed sources from Phases 1–3 |
 Phase 1 must not open a mutation endpoint, and the read-only guard that returns `405 read-only-mvp` stays in force until the Phase 2 entry condition is met. A phase is not entered by exception; if a control is urgent, the entry condition is what gets prioritized.
 ## 10. Security and workflow safety
 - **Fail closed** on unknown authentication, missing RBAC mapping, or ambiguous lease ownership. An unknown state renders as blocked, never as permitted.
 - **Redact by default**, per §5.
 - **Every privileged action** requires a resolved capability, an explicit operator confirmation, and a durable audit event naming actor, action, target, and outcome.
 - **Contamination surfaces.** Session contamination — including a manually killed MCP daemon (#630) — must be shown and must block clean claims rather than being silently repaired.
 - **Deployment boundary unchanged.** Loopback by default, with the existing refusal of public binds (#435). This ADR documents that target; it does not widen it.
 ## 11. Forbidden paths
 These are rejected designs, not preferences:
 1. **Raw provider incidents as work.** The allocator never receives an unclassified Sentry/GlitchTip incident; only the #612 bridge turns an observation into a Gitea issue.
 2. **Browser-held tokens.** No credential, keychain identifier, or Authorization header is ever sent to the browser or embedded in a client bundle.
 3. **Process-kill recovery.** The console must not expose `pkill`, process-identifier termination, or any host process kill as a recovery affordance (#630). Restart is the sanctioned, operator-owned path of #642 and the #655 umbrella.
 4. **Ungated browser mutations.** No review, approval, merge, close, or comment may originate from the browser without passing an MCP capability gate.
 5. **Policy invented in the console.** The console projects policy from the capability map and canonical workflows; it never encodes a second copy.
 6. **Recreating MVP scope.** Re-implementing a #426–#436 surface without an explicit evolve-or-extend statement is out of bounds.
 ## 12. Approval checklist (readable without chat history)
 A controller can accept or reject this ADR against these six points alone:
 1. Layers and their owners are defined (§2) and each authority is named (§4).
 2. The redaction boundary and the two invariants are stated (§5).
 3. API versioning is decided, including what happens to the existing unversioned routes (§6).
 4. A page map exists and names an owning child for every page (§7).
 5. Every #631 child maps to at least one component and one phase (§8).
 6. Phase gates and forbidden paths are explicit (§9, §11).
 ## 13. Open questions and follow-ups
 Unresolved choices are recorded here rather than settled by implication. Each needs its own durable issue before the phase that depends on it:
 - **Authentication mechanism.** Whether the console authenticates via an access proxy (Cloudflare Access or equivalent) or an application-level session is deferred to #633. This ADR requires only that it fail closed.
 - **Event model substrate.** Whether the #637 timeline reads the control-plane event log directly or through a projection is deferred to #637.
 - **CI path filter coverage.** `webui/ci_paths.py` triggers the web UI suite on `webui/`, `tests/test_webui_*`, and `docs/webui*`. This ADR lives under `docs/architecture/`, so editing it alone does not trigger that gate; the accompanying `tests/test_webui_architecture_docs.py` does run in the full suite. Widening the filter is a small follow-up, deliberately not bundled into a documentation-only change.
 - **Retention.** Audit-event retention duration is owned by #633.
 ## 14. Acceptance
 Accepting this ADR means:
 - Phase 1 children may proceed against the layers, page map, and API rules above.
 - Phase 2 children may not open a write path until #633 lands.
 - Any deviation is recorded as an amendment to this file with its own issue reference, not as an undocumented divergence in code.
@@ -100,6 +100,7 @@ that gates each call, not which tools exist.
 - `gitea_get_profile`
 - `gitea_get_runtime_context`
 - `gitea_get_shell_health`
 - `gitea_heartbeat_issue_lock`
 - `gitea_heartbeat_reviewer_pr_lease`
 - `gitea_inspect_workflow_lease`
 - `gitea_issue_irrecoverable_provenance_authorization`
@@ -37,12 +37,6 @@ Optional environment variables:
 See [webui-deployment.md](webui-deployment.md) for internal-only serving,
 Cloudflare Access/WARP/VPN guidance, and unsafe bind overrides (#435).
 See
 [architecture/webui-control-plane-console-architecture-adr.md](architecture/webui-control-plane-console-architecture-adr.md)
 for the console architecture: layer and authority boundaries, the redaction
 boundary, `/api/v1/...` versioning, the target page map, and the phase gates
 that govern when a write path may open (#632, epic #631).
 ## Routes (MVP)
 | Path | Description |
@@ -2007,6 +2007,7 @@ import allocator_dependencies  # noqa: E402
 import dependency_graph  # noqa: E402  # #784 durable dependency edges
 import control_plane_db  # noqa: E402
 import lease_lifecycle  # noqa: E402
 import lease_policy  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
@@ -2248,7 +2249,6 @@ import canonical_comment_validator as ccv  # noqa: E402
 # GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer.
 # Legacy global path retained only for test/doc references — do not seed manually.
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 WORK_LEASE_TTL_HOURS = 4
 AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
 VALID_WORK_LEASE_OPERATIONS = frozenset({
    AUTHOR_ISSUE_WORK_LEASE,
@@ -2548,7 +2548,12 @@ def _build_author_issue_work_lease(
    host: str | None,
 ) -> dict:
    created = _work_lease_now()
-    expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
+    # #790 Slice A: the window comes from the central policy, not a literal here.
    # It is also now a *sliding* window — the lease lives ``initial_ttl_minutes``
    # past its last valid heartbeat rather than a fixed four hours past its
    # creation, so an abandoned task stops holding the claim within one TTL.
    policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
    expires = created + timedelta(minutes=policy.initial_ttl_minutes)
    return {
        "operation_type": AUTHOR_ISSUE_WORK_LEASE,
        "issue_number": issue_number,
@@ -2559,6 +2564,15 @@ def _build_author_issue_work_lease(
        "created_at": _work_lease_timestamp(created),
        "expires_at": _work_lease_timestamp(expires),
        "last_heartbeat_at": _work_lease_timestamp(created),
        # #790 AC-N1: the ownership key for this task. Distinct from the recorded
        # PID, which is the shared daemon and identifies no individual task.
        "task_session_id": issue_lock_store.mint_task_session_id(
            AUTHOR_ISSUE_WORK_LEASE
        ),
        # #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,
    }
@@ -4328,6 +4342,138 @@ def gitea_lock_issue(
    return result
@mcp.tool()
 def gitea_heartbeat_issue_lock(
    issue_number: int,
    branch_name: str,
    task_session_id: str | None = None,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    worktree_path: str | None = None,
    expected_generation: int | None = None,
 ) -> dict:
    """Prove an owned author issue lease is still active (#790 Slice A).
    The task-liveness signal the lifecycle was missing. Before this, an author
    lease carried a fixed four-hour expiry that nothing could shorten, and the
    only liveness evidence was the recorded PID — the long-lived MCP daemon,
    which stays alive across every task it serves and so proved nothing about
    whether the authoring task still held the work.
    Each successful call slides the lease ``initial_ttl_minutes`` past *now*
    from the central policy, so an actively heartbeating session is never
    evicted while an abandoned one releases its claim within one TTL.
    What this tool cannot do, by construction:
    * **Acquire.** It refuses when no durable lock exists.
    * **Take over.** Exact issue, branch, realpath-normalized worktree,
      claimant username, claimant profile, and recorded task-session identifier
      must all match; a superseded session holding an older identifier is
      refused.
    * **Revive.** A lease already past its grace is not heartbeatable — that
      would let a session restore ownership it had stopped proving. It must use
      the sanctioned reclaim path, which mints a new generation.
    A lock predating the heartbeat lifecycle is rebound rather than heartbeated:
    its exact owner is re-verified and a genuine task-session identifier and
    first heartbeat are minted (#790 AC-N8). The rebind is decided server-side
    from the durable lifecycle marker; there is no caller-facing switch.
    Args:
        issue_number: The locked issue number.
        branch_name: The branch recorded on the lock.
        task_session_id: The identifier this session received when it acquired
            or rebound the lock. It is a fencing token, not an ownership
            assertion: it is compared against durable state and can only ever
            cause a refusal, never grant anything. Omitted only when rebinding a
            legacy lock, which has no identifier yet and mints one.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.
        worktree_path: Author worktree recorded on the lock.
        expected_generation: Optional fencing value. The per-issue flock already
            serializes the read and the write, so this is for a caller that
            wants to pin the generation it last observed across calls; a moved
            generation fails closed.
    Returns:
        dict with 'success', 'performed', the sliding 'expires_at',
        'last_heartbeat_at', 'lock_generation', 'task_session_id', the applied
        'policy', and post-write 'freshness'; on refusal 'success'/'performed'
        False with 'reasons' naming exactly what did not match.
    """
    blocked = _profile_permission_block(
        task_capability_map.required_permission("heartbeat_issue_lock"),
        issue_number=issue_number,
        remote=remote,
        host=host,
        org=org,
        repo=repo,
        org_explicit=org is not None,
        repo_explicit=repo is not None,
    )
    if blocked:
        return blocked
    resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
        worktree_path, _canonical_local_git_root()
    )
    h, o, r = _resolve(remote, host, org, repo)
    claimant = _work_lease_claimant(h)
    identity = claimant.get("username")
    profile = claimant.get("profile")
    existing = _load_existing_issue_lock(
        remote=remote, org=o, repo=r, issue_number=issue_number
    )
    if not existing:
        return {
            "success": False,
            "performed": False,
            "issue_number": issue_number,
            "reasons": [
                f"no durable lock for issue #{issue_number}; heartbeat cannot "
                "acquire a claim (fail closed)"
            ],
        }
    if issue_lock_store.is_legacy_lease(existing):
        # AC-N8 exit route one: canonical exact-owner rebinding. The other exit
        # is terminal retirement, which is Slice B.
        outcome = issue_lock_store.rebind_legacy_lock(
            remote=remote,
            org=o,
            repo=r,
            issue_number=issue_number,
            branch_name=branch_name,
            worktree_path=resolved_worktree,
            identity=identity,
            profile=profile,
            expected_generation=expected_generation,
        )
        outcome["operation"] = "legacy_rebind"
        return outcome
    outcome = issue_lock_store.heartbeat_session_lock(
        remote=remote,
        org=o,
        repo=r,
        issue_number=issue_number,
        branch_name=branch_name,
        worktree_path=resolved_worktree,
        identity=identity,
        profile=profile,
        task_session_id=str(task_session_id or ""),
        expected_generation=expected_generation,
    )
    outcome["operation"] = "heartbeat"
    return outcome
@mcp.tool()
 def gitea_assess_work_issue_duplicate(
    issue_number: int,
@@ -12536,39 +12682,10 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
    return False
 def _git_default_remote_name(root: str) -> str:
    """First configured git remote name for *root*, defaulting to 'origin'.
    Used to resolve the live remote master target for parity (#610). Best
    effort: any failure falls back to 'origin' so callers never raise.
    """
    try:
        res = subprocess.run(
            ["git", "-C", root, "remote"],
            capture_output=True, text=True, check=False,
        )
    except Exception:
        return "origin"
    if res.returncode != 0:
        return "origin"
    names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
    return names[0] if names else "origin"
 def _current_master_parity() -> dict:
-    """Assess this process's code against local and live remote master (#420/#610).
+    """Assess this process's code against the on-disk master HEAD (#420)."""
    Compares the daemon's startup commit, the on-disk checkout HEAD, and the
    live remote master target. A stale daemon relative to live master fails
    closed for mutations even when the local checkout HEAD still matches the
    startup commit. The live-remote read is best effort: an unresolved live
    head leaves read-only diagnostics unblocked but is never mutation-safe.
    """
    current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    live_head = master_parity_gate.read_remote_master_head(
+    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
        PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
    return master_parity_gate.assess_master_parity(
        _STARTUP_PARITY, current_head, live_remote_head=live_head)
 def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16509,34 +16626,15 @@ def gitea_get_runtime_context(
        "restart_required": parity["restart_required"],
        "startup_head": parity["startup_head"],
        "current_head": parity["current_head"],
        # #610 distinguished mutation-safety signals:
        "daemon_start_head": parity["daemon_start_head"],
        "local_head": parity["local_head"],
        "live_remote_head": parity["live_remote_head"],
        "live_known": parity["live_known"],
        "live_stale": parity["live_stale"],
        "mutation_safe": parity["mutation_safe"],
        "summary": master_parity_gate.format_parity(parity),
        "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
        # #610: the capability resolver is authoritative for mutation safety;
        # local parity alone must never authorize a mutation.
        "resolver_authoritative_for_mutation_safety": True,
    }
-    if parity["restart_required"] and not master_parity_gate.gate_disabled():
+    if parity["stale"] and not master_parity_gate.gate_disabled():
-        if parity["live_stale"]:
+        safe_next_action = (
-            safe_next_action = (
+            "Server code is stale relative to master; restart the Gitea MCP "
-                "Daemon is stale relative to LIVE remote master "
+            "server to load current capability gates before mutating. "
-                f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
+            f"({master_parity_gate.format_parity(parity)})"
-                f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
+        )
                "restart/reconnect the Gitea MCP server before mutating. The "
                "capability resolver is authoritative for mutation safety."
            )
        else:
            safe_next_action = (
                "Server code is stale relative to master; restart the Gitea MCP "
                "server to load current capability gates before mutating. "
                f"({master_parity_gate.format_parity(parity)})"
            )
        result["safe_next_action"] = safe_next_action
    if reveal and h:
@@ -16585,13 +16683,6 @@ def gitea_assess_master_parity(
        "determinable": parity["determinable"],
        "startup_head": parity["startup_head"],
        "current_head": parity["current_head"],
        # #610 distinguished mutation-safety signals:
        "daemon_start_head": parity["daemon_start_head"],
        "local_head": parity["local_head"],
        "live_remote_head": parity["live_remote_head"],
        "live_known": parity["live_known"],
        "live_stale": parity["live_stale"],
        "mutation_safe": parity["mutation_safe"],
        "mutation_gate_enforced": enforced,
        "summary": master_parity_gate.format_parity(parity),
        "reasons": parity["reasons"],
@@ -16608,7 +16699,7 @@ def gitea_assess_master_parity(
            source=canonical_source,
        ),
    }
-    if parity["restart_required"] and enforced:
+    if parity["stale"] and enforced:
        out["report"] = master_parity_gate.parity_report(parity)
    return out
@@ -15,15 +15,27 @@ import json
 import os
 import re
 import tempfile
 import uuid
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Any
 import lease_policy
 LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
 DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
 WORK_LEASE_TTL_HOURS = 4
 AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
 # Freshness classifications. ``STATUS_STALE`` remains the dead-PID band that
 # #753 recovery keys on; the two bands below are new in #790 Slice A and apply
 # only to leases minted under the heartbeat lifecycle.
 STATUS_LIVE = "live"
 STATUS_EXPIRED = "expired"
 STATUS_ABSENT = "absent"
 STATUS_STALE = "stale"
 STATUS_STALE_MISSED_HEARTBEAT = "stale_missed_heartbeat"
 STATUS_STALE_ABSOLUTE_CAP = "stale_absolute_cap"
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
@@ -253,6 +265,331 @@ def bind_session_lock(
    return path
 def _ownership_refusals(
    lock: dict[str, Any],
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    identity: str | None,
    profile: str | None,
 ) -> list[str]:
    """Exact-ownership mismatches between a durable lock and a live caller.
    Shared by the heartbeat writer and the legacy rebind path so the two cannot
    disagree about what "the same owner" means. Every field is compared against
    durable state; nothing is taken on the caller's word beyond the identity the
    server itself resolved.
    """
    reasons: list[str] = []
    if lock.get("issue_number") != issue_number:
        reasons.append(
            f"lock targets issue #{lock.get('issue_number')}, not #{issue_number}"
        )
    if str(lock.get("branch_name") or "") != str(branch_name or ""):
        reasons.append(
            f"lock branch '{lock.get('branch_name')}' does not match '{branch_name}'"
        )
    if not _same_realpath(str(lock.get("worktree_path") or ""), worktree_path):
        reasons.append(
            f"lock worktree '{lock.get('worktree_path')}' does not match "
            f"'{worktree_path}'"
        )
    lease = lock.get("work_lease") if isinstance(lock, dict) else None
    claimant = lease.get("claimant") if isinstance(lease, dict) else None
    claimant = claimant if isinstance(claimant, dict) else {}
    recorded_identity = str(claimant.get("username") or "").strip()
    recorded_profile = str(claimant.get("profile") or "").strip()
    if not recorded_identity or not recorded_profile:
        reasons.append("lock does not record both a claimant username and profile")
    if recorded_identity and recorded_identity != str(identity or "").strip():
        reasons.append(
            f"lock claimant '{recorded_identity}' does not match active identity "
            f"'{str(identity or '').strip() or 'unknown'}'"
        )
    if recorded_profile and recorded_profile != str(profile or "").strip():
        reasons.append(
            f"lock profile '{recorded_profile}' does not match active profile "
            f"'{str(profile or '').strip() or 'unknown'}'"
        )
    return reasons
 def _refusal(reasons: list[str], **extra: Any) -> dict[str, Any]:
    return {"success": False, "performed": False, "reasons": reasons, **extra}
 def heartbeat_session_lock(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    identity: str | None,
    profile: str | None,
    task_session_id: str,
    expected_generation: int | None = None,
    lock_dir: str | None = None,
    now: datetime | None = None,
 ) -> dict[str, Any]:
    """Slide a heartbeat-lifecycle lease forward (#790 Slice A, A4).
    The write happens inside the same per-issue ``flock`` that serializes
    acquisition, and under the #772 generation compare-and-swap, so a heartbeat
    can never race a concurrent reclaim: whichever lands first moves the
    generation and the other fails closed.
    Refuses — never revives — in every ambiguous case. A lease that has already
    lapsed past its grace is *not* heartbeatable: allowing that would let a
    session that stopped proving liveness restore ownership retroactively, which
    is precisely the revival AC-N5 forbids. Such a session must go through the
    sanctioned reclaim path, which mints a fresh generation.
    """
    current = _lease_now(now)
    root = _ensure_lock_dir(lock_dir)
    path = lock_file_path(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
    )
    declared_session = str(task_session_id or "").strip()
    if not declared_session:
        return _refusal(["no task_session_id supplied (fail closed)"])
    sentinel = flock_path(path)
    try:
        with _exclusive_file_lock(sentinel):
            lock = read_lock_file(path)
            if not lock:
                return _refusal([f"no durable lock for issue #{issue_number}"])
            if is_legacy_lease(lock):
                return _refusal(
                    [
                        "lock predates the heartbeat lifecycle; it must be rebound "
                        "by its exact owner before it can be heartbeated"
                    ],
                    lifecycle=lease_lifecycle_version(lock),
                    legacy_lease=True,
                )
            reasons = _ownership_refusals(
                lock,
                issue_number=issue_number,
                branch_name=branch_name,
                worktree_path=worktree_path,
                identity=identity,
                profile=profile,
            )
            recorded_session = lease_task_session_id(lock)
            if not recorded_session:
                reasons.append(
                    "lock declares the heartbeat lifecycle but records no "
                    "task_session_id (fail closed)"
                )
            elif recorded_session != declared_session:
                # A superseded session holding an old identifier cannot heartbeat
                # over the session that replaced it.
                reasons.append(
                    "task_session_id does not match the session recorded on the lock"
                )
            if reasons:
                return _refusal(reasons)
            current_generation = lock_generation(lock)
            if (
                expected_generation is not None
                and current_generation != expected_generation
            ):
                return _refusal(
                    [
                        f"lock generation changed: expected {expected_generation}, "
                        f"found {current_generation}; another session reclaimed or "
                        "replaced this claim (fail closed)"
                    ],
                    lock_generation=current_generation,
                )
            freshness = assess_lock_freshness(lock, now=current)
            if not freshness.get("live"):
                return _refusal(
                    [
                        f"lease is not live ({freshness.get('status')}): "
                        f"{freshness.get('reason')}; a lapsed lease must be "
                        "reclaimed, not heartbeated"
                    ],
                    freshness=freshness,
                )
            policy = lease_policy.policy_for(lease_task_class(lock))
            expires = current + timedelta(minutes=policy.initial_ttl_minutes)
            record = dict(lock)
            lease = dict(record.get("work_lease") or {})
            prior_heartbeat = lease.get("last_heartbeat_at")
            lease["last_heartbeat_at"] = _format_lease_timestamp(current)
            lease["expires_at"] = _format_lease_timestamp(expires)
            try:
                lease["heartbeat_count"] = int(lease.get("heartbeat_count") or 0) + 1
            except (TypeError, ValueError):
                lease["heartbeat_count"] = 1
            record["work_lease"] = lease
            record["lock_generation"] = current_generation + 1
            save_lock_file(path, record)
    except LockContentionError as exc:
        return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])
    return {
        "success": True,
        "performed": True,
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "task_session_id": declared_session,
        "lock_generation": record["lock_generation"],
        "prior_generation": current_generation,
        "prior_heartbeat_at": prior_heartbeat,
        "last_heartbeat_at": lease["last_heartbeat_at"],
        "expires_at": lease["expires_at"],
        "heartbeat_count": lease["heartbeat_count"],
        "lock_file_path": path,
        "policy": lease_policy.describe(lease_task_class(record)),
        "freshness": assess_lock_freshness(record, now=current),
    }
 def rebind_legacy_lock(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    identity: str | None,
    profile: str | None,
    expected_generation: int | None = None,
    lock_dir: str | None = None,
    now: datetime | None = None,
 ) -> dict[str, Any]:
    """Move a legacy lock into the heartbeat lifecycle (#790 AC-N8).
    One of the two sanctioned exits from the preserved-expiry legacy state; the
    other is terminal retirement, which is Slice B. Only the exact recorded
    owner may rebind, and only while the legacy lock is still live under its
    original absolute expiry — an already-expired legacy lease belongs to the
    #760 renewal path or #601 reclaim, and this must not become a second, weaker
    way to revive one.
    The rebind mints a genuine task-session identifier and a genuine first
    heartbeat. It does not fabricate history: the original creation and expiry
    are preserved under ``legacy_origin`` for audit, and the new lifecycle's
    absolute cap runs from the rebind, not from the legacy claim.
    """
    current = _lease_now(now)
    root = _ensure_lock_dir(lock_dir)
    path = lock_file_path(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
    )
    sentinel = flock_path(path)
    try:
        with _exclusive_file_lock(sentinel):
            lock = read_lock_file(path)
            if not lock:
                return _refusal([f"no durable lock for issue #{issue_number}"])
            if not is_legacy_lease(lock):
                return _refusal(
                    [
                        "lock is already on the heartbeat lifecycle; use the "
                        "heartbeat path"
                    ],
                    lifecycle=lease_lifecycle_version(lock),
                    legacy_lease=False,
                )
            reasons = _ownership_refusals(
                lock,
                issue_number=issue_number,
                branch_name=branch_name,
                worktree_path=worktree_path,
                identity=identity,
                profile=profile,
            )
            if reasons:
                return _refusal(reasons)
            current_generation = lock_generation(lock)
            if (
                expected_generation is not None
                and current_generation != expected_generation
            ):
                return _refusal(
                    [
                        f"lock generation changed: expected {expected_generation}, "
                        f"found {current_generation} (fail closed)"
                    ],
                    lock_generation=current_generation,
                )
            freshness = assess_lock_freshness(lock, now=current)
            if not freshness.get("live"):
                return _refusal(
                    [
                        f"legacy lease is not live ({freshness.get('status')}): "
                        f"{freshness.get('reason')}; rebinding is not a recovery "
                        "path for a lapsed lease"
                    ],
                    freshness=freshness,
                )
            policy = lease_policy.policy_for(lease_task_class(lock))
            expires = current + timedelta(minutes=policy.initial_ttl_minutes)
            session_id = mint_task_session_id(lease_task_class(lock))
            record = dict(lock)
            lease = dict(record.get("work_lease") or {})
            legacy_origin = {
                "created_at": lease.get("created_at"),
                "expires_at": lease.get("expires_at"),
                "last_heartbeat_at": lease.get("last_heartbeat_at"),
                "lifecycle": lease_policy.LIFECYCLE_LEGACY,
            }
            lease["lifecycle_version"] = lease_policy.LIFECYCLE_HEARTBEAT_V1
            lease["task_session_id"] = session_id
            lease["created_at"] = _format_lease_timestamp(current)
            lease["last_heartbeat_at"] = _format_lease_timestamp(current)
            lease["expires_at"] = _format_lease_timestamp(expires)
            lease["heartbeat_count"] = 1
            record["work_lease"] = lease
            record["legacy_rebind"] = {
                "rebound_at": _format_lease_timestamp(current),
                "task_session_id": session_id,
                "prior_generation": current_generation,
                "legacy_origin": legacy_origin,
                "reason": (
                    "legacy lock rebound into the heartbeat lifecycle by its exact "
                    "recorded owner"
                ),
            }
            record["lock_generation"] = current_generation + 1
            save_lock_file(path, record)
    except LockContentionError as exc:
        return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])
    return {
        "success": True,
        "performed": True,
        "issue_number": issue_number,
        "task_session_id": session_id,
        "lock_generation": record["lock_generation"],
        "prior_generation": current_generation,
        "lifecycle": lease_policy.LIFECYCLE_HEARTBEAT_V1,
        "legacy_rebind": record["legacy_rebind"],
        "expires_at": lease["expires_at"],
        "last_heartbeat_at": lease["last_heartbeat_at"],
        "lock_file_path": path,
        "freshness": assess_lock_freshness(record, now=current),
    }
 def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
    root = (lock_dir or default_lock_dir()).strip()
    pointer = read_lock_file(session_pointer_path(root))
@@ -336,6 +673,16 @@ def _parse_lease_timestamp(value: str | None) -> datetime | None:
        return None
 def _format_lease_timestamp(value: datetime) -> str:
    """Serialize a lease timestamp in the durable ``...Z`` form already on disk."""
    return (
        value.astimezone(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )
 def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
    if not lock:
        return None
@@ -356,60 +703,216 @@ def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -
    return assess_lock_freshness(lock, now=now)["live"]
 def lease_task_class(lock_data: dict[str, Any] | None) -> str:
    """Policy task class for a durable lock; author work when unrecorded."""
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    if isinstance(lease, dict):
        recorded = str(lease.get("operation_type") or "").strip()
        if recorded:
            return recorded
    return AUTHOR_ISSUE_WORK_LEASE
 def lease_lifecycle_version(lock_data: dict[str, Any] | None) -> str:
    """Read the durable lifecycle marker (#790 AC-N8).
    The marker is the *only* discriminator between a heartbeat-lifecycle lease
    and a legacy one. Timestamps are deliberately not consulted: a lock minted
    before this lifecycle existed has ``last_heartbeat_at == created_at``
    forever, and reading that equality as "recently heartbeated" would treat
    every never-heartbeated legacy lock as fresh — the precise inversion AC-N8
    forbids. A newly minted heartbeat lease also has the two equal, so the
    equality carries no information in either direction.
    """
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    if isinstance(lease, dict):
        recorded = str(lease.get("lifecycle_version") or "").strip()
        if recorded:
            return recorded
    return lease_policy.LIFECYCLE_LEGACY
 def is_legacy_lease(lock_data: dict[str, Any] | None) -> bool:
    """True when a lock predates the shared heartbeat lifecycle."""
    return lease_lifecycle_version(lock_data) != lease_policy.LIFECYCLE_HEARTBEAT_V1
 def lease_task_session_id(lock_data: dict[str, Any] | None) -> str:
    """Recorded per-task session identifier, or empty for a legacy lock."""
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    if isinstance(lease, dict):
        return str(lease.get("task_session_id") or "").strip()
    return ""
 def mint_task_session_id(task_class: str = AUTHOR_ISSUE_WORK_LEASE) -> str:
    """Mint an ownership key for one task (#790 AC-N1).
    Deliberately contains no process identifier. The recorded PID belongs to the
    long-lived MCP daemon, which outlives any individual task and is reused by
    every task it serves, so PID digits cannot identify *which* task holds a
    claim. The PID is still recorded alongside this value as evidence.
    """
    prefix = _sanitize_segment(str(task_class or AUTHOR_ISSUE_WORK_LEASE))
    return f"{prefix}-{uuid.uuid4().hex[:16]}"
 def _lease_heartbeat_at(lock_data: dict[str, Any] | None) -> datetime | None:
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    heartbeat_at = None
    if isinstance(lock_data, dict):
        heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
    if heartbeat_at is None and isinstance(lease, dict):
        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
    return heartbeat_at
 def assess_lock_freshness(
    lock_data: dict[str, Any] | None,
    *,
    now: datetime | None = None,
 ) -> dict[str, Any]:
-    """Classify a lock as live, expired, stale, or absent."""
+    """Classify a lock as live, expired, stale, or absent.
    #790 Slice A makes the heartbeat load-bearing. Before this change
    ``last_heartbeat_at`` was parsed and then never consulted: liveness was
    decided entirely by the absolute ``expires_at`` and by PID liveness, and
    since the recorded PID is the long-lived MCP daemon, an abandoned author
    task stayed "live" for the full four-hour TTL.
    Two rules govern the rewrite:
    * **An alive PID never establishes freshness** (AC-N2). It proves the daemon
      is up, nothing about the task. It is recorded as evidence and no branch
      returns ``live`` because of it.
    * **A dead PID still corroborates staleness.** The dead-PID band is
      unchanged and still precedes every heartbeat evaluation, so #753
      dead-session recovery keys on exactly the classification it always did.
    Legacy leases (AC-N8) keep their recorded absolute expiry and are never
    evaluated against the short heartbeat grace, so deploying this change cannot
    make an existing claim instantly reclaimable.
    """
    current = _lease_now(now)
    if not lock_data:
        return {
-            "status": "absent",
+            "status": STATUS_ABSENT,
            "live": False,
            "stale": False,
            "reason": "no lock record",
        }
    expires_at = lease_expires_at(lock_data)
    lease = lock_data.get("work_lease")
-    heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
+    expires_at = lease_expires_at(lock_data)
-    if heartbeat_at is None and isinstance(lease, dict):
+    heartbeat_at = _lease_heartbeat_at(lock_data)
-        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
+    created_at = (
        _parse_lease_timestamp(lease.get("created_at"))
        if isinstance(lease, dict)
        else None
    )
    pid = lock_data.get("session_pid")
    if pid is None:
        pid = lock_data.get("pid")
    # Evidence only. Never consulted to grant liveness (AC-N2).
    pid_alive = is_process_alive(pid) if pid is not None else False
-    if expires_at and expires_at <= current:
+    lifecycle = lease_lifecycle_version(lock_data)
-        return {
+    legacy = lifecycle != lease_policy.LIFECYCLE_HEARTBEAT_V1
-            "status": "expired",
+    policy = lease_policy.policy_for(lease_task_class(lock_data))
            "live": False,
            "stale": True,
            "reason": f"lease expired at {expires_at.isoformat()}",
            "pid_alive": pid_alive,
        }
-    if pid is not None and not pid_alive:
+    evidence: dict[str, Any] = {
        return {
            "status": "stale",
            "live": False,
            "stale": True,
            "reason": f"owner pid {pid} is not alive",
            "pid_alive": False,
        }
    return {
        "status": "live",
        "live": True,
        "stale": False,
        "reason": "lock heartbeat and lease are fresh",
        "pid_alive": pid_alive,
        "lifecycle": lifecycle,
        "legacy_lease": legacy,
        "task_session_id": lease_task_session_id(lock_data) or None,
        "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
        "expires_at": expires_at.isoformat() if expires_at else None,
    }
    def _result(status: str, *, live: bool, reason: str, **extra: Any) -> dict[str, Any]:
        return {
            "status": status,
            "live": live,
            "stale": not live and status != STATUS_ABSENT,
            "reason": reason,
            **evidence,
            **extra,
        }
    if legacy:
        # AC-N8: the preserved absolute expiry is the only clock for a lock
        # written before task-session heartbeats existed.
        if expires_at and expires_at <= current:
            return _result(
                STATUS_EXPIRED,
                live=False,
                reason=f"lease expired at {expires_at.isoformat()}",
            )
        if pid is not None and not pid_alive:
            return _result(
                STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive"
            )
        return _result(
            STATUS_LIVE,
            live=True,
            reason=(
                "legacy lease is within its recorded absolute expiry; the "
                "heartbeat grace does not apply retroactively"
            ),
            legacy_expiry_preserved=True,
        )
    # ── Heartbeat lifecycle ──
    if pid is not None and not pid_alive:
        # Unchanged dead-PID band: #753 recovery depends on this exact status.
        return _result(STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive")
    if heartbeat_at is None:
        # Contradictory: a heartbeat lease must carry a heartbeat. Fail closed.
        return _result(
            STATUS_STALE_MISSED_HEARTBEAT,
            live=False,
            reason=(
                f"lease declares lifecycle '{lifecycle}' but records no "
                "last_heartbeat_at (fail closed)"
            ),
        )
    if policy.absolute_cap_hours and created_at is not None:
        cap_at = created_at + timedelta(hours=policy.absolute_cap_hours)
        if cap_at <= current:
            return _result(
                STATUS_STALE_ABSOLUTE_CAP,
                live=False,
                reason=(
                    f"lease exceeded its {policy.absolute_cap_hours}h absolute cap "
                    f"at {cap_at.isoformat()}; canonical re-adoption is required"
                ),
                absolute_cap_at=cap_at.isoformat(),
            )
    grace_at = heartbeat_at + timedelta(minutes=policy.missed_heartbeat_grace_minutes)
    if grace_at <= current or (expires_at is not None and expires_at <= current):
        return _result(
            STATUS_STALE_MISSED_HEARTBEAT,
            live=False,
            reason=(
                f"no valid heartbeat since {heartbeat_at.isoformat()}; the "
                f"{policy.missed_heartbeat_grace_minutes}min grace lapsed at "
                f"{grace_at.isoformat()}"
            ),
            missed_heartbeat_since=grace_at.isoformat(),
        )
    warning_at = heartbeat_at + timedelta(minutes=policy.stale_warning_minutes)
    return _result(
        STATUS_LIVE,
        live=True,
        reason="lease heartbeat is fresh within the configured grace",
        heartbeat_warning=warning_at <= current,
    )
 def _same_realpath(left: str | None, right: str | None) -> bool:
    if not left or not right:
@@ -446,6 +949,27 @@ def assess_expired_lock_reclaim(
            "reasons": ["lock is still live; cannot reclaim (fail closed)"],
            "freshness": freshness,
        }
    status = str(freshness.get("status") or "")
    if status in (STATUS_STALE_MISSED_HEARTBEAT, STATUS_STALE_ABSOLUTE_CAP):
        # #790: under the heartbeat lifecycle the heartbeat *is* the liveness
        # proof, so a session that stopped heartbeating past its grace has
        # released its claim by definition. Requiring a dead PID on top of that
        # would reinstate the original defect — the recorded PID is the shared
        # daemon, which stays alive across every abandoned task it ever served.
        #
        # This band is unreachable for a legacy lease (AC-N8), so no lock
        # written before this lifecycle can be reclaimed by this path.
        return {
            "reclaim_allowed": True,
            "reasons": [
                f"heartbeat-lifecycle lease is {status}: {freshness.get('reason')}"
            ],
            "freshness": freshness,
            "prior_branch": existing_lock.get("branch_name"),
            "prior_worktree": existing_lock.get("worktree_path"),
            "prior_pid": existing_lock.get("session_pid") or existing_lock.get("pid"),
            "prior_task_session_id": lease_task_session_id(existing_lock) or None,
        }
    pid = existing_lock.get("session_pid")
    if pid is None:
        pid = existing_lock.get("pid")
@@ -0,0 +1,212 @@
 """Central lease policy configuration (#790 Slice A, AC-N7).
 The single authoritative source for every lease duration in the project. Before
 this module the numbers were scattered: a four-hour author TTL was declared
 twice (``issue_lock_store`` and ``gitea_mcp_server``), the reviewer/merger
 sliding window lived in ``reviewer_pr_lease``, the conflict-fix window in
 ``pr_work_lease``, and the control-plane default in ``control_plane_db``.
 Nothing tied them together, so tuning one class silently diverged from the
 others and no reader could answer "how long does a lease live?" without
 grepping four files.
 AC-N7 requires that this configuration exist *before* the first heartbeat and
 TTL behavior that reads from it, so it ships in Slice A rather than trailing the
 code it governs.
 Deliberate boundaries:
 * **Declaration is not rewiring.** Every task class is declared here, but only
  those with ``heartbeat_lifecycle_active`` were migrated onto the shared
  heartbeat lifecycle in Slice A — currently ``author_issue_work`` alone.
  Reviewer, merger, and conflict-fix leases keep their own existing behavior
  until Slice C moves them; their numbers are recorded here so the two cannot
  drift apart unnoticed, and ``tests/test_issue_790_lease_policy.py`` asserts
  the recorded values still equal the constants those modules use.
 * **No policy decision lives here.** This module answers "how long", never "may
  this session proceed". Freshness, reclaim, and renewal dispositions stay in
  ``issue_lock_store``.
 """
 from __future__ import annotations
 import os
 from dataclasses import dataclass
 from typing import Any
 # Task classes. Only the first is migrated onto the shared lifecycle in Slice A.
 TASK_CLASS_AUTHOR_ISSUE_WORK = "author_issue_work"
 TASK_CLASS_REVIEWER_PR = "reviewer_pr"
 TASK_CLASS_MERGER_PR = "merger_pr"
 TASK_CLASS_CONFLICT_FIX = "conflict_fix"
 # Durable marker for a lease minted under the shared heartbeat lifecycle.
 #
 # #790 AC-N8: this explicit marker — never a timestamp comparison — is what
 # distinguishes a heartbeat-lifecycle lease from a legacy one. A lock written
 # before this lifecycle existed carries no marker and reads as
 # ``LIFECYCLE_LEGACY``.
 LIFECYCLE_HEARTBEAT_V1 = "heartbeat-v1"
 LIFECYCLE_LEGACY = "legacy"
 _ENV_PREFIX = "GITEA_LEASE_POLICY"
@dataclass(frozen=True)
 class LeasePolicy:
    """Durations governing one task class.
    All intervals are minutes except ``absolute_cap_hours``. ``None`` for the
    cap means the class has no maximum continuous duration.
    """
    task_class: str
    initial_ttl_minutes: float
    heartbeat_cadence_minutes: float
    stale_warning_minutes: float
    missed_heartbeat_grace_minutes: float
    absolute_cap_hours: float | None
    recovery_grace_minutes: float
    terminal_race_drain_minutes: float
    terminal_retirement_eligible: bool
    heartbeat_lifecycle_active: bool
 # Defaults. ``author_issue_work`` adopts the reviewer window proven by #747
 # rather than inventing new numbers: a lease expires 10 minutes after its last
 # valid heartbeat, warns at half that, and an actively heartbeating session is
 # never evicted. The prior value was a fixed four hours (240 minutes) that no
 # heartbeat could shorten — the defect this issue exists to correct.
 _DEFAULTS: dict[str, LeasePolicy] = {
    TASK_CLASS_AUTHOR_ISSUE_WORK: LeasePolicy(
        task_class=TASK_CLASS_AUTHOR_ISSUE_WORK,
        initial_ttl_minutes=10.0,
        heartbeat_cadence_minutes=2.0,
        stale_warning_minutes=5.0,
        missed_heartbeat_grace_minutes=10.0,
        absolute_cap_hours=8.0,
        recovery_grace_minutes=10.0,
        terminal_race_drain_minutes=2.0,
        terminal_retirement_eligible=True,
        heartbeat_lifecycle_active=True,
    ),
    # Declared, not rewired. These mirror reviewer_pr_lease.LEASE_TTL_MINUTES
    # and STALE_WARNING_MINUTES; Slice C migrates the call sites.
    TASK_CLASS_REVIEWER_PR: LeasePolicy(
        task_class=TASK_CLASS_REVIEWER_PR,
        initial_ttl_minutes=10.0,
        heartbeat_cadence_minutes=2.0,
        stale_warning_minutes=5.0,
        missed_heartbeat_grace_minutes=10.0,
        absolute_cap_hours=None,
        recovery_grace_minutes=10.0,
        terminal_race_drain_minutes=2.0,
        terminal_retirement_eligible=False,
        heartbeat_lifecycle_active=False,
    ),
    TASK_CLASS_MERGER_PR: LeasePolicy(
        task_class=TASK_CLASS_MERGER_PR,
        initial_ttl_minutes=10.0,
        heartbeat_cadence_minutes=2.0,
        stale_warning_minutes=5.0,
        missed_heartbeat_grace_minutes=10.0,
        absolute_cap_hours=None,
        recovery_grace_minutes=10.0,
        terminal_race_drain_minutes=2.0,
        terminal_retirement_eligible=False,
        heartbeat_lifecycle_active=False,
    ),
    # Mirrors pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES. Deliberately left
    # at its current window; shortening it is Slice C's call, not this slice's.
    TASK_CLASS_CONFLICT_FIX: LeasePolicy(
        task_class=TASK_CLASS_CONFLICT_FIX,
        initial_ttl_minutes=120.0,
        heartbeat_cadence_minutes=2.0,
        stale_warning_minutes=5.0,
        missed_heartbeat_grace_minutes=10.0,
        absolute_cap_hours=None,
        recovery_grace_minutes=10.0,
        terminal_race_drain_minutes=2.0,
        terminal_retirement_eligible=False,
        heartbeat_lifecycle_active=False,
    ),
 }
 _NUMERIC_FIELDS = (
    "initial_ttl_minutes",
    "heartbeat_cadence_minutes",
    "stale_warning_minutes",
    "missed_heartbeat_grace_minutes",
    "absolute_cap_hours",
    "recovery_grace_minutes",
    "terminal_race_drain_minutes",
 )
 def env_var_name(task_class: str, field: str) -> str:
    """Environment variable that overrides one field of one task class."""
    return f"{_ENV_PREFIX}_{task_class.upper()}_{field.upper()}"
 def _override(task_class: str, field: str, default: float | None) -> float | None:
    """Read one override, falling back to *default* on anything unusable.
    A malformed or non-positive override is ignored rather than raised: a typo
    in an environment variable must not be able to mint a zero-length lease that
    makes every claim instantly reclaimable, nor crash the server at import.
    """
    raw = (os.environ.get(env_var_name(task_class, field)) or "").strip()
    if not raw:
        return default
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return default
    if value <= 0:
        return default
    return value
 def policy_for(task_class: str) -> LeasePolicy:
    """Return the effective policy for *task_class*.
    Unknown task classes fall back to the author policy, which is the most
    conservative migrated class, rather than raising — a new caller must never
    be able to crash a lock write by naming a class this table has not learned.
    """
    key = str(task_class or "").strip() or TASK_CLASS_AUTHOR_ISSUE_WORK
    base = _DEFAULTS.get(key) or _DEFAULTS[TASK_CLASS_AUTHOR_ISSUE_WORK]
    resolved = {
        field: _override(base.task_class, field, getattr(base, field))
        for field in _NUMERIC_FIELDS
    }
    if all(resolved[field] == getattr(base, field) for field in _NUMERIC_FIELDS):
        return base
    return LeasePolicy(
        task_class=base.task_class,
        terminal_retirement_eligible=base.terminal_retirement_eligible,
        heartbeat_lifecycle_active=base.heartbeat_lifecycle_active,
        **resolved,
    )
 def known_task_classes() -> tuple[str, ...]:
    """Every declared task class, migrated or not."""
    return tuple(_DEFAULTS)
 def describe(task_class: str) -> dict[str, Any]:
    """Serializable view of a policy, for audit records and tool payloads."""
    policy = policy_for(task_class)
    return {
        "task_class": policy.task_class,
        "initial_ttl_minutes": policy.initial_ttl_minutes,
        "heartbeat_cadence_minutes": policy.heartbeat_cadence_minutes,
        "stale_warning_minutes": policy.stale_warning_minutes,
        "missed_heartbeat_grace_minutes": policy.missed_heartbeat_grace_minutes,
        "absolute_cap_hours": policy.absolute_cap_hours,
        "recovery_grace_minutes": policy.recovery_grace_minutes,
        "terminal_race_drain_minutes": policy.terminal_race_drain_minutes,
        "terminal_retirement_eligible": policy.terminal_retirement_eligible,
        "heartbeat_lifecycle_active": policy.heartbeat_lifecycle_active,
        "lifecycle_version": LIFECYCLE_HEARTBEAT_V1,
    }
@@ -24,50 +24,12 @@ from __future__ import annotations
 import os
 import subprocess
 import time
 # Live-remote head cache: the parity gate runs on every mutation and every
 # runtime-context read, so the ``git ls-remote`` result is cached briefly to
 # avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
 _REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
 _REMOTE_HEAD_TTL = 60.0
 # When True, ``read_remote_master_head`` never performs ``git ls-remote`` unless
 # ``GITEA_TEST_LIVE_REMOTE_HEAD`` is set. Conftest enables this suite-wide so
 # feature worktrees (whose HEAD differs from live master) cannot flip legacy
 # runtime-context assertions to live_stale, and so unit tests never depend on
 # a live network (PR #788 F1/F2 / issue #610). Module-level (not env-only) so
 # ``patch.dict(os.environ, …, clear=True)`` cannot re-enable the probe.
 _HERMETIC_TEST_MODE: bool = False
 def _clear_remote_head_cache() -> None:
    """Reset the live-remote head cache (test isolation / forced refresh)."""
    _REMOTE_HEAD_CACHE.clear()
 def set_hermetic_test_mode(enabled: bool) -> None:
    """Enable or disable suite-wide hermetic live-remote reads (tests only)."""
    global _HERMETIC_TEST_MODE
    _HERMETIC_TEST_MODE = bool(enabled)
    _clear_remote_head_cache()
 def hermetic_test_mode() -> bool:
    """Return whether hermetic live-remote reads are active."""
    return bool(_HERMETIC_TEST_MODE)
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
 # GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
 ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
 # GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE -> opt a single test into a real ls-remote
 # even when hermetic mode is on (rare; prefer ENV_TEST_LIVE_REMOTE_HEAD).
 ENV_TEST_ALLOW_LIVE_REMOTE_PROBE = "GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE"
 def read_git_head(root: str) -> str | None:
@@ -96,75 +58,6 @@ def read_git_head(root: str) -> str | None:
    return (res.stdout or "").strip() or None
 def read_remote_master_head(
    root: str,
    remote: str = "origin",
    branch: str = "master",
    ttl: float = _REMOTE_HEAD_TTL,
 ) -> str | None:
    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
    a daemon that is behind the live remote master apart from one whose local
    checkout simply hasn't been pulled. ``None`` means the live head could not
    be resolved (offline, no such remote, git unavailable, error) -- callers
    must treat unknown live state as *not mutation-safe* while never blocking
    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
    precedence so the wiring can be exercised deterministically and offline.
    The result is cached for *ttl* seconds per (root, remote, branch) so the
    gate does not run a network probe on every mutation/read (``ttl=0`` forces
    a live probe). Both hits and ``None`` misses are cached to bound offline
    latency; the env override bypasses the cache and the subprocess entirely.
    Under suite hermetic mode (``set_hermetic_test_mode(True)``, set by
    conftest) a missing override returns ``None`` without network I/O so
    feature-worktree test runs cannot observe live_stale against real master
    (PR #788 F1) and unit tests stay offline (F2). Opt out with an explicit
    ``GITEA_TEST_LIVE_REMOTE_HEAD`` pin or ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
    """
    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
    if forced is not None:
        return forced.strip() or None
    if _HERMETIC_TEST_MODE and not (
        os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
    ).strip():
        # Hermetic default: live head unknown. live_stale stays False;
        # mutation_safe is False when live is unknown (documented #610 note).
        return None
    # Defense in depth: even without the module flag, never probe while pytest
    # is running unless the test opted into a real probe or set an override.
    if (os.environ.get("PYTEST_CURRENT_TEST") or "").strip() and not (
        os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
    ).strip():
        return None
    if not root:
        return None
    key = (root, remote, branch)
    now = time.monotonic()
    if ttl > 0:
        cached = _REMOTE_HEAD_CACHE.get(key)
        if cached is not None and (now - cached[0]) < ttl:
            return cached[1]
    sha: str | None = None
    try:
        res = subprocess.run(
            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5,
        )
        if res.returncode == 0:
            lines = (res.stdout or "").strip().splitlines()
            if lines:
                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
    except Exception:
        sha = None
    _REMOTE_HEAD_CACHE[key] = (now, sha)
    return sha
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
    """Capture the process source-tree baseline once at server startup.
@@ -179,38 +72,18 @@ def _short(sha: str | None) -> str:
    return sha[:12] if sha else "unknown"
-def assess_master_parity(
+def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
    startup: dict | None,
    current_head: str | None,
    live_remote_head: str | None = None,
 ) -> dict:
    """Compare the startup baseline against the current on-disk ``HEAD``.
-    Pure: all HEADs are supplied by the caller. Returns a structured result:
+    Pure: both HEADs are supplied by the caller. Returns a structured result:
    - ``in_parity``    -- server code matches the on-disk master (or parity
      could not be determined, which is not treated as stale).
    - ``stale``        -- the on-disk master has definitively advanced past the
      running process.
-    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
+    - ``restart_required`` -- alias of ``stale``; the recovery action.
-    - ``determinable`` -- whether both local HEADs were known well enough to
+    - ``determinable`` -- whether both HEADs were known well enough to compare.
      compare.
    - ``startup_head`` / ``current_head`` / ``reasons``.
    #610 adds live-remote awareness so a daemon that is stale relative to the
    *live* remote master cannot report a mutation-safe result even when the
    local checkout HEAD still matches the daemon's startup commit:
    - ``daemon_start_head`` -- the commit the running process started at
      (alias of ``startup_head``, named for clarity in reports).
    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
      could not be fetched.
    - ``live_known`` -- whether the live remote target was resolved.
    - ``live_stale`` -- the live remote master has advanced past the running
      process (daemon is behind live master) even if local parity is green.
    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
      target all agree; the only state in which a mutation may rely on parity.
    """
    startup_head = (startup or {}).get("startup_head")
    reasons: list[str] = []
@@ -218,56 +91,32 @@ def assess_master_parity(
    if startup_head is None:
        reasons.append(
            "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head,
+        return _result(True, False, False, startup_head, current_head, reasons)
                       live_remote_head, False, reasons)
    if current_head is None:
        reasons.append(
            "current workspace HEAD could not be read; code parity cannot be "
            "enforced")
-        return _result(True, False, False, startup_head, current_head,
+        return _result(True, False, False, startup_head, current_head, reasons)
                       live_remote_head, False, reasons)
-    local_in_parity = startup_head == current_head
+    if startup_head == current_head:
-    local_stale = not local_in_parity
+        return _result(True, False, True, startup_head, current_head, reasons)
    if local_stale:
        reasons.append(
            f"MCP server started at commit {_short(startup_head)} but the "
            f"workspace master is now {_short(current_head)}; restart the "
            f"server to load the current capability gates")
-    live_known = live_remote_head is not None
+    reasons.append(
-    live_stale = live_known and live_remote_head != startup_head
+        f"MCP server started at commit {_short(startup_head)} but the workspace "
-    if live_stale:
+        f"master is now {_short(current_head)}; restart the server to load the "
-        reasons.append(
+        f"current capability gates")
-            f"live remote master is {_short(live_remote_head)} but the MCP "
+    return _result(False, True, True, startup_head, current_head, reasons)
            f"server started at {_short(startup_head)}; the daemon is stale "
            f"relative to live master -- restart/reconnect before mutating")
    return _result(
        local_in_parity, local_stale, True, startup_head, current_head,
        live_remote_head, live_stale, reasons)
-def _result(in_parity, stale, determinable, startup_head, current_head,
+def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
            live_remote_head, live_stale, reasons):
    live_known = live_remote_head is not None
    mutation_safe = (
        determinable and in_parity and live_known and not live_stale)
    return {
        "in_parity": in_parity,
        "stale": stale,
-        "restart_required": stale or live_stale,
+        "restart_required": stale,
        "determinable": determinable,
        "startup_head": startup_head,
        "current_head": current_head,
        # #610 distinguished signals:
        "daemon_start_head": startup_head,
        "local_head": current_head,
        "live_remote_head": live_remote_head,
        "live_known": live_known,
        "live_stale": live_stale,
        "mutation_safe": mutation_safe,
        "reasons": list(reasons),
    }
@@ -281,13 +130,11 @@ def parity_block_reasons(assessment: dict) -> list[str]:
    """Block reasons for a mutation gate (empty when the mutation may proceed).
    A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons. A definitively stale server blocks, and (#610) a daemon that is
+    reasons; only a definitively stale server blocks.
    stale relative to the *live* remote master blocks even when the local
    checkout HEAD still matches the daemon's startup commit.
    """
    if gate_disabled():
        return []
-    if assessment.get("stale") or assessment.get("live_stale"):
+    if assessment.get("stale"):
        return list(assessment.get("reasons") or
                    ["server code is stale relative to master (fail closed)"])
    return []
@@ -300,10 +147,6 @@ def parity_report(assessment: dict) -> dict:
        "restart_required": True,
        "startup_head": assessment.get("startup_head"),
        "current_head": assessment.get("current_head"),
        # #610: name the live remote target so the report distinguishes a
        # local-code stale from a daemon-behind-live-master stale.
        "live_remote_head": assessment.get("live_remote_head"),
        "live_stale": bool(assessment.get("live_stale")),
        "reasons": list(assessment.get("reasons") or []),
        "recovery": [
            "The running MCP server is executing code older than the current "
@@ -314,45 +157,6 @@ def parity_report(assessment: dict) -> dict:
    }
 def parity_resolver_disagreement(
    assessment: dict,
    resolver_restart_required: bool,
 ) -> dict | None:
    """Typed blocker when the resolver requires restart but parity looks green.
    The capability resolver (``gitea_resolve_task_capability``) detects stale
    runtime authoritatively for mutation safety (#610). When it requires a
    restart, local-only parity must never override it: this returns a typed,
    fail-closed blocker that names the resolver as authoritative. Returns
    ``None`` when the resolver does not require a restart.
    """
    if not resolver_restart_required:
        return None
    parity_optimistic = bool(assessment.get("in_parity")) and not (
        assessment.get("stale") or assessment.get("live_stale"))
    return {
        "kind": "parity_resolver_disagreement",
        "restart_required": True,
        "resolver_authoritative": True,
        "parity_optimistic": parity_optimistic,
        "daemon_start_head": assessment.get("daemon_start_head"),
        "local_head": assessment.get("local_head"),
        "live_remote_head": assessment.get("live_remote_head"),
        "reasons": [
            "The capability resolver requires a restart/reconnect (stale "
            "runtime) but master-parity reported local code as in-parity. "
            "The resolver is authoritative for mutation safety; do not mutate "
            "on local parity alone. Restart/reconnect the Gitea MCP server "
            "and re-verify before mutating.",
        ],
        "recovery": [
            "Trust the resolver: treat this session as stale.",
            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
            "current master and live target state, then re-run preflight.",
        ],
    }
 def format_parity(assessment: dict) -> str:
    """One-line human summary for logs / runtime context."""
    if assessment.get("stale"):
@@ -32,6 +32,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
        "permission": "gitea.issue.comment",
        "role": "author",
    },
    # #790 Slice A: prove an owned author lease is still active. Strictly
    # narrower than lock_issue — it can only slide a lease this exact session
    # already owns, never acquire, take over, or revive one — so it gates on the
    # same authority rather than introducing an operation name that every
    # already-configured author profile would be missing.
    "heartbeat_issue_lock": {
        "permission": "gitea.issue.comment",
        "role": "author",
    },
    "set_issue_labels": {
        "permission": "gitea.issue.comment",
        "role": "author",
@@ -167,35 +167,6 @@ def _reset_mutation_authority(monkeypatch):
 import pytest
@pytest.fixture(autouse=True)
 def _hermetic_live_remote_master_head():
    """#610 / PR #788 F1/F2: keep live-remote parity reads offline in tests.
    ``read_remote_master_head`` would otherwise ``git ls-remote`` whenever
    ``GITEA_TEST_LIVE_REMOTE_HEAD`` is unset. Feature worktrees under
    ``branches/`` always differ from live master, so legacy suites that assert
    runtime-context ``safe_next_action`` flip to live_stale. Module-level
    hermetic mode survives ``patch.dict(os.environ, …, clear=True)``.
    Tests that exercise the real probe path call
    ``master_parity_gate.set_hermetic_test_mode(False)`` and/or set
    ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
    """
    try:
        import master_parity_gate as _mpg
        _mpg.set_hermetic_test_mode(True)
    except Exception:
        _mpg = None
    try:
        yield
    finally:
        if _mpg is not None:
            try:
                _mpg.set_hermetic_test_mode(False)
            except Exception:
                pass
@pytest.fixture(autouse=True)
 def _deterministic_workspace_remotes():
    try:
@@ -0,0 +1,444 @@
 """Task heartbeat through the native MCP author path (#790 Slice A, AC-N6).
 Assessor-level coverage is not sufficient here, and this project has already
 paid for learning that: in review #499 on PR #791 the #760 renewal waiver was
 computed correctly and then *discarded* at two later gates, so every real
 renewal still failed while the unit suite stayed green. AC-N6 exists because of
 that, and requires driving the real tools against a real git repository and a
 real durable lock file, composing the gates in production order.
 These tests therefore call ``gitea_lock_issue`` and
 ``gitea_heartbeat_issue_lock`` themselves and assert on what lands on disk,
 never on an assessor's return value alone.
 """
 from __future__ import annotations
 import os
 import subprocess
 import sys
 import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
 from unittest.mock import patch
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from mutation_profile_fixture import shared_mutation_env  # noqa: E402
 import issue_lock_provenance  # noqa: E402
 import issue_lock_store  # noqa: E402
 import lease_policy  # noqa: E402
 import mcp_server  # noqa: E402
 ISSUE = 9791
 BRANCH = f"fix/issue-{ISSUE}-heartbeat-mcp"
 IDENTITY = "example-user"
 PROFILE = "test-author-prgs"
 ORG = "Scaled-Tech-Consulting"
 REPO = "Gitea-Tools"
 def _ts(moment: datetime) -> str:
    return (
        moment.astimezone(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )
 class _HeartbeatMcpBase(unittest.TestCase):
    """Real git repo plus a real durable lock, driven through the real tools."""
    def setUp(self):
        self.lock_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.lock_dir.cleanup)
        self.repo = tempfile.mkdtemp(prefix="issue790-mcp-")
        self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
        self._init_worktree()
        self.remotes = patch.dict(
            mcp_server.REMOTES,
            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
        )
        self.remotes.start()
        self.addCleanup(patch.stopall)
        mcp_server._IDENTITY_CACHE.clear()
    def _git(self, *args):
        return subprocess.run(
            ["git", "-C", self.repo, *args], capture_output=True, text=True, check=True
        )
    def _init_worktree(self):
        self._git("init", "-q", "-b", "master")
        self._git("config", "user.email", "[email protected]")
        self._git("config", "user.name", "Test")
        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
            fh.write("seed\n")
        self._git("add", "seed.txt")
        self._git("commit", "-q", "-m", "seed")
        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
        # A fresh claim starts base-equivalent, which is the ordinary first-lock
        # shape and exercises assess_issue_lock_worktree on its normal path.
        self._git("checkout", "-q", "-b", BRANCH)
        self.head_sha = self.base_sha
        self.worktree = os.path.realpath(self.repo)
    def _lock_path(self):
        return issue_lock_store.lock_file_path(
            remote="prgs",
            org=ORG,
            repo=REPO,
            issue_number=ISSUE,
            lock_dir=self.lock_dir.name,
        )
    def _tool_env(self):
        env = shared_mutation_env(
            PROFILE, include_example_repo=True, GITEA_ISSUE_LOCK_DIR=self.lock_dir.name
        )
        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
        return env
    def _git_state(self, *, porcelain="", base_equivalent=True):
        return {
            "current_branch": BRANCH,
            "porcelain_status": porcelain,
            "base_equivalent": base_equivalent,
            "head_sha": self.head_sha,
            "inspected_git_root": self.worktree,
            "base_branch": "master",
        }
    def run_lock_issue(
        self,
        *,
        branch_entries=None,
        open_prs=None,
        git_state=None,
        identity=IDENTITY,
        profile=PROFILE,
    ):
        branch_entries = branch_entries if branch_entries is not None else []
        open_prs = open_prs if open_prs is not None else []
        git_state = git_state or self._git_state()
        env = self._tool_env()
        with patch(
            "mcp_server.api_get_all", return_value=list(branch_entries)
        ), patch(
            "mcp_server._list_open_pulls", return_value=list(open_prs)
        ), patch(
            "mcp_server.get_auth_header", return_value="token x"
        ), patch(
            "mcp_server._work_lease_claimant",
            return_value={"username": identity, "profile": profile},
        ), patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value=git_state,
        ), patch(
            "mcp_server.issue_duplicate_context_fetcher",
            side_effect=lambda h, o, r, auth, issue_number: (
                list(open_prs),
                [b.get("name") for b in branch_entries if isinstance(b, dict)],
                {"status": "not_claimed"},
            ),
        ), patch.dict(os.environ, env, clear=True):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_lock_issue(
                issue_number=ISSUE,
                branch_name=BRANCH,
                remote="prgs",
                worktree_path=self.worktree,
            )
    def run_heartbeat(
        self, *, task_session_id, identity=IDENTITY, profile=PROFILE, **kwargs
    ):
        env = self._tool_env()
        with patch(
            "mcp_server._work_lease_claimant",
            return_value={"username": identity, "profile": profile},
        ), patch("mcp_server.get_auth_header", return_value="token x"), patch.dict(
            os.environ, env, clear=True
        ):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_heartbeat_issue_lock(
                issue_number=ISSUE,
                branch_name=kwargs.pop("branch_name", BRANCH),
                task_session_id=task_session_id,
                remote="prgs",
                worktree_path=kwargs.pop("worktree_path", self.worktree),
                **kwargs,
            )
    def write_legacy_lock(self, *, hours_old: float = 3.0, ttl_hours: float = 4.0):
        """A durable lock in the shape the store wrote before this slice."""
        now = datetime.now(timezone.utc)
        claimant = {"username": IDENTITY, "profile": PROFILE}
        created = now - timedelta(hours=hours_old)
        record = {
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "remote": "prgs",
            "org": ORG,
            "repo": REPO,
            "worktree_path": self.worktree,
            "session_pid": os.getpid(),
            "pid": os.getpid(),
            "lock_generation": 1,
            "work_lease": {
                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
                "issue_number": ISSUE,
                "pr_number": None,
                "branch": BRANCH,
                "worktree_path": self.worktree,
                "claimant": claimant,
                "created_at": _ts(created),
                # The legacy signature: never advanced past creation.
                "last_heartbeat_at": _ts(created),
                "expires_at": _ts(created + timedelta(hours=ttl_hours)),
            },
            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
                tool="gitea_lock_issue", claimant=claimant
            ),
        }
        path = self._lock_path()
        record["lock_file_path"] = path
        issue_lock_store.save_lock_file(path, record)
        return record
 class TestLockIssueMintsTheLifecycle(_HeartbeatMcpBase):
    """Durable lock creation and read-back through the real tool."""
    def test_native_lock_writes_the_marker_and_a_task_session_id(self):
        result = self.run_lock_issue()
        self.assertTrue(result["success"], result)
        written = issue_lock_store.read_lock_file(result["lock_file_path"])
        lease = written["work_lease"]
        self.assertEqual(
            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
        )
        self.assertTrue(lease["task_session_id"])
        self.assertFalse(issue_lock_store.is_legacy_lease(written))
        # AC-N1: the ownership key is not the daemon pid, which is recorded
        # separately as evidence.
        self.assertNotIn(str(written["session_pid"]), lease["task_session_id"])
        self.assertEqual(written["session_pid"], os.getpid())
    def test_native_lease_uses_the_policy_window_not_four_hours(self):
        result = self.run_lock_issue()
        lease = result["work_lease"]
        created = datetime.fromisoformat(lease["created_at"].replace("Z", "+00:00"))
        expires = datetime.fromisoformat(lease["expires_at"].replace("Z", "+00:00"))
        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
        self.assertEqual(
            (expires - created).total_seconds() / 60.0, policy.initial_ttl_minutes
        )
    def test_freshness_of_a_new_native_lock_is_live(self):
        result = self.run_lock_issue()
        self.assertEqual(
            result["lock_freshness"]["status"], issue_lock_store.STATUS_LIVE
        )
        self.assertTrue(result["lock_freshness"]["live"])
 class TestHeartbeatThroughTheTool(_HeartbeatMcpBase):
    def _lock_and_session(self):
        result = self.run_lock_issue()
        self.assertTrue(result["success"], result)
        return result, result["work_lease"]["task_session_id"]
    def test_heartbeat_slides_the_lease_and_advances_the_generation(self):
        locked, session = self._lock_and_session()
        before = issue_lock_store.read_lock_file(locked["lock_file_path"])
        beat = self.run_heartbeat(task_session_id=session)
        self.assertTrue(beat["success"], beat)
        self.assertEqual(beat["operation"], "heartbeat")
        after = issue_lock_store.read_lock_file(locked["lock_file_path"])
        self.assertGreater(
            issue_lock_store.lock_generation(after),
            issue_lock_store.lock_generation(before),
        )
        self.assertGreaterEqual(
            after["work_lease"]["expires_at"], before["work_lease"]["expires_at"]
        )
        self.assertEqual(after["work_lease"]["heartbeat_count"], 2)
    def test_heartbeat_evidence_survives_the_downstream_mutation_gate(self):
        """The #499 F2 lesson, applied.
        A sanction that is computed and then discarded downstream is worthless.
        After a heartbeat the lock must still satisfy the gate every author
        mutation runs through.
        """
        locked, session = self._lock_and_session()
        self.run_heartbeat(task_session_id=session)
        written = issue_lock_store.read_lock_file(locked["lock_file_path"])
        verdict = issue_lock_store.verify_lock_for_mutation(
            written,
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
        )
        self.assertTrue(verdict["proven"], verdict)
        self.assertFalse(verdict["block"])
    def _duplicate_gate(self, *, open_prs, branches):
        env = self._tool_env()
        with patch("mcp_server.get_auth_header", return_value="token x"), patch(
            "mcp_server.issue_duplicate_context_fetcher",
            side_effect=lambda h, o, r, auth, issue_number: (
                list(open_prs),
                list(branches),
                {"status": "not_claimed"},
            ),
        ), patch.dict(os.environ, env, clear=True):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_assess_work_issue_duplicate(
                issue_number=ISSUE, branch_name=BRANCH, remote="prgs"
            )
    def test_heartbeat_does_not_change_the_duplicate_gate_verdict(self):
        """The gate must be invariant under heartbeating.
        The point is not that the gate passes — with a linked open PR at the
        lock phase it correctly blocks (#400), heartbeat or not. The property
        that matters is that sliding a lease neither loosens the gate nor
        corrupts the lock state it reads: the verdict before and after a
        heartbeat must be identical, for both the clear and the blocking shape.
        """
        _, session = self._lock_and_session()
        linked = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
        clear_before = self._duplicate_gate(open_prs=[], branches=[])
        blocked_before = self._duplicate_gate(open_prs=linked, branches=[BRANCH])
        self.assertTrue(self.run_heartbeat(task_session_id=session)["success"])
        clear_after = self._duplicate_gate(open_prs=[], branches=[])
        blocked_after = self._duplicate_gate(open_prs=linked, branches=[BRANCH])
        self.assertEqual(clear_before["outcome"], clear_after["outcome"])
        self.assertFalse(clear_after["block"])
        self.assertEqual(blocked_before["outcome"], blocked_after["outcome"])
        self.assertTrue(blocked_after["block"])
        self.assertEqual(blocked_after["linked_open_pr"], 4242)
    def test_foreign_session_id_is_refused_through_the_tool(self):
        self._lock_and_session()
        beat = self.run_heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
        self.assertFalse(beat["success"])
        self.assertIn("task_session_id does not match", " ".join(beat["reasons"]))
    def test_stale_generation_is_refused_through_the_tool(self):
        locked, session = self._lock_and_session()
        current = issue_lock_store.lock_generation(
            issue_lock_store.read_lock_file(locked["lock_file_path"])
        )
        beat = self.run_heartbeat(
            task_session_id=session, expected_generation=current + 5
        )
        self.assertFalse(beat["success"])
        self.assertIn("generation changed", beat["reasons"][0])
    def test_foreign_claimant_is_refused_through_the_tool(self):
        _, session = self._lock_and_session()
        beat = self.run_heartbeat(task_session_id=session, identity="someone-else")
        self.assertFalse(beat["success"])
    def test_heartbeat_cannot_acquire_a_missing_lock(self):
        beat = self.run_heartbeat(task_session_id="author_issue_work-000000000000")
        self.assertFalse(beat["success"])
        self.assertIn("no durable lock", beat["reasons"][0])
    def test_alive_pid_alone_does_not_keep_a_lease_live_through_the_tool(self):
        """PID-only refusal, end to end.
        The recorded pid is this live process. The lock is aged past its grace
        with no heartbeat, so the tool must refuse to slide it and the durable
        record must classify as a missed heartbeat rather than as live.
        """
        locked, session = self._lock_and_session()
        record = issue_lock_store.read_lock_file(locked["lock_file_path"])
        record["work_lease"]["last_heartbeat_at"] = _ts(
            datetime.now(timezone.utc) - timedelta(minutes=30)
        )
        record["work_lease"]["expires_at"] = _ts(
            datetime.now(timezone.utc) + timedelta(hours=2)
        )
        issue_lock_store.save_lock_file(locked["lock_file_path"], record)
        self.assertTrue(issue_lock_store.is_process_alive(record["session_pid"]))
        fresh = issue_lock_store.assess_lock_freshness(record)
        self.assertEqual(
            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
        )
        self.assertTrue(fresh["pid_alive"])
        beat = self.run_heartbeat(task_session_id=session)
        self.assertFalse(beat["success"])
        self.assertIn("reclaimed", " ".join(beat["reasons"]))
 class TestLegacyLocksThroughTheTool(_HeartbeatMcpBase):
    """AC-N8 end to end: protected on deployment, and rebindable."""
    def test_legacy_lock_stays_protected_after_deployment(self):
        record = self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        fresh = issue_lock_store.assess_lock_freshness(record)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
        self.assertTrue(fresh["legacy_lease"])
        self.assertTrue(fresh["legacy_expiry_preserved"])
        # It had never heartbeated, so under the new grace alone it would be
        # long gone; the preserved absolute expiry is what protects it.
        self.assertEqual(
            record["work_lease"]["created_at"],
            record["work_lease"]["last_heartbeat_at"],
        )
    def test_tool_rebinds_a_legacy_lock_and_mints_a_first_heartbeat(self):
        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        result = self.run_heartbeat(task_session_id=None)
        self.assertTrue(result["success"], result)
        self.assertEqual(result["operation"], "legacy_rebind")
        self.assertTrue(result["task_session_id"])
        written = issue_lock_store.read_lock_file(self._lock_path())
        lease = written["work_lease"]
        self.assertEqual(
            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
        )
        self.assertEqual(lease["heartbeat_count"], 1)
        self.assertNotEqual(
            lease["created_at"],
            written["legacy_rebind"]["legacy_origin"]["created_at"],
        )
        self.assertFalse(issue_lock_store.is_legacy_lease(written))
    def test_rebound_lock_then_heartbeats_through_the_tool(self):
        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        rebound = self.run_heartbeat(task_session_id=None)
        beat = self.run_heartbeat(task_session_id=rebound["task_session_id"])
        self.assertTrue(beat["success"], beat)
        self.assertEqual(beat["operation"], "heartbeat")
        self.assertEqual(beat["heartbeat_count"], 2)
    def test_rebind_refuses_a_foreign_owner_through_the_tool(self):
        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        result = self.run_heartbeat(task_session_id=None, identity="someone-else")
        self.assertFalse(result["success"])
        self.assertEqual(result["operation"], "legacy_rebind")
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,594 @@
 """Central lease policy and load-bearing heartbeat freshness (#790 Slice A).
 Before this slice, ``issue_lock_store.assess_lock_freshness`` parsed
 ``last_heartbeat_at`` and then never consulted it: liveness was decided by an
 absolute four-hour ``expires_at`` and by PID liveness. Because the recorded PID
 is the long-lived MCP daemon rather than the authoring task, an abandoned claim
 stayed "live" for the full four hours, and a claim whose work had already landed
 blocked reconciliation for just as long (Issue #787 / PR #789, and again Issue
 #760 / PR #791).
 These tests pin the corrected semantics, including the two asymmetries that are
 easy to lose in a refactor:
 * an **alive** PID must never make anything live (AC-N2), while
 * a **dead** PID must still mark a lease stale, because #753 dead-session
  recovery keys on exactly that classification.
 Durable-state helpers here write real lock files through the real flock path;
 they are not mocks of the store.
 """
 from __future__ import annotations
 import os
 import sys
 import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
 from unittest.mock import patch
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 import issue_lock_store  # noqa: E402
 import lease_policy  # noqa: E402
 import pr_work_lease  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
 ISSUE = 9790
 BRANCH = f"fix/issue-{ISSUE}-heartbeat"
 IDENTITY = "example-user"
 PROFILE = "test-author-prgs"
 ORG = "Example-Org"
 REPO = "Example-Repo"
 REMOTE = "prgs"
 DEAD_PID = 2**22  # far above any live pid on a test host
 def _ts(moment: datetime) -> str:
    return (
        moment.astimezone(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )
 class _LockFixture(unittest.TestCase):
    def setUp(self):
        self.lock_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.lock_dir.cleanup)
        self.now = datetime.now(timezone.utc)
        self.worktree = os.path.realpath(tempfile.mkdtemp(prefix="issue790-"))
        self.addCleanup(patch.stopall)
    def _path(self):
        return issue_lock_store.lock_file_path(
            remote=REMOTE,
            org=ORG,
            repo=REPO,
            issue_number=ISSUE,
            lock_dir=self.lock_dir.name,
        )
    def write_lock(
        self,
        *,
        lifecycle: str | None = lease_policy.LIFECYCLE_HEARTBEAT_V1,
        created_delta: timedelta = timedelta(minutes=1),
        heartbeat_delta: timedelta = timedelta(minutes=1),
        expires_delta: timedelta = timedelta(minutes=9),
        pid: int | None = None,
        task_session_id: str | None = "author_issue_work-aaaabbbbccccdddd",
        generation: int = 1,
        identity: str = IDENTITY,
        profile: str = PROFILE,
        branch: str = BRANCH,
        worktree: str | None = None,
    ) -> dict:
        """Write a real durable lock and return the record.
        Deltas are relative to ``self.now``; ``expires_delta`` is added, the
        others subtracted, so "in the past" reads naturally at each call site.
        """
        lease: dict = {
            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
            "issue_number": ISSUE,
            "pr_number": None,
            "branch": branch,
            "worktree_path": worktree or self.worktree,
            "claimant": {"username": identity, "profile": profile},
            "created_at": _ts(self.now - created_delta),
            "last_heartbeat_at": _ts(self.now - heartbeat_delta),
            "expires_at": _ts(self.now + expires_delta),
        }
        if lifecycle is not None:
            lease["lifecycle_version"] = lifecycle
            if task_session_id is not None:
                lease["task_session_id"] = task_session_id
        pid_value = os.getpid() if pid is None else pid
        record = {
            "issue_number": ISSUE,
            "branch_name": branch,
            "remote": REMOTE,
            "org": ORG,
            "repo": REPO,
            "worktree_path": worktree or self.worktree,
            "session_pid": pid_value,
            "pid": pid_value,
            "lock_generation": generation,
            "work_lease": lease,
        }
        path = self._path()
        record["lock_file_path"] = path
        issue_lock_store.save_lock_file(path, record)
        return record
 class TestPolicyIsTheSingleSource(unittest.TestCase):
    """AC-N7: one authoritative configuration source for every duration."""
    def test_author_policy_carries_the_agreed_values(self):
        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
        self.assertEqual(policy.initial_ttl_minutes, 10.0)
        self.assertEqual(policy.heartbeat_cadence_minutes, 2.0)
        self.assertEqual(policy.stale_warning_minutes, 5.0)
        self.assertEqual(policy.missed_heartbeat_grace_minutes, 10.0)
        self.assertEqual(policy.absolute_cap_hours, 8.0)
        self.assertEqual(policy.recovery_grace_minutes, 10.0)
        self.assertEqual(policy.terminal_race_drain_minutes, 2.0)
        self.assertTrue(policy.terminal_retirement_eligible)
        self.assertTrue(policy.heartbeat_lifecycle_active)
    def test_the_four_hour_author_ttl_literal_is_gone(self):
        """The duplicated literal AC-N7 exists to remove."""
        self.assertFalse(hasattr(issue_lock_store, "WORK_LEASE_TTL_HOURS"))
        import gitea_mcp_server
        self.assertFalse(hasattr(gitea_mcp_server, "WORK_LEASE_TTL_HOURS"))
    def test_declared_reviewer_values_match_the_module_still_using_them(self):
        """Slice A declares reviewer/merger numbers without rewiring them.
        Recording a value in two places is only safe if drift is detectable, so
        this asserts the declaration still equals the constants #747 owns. When
        Slice C migrates those call sites, this test becomes the proof the
        migration changed nothing.
        """
        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_REVIEWER_PR)
        self.assertEqual(
            policy.initial_ttl_minutes, float(reviewer_pr_lease.LEASE_TTL_MINUTES)
        )
        self.assertEqual(
            policy.stale_warning_minutes,
            float(reviewer_pr_lease.STALE_WARNING_MINUTES),
        )
        self.assertFalse(policy.heartbeat_lifecycle_active)
    def test_declared_conflict_fix_value_matches_its_module(self):
        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_CONFLICT_FIX)
        self.assertEqual(
            policy.initial_ttl_minutes,
            float(pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES),
        )
        self.assertFalse(policy.heartbeat_lifecycle_active)
    def test_environment_override_applies(self):
        var = lease_policy.env_var_name(
            lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes"
        )
        with patch.dict(os.environ, {var: "7"}):
            self.assertEqual(
                lease_policy.policy_for(
                    lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK
                ).initial_ttl_minutes,
                7.0,
            )
    def test_unusable_override_falls_back_instead_of_minting_a_zero_lease(self):
        """A typo must not make every claim instantly reclaimable."""
        var = lease_policy.env_var_name(
            lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes"
        )
        for bad in ("0", "-5", "not-a-number", "   "):
            with self.subTest(value=bad), patch.dict(os.environ, {var: bad}):
                self.assertEqual(
                    lease_policy.policy_for(
                        lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK
                    ).initial_ttl_minutes,
                    10.0,
                )
    def test_unknown_task_class_does_not_raise(self):
        policy = lease_policy.policy_for("something-new")
        self.assertEqual(policy.task_class, lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
 class TestLifecycleDiscrimination(_LockFixture):
    """AC-N8: the marker, never a timestamp, decides legacy vs heartbeat."""
    def test_missing_marker_reads_as_legacy(self):
        record = self.write_lock(lifecycle=None)
        self.assertTrue(issue_lock_store.is_legacy_lease(record))
        self.assertEqual(
            issue_lock_store.lease_lifecycle_version(record),
            lease_policy.LIFECYCLE_LEGACY,
        )
    def test_marker_present_reads_as_heartbeat_lifecycle(self):
        record = self.write_lock()
        self.assertFalse(issue_lock_store.is_legacy_lease(record))
    def test_equal_created_and_heartbeat_never_implies_a_fresh_heartbeat(self):
        """The exact inversion AC-N8 forbids.
        A legacy lock has ``last_heartbeat_at == created_at`` forever because
        nothing ever advanced it. Reading that equality as "recently
        heartbeated" would classify every never-heartbeated lock as fresh.
        """
        legacy = self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=3),
            heartbeat_delta=timedelta(hours=3),
        )
        lease = legacy["work_lease"]
        self.assertEqual(lease["created_at"], lease["last_heartbeat_at"])
        self.assertTrue(issue_lock_store.is_legacy_lease(legacy))
        # A brand-new heartbeat lease has them equal too, so the equality
        # carries no information in either direction.
        fresh = self.write_lock(
            created_delta=timedelta(seconds=0), heartbeat_delta=timedelta(seconds=0)
        )
        self.assertEqual(
            fresh["work_lease"]["created_at"],
            fresh["work_lease"]["last_heartbeat_at"],
        )
        self.assertFalse(issue_lock_store.is_legacy_lease(fresh))
    def test_minted_session_id_contains_no_pid(self):
        """AC-N1: the ownership key must not be derived from the daemon pid."""
        minted = issue_lock_store.mint_task_session_id()
        self.assertNotIn(str(os.getpid()), minted)
        self.assertNotEqual(minted, issue_lock_store.mint_task_session_id())
 class TestFreshnessIsHeartbeatDriven(_LockFixture):
    """AC-N2 and the new bands."""
    def test_fresh_heartbeat_is_live(self):
        record = self.write_lock(heartbeat_delta=timedelta(minutes=1))
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
        self.assertTrue(fresh["live"])
        self.assertFalse(fresh["heartbeat_warning"])
    def test_heartbeat_past_warning_is_still_live_but_flagged(self):
        record = self.write_lock(heartbeat_delta=timedelta(minutes=6))
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
        self.assertTrue(fresh["heartbeat_warning"])
    def test_missed_heartbeat_past_grace_is_classified_explicitly(self):
        record = self.write_lock(
            heartbeat_delta=timedelta(minutes=11),
            expires_delta=timedelta(minutes=30),
        )
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(
            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
        )
        self.assertFalse(fresh["live"])
        self.assertTrue(fresh["stale"])
    def test_alive_pid_never_establishes_freshness(self):
        """The defect in one assertion.
        The recorded PID is this very process, so it is unambiguously alive —
        and the lease is still not live, because the task stopped heartbeating.
        """
        record = self.write_lock(
            pid=os.getpid(),
            heartbeat_delta=timedelta(hours=4),
            expires_delta=timedelta(hours=4),
        )
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertTrue(fresh["pid_alive"])
        self.assertFalse(fresh["live"])
        self.assertEqual(
            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
        )
    def test_dead_pid_still_marks_stale_for_issue_753(self):
        """The opposite asymmetry: dead-PID corroboration is preserved."""
        record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1))
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE)
        self.assertFalse(fresh["live"])
        self.assertIn("not alive", fresh["reason"])
    def test_absolute_cap_requires_readoption(self):
        record = self.write_lock(
            created_delta=timedelta(hours=9), heartbeat_delta=timedelta(minutes=1)
        )
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE_ABSOLUTE_CAP)
        self.assertIn("re-adoption", fresh["reason"])
    def test_heartbeat_lifecycle_without_a_heartbeat_fails_closed(self):
        record = self.write_lock()
        del record["work_lease"]["last_heartbeat_at"]
        issue_lock_store.save_lock_file(self._path(), record)
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(
            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
        )
        self.assertIn("fail closed", fresh["reason"])
    def test_absent_lock(self):
        fresh = issue_lock_store.assess_lock_freshness(None)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_ABSENT)
        self.assertFalse(fresh["stale"])
 class TestLegacyLocksStayProtected(_LockFixture):
    """AC-N8: deployment must not retroactively shorten an existing claim."""
    def test_legacy_lock_with_a_stale_heartbeat_remains_live(self):
        """The deployment-safety case.
        A four-hour legacy lease minted three hours ago has not heartbeated
        once. Under the new grace it would be long gone; under its preserved
        absolute expiry it is still live, and must stay that way.
        """
        record = self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=3),
            heartbeat_delta=timedelta(hours=3),
            expires_delta=timedelta(hours=1),
        )
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
        self.assertTrue(fresh["live"])
        self.assertTrue(fresh["legacy_lease"])
        self.assertTrue(fresh["legacy_expiry_preserved"])
    def test_legacy_lock_past_its_absolute_expiry_is_expired_as_before(self):
        record = self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=5),
            heartbeat_delta=timedelta(hours=5),
            expires_delta=timedelta(hours=-1),
        )
        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_EXPIRED)
    def test_legacy_lock_is_never_reclaimed_by_the_heartbeat_band(self):
        record = self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=3),
            heartbeat_delta=timedelta(hours=3),
            expires_delta=timedelta(hours=1),
        )
        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
        self.assertFalse(reclaim["reclaim_allowed"])
 class TestReclaimAfterMissedHeartbeat(_LockFixture):
    def test_missed_heartbeat_makes_ownership_reclaimable(self):
        record = self.write_lock(
            pid=os.getpid(),
            heartbeat_delta=timedelta(minutes=15),
            expires_delta=timedelta(hours=3),
        )
        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
        self.assertTrue(reclaim["reclaim_allowed"])
        self.assertIn("stale_missed_heartbeat", reclaim["reasons"][0])
    def test_live_lease_is_never_reclaimable(self):
        record = self.write_lock(heartbeat_delta=timedelta(minutes=1))
        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
        self.assertFalse(reclaim["reclaim_allowed"])
    def test_dead_pid_reclaim_path_is_unchanged(self):
        """#753 must keep working through its original conditions."""
        record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1))
        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
        self.assertTrue(reclaim["reclaim_allowed"])
        self.assertTrue(reclaim["owner_pid_dead"])
 class TestHeartbeatWriter(_LockFixture):
    """A4: flock + CAS + exact verification, and no revival path."""
    def _heartbeat(self, **kwargs):
        params = {
            "remote": REMOTE,
            "org": ORG,
            "repo": REPO,
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "worktree_path": self.worktree,
            "identity": IDENTITY,
            "profile": PROFILE,
            "task_session_id": "author_issue_work-aaaabbbbccccdddd",
            "lock_dir": self.lock_dir.name,
            "now": self.now,
        }
        params.update(kwargs)
        return issue_lock_store.heartbeat_session_lock(**params)
    def test_heartbeat_slides_expiry_and_advances_generation(self):
        self.write_lock(heartbeat_delta=timedelta(minutes=4), generation=5)
        result = self._heartbeat()
        self.assertTrue(result["success"], result)
        self.assertEqual(result["prior_generation"], 5)
        self.assertEqual(result["lock_generation"], 6)
        self.assertEqual(result["heartbeat_count"], 1)
        self.assertEqual(result["last_heartbeat_at"], _ts(self.now))
        self.assertEqual(result["expires_at"], _ts(self.now + timedelta(minutes=10)))
        self.assertTrue(result["freshness"]["live"])
    def test_heartbeat_is_durable_and_repeatable(self):
        self.write_lock(heartbeat_delta=timedelta(minutes=4))
        self._heartbeat()
        second = self._heartbeat(now=self.now + timedelta(minutes=1))
        self.assertTrue(second["success"], second)
        self.assertEqual(second["heartbeat_count"], 2)
        written = issue_lock_store.read_lock_file(self._path())
        self.assertEqual(written["work_lease"]["heartbeat_count"], 2)
    def test_stale_generation_is_refused(self):
        self.write_lock(generation=5)
        result = self._heartbeat(expected_generation=4)
        self.assertFalse(result["success"])
        self.assertIn("generation changed", result["reasons"][0])
    def test_foreign_session_is_refused(self):
        self.write_lock()
        result = self._heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
        self.assertFalse(result["success"])
        self.assertIn("task_session_id does not match", " ".join(result["reasons"]))
    def test_missing_session_id_is_refused(self):
        self.write_lock()
        result = self._heartbeat(task_session_id="")
        self.assertFalse(result["success"])
    def test_foreign_claimant_is_refused(self):
        self.write_lock()
        for field, value in (
            ("identity", "someone-else"),
            ("profile", "other-profile"),
        ):
            with self.subTest(field=field):
                result = self._heartbeat(**{field: value})
                self.assertFalse(result["success"])
    def test_branch_and_worktree_mismatch_are_refused(self):
        self.write_lock()
        wrong_branch = self._heartbeat(branch_name=f"fix/issue-{ISSUE}-other")
        self.assertFalse(wrong_branch["success"])
        wrong_worktree = self._heartbeat(worktree_path="/tmp/not-the-worktree")
        self.assertFalse(wrong_worktree["success"])
    def test_lapsed_lease_cannot_be_heartbeated_back_to_life(self):
        """No revival path (A4).
        A session that stopped proving liveness must reclaim under a fresh
        generation, not restore ownership retroactively.
        """
        self.write_lock(
            heartbeat_delta=timedelta(minutes=30), expires_delta=timedelta(hours=1)
        )
        result = self._heartbeat()
        self.assertFalse(result["success"])
        self.assertIn("reclaimed", " ".join(result["reasons"]))
    def test_absent_lock_cannot_be_created_by_heartbeat(self):
        result = self._heartbeat()
        self.assertFalse(result["success"])
        self.assertIn("no durable lock", result["reasons"][0])
    def test_legacy_lock_is_refused_until_rebound(self):
        self.write_lock(lifecycle=None)
        result = self._heartbeat()
        self.assertFalse(result["success"])
        self.assertTrue(result["legacy_lease"])
        self.assertIn("rebound", " ".join(result["reasons"]))
 class TestLegacyRebind(_LockFixture):
    """AC-N8 exit route: canonical exact-owner rebinding."""
    def _rebind(self, **kwargs):
        params = {
            "remote": REMOTE,
            "org": ORG,
            "repo": REPO,
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "worktree_path": self.worktree,
            "identity": IDENTITY,
            "profile": PROFILE,
            "lock_dir": self.lock_dir.name,
            "now": self.now,
        }
        params.update(kwargs)
        return issue_lock_store.rebind_legacy_lock(**params)
    def test_rebind_mints_a_session_and_a_genuine_first_heartbeat(self):
        self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=3),
            heartbeat_delta=timedelta(hours=3),
            expires_delta=timedelta(hours=1),
            generation=2,
        )
        result = self._rebind()
        self.assertTrue(result["success"], result)
        self.assertTrue(result["task_session_id"])
        self.assertEqual(result["lock_generation"], 3)
        written = issue_lock_store.read_lock_file(self._path())
        lease = written["work_lease"]
        self.assertEqual(
            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
        )
        self.assertEqual(lease["last_heartbeat_at"], _ts(self.now))
        self.assertEqual(lease["expires_at"], _ts(self.now + timedelta(minutes=10)))
        self.assertFalse(issue_lock_store.is_legacy_lease(written))
        # The original claim is preserved for audit rather than overwritten.
        origin = written["legacy_rebind"]["legacy_origin"]
        self.assertTrue(origin["created_at"])
        self.assertEqual(origin["lifecycle"], lease_policy.LIFECYCLE_LEGACY)
    def test_rebound_lock_can_then_heartbeat(self):
        self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=3),
            heartbeat_delta=timedelta(hours=3),
            expires_delta=timedelta(hours=1),
        )
        rebound = self._rebind()
        beat = issue_lock_store.heartbeat_session_lock(
            remote=REMOTE,
            org=ORG,
            repo=REPO,
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
            identity=IDENTITY,
            profile=PROFILE,
            task_session_id=rebound["task_session_id"],
            lock_dir=self.lock_dir.name,
            now=self.now + timedelta(minutes=1),
        )
        self.assertTrue(beat["success"], beat)
    def test_rebind_refuses_a_foreign_owner(self):
        self.write_lock(lifecycle=None, expires_delta=timedelta(hours=1))
        result = self._rebind(identity="someone-else")
        self.assertFalse(result["success"])
    def test_rebind_refuses_a_lock_already_on_the_lifecycle(self):
        self.write_lock()
        result = self._rebind()
        self.assertFalse(result["success"])
        self.assertFalse(result["legacy_lease"])
    def test_rebind_is_not_a_recovery_path_for_a_lapsed_legacy_lease(self):
        """An expired legacy lease belongs to #760 renewal or #601 reclaim."""
        self.write_lock(
            lifecycle=None,
            created_delta=timedelta(hours=5),
            heartbeat_delta=timedelta(hours=5),
            expires_delta=timedelta(hours=-1),
        )
        result = self._rebind()
        self.assertFalse(result["success"])
        self.assertIn("not a recovery path", " ".join(result["reasons"]))
 if __name__ == "__main__":
    unittest.main()
@@ -78,95 +78,6 @@ class TestBlockReasonsAndReport(unittest.TestCase):
        self.assertTrue(report["recovery"])
 class TestLiveRemoteParity(unittest.TestCase):
    """#610: parity must account for the live remote master, not just local.
    The daemon can be stale relative to the live remote target while the local
    checkout HEAD still matches the daemon's startup commit, so local parity
    reports green even though a mutation would run against outdated code.
    """
    SHA_C = "c" * 40
    def test_distinguishes_three_shas(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertEqual(res["daemon_start_head"], SHA_A)
        self.assertEqual(res["local_head"], SHA_A)
        self.assertEqual(res["live_remote_head"], SHA_B)
    def test_mutation_safe_only_when_all_three_match(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
        self.assertTrue(res["mutation_safe"])
        self.assertTrue(res["live_known"])
        self.assertFalse(res["live_stale"])
    def test_live_stale_when_remote_advanced_past_daemon(self):
        # Local checkout still matches the daemon start (local parity green),
        # but the live remote master has advanced -> daemon is live-stale.
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertTrue(res["in_parity"])  # local parity still green
        self.assertTrue(res["live_stale"])
        self.assertFalse(res["mutation_safe"])
        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
        # Non-goal: unfetchable live remote must not be treated as stale for
        # read-only, but a mutation-safe claim fails closed.
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
        self.assertFalse(res["live_known"])
        self.assertFalse(res["mutation_safe"])
        self.assertFalse(res["live_stale"])
        self.assertTrue(res["in_parity"])
    def test_default_live_remote_preserves_legacy_shape(self):
        # Callers that do not supply a live head keep the pre-#610 behavior:
        # in-parity, not live-stale, no live-derived block.
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        self.assertFalse(res["live_stale"])
        self.assertEqual(mp.parity_block_reasons(res), [])
 class TestLiveStaleBlockAndReport(unittest.TestCase):
    """#610: live-staleness must block mutations and surface a typed blocker."""
    def test_live_stale_produces_block_reasons(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        self.assertTrue(mp.parity_block_reasons(res))
    def test_disable_env_suppresses_live_stale_block(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
            self.assertEqual(mp.parity_block_reasons(res), [])
    def test_resolver_disagreement_returns_typed_blocker(self):
        # Parity says local-green, resolver says restart required -> disagreement
        # is a typed, fail-closed blocker naming the resolver as authoritative.
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
        self.assertIsNotNone(blocker)
        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
        self.assertTrue(blocker["restart_required"])
        self.assertTrue(blocker["resolver_authoritative"])
    def test_no_disagreement_when_resolver_agrees(self):
        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
        self.assertIsNone(
            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
    def test_live_stale_report_names_live_remote(self):
        res = mp.assess_master_parity(
            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
        report = mp.parity_report(res)
        self.assertEqual(report["live_remote_head"], SHA_B)
        self.assertTrue(report["restart_required"])
 class TestReadGitHead(unittest.TestCase):
    def test_test_override_takes_precedence(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -184,149 +95,6 @@ class TestReadGitHead(unittest.TestCase):
            self.assertIsNone(mp.read_git_head(""))
 class TestReadRemoteMasterHead(unittest.TestCase):
    """#610: live remote master head reader (env-overridable, fails to None)."""
    def test_test_override_takes_precedence(self):
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
    def test_blank_override_is_none(self):
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
    def test_unfetchable_remote_is_none(self):
        # No override; a bogus root/remote must fail closed to None, never raise.
        env = {k: v for k, v in os.environ.items()
               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
        with patch.dict(os.environ, env, clear=True):
            self.assertIsNone(
                mp.read_remote_master_head("/nonexistent", remote="nope"))
 class TestRemoteHeadCache(unittest.TestCase):
    """#610: live remote reads are cached with a TTL to stay off the network.
    The parity gate runs on every mutation and every runtime-context read, so an
    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
    """
    def setUp(self):
        # These cases intentionally exercise the subprocess/cache path, so they
        # opt out of suite-wide hermetic mode (PR #788 F1).
        self._saved_hermetic = mp.hermetic_test_mode()
        mp.set_hermetic_test_mode(False)
        mp._clear_remote_head_cache()
        env = {
            k: v for k, v in os.environ.items()
            if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
                         mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE,
                         "PYTEST_CURRENT_TEST")
        }
        # Allow the probe path under hermetic defenses while still mocking
        # subprocess so no real network call runs.
        env[mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE] = "1"
        self._env = patch.dict(os.environ, env, clear=True)
        self._env.start()
        self.addCleanup(self._env.stop)
        self.addCleanup(mp._clear_remote_head_cache)
        self.addCleanup(
            lambda: mp.set_hermetic_test_mode(self._saved_hermetic)
        )
    def _fake_run(self, sha):
        class _R:
            returncode = 0
            stdout = f"{sha}\trefs/heads/master\n"
        calls = {"n": 0}
        def run(*args, **kwargs):
            calls["n"] += 1
            return _R()
        return run, calls
    def test_second_call_within_ttl_uses_cache(self):
        run, calls = self._fake_run(SHA_B)
        with patch.object(mp.subprocess, "run", run):
            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
        self.assertEqual(a, SHA_B)
        self.assertEqual(b, SHA_B)
        self.assertEqual(calls["n"], 1)
    def test_zero_ttl_bypasses_cache(self):
        run, calls = self._fake_run(SHA_B)
        with patch.object(mp.subprocess, "run", run):
            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
        self.assertEqual(calls["n"], 2)
    def test_env_override_never_touches_subprocess(self):
        run, calls = self._fake_run(SHA_B)
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
            with patch.object(mp.subprocess, "run", run):
                self.assertEqual(
                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
        self.assertEqual(calls["n"], 0)
 class TestHermeticLiveRemoteReads(unittest.TestCase):
    """#610 / PR #788 F1/F2: suite hermetic mode never hits the network."""
    def setUp(self):
        self._saved = mp.hermetic_test_mode()
        mp.set_hermetic_test_mode(True)
        mp._clear_remote_head_cache()
        self.addCleanup(lambda: mp.set_hermetic_test_mode(self._saved))
        self.addCleanup(mp._clear_remote_head_cache)
    def test_hermetic_mode_returns_none_without_subprocess(self):
        run_calls = {"n": 0}
        def boom(*args, **kwargs):
            run_calls["n"] += 1
            raise AssertionError("ls-remote must not run under hermetic mode")
        env = {
            k: v for k, v in os.environ.items()
            if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
                         mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE)
        }
        with patch.dict(os.environ, env, clear=True):
            with patch.object(mp.subprocess, "run", boom):
                self.assertIsNone(
                    mp.read_remote_master_head("/repo", remote="prgs")
                )
        self.assertEqual(run_calls["n"], 0)
    def test_hermetic_mode_survives_clear_true_env(self):
        """Module flag, not env pin: clear=True cannot re-enable the probe."""
        run_calls = {"n": 0}
        def boom(*args, **kwargs):
            run_calls["n"] += 1
            raise AssertionError("ls-remote must not run after clear=True")
        with patch.dict(os.environ, {}, clear=True):
            with patch.object(mp.subprocess, "run", boom):
                self.assertIsNone(mp.read_remote_master_head("/repo"))
        self.assertEqual(run_calls["n"], 0)
    def test_explicit_override_still_wins_under_hermetic(self):
        run_calls = {"n": 0}
        def boom(*args, **kwargs):
            run_calls["n"] += 1
            raise AssertionError("override must bypass subprocess")
        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            with patch.object(mp.subprocess, "run", boom):
                self.assertEqual(
                    mp.read_remote_master_head("/repo"), SHA_B
                )
        self.assertEqual(run_calls["n"], 0)
 class TestServerWiring(unittest.TestCase):
    """Integration with the gate choke point in the server namespace."""
@@ -337,13 +105,6 @@ class TestServerWiring(unittest.TestCase):
        self._saved = self.srv._STARTUP_PARITY
        self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                    "startup_head": SHA_A}
        # Keep the live-remote read hermetic (no real ls-remote network call):
        # default the live master to the daemon start so parity is fully green
        # unless a test overrides the live head explicitly (#610).
        self._live_patch = patch.dict(
            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
        self._live_patch.start()
        self.addCleanup(self._live_patch.stop)
    def tearDown(self):
        self.srv._STARTUP_PARITY = self._saved
@@ -386,36 +147,6 @@ class TestServerWiring(unittest.TestCase):
            self.assertTrue(out["in_parity"])
            self.assertNotIn("report", out)
    # --- #610: live-remote wiring -------------------------------------------
    def test_live_stale_blocks_mutation_though_local_green(self):
        # Local checkout matches the daemon start (local parity green) but the
        # live remote master has advanced -> mutations must fail closed.
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
            self.assertTrue(
                self.srv._master_parity_block("gitea.pr.create"))
    def test_assess_tool_exposes_three_distinct_shas(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
            out = self.srv.gitea_assess_master_parity(remote="prgs")
            self.assertEqual(out["daemon_start_head"], SHA_A)
            self.assertEqual(out["local_head"], SHA_A)
            self.assertEqual(out["live_remote_head"], SHA_B)
            self.assertTrue(out["live_stale"])
            self.assertFalse(out["mutation_safe"])
            self.assertIn("report", out)
    def test_assess_tool_mutation_safe_when_all_three_match(self):
        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
            out = self.srv.gitea_assess_master_parity(remote="prgs")
            self.assertTrue(out["mutation_safe"])
            self.assertFalse(out["live_stale"])
            self.assertNotIn("report", out)
 if __name__ == "__main__":
    unittest.main()
@@ -1,149 +0,0 @@
 """Documentation acceptance for the web console architecture ADR (#632 / epic #631).
 Enforces the acceptance criteria of issue #632:
 * AC1 — the ADR exists and covers layers, authority, phases, API versioning,
  and a page map.
 * AC2 — every #631 child (#632–#651) maps to at least one architectural
  component.
 * AC3 — the closed MVP (#425–#436) is stated as foundation, not recreated.
 * AC4 — forbidden paths are explicit: raw provider incidents as work,
  browser-held tokens, process-kill recovery.
 * AC5 — a controller can approve the document without reading chat history.
 Plus the linkage requirement: ``docs/webui-local-dev.md`` cross-links the ADR.
 """
 from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parent.parent
 ADR = (
    REPO_ROOT
    / "docs"
    / "architecture"
    / "webui-control-plane-console-architecture-adr.md"
 )
 ADR_BASENAME = "webui-control-plane-console-architecture-adr.md"
 LOCAL_DEV = REPO_ROOT / "docs" / "webui-local-dev.md"
 # Epic #631 children, phases 1-4 (twenty capability areas).
 EPIC_CHILDREN = tuple(f"#{number}" for number in range(632, 652))
 def _read(path: Path) -> str:
    assert path.is_file(), f"missing {path.relative_to(REPO_ROOT)}"
    return path.read_text(encoding="utf-8")
 def test_ac1_adr_exists_with_required_sections():
    text = _read(ADR)
    lower = text.lower()
    assert text.lstrip().startswith("#"), "ADR lacks a title"
    assert "#631" in text and "#632" in text
    for heading in (
        "## 2. Decision summary",
        "## 4. Authority boundaries",
        "## 5. Request flow and the redaction boundary",
        "## 6. API naming and versioning",
        "## 7. Page map",
        "## 8. Component ownership",
        "## 9. Phase gates",
        "## 11. Forbidden paths",
    ):
        assert heading in text, f"ADR must contain section {heading!r}"
    assert "browser ui" in lower and "domain loader" in lower
    assert "control-plane db" in lower and "capability gate" in lower
 def test_ac1_api_versioning_is_decided_including_legacy_routes():
    text = _read(ADR)
    assert "/api/v1/" in text, "ADR must decide the versioned API prefix"
    assert "/api/v2/" in text, "ADR must state how breaking changes are handled"
    lower = text.lower()
    assert "compatibility alias" in lower, (
        "ADR must say what happens to the existing unversioned MVP exports"
    )
 def test_ac1_page_map_covers_mvp_routes():
    text = _read(ADR)
    for route in ("`/`", "`/health`", "`/projects`", "`/prompts`", "`/runtime`",
                  "`/audit`", "`/actions`"):
        assert route in text, f"page map must account for MVP route {route}"
 def test_ac2_every_epic_child_maps_to_a_component():
    text = _read(ADR)
    ownership = text.split("## 8. Component ownership", 1)[-1].split("## 9.", 1)[0]
    missing = [child for child in EPIC_CHILDREN if child not in ownership]
    assert not missing, (
        f"epic #631 children without an architectural component: {missing}"
    )
 def test_ac2_every_child_row_declares_a_phase():
    text = _read(ADR)
    ownership = text.split("## 8. Component ownership", 1)[-1].split("## 9.", 1)[0]
    for child in EPIC_CHILDREN:
        row = next(
            (line for line in ownership.splitlines() if line.startswith(f"| {child} ")),
            None,
        )
        assert row is not None, f"no ownership row for {child}"
        assert row.rstrip().endswith(("| 1 |", "| 2 |", "| 3 |", "| 4 |")), (
            f"ownership row for {child} must end with its phase: {row!r}"
        )
 def test_ac3_mvp_is_foundation_not_recreated():
    text = _read(ADR)
    assert "#425" in text and "#436" in text
    lower = text.lower()
    assert "do not recreate" in lower or "recreating mvp scope" in lower
    assert "retained and evolved" in lower
 def test_ac4_forbidden_paths_are_explicit():
    text = _read(ADR)
    forbidden = text.split("## 11. Forbidden paths", 1)[-1].split("## 12.", 1)[0]
    lower = forbidden.lower()
    assert "raw provider incidents" in lower and "#612" in forbidden
    assert "browser-held tokens" in lower
    assert "process-kill recovery" in lower and "#630" in forbidden
    assert "ungated browser mutations" in lower
 def test_ac5_approval_checklist_is_self_contained():
    text = _read(ADR)
    assert "## 12. Approval checklist" in text
    checklist = text.split("## 12. Approval checklist", 1)[-1].split("## 13.", 1)[0]
    for marker in ("1.", "2.", "3.", "4.", "5.", "6."):
        assert marker in checklist, f"approval checklist missing item {marker}"
 def test_adr_states_the_two_boundary_invariants():
    text = _read(ADR)
    lower = text.lower()
    assert "no secrets to the browser" in lower
    assert "no ungated mutations" in lower
 def test_open_questions_are_recorded_not_implied():
    text = _read(ADR)
    assert "## 13. Open questions and follow-ups" in text
    section = text.split("## 13. Open questions and follow-ups", 1)[-1]
    assert "#633" in section, "deferred authorization work must name its issue"
 def test_local_dev_doc_cross_links_the_adr():
    text = _read(LOCAL_DEV)
    assert ADR_BASENAME in text, (
        "docs/webui-local-dev.md must cross-link the console architecture ADR "
        "(issue #632 scope)"
    )
 def test_docs_do_not_embed_secrets():
    for path in (ADR, LOCAL_DEV):
        text = _read(path)
        for marker in ("ghp_", "BEGIN PRIVATE KEY", "Authorization: Bearer"):
            assert marker not in text, f"{path.name} contains {marker!r}"