Add trusted auto-approval migration inventory

This commit is contained in:
2026-05-15 16:56:42 +02:00
parent a2c8ba9442
commit de8d184a4b
11 changed files with 381 additions and 3 deletions

View File

@@ -1,6 +1,8 @@
from __future__ import annotations
import argparse
import json
from dataclasses import asdict
from pathlib import Path
from typing import Sequence
@@ -171,6 +173,19 @@ def build_parser() -> argparse.ArgumentParser:
default="markdown",
help="Criteria output format.",
)
legacy = subparsers.add_parser(
"list-legacy-auto-approvals",
help="List historical trusted deterministic auto-approval records.",
)
legacy.add_argument("--database-path", help="Override REPO_REGISTRY_DATABASE_PATH.")
legacy.add_argument("--checkout-root", help="Override REPO_REGISTRY_CHECKOUT_ROOT.")
legacy.add_argument("--output", help="Write inventory output to this path instead of stdout.")
legacy.add_argument(
"--format",
choices=["json", "markdown"],
default="markdown",
help="Inventory output format.",
)
return parser
@@ -187,6 +202,8 @@ def main(argv: Sequence[str] | None = None) -> int:
return self_assess_command(args, parser)
if args.command == "list-quality-criteria":
return list_quality_criteria_command(args)
if args.command == "list-legacy-auto-approvals":
return list_legacy_auto_approvals_command(args)
parser.error(f"unknown command: {args.command}")
return 2
@@ -254,6 +271,39 @@ def list_quality_criteria_command(args: argparse.Namespace) -> int:
return 0
def list_legacy_auto_approvals_command(args: argparse.Namespace) -> int:
service = service_from_args(args)
records = service.list_trusted_auto_approval_migration_records()
if args.format == "json":
content = json.dumps([asdict(record) for record in records], indent=2) + "\n"
else:
content = legacy_auto_approval_records_markdown(records)
if args.output:
write_text(args.output, content)
else:
print(content, end="" if content.endswith("\n") else "\n")
return 0
def legacy_auto_approval_records_markdown(records) -> str:
if not records:
return "No legacy trusted auto-approval records found.\n"
lines = ["# Legacy Trusted Auto-Approval Records", ""]
for record in records:
lines.extend(
[
(
f"- repo={record.repository_id}:{record.repository_name} "
f"run={record.analysis_run_id} decision={record.review_decision_id}"
),
f" status={record.analysis_run_status} scanner={record.scanner_version or 'unknown'}",
f" approved_abilities={record.current_approved_ability_count}",
f" next={record.recommended_next_step}",
]
)
return "\n".join(lines) + "\n"
def self_assess_command(
args: argparse.Namespace,
parser: argparse.ArgumentParser,

View File

@@ -63,6 +63,22 @@ class ReviewDecision:
decision_kind: str = "other"
@dataclass(frozen=True)
class TrustedAutoApprovalMigrationRecord:
repository_id: int
repository_name: str
repository_url: str
repository_status: str
analysis_run_id: int | None
analysis_run_status: str
scanner_version: str
review_decision_id: int
decision_created_at: str
notes: str
current_approved_ability_count: int
recommended_next_step: str
def enrich_review_decision(decision: ReviewDecision) -> ReviewDecision:
fields = review_decision_audit_fields(decision.action, decision.notes)
return replace_review_decision(decision, **fields)

View File

@@ -40,6 +40,7 @@ from repo_registry.core.models import (
ReviewDecision,
ScanSummary,
SearchResult,
TrustedAutoApprovalMigrationRecord,
enrich_review_decision,
)
from repo_registry.candidate_graph.generator import CandidateGraphGenerator
@@ -52,7 +53,7 @@ from repo_registry.repo_ingestion.git import GitIngestionService
from repo_registry.repo_ingestion.metadata import RepositoryMetadataExtractor
from repo_registry.repo_scanning.scanner import DeterministicScanner
from repo_registry.semantic import EmbeddingProvider, cosine_similarity
from repo_registry.storage.sqlite import RegistryStore
from repo_registry.storage.sqlite import NotFoundError, RegistryStore
class RegistryService:
@@ -426,6 +427,54 @@ class RegistryService:
)
]
def list_trusted_auto_approval_migration_records(
self,
) -> list[TrustedAutoApprovalMigrationRecord]:
records: list[TrustedAutoApprovalMigrationRecord] = []
for repository in self.list_repositories():
approved_ability_count = len(self.ability_map(repository.id).abilities)
for decision in self.list_review_decisions(repository.id):
if decision.action != "trusted_auto_approve_candidate_graph":
continue
run_status = "unknown"
scanner_version = ""
if decision.analysis_run_id is not None:
try:
run = self.get_analysis_run(
repository.id,
decision.analysis_run_id,
)
except NotFoundError:
run_status = "missing"
else:
run_status = run.status
scanner_version = run.scanner_version
records.append(
TrustedAutoApprovalMigrationRecord(
repository_id=repository.id,
repository_name=repository.name,
repository_url=repository.url,
repository_status=repository.status,
analysis_run_id=decision.analysis_run_id,
analysis_run_status=run_status,
scanner_version=scanner_version,
review_decision_id=decision.id,
decision_created_at=decision.created_at,
notes=decision.notes,
current_approved_ability_count=approved_ability_count,
recommended_next_step=(
"Dry-run rebuild-characteristics, then confirm a "
"rebuild with manual or agentic review before "
"publishing replacement registry truth."
),
)
)
return sorted(
records,
key=lambda record: (record.decision_created_at, record.review_decision_id),
reverse=True,
)
def record_expectation_gap(
self,
repository_id: int,
@@ -626,7 +675,15 @@ class RegistryService:
analysis_run_id: int,
*,
notes: str = "",
allow_deprecated_migration_mode: bool = False,
) -> RepositoryAbilityMap:
if not allow_deprecated_migration_mode:
raise ValueError(
"trusted deterministic auto-approval is deprecated. Use "
"request_agentic_review() or approve_candidate_graph() after "
"human review; pass allow_deprecated_migration_mode=True only "
"when replaying historical migration behavior."
)
graph = self.store.get_candidate_graph(repository_id, analysis_run_id)
approved_count = 0
skipped_count = 0

View File

@@ -75,6 +75,7 @@ from repo_registry.web_api.schemas import (
ReviewDecisionResponse,
ScanSummaryResponse,
SearchResultResponse,
TrustedAutoApprovalMigrationRecordResponse,
)
@@ -208,6 +209,20 @@ def list_quality_criteria() -> dict[str, object]:
return criteria_registry_dict(load_quality_criteria())
@app.get(
"/review/migrations/trusted-auto-approvals",
tags=["review"],
response_model=list[TrustedAutoApprovalMigrationRecordResponse],
)
def list_trusted_auto_approval_migration_records(
service: RegistryService = Depends(get_service),
) -> list[dict[str, object]]:
return [
asdict(record)
for record in service.list_trusted_auto_approval_migration_records()
]
@app.post(
"/repos",
status_code=201,

View File

@@ -524,6 +524,21 @@ class ReviewDecisionResponse(BaseModel):
decision_kind: str = "other"
class TrustedAutoApprovalMigrationRecordResponse(BaseModel):
repository_id: int
repository_name: str
repository_url: str
repository_status: str
analysis_run_id: int | None
analysis_run_status: str
scanner_version: str
review_decision_id: int
decision_created_at: str
notes: str
current_approved_ability_count: int
recommended_next_step: str
class QualityCriterionResponse(BaseModel):
id: str
title: str