Use split background reports in promotion gate
GeoIntel CI / docs-smoke (push) Has been cancelled
GeoIntel CI / contract-smoke (push) Has been cancelled

This commit is contained in:
Codex
2026-07-10 02:44:11 +02:00
parent 251aa7b044
commit dea65666ec
7 changed files with 360 additions and 3 deletions
@@ -18,6 +18,10 @@ from statistics import mean
from typing import Any
PURE_EMPTY_CATEGORY = "pure_empty_negative"
SPARSE_CONTEXT_CATEGORY = "sparse_building_context"
@dataclass(frozen=True)
class CandidateKey:
model_asset_id: str
@@ -39,9 +43,18 @@ def parse_args() -> argparse.Namespace:
parser.add_argument(
"--hard-negative-summary",
action="append",
required=True,
default=[],
help="Path to hard_negative_matrix_summary.json. May be supplied multiple times.",
)
parser.add_argument(
"--background-split-summary",
action="append",
default=[],
help=(
"Path to background_corpus_split_summary.json. The pure-empty source summary is used as the "
"strict default gate; sparse-context evidence is kept as review-only context."
),
)
parser.add_argument("--output-dir", required=True, help="Directory for JSON and Markdown report output.")
parser.add_argument("--min-positive-samples", type=int, default=3)
parser.add_argument("--min-background-samples", type=int, default=3)
@@ -59,7 +72,10 @@ def parse_args() -> argparse.Namespace:
default=None,
help="Tile overlap to use for positive portfolio runs that do not record tile_overlap.",
)
return parser.parse_args()
args = parser.parse_args()
if not args.hard_negative_summary and not args.background_split_summary:
parser.error("at least one --hard-negative-summary or --background-split-summary is required")
return args
def read_json(path: Path) -> dict[str, Any]:
@@ -168,6 +184,65 @@ def collect_background_runs(summary_paths: list[Path]) -> dict[CandidateKey, lis
return grouped
def resolve_split_source_path(source_path: Any, split_summary_path: Path) -> Path:
path = Path(str(source_path))
if path.is_file():
return path
if not path.is_absolute():
candidate = split_summary_path.parent / path
if candidate.is_file():
return candidate
return path
def collect_split_background_inputs(
split_summary_paths: list[Path],
) -> tuple[list[Path], list[dict[str, Any]]]:
strict_background_paths: list[Path] = []
context_reviews: list[dict[str, Any]] = []
for split_summary_path in split_summary_paths:
payload = read_json(split_summary_path)
source_summaries = payload.get("source_summaries") or {}
pure_empty_source = source_summaries.get(PURE_EMPTY_CATEGORY)
if not pure_empty_source:
raise SystemExit(f"Split summary has no {PURE_EMPTY_CATEGORY} source: {split_summary_path}")
strict_block = payload.get("strict_default_gate") or {}
strict_category = strict_block.get("category")
if strict_category and strict_category != PURE_EMPTY_CATEGORY:
raise SystemExit(
f"Split summary strict gate must be {PURE_EMPTY_CATEGORY}; found {strict_category}: {split_summary_path}"
)
strict_background_paths.append(resolve_split_source_path(pure_empty_source, split_summary_path))
context_block = payload.get("context_review") or {}
sparse_context_source = source_summaries.get(SPARSE_CONTEXT_CATEGORY)
if context_block or sparse_context_source:
context_entry = dict(context_block)
context_entry["source_split_summary_path"] = str(split_summary_path)
if sparse_context_source:
context_entry["source_summary_path"] = str(
resolve_split_source_path(sparse_context_source, split_summary_path)
)
context_reviews.append(context_entry)
return strict_background_paths, context_reviews
def unique_paths(paths: list[Path]) -> list[Path]:
seen: set[str] = set()
unique: list[Path] = []
for path in paths:
marker = str(path)
if marker in seen:
continue
seen.add(marker)
unique.append(path)
return unique
def average_metric(runs: list[dict[str, Any]], metric: str) -> float | None:
values = [numeric(run, metric) for run in runs]
values = [value for value in values if value is not None]
@@ -177,7 +252,10 @@ def average_metric(runs: list[dict[str, Any]], metric: str) -> float | None:
def build_decisions(args: argparse.Namespace) -> dict[str, Any]:
portfolio_path = Path(args.positive_portfolio)
positive_portfolio = read_json(portfolio_path)
background_paths = [Path(path) for path in args.hard_negative_summary]
direct_background_paths = [Path(path) for path in args.hard_negative_summary]
split_summary_paths = [Path(path) for path in args.background_split_summary]
split_background_paths, context_reviews = collect_split_background_inputs(split_summary_paths)
background_paths = unique_paths([*direct_background_paths, *split_background_paths])
positive = collect_positive_runs(
positive_portfolio,
default_tile_size=args.default_positive_tile_size,
@@ -249,6 +327,8 @@ def build_decisions(args: argparse.Namespace) -> dict[str, Any]:
"generated_at": datetime.now(timezone.utc).isoformat(),
"positive_portfolio_path": str(portfolio_path),
"hard_negative_summary_paths": [str(path) for path in background_paths],
"background_split_summary_paths": [str(path) for path in split_summary_paths],
"background_context_reviews": context_reviews,
"gates": {
"min_positive_samples": args.min_positive_samples,
"min_background_samples": args.min_background_samples,
@@ -268,6 +348,7 @@ def write_markdown(report: dict[str, Any], path: Path) -> None:
f"- Generated: {report['generated_at']}",
f"- Positive portfolio: `{report['positive_portfolio_path']}`",
f"- Hard-negative summaries: {len(report['hard_negative_summary_paths'])}",
f"- Background split summaries: {len(report['background_split_summary_paths'])}",
f"- Candidates: {report['candidate_count']}",
"",
"## Gates",
@@ -281,6 +362,25 @@ def write_markdown(report: dict[str, Any], path: Path) -> None:
lines.append(f"- Promote candidate for operator review: `{recommended['candidate_key']}`")
else:
lines.append("- No candidate passed all positive and hard-negative gates.")
if report.get("background_context_reviews"):
lines.extend(
[
"",
"## Sparse-context review evidence",
"",
(
f"- Sparse-context review evidence: {len(report['background_context_reviews'])} "
"(not used as a default-promotion gate)"
),
]
)
for item in report["background_context_reviews"]:
lines.append(
"- "
f"`{item.get('category', SPARSE_CONTEXT_CATEGORY)}` from "
f"`{item.get('source_summary_path', 'unknown')}`; "
f"max detections `{item.get('max_detection_count')}`"
)
lines.extend(["", "## Candidate Decisions", ""])
for item in report["candidate_decisions"]:
reasons = ", ".join(item["rejection_reasons"]) or "none"