先看到“还在运行”，再拿到真的结果
这次仍用 main.py 和 expected.txt。区别是，运行工具先交回启动收据，主循环马上读取预期文件。程序还没结束时，固定模型会如实回答“还在运行”；宿主随后取得输出，标明这份结果还没有交给模型。思想篇解释了这几个交接为什么需要分开。
源码位于独立公开仓库 learn-agent-harness 的第十三课。Python3.10+，标准库即可运行，不需要 API Key：
bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s13-background-tasks/user_code/practice.py --decision allow
python3 examples/s13-background-tasks/user_code/practice.py --decision deny允许路径先保存 bg-1 / started，再读到 expected.txt 中的 42。默认程序等待 0.15 秒后输出 42；通常模型先结束，CLI 最后打印 completed、exit_code=0、output=42，以及一份未交给模型的新通知。机器调度也可能让结果在下一轮前到达，因此测试用文件信号精确控制先后，而不拿固定延迟当正确性依据。
模型响应是固定替身，后台进程与文件读取是真实执行。下面三块交互则直接执行公开函数正文，进程对象替换为可查看的 TeachingHandle：它返回预设快照，不在浏览器里启动进程。代码断点是唯一游标，右侧图跟随实际执行轨迹；切换输入可以观察同一函数走向不同分支。
start_background：先检查，再启动，最后交回收据
输入包含原调用 call、会话登记 jobs、宿主绑定的 launch，以及本次许可 allowed。模型只能请求 run_background 和空参数，不能传任意命令或目录。launch 由装配层固定为当前解释器运行工作区的 main.py。
函数先拒绝错误工具与参数，然后检查这个调用 ID 是否已登记。重复 ID 直接取旧收据；新 ID 才继续检查许可和容量。只有布尔 True 才允许启动，最多登记四项，已经完成的项也算在内，防止模型不断换 ID 启动程序。
真实执行轨迹回放；正式函数原样执行，进程对象是公开快照替身，不启动进程或访问模型。
Pys13/core/runtime.py · start_background
未运行
def start_background(call, jobs, launch, allowed=False):
"""只启动固定main.py；收据只说明已启动，不说明运行成功。"""
receipt = {"type": "tool_result", "tool_use_id": call["id"]}
if call.get("name") != "run_background" or call.get("input") != {}:
return {**receipt, "content": "未执行：工具或参数不支持。", "is_error": True}
if call["id"] in jobs:
return {**receipt, "content": jobs[call["id"]]["receipt"], "is_error": False}
if allowed is not True or len(jobs) >= 4:
return {**receipt, "content": "未执行：没有本次许可或会话登记已满。", "is_error": True}
job_id = "bg-" + str(len(jobs) + 1)
try:
handle = launch()
except OSError as error:
return {**receipt, "content": "无法启动：" + str(error)[:160], "is_error": True}
text = json.dumps({"job_id": job_id, "status": "started"})
jobs[call["id"]] = {"job_id": job_id, "handle": handle, "receipt": text, "result": None}
return {**receipt, "content": text, "is_error": False}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
def start_background(call, jobs, launch, allowed=False):
"""只启动固定main.py；收据只说明已启动，不说明运行成功。"""
receipt = {"type": "tool_result", "tool_use_id": call["id"]}
if call.get("name") != "run_background" or call.get("input") != {}:
return {**receipt, "content": "未执行：工具或参数不支持。", "is_error": True}
if call["id"] in jobs:
return {**receipt, "content": jobs[call["id"]]["receipt"], "is_error": False}
if allowed is not True or len(jobs) >= 4:
return {**receipt, "content": "未执行：没有本次许可或会话登记已满。", "is_error": True}
job_id = "bg-" + str(len(jobs) + 1)
try:
handle = launch()
except OSError as error:
return {**receipt, "content": "无法启动：" + str(error)[:160], "is_error": True}
text = json.dumps({"job_id": job_id, "status": "started"})
jobs[call["id"]] = {"job_id": job_id, "handle": handle, "receipt": text, "result": None}
return {**receipt, "content": text, "is_error": False}
def collect_background(jobs):
"""非阻塞检查；完成记录留在登记表，直到会话结束。"""
completed = []
for call_id, job in jobs.items():
if job["result"] is None:
snapshot = job["handle"].snapshot()
if snapshot is not None:
job["result"] = {"job_id": job["job_id"], "call_id": call_id, **snapshot}
if job["result"] is not None:
completed.append(deepcopy(job["result"]))
return completed
def prepare_background(history, jobs):
"""生成历史候选；通知是普通文本，不是原调用的第二个tool_result。"""
candidate = deepcopy(history)
for event in collect_background(jobs):
text = "[background-result] " + json.dumps(event, ensure_ascii=False, sort_keys=True)
present = any(message.get("role") == "user" and message.get("content") == text
for message in candidate)
if not present:
candidate.append({"role": "user", "content": text})
return candidate
class TeachingHandle:
def __init__(self, result=None):
self.result = result
self.checks = 0
def snapshot(self):
self.checks += 1
return deepcopy(self.result)
scenario = 'start'
name = 'start_background'
jobs = {}
call = {'id': 'call_1', 'name': 'run_background', 'input': {}}
allowed = scenario != 'denied'
launched = []
finished = {'status': 'completed', 'exit_code': 0, 'output': '42', 'is_error': False}
if scenario == 'failure':finished.update(exit_code=1, output='ZeroDivisionError', is_error=True)
if scenario == 'timeout':finished.update(status='timeout', exit_code=-9, output='', is_error=True)
def launch():
if scenario == 'launch_error':raise OSError('教学启动失败')
launched.append('main.py')
return TeachingHandle()
if scenario == 'invalid':call['input'] = {'command': 'other'}
if name != 'start_background' or scenario in ('duplicate', 'full'):
start_background(call, jobs, launch, True)
if scenario == 'full':
for i in range(2, 5):start_background({**call, 'id': 'call_'+str(i)}, jobs, launch, True)
call['id'] = 'call_5'
if name != 'start_background' and scenario not in ('running', 'pending'):
jobs['call_1']['handle'].result = finished
if scenario == 'cached':collect_background(jobs)
history = [{'role': 'user', 'content': '后台运行main.py。'},
{'role': 'assistant', 'content': [{'type': 'tool_use', **call}]},
{'role': 'user', 'content': [{'type': 'tool_result', 'tool_use_id': 'call_1', 'content': 'started'}]}]
if scenario == 'already':history = prepare_background(history, jobs)
if scenario == 'discarded':prepare_background(history, jobs)
before = deepcopy(history)
# 调用入口
observed = start_background(call, jobs, launch, allowed)
assert history == before
print(observed)
- 先校验和检查已有调用ID。
- 取得许可后启动，登记进程对象。
- 交回started收据。
在 handle = launch() 停住时，程序尚未启动。单步过去，才获得真实进程对象；再进入 jobs 的赋值行，下一步才把进程对象、后台 ID、收据和空结果登记起来。最后的 tool_result 内容只有 job_id 和 started，没有伪造退出码。
“相同调用ID”只有一次 launch；“启动失败”没有登记。“登记已满”保留原四项，不启动第五项。调用 ID 去重只覆盖同一会话中的同一个 ID，不能把它描述成跨重启或业务层的恰好一次执行。
collect_background：None 表示现在没有结果
主循环不用等待每一个程序。collect_background 遍历 jobs，仅对还没有 result 的项调用 snapshot。未结束的进程对象返回 None，检查立即转到下一项；已经结束则把后台 ID、原调用 ID 和退出信息固定成记录。
真实执行轨迹回放；正式函数原样执行，进程对象是公开快照替身，不启动进程或访问模型。
Pys13/core/runtime.py · collect_background
未运行
def collect_background(jobs):
"""非阻塞检查；完成记录留在登记表，直到会话结束。"""
completed = []
for call_id, job in jobs.items():
if job["result"] is None:
snapshot = job["handle"].snapshot()
if snapshot is not None:
job["result"] = {"job_id": job["job_id"], "call_id": call_id, **snapshot}
if job["result"] is not None:
completed.append(deepcopy(job["result"]))
return completed输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
def start_background(call, jobs, launch, allowed=False):
"""只启动固定main.py；收据只说明已启动，不说明运行成功。"""
receipt = {"type": "tool_result", "tool_use_id": call["id"]}
if call.get("name") != "run_background" or call.get("input") != {}:
return {**receipt, "content": "未执行：工具或参数不支持。", "is_error": True}
if call["id"] in jobs:
return {**receipt, "content": jobs[call["id"]]["receipt"], "is_error": False}
if allowed is not True or len(jobs) >= 4:
return {**receipt, "content": "未执行：没有本次许可或会话登记已满。", "is_error": True}
job_id = "bg-" + str(len(jobs) + 1)
try:
handle = launch()
except OSError as error:
return {**receipt, "content": "无法启动：" + str(error)[:160], "is_error": True}
text = json.dumps({"job_id": job_id, "status": "started"})
jobs[call["id"]] = {"job_id": job_id, "handle": handle, "receipt": text, "result": None}
return {**receipt, "content": text, "is_error": False}
def collect_background(jobs):
"""非阻塞检查；完成记录留在登记表，直到会话结束。"""
completed = []
for call_id, job in jobs.items():
if job["result"] is None:
snapshot = job["handle"].snapshot()
if snapshot is not None:
job["result"] = {"job_id": job["job_id"], "call_id": call_id, **snapshot}
if job["result"] is not None:
completed.append(deepcopy(job["result"]))
return completed
def prepare_background(history, jobs):
"""生成历史候选；通知是普通文本，不是原调用的第二个tool_result。"""
candidate = deepcopy(history)
for event in collect_background(jobs):
text = "[background-result] " + json.dumps(event, ensure_ascii=False, sort_keys=True)
present = any(message.get("role") == "user" and message.get("content") == text
for message in candidate)
if not present:
candidate.append({"role": "user", "content": text})
return candidate
class TeachingHandle:
def __init__(self, result=None):
self.result = result
self.checks = 0
def snapshot(self):
self.checks += 1
return deepcopy(self.result)
scenario = 'running'
name = 'collect_background'
jobs = {}
call = {'id': 'call_1', 'name': 'run_background', 'input': {}}
allowed = scenario != 'denied'
launched = []
finished = {'status': 'completed', 'exit_code': 0, 'output': '42', 'is_error': False}
if scenario == 'failure':finished.update(exit_code=1, output='ZeroDivisionError', is_error=True)
if scenario == 'timeout':finished.update(status='timeout', exit_code=-9, output='', is_error=True)
def launch():
if scenario == 'launch_error':raise OSError('教学启动失败')
launched.append('main.py')
return TeachingHandle()
if scenario == 'invalid':call['input'] = {'command': 'other'}
if name != 'start_background' or scenario in ('duplicate', 'full'):
start_background(call, jobs, launch, True)
if scenario == 'full':
for i in range(2, 5):start_background({**call, 'id': 'call_'+str(i)}, jobs, launch, True)
call['id'] = 'call_5'
if name != 'start_background' and scenario not in ('running', 'pending'):
jobs['call_1']['handle'].result = finished
if scenario == 'cached':collect_background(jobs)
history = [{'role': 'user', 'content': '后台运行main.py。'},
{'role': 'assistant', 'content': [{'type': 'tool_use', **call}]},
{'role': 'user', 'content': [{'type': 'tool_result', 'tool_use_id': 'call_1', 'content': 'started'}]}]
if scenario == 'already':history = prepare_background(history, jobs)
if scenario == 'discarded':prepare_background(history, jobs)
before = deepcopy(history)
# 调用入口
observed = collect_background(jobs)
assert history == before
print(observed)
- 非阻塞检查进程对象。
- 固定退出信息。
- 结果保留，返回副本。
“仍在运行”交回空列表，不会发起另一次模型请求。“程序错误”仍然有完成记录，只是 exit_code 非零、is_error 为 True；“进程超时”则保留 timeout，不能画成正常成功。“已经收集过”不再次检查进程对象，而是复制已有结果。
这里刻意没有收完就删除 jobs 里的结果。后面的历史候选可能被丢弃，若先把结果取走并清空，就会出现“程序确实完成了，模型却再也收不到”的空档。当前结果最多四份，保留到会话结束，比提前引入复杂确认协议更容易验证。
正式 ProcessHandle 用 Popen.poll检查进程；未结束且未超时返回 None。到达超时就 kill/wait 直接子进程，再读取输出。stdout/stderr 写入临时文件，避免子进程因为输出管道填满而等着没人读；结束时只读取前 2000 字节并关闭文件。截取不等于磁盘配额，也不提供后代进程管理，示例固定运行可信、无子进程的教学文件。
prepare_background：通知进入候选，还不算已经提交
这个函数先深复制 history，再取完成记录，为每份结果生成一条带 [background-result] 标记的普通 user 文本。通知里的 JSON 关联 job_id 与 call_id，模型能把它和早先的启动收据对上。
真实执行轨迹回放；正式函数原样执行，进程对象是公开快照替身，不启动进程或访问模型。
Pys13/core/runtime.py · prepare_background
未运行
def prepare_background(history, jobs):
"""生成历史候选；通知是普通文本，不是原调用的第二个tool_result。"""
candidate = deepcopy(history)
for event in collect_background(jobs):
text = "[background-result] " + json.dumps(event, ensure_ascii=False, sort_keys=True)
present = any(message.get("role") == "user" and message.get("content") == text
for message in candidate)
if not present:
candidate.append({"role": "user", "content": text})
return candidate输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
def start_background(call, jobs, launch, allowed=False):
"""只启动固定main.py；收据只说明已启动，不说明运行成功。"""
receipt = {"type": "tool_result", "tool_use_id": call["id"]}
if call.get("name") != "run_background" or call.get("input") != {}:
return {**receipt, "content": "未执行：工具或参数不支持。", "is_error": True}
if call["id"] in jobs:
return {**receipt, "content": jobs[call["id"]]["receipt"], "is_error": False}
if allowed is not True or len(jobs) >= 4:
return {**receipt, "content": "未执行：没有本次许可或会话登记已满。", "is_error": True}
job_id = "bg-" + str(len(jobs) + 1)
try:
handle = launch()
except OSError as error:
return {**receipt, "content": "无法启动：" + str(error)[:160], "is_error": True}
text = json.dumps({"job_id": job_id, "status": "started"})
jobs[call["id"]] = {"job_id": job_id, "handle": handle, "receipt": text, "result": None}
return {**receipt, "content": text, "is_error": False}
def collect_background(jobs):
"""非阻塞检查；完成记录留在登记表，直到会话结束。"""
completed = []
for call_id, job in jobs.items():
if job["result"] is None:
snapshot = job["handle"].snapshot()
if snapshot is not None:
job["result"] = {"job_id": job["job_id"], "call_id": call_id, **snapshot}
if job["result"] is not None:
completed.append(deepcopy(job["result"]))
return completed
def prepare_background(history, jobs):
"""生成历史候选；通知是普通文本，不是原调用的第二个tool_result。"""
candidate = deepcopy(history)
for event in collect_background(jobs):
text = "[background-result] " + json.dumps(event, ensure_ascii=False, sort_keys=True)
present = any(message.get("role") == "user" and message.get("content") == text
for message in candidate)
if not present:
candidate.append({"role": "user", "content": text})
return candidate
class TeachingHandle:
def __init__(self, result=None):
self.result = result
self.checks = 0
def snapshot(self):
self.checks += 1
return deepcopy(self.result)
scenario = 'new'
name = 'prepare_background'
jobs = {}
call = {'id': 'call_1', 'name': 'run_background', 'input': {}}
allowed = scenario != 'denied'
launched = []
finished = {'status': 'completed', 'exit_code': 0, 'output': '42', 'is_error': False}
if scenario == 'failure':finished.update(exit_code=1, output='ZeroDivisionError', is_error=True)
if scenario == 'timeout':finished.update(status='timeout', exit_code=-9, output='', is_error=True)
def launch():
if scenario == 'launch_error':raise OSError('教学启动失败')
launched.append('main.py')
return TeachingHandle()
if scenario == 'invalid':call['input'] = {'command': 'other'}
if name != 'start_background' or scenario in ('duplicate', 'full'):
start_background(call, jobs, launch, True)
if scenario == 'full':
for i in range(2, 5):start_background({**call, 'id': 'call_'+str(i)}, jobs, launch, True)
call['id'] = 'call_5'
if name != 'start_background' and scenario not in ('running', 'pending'):
jobs['call_1']['handle'].result = finished
if scenario == 'cached':collect_background(jobs)
history = [{'role': 'user', 'content': '后台运行main.py。'},
{'role': 'assistant', 'content': [{'type': 'tool_use', **call}]},
{'role': 'user', 'content': [{'type': 'tool_result', 'tool_use_id': 'call_1', 'content': 'started'}]}]
if scenario == 'already':history = prepare_background(history, jobs)
if scenario == 'discarded':prepare_background(history, jobs)
before = deepcopy(history)
# 调用入口
observed = prepare_background(history, jobs)
assert history == before
print(observed)
- 复制历史并生成文本通知。
- 只跳过已提交历史中的相同通知。
- 整份候选交回原循环。
在 candidate.append 停住时，传入的 history 没变，通知也还没追加。单步后只有 candidate 多出一条；返回后原 run_loop 才把候选提交为新的 messages。它仍是唯一的历史写入位置。
“通知已经提交”检查到历史里已有完全相同的 user 文本，就不再追加。“上次候选被丢弃”虽然之前生成过通知，但传入的原历史没有它，因此这次重新生成。函数不在 jobs 上提前写 notified=True，也不因为曾经读过结果就吞掉它。
原调用只有一份 tool_result 收据，后续通知是普通文本。标记和 JSON 是本例约定，SDK 接收的仍是常规消息；输出中出现的命令、建议或伪造提示都只是程序数据，不能越过原工具权限。
怎么接回已有循环？
本课没有再写一个后台专用 Agent Loop。装配继续复用第一课的 run_loop，把 prepare_background 传入第八课已有的 prepare 边界。新启动工具经过已有 Hook 包装和显式宿主许可，旧工具沿用原分发、许可和 Hook；system 由实际注册组装，父模型请求仍使用第十一课的有界恢复。
python
# user_code/practice.py 的装配，完整实现见公开源码
out = run_loop(
ask, execute, max_calls=max_calls,
question="后台运行main.py，等待时先读expected.txt；只根据真实结果报告。",
prepare=lambda h: prepare_background(h, jobs),
)短会话不叠加历史压缩与跨会话记忆，否则还得先解决“去重依据被压缩移走以后怎么办”。上一课持久任务图也不会被本课进程状态偷偷改写。这里保留的是已有能力的接入点，不是把此前每个机制都塞进一次演示。
模型回答或达到调用上限后，入口停止模型循环。宿主继续以本地检查等到进程结束或超时，显示最后结果与未消费通知，finally 回收所有直接子进程和输出文件。它没有用模型反复查询，也没有暗中增加模型预算。若要结果到达后自动唤醒模型，应该另设计显式的会话恢复与额度规则。
怎样证明没有重复启动，也没有丢通知？
bash
python3 examples/s13-background-tasks/user_code/verify_practice.py
python3 examples/s13-background-tasks/user_code/verify_fixtures.py
.venv/bin/python examples/s13-background-tasks/user_code/verify_sdk.py五组标准库测试涵盖真实进程交错、拒绝与调用上限、除零与超时回收、候选丢弃后重投和模型先结束。交错测试让 main.py 等待一个文件信号：读取 expected.txt 后宿主才发出信号，等真实进程结束，再进入下一轮准备。最终只保留一份原调用收据，并在下一轮看到完成通知。
十五组教学输入对应上面三个原样函数，站内采集器验证了 128 个真实快照。API 与进程边界没有被混称为浏览器真实运行：TeachingHandle 的 snapshot 是明确替身，正式 ProcessHandle 另由真实进程测试覆盖。
SDK 验证使用固定 Anthropic1.13.0 与模拟 HTTP，在携带完成通知的模型请求注入一次 529。允许路径四次 HTTP、拒绝路径两次；重发请求保持同一历史，程序计数文件只写一个 x。没有在线模型调用，也没有把后台任务升级成跨重启恢复系统。