Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
236 lines
9.8 KiB
Python
236 lines
9.8 KiB
Python
import pytest
|
|
import yaml
|
|
|
|
from tht.evidence.authoring import (
|
|
EvidencePreparationError,
|
|
normalize_source_text,
|
|
prepare_workspace_evidence,
|
|
)
|
|
from tht.evidence.canonical import (
|
|
CuratedEvidence,
|
|
ManualEvidenceProvenance,
|
|
dump_curated_markdown,
|
|
parse_curated_markdown,
|
|
)
|
|
from tht.evidence.local_archive import ArchiveConflict, LocalEvidenceArchive, _digest
|
|
|
|
|
|
def write_unit(root, *, rule="Use order ID and year.", source=False):
|
|
provenance = {"kind": "manual", "declared_by": "local curator"}
|
|
if source:
|
|
path = root / "evidence/source/orders.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(rule)
|
|
provenance = {
|
|
"source_file": "source/orders.md",
|
|
"source_sha256": "sha256:" + _digest(normalize_source_text(rule).encode()),
|
|
"supporting_excerpts": [rule],
|
|
}
|
|
unit = CuratedEvidence.model_validate(
|
|
{
|
|
"schema_version": 4,
|
|
"id": "evidence:order-key",
|
|
"title": "Order key",
|
|
"kind": "domain",
|
|
"language": "en",
|
|
"purposes": ["sql_generation"],
|
|
"payload": {"rule": rule},
|
|
"provenance": provenance,
|
|
}
|
|
)
|
|
path = root / "evidence/curated/domain/order-key.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(dump_curated_markdown(unit))
|
|
return path
|
|
|
|
|
|
def read_active(archive):
|
|
return parse_curated_markdown(
|
|
(archive.active_snapshot() / "curated/domain/order-key.md").read_text()
|
|
)
|
|
|
|
|
|
def test_manual_creation_and_visible_edit_reach_only_the_consolidated_snapshot(tmp_path):
|
|
path = write_unit(tmp_path)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
first = archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
assert first["status"] == "active"
|
|
assert read_active(archive).provenance.declared_by == "Alice"
|
|
path.write_text(path.read_text().replace("Use order ID and year.", "Use ID, year and company."))
|
|
assert read_active(archive).payload.rule == "Use order ID and year."
|
|
candidate = archive.consolidate(actor="Bob")
|
|
assert candidate["status"] == "pending_activation"
|
|
assert read_active(archive).payload.rule == "Use order ID and year."
|
|
archive.consolidate(actor="Bob", activate=lambda _: None)
|
|
assert read_active(archive).payload.rule == "Use ID, year and company."
|
|
assert read_active(archive).provenance.declared_by == "Bob"
|
|
|
|
|
|
def test_document_correction_becomes_manual_and_retains_original_source_separately(tmp_path):
|
|
path = write_unit(tmp_path, source=True)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.initialize()
|
|
original = parse_curated_markdown(path.read_text()).provenance
|
|
path.write_text(
|
|
path.read_text().replace(
|
|
"## Rule\n\nUse order ID and year.",
|
|
"## Rule\n\nThe approved key also includes company.",
|
|
)
|
|
)
|
|
archive.consolidate(actor="Curator", activate=lambda _: None)
|
|
current = read_active(archive)
|
|
assert isinstance(current.provenance, ManualEvidenceProvenance)
|
|
assert current.provenance.original == original
|
|
assert current.provenance.declared_by == "Curator"
|
|
assert (archive.active_snapshot() / "source/orders.md").read_text() == "Use order ID and year."
|
|
|
|
|
|
def test_failed_activation_and_restart_retry_reuse_the_candidate(tmp_path):
|
|
path = write_unit(tmp_path)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
previous = archive.active_snapshot()
|
|
path.write_text(
|
|
path.read_text().replace("Use order ID and year.", "Use the reviewed composite key.")
|
|
)
|
|
calls = []
|
|
|
|
def fail(candidate):
|
|
calls.append(candidate)
|
|
raise RuntimeError("Index unavailable")
|
|
|
|
with pytest.raises(RuntimeError, match="Index unavailable"):
|
|
archive.consolidate(actor="Alice", activate=fail)
|
|
recovered = LocalEvidenceArchive(tmp_path)
|
|
assert recovered.active_snapshot() == previous
|
|
result = recovered.consolidate(actor="Alice", activate=calls.append)
|
|
assert calls[0] == calls[1]
|
|
assert result["status"] == "active"
|
|
|
|
|
|
def test_deleted_source_unit_stays_deleted_after_restart_and_refinement_attempt(tmp_path):
|
|
path = write_unit(tmp_path, source=True)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.initialize()
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
path.unlink()
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
state = yaml.safe_load((tmp_path / "evidence/.local/state.yaml").read_text())
|
|
assert state["deleted_ids"] == ["evidence:order-key"]
|
|
assert state["suppressed_sources"] == ["source/orders.md"]
|
|
|
|
class MustNotCall:
|
|
def restructure(self, request):
|
|
raise AssertionError("Deleted knowledge must not be regenerated")
|
|
|
|
with pytest.raises(EvidencePreparationError, match="explicit_source_refresh"):
|
|
prepare_workspace_evidence(tmp_path, restructurer=MustNotCall(), git_status=lambda _: ())
|
|
assert not path.exists()
|
|
assert not list(LocalEvidenceArchive(tmp_path).active_snapshot().rglob("*.md"))
|
|
|
|
|
|
def test_invalid_edit_does_not_change_files_metadata_or_active_snapshot(tmp_path):
|
|
path = write_unit(tmp_path)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
old = archive.active_snapshot()
|
|
state = (tmp_path / "evidence/.local/state.yaml").read_bytes()
|
|
path.write_text(path.read_text().replace("## Rule", "## Wrong heading"))
|
|
invalid = path.read_bytes()
|
|
with pytest.raises(ValueError, match="curated/domain/order-key.md.*section"):
|
|
archive.consolidate(actor="Alice", activate=lambda _: pytest.fail("Must not activate"))
|
|
assert path.read_bytes() == invalid
|
|
assert (tmp_path / "evidence/.local/state.yaml").read_bytes() == state
|
|
assert archive.active_snapshot() == old
|
|
|
|
|
|
def test_workflow_correction_checks_revision_and_never_overwrites_a_manual_edit(tmp_path):
|
|
path = write_unit(tmp_path)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
before = archive.get("evidence:order-key")
|
|
proposed = before["unit"].model_copy(update={"title": "Workflow correction"})
|
|
path.write_text(path.read_text().replace("Use order ID and year.", "Manual edit in progress."))
|
|
with pytest.raises(ArchiveConflict):
|
|
archive.save(proposed, expected_revision=before["revision"], actor="Reviewer")
|
|
assert "Manual edit in progress." in path.read_text()
|
|
latest = archive.get("evidence:order-key")
|
|
archive.save(
|
|
latest["unit"].model_copy(update={"title": "Approved title"}),
|
|
expected_revision=latest["revision"],
|
|
actor="Reviewer",
|
|
)
|
|
assert archive.get("evidence:order-key")["unit"].title == "Approved title"
|
|
|
|
|
|
def test_source_refresh_does_not_replace_manual_content_or_historical_citation(tmp_path):
|
|
path = write_unit(tmp_path, source=True)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.initialize()
|
|
path.write_text(
|
|
path.read_text().replace("## Rule\n\nUse order ID and year.", "## Rule\n\nManual rule.")
|
|
)
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
(tmp_path / "evidence/source/orders.md").write_text("Contradictory newer source.")
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
assert read_active(archive).payload.rule == "Manual rule."
|
|
assert (archive.active_snapshot() / "source/orders.md").read_text() == "Use order ID and year."
|
|
|
|
|
|
def test_symlinks_cannot_escape_the_archive(tmp_path):
|
|
path = write_unit(tmp_path)
|
|
path.unlink()
|
|
outside = tmp_path / "outside.md"
|
|
outside.write_text("Private unrelated file")
|
|
path.symlink_to(outside)
|
|
with pytest.raises(ValueError, match="symlinks"):
|
|
LocalEvidenceArchive(tmp_path).consolidate(actor="Alice")
|
|
|
|
|
|
@pytest.mark.parametrize("edit_after_interruption", [False, True])
|
|
def test_interrupted_normalization_recovers_without_losing_new_edits(
|
|
tmp_path, monkeypatch, edit_after_interruption
|
|
):
|
|
import tht.evidence.local_archive as module
|
|
|
|
path = write_unit(tmp_path, source=True)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.initialize()
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
old_active = archive.active_snapshot()
|
|
path.write_text(
|
|
path.read_text().replace("## Rule\n\nUse order ID and year.", "## Rule\n\nReviewed rule.")
|
|
)
|
|
atomic = module._atomic
|
|
|
|
def interrupted(target, data):
|
|
if target == path:
|
|
raise OSError("Interrupted file write")
|
|
atomic(target, data)
|
|
|
|
monkeypatch.setattr(module, "_atomic", interrupted)
|
|
with pytest.raises(OSError, match="Interrupted"):
|
|
archive.consolidate(actor="Bob", activate=lambda _: pytest.fail("Must not activate"))
|
|
assert archive.active_snapshot() == old_active
|
|
if edit_after_interruption:
|
|
path.write_text(path.read_text().replace("Reviewed rule.", "Further manual correction."))
|
|
monkeypatch.setattr(module, "_atomic", atomic)
|
|
recovered = LocalEvidenceArchive(tmp_path)
|
|
recovered.consolidate(actor="Bob", activate=lambda _: None)
|
|
assert read_active(recovered).payload.rule == (
|
|
"Further manual correction." if edit_after_interruption else "Reviewed rule."
|
|
)
|
|
assert read_active(recovered).provenance.declared_by == "Bob"
|
|
assert read_active(recovered).provenance.original.source_file == "source/orders.md"
|
|
|
|
|
|
def test_missing_archive_directory_is_not_interpreted_as_deletion(tmp_path):
|
|
write_unit(tmp_path)
|
|
archive = LocalEvidenceArchive(tmp_path)
|
|
archive.consolidate(actor="Alice", activate=lambda _: None)
|
|
active = archive.active_snapshot()
|
|
(archive.evidence / "curated").rename(archive.evidence / "unmounted")
|
|
with pytest.raises(ValueError, match="absence is not deletion"):
|
|
archive.consolidate(actor="Alice", activate=lambda _: pytest.fail("Must not activate"))
|
|
assert archive.active_snapshot() == active
|