From b0868be6b34f03338dd9fb3c3eda65f42b0edca3 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 23:41:37 -0400 Subject: [PATCH] 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) --- gitea_config.py | 6 + ...est_issue_975_client_identity_heartbeat.py | 397 +++++++++++++++++- 2 files changed, 402 insertions(+), 1 deletion(-) diff --git a/gitea_config.py b/gitea_config.py index 86e721f..f99362d 100644 --- a/gitea_config.py +++ b/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 = ( diff --git a/tests/test_issue_975_client_identity_heartbeat.py b/tests/test_issue_975_client_identity_heartbeat.py index 6ad42a0..527d13d 100644 --- a/tests/test_issue_975_client_identity_heartbeat.py +++ b/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()