142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
"""Curator-facing Evidence authoring commands, separate from runtime preprocessing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
import typer
|
|
|
|
from tht.evidence import (
|
|
EvidencePreparationError,
|
|
PiEvidenceRestructurer,
|
|
prepare_workspace_evidence,
|
|
resolve_workspace_evidence,
|
|
validate_workspace_evidence,
|
|
)
|
|
|
|
evidence_app = typer.Typer(help="Prepare and validate workspace Evidence", no_args_is_help=True)
|
|
|
|
|
|
def _canonical_worktree(workspace_root: Path) -> Path:
|
|
if workspace_root.is_symlink():
|
|
raise typer.BadParameter("workspace-root must not be a symlink")
|
|
root = workspace_root.resolve()
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--show-toplevel"],
|
|
cwd=root,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0 or Path(result.stdout.strip()).resolve() != root:
|
|
raise typer.BadParameter("workspace-root must be the root of a canonical Git worktree")
|
|
return root
|
|
|
|
|
|
def _emit(payload: dict[str, object], json_output: bool) -> None:
|
|
if json_output:
|
|
typer.echo(json.dumps(payload, ensure_ascii=False, sort_keys=True))
|
|
return
|
|
for key, value in payload.items():
|
|
typer.echo(f"{key}: {value}")
|
|
|
|
|
|
def _preparation_payload(report) -> dict[str, object]:
|
|
return {
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_prepare",
|
|
"status": "prepared",
|
|
"changed": list(report.changed),
|
|
"unchanged": list(report.unchanged),
|
|
"created": list(report.created),
|
|
"orphaned": list(report.orphaned),
|
|
"findings": _findings_payload(report.findings),
|
|
"modelCalls": report.model_calls,
|
|
}
|
|
|
|
|
|
def _findings_payload(findings) -> list[dict[str, object]]:
|
|
return [finding.__dict__ for finding in findings]
|
|
|
|
|
|
@evidence_app.command("prepare")
|
|
def prepare_cmd(
|
|
workspace_root: Path,
|
|
upgrade: Annotated[bool, typer.Option(help="Reprocess every source with the installed pipeline.")] = False,
|
|
json_output: Annotated[bool, typer.Option("--json", help="Write machine JSON to stdout.")] = False,
|
|
) -> None:
|
|
"""Prepare changed Source Evidence without committing or publishing it."""
|
|
root = _canonical_worktree(workspace_root)
|
|
skill_path = Path(__file__).resolve().parents[2] / ".pi" / "skills" / "tht-evidence-authoring" / "SKILL.md"
|
|
restructurer = PiEvidenceRestructurer(os.environ.get("THT_PI_EXECUTABLE", "pi"), skill_path)
|
|
try:
|
|
report = prepare_workspace_evidence(root, restructurer=restructurer, upgrade=upgrade)
|
|
except EvidencePreparationError as error:
|
|
_emit({"schemaVersion": 1, "operation": "evidence_prepare", "status": "failed", "code": error.code}, json_output)
|
|
raise typer.Exit(code=1) from error
|
|
_emit(_preparation_payload(report), json_output)
|
|
if report.findings:
|
|
raise typer.Exit(code=3)
|
|
|
|
|
|
@evidence_app.command("validate")
|
|
def validate_cmd(
|
|
workspace_root: Path,
|
|
json_output: Annotated[bool, typer.Option("--json", help="Write machine JSON to stdout.")] = False,
|
|
) -> None:
|
|
"""Validate the authoring tree without writing, committing, or publishing it."""
|
|
root = _canonical_worktree(workspace_root)
|
|
report = validate_workspace_evidence(root)
|
|
payload = {
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_validate",
|
|
"status": "valid" if report.publishable else "review_required",
|
|
"publishable": report.publishable,
|
|
"findings": _findings_payload(report.findings),
|
|
}
|
|
_emit(payload, json_output)
|
|
if report.publishable:
|
|
return
|
|
if all(finding.code in {"orphaned_unit", "unresolved_review_item"} for finding in report.findings):
|
|
raise typer.Exit(code=3)
|
|
raise typer.Exit(code=1)
|
|
|
|
|
|
@evidence_app.command("resolve")
|
|
def resolve_cmd(
|
|
workspace_root: Path,
|
|
evidence_id: str,
|
|
retire: Annotated[bool, typer.Option(help="Retire the Evidence unit.")] = False,
|
|
source: Annotated[Path | None, typer.Option(help="Relink the unit to this source/ path.")] = None,
|
|
json_output: Annotated[bool, typer.Option("--json", help="Write machine JSON to stdout.")] = False,
|
|
) -> None:
|
|
"""Explicitly retire or relink one Evidence unit without publishing it."""
|
|
if retire == (source is not None):
|
|
raise typer.BadParameter("choose exactly one of --retire or --source")
|
|
root = _canonical_worktree(workspace_root)
|
|
try:
|
|
report = resolve_workspace_evidence(root, evidence_id, retire=retire, source=source)
|
|
except EvidencePreparationError as error:
|
|
_emit({"schemaVersion": 1, "operation": "evidence_resolve", "status": "failed", "code": error.code}, json_output)
|
|
exit_code = 2 if error.code in {
|
|
"evidence_id_invalid", "resolve_mode_invalid", "source_invalid",
|
|
} else 1
|
|
raise typer.Exit(code=exit_code) from error
|
|
_emit({
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_resolve",
|
|
"status": "resolved",
|
|
"action": report.action,
|
|
"evidenceId": report.evidence_id,
|
|
"sourceFile": report.source_file,
|
|
"findings": _findings_payload(report.findings),
|
|
}, json_output)
|
|
if report.findings:
|
|
if all(finding.code in {"orphaned_unit", "unresolved_review_item"} for finding in report.findings):
|
|
raise typer.Exit(code=3)
|
|
raise typer.Exit(code=1)
|