from __future__ import annotations import argparse import csv from datetime import date, datetime, timezone import hashlib import json import os from pathlib import Path import re import subprocess import sys import time from typing import Any, Callable, Iterable, Mapping, Sequence BATCH_SCHEMA = "HIBOR_EXACT_BATCH_V001" TASK_SCHEMA = "HIBOR_FAST_TASK_SPEC_V004" MAX_ITEMS = 10 MAX_GENERATED_PATH_CHARS = 220 DEFAULT_ITEM_TIMEOUT_SECONDS = 660 ADB_PREFLIGHT_TIMEOUT_SECONDS = 15 AUTO_DEVICE_SERIAL = "auto" PACKAGE_NAME = "cn.com.hibor" CACHE_ROOT = "/sdcard/Android/data/cn.com.hibor/files/myfile/" SHORT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,23}$") SOURCE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,119}$") MANIFEST_FIELDS = ( "batch_id", "item_index", "item_id", "source_id", "query", "title", "publisher", "publication_date", "pages", "pages_assumed", "raw_destination", "status", "kernel_exit_code", "kernel_status", "kernel_stop_code", "elapsed_ms", "task_path", "stdout_path", "stderr_path", "kernel_manifest_path", "kernel_delivery_path", "kernel_timing_path", "kernel_terminal_path", "kernel_terminal_present", "error", ) class BatchInputError(ValueError): pass Runner = Callable[ [Sequence[str], Path, Mapping[str, str], int], subprocess.CompletedProcess[bytes] ] def _utc_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _json_bytes(value: Any, *, compact: bool = False) -> bytes: if compact: text = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=False) else: text = json.dumps(value, ensure_ascii=False, indent=2, sort_keys=False) return (text + "\n").encode("utf-8") def _write_create_new(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0) try: descriptor = os.open(path, flags, 0o600) except FileExistsError as exc: raise BatchInputError(f"refusing to overwrite existing output: {path}") from exc try: view = memoryview(data) while view: written = os.write(descriptor, view) if written <= 0: raise OSError(f"short write: {path}") view = view[written:] os.fsync(descriptor) finally: os.close(descriptor) def _read_json_file(path: Path) -> Mapping[str, Any]: try: raw = path.read_bytes() except OSError as exc: raise BatchInputError(f"cannot read batch JSON: {path}") from exc if raw.startswith(b"\xef\xbb\xbf") or b"\x00" in raw: raise BatchInputError("batch JSON must be UTF-8 without BOM or NUL") try: value = json.loads(raw.decode("utf-8")) except (UnicodeError, json.JSONDecodeError) as exc: raise BatchInputError("batch JSON is not valid UTF-8 JSON") from exc if not isinstance(value, Mapping): raise BatchInputError("batch JSON root must be an object") return value def _required_text(value: Mapping[str, Any], field: str, context: str) -> str: result = value.get(field) if not isinstance(result, str) or not result.strip(): raise BatchInputError(f"{context}.{field} must be a non-empty string") return result.strip() def _resolve_under(project_root: Path, raw: str, allowed_root: Path, field: str) -> Path: candidate = Path(raw) if not candidate.is_absolute(): candidate = project_root / candidate resolved = candidate.resolve(strict=False) allowed = allowed_root.resolve(strict=False) try: resolved.relative_to(allowed) except ValueError as exc: raise BatchInputError(f"{field} must stay under {allowed}") from exc return resolved def validate_batch(value: Mapping[str, Any], project_root: Path) -> dict[str, Any]: expected_root_keys = ( "schema_version", "batch_id", "source_thread_id", "reply_thread_id", "runtime", "items", ) if set(value.keys()) != set(expected_root_keys): raise BatchInputError(f"batch root keys must be exactly {expected_root_keys}") if value["schema_version"] != BATCH_SCHEMA: raise BatchInputError(f"schema_version must be {BATCH_SCHEMA}") batch_id = _required_text(value, "batch_id", "batch") if not SHORT_ID_RE.fullmatch(batch_id): raise BatchInputError("batch_id must be 1..24 safe ASCII characters") source_thread_id = _required_text(value, "source_thread_id", "batch") reply_thread_id = _required_text(value, "reply_thread_id", "batch") runtime = value.get("runtime") runtime_keys = ( "quota_ledger", "adb_executable", "pdfinfo_executable", "device_serial", "source_scope", ) if not isinstance(runtime, Mapping) or set(runtime.keys()) != set(runtime_keys): raise BatchInputError(f"runtime keys must be exactly {runtime_keys}") quota_ledger = _resolve_under( project_root, _required_text(runtime, "quota_ledger", "runtime"), project_root / "ana-data" / "tmp", "runtime.quota_ledger", ) adb_executable = _required_text(runtime, "adb_executable", "runtime") pdfinfo_executable = _required_text(runtime, "pdfinfo_executable", "runtime") device_serial = _required_text(runtime, "device_serial", "runtime") source_scope = runtime.get("source_scope") if not isinstance(source_scope, Mapping): raise BatchInputError("runtime.source_scope must be an object") items = value.get("items") if not isinstance(items, list) or not 1 <= len(items) <= MAX_ITEMS: raise BatchInputError(f"items must contain 1..{MAX_ITEMS} reports") normalized_items: list[dict[str, Any]] = [] item_ids: set[str] = set() for index, item in enumerate(items, 1): context = f"items[{index - 1}]" if not isinstance(item, Mapping): raise BatchInputError(f"{context} must be an object") keys = set(item) required = { "item_id", "query", "title", "publisher", "publication_date", "source_id", "raw_destination", } if not required.issubset(keys) or keys.difference(required | {"pages"}): missing = sorted(required.difference(keys)) extra = sorted(keys.difference(required | {"pages"})) raise BatchInputError(f"{context} missing={missing} extra={extra}") item_id = _required_text(item, "item_id", context) if not SHORT_ID_RE.fullmatch(item_id) or item_id in item_ids: raise BatchInputError(f"{context}.item_id must be unique safe ASCII, 1..24 chars") item_ids.add(item_id) source_id = _required_text(item, "source_id", context) if not SOURCE_ID_RE.fullmatch(source_id): raise BatchInputError(f"{context}.source_id must be safe ASCII, 1..120 chars") publication_date = _required_text(item, "publication_date", context) try: if date.fromisoformat(publication_date).isoformat() != publication_date: raise ValueError except ValueError as exc: raise BatchInputError(f"{context}.publication_date must be YYYY-MM-DD") from exc pages = item.get("pages") if pages is not None and (type(pages) is not int or pages < 1): raise BatchInputError(f"{context}.pages must be a positive integer when supplied") raw_destination = _resolve_under( project_root, _required_text(item, "raw_destination", context), project_root / "ana-data" / "cases", f"{context}.raw_destination", ) normalized_items.append({ "item_id": item_id, "query": _required_text(item, "query", context), "title": _required_text(item, "title", context), "publisher": _required_text(item, "publisher", context), "publication_date": publication_date, "pages": pages, "source_id": source_id, "raw_destination": raw_destination, }) return { "batch_id": batch_id, "source_thread_id": source_thread_id, "reply_thread_id": reply_thread_id, "runtime": { "quota_ledger": quota_ledger, "adb_executable": adb_executable, "pdfinfo_executable": pdfinfo_executable, "device_serial": device_serial, "source_scope": dict(source_scope), }, "items": normalized_items, } def _report_identity(item: Mapping[str, Any]) -> str: text = "|".join((item["title"], item["publisher"], item["publication_date"])) return hashlib.sha256(text.encode("utf-8")).hexdigest() def build_task_spec( batch: Mapping[str, Any], item: Mapping[str, Any], item_index: int, item_output_root: Path, ) -> dict[str, Any]: batch_id = batch["batch_id"] task_id = f"HB-{batch_id}-{item_index:02d}" pages = item["pages"] if item["pages"] is not None else 1 runtime = batch["runtime"] observed_at = _utc_now() return { "schema_version": TASK_SCHEMA, "contract_version": "REPORT-COLLECTION-CAPABILITY-V1", "project_id": "project-info", "handoff_id": f"HIBOR-BATCH-{batch_id}-{item_index:02d}", "task_id": task_id, "source_role_instance_id": "case_analysis.report_collector", "source_thread_id": batch["source_thread_id"], "target_role_instance_id": "case_analysis.report_collector", "target_thread_id": batch["source_thread_id"], "reply_thread_id": batch["reply_thread_id"], "requester": "case_analysis.report_collector", "review_owner": "case_analysis.report_collector", "mode": "collect-one", "query": item["query"], "quantity": 1, "aliases": [item["title"]], "analysts": [], "institutions": [item["publisher"]], "report_types": [], "date_range": {"start": item["publication_date"], "end": item["publication_date"]}, "minimum_pages": 1, "exclude": [], "source_scope": runtime["source_scope"], "destination": str(item["raw_destination"]), "priority": "NORMAL", "naming_requirement": "standard", "output_root": str(item_output_root), "quota_ledger": str(runtime["quota_ledger"]), "adb_executable": runtime["adb_executable"], "pdfinfo_executable": runtime["pdfinfo_executable"], "package_name": PACKAGE_NAME, "cache_root": CACHE_ROOT, "device_serial": runtime["device_serial"], "observed_at_utc": observed_at, "total_budget_ms": 600_000, "close_reserve_ms": 30_000, "batch_increment_budget_ms": 240_000, "min_screens": 1, "normal_max_screens": 1, "hard_max_screens": 20, "hard_max_candidates": 100, "performance_slot_id": item["item_id"], "performance_plan_id": batch_id, "expected_reports": [{ "report_identity": _report_identity(item), "title": item["title"], "institution": item["publisher"], "report_date": item["publication_date"], "page_count": pages, }], } def _predicted_run_root(task_spec: Mapping[str, Any]) -> Path: canonical = json.dumps( task_spec, ensure_ascii=False, separators=(",", ":"), sort_keys=False ).encode("utf-8") token = hashlib.sha256(b"HIBOR-RUN-V001\x00" + canonical).hexdigest()[:24] task = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_spec["task_id"]))[:80] return Path(task_spec["output_root"]) / f"RUN-{task}-{token}" def _planned_paths(batch_root: Path, task_specs: Sequence[Mapping[str, Any]]) -> list[Path]: paths = [ batch_root / "batch_manifest.csv", batch_root / "batch_delivery.md", batch_root / "batch_timing.json", batch_root / "batch_terminal.json", ] for index, task in enumerate(task_specs, 1): paths.extend(( batch_root / "t" / f"{index:02d}.json", batch_root / "l" / f"{index:02d}.stdout.log", batch_root / "l" / f"{index:02d}.stderr.log", _predicted_run_root(task) / "report_collection_terminal.json", )) return paths def _default_runner( command: Sequence[str], cwd: Path, env: Mapping[str, str], timeout_seconds: int, ) -> subprocess.CompletedProcess[bytes]: return subprocess.run( list(command), cwd=cwd, env=dict(env), stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, timeout=timeout_seconds, ) def _parse_adb_devices(stdout: bytes) -> dict[str, str]: try: text = stdout.decode("utf-8") except UnicodeError as exc: raise BatchInputError("adb devices output is not valid UTF-8") from exc lines = [line.strip() for line in text.splitlines()] try: header_index = lines.index("List of devices attached") except ValueError as exc: raise BatchInputError("adb devices output is missing its header") from exc devices: dict[str, str] = {} for line in lines[header_index + 1:]: if not line or line.startswith("*"): continue fields = line.split() if len(fields) < 2: raise BatchInputError(f"cannot parse adb device row: {line}") serial, state = fields[0], fields[1] if serial in devices: raise BatchInputError(f"adb returned duplicate device serial: {serial}") devices[serial] = state return devices def resolve_device_serial( adb_executable: str, requested_serial: str, *, cwd: Path, environment: Mapping[str, str], runner: Runner | None = None, timeout_seconds: int = ADB_PREFLIGHT_TIMEOUT_SECONDS, ) -> str: if timeout_seconds < 1: raise BatchInputError("ADB preflight timeout must be positive") runner = runner or _default_runner try: completed = runner( [adb_executable, "devices", "-l"], cwd, environment, timeout_seconds ) except subprocess.TimeoutExpired as exc: raise BatchInputError( f"ADB device preflight timed out after {timeout_seconds}s" ) from exc except OSError as exc: raise BatchInputError( f"cannot run ADB device preflight: {type(exc).__name__}" ) from exc if int(completed.returncode) != 0: stderr = _as_bytes(completed.stderr).decode("utf-8", errors="replace").strip() detail = stderr[-500:] if stderr else f"exit code {completed.returncode}" raise BatchInputError(f"ADB device preflight failed: {detail}") devices = _parse_adb_devices(_as_bytes(completed.stdout)) if requested_serial == AUTO_DEVICE_SERIAL: if len(devices) != 1: raise BatchInputError( "runtime.device_serial=auto requires exactly one connected ADB device; " f"found {len(devices)}" ) serial, state = next(iter(devices.items())) if state != "device": raise BatchInputError( f"the only connected ADB device is not ready: {serial} ({state})" ) return serial state = devices.get(requested_serial) if state is None: raise BatchInputError( f"configured ADB device is not connected: {requested_serial}" ) if state != "device": raise BatchInputError( f"configured ADB device is not ready: {requested_serial} ({state})" ) return requested_serial def _as_bytes(value: bytes | str | None) -> bytes: if value is None: return b"" return value if isinstance(value, bytes) else value.encode("utf-8", errors="replace") def _parse_terminal(stdout: bytes) -> tuple[dict[str, Any] | None, str | None]: try: value = json.loads(stdout.decode("utf-8").strip()) except (UnicodeError, json.JSONDecodeError) as exc: return None, f"invalid kernel stdout JSON: {type(exc).__name__}" if not isinstance(value, dict): return None, "kernel stdout JSON is not an object" return value, None def _stage_group(stage: str) -> str: if re.fullmatch(r"item_\d+_detail", stage): return "detail" if re.fullmatch(r"item_\d+_cache", stage): return "cache" if re.fullmatch(r"item_\d+_copy", stage): return "copy" if re.fullmatch(r"item_\d+_validation_publish", stage): return "validation_publish" return stage def summarize_timing_files(paths: Iterable[Path]) -> dict[str, Any]: totals: dict[str, int] = {} files_read = 0 rows_read = 0 for path in paths: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): continue rows = value.get("rows") if isinstance(value, Mapping) else None if not isinstance(rows, list): continue files_read += 1 for row in rows: if not isinstance(row, Mapping): continue stage = row.get("stage") elapsed_ms = row.get("elapsed_ms") if not isinstance(stage, str) or type(elapsed_ms) is not int or elapsed_ms < 0: continue grouped = _stage_group(stage) totals[grouped] = totals.get(grouped, 0) + elapsed_ms rows_read += 1 return {"files_read": files_read, "rows_read": rows_read, "stage_totals_ms": totals} def _manifest_bytes(rows: Sequence[Mapping[str, Any]]) -> bytes: import io buffer = io.StringIO(newline="") writer = csv.DictWriter(buffer, fieldnames=MANIFEST_FIELDS, lineterminator="\n") writer.writeheader() for row in rows: writer.writerow({field: row.get(field, "") for field in MANIFEST_FIELDS}) return buffer.getvalue().encode("utf-8") def _delivery_text(batch_id: str, rows: Sequence[Mapping[str, Any]]) -> str: succeeded = sum(row["status"] == "SUCCESS" for row in rows) failed = len(rows) - succeeded lines = [ f"# 慧博精确研报批次 {batch_id}", "", f"- 总数：{len(rows)}", f"- 成功：{succeeded}", f"- 失败：{failed}", "", "| item_id | source_id | 状态 | 正式目录 | 错误 |", "|---|---|---|---|---|", ] for row in rows: error = str(row.get("error") or "").replace("|", "\\|").replace("\n", " ") lines.append( f"| {row['item_id']} | {row['source_id']} | {row['status']} | " f"{row['raw_destination']} | {error} |" ) return "\n".join(lines) + "\n" def _path_or_none(value: Any) -> Path | None: return Path(value) if isinstance(value, str) and value else None def execute_batch( batch_file: Path, *, project_root: Path | None = None, kernel_root: Path | None = None, python_executable: str | None = None, kernel_entry: Path | None = None, runner: Runner | None = None, device_runner: Runner | None = None, item_timeout_seconds: int = DEFAULT_ITEM_TIMEOUT_SECONDS, ) -> dict[str, Any]: if item_timeout_seconds < 1: raise BatchInputError("item timeout must be positive") project_root = (project_root or Path(__file__).resolve().parents[2]).resolve() kernel_root = (kernel_root or project_root / "dev" / "ana-dev").resolve() python_executable = python_executable or sys.executable runner = runner or _default_runner batch = validate_batch(_read_json_file(batch_file.resolve()), project_root) batch_root = project_root / "ana-data" / "tmp" / "hibor-runs" / batch["batch_id"] if batch_root.exists() or batch_root.is_symlink(): raise BatchInputError(f"batch output already exists; choose a new batch_id: {batch_root}") environment = dict(os.environ) environment["PYTHONUTF8"] = "1" environment["PYTHONIOENCODING"] = "utf-8" batch["runtime"]["device_serial"] = resolve_device_serial( batch["runtime"]["adb_executable"], batch["runtime"]["device_serial"], cwd=project_root, environment=environment, runner=device_runner, ) task_specs = [ build_task_spec(batch, item, index, batch_root / "r" / f"{index:02d}") for index, item in enumerate(batch["items"], 1) ] planned = _planned_paths(batch_root, task_specs) longest_planned = max(planned, key=lambda path: len(str(path.resolve(strict=False)))) if len(str(longest_planned.resolve(strict=False))) > MAX_GENERATED_PATH_CHARS: raise BatchInputError( f"generated evidence path would exceed {MAX_GENERATED_PATH_CHARS} chars: " f"{longest_planned}" ) batch_root.mkdir(parents=True, exist_ok=False) started_at = _utc_now() started_ns = time.monotonic_ns() rows: list[dict[str, Any]] = [] item_timings: list[dict[str, Any]] = [] for index, (item, task_spec) in enumerate(zip(batch["items"], task_specs), 1): task_path = batch_root / "t" / f"{index:02d}.json" stdout_path = batch_root / "l" / f"{index:02d}.stdout.log" stderr_path = batch_root / "l" / f"{index:02d}.stderr.log" _write_create_new(task_path, _json_bytes(task_spec, compact=True)) if kernel_entry is None: command = [ python_executable, "-m", "hibor_fast_collection", "--task", str(task_path), "--execute", ] else: command = [ python_executable, str(kernel_entry), "--task", str(task_path), "--execute", ] item_started = time.monotonic_ns() completed: subprocess.CompletedProcess[bytes] | None = None run_error: str | None = None try: completed = runner(command, kernel_root, environment, item_timeout_seconds) stdout = _as_bytes(completed.stdout) stderr = _as_bytes(completed.stderr) kernel_exit_code = int(completed.returncode) except subprocess.TimeoutExpired as exc: stdout = _as_bytes(exc.stdout) stderr = _as_bytes(exc.stderr) kernel_exit_code = 124 run_error = f"kernel timeout after {item_timeout_seconds}s" except OSError as exc: stdout = b"" stderr = str(exc).encode("utf-8", errors="replace") kernel_exit_code = 127 run_error = f"kernel process error: {type(exc).__name__}" elapsed_ms = max(0, (time.monotonic_ns() - item_started) // 1_000_000) _write_create_new(stdout_path, stdout) _write_create_new(stderr_path, stderr) terminal, parse_error = _parse_terminal(stdout) if stdout.strip() else (None, "empty kernel stdout") if run_error is None: run_error = parse_error terminal = terminal or {} kernel_terminal_path = _path_or_none(terminal.get("terminal_path")) kernel_terminal_present = bool( kernel_terminal_path is not None and kernel_terminal_path.is_file() and terminal.get("terminal_present") is True ) kernel_status = terminal.get("status") if isinstance(terminal.get("status"), str) else None success = kernel_exit_code == 0 and kernel_status == "SUCCESS" and kernel_terminal_present if not success and run_error is None: details = terminal.get("stop_code") or terminal.get("blocker") if not details and stderr: details = stderr.decode("utf-8", errors="replace").strip()[-500:] run_error = str(details or "kernel did not return a persisted SUCCESS terminal") kernel_timing_path = _path_or_none(terminal.get("timing_path")) timing_summary = summarize_timing_files( [kernel_timing_path] if kernel_timing_path is not None else [] ) item_timings.append({ "item_index": index, "item_id": item["item_id"], "cli_elapsed_ms": elapsed_ms, "kernel_timing_path": str(kernel_timing_path) if kernel_timing_path else None, **timing_summary, }) rows.append({ "batch_id": batch["batch_id"], "item_index": index, "item_id": item["item_id"], "source_id": item["source_id"], "query": item["query"], "title": item["title"], "publisher": item["publisher"], "publication_date": item["publication_date"], "pages": item["pages"] if item["pages"] is not None else "", "pages_assumed": str(item["pages"] is None).lower(), "raw_destination": str(item["raw_destination"]), "status": "SUCCESS" if success else "FAILED", "kernel_exit_code": kernel_exit_code, "kernel_status": kernel_status or "", "kernel_stop_code": terminal.get("stop_code") or "", "elapsed_ms": elapsed_ms, "task_path": str(task_path), "stdout_path": str(stdout_path), "stderr_path": str(stderr_path), "kernel_manifest_path": terminal.get("manifest_path") or "", "kernel_delivery_path": terminal.get("delivery_path") or "", "kernel_timing_path": terminal.get("timing_path") or "", "kernel_terminal_path": str(kernel_terminal_path) if kernel_terminal_path else "", "kernel_terminal_present": str(kernel_terminal_present).lower(), "error": "" if success else run_error, }) print( f"[{index}/{len(task_specs)}] {item['item_id']}: " f"{'SUCCESS' if success else 'FAILED'} ({elapsed_ms / 1000:.3f}s)", file=sys.stderr, flush=True, ) succeeded = sum(row["status"] == "SUCCESS" for row in rows) failed = len(rows) - succeeded status = "SUCCESS" if failed == 0 else "PARTIAL_SUCCESS" if succeeded else "FAILED" exit_code = 0 if failed == 0 else 2 if succeeded else 3 manifest_path = batch_root / "batch_manifest.csv" delivery_path = batch_root / "batch_delivery.md" timing_path = batch_root / "batch_timing.json" terminal_path = batch_root / "batch_terminal.json" _write_create_new(manifest_path, _manifest_bytes(rows)) _write_create_new(delivery_path, _delivery_text(batch["batch_id"], rows).encode("utf-8")) stage_totals: dict[str, int] = {} for item_timing in item_timings: for stage, milliseconds in item_timing["stage_totals_ms"].items(): stage_totals[stage] = stage_totals.get(stage, 0) + int(milliseconds) ended_at = _utc_now() total_elapsed_ms = max(0, (time.monotonic_ns() - started_ns) // 1_000_000) timing_value = { "schema_version": "HIBOR_EXACT_BATCH_TIMING_V001", "batch_id": batch["batch_id"], "started_at_utc": started_at, "ended_at_utc": ended_at, "total_elapsed_ms": total_elapsed_ms, "stage_totals_ms": stage_totals, "items": item_timings, } _write_create_new(timing_path, _json_bytes(timing_value)) generated_paths = [path for path in batch_root.rglob("*") if path.is_file()] generated_paths.append(terminal_path) max_generated_path_chars = max(len(str(path.resolve(strict=False))) for path in generated_paths) if max_generated_path_chars > MAX_GENERATED_PATH_CHARS: raise BatchInputError( f"generated path exceeded {MAX_GENERATED_PATH_CHARS} chars after execution" ) terminal_value = { "schema_version": "HIBOR_EXACT_BATCH_TERMINAL_V001", "batch_id": batch["batch_id"], "status": status, "exit_code": exit_code, "started_at_utc": started_at, "ended_at_utc": ended_at, "total_elapsed_ms": total_elapsed_ms, "requested": len(rows), "succeeded": succeeded, "failed": failed, "manifest_path": str(manifest_path), "delivery_path": str(delivery_path), "timing_path": str(timing_path), "terminal_path": str(terminal_path), "terminal_present": True, "max_generated_path_chars": max_generated_path_chars, "path_limit_chars": MAX_GENERATED_PATH_CHARS, "items": [{ "item_id": row["item_id"], "source_id": row["source_id"], "status": row["status"], "kernel_exit_code": row["kernel_exit_code"], "kernel_terminal_path": row["kernel_terminal_path"] or None, "error": row["error"] or None, } for row in rows], } _write_create_new(terminal_path, _json_bytes(terminal_value)) return terminal_value def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Run 1..10 preselected exact Hibor reports through the existing CLI." ) parser.add_argument("batch", type=Path, help="UTF-8 HIBOR_EXACT_BATCH_V001 JSON") parser.add_argument("--project-root", type=Path, default=None) parser.add_argument("--kernel-root", type=Path, default=None) parser.add_argument("--kernel-python", default=sys.executable) parser.add_argument("--kernel-entry", type=Path, default=None, help=argparse.SUPPRESS) parser.add_argument( "--item-timeout-seconds", type=int, default=DEFAULT_ITEM_TIMEOUT_SECONDS ) return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: terminal = execute_batch( args.batch, project_root=args.project_root, kernel_root=args.kernel_root, python_executable=args.kernel_python, kernel_entry=args.kernel_entry, item_timeout_seconds=args.item_timeout_seconds, ) except (BatchInputError, OSError) as exc: print(f"HIBOR_BATCH_INPUT_ERROR:{exc}", file=sys.stderr) return 12 sys.stdout.buffer.write(_json_bytes(terminal, compact=True)) return int(terminal["exit_code"]) if __name__ == "__main__": raise SystemExit(main())