diff --git a/internal/capability/mcp/discovery_test.go b/internal/capability/mcp/discovery_test.go index f68dd9c..cc632f1 100644 --- a/internal/capability/mcp/discovery_test.go +++ b/internal/capability/mcp/discovery_test.go @@ -249,18 +249,44 @@ func TestRegistry_Unregister_RemovesTools(t *testing.T) { } } +const ( + // methodSubscriptionsAck 是 SEP-2575 的订阅确认通知方法名（SDK 未导出该常量）。 + methodSubscriptionsAck = "notifications/subscriptions/acknowledged" + // methodToolListChanged 是工具清单变更通知方法名（SDK 未导出该常量）。 + methodToolListChanged = "notifications/tools/list_changed" + // sep2575ProtocolVersion 是引入 subscriptions/listen 订阅流的协议版本：自它起 + // 服务器只向**已登记订阅**的会话投递 list_changed 通知（更早的协议版本对全部 + // 会话广播，不存在下面这个窗口）。 + sep2575ProtocolVersion = "2026-07-28" + // notificationWaitBudget 是等上述事件的兜底时限：只用于避免测试挂死， + // 事件本身才是断言对象。 + notificationWaitBudget = 5 * time.Second +) + // TestRegistry_ToolListChangedRefreshes 服务器运行中加工具 → 通知 → registry 增量刷新， // 同时 schema 缓存按服务器整表失效（design/09 §1.1）。 +// +// 确定性要点：现代协议（≥ sep2575ProtocolVersion）下服务器**只**把 tools/list_changed +// 投给已登记 subscriptions/listen 订阅的会话，而客户端发订阅是 fire-and-forget +// （Connect 返回 ≠ 服务器已登记），通知本身又是「变更后 10ms 发一次、丢了不补」—— +// 订阅登记之前触发的变更会被永久静默丢弃。旧写法（改工具 → 轮询 5s）等于拿这个竞态 +// 与机器负载对赌，输的那次等多久都等不到 mul。现在先等服务器回订阅确认（服务器先登记 +// 订阅、再回确认）再触发变更，把「通知会不会来」变回确定的事。 func TestRegistry_ToolListChangedRefreshes(t *testing.T) { t.Parallel() srv := newFakeServer() reg := registry.New() - tr := startFakeServer(t, srv) - r := NewRegistry(reg, MCPConfig{Name: "calc"}, WithTransport(tr)) + sn := &sniffTransport{inner: startFakeServer(t, srv)} + r := NewRegistry(reg, MCPConfig{Name: "calc"}, WithTransport(sn)) if err := r.Register(context.Background()); err != nil { t.Fatalf("Register: %v", err) } + if v := r.mcp.ServerInfo().ProtocolVersion; v >= sep2575ProtocolVersion { + waitFor(t, notificationWaitBudget, "服务器回订阅确认 "+methodSubscriptionsAck, + func() bool { return sn.received(methodSubscriptionsAck) }) + } + _, _ = reg.Get("echo") // 让缓存先有命中记录，便于确认失效计数 srv.AddTool(&mcpsdk.Tool{ Name: "mul", @@ -268,10 +294,14 @@ func TestRegistry_ToolListChangedRefreshes(t *testing.T) { InputSchema: addToolSpec().InputSchema, }, addHandler) - waitFor(t, 5*time.Second, "registry 出现新工具 mul", func() bool { + waitFor(t, notificationWaitBudget, "registry 出现新工具 mul", func() bool { _, err := reg.Get("mul") return err == nil }) + // 新工具必须由变更通知驱动刷进来，而不是别的路径顺手带进来的。 + if !sn.received(methodToolListChanged) { + t.Fatalf("registry 出现了 mul，但客户端没收到 %s", methodToolListChanged) + } if _, err := reg.Get("echo"); err != nil { t.Fatalf("刷新后 echo 丢失：%v", err) } diff --git a/internal/capability/mcp/keepalive_test.go b/internal/capability/mcp/keepalive_test.go index 4ce7247..eb633f7 100644 --- a/internal/capability/mcp/keepalive_test.go +++ b/internal/capability/mcp/keepalive_test.go @@ -51,12 +51,14 @@ func TestClient_NoKeepAlive_NoPings(t *testing.T) { } } -// sniffTransport 包一层 SDK 传输，记录客户端发出的 JSON-RPC 请求方法， -// 用于在线路层断言客户端确实发了某个请求。 +// sniffTransport 包一层 SDK 传输，在线路层记录两个方向的 JSON-RPC 请求方法： +// seen 是客户端发出的（断言客户端确实发了某个请求），fromServer 是服务器发来的 +// （断言服务器确实回了某条通知，例如 SEP-2575 的订阅确认）。 type sniffTransport struct { - inner mcpsdk.Transport - mu sync.Mutex - seen []string + inner mcpsdk.Transport + mu sync.Mutex + seen []string + fromServer []string } // Connect 实现 mcpsdk.Transport。 @@ -94,6 +96,25 @@ func (t *sniffTransport) record(method string) { t.seen = append(t.seen, method) } +// received 报告服务器是否发来过某个方法。 +func (t *sniffTransport) received(method string) bool { + t.mu.Lock() + defer t.mu.Unlock() + for _, m := range t.fromServer { + if m == method { + return true + } + } + return false +} + +// recordReceived 记录一个服务器发来的方法名。 +func (t *sniffTransport) recordReceived(method string) { + t.mu.Lock() + defer t.mu.Unlock() + t.fromServer = append(t.fromServer, method) +} + // sniffConn 拦截写入的请求消息，记录方法名后交给内层连接。 type sniffConn struct { mcpsdk.Connection @@ -107,3 +128,17 @@ func (c *sniffConn) Write(ctx context.Context, msg sdkjsonrpc.Message) error { } return c.Connection.Write(ctx, msg) } + +// Read 记录服务器发来的请求方法后转发。通知也是不带 ID 的请求帧，因此 +// 订阅确认、tools/list_changed 这类服务器通知都在这里现形。 +func (c *sniffConn) Read(ctx context.Context) (sdkjsonrpc.Message, error) { + msg, err := c.Connection.Read(ctx) + if err != nil { + // 出错时内层消息未被上层使用（jsonrpc2 读循环遇错即退出），原样透传错误。 + return nil, err + } + if req, ok := msg.(*sdkjsonrpc.Request); ok { + c.owner.recordReceived(req.Method) + } + return msg, nil +} diff --git a/internal/runtime/multimodal/vision/cache_test.go b/internal/runtime/multimodal/vision/cache_test.go index f430506..e7eac80 100644 --- a/internal/runtime/multimodal/vision/cache_test.go +++ b/internal/runtime/multimodal/vision/cache_test.go @@ -87,11 +87,45 @@ func TestImageCache_SecondResolveFetchesOnce(t *testing.T) { } } +// testClock 是可手动推进的测试时钟：注入 imageCache.now 之后，TTL 判据只随 +// 测试显式推进而前进，用例不再依赖「两次解析之间的真实耗时」这种墙钟前提。 +type testClock struct { + mu sync.Mutex + now time.Time +} + +// newTestClock 返回一个从固定时刻起步的测试时钟。 +func newTestClock() *testClock { + return &testClock{now: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)} +} + +// Now 返回当前测试时刻；签名与 time.Now 一致，可直接赋给 imageCache.now。 +func (c *testClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +// Advance 把测试时刻前进 d。 +func (c *testClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(d) +} + // TestImageCache_RefetchesAfterTTLWhenContentChanges 钉住失效策略：TTL 内 // 返回缓存（旧内容，这是缓存换取的确定性代价），TTL 到期后重新取回，上游 // 换了内容也能自愈 —— 新字节是新内容键，不会被旧条目命中。 +// +// 时间轴由**注入的测试时钟**驱动：只有测试显式推进时钟时 TTL 才会到期， +// 因此「TTL 内命中」不再取决于「两次解析之间的真实耗时是否小于 TTL」—— +// 取回走真实 HTTP，并行负载下这个前提会失效（旧写法 ttl=20ms 就是把结论 +// 交给机器负载决定）。 func TestImageCache_RefetchesAfterTTLWhenContentChanges(t *testing.T) { t.Parallel() + // testCacheTTL 的取值本身不重要，只需大于零：时间只在测试显式推进时前进。 + const testCacheTTL = 20 * time.Millisecond + original := pngBytes(t) changed := pngBytesAlt(t) var mu sync.Mutex @@ -107,8 +141,10 @@ func TestImageCache_RefetchesAfterTTLWhenContentChanges(t *testing.T) { })) t.Cleanup(srv.Close) + clock := newTestClock() res := newMainModelResolver() - res.cache.ttl = 20 * time.Millisecond + res.cache.ttl = testCacheTTL + res.cache.now = clock.Now src := srv.URL + "/cat.png" first, err := resolveMainImage(res, src) @@ -123,7 +159,8 @@ func TestImageCache_RefetchesAfterTTLWhenContentChanges(t *testing.T) { payload = changed mu.Unlock() - // TTL 内：仍返回旧图，且没有再取回。 + // TTL 内：仍返回旧图，且没有再取回。时钟没有前进，所以这一段的结论由 + // 缓存自身的判据决定，与机器快慢无关。 if got, err := resolveMainImage(res, src); err != nil || got != first { t.Fatalf("TTL 内 = (%q, %v)，want 旧缓存 %q", got, err, first) } @@ -131,7 +168,9 @@ func TestImageCache_RefetchesAfterTTLWhenContentChanges(t *testing.T) { t.Fatalf("TTL 内上游被取回 %d 次，want 1", got) } - time.Sleep(60 * time.Millisecond) + // 把注入的时钟推进到 TTL 边界：get 用 !now.Before(expiresAt) 判定过期， + // 到达边界即为过期。取代旧写法的 time.Sleep(60ms) 墙钟等待。 + clock.Advance(testCacheTTL) third, err := resolveMainImage(res, src) if err != nil { diff --git a/internal/services/gateway/voice_turn_test.go b/internal/services/gateway/voice_turn_test.go index 61f9313..c7fe4af 100644 --- a/internal/services/gateway/voice_turn_test.go +++ b/internal/services/gateway/voice_turn_test.go @@ -35,9 +35,14 @@ import ( // waitFrame 排空连接上的帧直到出现给定类型（语音回合的收尾帧会排在若干状态 // 信号广播之后），超时即失败。 +// +// 兜底期限取 15s：**断言的是「帧确实到达」，不是「3 秒内到达」** —— 语音回合 +// 要经编排器 → 假 runner（带门控）→ TTS 假实现 → 定向回连接，在整树并行 +// （多包同时跑）或宿主负载高时 3s 会被误判成超时（实测：单跑 -count=20 全绿、 +// 只在整树并行下偶红）。放宽兜底不放宽断言：帧没到照样 Fatal。 func waitFrame(t *testing.T, c *wsConn, typ string) wsEvent { t.Helper() - deadline := time.After(3 * time.Second) + deadline := time.After(15 * time.Second) for { select { case b := <-c.send: