bowei/skills-win
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: SHANGHAI-U9 2026-07-27

Browse Source
This commit is contained in:
bowei
2026-07-27 01:08:42 +08:00
parent 17632bea27
commit 8080703a0e
5 changed files with 1385 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
langgraph-plan-review/scripts/agents.py
+5
View File
@@ -30,6 +30,11 @@ def _run_command(command: Sequence[str], values: Mapping[str, str], timeout: int
capture_output=True, capture_output=True,
check=False, check=False,
text=True, 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, timeout=timeout,
) )
elapsed = time.monotonic() - t0 elapsed = time.monotonic() - t0
sim-rtl-flow/SKILL.md
+314
View File
@@ -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** | **`<work_dir>\.sim_rtl_flow\runs\<thread-id>\`** — 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="<CONDA_LANGGRAPH_PYTHON — see ~\skills_env.md>"
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 <proj> --thread-id <id> record --key review:sim_performance#1 --approve
"$PY" "$RUN" --work-dir <proj> --thread-id <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 (`<project>_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
`<work_dir>/.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 — `<name>-x.slx`, `<name>_A-E.slx`, `<name>.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:<M>#n` | nativize M into HDL-capable blocks, save `<project>_M-E.slx`, generate RTL | request says `simulink-rtl-dev` — **not installed**; use MATLAB MCP + model-based-design-core |
| `ip_replace:<M>#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:<M>#1` per module (3 for `["A","B","C"]`) |
| next module stage | one `ip_replace:<M>#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 `<project_name>-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**
(`<SIM_RTL_FLOW_ROOT>\runs\<id>\`) 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\<id>` | 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 "<user's reason>"` |
| `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` |
sim-rtl-flow/scripts/run_sim_rtl_flow.py
+680
View File
@@ -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 ``<work_dir>/.sim_rtl_flow/`` -- i.e. inside the
project being designed, never inside the sim-rtl-flow repository (Notes/PR01-2.md).
Commands
--------
start --input <json> [--thread-id ID] begin a run
resume continue after record
record --key K --result '<json>' 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 <json> 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 = <sim-rtl-flow 检出目录>\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:
"""``<work_dir>/.sim_rtl_flow/runs/<thread-id>/`` -- 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 <file>` 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 '<json>' (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())
zynq-imx6-flow/SKILL.md
+199
View File
@@ -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 | `<project>\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-<short-task-name> \
--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-<name> --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-<name> --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-<name>` 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
`<project>\runs\` artifacts or the per-step `history`.
To abandon a paused run and start clean, delete `<project>\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. |
zynq-imx6-flow/scripts/run_zynq_flow.py
+187
View File
@@ -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 <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 <dir> --resume-json '<json>' [--thread-id ..] [--real]
status --project-dir <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()