为本地优先 Agent 定义断网降级契约

事故报告只有三行：一次重构任务正在远程服务实例上执行，办公室 Wi-Fi 中断了 11 秒，随后这轮任务没有返回任何结果。本地模型依然加载着，代码仓库依然在磁盘上，用户依然对着转圈的加载图标等待。代码库从未定义过接下来该怎么办，于是用户得到的只有一段堆栈跟踪，以及一轮丢失的任务。

这个缺口不是网络 bug，再怎么重试也补不上。真正缺失的是一份契约，约定两个都能处理同一请求的服务实例该如何协作。下面的模式综合了常见的 local-first Agent 配置，并非某个团队的单次事故，但这类故障反复出现，值得专门写下来。

为什么回退契约比重试循环更有效

重试循环只回答一个问题：再试多少次。回退契约则回答四个问题：哪些故障值得重试，哪些故障应该把这一轮任务交给本地实例，每种情况下用户需要等多久，以及界面如何告诉用户究竟是哪个实例给出了回答。最后一个问题是大多数 local-first 项目都会忽略的，而它恰恰最容易损害用户的信任。

当回答来源不可见时，本地降级回答看起来就和正常的远程回答没有区别。于是，用户会把小模型输出质量较弱的问题归咎于产品，而不是网络。通过契约标明回答来源，就能把一次不可见的降级变成可以解释的事件。相比让本地路径达到远程路径的效果，这种修复的成本低得多。

第 1 步：先给故障分类，再选择执行位置

设计时真正有用的划分不是“在线还是离线”，而是一小组需要采取不同应对方式的故障类别。表中的预算只是供讨论的示例起点，并非实测值；每个团队都应该根据自己的延迟数据来设定。

关键在于，超时不等于取消。请求超过截止时间后，即使调用方已经放弃等待，它仍可能占用远程实例，继续消耗该实例的可用预算。因此，契约需要在第一次尝试之前，就为这一轮任务附上一个幂等键。这样，后续本地回答返回后，就能抑制重复的远程结果，而不是把两份存在分歧的输出合并起来。

第 2 步：把预算编码为数据

只写在文字说明里的预算，一个 sprint 之后就会和代码脱节。把预算放进一个冻结的 dataclass，测试就能构造更紧的预算，而不必修改生产环境的默认值；所有参数也能集中在一个方便审查的位置。

# fallback_contract.py
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field

@dataclass(frozen=True)
class Budget:
    soft_deadline_s: float = 1.5    # local seat must answer within this
    hard_deadline_s: float = 6.0    # remote seat gets no longer than this
    circuit_window_s: float = 60.0
    failure_threshold: int = 3

@dataclass
class Turn:
    prompt: str
    idempotency_key: str = field(default_factory=lambda: uuid.uuid4().hex)
    provenance: str = "unset"
    degraded: bool = False

def run_turn(turn: Turn, *, local, remote, budget: Budget, clock=time.monotonic) -> str:
    start = clock()
    try:
        text = remote.send(
            turn.prompt,
            deadline_s=budget.hard_deadline_s,
            idempotency_key=turn.idempotency_key,
        )
        turn.provenance = "remote"
        return text
    except (ConnectionError, TimeoutError):
        turn.degraded = True

    elapsed = clock() - start
    remaining = max(0.0, budget.soft_deadline_s - elapsed)
    text = local.send(turn.prompt, deadline_s=remaining, idempotency_key=None)
    turn.provenance = "local"
    return text

剩余预算的计算比看起来更重要。如果远程尝试已经耗掉两秒，却又给本地实例完整的 1.5 秒，就会悄悄突破用户愿意等待的时间上限。把预算限制在剩余时间内，才能让整轮任务保持在对外承诺的时间范围内。

第 3 步：主动注入故障，而不是等故障发生

等待真正的网络分区发生，算不上测试策略。几个模拟传输实现就能在毫秒级复现每一类故障，而且完全不需要网络，也能在 CI 中运行。

# fake_seats.py
import time

class HealthySeat:
    def send(self, prompt, *, deadline_s, idempotency_key):
        return "remote answer"

class PartitionedSeat:
    def send(self, prompt, *, deadline_s, idempotency_key):
        raise ConnectionRefusedError("no route to seat")

class SlowSeat:
    def send(self, prompt, *, deadline_s, idempotency_key):
        time.sleep(deadline_s + 0.5)
        raise TimeoutError("deadline exceeded")

