diff --git a/docs/maintainers/MARCH_FEEL_AND_READABILITY.md b/docs/maintainers/MARCH_FEEL_AND_READABILITY.md new file mode 100644 index 000000000..9c91b96c9 --- /dev/null +++ b/docs/maintainers/MARCH_FEEL_AND_READABILITY.md @@ -0,0 +1,188 @@ +# learn-claude-code：章节可读性与文风恢复说明（给维护者） + +**读者**：本仓库维护者 / 章节作者 / Reviewer +**基准对照**：2026-03-29 `16b927c`（12 课 + `docs/zh` 心智模型短章）vs 当前 `main`（17 课根目录） +**诉求来源**：课程创始人侧反馈——当前中后章阅读感、创作感、上头感明显变差；版式固定、AI 味、行文尴尬、难读难懂。 +**状态**：本文为写作与改写的权威约束。机制正确性仍要守；**文风与信息优先级以本文为准**。 + +--- + +## 1. 一句话 + +请把章节从「工程规格书 + AI 润色课包」拉回「短讲义卡片」： + +> **问题砸人 → 一张图/ASCII → 最小代码 → 试一试** + +深度可以保留，但必须住在折叠/附录；不能占首屏和主呼吸。 + +--- + +## 2. 客观诊断（不是口味吵架） + +### 2.1 分层结论 + +| 范围 | 阅读感 | 创作感 | 上头感 | 说明 | +|------|------:|------:|------:|------| +| 2026-03 短章 | ~8.5 | ~8 | ~8 | 短、具体、对「你」说话 | +| main 早章（约 s01–s07） | ~7 | ~6 | ~6.5 | 仍可读，已被模板镀铬 | +| main 晚章（s08/s13/s15/s16…） | ~3.5 | ~2.5 | ~2 | 手册化 / 抽象词 / 标题墙 | + +### 2.2 关键数字（main 中文 README） + +- 三月章均约 **4KB / 5 个 H2**；main 中位约 **6.9KB**，重灾章 **11–18KB**。 +- 分水岭在 **s08**：此前多为「镀铬但仍短」，此后跳变。 +- 「你」：s01 仍在；多数中后章 **≈0**。 +- 全部 17 章：双语导航 + `translation-sync`；**`
` 使用率曾长期为 0**（深度只能堆主文）。 +- 重灾优先序（分诊）：**s13 → s16 → s08 → s15 → s10**（s17/s09 紧随）。 + +### 2.3 根因 Top 5（不是「单纯更长」） + +1. **声音从「对你说话」变成系统旁白** +2. **taxonomy / 边界表 / 组件目录压过洞见** +3. **模板铬 + 三语文案生产线**（章感雷同） +4. **工程词典取代叙事动词**（宿主/registry/生命周期/语义 key…） +5. **覆盖焦虑**把教学文写成 runbook + +### 2.4 高 AI 味的机械尾槽（特别刺眼） + +章章同一套收尾流水线会让人一眼看出「流水线产物」： + +| 槽位 | 机械套路 | 读感 | 要求 | +|------|----------|------|------| +| 试一下 | cd → 编号 prompt →「观察重点：是否…是否…」 | QA/CI checklist | 命令 + ≤3 prompt；**禁止观察重点勾选清单** | +| 接下来 | 「现在能 X 了。但 Y 又爆了。」 | 万能悬念工厂 | **可选**；≤3 句且换写法；可整段删 | +| 相对 sN | 组件/之前/之后大表 | PR 变更表塞进教材 | **默认撤出主文**；改一句能力增量或进 `
` | + +三月同题常在「试一试」后**戛然而止**——没有「接下来」，没有观察审问；更像人写的。 + +### 2.5 同题对比（摘录） + +**三月 s01** +> 没有循环, 每次工具调用你都得手动把结果粘回去。**你自己就是那个循环。** + +**main s06（同主题后继）** +> …多数中间细节不再需要，却仍然占用上下文。 +（正确、完整、无趣；「pytest 一个词」那种 punch 没了。） + +**main s08** +> **本节将实现一条四步压缩管线。** +（预告腔直接杀死好奇心。） + +**main s15** +开篇 9 条「需要同时拥有」功能 backlog +「组件在循环中的位置」大表 + 观察重点 8 条——组件目录，不是故事。 + +**main s16** +抽象工程词命中可到数十上百；开篇像架构演进史，不像痛点。 + +--- + +## 3. 写作要求（必须遵守） + +### 3.1 主文硬指标（不含 `
`） + +| 指标 | 达标 | 重灾线 | +|------|------|--------| +| 字节 | ≤7KB | >10KB | +| 行数 | ≤180 | >260 | +| H2 | ≤7 | >10 | +| 主文代码围栏 | ≤4 | >8 | +| 主文表格 | ≤1 | ≥3 | +| 「你」（问题/试跑） | ≥1 | 全程无人称 | +| 「本节将/本章将」 | 0 | ≥1 | +| 抽象工程词* | ≤15 | >40 | +| 首屏 | 能看到「问题」/痛点 | 只见导航或目录 | + +\*词表示例：管线|拓扑|适配器|原语|宿主|生命周期|registry|schema|journal|元数据|编排|语义 + +图片：主路径 0–2 张；ASCII 能讲清就不上大图；次要图进折叠。 + +### 3.2 推荐骨架 + +```text +标题（短、可记） +一行语言切换（若课程需要）+ 一行面包屑 +格言 1 句 + Harness 层 1 行（要狠） + +## 问题 ← 2–5 句；有「你」或可感场景；禁止功能清单开场 +## 解决方案 ← 1 ASCII/1 关键图 + 2–4 句 +## 工作原理 ← ≤4 步；人话 → 再术语；短代码 +## 试一试 ← 命令 + ≤3 prompt；禁止观察重点清单 +## 接下来（可选） + +
…深度、模式库、边界、相对前章细表…
+``` + +复杂章只保证主文讲清「**这一章只加一件东西**」。 + +### 3.3 声音 + +- 先洞见，后术语。 +- 每章至少一句 punch（删掉它章就塌）。 +- 隐喻最多开篇一小段；禁止全章跟隐喻跑。 +- 中文像人讲，不要英译说明书。 +- 90 秒说不清「只加了一件东西」→ 再砍。 + +### 3.4 机制 fidelity（教学可简化，不可说错） + +以各章真实教学代码与上游产品契约为准。以 s16 为例必须诚实： + +- Dynamic = 模型写脚本（Claude Code：`script` / `scriptPath`）；Saved = `name` + `args` +- 不得再暗示「模型不能提交可执行代码」仿佛是产品事实 +- `parallel`/`pipeline` 失败隔离为 null；resume = 最长未改前缀 +- 说明为何真 JS runtime 忌 `Date.now` / `Math.random` +- 本章若是 Python 教学 runtime，要标明：思想对齐，不是 bit-perfect 复刻 + +思想脊梁可参考官方文 +[A harness for every task: dynamic workflows in Claude Code](https://claude.com/blog/a-harness-for-every-task-dynamic-workflows-in-claude-code)。 + +### 3.5 Reviewer 清单 + +- [ ] 首屏出现痛点 +- [ ] 主文硬指标达标 +- [ ] 有 punch；格言与正文咬合 +- [ ] 对「你」说话 +- [ ] 无「本节将」、无标题墙、无主文组件目录 +- [ ] 无观察重点勾选、无万能「接下来」、无主文「相对 sN」大表 +- [ ] 深度在 `
` +- [ ] GitHub 预览像讲义，不像 API 手册 +- [ ] 机制正确 + +--- + +## 4. 建议改写队列 + +1. **P1 重灾**：s13_agent_teams，s16_workflow_runtime，s08_context_compact，s15_integrated_harness +2. **P2 手册化**：s10_task_system，s17_goal_loop，s09_memory，s14_mcp_plugin，s11… +3. **P4 镀铬早章**：少动结构；去预告腔、补 punch、减轻尾槽机械感 +4. **全局政策**：主文短 + `
` 分层；尾槽去流水线化 + +s16 无三月祖先：叙事节奏应对标 **s01/s06 三月短章**，不要对标 s13/s15 说明书骨架。 + +--- + +## 5. 证据与附件（工作区） + +| 文件 | 内容 | +|------|------| +| `MARCH_FEEL_RECOVERY_PLAYBOOK.md` | 恢复标准细则 | +| `DEEP_READABILITY_WHY_BAD.md` | 文风深挖与打分 | +| `MAIN_CHAPTER_TRIAGE.md` | 17 章分诊表 | +| `TEMPLATE_SLOTS_AI_SMELL.md` | 试一下/接下来/相对前章并置 | +| `MARCH_VS_MAIN_READABILITY.md` | 量化简报 | +| `AGENT_WRITING_BRIEF.md` | s16 思想/fidelity 总纲 | +| `/workspace/lcc_shots2/*.png` | GitHub 渲染截图对照 | + +三月对照 commit：`16b927c8ee7befa07caf8844d22f86ffef0aea05`。 + +--- + +## 6. 非目标 + +- 不是要求删掉三语支持或测试。 +- 不是要求章节变浅、变错。 +- 不是要求每章都用同一套「精修散文」或同一套隐喻。 +- 是要求：**可读、有创作棱角、想翻下一章**；正确性放在正确的信息层级里。 + +--- + +*维护者文档版本：2026-08-12* diff --git a/s16_workflow_runtime/README.ja.md b/s16_workflow_runtime/README.ja.md index 1fcb6e9b3..f7c0a8118 100644 --- a/s16_workflow_runtime/README.ja.md +++ b/s16_workflow_runtime/README.ja.md @@ -1,245 +1,139 @@ -# s16: Workflow Runtime — モデルが単一 step を決め、script が orchestration を決める +# s16: Workflow Runtime — オーケストレーションをコードに書く [English](README.md) · [中文](README.zh.md) · [日本語](README.ja.md) -s01 → ... → s14 → [s15](../s15_integrated_harness/) → `s16` → [s17](../s17_goal_loop/) +[s15](../s15_integrated_harness/) → `s16` → [s17](../s17_goal_loop/) -> *「1 回の tool_use で、一式の orchestration を実行する」* — `Workflow` ツールが復元可能な script runtime を起動し、多数の agent call を協調させます。 +> *計画をチャットの中だけに置かない。* 順番はスクリプトが持ち、一歩ごとの判断はモデルが持つ。 > -> **Harness 層**: Orchestration — single-agent loop の上で保存済み multi-agent script を実行します。 +> **Harness 層**: オーケストレーション — 単一 agent ループの上で、マルチ agent スクリプトを走らせる。 ---- +## 問題 -s01 から s15 まで、各 round で model が呼び出す tools を決めます。tool results が `messages[]` に入ると、model は更新された context から次の step を決めます。次の経路が前の step の発見に依存する task に向いています。 +あなたはすでに、ひとつのループの中でモデルにファイルを読ませ、コードを直し、エラーを見させることができます。ところが仕事によっては、**順番が最初から分かっている**ものがあります。次元ごとにレビューし、次に意地悪な検証、最後にまとめる——という具合です。その順番をチャットの記憶にだけ預けていると、モデルは途中で「完了」と言い、自分の宿題を甘く採点し、圧縮を何度か経ると「X を触るな」さえ消えます。 -一方、固定された流れを繰り返す task もあります。code review なら、複数の観点を同時に調べ、各 finding を検証し、重複をまとめて severity 順に並べます。実行前に step と順序が分かっている場合、host には次の 3 つが必要です。 +柔らかい会話は、並列も、結果の形の安定も、落ちてからの再開も支えきれません。もっとおしゃべりの上手なモデルが欲しいのではありません。**書き下ろされたオーケストレーション**が欲しいのです。 -- **並行性**: 1 件ずつ順番に待たないこと。 -- **安定した結果構造**: 個々の agent answer が変わっても構造を保つこと。 -- **復元可能性**: 途中で止まっても、完了済みの部分を最初からやり直さないこと。 +## 解決策 -この orchestration が conversation history にしか存在しなければ、順序と checkpoint も history にしか残りません。saved workflow は固定 flow を code に置き、完了した call を journal に記録します。 - -## 計画は chat のラウンドを重ねず、コードに書く - -harness の tool pool に `Workflow` ツールを追加します。host は `agent() / parallel() / pipeline() / phase()` で構成した trusted script を登録します。model が渡すのは saved workflow name、argument、任意の resume run ID だけで、実行可能 code や metadata は渡しません。 - -workflow は 1 回の `tool_use` として main loop に入ります。script の実行中、runtime は lifecycle event と progress event を出し、各 step を disk journal へ記録します。script が終わると、この call は launch 情報、result、task state を返します。script の中間結果は変数に保存され、conversation history を使いません。`resume_from_run_id` で再開すると、変更されていない `agent()` は journal の結果を再利用します。 - -![Workflow Runtime Overview](images/workflow-runtime-overview.svg) - -```python -SAMPLE_META = {"name": "review-changes", "description": "コード変更を review", "phases": ["Review", "Verify"]} - -async def sample_workflow(ctx, args): - ctx.phase("Review") - results = await ctx.pipeline(DIMENSIONS, audit, verify) # 各 dimension が独立して audit → verify を通る - confirmed = [f for r in results if r for f in r["confirmed"]] - ctx.log(f"{len(confirmed)} 件の実在する問題を確認") - return {"confirmed": confirmed} +```text + あなたの会話 ──► Workflow(...) ──► 結果が一条で戻る + │ + ▼ + スクリプト: agent / pipeline / parallel + │ + ▼ + 変数 + journal（半成品はここに。スレに詰め込まない） ``` -## Workflow ツール: 1 回の call で run 全体を実行する - -`Workflow` は s15 host の既存 tool pool に追加されます。ユーザーが保存済み workflow の実行を求めるか、タスクが既知の orchestration に一致したときにモデルがこのツールを選びます。adapter は name を host-owned `WORKFLOWS` registry で解決し、trusted metadata と function を runtime へ渡します。s15 の他の tools も同じ loop で利用できます。 +ヘルパー（サブ agent）は相変わらず考えます。**スクリプト**がループ・分配・マージを持ちます。中間結果は変数と journal に置き、親の対話には入れません。 -model-facing schema が受け取るのは `name`、`args`、`resume_from_run_id` です。unknown name や不正 argument は error tool result として返し、host loop を終了させません。その後 runtime が登録済み metadata を検証し、permission check を通し、local workflow task を登録して、script の実行前に `async_launched` を出します。progress event と最後の `task_notification` が続き、call は JSON-safe な launch 情報、result、task state を返します。 +一言でいうと：**オーケストレーションを「知性」から「構造」へ移す。** -```python -WORKFLOW_TOOL = { - "name": "Workflow", - "input_schema": { - "type": "object", - "properties": { - "name": {"type": "string"}, - "args": {"type": "object"}, - "resume_from_run_id": {"type": "string"}, - }, - "required": ["name"], - "additionalProperties": False, - }, -} - -async def run_workflow(name, args=None, resume_from_run_id=None): - meta, script_fn = WORKFLOWS[name] - out = await WorkflowTool().call( - meta, script_fn, - args=args, - resume_from_run_id=resume_from_run_id, - ) - return {"launched": out["launched"], "result": out["result"], - "task": serialize_task(out["task"])} -``` +![静的 harness と動的 workflow](images/dynamic-vs-static.png) -## Workflow metadata: 起動前に検証する +*左：汎用の固定パイプライン。右：このタスク向けに裁断したオーケストレーション。* -各 saved workflow は `name`、`description`、任意の `phases` を持つ trusted metadata を登録します。runtime は workflow code を実行する前に検証します。`name` と `description` は task と UI の表示に使い、`phases` は progress 表示の group 名を定義します。これらは model input ではなく host registry に属します。 +Claude Code には二つの扉があります。**動的**——モデルがこのタスク用に JS を書く（`script` / `scriptPath`）。**保存済み**——良いスクリプトを `name` + `args` で再実行。外側には SDK で先に書き切る静的オーケストレーションもあります。本章は **Python の教材用 runtime**（JS VM なし）です。考えは揃え、デモは「保存済み」の扉を使います。製品ではモデルはスクリプトを出せます——ここでは JS を走らせないだけです。 -不正な登録内容は launch 前に `WorkflowInputError` になります。s12 の cron 式検証と同じ考えです。不正な saved workflow が実行時まで進んでから壊れないようにします。 +## 仕組み -runtime は `meta.name` をローカル artifact のファイル名に使うため、英数字で始まり、英数字、`.`、`_`、`-` のみからなる 1-64 文字の安全な slug も要求する。 +**1. 三つの動詞** -```python -def validate_meta(meta): - if not isinstance(meta, dict): - raise WorkflowInputError("meta は object literal でなければなりません") - if not meta.get("name") or not meta.get("description"): - raise WorkflowInputError("meta には name と description が必要です") - if not isinstance(meta["name"], str) or not WORKFLOW_NAME_RE.fullmatch(meta["name"]): - raise WorkflowInputError("meta.name は安全な 1-64 文字の slug が必要です") - if "phases" in meta and ( - not isinstance(meta["phases"], list) - or not all(isinstance(p, str) and p for p in meta["phases"]) - ): - raise WorkflowInputError("meta.phases は空でない文字列だけを含む必要があります") - return meta +```text + agent 一人のヘルパー、一件の仕事（schema で次に渡せる JSON も可） + pipeline 各 item が自分の段階を進む（既定。同期しない） + parallel 全部揃ってから先へ（バリア。多用しない） ``` -## Orchestration primitive - -script は少数の orchestration primitive だけを公開する `ExecutionState` を受け取り、ファイルを直接読み書きせず、shell も実行しません。default の interactive mode では `agent()` を host と同じ real API client に接続し、各 workflow agent は arguments で渡された内容だけを読みます。`demo` と unit test は `MockAgentRunner` を使い、event と journal replay を繰り返し確認できるようにします。 +失敗時：`parallel` のその枠は `null`。`pipeline` はその item を捨てます。艦隊は丸ごと沈みません。マージ前にフィルタしてください。 -| Primitive | 役割 | -|------|------| -| `agent(prompt, {schema, label, phase})` | 1 つの subagent を派遣 | -| `parallel(thunks)` | **barrier**: すべての task を並行実行し、全結果が戻るまで待つ | -| `pipeline(items, *stages)` | 各 item を **barrier なし**で stage ごとに実行し、終わった item から先へ進める | -| `phase(title)` | 現在の progress phase を記録し、progress bar を更新 | -| `log(message)` | progress log を 1 行出力 | -| `workflow(name, args)` | nested sub-workflow（1 階層だけ） | +**2. 再開はノートで。チャット記憶ではない** -各 item が同じ stage を独立して通る場合は `pipeline` を使えます。item A が stage 3 にいる間、item B はまだ stage 1 かもしれません。次の処理が前の group の全結果を必要とする場合は `parallel` を使います。 +journal は `agent()` の**呼び出し順**で記帳します。再開は最長の未変更プレフィックスを再生し、最初の変更から先は全部ライブです。本番の JS runtime は `Date.now()` / `Math.random()` を禁じます——ノートがずれないように。教材スクリプトも決定的に書いてください。 -```python -async def pipeline(self, items, *stages): - async def run_item(item, idx): - value = item - for stage in stages: # 各 item がすべての stage を独立して完走 - value = await stage(value, item, idx) - return value - return await asyncio.gather(*[run_item(it, i) for i, it in enumerate(items)]) +```text + journal [A] [B] [C] [D] + 再開 命中 命中 ✂ ライブ ``` -## 構造化出力: Subagent に散文を返させない - -`agent({schema})` は、schema に一致する JSON object だけを返すよう workflow agent に要求します。runtime は結果を parse、validate し、不一致なら 1 回 retry します。下流コードは prose から field を取り出さず、object を受け取れます。 +**3. サンプル一つ：Fanout + Adversarial** -s05 では tool argument を全面的に信頼できないと説明しました。ここでは同じ教訓を逆向きに使います。subagent の出力も全面的には信頼できません。orchestration boundary で検証し、1 回 retry の機会を与え、不確実性を後続 flow の外へ止めます。 +`review-changes` は「一つのパターン」ではありません。**Fanout** の中に **Adversarial** が入ります。次元ごとに `pipeline(audit, verify)`、検証で `parallel` に意地悪させ、残った finding だけ残します。 -```python -run = await asyncio.to_thread(self.runner.run, prompt, schema, label) -result = run.value -if schema is not None: - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: # 1 回だけ注意して retry、それでも不正なら error - retry = await asyncio.to_thread( - self.runner.run, prompt + "\n\n有効な JSON を返してください。", schema, label - ) - result = retry.value - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: - raise WorkflowInputError(f"agent({{schema}}) の出力が不正です: {err}") +```text + correctness ── 監査 ── 検証 ──┐ + security ── 監査 ── 検証 ──┤── confirmed + performance ── 監査 ── 検証 ──┤ + style ── 監査 ── 検証 ──┘ ``` -## Task state と progress event - -`LocalWorkflowTask` は status と token usage を管理し、SDK style の event stream を外へ出します。`task_started` → phase change、subagent start、log を含む一連の `task_progress` → 完了または失敗に加え、output file、agent 数、token 数を含む最後の `task_notification` です。 - -demo はこれらの event を順番に表示し、最後の notification の後で task state を返します。 - ```python -class LocalWorkflowTask: - def progress_event(self, ptype, **data): # phase/subagent/log - self.progress.append({"type": ptype, **data}) - print(f" progress {ptype} ...") +# code.py 抜粋 — 形だけ見ればよい +async def sample_workflow(ctx, args): + ctx.phase("Review") + results = await ctx.pipeline(DIMENSIONS, audit, verify) + confirmed = [f for r in results if r for f in r["confirmed"]] + return {"confirmed": confirmed} ``` -## 保存: Snapshot + journal で中断から再開する +艦隊は早逃げできず、著者は自分の審判にならず、トポロジも疲れたチャットのたびに書き換わりません。 -runtime は各 run を `s16_workflow_runtime/.runtime/` に保存します。`.json` snapshot、`.output.json` output、`.journal.jsonl` journal、`.lock` coordination file です。fresh run は journal を開く前に exclusive file creation で新しい `runId` を予約します。run lock は実行と最終永続化が終わるまで保持するため、別 process は同じ run を同時に resume できません。snapshot に workflow name、arguments、task state を記録し、resume は保存済み snapshot と journal を先に検証してから、成功済み artifact を変更します。 +
+よくある六つの形（パターン庫） -journal は checkpoint resume の中心で、各 `agent()` の結果を 1 行ずつ記録します。 +![六種 Workflow モード](images/six-workflow-patterns.png) -```python -class WorkflowJournal: - def record(self, key, value): - self._f.write(json.dumps({"key": key, "value": value}) + "\n") - self._f.flush() - self.cache[key] = value -``` +| パターン | 人の言葉 | 原語のスケッチ | +|------|------|----------| +| Classify-And-Act | 仕分けしてから適任へ | `agent` → 分岐 → `agent` | +| Fanout-And-Synthesize | 分けてやり、またまとめる | `pipeline` / `parallel` → 統合 | +| Adversarial Verification | 狐に鶏小屋を採点させない | 産出 → `parallel(verify)` → フィルタ | +| Generate-And-Filter | まず多く作り、それから篩 | `parallel(gens)` → フィルタ | +| Tournament | 一対一で優勝を決める | 審判 `agent` | +| Loop Until Done | 「まだ新しい？」なら続ける | `while` + 停止 + `budget` | -## Resume: runId から続行し、変更のないものを再利用する +`review-changes` ≈ Fanout + Adversarial。調査系はよく 分配 → フィルタ → 検証 → 統合 と積みます。 -`resume_from_run_id` を渡して workflow を再度呼ぶと script を再実行しますが、各 `agent()` は決定的な semantic key を計算します。journal に key があれば、再実行せず cached result を返します。変更された call と、それに依存する後続 step だけが本当に動きます。 +
-key は concurrency の完了順に依存してはいけません。`parallel` と `pipeline` の Agent は不定の順番で完了します。「何番目に完了したか」を key にすると、次回の cache が別の call へ対応してしまいます。そのため key は競合する counter ではなく、call の内容、つまり type、label、prompt、schema の stable hash です。 +
+動的 / 保存済み / 静的と公式原語図 ```python -def key(self, kind, label, prompt, schema): - basis = f"{kind}|{label}|{prompt}|{json.dumps(schema, sort_keys=True)}" - return f"{kind}-{_stable_hash(basis) % 10**10:010d}" - -# agent() の内部: -cached = self.journal.cached(key) -if cached is not MISS: - self.task.progress_event("workflow_agent", label=label, status="cached") - return cached +# 教材スケッチ +Workflow({ "name": "review-changes", "args": { "changes": "..." } }) +# Claude Code はさらに: script | scriptPath | resumeFromRunId ``` -## Stable call key +![Workflow 原語](images/workflow-primitives.png) -resume では、現在の各 `agent()` call を以前の journal record と対応付ける必要があります。stable hash は変更されていない workflow code と arguments に同じ call key を与えます。real model の出力は変化しても、call 内容が同じなら journal に保存済みの result を使います。 +
-## 実際に動かす +
+信頼できない入力：読み書きを隔離 -sample workflow `review-changes` は `pipeline` を使い、各 review dimension を独立して audit → verify へ通します。interactive mode は real API を使い、`args.changes` から review 対象を読みます。`demo` は固定 runner data で pipeline、validation、journal、resume を示します。 - -```python -async def sample_workflow(ctx, args): - ctx.phase("Review") - changes = args.get("changes", "") - - async def audit(_v, dimension, _i): - out = await ctx.agent(f"この変更に {dimension} 関連の問題がないか確認してください:\n{changes}", - schema=FINDINGS_SCHEMA, label=f"audit:{dimension}", phase="Review") - return {"dimension": dimension, "findings": out["findings"]} - - async def verify(audited, dimension, _i): - ctx.phase("Verify") - verdicts = await ctx.parallel([ # 各 finding を独立して verify - (lambda f=f: ctx.agent(f"変更内容に照らして finding を検証してください:\n{changes}\n\n{f}", - schema=VERDICT_SCHEMA, label=f"verify:{dimension}:{f['title']}")) - for f in audited["findings"]]) - return {"dimension": dimension, - "confirmed": [f for f, v in zip(audited["findings"], verdicts) if v and v["isReal"]]} +チケットを読む agent が、同時に PR を開ける鍵を持ってはいけません。読み手は読むだけ → 要約。信頼側は要約だけ見て動きます。 - results = await ctx.pipeline(DIMENSIONS, audit, verify) - ... +```text + バックログ → [隔離: 読 / 重複除去 / 要約] → [信頼: 実行] ``` -## s15 からの変更点 +![隔離分流](images/quarantine-triage.png) -| | s15 Integrated Harness | s16 Workflow Runtime | -|--|-----------|---------------------| -| loop | 1 つ、モデル駆動 | main loop は不変。tool の背後で script orchestration を実行 | -| 次の step を決めるもの | モデルが毎ラウンド判断 | script が orchestration flow を事前に定義 | -| multi-agent | s06 subagent を一度だけ派遣 | agent-runner boundary を通る scripted、resumable call | -| 新しい仕組み | — | orchestration primitive、host registry と tool adapter、task lifecycle、progress event、journal/resume、structured output | +
-s16 は main loop を置き換えません。tool layer に `Workflow` を公開し、背後で local workflow runtime を起動します。saved script が agent-runner boundary を通じて N 回の call を協調させます。s06 の subagent はモデルがその場で 1 回派遣し、s16 は orchestration を resumable な host code にします。 +計画を握るのは誰か。s06 は一回きりの委譲、s13 はメール箱つきの仲間、s15 は単一ループのチャット、**s16 はスクリプト + journal**、s17 は入り口で「全体は終わったか」と聞きます。普通の数ファイルの修正なら s15 か一つの s06 で足りることが多い。Workflow は token と調整のコストが要ります——**構造が一度の会話より長生きしなければならない**ときだけ手を伸ばしてください。 ## 試してみる ```bash -python s16_workflow_runtime/code.py # main model と Workflow agent の両方が real API を使う -python s16_workflow_runtime/code.py demo # deterministic fixture と event stream を確認 -python s16_workflow_runtime/code.py resume # 前回の runId から resume。すべての agent() が journal cache に当たる +python s16_workflow_runtime/code.py demo +python s16_workflow_runtime/code.py resume ``` -default command では、model に changes を読ませ、その text を `args.changes` に入れて保存済み `review-changes` workflow を実行させます。main model と workflow agent の両方が real API を使います。`demo` は固定 runner data で lifecycle と resume を繰り返し観察でき、すべて cache hit した resume は `agents=0 tokens=0` と表示されます。 - -## 次へ +一回目は Review → Verify を見てください。同じ run の二回目は `cached` がほとんど（理想は `agents=0 tokens=0`）。完全なホストに載せるなら引数なしで `code.py` を。 -[s17 Goal Loop](../s17_goal_loop/) は、より小さな独立 loop で goal が達成されたかを確認し、次の round が必要かを判断します。 +s15 はあいかわらずそのループです。ここに増えるのは `Workflow` ツールだけ。[s17](../s17_goal_loop/) は別の問いをします。もう止まっていい？ - + diff --git a/s16_workflow_runtime/README.md b/s16_workflow_runtime/README.md index b69a2df09..d7424f6da 100644 --- a/s16_workflow_runtime/README.md +++ b/s16_workflow_runtime/README.md @@ -1,245 +1,139 @@ -# s16: Workflow Runtime — The Model Decides Each Step; a Script Decides the Orchestration +# s16: Workflow Runtime — put the plan in code [English](README.md) · [中文](README.zh.md) · [日本語](README.ja.md) -s01 → ... → s14 → [s15](../s15_integrated_harness/) → `s16` → [s17](../s17_goal_loop/) +[s15](../s15_integrated_harness/) → `s16` → [s17](../s17_goal_loop/) -> *"One tool_use runs an entire orchestration"* — The `Workflow` tool starts a recoverable script runtime that coordinates many agent calls. +> *Don't keep the plan only in chat.* The script owns order; the model owns each judgment. > -> **Harness layer**: Orchestration — run saved multi-agent scripts above the single-agent loop. +> **Harness layer**: orchestration — a multi-agent script on top of the single agent loop. ---- +## Problem -From s01 through s15, the model decides which tools to call in each round. Their results enter `messages[]`, and the model decides the next step from the updated context. This works well when the path depends on what the previous step discovers. +You already know how to let a model read files, edit code, and read errors in one loop. Some jobs, though, have an order you **already know**: review by dimension, then adversarial checks, then merge. If that order lives only in chat, the model stops halfway and calls it done, grades its own homework too kindly, and after a few compressions even "don't touch X" disappears. -Some tasks repeat a fixed sequence. A code review may inspect several dimensions concurrently, verify each finding, combine duplicates, and sort the result. The sequence and dependencies are known before execution. Here the host needs three things: +Soft conversation can't carry parallelism, stable result shapes, or crash-and-resume. You don't need a chattier model. You need **orchestration written down**. -- **Parallelism**, rather than waiting for one item at a time; -- **A stable result structure**, even when individual agent answers vary; -- **Recoverability**, so an interruption does not rerun work that is already complete. +## Solution -If this orchestration exists only in conversation history, its ordering and checkpoints also exist only in that history. A saved workflow puts the fixed sequence in code and records completed calls in a journal. - -## Put the Plan in Code, Not in a Sequence of Chat Turns - -Add a `Workflow` tool to the harness tool pool. The host registers trusted scripts built from `agent()`, `parallel()`, `pipeline()`, and `phase()`. The model supplies only a saved workflow name, arguments, and an optional run ID to resume; it does not send executable code or metadata. - -The workflow enters the main loop as one `tool_use`. As the script runs, the runtime emits lifecycle and progress events and records every step in a journal on disk. When the script finishes, the call returns the launch envelope, result, and task state. Intermediate script results live in variables instead of taking space in conversation history. When restarted with `resume_from_run_id`, unchanged `agent()` calls hit the journal cache and reuse previous results. - -![Workflow Runtime Overview](images/workflow-runtime-overview.svg) - -```python -SAMPLE_META = {"name": "review-changes", "description": "Review code changes", "phases": ["Review", "Verify"]} - -async def sample_workflow(ctx, args): - ctx.phase("Review") - results = await ctx.pipeline(DIMENSIONS, audit, verify) # Each dimension independently runs audit → verify - confirmed = [f for r in results if r for f in r["confirmed"]] - ctx.log(f"Confirmed {len(confirmed)} real issues") - return {"confirmed": confirmed} +```text + your chat ──► Workflow(...) ──► one result back + │ + ▼ + script: agent / pipeline / parallel + │ + ▼ + vars + journal (keep intermediates here, not in the thread) ``` -## The Workflow Tool: One Call, One Complete Run - -`Workflow` is added to the s15 host's existing tool pool. The user can request a saved workflow, or the model can select it when a task matches a known orchestration. The adapter resolves the name through the host-owned `WORKFLOWS` registry, then passes its trusted metadata and function to the runtime. The other s15 tools remain available in the same loop. +Subagents still think; the **script** owns loops, fan-out, and merge. Intermediates live in variables and a journal, not the host dialogue. -The model-facing schema accepts `name`, `args`, and `resume_from_run_id`. Unknown names and malformed arguments become an error tool result instead of ending the host loop. The runtime then validates the registered metadata, checks permissions, registers a local workflow task, and emits `async_launched` before running the script. Progress events follow, then the final `task_notification`; the call returns JSON-safe launch information, result, and task state. +One line: **move orchestration from intelligence to structure.** -```python -WORKFLOW_TOOL = { - "name": "Workflow", - "input_schema": { - "type": "object", - "properties": { - "name": {"type": "string"}, - "args": {"type": "object"}, - "resume_from_run_id": {"type": "string"}, - }, - "required": ["name"], - "additionalProperties": False, - }, -} - -async def run_workflow(name, args=None, resume_from_run_id=None): - meta, script_fn = WORKFLOWS[name] - out = await WorkflowTool().call( - meta, script_fn, - args=args, - resume_from_run_id=resume_from_run_id, - ) - return {"launched": out["launched"], "result": out["result"], - "task": serialize_task(out["task"])} -``` +![static harness vs dynamic workflow](images/dynamic-vs-static.png) -## Workflow Metadata: Validate Before Launch +*Left: a generic fixed pipeline. Right: a harness cut for this task.* -Each saved workflow registers trusted metadata with `name`, `description`, and optional `phases`. The runtime validates it before executing workflow code. `name` and `description` identify the task in the UI, while `phases` names groups in the progress display. These fields belong to the host registry, not to model input. +Claude Code has two doors: **dynamic** — the model writes JS for this task (`script` / `scriptPath`); **saved** — rerun a good script with `name` + `args`. Outside sits static SDK / `claude -p` orchestration. This lesson is a **Python teaching runtime** (no JS VM): same ideas, demo on the saved door. In the product the model can submit scripts — we just don't run JS here. -Invalid registration raises `WorkflowInputError` before launch. This is the same idea as validating cron expressions in s12: do not wait until execution to discover a bad saved workflow. +## How it works -Because the runtime uses `meta.name` in local artifact filenames, it also requires a 1-64 character safe slug containing letters, numbers, `.`, `_`, or `-`. +**1. Three verbs** -```python -def validate_meta(meta): - if not isinstance(meta, dict): - raise WorkflowInputError("meta must be an object literal") - if not meta.get("name") or not meta.get("description"): - raise WorkflowInputError("meta requires name and description") - if not isinstance(meta["name"], str) or not WORKFLOW_NAME_RE.fullmatch(meta["name"]): - raise WorkflowInputError("meta.name must be a safe 1-64 character slug") - if "phases" in meta and ( - not isinstance(meta["phases"], list) - or not all(isinstance(p, str) and p for p in meta["phases"]) - ): - raise WorkflowInputError("meta.phases must contain non-empty strings") - return meta +```text + agent one helper, one job (optional schema → JSON you can pass on) + pipeline each item walks stages on its own (default; no barrier) + parallel wait for every result (barrier; use sparingly) ``` -## Orchestration Primitives - -A script receives an `ExecutionState` exposing a small set of orchestration primitives. It does not read files or run shell commands directly. The default interactive mode connects `agent()` to the same real API client as the host, and each workflow agent reads only the content supplied through workflow arguments. `demo` and unit tests use `MockAgentRunner` so events and journal replay are repeatable. +On failure: a `parallel` slot becomes `null`; `pipeline` drops that item. The fleet does not sink. Filter before you merge. -| Primitive | Purpose | -|------|------| -| `agent(prompt, {schema, label, phase})` | Dispatch one subagent | -| `parallel(thunks)` | **Barrier**: run every task concurrently and wait until all results return | -| `pipeline(items, *stages)` | Run each item through stages **without a barrier**; finished items proceed immediately | -| `phase(title)` | Mark the current progress phase and update the progress display | -| `log(message)` | Emit a progress log line | -| `workflow(name, args)` | Run a nested sub-workflow, one level only | +**2. Resume from a notebook, not chat memory** -Use `pipeline` when each item independently crosses the same stages. Item A may reach stage three while item B is still in stage one. Use `parallel` when the next step needs every result from the preceding group. +The journal records `agent()` calls in **invocation order**. Resume replays the longest unchanged prefix; after the first change, everything runs live. Real JS runtimes ban `Date.now()` / `Math.random()` so the notebook can match — keep teaching scripts deterministic too. -```python -async def pipeline(self, items, *stages): - async def run_item(item, idx): - value = item - for stage in stages: # Each item independently completes every stage - value = await stage(value, item, idx) - return value - return await asyncio.gather(*[run_item(it, i) for i, it in enumerate(items)]) +```text + journal [A] [B] [C] [D] + resume hit hit ✂ live ``` -## Structured Output: Do Not Let Subagents Return Essays - -`agent({schema})` asks a workflow agent to return only a JSON object matching the schema. The runtime parses and validates the result, then retries once if it does not match. Downstream code receives an object instead of extracting fields from prose. +**3. One sample: fan-out + adversarial** -s05 warned that tool arguments cannot be trusted completely. This is the same lesson in reverse: subagent output cannot be trusted completely either. Validate at the orchestration boundary, give one retry, and keep uncertainty out of the rest of the flow. +`review-changes` is not "one pattern". It is **Fanout** with **Adversarial** inside: `pipeline(audit, verify)` per dimension, then `parallel` verifiers, keep only findings that still stand. -```python -run = await asyncio.to_thread(self.runner.run, prompt, schema, label) -result = run.value -if schema is not None: - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: # Retry once with a reminder, then fail - retry = await asyncio.to_thread( - self.runner.run, prompt + "\n\nReturn valid JSON.", schema, label - ) - result = retry.value - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: - raise WorkflowInputError(f"agent({{schema}}) returned invalid output: {err}") +```text + correctness ── audit ── verify ──┐ + security ── audit ── verify ──┤── confirmed + performance ── audit ── verify ──┤ + style ── audit ── verify ──┘ ``` -## Task State and Progress Events - -`LocalWorkflowTask` maintains status and token usage and emits an SDK-style event stream: `task_started` → a sequence of `task_progress` events containing phase changes, subagent starts, and log batches → one final `task_notification` reporting completion or failure, plus the output file and agent and token counts. - -The demo prints these events in order and returns the task state after the final notification. - ```python -class LocalWorkflowTask: - def progress_event(self, ptype, **data): # Phase/subagent/log - self.progress.append({"type": ptype, **data}) - print(f" progress {ptype} ...") +# from code.py — the shape is the point +async def sample_workflow(ctx, args): + ctx.phase("Review") + results = await ctx.pipeline(DIMENSIONS, audit, verify) + confirmed = [f for r in results if r for f in r["confirmed"]] + return {"confirmed": confirmed} ``` -## Storage: Snapshot + Journal for Resuming after Interruptions +The fleet can't stop early, authors don't referee themselves, and topology isn't rewritten by a tired chat turn. -The runtime stores each run under `s16_workflow_runtime/.runtime/`: a `.json` snapshot, `.output.json` output, `.journal.jsonl` journal, and `.lock` coordination file. Every fresh run reserves a new `runId` with exclusive file creation before opening its journal. The run lock stays held through execution and final persistence, so another process cannot resume the same run at the same time. Its snapshot records the workflow name, arguments, and task state; resume validates the saved snapshot and journal before changing either successful artifact. +
+Six common shapes (pattern toolbox) -The journal is the core of checkpointed resume. It records every `agent()` result one line at a time: +![Six workflow patterns](images/six-workflow-patterns.png) -```python -class WorkflowJournal: - def record(self, key, value): - self._f.write(json.dumps({"key": key, "value": value}) + "\n") - self._f.flush() - self.cache[key] = value -``` +| Pattern | In plain words | Primitives | +|------|------|----------| +| Classify-And-Act | Sort, then hand off | `agent` → branch → `agent` | +| Fanout-And-Synthesize | Split, then merge | `pipeline` / `parallel` → synthesize | +| Adversarial Verification | Don't let the fox grade the henhouse | produce → `parallel(verify)` → filter | +| Generate-And-Filter | Many drafts, then a ruler | `parallel(gens)` → filter | +| Tournament | Pairwise to a winner | judge `agent` | +| Loop Until Done | Keep going while "anything new?" | `while` + stop + `budget` | -## Resume: Continue by runId and Reuse Everything Unchanged +`review-changes` ≈ Fanout + Adversarial. Research stacks often go fan-out → filter → verify → synthesize. -Calling the workflow again with `resume_from_run_id` reruns the script, but every `agent()` computes a deterministic semantic key. If that key is present in the journal, it returns the cached result without executing again. Every unchanged call hits the cache; only a changed call and the downstream steps that depend on it actually rerun. +
-The key detail is that keys cannot depend on concurrency order. Agents in `parallel` and `pipeline` finish in nondeterministic order. If "the nth completion" became the key, cache entries would map to the wrong calls on the next run. A key therefore uses a stable hash of call content, including type, label, prompt, and schema, rather than a shared counter: +
+Dynamic / saved / static & official primitives ```python -def key(self, kind, label, prompt, schema): - basis = f"{kind}|{label}|{prompt}|{json.dumps(schema, sort_keys=True)}" - return f"{kind}-{_stable_hash(basis) % 10**10:010d}" - -# Inside agent(): -cached = self.journal.cached(key) -if cached is not MISS: - self.task.progress_event("workflow_agent", label=label, status="cached") - return cached +# teaching sketch +Workflow({ "name": "review-changes", "args": { "changes": "..." } }) +# Claude Code also accepts: script | scriptPath | resumeFromRunId ``` -## Stable Call Keys +![Workflow primitives](images/workflow-primitives.png) -On resume, the runtime must match each current `agent()` call with its earlier journal record. A stable hash gives unchanged workflow code and arguments the same call key. Real model output may vary; when the call content has not changed, resume uses the result already saved in the journal. +
-## See It Run +
+Untrusted input: quarantine reads -The sample `review-changes` workflow uses `pipeline` to send each review dimension independently through audit → verify. Interactive mode uses the real API and reads the material to review from `args.changes`. `demo` uses fixed runner data to show pipeline, validation, journal, and resume behavior. - -```python -async def sample_workflow(ctx, args): - ctx.phase("Review") - changes = args.get("changes", "") - - async def audit(_v, dimension, _i): - out = await ctx.agent(f"Inspect this change for {dimension} issues:\n{changes}", - schema=FINDINGS_SCHEMA, label=f"audit:{dimension}", phase="Review") - return {"dimension": dimension, "findings": out["findings"]} - - async def verify(audited, dimension, _i): - ctx.phase("Verify") - verdicts = await ctx.parallel([ # Verify every finding independently - (lambda f=f: ctx.agent(f"Verify this finding against the change:\n{changes}\n\n{f}", - schema=VERDICT_SCHEMA, label=f"verify:{dimension}:{f['title']}")) - for f in audited["findings"]]) - return {"dimension": dimension, - "confirmed": [f for f, v in zip(audited["findings"], verdicts) if v and v["isReal"]]} +The agent that reads tickets should not also hold the keys to open a PR. Readers only read → summary; the trusted side acts on the summary. - results = await ctx.pipeline(DIMENSIONS, audit, verify) - ... +```text + backlog → [quarantine: read / dedupe / summarize] → [trusted: act] ``` -## Changes from s15 +![quarantine triage](images/quarantine-triage.png) -| | s15 Integrated Harness | s16 Workflow Runtime | -|--|-----------|---------------------| -| Loop | One model-driven loop | Main loop unchanged; a tool runs scripted orchestration | -| Who decides the next step | Model decides each round | Script declares the orchestration in advance | -| Multiple agents | One-shot s06 subagents | Scripted, resumable calls through an agent-runner boundary | -| New mechanisms | — | Script primitives, host registry and tool adapter, task lifecycle, progress events, journal/resume, structured output | +
-s16 does not replace the main loop. It exposes `Workflow` at the tool layer and starts a local workflow runtime behind it: one saved script coordinates N calls through an agent-runner boundary. An s06 subagent is dispatched once at the model's discretion; s16 turns the orchestration into resumable host code. +Who owns the plan? s06 is one-shot dispatch, s13 is teammates with a mailbox, s15 is one chat loop, **s16 is script + journal**, s17 asks at the door whether the whole goal is done. For ordinary file edits, s15 or one s06 is often enough. Workflows cost tokens and coordination — reach for them when **structure must outlive a single conversation**. -## Try It +## Try it ```bash -python s16_workflow_runtime/code.py # Both the main model and Workflow agents use the real API -python s16_workflow_runtime/code.py demo # Deterministic review-changes fixture and event stream -python s16_workflow_runtime/code.py resume # Resume by the last runId; every agent() hits the journal cache +python s16_workflow_runtime/code.py demo +python s16_workflow_runtime/code.py resume ``` -In the default command, ask the model to read the changes, place that text in `args.changes`, and run the saved `review-changes` workflow. Both the main model and workflow agents use the real API. The `demo` command uses fixed runner data so lifecycle and resume behavior can be observed repeatedly. A resumed demo reports `agents=0 tokens=0` when every call hits the cache. - -## Next +First run: watch Review → Verify. Second run on the same id: expect mostly `cached` (ideally `agents=0 tokens=0`). For the full host loop, run `code.py` with no args. -[s17 Goal Loop](../s17_goal_loop/) uses a smaller, independent loop to check whether a stated goal has been reached and decide whether another turn is needed. +s15 is still the loop; this chapter only adds a `Workflow` tool. [s17](../s17_goal_loop/) asks a different question: should we stop? - + diff --git a/s16_workflow_runtime/README.zh.md b/s16_workflow_runtime/README.zh.md index 81f0108b3..0dcd2640f 100644 --- a/s16_workflow_runtime/README.zh.md +++ b/s16_workflow_runtime/README.zh.md @@ -1,245 +1,143 @@ -# s16: Workflow Runtime — 模型决定单步，脚本决定编排 +# s16: Workflow Runtime — 把步骤写进代码 [English](README.md) · [中文](README.zh.md) · [日本語](README.ja.md) -s01 → ... → s14 → [s15](../s15_integrated_harness/) → `s16` → [s17](../s17_goal_loop/) +[s15](../s15_integrated_harness/) → `s16` → [s17](../s17_goal_loop/) -> *"一次 tool_use，跑完一整套编排"* — `Workflow` 工具启动一个可恢复的脚本运行时，协调多次 agent 调用。 +> *计划不必只活在对话里。* 先后顺序交给脚本，每一步的判断留给模型。 > -> **Harness 层**: 编排 — 在单 agent 循环之上，执行保存好的多 agent 脚本。 +> **Harness 层**: 编排 — 在单个 Agent 循环之上，再运行一套多 Agent 脚本。 ---- +## 问题 -从 s01 到 s15，每一轮都由模型决定调用哪些工具。工具结果进入 `messages[]` 后，模型再根据更新后的上下文决定下一步。当后续路径取决于上一步发现了什么时，这种方式很合适。 +一路学到这里，你已经见过模型如何在循环里读文件、改代码、查看报错。 -有些任务会重复一套固定流程。例如代码审查可以同时检查多个维度，再逐条验证发现、合并重复项并按严重程度排序。执行前已经知道步骤及其先后关系，这时宿主需要三样东西： +可有些事情，顺序其实一开始就清楚：先分几个角度看看，再请另一位核对，最后汇总。若这些步骤只靠聊天记着，时间一长就容易乱——做到一半以为结束了；自己复查时又容易放过问题；上下文压缩几轮之后，原先那句「请别改动 X」也可能渐渐淡掉。 -- **并行**，别一个一个串着等； -- **稳定的结果结构**，即使每个 agent 的回答会变化； -- **可恢复**，跑到一半断了，已经做完的部分别从头再来。 +对话很适合探索。它不太擅长稳住并行、固定结果的形状，也不太擅长中断之后从同一个地方接着做。这时你需要的，往往不是一个更会聊天的模型，而是一份**写下来的步骤**。 -如果这套编排只存在于对话历史里，步骤顺序和检查点也只存在于历史里。保存好的 workflow 把固定流程写进代码，并在 journal 中记录已经完成的调用。 +## 解决方案 -## 计划写在代码里，不是靠聊天一轮轮凑 - -在 harness 的工具池里加入一个 `Workflow` 工具。宿主注册由 `agent() / parallel() / pipeline() / phase()` 组成的可信脚本。模型只提供保存好的 workflow 名称、参数和可选的续跑 run ID，不会提交可执行代码或元数据。 - -workflow 以一次 `tool_use` 进入主循环。脚本运行时，runtime 会发出生命周期和进度事件，并把每一步写进磁盘上的 journal。脚本结束后，这次调用返回启动信息、结果和任务状态。脚本里的中间结果存在变量里，不会塞进对话历史。下次用 `resume_from_run_id` 重启时，没改过的 `agent()` 会直接使用 journal 中的结果。 - -![Workflow Runtime 总览](images/workflow-runtime-overview.svg) - -```python -SAMPLE_META = {"name": "review-changes", "description": "审查代码改动", "phases": ["Review", "Verify"]} - -async def sample_workflow(ctx, args): - ctx.phase("Review") - results = await ctx.pipeline(DIMENSIONS, audit, verify) # 每个维度独立走 审计 → 验证 - confirmed = [f for r in results if r for f in r["confirmed"]] - ctx.log(f"确认了 {len(confirmed)} 个真实问题") - return {"confirmed": confirmed} +```text + 对话 ──► Workflow(...) ──► 结果回来 + │ + ▼ + 脚本：agent / pipeline / parallel + │ + ▼ + 变量 + journal（中间结果放这里） ``` -## Workflow 工具：一次调用，完成整次运行 +子 Agent 仍然负责思考。**脚本**负责循环、分发与合并。中间结果放进变量和 journal，不必再挤回主对话。 -`Workflow` 会加入 s15 宿主已有的工具池。用户可以要求运行一个保存好的 workflow，模型也可以在任务匹配已知编排时选择这个工具。适配器会用名称查询宿主管理的 `WORKFLOWS` registry，再把可信的元数据和函数交给运行时；s15 的其他工具仍在同一个循环里可用。 +可以记一句：**让结构来保管步骤，让模型来做判断。** -模型可见的 schema 只接受 `name`、`args` 和 `resume_from_run_id`。名称未知或参数格式错误时，适配器会返回错误工具结果，不会让宿主循环退出。随后运行时校验已经注册的元数据、经过权限检查、注册本地 workflow 任务，并在执行脚本前发出 `async_launched`。进度事件和最终的 `task_notification` 随后到达；调用返回可写入 JSON 的启动信息、结果和任务状态。 +![静态 harness 与动态 workflow](images/dynamic-vs-static.png) -```python -WORKFLOW_TOOL = { - "name": "Workflow", - "input_schema": { - "type": "object", - "properties": { - "name": {"type": "string"}, - "args": {"type": "object"}, - "resume_from_run_id": {"type": "string"}, - }, - "required": ["name"], - "additionalProperties": False, - }, -} - -async def run_workflow(name, args=None, resume_from_run_id=None): - meta, script_fn = WORKFLOWS[name] - out = await WorkflowTool().call( - meta, script_fn, - args=args, - resume_from_run_id=resume_from_run_id, - ) - return {"launched": out["launched"], "result": out["result"], - "task": serialize_task(out["task"])} -``` +*左：事先写好的通用流程。右：为这一次任务量身写下的编排。* -## Workflow 元数据：启动前先校验 +在 Claude Code 里，常见有两种用法。**动态**：模型为这次任务写一段 JavaScript（`script` / `scriptPath`）。**已保存**：跑顺了的脚本留在仓库里，用 `name` 和 `args` 再次唤起。此外还有用 SDK 事先写好的静态编排。本章是一个 **Python 教学运行时**，不内嵌 JS 引擎：想法与产品对齐，演示走「已保存」这一路。产品里模型本来就可以提交脚本，我们只是在这里用更易读的 Python 来讲清楚。 -每个保存好的 workflow 都会注册一份可信元数据，包含 `name`、`description` 和可选的 `phases`。运行时会在执行 workflow 代码前校验它：`name` 和 `description` 用来标识任务，`phases` 给进度显示分组命名。这些字段属于宿主 registry，不是模型输入。 +## 工作原理 -注册内容不合法时，运行时会在启动前抛出 `WorkflowInputError`。这和 s12 校验 cron 表达式是一个思路：保存好的 workflow 有问题，就不要等到执行时才发现。 +**1. 先认识三个词** -运行时会把 `meta.name` 用在本地产物文件名中，因此还要求它是 1-64 个字符的安全 slug，只能包含字母、数字、`.`、`_`、`-`。 - -```python -def validate_meta(meta): - if not isinstance(meta, dict): - raise WorkflowInputError("meta 必须是对象字面量") - if not meta.get("name") or not meta.get("description"): - raise WorkflowInputError("meta 必须包含 name 和 description") - if not isinstance(meta["name"], str) or not WORKFLOW_NAME_RE.fullmatch(meta["name"]): - raise WorkflowInputError("meta.name 必须是 1-64 字符的安全 slug") - if "phases" in meta and ( - not isinstance(meta["phases"], list) - or not all(isinstance(p, str) and p for p in meta["phases"]) - ): - raise WorkflowInputError("meta.phases 必须包含非空字符串") - return meta +```text + agent 一位帮手，完成一件事（可带 schema，得到便于传递的 JSON） + pipeline 每一项各自走完各阶段（默认如此，不必互相等待） + parallel 等齐所有结果再继续（屏障，偶尔使用） ``` -## 编排原语 +某一步不顺利时，不必让整次运行停住：`parallel` 里出问题的那一格会变成 `null`；`pipeline` 则跳过那一项。合并之前，先把空位滤掉即可。 -脚本收到一个只暴露少量编排原语的 `ExecutionState`，本身不直接读写文件，也不运行 shell。默认交互模式把 `agent()` 接到与宿主相同的真实 API client；每个子 agent 只读取 workflow 参数中提供的内容。`demo` 和单元测试使用 `MockAgentRunner`，便于重复观察事件和 journal。 +**2. 若要接着跑，用小本记下进度** -| 原语 | 作用 | -|------|------| -| `agent(prompt, {schema, label, phase})` | 派一个子 agent 干活 | -| `parallel(thunks)` | **等齐屏障**：所有任务并行跑完，一起等结果回来 | -| `pipeline(items, *stages)` | 每个 item 分阶段跑，**不等齐**，跑完一个往下走一个 | -| `phase(title)` | 标记当前进度阶段（更新进度条） | -| `log(message)` | 打一行进度日志 | -| `workflow(name, args)` | 嵌套子工作流（只支持一层） | +每调用一次 `agent()`，journal 就按**调用顺序**记一笔。再次运行时，从前往后核对：尚未改动的前缀可以直接复用；遇到第一处变化，之后的步骤再重新执行。真实的 JS 运行时不宜使用 `Date.now()` / `Math.random()`，否则记录很难对齐。教学脚本也尽量写成确定的。 -每个 item 都要独立经过相同步骤时，可以使用 `pipeline`。item A 跑到第 3 阶段时，item B 可能还在第 1 阶段；下一步必须同时使用上一阶段全部结果时，再使用 `parallel` 等待所有调用完成。 - -```python -async def pipeline(self, items, *stages): - async def run_item(item, idx): - value = item - for stage in stages: # 每个 item 独立跑完所有 stage - value = await stage(value, item, idx) - return value - return await asyncio.gather(*[run_item(it, i) for i, it in enumerate(items)]) +```text + journal [A] [B] [C] [D] + 续跑 复用 复用 ✂ 重跑 ``` -## 结构化输出：别让子 agent 回来写散文 +**3. 用一个例子串起来** -`agent({schema})` 会要求子 agent 只返回匹配 schema 的 JSON 对象。运行时解析并校验结果，不符合时重试一次。这样下游代码拿到的是对象，不必再从自然语言中提取字段。 +`review-changes` 并不只是一种固定招式。它先把工作拆开，再请另一路核对：多个维度各自走 `pipeline(audit, verify)`，在验证阶段用 `parallel` 并行检查，最后只留下仍然成立的发现。 -s05 就说过，工具的参数不能全信；这里是同一个道理反过来：子 agent 的输出也不能全信。加一层校验，不对就给一次机会重试，把不确定性挡在编排层外面。 - -```python -run = await asyncio.to_thread(self.runner.run, prompt, schema, label) -result = run.value -if schema is not None: - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: # 提醒一次重试，再不对就报错 - retry = await asyncio.to_thread( - self.runner.run, prompt + "\n\n返回合法的 JSON。", schema, label - ) - result = retry.value - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: - raise WorkflowInputError(f"agent({{schema}}) 输出不合法: {err}") +```text + correctness ── 审阅 ── 核对 ──┐ + security ── 审阅 ── 核对 ──┤── confirmed + performance ── 审阅 ── 核对 ──┤ + style ── 审阅 ── 核对 ──┘ ``` -## 任务状态和进度事件 - -`LocalWorkflowTask` 维护状态和 token 用量，向外发一条 SDK 风格的事件流：`task_started` → 一串 `task_progress`（包含阶段切换、子 agent 启动和日志输出）→ 最后一个 `task_notification`（完成或失败，带输出文件、agent 数和 token 数）。 - -演示会按顺序打印这些事件，并在最终通知后返回任务状态。 - ```python -class LocalWorkflowTask: - def progress_event(self, ptype, **data): # 阶段/子agent/日志 - self.progress.append({"type": ptype, **data}) - print(f" 进度 {ptype} ...") +# 摘自 code.py，先看形状 +async def sample_workflow(ctx, args): + ctx.phase("Review") + results = await ctx.pipeline(DIMENSIONS, audit, verify) + confirmed = [f for r in results if r for f in r["confirmed"]] + return {"confirmed": confirmed} ``` -## 存储：快照 + journal，断了能续 +这样安排之后，步骤不容易半途收束，作者也不必兼任唯一的检查者，流程也不必随着聊天一轮轮改写。 -运行时把每次运行的数据存在 `s16_workflow_runtime/.runtime/`：快照 `.json`、输出 `.output.json`、journal `.journal.jsonl` 和协调文件 `.lock`。每次新运行都会在打开 journal 前，用排他式文件创建预留新的 `runId`。整次执行和最终持久化期间都持有 run lock，另一个进程不能同时 resume 同一次运行。快照记录 workflow 名称、参数和任务状态；resume 会先验证已保存的快照和 journal，再改动原有的成功产物。 +
+六种常见形状 -journal 是断点续跑的核心，它一条一条记下来每个 `agent()` 的结果： +![六种 Workflow 模式](images/six-workflow-patterns.png) -```python -class WorkflowJournal: - def record(self, key, value): - self._f.write(json.dumps({"key": key, "value": value}) + "\n") - self._f.flush() - self.cache[key] = value -``` +| 名称 | 含义 | 如何组合 | +|------|------|----------| +| Classify-And-Act | 先分类，再交给合适的帮手 | `agent` → 分支 → `agent` | +| Fanout-And-Synthesize | 分头进行，再汇总 | `pipeline` / `parallel` → 汇总 | +| Adversarial Verification | 请另一路来核对，而不是只听自己 | 产出 → `parallel(verify)` → 过滤 | +| Generate-And-Filter | 先多准备几份，再筛选 | `parallel` 生成 → 过滤 | +| Tournament | 两两比较，留下更好的 | 裁判 `agent` | +| Loop Until Done | 仍有新发现就继续 | `while` + 停止条件 + `budget` | -## resume：用 runId 续跑，没改的直接用缓存 +`review-changes` 大约是 Fanout 与 Adversarial 的组合。做调研时，也常再叠上：分发 → 过滤 → 核对 → 汇总。 -带着 `resume_from_run_id` 再次调用 workflow 时，脚本会重新执行，但每个 `agent()` 都会计算一个确定的语义 key：key 在 journal 里有记录，就直接返回缓存结果；只有改过的调用以及依赖它的后续步骤才会真的运行。 +
-这里有个关键点：key 不能依赖并发顺序。`parallel` 和 `pipeline` 里 agent 完成的顺序是不确定的，用"第几个完成"当 key，两次跑缓存就对错位了。所以 key 是根据调用内容（类型、标签、prompt、schema）算的稳定哈希，不是一个会竞争的计数器： +
+动态、已保存、静态，以及官方示意 ```python -def key(self, kind, label, prompt, schema): - basis = f"{kind}|{label}|{prompt}|{json.dumps(schema, sort_keys=True)}" - return f"{kind}-{_stable_hash(basis) % 10**10:010d}" - -# agent() 内部： -cached = self.journal.cached(key) -if cached is not MISS: - self.task.progress_event("workflow_agent", label=label, status="cached") - return cached +# 教学示意 +Workflow({ "name": "review-changes", "args": { "changes": "..." } }) +# Claude Code 还接受：script | scriptPath | resumeFromRunId ``` -## 稳定调用键 - -续跑时，运行时需要把当前 `agent()` 与 journal 中的旧调用对应起来。稳定哈希让同一份 workflow 和同样的参数产生相同的调用 key。真实模型的回答可以变化；只要调用内容没有变化，resume 就直接使用 journal 中已经保存的结果。 +![Workflow 原语](images/workflow-primitives.png) -## 跑起来看看 +
-示例 workflow `review-changes` 用 `pipeline` 让每个审查维度独立走“审计 → 验证”。默认交互模式使用真实 API，并从 `args.changes` 读取待审查内容；`demo` 使用固定 runner 数据来展示 pipeline、结构校验、journal 和续跑。 +
+遇到不可信输入时，把读取与行动分开 -```python -async def sample_workflow(ctx, args): - ctx.phase("Review") - changes = args.get("changes", "") - - async def audit(_v, dimension, _i): - out = await ctx.agent(f"检查这段变更里有没有{dimension}相关的问题：\n{changes}", - schema=FINDINGS_SCHEMA, label=f"audit:{dimension}", phase="Review") - return {"dimension": dimension, "findings": out["findings"]} - - async def verify(audited, dimension, _i): - ctx.phase("Verify") - verdicts = await ctx.parallel([ # 每条发现独立做对抗性验证 - (lambda f=f: ctx.agent(f"根据变更内容验证这条 finding：\n{changes}\n\n{f}", - schema=VERDICT_SCHEMA, label=f"verify:{dimension}:{f['title']}")) - for f in audited["findings"]]) - return {"dimension": dimension, - "confirmed": [f for f, v in zip(audited["findings"], verdicts) if v and v["isReal"]]} +负责阅读工单的 Agent，不宜同时握有发起变更的权限。可以让一侧只读、整理成摘要；另一侧只根据摘要行动。 - results = await ctx.pipeline(DIMENSIONS, audit, verify) - ... +```text + 待办 → [阅读区: 读取 / 去重 / 摘要] → [行动区: 执行] ``` -## 相对 s15 的变更 +![隔离分流](images/quarantine-triage.png) -| | s15 Agent Harness 集成 | s16 Workflow Runtime | -|--|-----------|---------------------| -| 循环 | 单个、模型驱动 | 主循环不变；工具背后执行脚本编排 | -| 谁决定下一步 | 模型逐轮决定 | 脚本预先写好编排流程 | -| 多 agent | s06 子 agent，一次性派出去 | 通过 agent-runner 边界执行脚本化、可续跑的调用 | -| 新增机制 | — | 编排原语、宿主 registry 与工具适配器、任务生命周期、进度事件、journal/续跑、结构化输出 | +
-s16 不替换主循环，它只是在工具层暴露 `Workflow`，背后启动一个本地 workflow 运行时：一份保存好的脚本通过 agent-runner 边界协调 N 次调用。s06 的子 agent 是模型临场派一次；s16 把编排写成可续跑的宿主代码。 +不妨问一句：计划由谁保管？s06 是一次性委托；s13 是长期协作的队友；s15 在单次循环的对话里决定下一步；**s16 把步骤写进脚本，把进度记在 journal**；s17 则关心整件事是否已经完成。 -## 试一下 +日常改几个文件，s15 或一次 s06 往往就够。Workflow 会多花一些 token，也需要一点协调——**当步骤需要比单次对话活得更久**，再请它来帮忙。 + +## 试一试 ```bash -python s16_workflow_runtime/code.py # 主模型和 Workflow 子 agent 都使用真实 API -python s16_workflow_runtime/code.py demo # 运行确定性的 review-changes 测试数据并观察事件流 -python s16_workflow_runtime/code.py resume # 用上次的 runId 续跑，每个 agent() 都命中 journal 缓存 +python s16_workflow_runtime/code.py demo +python s16_workflow_runtime/code.py resume ``` -默认命令里，可以先让模型读取改动，再把内容放进 `args.changes` 并运行保存好的 `review-changes` workflow。主模型和 workflow 子 agent 都使用真实 API。`demo` 命令使用固定 runner 数据，便于重复观察生命周期和续跑；续跑命中全部缓存时显示 `agents=0 tokens=0`。 - -## 接下来 +第一次运行，可以留意从 Review 走到 Verify。同一 run 再执行一次，多数步骤应显示为 `cached`（理想情况是 `agents=0 tokens=0`）。若想放进上一章的完整程序，不加参数直接运行 `code.py` 即可。 -[s17 Goal Loop](../s17_goal_loop/) 会使用一个更小、独立的循环检查既定目标是否已经达成，并据此决定是否还需要下一轮。 +s15 仍是那个循环；这里只是多了一个 `Workflow` 工具。[s17](../s17_goal_loop/) 会接着问：是否可以停下了？ - + diff --git a/s16_workflow_runtime/code.py b/s16_workflow_runtime/code.py index 3c0a7623e..f1b59bceb 100644 --- a/s16_workflow_runtime/code.py +++ b/s16_workflow_runtime/code.py @@ -1,6 +1,20 @@ #!/usr/bin/env python3 """ -s16: Workflow Runtime - run a saved orchestration through one tool call. +s16: Workflow Runtime — teaching model of Dynamic Agentic Workflows. + +Idea: move orchestration from intelligence to structure. Subagents judge; +the script owns topology (loops, fan-out, merge). Intermediate results live +in variables + a journal, not in the host conversation. + +Claude Code accepts script / scriptPath / name / args / resumeFromRunId. +Dynamic = model writes a JS harness for this task; Saved = rerun by name. +This lesson is a small Python runtime with the same ideas, line-readable. +Demo registers one saved workflow; we do not embed a JS interpreter. + +Fidelity notes baked into this file: + - parallel/pipeline failures isolate to null slots (fleet continues) + - resume = longest unchanged agent() call-order prefix + - real JS runtimes ban Date.now/Math.random so journals stay aligned Run: python s16_workflow_runtime/code.py @@ -319,15 +333,32 @@ def run(self, prompt, schema=None, label=None): # -- Journal -- class WorkflowJournal: - """Append-only .journal.jsonl. On resume, agent() calls whose - semantic key is already present are replayed from cache instead of re-run.""" + """Append-only .journal.jsonl with longest-unchanged-prefix resume. + + Replay walks agent() calls in *invocation* order (the contract Claude Code + and mature Pi ports use). Cache hits continue only while each call's + semantic key matches the next journal entry. The first mismatch or missing + entry breaks the prefix: every later call runs live, even if an older key + still exists further down the journal. + + Why real JS runtimes ban Date.now() / Math.random() / bare new Date(): those + make call order or prompts nondeterministic, so resume cannot match the + journal. This Python teaching runtime does not sandbox that — write + deterministic scripts anyway. + """ def __init__(self, run_id, resume, store=None): store = STORE if store is None else store store.mkdir(parents=True, exist_ok=True) self.path = store / f"{run_id}.journal.jsonl" self.resume = resume + self.entries = [] self.cache = {} + self._cursor = 0 # next call-order index to assign + self._write_cursor = 0 # next index to flush to disk + self._pending = {} # idx -> (key, value) awaiting in-order flush + self._prefix_broken = False + self._did_truncate = False if resume: if not self.path.exists(): raise WorkflowInputError(f"resume journal not found for {run_id}") @@ -344,24 +375,54 @@ def __init__(self, run_id, resume, store=None): raise WorkflowInputError( f"invalid resume journal record at line {line_number}" ) from exc + self.entries.append({"key": rec["key"], "value": rec["value"]}) self.cache[rec["key"]] = rec["value"] self._f = self.path.open("a") else: self._f = self.path.open("w") # fresh run truncates def key(self, kind, label, prompt, schema): - # Deterministic semantic key, independent of concurrency order, so a - # parallel/pipeline call gets the same key on resume. + # Content hash identifies *this* call. Prefix matching uses call order; + # the hash must stay stable across resume (not Python's salted hash()). basis = f"{kind}|{label}|{prompt}|{json.dumps(schema, sort_keys=True)}" return f"{kind}-{_stable_hash(basis) % 10**10:010d}" - def cached(self, key): - return self.cache.get(key, MISS) - - def record(self, key, value): - self._f.write(json.dumps({"key": key, "value": value}) + "\n") - self._f.flush() - self.cache[key] = value + def try_replay(self, key): + """Reserve the next call-order slot. Return (cached_value_or_MISS, idx).""" + idx = self._cursor + self._cursor += 1 + if self._prefix_broken or not self.resume: + return MISS, idx + if idx >= len(self.entries) or self.entries[idx]["key"] != key: + self._prefix_broken = True + return MISS, idx + return self.entries[idx]["value"], idx + + def record(self, idx, key, value): + """Record a live result at call-order index idx; flush in order.""" + if self.resume and self._prefix_broken and not self._did_truncate: + # Keep only the unchanged prefix; rewrite the file from there. + self.entries = self.entries[:idx] + self._f.close() + self._f = self.path.open("w") + for rec in self.entries: + self._f.write(json.dumps({"key": rec["key"], "value": rec["value"]}) + "\n") + self._f.flush() + self._write_cursor = len(self.entries) + self._did_truncate = True + + self._pending[idx] = (key, value) + while self._write_cursor in self._pending: + k, v = self._pending.pop(self._write_cursor) + self._f.write(json.dumps({"key": k, "value": v}) + "\n") + self._f.flush() + rec = {"key": k, "value": v} + if self._write_cursor < len(self.entries): + self.entries[self._write_cursor] = rec + else: + self.entries.append(rec) + self.cache[k] = v + self._write_cursor += 1 def close(self): self._f.close() @@ -439,6 +500,9 @@ def __init__(self, task, journal, runner, budget, args, depth=0, limits=None): self._phase = None self._phases_seen = set() self._limits = limits or ExecutionLimits() + # Serializes call-order tickets so parallel agent() invocations still + # get a stable prefix position (creation order, not completion order). + self._order_lock = asyncio.Lock() def phase(self, title): """Start a phase; subsequent agent()s group under it. Upsert: emitting the @@ -453,69 +517,125 @@ def log(self, message): self.task.progress_event("workflow_log", message=message) async def agent(self, prompt, schema=None, label=None, phase=None): - """Spawn one subagent. With a schema, force StructuredOutput + validate - (retry once). On resume, a cached key short-circuits the run.""" + """Spawn one subagent. With a schema, validate (+ one retry). + + Resume uses longest unchanged prefix in agent() call order: hits continue + until the first changed/missing call; everything after runs live. + """ label = label or (prompt[:24] + "...") self._limits.claim_agent() if self.budget.remaining() <= 0: raise WorkflowInputError("token budget exceeded") key = self.journal.key("agent", label, prompt, schema) - cached = self.journal.cached(key) - if cached is not MISS: - if schema is not None: - ok, err = SimpleJsonSchema(schema).validate(cached) - if not ok: - raise WorkflowInputError( - f"cached agent output failed schema validation: {err}" - ) - self.task.progress_event("workflow_agent", label=label, - phase=phase or self._phase, status="cached") - return cached - - async with self._limits.semaphore: - run = await asyncio.to_thread( - self.runner.run, prompt, schema, label - ) - result = run.value - tokens = run.tokens + async with self._order_lock: + cached, call_idx = self.journal.try_replay(key) + if cached is not MISS: + # None is a recorded failure (null-isolation); replay it as-is. + if cached is not None and schema is not None: + ok, err = SimpleJsonSchema(schema).validate(cached) + if not ok: + raise WorkflowInputError( + f"cached agent output failed schema validation: {err}" + ) + self.task.progress_event( + "workflow_agent", + label=label, + phase=phase or self._phase, + status="cached", + ) + return cached - if schema is not None: - ok, err = SimpleJsonSchema(schema).validate(result) - if not ok: - retry = await asyncio.to_thread( - self.runner.run, - prompt + "\n\nReturn valid JSON.", - schema, - label, + result = None + tokens = 0 + live_error = None + try: + async with self._limits.semaphore: + run = await asyncio.to_thread( + self.runner.run, prompt, schema, label ) - result = retry.value - tokens += retry.tokens + result = run.value + tokens = run.tokens + + if schema is not None: ok, err = SimpleJsonSchema(schema).validate(result) if not ok: - raise WorkflowInputError(f"agent({{schema}}) invalid output: {err}") + retry = await asyncio.to_thread( + self.runner.run, + prompt + "\n\nReturn valid JSON.", + schema, + label, + ) + result = retry.value + tokens += retry.tokens + ok, err = SimpleJsonSchema(schema).validate(result) + if not ok: + raise WorkflowInputError( + f"agent({{schema}}) invalid output: {err}" + ) + + self.budget.add(tokens) + self.task.usage["agents"] += 1 + self.task.usage["tokens"] += tokens + except Exception as exc: + # Fill the call-order slot with null so later parallel records can + # flush; parallel/pipeline turn the raise into a null slot. + result = None + live_error = exc + + async with self._order_lock: + self.journal.record(call_idx, key, result) + + if live_error is not None: + self.task.progress_event( + "workflow_agent", + label=label, + phase=phase or self._phase, + status="null", + ) + raise live_error - self.budget.add(tokens) - self.task.usage["agents"] += 1 - self.task.usage["tokens"] += tokens - self.journal.record(key, result) self.task.progress_event("workflow_agent", label=label, phase=phase or self._phase, status="done") return result async def parallel(self, thunks): - """BARRIER: run all thunks concurrently and fail if any thunk fails.""" - return await asyncio.gather(*[thunk() for thunk in thunks]) + """BARRIER: run all thunks concurrently; wait for every result. + + A failing thunk becomes None in that slot — the gather itself does not + reject. Filter with care (e.g. [x for x in results if x]). + """ + async def isolate(thunk): + try: + return await thunk() + except Exception as exc: + self.log(f"parallel step → null ({type(exc).__name__})") + return None + + return await asyncio.gather(*[isolate(thunk) for thunk in thunks]) async def pipeline(self, items, *stages): """Per-item staged flow, NO barrier between stages: item A can be in stage 3 while item B is still in stage 1. Each stage gets - (prev_result, original_item, index). A throwing stage fails the workflow.""" + (prev_result, original_item, index). + + A failing stage drops that item to None and skips its remaining stages; + other items keep going. + """ async def run_item(item, idx): value = item for stage in stages: - value = await stage(value, item, idx) + try: + value = await stage(value, item, idx) + except Exception as exc: + self.log( + f"pipeline item {idx} → null ({type(exc).__name__})" + ) + return None + if value is None: + return None return value + return await asyncio.gather(*[run_item(it, i) for i, it in enumerate(items)]) async def workflow(self, name, args=None): @@ -713,15 +833,22 @@ async def verify(audited, dimension, _idx): # Saved workflow registry WORKFLOWS = {SAMPLE_META["name"]: (SAMPLE_META, sample_workflow)} +# Teaching adapter: saved-workflow path (name + args). Claude Code also accepts +# script / scriptPath for dynamic JS the model writes; we keep this surface +# small and map the same resume idea via resume_from_run_id / resumeFromRunId. WORKFLOW_TOOL = { "name": "Workflow", - "description": "Run a saved workflow by name. Pass input in args.", + "description": ( + "Run a saved workflow by name. Pass input in args. " + "Optional resume_from_run_id (alias: resumeFromRunId) continues a prior run." + ), "input_schema": { "type": "object", "properties": { "name": {"type": "string"}, "args": {"type": "object"}, "resume_from_run_id": {"type": "string"}, + "resumeFromRunId": {"type": "string"}, }, "required": ["name"], "additionalProperties": False, @@ -741,20 +868,34 @@ def serialize_task(task): } -async def run_workflow(name, args=None, resume_from_run_id=None): - """Model-facing adapter: resolve trusted code from the host registry.""" +async def run_workflow( + name, + args=None, + resume_from_run_id=None, + resumeFromRunId=None, +): + """Model-facing adapter: resolve a saved workflow from the host registry. + + Claude Code's dynamic door passes script/scriptPath instead of name; this + teaching runtime stays on the saved-name path so every line stays readable. + """ if not isinstance(name, str): raise WorkflowInputError("workflow name must be a string") if name not in WORKFLOWS: raise WorkflowInputError(f"unknown workflow '{name}'") if args is not None and not isinstance(args, dict): raise WorkflowInputError("workflow args must be an object") + if resume_from_run_id and resumeFromRunId and resume_from_run_id != resumeFromRunId: + raise WorkflowInputError( + "resume_from_run_id and resumeFromRunId disagree" + ) + resume_id = resume_from_run_id or resumeFromRunId meta, script_fn = WORKFLOWS[name] out = await WorkflowTool().call( meta, script_fn, args=args, - resume_from_run_id=resume_from_run_id, + resume_from_run_id=resume_id, ) return { "launched": out["launched"], @@ -814,9 +955,14 @@ async def run_demo(argv): if not resume_id: print("nothing to resume; run `python code.py demo` first.") return - print(f"resuming {resume_id}; unchanged agent() calls use the journal cache\n") + print( + f"resuming {resume_id}\n" + "longest unchanged agent() prefix → cache hit; " + "first change breaks the prefix\n" + ) else: - print("launching workflow `review-changes`\n") + print("launching saved workflow `review-changes`") + print("watch phases: Review → Verify, then a confirmed list\n") out = await WORKFLOW_HANDLERS["Workflow"]( name="review-changes", diff --git a/s16_workflow_runtime/images/dynamic-vs-static.png b/s16_workflow_runtime/images/dynamic-vs-static.png new file mode 100644 index 000000000..a3e09c92d Binary files /dev/null and b/s16_workflow_runtime/images/dynamic-vs-static.png differ diff --git a/s16_workflow_runtime/images/quarantine-triage.png b/s16_workflow_runtime/images/quarantine-triage.png new file mode 100644 index 000000000..fcbbac0d6 Binary files /dev/null and b/s16_workflow_runtime/images/quarantine-triage.png differ diff --git a/s16_workflow_runtime/images/six-workflow-patterns.png b/s16_workflow_runtime/images/six-workflow-patterns.png new file mode 100644 index 000000000..a4f4c06a8 Binary files /dev/null and b/s16_workflow_runtime/images/six-workflow-patterns.png differ diff --git a/s16_workflow_runtime/images/workflow-primitives.png b/s16_workflow_runtime/images/workflow-primitives.png new file mode 100644 index 000000000..06f4de8af Binary files /dev/null and b/s16_workflow_runtime/images/workflow-primitives.png differ diff --git a/tests/test_workflow_goal_lessons.py b/tests/test_workflow_goal_lessons.py index e087466bc..87f49b72a 100644 --- a/tests/test_workflow_goal_lessons.py +++ b/tests/test_workflow_goal_lessons.py @@ -117,8 +117,8 @@ async def run() -> None: async def fail_stage(_value, _item, _index): raise RuntimeError("stage failed") - with pytest.raises(RuntimeError, match="stage failed"): - await state.pipeline(["item"], fail_stage) + # Null-isolation: a failing stage drops that item, not the whole run. + assert await state.pipeline(["item"], fail_stage) == [None] try: asyncio.run(run()) @@ -482,16 +482,139 @@ def test_workflow_default_entry_extends_the_real_s15_host( ) -def test_workflow_tool_adapter_rejects_model_supplied_code() -> None: +def test_workflow_tool_adapter_uses_saved_name_surface() -> None: + """Teaching adapter stays on name/args; Claude Code also has script/scriptPath.""" workflow = load_lesson( "workflow_schema_test", ROOT / "s16_workflow_runtime" / "code.py" ) properties = workflow.WORKFLOW_TOOL["input_schema"]["properties"] - assert set(properties) == {"name", "args", "resume_from_run_id"} + assert set(properties) == { + "name", + "args", + "resume_from_run_id", + "resumeFromRunId", + } assert "description" not in properties assert "script" not in properties + assert "scriptPath" not in properties with pytest.raises(workflow.WorkflowInputError, match="name must be a string"): asyncio.run(workflow.run_workflow({"name": "review-changes"})) with pytest.raises(workflow.WorkflowInputError, match="unknown workflow"): asyncio.run(workflow.run_workflow("missing")) + + +def test_parallel_and_pipeline_isolate_failures(tmp_path: Path) -> None: + workflow = load_lesson( + "workflow_null_isolation_test", ROOT / "s16_workflow_runtime" / "code.py" + ) + journal = workflow.WorkflowJournal( + "wf_null-iso_0001", resume=False, store=tmp_path + ) + task = workflow.LocalWorkflowTask("task", "wf_null-iso_0001", {}) + state = workflow.ExecutionState( + task, journal, workflow.MockAgentRunner(), workflow.Budget(), {} + ) + + async def ok(): + return await state.agent("ok", label="ok") + + async def boom(): + raise RuntimeError("helper crashed") + + async def stage_ok(value, _item, _index): + return f"ok:{value}" + + async def stage_fail(value, _item, _index): + if value == "bad": + raise RuntimeError("stage crashed") + return f"next:{value}" + + async def stage_later(value, _item, _index): + return f"final:{value}" + + async def run(): + parallel_out = await state.parallel([ok, boom, ok]) + pipeline_out = await state.pipeline( + ["good", "bad", "also"], stage_fail, stage_later + ) + return parallel_out, pipeline_out + + try: + parallel_out, pipeline_out = asyncio.run(run()) + finally: + journal.close() + + assert parallel_out[1] is None + assert parallel_out[0] is not None and parallel_out[2] is not None + assert pipeline_out == ["final:next:good", None, "final:next:also"] + + +def test_resume_uses_longest_unchanged_prefix( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workflow = load_lesson( + "workflow_prefix_resume_test", ROOT / "s16_workflow_runtime" / "code.py" + ) + monkeypatch.setattr(workflow, "STORE", tmp_path) + run_id = "wf_prefix-test_0000000000001a7b" + monkeypatch.setattr(workflow, "create_run_id", lambda _meta: run_id) + calls: list[str] = [] + + class TrackingRunner: + def run(self, prompt, schema=None, label=None): + calls.append(label or prompt) + return workflow.RunnerOutput({"label": label, "prompt": prompt}, 1) + + monkeypatch.setattr(workflow, "RUNNER_FACTORY", TrackingRunner) + meta = {"name": "prefix-test", "description": "prefix resume"} + + async def script_v1(ctx, _args): + a = await ctx.agent("one", label="a") + b = await ctx.agent("two", label="b") + c = await ctx.agent("three", label="c") + return [a, b, c] + + async def script_v2(ctx, _args): + # a unchanged, b's prompt changes → prefix breaks; c must run live + # even though an old "c" key exists further down the journal. + a = await ctx.agent("one", label="a") + b = await ctx.agent("two-changed", label="b") + c = await ctx.agent("three", label="c") + return [a, b, c] + + first = asyncio.run(workflow.WorkflowTool().call(meta, script_v1)) + assert first["task"].status == "completed" + assert calls == ["a", "b", "c"] + + calls.clear() + resumed = asyncio.run( + workflow.WorkflowTool().call( + meta, script_v2, resume_from_run_id=run_id + ) + ) + assert resumed["task"].status == "completed" + assert calls == ["b", "c"], "after the first miss, later calls must run live" + assert resumed["result"][0] == {"label": "a", "prompt": "one"} + assert resumed["result"][1]["prompt"] == "two-changed" + assert resumed["task"].usage["agents"] == 2 + + +def test_resume_from_run_id_camel_case_alias( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workflow = load_lesson( + "workflow_resume_alias_test", ROOT / "s16_workflow_runtime" / "code.py" + ) + monkeypatch.setattr(workflow, "STORE", tmp_path) + first = asyncio.run( + workflow.run_workflow("review-changes", {"budget": None}) + ) + resumed = asyncio.run( + workflow.run_workflow( + "review-changes", + resumeFromRunId=first["task"]["runId"], + ) + ) + assert resumed["task"]["status"] == "completed" + assert resumed["task"]["usage"]["agents"] == 0 diff --git a/web/public/course-assets/s16_workflow_runtime/dynamic-vs-static.png b/web/public/course-assets/s16_workflow_runtime/dynamic-vs-static.png new file mode 100644 index 000000000..a3e09c92d Binary files /dev/null and b/web/public/course-assets/s16_workflow_runtime/dynamic-vs-static.png differ diff --git a/web/public/course-assets/s16_workflow_runtime/quarantine-triage.png b/web/public/course-assets/s16_workflow_runtime/quarantine-triage.png new file mode 100644 index 000000000..fcbbac0d6 Binary files /dev/null and b/web/public/course-assets/s16_workflow_runtime/quarantine-triage.png differ diff --git a/web/public/course-assets/s16_workflow_runtime/six-workflow-patterns.png b/web/public/course-assets/s16_workflow_runtime/six-workflow-patterns.png new file mode 100644 index 000000000..a4f4c06a8 Binary files /dev/null and b/web/public/course-assets/s16_workflow_runtime/six-workflow-patterns.png differ diff --git a/web/public/course-assets/s16_workflow_runtime/workflow-primitives.png b/web/public/course-assets/s16_workflow_runtime/workflow-primitives.png new file mode 100644 index 000000000..06f4de8af Binary files /dev/null and b/web/public/course-assets/s16_workflow_runtime/workflow-primitives.png differ