这一章回答：前缀准备真的发生在源码流关闭之前吗？准备出的 handle 是否只在正确的最终调用处领取，取消和最终错误又是否能收回 worker？
测试的完整因果路径是：读取已有 dist/spine.wasm，用当前的 Manifest{"lookup": {Call: read, AllowEarlyRead: true}} 创建 Runner；read 闭包记录 Host 实际收到的 key，并用 started/stopped 发出时序信号。第一个子测试让 producer 发送一条完整首行后停住，只有当真实 Host read 已经开始才发送剩余源码。若 prefix 在 EOF 前准备，首行会沿 RunPrefix → prefix_feed → Prefix.feed → prepare → runState.prepare → worker → read 到达 started，producer 才能继续。随后 channel 关闭，同一个 Guest execute 完整 source；最终调用 claim/resolve 之前的 Future，结果为 26。
其余子测试再分别固定边界：PLM 不接受最终程序时仍可复用 prefix handle；无换行的半句不提前读取；重复 request 不合并；依赖变量留给最终执行；未选择分支和非 early-read 工具停止 prefix 扫描；取消必须传到 Host 工具并等待它退出；最终编译错误也要清理已准备读取；输入上限和空流各有明确结果。这些是源码中断言的期望，不是本次运行结果；本任务按要求没有构建或执行测试。
1. 测试夹具先准备真实执行环境
prefix_test.go · L1–L13 · 打开完整源码
package spine
import (
"context"
"encoding/json"
"errors"
"os"
"strings"
"sync"
"testing"
"time"
)导入区显示测试边界：context 和 time 控制本轮与总测试时限，encoding/json 解码 Host 工具参数，os 读取真实 Wasm artifact，sync 保护并发计数，strings 检查错误和构造超限输入，testing 提供测试入口。测试函数名为 TestPrefixGuest，属于 spine 包，可以直接调用 RunPrefix 并访问包内的 maxMessage。
它不是 Python parser 单测。没有 dist/spine.wasm 时，测试在进入任何子测试前就失败；因此源码只能说明它打算观察真实 Guest 路径，不能把期望断言写成已通过。
prefix_test.go · L14–L21 · 打开完整源码
func TestPrefixGuest(t *testing.T) {
wasm, err := os.ReadFile("dist/spine.wasm")
if err != nil {
t.Fatal(err)
}
var mu sync.Mutex
calls := map[string]int{}
var started, stopped chan struct{}测试读取 artifact 后创建 mu、按 key 计数的 calls，以及可由各子测试设置的 started、stopped 信号。多个 Future worker 可能并发访问 calls，锁只保护 map，不把工具调用强制串行。
2. read 闭包标出真正的并发起点
prefix_test.go · L22–L37 · 打开完整源码
read := func(ctx context.Context, raw json.RawMessage) (any, error) {
var a struct {
Key string `json:"key"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return nil, err
}
mu.Lock()
calls[a.Key]++
mu.Unlock()
if started != nil {
select {
case started <- struct{}{}:
default:
}
}read 是注册到当前 Manifest 的 Host 工具实现。它先把原始 JSON 解码成带 key 字段的参数；解码失败立即返回。成功后加锁递增 calls[a.Key]，释放锁，再尝试向 started 发送一个空事件。这个发送使用 select 的 default 分支，不会因为重复通知或没有接收者把工具卡住。
因此本测试里的 started 不是“producer 已经发出 chunk”的信号，也不是“RunPrefix 已经被调用”的信号。它发生在真实 Host read 已进入、参数已解码、调用计数已更新之后。对第一个子测试的 early-read 路径而言，这意味着 runState.prepare 已经创建 Future 并启动 worker，worker 至少已经走到 Host 工具入口。
prefix_test.go · L38–L52 · 打开完整源码
switch a.Key {
case "book":
return 21, nil
case "shipping":
return 5, nil
case "choose":
return "book", nil
case "wait":
<-ctx.Done()
close(stopped)
return nil, ctx.Err()
default:
return nil, errors.New("missing key")
}
}switch 为测试场景提供确定的工具行为：book 返回 21，shipping 返回 5，choose 返回字符串 book。wait 不返回业务值，而是等到传入的 context 被取消，关闭 stopped 后返回 ctx.Err()；未知 key 返回错误。
stopped 证明的是等待取消的 Host 工具已经离开，而不是 Future 结果是否被最终调用领取。它与 started 分开，正好让测试分别观察“worker 到达工具”和“取消后 worker 退出”。
3. Runner 使用当前 Manifest，stream 表示正常 EOF
prefix_test.go · L53–L68 · 打开完整源码
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
r, err := New(ctx, wasm, Manifest{"lookup": {Call: read, AllowEarlyRead: true}})
if err != nil {
t.Fatal(err)
}
defer r.Close(context.Background())
count := func(key string) int { mu.Lock(); defer mu.Unlock(); return calls[key] }
stream := func(parts ...string) <-chan string {
ch := make(chan string, len(parts))
for _, p := range parts {
ch <- p
}
close(ch)
return ch
}测试建立两分钟根 context，并用当前的具名工具清单创建 Runner：lookup 的 Call 是上面的 read，AllowEarlyRead: true 明确允许 prefix 提前准备。New 会把 Go 侧 Manifest 转成 Guest 看到的 [{"name":"lookup","allow_early_read":true}]；Guest Prefix 依据这份清单建立 early_names。这里没有旧版的位置参数或单独的 tool 注册调用。
Runner 在所有子测试结束后关闭；计数器闭包加锁读取某个 key 的快照。stream 创建容量等于片段数的 buffered channel，按给定顺序填入片段后关闭。它表示合法的已结束源码流；stream() 返回的是已关闭 channel，不是 nil channel。nil stream 的拒绝属于 RunPrefix 入口行为，空 stream 则会继续执行空程序。
4. 首个场景：源码未关闭，Host 已经开始读取
prefix_test.go · L69–L91 · 打开完整源码
t.Run("starts before source closes and is claimed once", func(t *testing.T) {
runCtx, stop := context.WithCancel(ctx)
defer stop()
started = make(chan struct{}, 2)
chunks := make(chan string)
// Producer refuses to send the rest until the real Host tool has started.
go func() {
defer close(chunks)
select {
case chunks <- "price=lookup(key=inputs['item'])\n":
case <-runCtx.Done():
return
}
select {
case <-started:
case <-runCtx.Done():
return
}
select {
case chunks <- "shipping=lookup(key='shipping')\nresult=price+shipping":
case <-runCtx.Done():
}
}()子测试从根 context 派生 runCtx，设置容量为 2 的 started，并使用无缓冲 chunks。无缓冲 channel 让首段发送必须与 receiveSource 的接收相遇，也让 producer 在发送完首段后可以明确停在第二次发送之前。
producer 的第一段是带换行的完整候选：
price=lookup(key=inputs['item'])\n它发送成功后不关闭 chunks，而是等待 started 或取消。当前 Manifest 允许 lookup，Prefix 可以从固定 inputs['item'] 得到 book，并通过 Host prepare 启动读取。read 进入后发出 started，producer 才发送第二段：
shipping=lookup(key='shipping')\nresult=price+shipping第二段包含换行后的 shipping 候选和没有换行的最终赋值；关闭 channel 发生在 producer 返回时。这个握手把并发窗口钉在源码 EOF 之前：如果实现只有收齐源码才 prepare，就不会先得到 started，也就不能按该协议推进到第二段。
prefix_test.go · L92–L98 · 打开完整源码
before := count("book")
out, err := r.RunPrefix(runCtx, chunks, map[string]any{"item": "book"})
if err != nil || string(out.Value) != "26" || count("book") != before+1 {
t.Fatalf("%+v %v", out, err)
}
started = nil
})测试在调用前记录 book 次数，再同步运行 RunPrefix。它要求无错误、最终 JSON 值为 26，且 book 次数只增加一次。增量断言允许所有子测试共享同一个 Runner，同时仍能确认本场景没有因为“提前 worker + 最终调用”而把同一读取执行两次。清空 started 是测试夹具收尾，避免后续子测试误收到旧信号。
5. 最终执行阶段仍能领取 prefix 结果
prefix_test.go · L99–L105 · 打开完整源码
t.Run("unsupported final syntax reuses the leading read", func(t *testing.T) {
before := count("book")
out, err := r.RunPrefix(ctx, stream("a=lookup(key='book')\n", "import math\nresult=a"), nil)
if err != nil || string(out.Value) != "21" || out.Transformed != "" || count("book") != before+1 {
t.Fatalf("%+v %v", out, err)
}
})第一场景之后，这个子测试把首个可准备读取和第二段 import math/result 一起送入已关闭的 buffered stream。import 会使 whole-program PLM 保持原程序而不产生转换文本，但这不抹掉 Prefix 已经保存的 ready 条目。最终普通 Python 工具代理仍会构造 request，按队首 claim book 的 handle，所以结果为 21、out.Transformed 为空，Host 次数只增加一次。
这里的“复用”不是重新调用 prepare，也不是把一个 Future 结果复制到全局缓存；它是同一 Guest 的 Prefix 队列与同一 runState 之间完成一次匹配。out.Transformed == "" 只说明没有产生可展示的 AST 文本，不说明程序没有执行。
6. 完整性、重复身份和依赖边界
prefix_test.go · L106–L125 · 打开完整源码
t.Run("partial statement is not executed", func(t *testing.T) {
before := count("book")
_, err := r.RunPrefix(ctx, stream("a=lookup(key='book'"), nil)
if err == nil || !strings.Contains(err.Error(), "SyntaxError") || count("book") != before {
t.Fatalf("%v", err)
}
})
t.Run("duplicate requests have distinct handles", func(t *testing.T) {
before := count("book")
out, err := r.RunPrefix(ctx, stream("a=lookup(key='book')\n", "b=lookup(key='book')\nresult=a+b"), nil)
if err != nil || string(out.Value) != "42" || count("book") != before+2 {
t.Fatalf("%+v %v", out, err)
}
})
t.Run("dependency waits for final execution", func(t *testing.T) {
out, err := r.RunPrefix(ctx, stream("key=lookup(key='choose')\n", "a=lookup(key=key)\nresult=a"), nil)
if err != nil || string(out.Value) != "21" {
t.Fatalf("%+v %v", out, err)
}
})三个连续场景各自固定一种因果关系：
- 半句：只发送没有闭合括号、也没有换行的
a=lookup(...)。Prefix 没有完整文本可扫描，因此book次数不变；最终execute编译完整 source 时返回包含SyntaxError的错误。 - 重复 request：两个变量分别调用完全相同的
lookup(key='book')。测试要求结果 42 且次数增加 2，说明两个源码位置对应两个独立 Future，不能按 JSON request 或工具名去重。 - 依赖：第一条
choose返回字符串book，第二条把变量key作为参数。Prefix 的窄规则不会为第二条提前求值；最终执行在第一条产生key后处理第二条，得到 21。测试只要求最终语义成立，不把第二条误称为 prefix 阶段的提前读取。
它们分别观察“尚未完整就不开始”“相同内容仍按事件分配 handle”“需要程序状态的参数不能由 prefix 猜测”。
7. 分支和取消：停止发现与退出 worker
prefix_test.go · L126–L156 · 打开完整源码
t.Run("unselected branch is not prepared", func(t *testing.T) {
before := count("book")
out, err := r.RunPrefix(ctx, stream("if False:\n a=lookup(key='book')\n", "result=7"), nil)
if err != nil || string(out.Value) != "7" || count("book") != before {
t.Fatalf("%+v %v", out, err)
}
})
t.Run("cancellation during source intake joins workers", func(t *testing.T) {
runCtx, stop := context.WithCancel(ctx)
defer stop()
started, stopped = make(chan struct{}, 1), make(chan struct{})
chunks := make(chan string, 1)
chunks <- "a=lookup(key='wait')\n"
go func() {
select {
case <-started:
stop()
case <-runCtx.Done():
}
}()
_, err := r.RunPrefix(runCtx, chunks, nil)
if err == nil {
t.Fatal("expected cancellation")
}
select {
case <-stopped:
default:
t.Fatal("worker still alive")
}
started = nil
})分支场景把 if False: 放在第一段，分支内部虽有 lookup，但顶层 statement 是 ast.If，不是 Prefix 接受的普通赋值候选。Prefix 因此在第一个 statement 停止扫描，Host 不提前收到 book；最终程序正常执行 result=7。
取消场景把 wait 读取放进一个容量为 1 的 chunk channel，先让 RunPrefix 拿到完整首行。read 发出 started 后阻塞在 ctx.Done()；协调 goroutine 收到 started 就调用 stop。此时源码接收也可能仍在等更多 chunk，但 receiveSource 的 context case 和 Host worker 使用的是同一轮派生 context。
测试要求 RunPrefix 返回错误，并且随后 stopped 已关闭。最后这个非阻塞检查很关键：只返回错误而不等待 worker，不能证明取消已经传到工具。共享的 started 在场景结束时清空。
8. 最终错误、输入上限和空流
prefix_test.go · L157–L181 · 打开完整源码
t.Run("final syntax error cancels prepared reads", func(t *testing.T) {
stopped = make(chan struct{})
_, err := r.RunPrefix(ctx, stream("a=lookup(key='wait')\n", "result = ("), nil)
if err == nil || !strings.Contains(err.Error(), "SyntaxError") {
t.Fatalf("%v", err)
}
select {
case <-stopped:
default:
t.Fatal("worker still alive")
}
})
t.Run("source bound", func(t *testing.T) {
_, err := r.RunPrefix(ctx, stream(strings.Repeat("#", maxMessage+1)), nil)
if err == nil || !strings.Contains(err.Error(), "1 MiB") {
t.Fatalf("%v", err)
}
})
t.Run("empty stream executes empty program", func(t *testing.T) {
out, err := r.RunPrefix(ctx, stream(), nil)
if err != nil || string(out.Value) != "null" {
t.Fatalf("%+v %v", out, err)
}
})
}最终语法错误场景先送入会等待取消的完整首行，再送入没有闭合括号的 result = (。第一条读取可能已经启动但尚未领取；最终 compile 返回 SyntaxError 后，测试仍要求 stopped 已关闭，验证失败路径同样经过统一取消和 worker 等待。
source bound 用 strings.Repeat("#", maxMessage+1) 生成超出 1 MiB 的片段。Host 应在片段进入 Guest 前按累计字节数拒绝，错误文本包含 1 MiB；这不是 Python 语法测试。
空流场景使用 stream() 返回的已关闭 channel。Prefix 初始化后没有候选，最终执行空程序；没有 result 时协议编码为 JSON null，所以要求无错误且 out.Value == "null"。
这三个场景把错误时间点分开：超限发生在接收 Host 阶段，语法错误发生在最终 Guest compile 阶段，已启动读取的取消和等待发生在 Run 收尾阶段。它们不能用一个最终数值断言互相替代。
9. 读测试时的检查顺序
读每个子测试时，先找它构造的源码状态，再问四个问题：
- 哪个 chunk 让 Prefix 看到最后一个换行之前的完整文本？
- 当前第一个非候选 statement 是否让扫描停止？
prepare是否真的创建了 handle，最终调用是否按队首和完整 request claim？- 测试是否检查了错误所属阶段，以及取消后 worker 是否已经退出？
按这个顺序，Go 测试中的 calls 是 Host 实际调用计数，started 是工具入口时序，stopped 是取消收尾时序；三者不是同一个观察量。下一章 prefix-python-tests.md 会用不启动 Wasm 的替身，隔离同样的 AST 过滤、Manifest 门槛、FIFO 领取和零 handle 行为。