class LocalSeat:
    def __init__(self) -> None:
        self.calls = 0

    def send(self, prompt, *, deadline_s, idempotency_key):
        self.calls += 1
        return "local answer"

每个模拟实现都与真实传输实现保持相同的返回结构，因此生产代码的调用位置无需改变。这条约束值得严格执行：一旦某个模拟实现需要在 run_turn 中增加特殊分支，这套测试装置测的就不再是实际交付的代码了。

第 4 步：不只断言输出，还要断言回答来源

最有价值的断言，是检查用户会被告知什么。一轮任务如果悄悄回退，却没有任何提示，即使返回的文本完全合理，测试也应该判定失败。

# test_fallback_contract.py
import pytest
from fallback_contract import Budget, Turn, run_turn
from fake_seats import HealthySeat, LocalSeat, PartitionedSeat, SlowSeat

@pytest.mark.parametrize("seat", [PartitionedSeat(), SlowSeat()])
def test_local_takes_over_and_says_so(seat):
    local = LocalSeat()
    turn = Turn(prompt="refactor auth.py")
    out = run_turn(turn, local=local, remote=seat, budget=Budget())
    assert out == "local answer"
    assert turn.provenance == "local"
    assert turn.degraded is True
    assert local.calls == 1

def test_healthy_seat_keeps_provenance_remote():
    turn = Turn(prompt="summarise diff")
    out = run_turn(turn, local=LocalSeat(), remote=HealthySeat(), budget=Budget())
    assert out == "remote answer"
    assert turn.provenance == "remote"
    assert turn.degraded is False

def test_idempotency_key_is_stable_across_fallback():
    turn = Turn(prompt="migrate schema")
    key = turn.idempotency_key
    run_turn(turn, local=LocalSeat(), remote=PartitionedSeat(), budget=Budget())
    assert turn.idempotency_key == key

运行测试套件时启用耗时报告，让注入的延迟清晰可见，而不是被隐藏起来：

python -m pytest test_fallback_contract.py -q --durations=5

模拟慢速实例的用例会占据大部分运行时间，这本身就是诊断信号。如果这一项从耗时报告中消失，就说明模拟实现不再执行等待，超时路径也就没有被真正覆盖。

免费服务器在哪些地方确实有优势，哪些地方没有

Local-first 不是一种道德立场，决定是否采用它的是三项取舍。首先是延迟：小型本地模型的首 token 延迟比较稳定，但生成较长内容时可能明显落后；远程实例的网络开销存在波动，但开始生成后的吞吐量要强得多。其次是敏感信息：凭据和私有源码应留在本机，只有经过精简的请求载荷才能通过网络传输，因此远程路径的数据边界取决于本地脱敏步骤执行得如何。第三是离线能力，而这也是本地实例唯一无条件胜出的方面。

MonkeyCode 的免费模型访问和免费服务器选项，正是在这个具体环节与工作流产生了实际关联，而不是被生硬地附加进来。免费的远程实例让你能以很低的成本，针对真实传输行为运行契约测试，而不只是测试模拟实现；这恰恰是上面那套测试装置无法提供的观察结果。可用性、模型选择以及任何使用限制都会变化，阅读时应以产品文档为准。利益披露：本文是 MonkeyCode 产品推广活动的一部分。

局限，以及哪些团队不适合采用

这套测试装置验证的是回退行为，不是性能。预算表中的每个数字都是为了让测试易于理解而选择的占位值，文中任何延迟或成本对比都不应被当作实测结果引用。此外，这个同步示例无法取消正在执行的远程任务，因此必须通过幂等键抑制重复结果，这不是可选项。

没有本地推理路径的团队不应该采用这种模式；它们只有一个服务实例，正确的设计是一个能明确展示重试状态的队列。受到严格数据驻留规则约束的团队，则应该彻底反转这份契约，从设计上就把远程实例视为不可用。这样，本地路径就是唯一路径，来源标签也就没有必要了。其他团队可以先从两个模拟传输实现、一个参数化测试和一个来源字段开始，再随着真实事故的发生，逐步扩充故障分类表。

如果你正在梳理自己的故障类别，花一个下午，把同一套测试装置跑在真实的免费实例上，就是一种成本很低的方法，能让预算从猜测变成实际观察的结果。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。