from __future__ import annotations import fcntl import hashlib import json import os import stat import threading from contextlib import contextmanager from datetime import date, time from pathlib import Path import pytest import checkin_cli.customer_admin as customer_admin_module from checkin_cli import adaptive_nutrition as adaptive_module from checkin_cli import customer_schedule as schedule_module from checkin_cli import diagnostic_evidence as evidence_module from checkin_cli.adaptive_nutrition import ( AdaptiveEventStore, AdaptiveOverlay, CanonicalSequenceJournal, OverlayJournal, ) from checkin_cli.customer_admin import ( CustomerDraft, register_customer, set_customer_ai_consent, ) from checkin_cli.customer_coaching import ( AiProcessingConsent, CONSENT_VERSION, CustomerRuntime, load_customer_registry, ) from checkin_cli.customer_schedule import ( initialize_schedule_delivery_fence, reserve_customer_task_delivery, ) from checkin_cli.weekly_operations_schedule_host_models_r4 import CustomerScheduleTask from checkin_cli.diagnostic_evidence import ( DiagnosticEvidenceError, DiagnosticEvidenceSource, build_diagnostic_promotion_manifest, diagnostic_source, existing_read_lock, require_synthetic_provenance, snapshot_diagnostic_source, synthetic_replay_fixture, ) from checkin_cli.store import CanonicalEventTransaction from checkin_cli.wizard_models import WizardFlow, WizardSession from checkin_cli.wizard_storage import WizardStorage def _files(tmp_path: Path): lock = tmp_path / ".events.lock" lock.write_bytes(b"") lock.chmod(0o600) data = tmp_path / "events.jsonl" data.write_text(json.dumps({"event_type": "x", "customer_key": "secret", "body": "private"}) + "\n") return data, lock def _registered_runtime_tree(tmp_path: Path) -> tuple[Path, CustomerRuntime]: profile = tmp_path / "profile" registry = profile / "customers" / "registry.json" registry.parent.mkdir(parents=True) weeks = [ { "week": week, "calories_kcal": 2300, "protein_g": 150, "meal_structure": ["아침", "점심", "저녁"], } for week in range(1, 13) ] registry.write_text( json.dumps( { "version": 1, "owner": {"user_id": "owner", "chat_id": "owner-chat", "topic_id": "owner-topic"}, "customers": [ { "customer_key": "client_001", "display_name": "고객 001", "enabled": False, "telegram": { "user_id": "customer", "chat_id": "customer-chat", "topic_id": "customer-topic", }, "schedule": { "daily_time": "08:00", "weekly_weekday": 0, "monthly_day": 1, }, "plan": { "starts_on": "2026-08-03", "focus": "nutrition_90_training_10", "weeks": weeks, }, } ], }, ensure_ascii=False, ), encoding="utf-8", ) registry.chmod(0o600) runtime = load_customer_registry(registry, profile).customers[0] runtime.data_root.mkdir(parents=True) return profile, runtime def _typed_source_tree(tmp_path: Path): profile, runtime = _registered_runtime_tree(tmp_path) customer = runtime.data_root (customer / "wizard" / "drafts").mkdir(parents=True) (customer / "nutrition-plans").mkdir() descriptor = diagnostic_source(profile, runtime, DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE) descriptor.lock_path.parent.mkdir(parents=True, exist_ok=True) descriptor.lock_path.write_bytes(b"") descriptor.lock_path.chmod(0o600) for path in descriptor.data_paths: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({"event_type": "bounded"}) + "\n") path.chmod(0o600) return profile, customer, runtime, descriptor def _schedule_source_tree(tmp_path: Path): profile = tmp_path / "profile" initialize_schedule_delivery_fence(profile) task = CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) reserve_customer_task_delivery( profile, task, body="private schedule body", destination={"user_id": "raw-user-id", "chat_id": "raw-chat-id"}, template_digest="1" * 64, registry_digest="2" * 64, config_digest="3" * 64, reservation_id="reservation-00000001", ) descriptor = diagnostic_source(profile, None, DiagnosticEvidenceSource.SCHEDULE) return profile, descriptor def _tree_state(root: Path): result = [] paths = sorted((root, *root.rglob("*")), key=lambda path: str(path.relative_to(root))) for path in paths: info = path.lstat() if stat.S_ISREG(info.st_mode): value = path.read_bytes() elif stat.S_ISLNK(info.st_mode): value = os.readlink(path) else: value = None result.append( ( str(path.relative_to(root)), info.st_mode, info.st_size, info.st_mtime_ns, value, ) ) return tuple(result) def _canonical_event(index: int) -> dict[str, object]: return { "event_id": f"event-{index:04d}", "event_type": "morning_checkin", "occurred_at_kst": f"2026-08-{index + 2:02d}T09:00:00+09:00", "recorded_at_kst": f"2026-08-{index + 2:02d}T09:00:00+09:00", "provenance": { "source_type": "manual", "source_ref": f"diagnostic:{index}", "content_sha256": "0" * 64, }, "status": "accepted", "dedupe_key": f"diagnostic-event-{index}", "check_in": { "body_weight_kg": 80, "calories_kcal": 2300, "sleep_hours": 8, "sleep_quality_1to5": 4, "readiness_1to5": 4, "pain_summary": "없음", "training_plan": "계획대로 진행", }, } def _canonical_source_tree(tmp_path: Path): profile, runtime = _registered_runtime_tree(tmp_path) customer = runtime.data_root (customer / "wizard" / "drafts").mkdir(parents=True) (customer / "nutrition-plans").mkdir() transaction = CanonicalEventTransaction( customer / "wizard" / "events.jsonl", customer / "nutrition-plans" / "canonical-sequence.jsonl", ) transaction.append(_canonical_event(1)) descriptor = diagnostic_source( profile, runtime, DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE, ) return profile, customer, runtime, descriptor, transaction def _adaptive_source_tree(tmp_path: Path): profile, runtime = _registered_runtime_tree(tmp_path) customer = runtime.data_root (customer / "wizard" / "drafts").mkdir(parents=True) nutrition = customer / "nutrition-plans" nutrition.mkdir() store = AdaptiveEventStore.for_registered(runtime) store.append( "plan_proposed", {"proposal_digest": "a" * 64, "revision": 1}, dedupe_key="diagnostic-adaptive-1", ) descriptor = diagnostic_source( profile, runtime, DiagnosticEvidenceSource.ADAPTIVE_JOURNALS, ) return profile, customer, runtime, descriptor, store def _overlay_source_tree(tmp_path: Path): profile, runtime = _registered_runtime_tree(tmp_path) customer = runtime.data_root (customer / "wizard" / "drafts").mkdir(parents=True) nutrition = customer / "nutrition-plans" nutrition.mkdir() journal = OverlayJournal(nutrition / "adaptive-overlays.jsonl", root=nutrition) journal.append( AdaptiveOverlay( "revision-1", "a" * 64, "2026-08-03T00:00:00+09:00", authority_snapshot_id="diagnostic-authority-1", ) ) descriptor = diagnostic_source( profile, runtime, DiagnosticEvidenceSource.OVERLAY, ) return profile, customer, runtime, descriptor, journal def _wizard_session(session_id: str, version: int) -> WizardSession: return WizardSession( session_id=session_id, flow=WizardFlow.MORNING, owner_id="owner", topic_id="topic", kst_day="2026-08-03", version=version, step="sleep", answers={"sleep_hours": str(8 + version)}, ) def _wizard_domain_source_tree(tmp_path: Path): profile, runtime = _registered_runtime_tree(tmp_path) customer = runtime.data_root storage = WizardStorage(customer / "wizard") with storage.locked(): storage.save(_wizard_session("a" * 32, 1)) descriptor = diagnostic_source( profile, runtime, DiagnosticEvidenceSource.WIZARD_DRAFT, ) return { "root": customer, "descriptor": descriptor, "writer_lock": storage.locked, "writer_update": lambda: storage.save(_wizard_session("b" * 32, 2)), } def _canonical_domain_source_tree(tmp_path: Path): profile, customer, _, descriptor, transaction = _canonical_source_tree(tmp_path) return { "root": customer, "descriptor": descriptor, "writer_lock": transaction._locked, } def _adaptive_domain_source_tree(tmp_path: Path): profile, customer, _, descriptor, store = _adaptive_source_tree(tmp_path) return { "root": customer, "descriptor": descriptor, "writer_lock": store.locked, } def _overlay_domain_source_tree(tmp_path: Path): profile, customer, _, descriptor, journal = _overlay_source_tree(tmp_path) return { "root": customer, "descriptor": descriptor, "writer_lock": journal._locked, } def _schedule_domain_source_tree(tmp_path: Path): profile, descriptor = _schedule_source_tree(tmp_path) return { "root": profile, "descriptor": descriptor, "writer_lock": lambda: schedule_module._schedule_lock(profile), } def _profile_authority_domain_source_tree(tmp_path: Path): profile = tmp_path / "profile" (profile / "customers").mkdir(parents=True) registry = profile / "customers" / "registry.json" registry.write_text( json.dumps( { "version": 1, "owner": {"user_id": "1", "chat_id": "-100", "topic_id": "10"}, "customers": [], } ), encoding="utf-8", ) registry.chmod(0o600) register_customer( registry, CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), ), ) descriptor = diagnostic_source( profile, None, DiagnosticEvidenceSource.PROFILE_AUTHORITY, ) return { "root": profile, "descriptor": descriptor, "writer_lock": lambda: customer_admin_module.profile_authority_lock(profile), "writer_update": lambda: set_customer_ai_consent( registry, "client_001", AiProcessingConsent( granted=True, recorded_on=date(2026, 8, 3), notice_version=CONSENT_VERSION, ), ), } _DOMAIN_SOURCE_BUILDERS = ( pytest.param(_wizard_domain_source_tree, id="wizard_draft"), pytest.param(_canonical_domain_source_tree, id="canonical_event_sequence"), pytest.param(_adaptive_domain_source_tree, id="adaptive_journals"), pytest.param(_overlay_domain_source_tree, id="overlay"), pytest.param(_schedule_domain_source_tree, id="schedule"), pytest.param(_profile_authority_domain_source_tree, id="profile_authority"), ) def _replace_with_symlink(path: Path, target: Path) -> None: if path.is_dir(): os.replace(path, target) path.symlink_to(target, target_is_directory=True) return raw = path.read_bytes() target.write_bytes(raw) target.chmod(stat.S_IMODE(path.stat().st_mode)) path.unlink() path.symlink_to(target) def _replace_lock_inode(path: Path) -> None: replacement = path.with_name(f".{path.name}.replacement") replacement.write_bytes(path.read_bytes()) replacement.chmod(stat.S_IMODE(path.stat().st_mode)) os.replace(replacement, path) def _deterministic_writer_reader_snapshot(fixture: dict[str, object]): ready = threading.Event() reader_lock_attempt = threading.Event() writer_gate = threading.Event() writer_errors: list[BaseException] = [] reader_errors: list[BaseException] = [] values: list[object] = [] writer_state: dict[str, object] = {} original_flock = evidence_module.fcntl.flock def observed_flock(descriptor: int, operation: int) -> None: if operation == fcntl.LOCK_SH: reader_lock_attempt.set() original_flock(descriptor, operation) evidence_module.fcntl.flock = observed_flock def run_writer() -> None: try: writer_lock = fixture["writer_lock"] assert callable(writer_lock) with writer_lock(): ready.set() assert writer_gate.wait(timeout=5) update = fixture.get("writer_update") if callable(update): update() writer_state["tree"] = _tree_state(fixture["root"]) except BaseException as exc: writer_errors.append(exc) def run_reader() -> None: try: values.append(snapshot_diagnostic_source(fixture["descriptor"])) except BaseException as exc: reader_errors.append(exc) writer_thread = threading.Thread(target=run_writer) writer_thread.start() assert ready.wait(timeout=5) reader_thread = threading.Thread(target=run_reader) reader_thread.start() assert reader_lock_attempt.wait(timeout=5) writer_gate.set() writer_thread.join(timeout=5) reader_thread.join(timeout=5) evidence_module.fcntl.flock = original_flock assert not writer_thread.is_alive() assert not reader_thread.is_alive() assert not writer_errors assert not reader_errors assert len(values) == 1 assert "tree" in writer_state return values[0], writer_state["tree"] def _barriered_snapshot(writer, snapshotter): gate = threading.Barrier(2) writer_ready = threading.Event() reader_started = threading.Event() writer_errors: list[BaseException] = [] reader_errors: list[BaseException] = [] reader_values: list[object] = [] def run_writer() -> None: try: writer(writer_ready, gate) except BaseException as exc: writer_errors.append(exc) def run_reader() -> None: reader_started.set() try: reader_values.append(snapshotter()) except BaseException as exc: reader_errors.append(exc) writer_thread = threading.Thread(target=run_writer) writer_thread.start() assert writer_ready.wait(timeout=5) reader_thread = threading.Thread(target=run_reader) reader_thread.start() assert reader_started.wait(timeout=5) gate.wait(timeout=5) writer_thread.join(timeout=5) reader_thread.join(timeout=5) assert not writer_thread.is_alive() assert not reader_thread.is_alive() assert not writer_errors if reader_errors: assert len(reader_errors) == 1 error = reader_errors[0] assert isinstance(error, DiagnosticEvidenceError) assert str(error) == "snapshot_unstable" return None assert len(reader_values) == 1 return reader_values[0] def _snapshot_after_lock_inode_replacement(fixture: dict[str, object]): writer_lock = fixture["writer_lock"] assert callable(writer_lock) writer_context = writer_lock() writer_context.__enter__() reader_lock_attempt = threading.Event() release_reader = threading.Event() reader_errors: list[BaseException] = [] original_flock = evidence_module.fcntl.flock def observed_flock(descriptor: int, operation: int) -> None: if operation == fcntl.LOCK_SH: reader_lock_attempt.set() assert release_reader.wait(timeout=5) original_flock(descriptor, operation) evidence_module.fcntl.flock = observed_flock reader_thread = threading.Thread( target=lambda: _capture_snapshot_error(fixture["descriptor"], reader_errors) ) reader_thread.start() try: assert reader_lock_attempt.wait(timeout=5) _replace_lock_inode(fixture["descriptor"].lock_path) baseline = _tree_state(fixture["root"]) release_reader.set() writer_context.__exit__(None, None, None) writer_context = None finally: release_reader.set() if writer_context is not None: writer_context.__exit__(None, None, None) evidence_module.fcntl.flock = original_flock reader_thread.join(timeout=5) assert not reader_thread.is_alive() assert len(reader_errors) == 1 error = reader_errors[0] assert isinstance(error, DiagnosticEvidenceError) assert str(error) == "snapshot_unstable" return baseline def _capture_snapshot_error(descriptor: object, errors: list[BaseException]) -> None: try: snapshot_diagnostic_source(descriptor) except BaseException as exc: errors.append(exc) def test_profile_authority_descriptor_excludes_unlocked_config(tmp_path): fixture = _profile_authority_domain_source_tree(tmp_path) descriptor = fixture["descriptor"] config = fixture["root"] / "config.json" config.write_text("{}\n", encoding="utf-8") config.chmod(0o600) assert config not in descriptor.data_paths assert all(path.name not in {"config.json", "config.yaml", "config.yml"} for path in descriptor.data_paths) @pytest.mark.parametrize( "builder", [ pytest.param(_wizard_domain_source_tree, id="wizard_draft"), pytest.param(_profile_authority_domain_source_tree, id="profile_authority"), ], ) def test_writer_inodes_are_shared_with_typed_readers_without_mutation(tmp_path, builder): fixture = builder(tmp_path) descriptor = fixture["descriptor"] lock = descriptor.lock_path lock_info = lock.lstat() assert stat.S_ISREG(lock_info.st_mode) assert stat.S_IMODE(lock_info.st_mode) == 0o600 receipt, before_reader = _deterministic_writer_reader_snapshot(fixture) assert receipt.kind is descriptor.kind assert _tree_state(fixture["root"]) == before_reader assert snapshot_diagnostic_source(descriptor) == receipt @pytest.mark.parametrize("builder", _DOMAIN_SOURCE_BUILDERS) def test_missing_writer_lock_is_noncreating_and_nonmutating(tmp_path, builder): fixture = builder(tmp_path) descriptor = fixture["descriptor"] descriptor.lock_path.unlink() before_reader = _tree_state(fixture["root"]) with pytest.raises(DiagnosticEvidenceError, match="snapshot_unstable"): snapshot_diagnostic_source(descriptor) assert not descriptor.lock_path.exists() assert _tree_state(fixture["root"]) == before_reader @pytest.mark.parametrize("mutation", ["lock", "data"]) @pytest.mark.parametrize("builder", _DOMAIN_SOURCE_BUILDERS) def test_symlinked_writer_lock_or_data_is_rejected_nonmutating( tmp_path, builder, mutation, ): fixture = builder(tmp_path) descriptor = fixture["descriptor"] if mutation == "lock": path = descriptor.lock_path else: path = next(path for path in descriptor.data_paths if path.exists()) target = tmp_path / f"outside-{descriptor.kind.value}-{mutation}" _replace_with_symlink(path, target) before_reader = _tree_state(fixture["root"]) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(descriptor) assert _tree_state(fixture["root"]) == before_reader @pytest.mark.parametrize("builder", _DOMAIN_SOURCE_BUILDERS) def test_writer_lock_inode_replacement_is_snapshot_unstable_nonmutating(tmp_path, builder): fixture = builder(tmp_path) before_reader = _snapshot_after_lock_inode_replacement(fixture) assert _tree_state(fixture["root"]) == before_reader def _promotion_roots(tmp_path: Path): profile = tmp_path / "profile" hermes = tmp_path / "hermes" (profile / "checkin_cli").mkdir(parents=True) (profile / "tests").mkdir() (hermes / "gateway" / "platforms").mkdir(parents=True) (hermes / "tests" / "gateway").mkdir(parents=True) (profile / "pyproject.toml").write_text("[project]\n") (profile / "checkin_cli" / "__init__.py").write_text("") (hermes / "gateway" / "__init__.py").write_text("") (hermes / "gateway" / "platforms" / "nutrition_coaching.py").write_text("") return profile, hermes def _entry(root: Path, prefix: str, relative: str, *, kind: str = "code"): raw = (root / relative).read_bytes() return { "kind": kind, "relative_path": f"{prefix}/{relative}", "sha256": hashlib.sha256(raw).hexdigest(), "schema_version": "diagnostic_promotion_manifest_v1", } def test_snapshot_is_bounded_and_nonmutating(tmp_path): _, _, _, descriptor = _typed_source_tree(tmp_path) before = { str(path): (path.read_bytes(), path.stat().st_mode, path.stat().st_mtime_ns) for path in (*descriptor.data_paths, descriptor.lock_path) } receipt = snapshot_diagnostic_source(descriptor) after = { str(path): (path.read_bytes(), path.stat().st_mode, path.stat().st_mtime_ns) for path in (*descriptor.data_paths, descriptor.lock_path) } assert before == after assert "bounded" not in json.dumps(receipt.to_dict()) def test_missing_or_symlink_lock_fails_without_creation(tmp_path): missing = tmp_path / ".missing" with pytest.raises(DiagnosticEvidenceError): with existing_read_lock(missing): pass assert not missing.exists() target = tmp_path / "target" target.write_bytes(b"") target.chmod(0o600) link = tmp_path / ".link" link.symlink_to(target) with pytest.raises(DiagnosticEvidenceError): with existing_read_lock(link): pass def test_typed_descriptor_rejects_unrelated_lock_and_symlink_data(tmp_path): _, _, _, descriptor = _typed_source_tree(tmp_path) unrelated = descriptor.lock_path.parent / ".unrelated.lock" unrelated.write_bytes(b"") unrelated.chmod(0o600) object.__setattr__(descriptor, "lock_path", unrelated) with pytest.raises(DiagnosticEvidenceError, match="unrelated"): snapshot_diagnostic_source(descriptor) _, _, _, descriptor = _typed_source_tree(tmp_path / "symlink") data = descriptor.data_paths[0] target = data.with_name("events-target.jsonl") target.write_text("{}\n") target.chmod(0o600) data.unlink() data.symlink_to(target) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(descriptor) def test_typed_snapshot_uses_fixed_writer_domain_and_bounds_rows(tmp_path): profile, _, runtime, descriptor = _typed_source_tree(tmp_path) receipt = snapshot_diagnostic_source(descriptor) assert receipt.kind is DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE assert receipt.row_count == 2 assert "bounded" not in json.dumps(receipt.to_dict()) assert diagnostic_source(profile, runtime, DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE).lock_path.name == ".events.lock" def test_diagnostic_source_requires_registry_derived_runtime(tmp_path): profile, runtime = _registered_runtime_tree(tmp_path) source_kind = DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE standalone = CustomerRuntime(runtime.spec, runtime.data_root) uninitialized = object.__new__(CustomerRuntime) _, other_runtime = _registered_runtime_tree(tmp_path / "other") forged_binding = CustomerRuntime( runtime.spec, runtime.data_root, other_runtime.registered_binding, ) wrong_root = tmp_path / "wrong-root" wrong_root.mkdir() forged_root = CustomerRuntime(runtime.spec, wrong_root, runtime.binding) for candidate in (standalone, uninitialized, forged_binding, forged_root): with pytest.raises(DiagnosticEvidenceError): diagnostic_source(profile, candidate, source_kind) descriptor = diagnostic_source( profile, runtime, DiagnosticEvidenceSource.WIZARD_DRAFT, ) forged_descriptor = object.__new__(type(descriptor)) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(forged_descriptor) def test_canonical_sequence_read_locked_missing_lock_is_noncreating(tmp_path): customer = tmp_path / "customer" (customer / "wizard").mkdir(parents=True, mode=0o700) (customer / "nutrition-plans").mkdir(mode=0o700) journal = CanonicalSequenceJournal( customer / "wizard" / "events.jsonl", customer / "nutrition-plans" / "canonical-sequence.jsonl", ) before = _tree_state(customer) with pytest.raises(FileNotFoundError): with journal.read_locked(): pass assert not journal.lock_path.exists() assert _tree_state(customer) == before @pytest.mark.parametrize( "builder", [_canonical_source_tree, _adaptive_source_tree, _overlay_source_tree], ids=["canonical", "adaptive", "overlay"], ) def test_missing_domain_lock_is_snapshot_unstable_without_creation(tmp_path, builder): _, customer, _, descriptor, _ = builder(tmp_path) descriptor.lock_path.unlink() before = _tree_state(customer) with pytest.raises(DiagnosticEvidenceError, match="snapshot_unstable"): snapshot_diagnostic_source(descriptor) assert not descriptor.lock_path.exists() assert _tree_state(customer) == before def test_canonical_pair_reader_waits_for_writer_and_is_nonmutating(tmp_path): _, customer, _, descriptor, transaction = _canonical_source_tree(tmp_path) before = snapshot_diagnostic_source(descriptor) state: dict[str, object] = {} original_append_pair = transaction._append_pair_locked def paused_append(event, token, *, intent_id=None): ready = state["ready"] gate = state["gate"] assert isinstance(ready, threading.Event) assert isinstance(gate, threading.Barrier) ready.set() gate.wait(timeout=5) return original_append_pair(event, token, intent_id=intent_id) setattr(transaction, "_append_pair_locked", paused_append) def writer(ready, gate): state["ready"] = ready state["gate"] = gate transaction.append(_canonical_event(2)) try: result = _barriered_snapshot( writer, lambda: snapshot_diagnostic_source(descriptor), ) finally: setattr(transaction, "_append_pair_locked", original_append_pair) after_writer = _tree_state(customer) final = snapshot_diagnostic_source(descriptor) assert after_writer == _tree_state(customer) assert final.row_count == before.row_count + 2 if result is not None: assert result == final def test_adaptive_journal_reader_waits_for_writer_and_is_nonmutating(tmp_path): _, customer, _, descriptor, store = _adaptive_source_tree(tmp_path) before = snapshot_diagnostic_source(descriptor) state: dict[str, object] = {} original_append_row = store._append_row def paused_append_row(path, row): ready = state["ready"] gate = state["gate"] assert isinstance(ready, threading.Event) assert isinstance(gate, threading.Barrier) ready.set() gate.wait(timeout=5) return original_append_row(path, row) setattr(store, "_append_row", paused_append_row) def writer(ready, gate): state["ready"] = ready state["gate"] = gate store.append( "plan_proposed", {"proposal_digest": "b" * 64, "revision": 2}, dedupe_key="diagnostic-adaptive-2", ) try: result = _barriered_snapshot( writer, lambda: snapshot_diagnostic_source(descriptor), ) finally: setattr(store, "_append_row", original_append_row) after_writer = _tree_state(customer) final = snapshot_diagnostic_source(descriptor) assert after_writer == _tree_state(customer) assert final.row_count == before.row_count + 1 if result is not None: assert result == final def test_overlay_reader_waits_for_writer_and_is_nonmutating(tmp_path): _, customer, _, descriptor, journal = _overlay_source_tree(tmp_path) before = snapshot_diagnostic_source(descriptor) state: dict[str, object] = {} original_lock = adaptive_module._verified_private_lock @contextmanager def paused_lock(path, *, exclusive, create): with original_lock(path, exclusive=exclusive, create=create) as descriptor_handle: if Path(path) == journal.lock_path and exclusive: ready = state["ready"] gate = state["gate"] assert isinstance(ready, threading.Event) assert isinstance(gate, threading.Barrier) ready.set() gate.wait(timeout=5) yield descriptor_handle setattr(adaptive_module, "_verified_private_lock", paused_lock) def writer(ready, gate): state["ready"] = ready state["gate"] = gate journal.replace( "revision-1", AdaptiveOverlay( "revision-2", "b" * 64, "2026-08-04T00:00:00+09:00", supersedes_revision_id="revision-1", authority_snapshot_id="diagnostic-authority-1", ), ) try: result = _barriered_snapshot( writer, lambda: snapshot_diagnostic_source(descriptor), ) finally: setattr(adaptive_module, "_verified_private_lock", original_lock) after_writer = _tree_state(customer) final = snapshot_diagnostic_source(descriptor) assert after_writer == _tree_state(customer) assert final.row_count == before.row_count + 1 if result is not None: assert result == final def test_schedule_reader_waits_for_writer_and_is_nonmutating(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) before = snapshot_diagnostic_source(descriptor) state: dict[str, object] = {} original_lock = schedule_module._schedule_lock @contextmanager def paused_schedule_lock(profile_root): with original_lock(profile_root) as resources: ready = state["ready"] gate = state["gate"] assert isinstance(ready, threading.Event) assert isinstance(gate, threading.Barrier) ready.set() gate.wait(timeout=5) yield resources setattr(schedule_module, "_schedule_lock", paused_schedule_lock) def writer(ready, gate): state["ready"] = ready state["gate"] = gate reserve_customer_task_delivery( profile, CustomerScheduleTask("client_001", "daily", date(2026, 8, 4)), body="second schedule body", destination={"user_id": "second-user", "chat_id": "second-chat"}, template_digest="4" * 64, registry_digest="5" * 64, config_digest="6" * 64, reservation_id="reservation-00000002", ) try: result = _barriered_snapshot( writer, lambda: snapshot_diagnostic_source(descriptor), ) finally: setattr(schedule_module, "_schedule_lock", original_lock) after_writer = _tree_state(profile) final = snapshot_diagnostic_source(descriptor) assert after_writer == _tree_state(profile) assert final.row_count == before.row_count + 1 if result is not None: assert result == final def test_schedule_missing_lock_is_snapshot_unstable_without_creation(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) descriptor.lock_path.unlink() before = _tree_state(profile) with pytest.raises(DiagnosticEvidenceError, match="snapshot_unstable"): snapshot_diagnostic_source(descriptor) assert not descriptor.lock_path.exists() assert _tree_state(profile) == before def test_schedule_snapshot_success_is_bounded_and_nonmutating(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) before = _tree_state(profile) receipt = snapshot_diagnostic_source(descriptor) assert _tree_state(profile) == before assert receipt.kind is DiagnosticEvidenceSource.SCHEDULE assert receipt.row_count == 1 assert "private schedule body" not in json.dumps(receipt.to_dict()) assert "raw-user-id" not in json.dumps(receipt.to_dict()) assert "raw-chat-id" not in json.dumps(receipt.to_dict()) def test_schedule_snapshot_corrupt_ledger_is_nonmutating(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) ledger = profile / "data" / "scheduled-deliveries.jsonl" ledger.write_text("{not-json}\n", encoding="utf-8") before = _tree_state(profile) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(descriptor) assert _tree_state(profile) == before def test_schedule_snapshot_nonready_fence_is_nonmutating(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) fence = profile / "data" / "scheduled-deliveries-fence.json" payload = json.loads(fence.read_text(encoding="utf-8")) payload["state"] = "preparing" fence.write_text(json.dumps(payload) + "\n", encoding="utf-8") fence.chmod(0o600) before = _tree_state(profile) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(descriptor) assert _tree_state(profile) == before def test_schedule_snapshot_orphan_tombstone_is_nonmutating(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) orphan = ( profile / "data" / "customer-schedule-claims" / "client_001" / "2026-08-04" / "daily.claim" ) orphan.parent.mkdir(parents=True) orphan.write_bytes(b"scheduled-delivery-tombstone-v1\norphan-reservation\n") orphan.chmod(0o600) before = _tree_state(profile) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(descriptor) assert _tree_state(profile) == before def test_schedule_snapshot_symlink_claim_is_nonmutating(tmp_path): profile, descriptor = _schedule_source_tree(tmp_path) claim = next((profile / "data" / "customer-schedule-claims").rglob("*.claim")) outside = tmp_path / "outside.claim" outside.write_bytes(claim.read_bytes()) outside.chmod(0o600) claim.unlink() claim.symlink_to(outside) before = _tree_state(profile) with pytest.raises(DiagnosticEvidenceError): snapshot_diagnostic_source(descriptor) assert _tree_state(profile) == before def test_synthetic_replay_has_no_raw_rows(tmp_path): _, _, _, descriptor = _typed_source_tree(tmp_path) receipt = snapshot_diagnostic_source(descriptor) fixture = synthetic_replay_fixture([receipt]) require_synthetic_provenance(fixture) assert "bounded" not in json.dumps(fixture) with pytest.raises(DiagnosticEvidenceError): require_synthetic_provenance({"provenance": "production"}) def test_promotion_manifest_rejects_unlisted_field(tmp_path): profile, hermes = _promotion_roots(tmp_path) candidate = profile / "checkin_cli" / "diagnostic.py" candidate.write_text("SCHEMA = 'diagnostic'\n") entry = _entry(profile, "profile", "checkin_cli/diagnostic.py") entry["runtime_data"] = False with pytest.raises(DiagnosticEvidenceError, match="closed"): build_diagnostic_promotion_manifest(entry and [entry], approved_roots={"profile": profile, "hermes": hermes}) def test_promotion_manifest_rejects_path_traversal_and_symlink(tmp_path): profile, hermes = _promotion_roots(tmp_path) outside = tmp_path / "outside.py" outside.write_text("outside = True\n") traversal = { "kind": "code", "relative_path": "profile/../outside.py", "sha256": hashlib.sha256(outside.read_bytes()).hexdigest(), "schema_version": "diagnostic_promotion_manifest_v1", } with pytest.raises(DiagnosticEvidenceError): build_diagnostic_promotion_manifest([traversal], approved_roots={"profile": profile, "hermes": hermes}) link = profile / "checkin_cli" / "linked.py" link.symlink_to(outside) with pytest.raises(DiagnosticEvidenceError): build_diagnostic_promotion_manifest( [_entry(profile, "profile", "checkin_cli/linked.py")], approved_roots={"profile": profile, "hermes": hermes}, ) def test_promotion_manifest_rejects_bad_digest_and_suffix(tmp_path): profile, hermes = _promotion_roots(tmp_path) candidate = profile / "checkin_cli" / "diagnostic.py" candidate.write_text("SCHEMA = 'diagnostic'\n") bad = _entry(profile, "profile", "checkin_cli/diagnostic.py") bad["sha256"] = "0" * 64 with pytest.raises(DiagnosticEvidenceError, match="digest"): build_diagnostic_promotion_manifest([bad], approved_roots={"profile": profile, "hermes": hermes}) config = profile / "checkin_cli" / "runtime.json" config.write_text("{}\n") with pytest.raises(DiagnosticEvidenceError, match="suffix"): build_diagnostic_promotion_manifest( [_entry(profile, "profile", "checkin_cli/runtime.json")], approved_roots={"profile": profile, "hermes": hermes}, )