# SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc import hashlib import json import multiprocessing import os import re import stat import threading import traceback from pathlib import Path import pytest from solstone.think.identity import ( STEWARD_SECTION_ATTENTION, STEWARD_SECTION_AUTO_REPAIRS, STEWARD_SECTION_STATUS, ensure_identity_directory, update_identity_section, write_identity, ) @pytest.fixture(autouse=True) def _temp_journal(monkeypatch, tmp_path): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) config_dir = tmp_path / "config" config_dir.mkdir() (config_dir / "journal.json").write_text("{}", encoding="utf-8") return tmp_path def _history_path(journal_path: Path) -> Path: return journal_path / "identity" / "history.jsonl" def _read_history(journal_path: Path) -> list[dict]: return [ json.loads(line) for line in _history_path(journal_path).read_text().splitlines() ] def test_write_identity_first_write(tmp_path): write_identity( "partner.md", actor="test writer", op="replace", section=None, content="first partner\n", reason="test", ) partner_path = tmp_path / "identity" / "partner.md" assert partner_path.read_text(encoding="utf-8") == "first partner\n" records = _read_history(tmp_path) assert len(records) == 1 record = records[0] assert record["before_hash"] == hashlib.sha256(b"").hexdigest() assert record["bytes_before"] == 0 assert record["after_hash"] == hashlib.sha256(b"first partner\n").hexdigest() assert record["bytes_after"] == len("first partner\n".encode("utf-8")) def test_write_identity_atomic_failure(tmp_path, monkeypatch): identity_dir = tmp_path / "identity" identity_dir.mkdir() target = identity_dir / "partner.md" target.write_text("original\n", encoding="utf-8") def fail_replace(src, dst): raise OSError("replace failed") monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) with pytest.raises(OSError, match="replace failed"): write_identity( "partner.md", actor="test writer", op="replace", section=None, content="updated\n", reason="test", ) assert target.read_text(encoding="utf-8") == "original\n" assert not _history_path(tmp_path).exists() assert list(identity_dir.glob(".tmp_*.tmp")) == [] def _identity_worker(journal_path, barrier, errors, index): os.environ["SOLSTONE_JOURNAL"] = journal_path try: from solstone.think.identity import write_identity barrier.wait(timeout=5) write_identity( "partner.md", actor=f"writer-{index}", op="replace", section=None, content=f"content-{index}\n", reason="test", ) except BaseException: errors.put(traceback.format_exc()) raise def test_write_identity_serializes_process_writers(tmp_path): ctx = multiprocessing.get_context("spawn") barrier = ctx.Barrier(4) errors = ctx.Queue() procs = [ ctx.Process(target=_identity_worker, args=(str(tmp_path), barrier, errors, i)) for i in range(4) ] for p in procs: p.start() for p in procs: p.join(timeout=15) for p in procs: if p.is_alive(): p.terminate() drained = [] while True: try: drained.append(errors.get_nowait()) except Exception: break assert all(p.exitcode == 0 for p in procs), "\n".join(drained) records = _read_history(tmp_path) assert len(records) == 4 assert {r["actor"] for r in records} == {f"writer-{i}" for i in range(4)} final = (tmp_path / "identity" / "partner.md").read_text(encoding="utf-8") assert final in {f"content-{i}\n" for i in range(4)} def test_write_identity_lock_serializes(tmp_path): def writer(actor: str, content: str) -> None: write_identity( "partner.md", actor=actor, op="replace", section=None, content=content, reason="test", ) thread_one = threading.Thread(target=writer, args=("writer-1", "first\n")) thread_two = threading.Thread(target=writer, args=("writer-2", "second\n")) thread_one.start() thread_two.start() thread_one.join() thread_two.join() final_content = (tmp_path / "identity" / "partner.md").read_text(encoding="utf-8") assert final_content in {"first\n", "second\n"} records = _read_history(tmp_path) assert len(records) == 2 assert {records[0]["actor"], records[1]["actor"]} == {"writer-1", "writer-2"} def test_write_identity_history_schema(tmp_path): write_identity( "partner.md", actor="schema test", op="replace", section=None, content="partner\n", reason="test", ) record = _read_history(tmp_path)[0] assert list(record) == [ "ts", "file", "actor", "op", "section", "reason", "before_hash", "after_hash", "bytes_before", "bytes_after", ] assert record["file"] == "partner.md" assert record["actor"] == "schema test" assert record["op"] == "replace" assert record["section"] is None assert isinstance(record["bytes_before"], int) assert isinstance(record["bytes_after"], int) assert isinstance(record["before_hash"], str) assert isinstance(record["after_hash"], str) assert re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", record["ts"]) def test_write_identity_mode_0600(tmp_path): write_identity( "partner.md", actor="mode test", op="replace", section=None, content="partner\n", reason="test", ) mode = stat.S_IMODE(os.stat(tmp_path / "identity" / "partner.md").st_mode) assert mode == 0o600 def test_update_identity_section_returns_false_no_change(tmp_path): identity_dir = tmp_path / "identity" identity_dir.mkdir() (identity_dir / "partner.md").write_text( "# partner\n\n## work patterns\nPrefers mornings\n", encoding="utf-8", ) changed = update_identity_section( "partner.md", "work patterns", "Prefers mornings", actor="section test", reason="test", ) assert changed is False assert not _history_path(tmp_path).exists() def test_health_md_bootstrap_creates_via_ensure_identity_directory(tmp_path): identity_dir = ensure_identity_directory() health = identity_dir / "health.md" assert health.read_text(encoding="utf-8") == "\n".join( [ STEWARD_SECTION_STATUS, "", "not yet generated", "", STEWARD_SECTION_ATTENTION, "", STEWARD_SECTION_AUTO_REPAIRS, "", ] ) record = [row for row in _read_history(tmp_path) if row["file"] == "health.md"][0] assert record["actor"] == "ensure_identity_directory" assert record["op"] == "create"