feat(arch01): atomic platform install + authority kernel (Closes #822)

ARCH-01 Foundation Slice A — first implementation leaf of the #820/#821
program. Adds a new, disabled-by-default SQLite kernel:

- Connection-bound trusted-service actor context (cp_actor_principal/kind,
  cp_operation_mode, cp_service_session, cp_context_epoch) that SQL may read
  but never set; every mutating trigger runs the actor protocol and fails
  closed on a missing, stale, or epoch-shifted context.
- Immutable authority-dominance lattice with the exact seeded tuple set,
  validated individually by the install-state trigger.
- Principal-equivalence root: a class exists before its first principal and
  current_class_id is NOT NULL.
- Single atomic BEGIN IMMEDIATE install seeding the installer principal, the
  distinguished operator-key issuer, dominance, the initial NULL-grantor
  bootstrap grant, the active invariant, and the immutable installed marker
  last; second install returns ALREADY_INSTALLED with no mutation.
- Immutability triggers on marker/seed/dominance/issuer/installer
  registration/initial-grant identity; append-only audit_records; last-active
  grant floored by CHECK so concurrent revokes cannot drop it to zero.

Files: arch01_platform.py (new), tests/test_arch01_platform.py (new).

Tests: 25 tests + 21 subtests cover ACs 1-14 including negative, rollback,
raw-write-bypass, and concurrency. Focused suite green; full suite 4422
passed / 11 pre-existing baseline failures at 53c2c92, zero new.

Subsystem disabled by default until readiness checks pass. SQLite-first;
PostgreSQL parity is #827.

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

branch_cleanup_guard.py

@@ -163,19 +163,7 @@ _TERMINAL_OWNERSHIP_STATUSES = frozenset(
     {"released", "abandoned", "done", "blocked", "terminal", "closed"}
 )
 _EXPIRED_STATUSES = frozenset({"expired"})
-_STALE_STATUSES = frozenset(
+_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"})
-    {
-        "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:

docs/mcp-tool-inventory.md

@@ -100,7 +100,6 @@ 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`

docs/webui-authz-audit.md

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

docs/webui-deployment.md

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

gitea_mcp_server.py

@@ -2007,7 +2007,6 @@ 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)
@@ -2250,6 +2249,7 @@ 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,
@@ -2549,12 +2549,7 @@ def _build_author_issue_work_lease(
     host: str | None,
 ) -> dict:
     created = _work_lease_now()
-    # #790 Slice A: the window comes from the central policy, not a literal here.
+    expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
-    # 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,
@@ -2565,15 +2560,6 @@ 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,
     }
 
 
@@ -4343,138 +4329,6 @@ 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,

issue_lock_store.py

@@ -15,27 +15,15 @@ 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._+-]+")
 
 
@@ -265,331 +253,6 @@ 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))
@@ -673,16 +336,6 @@ 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
@@ -703,216 +356,60 @@ 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": STATUS_ABSENT,
+            "status": "absent",
             "live": False,
             "stale": False,
             "reason": "no lock record",
         }
 
-    lease = lock_data.get("work_lease")
     expires_at = lease_expires_at(lock_data)
-    heartbeat_at = _lease_heartbeat_at(lock_data)
+    lease = lock_data.get("work_lease")
-    created_at = (
+    heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
-        _parse_lease_timestamp(lease.get("created_at"))
+    if heartbeat_at is None and isinstance(lease, dict):
-        if isinstance(lease, dict)
+        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
-        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
 
-    lifecycle = lease_lifecycle_version(lock_data)
+    if expires_at and expires_at <= current:
-    legacy = lifecycle != lease_policy.LIFECYCLE_HEARTBEAT_V1
+        return {
-    policy = lease_policy.policy_for(lease_task_class(lock_data))
+            "status": "expired",
+            "live": False,
+            "stale": True,
+            "reason": f"lease expired at {expires_at.isoformat()}",
+            "pid_alive": pid_alive,
+        }
 
-    evidence: dict[str, Any] = {
+    if pid is not None and not pid_alive:
+        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:
@@ -949,27 +446,6 @@ 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")

lease_policy.py

@@ -1,212 +0,0 @@
-"""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,
-    }

task_capability_map.py

@@ -32,15 +32,6 @@ 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",

tests/test_issue_790_heartbeat_mcp_path.py

@@ -1,444 +0,0 @@
-"""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()

tests/test_issue_790_lease_policy.py

@@ -1,594 +0,0 @@
-"""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()

tests/test_webui_console_authz_audit.py

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

webui/app.py

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

webui/console_audit.py

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

webui/console_authz.py

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

webui/console_redaction.py

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