S16 Workflow Runtime：用可恢复脚本编排多步工作

changes = """def load_user(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    return db.execute(query).fetchone()
"""

out = await ctx.agent(
    f"检查以下代码改动中的安全问题，只报告有依据的问题：\n{changes}"
)

候选问题：SQL 查询直接拼入了 user_id。
如果 user_id 可被外部控制，可能存在 SQL 注入风险。
建议使用参数化查询。

{
  "findings": [
    {"title": "user_id 直接拼接进 SQL，可能引入注入风险", "severity": "high"}
  ]
}

FINDINGS_SCHEMA = {
    "type": "object",                  # 整个回答是对象，对应 Python 字典
    "required": ["findings"],           # 必须包含 findings 字段
    "properties": {
        "findings": {
            "type": "array",           # findings 是列表
            "items": {                 # 列表中每一项的规则
                "type": "object",
                "required": ["title", "severity"],
                "properties": {
                    "title": {"type": "string"},
                    "severity": {
                        "type": "string",
                        "enum": ["high", "medium", "low"],
                    },
                },
            },
        },
    },
}

out = await ctx.agent(
    f"检查以下代码改动中的安全问题，只报告有依据的问题：\n{changes}",
    schema=FINDINGS_SCHEMA,
)

findings = out["findings"]

finding = findings[0]

VERDICT_SCHEMA = {
    "type": "object",
    "required": ["isReal", "reason"],   # 两个字段都必须提供
    "properties": {
        "isReal": {"type": "boolean"}, # 是否成立：布尔值
        "reason": {"type": "string"},  # 判断理由：字符串
    },
}

import json

verdict = await ctx.agent(
    f"根据代码改动，复核这条候选问题是否成立。\n"
    f"代码改动：\n{changes}\n"
    f"候选问题：\n{json.dumps(finding, ensure_ascii=False)}",
    schema=VERDICT_SCHEMA,
)

{
  "isReal": true,
  "reason": "代码将 user_id 直接拼入 SQL；若它来自外部输入，可能改变查询语义。"
}

confirmed = []
if verdict["isReal"]:
    confirmed.append(finding)

confirmed = []
for finding in findings:
    verdict = await ctx.agent(
        f"根据代码改动，复核这条候选问题是否成立。\n"
        f"代码改动：\n{changes}\n"
        f"候选问题：\n{json.dumps(finding, ensure_ascii=False)}",
        schema=VERDICT_SCHEMA,
    )
    if verdict["isReal"]:
        confirmed.append(finding)

import json


async def sample_workflow(ctx, args):
    changes = args.get("changes", "")

    # 检查：得到候选问题列表
    out = await ctx.agent(
        f"检查以下代码改动中的安全问题，只报告有依据的问题：\n{changes}",
        schema=FINDINGS_SCHEMA,
    )
    findings = out["findings"]

    # 逐条复核，并保留被判定成立的问题
    confirmed = []
    for finding in findings:
        verdict = await ctx.agent(
            f"根据代码改动，复核这条候选问题是否成立。\n"
            f"代码改动：\n{changes}\n"
            f"候选问题：\n{json.dumps(finding, ensure_ascii=False)}",
            schema=VERDICT_SCHEMA,
        )
        if verdict["isReal"]:
            confirmed.append(finding)

    return {"confirmed": confirmed}

检查 → 复核 A → 筛选 A → 复核 B → 筛选 B → 返回结果

tasks = [
    lambda f=f: ctx.agent(
        f"根据代码改动，复核这条候选问题是否成立。\n"
        f"代码改动：\n{changes}\n"
        f"候选问题：\n{json.dumps(f, ensure_ascii=False)}",
        schema=VERDICT_SCHEMA,
    )
    for f in findings
]

async def sample_workflow(ctx, args):
    changes = args.get("changes", "")
    # ...沿用第 5 节的检查调用，得到 out...
    ...
    findings = out["findings"]

    # ...使用上面的代码，为 findings 中每条问题准备 tasks...
    ...
    verdicts = await ctx.parallel(tasks)

    confirmed = []
    for finding, verdict in zip(findings, verdicts):
        if verdict["isReal"]:
            confirmed.append(finding)

    return {"confirmed": confirmed}

检查 → 得到 A、B → 准备 tasks
                         ↓
                 ctx.parallel(tasks)
                   ├→ 复核 A ─┐
                   └→ 复核 B ─┴→ 等齐 verdicts
                                      ↓
                        与 findings 按位置配对
                                      ↓
                             筛选 → 返回结果

