diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index a4b7274fd6..e470a0e1dc 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -777,6 +777,7 @@ jobs: run: >- python -m pytest -q tests/test_command_invocation.py + tests/test_context_provider_service_ownership.py tests/test_doctor_git_encoding.py tests/test_doctor_install_freshness.py tests/test_file_lock.py diff --git a/examples/context-provider-service-ownership-smoke.py b/examples/context-provider-service-ownership-smoke.py index 536c56893f..284d9279d7 100644 --- a/examples/context-provider-service-ownership-smoke.py +++ b/examples/context-provider-service-ownership-smoke.py @@ -5,6 +5,7 @@ import json import os +import subprocess import sys import tempfile from pathlib import Path @@ -48,6 +49,19 @@ def main() -> int: assert missing.verified is False assert missing.reason_code == "provider_service_ownership_receipt_unavailable" + exited_provider = subprocess.Popen([sys.executable, "-c", "pass"]) + exited_provider.wait() + write_receipt(receipt, generation="generation-1", pid=exited_provider.pid) + exited = load_context_provider_service_ownership( + receipt, + expected_provider="contract-provider", + ) + assert exited.verified is False + assert exited.reason_code == "provider_service_process_unavailable" + exited_public = exited.public_packet(required=True) + assert exited_public["ok"] is False + assert exited_public["progress_disposition"] == "not_started" + write_receipt(receipt, generation="generation-1", pid=os.getpid()) before = load_context_provider_service_ownership( receipt, diff --git a/loopx/capabilities/context_providers/service_ownership.py b/loopx/capabilities/context_providers/service_ownership.py index f532672bbf..c4369f5a38 100644 --- a/loopx/capabilities/context_providers/service_ownership.py +++ b/loopx/capabilities/context_providers/service_ownership.py @@ -2,12 +2,13 @@ import hashlib import json -import os import re from dataclasses import dataclass from pathlib import Path from typing import Mapping +from ...file_lock import process_is_alive + CONTEXT_PROVIDER_SERVICE_RECEIPT_SCHEMA_VERSION = ( "context_provider_service_ownership_receipt_v0" @@ -29,16 +30,6 @@ def _opaque_ref(kind: str, value: str) -> str: return f"{kind}-{digest}" -def _process_alive(pid: int) -> bool: - try: - os.kill(pid, 0) - except ProcessLookupError: - return False - except PermissionError: - return True - return True - - @dataclass(frozen=True) class ContextProviderServiceOwnership: provider: str @@ -158,7 +149,7 @@ def load_context_provider_service_ownership( reason_code="provider_service_ownership_receipt_invalid", ownership_mode=ownership_mode or "unknown", ) - alive = _process_alive(pid) + alive = process_is_alive(pid) return ContextProviderServiceOwnership( provider=provider, status="verified" if alive else "unavailable", diff --git a/tests/test_context_provider_service_ownership.py b/tests/test_context_provider_service_ownership.py new file mode 100644 index 0000000000..25e4e6feb1 --- /dev/null +++ b/tests/test_context_provider_service_ownership.py @@ -0,0 +1,95 @@ +"""Regression tests for the context-provider ownership liveness probe.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from loopx import file_lock +from loopx.capabilities.context_providers import service_ownership + + +def write_receipt(path: Path, *, pid: int) -> Path: + path.write_text( + json.dumps( + { + "schema_version": "context_provider_service_ownership_receipt_v0", + "provider": "contract-provider", + "service_ref": "public-index-service", + "ownership_mode": "persistent_external", + "generation": "generation-1", + "pid": pid, + "observed_at": "2026-07-12T04:30:00Z", + } + ), + encoding="utf-8", + ) + return path + + +def reaped_pid() -> int: + process = subprocess.Popen([sys.executable, "-c", "pass"]) + process.wait() + return process.pid + + +def load(path: Path) -> service_ownership.ContextProviderServiceOwnership: + return service_ownership.load_context_provider_service_ownership( + path, + expected_provider="contract-provider", + ) + + +def test_exited_provider_pid_is_typed_unavailable(tmp_path: Path) -> None: + ownership = load(write_receipt(tmp_path / "service.json", pid=reaped_pid())) + assert ownership.verified is False + assert ownership.status == "unavailable" + assert ownership.reason_code == "provider_service_process_unavailable" + assert ownership.process_alive is False + packet = ownership.public_packet(required=True) + assert packet["ok"] is False + assert packet["progress_disposition"] == "not_started" + + +def test_ownership_probe_delegates_to_the_shared_non_signaling_probe( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + probed: list[int] = [] + + def probe(pid: int) -> bool: + probed.append(pid) + return True + + monkeypatch.setattr(file_lock, "os", SimpleNamespace(name="nt")) + monkeypatch.setattr(file_lock, "_windows_process_is_alive", probe) + + ownership = load(write_receipt(tmp_path / "service.json", pid=os.getpid())) + assert probed == [os.getpid()] + assert ownership.verified is True + + +@pytest.mark.skipif( + os.name != "nt", + reason="signal 0 is a console control event on Windows, not a liveness probe", +) +def test_native_windows_ownership_probe_sends_no_process_signal( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def refuse_signal(*args: object, **kwargs: object) -> None: + raise AssertionError("the ownership probe must not send process signals") + + monkeypatch.setattr(os, "kill", refuse_signal) + + live = load(write_receipt(tmp_path / "live.json", pid=os.getpid())) + assert live.verified is True + assert live.process_alive is True + + exited = load(write_receipt(tmp_path / "exited.json", pid=reaped_pid())) + assert exited.verified is False + assert exited.reason_code == "provider_service_process_unavailable"