Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/python-tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
14 changes: 14 additions & 0 deletions examples/context-provider-service-ownership-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
Expand Down Expand Up @@ -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,
Expand Down
15 changes: 3 additions & 12 deletions loopx/capabilities/context_providers/service_ownership.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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",
Expand Down
95 changes: 95 additions & 0 deletions tests/test_context_provider_service_ownership.py
Original file line number Diff line number Diff line change
@@ -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"
Loading