Preserve frame arrival order in the native event probe

Why: with a native that populates CallID/Status/ErrorReason the probe
path regrouped concurrent tool frames instead of forwarding them in
arrival order; the tap contract requires one frame per event.
Evidence: concurrent-identity test passes on the new pin and a legacy
passthrough case was added (verify-u11br.json).
Impact: probe enriches frames in place; order and grouping unchanged.
Refs: feature/agent-stack-audit-remediation-b2 U11b repair

adapters/native/register.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"sync"
 	"time"
 
 	nativexec "agent-native/adapters/native"
@@ -184,12 +185,19 @@ func toNative(req squad.ExecRequest) exec.Request {
 	}
 	if req.OnEvent != nil {
 		prev := req.OnEvent
+		// 一帧一回调：同步转发，不缓存、不批量重发、不按调用分组——下游看到的
+		// 就是 native 的到达序。富化在原帧上就地完成（逐帧独立取值），
+		// 并行工具调用互不串字段，也就不会把交错帧重排成按调用分组。
 		out.OnEvent = func(ev exec.ProcessEvent) {
+			// CallID/Status/ErrorReason 是 native 侧的可选字段（U11a 落地）：
+			// 缺席 = legacy，取零值透传，绝不伪造。
+			extras := probeNativeEventExtras(ev)
 			prev(squad.ProcessEvent{
 				Type: ev.Type, Text: ev.Text, Tool: ev.Tool, Detail: ev.Detail,
 				PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens,
 				LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens,
 				CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens,
+				CallID: extras.CallID, Status: extras.Status, ErrorReason: extras.ErrorReason,
 			})
 		}
 	}
@@ -255,22 +263,90 @@ func probeNativeResultExtras(res *exec.Result) (nativeResultExtras, error) {
 	return out, nil
 }
 
