ThothII/harness/tests/test_decision_join_set_cli.py

import fcntl
import json
import multiprocessing
import time

import pytest
from typer.testing import CliRunner

from tht.cli.decision_cmd import decision_app
from tht.decisions import append_decision, append_decisions, list_decisions
from tht.phase import current_phase


def _concurrent_append_worker(session, subject, start, ready, done):
    ready.put(subject)
    start.wait()
    try:
        append_decision(session, type="concept_clarified", subject=subject)
        done.put((subject, None))
    except Exception as error:  # pragma: no cover - surfaced through the parent assertion
        done.put((subject, repr(error)))


def _walk_to_phase(session, target):
    while current_phase(session) < target:
        append_decision(session, type="phase_approved", subject=f"phase:{current_phase(session)}")


def _configure_command(monkeypatch, sessions):
    import tht.cli.decision_cmd as mod
    import tht.cli.session_cmd as session_mod
    from tht.session.models import SessionManifest, SessionSnapshot

    class _Repository:
        def get(self, session_id):
            return SessionSnapshot(
                manifest=SessionManifest(id=session_id, created_at="2026-01-01T00:00:00Z", question="q", database="d", schema="s"),
                decisions=list_decisions(sessions / session_id),
            )

        def append_decisions(self, session_id, decisions):
            return append_decisions(sessions / session_id, list(decisions))

    class _Cfg:
        pass

    repository = _Repository()
    monkeypatch.setattr(mod, "_load_config_or_exit", lambda _c: _Cfg())
    monkeypatch.setattr(mod, "load_session_or_exit", lambda _cfg, _s: None)
    monkeypatch.setattr(mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
    monkeypatch.setattr(mod, "session_repository", lambda _cfg: repository)
    monkeypatch.setattr(session_mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))


def test_add_join_set_rejects_the_whole_batch_when_one_item_is_invalid(tmp_path, monkeypatch):
    session_id = "s1"
    session = tmp_path / session_id
    session.mkdir()
    _walk_to_phase(session, 4)
    _configure_command(monkeypatch, tmp_path)

    payload = [
        {"type": "join_modified", "subject": "a-b", "detail": "a.id = b.a_id"},
        {"type": "not_a_decision", "subject": "b-c", "detail": "b.id = c.b_id"},
    ]
    result = CliRunner().invoke(
        decision_app,
        ["add-join-set", "--session", session_id, "--doc", "-"],
        input=json.dumps(payload),
    )

    assert result.exit_code != 0
    assert not [decision for decision in list_decisions(session) if decision.type == "join_modified"]


def test_add_join_set_appends_the_complete_valid_batch(tmp_path, monkeypatch):
    session_id = "s1"
    session = tmp_path / session_id
    session.mkdir()
    _walk_to_phase(session, 4)
    _configure_command(monkeypatch, tmp_path)

    payload = [
        {"type": "join_modified", "subject": "a-b", "detail": "a.id = b.a_id"},
        {"type": "join_modified", "subject": "b-c", "detail": "b.id = c.b_id"},
    ]
    result = CliRunner().invoke(
        decision_app,
        ["add-join-set", "--session", session_id, "--doc", "-"],
        input=json.dumps(payload),
    )

    assert result.exit_code == 0, result.output
    joins = [decision for decision in list_decisions(session) if decision.type == "join_modified"]
    assert [decision.subject for decision in joins] == ["a-b", "b-c"]


def test_atomic_replace_failure_keeps_the_original_ledger(tmp_path, monkeypatch):
    session = tmp_path / "s1"
    session.mkdir()
    append_decision(session, type="phase_approved", subject="phase:1")
    original = (session / "review_decisions.jsonl").read_text()

    import tht.decisions as mod

    def fail_replace(_source, _target):
        raise OSError("disk failure")

    monkeypatch.setattr(mod.os, "replace", fail_replace)
    with pytest.raises(OSError, match="disk failure"):
        append_decisions(
            session,
            [
                {"type": "join_modified", "subject": "a-b"},
                {"type": "join_modified", "subject": "b-c"},
            ],
        )

    assert (session / "review_decisions.jsonl").read_text() == original


def test_concurrent_appends_wait_for_the_session_lock_and_keep_both_records(tmp_path):
    session = tmp_path / "s1"
    session.mkdir()
    context = multiprocessing.get_context("fork")
    start = context.Event()
    ready = context.Queue()
    done = context.Queue()
    processes = [
        context.Process(
            target=_concurrent_append_worker,
            args=(session, subject, start, ready, done),
        )
        for subject in ("first", "second")
    ]
    for process in processes:
        process.start()
    assert {ready.get(timeout=5), ready.get(timeout=5)} == {"first", "second"}

    lock_path = session / ".review_decisions.lock"
    with lock_path.open("a+") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        start.set()
        time.sleep(0.2)
        writers_waited = not (session / "review_decisions.jsonl").exists()
        fcntl.flock(lock, fcntl.LOCK_UN)

    results = [done.get(timeout=5), done.get(timeout=5)]
    for process in processes:
        process.join(timeout=5)
        assert process.exitcode == 0

    assert writers_waited
    assert all(error is None for _subject, error in results)
    records = list_decisions(session)
    assert {record.subject for record in records} == {"first", "second"}
    assert [record.seq for record in records] == [1, 2]