Merge pull request 'fix(runtime): recognize client identity environment and refresh worker registrations' (#976) from fix/issue-975-client-identity-heartbeat into master

Address REQUEST_CHANGES review #652 on PR #976 (issue #975).

Blocker 1: name GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS individually in
RECOGNIZED_GITEA_ENV_KEYS so the production-consumed operator override is no
longer classified as unsupported-env / runtime_reconnect_required. No prefix
broadening; unknown overrides remain fail-closed.

Blocker 2: add focused production-path tests that call _active_worker_identity
and _start_worker_heartbeat, proving supervisor attachment, identity/session/
generation/pid/epoch agreement, no duplicates, failed/fenced paths, orderly
shutdown, configured interval, and off-loop sqlite heartbeats.

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

gitea_config.py

@@ -1193,6 +1193,22 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({
     "GITEA_IRRECOVERABLE_HMAC_SECRET",
     "GITEA_FORCE_MCP_RUNTIME_CHECK",
     "GITEA_FORCE_CLIENT_MANAGED",
+    # #975: the client-identity inputs the server actually consumes at startup
+    # (CLIENT_NAME_ENV / CLIENT_INSTANCE_ENV / CLIENT_SESSION_ENV in
+    # gitea_mcp_server). Production read them while this allowlist omitted them,
+    # so the peer-env scan classified them as unsupported overrides and the
+    # capability resolver refused every mutation fleet-wide. Named individually
+    # on purpose: no prefix is added, so an unrecognised GITEA_* override is
+    # still refused exactly as it was before.
+    "GITEA_MCP_CLIENT",
+    "GITEA_MCP_CLIENT_INSTANCE",
+    "GITEA_MCP_CLIENT_SESSION",
+    # #975 review 652 B1: production also consumes HEARTBEAT_INTERVAL_ENV from
+    # mcp_worker_identity via gitea_mcp_server._start_worker_heartbeat. Omitting
+    # it reproduced the same unsupported-env → runtime_reconnect_required
+    # failure mode for the documented operator override. Named individually;
+    # no GITEA_* / GITEA_WORKER_* prefix is added.
+    "GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS",
 })
 
 RECOGNIZED_GITEA_ENV_PREFIXES = (

gitea_mcp_server.py

@@ -1916,8 +1916,16 @@ def _verify_role_mutation_workspace(
             if runtime_reasons:
                 raise RuntimeError("; ".join(runtime_reasons))
     except Exception as exc:
-        if "stale-runtime:" in str(exc):
+        # #975: ``_check_mcp_runtimes_diagnostics`` raises every reason it
-            raise RuntimeError(str(exc))
+        # produces through this one RuntimeError, but only ``stale-runtime:``
+        # was re-raised here — an ``unsupported-env:`` reason was swallowed
+        # while still failing the capability resolver, so this preflight and
+        # the resolver disagreed about the identical diagnostic. Both
+        # authoritative prefixes now propagate the same way. This can only ever
+        # widen what is refused, never widen what is permitted.
+        message = str(exc)
+        if any(prefix in message for prefix in RUNTIME_DIAGNOSTIC_HARD_PREFIXES):
+            raise RuntimeError(message)
         pass
 
     role = _effective_workspace_role()
@@ -15714,6 +15722,10 @@ _WORKER_REGISTRY = None
 _WORKER_IDENTITY: str | None = None
 _WORKER_GENERATION: str | None = None
 _WORKER_REGISTRATION_ATTEMPTED = False
+#: #975: the one heartbeat supervisor for this process's registration. One
+#: process registers exactly one worker identity, so there is exactly one
+#: supervisor and it is never replaced.
+_WORKER_HEARTBEAT_SUPERVISOR = None
 
 #: Env a client launcher may set to name itself and its session. Absent values
 #: are reported as unknown; they are never guessed at, because guessing is what
@@ -15800,6 +15812,17 @@ def _active_worker_identity() -> str | None:
             if outcome.get("registered"):
                 _WORKER_IDENTITY = identity
                 _WORKER_GENERATION = generation
+                # #975: registration is the only moment identity and fencing
+                # epoch are both known, so the heartbeat supervisor is started
+                # here. Without it ``last_heartbeat_at`` never left
+                # ``started_at`` and every healthy client lost ownership at the
+                # TTL.
+                _start_worker_heartbeat(
+                    identity=identity,
+                    fencing_epoch=outcome.get("fencing_epoch"),
+                    hints=hints,
+                    generation=generation,
+                )
                 return identity
             if not outcome.get("collision"):
                 return None
@@ -15810,6 +15833,78 @@ def _active_worker_identity() -> str | None:
     return None
 
 
+def _start_worker_heartbeat(
+    *,
+    identity: str,
+    fencing_epoch,
+    hints: dict,
+    generation: str | None,
+) -> None:
+    """Attach a heartbeat supervisor to the registration just created (#975).
+
+    Never raises: a supervisor that cannot start leaves the row exactly as
+    ``register()`` wrote it, which is the pre-#975 behaviour, rather than
+    failing the tool call that happened to trigger lazy registration.
+
+    Under pytest the thread is deliberately not started. Tests drive
+    ``WorkerHeartbeatSupervisor`` directly with an injected clock, so the suite
+    proves the lifecycle without leaving background sqlite writers behind.
+    """
+    global _WORKER_HEARTBEAT_SUPERVISOR
+    if _WORKER_HEARTBEAT_SUPERVISOR is not None:
+        return
+    registry = _worker_registry()
+    if registry is None or fencing_epoch is None:
+        return
+    try:
+        supervisor = mcp_worker_identity.WorkerHeartbeatSupervisor(
+            registry,
+            worker_identity=identity,
+            fencing_epoch=int(fencing_epoch),
+            session_id=hints.get("session_id"),
+            generation_id=generation,
+            client_name=hints.get("client_name"),
+            pid=os.getpid(),
+            ttl_seconds=mcp_worker_identity.DEFAULT_HEARTBEAT_TTL_SECONDS,
+            interval_seconds=os.environ.get(
+                mcp_worker_identity.HEARTBEAT_INTERVAL_ENV
+            ),
+        )
+        _WORKER_HEARTBEAT_SUPERVISOR = supervisor
+        if not mcp_daemon_guard.is_pytest_runtime():
+            supervisor.start()
+    except Exception:
+        return
+
+
+def _worker_heartbeat_status() -> dict:
+    """Read-only heartbeat observability for ``gitea_get_runtime_context`` (#975).
+
+    Reports the unsupervised case explicitly rather than omitting the key, so an
+    operator can tell "no supervisor" apart from "supervisor with no beats yet".
+    """
+    supervisor = _WORKER_HEARTBEAT_SUPERVISOR
+    if supervisor is None:
+        return {
+            "supervised": False,
+            "running": False,
+            "reasons": [
+                "no worker heartbeat supervisor is attached to this process; the "
+                "registration is not being renewed and will go stale at its TTL"
+            ],
+        }
+    try:
+        status = dict(supervisor.status())
+    except Exception as exc:
+        return {
+            "supervised": True,
+            "running": False,
+            "reasons": [f"heartbeat status unavailable: {type(exc).__name__}: {exc}"],
+        }
+    status.setdefault("reasons", [])
+    return status
+
+
 def _active_role_kind_safe() -> str | None:
     """Best-effort role for the registry record; never raises into a tool call.
 
@@ -19451,6 +19546,11 @@ def gitea_get_runtime_context(
         "fencing_epoch": provenance_assessment["fencing_epoch"],
         "conflicting_live_sessions": provenance_assessment["conflicting_live_sessions"],
         "provenance_assessment": provenance_assessment,
+        # #975: whether this registration is actually being renewed. Read-only,
+        # and it grants nothing — ownership still comes from the attachment
+        # record above. It exists so "my heartbeat stopped" is diagnosable
+        # before the TTL turns it into session_attachment_missing.
+        "worker_heartbeat": _worker_heartbeat_status(),
         "unconsumed_gitea_env": unconsumed_env,
         "preflight_ready": preflight["preflight_ready"],
         "preflight_block_reasons": preflight["preflight_block_reasons"],
@@ -21959,6 +22059,17 @@ def gitea_route_task_session(
 # self-recovery was removed from the read-only path (was _trigger_mcp_auto_restart).
 # Recovery is owned exclusively by the IDE/client reconnect path.
 
+#: Every reason prefix ``_check_mcp_runtimes_diagnostics`` can emit. Callers
+#: raise its reasons as one RuntimeError, and #975 found that the preflight
+#: re-raise recognised only ``stale-runtime:``, silently dropping
+#: ``unsupported-env:`` while the capability resolver still failed on it. This
+#: lives beside the producer so a newly added reason family cannot be forgotten
+#: by a distant re-raise predicate again.
+RUNTIME_DIAGNOSTIC_HARD_PREFIXES: tuple[str, ...] = (
+    "stale-runtime:",
+    "unsupported-env:",
+)
+
 
 def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
     """Read-only: report missing or stale MCP runtimes (no config or process mutation).

mcp_worker_identity.py

@@ -29,6 +29,7 @@ implements:
 
 from __future__ import annotations
 
+import atexit
 import hashlib
 import os
 import re
@@ -123,6 +124,103 @@ DEFAULT_REGISTRY_PATH = os.path.expanduser(
 #: worker within a single operator coffee break.
 DEFAULT_HEARTBEAT_TTL_SECONDS = 900.0
 
+#: Optional operator override for the beat interval, in seconds. Clamped by
+#: :func:`heartbeat_interval_for` so it can never be set at or above the TTL.
+HEARTBEAT_INTERVAL_ENV = "GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS"
+
+#: Never beat faster than this, so a misconfigured interval cannot turn the
+#: supervisor into a busy sqlite writer.
+MIN_HEARTBEAT_INTERVAL_SECONDS = 1.0
+
+
+def heartbeat_interval_for(
+    ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS,
+    override: str | float | None = None,
+) -> float:
+    """Beat interval for *ttl_seconds*: one third of the TTL, capped at a half.
+
+    #975. A third means two consecutive beats can be lost before the row is
+    allowed to look stale, and the half-TTL cap is a hard ceiling so no
+    override can produce an interval that expires the registration it is
+    supposed to renew. That is the whole liveness contract: a healthy worker
+    stays owned, and a worker that has genuinely stopped beating still expires
+    on the configured TTL — this function never touches the TTL itself.
+    """
+    try:
+        ttl = float(ttl_seconds)
+    except (TypeError, ValueError):
+        ttl = DEFAULT_HEARTBEAT_TTL_SECONDS
+    if not ttl > 0:
+        ttl = DEFAULT_HEARTBEAT_TTL_SECONDS
+
+    interval = ttl / 3.0
+    if override is not None:
+        candidate = str(override).strip()
+        if candidate:
+            try:
+                parsed = float(candidate)
+            except (TypeError, ValueError):
+                parsed = None
+            if parsed is not None and parsed > 0:
+                interval = parsed
+
+    ceiling = ttl / 2.0
+    if interval > ceiling:
+        interval = ceiling
+    if interval < MIN_HEARTBEAT_INTERVAL_SECONDS:
+        interval = min(MIN_HEARTBEAT_INTERVAL_SECONDS, ceiling)
+    return interval
+
+
+#: Recorded-vs-presented pairs compared by :func:`_heartbeat_expectation_drift`.
+_HEARTBEAT_EXPECTATION_FIELDS = (
+    "session_id",
+    "generation_id",
+    "client_name",
+    "pid",
+)
+
+
+def _heartbeat_expectation_drift(
+    record: dict[str, Any],
+    *,
+    expected_session_id: str | None = None,
+    expected_generation_id: str | None = None,
+    expected_client_name: str | None = None,
+    expected_pid: int | None = None,
+) -> list[tuple[str, Any, Any]]:
+    """Return ``(field, recorded, presented)`` for every mismatched expectation.
+
+    Only supplied expectations are compared, so a caller that presents nothing
+    gets the pre-#975 behaviour. ``client_name`` is compared through
+    :func:`normalize_client_name` so one application's namespaces stay one
+    client while genuinely different clients stay distinct.
+    """
+    presented = {
+        "session_id": expected_session_id,
+        "generation_id": expected_generation_id,
+        "client_name": (
+            normalize_client_name(expected_client_name)
+            if expected_client_name is not None
+            else None
+        ),
+        "pid": expected_pid,
+    }
+    drift: list[tuple[str, Any, Any]] = []
+    for field in _HEARTBEAT_EXPECTATION_FIELDS:
+        want = presented[field]
+        if want is None:
+            continue
+        got = record.get(field)
+        if field == "pid":
+            match = got is not None and int(got) == int(want)
+        else:
+            match = got == want
+        if not match:
+            drift.append((field, got, want))
+    return drift
+
+
 STATUS_ACTIVE = "active"
 STATUS_SUPERSEDED = "superseded"
 STATUS_RELEASED = "released"
@@ -786,11 +884,25 @@ class WorkerRegistry:
         worker_identity: str,
         fencing_epoch: int,
         now: datetime | None = None,
+        expected_session_id: str | None = None,
+        expected_generation_id: str | None = None,
+        expected_client_name: str | None = None,
+        expected_pid: int | None = None,
     ) -> dict[str, Any]:
         """Renew only the owning registration (#948 AC11).
 
         A stale epoch is refused rather than silently renewed, so a superseded
         session that resumes cannot heartbeat its way back into ownership.
+
+        #975 adds optional keyword-only *expectations*. Every one supplied must
+        match the recorded row or the renewal is refused. They exist because
+        identity plus epoch cannot express "renew the row I registered, and only
+        that row": a recycled PID, or a second session of the same client, would
+        otherwise be renewable by the wrong beater. They are fencing tokens,
+        never assertions — supplying one can only cause a refusal, never grant
+        anything, and omitting them preserves the pre-#975 behaviour exactly.
+        Drift reports the existing ``BLOCKER_FENCED`` rather than a new
+        ``blocker_kind``, because consumers switch on that value.
         """
         stamp = _ts(now or _utc_now())
         with self._tx() as conn:
@@ -807,6 +919,30 @@ class WorkerRegistry:
                     "reasons": [f"no registration for {worker_identity!r}"],
                 }
             record = self._row_to_record(row)
+            drift = _heartbeat_expectation_drift(
+                record,
+                expected_session_id=expected_session_id,
+                expected_generation_id=expected_generation_id,
+                expected_client_name=expected_client_name,
+                expected_pid=expected_pid,
+            )
+            if drift:
+                return {
+                    "success": False,
+                    "renewed": False,
+                    "mutation_performed": False,
+                    "blocker_kind": BLOCKER_FENCED,
+                    "expectation_drift": drift,
+                    "reasons": [
+                        "presented worker expectations do not match the recorded "
+                        "registration, so this beater does not own the row: "
+                        + "; ".join(
+                            f"{field} recorded {recorded!r}, presented {presented!r}"
+                            for field, recorded, presented in drift
+                        )
+                        + " (#975)"
+                    ],
+                }
             if record["status"] != STATUS_ACTIVE:
                 return {
                     "success": False,
@@ -1430,3 +1566,253 @@ def resolve_bound_remote(
             "explicitly to avoid host drift (#948)."
         ],
     }
+
+
+# --- Production heartbeat lifecycle (#975) --------------------------------
+#
+# #948 delivered ``WorkerRegistry.heartbeat()`` and nothing ever called it, so
+# ``last_heartbeat_at`` stayed pinned to ``started_at`` for every production
+# worker and ``heartbeat_ttl_seconds`` stopped being a liveness window at all —
+# it became a hard cap on how long any client could stay attached. This
+# supervisor is the missing caller.
+#
+# It is a daemon thread rather than an asyncio task or a per-request refresh
+# because the renewal has to survive an *idle* session: a client waiting on a
+# lease with no tool call in flight must not lose ownership, which rules out
+# request-driven refresh. Registration itself happens lazily inside a tool call
+# on the server's event loop, and the registry performs blocking
+# ``BEGIN IMMEDIATE`` sqlite writes, which must not run on that loop.
+# ``daemon=True`` is deliberate: a hard kill takes the thread down with the
+# process, so a dead worker still goes stale on the normal TTL.
+
+#: Refusals that mean this worker no longer owns its row. Beating again could
+#: only ever be an attempt to renew ownership it has already lost, so the
+#: supervisor stops permanently instead of retrying.
+TERMINAL_HEARTBEAT_BLOCKERS = frozenset({BLOCKER_FENCED, BLOCKER_NO_ATTACHMENT})
+
+
+class WorkerHeartbeatSupervisor:
+    """Periodically renew exactly one worker registration.
+
+    Contract:
+
+    * It never registers. A supervisor exists only for an already-registered
+      identity, so it cannot create a second registration or a second identity
+      system.
+    * Every beat presents the full expectation set, so it can renew only the row
+      matching this exact client name, worker identity, session, generation and
+      pid.
+    * A terminal refusal (fenced or missing) stops it permanently and records
+      why. A fenced session must never beat its way back into ownership.
+    * A transient failure (a locked database, say) is counted and the loop
+      continues, so one contended write does not silently end the heartbeat.
+    * Nothing here raises into a caller. ``beat_once`` returns its outcome and
+      the thread body swallows everything, because a heartbeat failure must
+      degrade to "not renewed" and never crash a tool call.
+    """
+
+    def __init__(
+        self,
+        registry: WorkerRegistry,
+        *,
+        worker_identity: str,
+        fencing_epoch: int,
+        session_id: str | None = None,
+        generation_id: str | None = None,
+        client_name: str | None = None,
+        pid: int | None = None,
+        ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS,
+        interval_seconds: float | None = None,
+        clock=None,
+    ) -> None:
+        self._registry = registry
+        self.worker_identity = worker_identity
+        self.fencing_epoch = int(fencing_epoch)
+        self.session_id = session_id
+        self.generation_id = generation_id
+        self.client_name = (
+            normalize_client_name(client_name) if client_name is not None else None
+        )
+        self.pid = int(pid) if pid is not None else None
+        self.ttl_seconds = float(ttl_seconds)
+        self.interval_seconds = (
+            heartbeat_interval_for(self.ttl_seconds)
+            if interval_seconds is None
+            else heartbeat_interval_for(self.ttl_seconds, interval_seconds)
+        )
+        #: Injectable so every TTL test uses controlled time and no test waits
+        #: for a real interval or a real TTL to elapse.
+        self._clock = clock or _utc_now
+
+        self._stop_event = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._lock = threading.Lock()
+        self._atexit_registered = False
+
+        self.started = False
+        self.stopped_reason: str | None = None
+        self.beats_attempted = 0
+        self.beats_renewed = 0
+        self.transient_failures = 0
+        self.last_beat_at: str | None = None
+        self.last_result: dict[str, Any] | None = None
+
+    # -- one beat --
+
+    def beat_once(self, now: datetime | None = None) -> dict[str, Any]:
+        """Renew once. Never raises; returns the registry outcome or a failure."""
+        if self.stopped_reason is not None:
+            return {
+                "success": False,
+                "renewed": False,
+                "beat_attempted": False,
+                "reasons": [
+                    f"supervisor already stopped: {self.stopped_reason}"
+                ],
+            }
+
+        stamp = now or self._clock()
+        self.beats_attempted += 1
+        try:
+            result = self._registry.heartbeat(
+                worker_identity=self.worker_identity,
+                fencing_epoch=self.fencing_epoch,
+                now=stamp,
+                expected_session_id=self.session_id,
+                expected_generation_id=self.generation_id,
+                expected_client_name=self.client_name,
+                expected_pid=self.pid,
+            )
+        except Exception as exc:
+            # Transient by assumption: an unexpected error is not proof this
+            # worker lost ownership, so it must not silently end the heartbeat.
+            self.transient_failures += 1
+            result = {
+                "success": False,
+                "renewed": False,
+                "transient": True,
+                "blocker_kind": None,
+                "reasons": [f"heartbeat raised {type(exc).__name__}: {exc}"],
+            }
+            self.last_result = result
+            return result
+
+        self.last_result = result
+        if result.get("renewed"):
+            self.beats_renewed += 1
+            self.last_beat_at = result.get("last_heartbeat_at") or _ts(stamp)
+            return result
+
+        blocker = result.get("blocker_kind")
+        if blocker in TERMINAL_HEARTBEAT_BLOCKERS:
+            self._stop_internal(
+                reason=(
+                    f"refused with blocker_kind={blocker!r}: "
+                    + "; ".join(result.get("reasons") or [])
+                )
+            )
+        else:
+            self.transient_failures += 1
+        return result
+
+    # -- lifecycle --
+
+    def start(self) -> dict[str, Any]:
+        """Start the beat thread. Idempotent; safe to call from a tool call."""
+        with self._lock:
+            if self.stopped_reason is not None:
+                return {
+                    "started": False,
+                    "reasons": [f"supervisor stopped: {self.stopped_reason}"],
+                }
+            if self._thread is not None and self._thread.is_alive():
+                return {"started": True, "already_running": True, "reasons": []}
+
+            self._stop_event.clear()
+            thread = threading.Thread(
+                target=self._run,
+                name=f"gitea-worker-heartbeat-{self.worker_identity}",
+                daemon=True,
+            )
+            self._thread = thread
+            self.started = True
+            if not self._atexit_registered:
+                # Orderly shutdown stops the heartbeat. A hard kill does not
+                # run this, which is correct: the row must then go stale.
+                atexit.register(self._atexit_stop)
+                self._atexit_registered = True
+            thread.start()
+            return {"started": True, "already_running": False, "reasons": []}
+
+    def _run(self) -> None:
+        while not self._stop_event.is_set():
+            # Wait first: registration already stamped a fresh heartbeat, so an
+            # immediate beat would be a redundant write on every server launch.
+            if self._stop_event.wait(self.interval_seconds):
+                return
+            try:
+                self.beat_once()
+            except Exception:
+                # beat_once is already total; this is the last-resort guard that
+                # keeps a supervisor thread from dying silently.
+                self.transient_failures += 1
+            if self.stopped_reason is not None:
+                return
+
+    def stop(self, reason: str = "stopped") -> dict[str, Any]:
+        """Stop beating. Idempotent, and prompt because the loop waits on an Event."""
+        self._stop_internal(reason=reason)
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=max(1.0, min(5.0, self.interval_seconds)))
+        return {
+            "stopped": True,
+            "reason": self.stopped_reason,
+            "thread_alive": bool(thread is not None and thread.is_alive()),
+        }
+
+    def _stop_internal(self, *, reason: str) -> None:
+        if self.stopped_reason is None:
+            self.stopped_reason = reason
+        self._stop_event.set()
+        if self._atexit_registered:
+            try:
+                atexit.unregister(self._atexit_stop)
+            except Exception:
+                pass
+            self._atexit_registered = False
+
+    def _atexit_stop(self) -> None:
+        try:
+            self.stop(reason="process exit")
+        except Exception:
+            pass
+
+    # -- observability --
+
+    def status(self) -> dict[str, Any]:
+        """Read-only observability payload; safe to embed in a tool result."""
+        thread = self._thread
+        return {
+            "supervised": True,
+            "worker_identity": self.worker_identity,
+            "fencing_epoch": self.fencing_epoch,
+            "session_id": self.session_id,
+            "generation_id": self.generation_id,
+            "client_name": self.client_name,
+            "pid": self.pid,
+            "heartbeat_ttl_seconds": self.ttl_seconds,
+            "heartbeat_interval_seconds": self.interval_seconds,
+            "started": self.started,
+            "running": bool(
+                thread is not None
+                and thread.is_alive()
+                and self.stopped_reason is None
+            ),
+            "stopped_reason": self.stopped_reason,
+            "beats_attempted": self.beats_attempted,
+            "beats_renewed": self.beats_renewed,
+            "transient_failures": self.transient_failures,
+            "last_heartbeat_at": self.last_beat_at,
+            "last_blocker_kind": (self.last_result or {}).get("blocker_kind"),
+        }