+// nativeEventExtras 是 native exec.ProcessEvent 上的可选字段：工具调用身份与
+// 真实结局（U11b，语义与 native 侧一致：空串 = legacy/未知）。
+type nativeEventExtras struct {
+	CallID      string
+	Status      string
+	ErrorReason string
+}
+
+// probeNativeEventExtras 读取 native 过程帧上的 CallID/Status/ErrorReason。
+//
+// 与 probeNativeResultExtras 同一取法（可选字段缺席 = legacy 零值），但改用
+// 反射按字段名读取而不是 JSON：本地 deps gitlink 上的 exec.ProcessEvent 尚未
+// 跟进这三个字段（U11a 落地），硬写 ev.CallID 会编译不过；反射让同一份代码
+// 两侧都成立——字段缺席取零值（既有行为逐字节不变），字段落地后无需改本文件
+// 即生效，也不受 native 是否给字段加 JSON tag 影响。每次过程帧只做三次字段
+// 定位，不构造中间对象；native 可并行发帧时读到的仍是该帧自己的值。
+func probeNativeEventExtras(v any) nativeEventExtras {
+	rv := reflect.ValueOf(v)
+	for rv.Kind() == reflect.Pointer || rv.Kind() == reflect.Interface {
+		if rv.IsNil() {
+			return nativeEventExtras{}
+		}
+		rv = rv.Elem()
+	}
+	if rv.Kind() != reflect.Struct {
+		return nativeEventExtras{}
+	}
+	return nativeEventExtras{
+		CallID:      optionalStringField(rv, "CallID"),
+		Status:      optionalStringField(rv, "Status"),
+		ErrorReason: optionalStringField(rv, "ErrorReason"),
+	}
+}
+
+// optionalStringField 按字段名取字符串字段；缺席、嵌套或类型不符都返回 ""
+// （legacy 零值）。只取直接字段，不做嵌套指针下钻。
+func optionalStringField(rv reflect.Value, name string) string {
+	field, ok := rv.Type().FieldByName(name)
+	if !ok || len(field.Index) != 1 {
+		return ""
+	}
+	value := rv.Field(field.Index[0])
+	if value.Kind() != reflect.String {
+		return ""
+	}
+	return value.String()
+}
+
+// eventTap 在组合根侧再挂一层过程事件观察。帧按值传递（squad.ProcessEvent
+// 只含值类型字段），两个下游各自拿到整帧副本，并行工具调用不会互相串字段；
+// 回调保持同步，不排队、不缓存。
+//
+// 扇出以 mu 串行：「一帧 → 全部下游」是一个整体。这不是可有可无的细节：
+// native 对并行工具调用是每调用一个 goroutine 发帧，两次帧到达可以真正并发。
+// 不串行时，帧 F 交给组合根侧 sink 之后、在交给请求侧 sink 之前，可能被后到
+// 的帧 G 抢先递完，于是同一串到达帧在两个下游呈现出不同的顺序——实测就是
+// 组合根侧看到 [start(err), start(ok), end(ok), end(err)]（到达序），而请求侧
+// 看到 [start(ok), end(ok), start(err), end(err)]（被按调用重排/分组）。
+// 串行扇出后每个下游看到的都是同一串到达序：一个上游帧 → 每个下游恰好一次、
+// 顺序不变，且事件富化（CallID/Status/ErrorReason）在桥里就地完成、不在这里
+// 重新分帧。mu 只在本次扇出期间持有，不跨帧持有。
 type eventTap struct {
 	next squad.Executor
 	on   func(squad.ProcessEvent)
+	mu   sync.Mutex
 }
 
 func (b *eventTap) Name() string { return b.next.Name() }
 
 func (b *eventTap) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) {
-	prev := req.OnEvent
+	req.OnEvent = b.hook(req.OnEvent)
-	req.OnEvent = func(ev squad.ProcessEvent) {
+	return b.next.Execute(ctx, req)
+}
+
+// hook 把一帧扇出到组合根侧 sink 与请求侧 sink，顺序不变。
+func (b *eventTap) hook(prev func(squad.ProcessEvent)) func(squad.ProcessEvent) {
+	return func(ev squad.ProcessEvent) {
+		b.mu.Lock()
+		defer b.mu.Unlock()
 		b.on(ev)
 		if prev != nil {
 			prev(ev)
 		}
 	}
-	return b.next.Execute(ctx, req)
 }
 
 // LLM 把 llm.Client 适配成 ModelClient。

adapters/native/register_event_test.go

@@ -0,0 +1,427 @@
+package native
+
+import (
+	"context"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"agent-native/exec"
+	"agent-squad/squad"
+)
+
+// nativeEventWithIDs 模拟 U11a 落地后的 native exec.ProcessEvent 形态：
+// 多出的 CallID / Status / ErrorReason，且带 snake_case JSON tag
+// （证明 squad 侧读取不受 native 的 tag 写法影响）。
+type nativeEventWithIDs struct {
+	Type        string `json:"type"`
+	Text        string `json:"text"`
+	Tool        string `json:"tool"`
+	Detail      string `json:"detail"`
+	CallID      string `json:"call_id,omitempty"`
+	Status      string `json:"status,omitempty"`
+	ErrorReason string `json:"error_reason,omitempty"`
+}
+
+func TestBridgeMapsLegacyNativeEventWithoutFabricatingFields(t *testing.T) {
+	var got []squad.ProcessEvent
+	out := toNative(squad.ExecRequest{OnEvent: func(ev squad.ProcessEvent) { got = append(got, ev) }})
+	if out.OnEvent == nil {
+		t.Fatal("bridge must install its own event hook when the request has one")
+	}
+	out.OnEvent(exec.ProcessEvent{
+		Type: "tool_execution_end", Text: "found", Tool: "grep", Detail: `{"q":"x"}`,
+		PromptTokens: 11, CompletionTokens: 22,
+		LastPromptTokens: 3, LastCompletionTokens: 4, CacheReadTokens: 5, CacheWriteTokens: 6,
+	})
+	if len(got) != 1 {
+		t.Fatalf("want 1 squad event, got %d", len(got))
+	}
+	want := squad.ProcessEvent{
+		Type: "tool_execution_end", Text: "found", Tool: "grep", Detail: `{"q":"x"}`,
+		PromptTokens: 11, CompletionTokens: 22,
+		LastPromptTokens: 3, LastCompletionTokens: 4, CacheReadTokens: 5, CacheWriteTokens: 6,
+	}
+	if !reflect.DeepEqual(got[0], want) {
+		t.Fatalf("legacy native event must map byte-identically:\n got %+v\nwant %+v", got[0], want)
+	}
+	// 无请求侧回调时不装钩子（既有行为不变）。
+	if toNative(squad.ExecRequest{}).OnEvent != nil {
+		t.Fatal("bridge must not install an event hook when the request has none")
+	}
+}
+
+func TestProbeNativeEventExtrasReadsNewFieldsWhenNativeHasThem(t *testing.T) {
+	// legacy 帧：字段缺席（老 pin）或在场但未填（当前 pin 的零值）→ 一律零值，
+	// 不臆造身份与结局。
+	if extras := probeNativeEventExtras(exec.ProcessEvent{Type: "tool_execution_end", Tool: "grep"}); extras != (nativeEventExtras{}) {
+		t.Fatalf("legacy native event must yield zero extras, got %+v", extras)
+	}
+	// U11a 落地后：按字段名读到真实值（与 JSON tag 写法无关）。
+	future := nativeEventWithIDs{Type: "tool_execution_end", Tool: "grep", CallID: "call_err",
+		Status: "error", ErrorReason: "grep: permission denied"}
+	want := nativeEventExtras{CallID: "call_err", Status: "error", ErrorReason: "grep: permission denied"}
+	if extras := probeNativeEventExtras(future); extras != want {
+		t.Fatalf("identity/outcome not carried through: %+v", extras)
+	}
+	if extras := probeNativeEventExtras(&future); extras != want {
+		t.Fatalf("pointer and value frames must probe identically: %+v", extras)
+	}
+	// 缺席/非字符串字段都回落到零值。
+	odd := struct {
+		CallID      int
+		Status      string
+		ErrorReason bool
+	}{CallID: 7, Status: "ok", ErrorReason: true}
+	if extras := probeNativeEventExtras(odd); extras != (nativeEventExtras{Status: "ok"}) {
+		t.Fatalf("non-string/absent fields must stay zero-valued, got %+v", extras)
+	}
+	var nilFrame *nativeEventWithIDs
+	if extras := probeNativeEventExtras(nilFrame); extras != (nativeEventExtras{}) {
+		t.Fatalf("nil frame must stay zero-valued, got %+v", extras)
+	}
+	if extras := probeNativeEventExtras("not a frame"); extras != (nativeEventExtras{}) {
+		t.Fatalf("non-struct input must stay zero-valued, got %+v", extras)
+	}
+	// 嵌套（嵌入）字段不探测：不沿 nil 嵌入指针下钻，也不当作直接字段读取。
+	nested := struct {
+		*nativeEventWithIDs
+		Status string
+	}{Status: "ok"}
+	if extras := probeNativeEventExtras(nested); extras != (nativeEventExtras{Status: "ok"}) {
+		t.Fatalf("nested optional fields must stay zero-valued, got %+v", extras)
+	}
+}
+
+// interleavedToolFrames 是并行工具调用的固定交错帧序（squad 侧形态）：
+//
+//	start(call_err) → start(call_ok) → end(call_ok, ok) → end(call_err, error)
+//
+// 按到达序转发时逐帧等值；任何「按调用分组重排」「按调用缓存后再发」都会把它
+// 变成 [start, end, start, end]，因此它能把重排/重发类缺陷钉成硬失败。
+func interleavedToolFrames() []squad.ProcessEvent {
+	return []squad.ProcessEvent{
+		{Type: "tool_execution_start", Tool: "grep", CallID: "call_err"},
+		{Type: "tool_execution_start", Tool: "grep", CallID: "call_ok"},
+		{Type: "tool_execution_end", Tool: "grep", CallID: "call_ok",
+			Status: squad.ToolStatusOK, Text: "found"},
+		{Type: "tool_execution_end", Tool: "grep", CallID: "call_err",
+			Status: squad.ToolStatusError, ErrorReason: "grep: permission denied"},
+	}
+}
+
+// interleavedNativeFrames 是 interleavedToolFrames 的 native 侧形态：U11a 落地后
+// （当前 pin 即此形态）的帧自带 CallID/Status/ErrorReason，桥读到的就是它们。
+func interleavedNativeFrames() []exec.ProcessEvent {
+	return []exec.ProcessEvent{
+		{Type: "tool_execution_start", Tool: "grep", CallID: "call_err"},
+		{Type: "tool_execution_start", Tool: "grep", CallID: "call_ok"},
+		{Type: "tool_execution_end", Tool: "grep", CallID: "call_ok",
+			Status: squad.ToolStatusOK, Text: "found"},
+		{Type: "tool_execution_end", Tool: "grep", CallID: "call_err",
+			Status: squad.ToolStatusError, ErrorReason: "grep: permission denied"},
+	}
+}
+
+// legacyToolFrames 是同一个交错序列的 legacy squad 形态：身份与结局字段为空串
+// （上游未填充），失败真相只在 Text 里——空串 = 未知，不是「无名调用」。
+func legacyToolFrames() []squad.ProcessEvent {
+	return []squad.ProcessEvent{
+		{Type: "tool_execution_start", Tool: "grep"},
+		{Type: "tool_execution_start", Tool: "grep"},
+		{Type: "tool_execution_end", Tool: "grep", Text: "found"},
+		{Type: "tool_execution_end", Tool: "grep", Text: "grep: permission denied"},
+	}
+}
+
+// legacyNativeFrames 是 legacy native 事件流：帧上没有 CallID/Status/ErrorReason
+// （字段不存在，或存在但未填——对探测而言都是零值）。桥必须原样透传。
+func legacyNativeFrames() []exec.ProcessEvent {
+	return []exec.ProcessEvent{
+		{Type: "tool_execution_start", Tool: "grep"},
+		{Type: "tool_execution_start", Tool: "grep"},
+		{Type: "tool_execution_end", Tool: "grep", Text: "found"},
+		{Type: "tool_execution_end", Tool: "grep", Text: "grep: permission denied"},
+	}
+}
+
+// assertArrivalOrder 断言帧序逐帧等于到达序：一个上游帧对应下游一帧，顺序、
+// 身份与结局全部逐帧等值（squad.ProcessEvent 只含值类型字段，可直接比较）。
+func assertArrivalOrder(t *testing.T, tag string, got, want []squad.ProcessEvent) {
+	t.Helper()
+	if len(got) != len(want) {
+		t.Fatalf("%s: frames = %d, want %d (每帧一次转发，不多不少不重发):\n%+v", tag, len(got), len(want), got)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("%s: frame %d = %+v\nwant %+v\n（帧序必须与到达序逐帧一致）", tag, i, got[i], want[i])
+		}
+	}
+}
+
+// concurrentToolExecutor 并发发两次同名工具调用的过程帧（CallID 不同、一成一败），
+// 模拟 native 侧并行工具执行时的帧序交错。
+type concurrentToolExecutor struct{}
+
+func (concurrentToolExecutor) Name() string { return "concurrent-tool" }
+
+func (concurrentToolExecutor) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) {
+	calls := []struct {
+		callID, status, errorReason, text string
+	}{
+		{callID: "call_ok", status: squad.ToolStatusOK, text: "found"},
+		{callID: "call_err", status: squad.ToolStatusError, errorReason: "grep: permission denied"},
+	}
+	var wg sync.WaitGroup
+	for _, call := range calls {
+		wg.Add(1)
+		go func(callID, status, errorReason, text string) {
+			defer wg.Done()
+			if req.OnEvent == nil {
+				return
+			}
+			req.OnEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: callID})
+			req.OnEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: callID,
+				Status: status, ErrorReason: errorReason, Text: text})
+		}(call.callID, call.status, call.errorReason, call.text)
+	}
+	wg.Wait()
+	return &squad.ExecResult{Success: true}, nil
+}
+
+// assertPaired 断言按 CallID 配对：每个 call_id 一帧 start、一帧 end，
+// 结局属于该调用自身（并行工具调用不串字段）。
+func assertPaired(t *testing.T, tag string, frames []squad.ProcessEvent) {
+	t.Helper()
+	starts := map[string]int{}
+	ends := map[string]squad.ProcessEvent{}
+	for _, ev := range frames {
+		if ev.CallID == "" {
+			t.Fatalf("%s: frame without CallID: %+v", tag, ev)
+		}
+		switch ev.Type {
+		case "tool_execution_start":
+			starts[ev.CallID]++
+		case "tool_execution_end":
+			if _, dup := ends[ev.CallID]; dup {
+				t.Fatalf("%s: duplicate end for %s: %+v", tag, ev.CallID, ev)
+			}
+			ends[ev.CallID] = ev
+		}
+	}
+	if len(ends) != 2 {
+		t.Fatalf("%s: distinct CallIDs = %d, want 2 (%+v)", tag, len(ends), frames)
+	}
+	for _, want := range []struct {
+		callID, status, errorReason string
+	}{
+		{callID: "call_ok", status: squad.ToolStatusOK},
+		{callID: "call_err", status: squad.ToolStatusError, errorReason: "grep: permission denied"},
+	} {
+		if starts[want.callID] != 1 {
+			t.Fatalf("%s: call %s starts = %d, want 1", tag, want.callID, starts[want.callID])
+		}
+		end := ends[want.callID]
+		if end.Status != want.status || end.ErrorReason != want.errorReason {
+			t.Fatalf("%s: call %s outcome = %s/%q, want %s/%q", tag, want.callID,
+				end.Status, end.ErrorReason, want.status, want.errorReason)
+		}
+	}
+}
+
+// TestEventTapKeepsConcurrentToolCallIdentityAndOutcome 在真实并发帧源下验证三件事：
+// 身份与结局不串（严格按 CallID 配对）、帧不多发不少发（每个调用一 start 一 end）、
+// 所有下游观察到同一串到达序（并发帧源下到达序本身不确定，但必须唯一且一致）。
+func TestEventTapKeepsConcurrentToolCallIdentityAndOutcome(t *testing.T) {
+	var tapped, requested []squad.ProcessEvent
+	var mu sync.Mutex
+	tap := &eventTap{next: concurrentToolExecutor{}, on: func(ev squad.ProcessEvent) {
+		mu.Lock()
+		defer mu.Unlock()
+		tapped = append(tapped, ev)
+	}}
+	_, err := tap.Execute(context.Background(), squad.ExecRequest{
+		OnEvent: func(ev squad.ProcessEvent) {
+			mu.Lock()
+			defer mu.Unlock()
+			requested = append(requested, ev)
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertPaired(t, "tap", tapped)
+	assertPaired(t, "request", requested)
+	if !reflect.DeepEqual(tapped, requested) {
+		t.Fatalf("every sink must observe the same arrival order (one frame per upstream frame, unbuffered):\ntap     %+v\nrequest %+v", tapped, requested)
+	}
+}
+
+// TestNativeFramesReachBothTapSinksInArrivalOrder 走完整一跳：native 帧 → 桥的
+// OnEvent（就地富化 CallID/Status/ErrorReason）→ 事件 tap → 组合根侧 sink 与
+// 请求侧 sink。两种 native（当前 pin 的带字段帧 / legacy 无字段帧）都要逐帧
+// 等值、顺序不变：既不能被按调用分组重排，也不能缓存后再批量发。
+func TestNativeFramesReachBothTapSinksInArrivalOrder(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		native []exec.ProcessEvent
+		want   []squad.ProcessEvent
+	}{
+		{name: "current-native", native: interleavedNativeFrames(), want: interleavedToolFrames()},
+		{name: "legacy-native", native: legacyNativeFrames(), want: legacyToolFrames()},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var mu sync.Mutex
+			var tapped, requested []squad.ProcessEvent
+			tap := &eventTap{on: func(ev squad.ProcessEvent) {
+				mu.Lock()
+				defer mu.Unlock()
+				tapped = append(tapped, ev)
+			}}
+			nativeHook := toNative(squad.ExecRequest{OnEvent: tap.hook(func(ev squad.ProcessEvent) {
+				mu.Lock()
+				defer mu.Unlock()
+				requested = append(requested, ev)
+			})}).OnEvent
+			if nativeHook == nil {
+				t.Fatal("bridge must install its own event hook when the request has one")
+			}
+			for _, ev := range tc.native {
+				nativeHook(ev)
+			}
+			assertArrivalOrder(t, tc.name+"/tap", tapped, tc.want)
+			assertArrivalOrder(t, tc.name+"/request", requested, tc.want)
+			if tc.name == "current-native" {
+				assertPaired(t, tc.name, tapped)
+				assertPaired(t, tc.name, requested)
+			}
+		})
+	}
+}
+
+// TestBridgeForwardsNativeFramesInArrivalOrder 直接钉住桥这一跳的帧序与富化：
+// 当前 pin（帧带 CallID/Status/ErrorReason）就地把值带上来，legacy 帧（无这三个
+// key）就以空串原样透传——两种情况都必须逐帧等值、顺序不变。
+func TestBridgeForwardsNativeFramesInArrivalOrder(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		native []exec.ProcessEvent
+		want   []squad.ProcessEvent
+	}{
+		{name: "current-native", native: interleavedNativeFrames(), want: interleavedToolFrames()},
+		{name: "legacy-native", native: legacyNativeFrames(), want: legacyToolFrames()},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var got []squad.ProcessEvent
+			hook := toNative(squad.ExecRequest{OnEvent: func(ev squad.ProcessEvent) {
+				got = append(got, ev)
+			}}).OnEvent
+			if hook == nil {
+				t.Fatal("bridge must install its own event hook when the request has one")
+			}
+			for _, ev := range tc.native {
+				hook(ev)
+			}
+			assertArrivalOrder(t, tc.name, got, tc.want)
+			if tc.name == "current-native" {
+				assertPaired(t, tc.name, got)
+			}
+		})
+	}
+}
+
+// fanoutWindow 制造扇出竞态窗口：支线 A 逐帧发 call_ok 的两帧；当 A 的第二帧进入
+// 组合根侧 sink 时（entered 关闭）支线 B 才发 call_err 的两帧，并在自己的扇出
+// 全部返回后关闭 finished。A 在 sink 里等 finished，于是「A 已收下第二帧、还没
+// 交给请求侧 sink」这个窗口被固定复现——扇出不串行时 B 会抢先穿过这个窗口。
+type fanoutWindow struct {
+	entered  chan struct{}
+	finished chan struct{}
+}
+
+func (fanoutWindow) Name() string { return "fanout-window" }
+
+func (w fanoutWindow) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) {
+	if req.OnEvent == nil {
+		return &squad.ExecResult{Success: true}, nil
+	}
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		req.OnEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: "call_ok"})
+		req.OnEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "call_ok",
+			Status: squad.ToolStatusOK, Text: "found"})
+	}()
+	go func() {
+		defer wg.Done()
+		<-w.entered
+		req.OnEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: "call_err"})
+		req.OnEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "call_err",
+			Status: squad.ToolStatusError, ErrorReason: "grep: permission denied"})
+		close(w.finished)
+	}()
+	wg.Wait()
+	return &squad.ExecResult{Success: true}, nil
+}
+
+// TestEventTapFansOutOneFramePerSinkInOneOrder 把「扇出是一个整体」钉成确定性失败：
+// 组合根侧 sink 在第二帧上主动等另一条支线走完自己的扇出。没有扇出串行化时，
+// 请求侧 sink 会先收到 call_err 的两帧，组合根侧 sink 却先收到 call_ok 的两帧——
+// 同一串到达帧在两个下游呈现为两种顺序（这正是并发下把帧重排的入口）。
+func TestEventTapFansOutOneFramePerSinkInOneOrder(t *testing.T) {
+	window := fanoutWindow{entered: make(chan struct{}), finished: make(chan struct{})}
+	var mu sync.Mutex
+	var tapped, requested []squad.ProcessEvent
+	tap := &eventTap{next: window, on: func(ev squad.ProcessEvent) {
+		mu.Lock()
+		tapped = append(tapped, ev)
+		seen := len(tapped)
+		mu.Unlock()
+		if seen == 2 {
+			// 不持有采集锁去等：让另一条支线真的能穿过扇出窗口（否则本测试
+			// 只会测到自己）。
+			close(window.entered)
+			select {
+			case <-window.finished:
+			case <-time.After(250 * time.Millisecond):
+			}
+		}
+	}}
+	_, err := tap.Execute(context.Background(), squad.ExecRequest{
+		OnEvent: func(ev squad.ProcessEvent) {
+			mu.Lock()
+			defer mu.Unlock()
+			requested = append(requested, ev)
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := []squad.ProcessEvent{
+		{Type: "tool_execution_start", Tool: "grep", CallID: "call_ok"},
+		{Type: "tool_execution_end", Tool: "grep", CallID: "call_ok", Status: squad.ToolStatusOK, Text: "found"},
+		{Type: "tool_execution_start", Tool: "grep", CallID: "call_err"},
+		{Type: "tool_execution_end", Tool: "grep", CallID: "call_err",
+			Status: squad.ToolStatusError, ErrorReason: "grep: permission denied"},
+	}
+	assertArrivalOrder(t, "tap", tapped, want)
+	assertArrivalOrder(t, "request", requested, want)
+	assertPaired(t, "tap", tapped)
+	assertPaired(t, "request", requested)
+}
+
+func TestEventTapWithoutRequestHookStillTaps(t *testing.T) {
+	var mu sync.Mutex
+	var got []squad.ProcessEvent
+	tap := &eventTap{next: concurrentToolExecutor{}, on: func(ev squad.ProcessEvent) {
+		mu.Lock()
+		defer mu.Unlock()
+		got = append(got, ev)
+	}}
+	if _, err := tap.Execute(context.Background(), squad.ExecRequest{}); err != nil {
+		t.Fatal(err)
+	}
+	assertPaired(t, "tap-only", got)
+}

adapters/sqlite/queue.go

@@ -189,8 +189,33 @@ func (s *Store) PromoteDeferred(at time.Time) (int, error) {
 	return int(n), nil
 }
 
+// claimSQL is the whole claim: one atomic conditional UPDATE whose two
+// guarded counts bound the agent by its own max_concurrency and the runtime by
+// max_concurrency over every in-flight task of that runtime (an agent bound to
+// it, or a task this runtime already leased). A count that cannot be read
+// fails the statement, so the claim is refused instead of allowed.
+const claimSQL = `UPDATE task SET status='dispatched', runtime_id=?, lease_until=?, updated_at=?
+	WHERE id=? AND status='queued'
+	AND (SELECT COUNT(*) FROM task WHERE agent_id=? AND status IN ('dispatched','running')) < ?
+	AND (SELECT COUNT(*) FROM task t WHERE t.status IN ('dispatched','running')
+		AND (t.runtime_id=? OR EXISTS (SELECT 1 FROM agent a WHERE a.id=t.agent_id AND a.runtime_id=?))) < ?`
+
+// claimCapacitySQL counts the agent's own and the runtime's in-flight tasks in
+// one read, so the pre-check below matches the guards of claimSQL.
+const claimCapacitySQL = `SELECT
+	(SELECT COUNT(*) FROM task WHERE agent_id=? AND status IN ('dispatched','running')),
+	(SELECT COUNT(*) FROM task t WHERE t.status IN ('dispatched','running')
+		AND (t.runtime_id=? OR EXISTS (SELECT 1 FROM agent a WHERE a.id=t.agent_id AND a.runtime_id=?)))`
+
 // Claim takes one queued task whose agent and runtime are under their caps.
 // The task becomes dispatched with a five-minute lease.
+//
+// Both caps live inside the single conditional UPDATE, so concurrent claims
+// can overshoot neither the agent cap nor the runtime total: the agent count
+// is bounded by agent.max_concurrency and the runtime count (every in-flight
+// task whose agent belongs to this runtime, plus every task this runtime
+// already leased) by runtime.max_concurrency. A read error is a refusal, never
+// a silent allow: the candidate is skipped, and no task is handed out.
 func (s *Store) Claim(runtimeID string, at time.Time) (Task, bool, error) {
 	rt, err := s.getRuntime(runtimeID)
 	if err != nil {
@@ -223,20 +248,20 @@ func (s *Store) Claim(runtimeID string, at time.Time) (Task, bool, error) {
 		if agent.RuntimeID != "" && agent.RuntimeID != runtimeID {
 			continue
 		}
-		cap := agent.MaxConcurrency
+		// Pre-check only: the UPDATE below re-counts both caps under the write
-		if rt.MaxConcurrency < cap {
+		// lock. A count that cannot be read means unknown capacity, so refuse.
-			cap = rt.MaxConcurrency
+		var agentBusy, runtimeBusy int
+		err = s.db.QueryRow(claimCapacitySQL, cand.agentID, runtimeID, runtimeID).Scan(&agentBusy, &runtimeBusy)
+		if err != nil {
+			return Task{}, false, fmt.Errorf("claim capacity count: %w", err)
 		}
-		var running int
+		if agentBusy >= agent.MaxConcurrency || runtimeBusy >= rt.MaxConcurrency {
-		_ = s.db.QueryRow(`SELECT COUNT(*) FROM task WHERE agent_id=? AND status IN ('dispatched','running')`, cand.agentID).Scan(&running)
-		if running >= cap {
 			continue
 		}
 		lease := at.UTC().Add(5 * time.Minute).Format(time.RFC3339Nano)
-		res, err := s.db.Exec(`UPDATE task SET status='dispatched', runtime_id=?, lease_until=?, updated_at=?
+		res, err := s.db.Exec(claimSQL,
-			WHERE id=? AND status='queued'
+			runtimeID, lease, now(), cand.id, cand.agentID, agent.MaxConcurrency,
-			AND (SELECT COUNT(*) FROM task WHERE agent_id=? AND status IN ('dispatched','running')) < ?`,
+			runtimeID, runtimeID, rt.MaxConcurrency)
-			runtimeID, lease, now(), cand.id, cand.agentID, cap)
 		if err != nil {
 			return Task{}, false, err
 		}

adapters/sqlite/queue_test.go

@@ -0,0 +1,325 @@
+package platform
+
+import (
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+)
+
+// openQueue opens a hermetic store in a temp dir and returns it with its path.
+func openQueue(t *testing.T) (*Store, string) {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "queue.db")
+	s, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = s.Close() })
+	return s, path
+}
+
+// newQueueRuntime registers a runtime with the given cap.
+func newQueueRuntime(t *testing.T, s *Store, workspaceID, name string, cap int) Runtime {
+	t.Helper()
+	rt, err := s.RegisterRuntime(Runtime{
+		WorkspaceID: workspaceID, Name: name, Command: "in-process",
+		Protocol: "agent-native", MaxConcurrency: cap,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return rt
+}
+
+// newQueueAgent creates an agent with the given cap, pinned to runtimeID
+// (an empty runtimeID leaves the agent unbound, claimable by any runtime).
+func newQueueAgent(t *testing.T, s *Store, workspaceID, name, runtimeID string, cap int) Agent {
+	t.Helper()
+	agent, err := s.CreateAgent(Agent{WorkspaceID: workspaceID, Name: name, RuntimeID: runtimeID, MaxConcurrency: cap})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return agent
+}
+
+// queueTask creates an issue and queues one task for agentID.
+func queueTask(t *testing.T, s *Store, workspaceID, agentID, title string) Task {
+	t.Helper()
+	issue, err := s.CreateIssue(Issue{WorkspaceID: workspaceID, Title: title, Body: title, Status: "todo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task, err := s.enqueue(issue, agentID, false, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return task
+}
+
+// taskStatus reads the durable status of one task.
+func taskStatus(t *testing.T, s *Store, id string) string {
+	t.Helper()
+	var status string
+	if err := s.db.QueryRow(`SELECT status FROM task WHERE id=?`, id).Scan(&status); err != nil {
+		t.Fatal(err)
+	}
+	return status
+}
+
+// R6: the runtime cap bounds every agent on that runtime, not one agent only.
+// A per-agent cap of 4 each still yields a single in-flight task at cap 1.
+func TestClaimEnforcesRuntimeCapAcrossAgents(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		bindToRT bool
+	}{
+		{name: "agents bound to the runtime", bindToRT: true},
+		{name: "agents not bound to the runtime", bindToRT: false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s, _ := openQueue(t)
+			ws, err := s.CreateWorkspace("lab")
+			if err != nil {
+				t.Fatal(err)
+			}
+			rt := newQueueRuntime(t, s, ws.ID, "rt", 1)
+			runtimeID := ""
+			if tc.bindToRT {
+				runtimeID = rt.ID
+			}
+			first := newQueueAgent(t, s, ws.ID, "a1", runtimeID, 4)
+			second := newQueueAgent(t, s, ws.ID, "a2", runtimeID, 4)
+			t1 := queueTask(t, s, ws.ID, first.ID, "one")
+			t2 := queueTask(t, s, ws.ID, second.ID, "two")
+
+			claimed, ok, err := s.Claim(rt.ID, time.Now())
+			if err != nil || !ok || claimed.ID != t1.ID || claimed.RuntimeID != rt.ID {
+				t.Fatalf("first claim %+v ok=%v err=%v want task %s on %s", claimed, ok, err, t1.ID, rt.ID)
+			}
+			if _, ok, err := s.Claim(rt.ID, time.Now()); err != nil || ok {
+				t.Fatalf("runtime cap 1 must refuse the other agent: ok=%v err=%v", ok, err)
+			}
+			if got := taskStatus(t, s, t2.ID); got != "queued" {
+				t.Fatalf("second task %q must stay queued while the runtime is full", got)
+			}
+
+			if err := s.FinishTask(claimed.ID, "completed", ""); err != nil {
+				t.Fatal(err)
+			}
+			next, ok, err := s.Claim(rt.ID, time.Now())
+			if err != nil || !ok || next.ID != t2.ID {
+				t.Fatalf("claim after release %+v ok=%v err=%v want task %s", next, ok, err, t2.ID)
+			}
+		})
+	}
+}
+
+// The agent cap holds when two runtimes compete for the same agent.
+func TestClaimEnforcesAgentCapAcrossRuntimes(t *testing.T) {
+	s, _ := openQueue(t)
+	ws, err := s.CreateWorkspace("lab")
+	if err != nil {
+		t.Fatal(err)
+	}
+	rtA := newQueueRuntime(t, s, ws.ID, "rt-a", 2)
+	rtB := newQueueRuntime(t, s, ws.ID, "rt-b", 2)
+	agent := newQueueAgent(t, s, ws.ID, "solo", "", 1)
+	t1 := queueTask(t, s, ws.ID, agent.ID, "one")
+	t2 := queueTask(t, s, ws.ID, agent.ID, "two")
+
+	claimed, ok, err := s.Claim(rtA.ID, time.Now())
+	if err != nil || !ok || claimed.ID != t1.ID {
+		t.Fatalf("rt-a claim %+v ok=%v err=%v want task %s", claimed, ok, err, t1.ID)
+	}
+	if _, ok, err := s.Claim(rtB.ID, time.Now()); err != nil || ok {
+		t.Fatalf("agent cap 1 must bound the second runtime: ok=%v err=%v", ok, err)
+	}
+	if got := taskStatus(t, s, t2.ID); got != "queued" {
+		t.Fatalf("second task %q must stay queued while the agent is busy", got)
+	}
+}
+
+// Concurrent claims never overshoot the runtime cap or an agent cap.
+func TestClaimConcurrentNeverExceedsCaps(t *testing.T) {
+	s, _ := openQueue(t)
+	ws, err := s.CreateWorkspace("lab")
+	if err != nil {
+		t.Fatal(err)
+	}
+	rt := newQueueRuntime(t, s, ws.ID, "rt", 2)
+	tight := newQueueAgent(t, s, ws.ID, "tight", rt.ID, 1)
+	loose := newQueueAgent(t, s, ws.ID, "loose", rt.ID, 8)
+	other := newQueueAgent(t, s, ws.ID, "other", rt.ID, 8)
+	for _, agent := range []Agent{tight, loose, other} {
+		queueTask(t, s, ws.ID, agent.ID, "task for "+agent.Name)
+	}
+
+	const workers = 12
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	claimed := map[string]string{} // task id -> agent id
+	failures := []string{}
+	for i := 0; i < workers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			task, ok, err := s.Claim(rt.ID, time.Now())
+			mu.Lock()
+			defer mu.Unlock()
+			switch {
+			case err != nil:
+				failures = append(failures, err.Error())
+			case ok:
+				claimed[task.ID] = task.AgentID
+			}
+		}()
+	}
+	wg.Wait()
+
+	if len(failures) > 0 {
+		t.Fatalf("concurrent claims failed: %v", failures)
+	}
+	if len(claimed) == 0 {
+		t.Fatal("no claim landed")
+	}
+	if len(claimed) > rt.MaxConcurrency {
+		t.Fatalf("runtime cap %d exceeded: %v", rt.MaxConcurrency, claimed)
+	}
+	perAgent := map[string]int{}
+	for _, agentID := range claimed {
+		perAgent[agentID]++
+	}
+	caps := map[string]int{tight.ID: tight.MaxConcurrency, loose.ID: loose.MaxConcurrency, other.ID: other.MaxConcurrency}
+	for agentID, n := range perAgent {
+		if cap, known := caps[agentID]; !known {
+			t.Fatalf("unknown agent %s claimed", agentID)
+		} else if n > cap {
+			t.Fatalf("agent cap %d exceeded by %d for %s: %v", cap, n, agentID, claimed)
+		}
+	}
+	if got := perAgent[tight.ID]; got > 1 {
+		t.Fatalf("agent cap 1 exceeded by %d: %v", got, claimed)
+	}
+}
+
+// A completed or lease-expired task gives its slot back.
+func TestClaimCapacityFreedByReleaseAndReclaim(t *testing.T) {
+	s, _ := openQueue(t)
+	ws, err := s.CreateWorkspace("lab")
+	if err != nil {
+		t.Fatal(err)
+	}
+	rt := newQueueRuntime(t, s, ws.ID, "rt", 1)
+	agent := newQueueAgent(t, s, ws.ID, "solo", rt.ID, 1)
+	t1 := queueTask(t, s, ws.ID, agent.ID, "one")
+	t2 := queueTask(t, s, ws.ID, agent.ID, "two")
+
+	claimed, ok, err := s.Claim(rt.ID, time.Now())
+	if err != nil || !ok || claimed.ID != t1.ID {
+		t.Fatalf("first claim %+v ok=%v err=%v want task %s", claimed, ok, err, t1.ID)
+	}
+	if _, ok, err := s.Claim(rt.ID, time.Now()); err != nil || ok {
+		t.Fatalf("the agent is at its cap: ok=%v err=%v", ok, err)
+	}
+
+	if err := s.FinishTask(claimed.ID, "completed", ""); err != nil {
+		t.Fatal(err)
+	}
+	second, ok, err := s.Claim(rt.ID, time.Now())
+	if err != nil || !ok || second.ID != t2.ID {
+		t.Fatalf("claim after release %+v ok=%v err=%v want task %s", second, ok, err, t2.ID)
+	}
+
+	n, err := s.RecoverOrphans(time.Now().Add(10 * time.Minute))
+	if err != nil || n != 1 {
+		t.Fatalf("reclaim %d %v", n, err)
+	}
+	if got := taskStatus(t, s, t2.ID); got != "queued" {
+		t.Fatalf("reclaimed task %q must be queued again", got)
+	}
+	third, ok, err := s.Claim(rt.ID, time.Now())
+	if err != nil || !ok || third.ID != t2.ID || third.Attempts != 1 {
+		t.Fatalf("claim after reclaim %+v ok=%v err=%v want task %s with one attempt", third, ok, err, t2.ID)
+	}
+}
+
+// A capacity read that fails refuses the claim; it never allows it.
+func TestClaimRefusesWhenReadFails(t *testing.T) {
+	t.Run("unreadable store", func(t *testing.T) {
+		s, path := openQueue(t)
+		ws, err := s.CreateWorkspace("lab")
+		if err != nil {
+			t.Fatal(err)
+		}
+		rt := newQueueRuntime(t, s, ws.ID, "rt", 2)
+		agent := newQueueAgent(t, s, ws.ID, "a", rt.ID, 2)
+		task := queueTask(t, s, ws.ID, agent.ID, "one")
+		if err := s.Close(); err != nil {
+			t.Fatal(err)
+		}
+		if claimed, ok, err := s.Claim(rt.ID, time.Now()); ok {
+			t.Fatalf("an unreadable store handed out %+v (err=%v)", claimed, err)
+		}
+		again, err := Open(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer again.Close()
+		if got := taskStatus(t, again, task.ID); got != "queued" {
+			t.Fatalf("task %q after a failed capacity read", got)
+		}
+	})
+
+	t.Run("agent lookup fails", func(t *testing.T) {
+		s, _ := openQueue(t)
+		ws, err := s.CreateWorkspace("lab")
+		if err != nil {
+			t.Fatal(err)
+		}
+		rt := newQueueRuntime(t, s, ws.ID, "rt", 2)
+		agent := newQueueAgent(t, s, ws.ID, "a", rt.ID, 2)
+		task := queueTask(t, s, ws.ID, agent.ID, "one")
+		if _, err := s.db.Exec(`DROP TABLE agent`); err != nil {
+			t.Fatal(err)
+		}
+		claimed, ok, err := s.Claim(rt.ID, time.Now())
+		if ok || err != nil {
+			t.Fatalf("a failed agent read must skip the candidate: %+v ok=%v err=%v", claimed, ok, err)
+		}
+		if _, err := s.db.Exec(`CREATE TABLE agent (id TEXT PRIMARY KEY, runtime_id TEXT)`); err != nil {
+			t.Fatal(err)
+		}
+		if got := taskStatus(t, s, task.ID); got != "queued" {
+			t.Fatalf("task %q after a skipped candidate", got)
+		}
+	})
+
+	// The guarded count runs inside the atomic claim: when it cannot be read
+	// the statement must fail rather than hand the task out.
+	t.Run("guard count fails", func(t *testing.T) {
+		s, _ := openQueue(t)
+		ws, err := s.CreateWorkspace("lab")
+		if err != nil {
+			t.Fatal(err)
+		}
+		rt := newQueueRuntime(t, s, ws.ID, "rt", 2)
+		agent := newQueueAgent(t, s, ws.ID, "a", rt.ID, 2)
+		task := queueTask(t, s, ws.ID, agent.ID, "one")
+		if _, err := s.db.Exec(`DROP TABLE agent`); err != nil {
+			t.Fatal(err)
+		}
+		lease := time.Now().UTC().Add(5 * time.Minute).Format(time.RFC3339Nano)
+		res, err := s.db.Exec(claimSQL, rt.ID, lease, now(), task.ID, agent.ID, 2, rt.ID, rt.ID, rt.MaxConcurrency)
+		if err == nil {
+			n, _ := res.RowsAffected()
+			t.Fatalf("a failed guard count claimed %d task(s)", n)
+		}
+		if _, err := s.db.Exec(`CREATE TABLE agent (id TEXT PRIMARY KEY, runtime_id TEXT)`); err != nil {
+			t.Fatal(err)
+		}
+		if got := taskStatus(t, s, task.ID); got != "queued" {
+			t.Fatalf("task %q after a failed guard count", got)
+		}
+	})
+}

adapters/sqlite/runstore.go

@@ -22,13 +22,19 @@ func (s *Store) RunStore() *SQLiteStore {
 	return &SQLiteStore{db: s.db}
 }
 
+// runSchemaVersion 是 squad run 存储的 schema 版本（squad_run_schema_version 记档）。
+// v1 起步，v2 补齐协作态列（mailbox/handoffs/board）。迁移只做增量：
+// 新库直接建到最新形态，旧库经 ALTER TABLE ADD COLUMN 补列（既有行读出为空默认，不丢数据）。
+const runSchemaVersion = 2
+
 const runSchema = `
 CREATE TABLE IF NOT EXISTS squad_run_state (
   ws_dir TEXT PRIMARY KEY,
   meta TEXT NOT NULL, workflow TEXT NOT NULL, squad TEXT NOT NULL, bb TEXT NOT NULL,
   template TEXT, approved TEXT, repair_budget_remaining INTEGER NOT NULL DEFAULT 0,
   repair_budget_granted INTEGER NOT NULL DEFAULT 0,
-  clarifications TEXT, steering TEXT, usage TEXT, updated_at TEXT NOT NULL
+  clarifications TEXT, steering TEXT, usage TEXT, updated_at TEXT NOT NULL,
+  mailbox TEXT, handoffs TEXT, board TEXT
 );
 CREATE TABLE IF NOT EXISTS squad_run_gate (
   ws_dir TEXT PRIMARY KEY, gate TEXT NOT NULL
@@ -41,13 +47,90 @@ CREATE TABLE IF NOT EXISTS squad_run_checkpoint (
 CREATE TABLE IF NOT EXISTS squad_run_leader (
   ws_dir TEXT PRIMARY KEY, data TEXT NOT NULL
 );
+CREATE TABLE IF NOT EXISTS squad_run_schema_version (
+  id INTEGER PRIMARY KEY CHECK (id = 1), version INTEGER NOT NULL
+);
 `
 
+// runStateV2Columns 是 v2 在 squad_run_state 上新增的协作态列（列名为代码内常量，非外部输入）。
+var runStateV2Columns = []string{"mailbox", "handoffs", "board"}
+
+// migrate 建表并按版本步进：老库补列后再记版本号，逐次读写都是幂等的
+// （每次 Save/Load 都调；补列只在版本落后时发生，且既有行不被改写）。
 func (s *SQLiteStore) migrate() error {
-	_, err := s.db.Exec(runSchema)
+	if _, err := s.db.Exec(runSchema); err != nil {
+		return err
+	}
+	cur, err := s.runSchemaVersion()
+	if err != nil {
+		return err
+	}
+	if cur >= runSchemaVersion {
+		return nil
+	}
+	if err := s.addRunStateColumns(runStateV2Columns); err != nil {
+		return err
+	}
+	_, err = s.db.Exec(`INSERT INTO squad_run_schema_version (id, version) VALUES (1, ?)
+	  ON CONFLICT(id) DO UPDATE SET version = excluded.version`, runSchemaVersion)
 	return err
 }
 
+// runSchemaVersion 读取当前 schema 版本；无记录 = 0（v1 及更早的库）。
+func (s *SQLiteStore) runSchemaVersion() (int, error) {
+	var v int
+	err := s.db.QueryRow(`SELECT version FROM squad_run_schema_version WHERE id = 1`).Scan(&v)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return 0, nil
+		}
+		return 0, err
+	}
+	return v, nil
+}
+
+// runStateColumns 读出 squad_run_state 现存列名（迁移判存在性用）。
+func (s *SQLiteStore) runStateColumns() (map[string]bool, error) {
+	rows, err := s.db.Query(`PRAGMA table_info(squad_run_state)`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := map[string]bool{}
+	for rows.Next() {
+		var cid, notnull, pk int
+		var name, ctype string
+		var dflt sql.NullString
+		if err := rows.Scan(&cid, &name, &ctype, &notnull, &dflt, &pk); err != nil {
+			return nil, err
+		}
+		out[name] = true
+	}
+	return out, rows.Err()
+}
+
+// addRunStateColumns 为缺失的列补 ALTER TABLE ADD COLUMN（既有行保持原值，
+// 新列读出为 NULL → 空默认）。
+func (s *SQLiteStore) addRunStateColumns(cols []string) error {
+	have, err := s.runStateColumns()
+	if err != nil {
+		return err
+	}
+	for _, col := range cols {
+		if have[col] {
+			continue
+		}
+		if _, err := s.db.Exec(`ALTER TABLE squad_run_state ADD COLUMN ` + col + ` TEXT`); err != nil {
+			// 并发迁移：另一进程可能刚补上同一列，复核后不算失败。
+			if again, err2 := s.runStateColumns(); err2 == nil && again[col] {
+				continue
+			}
+			return fmt.Errorf("migrate squad_run_state: add column %s: %w", col, err)
+		}
+	}
+	return nil
+}
+
 // OpenSQLiteStore 独立打开一个 squad.Store（不经 platform.Store 的场景，如测试）。
 func OpenSQLiteStore(path string) (*SQLiteStore, error) {
 	db, err := sql.Open("sqlite", path+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)")
@@ -70,54 +153,114 @@ func j(v any) string {
 	return string(b)
 }
 
+// runStateColumn 是 squad_run_state 的一列：列名 + 待编码值（编码失败即报错，
+// 不用空串顶替——空串读回来是「空默认」，等于静默丢数据）。
+type runStateColumn struct {
+	name string
+	val  any
+}
+
+// runStateColumnsOf 列出 RunState 的全部持久化字段（含协作态 mailbox/handoffs/board）。
+func runStateColumnsOf(st *squad.RunState) []runStateColumn {
+	return []runStateColumn{
+		{"meta", st.Meta},
+		{"workflow", st.Plan.Workflow},
+		{"squad", st.Plan.Squad},
+		{"bb", st.BB},
+		{"approved", st.Approved},
+		{"clarifications", st.Clarifications},
+		{"steering", st.Steering},
+		{"usage", st.Usage},
+		{"mailbox", st.Mailbox},
+		{"handoffs", st.Handoffs},
+		{"board", st.Board},
+	}
+}
+
 func (s *SQLiteStore) SaveRun(st *squad.RunState) error {
 	if err := s.migrate(); err != nil {
 		return err
 	}
+	encoded := map[string]string{}
+	for _, col := range runStateColumnsOf(st) {
+		data, err := json.Marshal(col.val)
+		if err != nil {
+			return fmt.Errorf("save run %s: encode column %s: %w", st.Meta.WorkDir, col.name, err)
+		}
+		encoded[col.name] = string(data)
+	}
 	_, err := s.db.Exec(`INSERT INTO squad_run_state
-	  (ws_dir, meta, workflow, squad, bb, template, approved, repair_budget_remaining, repair_budget_granted, clarifications, steering, usage, updated_at)
+	  (ws_dir, meta, workflow, squad, bb, template, approved, repair_budget_remaining, repair_budget_granted, clarifications, steering, usage, mailbox, handoffs, board, updated_at)
-	  VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)
+	  VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
 	  ON CONFLICT(ws_dir) DO UPDATE SET
 	    meta=excluded.meta, workflow=excluded.workflow, squad=excluded.squad, bb=excluded.bb,
 	    template=excluded.template, approved=excluded.approved,
 	    repair_budget_remaining=excluded.repair_budget_remaining,
 	    repair_budget_granted=excluded.repair_budget_granted,
 	    clarifications=excluded.clarifications, steering=excluded.steering,
-	    usage=excluded.usage, updated_at=excluded.updated_at`,
+	    usage=excluded.usage, mailbox=excluded.mailbox, handoffs=excluded.handoffs,
-		st.Meta.WorkDir, j(st.Meta), j(st.Plan.Workflow), j(st.Plan.Squad), j(st.BB),
+	    board=excluded.board, updated_at=excluded.updated_at`,
-		st.Template, j(st.Approved), st.RepairBudgetRemaining, st.RepairBudgetGranted,
+		st.Meta.WorkDir, encoded["meta"], encoded["workflow"], encoded["squad"], encoded["bb"],
-		j(st.Clarifications), j(st.Steering), j(st.Usage), now())
+		st.Template, encoded["approved"], st.RepairBudgetRemaining, st.RepairBudgetGranted,
+		encoded["clarifications"], encoded["steering"], encoded["usage"],
+		encoded["mailbox"], encoded["handoffs"], encoded["board"], now())
 	return err
 }
 
+// LoadRun 按列位置读回 RunState。解码目标逐列具名指定——绝不用 JSON 文本当键：
+// 文本相同的两列（例如两条一模一样的澄清/补充指令）会互相覆盖，把一列丢掉。
 func (s *SQLiteStore) LoadRun(wsDir string) (*squad.RunState, error) {
 	if err := s.migrate(); err != nil {
 		return nil, err
 	}
 	st := &squad.RunState{}
-	var meta, wf, sq, bb, template, approved, clar, steering, usage string
+	// 列 → 扫描目标（顺序与 SQL 一一对应）。
-	row := s.db.QueryRow(`SELECT meta, workflow, squad, bb, template, approved,
+	var tpl sql.NullString
-	  repair_budget_remaining, repair_budget_granted, clarifications, steering, usage
-	  FROM squad_run_state WHERE ws_dir = ?`, wsDir)
 	var granted int
-	err := row.Scan(&meta, &wf, &sq, &bb, &template, &approved,
+	var meta, wf, sq, bb, approved, clar, steering, usage, mailbox, handoffs, board sql.NullString
-		&st.RepairBudgetRemaining, &granted, &clar, &steering, &usage)
+	row := s.db.QueryRow(`SELECT meta, workflow, squad, bb, template, approved,
-	if err != nil {
+	  repair_budget_remaining, repair_budget_granted, clarifications, steering, usage,
+	  mailbox, handoffs, board
+	  FROM squad_run_state WHERE ws_dir = ?`, wsDir)
+	if err := row.Scan(&meta, &wf, &sq, &bb, &tpl, &approved,
+		&st.RepairBudgetRemaining, &granted, &clar, &steering, &usage,
+		&mailbox, &handoffs, &board); err != nil {
 		if err == sql.ErrNoRows {
 			return nil, fmt.Errorf("load run %s: %w", wsDir, os.ErrNotExist)
 		}
 		return nil, err
 	}
 	st.RepairBudgetGranted = granted
-	for pair, dst := range map[string]any{
+	for _, dec := range []struct {
-		meta: &st.Meta, wf: &st.Plan.Workflow, sq: &st.Plan.Squad, bb: &st.BB,
+		col string
-		approved: &st.Approved, clar: &st.Clarifications, steering: &st.Steering, usage: &st.Usage,
+		raw sql.NullString
+		dst any
+	}{
+		{"meta", meta, &st.Meta},
+		{"workflow", wf, &st.Plan.Workflow},
+		{"squad", sq, &st.Plan.Squad},
+		{"bb", bb, &st.BB},
+		{"approved", approved, &st.Approved},
+		{"clarifications", clar, &st.Clarifications},
+		{"steering", steering, &st.Steering},
+		{"usage", usage, &st.Usage},
+		{"mailbox", mailbox, &st.Mailbox},
+		{"handoffs", handoffs, &st.Handoffs},
+		{"board", board, &st.Board},
 	} {
-		if err := json.Unmarshal([]byte(pair), dst); err != nil {
+		// 旧行缺列（NULL）或空串 = 该字段的空默认，不报错、不影响其他列。
-			return nil, fmt.Errorf("load run %s: parse: %w", wsDir, err)
+		if !dec.raw.Valid || dec.raw.String == "" {
+			continue
+		}
+		if err := json.Unmarshal([]byte(dec.raw.String), dec.dst); err != nil {
+			return nil, fmt.Errorf("load run %s: parse column %s: %w", wsDir, dec.col, err)
 		}
 	}
-	st.Template = template
+	st.Template = tpl.String
+	if st.Meta.WorkDir == "" {
+		// 行键（ws_dir）即权威工作区路径：meta 旧数据缺该字段时由行键兜底。
+		st.Meta.WorkDir = wsDir
+	}
 	if st.Approved == nil {
 		st.Approved = map[string]bool{}
 	}

adapters/sqlite/runstore_test.go

@@ -2,9 +2,14 @@ package platform
 
 import (
 	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -12,10 +17,12 @@ import (
 )
 
 func newTestRunState(wsDir string) *squad.RunState {
+	// 固定时刻：多次构造出的状态除工作区路径外完全一致，可直接比对两个后端的读回结果。
+	now := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
 	return &squad.RunState{
 		Meta: squad.RunMetadata{
 			ID: "run_test", Status: "executing", WorkDir: wsDir,
-			CreatedAt: time.Now(), UpdatedAt: time.Now(),
+			CreatedAt: now, UpdatedAt: now,
 			Metadata: map[string]string{"mode": ""},
 		},
 		Plan: squad.SquadPlan{
@@ -28,14 +35,28 @@ func newTestRunState(wsDir string) *squad.RunState {
 			},
 			Squad: squad.SquadConfig{SquadName: "demo"},
 		},
-		BB:                    squad.BlackboardState{Facts: []squad.FactItem{{ID: "f1", Claim: "x"}}},
+		BB:                    squad.BlackboardState{Facts: []squad.FactItem{{ID: "f1", Claim: "x"}}, Fields: map[string]string{"topic": "demo"}},
 		Template:              "tpl",
 		Approved:              map[string]bool{"step:s1": true},
 		RepairBudgetRemaining: 1,
 		RepairBudgetGranted:   0,
-		Clarifications:        []string{"c1"},
+		// 内容完全相同的重复条目（且 clarifications 与 steering 的 JSON 文本一样）：
-		Steering:              []string{"st1"},
+		// 内容键寻址的解码会把它们互相覆盖掉——这里逐条都要活下来。
-		Usage:                 map[string]*squad.Usage{"writer": {PromptTokens: 10, CompletionTokens: 5, Steps: 1}},
+		Clarifications: []string{"同一句澄清", "同一句澄清"},
+		Steering:       []string{"同一句澄清", "同一句澄清"},
+		Mailbox: []squad.MailMessage{
+			{ID: "m1", From: "researcher", To: "writer", Text: "同一句澄清", At: now},
+			{ID: "m2", From: "researcher", To: "writer", Text: "同一句澄清", At: now},
+		},
+		Handoffs: []squad.HandoffRecord{
+			{FromRole: "researcher", FromStep: "s1", ToRole: "writer", Output: "同一句澄清", At: now},
+			{FromRole: "writer", FromStep: "s2", ToRole: "reviewer", Output: "同一句澄清", At: now},
+		},
+		Board: []squad.BoardTask{
+			{ID: "s1", Status: "done", Attempts: 1},
+			{ID: "s2", Status: "claimed", ClaimedBy: "writer", LeaseExpires: now.Add(time.Minute), Attempts: 2},
+		},
+		Usage: map[string]*squad.Usage{"writer": {PromptTokens: 10, CompletionTokens: 5, Steps: 1}},
 	}
 }
 
@@ -58,9 +79,20 @@ func TestSQLiteStoreRoundTrip(t *testing.T) {
 	if got.Meta.ID != "run_test" || got.Plan.Workflow.WorkflowID != "wf1" || got.Template != "tpl" {
 		t.Fatalf("loaded = %+v", got.Meta)
 	}
-	if !got.Approved["step:s1"] || got.RepairBudgetRemaining != 1 || len(got.Clarifications) != 1 || len(got.Steering) != 1 {
+	if !got.Approved["step:s1"] || got.RepairBudgetRemaining != 1 ||
+		len(got.Clarifications) != 2 || len(got.Steering) != 2 {
 		t.Fatalf("scalars = %+v", got)
 	}
+	// 协作态必须真的落库（F11a）：重启后邮箱/移交/任务板不能静默丢。
+	if len(got.Mailbox) != 2 || got.Mailbox[0].ID != "m1" || got.Mailbox[1].ID != "m2" {
+		t.Fatalf("mailbox = %+v", got.Mailbox)
+	}
+	if len(got.Handoffs) != 2 || got.Handoffs[1].ToRole != "reviewer" {
+		t.Fatalf("handoffs = %+v", got.Handoffs)
+	}
+	if len(got.Board) != 2 || got.Board[1].Status != "claimed" || got.Board[1].ClaimedBy != "writer" {
+		t.Fatalf("board = %+v", got.Board)
+	}
 	if got.Usage["writer"].PromptTokens != 10 {
 		t.Fatalf("usage = %+v", got.Usage)
 	}
@@ -261,3 +293,467 @@ func TestEngineFailedResultWithCompletePatchesStaysFailedOnSQLiteStore(t *testin
 }
 
 const goodReportLite = "# 测试交付物\n\n## 概述\n\n这是由脚本后端生成的合格交付物，长度超过五十个字符以满足最小质检要求。\n"
+
+// ---- F11a/F11b：SQLite 与 FileStore 同一契约、重复条目不丢、旧库升级 ----
+
+// assertSameRunState 逐字段比对两个后端读回的运行状态（同一契约）。
+// 字段按规范 JSON 比对：时间等值字段（创建/更新时间、消息时刻、租约到期）在两侧
+// 都经历过 JSON 往返，按墙钟等价判定，不受内部表示（时区指针/单调时钟）影响。
+func assertSameRunState(t *testing.T, got, want *squad.RunState) {
+	t.Helper()
+	sameJSON := func(label string, a, b any) {
+		t.Helper()
+		ab, err := json.Marshal(a)
+		if err != nil {
+			t.Fatalf("%s: marshal got: %v", label, err)
+		}
+		bb, err := json.Marshal(b)
+		if err != nil {
+			t.Fatalf("%s: marshal want: %v", label, err)
+		}
+		if string(ab) != string(bb) {
+			t.Fatalf("%s differs:\n got %s\nwant %s", label, ab, bb)
+		}
+	}
+	sameJSON("meta", got.Meta, want.Meta)
+	sameJSON("plan", got.Plan, want.Plan)
+	sameJSON("blackboard", got.BB, want.BB)
+	sameJSON("template", got.Template, want.Template)
+	sameJSON("approved", got.Approved, want.Approved)
+	sameJSON("repair_budget_remaining", got.RepairBudgetRemaining, want.RepairBudgetRemaining)
+	sameJSON("repair_budget_granted", got.RepairBudgetGranted, want.RepairBudgetGranted)
+	sameJSON("clarifications", got.Clarifications, want.Clarifications)
+	sameJSON("steering", got.Steering, want.Steering)
+	sameJSON("mailbox", got.Mailbox, want.Mailbox)
+	sameJSON("handoffs", got.Handoffs, want.Handoffs)
+	sameJSON("board", got.Board, want.Board)
+	sameJSON("usage", got.Usage, want.Usage)
+}
+
+// TestRunStoreContractParity：FileStore 与 SQLiteStore 满足同一份 Store 契约——
+// 同一份运行状态（含 mailbox/handoffs/board）写进去，读回来必须一致。
+func TestRunStoreContractParity(t *testing.T) {
+	sq, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sq.Close()
+
+	fileWS := filepath.Join(t.TempDir(), "runs", "run_file")
+	sqlWS := filepath.Join(t.TempDir(), "runs", "run_sql")
+	// FileStore 侧的模板文件由工作区装配期写出（PrepareWorkspace），先补齐，
+	// 使两侧的 SaveRun/LoadRun 覆盖同一批字段。
+	if err := os.MkdirAll(fileWS, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(fileWS, "template.md"), []byte("tpl"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	want := newTestRunState(fileWS)
+	if err := (squad.FileStore{}).SaveRun(want); err != nil {
+		t.Fatal(err)
+	}
+	gotFile, err := (squad.FileStore{}).LoadRun(fileWS)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sqlState := newTestRunState(sqlWS)
+	if err := sq.SaveRun(sqlState); err != nil {
+		t.Fatal(err)
+	}
+	gotSQL, err := sq.LoadRun(sqlWS)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// SQLite 读回的内容与写进去的一致。
+	assertSameRunState(t, gotSQL, sqlState)
+	// 两个后端的读回结果逐字段一致（除工作区路径这一运行自身的键）。
+	assertSameRunState(t, gotFile, want)
+	fileMeta, sqlMeta := gotFile.Meta, gotSQL.Meta
+	fileMeta.WorkDir, sqlMeta.WorkDir = "", ""
+	fileJSON, _ := json.Marshal(fileMeta)
+	sqlJSON, _ := json.Marshal(sqlMeta)
+	if string(fileJSON) != string(sqlJSON) {
+		t.Fatalf("meta parity broken:\n file %s\n  sql %s", fileJSON, sqlJSON)
+	}
+}
+
+// TestSQLiteRunStoreKeepsIdenticalEntries：内容相同的两条澄清/补充指令必须都活着
+// （旧实现拿 JSON 文本当键，两条同文条目只剩一条）。
+func TestSQLiteRunStoreKeepsIdenticalEntries(t *testing.T) {
+	sq, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sq.Close()
+	ws := filepath.Join(t.TempDir(), "runs", "run_dup")
+	st := newTestRunState(ws)
+	if err := sq.SaveRun(st); err != nil {
+		t.Fatal(err)
+	}
+	got, err := sq.LoadRun(ws)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(got.Clarifications, []string{"同一句澄清", "同一句澄清"}) {
+		t.Fatalf("clarifications = %#v, want both identical entries", got.Clarifications)
+	}
+	if !reflect.DeepEqual(got.Steering, []string{"同一句澄清", "同一句澄清"}) {
+		t.Fatalf("steering = %#v, want both identical entries", got.Steering)
+	}
+	// 邮箱正文相同、ID 不同：两条都要在（不靠正文寻址）。
+	if len(got.Mailbox) != 2 || got.Mailbox[0].ID != "m1" || got.Mailbox[1].ID != "m2" {
+		t.Fatalf("mailbox = %+v", got.Mailbox)
+	}
+}
+
+// legacyRunStateDDL 是 F11 修复前的 squad_run_state（无协作态列），用于验证升级路径。
+const legacyRunStateDDL = `
+CREATE TABLE squad_run_state (
+  ws_dir TEXT PRIMARY KEY,
+  meta TEXT NOT NULL, workflow TEXT NOT NULL, squad TEXT NOT NULL, bb TEXT NOT NULL,
+  template TEXT, approved TEXT, repair_budget_remaining INTEGER NOT NULL DEFAULT 0,
+  repair_budget_granted INTEGER NOT NULL DEFAULT 0,
+  clarifications TEXT, steering TEXT, usage TEXT, updated_at TEXT NOT NULL
+);
+`
+
+// TestSQLiteRunStoreUpgradesLegacySchema：旧库（缺 mailbox/handoffs/board 列）打开后
+// 补列升级，既有一行原样读出，协作态为空默认不报错、不丢其他字段。
+func TestSQLiteRunStoreUpgradesLegacySchema(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "runs.db")
+	legacy, err := sql.Open("sqlite", path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := legacy.Exec(legacyRunStateDDL); err != nil {
+		t.Fatal(err)
+	}
+	ws := filepath.Join(t.TempDir(), "runs", "run_legacy")
+	meta, _ := json.Marshal(squad.RunMetadata{ID: "run_legacy", WorkDir: ws, Status: "executing"})
+	wf, _ := json.Marshal(squad.WorkflowPlan{WorkflowID: "wf_old", Status: "in_progress"})
+	sqCfg, _ := json.Marshal(squad.SquadConfig{SquadName: "legacy"})
+	bb, _ := json.Marshal(squad.BlackboardState{Fields: map[string]string{"k": "v"}})
+	approved, _ := json.Marshal(map[string]bool{"step:s1": true})
+	clar, _ := json.Marshal([]string{"旧澄清", "旧澄清"})
+	usage, _ := json.Marshal(map[string]*squad.Usage{"writer": {PromptTokens: 7, CompletionTokens: 3, Steps: 2}})
+	if _, err := legacy.Exec(`INSERT INTO squad_run_state
+	  (ws_dir, meta, workflow, squad, bb, template, approved, repair_budget_remaining,
+	   repair_budget_granted, clarifications, steering, usage, updated_at)
+	  VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`,
+		ws, meta, wf, sqCfg, bb, "旧模板", approved, 2, 1, clar, `["旧补充"]`, usage, "2026-01-01T00:00:00Z"); err != nil {
+		t.Fatal(err)
+	}
+	if err := legacy.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	sq, err := OpenSQLiteStore(path)
+	if err != nil {
+		t.Fatalf("open legacy db: %v", err)
+	}
+	defer sq.Close()
+
+	got, err := sq.LoadRun(ws)
+	if err != nil {
+		t.Fatalf("load upgraded row: %v", err)
+	}
+	// 旧数据无损。
+	if got.Meta.ID != "run_legacy" || got.Plan.Workflow.WorkflowID != "wf_old" ||
+		got.Plan.Squad.SquadName != "legacy" || got.Template != "旧模板" {
+		t.Fatalf("legacy row corrupted: %+v", got)
+	}
+	if !got.Approved["step:s1"] || got.RepairBudgetRemaining != 2 || got.RepairBudgetGranted != 1 {
+		t.Fatalf("legacy scalars lost: %+v", got)
+	}
+	if !reflect.DeepEqual(got.Clarifications, []string{"旧澄清", "旧澄清"}) {
+		t.Fatalf("legacy clarifications lost: %#v", got.Clarifications)
+	}
+	if !reflect.DeepEqual(got.Steering, []string{"旧补充"}) {
+		t.Fatalf("legacy steering lost: %#v", got.Steering)
+	}
+	if got.Usage["writer"].PromptTokens != 7 || got.BB.Fields["k"] != "v" {
+		t.Fatalf("legacy usage/blackboard lost: %+v %+v", got.Usage, got.BB)
+	}
+	// 缺列读出为空默认。
+	if len(got.Mailbox) != 0 || len(got.Handoffs) != 0 || len(got.Board) != 0 {
+		t.Fatalf("new columns should default to empty: %+v %+v %+v", got.Mailbox, got.Handoffs, got.Board)
+	}
+	// 版本已记档，补列只做一次。
+	var version int
+	if err := sq.db.QueryRow(`SELECT version FROM squad_run_schema_version WHERE id = 1`).Scan(&version); err != nil {
+		t.Fatalf("schema version not recorded: %v", err)
+	}
+	if version != runSchemaVersion {
+		t.Fatalf("schema version = %d, want %d", version, runSchemaVersion)
+	}
+	// 升级后写入协作态并再次读回。
+	got.Mailbox = []squad.MailMessage{{ID: "m1", From: "a", To: "b", Text: "升级后写入"}}
+	got.Board = []squad.BoardTask{{ID: "s1", Status: "done"}}
+	got.Meta.Status = "completed"
+	if err := sq.SaveRun(got); err != nil {
+		t.Fatalf("save after upgrade: %v", err)
+	}
+	again, err := sq.LoadRun(ws)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(again.Mailbox) != 1 || again.Mailbox[0].Text != "升级后写入" ||
+		len(again.Board) != 1 || again.Board[0].ID != "s1" {
+		t.Fatalf("post-upgrade collaboration state lost: %+v %+v", again.Mailbox, again.Board)
+	}
+	if !reflect.DeepEqual(again.Clarifications, []string{"旧澄清", "旧澄清"}) {
+		t.Fatalf("upgrade write clobbered legacy clarifications: %#v", again.Clarifications)
+	}
+}
+
+// TestSQLiteStoreKeepsCollaborationStateAcrossRestart：进程重启（新连接）后，
+// 邮箱消息、所有权移交记录与任务板状态都还在——恢复不必从头重建协作上下文。
+func TestSQLiteStoreKeepsCollaborationStateAcrossRestart(t *testing.T) {
+	dbPath := filepath.Join(t.TempDir(), "runs.db")
+	runStore, err := OpenSQLiteStore(dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer runStore.Close()
+
+	dir := t.TempDir()
+	reg := squad.NewRoleRegistry()
+	for _, role := range []string{"researcher", "writer", "reviewer"} {
+		if err := reg.Register(squad.RoleSpec{Role: role, Backend: "script"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	backends := squad.NewBackendResolver()
+	backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{
+		Success: true, Logs: "ok",
+		Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")},
+	}})
+	blueprints := squad.NewBlueprintRegistry()
+	if err := blueprints.Register(squad.Blueprint{
+		Key: "handoff", Title: "移交演示", Match: []string{"移交演示"},
+		Steps: []squad.StepSpec{
+			{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"notes.md", "fields.json"}},
+			{ID: "s2", Role: "writer", Title: "成文", Targets: []string{"output.md"}, Handoff: "reviewer"},
+			{ID: "s3", Role: "reviewer", Title: "终审", Targets: []string{"output.md"}},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := blueprints.Register(squad.Blueprint{
+		Key: "board", Title: "任务板演示", Match: []string{"任务板演示"},
+		Steps: []squad.StepSpec{
+			{ID: "a", Role: "researcher", Title: "任务a", Targets: []string{"notes.md"}},
+			{ID: "b", Role: "writer", Title: "任务b", Needs: []string{"a"}, Targets: []string{"output.md"}},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	eng, err := squad.NewEngine(squad.Options{
+		DataDir: dir, Registry: reg, Blueprints: blueprints,
+		Planner:  squad.NewRulePlanner(),
+		Backends: backends, Sink: squad.NopSink{},
+		Store: runStore,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx := context.Background()
+	handed, err := eng.Run(ctx, squad.Goal{Text: "做一个移交演示", HitlMode: squad.HitlAuto}, squad.RunOptions{})
+	if err != nil || handed.Status != squad.RunCompleted {
+		t.Fatalf("handoff run: %v %+v", err, handed)
+	}
+	if err := eng.PostMessage(ctx, handed.RunID, squad.MailMessage{From: "human", To: "writer", Text: "补充口径"}); err != nil {
+		t.Fatalf("post message: %v", err)
+	}
+	boarded, err := eng.Run(ctx, squad.Goal{Text: "做一个任务板演示", HitlMode: squad.HitlAuto},
+		squad.RunOptions{Topology: squad.TopologyTaskBoard})
+	if err != nil || boarded.Status != squad.RunCompleted {
+		t.Fatalf("board run: %v %+v", err, boarded)
+	}
+
+	// 重启 = 新连接（新进程会重新 migrate）。
+	restarted, err := OpenSQLiteStore(dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer restarted.Close()
+
+	gotHanded, err := restarted.LoadRun(handed.WorkDir)
+	if err != nil {
+		t.Fatalf("reload handoff run: %v", err)
+	}
+	if len(gotHanded.Handoffs) != 1 || gotHanded.Handoffs[0].FromRole != "writer" || gotHanded.Handoffs[0].ToRole != "reviewer" {
+		t.Fatalf("handoffs lost across restart: %+v", gotHanded.Handoffs)
+	}
+	if len(gotHanded.Mailbox) != 1 || gotHanded.Mailbox[0].Text != "补充口径" || gotHanded.Mailbox[0].To != "writer" {
+		t.Fatalf("mailbox lost across restart: %+v", gotHanded.Mailbox)
+	}
+	gotBoarded, err := restarted.LoadRun(boarded.WorkDir)
+	if err != nil {
+		t.Fatalf("reload board run: %v", err)
+	}
+	if len(gotBoarded.Board) != 2 {
+		t.Fatalf("board lost across restart: %+v", gotBoarded.Board)
+	}
+	for _, task := range gotBoarded.Board {
+		if task.Status != "done" {
+			t.Fatalf("board task %s = %s, want done", task.ID, task.Status)
+		}
+	}
+}
+
+// faultingStore 在真实 SQLiteStore 之上注入落盘故障（其余方法透传）。
+type faultingStore struct {
+	inner          squad.Store
+	failCheckpoint bool
+	failTerminal   bool
+}
+
+func (f *faultingStore) SaveRun(st *squad.RunState) error {
+	if f.failTerminal && squad.RunStatus(st.Meta.Status).Terminal() {
+		return fmt.Errorf("injected sqlite run state save failure (%s)", st.Meta.Status)
+	}
+	return f.inner.SaveRun(st)
+}
+
+func (f *faultingStore) LoadRun(wsDir string) (*squad.RunState, error) { return f.inner.LoadRun(wsDir) }
+
+func (f *faultingStore) SaveCheckpoint(wsDir string, cp squad.Checkpoint) error {
+	if f.failCheckpoint {
+		return fmt.Errorf("injected sqlite checkpoint save failure (superstep %d)", cp.Superstep)
+	}
+	return f.inner.SaveCheckpoint(wsDir, cp)
+}
+
+func (f *faultingStore) LoadLatestCheckpoint(wsDir string) (*squad.Checkpoint, error) {
+	return f.inner.LoadLatestCheckpoint(wsDir)
+}
+
+func (f *faultingStore) SaveGate(wsDir string, gate *squad.HitlGate) error {
+	return f.inner.SaveGate(wsDir, gate)
+}
+
+func (f *faultingStore) LoadGate(wsDir string) (*squad.HitlGate, error) {
+	return f.inner.LoadGate(wsDir)
+}
+
+func (f *faultingStore) SaveLeader(wsDir string, data []byte) error {
+	return f.inner.SaveLeader(wsDir, data)
+}
+
+func (f *faultingStore) LoadLeader(wsDir string) ([]byte, error) { return f.inner.LoadLeader(wsDir) }
+
+// eventTally 收集事件类型计数（断言没有成功事件）。
+type eventTally struct {
+	mu   sync.Mutex
+	seen []string
+	sink squad.ProgressSink
+}
+
+func newEventTally() *eventTally {
+	t := &eventTally{}
+	t.sink = squad.ProgressFunc(func(ev squad.Event) {
+		t.mu.Lock()
+		defer t.mu.Unlock()
+		t.seen = append(t.seen, ev.Type)
+	})
+	return t
+}
+
+func (t *eventTally) has(typ string) bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	for _, got := range t.seen {
+		if got == typ {
+			return true
+		}
+	}
+	return false
+}
+
+func (t *eventTally) types() []string {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return append([]string(nil), t.seen...)
+}
+
+// TestSQLiteStoreSurfacesSaveFailures：SQLite 后端上的落盘故障必须可见——
+// 三档 checkpoint 模式下都不发 run_completed，运行要么报错、要么以失败收场。
+func TestSQLiteStoreSurfacesSaveFailures(t *testing.T) {
+	for _, mode := range []string{"sync", "async", "on_exit"} {
+		for _, fault := range []string{"checkpoint", "terminal"} {
+			t.Run(mode+"/"+fault, func(t *testing.T) {
+				runStore, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer runStore.Close()
+				dir := t.TempDir()
+				reg := squad.NewRoleRegistry()
+				for _, role := range []string{"researcher", "writer", "reviewer"} {
+					if err := reg.Register(squad.RoleSpec{Role: role, Backend: "script"}); err != nil {
+						t.Fatal(err)
+					}
+				}
+				backends := squad.NewBackendResolver()
+				backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{
+					Success: true, Logs: "ok",
+					Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")},
+				}})
+				blueprints := squad.NewBlueprintRegistry()
+				if err := blueprints.Register(squad.Blueprint{
+					Key: "demo", Title: "故障演示", Match: []string{"故障演示"},
+					Steps: []squad.StepSpec{
+						{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"notes.md", "fields.json"}},
+						{ID: "s2", Role: "writer", Title: "成文", Targets: []string{"output.md"}},
+						{ID: "s3", Role: "reviewer", Title: "终审", Targets: []string{"output.md"}},
+					},
+				}); err != nil {
+					t.Fatal(err)
+				}
+				store := &faultingStore{inner: runStore,
+					failCheckpoint: fault == "checkpoint", failTerminal: fault == "terminal"}
+				tally := newEventTally()
+				eng, err := squad.NewEngine(squad.Options{
+					DataDir: dir, Registry: reg, Blueprints: blueprints,
+					Planner: squad.NewRulePlanner(), Backends: backends,
+					Sink: tally.sink, Store: store,
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				res, err := eng.Run(context.Background(),
+					squad.Goal{Text: "做一个故障演示", HitlMode: squad.HitlAuto},
+					squad.RunOptions{CheckpointMode: mode})
+				if tally.has(squad.EvRunCompleted) {
+					t.Fatalf("run_completed emitted despite %s save failure: %v", fault, tally.types())
+				}
+				if fault == "checkpoint" && mode == "sync" {
+					if err == nil {
+						t.Fatalf("sync boundary failure must surface as error: %+v", res)
+					}
+					if !strings.Contains(err.Error(), "persistence failure") {
+						t.Fatalf("error not marked as persistence failure: %v", err)
+					}
+					return
+				}
+				if err != nil {
+					t.Fatalf("%s/%s: unexpected infra error: %v", mode, fault, err)
+				}
+				if res.Status != squad.RunFailed {
+					t.Fatalf("%s/%s: want failed, got %s", mode, fault, res.Status)
+				}
+				if !strings.Contains(res.Error, "persistence failure") {
+					t.Fatalf("%s/%s: failure not explained: %q", mode, fault, res.Error)
+				}
+				if !tally.has(squad.EvRunFailed) {
+					t.Fatalf("%s/%s: no run_failed event: %v", mode, fault, tally.types())
+				}
+			})
+		}
+	}
+}

squad/checkpoint_test.go

@@ -2,9 +2,11 @@ package squad
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 )
 
@@ -111,3 +113,225 @@ func TestGraphContinuityRejectsStructuralChange(t *testing.T) {
 		t.Fatalf("expected continuity rejection, got %v", err)
 	}
 }
+
+// ---- F11c：落盘失败的诚实性 ----
+
+// faultStore 包装底层 Store 注入落盘故障，其余路径照常透传。
+// failCheckpoint：任何 SaveCheckpoint 失败（async 档在出口 flush 时暴露）。
+// failTerminalSave：Meta 进入终态时的 SaveRun 失败（finalize/manualPass 的收口写）。
+type faultStore struct {
+	Store
+	mu               sync.Mutex
+	failCheckpoint   bool
+	failTerminalSave bool
+	checkpointWrites int
+	terminalAttempts int
+}
+
+func (f *faultStore) SaveCheckpoint(wsDir string, cp Checkpoint) error {
+	f.mu.Lock()
+	f.checkpointWrites++
+	shouldFail := f.failCheckpoint
+	f.mu.Unlock()
+	if shouldFail {
+		return fmt.Errorf("injected checkpoint write failure (superstep %d)", cp.Superstep)
+	}
+	return f.Store.SaveCheckpoint(wsDir, cp)
+}
+
+func (f *faultStore) SaveRun(st *RunState) error {
+	f.mu.Lock()
+	f.terminalAttempts++
+	shouldFail := f.failTerminalSave && RunStatus(st.Meta.Status).Terminal()
+	f.mu.Unlock()
+	if shouldFail {
+		return fmt.Errorf("injected terminal run state failure (%s)", st.Meta.Status)
+	}
+	return f.Store.SaveRun(st)
+}
+
+func (f *faultStore) writes() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.checkpointWrites
+}
+
+// eventRecorder 按序收集引擎事件（顺序断言用）。
+type eventRecorder struct {
+	mu   sync.Mutex
+	evs  []Event
+	sink ProgressSink
+}
+
+func newEventRecorder() *eventRecorder {
+	r := &eventRecorder{}
+	r.sink = ProgressFunc(func(ev Event) {
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		r.evs = append(r.evs, ev)
+	})
+	return r
+}
+
+func (r *eventRecorder) types() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]string, 0, len(r.evs))
+	for _, ev := range r.evs {
+		out = append(out, ev.Type)
+	}
+	return out
+}
+
+func (r *eventRecorder) has(t string) bool {
+	for _, got := range r.types() {
+		if got == t {
+			return true
+		}
+	}
+	return false
+}
+
+// newTestEngineWithStore 与 newTestEngine 同装配，额外注入 Store 与事件收集 sink。
+func newTestEngineWithStore(t *testing.T, store Store, rec *eventRecorder) *SquadEngine {
+	t.Helper()
+	dir := t.TempDir()
+	reg := NewRoleRegistry()
+	for _, role := range []string{"researcher", "writer", "reviewer"} {
+		if err := reg.Register(RoleSpec{Role: role, Backend: "script"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	backends := NewBackendResolver()
+	backends.RegisterInstance("script", &scriptBackend{script: map[string]map[string]string{
+		"01_researcher": {"notes.md": goodReport, "fields.json": "{}"},
+		"02_writer":     {"output.md": goodReport},
+		"03_reviewer":   {"output.md": goodReport},
+	}})
+	eng, err := NewEngine(Options{DataDir: dir, Registry: reg,
+		Planner: mockScriptPlanner(baseSteps()...), Backends: backends,
+		Sink: rec.sink, Store: store})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return eng
+}
+
+// assertNoCleanCompletion：落盘失败后不得再以「干净完成」面世——
+// 没有 run_completed 事件、也没有 completed/degraded 的返回值。
+func assertNoCleanCompletion(t *testing.T, res *RunResult, err error, rec *eventRecorder) {
+	t.Helper()
+	if rec.has(EvRunCompleted) {
+		t.Fatalf("run_completed emitted after persistence failure: %v", rec.types())
+	}
+	if res != nil {
+		if res.Status == RunCompleted || res.Status == RunDegraded {
+			t.Fatalf("clean completion after persistence failure: status=%s", res.Status)
+		}
+		if res.Status != RunFailed {
+			t.Fatalf("status = %s, want failed or an error return", res.Status)
+		}
+		if !strings.Contains(res.Error, persistenceFailurePrefix) {
+			t.Fatalf("failure not explained in result: %q", res.Error)
+		}
+	}
+	if err != nil && !strings.Contains(err.Error(), persistenceFailurePrefix) {
+		t.Fatalf("error does not name the persistence failure: %v", err)
+	}
+	if res == nil && err == nil {
+		t.Fatal("persistence failure vanished: neither result nor error")
+	}
+}
+
+// TestCheckpointFlushFailureIsVisible：出口 flush 的落盘失败必须可见，
+// 且先于任何完成事件/返回（三档 sync/async/on_exit 都一样）。
+func TestCheckpointFlushFailureIsVisible(t *testing.T) {
+	for _, mode := range []string{"sync", "async", "on_exit"} {
+		t.Run(mode, func(t *testing.T) {
+			store := &faultStore{Store: FileStore{}, failCheckpoint: true}
+			rec := newEventRecorder()
+			eng := newTestEngineWithStore(t, store, rec)
+			res, err := eng.Run(context.Background(), Goal{Text: "快照"}, RunOptions{CheckpointMode: mode})
+			if store.writes() == 0 {
+				t.Fatal("test did not reach any checkpoint write")
+			}
+			assertNoCleanCompletion(t, res, err, rec)
+			// 失败在推进退出处如实标注；async/on_exit 由 finalizeEvent 发 run_failed，
+			// sync 在第一个边界就报错返回。
+			if mode == "sync" {
+				if err == nil {
+					t.Fatalf("sync boundary write failure must surface as error, got %+v", res)
+				}
+				return
+			}
+			if res == nil || res.Status != RunFailed {
+				t.Fatalf("%s: want failed result, got %+v err=%v", mode, res, err)
+			}
+			if !rec.has(EvRunFailed) {
+				t.Fatalf("%s: no run_failed event: %v", mode, rec.types())
+			}
+		})
+	}
+}
+
+// TestTerminalSaveFailureBlocksCompletion：终态运行态写盘失败时不得宣布完成
+// （事件与返回值都必须可区分）；业务动作已发生（交付物文件仍在），但不声称成功。
+func TestTerminalSaveFailureBlocksCompletion(t *testing.T) {
+	for _, mode := range []string{"sync", "async", "on_exit"} {
+		t.Run(mode, func(t *testing.T) {
+			store := &faultStore{Store: FileStore{}, failTerminalSave: true}
+			rec := newEventRecorder()
+			eng := newTestEngineWithStore(t, store, rec)
+			res, err := eng.Run(context.Background(), Goal{Text: "快照"}, RunOptions{CheckpointMode: mode})
+			if err != nil {
+				t.Fatalf("terminal save failure must not abort the run as infra error: %v", err)
+			}
+			assertNoCleanCompletion(t, res, err, rec)
+			if res.Status != RunFailed {
+				t.Fatalf("%s: want failed, got %s", mode, res.Status)
+			}
+			if !rec.has(EvRunFailed) {
+				t.Fatalf("%s: no run_failed event: %v", mode, rec.types())
+			}
+			// 完成事件必须晚于/缺席于失败：事件序里终态只有 run_failed。
+			types := rec.types()
+			lastTerminal := ""
+			for _, ty := range types {
+				switch ty {
+				case EvRunFailed, EvRunCompleted, EvRunCancelled:
+					lastTerminal = ty
+				}
+			}
+			if lastTerminal != EvRunFailed {
+				t.Fatalf("%s: last terminal event = %q, want run_failed", mode, lastTerminal)
+			}
+			// 业务动作已发生且未被回滚：交付物文件确实写出来了（但运行不声称完成）。
+			if _, err := os.Stat(filepath.Join(res.WorkDir, "output.md")); err != nil {
+				t.Fatalf("deliverable written before terminal save should remain: %v", err)
+			}
+		})
+	}
+}
+
+// TestCheckpointFlushSucceedsInAllModes（对照）：故障注入关闭时三档都正常收口，
+// 出口快照确实落盘（避免上面的失败断言因「压根没写」而空过）。
+func TestCheckpointFlushSucceedsInAllModes(t *testing.T) {
+	for _, mode := range []string{"sync", "async", "on_exit"} {
+		t.Run(mode, func(t *testing.T) {
+			store := &faultStore{Store: FileStore{}}
+			rec := newEventRecorder()
+			eng := newTestEngineWithStore(t, store, rec)
+			res, err := eng.Run(context.Background(), Goal{Text: "快照"}, RunOptions{CheckpointMode: mode})
+			if err != nil || res.Status != RunCompleted {
+				t.Fatalf("%s: %v %+v", mode, err, res)
+			}
+			if !rec.has(EvRunCompleted) {
+				t.Fatalf("%s: no run_completed: %v", mode, rec.types())
+			}
+			latest, err := FileStore{}.LoadLatestCheckpoint(res.WorkDir)
+			if err != nil || latest == nil || latest.Superstep < 3 {
+				t.Fatalf("%s: exit snapshot not flushed: %v %+v", mode, err, latest)
+			}
+		})
+	}
+}

squad/engine.go

@@ -973,9 +973,18 @@ func (e *SquadEngine) manualPass(st *RunState, decision RespondDecision) (*RunRe
 	wf := st.Workflow()
 	wf.Status = string(status)
 	st.setWorkflow(wf)
-	_ = st.save()
+	// F11c：终态落盘先于完成事件——写不进去不得宣布完成（业务动作不回滚）。
+	if serr := st.save(); serr != nil {
+		msg := persistenceFailureText("save terminal run state", serr)
+		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			WorkDir: st.Meta.WorkDir, Quality: report, Error: msg, Summary: msg}, nil
+	}
 	manifest := buildManifest(st, "output.md", *report)
-	_ = writeJSON(filepath.Join(st.Meta.WorkDir, "manifest.json"), manifest)
+	if merr := writeJSON(filepath.Join(st.Meta.WorkDir, "manifest.json"), manifest); merr != nil {
+		msg := persistenceFailureText("write manifest", merr)
+		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			WorkDir: st.Meta.WorkDir, Quality: report, Error: msg, Summary: msg}, nil
+	}
 	e.deps.Sink.Emit(Event{Type: EvRunCompleted, RunID: st.Meta.ID, Text: string(status), At: time.Now()})
 	return &RunResult{RunID: st.Meta.ID, Status: status, Plan: planPtr(st),
 		Deliverable: "output.md", Quality: report, WorkDir: st.Meta.WorkDir,
@@ -989,10 +998,19 @@ func (e *SquadEngine) cancelRun(st *RunState, reason string) *RunResult {
 	wf := st.Workflow()
 	wf.Status = "cancelled"
 	st.setWorkflow(wf)
-	_ = st.save()
+	// 取消语义不变（运行确实被中止），但取消态落盘失败必须可见，不静默。
+	saveErr := st.save()
+	if saveErr != nil {
+		e.deps.Sink.Emit(Event{Type: EvPhase, RunID: st.Meta.ID,
+			Text: persistenceFailureText("save cancelled run state", saveErr), At: time.Now()})
+	}
 	e.deps.Sink.Emit(Event{Type: EvRunCancelled, RunID: st.Meta.ID, Text: reason, At: time.Now()})
-	return &RunResult{RunID: st.Meta.ID, Status: RunCancelled, Plan: planPtr(st),
+	out := &RunResult{RunID: st.Meta.ID, Status: RunCancelled, Plan: planPtr(st),
 		Summary: reason, WorkDir: st.Meta.WorkDir}
+	if saveErr != nil {
+		out.Error = persistenceFailureText("save cancelled run state", saveErr)
+	}
+	return out
 }
 
 // failStepResume 权限拒绝后的失败步骤处理（关键失败/非关键降级）。

squad/groupchat.go

@@ -339,7 +339,13 @@ func (e *SquadEngine) gcFinish(st *RunState, drv *flowDriver) *RunResult {
 	st.Meta.Status = string(RunCompleted)
 	now := time.Now()
 	st.Meta.CompletedAt = &now
-	_ = st.save()
+	// F11c：终态落盘先于返回 RunCompleted——写不进去就不能报完成。
+	if serr := st.save(); serr != nil {
+		st.mu.Unlock()
+		msg := persistenceFailureText("save terminal run state", serr)
+		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			WorkDir: st.Meta.WorkDir, Error: msg, Summary: msg}
+	}
 	st.mu.Unlock()
 	summary := fmt.Sprintf("group_chat：%d 轮发言，转录已收口", turns)
 	drv.emit(st, Event{Type: EvPhase, Text: summary})

squad/leader.go

@@ -33,7 +33,7 @@ func leaderExecRequest(drv *flowDriver, st *RunState, leader RoleSpec, workDir,
 		Budget:        budget,
 		OnEvent: func(ev ProcessEvent) {
 			drv.emit(st, Event{Type: EvDetail, StepID: stepID, Role: leader.Role, Text: ev.Text,
-				Data: marshalJSON(map[string]string{"process_type": ev.Type, "tool": ev.Tool, "detail": ev.Detail})})
+				Data: processEventData(ev)})
 		},
 	}
 }

squad/port.go

@@ -75,8 +75,27 @@ type ProcessEvent struct {
 	LastCompletionTokens int
 	CacheReadTokens      int
 	CacheWriteTokens     int
+	// CallID 是本次工具调用的身份（与 native exec.ProcessEvent.CallID 同语义）：
+	// tool_call_delta / tool_execution_start / tool_execution_end 携带同一 ID，
+	// 消费方按 ID 配对而不是按到达顺序——并行工具调用会交错到达。
+	// 空串 = 上游未填充（legacy 版本），不是「一次无名调用」。
+	CallID string `json:"call_id,omitempty"`
+	// Status 是 tool_execution_end 的真实结局，取值见 ToolStatus*；
+	// 空串 = legacy/未知，不据此判定失败（与既有行为一致）。
+	Status string `json:"status,omitempty"`
+	// ErrorReason 是 Status=error 时的真实失败原因原文（与 native 同语义）；
+	// 空串 = 无失败或上游未填充。
+	ErrorReason string `json:"error_reason,omitempty"`
 }
 
+// ProcessEvent.Status 的取值域（与 native 侧一致；空串 = legacy/未知）。
+const (
+	ToolStatusOK       = "ok"
+	ToolStatusError    = "error"
+	ToolStatusDenied   = "denied"
+	ToolStatusCanceled = "canceled"
+)
+
 // ExecResult 是一次步骤执行的终态。
 //
 // 语义契约（F05/U3b）：Success=true 才是经过验证的成功；仅在交付物齐全且

squad/process_event_test.go

@@ -0,0 +1,222 @@
+package squad
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+)
+
+const toolCallFailure = "grep: permission denied"
+
+// concurrentToolBackend 在一次执行里并发发两次同名工具调用（start + end）：
+// CallID 每次执行唯一、一成一败。用于验证过程帧经扇出后仍按 CallID 配对、
+// 结局不串（并行工具调用交错到达时不再靠到达顺序猜）。
+type concurrentToolBackend struct {
+	inner Executor
+
+	mu   sync.Mutex
+	exec int
+}
+
+func (b *concurrentToolBackend) Name() string { return "concurrent-tool" }
+
+func (b *concurrentToolBackend) Execute(ctx context.Context, req ExecRequest) (*ExecResult, error) {
+	b.mu.Lock()
+	b.exec++
+	round := b.exec
+	b.mu.Unlock()
+
+	if req.OnEvent != nil {
+		var wg sync.WaitGroup
+		for _, suffix := range []string{"ok", "err"} {
+			wg.Add(1)
+			go func(suffix string) {
+				defer wg.Done()
+				callID := fmt.Sprintf("call_%d_%s", round, suffix)
+				req.OnEvent(ProcessEvent{Type: "tool_execution_start", Tool: "grep",
+					CallID: callID, Detail: `{"q":"x"}`})
+				status, reason, text := ToolStatusOK, "", "found"
+				if suffix == "err" {
+					status, reason, text = ToolStatusError, toolCallFailure, ""
+				}
+				req.OnEvent(ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: callID,
+					Status: status, ErrorReason: reason, Text: text})
+			}(suffix)
+		}
+		wg.Wait() // 后端自己等待：并行工具调用在同一执行内收敛
+	}
+	if b.inner == nil {
+		return &ExecResult{Success: true}, nil
+	}
+	return b.inner.Execute(ctx, req)
+}
+
+// detailFrame 是 detail 事件 Data 的读取视图。
+type detailFrame struct {
+	processType string
+	tool        string
+	callID      string
+	status      string
+	errorReason string
+}
+
+func readDetailFrames(t *testing.T, events []Event) []detailFrame {
+	t.Helper()
+	var out []detailFrame
+	for _, ev := range events {
+		if ev.Type != EvDetail {
+			continue
+		}
+		var data map[string]string
+		if err := json.Unmarshal(ev.Data, &data); err != nil {
+			t.Fatalf("detail data must stay a JSON object: %v (%s)", err, ev.Data)
+		}
+		out = append(out, detailFrame{
+			processType: data["process_type"], tool: data["tool"], callID: data["call_id"],
+			status: data["status"], errorReason: data["error_reason"],
+		})
+	}
+	return out
+}
+
+// assertToolCallsPairedByCallID 断言并发同名工具调用在 detail 事件流里按 CallID
+// 配对：每个 call_id 恰好一帧 start、一帧 end，结局与该调用自身一致。
+func assertToolCallsPairedByCallID(t *testing.T, frames []detailFrame) {
+	t.Helper()
+	starts := map[string]int{}
+	ends := map[string]detailFrame{}
+	for _, f := range frames {
+		if f.callID == "" {
+			t.Fatalf("frame lost its CallID: %+v", f)
+		}
+		if f.tool != "grep" {
+			t.Fatalf("frame tool = %q, want grep", f.tool)
+		}
+		switch f.processType {
+		case "tool_execution_start":
+			starts[f.callID]++
+		case "tool_execution_end":
+			if _, dup := ends[f.callID]; dup {
+				t.Fatalf("duplicate end frame for %s: %+v", f.callID, f)
+			}
+			ends[f.callID] = f
+		}
+	}
+	if len(ends) == 0 {
+		t.Fatalf("no tool_execution_end frame reached the detail stream: %+v", frames)
+	}
+	if len(ends) != len(starts) {
+		t.Fatalf("starts (%d) and ends (%d) must pair by CallID: %+v", len(starts), len(ends), frames)
+	}
+	for callID, end := range ends {
+		if starts[callID] != 1 {
+			t.Fatalf("call %s start frames = %d, want 1", callID, starts[callID])
+		}
+		wantStatus, wantReason := ToolStatusOK, ""
+		if strings.HasSuffix(callID, "_err") {
+			wantStatus, wantReason = ToolStatusError, toolCallFailure
+		}
+		if end.status != wantStatus || end.errorReason != wantReason {
+			t.Fatalf("call %s outcome = %s/%q, want %s/%q（结局必须属于本次调用）",
+				callID, end.status, end.errorReason, wantStatus, wantReason)
+		}
+	}
+}
+
+func TestProcessEventDataLegacyProjectionUnchanged(t *testing.T) {
+	// legacy 上游（无 CallID/Status/ErrorReason）的投影必须逐字节不变。
+	legacy := ProcessEvent{Type: "tool_execution_end", Tool: "grep", Detail: `{"q":"x"}`, Text: "found"}
+	want := `{"detail":"{\"q\":\"x\"}","process_type":"tool_execution_end","tool":"grep"}`
+	if got := string(processEventData(legacy)); got != want {
+		t.Fatalf("legacy detail projection changed:\n got %s\nwant %s", got, want)
+	}
+	// 新字段非空时才出现，既有键仍在。
+	withID := legacy
+	withID.CallID, withID.Status, withID.ErrorReason = "call_1", ToolStatusOK, ""
+	var data map[string]string
+	if err := json.Unmarshal(processEventData(withID), &data); err != nil {
+		t.Fatal(err)
+	}
+	if data["call_id"] != "call_1" || data["status"] != ToolStatusOK {
+		t.Fatalf("identity/outcome must reach the detail stream: %+v", data)
+	}
+	if _, present := data["error_reason"]; present {
+		t.Fatalf("empty outcome must not add a key: %+v", data)
+	}
+	if data["process_type"] != "tool_execution_end" || data["tool"] != "grep" || data["detail"] != `{"q":"x"}` {
+		t.Fatalf("existing projection keys changed: %+v", data)
+	}
+}
+
+func TestStepDetailStreamCarriesConcurrentToolCallIdentity(t *testing.T) {
+	script := map[string]map[string]string{
+		"01_researcher": {"notes.md": "# 笔记\n内容"},
+		"02_writer":     {"output.md": goodReport},
+		"03_reviewer":   {"output.md": goodReport},
+	}
+	dir := t.TempDir()
+	reg := NewRoleRegistry()
+	for _, spec := range []RoleSpec{
+		{Role: "researcher", Name: "调研", Backend: "concurrent-tool"},
+		{Role: "writer", Name: "撰写", Backend: "concurrent-tool"},
+		{Role: "reviewer", Name: "审核", Backend: "concurrent-tool"},
+	} {
+		if err := reg.Register(spec); err != nil {
+			t.Fatal(err)
+		}
+	}
+	backends := NewBackendResolver()
+	backends.RegisterInstance("concurrent-tool", &concurrentToolBackend{
+		inner: &scriptBackend{script: script},
+	})
+	collector := &sliceSink{}
+	engine, err := NewEngine(Options{DataDir: dir, Registry: reg, Planner: mockScriptPlanner(baseSteps()...),
+		Backends: backends, Sink: collector})
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err := engine.Run(context.Background(), Goal{Text: "生成测试报告"}, RunOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != RunCompleted {
+		t.Fatalf("want completed, got %s (err=%s)", res.Status, res.Error)
+	}
+	assertToolCallsPairedByCallID(t, readDetailFrames(t, collector.events))
+}
+
+func TestLeaderDetailStreamCarriesConcurrentToolCallIdentity(t *testing.T) {
+	inner := &leaderScriptBackend{
+		decisions: []string{delegateJSON("writer", "撰写"), closeJSON},
+		script:    map[string]map[string]string{"01_writer": {"output.md": goodReport}},
+	}
+	dir := t.TempDir()
+	reg := NewRoleRegistry()
+	for _, spec := range []RoleSpec{
+		{Role: "leader", Name: "主管", Backend: "concurrent-tool"},
+		{Role: "writer", Name: "撰写", Backend: "concurrent-tool"},
+	} {
+		if err := reg.Register(spec); err != nil {
+			t.Fatal(err)
+		}
+	}
+	backends := NewBackendResolver()
+	backends.RegisterInstance("concurrent-tool", &concurrentToolBackend{inner: inner})
+	collector := &sliceSink{}
+	engine, err := NewEngine(Options{DataDir: dir, Registry: reg, Backends: backends, Sink: collector})
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err := engine.Run(context.Background(), Goal{Text: "产出简报"},
+		RunOptions{Mode: "leader", LeaderRole: "leader"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != RunCompleted {
+		t.Fatalf("want completed, got %s err=%s", res.Status, res.Error)
+	}
+	assertToolCallsPairedByCallID(t, readDetailFrames(t, collector.events))
+}

squad/progress.go

@@ -136,3 +136,22 @@ func (s *FanoutSink) Snapshot() ([]Event, int) {
 type NopSink struct{}
 
 func (NopSink) Emit(Event) {}
+
+// processEventData 把过程帧投影成 detail 事件的 Data（JSON 对象）。
+//
+// 工具调用身份与真实结局（call_id / status / error_reason）非空时随帧透传，
+// 使 detail 事件流能按 CallID 配对交错的并行工具调用，而不是按到达顺序猜。
+// 三者全空（legacy 上游未填充）时输出与既有投影逐字节一致，不引入新键。
+func processEventData(ev ProcessEvent) json.RawMessage {
+	data := map[string]string{"process_type": ev.Type, "tool": ev.Tool, "detail": ev.Detail}
+	if ev.CallID != "" {
+		data["call_id"] = ev.CallID
+	}
+	if ev.Status != "" {
+		data["status"] = ev.Status
+	}
+	if ev.ErrorReason != "" {
+		data["error_reason"] = ev.ErrorReason
+	}
+	return marshalJSON(data)
+}

squad/review.go

@@ -326,7 +326,13 @@ func (e *SquadEngine) runReviewDriver(ctx context.Context, st *RunState) (*RunRe
 	st.Meta.Status = string(status)
 	now := time.Now()
 	st.Meta.CompletedAt = &now
-	_ = st.save()
+	// F11c：终态落盘先于返回 RunCompleted——写不进去就不能报完成。
+	if serr := st.save(); serr != nil {
+		st.mu.Unlock()
+		msg := persistenceFailureText("save terminal run state", serr)
+		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			WorkDir: st.Meta.WorkDir, Error: msg, Summary: msg}, nil
+	}
 	st.mu.Unlock()
 	summary := fmt.Sprintf("review：保留 %d 条 / 丢弃 %d 条未验证", len(verified), dropped)
 	if finderFailed > 0 {

squad/runstate.go

@@ -96,6 +96,11 @@ func (st *RunState) persist() error {
 			return err
 		}
 	}
+	// steering.json 与其他 RunState 字段同权：SaveRun 后的 LoadRun 必须能读回
+	// 未消费的补充指令（与 SQLiteStore 的 steering 列同一契约）。
+	if err := writeJSON(filepath.Join(st.Meta.WorkDir, "steering.json"), st.Steering); err != nil {
+		return err
+	}
 	return writeJSON(filepath.Join(st.Meta.WorkDir, "runstate.json"), diskRunState{
 		Approved:              st.Approved,
 		RepairBudgetRemaining: st.RepairBudgetRemaining,

squad/runstore.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"time"
 )
 
@@ -140,6 +141,10 @@ type checkpointWriter struct {
 	store Store
 	ch    chan Checkpoint
 	done  chan error
+	// flushOnce/flushErr 让出口 flush 幂等：正常收口路径先 flush 一次（失败必须
+	// 先于任何完成事件可见），defer 再兜底 flush 时不会二次写、也不会重复报错。
+	flushOnce sync.Once
+	flushErr  error
 }
 
 func newCheckpointWriter(mode CheckpointMode, store Store) *checkpointWriter {
@@ -189,7 +194,13 @@ func (w *checkpointWriter) boundary(wsDir string, cp Checkpoint) error {
 }
 
 // flush 在推进退出前调用：async 落完并报告首个错误；on_exit 写最终快照。
+// 幂等：重复调用返回首次结果（async 的 chan 只关一次）。
 func (w *checkpointWriter) flush(wsDir string, cp Checkpoint) error {
+	w.flushOnce.Do(func() { w.flushErr = w.flushOnceImpl(wsDir, cp) })
+	return w.flushErr
+}
+
+func (w *checkpointWriter) flushOnceImpl(wsDir string, cp Checkpoint) error {
 	cp.wsDir = wsDir
 	switch w.mode {
 	case CheckpointAsync:
@@ -202,6 +213,15 @@ func (w *checkpointWriter) flush(wsDir string, cp Checkpoint) error {
 	}
 }
 
+// persistenceFailurePrefix 标记落盘失败。这类失败必须让运行以可区分的面目收场：
+// 业务动作（步骤执行、交付物写出）已发生，不回滚、不声称完成。
+const persistenceFailurePrefix = "persistence failure"
+
+// persistenceFailureText 组装落盘失败的可读说明（stage = 落盘环节）。
+func persistenceFailureText(stage string, err error) string {
+	return persistenceFailurePrefix + ": " + stage + ": " + err.Error()
+}
+
 // validateGraphContinuity 恢复连续性校验（Q1 WP4；WP1 的图版本裁定）。
 // 最新检查点存在且其图指纹与当前 Steps 重编译不一致时拒绝恢复——
 // 崩溃期间图形态被改过，继续推进会把已完成的结果接到错误的图上。

squad/workflow.go

@@ -42,16 +42,41 @@ func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult,
 	cpw := newCheckpointWriter(CheckpointMode(st.Meta.Metadata["checkpoint_mode"]), st.storeOr())
 	// 推进退出前 flush：async 落完、on_exit 写最终快照（Q1 WP4 三档的出口语义）。
 	var lastGraph *graph
-	defer func() {
+	// exitSnapshot 组装出口快照（尚未编译图时现编译）。
+	exitSnapshot := func() (Checkpoint, bool) {
 		if lastGraph == nil {
-			if g, gerr := compileGraph(st.Plan.Workflow.Steps); gerr == nil {
+			g, gerr := compileGraph(st.Plan.Workflow.Steps)
-				lastGraph = g
+			if gerr != nil {
+				return Checkpoint{}, false
+			}
+			lastGraph = g
+		}
+		return snapshotOf(superNo, lastGraph, st.Plan.Workflow.Steps), true
+	}
+	// 兜底出口：任何非正常退出路径（挂起、图编译失败、infra 错误）也在此 flush。
+	// flush 幂等——收口路径已提前 flush 过，不会二次写。
+	defer func() {
+		if cp, ok := exitSnapshot(); ok {
+			if ferr := cpw.flush(st.Meta.WorkDir, cp); ferr != nil {
+				// F11c：出口快照落盘失败不得静默——翻成可区分的失败。
+				res, gate, err = d.persistenceFailure(st, res, gate, err, "exit checkpoint flush", ferr)
 			}
 		}
-		if lastGraph != nil {
-			_ = cpw.flush(st.Meta.WorkDir, snapshotOf(superNo, lastGraph, st.Plan.Workflow.Steps))
-		}
 	}()
+	// flushExitBeforeFinalize 在收口前落出口快照（async/on_exit 的最终落盘），
+	// 失败立即交出可区分的失败：落盘失败必须先于任何完成事件/RunCompleted 返回。
+	// 返回的 ok=false 表示快照已落盘（或本次无需落盘），调用方继续正常收口。
+	flushExitBeforeFinalize := func() (*RunResult, *HitlGate, error, bool) {
+		cp, ok := exitSnapshot()
+		if !ok {
+			return nil, nil, nil, false
+		}
+		if ferr := cpw.flush(st.Meta.WorkDir, cp); ferr != nil {
+			out, g2, e2 := d.persistenceFailure(st, nil, nil, nil, "exit checkpoint flush", ferr)
+			return out, g2, e2, true
+		}
+		return nil, nil, nil, false
+	}
 	for {
 		if !runDeadline.IsZero() && time.Now().After(runDeadline) {
 			d.emit(st, Event{Type: EvPhase, Text: "运行截止时刻已到，未完步骤按 skipped 收口"})
@@ -64,6 +89,9 @@ func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult,
 			}
 			_ = st.save()
 			st.mu.Unlock()
+			if out, g2, e2, failed := flushExitBeforeFinalize(); failed {
+				return out, g2, e2
+			}
 			return d.finalize(ctx, st)
 		}
 		g, gerr := compileGraph(st.Plan.Workflow.Steps)
@@ -152,13 +180,20 @@ func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult,
 			}
 		}
 
-		// superstep 边界快照（sync 立即 / async 入队；on_exit 由 defer 的 flush 写）。
+		// superstep 边界快照（sync 立即 / async 入队；on_exit 由出口 flush 写）。
 		lastGraph = g
 		if err := cpw.boundary(st.Meta.WorkDir, snapshotOf(superNo, g, st.Plan.Workflow.Steps)); err != nil {
-			return nil, nil, err
+			// F11c：边界快照写盘失败已经是可见错误，补上落盘失败标记以便识别。
+			return nil, nil, fmt.Errorf("%s: %w", persistenceFailurePrefix, err)
 		}
 	}
 
+	// 全部步骤已收口：先落出口快照（async/on_exit 的最终落盘），失败必须先于收口
+	// 可见——不得先宣布完成、再报落盘失败。
+	if out, g2, e2, failed := flushExitBeforeFinalize(); failed {
+		return out, g2, e2
+	}
+
 	return d.finalize(ctx, st)
 }
 
@@ -394,7 +429,7 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru
 			ToolProfile:  orDefault(role.ToolProfile, ToolProfileWorkspace),
 			OnEvent: func(ev ProcessEvent) {
 				d.emit(st, Event{Type: EvDetail, StepID: step.ID, Role: step.Role, Text: ev.Text,
-					Data: marshalJSON(map[string]string{"process_type": ev.Type, "tool": ev.Tool, "detail": ev.Detail})})
+					Data: processEventData(ev)})
 			},
 			// RoleSpec v2 贯通：工具白名单 / 技能 / 标识（原为死字段或缺失）。
 			ToolAllowlist: role.Tools,
@@ -955,10 +990,22 @@ func (d *flowDriver) finalize(ctx context.Context, st *RunState) (*RunResult, *H
 	st.setWorkflow(wf)
 	st.Meta.Status = string(status)
 	st.Meta.CompletedAt = &now
-	_ = st.save()
+	// F11c：终态落盘先于完成事件。写不进去就不能宣布完成——
+	// 业务动作已发生，不回滚，但运行以可区分的失败面世。
+	if serr := st.save(); serr != nil {
+		msg := persistenceFailureText("save terminal run state", serr)
+		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			Quality: report, WorkDir: st.Meta.WorkDir, Usage: st.usageSnapshot(),
+			Error: msg, Summary: msg}, nil, nil
+	}
 
 	manifest := buildManifest(st, deliverable, *report)
-	_ = writeJSON(filepath.Join(st.Meta.WorkDir, "manifest.json"), manifest)
+	if merr := writeJSON(filepath.Join(st.Meta.WorkDir, "manifest.json"), manifest); merr != nil {
+		msg := persistenceFailureText("write manifest", merr)
+		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			Quality: report, WorkDir: st.Meta.WorkDir, Usage: st.usageSnapshot(),
+			Error: msg, Summary: msg}, nil, nil
+	}
 
 	d.emit(st, Event{Type: EvRunCompleted, Text: string(status),
 		Data: marshalJSON(map[string]any{"deliverable": deliverable, "score": report.Score})})
@@ -1078,6 +1125,41 @@ func (d *flowDriver) emitHuman(st *RunState, req *HumanRequest) {
 	d.emit(st, Event{Type: EvHumanInputRequired, Text: req.Question, Data: marshalJSON(req)})
 }
 
+// persistenceFailure 把一次落盘失败折算成可见的失败（F11c）：
+//   - 已有 infra 错误 → 附加到错误里返回，不吞掉原错误；
+//   - 挂起中 → 挂起语义不变（运行未收口），但结果与事件流如实标注；
+//   - 其余（收口完成/无结果）→ 结果翻成 RunFailed，绝不再以 completed 面世。
+//
+// 业务动作（步骤执行、交付物写出）已发生，不回滚，也不声称干净完成。
+func (d *flowDriver) persistenceFailure(st *RunState, res *RunResult, gate *HitlGate, prev error, stage string, cause error) (*RunResult, *HitlGate, error) {
+	msg := persistenceFailureText(stage, cause)
+	if prev != nil {
+		return res, gate, fmt.Errorf("%w（同时发生 %s）", prev, msg)
+	}
+	if res == nil {
+		res = &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
+			WorkDir: st.Meta.WorkDir, Summary: msg}
+	}
+	if !strings.Contains(res.Error, msg) {
+		// 同一次失败会经出口 flush 与兜底 flush 各报一次，只记一遍。
+		if res.Error != "" {
+			msg = res.Error + "；" + msg
+		}
+		res.Error = msg
+	}
+	if gate != nil {
+		// 挂起路径不发终态事件（finalizeEvent 只处理 completed/degraded/failed），
+		// 所以在这里补一条可观察的标注事件。
+		d.emit(st, Event{Type: EvPhase, Text: msg})
+		return res, gate, nil
+	}
+	switch res.Status {
+	case RunCompleted, RunDegraded:
+		res.Status = RunFailed
+	}
+	return res, nil, nil
+}
+
 func summarize(st *RunState, report QualityReport, status RunStatus) string {
 	var b strings.Builder
 	fmt.Fprintf(&b, "运行 %s：%s。步骤 %d 个（完成 %d/跳过 %d/失败 %d）。质检得分 %.0f。",