DIMENSIONS = ["correctness", "security", "performance", "style"]

async def audit(_value, dimension, _idx):
    out = await ctx.agent(
        f"检查以下代码改动中的 {dimension} 问题，只报告有依据的问题：\n{changes}",
        schema=FINDINGS_SCHEMA,
    )
    return {"dimension": dimension, "findings": out["findings"]}

async def verify(audited, dimension, _idx):
    tasks = [
        lambda f=f: ctx.agent(
            f"根据代码改动，复核这条 {dimension} 候选问题是否成立。\n"
            f"代码改动：\n{changes}\n"
            f"候选问题：\n{json.dumps(f, ensure_ascii=False)}",
            schema=VERDICT_SCHEMA,
        )
        for f in audited["findings"]
    ]
    verdicts = await ctx.parallel(tasks)

    confirmed = []
    for finding, verdict in zip(audited["findings"], verdicts):
        if verdict["isReal"]:
            confirmed.append(finding)
    return {"dimension": dimension, "confirmed": confirmed}

async def sample_workflow(ctx, args):
    changes = args.get("changes", "")

    async def audit(_value, dimension, _idx):
        # ...上面的检查代码：使用 changes 和 dimension，得到 out...
        ...
        return {"dimension": dimension, "findings": out["findings"]}

    async def verify(audited, dimension, _idx):
        # ...上面的复核代码：读取 audited["findings"]，得到 confirmed...
        ...
        return {"dimension": dimension, "confirmed": confirmed}

    results = await ctx.pipeline(DIMENSIONS, audit, verify)
    # 下一节在这里接上汇总、排序和最终返回。

ctx.pipeline(DIMENSIONS, audit, verify)

安全性：audit → audited → verify → 该维度结果 ─┐
性能：  audit → audited → verify → 该维度结果 ─┴→ 等齐 → results

confirmed = []
for result in results:
    for finding in result["confirmed"]:
        confirmed.append({"dimension": result["dimension"], **finding})

async def sample_workflow(ctx, args):
    # ...读取 changes，定义上一节的 audit 和 verify...
    ...
    results = await ctx.pipeline(DIMENSIONS, audit, verify)

    confirmed = []
    for result in results:
        for finding in result["confirmed"]:
            confirmed.append({"dimension": result["dimension"], **finding})

    severity_order = {"high": 0, "medium": 1, "low": 2}
    confirmed.sort(key=lambda finding: severity_order.get(finding["severity"], 3))
    return {"confirmed": confirmed}

results：按维度分组的确认项
          ↓ 展开各组，为每条问题补上 dimension
confirmed：合并后的问题列表
          ↓ 按 high → medium → low 排序
返回 {"confirmed": confirmed}

{
  "confirmed": [
    {
      "dimension": "security",
      "title": "user_id 直接拼接进 SQL，可能引入注入风险",
      "severity": "high"
    }
  ]
}

async def audit(_value, dimension, _idx):
    out = await ctx.agent(
        f"检查以下代码改动中的 {dimension} 问题，只报告有依据的问题：\n{changes}",
        schema=FINDINGS_SCHEMA,
        label=f"audit:{dimension}",  # 本节新增：为这次检查命名
    )
    return {"dimension": dimension, "findings": out["findings"]}

async def verify(audited, dimension, _idx):
    tasks = [
        lambda f=f: ctx.agent(
            f"根据代码改动，复核这条 {dimension} 候选问题是否成立。\n"
            f"代码改动：\n{changes}\n"
            f"候选问题：\n{json.dumps(f, ensure_ascii=False)}",
            schema=VERDICT_SCHEMA,
            label=f"verify:{dimension}:{f['title']}",  # 本节新增
        )
        for f in audited["findings"]
    ]
    # ...await ctx.parallel(tasks)，再配对筛选得到 confirmed...
    ...
    return {"dimension": dimension, "confirmed": confirmed}

audit:security
verify:security:user_id 直接拼接进 SQL，可能引入注入风险

async def sample_workflow(ctx, args):
    ctx.phase("Review")  # 本节新增：记录检查阶段
    changes = args.get("changes", "")

    async def audit(_value, dimension, _idx):
        out = await ctx.agent(
            ...,  # 沿用第 9 节的检查提示词，包含 changes
            schema=FINDINGS_SCHEMA,
            label=f"audit:{dimension}",
            phase="Review",  # 本节新增：这次调用属于检查
        )
        return {"dimension": dimension, "findings": out["findings"]}

    async def verify(audited, dimension, _idx):
        ctx.phase("Verify")  # 本节新增：记录复核阶段
        tasks = [
            lambda f=f: ctx.agent(
                ...,  # 沿用第 9 节的复核提示词，包含 changes 和 f
                schema=VERDICT_SCHEMA,
                label=f"verify:{dimension}:{f['title']}",
                phase="Verify",  # 本节新增：这次调用属于复核
            )
            for f in audited["findings"]
        ]
        # ...并发执行 tasks，再配对筛选得到 confirmed...
        ...
        return {"dimension": dimension, "confirmed": confirmed}

    results = await ctx.pipeline(DIMENSIONS, audit, verify)
    # ...沿用第 8 节：合并、排序，得到总的 confirmed...
    ...
    return {"confirmed": confirmed}

label=audit:security          phase=Review
label=verify:security:某条问题  phase=Verify

async def sample_workflow(ctx, args):
    # ...沿用上一节的输入读取、阶段标记，以及 audit、verify 定义...
    ...
    results = await ctx.pipeline(DIMENSIONS, audit, verify)

    confirmed = []
    for result in results:
        for finding in result["confirmed"]:
            confirmed.append({"dimension": result["dimension"], **finding})

    severity_order = {"high": 0, "medium": 1, "low": 2}
    confirmed.sort(key=lambda finding: severity_order.get(finding["severity"], 3))

    ctx.log(f"confirmed {len(confirmed)} real finding(s)")  # 本节新增
    return {"confirmed": confirmed}

confirmed 3 real finding(s)

安全性检查：已完成，得到两条候选问题
问题 A 复核：已完成，得到复核结论
问题 B 复核：请求失败，没有取得复核结论

此前讲到：取得结果 → 校验结果 → 返回给业务脚本
本节补上：取得结果 → 校验结果 →【保存到 journal】→ 返回给业务脚本

class ExecutionState:
    async def agent(self, prompt, schema=None, label=None, phase=None):
        label = label or (prompt[:24] + "...")
        # ...检查调用次数与预算...

        # 本节展开：请求模型前，用调用参数生成标识
        key = self.journal.key("agent", label, prompt, schema)
        # ...按 key 查询已保存的结果，复用流程后面展开...

        # 1. 调用模型，取得结果
        async with self._limits.semaphore:
            run = await asyncio.to_thread(
                self.runner.run, prompt, schema, label
            )
            result = run.value
            tokens = run.tokens

        # 2. 已讲过的结构校验：不符合 schema 时重试一次
        if schema is not None:
            ...

        # ...记录本次用量...

        # 3. 本节展开：返回之前，把这次调用的结果保存下来
        self.journal.record(key, result)

        # 4. 记录这次调用已经完成
        self.task.progress_event("workflow_agent", label=label,
                                 phase=phase or self._phase, status="done")

        # 5. 返回业务脚本，成为 out 的值
        return result

"agent" + label + 完整 prompt + schema → key ─┐
模型返回值 → 通过结构校验 → result ────────────┴→ journal：{"key": key, "value": result}

{
  "key": "agent-2419530705",
  "value": {
    "findings": [
      {"title": "候选问题 A", "severity": "high"},
      {"title": "候选问题 B", "severity": "medium"}
    ]
  }
}

                 ┌─ 有记录 → 校验已有结果 → 返回给脚本
计算 key → 查找 ─┤
                 └─ 无记录 → 调用模型 → 校验、保存 → 返回给脚本

class ExecutionState:
    async def agent(self, prompt, schema=None, label=None, phase=None):
        label = label or (prompt[:24] + "...")
        # ...检查调用次数与预算...
        key = self.journal.key("agent", label, prompt, schema)

        # 本节展开：先按 key 查找已保存的结果
        cached = self.journal.cached(key)
        if cached is not MISS:
            # ...如果指定了 schema，就校验 cached；不合格则报错...
            ...
            self.task.progress_event("workflow_agent", label=label,
                                     phase=phase or self._phase, status="cached")
            return cached

        # 没有记录才走这里：调用模型，校验结果并统计用量
        ...
        self.journal.record(key, result)
        self.task.progress_event("workflow_agent", label=label,
                                 phase=phase or self._phase, status="done")
        return result

first = await run_workflow(
    name="review-changes",
    args={"changes": changes},
)
run_id = first["launched"]["runId"]

resumed = await run_workflow(
    name="review-changes",
    resume_from_run_id=run_id,
)

class WorkflowJournal:
    def __init__(self, run_id, resume, store=None):
        # ...确定并创建记录目录 store...
        self.path = store / f"{run_id}.journal.jsonl"
        self.cache = {}

        if resume:
            # ...检查记录文件存在...
            for line_number, line in enumerate(
                self.path.read_text(encoding="utf-8").splitlines(), start=1
            ):
                rec = json.loads(line)
                # ...检查这条记录包含合法的 key 和 value...
                self.cache[rec["key"]] = rec["value"]
            self._f = self.path.open("a", encoding="utf-8")
        else:
            self._f = self.path.open("w", encoding="utf-8")

    def cached(self, key):
        return self.cache.get(key, MISS)

原 runId → 读取原输入与 journal → 从头执行 sample_workflow(ctx, args)
                                      ↓
                          每次 ctx.agent() 按 key 查找
                            ├─ 命中：复用已有结果
                            └─ 未命中：调用模型，保存新结果

SAMPLE_META = {
    "name": "review-changes",
    "description": "Review changed files across dimensions, verify each finding",
    "phases": ["Review", "Verify"],
}

WORKFLOWS = {SAMPLE_META["name"]: (SAMPLE_META, sample_workflow)}

async def run_workflow(name, args=None, resume_from_run_id=None):
    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")

    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"]),
    }

class WorkflowTool:
    async def _call_locked(self, meta, script_fn, args, run_id, resuming):
        # ...准备本次运行需要的 ctx...
        ...
        result = await script_fn(ctx, args)
        # ...记录运行状态，保存并返回 result...

run_workflow(name="review-changes", args=...)
                    ↓ 用 name 查找
       WORKFLOWS[name] → meta + script_fn
                    ↓
       WorkflowTool().call(meta, script_fn, args=...)
                    ↓ 准备 ctx，执行 script_fn(ctx, args)
       run_workflow 整理并返回 launched / result / task

async def review_and_count(ctx, args):
    changes = args.get("changes", "")

    review = await ctx.workflow(
        "review-changes",
        {"changes": changes},
    )

    return {
        "confirmed": review["confirmed"],
        "count": len(review["confirmed"]),
    }

review_and_count(ctx, args)
          ↓ 等待 ctx.workflow("review-changes", {"changes": changes})
      子流程 sample_workflow(child, 子流程参数)
          ↓ 返回 {"confirmed": ...}
      父流程中的 review 接收结果
          ↓ 计算 count
      父流程返回 {"confirmed": ..., "count": ...}

REVIEW_AND_COUNT_META = {
    "name": "review-and-count",
    "description": "Run the saved review and count confirmed findings",
}
WORKFLOWS[REVIEW_AND_COUNT_META["name"]] = (
    REVIEW_AND_COUNT_META,
    review_and_count,
)

out = await run_workflow(
    name="review-and-count",
    args={"changes": changes},
)

class ExecutionState:
    async def workflow(self, name, args=None):
        if self._depth >= 1:
            raise WorkflowInputError("workflow() nesting is one level only")
        if name not in WORKFLOWS:
            raise WorkflowInputError(f"unknown workflow '{name}'")
        meta, fn = WORKFLOWS[name]
        child = ExecutionState(self.task, self.journal, self.runner, self.budget,
                               args or {}, depth=self._depth + 1,
                               limits=self._limits)
        return await fn(child, args or {})

class WorkflowTool:
    async def _call_locked(self, meta, script_fn, args, run_id, resuming):
        if resuming:
            # ...校验原运行信息，并恢复或核对 args（第 14 节）...
            ...
            journal = WorkflowJournal(run_id, resume=True)
        else:
            args = args or {}
            journal = WorkflowJournal(run_id, resume=False)

        task_id = create_task_id(run_id)
        task = LocalWorkflowTask(task_id, run_id, meta)
        # ...发出启动事件，保存本次运行的初始快照...

        ctx = ExecutionState(
            task, journal, RUNNER_FACTORY(), Budget(args.get("budget")), args
        )
        result = await script_fn(ctx, args)
        # ...处理成功或失败，关闭 journal，保存并返回结果...

class ExecutionState:
    def __init__(self, task, journal, runner, budget, args, depth=0, limits=None):
        self.task = task
        self.journal = journal
        self.runner = runner
        self.budget = budget
        self.args = args
        self._depth = depth
        self._phase = None
        self._phases_seen = set()
        self._limits = limits or ExecutionLimits()

WorkflowTool 准备本次运行的对象
                ↓
        创建 ExecutionState，得到 ctx
                ↓
        调用 sample_workflow(ctx, args)
                ↓
           取得脚本返回的 result

class WorkflowTool:
    async def _call_locked(self, meta, script_fn, args, run_id, resuming):
        # ...准备 journal、task 和启动信息 launched（第 17 节）...
        ...

        try:
            ctx = ExecutionState(
                task, journal, RUNNER_FACTORY(), Budget(args.get("budget")), args
            )
            result = await script_fn(ctx, args)
            task.status = "completed"
        except Exception as e:
            task.status = "failed"
            result = {"error": str(e)}
        finally:
            journal.close()

        # 接下来保存最终状态与输出，再返回。

class WorkflowTool:
    async def _call_locked(self, meta, script_fn, args, run_id, resuming):
        # ...上面的准备与 try/except/finally，得到 task 和 result...
        ...

        _write_json(STORE / f"{run_id}.output.json", result)
        _write_json(STORE / f"{run_id}.json", {
            "runId": run_id,
            "workflowName": meta["name"],
            "args": args,
            "task": serialize_task(task),
        })
        _save_last_run(run_id)
        task.event("task_notification", status=task.status,
                   agents=task.usage["agents"], tokens=task.usage["tokens"],
                   outputFile=f".runtime/{run_id}.output.json")
        return {"launched": launched, "result": result, "task": task}

执行 script_fn(ctx, args)
    ├─ 正常返回 ──────→ completed + 脚本结果 ─┐
    └─ 抛出 Exception → failed + 错误说明 ────┤
                                             ↓
                                  finally：关闭 journal
                                             ↓
                                  保存最终状态与输出
                                             ↓
                                  返回 launched / result / task

resumed = await run_workflow(
    name="review-changes",
    resume_from_run_id=out["task"]["runId"],
)

out = await run_workflow(
    name="review-changes",
    args={"changes": changes, "budget": 1000},
)

class ExecutionState:
    async def agent(self, prompt, schema=None, label=None, phase=None):
        # ...次数与预算检查；若命中 journal 缓存，直接返回已有结果...
        ...

        async with self._limits.semaphore:
            run = await asyncio.to_thread(
                self.runner.run, prompt, schema, label
            )
            result = run.value
            tokens = run.tokens

        # ...按 schema 校验，必要时重试；记账、保存 journal、更新进度...
        ...
        return result

RUNNER_FACTORY = MockAgentRunner


def install_workflow_tool(host):
    global RUNNER_FACTORY
    RUNNER_FACTORY = lambda: AnthropicAgentRunner(host.client, host.MODEL)
    # ...将 Workflow 工具注册到宿主...
    ...

S15 模型选择 Workflow(name="review-changes", args=...)
    ↓
run_workflow_sync() → run_workflow() → sample_workflow(ctx, args)
    ↓
返回 launched / result / task
    ↓
S15 写入 tool_result → messages[] → 下一轮模型调用

def install_workflow_tool(host):
    global RUNNER_FACTORY
    RUNNER_FACTORY = lambda: AnthropicAgentRunner(host.client, host.MODEL)
    if getattr(host, "_workflow_tool_installed", False):
        return
    base_assemble = host.assemble_tool_pool

    def assemble_with_workflow():
        tools, handlers = base_assemble()
        if not any(tool.get("name") == "Workflow" for tool in tools):
            tools.append(WORKFLOW_TOOL)
        handlers["Workflow"] = run_workflow_sync
        return tools, handlers

    host.assemble_tool_pool = assemble_with_workflow
    host._workflow_tool_installed = True

def run_workflow_sync(**tool_input):
    try:
        return json.dumps(asyncio.run(run_workflow(**tool_input)), default=str)
    except WorkflowInputError as exc:
        return f"Error: {exc}"

python3 s16_workflow_runtime/code.py demo
python3 s16_workflow_runtime/code.py resume