Apply the image decision tree to missing hall portraits
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Has been cancelled

A hall visual that stays on the general branch uses the Efficient model
from the 2026-10-08 intelligence-radar tree. render-hall writes only
portraits whose files are still absent.

Assistant: grok
Assistant-Session: 01a11ab6-1534-7d63-8606-662b92a80ed6
This commit is contained in:
tegwick 2026-10-08 16:44:43 +02:00
parent 7073d2543c
commit e1bab67bd1
10 changed files with 675 additions and 11 deletions

View file

@ -23,6 +23,7 @@ from llm_connect.exceptions import (
)
from llm_connect.factory import create_adapter
from llm_connect.fx import resolve_fx_rate
from llm_connect.hall_portraits import plan_missing_portraits, reveal_portrait
from llm_connect.images import (
ImageRenderRequest,
ImageSpendGate,
@ -182,6 +183,36 @@ def _build_parser() -> argparse.ArgumentParser:
image_render.add_argument("--json", action="store_true", help="Emit receipt JSON")
image_render.set_defaults(func=_image_render)
image_plan = image_commands.add_parser(
"plan",
help="Choose models for hall entries that have no portrait file yet",
)
image_plan.add_argument("--hall", type=Path, required=True, help="Hall of Helix checkout")
image_plan.add_argument(
"--tier",
default=None,
help="Override the general-branch tier (default efficient)",
)
image_plan.add_argument("--json", action="store_true", help="Emit the plan as JSON")
image_plan.set_defaults(func=_image_plan)
image_hall = image_commands.add_parser(
"render-hall",
help="Render missing hall portraits using the image decision tree",
)
image_hall.add_argument("--hall", type=Path, required=True, help="Hall of Helix checkout")
image_hall.add_argument("--tier", default=None, help="Override the general-branch tier")
image_hall.add_argument("--limit", type=int, default=None, help="Render at most this many missing portraits")
image_hall.add_argument("--dry-run", action="store_true", help="Print the plan and do not call a provider")
image_hall.add_argument("--provider", default="openrouter", help="Image provider (default openrouter)")
image_hall.add_argument("--max-batch-cost-usd", type=float, default=None, help="Override the batch USD cap")
image_hall.add_argument("--force", action="store_true", help="Replace an existing output file")
image_hall.add_argument("--ledger", type=Path, default=None, help="Usage ledger JSONL path")
image_hall.add_argument("--no-ledger", action="store_true", help="Do not append to the usage ledger")
image_hall.add_argument("--eur-per-usd", type=float, default=None, help="Override FX rate")
image_hall.add_argument("--json", action="store_true", help="Emit receipt JSON")
image_hall.set_defaults(func=_image_render_hall)
return parser
@ -235,6 +266,138 @@ def _image_render(args: argparse.Namespace) -> int:
return exit_code
def _image_plan(args: argparse.Namespace) -> int:
try:
jobs = plan_missing_portraits(args.hall, tier=args.tier)
except LLMError as exc:
print(str(exc), file=sys.stderr)
return 1
payload = {
"count": len(jobs),
"quote_usd": round(sum(job.selection.quote_usd for job in jobs), 4),
"max_batch_cost_usd": round(sum(job.selection.max_cost_usd for job in jobs), 4),
"by_model": _hall_model_counts(jobs),
"jobs": [job.to_dict() for job in jobs],
}
print(json.dumps(payload, indent=2 if args.json else None, sort_keys=True))
return 0
def _image_render_hall(args: argparse.Namespace) -> int:
if args.limit is not None and args.limit <= 0:
raise SystemExit("--limit must be positive")
try:
jobs = plan_missing_portraits(args.hall, tier=args.tier)
except LLMError as exc:
print(str(exc), file=sys.stderr)
return 1
if args.limit is not None:
jobs = jobs[: args.limit]
if not jobs:
print(json.dumps({"count": 0, "receipts": []}, sort_keys=True))
return 0
if args.dry_run:
return _image_plan(args) if args.limit is None else _print_hall_jobs(jobs, json_indent=args.json)
fx = resolve_fx_rate(args.eur_per_usd)
groups: dict[tuple[str, float, float], list[Any]] = {}
for job in jobs:
key = (job.selection.model, job.selection.quote_usd, job.selection.max_cost_usd)
groups.setdefault(key, []).append(job)
receipts: list[dict[str, Any]] = []
exit_code = 0
for (model, quote, cap), group in groups.items():
batch_cap = args.max_batch_cost_usd
if batch_cap is None:
batch_cap = round(cap * len(group), 6)
try:
client = create_image_client(args.provider, model=model)
gate = ImageSpendGate(
ImageSpendPolicy(
max_cost_usd=cap,
max_batch_cost_usd=batch_cap,
quote_usd=quote,
)
)
except LLMError as exc:
print(str(exc), file=sys.stderr)
exit_code = 1
break
for job in group:
try:
result = render_image(
client,
ImageRenderRequest(
prompt=job.prompt,
output_format="jpeg" if job.output_path.suffix.lower() in {".jpg", ".jpeg"} else "png",
output_path=job.output_path,
),
gate=gate,
fx=fx,
force=args.force,
)
except LLMBudgetExceededError as exc:
print(str(exc), file=sys.stderr)
exit_code = 2
break
except LLMError as exc:
print(str(exc), file=sys.stderr)
exit_code = 1
break
if not args.no_ledger:
ledger_path = args.ledger or default_usage_ledger_path()
UsageLedger(ledger_path).append(
usage_event_from_receipt(result.receipt, source="cli-hall")
)
record = {
"entry_id": job.entry_id,
"branch": job.selection.branch,
"tier": job.selection.tier,
"receipt": result.receipt.to_dict(),
}
receipt_dir = job.output_path.parent / "receipts"
receipt_dir.mkdir(parents=True, exist_ok=True)
receipt_path = receipt_dir / f"{job.output_path.name}.json"
receipt_path.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8")
updated = reveal_portrait(
job.entry_path.read_text(encoding="utf-8"),
job.relative_path,
job.alt,
)
job.entry_path.write_text(updated, encoding="utf-8")
receipts.append(record)
if not result.receipt.within_cap:
print("Image cost exceeded the configured cap; batch stopped", file=sys.stderr)
exit_code = 2
break
if exit_code:
break
indent = 2 if args.json else None
print(json.dumps({"count": len(receipts), "receipts": receipts}, indent=indent, sort_keys=True))
return exit_code
def _print_hall_jobs(jobs: list[Any], *, json_indent: bool) -> int:
payload = {
"count": len(jobs),
"quote_usd": round(sum(job.selection.quote_usd for job in jobs), 4),
"max_batch_cost_usd": round(sum(job.selection.max_cost_usd for job in jobs), 4),
"by_model": _hall_model_counts(jobs),
"jobs": [job.to_dict() for job in jobs],
}
print(json.dumps(payload, indent=2 if json_indent else None, sort_keys=True))
return 0
def _hall_model_counts(jobs: list[Any]) -> dict[str, int]:
counts: dict[str, int] = {}
for job in jobs:
counts[job.selection.model] = counts.get(job.selection.model, 0) + 1
return counts
def _image_jobs(args: argparse.Namespace) -> list[ImageRenderRequest]:
if bool(args.prompt) == bool(args.batch):
raise SystemExit("Pass exactly one of --prompt or --batch")