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.
314 lines
12 KiB
Python
314 lines
12 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.cli.config_cmd import CONFIG_OPT
|
|
from tht.evidence import (
|
|
EvidencePreparationError,
|
|
PiEvidenceRestructurer,
|
|
migrate_workspace_evidence,
|
|
prepare_workspace_evidence,
|
|
resolve_workspace_evidence,
|
|
validate_workspace_evidence,
|
|
)
|
|
|
|
evidence_app = typer.Typer(help="Prepare and validate workspace Evidence", no_args_is_help=True)
|
|
|
|
|
|
@evidence_app.command("sources", hidden=True)
|
|
def sources_cmd(action: str, config: Path = CONFIG_OPT,
|
|
source_id: str | None = typer.Option(None), revision: str | None = typer.Option(None),
|
|
decision: str | None = typer.Option(None), actor: str = typer.Option("installation operator"),
|
|
json_output: bool = typer.Option(False, "--json")):
|
|
from tht.evidence.administration import ConsolidationError, source_action
|
|
try:
|
|
if action not in {"refresh", "decide"} or len(actor) > 256 or not actor.strip():
|
|
raise ValueError("Invalid source action")
|
|
if action == "refresh" and any(v is not None for v in (source_id, revision, decision)):
|
|
raise ValueError("Refresh does not accept decision options")
|
|
if action == "decide" and (decision not in {"keep", "replace"} or not source_id or not revision):
|
|
raise ValueError("A source decision requires source-id, revision and keep or replace")
|
|
payload = source_action(config, action=action, source_id=source_id, revision=revision,
|
|
decision=decision, actor=actor)
|
|
except Exception as error: # noqa: BLE001 - never expose connector/provider exception details
|
|
safe = isinstance(error, (ValueError, EvidencePreparationError, ConsolidationError))
|
|
_emit({"status": "failed", "code": "evidence_source_failed",
|
|
"error": str(error)[:1500] if safe else "Source acquisition or refinement failed; existing Evidence is preserved.",
|
|
"saved": isinstance(error, ConsolidationError) and error.saved}, json_output)
|
|
raise typer.Exit(1) from None
|
|
_emit(payload, json_output)
|
|
|
|
|
|
@evidence_app.command("admin", hidden=True)
|
|
def admin_cmd(workspace: str = typer.Option(...), config: Path = CONFIG_OPT):
|
|
from tht.evidence.administration import browse
|
|
try:
|
|
if os.environ.get("THT_PRINCIPAL_IS_ADMIN", "").lower() not in {"true", "1"}:
|
|
raise ValueError("Evidence administration requires an administrator")
|
|
payload = json.loads(config.read_text())
|
|
root = Path(payload["root"])
|
|
if not root.is_absolute() or root.name != workspace:
|
|
raise ValueError("Invalid workspace archive identity")
|
|
_emit(browse(root, payload.get("query", {})), True)
|
|
except (ValueError, OSError) as error:
|
|
_emit({"code": "evidence_unavailable", "message": str(error)[:1500]}, True)
|
|
raise typer.Exit(1) from None
|
|
|
|
|
|
def _canonical_worktree(workspace_root: Path) -> Path:
|
|
requested = workspace_root.absolute()
|
|
root = workspace_root.resolve()
|
|
if workspace_root.is_symlink() or requested != root:
|
|
raise typer.BadParameter("workspace-root must not be a symlink")
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--show-toplevel"],
|
|
cwd=root,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise typer.BadParameter("workspace-root must be inside a canonical Git worktree")
|
|
git_root = Path(result.stdout.strip()).resolve()
|
|
try:
|
|
root.relative_to(git_root)
|
|
except ValueError as error:
|
|
raise typer.BadParameter(
|
|
"workspace-root must be inside a canonical Git worktree",
|
|
) from error
|
|
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]
|
|
|
|
|
|
def evaluate_from_config(
|
|
workspace_root: Path,
|
|
config: Path,
|
|
*,
|
|
generation: str | None = None,
|
|
) -> dict[str, object]:
|
|
"""Evaluate one stored Evidence generation without changing corpus or vectors."""
|
|
from tht.adapters.factory import build_vector_store
|
|
from tht.cli.schema_cmd import _load_config_or_exit
|
|
from tht.cli.vector_cmd import make_embedder
|
|
from tht.evidence.canonical import load_curated_tree
|
|
from tht.evidence.corpus.store import CorpusStore
|
|
from tht.evidence.evaluation import evaluate_retrieval, load_evaluation_fixture
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
store = CorpusStore(cfg.paths.artifacts.parent / "corpus")
|
|
manifest = store.manifest(generation) if generation is not None else store.active_manifest()
|
|
if manifest is None:
|
|
raise RuntimeError("active Evidence generation is unavailable")
|
|
document_generations = manifest.metadata.get("document_generations")
|
|
if not isinstance(document_generations, dict):
|
|
raise TypeError("Evidence generation is invalid")
|
|
language = {"en": "english", "it": "italian"}.get(cfg.language)
|
|
if language is None:
|
|
raise RuntimeError("workspace language is unsupported for Qdrant BM25")
|
|
report = evaluate_retrieval(
|
|
load_evaluation_fixture(workspace_root / "evidence" / "evaluation.yaml"),
|
|
workspace_revision=cfg._workspace_revision,
|
|
vector_generation=manifest.vector_generation,
|
|
document_generations=document_generations,
|
|
workspace_id=cfg._workspace_id,
|
|
language=language,
|
|
searcher=build_vector_store(cfg),
|
|
embedder=make_embedder(cfg.embeddings),
|
|
expected_kinds={
|
|
evidence.id: evidence.kind
|
|
for evidence in load_curated_tree(workspace_root / "evidence" / "curated")
|
|
},
|
|
)
|
|
return report.model_dump()
|
|
|
|
|
|
@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)
|
|
from tht.evidence.authoring import authoring_skill_path
|
|
skill_path = authoring_skill_path()
|
|
restructurer = PiEvidenceRestructurer(os.environ.get("THT_PI_EXECUTABLE", "pi"), skill_path)
|
|
try:
|
|
try:
|
|
max_workers = int(os.environ.get("THT_EVIDENCE_AUTHORING_WORKERS", "1"))
|
|
except ValueError as error:
|
|
raise EvidencePreparationError("authoring_workers_invalid") from error
|
|
report = prepare_workspace_evidence(
|
|
root,
|
|
restructurer=restructurer,
|
|
upgrade=upgrade,
|
|
max_workers=max_workers,
|
|
)
|
|
except EvidencePreparationError as error:
|
|
payload = {
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_prepare",
|
|
"status": "failed",
|
|
"code": error.code,
|
|
}
|
|
if error.source_file is not None:
|
|
payload["sourceFile"] = error.source_file
|
|
_emit(payload, 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("migrate")
|
|
def migrate_cmd(
|
|
workspace_root: Path,
|
|
json_output: Annotated[bool, typer.Option("--json", help="Write machine JSON to stdout.")] = False,
|
|
) -> None:
|
|
"""Convert legacy units to editable v4 Markdown and establish a local baseline."""
|
|
root = _canonical_worktree(workspace_root)
|
|
try:
|
|
report = migrate_workspace_evidence(root)
|
|
except EvidencePreparationError as error:
|
|
_emit({
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_migrate",
|
|
"status": "failed",
|
|
"code": error.code,
|
|
}, json_output)
|
|
raise typer.Exit(code=1) from error
|
|
_emit({
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_migrate",
|
|
"status": "migrated",
|
|
"migrated": list(report.migrated),
|
|
"unchanged": list(report.unchanged),
|
|
"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)
|
|
|
|
|
|
@evidence_app.command("evaluate")
|
|
def evaluate_cmd(
|
|
workspace_root: Path,
|
|
config: Path = CONFIG_OPT,
|
|
generation: str | None = typer.Option(None, "--generation"),
|
|
json_output: bool = typer.Option(False, "--json"),
|
|
) -> None:
|
|
"""Evaluate active or selected Evidence retrieval generation without publishing it."""
|
|
root = _canonical_worktree(workspace_root)
|
|
try:
|
|
payload = evaluate_from_config(root, config, generation=generation)
|
|
except Exception: # noqa: BLE001 - CLI reports a safe operational failure.
|
|
_emit({
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_evaluate",
|
|
"status": "failed",
|
|
"code": "evaluation_failed",
|
|
}, json_output)
|
|
raise typer.Exit(code=1) from None
|
|
payload = {
|
|
"schemaVersion": 1,
|
|
"operation": "evidence_evaluate",
|
|
"status": "passed" if payload["passed"] else "failed",
|
|
**payload,
|
|
}
|
|
_emit(payload, json_output)
|
|
if not payload["passed"]:
|
|
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)
|