fix(runtime): recognize heartbeat interval env and wire production lifecycle tests

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

@@ -1203,6 +1203,12 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({
     "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 = (

tests/test_issue_975_client_identity_heartbeat.py

@@ -26,7 +26,9 @@ import tempfile
 import threading
 import time
 import unittest
+from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
+from typing import Iterator
 
 import gitea_config
 import mcp_worker_identity as mwi
@@ -43,6 +45,13 @@ CLIENT_IDENTITY_ENV_KEYS = (
     "GITEA_MCP_CLIENT_SESSION",
 )
 
+#: Every production-consumed GITEA_* constant this change must keep allowlisted.
+#: Drift between these names and RECOGNIZED_GITEA_ENV_KEYS reopens the #975
+#: unsupported-env mutation-gate failure mode.
+PRODUCTION_CONSUMED_GITEA_ENV_KEYS = CLIENT_IDENTITY_ENV_KEYS + (
+    mwi.HEARTBEAT_INTERVAL_ENV,
+)
+
 
 def _registry() -> mwi.WorkerRegistry:
     """A registry on a throwaway path; never the operator's real one."""
@@ -146,19 +155,48 @@ class ClientIdentityEnvRecognitionTests(unittest.TestCase):
             "GITEA_MCP_CLIENT": "claude_code",
             "GITEA_MCP_CLIENT_INSTANCE": "inst-0001",
             "GITEA_MCP_CLIENT_SESSION": "sess-0001",
+            mwi.HEARTBEAT_INTERVAL_ENV: "30",
         }
         self.assertEqual(gitea_config.get_unconsumed_gitea_env_overrides(env), {})
 
+    def test_heartbeat_interval_key_is_recognized(self):
+        # #975 review 652 B1: the documented operator override must be accepted.
+        self.assertEqual(
+            mwi.HEARTBEAT_INTERVAL_ENV, "GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS"
+        )
+        self.assertIn(
+            mwi.HEARTBEAT_INTERVAL_ENV, gitea_config.RECOGNIZED_GITEA_ENV_KEYS
+        )
+        self.assertEqual(
+            gitea_config.get_unconsumed_gitea_env_overrides(
+                {mwi.HEARTBEAT_INTERVAL_ENV: "30"}
+            ),
+            {},
+        )
+
+    def test_nearby_unknown_heartbeat_key_remains_rejected(self):
+        # Exact key only — no broad GITEA_WORKER_* prefix that would admit typos.
+        nearby = "GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS_EXTRA"
+        env = {
+            mwi.HEARTBEAT_INTERVAL_ENV: "30",
+            nearby: "1",
+        }
+        unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
+        self.assertIn(nearby, unconsumed)
+        self.assertNotIn(mwi.HEARTBEAT_INTERVAL_ENV, unconsumed)
+
     def test_unknown_gitea_key_is_still_unsupported(self):
         # AC2: the gate is not broadened. An unrecognised override is refused
         # exactly as before, and the recognised siblings do not shield it.
         env = {
             "GITEA_MCP_CLIENT": "claude_code",
+            mwi.HEARTBEAT_INTERVAL_ENV: "30",
             "GITEA_TOTALLY_INVENTED_OVERRIDE": "1",
         }
         unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
         self.assertIn("GITEA_TOTALLY_INVENTED_OVERRIDE", unconsumed)
         self.assertNotIn("GITEA_MCP_CLIENT", unconsumed)
+        self.assertNotIn(mwi.HEARTBEAT_INTERVAL_ENV, unconsumed)
 
     def test_no_new_prefix_admits_arbitrary_client_keys(self):
         # A prefix would have been the lazy fix; it would admit anything
@@ -171,16 +209,19 @@ class ClientIdentityEnvRecognitionTests(unittest.TestCase):
 
     def test_server_consumes_exactly_the_recognized_key_names(self):
         # Guards the actual defect class: the constants production reads and the
-        # allowlist drifting apart again.
+        # allowlist drifting apart again. Includes HEARTBEAT_INTERVAL_ENV so the
+        # #975 review 652 B1 regression cannot reappear silently.
         import gitea_mcp_server as mcp_server
 
         for name in (
             mcp_server.CLIENT_NAME_ENV,
             mcp_server.CLIENT_INSTANCE_ENV,
             mcp_server.CLIENT_SESSION_ENV,
+            mwi.HEARTBEAT_INTERVAL_ENV,
         ):
             with self.subTest(env_key=name):
                 self.assertIn(name, gitea_config.RECOGNIZED_GITEA_ENV_KEYS)
+                self.assertIn(name, PRODUCTION_CONSUMED_GITEA_ENV_KEYS)
 
 
 class RuntimeDiagnosticReasonPropagationTests(unittest.TestCase):
@@ -203,6 +244,27 @@ class RuntimeDiagnosticReasonPropagationTests(unittest.TestCase):
             any(p in message for p in mcp_server.RUNTIME_DIAGNOSTIC_HARD_PREFIXES)
         )
 
