from __future__ import annotations import importlib.util import json import os import subprocess import sys from pathlib import Path import pytest HERE = Path(__file__).parent HARNESS = HERE / "lifecycle_observer.py" SPEC = importlib.util.spec_from_file_location("lifecycle_observer", HARNESS) assert SPEC and SPEC.loader observer = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = observer SPEC.loader.exec_module(observer) D = "a" * 64 def private_dir(path: Path) -> Path: path.mkdir(parents=True, mode=0o700) path.chmod(0o700) return path def atomic(path: Path, value: object) -> None: atomic_raw(path, json.dumps(value, sort_keys=True, separators=(",", ":")).encode()) def atomic_raw(path: Path, raw: bytes) -> None: tmp = path.with_name(".next") tmp.write_bytes(raw) tmp.chmod(0o600) os.replace(tmp, path) def profile(tmp_path: Path) -> Path: root = private_dir(tmp_path / "profile") private_dir(root / "data") private_dir(root / "data/onboarding") out = private_dir(root / "data/onboarding/telegram-publication-outbox-v1") private_dir(root / "data/owner-actions") atomic(out / "ledger.json", {"schema": "telegram-nutrition-onboarding-publication-outbox-v2", "records": []}) return root def manifest(tmp_path: Path, modes: list[str]) -> Path: evidence = private_dir(tmp_path / "evidence") expected = { "session_id": "new-session", "generation": 1, "customer_key": "new-customer", "transaction_id": "new-transaction", "draft_id": "new-draft", "delivery_key": "new-delivery", } observers = [] for mode in modes: state = {"subscribe-outbox": "DISPATCHING", "audit-tail": "customer_activation", "watch-deliveries": "sent_audited"}[mode] observers.append({"name": mode, "mode": mode, "after_seq": 0, "expected": state, "output": f"{mode}.receipt.json"}) doc = { "schema": "task26-lifecycle-observer-manifest-v1", "bindings": {"candidate": observer.CANDIDATE, "wheel_sha256": observer.WHEEL, "plan_sha256": observer.PLAN, "runbook_sha256": {"golden_v4": observer.GOLDEN_V4, "recovery_v4": observer.RECOVERY_V4}}, "receipts": {name: D for name in ("invite", "reset", "cleanup", "provider")}, "expected_ids": expected, "observers": observers, } path = evidence / "manifest.json" atomic(path, doc) return path def outbox_record(session: str = "new-session", generation: int = 1, state: str = "DISPATCHING") -> dict[str, object]: row: dict[str, object] = { "session_id": session, "generation": generation, "payload": {}, "route": ["chat", "topic"], "role": "owner", "render_identity": D, "payload_digest": D, "dispatch_identity": D, "state": state, "message_id": None, "receipt_integrity": None, } return row def run_armed(root: Path, manifest_path: Path, mode: str, trigger) -> subprocess.CompletedProcess[str]: read_fd, write_fd = os.pipe() cmd = [sys.executable, "-B", str(HARNESS), mode, "--profile", str(root), "--manifest", str(manifest_path), "--after-seq", "0", "--timeout", "2", "--output", str(manifest_path.parent / f"{mode}-cli.json"), "--ready-fd", str(write_fd)] process = subprocess.Popen(cmd, pass_fds=(write_fd,), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) os.close(write_fd) ready = os.read(read_fd, 4096) os.close(read_fd) assert json.loads(ready)["status"] == "READY" trigger() stdout, stderr = process.communicate(timeout=3) return subprocess.CompletedProcess(cmd, process.returncode, stdout, stderr) def test_subscribe_before_trigger_atomic_outbox(tmp_path: Path) -> None: root, mf = profile(tmp_path), manifest(tmp_path, ["subscribe-outbox"]) target = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json" result = run_armed(root, mf, "subscribe-outbox", lambda: atomic(target, { "schema": "telegram-nutrition-onboarding-publication-outbox-v2", "records": [outbox_record()]})) assert result.returncode == 0, result.stderr receipt = json.loads((mf.parent / "subscribe-outbox-cli.json").read_text()) assert receipt["status"] == "PASS" and receipt["transition"]["state"] == "DISPATCHING" def test_timeout_is_bounded_and_monotonic(tmp_path: Path) -> None: root, mf = profile(tmp_path), manifest(tmp_path, ["subscribe-outbox"]) result = subprocess.run([sys.executable, "-B", str(HARNESS), "subscribe-outbox", "--profile", str(root), "--manifest", str(mf), "--after-seq", "0", "--timeout", "0.03", "--output", str(mf.parent / "timeout.json")], text=True, capture_output=True, timeout=2, check=False) assert result.returncode == 3 assert "monotonic deadline expired" in result.stderr @pytest.mark.parametrize("rows,error", [ ([outbox_record(), outbox_record()], "duplicate"), ([outbox_record("old-session")], "wrong lifecycle ID"), ([outbox_record(state="UNKNOWN")], "unknown outcome"), ]) def test_rejects_duplicate_wrong_id_and_unknown_outcome(tmp_path: Path, rows: list[dict[str, object]], error: str) -> None: root, mf = profile(tmp_path), manifest(tmp_path, ["subscribe-outbox"]) target = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json" result = run_armed(root, mf, "subscribe-outbox", lambda: atomic(target, { "schema": "telegram-nutrition-onboarding-publication-outbox-v2", "records": rows})) assert result.returncode == 2 assert error in result.stderr def test_rejects_stale_no_transition(tmp_path: Path) -> None: root, mf = profile(tmp_path), manifest(tmp_path, ["subscribe-outbox"]) target = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json" result = run_armed(root, mf, "subscribe-outbox", lambda: atomic(target, { "schema": "telegram-nutrition-onboarding-publication-outbox-v2", "records": []})) assert result.returncode == 2 and "stale" in result.stderr def test_audit_and_delivery_exact_candidate_transitions(tmp_path: Path) -> None: root = profile(tmp_path) for mode in ("audit-tail", "watch-deliveries"): mf = manifest(tmp_path / mode, [mode]) if mode == "audit-tail": target = root / "data/customer-activation-audit.jsonl" payload = {"event": "customer_activation", "customer_id": "new-customer", "enabled": True, "transaction_id": "new-transaction", "registry_path": "redacted", "data_root": "redacted", "checklist_evidence_path": "redacted", "recorded_at": "2026-08-15T00:00:00+00:00"} def trigger() -> None: atomic_raw(target, (json.dumps(payload, sort_keys=True) + "\n").encode()) else: target = root / "data/owner-actions/draft-deliveries.json" payload = {"draft_id": "new-draft", "customer_key": "new-customer", "session_id": "new-session", "status": "sent_audited", "message_id": "receipt", "provider_receipt": "provider", "idempotency_key": "new-delivery"} def trigger() -> None: atomic(target, {"new-delivery": payload}) result = run_armed(root, mf, mode, trigger) assert result.returncode == 0, result.stderr def test_arm_only_and_multi_ready_barrier(tmp_path: Path) -> None: root, mf = profile(tmp_path), manifest(tmp_path, ["subscribe-outbox", "audit-tail", "watch-deliveries"]) receipt = mf.parent / "armed.json" result = subprocess.run([sys.executable, "-B", str(HARNESS), "arm-only", "--profile", str(root), "--manifest", str(mf), "--receipt", str(receipt)], text=True, capture_output=True, check=False) assert result.returncode == 0, result.stderr doc = json.loads(receipt.read_text()) assert doc["status"] == "ARMED_READ_ONLY" and len(doc["observers"]) == 3 def test_rejects_symlink_hardlink_public_and_schema_drift(tmp_path: Path) -> None: root, mf = profile(tmp_path), manifest(tmp_path, ["subscribe-outbox"]) target = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json" target.chmod(0o644) with pytest.raises(observer.ObserverError, match="private"): observer.Observer(root, observer.load_manifest(mf), observer.observer_specs(observer.load_manifest(mf))[0]).arm() target.chmod(0o600) link = target.with_name("hard") os.link(target, link) with pytest.raises(observer.ObserverError, match="hard link"): observer.Observer(root, observer.load_manifest(mf), observer.observer_specs(observer.load_manifest(mf))[0]).arm() link.unlink() atomic(target, {"schema": "drift", "records": []}) with pytest.raises(observer.ObserverError, match="schema drift"): observer.Observer(root, observer.load_manifest(mf), observer.observer_specs(observer.load_manifest(mf))[0]).arm()