From 8080703a0e17238c1cff0b7d375cb48d5fba4ecf Mon Sep 17 00:00:00 2001 From: bowei Date: Mon, 27 Jul 2026 01:08:42 +0800 Subject: [PATCH] sync: SHANGHAI-U9 2026-07-27 --- langgraph-plan-review/scripts/agents.py | 5 + sim-rtl-flow/SKILL.md | 314 +++++++++++ sim-rtl-flow/scripts/run_sim_rtl_flow.py | 680 +++++++++++++++++++++++ zynq-imx6-flow/SKILL.md | 199 +++++++ zynq-imx6-flow/scripts/run_zynq_flow.py | 187 +++++++ 5 files changed, 1385 insertions(+) create mode 100644 sim-rtl-flow/SKILL.md create mode 100644 sim-rtl-flow/scripts/run_sim_rtl_flow.py create mode 100644 zynq-imx6-flow/SKILL.md create mode 100644 zynq-imx6-flow/scripts/run_zynq_flow.py diff --git a/langgraph-plan-review/scripts/agents.py b/langgraph-plan-review/scripts/agents.py index 9d48e83..8869199 100644 --- a/langgraph-plan-review/scripts/agents.py +++ b/langgraph-plan-review/scripts/agents.py @@ -30,6 +30,11 @@ def _run_command(command: Sequence[str], values: Mapping[str, str], timeout: int capture_output=True, check=False, text=True, + # Windows with a CJK locale defaults text mode to GBK; the codex review + # output is UTF-8 (Chinese) -> UnicodeDecodeError in the reader thread and + # result.stdout comes back None. Pin UTF-8 and never hard-fail on decode. + encoding="utf-8", + errors="replace", timeout=timeout, ) elapsed = time.monotonic() - t0 diff --git a/sim-rtl-flow/SKILL.md b/sim-rtl-flow/SKILL.md new file mode 100644 index 0000000..da6a13d --- /dev/null +++ b/sim-rtl-flow/SKILL.md @@ -0,0 +1,314 @@ +--- +name: sim-rtl-flow +description: > + Run the sim-rtl-flow LangGraph workflow at E:\LangGraph_Design\sim-rtl-flow — the + Simulink-to-RTL-to-hardware design pipeline: Simulink -x principle sim (sim-golden) → + HUMAN sign-off on simulation performance → optional -q fixed-point → parallel + functional partition + RTL/SoC interface spec → HUMAN sign-off on the partition → + per-module -E.slx nativization and Xilinx-IP replacement (parallel Send fanout) → + Vivado build + timing → hardware-in-the-loop via fpga-imx6-flow or zynq-imx6-flow. + It drives the compiled graph in-process (no langgraph dev server) and hands every + subagent task back to THIS Claude session — and every human checkpoint to the USER — + through a record/replay ledger stored in the design project being worked on. + + ALWAYS use this skill whenever the user wants to run, start, drive, resume or continue + the sim-rtl-flow workflow / Simulink→RTL design flow, or says things like "启动 + sim-rtl-flow", "跑 sim-rtl-flow", "运行仿真到RTL的设计流程", "执行 sim-rtl-flow 设计任务", + "继续 sim-rtl-flow", "run the sim to rtl flow", "resume the sim-rtl-flow run", even if + they don't name the project path. Also use it when the user asks to record a subagent + result, to approve or reject one of the two human checkpoints, or to rehearse the graph + with the scripted backend. +--- + +## What this skill does + +Drives the **sim-rtl-flow** graph (21 top-level nodes + a 3-node per-module subgraph, +1:1 with `Notes/PR01-1.md`, `Notes/PR01-2.md` and `Notes/PR01_sim_rtl_flow.drawio`). + +The graph contains **no** agent logic — it talks to the outside world through two +protocols: `SubagentBackend` (design work) and `ReviewGate` (the two mandatory human +sign-offs). This skill plugs a **record/replay** implementation into both: the graph +runs in-process until it needs an answer that isn't recorded yet, then stops and prints +the request. **You are the subagent; the USER is the reviewer.** You do the work, +`record` the answer, and `resume` — the driver replays the ledger, deterministically +walks back to where it stopped, and asks for the next thing. + +| Item | Value | +|---|---| +| Project root | 见 `~\skills_env.md` 的 `SIM_RTL_FLOW_ROOT`（driver 自动读取，也可 `--project` 覆盖） | +| Conda env | `langgraph` (path 见 `~\skills_env.md` 的 `CONDA_LANGGRAPH_PYTHON`；装环境用 `/langgraph-vivado-install`) | +| Graph | `sim_rtl_flow.graph:build_workflow` (in-process, no `langgraph dev`) | +| Driver | `%USERPROFILE%\.claude\skills\sim-rtl-flow\scripts\run_sim_rtl_flow.py` | +| **Run state** | **`\.sim_rtl_flow\runs\\`** — inside the *design project*, never in the workflow repo | +| Max attempts | 10 per gate (default) | + +There is **no LangGraph checkpointer** in the driver, on purpose: graph nodes have no +side effects of their own (all side effects live in the subagents, which are replayed), +so re-invoking from scratch is cheaper and immune to partial-superstep resume semantics. +`ledger.json` *is* the persistence. + +**Drive the driver with the Bash tool (Git Bash), not PowerShell** — PowerShell 5.1 +mangles the double quotes inside `--result '{...}'` payloads. Every command below was +run this way and verified. + +```bash +PY="" +RUN="$USERPROFILE/.claude/skills/sim-rtl-flow/scripts/run_sim_rtl_flow.py" +export PYTHONIOENCODING=utf-8 +``` + +The driver resolves the workflow checkout from `SIM_RTL_FLOW_ROOT` in +`~\skills_env.md`, so no path is baked into this skill. + +## ⚠️ The two human checkpoints are the user's decision, not yours + +The graph stops at two points that **an agent may not decide**: + +| checkpoint | where | what the user confirms | +|---|---|---| +| `sim_performance` | after step 1 | the `-x` sim meets the requirement and may be frozen as **sim-golden** | +| `functional_partition` | after step 3 | the pure-sim / RTL / SoC split and the RTL↔SoC interface doc | + +When one comes up the driver prints a `!!!` banner. **Relay the question and its context +to the user verbatim, then wait.** Do not approve on your own judgement, do not infer +approval from "the numbers look fine", and do not use `--all-ok` (the driver refuses it +for review keys anyway). A rejection **must** carry the user's reason. + +```bash +"$PY" "$RUN" --work-dir --thread-id record --key review:sim_performance#1 --approve +"$PY" "$RUN" --work-dir --thread-id record --key review:sim_performance#1 \ + --reject --notes "带外抑制不足 6dB" +``` + +Rejecting `sim_performance` re-runs step 1; rejecting `functional_partition` re-runs the +whole partition + interface superstep. + +## Non-negotiable design rules the graph enforces + +From `Notes/PR01.md` and `Notes/PR01-2.md`; the graph routes on these. + +1. **sim-golden is the only acceptance criterion.** `-q`, the partition, and every `-E` + version must match the `-x` sim. If it doesn't, record `gate_passed: false` — never + record a pass you didn't verify. +2. **One `-E.slx` per RTL submodule**, saved separately (`_A-E.slx`, …). Do not + nativize all submodules in one Simulink model. +3. **Never hand-edit MATLAB-generated RTL.** If IP replacement, Vivado timing, or HIL + needs a change there, return `change_scope: "simulink_module"` + `affected_modules`. + The graph then forces a `modify_e:*` step — **edit the `-E.slx` and re-simulate + against sim-golden** — before any RTL is regenerated. +4. **Timing fixes change the RTL, not the constraints.** Prefer adding pipeline/delay + stages. Only touch timing constraints when the constraint itself is plainly wrong, + and say why in `notes`. +5. **HIL is strictly read-only.** `hil:*` arrives with `allow_code_changes=false`. Run + `fpga-imx6-flow` / `zynq-imx6-flow` for *diagnosis only*; the moment a code change is + needed, stop that sub-flow and report it back here — the fix happens in the + sim-rtl-flow main graph, then re-enters Vivado and HIL. + +## Run (agent path) + +### 1. Write the workflow input + +`work_dir` is the design project. **All run state goes under +`/.sim_rtl_flow/`** — that is where `resume`/`record`/`status` will look. + +```bash +cat > /c/Users/Administrator/AppData/Local/Temp/claude/input.json <<'EOF' +{ + "requirements_path": "Notes/requirements.md", + "work_dir": "E:/Xilinx_design/my_project", + "project_name": "demo", + "device_family": "fpga-imx6", + "fixed_point_required": true, + "rtl_modules": ["A", "B", "C"], + "soc_modules": ["control"] +} +EOF +``` + +| field | meaning | +|---|---| +| `requirements_path` | user's requirement `.md`, relative to `work_dir` | +| `work_dir` | the design project — **absolute path**; also where run state is written | +| `project_name` | artifact prefix — `-x.slx`, `_A-E.slx`, `.bit` | +| `device_family` | `fpga-imx6` → HIL via `fpga-imx6-flow`; `zynq-imx6` → `zynq-imx6-flow`. Anything else raises at `initialize`. | +| `fixed_point_required` | `false` skips the `-q` stage | +| `rtl_modules` | submodule names; each gets its own parallel `native → ip_replace` chain. Non-empty. | +| `soc_modules` | MCU/SoC-side modules (interface-spec input) | + +### 2. Start — exits 10 with a handoff + +```bash +"$PY" "$RUN" --thread-id my-design-001 start --input /c/Users/Administrator/AppData/Local/Temp/claude/input.json +``` + +`start` takes `work_dir` from the input JSON and prints where state landed. Verified: + +``` +work dir : E:\Xilinx_design\my_project +run dir : E:\Xilinx_design\my_project\.sim_rtl_flow\runs\my-design-001 +======================================================================== +HANDOFF — 1 subagent task(s), 0 human review(s) +======================================================================== +--- key: simulink_x#1 + role : simulink-principle-agent + permission : WRITE ALLOWED + prompt : 根据需求文档创建或修订 Simulink -x 原理仿真；... +``` + +Exit codes: **10** = handoff pending, **0** = completed, **1** = `failed`, **2** = crash. + +### 3. Answer, then resume + +Every later command needs to find the run, so pass `--work-dir` (or `cd` into it): + +```bash +W="--work-dir E:/Xilinx_design/my_project --thread-id my-design-001" +"$PY" "$RUN" $W record --key simulink_x#1 \ + --result '{"gate_passed": true, "artifacts": ["demo-x.slx"], "notes": "EVM -38dB"}' +"$PY" "$RUN" $W resume +``` + +Loop until `WORKFLOW FINISHED — status: completed`. When several keys are pending, +record **each**, then `resume` once. `record --all-ok` fast-forwards pending *subagent* +keys with a default success — rehearsal only, and it refuses review keys. + +### 4. What you get handed + +| key | what you actually do | which skill to use | +|---|---|---| +| `simulink_x#n` | build/revise the `-x` sim, document the performance envelope | MATLAB MCP + `model-based-design-core:building-simulink-models`, `simulating-simulink-models` | +| `review:sim_performance#n` | **ask the user** to sign off on sim-golden | — (relay to user) | +| `fixed_point#n` | `-q` fixed-point, compare to `-x` | same MATLAB skills | +| `partition#n` | pure-sim / RTL / SoC split, must not degrade sim-golden | same | +| `interfaces#n` | RTL↔SoC interface doc — register-config or data-stream only | write the doc under `work_dir` | +| `review:functional_partition#n` | **ask the user** to sign off on the split + interfaces | — (relay to user) | +| `module_plan#n` | split the RTL side into A/B/C…, give order + dependencies (**first pass only**) | inline | +| `native:#n` | nativize M into HDL-capable blocks, save `_M-E.slx`, generate RTL | request says `simulink-rtl-dev` — **not installed**; use MATLAB MCP + model-based-design-core | +| `ip_replace:#n` | swap in Xilinx IP, cycle-true compare vs MATLAB RTL | request says `rtl-ip-replace` — **not installed**; use `vivado-sim` + `vivado-mcp` | +| `modify_e:vivado#n` / `modify_e:hil#n` | **edit the named `-E.slx` and re-simulate** against sim-golden | MATLAB MCP + model-based-design-core | +| `vivado#n` | build the Vivado project, synth+impl+bitstream, check all timing | `vivado-prj-create`, `vivado-mcp` | +| `hil:fpga-imx6-flow#n` / `hil:zynq-imx6-flow#n` | HIL debug, **read-only** | `fpga-imx6-flow` / `zynq-imx6-flow` | +| `timing_rework#n` | fix non-Simulink RTL / SoC / project timing | inline | +| `hil_rework#n` | apply the code change HIL asked for | inline | + +`#n` is the n-th time that key has been requested this run — retries and rework produce +`#2`, `#3`, … + +### 5. Answer JSON + +`"$PY" "$RUN" schema` prints both templates. All fields optional except `approved`. + +```json +subagent: {"success": true, "gate_passed": true, "artifacts": [], "notes": "", + "timing_met": true, "hil_passed": true, "needs_code_change": false, + "change_scope": "none", "affected_modules": [], "metadata": {}} +review: {"approved": true, "notes": "", "metadata": {}} +``` + +| field | who reads it | +|---|---| +| `success` + `gate_passed` | every design node — `false` → retry, then `failed` after 10 attempts | +| `artifacts` | `simulink_x`'s first artifact **becomes sim-golden**; `vivado`'s go to HIL as the bitstream | +| `timing_met` | `vivado` only | +| `hil_passed` | `hil:*` only | +| `needs_code_change` + `change_scope` + `affected_modules` | `vivado` and `hil:*` — the rework router | + +Routing on `change_scope`: + +- `simulink_module` → **`modify_e:*`** (edit `-E.slx` + re-simulate) → only the affected + modules re-run `native → ip_replace` → Vivado → HIL. +- `non_simulink_rtl` / `soc` / `hardware` → `timing_rework` (from Vivado) or `hil_rework` + (from HIL) → Vivado. +- The driver **rejects** `needs_code_change: true` with `change_scope: "none"`. + +### 6. Parallel batches (verified) + +| stage | keys handed over at once | +|---|---| +| after `-x` / `-q` | `interfaces#1` **and** `partition#1` | +| module fanout | one `native:#1` per module (3 for `["A","B","C"]`) | +| next module stage | one `ip_replace:#1` per module | + +Work batched tasks in any order; record all, then `resume`. + +## Rehearsal without touching hardware + +```bash +"$PY" "$RUN" dry-run --input examples/happy_path.json +``` + +Runs the whole graph against `ScriptedBackend` + auto-approving `ScriptedReviewGate` — +no handoffs, no side effects. Verified: `status: completed`, 13 subagent calls listed +with their skill and write permission, plus both review checkpoints. To rehearse a +branch, add `outcomes` (task → list of results) and/or `reviews` (checkpoint → list of +decisions) to the input JSON; `start`/`resume` strip both. + +```json +"outcomes": {"hil:fpga-imx6-flow": [ + {"success": true, "hil_passed": false, "needs_code_change": true, + "change_scope": "simulink_module", "affected_modules": ["A"]}, + {"success": true, "hil_passed": true}]}, +"reviews": {"sim_performance": [{"approved": false, "notes": "EVM 未达标"}]} +``` + +## Other commands + +```bash +"$PY" "$RUN" $W status # ledger + pending + final status +"$PY" "$RUN" schema # both answer templates +"$PY" "$RUN" mermaid # graph as Mermaid (see gotcha) +"$PY" -m unittest discover -s tests # project's own 16 tests — all pass +``` + +Global flags go **before** the subcommand: `--project`, `--work-dir`, `--thread-id`, +`--recursion-limit` (200), `--max-attempts` (10). + +## Gotchas + +- **`resume`/`record`/`status` need `--work-dir`** (or a `cd` into the design project). + Only `start` can infer it, from the input JSON. Without it the driver looks in the cwd + and reports `no input.json in …`. +- **The driver refuses a `work_dir` inside the sim-rtl-flow repo.** Run state belongs to + the design project (PR01-2). `examples/happy_path.json` therefore carries a placeholder + `E:/Xilinx_design/your_project` — fine for `dry-run`, but replace it before `start`. +- **`mermaid` output is still partly misleading.** The forward and rework edges now + render (path maps were added), but `plan_modules`' `Send` fanout is a dynamic edge + Mermaid can't show. Use `Notes/Result00.md` for the authoritative picture. +- **`--max-attempts` must precede the subcommand** and stay the *same value* across + `start` and every `resume` — it is runtime context, not persisted state. +- **The two skills the graph names don't exist.** `native:*` suggests `simulink-rtl-dev`, + `ip_replace:*` suggests `rtl-ip-replace`; neither is installed. Fall back to the MATLAB + MCP + `model-based-design-core:*` skills and `vivado-sim` / `vivado-mcp`. Don't abort. +- **`simulink_x`'s first artifact silently becomes sim-golden.** Record `artifacts: []` + and the graph substitutes `-x.slx`. Always record the real filename. +- **`module_plan` runs only on the first pass.** Rework rounds reuse the plan, so you + will not be asked to re-partition the module list mid-run. +- **Re-recording a key requires editing `ledger.json` by hand.** `record` only accepts + currently-pending keys; anything already in the ledger is replayed forever. To redo a + step, delete its entry *and everything after it*, then `resume`. +- **Runs created before the PR01-2 change live in the old location** + (`\runs\\`) and their ledgers have no `review:*` or + `module_plan` keys, so they cannot be resumed under the current graph. Start a fresh + thread instead. +- **`work_dir` is passed straight through to every subagent request.** Forward slashes + are fine on Windows; a relative `"."` resolves against wherever the driver was + launched — always use an absolute path. +- **`start` wipes the ledger** unless `--keep-ledger`. Use a fresh `--thread-id` per run. + +## Troubleshooting + +| symptom | cause / fix | +|---|---| +| `sim_rtl_flow package not found under …\src` | `SIM_RTL_FLOW_ROOT` in `~\skills_env.md` is wrong/missing; fix it or pass `--project <检出目录>` | +| `sim-rtl-flow 工作流仓库路径未配置` | add `SIM_RTL_FLOW_ROOT = <检出目录>` to `~\skills_env.md` | +| `no input.json in …\.sim_rtl_flow\runs\` | missing/incorrect `--work-dir`, or `start` was never run | +| `nothing pending — run 'start' or 'resume' first` | the previous invoke completed or crashed; check `status` | +| `--key X is not pending; pending = [...]` | typo, or already recorded (see the re-record gotcha) | +| `--all-ok 不能用于人工审核断点` | working as intended — the user must decide; use `--approve` / `--reject` | +| `驳回必须写明理由：请提供 notes` | add `--notes ""` | +| `unknown review fields: ['gate_passed']` | you sent a subagent result to a `review:*` key; use `--approve`/`--reject` | +| `needs_code_change=true requires an explicit change_scope` | add `change_scope` | +| `MissingReviewGateError` | only if the graph is driven without a review gate — the driver always supplies one | +| `device_family must be 'fpga-imx6' or 'zynq-imx6'` | raised in `initialize`; exit 2 | +| exit 2 + `recursion_limit` | a rework loop isn't converging — raise `--recursion-limit`, or stop recording `needs_code_change` for the same module | +| mojibake in the Chinese prompts | `export PYTHONIOENCODING=utf-8` | diff --git a/sim-rtl-flow/scripts/run_sim_rtl_flow.py b/sim-rtl-flow/scripts/run_sim_rtl_flow.py new file mode 100644 index 0000000..3584966 --- /dev/null +++ b/sim-rtl-flow/scripts/run_sim_rtl_flow.py @@ -0,0 +1,680 @@ +"""Drive the sim-rtl-flow LangGraph workflow from a Claude Code session. + +The graph never talks to an agent host directly: subagent work goes through the +``SubagentBackend`` protocol and the two mandatory human checkpoints go through +the ``ReviewGate`` protocol. This driver plugs a *record/replay* implementation +into both, so the agent host can be **this Claude session** and the reviewer can +be **the user**: + +* when the graph asks for something that has no recorded answer yet, the driver + stops the run, writes the pending request(s) to ``pending.json`` and prints + them; +* Claude performs the subagent task (or relays the review request to the user) + and calls ``record``; +* ``resume`` re-invokes the graph from scratch -- every already-recorded call is + replayed from the ledger, so the graph deterministically walks back to where + it stopped and asks for the next thing. + +Replay-from-scratch is why no LangGraph checkpointer is used here: graph nodes +have no side effects of their own (all side effects live in the subagents, which +are replayed from the ledger), so a fresh invoke is both cheaper and immune to +partial-superstep resume semantics. + +All run state lives under ``/.sim_rtl_flow/`` -- i.e. inside the +project being designed, never inside the sim-rtl-flow repository (Notes/PR01-2.md). + +Commands +-------- + start --input [--thread-id ID] begin a run + resume continue after record + record --key K --result '' record one subagent result + record --key K --approve | --reject answer a human review checkpoint + record --all-ok default-success for pending SUBAGENTS + status show ledger / pending / last state + dry-run --input end-to-end ScriptedBackend rehearsal + mermaid print the graph as Mermaid +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import threading +import traceback +from collections import defaultdict +from pathlib import Path +from typing import Any + + +def _env_lookup(key: str, default: str | None = None) -> str | None: + """Read KEY from the OS environment, else from a ``KEY = value`` line in + ~/skills_env.md. Machine-local paths are never hardcoded in a skill; see + the same helper in fpga_flash/scripts/vivado_env.py.""" + value = os.environ.get(key) + if value: + return value + path = os.path.join(os.path.expanduser("~"), "skills_env.md") + try: + with open(path, encoding="utf-8") as handle: + for line in handle: + match = re.match(rf"\s*{re.escape(key)}\s*=\s*(.+?)\s*(?:#.*)?$", line) + if match: + found = match.group(1).strip() + if found and not found.startswith("<"): + return found + except OSError: + pass + return default + + +def _default_project() -> Path: + """The sim-rtl-flow checkout: SIM_RTL_FLOW_ROOT in ~/skills_env.md.""" + root = _env_lookup("SIM_RTL_FLOW_ROOT") + if root: + return Path(root) + raise SystemExit( + "sim-rtl-flow 工作流仓库路径未配置：请在 ~/skills_env.md 增加\n" + " SIM_RTL_FLOW_ROOT = \n" + "或用 --project 显式指定。" + ) + +#: Run state lives inside the *calling* project, not in the workflow repo. +STATE_DIRNAME = ".sim_rtl_flow" + +CHANGE_SCOPES = { + "none", + "simulink_module", + "non_simulink_rtl", + "soc", + "hardware", + "unknown", +} + +RESULT_DEFAULTS: dict[str, Any] = { + "success": True, + "gate_passed": True, + "artifacts": [], + "notes": "", + "timing_met": True, + "hil_passed": True, + "needs_code_change": False, + "change_scope": "none", + "affected_modules": [], + "metadata": {}, +} + +REVIEW_DEFAULTS: dict[str, Any] = {"approved": True, "notes": "", "metadata": {}} + + +def _bootstrap(project: Path): + src = project / "src" + if not (src / "sim_rtl_flow").is_dir(): + raise SystemExit(f"sim_rtl_flow package not found under {src}") + sys.path.insert(0, str(src)) + from sim_rtl_flow.backend import ( # noqa: E402 + ScriptedBackend, + SubagentResult, + ) + from sim_rtl_flow.graph import WorkflowContext, build_workflow # noqa: E402 + from sim_rtl_flow.review import ReviewDecision, ScriptedReviewGate # noqa: E402 + + return { + "ScriptedBackend": ScriptedBackend, + "ScriptedReviewGate": ScriptedReviewGate, + "SubagentResult": SubagentResult, + "ReviewDecision": ReviewDecision, + "WorkflowContext": WorkflowContext, + "build_workflow": build_workflow, + } + + +class HandoffRequired(Exception): + """Raised inside a graph node when an answer is not recorded yet.""" + + def __init__(self, key: str) -> None: + super().__init__(f"handoff required: {key}") + self.key = key + + +class Handoffs: + """Shared record/replay ledger for subagent tasks and human reviews.""" + + def __init__(self, ledger: dict[str, Any]) -> None: + self._ledger = ledger + self._counter: dict[str, int] = defaultdict(int) + self._lock = threading.Lock() + self.pending: list[dict[str, Any]] = [] + self.replayed: list[str] = [] + + def resolve(self, name: str, describe) -> dict[str, Any]: + """Return the recorded payload for ``name``'s next occurrence, or halt.""" + with self._lock: + self._counter[name] += 1 + key = f"{name}#{self._counter[name]}" + recorded = self._ledger.get(key) + if recorded is not None: + self.replayed.append(key) + return recorded + self.pending.append({"key": key, **describe()}) + raise HandoffRequired(key) + + +class ClaudeSessionBackend: + """Replay recorded subagent results; hand unknown requests to Claude.""" + + def __init__(self, handoffs: Handoffs, SubagentResult) -> None: + self._handoffs = handoffs + self._result_cls = SubagentResult + + def run(self, request): # noqa: ANN001 - duck-typed SubagentRequest + payload = self._handoffs.resolve( + request.task, + lambda: { + "kind": "subagent", + "task": request.task, + "role": request.role, + "skill": request.skill, + "module": request.module, + "allow_code_changes": request.allow_code_changes, + "work_dir": request.work_dir, + "prompt": request.prompt, + "context": request.context, + }, + ) + values = {**RESULT_DEFAULTS, **payload} + values["artifacts"] = tuple(values["artifacts"]) + values["affected_modules"] = tuple(values["affected_modules"]) + return self._result_cls(**values) + + +class ClaudeSessionReviewGate: + """Relay the two mandatory human checkpoints to the user via Claude.""" + + def __init__(self, handoffs: Handoffs, ReviewDecision) -> None: + self._handoffs = handoffs + self._decision_cls = ReviewDecision + + def review(self, request): # noqa: ANN001 - duck-typed ReviewRequest + payload = self._handoffs.resolve( + f"review:{request.checkpoint}", + lambda: { + "kind": "review", + "checkpoint": request.checkpoint, + "title": request.title, + "work_dir": request.work_dir, + "prompt": request.prompt, + "context": request.context, + }, + ) + values = {**REVIEW_DEFAULTS, **payload} + return self._decision_cls(**values) + + +# -------------------------------------------------------------------------- +# run directory helpers +# -------------------------------------------------------------------------- + + +class RunDir: + """``/.sim_rtl_flow/runs//`` -- inside the design project.""" + + def __init__(self, work_dir: Path, thread_id: str, *, create: bool = True) -> None: + self.work_dir = work_dir.expanduser().resolve() + self.thread_id = thread_id + self.path = self.work_dir / STATE_DIRNAME / "runs" / thread_id + if create: + self.path.mkdir(parents=True, exist_ok=True) + + def _read(self, name: str, default: Any) -> Any: + target = self.path / name + if not target.exists(): + return default + return json.loads(target.read_text(encoding="utf-8")) + + def _write(self, name: str, payload: Any) -> None: + self.path.mkdir(parents=True, exist_ok=True) + (self.path / name).write_text( + json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + @property + def workflow_input(self) -> dict[str, Any]: + data = self._read("input.json", None) + if data is None: + raise SystemExit( + f"no input.json in {self.path}\n" + "run `start --input ` first, or point --work-dir at the " + "design project whose .sim_rtl_flow/ holds this run" + ) + return data + + def set_input(self, payload: dict[str, Any]) -> None: + self._write("input.json", payload) + + @property + def ledger(self) -> dict[str, Any]: + return self._read("ledger.json", {}) + + def set_ledger(self, payload: dict[str, Any]) -> None: + self._write("ledger.json", payload) + + @property + def pending(self) -> list[dict[str, Any]]: + return self._read("pending.json", []) + + def set_pending(self, payload: list[dict[str, Any]]) -> None: + self._write("pending.json", payload) + + def set_state(self, payload: Any) -> None: + self._write("last_state.json", payload) + + @property + def state(self) -> Any: + return self._read("last_state.json", None) + + +def _resolve_work_dir(args, payload: dict[str, Any] | None = None) -> Path: + """The calling project's directory -- where all run state must live.""" + if payload is not None: + work_dir = payload.get("work_dir") + if not work_dir: + raise SystemExit("workflow input must define work_dir") + resolved = Path(work_dir) + elif args.work_dir is not None: + resolved = Path(args.work_dir) + else: + resolved = Path.cwd() + + resolved = resolved.expanduser().resolve() + project = Path(args.project).expanduser().resolve() + if resolved == project or project in resolved.parents: + raise SystemExit( + f"work_dir 不能位于 sim-rtl-flow 工作流仓库内（{resolved}）。\n" + "运行状态必须保存在调用这个工作流的那个工程目录下 —— " + "请把 work_dir 指向真实的设计工程。" + ) + return resolved + + +def _jsonable(value: Any) -> Any: + if isinstance(value, dict): + return {str(k): _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + if isinstance(value, (str, int, float, bool)) or value is None: + return value + return str(value) + + +def _unwrap(exc: BaseException) -> HandoffRequired | None: + seen: set[int] = set() + stack: list[BaseException] = [exc] + while stack: + current = stack.pop() + if current is None or id(current) in seen: + continue + seen.add(id(current)) + if isinstance(current, HandoffRequired): + return current + if isinstance(current, BaseExceptionGroup): # noqa: F821 - py3.11+ + stack.extend(current.exceptions) + stack.append(current.__cause__) + stack.append(current.__context__) + return None + + +# -------------------------------------------------------------------------- +# printing +# -------------------------------------------------------------------------- + + +def _print_pending(pending: list[dict[str, Any]]) -> None: + reviews = [item for item in pending if item.get("kind") == "review"] + tasks = [item for item in pending if item.get("kind") != "review"] + + print("\n" + "=" * 72) + print( + f"HANDOFF — {len(tasks)} subagent task(s), {len(reviews)} human review(s)" + ) + print("=" * 72) + + for item in reviews: + print("\n" + "!" * 72) + print(f"!! 人工审核确认 — key: {item['key']}") + print(f"!! {item['title']}") + print("!! 必须把该问题原样转达用户并等待用户决定；") + print("!! Claude 不得代替用户批准，也不得依据自己的判断放行。") + print("!" * 72) + print(f" checkpoint : {item['checkpoint']}") + print(f" work_dir : {item['work_dir']}") + print(f" 问题 : {item['prompt']}") + print(" context : " + json.dumps(item["context"], ensure_ascii=False)) + + for item in tasks: + print(f"\n--- key: {item['key']}") + print(f" role : {item['role']}") + print(f" skill : {item['skill'] or '(none — do it inline)'}") + print(f" module : {item['module'] or '-'}") + print(f" work_dir : {item['work_dir']}") + write = ( + "WRITE ALLOWED" + if item["allow_code_changes"] + else "READ-ONLY (no code edits!)" + ) + print(f" permission : {write}") + print(f" prompt : {item['prompt']}") + print(" context : " + json.dumps(item["context"], ensure_ascii=False)) + + print("\n" + "-" * 72) + print("Answer each, then resume:") + for item in reviews: + print(f" record --key {item['key']} --approve # 用户批准后") + print(f" record --key {item['key']} --reject --notes '…' # 用户驳回后") + for item in tasks: + print( + f" record --key {item['key']} " + "--result '{\"gate_passed\": true, \"notes\": \"...\"}'" + ) + print(" resume") + print("-" * 72) + + +def _print_final(state: Any, replayed: list[str]) -> None: + status = state.get("status") if isinstance(state, dict) else None + print("\n" + "=" * 72) + print(f"WORKFLOW FINISHED — status: {status}") + print("=" * 72) + if isinstance(state, dict): + if state.get("failure_reason"): + print(f"failure_reason : {state['failure_reason']}") + for label, key in (("人工确认 · 仿真性能", "sim_review"), + ("人工确认 · 功能划分", "partition_review")): + review = state.get(key) + if review: + mark = "批准" if review.get("approved") else "驳回" + print(f"{label} : {mark} {review.get('notes', '')}".rstrip()) + print(f"modules : {state.get('rtl_modules')}") + for item in state.get("module_results", []): + print( + f" - {item['module']}: stage={item['stage']} " + f"success={item['success']} artifacts={item.get('artifacts')}" + ) + print(f"vivado : {_jsonable(state.get('vivado_result'))}") + print(f"hil : {_jsonable(state.get('hil_result'))}") + print(f"\nevents ({len(state.get('events', []))}):") + for event in state.get("events", []): + print(" " + json.dumps(_jsonable(event), ensure_ascii=False)) + print(f"\nreplayed answers: {len(replayed)}") + + +# -------------------------------------------------------------------------- +# core invoke +# -------------------------------------------------------------------------- + + +def _invoke(run: RunDir, api: dict[str, Any], args) -> int: + handoffs = Handoffs(run.ledger) + graph = api["build_workflow"]() + context = api["WorkflowContext"]( + backend=ClaudeSessionBackend(handoffs, api["SubagentResult"]), + max_attempts=args.max_attempts, + review_gate=ClaudeSessionReviewGate(handoffs, api["ReviewDecision"]), + ) + try: + state = graph.invoke( + run.workflow_input, + {"recursion_limit": args.recursion_limit}, + context=context, + ) + except BaseException as exc: # noqa: BLE001 - re-raised unless it is a handoff + handoff = _unwrap(exc) + if handoff is None: + traceback.print_exc() + return 2 + run.set_pending(handoffs.pending) + _print_pending(handoffs.pending) + return 10 + + run.set_pending([]) + run.set_state(_jsonable(state)) + _print_final(state, handoffs.replayed) + return 0 if state.get("status") == "completed" else 1 + + +# -------------------------------------------------------------------------- +# commands +# -------------------------------------------------------------------------- + + +def cmd_start(args, api) -> int: + payload = json.loads(Path(args.input).read_text(encoding="utf-8")) + payload.pop("outcomes", None) + run = RunDir(_resolve_work_dir(args, payload), args.thread_id) + run.set_input(payload) + if not args.keep_ledger: + run.set_ledger({}) + run.set_pending([]) + print(f"work dir : {run.work_dir}") + print(f"run dir : {run.path}") + print(f"thread : {run.thread_id}") + print("input : " + json.dumps(payload, ensure_ascii=False)) + return _invoke(run, api, args) + + +def cmd_resume(args, api) -> int: + run = RunDir(_resolve_work_dir(args), args.thread_id, create=False) + return _invoke(run, api, args) + + +def _validate_result(payload: dict[str, Any]) -> dict[str, Any]: + unknown = set(payload) - set(RESULT_DEFAULTS) + if unknown: + raise SystemExit(f"unknown result fields: {sorted(unknown)}") + scope = payload.get("change_scope", "none") + if scope not in CHANGE_SCOPES: + raise SystemExit(f"change_scope must be one of {sorted(CHANGE_SCOPES)}") + if payload.get("needs_code_change") and scope == "none": + raise SystemExit( + "needs_code_change=true requires an explicit change_scope " + "(simulink_module / non_simulink_rtl / soc / hardware)" + ) + return payload + + +def _validate_review(payload: dict[str, Any]) -> dict[str, Any]: + unknown = set(payload) - set(REVIEW_DEFAULTS) + if unknown: + raise SystemExit(f"unknown review fields: {sorted(unknown)}") + if "approved" not in payload: + raise SystemExit("review decision must set approved (true/false)") + if not payload["approved"] and not payload.get("notes"): + raise SystemExit("驳回必须写明理由：请提供 notes") + return payload + + +def cmd_record(args, api) -> int: + run = RunDir(_resolve_work_dir(args), args.thread_id, create=False) + pending = run.pending + if not pending: + raise SystemExit("nothing pending — run `start` or `resume` first") + ledger = run.ledger + by_key = {item["key"]: item for item in pending} + + if args.all_ok: + reviews = [k for k, v in by_key.items() if v.get("kind") == "review"] + if reviews: + raise SystemExit( + "--all-ok 不能用于人工审核断点，必须由用户决定：" + ", ".join(reviews) + ) + for key in by_key: + ledger[key] = dict(RESULT_DEFAULTS) + run.set_ledger(ledger) + print(f"recorded default success for: {sorted(by_key)}") + return 0 + + if not args.key: + raise SystemExit("--key is required (or use --all-ok)") + if args.key not in by_key: + raise SystemExit(f"--key {args.key} is not pending; pending = {sorted(by_key)}") + item = by_key[args.key] + is_review = item.get("kind") == "review" + + if args.approve or args.reject: + if not is_review: + raise SystemExit("--approve/--reject 只用于人工审核断点") + payload: dict[str, Any] = { + "approved": bool(args.approve), + "notes": args.notes or "", + } + else: + if args.result_file: + raw = Path(args.result_file).read_text(encoding="utf-8") + elif args.result is not None: + raw = args.result + else: + raise SystemExit( + "provide --approve/--reject (review) or --result '' (subagent)" + ) + payload = json.loads(raw) + if not isinstance(payload, dict): + raise SystemExit("result must be a JSON object") + + payload = _validate_review(payload) if is_review else _validate_result(payload) + ledger[args.key] = payload + run.set_ledger(ledger) + remaining = sorted(set(by_key) - set(ledger)) + print(f"recorded {args.key}: " + json.dumps(payload, ensure_ascii=False)) + print(f"still pending: {remaining if remaining else '(none — run `resume`)'}") + return 0 + + +def cmd_status(args, api) -> int: + run = RunDir(_resolve_work_dir(args), args.thread_id, create=False) + print(f"work dir: {run.work_dir}") + print(f"run dir : {run.path}") + if not run.path.exists(): + print("(no such run — check --work-dir / --thread-id)") + return 1 + ledger = run.ledger + print(f"ledger : {len(ledger)} recorded answer(s)") + for key, value in ledger.items(): + print(" " + key + " -> " + json.dumps(value, ensure_ascii=False)) + pending = run.pending + if pending: + _print_pending(pending) + else: + state = run.state + print("pending : none") + if state: + print(f"status : {state.get('status')}") + return 0 + + +def cmd_dry_run(args, api) -> int: + payload = json.loads(Path(args.input).read_text(encoding="utf-8")) + outcomes = payload.pop("outcomes", None) + reviews = payload.pop("reviews", None) + backend = api["ScriptedBackend"](outcomes) if outcomes else api["ScriptedBackend"]() + gate = ( + api["ScriptedReviewGate"](reviews) if reviews else api["ScriptedReviewGate"]() + ) + graph = api["build_workflow"]() + state = graph.invoke( + payload, + {"recursion_limit": args.recursion_limit}, + context=api["WorkflowContext"]( + backend=backend, max_attempts=args.max_attempts, review_gate=gate + ), + ) + print("scripted subagent calls:") + for call in backend.calls: + print( + f" {call.task:<22} skill={call.skill or '-':<18} " + f"write={call.allow_code_changes}" + ) + print("human review checkpoints (auto-approved in rehearsal):") + for request in gate.requests: + print(f" {request.checkpoint:<22} {request.title}") + _print_final(state, []) + return 0 if state.get("status") == "completed" else 1 + + +def cmd_mermaid(args, api) -> int: + print(api["build_workflow"]().get_graph().draw_mermaid()) + return 0 + + +def cmd_schema(args, api) -> int: + print("subagent result:") + print(json.dumps(RESULT_DEFAULTS, ensure_ascii=False, indent=2)) + print("\nhuman review decision:") + print(json.dumps(REVIEW_DEFAULTS, ensure_ascii=False, indent=2)) + return 0 + + +def main(argv: list[str] | None = None) -> int: + try: + sys.stdout.reconfigure(encoding="utf-8") + except Exception: # noqa: BLE001 - older streams + pass + + parser = argparse.ArgumentParser(prog="run_sim_rtl_flow") + parser.add_argument("--project", type=Path, default=None) + parser.add_argument( + "--work-dir", + default=None, + help=( + "the design project whose .sim_rtl_flow/ holds run state; " + "defaults to cwd (on `start` it comes from the input's work_dir)" + ), + ) + parser.add_argument("--thread-id", default="default") + parser.add_argument("--recursion-limit", type=int, default=200) + parser.add_argument("--max-attempts", type=int, default=10) + sub = parser.add_subparsers(dest="command", required=True) + + p = sub.add_parser("start") + p.add_argument("--input", required=True) + p.add_argument("--keep-ledger", action="store_true") + p.set_defaults(func=cmd_start) + + p = sub.add_parser("resume") + p.set_defaults(func=cmd_resume) + + p = sub.add_parser("record") + p.add_argument("--key") + p.add_argument("--result") + p.add_argument("--result-file") + p.add_argument("--approve", action="store_true") + p.add_argument("--reject", action="store_true") + p.add_argument("--notes") + p.add_argument("--all-ok", action="store_true") + p.set_defaults(func=cmd_record) + + p = sub.add_parser("status") + p.set_defaults(func=cmd_status) + + p = sub.add_parser("dry-run") + p.add_argument("--input", required=True) + p.set_defaults(func=cmd_dry_run) + + p = sub.add_parser("mermaid") + p.set_defaults(func=cmd_mermaid) + + p = sub.add_parser("schema") + p.set_defaults(func=cmd_schema) + + args = parser.parse_args(argv) + if getattr(args, "approve", False) and getattr(args, "reject", False): + raise SystemExit("--approve 与 --reject 不能同时使用") + if args.project is None: + args.project = _default_project() + api = _bootstrap(args.project) + return args.func(args, api) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/zynq-imx6-flow/SKILL.md b/zynq-imx6-flow/SKILL.md new file mode 100644 index 0000000..5b04f7d --- /dev/null +++ b/zynq-imx6-flow/SKILL.md @@ -0,0 +1,199 @@ +--- +name: zynq-imx6-flow +description: > + Run the zynq-imx6-flow LangGraph workflow at E:\LangGraph_Design\zynq-imx6-flow — + the closed-loop xmod (Zynq PS+PL + i.MX6) hardware-in-the-loop design/debug pipeline: + cold power-cycle, connect JTAG, program bitstream/ELF or burn flash, restart the + ps/imx6 app, analyze (ILA + PS/imx6 serial consoles), judge against design + requirements, and iterate by modifying RTL / imx6 app / PS baremetal code. It drives + the graph in-process (no langgraph dev) and hands the 6 interrupts (analyze / + meets_requirement / propose / modify_fpga / modify_imx6 / modify_ps) back to this + Claude session to execute. Also supports zynq-only boards (no i.MX6) and PS + baremetal/petalinux modes. + + ALWAYS use this skill whenever the user wants to run, start, drive, resume, or + continue the zynq-imx6-flow workflow / xmod Zynq+imx6 design loop, or says things + like "启动 zynq-imx6-flow", "跑 zynq 设计流程", "run the zynq FPGA flow", "执行 xmod + 设计任务", "继续 zynq 工作流", "resume the zynq design loop", even if they don't name + the project path. Before driving, this skill confirms the langgraph conda env from + /langgraph-vivado-install is in place. +--- + +## What this skill does + +Drives the **zynq-imx6-flow** graph (28 nodes / 39 flow-edges, 1:1 with +`Notes/zynq-imx6-flow.excalidraw`, plan `Notes/Task00.md`). The deterministic +hardware/check nodes (power, JTAG, program, restart) run in-process; the **6 interrupt +nodes** pause and hand control to **this Claude session**, which does the real work +(analysis, judgment, RTL/app/PS edits) and resumes the same thread. Persistence is a +SqliteSaver file, so `start`/`resume` survive across separate process calls. + +| Item | Value | +|---|---| +| Project root | 见 `~\skills_env.md` 的 `ZYNQ_IMX6_FLOW_ROOT`（driver 自动读取，也可 `--project-dir` 覆盖） | +| Conda env | `langgraph` (path 见 `~\skills_env.md` 的 `CONDA_LANGGRAPH_PYTHON`；装环境用 `/langgraph-vivado-install`) | +| Graph | `app.main_graph:build_graph` (in-process; no langgraph dev) | +| Driver | `%USERPROFILE%\.claude\skills\zynq-imx6-flow\scripts\run_zynq_flow.py` | +| Checkpoint | `\runs\skill_checkpoint.sqlite` (keyed by `--thread-id`) | + +**Drive the driver with the Bash tool (Git Bash), not PowerShell.** Windows PowerShell +5.1 strips the double-quotes out of `--resume-json` payloads (see Gotchas), corrupting +the JSON. Git Bash single-quoting passes it intact. All commands below are Bash and were +run this way to verify them. + +```bash +PY="/c/Users/Administrator/miniconda3/envs/langgraph/python.exe" +RUN="/c/Users/Administrator/.claude/skills/zynq-imx6-flow/scripts/run_zynq_flow.py" +export PYTHONIOENCODING=utf-8 +``` + +## ⚠️ Hardware timing rule (power / USB-Blast) + +**Before every power change to the board, USB-Blast must already be disconnected; only +reconnect JTAG after power is up.** Never power the board up/down while USB-Blast is +connected (protects the JTAG adapter and board). The graph topology already enforces +this: `power_off → check_power_off →(是) blast_off_a → check_blast_off →(是) power_on → +check_power →(是) blast_on → check_hw`. After `flash_burn` the graph returns to +`power_off` for a full cold-restart before re-analyzing. Serial port numbers are **never +hardcoded** — they come from `~/zynq_imx6_flow.conf` (see "Mock vs real"). + +## Step 0 — Confirm the environment (REQUIRED before driving) + +The workflow runs in the `langgraph` conda env created by **`/langgraph-vivado-install`**. +Verify it first so a missing env is a clear message, not a cryptic mid-run import error: + +```bash +"$PY" -c "from langgraph.checkpoint.sqlite import SqliteSaver; from langgraph.types import Command; import langgraph; print('[OK] langgraph')" +``` + +- **`[OK]`** → proceed. +- **import fails / no python** → tell the user to run **`/langgraph-vivado-install`** + first (it creates the `langgraph` env with `langgraph` + `langgraph-checkpoint-sqlite`). + Do not proceed until it succeeds. + +Optional sanity check — the project's own acceptance suite (all mock, ~2 s): + +```bash +cd /e/LangGraph_Design/zynq-imx6-flow && "$PY" smoke_test.py # prints [PASS] +``` + +## Step 1 — Start the workflow + +Pick the board + first-pass program mode (defaults shown), then start on a fresh +`--thread-id`: + +```bash +"$PY" "$RUN" start --thread-id zynq- \ + --board-type xmod --ps-mode baremetal --program-mode bitstream --bit-path mock.bit +``` + +- `--board-type` `xmod` (Zynq+i.MX6, default) | `zynq_only` (no i.MX6 — imx6 branch is skipped). +- `--ps-mode` `baremetal` (default) | `petalinux`. +- `--program-mode` `bitstream` (default, downloads bit → restart → analyze) | + `debug_flash` (image already fused in flash → straight to analyze, no download). +- `--bit-path` / `--elf-path` / `--flash-path` optional artifact paths. +- Add `--real` only with hardware wired + board connected (default is MOCK — see below). + +The driver prints **one JSON line** and exits: + +| Exit | `status` | Meaning | +|---|---|---| +| 0 | `done` | Workflow finished (符合要求 / iteration cap) → Step 3 | +| 1 | `error` | Read `error`, diagnose, report | +| 2 | `interrupt` | An interrupt fired → read `interrupt.action` (and `pending_count`) → Step 2 | + +## Step 2 — Handle the interrupt(s), then resume + +On exit 2, `interrupt.action` (or the `pending_interrupts` list) tells you what to do. +Perform it, then resume the **same `--thread-id`**. Repeat until exit 0. + +| `action` | excalidraw node | What you (the session) do | Resume JSON | +|---|---|---|---| +| `analyze` | 通过ila/串口分析 | Grab ILA + the `sources` consoles (ps_console, and imx6_console if xmod); summarize | `{"analysis_log":"<摘要/路径>"}` | +| `meets_requirement` | 符合设计要求? | Judge analysis vs the design requirement | `{"meets": true}` or `{"meets": false}` | +| `propose` | 分析原因，提方案 | Diagnose; **review the plan with `/langgraph-plan-review`**; decide what to change + next program mode | `{"reason":"...","need_fpga":bool,"need_imx6":bool,"need_ps":bool,"program_mode":"bitstream"或"flash"}` | +| `modify_fpga` | 改 RTL→综合→bit/flash | Edit RTL, run Vivado synth/impl, produce bit + flash | `{"bit_path":"...","flash_path":"..."}` | +| `modify_imx6` | VM 改 app→交叉编译 | In the Ubuntu VM edit imx6 app + cross-compile | `{"app_path":"..."}` | +| `modify_ps` | sdk 改 PS 裸机→ELF | In Xilinx SDK edit PS baremetal code + build (petalinux: rebuild app/image) | `{"elf_path":"..."}` | + +**Single interrupt** — resume with the payload directly: + +```bash +"$PY" "$RUN" resume --thread-id zynq- --resume-json '{"meets": false}' +``` + +**Three-way parallel fan-out** — after a `propose` that sets `need_fpga`/`need_imx6`/ +`need_ps` true, `modify_fpga` + `modify_imx6` + `modify_ps` fire **in one superstep** +(`pending_count: 3`). Resume them with **one map keyed by `action`** (subset allowed; +unresumed ones stay pending and the deferred `round_done` still waits for every branch): + +```bash +"$PY" "$RUN" resume --thread-id zynq- --resume-json \ + '{"modify_fpga":{"bit_path":"new.bit","flash_path":"new.bin"},"modify_imx6":{"app_path":"new-app"},"modify_ps":{"elf_path":"new.elf"}}' +``` + +Notes: +- Always reuse the **same `--thread-id`** from `start` — that's how the sqlite checkpoint + reattaches (different tasks → different thread ids → isolated state). +- `propose` fans out to `decide_fpga`/`decide_imx6`/`decide_ps`; all-false goes straight + to `round_done` → re-program. `board_type=zynq_only` forces `need_imx6=false`. +- `program_mode` from `propose` chooses the re-program path after `round_done`: + `bitstream` = download bit; `flash` = burn flash then full cold power/connect + re-sequence (`flash_burn → power_off`). This is the big closed loop. +- `"$PY" "$RUN" status --thread-id zynq-` re-prints the current state without advancing. + +## Step 3 — Finish + +On exit 0 (`status: done`), report iterations completed and whether the design met +requirements (`meets_requirement`) or hit the iteration cap. Offer to inspect +`\runs\` artifacts or the per-step `history`. + +To abandon a paused run and start clean, delete `\runs\skill_checkpoint.sqlite` +or use a fresh `--thread-id`. + +## Mock vs real + +- **MOCK (default)** — sets `ZYNQ_IMX6_MOCK=1`; hardware/check nodes return success + immediately. Dry-runs the *design-loop logic* and all 6 interrupts with no board. Ideal + for demonstrating/validating the flow. Every command in this doc was verified in mock. +- **Real (`--real`)** — sets `ZYNQ_IMX6_MOCK=0` and drives actual hardware. The `tools_hw` + paths are wired (Modbus power relay `0x05`, USB-Blast `0xA0` frames over pyserial). Real + mode additionally requires: + 1. **pyserial** in the env (`"$PY" -m pip install pyserial`) — not installed by default; + without it `--real` returns `{"status":"error","error":"Real mode requires pyserial"}`. + 2. **`~/zynq_imx6_flow.conf`** with 4 sections `[power] [usb_blast] [imx6_console] + [ps_console]` (template: project's `zynq_imx6_flow.example.conf`). Relay ports + (`power`/`usb_blast`) may fall back to `~/vivado_mcp_usb.conf`. `zynq_only` boards + don't need `[imx6_console]`. COM numbers differ per machine — find them in Device + Manager → Ports (COM & LPT). + 3. A connected board + Vivado for the JTAG program/flash nodes. + +## Gotchas + +- **PowerShell 5.1 corrupts `--resume-json`.** `& $py $RUN resume --resume-json '{"meets": false}'` + arrives at Python as `{meets: false}` — PS strips the double-quotes (and `\"`-escaping + splits it on spaces instead). This is the known PSNativeCommandArgumentPassing bug, + unfixed in 5.1. **Use the Bash tool.** Verified: the same JSON through Git Bash arrives + intact (`analysis_log` values with spaces round-trip correctly). +- **`round_done` is a deferred node** (`defer=True`). With the three-way fan-out it runs + **exactly once** after all active modify/deploy branches finish (verified: iteration + advances 0→1 once even with all three branches on). Don't expect it after only the + first branch. +- **`propose` must return `program_mode` ∈ {bitstream, flash}.** Any other value raises + `propose returned invalid program_mode`. (The `start` `--program-mode` has a third + option `debug_flash` for the first pass only; `propose` does not.) +- **`langgraph.__version__` doesn't exist** on 1.2.9 — check the version via + `importlib.metadata.version("langgraph")` if needed (smoke_test asserts `>= 0.2.39`). + +## Troubleshooting + +| Symptom | Fix | +|---|---| +| `--resume-json` value shows up without quotes / split on spaces | You used PowerShell. Drive via the Bash tool (single-quoted JSON). | +| `UnicodeEncodeError: 'gbk' ...` | Don't use bare `conda run`; use the env `python.exe` directly (the driver reconfigures stdout to UTF-8). | +| `{"status":"error","error":"Real mode requires pyserial"}` | `--real` needs pyserial: `"$PY" -m pip install pyserial`. Or drop `--real` for a mock run. | +| `SerialConfigError: Missing serial sections [...]` | Create `~/zynq_imx6_flow.conf` from `zynq_imx6_flow.example.conf` (relay ports may come from `~/vivado_mcp_usb.conf`). | +| `import failed ... Run /langgraph-vivado-install` | The `langgraph` env is missing/incomplete → run `/langgraph-vivado-install`. | +| `propose returned invalid program_mode` | The `propose` resume used something other than `bitstream`/`flash`. | +| `resume` seems to start over | You passed a different `--thread-id` than `start`. Use the same one. | +| Want to see the graph visually | Optional: `langgraph dev` via the project's `langgraph.json` opens Studio — not required to run. | diff --git a/zynq-imx6-flow/scripts/run_zynq_flow.py b/zynq-imx6-flow/scripts/run_zynq_flow.py new file mode 100644 index 0000000..020ca15 --- /dev/null +++ b/zynq-imx6-flow/scripts/run_zynq_flow.py @@ -0,0 +1,187 @@ +"""In-process driver for the zynq-imx6-flow LangGraph workflow. + +Builds the compiled graph from the project's ``app`` package and drives it with a +persistent SqliteSaver — no ``langgraph dev`` server. Mirrors the sibling +fpga-imx6-flow driver: exit code 0 = done, 1 = error, 2 = interrupt fired. + +Commands: + start --project-dir [--program-mode bitstream|debug_flash] + [--ps-mode baremetal|petalinux] [--board-type xmod|zynq_only] + [--bit-path ..] [--elf-path ..] [--flash-path ..] + [--thread-id ..] [--real] [--max-iterations N] + resume --project-dir --resume-json '' [--thread-id ..] [--real] + status --project-dir [--thread-id ..] + +Output: one JSON line on stdout describing status + the pending interrupt(s). + +Resume payloads: + * Single pending interrupt -> pass its payload directly, e.g. '{"meets": true}'. + * Multiple pending (the three-way modify fan-out: modify_fpga + modify_imx6 + + modify_ps can fire in one superstep) -> pass a map keyed by each interrupt's + "action", e.g. '{"modify_fpga": {...}, "modify_imx6": {...}, "modify_ps": {...}}'. + You may resume a subset; unresumed interrupts stay pending and round_done + (defer=True) still waits for every active branch. + +Run via the env python directly or ``conda run --no-capture-output`` to avoid the +Windows GBK console re-encode crash on the Chinese interrupt payloads. +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from pathlib import Path + +try: + sys.stdout.reconfigure(encoding="utf-8") # emit UTF-8 even on a GBK console +except Exception: + pass + + +def _env_lookup(key: str, default: str | None = None) -> str | None: + """Read KEY from the OS environment, else from a ``KEY = value`` line in + ~/skills_env.md. Machine-local paths are never hardcoded in a skill; see + the same helper in fpga_flash/scripts/vivado_env.py.""" + value = os.environ.get(key) + if value: + return value + path = os.path.join(os.path.expanduser("~"), "skills_env.md") + try: + with open(path, encoding="utf-8") as handle: + for line in handle: + match = re.match(rf"\s*{re.escape(key)}\s*=\s*(.+?)\s*(?:#.*)?$", line) + if match: + found = match.group(1).strip() + if found and not found.startswith("<"): + return found + except OSError: + pass + return default + + +#: The zynq-imx6-flow checkout: ZYNQ_IMX6_FLOW_ROOT in ~/skills_env.md. +DEFAULT_PROJECT = _env_lookup("ZYNQ_IMX6_FLOW_ROOT") + + +def _emit(obj: dict, code: int) -> None: + print(json.dumps(obj, ensure_ascii=False)) + sys.exit(code) + + +def _pending(snap): + return [intr for task in snap.tasks for intr in task.interrupts] + + +def main() -> None: + ap = argparse.ArgumentParser(description="Drive the zynq-imx6-flow graph in-process.") + ap.add_argument("command", choices=["start", "resume", "status"]) + ap.add_argument("--project-dir", default=DEFAULT_PROJECT) + ap.add_argument("--thread-id", default="zynq-skill-1") + ap.add_argument("--resume-json", default="") + ap.add_argument("--program-mode", default="bitstream", + choices=["bitstream", "debug_flash"]) + ap.add_argument("--ps-mode", default="baremetal", choices=["baremetal", "petalinux"]) + ap.add_argument("--board-type", default="xmod", choices=["xmod", "zynq_only"]) + ap.add_argument("--bit-path", default="") + ap.add_argument("--elf-path", default="") + ap.add_argument("--flash-path", default="") + ap.add_argument("--real", action="store_true", + help="drive real hardware (tools_* must be wired; default is MOCK dry-run)") + ap.add_argument("--max-iterations", type=int, default=8) + args = ap.parse_args() + + if not args.project_dir: + _emit({"status": "error", + "error": "zynq-imx6-flow 仓库路径未配置：在 ~/skills_env.md 增加 " + "`ZYNQ_IMX6_FLOW_ROOT = <检出目录>`，或用 --project-dir 指定"}, 1) + project = Path(args.project_dir) + if not (project / "app" / "main_graph.py").exists(): + _emit({"status": "error", + "error": f"app/main_graph.py not found under {project}; check --project-dir"}, 1) + + # MOCK by default so the design loop can be dry-run with no hardware attached. + os.environ["ZYNQ_IMX6_MOCK"] = "0" if args.real else "1" + sys.path.insert(0, str(project)) + + try: + from langgraph.checkpoint.sqlite import SqliteSaver + from langgraph.types import Command + from app.config import initial_state + from app.main_graph import build_graph + except Exception as exc: # noqa: BLE001 + _emit({"status": "error", + "error": f"import failed: {exc}. Run /langgraph-vivado-install to set up the env."}, 1) + + runs = project / "runs" + runs.mkdir(exist_ok=True) + db = str(runs / "skill_checkpoint.sqlite") + cfg = {"configurable": {"thread_id": args.thread_id}} + + with SqliteSaver.from_conn_string(db) as cp: + graph = build_graph(checkpointer=cp) + try: + if args.command == "start": + init = initial_state( + program_mode=args.program_mode, + ps_mode=args.ps_mode, + board_type=args.board_type, + max_iterations=args.max_iterations, + ) + if args.bit_path: + init["bit_path"] = args.bit_path + if args.elf_path: + init["elf_path"] = args.elf_path + if args.flash_path: + init["flash_path"] = args.flash_path + graph.invoke(init, cfg) + elif args.command == "resume": + payload = json.loads(args.resume_json) if args.resume_json else {} + pend = _pending(graph.get_state(cfg)) + if len(pend) <= 1: + graph.invoke(Command(resume=payload), cfg) + else: + # Multiple interrupts pending (three-way modify fan-out). payload + # must be a map keyed by each interrupt's "action" (or its id). + resume_map = {} + for intr in pend: + val = intr.value if isinstance(intr.value, dict) else {} + action = val.get("action") + if action in payload: + resume_map[intr.id] = payload[action] + elif intr.id in payload: + resume_map[intr.id] = payload[intr.id] + if not resume_map: + _emit({"status": "error", + "error": f"resume-json keys {list(payload)} matched none of the " + f"pending actions {[i.value.get('action') for i in pend]}"}, 1) + graph.invoke(Command(resume=resume_map), cfg) + snap = graph.get_state(cfg) + except Exception as exc: # noqa: BLE001 + _emit({"status": "error", "error": str(exc)}, 1) + + pend = _pending(snap) + if snap.next: + _emit({ + "status": "interrupt", + "thread_id": args.thread_id, + "mock": not args.real, + "next": list(snap.next), + "iteration": snap.values.get("iteration"), + "interrupt": pend[0].value if pend else None, + "pending_interrupts": [i.value for i in pend], + "pending_count": len(pend), + }, 2) + else: + _emit({ + "status": "done", + "thread_id": args.thread_id, + "iteration": snap.values.get("iteration"), + "meets_requirement": snap.values.get("meets_requirement"), + "steps": len(snap.values.get("history", [])), + }, 0) + + +if __name__ == "__main__": + main()