diff --git a/tests/control_plane/test_checkpoint_retention.py b/tests/control_plane/test_checkpoint_retention.py index 4e9b510234..09a01470af 100644 --- a/tests/control_plane/test_checkpoint_retention.py +++ b/tests/control_plane/test_checkpoint_retention.py @@ -1,7 +1,6 @@ """Real local artifact owner, retention CLI and filesystem recovery negatives.""" import json import os -from pathlib import Path import threading from types import SimpleNamespace diff --git a/tests/control_plane_ts/content_digest_single_owner.test.ts b/tests/control_plane_ts/content_digest_single_owner.test.ts index d0f9cc73e4..8a683c4c3c 100644 --- a/tests/control_plane_ts/content_digest_single_owner.test.ts +++ b/tests/control_plane_ts/content_digest_single_owner.test.ts @@ -138,6 +138,7 @@ const CANONICAL_CONSUMERS = [ "control_plane/quota/refresh_external_delivery.ts", "control_plane/quota/turn_envelope.ts", "control_plane/runtime/checkpoint_artifacts.ts", + "control_plane/runtime/checkpoint_retention.ts", "control_plane/runtime/execution_identity.ts", "control_plane/runtime/usage_statistics_cycles.ts", "control_plane/runtime/usage_statistics_goal_contract.ts", diff --git a/tests/control_plane_ts/replan_semantics.test.ts b/tests/control_plane_ts/replan_semantics.test.ts index 563ecbbd30..d45137dcf7 100644 --- a/tests/control_plane_ts/replan_semantics.test.ts +++ b/tests/control_plane_ts/replan_semantics.test.ts @@ -84,7 +84,11 @@ test("receipt-bound reentry keeps original canonical proof separate from the nex transition_candidates: []}).obligation as JsonObject; assert.equal(unchanged.selection_binding, "heartbeat_receipt"); assert.equal(unchanged.resolution_mode, undefined); - assert.equal(projectReplanSemantics({...request, selected_obligation_id: null}).obligation, current); + // A receipt admitted without replan work cannot adopt a later frontier. + assert.equal(projectReplanSemantics({...request, selected_obligation_id: null}).obligation, null); + // A fresh, unguarded read still exposes that current frontier for admission. + assert.equal(projectReplanSemantics({...request, guard_scoped: false, + selected_obligation_id: null}).obligation, current); }); test("verified original writeback projects only its outstanding debit even after frontier drift", () => { diff --git a/tests/test_managed_effect_runtime_fixture.py b/tests/test_managed_effect_runtime_fixture.py index 1be72ed3c9..71441cc2fc 100644 --- a/tests/test_managed_effect_runtime_fixture.py +++ b/tests/test_managed_effect_runtime_fixture.py @@ -14,6 +14,17 @@ from loopx.control_plane import effect_runtime +@pytest.fixture +def nested_pytest_environment(tmp_path): + repository = Path(__file__).resolve().parents[1] + # The nested session's import-time leak census must not claim caller writes. + home = tmp_path / "nested-home" + home.mkdir() + return {**os.environ, "HOME": str(home), "USERPROFILE": str(home), "PYTHONPATH": os.pathsep.join([ + str(repository / "tests"), str(repository), + ])} + + def _has_exited(pid): if not effect_runtime._pid_is_alive(pid): return True @@ -30,7 +41,9 @@ def _has_exited(pid): @pytest.mark.parametrize("failing", [False, True]) @pytest.mark.parametrize("retention", ["all", "none"]) -def test_pytest_retires_source_and_copied_package_runtimes(tmp_path, failing, retention): +def test_pytest_retires_source_and_copied_package_runtimes( + tmp_path, failing, retention, nested_pytest_environment, +): repository = Path(__file__).resolve().parents[1] records = tmp_path / "runtimes.jsonl" probe = tmp_path / "test_disposable_namespace.py" @@ -66,15 +79,12 @@ def test_owned_runtime(tmp_path, copied): stream.write(child.stdout.strip() + "\\n") assert not {failing!r}, "fixture failure retained" '''), encoding="utf-8") - environment = {**os.environ, "PYTHONPATH": os.pathsep.join([ - str(repository / "tests"), str(repository), - ])} observed = [] try: result = subprocess.run([ sys.executable, "-m", "pytest", "-q", "-p", "conftest", "-o", f"tmp_path_retention_policy={retention}", str(probe), - ], cwd=tmp_path, env=environment, capture_output=True, text=True, timeout=120) + ], cwd=tmp_path, env=nested_pytest_environment, capture_output=True, text=True, timeout=120) assert result.returncode == (1 if failing else 0), result.stdout + result.stderr if failing: assert "2 failed" in result.stdout and "fixture failure retained" in result.stdout @@ -94,6 +104,28 @@ def test_owned_runtime(tmp_path, copied): ) +@pytest.mark.parametrize("writer_home", ["caller", "nested"]) +def test_nested_pytest_attributes_default_route_writes_to_its_own_home( + tmp_path, nested_pytest_environment, writer_home, +): + home = Path(os.environ["HOME"] if writer_home == "caller" else nested_pytest_environment["HOME"]) + leaked_route = home / ".loopx" + assert not leaked_route.exists() + probe = tmp_path / "test_default_route_writer.py" + probe.write_text(textwrap.dedent(f''' + from pathlib import Path + + def test_writer(): + Path({str(leaked_route)!r}).mkdir() + '''), encoding="utf-8") + result = subprocess.run([ + sys.executable, "-m", "pytest", "-q", "-p", "conftest", str(probe), + ], cwd=tmp_path, env=nested_pytest_environment, capture_output=True, text=True, timeout=120) + owns_write = writer_home == "nested" + assert result.returncode == (1 if owns_write else 0), result.stdout + result.stderr + assert ("loopx default runtime route leak" in result.stdout) == owns_write + + def test_cleanup_preserves_foreign_namespace_and_rejects_wrong_pid(tmp_path, monkeypatch): foreign = tmp_path / "foreign" owned = tmp_path / "owned" diff --git a/tests/test_usage_goal.py b/tests/test_usage_goal.py index a8f1d99f0b..8d8567af04 100644 --- a/tests/test_usage_goal.py +++ b/tests/test_usage_goal.py @@ -1,5 +1,9 @@ """Host observer remains outside execution authority and never waits for HTTP.""" import json +import os +from pathlib import Path +import subprocess +import sys import threading import time @@ -8,6 +12,51 @@ from loopx import usage_goal, usage_ping +@pytest.fixture +def usage_transport(tmp_path, monkeypatch): + """Own opt-in and detached children through failed assertions as well as success.""" + monkeypatch.setattr(usage_ping, "select_default_runtime_root", lambda: tmp_path) + for name in ("CI", "DO_NOT_TRACK", "LOOPX_USAGE_PING", "LOOPX_USAGE_POLICY"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("LOOPX_USAGE_PING_ENDPOINT", "http://127.0.0.1:1/v1/ping") + popen = subprocess.Popen + children = [] + + def spawn(command, *args, **kwargs): + # Preserve the real detach flags, cwd, stdin and process entry point. + # Only replace discarded stderr in this explicitly opted-in fixture. + if kwargs.get("stderr") != subprocess.DEVNULL: + return popen(command, *args, **kwargs) + error = tmp_path / f"usage-child-{len(children)}.stderr" + with error.open("wb") as stream: + child = popen(command, *args, **{**kwargs, "stderr": stream}) + children.append((child, error)) + return child + + def diagnostics(): + return json.dumps([ + {"returncode": child.poll(), "stderr": error.read_text(errors="replace")[-2048:]} + for child, error in children + ]) + + monkeypatch.setattr(subprocess, "Popen", spawn) + try: + usage_ping.control("enable") + yield diagnostics + finally: + try: + # The TS consent lock fences later descendants too. Reaping the + # first Python hop alone would not stop its detached Node sender. + usage_ping.control("disable") + finally: + for child, _ in children: + try: + child.wait(timeout=4) + except subprocess.TimeoutExpired: + child.kill() + child.wait(timeout=4) + + def test_telemetry_failure_cannot_replace_host_exception(tmp_path, monkeypatch): monkeypatch.setattr(usage_ping, "select_default_runtime_root", lambda: tmp_path) usage_ping.state_path().write_text(json.dumps({"generation": "fixture", "consent": "enabled"})) @@ -67,15 +116,7 @@ def set(self): assert str(tmp_path) not in json.dumps([row["observation"] for row in observed]) -def test_real_host_call_matches_wall_duration_through_detached_ts(tmp_path, monkeypatch): - import subprocess - import sys - from pathlib import Path - monkeypatch.setattr(usage_ping, "select_default_runtime_root", lambda: tmp_path) - for name in ("CI", "DO_NOT_TRACK", "LOOPX_USAGE_PING", "LOOPX_USAGE_POLICY"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setenv("LOOPX_USAGE_PING_ENDPOINT", "http://127.0.0.1:1/v1/ping") - usage_ping.control("enable") +def test_real_host_call_matches_wall_duration_through_detached_ts(tmp_path, monkeypatch, usage_transport): before = time.monotonic_ns() with usage_goal.observe_goal_execution(tmp_path / "runtime", "fixture-goal", host="generic-cli"): child_start = time.time_ns() // 1_000_000 @@ -93,7 +134,7 @@ def test_real_host_call_matches_wall_duration_through_detached_ts(tmp_path, monk break time.sleep(0.03) else: - pytest.fail("real host observation did not reach detached TS") + pytest.fail(f"real host observation did not reach detached TS: {usage_transport()}") intervals = state["goals"][0]["intervals"] measured = sum(end - start for start, end in intervals) assert state["goals"][0]["measurement"] == "host_call" @@ -156,7 +197,7 @@ def test_metadata_cannot_redirect_timing_read_outside_selected_home(tmp_path, mo assert usage_goal._bound_codex_session(registry, "goal", "agent") is None -def _bound_cycle_through_detached_ts(tmp_path, monkeypatch, *, header_characters: int = 0, numeric_seconds: bool = False, +def _bound_cycle_through_detached_ts(tmp_path, monkeypatch, usage_transport, *, header_characters: int = 0, numeric_seconds: bool = False, caller_relative_paths: bool = False): """Drive the real quota observer -> detached TS chain over one bound session. @@ -167,19 +208,13 @@ def _bound_cycle_through_detached_ts(tmp_path, monkeypatch, *, header_characters call does: relative to the caller, not to the detached child. """ from datetime import datetime, timezone - from pathlib import Path registry, rollout = _bound_fixture(tmp_path, monkeypatch) if header_characters: rollout.write_text(json.dumps({"type": "session_meta", "payload": { "id": "thread-a", "base_instructions": {"text": "i" * header_characters}}}) + "\n") - monkeypatch.setattr(usage_ping, "select_default_runtime_root", lambda: tmp_path) - for name in ("CI", "DO_NOT_TRACK", "LOOPX_USAGE_PING", "LOOPX_USAGE_POLICY"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setenv("LOOPX_USAGE_PING_ENDPOINT", "http://127.0.0.1:1/v1/ping") - usage_ping.control("enable") cycle_path = Path(str(usage_ping.state_path()) + ".cycles") - def wait_for(predicate): + def wait_for(predicate, phase): deadline = time.monotonic() + 8 while time.monotonic() < deadline: try: @@ -188,7 +223,7 @@ def wait_for(predicate): except (OSError, ValueError): pass time.sleep(0.03) - pytest.fail("detached timing did not reach TS owner") + pytest.fail(f"{phase}: detached timing did not reach TS owner: {usage_transport()}") now = time.time_ns() // 1_000_000 registry_path, runtime_root = registry, tmp_path / "runtime" @@ -203,7 +238,7 @@ def publish(phase): agent_id="agent", turn_id="turn-a", phase=phase, at=time.time_ns() // 1_000_000, host="unknown") publish("start") - wait_for(lambda: bool(json.loads(cycle_path.read_text())["cursors"])) + wait_for(lambda: bool(json.loads(cycle_path.read_text())["cursors"]), "binding cursor") if numeric_seconds: # Wait for binding before choosing a whole-second provider interval. # Slow subprocess startup must not put its start before the cursor's @@ -219,13 +254,13 @@ def publish(phase): }}) + "\n") publish("spend") goals = Path(str(usage_ping.state_path()) + ".goals") - wait_for(lambda: len(json.loads(goals.read_text())["goals"]) == 2) + wait_for(lambda: len(json.loads(goals.read_text())["goals"]) == 2, "paired observations") preview = usage_ping.control("status")["goal_preview"] return preview, cycle_path, goals, publish -def test_real_detached_cycle_and_bound_codex_event_reach_shared_ts_aggregator(tmp_path, monkeypatch): - preview, cycle_path, goals, publish = _bound_cycle_through_detached_ts(tmp_path, monkeypatch) +def test_real_detached_cycle_and_bound_codex_event_reach_shared_ts_aggregator(tmp_path, monkeypatch, usage_transport): + preview, cycle_path, goals, publish = _bound_cycle_through_detached_ts(tmp_path, monkeypatch, usage_transport) assert {row["measurement"] for row in preview["counters"]} == {"quota_cycle", "codex_turn"} assert all(row["host"] == "codex_app" for row in preview["counters"]) assert "PRIVATE CONTENT" not in cycle_path.read_text() + goals.read_text() @@ -236,17 +271,16 @@ def test_real_detached_cycle_and_bound_codex_event_reach_shared_ts_aggregator(tm assert not cycle_path.exists() and not goals.exists() -def test_caller_relative_registry_and_runtime_reach_the_detached_observer(tmp_path, monkeypatch): +def test_caller_relative_registry_and_runtime_reach_the_detached_observer(tmp_path, monkeypatch, usage_transport): """Leaving the caller's directory must not re-resolve the caller's own paths.""" - preview, _, _, _ = _bound_cycle_through_detached_ts(tmp_path, monkeypatch, caller_relative_paths=True) + preview, _, _, _ = _bound_cycle_through_detached_ts(tmp_path, monkeypatch, usage_transport, caller_relative_paths=True) assert {row["measurement"] for row in preview["counters"]} == {"quota_cycle", "codex_turn"} assert all(row["host"] == "codex_app" for row in preview["counters"]) usage_ping.control("disable") -def test_provider_numeric_timing_reaches_real_detached_installation_aggregator(tmp_path, monkeypatch): - from pathlib import Path - preview, _, _, _ = _bound_cycle_through_detached_ts(tmp_path, monkeypatch, numeric_seconds=True) +def test_provider_numeric_timing_reaches_real_detached_installation_aggregator(tmp_path, monkeypatch, usage_transport): + preview, _, _, _ = _bound_cycle_through_detached_ts(tmp_path, monkeypatch, usage_transport, numeric_seconds=True) assert {row["measurement"] for row in preview["counters"]} == {"quota_cycle", "codex_turn"} installation = json.loads(Path(str(usage_ping.state_path()) + ".installation").read_text()) timed = [day for day in installation["days"] if "codex_turn" in day["intervals"]] @@ -257,16 +291,68 @@ def test_provider_numeric_timing_reaches_real_detached_installation_aggregator(t usage_ping.control("disable") -def test_bound_session_with_a_long_metadata_header_still_reaches_ts(tmp_path, monkeypatch): +def test_bound_session_with_a_long_metadata_header_still_reaches_ts(tmp_path, monkeypatch, usage_transport): """Codex records base_instructions in session_meta; a long header must bind.""" characters = 72_000 header = json.dumps({"type": "session_meta", "payload": { "id": "thread-a", "base_instructions": {"text": "i" * characters}}}) + "\n" assert len(header) > 65536, "fixture header must exceed one fixed read" preview, _, _, publish = _bound_cycle_through_detached_ts( - tmp_path, monkeypatch, header_characters=characters) + tmp_path, monkeypatch, usage_transport, header_characters=characters) assert {row["measurement"] for row in preview["counters"]} == {"quota_cycle", "codex_turn"} assert all(row["host"] == "codex_app" for row in preview["counters"]) usage_ping.control("disable") monkeypatch.setattr(usage_ping, "_detach", lambda *a, **k: pytest.fail("disabled observer spawned")) publish("start") + + +def test_usage_transport_reports_real_child_exit(usage_transport): + usage_ping._detach({}, command=[sys.executable, "-c", + "import sys; sys.stdin.read(); sys.stderr.write('fixture child failure'); sys.exit(7)"]) + deadline = time.monotonic() + 4 + while time.monotonic() < deadline: + records = json.loads(usage_transport()) + if records and records[0]["returncode"] is not None: + break + time.sleep(0.03) + assert records == [{"returncode": 7, "stderr": "fixture child failure"}] + + +def test_failed_observer_test_revokes_opt_in_and_fences_late_dispatch(tmp_path): + """A failing assertion must not leave a valid generation for a late child.""" + nested = tmp_path / "nested" + nested.mkdir() + home = nested / "home" + home.mkdir() + request_file = nested / "request.json" + (nested / "test_failure.py").write_text('''import json +from pathlib import Path +from test_usage_goal import usage_transport +from loopx import usage_ping + +def test_failed_assertion(tmp_path, usage_transport): + path = usage_ping.state_path() + request = {"path": str(path), "generation": json.loads(path.read_text())["generation"]} + Path("request.json").write_text(json.dumps(request)) + raise AssertionError("fixture assertion failed") +''') + env = {**os.environ, "HOME": str(home), "USERPROFILE": str(home), + "PYTHONPATH": os.pathsep.join([str(Path(__file__).parent), str(Path(__file__).parents[1])])} + result = subprocess.run([sys.executable, "-m", "pytest", "-q", "--no-cov", "--confcutdir", str(nested), + "test_failure.py"], cwd=nested, env=env, capture_output=True, text=True, timeout=30) + assert result.returncode == 1, result.stdout + result.stderr + assert "fixture assertion failed" in result.stdout + request = json.loads(request_file.read_text()) + path = Path(request["path"]) + state = json.loads(path.read_text()) + assert state["consent"] == "disabled" + assert state["generation"] != request["generation"] + request.update({"registry": str(nested / "unused.json"), "runtime": str(nested / "runtime"), + "goal": "fixture-goal", "agent": "fixture-agent", "turn": "fixture-turn", + "phase": "start", "at": time.time_ns() // 1_000_000, "host": "unknown"}) + late = subprocess.run([sys.executable, "-m", "loopx.usage_goal"], input=json.dumps(request), env=env, + capture_output=True, text=True, timeout=8) + assert late.returncode == 0, late.stderr + assert json.loads(path.read_text()) == state + assert not Path(str(path) + ".cycles").exists() + assert not Path(str(path) + ".goals").exists()