"""Focused tests for the no-send clone lifecycle checkpoint.""" from __future__ import annotations import argparse import asyncio import json import os import subprocess import sys import uuid from pathlib import Path from typing import Any import pytest REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import apply_proposal as ap # noqa: E402 import run_leo_clone_lifecycle_checkpoint as lifecycle # noqa: E402 RUN_MARKER = "leo-lifecycle-unit-20260710" CONVERSATION_MARKER = "conversation-memory-unit-20260710" @pytest.fixture(autouse=True) def _clear_bound_registries() -> None: lifecycle.bound._ACTIVE_TEMP_ROOTS.clear() lifecycle.bound._ACTIVE_GATEWAY_CHILDREN.clear() lifecycle.bound._LAST_TERMINATION_SIGNAL = None yield lifecycle.bound._ACTIVE_TEMP_ROOTS.clear() lifecycle.bound._ACTIVE_GATEWAY_CHILDREN.clear() lifecycle.bound._LAST_TERMINATION_SIGNAL = None def _args(tmp_path: Path, **overrides: Any) -> argparse.Namespace: review_secret = tmp_path / "kb-review.env" apply_secret = tmp_path / "kb-apply.env" review_secret.write_text("KB_REVIEW_PASSWORD=private\n", encoding="utf-8") apply_secret.write_text("KB_APPLY_PASSWORD=private\n", encoding="utf-8") values: dict[str, Any] = { "container": "teleo-lifecycle-clone", "db": "teleo", "output": tmp_path / "receipt.json", "run_marker": RUN_MARKER, "conversation_marker": CONVERSATION_MARKER, "chat_id": "clone-chat", "user_id": "clone-user", "user_name": "clone operator", "turn_timeout": 30, "review_secrets_file": str(review_secret), "apply_secrets_file": str(apply_secret), "copy_model_auth": True, "operator_review": True, "guarded_apply": True, } values.update(overrides) return argparse.Namespace(**values) def _proposal(fixture: dict[str, Any], state: str) -> dict[str, Any]: row = { "id": fixture["proposal_id"], "proposal_type": "approve_claim", "status": state, "channel": "clone_lifecycle", "source_ref": fixture["source_ref"], "rationale": fixture["rationale"], "payload": fixture["proposal_payload"], "reviewed_by_handle": None, "reviewed_by_agent_id": None, "reviewed_at": None, "review_note": None, "applied_by_handle": None, "applied_by_agent_id": None, "applied_at": None, } if state in {"approved", "applied"}: row.update( { "reviewed_by_handle": lifecycle.REVIEWER_HANDLE, "reviewed_at": "2026-07-10T10:00:00+00:00", "review_note": lifecycle.REVIEW_NOTE, } ) if state == "applied": row.update( { "applied_by_handle": lifecycle.APPLIED_BY_HANDLE, "applied_by_agent_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", "applied_at": "2026-07-10T10:01:00+00:00", } ) return row def _approval_snapshot(proposal: dict[str, Any]) -> dict[str, Any]: return { "proposal_id": proposal["id"], "proposal_type": proposal["proposal_type"], "payload": proposal["payload"], "reviewed_by_handle": proposal["reviewed_by_handle"], "reviewed_by_agent_id": proposal["reviewed_by_agent_id"], "reviewed_by_db_role": "kb_review", "reviewed_at": proposal["reviewed_at"], "review_note": proposal["review_note"], } def test_fixture_is_deterministic_narrow_and_accepted_by_guarded_apply() -> None: first = lifecycle.build_lifecycle_fixture(RUN_MARKER) second = lifecycle.build_lifecycle_fixture(RUN_MARKER) assert first == second assert len(first["apply_payload"]["claims"]) == 1 assert len(first["apply_payload"]["sources"]) == 1 assert len(first["apply_payload"]["evidence"]) == 1 assert first["apply_payload"]["edges"] == [] assert first["apply_payload"]["reasoning_tools"] == [] assert first["proposal_payload"]["lifecycle_checkpoint"]["model_authority"] == "stage_only" for key in ("proposal_id", "claim_id", "source_id"): uuid.UUID(first[key]) approved = _proposal(first, "approved") sql = ap.build_apply_sql(approved, lifecycle.APPLIED_BY_HANDLE) assert "kb_stage.assert_approved_proposal" in sql assert "kb_stage.finish_approved_proposal" in sql assert first["claim_id"] in sql assert first["source_id"] in sql def test_stage_sql_is_idempotent_and_cannot_write_canonical_tables() -> None: fixture = lifecycle.build_lifecycle_fixture(RUN_MARKER) sql = lifecycle.build_stage_sql(fixture) lowered = sql.lower() assert "insert into kb_stage.kb_proposals" in lowered assert "on conflict (id) do nothing" in lowered assert "pending_review" in sql assert fixture["source_ref"] in sql assert "lifecycle stage validation failed" in lowered assert "marker_count <> 1 or exact_count <> 1" in lowered assert lowered.index("raise exception") < lowered.index("commit;") assert "insert into public." not in lowered assert "update public." not in lowered assert "delete from public." not in lowered def test_unrelated_row_guard_is_read_only_and_excludes_only_deterministic_rows( monkeypatch: pytest.MonkeyPatch, ) -> None: fixture = lifecycle.build_lifecycle_fixture(RUN_MARKER) captured: dict[str, str] = {} def fake_psql(_container: str, _database: str, sql: str) -> dict[str, Any]: captured["sql"] = sql return {table: {"count": 1, "rowset_md5": "same"} for table in lifecycle.LIFECYCLE_TABLES} monkeypatch.setattr(lifecycle.bound, "_psql_json", fake_psql) snapshot = lifecycle.target_unrelated_guard_snapshot("clone", "teleo", fixture) sql = captured["sql"].lower() assert set(snapshot) == set(lifecycle.LIFECYCLE_TABLES) assert "begin transaction read only" in sql assert fixture["proposal_id"] in sql assert fixture["claim_id"] in sql assert fixture["source_id"] in sql assert lifecycle.APPROVAL_TABLE in sql assert "insert into" not in sql assert "update " not in sql assert "delete from" not in sql def test_stage_replay_returns_the_same_pending_proposal( monkeypatch: pytest.MonkeyPatch, ) -> None: fixture = lifecycle.build_lifecycle_fixture(RUN_MARKER) pending = _proposal(fixture, "pending_review") monkeypatch.setattr( lifecycle, "assert_disposable_target", lambda _container, _database: ({"id": "clone"}, {"id": "production"}), ) monkeypatch.setattr( lifecycle.bound, "_psql_json", lambda _container, _database, _sql: { "created": False, "marker_match_count": 1, "proposal": pending, }, ) replay = lifecycle.stage_lifecycle_proposal("clone", "teleo", fixture) assert replay["created"] is False assert replay["idempotent_existing"] is True assert replay["proposal"] == pending assert replay["canonical_row_ids"]["public.claims"] == [fixture["claim_id"]] def test_staging_wrapper_hard_binds_target_and_exposes_no_review_apply_or_send(tmp_path: Path) -> None: wrapper = lifecycle.build_lifecycle_wrapper( kb_tool=tmp_path / "kb_tool.py", tool_log=tmp_path / "tool.jsonl", container="clone-container", database="teleo_clone", run_marker=RUN_MARKER, run_nonce="unit-test-nonce", ) assert "TARGET_CONTAINER=clone-container" in wrapper assert "TARGET_DATABASE=teleo_clone" in wrapper assert wrapper.startswith("#!/bin/bash\n") assert f"DOCKER={lifecycle.bound.SYSTEM_DOCKER}" in wrapper assert f"export PATH={lifecycle.bound.SYSTEM_EXEC_PATH}" in wrapper assert "/usr/bin/env bash" not in wrapper assert "$(docker " not in wrapper assert lifecycle.INTERNAL_STAGE_COMMAND in wrapper assert f'${{1:-}}" == {lifecycle.STAGE_COMMAND}' in wrapper assert "accepts no model-supplied arguments" in wrapper assert "approve_proposal.py" not in wrapper assert "apply_proposal.py" not in wrapper assert "send_message" not in wrapper def test_lifecycle_surface_temporarily_adds_only_the_stage_verb( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: original_commands = lifecycle.bound.READ_ONLY_BRIDGE_COMMANDS def fake_install(_profile: Path) -> dict[str, Any]: return { "allowed_tools": ["skills_list", "skill_view", "terminal"], "read_only_bridge_commands": sorted(lifecycle.bound.READ_ONLY_BRIDGE_COMMANDS), "send_message_tool_enabled": False, } monkeypatch.setattr(lifecycle.bound, "install_checkpoint_tool_surface", fake_install) with lifecycle.lifecycle_gateway_surface(allow_staging=True): assert lifecycle.STAGE_COMMAND in lifecycle.bound.READ_ONLY_BRIDGE_COMMANDS surface = lifecycle.bound.install_checkpoint_tool_surface(tmp_path) assert surface["staging_bridge_commands"] == [lifecycle.STAGE_COMMAND] assert surface["model_can_stage_pending_proposal"] is True assert surface["model_can_approve_or_apply"] is False assert surface["operator_review_apply_tools_exposed"] is False assert "send_message" not in surface["allowed_tools"] assert original_commands == lifecycle.bound.READ_ONLY_BRIDGE_COMMANDS assert lifecycle.STAGE_COMMAND not in lifecycle.bound.READ_ONLY_BRIDGE_COMMANDS with lifecycle.lifecycle_gateway_surface(allow_staging=False): surface = lifecycle.bound.install_checkpoint_tool_surface(tmp_path) assert surface["staging_bridge_commands"] == [] assert surface["model_can_stage_pending_proposal"] is False with pytest.raises(lifecycle.bound.CheckpointError): lifecycle.bound.validate_read_only_bridge_argv([lifecycle.STAGE_COMMAND]) def test_prompts_keep_conversation_marker_out_of_mutation_and_recall_prompt() -> None: fixture = lifecycle.build_lifecycle_fixture(RUN_MARKER) prompts = lifecycle.lifecycle_prompts(fixture, CONVERSATION_MARKER) assert CONVERSATION_MARKER in prompts["T1_conversation_marker"] assert CONVERSATION_MARKER not in prompts["T2_stage_pending_proposal"] assert CONVERSATION_MARKER not in prompts["T5_recall_and_applied_readback"] assert CONVERSATION_MARKER not in json.dumps(fixture, sort_keys=True) assert RUN_MARKER in prompts["T2_stage_pending_proposal"] assert RUN_MARKER not in prompts["T5_recall_and_applied_readback"] def test_lifecycle_state_receipt_is_exact_and_rejects_extra_or_contradictory_state() -> None: expected = { "phase": "T2", "proposal_id": "537c3087-b795-5382-8335-e0928d1ac96c", "status": "pending_review", "canonical": False, } reply = "Proposal staged and not canonical.\nLIFECYCLE_STATE: " + json.dumps(expected) assert lifecycle.parse_lifecycle_state(reply) == {"valid": True, "value": expected} assert lifecycle.lifecycle_state_matches(reply, expected) is True assert lifecycle.lifecycle_state_matches(reply.replace("}", ', "unverified": true}', 1), expected) is False assert lifecycle._reply_has_phase_contradiction("T2", reply) is False assert ( lifecycle._reply_has_phase_contradiction("T2", "The proposal is now canonical.\n" + reply.splitlines()[-1]) is True ) assert lifecycle.parse_lifecycle_state("LIFECYCLE_STATE: not-json")["valid"] is False def test_phase_row_read_requires_the_exact_full_uuid() -> None: turn = {"tool_invocations": [{"argv": ["show-proposal", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"]}]} assert lifecycle._phase_has_bound_read(turn, "show-proposal", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa") is True assert lifecycle._phase_has_bound_read(turn, "show-proposal", "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb") is False def test_validation_requires_explicit_clone_review_apply_and_auth(tmp_path: Path) -> None: lifecycle.validate_args(_args(tmp_path)) with pytest.raises(lifecycle.bound.CheckpointError, match="production"): lifecycle.validate_args(_args(tmp_path, container=lifecycle.bound.PRODUCTION_CONTAINER)) with pytest.raises(lifecycle.bound.CheckpointError, match="operator-review"): lifecycle.validate_args(_args(tmp_path, operator_review=False)) with pytest.raises(lifecycle.bound.CheckpointError, match="guarded-apply"): lifecycle.validate_args(_args(tmp_path, guarded_apply=False)) with pytest.raises(lifecycle.bound.CheckpointError, match="copy-model-auth"): lifecycle.validate_args(_args(tmp_path, copy_model_auth=False)) def test_validation_rejects_receipt_inside_live_profile( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: live = tmp_path / "live-profile" live.mkdir() monkeypatch.setattr(lifecycle.bound, "LIVE_PROFILE", live) with pytest.raises(lifecycle.bound.CheckpointError, match="outside the live leoclean profile"): lifecycle.validate_args(_args(tmp_path, output=live / "receipt.json")) def test_disposable_identity_rejects_an_alias_of_production( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( lifecycle.bound, "container_identity", lambda name: {"id": "same-id", "name": name, "image": "postgres"}, ) with pytest.raises(lifecycle.bound.CheckpointError, match="production container identity"): lifecycle.assert_disposable_target("alias-container", "teleo") def _install_fake_runtime( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, args: argparse.Namespace, ) -> tuple[dict[str, str], dict[str, Any]]: fixture = lifecycle.build_lifecycle_fixture(args.run_marker) expected_rows = lifecycle.guarded._expected_bundle_rows(fixture["apply_payload"]) empty_rows = {key: [] for key in expected_rows} base_counts = { "public.claims": 100, "public.sources": 200, "public.claim_evidence": 300, "public.claim_edges": 400, "public.reasoning_tools": 5, "kb_stage.kb_proposals": 9, lifecycle.APPROVAL_TABLE: 4, } stage_counts = dict(base_counts) stage_counts["kb_stage.kb_proposals"] += 1 approved_counts = dict(stage_counts) approved_counts[lifecycle.APPROVAL_TABLE] += 1 final_counts = dict(approved_counts) final_counts["public.claims"] += 1 final_counts["public.sources"] += 1 final_counts["public.claim_evidence"] += 1 state: dict[str, Any] = {"phase": "initial", "invocations": []} base_snapshot = { "counts": base_counts, "rowset_md5": {table: f"production-{index}" for index, table in enumerate(base_counts)}, } target_identity = { "id": "clone-id", "name": args.container, "image": "postgres", "labels": { lifecycle.bound.DISPOSABLE_LABEL_KEY: lifecycle.bound.DISPOSABLE_LABEL_VALUE, }, "network_mode": "none", "ports": {}, "mount_sources": ["/clone-data"], "started_at": "clone-start", } production_identity = { "id": "production-id", "name": lifecycle.bound.PRODUCTION_CONTAINER, "image": "postgres", "labels": {}, "network_mode": "bridge", "ports": {}, "mount_sources": ["/production-data"], "started_at": "production-start", } monkeypatch.setattr( lifecycle, "assert_disposable_target", lambda _container, _database: ( dict(target_identity), dict(production_identity), ), ) monkeypatch.setattr( lifecycle.bound, "container_identity", lambda container: dict( production_identity if container == lifecycle.bound.PRODUCTION_CONTAINER else target_identity ), ) def fake_guard_snapshot(container: str, _database: str, **_kwargs: Any) -> dict[str, Any]: if container == lifecycle.bound.PRODUCTION_CONTAINER: snapshot = json.loads(json.dumps(base_snapshot)) snapshot["database_identity"] = { "current_database": args.db, "system_identifier": "production-system", "transaction_read_only": "off", } return snapshot identity = { "current_database": args.db, "system_identifier": "clone-system", "transaction_read_only": "off", } if state["phase"] == "applied": snapshot = json.loads(json.dumps(base_snapshot)) snapshot["counts"] = dict(final_counts) for table in ( "public.claims", "public.sources", "public.claim_evidence", "kb_stage.kb_proposals", lifecycle.APPROVAL_TABLE, ): snapshot["rowset_md5"][table] = "changed" snapshot["database_identity"] = identity return snapshot snapshot = json.loads(json.dumps(base_snapshot)) snapshot["database_identity"] = identity return snapshot monkeypatch.setattr(lifecycle.bound, "database_guard_snapshot", fake_guard_snapshot) monkeypatch.setattr( lifecycle.bound, "service_state", lambda: {"returncode": 0, "ActiveState": "active", "MainPID": "123", "NRestarts": "0"}, ) monkeypatch.setattr( lifecycle.bound, "bridge_hashes", lambda: {"wrapper": "live-wrapper-hash", "bridge_skill": "live-skill-hash"}, ) monkeypatch.setattr( lifecycle, "target_unrelated_guard_snapshot", lambda _container, _database, _fixture: { table: {"count": base_counts[table], "rowset_md5": f"unrelated-{table}"} for table in lifecycle.LIFECYCLE_TABLES }, ) monkeypatch.setattr(lifecycle, "_source_manifest", lambda: [{"repo_relative_path": "stable", "sha256": "x"}]) def fake_proposal_readback(_container: str, _database: str, _proposal_id: str) -> dict[str, Any] | None: if state["phase"] == "initial": return None return _proposal(fixture, state["phase"]) monkeypatch.setattr(lifecycle.guarded, "_proposal_readback", fake_proposal_readback) monkeypatch.setattr( lifecycle.guarded, "_exact_bundle_readback", lambda _container, _database, _payload: expected_rows if state["phase"] == "applied" else empty_rows, ) monkeypatch.setattr( lifecycle.guarded, "_approval_snapshot_readback", lambda _container, _database, _proposal_id: _approval_snapshot(_proposal(fixture, "approved")), ) runtime_root = tmp_path / "private-runtime" runtime_profile = runtime_root / "profile" def fake_copy_profile(**_kwargs: Any) -> tuple[Path, dict[str, Any]]: runtime_profile.mkdir(parents=True) lifecycle.bound.register_temp_root(runtime_root) return runtime_profile, {"passes": True, "secret_contents_read_or_recorded": False} monkeypatch.setattr(lifecycle.bound, "copy_profile", fake_copy_profile) monkeypatch.setattr( lifecycle.bound, "copy_ephemeral_model_auth", lambda _profile: { "mode": "ephemeral_file_copy", "source_contents_recorded": False, "source_fingerprint_recorded": False, }, ) monkeypatch.setattr( lifecycle.bound, "load_ephemeral_provider_environment", lambda _profile: ({}, {"bound_credential_count": 0, "credential_contents_recorded": False}), ) tool_log = runtime_profile / "checkpoint-tool-calls.jsonl" monkeypatch.setattr( lifecycle, "patch_lifecycle_bridge", lambda _profile, **_kwargs: { "wrapper_path": str(runtime_profile / "bin" / "teleo-kb"), "bridge_skill_path": str(runtime_profile / "skills" / "teleo-kb-bridge" / "SKILL.md"), "tool_log_path": str(tool_log), "run_nonce": "lifecycle-test-nonce", "model_can_approve": False, "model_can_apply": False, }, ) async def fake_turn( _args: argparse.Namespace, _profile: Path, *, phase: str, prompt: str, provider_environment: dict[str, str], tool_log: Path, ) -> dict[str, Any]: assert provider_environment == {} assert tool_log == runtime_profile / "checkpoint-tool-calls.jsonl" command_argvs: list[list[str]] if phase == "T1_conversation_marker": reply = ( f"I remember {args.conversation_marker}; conversation memory is not canonical Postgres.\n" f'{lifecycle.LIFECYCLE_STATE_PREFIX} {{"phase":"T1","conversation_marker":' f'"{args.conversation_marker}","kb_mutated":false}}' ) command_argvs = [] elif phase == "T2_stage_pending_proposal": state["phase"] = "pending_review" reply = ( f"Staged proposal {fixture['proposal_id']} as pending_review, not canonical.\n" f'{lifecycle.LIFECYCLE_STATE_PREFIX} {{"phase":"T2","proposal_id":"{fixture["proposal_id"]}",' '"status":"pending_review","canonical":false}' ) command_argvs = [ [lifecycle.STAGE_COMMAND], ["show-proposal", fixture["proposal_id"]], ] elif phase == "T3_approved_unapplied_readback": reply = ( f"Proposal {fixture['proposal_id']} is approved, applied_at NULL, and not canonical.\n" f'{lifecycle.LIFECYCLE_STATE_PREFIX} {{"phase":"T3","proposal_id":"{fixture["proposal_id"]}",' '"status":"approved","applied":false,"canonical":false}' ) command_argvs = [["show-proposal", fixture["proposal_id"]]] else: reply = ( f"Marker {args.conversation_marker}; proposal {fixture['proposal_id']} is applied; " f"canonical claim {fixture['claim_id']} and source {fixture['source_id']}.\n" f'{lifecycle.LIFECYCLE_STATE_PREFIX} {{"phase":"T5","conversation_marker":' f'"{args.conversation_marker}","proposal_id":"{fixture["proposal_id"]}","status":"applied",' f'"claim_id":"{fixture["claim_id"]}","source_id":"{fixture["source_id"]}"}}' ) command_argvs = [ ["show-proposal", fixture["proposal_id"]], ["show", fixture["claim_id"]], ] invocations = [ { "invocation_id": f"{phase}-{index}", "container": args.bound_container_id, "database": args.db, "argv": command_argv, "returncode": 0, } for index, command_argv in enumerate(command_argvs) ] state["invocations"].extend(invocations) pid = { "T1_conversation_marker": 101, "T2_stage_pending_proposal": 102, "T3_approved_unapplied_readback": 103, "T5_recall_and_applied_readback": 105, }[phase] gateway = { "authorized": True, "handler_invoked": True, "posted_to_telegram": False, "model_free_fallback_used": False, "reply": reply, "session_key": "telegram:clone-chat", "transcript_tool_trace": { "session_id": "persisted-private-session", "events": [ event for item in invocations for event in ( {"phase": "call", "tool_name": "terminal", "tool_call_id": item["invocation_id"]}, { "phase": "result", "tool_name": "terminal", "tool_call_id": item["invocation_id"], "content": '{"exit_code": 0}', }, ) ], }, "tool_surface": { "allowed_tools": ["skills_list", "skill_view", "terminal"], "actual_registry_tools": ["skills_list", "skill_view", "terminal"], "send_message_tool_enabled": False, "gateway_adapters_verified_mapping": True, "gateway_adapter_count": 0, "model_can_approve_or_apply": False, "operator_review_apply_tools_exposed": False, }, "child_process": { "pid": pid, "readiness_verified": True, "timed_out": False, "exitcode": 0, "termination": {"attempted": False}, "alive_after_readback": False, "process_group_alive_after_readback": False, "result_transport": "private_temp_file", }, } return { "phase": phase, "prompt": prompt, "prompt_sha256": "prompt-hash", "gateway": gateway, "tool_invocations": invocations, } monkeypatch.setattr(lifecycle, "run_gateway_turn", fake_turn) monkeypatch.setattr( lifecycle, "stage_lifecycle_proposal", lambda _container, _database, _fixture: { "created": False, "idempotent_existing": True, "marker_match_count": 1, "proposal": _proposal(fixture, "pending_review"), }, ) def fake_review( _args: argparse.Namespace, _proposal_id: str ) -> tuple[subprocess.CompletedProcess[str], subprocess.CompletedProcess[str]]: dry = subprocess.CompletedProcess([], 0, stdout="select kb_stage.approve_strict_proposal(...);", stderr="") state["phase"] = "approved" applied = subprocess.CompletedProcess([], 0, stdout="approved", stderr="") return dry, applied def fake_apply( _args: argparse.Namespace, _proposal_id: str ) -> tuple[subprocess.CompletedProcess[str], subprocess.CompletedProcess[str]]: dry = subprocess.CompletedProcess( [], 0, stdout=("select kb_stage.assert_approved_proposal(...); select kb_stage.finish_approved_proposal(...);"), stderr="", ) state["phase"] = "applied" applied = subprocess.CompletedProcess([], 0, stdout="applied", stderr="") return dry, applied monkeypatch.setattr(lifecycle, "run_operator_review", fake_review) monkeypatch.setattr(lifecycle, "run_guarded_apply", fake_apply) def fake_counts(_container: str, _database: str, **_kwargs: Any) -> dict[str, int]: if state["phase"] == "applied": return dict(final_counts) if state["phase"] == "approved": return dict(approved_counts) return dict(stage_counts) monkeypatch.setattr(lifecycle.bound, "database_counts", fake_counts) monkeypatch.setattr( lifecycle.bound, "read_tool_proof", lambda _path, _container, _database, **_kwargs: { "event_count": len(state["invocations"]) * 2, "parse_errors": [], "duplicate_invocation_ids": [], "complete_start_end_pairing": True, "invocations": list(state["invocations"]), "invocation_count": len(state["invocations"]), "all_bound_to_supplied_target": bool(state["invocations"]), "database_read_only_required": False, "all_completed_successfully": bool(state["invocations"]), }, ) return state, {"runtime_root": runtime_root, "expected_rows": expected_rows} def test_mocked_full_lifecycle_produces_private_passing_receipt_and_cleanup( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: args = _args(tmp_path) _state, proof = _install_fake_runtime(monkeypatch, tmp_path, args) report = asyncio.run(lifecycle.run_lifecycle(args)) assert report["status"] == "pass", { "errors": report["errors"], "failed_checks": [name for name, passed in report["checks"].items() if not passed], } assert all(report["checks"].values()) assert report["final_canonical_rows"] == proof["expected_rows"] assert report["target_deltas"]["baseline_to_final"] == { "kb_stage.kb_proposals": 1, lifecycle.APPROVAL_TABLE: 1, "public.claim_edges": 0, "public.claim_evidence": 1, "public.claims": 1, "public.reasoning_tools": 0, "public.sources": 1, } assert report["production_invariants"]["guard_snapshot_unchanged"] is True assert report["isolated_restart"]["same_persisted_session_id"] is True assert report["cleanup"]["active_gateway_registry_clear"] is True assert not os.path.lexists(proof["runtime_root"]) retained = json.loads(args.output.read_text(encoding="utf-8")) assert retained["status"] == "pass" assert args.output.stat().st_mode & 0o777 == 0o600 def test_mocked_lifecycle_fails_if_target_name_is_rebound_after_execution( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: args = _args(tmp_path) _install_fake_runtime(monkeypatch, tmp_path, args) calls = 0 def replaced_identity(_container: str) -> dict[str, Any]: nonlocal calls calls += 1 return { "id": "replacement-id", "name": args.container, "image": "postgres", "labels": { lifecycle.bound.DISPOSABLE_LABEL_KEY: lifecycle.bound.DISPOSABLE_LABEL_VALUE, }, "network_mode": "none", "ports": {}, "mount_sources": ["/replacement-data"], "started_at": "replacement-start", } monkeypatch.setattr(lifecycle.bound, "container_identity", replaced_identity) report = asyncio.run(lifecycle.run_lifecycle(args)) assert calls == 1 assert report["status"] == "fail" assert any("pinned container identity" in item["message"] for item in report["errors"]) assert report["checks"]["target_container_name_still_resolves_to_pinned_identity"] is False def test_mocked_lifecycle_fails_if_t1_changes_database_before_authorized_stage( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: args = _args(tmp_path) _install_fake_runtime(monkeypatch, tmp_path, args) original_snapshot = lifecycle.bound.database_guard_snapshot clone_reads = 0 def mutate_after_t1(container: str, database: str, **kwargs: Any) -> dict[str, Any]: nonlocal clone_reads snapshot = original_snapshot(container, database, **kwargs) if container != lifecycle.bound.PRODUCTION_CONTAINER: clone_reads += 1 if clone_reads == 2: snapshot = json.loads(json.dumps(snapshot)) snapshot["rowset_md5"]["kb_stage.kb_proposals"] = "unexpected-t1-write" return snapshot monkeypatch.setattr(lifecycle.bound, "database_guard_snapshot", mutate_after_t1) report = asyncio.run(lifecycle.run_lifecycle(args)) assert report["status"] == "fail" assert any("T1 changed clone database state" in item["message"] for item in report["errors"]) assert report["checks"]["T1_database_unchanged_and_staging_absent"] is False def test_source_contains_no_direct_canonical_write_service_restart_or_delivery_adapter() -> None: source = (REPO_ROOT / "scripts" / "run_leo_clone_lifecycle_checkpoint.py").read_text(encoding="utf-8").lower() assert "insert into public." not in source assert "update public." not in source assert "delete from public." not in source assert "systemctl restart" not in source assert "send_message(" not in source assert "guarded._approve" in source assert "guarded._apply" in source def test_parse_args_requires_one_explicit_command_boundary(tmp_path: Path) -> None: args = lifecycle.parse_args( [ "--container", "teleo-disposable-clone", "--db", "teleo", "--output", str(tmp_path / "receipt.json"), "--copy-model-auth", "--operator-review", "--guarded-apply", ] ) assert args.container == "teleo-disposable-clone" assert args.db == "teleo" assert args.copy_model_auth is True assert args.operator_review is True assert args.guarded_apply is True assert lifecycle.MARKER_RE.fullmatch(args.run_marker) assert lifecycle.MARKER_RE.fullmatch(args.conversation_marker) assert args.run_marker != args.conversation_marker