KEEL · 龙骨 · A CURRICULUM FOR THE AI ERA

08 · 长任务如何暂停、恢复和取消？ — keel 龙骨

## 现场：二十分钟的活，第七分钟被重启了

现场：二十分钟的活，第七分钟被重启了

Incident Bridge 上线后加了一个「跨系统根因诊断」功能：串起事件库、调用链、部署记录，最终产出一份带引用的结论。单次运行约二十分钟。

上线第五天，用户在任务进行到第七分钟时补充了服务名并提交，得到的回答是：

「找不到这个任务，请重新开始。」

后台一查，那次提交撞上了例行部署——进程被重启了。更糟的是：重启前所有处于 input_required、正等着用户补充信息的任务，全部消失了。

原因在 TaskStore 的默认构造：

def __init__(self, path: str | Path | None = None) -> None:
    self.path = Path(path) if path else None
    self.tasks: dict[str, TaskSnapshot] = {}

不传路径就是纯内存字典。进程一死，全部烟消云散。

本章要解决的是这一整类问题：长任务的生命周期怎么建模，才能让「等待」「重启」「重试」都不至于丢东西。

直觉模型：请求 ≠ 任务

先立一条分界线：

一句话：短请求返回答案，长任务返回承诺。 承诺的内容是一个可以在之后被查询、被补全、被取消的稳定句柄。

在 2026-07-28 版本里，这个能力不再是核心协议的一部分，而是扩展：

Tasks: Asynchronous execution of long-running operations, with polling, mid-flight input, and durable handles

（来源：modelcontextprotocol.io 规范总览 Extensions 章节，检索于 2026-09-29）

注意这段描述里的三个关键词，它们正好对应本章三节：polling（轮询）、mid-flight input（中途输入）、durable handles（持久句柄）。

既然是扩展，就要回到第 03 章那条规则：扩展必须协商（capabilities.extensions 里带 io.modelcontextprotocol/tasks），一方支持另一方不支持时，支持方要么退回核心行为，要么明确拒绝。

精确定义：状态机与终态

一个任务最小需要：task_id、status、input_schema、input_value、result、idempotency_key。本项目用一个 dataclass 表达：

@dataclass
class TaskSnapshot:
    task_id: str
    status: str = "working"
    input_schema: dict[str, Any] | None = None
    input_value: dict[str, Any] | None = None
    result: dict[str, Any] | None = None
    idempotency_key: str | None = None

TERMINAL_STATES = {"completed", "failed", "cancelled"}

状态迁移：

working ──► input_required ──► working ──► completed
   │              │               │
   └──────────────┴───────────────┴──► failed
                                   └─► cancelled

TERMINAL_STATES 的作用是把「已经结束」这件事变成可判定的。看它怎么被用：

def complete(self, task_id: str, result: dict[str, Any]) -> TaskSnapshot:
    task = self.get(task_id)
    if task.status in TERMINAL_STATES:
        return task          # 已终态：直接返回既有快照，不改写
    task.result = result
    task.status = "completed"
    self._save()
    return task

幂等性从这一行开始。 一个已经完成的任务再次被要求完成，不会报错也不会被改写，而是返回既有快照——结果不会被第二次调用覆盖。

一次完整运行

python courses/foundation/mcp-protocol-engineering/course/project/examples/07_task_recovery.py

实跑输出：

restored: 34df0ab7-eac3-429b-902f-efa8ea132d2f input_required
input: working
completed: completed

这个示例的关键在它的写法：它构造了两个 TaskStore 实例，指向同一个 JSON 文件。

first = TaskStore(path)
task = first.create({"type": "object", "required": ["service"]}, idempotency_key="incident-1")
second = TaskStore(path)                       # ← 模拟进程重启：全新的对象，"忘了"第一个
restored = second.get(task.task_id)

第二个实例内部字典是空的，但它在构造时把 JSON 读回来了：

if self.path and self.path.exists():
    payload = json.loads(self.path.read_text(encoding="utf-8"))
    self.tasks = {key: TaskSnapshot(**value) for key, value in payload.items()}

三行输出验证的是同一件事：状态跨实例存活。第一行证明它会带着 input_required 回来，第二行证明可以接着补输入，第三行证明能走到终态。

这正是现场那个事故的修复方向——把 TaskStore() 换成 TaskStore(path)，一行改动。

失败注入

注入 A：幂等键去重

first = store.create({"type": "object", "required": ["service"]}, idempotency_key="incident-1")
again = store.create({"type": "object", "required": ["service"]}, idempotency_key="incident-1")

实跑输出：

8A. idempotent create -> same task: True | status: input_required

同一个幂等键返回同一个 task_id，没有创建第二个任务。

为什么这一条不能省？因为网络层的不确定性意味着客户端无法区分「请求丢了」和「响应丢了」。前者需要重发，后者重发会重复执行副作用。幂等键让「重发一定安全」，于是客户端可以无脑重试，而不必先猜网络发生了什么——这正是第 06 章检查清单里「重试是否可能重复副作用」那一项的答案。

注入 B：错误状态下提交输入

working = store.create()                      # 没有 input_schema → 直接 working
store.provide_input(working.task_id, {"service": "payments"})

实跑输出：

8B. provide_input on working -> task is not waiting for input: working

状态机越界被响亮地拒绝，错误信息里带上当前状态，方便排查。

注入 C：重复完成

first = store.complete(task_id, {"summary": "no active incident"})
second = store.complete(task_id, {"summary": "rewritten!"})

实跑输出：

8C. complete twice -> status: completed | result unchanged: True

第二次传的结果被丢弃了，保留第一次的结论。这是对的：一个已经对外承诺过的结论，不该被后续迟到的写操作改写。

但这里有一条实践要求：调用方必须从返回值读真实状态，不能假设自己的写生效了。 沉默的成功最危险。

注入 D：完成后取消

实跑输出：

8D. cancel after completed -> status: completed

cancel 对终态静默无效。对照 provide_input 抛异常——本项目在这两处策略不一致，是个值得留意的设计不整齐。生产实现应该给出统一回答：要么都明确报错让用户知道，要么都返回当前状态让用户判断，不要一半一半。

注入 E：未知句柄

实跑输出：

8E. unknown task -> 'unknown task: no-such-task'

句柄不存在要明确报错，不要返回空。理由和第 05 章「权限错误不能装成查不到」一样：「不存在」和「存在但没有结果」是两件不同的事，混在一起会让上层做出错误决策。

持久化：这次实现的上限在哪

_save() 是这么写的：

def _save(self) -> None:
    if not self.path:
        return
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self.path.write_text(json.dumps({key: asdict(value) for key, value in self.tasks.items()},
                                    ensure_ascii=False, indent=2), encoding="utf-8")

每次操作后把整个字典重写一遍。 这个写法在教学场景里非常好：单进程、顺序执行、肉眼可读、易于讲解。但要知道它的三个硬限制：

1. 不是原子的。写到一半断电，文件是残缺 JSON，__init__ 里那句 json.loads 会抛异常，任务恢复直接失败。
2. 并发会丢写。两个操作同时算完、各自全量覆盖，后写赢，前一次的结果没了。
3. 规模不适配。任务数量上千后，每次全量重写整个文件的开销不可接受。

对应的生产替换很清楚：数据库行级更新 + 事务 + 乐观锁/租约。_save() 的职责在教学里是「让你看见持久化确实发生了」，不是「演示如何做高并发落盘」。

另外三个生产上必须有、本项目没有的东西：

与 input_required 的衔接

第 05 章讲过 Elicitation 的教学替身。这里补全它的服务端视角：TaskStore.create 有个细节——

task = TaskSnapshot(str(uuid.uuid4()),
                    "input_required" if input_schema else "working",
                    input_schema, idempotency_key=idempotency_key)

只要你传了 input_schema，任务创建时就直接进入 input_required，而不是先跑一段再饿着停下来。

这是个好的设计习惯：「需要什么」必须与任务一起持久化。 如果 schema 只存在内存里，进程重启后你连「该问用户要什么字段」都忘了——任务会卡在一个既不能继续也不能追问的状态。

生产替换点

练习与验收

练习（有可观察结果）：实现一个 reclaim(timeout_seconds) 方法：扫描所有非终态任务，把超过 timeout_seconds 未更新的（需要给 TaskSnapshot 加 updated_at）标记 为 failed，并返回一份能被解释清楚的清单。

验收标准：创建一个任务、睡到超时、调用 reclaim，断言它变成 failed；再对一个 completed 的任务调用 reclaim，断言它不受影响（终态判定生效）。

本章检查点

• 现场那个事故，只加一行 TaskStore(path) 就够了吗？还缺什么才能在真实部署里存活？
• 为什么幂等键必须让重发安全，而不是让客户端去区分「请求丢了」和「响应丢了」？
• complete 静默返回结果、而 provide_input 抛异常，这种不一致会带来什么排查困难？
• _save() 整体重写 JSON，在什么并发模式下会丢数据？

现在能解释什么

你现在能解释为什么长任务不能直接当成普通请求：它需要持久句柄、需要能中途等待输入、需要能在进程生命周期之外存活。你也亲手验证了这个状态机的行为——包括它静默的那些地方。幂等键部分你知道它的价值是让「重发一定安全」。下一章解决最后一个问题：你怎么向别人证明自己的实现对各种客户端都兼容，以及出了问题怎么查。

进入 keel 阅读