feat(evidence): resolve curated evidence explicitly

harness/tht/evidence/authoring.py

@@ -173,6 +173,16 @@ class EvidencePreparationReport:
     model_calls: int
 
 
+@dataclass(frozen=True)
+class EvidenceResolutionReport:
+    """The result of one curator-directed Evidence resolution."""
+
+    action: Literal["retired", "relinked"]
+    evidence_id: str
+    source_file: str | None
+    findings: tuple[ValidationFinding, ...]
+
+
 class ManifestSource(StrictModel):
     sha256: str
     units: tuple[str, ...]
@@ -531,6 +541,100 @@ def prepare_workspace_evidence(
     )
 
 
+def resolve_workspace_evidence(
+    workspace_root: Path,
+    evidence_id: str,
+    *,
+    retire: bool = False,
+    source: str | Path | None = None,
+    git_status: Callable[[Path], tuple[str, ...]] | None = None,
+) -> EvidenceResolutionReport:
+    """Retire or relink one Evidence unit as one staged curator update.
+
+    The operation intentionally neither creates a commit nor publishes anything.  Both
+    the curated tree and the managed manifest are replaced only after the staged tree
+    has been written and structurally validated.
+    """
+    if retire == (source is not None):
+        raise EvidencePreparationError("resolve_mode_invalid")
+    if not is_evidence_id(evidence_id):
+        raise EvidencePreparationError("evidence_id_invalid")
+
+    workspace_root = workspace_root.resolve()
+    evidence_root = workspace_root / "evidence"
+    _reject_dirty_worktree(workspace_root, git_status or _git_status)
+    try:
+        manifest = load_manifest(evidence_root / "manifest.yaml")
+        documents = load_curated_tree(evidence_root / "curated")
+    except (OSError, ValidationError, ValueError) as error:
+        raise EvidencePreparationError("authoring_state_invalid") from error
+
+    documents_by_id = {document.id: document for document in documents}
+    if len(documents_by_id) != len(documents):
+        raise EvidencePreparationError("duplicate_evidence_id")
+    if evidence_id not in documents_by_id:
+        raise EvidencePreparationError("evidence_not_found")
+
+    source_units = {path: tuple(entry.units) for path, entry in manifest.sources.items()}
+    orphaned = set(manifest.orphans)
+    source_file: str | None = None
+    action: Literal["retired", "relinked"]
+
+    if retire:
+        action = "retired"
+        del documents_by_id[evidence_id]
+        for path, units in tuple(source_units.items()):
+            source_units[path] = tuple(unit for unit in units if unit != evidence_id)
+        orphaned.discard(evidence_id)
+    else:
+        action = "relinked"
+        assert source is not None
+        source_file = _resolve_source_file(evidence_root, source)
+        source_text = _load_resolution_source(evidence_root, source_file)
+        document = documents_by_id[evidence_id]
+        review_items = document.review_items
+        if all(_normalize(excerpt) in source_text for excerpt in document.provenance.supporting_excerpts):
+            review_items = tuple(
+                item for item in review_items if item.code != "source_no_longer_supports_unit"
+            )
+        documents_by_id[evidence_id] = document.model_copy(update={
+            "provenance": document.provenance.model_copy(update={
+                "source_file": source_file,
+                "source_sha256": _source_hash(source_text),
+            }),
+            "review_items": review_items,
+        })
+        for path, units in tuple(source_units.items()):
+            source_units[path] = tuple(unit for unit in units if unit != evidence_id)
+        source_units[source_file] = tuple(sorted((*source_units.get(source_file, ()), evidence_id)))
+        orphaned.discard(evidence_id)
+
+    sources = {
+        path: ManifestSource(
+            sha256=(
+                _source_hash(_load_resolution_source(evidence_root, path))
+                if path == source_file
+                else manifest.sources[path].sha256
+            ),
+            units=units,
+        )
+        for path, units in sorted(source_units.items())
+    }
+    next_manifest = EvidenceManifest(
+        schema_version=1,
+        pipeline_version=manifest.pipeline_version,
+        sources=sources,
+        orphans=tuple(sorted(orphaned)),
+    )
+    findings = _stage_and_apply_authoring_tree(workspace_root, documents_by_id, next_manifest)
+    return EvidenceResolutionReport(
+        action=action,
+        evidence_id=evidence_id,
+        source_file=source_file,
+        findings=findings,
+    )
+
+
 def _empty_manifest() -> EvidenceManifest:
     return EvidenceManifest(
         schema_version=1,
@@ -558,6 +662,30 @@ def _load_preparation_manifest(path: Path, *, upgrade: bool) -> EvidenceManifest
             raise EvidencePreparationError("authoring_state_invalid") from error
 
 
+def _resolve_source_file(evidence_root: Path, source: str | Path) -> str:
+    raw_source = source.as_posix() if isinstance(source, Path) else source
+    try:
+        source_file = validate_source_file(raw_source)
+    except ValueError as error:
+        raise EvidencePreparationError("source_invalid", raw_source) from error
+    path = evidence_root / source_file
+    ancestor = evidence_root
+    for part in Path(source_file).parts:
+        ancestor /= part
+        if ancestor.is_symlink():
+            raise EvidencePreparationError("source_invalid", source_file)
+    if not path.is_file() or path.stat().st_size > MAX_AUTHORING_FILE_BYTES:
+        raise EvidencePreparationError("source_invalid", source_file)
+    return source_file
+
+
+def _load_resolution_source(evidence_root: Path, source_file: str) -> str:
+    try:
+        return normalize_source_text((evidence_root / source_file).read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError) as error:
+        raise EvidencePreparationError("source_invalid", source_file) from error
+
+
 def _git_status(workspace_root: Path) -> tuple[str, ...]:
     result = subprocess.run(
         ["git", "status", "--porcelain"],
@@ -581,6 +709,14 @@ def _reject_dirty_authoring_state(
             raise EvidencePreparationError("authoring_worktree_dirty")
 
 
+def _reject_dirty_worktree(
+    workspace_root: Path,
+    git_status: Callable[[Path], tuple[str, ...]],
+) -> None:
+    if git_status(workspace_root):
+        raise EvidencePreparationError("worktree_dirty")
+
+
 def _load_source_texts(evidence_root: Path) -> dict[str, str]:
     source_root = evidence_root / "source"
     if not source_root.is_dir():