#!/usr/bin/env python3 """Apply the independently approved L0 acceptance sync for NEWENERGY BATCH-002. The script is deliberately scoped to the frozen BATCH-002 artifact set. It updates review/package state, recalculates hashes affected by those state-only changes, and creates acceptance receipts. It does not alter research facts, evidence wording, candidate selection, the accepted BATCH-001 package, or the industry-root current release. """ from __future__ import annotations import csv import hashlib import json import re from copy import deepcopy from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo TASK_ID = "TASK-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-20260806-001" CASE_ID = "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002" BATCH_ID = "BATCH-002" RUN_ID = "RUN-ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260806-002-BATCH-002-001" FINAL_AUDIT = "AUDIT-ANA-NEWENERGY-FOUR-TRACK-ATLAS-BATCH002-EXECUTION-OUTPUT-REPAIR004-REREVIEW-20260807-001" ACCEPTED = "ACCEPTED_BY_INDEPENDENT_REVIEW" FINAL_OUTPUT = "FINAL_ACCEPTED" FINAL_ARTIFACT = "FINAL_ACCEPTED_BY_INDEPENDENT_REVIEW" PARENT_STATUS = "PENDING_PROJECT_ADMIN_APPEND_ONLY_TERMINAL_SYNC" ROOT = Path(__file__).resolve().parents[2] INDUSTRY = ROOT / "ana-data/cases/新能源案例" CASE = INDUSTRY / CASE_ID RESULT = ROOT / "ana-data/result/新能源案例" / CASE_ID ARTIFACT_MANIFEST = INDUSTRY / "manifest/artifact_manifest_BATCH002.csv" OUTPUT_MANIFEST = CASE / "manifest/output_manifest.csv" EXECUTION_RECEIPT = CASE / "manifest/execution_validation_receipt.json" ACCEPTANCE_RECEIPT = CASE / "manifest/acceptance_validation_receipt.json" ACCEPTANCE_RECORD = RESULT / "acceptance_record.md" CURRENT_STABLE_MANIFEST = INDUSTRY / "manifest/current_output_manifest.csv" AUDITED_HASHES = { ROOT / "ana-data/tools/newenergy_batch002_build.py": "537C8826CE2F49DEFA188E1E308643E816E5D3B3C8E34A6BCB1B02883FB5D389", INDUSTRY / "manifest/legacy_to_v1_adapter_contract_BATCH002.csv": "57F232EB9FF61893FE6D6130684A0BED01E9D2B6B6E76719691A9E5523DAB954", INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv": "A332F0C8F11AF8DDA9E5D77D389EF34502D2ABDB9D4B3663F6CA36D8A2D98055", INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv": "A25D32490C7DEFA14024B1FA46C6DA3CDE323A5CDBEF36D3C84D1DF6ECA2C45D", INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv": "D656AB6C5B5AF4FE2632FAC8F54C6239C4894D5DA81D150588FE67A3B52DAFAE", ARTIFACT_MANIFEST: "A3D9933AFBB4D2066290ACD5679A9AACE986985148BB2668FA430E38871C2B25", OUTPUT_MANIFEST: "B074F15BFF21EB7B3872369B26A0E20E245782BB8C852BAA2D2A0994660174DE", CASE / "evidence/case_evidence_map.csv": "27666694FA5A38E29AB518C19A970E20F5D7CE091DCAD15C0EB29CFA9CC7EF70", EXECUTION_RECEIPT: "41D663B73702CFAAD0C22AF59514062484EFB22FA3AE59DB61CDBE11373F2299", } IMMUTABLE_B001 = { INDUSTRY / "ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001/manifest/output_manifest.csv": "D0DC978D558CA56C9D32B9DC0F76489EDB0F643DB1B482585EF2D96DEE3DF6C4", ROOT / "ana-data/result/新能源案例/ANA-NEWENERGY-FOUR-TRACK-ATLAS-20260805-001/acceptance_record.md": "448E67471C53E1E59011A1ADBB44D2ED0BDA0068084D843251E2CCCD4A407CFA", CURRENT_STABLE_MANIFEST: "A2BEAF2CD0D3955ACF9DE8C6769D4BC9A3A7A1094FDD28C8447B66FD09372EB7", } def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest().upper() def project_relative(path: Path) -> str: return path.resolve().relative_to(ROOT.resolve()).as_posix() def read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) return list(reader.fieldnames or []), list(reader) def write_csv(path: Path, headers: list[str], rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=headers, extrasaction="ignore", lineterminator="\n") writer.writeheader() writer.writerows(rows) def write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8", newline="\n") def accepted_at() -> str: if ACCEPTANCE_RECORD.exists(): match = re.search(r"^- `accepted_at=([^`]+)`$", ACCEPTANCE_RECORD.read_text(encoding="utf-8"), flags=re.M) if match: return match.group(1) return datetime.now(ZoneInfo("Asia/Shanghai")).replace(microsecond=0).isoformat() def audit_gate() -> None: report = (ROOT / "ana-doc/新能源案例/案例审计报告.md").read_text(encoding="utf-8") marker = f"## 2026-08-07 `{FINAL_AUDIT}`" start = report.find(marker) if start < 0: raise RuntimeError("final audit section missing") tail = report[start:] end = tail.find("\n## ", 1) section = tail if end < 0 else tail[:end] required = ( "总体结论：`PASS / REPAIR004 聚焦复审通过`", "允许分析员按项目规范把本批 outputs、canonical shard、manifest 和 result 入口", ) if any(token not in section for token in required): raise RuntimeError("final audit does not release the acceptance sync") def preflight() -> tuple[list[str], list[dict[str, str]]]: audit_gate() for path, expected in {**AUDITED_HASHES, **IMMUTABLE_B001}.items(): if not path.is_file() or sha256(path) != expected: raise RuntimeError(f"preflight hash drift: {project_relative(path)}") headers, artifacts = read_csv(ARTIFACT_MANIFEST) if len(artifacts) != 83 or len({row["relative_path"] for row in artifacts}) != 83: raise RuntimeError("audited artifact exact-set is not 83 unique paths") for row in artifacts: path = ROOT / row["relative_path"] if not path.is_file() or path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"].upper(): raise RuntimeError(f"audited artifact mismatch: {row['relative_path']}") stable_rows = read_csv(CURRENT_STABLE_MANIFEST)[1] if any("BATCH-002" in row.get("source_path", "") or CASE_ID in row.get("source_path", "") for row in stable_rows): raise RuntimeError("BATCH-002 already entered current industry release") return headers, artifacts def replace_required(path: Path, old: str, new: str) -> int: text = path.read_text(encoding="utf-8") count = text.count(old) if count: path.write_text(text.replace(old, new), encoding="utf-8", newline="\n") elif new not in text: raise RuntimeError(f"required marker missing: {project_relative(path)} :: {old}") return count def sync_markdown(artifacts: list[dict[str, str]]) -> dict[str, int]: md_paths = [ROOT / row["relative_path"] for row in artifacts if row["file_ext"] == ".md"] if len(md_paths) != 42: raise RuntimeError(f"audited markdown count changed: {len(md_paths)}/42") counts = {"markdown_files": len(md_paths), "review_markers": 0, "terminal_sentences": 0} for path in md_paths: counts["review_markers"] += replace_required( path, "- review_status: `DRAFT_FOR_REVIEW`", f"- review_status: `{ACCEPTED}`" ) terminal = { CASE / "outputs/summary.md": ( "- 输出状态仍为 `DRAFT_FOR_REVIEW`，等待独立执行/输出审核。", f"- 本批执行与输出已由独立审核 `{FINAL_AUDIT}` 以 `PASS / 0 / 0` 接受；接受不扩展研究边界或完整覆盖声明。", ), CASE / "outputs/新能源报告索引.md": ( "当前全部产物为 `DRAFT_FOR_REVIEW`，独立执行/输出审核通过前不得标记完成或对外交付。", f"当前全部产物已由独立审核 `{FINAL_AUDIT}` 接受；本入口保留逐批审计血缘，行业默认阅读入口仍由行业根原子 promotion 单独控制。", ), RESULT / "result_index.md": ( "当前状态：`DRAFT_FOR_REVIEW`；独立执行/输出审核通过前不构成正式交付。", f"当前状态：`{ACCEPTED}`；接受审计：`{FINAL_AUDIT}=PASS`。本逐批入口仅保留审计血缘，不替代行业根当前成果入口。", ), CASE / "manifest/batch_summary.md": ( "执行/输出独立审核通过前不得回写父级终态或对外交付。", f"执行/输出独立审核 `{FINAL_AUDIT}` 已通过；父级 append-only 终态由项目管理员同步，行业根发布仍走独立 candidate/promotion。", ), } for path, (old, new) in terminal.items(): counts["terminal_sentences"] += replace_required(path, old, new) package = CASE / "manifest/package_exact_set_receipt.md" counts["package_count"] = replace_required( package, "- expected_artifact_exact_set_count: `83`", "- expected_artifact_exact_set_count: `85`" ) counts["package_audit"] = replace_required( package, "exact-set 包括本批新行业 shard、新 raw/converted、supplement 查询回执、案例 outputs/manifest/evidence、result 入口和获批 replay tool；不包括空 img/tmp 目录，也不复制前批 artifacts。", f"exact-set 包括本批新行业 shard、新 raw/converted、supplement 查询回执、案例 outputs/manifest/evidence、result 入口、接受记录、接受验证回执和获批 replay tool；独立接受审计为 `{FINAL_AUDIT}=PASS`，不包括空 img/tmp 目录，也不复制前批 artifacts。", ) leftovers = [(project_relative(path), path.read_text(encoding="utf-8").count("DRAFT_FOR_REVIEW")) for path in md_paths] leftovers = [item for item in leftovers if item[1]] if leftovers: raise RuntimeError(f"draft markdown markers remain: {leftovers[:3]}") return counts def sync_json_review_status(value: Any) -> Any: if isinstance(value, dict): return { key: ACCEPTED if key == "review_status" and item == "DRAFT_FOR_REVIEW" else sync_json_review_status(item) for key, item in value.items() } if isinstance(value, list): return [sync_json_review_status(item) for item in value] return value def sync_structured_statuses(artifacts: list[dict[str, str]]) -> dict[str, int]: counts = {"csv_files": 0, "csv_rows": 0, "json_files": 0, "review_rows": 0, "sample_rows": 0} for artifact in artifacts: path = ROOT / artifact["relative_path"] if path == ARTIFACT_MANIFEST: continue if artifact["file_ext"] == ".csv": headers, rows = read_csv(path) changed = False for row in rows: if "review_status" in headers and row.get("review_status") == "DRAFT_FOR_REVIEW": row["review_status"] = ACCEPTED counts["review_rows"] += 1 changed = True if "sample_status" in headers and row.get("sample_status") == "READY_FOR_INDEPENDENT_REVIEW": row["sample_status"] = ACCEPTED counts["sample_rows"] += 1 changed = True if changed: write_csv(path, headers, rows) counts["csv_files"] += 1 counts["csv_rows"] += len(rows) elif artifact["file_ext"] == ".json": value = json.loads(path.read_text(encoding="utf-8")) updated = sync_json_review_status(value) if updated != value: write_json(path, updated) counts["json_files"] += 1 return counts def update_output_manifest() -> list[dict[str, str]]: headers, rows = read_csv(OUTPUT_MANIFEST) if len(rows) != 42: raise RuntimeError("output manifest is not 42 rows") applicable = 0 not_applicable = 0 for row in rows: row["review_status"] = ACCEPTED if row["applicability"] == "APPLICABLE_FILE_OUTPUT": output = ROOT / row["output_path"] if not output.is_file(): raise FileNotFoundError(output) row["file_size"] = str(output.stat().st_size) row["sha256"] = sha256(output) row["output_status"] = FINAL_OUTPUT applicable += 1 elif row["output_status"] == "NOT_APPLICABLE": not_applicable += 1 else: raise RuntimeError(f"unexpected output applicability/status: {row['output_item_id']}") if (applicable, not_applicable) != (40, 2): raise RuntimeError(f"output distribution changed: {applicable}/{not_applicable}") write_csv(OUTPUT_MANIFEST, headers, rows) return rows def update_projection_receipts() -> None: adapter = INDUSTRY / "manifest/legacy_to_v1_adapter_contract_BATCH002.csv" period = INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv" projection = INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv" headers, rows = read_csv(projection) for row in rows: shard = ROOT / row["batch002_shard_path"] row["batch002_shard_sha256"] = sha256(shard) row["adapter_contract_sha256"] = sha256(adapter) if row.get("period_projection_receipt_path") not in ("", "NOT_APPLICABLE"): row["period_projection_receipt_sha256"] = sha256(period) row["review_status"] = ACCEPTED write_csv(projection, headers, rows) def update_registry() -> None: path = INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv" headers, rows = read_csv(path) for row in rows: shard = ROOT / row["shard_path"] actual_rows = read_csv(shard)[1] row["row_count"] = str(len(actual_rows)) row["sha256"] = sha256(shard) row["review_status"] = ACCEPTED if row["batch_id"] == BATCH_ID: row["immutable"] = "YES_ACCEPTED_IMMUTABLE" write_csv(path, headers, rows) def update_execution_receipt(at: str) -> None: receipt = json.loads(EXECUTION_RECEIPT.read_text(encoding="utf-8")) receipt["outputs_status"] = FINAL_OUTPUT receipt["review_status"] = ACCEPTED receipt["final_audit"] = FINAL_AUDIT receipt["acceptance_transition"] = "L0_STATUS_AND_AFFECTED_HASH_SYNC_ONLY" receipt["accepted_at"] = at receipt["parent_terminal_status"] = PARENT_STATUS receipt["legacy_adapter_contract_sha256"] = sha256(INDUSTRY / "manifest/legacy_to_v1_adapter_contract_BATCH002.csv") receipt["legacy_period_projection_receipt_sha256"] = sha256(INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv") receipt["legacy_projection_validation_sha256"] = sha256(INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv") write_json(EXECUTION_RECEIPT, receipt) def write_acceptance_record(at: str, output_rows: list[dict[str, str]]) -> None: text = f"""# {CASE_ID} 正式接受记录 - `task_id={TASK_ID}` - `case_id={CASE_ID}` - `batch_id={BATCH_ID}` - `run_id={RUN_ID}` - `status={ACCEPTED}` - `accepted_at={at}` - `final_audit={FINAL_AUDIT}` - `audit_result=PASS/0/0` - `parent_status={PARENT_STATUS}` - `accepted_output_manifest=ana-data/cases/新能源案例/{CASE_ID}/manifest/output_manifest.csv` - `accepted_output_manifest_sha256={sha256(OUTPUT_MANIFEST)}` - `applicable_outputs={sum(row['output_status'] == FINAL_OUTPUT for row in output_rows)}` - `conditional_not_applicable_outputs={sum(row['output_status'] == 'NOT_APPLICABLE' for row in output_rows)}` ## 接受范围 本记录接受本 case/BATCH/run 的 16 个新能源子行业深化节点、32 份节点正文、三类顶层视图、9 条增量公司—赛道映射、六组 legacy-to-V1 只读投影、期间语义回执和 40 份实物输出。7 个证据不足节点继续保留 GAP，2 个条件逻辑数据集继续 `NOT_APPLICABLE`。 独立复审 `{FINAL_AUDIT}` 已给出 `PASS / blocker 0 / non-blocking 0`。审核后只执行状态与受影响 hash 的 L0 同步；没有改变来源、证据原文、候选选择、结论强度、BATCH-001 接受包或行业根当前 release。 ## 边界 本接受不放行下一批、完整覆盖、估值、行情、交易、收益、市场反向补漏、核电敏感内容或数据库写入。逐批 result 入口只保留审计血缘；若要进入行业默认阅读入口，必须另建行业根 candidate，并通过既有新能源审核与原子 promotion。 """ ACCEPTANCE_RECORD.parent.mkdir(parents=True, exist_ok=True) ACCEPTANCE_RECORD.write_text(text, encoding="utf-8", newline="\n") def write_acceptance_validation(at: str, output_rows: list[dict[str, str]], sync_counts: dict[str, Any]) -> None: validation = { "schema_version": "NEWENERGY_BATCH002_ACCEPTANCE_VALIDATION_V1", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "status": ACCEPTED, "accepted_at": at, "final_audit": FINAL_AUDIT, "transition_type": "L0_REVIEW_STATUS_AND_AFFECTED_HASH_SYNC_ONLY", "outputs": { "manifest_rows": len(output_rows), "final_accepted": sum(row["output_status"] == FINAL_OUTPUT for row in output_rows), "not_applicable": sum(row["output_status"] == "NOT_APPLICABLE" for row in output_rows), "manifest_sha256": sha256(OUTPUT_MANIFEST), }, "accepted_scope": {"subindustry_nodes": 16, "node_documents": 32, "incremental_company_track_mappings": 9, "gap_nodes": 7}, "sync_counts": sync_counts, "core_hashes": { "case_evidence_map": sha256(CASE / "evidence/case_evidence_map.csv"), "canonical_registry": sha256(INDUSTRY / "manifest/canonical_shard_registry_BATCH002.csv"), "projection_validation": sha256(INDUSTRY / "manifest/legacy_to_v1_projection_validation_BATCH002.csv"), "period_projection_receipt": sha256(INDUSTRY / "manifest/legacy_period_projection_receipt_BATCH002.csv"), "execution_validation": sha256(EXECUTION_RECEIPT), "acceptance_record": sha256(ACCEPTANCE_RECORD), }, "immutable_b001_hash_match": len(IMMUTABLE_B001), "industry_current_release_changed": False, "parent_terminal_status": PARENT_STATUS, "validation_status": "PASS", } write_json(ACCEPTANCE_RECEIPT, validation) def rebuild_artifact_manifest(headers: list[str], old_rows: list[dict[str, str]], at: str) -> list[dict[str, str]]: by_path = {row["relative_path"]: deepcopy(row) for row in old_rows} additions = [ACCEPTANCE_RECEIPT, ACCEPTANCE_RECORD] for index, path in enumerate(additions, start=84): relative = project_relative(path) by_path[relative] = { "artifact_id": f"NEB2-ART-{index:04d}", "task_id": TASK_ID, "case_id": CASE_ID, "batch_id": BATCH_ID, "run_id": RUN_ID, "artifact_type": "RECEIPT_OR_VALIDATION" if path.suffix == ".json" else "HUMAN_READABLE_OUTPUT_OR_RECEIPT", "industry_case": "新能源案例", "industry_id": "IND-NEWENERGY", "subindustry_id": "", "company_id": "", "logical_path": relative, "relative_path": relative, "absolute_path": str(path.resolve()), "file_name": path.name, "file_ext": path.suffix.lower(), "file_size": "", "sha256": "", "source_doc_id": "", "source_url": "", "source_collected_at": "", "raw_pool_path": "", "source_file_name": "", "detected_type": path.suffix.lower().lstrip("."), "archive_file_name": path.name, "extension_added_by_archive_flag": "NO", "extension_mismatch_flag": "NO", "created_at": at, "created_by": "case_analysis.analyst.new_energy", "tool_or_method": "newenergy_batch002_accept.py", "tool_version": "BATCH002-ACCEPT-V1", "parameters_summary": "independent PASS terminal L0 status/hash sync only", "source_snapshot_id": "BATCH002_ACCEPTANCE", "artifact_status": FINAL_ARTIFACT, "sensitivity_screen": "LEGAL_PUBLIC_CIVIL_NUCLEAR_HIGH_LEVEL_ONLY", "schema_version": "NEWENERGY_EXTENSION_V1", "review_status": ACCEPTED, } if len(by_path) != 85: raise RuntimeError(f"post-acceptance artifact exact-set is not 85: {len(by_path)}") rows: list[dict[str, str]] = [] for relative in sorted(by_path): row = by_path[relative] path = ROOT / relative if not path.is_file(): raise FileNotFoundError(path) row["file_size"] = str(path.stat().st_size) row["sha256"] = sha256(path) row["artifact_status"] = FINAL_ARTIFACT row["review_status"] = ACCEPTED if row.get("parameters_summary") == "exact released batch/run; immutable B001 references; DRAFT outputs": row["parameters_summary"] = "exact released batch/run; immutable B001 references; independently accepted outputs" rows.append(row) write_csv(ARTIFACT_MANIFEST, headers, rows) return rows def verify_accepted() -> dict[str, Any]: for path, expected in IMMUTABLE_B001.items(): if sha256(path) != expected: raise RuntimeError(f"immutable BATCH-001/current release drift: {project_relative(path)}") output_rows = read_csv(OUTPUT_MANIFEST)[1] if len(output_rows) != 42: raise RuntimeError("accepted output manifest row count mismatch") if sum(row["output_status"] == FINAL_OUTPUT for row in output_rows) != 40: raise RuntimeError("accepted applicable output count mismatch") if sum(row["output_status"] == "NOT_APPLICABLE" for row in output_rows) != 2: raise RuntimeError("accepted N/A output count mismatch") if any(row["review_status"] != ACCEPTED for row in output_rows): raise RuntimeError("output review status mismatch") for row in output_rows: if row["applicability"] == "APPLICABLE_FILE_OUTPUT": path = ROOT / row["output_path"] if path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"].upper(): raise RuntimeError(f"accepted output hash mismatch: {row['output_path']}") artifact_rows = read_csv(ARTIFACT_MANIFEST)[1] if len(artifact_rows) != 85 or len({row["relative_path"] for row in artifact_rows}) != 85: raise RuntimeError("accepted artifact exact-set mismatch") for row in artifact_rows: path = ROOT / row["relative_path"] if not path.is_file() or path.stat().st_size != int(row["file_size"]) or sha256(path) != row["sha256"].upper(): raise RuntimeError(f"accepted artifact hash mismatch: {row['relative_path']}") if row["artifact_status"] != FINAL_ARTIFACT or row["review_status"] != ACCEPTED: raise RuntimeError(f"accepted artifact status mismatch: {row['relative_path']}") stable_rows = read_csv(CURRENT_STABLE_MANIFEST)[1] batch2_current = sum("BATCH-002" in row.get("source_path", "") or CASE_ID in row.get("source_path", "") for row in stable_rows) if batch2_current: raise RuntimeError("BATCH-002 polluted the current BATCH-001 industry release") if not ACCEPTANCE_RECORD.is_file() or not ACCEPTANCE_RECEIPT.is_file(): raise RuntimeError("acceptance records missing") return { "status": ACCEPTED, "final_audit": FINAL_AUDIT, "output_manifest_rows": len(output_rows), "final_accepted_outputs": 40, "not_applicable_outputs": 2, "artifact_exact_set": len(artifact_rows), "output_manifest_sha256": sha256(OUTPUT_MANIFEST), "artifact_manifest_sha256": sha256(ARTIFACT_MANIFEST), "acceptance_record_sha256": sha256(ACCEPTANCE_RECORD), "acceptance_validation_sha256": sha256(ACCEPTANCE_RECEIPT), "immutable_b001_and_current_release_hash_match": len(IMMUTABLE_B001), "batch002_current_release_paths": batch2_current, "parent_status": PARENT_STATUS, } def run() -> dict[str, Any]: if ACCEPTANCE_RECORD.exists(): result = verify_accepted() result["idempotent"] = True return result headers, artifacts = preflight() at = accepted_at() markdown_counts = sync_markdown(artifacts) structured_counts = sync_structured_statuses(artifacts) output_rows = update_output_manifest() update_projection_receipts() update_registry() update_execution_receipt(at) write_acceptance_record(at, output_rows) write_acceptance_validation(at, output_rows, {"markdown": markdown_counts, "structured": structured_counts}) rebuild_artifact_manifest(headers, artifacts, at) result = verify_accepted() result["idempotent"] = False result["accepted_at"] = at return result if __name__ == "__main__": print(json.dumps(run(), ensure_ascii=False, sort_keys=True, indent=2))