+    def test_unknown_override_still_produces_hard_unsupported_env_diagnostic(self):
+        # Fail-closed classification: an unknown override remains unconsumed and
+        # the hard-prefix family still matches the diagnostic reason the
+        # capability resolver raises as runtime_reconnect_required.
+        import gitea_mcp_server as mcp_server
+
+        env = {
+            mwi.HEARTBEAT_INTERVAL_ENV: "30",
+            "GITEA_TOTALLY_INVENTED_OVERRIDE": "1",
+        }
+        unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
+        self.assertIn("GITEA_TOTALLY_INVENTED_OVERRIDE", unconsumed)
+        rendered = (
+            "unsupported-env: Unsupported GITEA_* environment variable override(s) "
+            f"detected: {', '.join(f'{k}={v}' for k, v in sorted(unconsumed.items()))}. "
+            "Unknown env overrides are unsupported."
+        )
+        self.assertTrue(
+            any(p in rendered for p in mcp_server.RUNTIME_DIAGNOSTIC_HARD_PREFIXES)
+        )
+
 
 # --- Interval policy ------------------------------------------------------
 
@@ -755,5 +817,338 @@ class SupervisorThreadLoopTests(unittest.TestCase):
         self.assertFalse(supervisor.start()["started"])
 
 
+# --- #975 review 652 B2: production registration → heartbeat wiring ---------
+
+
+@contextmanager
+def _production_worker_lifecycle(
+    *,
+    client: str = "claude_code",
+    instance: str = "inst-prod-0001",
+    session: str = "sess-prod-0001",
+    interval: str | None = "30",
+    pin_registry: bool = True,
+) -> Iterator[tuple[str | None, object]]:
+    """Reset process-local worker state and drive the real production entry path.
+
+    Pins ``GITEA_WORKER_REGISTRY_DB`` so pytest may open a registry, then
+    restores every module global and environment key on exit. Never writes to
+    the operator's real registry.
+    """
+    import gitea_mcp_server as mcp_server
+
+    handle, path = tempfile.mkstemp(suffix=".sqlite3")
+    os.close(handle)
+    os.unlink(path)
+
+    env_keys = (
+        mwi.REGISTRY_PATH_ENV,
+        mwi.HEARTBEAT_INTERVAL_ENV,
+        mcp_server.CLIENT_NAME_ENV,
+        mcp_server.CLIENT_INSTANCE_ENV,
+        mcp_server.CLIENT_SESSION_ENV,
+    )
+    saved_env = {key: os.environ.get(key) for key in env_keys}
+    saved = {
+        "registry": mcp_server._WORKER_REGISTRY,
+        "identity": mcp_server._WORKER_IDENTITY,
+        "generation": mcp_server._WORKER_GENERATION,
+        "attempted": mcp_server._WORKER_REGISTRATION_ATTEMPTED,
+        "supervisor": mcp_server._WORKER_HEARTBEAT_SUPERVISOR,
+    }
+
+    mcp_server._WORKER_REGISTRY = None
+    mcp_server._WORKER_IDENTITY = None
+    mcp_server._WORKER_GENERATION = None
+    mcp_server._WORKER_REGISTRATION_ATTEMPTED = False
+    mcp_server._WORKER_HEARTBEAT_SUPERVISOR = None
+
+    if pin_registry:
+        os.environ[mwi.REGISTRY_PATH_ENV] = path
+    else:
+        os.environ.pop(mwi.REGISTRY_PATH_ENV, None)
+
+    os.environ[mcp_server.CLIENT_NAME_ENV] = client
+    os.environ[mcp_server.CLIENT_INSTANCE_ENV] = instance
+    os.environ[mcp_server.CLIENT_SESSION_ENV] = session
+    if interval is None:
+        os.environ.pop(mwi.HEARTBEAT_INTERVAL_ENV, None)
+    else:
+        os.environ[mwi.HEARTBEAT_INTERVAL_ENV] = str(interval)
+
+    try:
+        yield (path if pin_registry else None), mcp_server
+    finally:
+        supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+        if supervisor is not None:
+            try:
+                supervisor.stop(reason="test teardown")
+            except Exception:
+                pass
+        mcp_server._WORKER_REGISTRY = saved["registry"]
+        mcp_server._WORKER_IDENTITY = saved["identity"]
+        mcp_server._WORKER_GENERATION = saved["generation"]
+        mcp_server._WORKER_REGISTRATION_ATTEMPTED = saved["attempted"]
+        mcp_server._WORKER_HEARTBEAT_SUPERVISOR = saved["supervisor"]
+        for key, value in saved_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        for suffix in ("-wal", "-shm"):
+            try:
+                os.unlink(path + suffix)
+            except OSError:
+                pass
+
+
+class ProductionWorkerLifecycleWiringTests(unittest.TestCase):
+    """#975 review 652 B2: exercise the real production orchestration path.
+
+    These tests call ``_active_worker_identity()`` / ``_start_worker_heartbeat``
+    rather than the ``_attach`` / ``_supervisor`` factories. Deleting the
+    production wiring call must fail them.
+    """
+
+    def test_successful_registration_starts_heartbeat_supervision(self):
+        # (1) Successful worker registration starts heartbeat supervision.
+        with _production_worker_lifecycle() as (path, mcp_server):
+            self.assertIsNone(mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+            identity = mcp_server._active_worker_identity()
+            self.assertIsNotNone(identity)
+            supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            self.assertIsNotNone(
+                supervisor,
+                "production registration must call _start_worker_heartbeat; "
+                "removing that call must fail this test",
+            )
+            self.assertIs(supervisor, mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+            self.assertEqual(supervisor.worker_identity, identity)
+            status = mcp_server._worker_heartbeat_status()
+            self.assertTrue(status["supervised"])
+            self.assertEqual(status["worker_identity"], identity)
+            # Under pytest the thread is deliberately not auto-started; the
+            # supervisor is still attached by the production wiring.
+            registry = mwi.WorkerRegistry(path)
+            self.assertIsNotNone(registry.get(identity))
+
+    def test_supervision_not_started_before_registration_succeeds(self):
+        # (2) Heartbeat supervision is not started before registration succeeds.
+        with _production_worker_lifecycle(pin_registry=False) as (_path, mcp_server):
+            self.assertIsNone(mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+            self.assertIsNone(mcp_server._WORKER_IDENTITY)
+            identity = mcp_server._active_worker_identity()
+            self.assertIsNone(identity)
+            self.assertIsNone(mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+            status = mcp_server._worker_heartbeat_status()
+            self.assertFalse(status["supervised"])
+
+    def test_repeated_active_worker_identity_no_duplicate_supervisor_or_rows(self):
+        # (3) Repeated _active_worker_identity() calls create neither a second
+        # supervisor nor a second registry row.
+        with _production_worker_lifecycle() as (path, mcp_server):
+            first = mcp_server._active_worker_identity()
+            supervisor_first = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            second = mcp_server._active_worker_identity()
+            third = mcp_server._active_worker_identity()
+            self.assertEqual(first, second)
+            self.assertEqual(second, third)
+            self.assertIs(supervisor_first, mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+            registry = mwi.WorkerRegistry(path)
+            rows = registry.list_workers(status=None)
+            matching = [
+                row for row in rows if row.get("worker_identity") == first
+            ]
+            self.assertEqual(len(matching), 1)
+
+    def test_supervisor_receives_exact_registered_identity_fields(self):
+        # (4) Exact normalized identity, client, session, generation, process
+        # identity, and fencing epoch reach the supervisor from production
+        # expressions — not from shared helper constants on both sides.
+        client_raw = "Claude-Code"
+        session = "sess-prod-exact-fields"
+        instance = "inst-exact-fields"
+        with _production_worker_lifecycle(
+            client=client_raw, instance=instance, session=session, interval="45"
+        ) as (path, mcp_server):
+            identity = mcp_server._active_worker_identity()
+            self.assertIsNotNone(identity)
+            supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            self.assertIsNotNone(supervisor)
+            registry = mwi.WorkerRegistry(path)
+            record = registry.get(identity)
+            self.assertIsNotNone(record)
+
+            # Record side comes from register(); supervisor side from
+            # _start_worker_heartbeat. They must agree.
+            self.assertEqual(supervisor.worker_identity, record["worker_identity"])
+            self.assertEqual(int(supervisor.fencing_epoch), int(record["fencing_epoch"]))
+            self.assertEqual(supervisor.session_id, record["session_id"])
+            self.assertEqual(supervisor.generation_id, record["generation_id"])
+            self.assertEqual(
+                supervisor.client_name,
+                mwi.normalize_client_name(record["client_name"]),
+            )
+            self.assertEqual(supervisor.pid, int(record["pid"]))
+            self.assertEqual(supervisor.pid, os.getpid())
+            self.assertEqual(supervisor.session_id, session)
+            self.assertEqual(
+                supervisor.client_name, mwi.normalize_client_name(client_raw)
+            )
+            # Generation is process-minted, not a test helper constant.
+            self.assertEqual(supervisor.generation_id, mcp_server._WORKER_GENERATION)
+            self.assertTrue(str(supervisor.generation_id))
+
+    def test_failed_registration_leaves_no_active_supervisor(self):
+        # (5) Failed registration (no openable registry under pytest) leaves no
+        # supervisor. Fenced path covered below via claim_generation.
+        with _production_worker_lifecycle(pin_registry=False) as (_path, mcp_server):
+            identity = mcp_server._active_worker_identity()
+            self.assertIsNone(identity)
+            self.assertIsNone(mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+            self.assertTrue(mcp_server._WORKER_REGISTRATION_ATTEMPTED)
+
+    def test_fenced_registration_stops_production_supervisor(self):
+        # (5 continued) After production wiring attaches a supervisor, a fenced
+        # beat must stop it permanently and leave it non-running.
+        with _production_worker_lifecycle() as (path, mcp_server):
+            identity = mcp_server._active_worker_identity()
+            supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            self.assertIsNotNone(identity)
+            self.assertIsNotNone(supervisor)
+
+            registry = mwi.WorkerRegistry(path)
+            record = registry.get(identity)
+            claim = registry.claim_generation(
+                worker_identity=identity,
+                generation_id=record["generation_id"],
+            )
+            self.assertTrue(claim.get("claimed") or claim.get("success"))
+            # Bumped epoch fences the production-started supervisor's cached epoch.
+            self.assertNotEqual(
+                int(claim["fencing_epoch"]), int(supervisor.fencing_epoch)
+            )
+            result = supervisor.beat_once(now=NOW)
+            self.assertFalse(result.get("renewed"))
+            self.assertEqual(result.get("blocker_kind"), mwi.BLOCKER_FENCED)
+            self.assertIsNotNone(supervisor.stopped_reason)
+            self.assertFalse(supervisor.status()["running"])
+            # Restart is refused after a terminal stop.
+            self.assertFalse(supervisor.start()["started"])
+
+    def test_orderly_shutdown_stops_production_supervisor_idempotently(self):
+        # (6) Orderly shutdown stops the production-started supervisor idempotently.
+        with _production_worker_lifecycle() as (_path, mcp_server):
+            identity = mcp_server._active_worker_identity()
+            supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            self.assertIsNotNone(identity)
+            self.assertIsNotNone(supervisor)
+            # Simulate non-pytest start, then atexit-style stop.
+            started = supervisor.start()
+            self.assertTrue(started["started"])
+            first = supervisor.stop(reason="orderly shutdown")
+            second = supervisor.stop(reason="orderly shutdown")
+            self.assertTrue(first["stopped"])
+            self.assertTrue(second["stopped"])
+            self.assertEqual(supervisor.stopped_reason, "orderly shutdown")
+            self.assertFalse(supervisor.status()["running"])
+            self.assertFalse(supervisor.start()["started"])
+
+    def test_removing_production_wiring_would_fail_this_suite(self):
+        # (7) Explicit wiring proof: _active_worker_identity must invoke
+        # _start_worker_heartbeat. A spy confirms the production call site is
+        # reached with the registered identity; a missing call fails the assert.
+        with _production_worker_lifecycle() as (_path, mcp_server):
+            seen: list[dict] = []
+            real = mcp_server._start_worker_heartbeat
+
+            def spy(**kwargs):
+                seen.append(dict(kwargs))
+                return real(**kwargs)
+
+            mcp_server._start_worker_heartbeat = spy  # type: ignore[assignment]
+            try:
+                identity = mcp_server._active_worker_identity()
+            finally:
+                mcp_server._start_worker_heartbeat = real  # type: ignore[assignment]
+
+            self.assertIsNotNone(identity)
+            self.assertEqual(len(seen), 1, "production path must call _start_worker_heartbeat exactly once")
+            self.assertEqual(seen[0]["identity"], identity)
+            self.assertIsNotNone(seen[0]["fencing_epoch"])
+            self.assertEqual(seen[0]["hints"]["session_id"], "sess-prod-0001")
+            self.assertIsNotNone(mcp_server._WORKER_HEARTBEAT_SUPERVISOR)
+
+    def test_configured_interval_reaches_production_supervisor(self):
+        # (8) The configured heartbeat interval reaches the production supervisor.
+        with _production_worker_lifecycle(interval="30") as (_path, mcp_server):
+            identity = mcp_server._active_worker_identity()
+            supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            self.assertIsNotNone(identity)
+            self.assertIsNotNone(supervisor)
+            expected = mwi.heartbeat_interval_for(
+                mwi.DEFAULT_HEARTBEAT_TTL_SECONDS, "30"
+            )
+            self.assertAlmostEqual(supervisor.interval_seconds, expected)
+            self.assertAlmostEqual(supervisor.interval_seconds, 30.0)
+            status = mcp_server._worker_heartbeat_status()
+            self.assertAlmostEqual(status["heartbeat_interval_seconds"], 30.0)
+
+    def test_sqlite_work_remains_off_the_async_event_loop(self):
+        # (9) Heartbeat sqlite writes run on a daemon thread, not the asyncio loop.
+        with _production_worker_lifecycle(interval="0.05") as (path, mcp_server):
+            identity = mcp_server._active_worker_identity()
+            supervisor = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
+            self.assertIsNotNone(identity)
+            self.assertIsNotNone(supervisor)
+
+            registry = mwi.WorkerRegistry(path)
+            beat_threads: list[str] = []
+            real = registry.heartbeat
+
+            def observing(**kwargs):
+                beat_threads.append(threading.current_thread().name)
+                # Rebind the production supervisor's registry method path.
+                return real(**kwargs)
+
+            # Point the production supervisor at the observing registry methods
+            # by swapping the bound registry on the supervisor instance.
+            supervisor._registry.heartbeat = observing  # type: ignore[method-assign]
+            try:
+                started = supervisor.start()
+                self.assertTrue(started["started"])
+                thread = supervisor._thread
+                self.assertIsNotNone(thread)
+                self.assertTrue(thread.daemon)
+                self.assertIn("gitea-worker-heartbeat", thread.name)
+                # Force one beat immediately (the loop waits first).
+                supervisor.beat_once(now=NOW + timedelta(seconds=1))
+                # Also let the thread run at least one scheduled beat.
+                deadline = time.monotonic() + 5.0
+                while supervisor.beats_attempted < 2 and time.monotonic() < deadline:
+                    time.sleep(0.02)
+            finally:
+                supervisor.stop(reason="test teardown")
+                try:
+                    supervisor._registry.heartbeat = real  # type: ignore[method-assign]
+                except Exception:
+                    pass
+
+            self.assertGreaterEqual(supervisor.beats_attempted, 1)
+            # Synchronous beat_once may run on the test thread; the background
+            # loop must still be a distinct daemon worker-heartbeat thread.
+            self.assertTrue(
+                any("gitea-worker-heartbeat" in name for name in beat_threads)
+                or (
+                    supervisor._thread is not None
+                    and "gitea-worker-heartbeat" in supervisor._thread.name
+                )
+            )
+
+
 if __name__ == "__main__":
     unittest.main()