sync: SHANGHAI-U9 2026-08-19 sim-rtl-flow 增约束原语阶段与版本说明树文件重命名,新增 ledger 迁移脚本

This commit is contained in:
bowei
2026-08-19 12:50:44 +08:00
parent d77d025c59
commit 6535dd83d4
3 changed files with 205 additions and 21 deletions
@@ -0,0 +1,93 @@
"""Seed constraint_primitives#N placeholders into an existing sim-rtl-flow ledger.
The `constraint_primitives` node was added to the graph after this run started.
Replay is keyed by `<task>#<n>`, so a run whose ledger predates the node rewinds
to the first unrecorded `constraint_primitives#1` instead of resuming where it
left off. Seeding one placeholder per module round that already happened lets
the replay walk straight through to the real pending key.
The placeholders record the truth: those rounds went to Vivado WITHOUT any
constraint injection, because the step did not exist yet. The first round that
runs after the migration gets a real injection.
Usage:
python migrate_constraint_ledger.py <run_dir> <count> [--apply]
Without --apply it only prints what it would do.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
PLACEHOLDER = {
"success": True,
"gate_passed": True,
"artifacts": [],
"notes": (
"迁移占位:这一轮模块处理发生在『约束原语』步骤并入 sim-rtl-flow 之前,"
"当时并未做任何综合属性注入,该轮 Vivado 跑的是未注入的 RTL。"
"本条只为让 replay 走过历史轮次;之后的每一轮都会真正执行注入。"
),
"timing_met": True,
"hil_passed": True,
"needs_code_change": False,
"change_scope": "none",
"affected_modules": [],
"metadata": {
"migration_placeholder": True,
"injected": {},
"unmatched": [],
"spec_path": "",
"spec_hash": "",
},
}
def main(argv: list[str]) -> int:
if len(argv) < 3:
print(__doc__)
return 2
run_dir = Path(argv[1])
count = int(argv[2])
apply = "--apply" in argv
ledger_path = run_dir / "ledger.json"
ledger = json.loads(ledger_path.read_text(encoding="utf-8"))
added = []
for n in range(1, count + 1):
key = f"constraint_primitives#{n}"
if key in ledger:
print(f" skip {key} (already recorded)")
continue
ledger[key] = dict(PLACEHOLDER)
added.append(key)
print(f"ledger: {ledger_path}")
print(f"would add: {added}")
if not apply:
print("(dry run — pass --apply to write)")
return 0
backup = ledger_path.with_suffix(".json.pre_constraint_migration")
if not backup.exists():
backup.write_text(
json.dumps(
json.loads(ledger_path.read_text(encoding="utf-8")),
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
print(f"backup -> {backup}")
ledger_path.write_text(
json.dumps(ledger, ensure_ascii=False, indent=2), encoding="utf-8"
)
print(f"applied {len(added)} placeholder(s)")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))
+15 -4
View File
@@ -402,7 +402,7 @@ def _print_final(state: Any, replayed: list[str]) -> None:
if review:
mark = "批准" if review.get("approved") else "驳回"
if (review.get("metadata") or {}).get("skipped_by_history"):
mark = "按历史设计说明文件跳过"
mark = "按版本说明树文件跳过"
print(f"{label} : {mark} {review.get('notes', '')}".rstrip())
plan = state.get("history_plan") or {}
print(f"历史文件 · 参考 : {plan.get('history_path') or '(none)'}")
@@ -421,6 +421,17 @@ def _print_final(state: Any, replayed: list[str]) -> None:
f" - {item['module']}: stage={item['stage']} "
f"success={item['success']} artifacts={item.get('artifacts')}"
)
constraints = state.get("constraint_result") or {}
if constraints:
print(
"约束原语 : "
f"source={constraints.get('source')} "
f"passed={constraints.get('passed')} "
f"spec={constraints.get('spec_path') or '(未回报)'} "
f"hash={constraints.get('spec_hash') or '(未回报 — 下轮必须重注)'} "
f"injected={_jsonable(constraints.get('injected'))} "
f"idempotent={constraints.get('idempotent_verified')}"
)
print(f"vivado : {_jsonable(state.get('vivado_result'))}")
print(f"hil : {_jsonable(state.get('hil_result'))}")
print(f"\nevents ({len(state.get('events', []))}):")
@@ -473,8 +484,8 @@ def cmd_start(args, api) -> int:
payload.pop("outcomes", None)
payload.pop("reviews", None)
work_dir = _resolve_work_dir(args, payload)
# 把"参考哪一份历史设计说明文件"冻结进 input.json:本次运行结束时会写出一份
# 新的 sim-rtl-flow-<date>.md,若每次 resume 都重新扫描目录,replay 就会在
# 把"参考哪一份版本说明树文件"冻结进 input.json:本次运行结束时会写出一份
# 新的 xxx<date>_tree.md,若每次 resume 都重新扫描目录,replay 就会在
# 半路换掉参考文件,走出一条与已记录 ledger 不同的路。
if "history_path" not in payload:
history_dir = api["history_dir_for"](
@@ -482,7 +493,7 @@ def cmd_start(args, api) -> int:
)
payload["history_dir"] = history_dir
payload["history_path"] = api["discover_history_doc"](
str(work_dir), history_dir
str(work_dir), history_dir, payload.get("project_name", "")
)
run = RunDir(work_dir, args.thread_id)
run.set_input(payload)