Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b6d87638cc | ||
|
|
9edb6ba865 |
@@ -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 = func(ev squad.ProcessEvent) {
|
||||
req.OnEvent = b.hook(req.OnEvent)
|
||||
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。
|
||||
|
||||
@@ -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
|
||||
|
||||
Vendored
+1
-1
Submodule deps/agent-native updated: ac91b8436e...f2fb9ffa6d
Reference in New Issue
Block a user