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,6 +185,9 @@ 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，取零值透传，绝不伪造。
@@ -309,23 +313,40 @@ func optionalStringField(rv reflect.Value, name string) 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

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"sync"
 	"testing"
+	"time"
 
 	"agent-native/exec"
 	"agent-squad/squad"
@@ -52,7 +53,8 @@ func TestBridgeMapsLegacyNativeEventWithoutFabricatingFields(t *testing.T) {
 }
 
 func TestProbeNativeEventExtrasReadsNewFieldsWhenNativeHasThem(t *testing.T) {
-	// U11a 落地前：native 帧没有这三个字段 → 零值（legacy），不伪造。
+	// 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)
 	}
@@ -92,6 +94,72 @@ func TestProbeNativeEventExtrasReadsNewFieldsWhenNativeHasThem(t *testing.T) {
 	}
 }
 
+// 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{}
@@ -162,6 +230,9 @@ func assertPaired(t *testing.T, tag string, frames []squad.ProcessEvent) {
 	}
 }
 
+// TestEventTapKeepsConcurrentToolCallIdentityAndOutcome 在真实并发帧源下验证三件事：
+// 身份与结局不串（严格按 CallID 配对）、帧不多发不少发（每个调用一 start 一 end）、
+// 所有下游观察到同一串到达序（并发帧源下到达序本身不确定，但必须唯一且一致）。
 func TestEventTapKeepsConcurrentToolCallIdentityAndOutcome(t *testing.T) {
 	var tapped, requested []squad.ProcessEvent
 	var mu sync.Mutex
@@ -183,10 +254,164 @@ func TestEventTapKeepsConcurrentToolCallIdentityAndOutcome(t *testing.T) {
 	assertPaired(t, "tap", tapped)
 	assertPaired(t, "request", requested)
 	if !reflect.DeepEqual(tapped, requested) {
-		t.Fatalf("tap must forward whole frames unchanged:\n%+v\n%+v", 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