sysadminandClaude Opus 4.8
b0868be6b3
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]>
1155 lines
46 KiB
Python
1155 lines
46 KiB
Python
"""Client-identity env recognition and the production heartbeat lifecycle (#975).
|
|
|
|
Two defects left behind by #948, covered here against the issue's acceptance
|
|
criteria:
|
|
|
|
1. ``GITEA_MCP_CLIENT`` / ``GITEA_MCP_CLIENT_INSTANCE`` /
|
|
``GITEA_MCP_CLIENT_SESSION`` were consumed by production while absent from
|
|
the recognized-key allowlist, so the peer-env scan classified them as
|
|
unsupported overrides and the capability resolver refused every mutation
|
|
fleet-wide.
|
|
2. ``WorkerRegistry.heartbeat()`` had no production caller, so
|
|
``last_heartbeat_at`` never left ``started_at`` and ``heartbeat_ttl_seconds``
|
|
became a hard cap on attachment rather than a liveness window.
|
|
|
|
Every TTL assertion here uses an injected clock (AC13). No test waits for a
|
|
real TTL. The one test that exercises the real thread loop uses a
|
|
millisecond-scale interval and polls with a bounded timeout, never a TTL.
|
|
|
|
All client, session, and generation identifiers are synthetic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
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
|
|
|
|
|
|
NOW = datetime(2026, 7, 30, 0, 40, 9, tzinfo=timezone.utc)
|
|
|
|
TTL = 900.0
|
|
|
|
#: The three keys production reads as client identity.
|
|
CLIENT_IDENTITY_ENV_KEYS = (
|
|
"GITEA_MCP_CLIENT",
|
|
"GITEA_MCP_CLIENT_INSTANCE",
|
|
"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."""
|
|
handle, path = tempfile.mkstemp(suffix=".sqlite3")
|
|
os.close(handle)
|
|
os.unlink(path)
|
|
return mwi.WorkerRegistry(path)
|
|
|
|
|
|
def _attach(
|
|
registry: mwi.WorkerRegistry,
|
|
*,
|
|
client: str = "claude_code",
|
|
session: str = "sess-0001",
|
|
generation: str = "gen-0001",
|
|
profile: str = "prgs-author",
|
|
role: str = "author",
|
|
pid: int = 4242,
|
|
now: datetime = NOW,
|
|
ttl: float = TTL,
|
|
) -> dict:
|
|
"""Register one synthetic worker and return the outcome plus its identity."""
|
|
identity = mwi.generate_worker_identity(client, session, now=now)
|
|
outcome = registry.register(
|
|
worker_identity=identity,
|
|
client_name=client,
|
|
client_instance_id=f"inst-{session}",
|
|
session_id=session,
|
|
generation_id=generation,
|
|
role=role,
|
|
profile=profile,
|
|
pid=pid,
|
|
heartbeat_ttl_seconds=ttl,
|
|
now=now,
|
|
)
|
|
outcome["identity"] = identity
|
|
return outcome
|
|
|
|
|
|
def _supervisor(
|
|
registry: mwi.WorkerRegistry,
|
|
outcome: dict,
|
|
*,
|
|
client: str = "claude_code",
|
|
session: str = "sess-0001",
|
|
generation: str = "gen-0001",
|
|
pid: int = 4242,
|
|
ttl: float = TTL,
|
|
interval_seconds: float | None = None,
|
|
clock=None,
|
|
) -> mwi.WorkerHeartbeatSupervisor:
|
|
"""A supervisor bound to *outcome*'s registration. Never started here."""
|
|
return mwi.WorkerHeartbeatSupervisor(
|
|
registry,
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
session_id=session,
|
|
generation_id=generation,
|
|
client_name=client,
|
|
pid=pid,
|
|
ttl_seconds=ttl,
|
|
interval_seconds=interval_seconds,
|
|
clock=clock,
|
|
)
|
|
|
|
|
|
class _ManualClock:
|
|
"""Controlled time. Nothing in this suite sleeps to reach a TTL."""
|
|
|
|
def __init__(self, start: datetime = NOW) -> None:
|
|
self.now = start
|
|
|
|
def __call__(self) -> datetime:
|
|
return self.now
|
|
|
|
def advance(self, seconds: float) -> datetime:
|
|
self.now = self.now + timedelta(seconds=seconds)
|
|
return self.now
|
|
|
|
|
|
# --- AC1-AC3: environment recognition ------------------------------------
|
|
|
|
|
|
class ClientIdentityEnvRecognitionTests(unittest.TestCase):
|
|
"""AC1-AC3: the keys production consumes are the keys the gate accepts."""
|
|
|
|
def test_each_client_identity_key_is_recognized(self):
|
|
# AC1: individually asserted, so a partial fix cannot pass.
|
|
for key in CLIENT_IDENTITY_ENV_KEYS:
|
|
with self.subTest(key=key):
|
|
self.assertIn(
|
|
key,
|
|
gitea_config.RECOGNIZED_GITEA_ENV_KEYS,
|
|
f"{key} is read by the server but not recognized by the gate",
|
|
)
|
|
|
|
def test_recognized_client_identity_keys_are_not_unconsumed(self):
|
|
# AC3 at the classification layer the resolver reads: with only these
|
|
# keys set there is no unsupported override to raise a blocker from.
|
|
env = {
|
|
"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
|
|
# beginning GITEA_MCP_CLIENT, including typos and future unknowns.
|
|
env = {"GITEA_MCP_CLIENT_SOMETHING_ELSE": "x"}
|
|
self.assertIn(
|
|
"GITEA_MCP_CLIENT_SOMETHING_ELSE",
|
|
gitea_config.get_unconsumed_gitea_env_overrides(env),
|
|
)
|
|
|
|
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. 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):
|
|
"""`unsupported-env:` must not be swallowed where `stale-runtime:` is not."""
|
|
|
|
def test_both_reason_families_are_hard_prefixes(self):
|
|
import gitea_mcp_server as mcp_server
|
|
|
|
self.assertIn("stale-runtime:", mcp_server.RUNTIME_DIAGNOSTIC_HARD_PREFIXES)
|
|
self.assertIn("unsupported-env:", mcp_server.RUNTIME_DIAGNOSTIC_HARD_PREFIXES)
|
|
|
|
def test_unsupported_env_message_is_recognized_as_hard(self):
|
|
import gitea_mcp_server as mcp_server
|
|
|
|
message = (
|
|
"unsupported-env: Unsupported GITEA_* environment variable override(s) "
|
|
"detected: GITEA_INVENTED=1. Unknown env overrides are unsupported."
|
|
)
|
|
self.assertTrue(
|
|
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 ------------------------------------------------------
|
|
|
|
|
|
class HeartbeatIntervalPolicyTests(unittest.TestCase):
|
|
"""The interval must always leave room for lost beats inside the TTL."""
|
|
|
|
def test_default_interval_is_one_third_of_ttl(self):
|
|
self.assertAlmostEqual(mwi.heartbeat_interval_for(900.0), 300.0)
|
|
|
|
def test_interval_is_capped_at_half_the_ttl(self):
|
|
# An override may not exceed the ceiling, or it would expire the very
|
|
# registration it exists to renew.
|
|
self.assertAlmostEqual(mwi.heartbeat_interval_for(900.0, 5000), 450.0)
|
|
|
|
def test_override_below_the_ceiling_is_honoured(self):
|
|
self.assertAlmostEqual(mwi.heartbeat_interval_for(900.0, 120), 120.0)
|
|
|
|
def test_unparsable_or_nonpositive_override_falls_back_to_policy(self):
|
|
for bad in ("", "abc", "0", "-5", None):
|
|
with self.subTest(override=bad):
|
|
self.assertAlmostEqual(mwi.heartbeat_interval_for(900.0, bad), 300.0)
|
|
|
|
def test_interval_is_always_strictly_below_the_ttl(self):
|
|
for ttl in (1.0, 30.0, 900.0, 14400.0):
|
|
with self.subTest(ttl=ttl):
|
|
self.assertLess(mwi.heartbeat_interval_for(ttl), ttl)
|
|
|
|
|
|
# --- AC4-AC6: a healthy worker stays owned --------------------------------
|
|
|
|
|
|
class HealthyWorkerRemainsOwnedTests(unittest.TestCase):
|
|
"""AC4-AC6: the TTL becomes a liveness window instead of an attachment cap."""
|
|
|
|
def test_unsupervised_worker_goes_unowned_at_the_ttl(self):
|
|
# The regression itself, asserted so the fix cannot be mistaken for a
|
|
# TTL change: without a beater the row still expires exactly as before.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
record = registry.get(outcome["identity"])
|
|
self.assertEqual(record["started_at"], record["last_heartbeat_at"])
|
|
|
|
after_ttl = NOW + timedelta(seconds=TTL + 1)
|
|
liveness = registry.is_live(record, now=after_ttl, pid_alive=True)
|
|
self.assertFalse(liveness["live"])
|
|
|
|
assessment = mwi.assess_provenance(
|
|
registry=registry,
|
|
worker_identity=outcome["identity"],
|
|
generation_id="gen-0001",
|
|
env={"GITEA_CLIENT_MANAGED": "1"},
|
|
native_transport_bound=True,
|
|
now=after_ttl,
|
|
pid_alive_probe=lambda _pid: True,
|
|
)
|
|
self.assertEqual(assessment["session_ownership"], "unowned")
|
|
self.assertEqual(assessment["blocker_kind"], mwi.BLOCKER_NO_ATTACHMENT)
|
|
|
|
def test_supervised_worker_stays_owned_past_two_full_ttls(self):
|
|
# AC4: beyond two complete TTL periods of simulated time.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(registry, outcome, clock=clock)
|
|
|
|
interval = supervisor.interval_seconds
|
|
self.assertLess(interval, TTL)
|
|
|
|
elapsed = 0.0
|
|
target = TTL * 2 + interval
|
|
while elapsed < target:
|
|
clock.advance(interval)
|
|
elapsed += interval
|
|
result = supervisor.beat_once()
|
|
self.assertTrue(result["renewed"], result.get("reasons"))
|
|
|
|
assessment = mwi.assess_provenance(
|
|
registry=registry,
|
|
worker_identity=outcome["identity"],
|
|
generation_id="gen-0001",
|
|
env={"GITEA_CLIENT_MANAGED": "1"},
|
|
native_transport_bound=True,
|
|
now=clock.now,
|
|
pid_alive_probe=lambda _pid: True,
|
|
)
|
|
self.assertEqual(
|
|
assessment["session_ownership"],
|
|
"owned",
|
|
f"ownership lost after {elapsed:.0f}s of simulated time",
|
|
)
|
|
|
|
self.assertGreater(elapsed, TTL * 2)
|
|
|
|
def test_last_heartbeat_advances_and_diverges_from_started_at(self):
|
|
# AC5.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(registry, outcome, clock=clock)
|
|
|
|
before = registry.get(outcome["identity"])
|
|
clock.advance(300)
|
|
supervisor.beat_once()
|
|
after = registry.get(outcome["identity"])
|
|
|
|
self.assertEqual(before["started_at"], after["started_at"])
|
|
self.assertNotEqual(after["started_at"], after["last_heartbeat_at"])
|
|
self.assertGreater(after["last_heartbeat_at"], before["last_heartbeat_at"])
|
|
|
|
def test_repeated_heartbeats_create_no_duplicate_rows(self):
|
|
# AC6: heartbeating is not a second registration path.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(registry, outcome, clock=clock)
|
|
|
|
for _ in range(5):
|
|
clock.advance(300)
|
|
self.assertTrue(supervisor.beat_once()["renewed"])
|
|
|
|
workers = registry.list_workers()
|
|
self.assertEqual(len(workers), 1)
|
|
self.assertEqual(workers[0]["worker_identity"], outcome["identity"])
|
|
self.assertEqual(supervisor.status()["beats_renewed"], 5)
|
|
|
|
|
|
# --- AC7, AC8, AC12: only the owner may renew -----------------------------
|
|
|
|
|
|
class HeartbeatOwnershipFencingTests(unittest.TestCase):
|
|
"""AC7/AC8/AC12: one worker can never refresh another worker's row."""
|
|
|
|
def test_two_client_names_stay_distinct_rows_and_identities(self):
|
|
# AC7.
|
|
registry = _registry()
|
|
first = _attach(registry, client="claude_code", session="sess-a")
|
|
second = _attach(registry, client="gemini", session="sess-b")
|
|
|
|
self.assertNotEqual(first["identity"], second["identity"])
|
|
rows = {w["worker_identity"]: w for w in registry.list_workers()}
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual(rows[first["identity"]]["client_name"], "claude_code")
|
|
self.assertEqual(rows[second["identity"]]["client_name"], "gemini")
|
|
|
|
def test_wrong_session_cannot_refresh_the_row(self):
|
|
# AC8, session dimension.
|
|
registry = _registry()
|
|
outcome = _attach(registry, session="sess-real")
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
expected_session_id="sess-impostor",
|
|
)
|
|
self.assertFalse(result["renewed"])
|
|
self.assertEqual(result["blocker_kind"], mwi.BLOCKER_FENCED)
|
|
self.assertIn(
|
|
"session_id", [field for field, _r, _p in result["expectation_drift"]]
|
|
)
|
|
|
|
def test_wrong_generation_cannot_refresh_the_row(self):
|
|
# AC8, generation dimension.
|
|
registry = _registry()
|
|
outcome = _attach(registry, generation="gen-real")
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
expected_generation_id="gen-other",
|
|
)
|
|
self.assertFalse(result["renewed"])
|
|
self.assertEqual(result["blocker_kind"], mwi.BLOCKER_FENCED)
|
|
|
|
def test_wrong_client_name_cannot_refresh_the_row(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry, client="claude_code")
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
expected_client_name="gemini",
|
|
)
|
|
self.assertFalse(result["renewed"])
|
|
self.assertEqual(result["blocker_kind"], mwi.BLOCKER_FENCED)
|
|
|
|
def test_wrong_pid_cannot_refresh_the_row(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry, pid=4242)
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
expected_pid=9999,
|
|
)
|
|
self.assertFalse(result["renewed"])
|
|
self.assertEqual(result["blocker_kind"], mwi.BLOCKER_FENCED)
|
|
|
|
def test_namespaces_of_one_application_share_a_client_name(self):
|
|
# Several namespaces of one application must stay one client, while
|
|
# separate applications stay distinct. Aliases normalise, and the
|
|
# expectation check compares normalised values, so an author and a
|
|
# reviewer namespace of the same product are not treated as impostors.
|
|
registry = _registry()
|
|
outcome = _attach(registry, client="claude")
|
|
self.assertEqual(
|
|
registry.get(outcome["identity"])["client_name"], "claude_code"
|
|
)
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
expected_client_name="claude_desktop",
|
|
)
|
|
self.assertTrue(result["renewed"], result.get("reasons"))
|
|
|
|
def test_matching_expectations_renew_normally(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
expected_session_id="sess-0001",
|
|
expected_generation_id="gen-0001",
|
|
expected_client_name="claude_code",
|
|
expected_pid=4242,
|
|
)
|
|
self.assertTrue(result["renewed"], result.get("reasons"))
|
|
|
|
def test_omitted_expectations_preserve_pre_fix_behaviour(self):
|
|
# AC12: the #948 contract is unchanged for callers presenting nothing.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
self.assertTrue(
|
|
registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=outcome["fencing_epoch"],
|
|
)["renewed"]
|
|
)
|
|
|
|
def test_stale_fencing_epoch_is_still_refused(self):
|
|
# AC12: the pre-existing epoch fence is untouched.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
result = registry.heartbeat(
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=int(outcome["fencing_epoch"]) + 1,
|
|
)
|
|
self.assertFalse(result["renewed"])
|
|
self.assertEqual(result["blocker_kind"], mwi.BLOCKER_FENCED)
|
|
|
|
def test_fenced_supervisor_stops_permanently(self):
|
|
# A fenced session must never beat its way back into ownership.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
supervisor = mwi.WorkerHeartbeatSupervisor(
|
|
registry,
|
|
worker_identity=outcome["identity"],
|
|
fencing_epoch=int(outcome["fencing_epoch"]) + 1,
|
|
session_id="sess-0001",
|
|
generation_id="gen-0001",
|
|
client_name="claude_code",
|
|
pid=4242,
|
|
ttl_seconds=TTL,
|
|
clock=_ManualClock(),
|
|
)
|
|
first = supervisor.beat_once()
|
|
self.assertFalse(first["renewed"])
|
|
self.assertIsNotNone(supervisor.stopped_reason)
|
|
|
|
# A second attempt does not even reach the registry.
|
|
second = supervisor.beat_once()
|
|
self.assertFalse(second["renewed"])
|
|
self.assertFalse(second["beat_attempted"])
|
|
self.assertEqual(supervisor.status()["beats_attempted"], 1)
|
|
|
|
def test_missing_registration_stops_the_supervisor(self):
|
|
registry = _registry()
|
|
supervisor = mwi.WorkerHeartbeatSupervisor(
|
|
registry,
|
|
worker_identity=mwi.generate_worker_identity("grok", "sess-gone", now=NOW),
|
|
fencing_epoch=1,
|
|
ttl_seconds=TTL,
|
|
clock=_ManualClock(),
|
|
)
|
|
result = supervisor.beat_once()
|
|
self.assertEqual(result["blocker_kind"], mwi.BLOCKER_NO_ATTACHMENT)
|
|
self.assertIsNotNone(supervisor.stopped_reason)
|
|
|
|
|
|
# --- AC9, AC10, AC11: stopping and expiry ---------------------------------
|
|
|
|
|
|
class ShutdownAndExpiryTests(unittest.TestCase):
|
|
"""AC9-AC11: orderly shutdown stops beating; dead workers still expire."""
|
|
|
|
def test_stop_ends_the_heartbeat(self):
|
|
# AC9.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(registry, outcome, clock=clock)
|
|
|
|
clock.advance(300)
|
|
self.assertTrue(supervisor.beat_once()["renewed"])
|
|
supervisor.stop(reason="orderly shutdown")
|
|
|
|
self.assertFalse(supervisor.status()["running"])
|
|
self.assertEqual(supervisor.stopped_reason, "orderly shutdown")
|
|
|
|
clock.advance(300)
|
|
after_stop = supervisor.beat_once()
|
|
self.assertFalse(after_stop["renewed"])
|
|
self.assertFalse(after_stop["beat_attempted"])
|
|
|
|
def test_stop_is_idempotent_and_keeps_the_first_reason(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
supervisor = _supervisor(registry, outcome, clock=_ManualClock())
|
|
supervisor.stop(reason="orderly shutdown")
|
|
supervisor.stop(reason="second call")
|
|
self.assertEqual(supervisor.stopped_reason, "orderly shutdown")
|
|
|
|
def test_stopped_worker_becomes_unowned_once_the_ttl_elapses(self):
|
|
# AC10: stopping beating restores normal expiration; the fix does not
|
|
# make a worker immortal.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(registry, outcome, clock=clock)
|
|
|
|
clock.advance(300)
|
|
supervisor.beat_once()
|
|
supervisor.stop(reason="orderly shutdown")
|
|
|
|
last_beat = clock.now
|
|
clock.advance(TTL + 1)
|
|
record = registry.get(outcome["identity"])
|
|
liveness = registry.is_live(record, now=clock.now, pid_alive=True)
|
|
self.assertFalse(liveness["live"])
|
|
self.assertGreater(liveness["heartbeat_age_seconds"], TTL)
|
|
|
|
assessment = mwi.assess_provenance(
|
|
registry=registry,
|
|
worker_identity=outcome["identity"],
|
|
generation_id="gen-0001",
|
|
env={"GITEA_CLIENT_MANAGED": "1"},
|
|
native_transport_bound=True,
|
|
now=clock.now,
|
|
pid_alive_probe=lambda _pid: True,
|
|
)
|
|
self.assertEqual(assessment["session_ownership"], "unowned")
|
|
self.assertLess(last_beat, clock.now)
|
|
|
|
def test_dead_worker_expires_even_with_a_live_pid(self):
|
|
# A recycled or long-lived pid must not keep a silent worker owned.
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
record = registry.get(outcome["identity"])
|
|
liveness = registry.is_live(
|
|
record, now=NOW + timedelta(seconds=TTL + 1), pid_alive=True
|
|
)
|
|
self.assertFalse(liveness["live"])
|
|
self.assertFalse(liveness["heartbeat_fresh"])
|
|
|
|
def test_one_worker_expiring_does_not_stale_healthy_workers(self):
|
|
# AC11.
|
|
registry = _registry()
|
|
healthy = _attach(registry, client="claude_code", session="sess-live")
|
|
abandoned = _attach(registry, client="gemini", session="sess-dead")
|
|
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(
|
|
registry,
|
|
healthy,
|
|
client="claude_code",
|
|
session="sess-live",
|
|
clock=clock,
|
|
)
|
|
|
|
elapsed = 0.0
|
|
while elapsed < TTL + 300:
|
|
clock.advance(supervisor.interval_seconds)
|
|
elapsed += supervisor.interval_seconds
|
|
self.assertTrue(supervisor.beat_once()["renewed"])
|
|
|
|
healthy_live = registry.is_live(
|
|
registry.get(healthy["identity"]), now=clock.now, pid_alive=True
|
|
)
|
|
abandoned_live = registry.is_live(
|
|
registry.get(abandoned["identity"]), now=clock.now, pid_alive=True
|
|
)
|
|
self.assertTrue(healthy_live["live"])
|
|
self.assertFalse(abandoned_live["live"])
|
|
|
|
|
|
# --- Failure handling and observability -----------------------------------
|
|
|
|
|
|
class HeartbeatFailureHandlingTests(unittest.TestCase):
|
|
"""A heartbeat failure never grants ownership and never crashes a caller."""
|
|
|
|
def test_transient_exception_is_counted_and_beating_continues(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
clock = _ManualClock()
|
|
supervisor = _supervisor(registry, outcome, clock=clock)
|
|
|
|
calls = {"n": 0}
|
|
real = registry.heartbeat
|
|
|
|
def flaky(**kwargs):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise RuntimeError("database is locked")
|
|
return real(**kwargs)
|
|
|
|
registry.heartbeat = flaky # type: ignore[method-assign]
|
|
try:
|
|
clock.advance(300)
|
|
first = supervisor.beat_once()
|
|
self.assertFalse(first["renewed"])
|
|
self.assertTrue(first["transient"])
|
|
self.assertIsNone(supervisor.stopped_reason)
|
|
|
|
clock.advance(300)
|
|
second = supervisor.beat_once()
|
|
self.assertTrue(second["renewed"], second.get("reasons"))
|
|
finally:
|
|
registry.heartbeat = real # type: ignore[method-assign]
|
|
|
|
status = supervisor.status()
|
|
self.assertEqual(status["transient_failures"], 1)
|
|
self.assertEqual(status["beats_renewed"], 1)
|
|
self.assertTrue(status["supervised"])
|
|
|
|
def test_status_reports_an_unstarted_supervisor_honestly(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
status = _supervisor(registry, outcome, clock=_ManualClock()).status()
|
|
self.assertTrue(status["supervised"])
|
|
self.assertFalse(status["started"])
|
|
self.assertFalse(status["running"])
|
|
self.assertEqual(status["beats_renewed"], 0)
|
|
self.assertIsNone(status["stopped_reason"])
|
|
|
|
def test_runtime_context_reports_unsupervised_when_none_attached(self):
|
|
# The observability surface must distinguish "no supervisor" from
|
|
# "supervisor with no beats yet", or the regression is invisible again.
|
|
import gitea_mcp_server as mcp_server
|
|
|
|
saved = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
|
|
mcp_server._WORKER_HEARTBEAT_SUPERVISOR = None
|
|
try:
|
|
status = mcp_server._worker_heartbeat_status()
|
|
finally:
|
|
mcp_server._WORKER_HEARTBEAT_SUPERVISOR = saved
|
|
self.assertFalse(status["supervised"])
|
|
self.assertFalse(status["running"])
|
|
self.assertTrue(status["reasons"])
|
|
|
|
def test_runtime_context_surfaces_an_attached_supervisor(self):
|
|
import gitea_mcp_server as mcp_server
|
|
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
supervisor = _supervisor(registry, outcome, clock=_ManualClock())
|
|
|
|
saved = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
|
|
mcp_server._WORKER_HEARTBEAT_SUPERVISOR = supervisor
|
|
try:
|
|
status = mcp_server._worker_heartbeat_status()
|
|
finally:
|
|
mcp_server._WORKER_HEARTBEAT_SUPERVISOR = saved
|
|
|
|
self.assertTrue(status["supervised"])
|
|
self.assertEqual(status["worker_identity"], outcome["identity"])
|
|
self.assertEqual(status["heartbeat_ttl_seconds"], TTL)
|
|
self.assertLess(status["heartbeat_interval_seconds"], TTL)
|
|
|
|
def test_status_failure_degrades_instead_of_raising(self):
|
|
import gitea_mcp_server as mcp_server
|
|
|
|
class Exploding:
|
|
def status(self):
|
|
raise RuntimeError("boom")
|
|
|
|
saved = mcp_server._WORKER_HEARTBEAT_SUPERVISOR
|
|
mcp_server._WORKER_HEARTBEAT_SUPERVISOR = Exploding()
|
|
try:
|
|
status = mcp_server._worker_heartbeat_status()
|
|
finally:
|
|
mcp_server._WORKER_HEARTBEAT_SUPERVISOR = saved
|
|
|
|
self.assertTrue(status["supervised"])
|
|
self.assertFalse(status["running"])
|
|
self.assertTrue(status["reasons"])
|
|
|
|
|
|
class SupervisorThreadLoopTests(unittest.TestCase):
|
|
"""The real thread, on a millisecond interval. Never waits for a TTL."""
|
|
|
|
def test_thread_beats_repeatedly_then_stops_on_request(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
beats = threading.Event()
|
|
seen = {"n": 0}
|
|
real = registry.heartbeat
|
|
|
|
def counting(**kwargs):
|
|
result = real(**kwargs)
|
|
seen["n"] += 1
|
|
if seen["n"] >= 3:
|
|
beats.set()
|
|
return result
|
|
|
|
registry.heartbeat = counting # type: ignore[method-assign]
|
|
supervisor = _supervisor(registry, outcome, ttl=1.0, interval_seconds=0.01)
|
|
try:
|
|
started = supervisor.start()
|
|
self.assertTrue(started["started"])
|
|
self.assertTrue(
|
|
beats.wait(timeout=10.0), "heartbeat thread produced no beats"
|
|
)
|
|
self.assertTrue(supervisor.status()["running"])
|
|
finally:
|
|
supervisor.stop(reason="test teardown")
|
|
registry.heartbeat = real # type: ignore[method-assign]
|
|
|
|
self.assertGreaterEqual(supervisor.status()["beats_renewed"], 3)
|
|
self.assertFalse(supervisor.status()["running"])
|
|
|
|
# Stopping is prompt: the loop waits on an Event, not a sleep.
|
|
deadline = time.monotonic() + 5.0
|
|
while supervisor._thread is not None and supervisor._thread.is_alive():
|
|
if time.monotonic() > deadline:
|
|
self.fail("heartbeat thread did not exit after stop()")
|
|
time.sleep(0.01)
|
|
|
|
def test_start_is_idempotent(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
supervisor = _supervisor(registry, outcome, ttl=1.0, interval_seconds=0.05)
|
|
try:
|
|
self.assertFalse(supervisor.start()["already_running"])
|
|
self.assertTrue(supervisor.start()["already_running"])
|
|
finally:
|
|
supervisor.stop(reason="test teardown")
|
|
|
|
def test_start_refuses_after_a_terminal_stop(self):
|
|
registry = _registry()
|
|
outcome = _attach(registry)
|
|
supervisor = _supervisor(registry, outcome, clock=_ManualClock())
|
|
supervisor.stop(reason="orderly shutdown")
|
|
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()
|