finish three custodian workplans from live hub evidence
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Python Tests / pytest (push) Successful in 21s

Close CUST-WP-0064 after the 2026-08-24 unassisted fire ingested
clay-borg, close CUST-WP-0065 now that all 120 active repos project a
classification, and close ADHOC-2026-08-25. Mark CUST-WP-0067 T02/T10
done (reverse relays already gone; work-record recovery lives on 0068).
Park the later no-checkout SBOM regression as CUST-IN-0015. Teach the
classification gate to use this host's checkout path.
This commit is contained in:
codex 2026-08-28 20:27:05 +02:00
parent 349238cf8f
commit 93b8174abd
8 changed files with 143 additions and 21 deletions

View file

@ -10,6 +10,7 @@ from __future__ import annotations
import argparse
import json
import os
import socket
import sys
import urllib.error
import urllib.request
@ -50,17 +51,48 @@ def fetch_repositories(
return payload
def _source_path(record: dict[str, Any]) -> Path | None:
local_path = record.get("local_path")
if not isinstance(local_path, str) or not local_path.strip():
return None
return Path(local_path) / ".repo-classification.yaml"
def _checkout_candidates(record: dict[str, Any], *, hostname: str) -> list[Path]:
"""Prefer this host's recorded checkout over a `local_path` from another machine."""
seen: set[str] = set()
candidates: list[Path] = []
def add(raw: object) -> None:
if not isinstance(raw, str):
return
text = raw.strip()
if not text or text in seen:
return
seen.add(text)
candidates.append(Path(text))
host_paths = record.get("host_paths")
if isinstance(host_paths, dict):
short = hostname.split(".")[0]
add(host_paths.get(hostname))
add(host_paths.get(short))
for value in host_paths.values():
if isinstance(value, str) and Path(value).is_dir():
add(value)
add(record.get("local_path"))
return candidates
def _source_path(record: dict[str, Any], *, hostname: str) -> Path | None:
preferred: Path | None = None
for checkout in _checkout_candidates(record, hostname=hostname):
path = checkout / ".repo-classification.yaml"
if preferred is None:
preferred = path
if path.is_file():
return path
return preferred
def analyze_repositories(
records: list[dict[str, Any]],
*,
validate_sources: bool = True,
hostname: str | None = None,
) -> dict[str, Any]:
active = [record for record in records if record.get("status") == "active"]
classified = [record for record in active if record.get("category") is not None]
@ -74,10 +106,11 @@ def analyze_repositories(
slug_counts: dict[str, int] = {}
allowed = load_allowed() if validate_sources else None
current_host = hostname or socket.gethostname()
for record in active:
slug = str(record.get("slug") or "<missing-slug>")
slug_counts[slug] = slug_counts.get(slug, 0) + 1
path = _source_path(record)
path = _source_path(record, hostname=current_host)
source_present = path is not None and path.is_file()
if record.get("category") is None: