From 13a38dd3a524db05fd27d4e46394beac092d3bca Mon Sep 17 00:00:00 2001 From: sam Date: Thu, 29 Oct 2026 21:07:39 +0800 Subject: [PATCH 1/2] fix: detect unresponsive runtime connections and log transport failures --- .../internal/cli/connect_cleanup_test.go | 77 ++++++++++++-- apps/daemon/internal/transport/liveness.go | 19 ++++ .../internal/transport/liveness_test.go | 100 ++++++++++++++++++ apps/daemon/internal/transport/ws.go | 26 +++++ docs/runtime-protocol.md | 1 + docs/zh/runtime-protocol.md | 3 +- .../core/internal/runtimegateway/session.go | 8 +- .../runtimegateway/transport_failure.go | 38 +++++++ .../runtimegateway/transport_failure_test.go | 57 ++++++++++ 9 files changed, 317 insertions(+), 12 deletions(-) create mode 100644 apps/daemon/internal/transport/liveness.go create mode 100644 apps/daemon/internal/transport/liveness_test.go create mode 100644 services/core/internal/runtimegateway/transport_failure.go create mode 100644 services/core/internal/runtimegateway/transport_failure_test.go diff --git a/apps/daemon/internal/cli/connect_cleanup_test.go b/apps/daemon/internal/cli/connect_cleanup_test.go index b0df88163..e5122a48c 100644 --- a/apps/daemon/internal/cli/connect_cleanup_test.go +++ b/apps/daemon/internal/cli/connect_cleanup_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "os" "strings" + "sync" "sync/atomic" "testing" "time" @@ -68,13 +69,40 @@ func TestReconnectWaitsForConfirmedRouterCleanup(t *testing.T) { } type cleanupExecutor struct { - closes atomic.Int32 - retry chan struct{} - confirm chan struct{} + started chan struct{} + cancelled chan struct{} + closes atomic.Int32 + retry chan struct{} + confirm chan struct{} } -func (e *cleanupExecutor) StartTurn(context.Context, string, proto.MessageInput, chan<- proto.Envelope) (agent.Turn, error) { - return nil, errors.New("unexpected Turn") +func (e *cleanupExecutor) StartTurn(_ context.Context, _ string, _ proto.MessageInput, out chan<- proto.Envelope) (agent.Turn, error) { + if e.started == nil { + return nil, errors.New("unexpected Turn") + } + close(e.started) + return &cleanupActiveTurn{out: out, cancelled: e.cancelled}, nil +} + +type cleanupActiveTurn struct { + agent.Turn + out chan<- proto.Envelope + cancelled chan struct{} + once sync.Once +} + +func (t *cleanupActiveTurn) Cancel(context.Context) error { + t.once.Do(func() { close(t.cancelled); close(t.out) }) + return nil +} +func (t *cleanupActiveTurn) CancellationOutcome() proto.DonePayload { return proto.DonePayload{} } +func (t *cleanupActiveTurn) AwaitSettlement(ctx context.Context) (agent.TurnSettlement, error) { + select { + case <-t.cancelled: + return agent.TurnSettlement{Reusable: true}, nil + case <-ctx.Done(): + return agent.TurnSettlement{}, ctx.Err() + } } func (e *cleanupExecutor) Close(context.Context) error { if e.closes.Add(1) == 1 { @@ -91,11 +119,15 @@ func TestDisconnectedPumpRetainsExactExecutorUntilCleanup(t *testing.T) { if suspend { name = "suspension-enabled" } - t.Run(name, func(t *testing.T) { testDisconnectedPumpCleanup(t, suspend) }) + t.Run(name, func(t *testing.T) { testDisconnectedPumpCleanup(t, suspend, false) }) } } -func testDisconnectedPumpCleanup(t *testing.T, suspend bool) { +func TestSilentConnectionStopsOwnerBeforeReconnect(t *testing.T) { + testDisconnectedPumpCleanup(t, true, true) +} + +func testDisconnectedPumpCleanup(t *testing.T, suspend, silent bool) { peers := make(chan *websocket.Conn, 1) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { peer, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) @@ -113,6 +145,10 @@ func testDisconnectedPumpCleanup(t *testing.T, suspend bool) { }) } owner := &cleanupExecutor{retry: make(chan struct{}), confirm: make(chan struct{})} + if silent { + owner.started = make(chan struct{}) + owner.cancelled = make(chan struct{}) + } registry := agent.NewRegistry() registry.RegisterKind(proto.SupportedAgentKind{Kind: "cleanup", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{EnvironmentNone: proto.CapabilitySupported})}, harnessconfig.Configuration{}) var factories atomic.Int32 @@ -165,16 +201,39 @@ func testDisconnectedPumpCleanup(t *testing.T, suspend bool) { t.Fatal(err) } if status.State == "ready" { + if silent { + start, _ := proto.NewEnvelope(proto.TypeExecutionStart, "prepare", proto.ExecutionStartPayload{Handle: status.Handle, ExecutorID: status.ExecutorID, RunID: "active-run", Input: proto.TextInput("controlled fixture")}) + if err := peer.WriteJSON(start); err != nil { + t.Fatal(err) + } + select { + case <-owner.started: + case <-time.After(3 * time.Second): + t.Fatal("active Turn did not start") + } + } break } if status.State == "failed" { t.Fatalf("preparation failed: %+v", status) } } - _ = peer.Close() + wait := 3 * time.Second + if silent { + wait = 70 * time.Second + } else { + _ = peer.Close() + } select { case <-owner.retry: - case <-time.After(3 * time.Second): + if silent { + select { + case <-owner.cancelled: + default: + t.Fatal("owner cleanup did not cancel active Turn") + } + } + case <-time.After(wait): t.Fatal("original executor cleanup was not retried") } select { diff --git a/apps/daemon/internal/transport/liveness.go b/apps/daemon/internal/transport/liveness.go new file mode 100644 index 000000000..88f8314b6 --- /dev/null +++ b/apps/daemon/internal/transport/liveness.go @@ -0,0 +1,19 @@ +package transport + +import "time" + +const ( + peerProbeInterval = 15 * time.Second + peerIdleTimeout = 60 * time.Second +) + +// A successful local write is not evidence that Core is still reachable through +// a proxy. Require inbound traffic or WebSocket pong replies, even when a Turn +// is busy producing output. Only the reader updates its deadline after setup. +func (c *Conn) configureLiveness(timeout time.Duration) { + c.peerTimeout = timeout + _ = c.ws.SetReadDeadline(time.Now().Add(timeout)) + c.ws.SetPongHandler(func(string) error { + return c.ws.SetReadDeadline(time.Now().Add(timeout)) + }) +} diff --git a/apps/daemon/internal/transport/liveness_test.go b/apps/daemon/internal/transport/liveness_test.go new file mode 100644 index 000000000..4c70bbef3 --- /dev/null +++ b/apps/daemon/internal/transport/liveness_test.go @@ -0,0 +1,100 @@ +package transport + +import ( + "errors" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +func TestLivenessDetectsSilentPeerAndAllowsResponsiveIdlePeer(t *testing.T) { + for _, responsive := range []bool{false, true} { + name := "silent" + if responsive { + name = "responsive" + } + t.Run(name, func(t *testing.T) { + stop := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ws, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + return + } + defer ws.Close() + if !responsive { + <-stop + return + } + for { + if _, _, err := ws.ReadMessage(); err != nil { + return + } + } + })) + defer srv.Close() + defer close(stop) + c, err := dial(t.Context(), DialOptions{WSURL: "ws" + strings.TrimPrefix(srv.URL, "http"), DeviceID: "test", Credential: "test", DaemonVersion: "0.22.0"}, 20*time.Millisecond, 200*time.Millisecond) + if err != nil { + t.Fatal(err) + } + defer c.Close() + deadline := time.NewTimer(800 * time.Millisecond) + defer deadline.Stop() + for { + select { + case <-c.Done(): + if responsive { + t.Fatalf("healthy idle connection closed: %v", c.Err()) + } + var ne net.Error + if !errors.As(c.Err(), &ne) || !ne.Timeout() { + t.Fatalf("expected timeout, got %v", c.Err()) + } + return + case <-deadline.C: + if !responsive { + t.Fatal("silent peer remained connected") + } + return + } + } + }) + } +} + +func TestLivenessBoundsBlockedInboundDispatch(t *testing.T) { + stop := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ws, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + return + } + defer ws.Close() + for i := 0; i < 65; i++ { + if err := ws.WriteMessage(websocket.TextMessage, []byte(`{"type":"heartbeat","id":"fixture","payload":{}}`)); err != nil { + return + } + } + <-stop + })) + defer srv.Close() + defer close(stop) + c, err := dial(t.Context(), DialOptions{WSURL: "ws" + strings.TrimPrefix(srv.URL, "http"), DeviceID: "test", Credential: "test", DaemonVersion: "0.22.0"}, 20*time.Millisecond, 200*time.Millisecond) + if err != nil { + t.Fatal(err) + } + defer c.Close() + select { + case <-c.Done(): + if c.Err() == nil || !strings.Contains(c.Err().Error(), "inbound dispatch timeout") { + t.Fatalf("unexpected failure: %v", c.Err()) + } + case <-time.After(2 * time.Second): + t.Fatal("full receive queue prevented connection cleanup") + } +} diff --git a/apps/daemon/internal/transport/ws.go b/apps/daemon/internal/transport/ws.go index a55bb47f8..e20e80e87 100644 --- a/apps/daemon/internal/transport/ws.go +++ b/apps/daemon/internal/transport/ws.go @@ -65,6 +65,9 @@ type Conn struct { hbCancel context.CancelFunc firstHeartbeat heartbeatObservation + probeInterval time.Duration + peerTimeout time.Duration // Immutable once the reader starts. + eventCredit bool // Immutable selection confirmed by the upgrade response. flowMu sync.Mutex flows map[string]*eventFlow @@ -84,6 +87,10 @@ type envelopeWithAck struct { // Dial opens the reverse WebSocket. Read/write goroutines are live // on return; caller MUST eventually Close to avoid goroutine leaks. func Dial(ctx context.Context, opts DialOptions) (*Conn, error) { + return dial(ctx, opts, peerProbeInterval, peerIdleTimeout) +} + +func dial(ctx context.Context, opts DialOptions, probeInterval, idleTimeout time.Duration) (*Conn, error) { if opts.WSURL == "" || opts.DeviceID == "" || opts.Credential == "" || opts.DaemonVersion == "" { return nil, fmt.Errorf("transport.Dial: WSURL, DeviceID, Credential, DaemonVersion are all required") } @@ -144,6 +151,8 @@ func Dial(ctx context.Context, opts DialOptions) (*Conn, error) { sendCh: make(chan envelopeWithAck, 64), closed: make(chan struct{}), } + c.probeInterval = probeInterval + c.configureLiveness(idleTimeout) go c.writeLoop() go c.readLoop() return c, nil @@ -320,6 +329,8 @@ func (c *Conn) StartHeartbeats(parentCtx context.Context, interval time.Duration // from the defer rather than from Conn.Close() so gorilla/websocket's // internal write state isn't raced. func (c *Conn) writeLoop() { + probe := time.NewTicker(c.probeInterval) + defer probe.Stop() defer func() { // Best-effort close frame so the peer sees code 1000 rather // than EOF. @@ -332,6 +343,12 @@ func (c *Conn) writeLoop() { select { case <-c.closed: return + case <-probe.C: + if err := c.ws.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil { + c.fail(fmt.Errorf("transport: liveness probe: %w", err)) + _ = c.Close() + return + } case msg := <-c.sendCh: msg.stamp.Received(msg.observation, workspaceobs.OutboundQueue) started := msg.observation.Begin(workspaceobs.OutboundEncode) @@ -392,6 +409,9 @@ func (c *Conn) readLoop() { return } readReturned := time.Now() + if c.peerTimeout > 0 { + _ = c.ws.SetReadDeadline(readReturned.Add(c.peerTimeout)) + } log.Bg().Debug("transport: received WS frame", "bytes", len(raw)) decodeStarted := time.Now() var env proto.Envelope @@ -411,9 +431,15 @@ func (c *Conn) readLoop() { } received := ReceivedEnvelope{Envelope: env, decoded: time.Now(), readReturned: readReturned, decodeStarted: decodeStarted, connectionID: c.connectionID} log.Bg().Info("transport: dispatching envelope", "type", env.Type, "id", env.ID) + delivery := time.NewTimer(c.peerTimeout) select { case c.recvCh <- received: + delivery.Stop() case <-c.closed: + delivery.Stop() + return + case <-delivery.C: + c.fail(errors.New("transport: inbound dispatch timeout")) return } } diff --git a/docs/runtime-protocol.md b/docs/runtime-protocol.md index 025e88180..19e9365dd 100644 --- a/docs/runtime-protocol.md +++ b/docs/runtime-protocol.md @@ -32,6 +32,7 @@ The wire version is [`proto.Version`](https://github.com/MiniMax-AI/OpenAgentCor - Both requests carry the credential as a Bearer header. A credential never appears in a URL, a payload log or a trace. - The first heartbeat is sent at once, then at the interval that bootstrap returned. Each heartbeat declares `supported_agent_kinds`, their availability and their [capabilities](#capability-declarations). Before the first heartbeat, capabilities are unknown; a kind missing from a heartbeat is not advertised. Neither permits inference. - Heartbeats establish liveness only, never execution progress or a receipt for an earlier message. +- The Runtime probes the WebSocket with a standard Ping every 15 seconds and requires an inbound message or Pong within 60 seconds. Local write success alone does not prove that Core remains reachable through a proxy. An expired read deadline closes the connection and enters ordinary disconnect cleanup; it is not a Turn duration limit or permission to replay execution. Core records transport failure classes and WebSocket close codes without peer-controlled close text or frame contents. Each physical connection has fresh routing, admission handles and transfer state. A newer connection for the same device replaces the previous one: Core evicts the old Run routes, and the new connection inherits none of them. A valid credential and connection are never authority to choose another Session or Environment binding. diff --git a/docs/zh/runtime-protocol.md b/docs/zh/runtime-protocol.md index 68f532f42..088dbbaaa 100644 --- a/docs/zh/runtime-protocol.md +++ b/docs/zh/runtime-protocol.md @@ -1,7 +1,7 @@ --- title: "Core–Runtime 协议" source: docs/runtime-protocol.md -source_hash: 14bba94dc1617430e0fa049591645e2d1084b4ce2d976bb9bebe54c1d338fb6c +source_hash: 89038da4d9b924fe2bd0e28dd2420515e8d9cf38988dd53e1fd9370d66655a75 --- Core–Runtime 协议连接 Core 与 Runtime daemon，定义 daemon 连接上消息的含义和顺序。托管和自托管 Runtime 使用同一协议。 @@ -34,6 +34,7 @@ wire 版本为 [`proto.Version`](https://github.com/MiniMax-AI/OpenAgentCore/blo - 两个请求都通过 Bearer header 提供凭据。凭据不得出现在 URL、payload 日志或 trace 中。 - 立即发送第一个 heartbeat，之后按 bootstrap 返回的间隔发送。每个 heartbeat 声明 `supported_agent_kinds`、其可用性和[能力](#capability-declarations)。第一个 heartbeat 之前能力未知；heartbeat 中缺失的 kind 视为未声明。两者都不允许推断。 - Heartbeat 仅证明存活，不证明执行进度，也不充当此前消息的回执。 +- Runtime 每 15 秒发送标准 WebSocket Ping，并要求在 60 秒内收到入站消息或 Pong。仅本地写入成功不足以证明经过代理仍能到达 Core。读取截止时间到期会关闭连接并进入常规断连清理；它不是 Turn 总时长限制，也不授权重放执行。Core 记录传输故障类别及 WebSocket 关闭码，不记录对端提供的关闭文本或帧内容。 每条物理连接拥有新的路由、admission handle 和传输状态。同一设备的新连接替代旧连接：Core 移除旧 Run 路由，新连接不继承这些状态。有效凭据和连接不授权选择其他 Session 或 Environment 绑定。 diff --git a/services/core/internal/runtimegateway/session.go b/services/core/internal/runtimegateway/session.go index 69c61802d..2a88c5a1d 100644 --- a/services/core/internal/runtimegateway/session.go +++ b/services/core/internal/runtimegateway/session.go @@ -388,7 +388,9 @@ func (s *Session) writeLoop() { o.end(artifactWrite, writeStarted, writeErr) o.frame(-1) if err := writeErr; err != nil { - s.log("agentdaemon gateway: write %s frame: %v", env.Type, err) + if !s.IsClosed() { + s.logTransportFailure("write", err) + } s.Close("write error: " + err.Error()) return } @@ -409,7 +411,9 @@ func (s *Session) readLoop() { _ = s.conn.SetReadDeadline(time.Now().Add(HeartbeatTimeout)) _, raw, err := s.conn.ReadMessage() if err != nil { - s.log("agentdaemon gateway: read frame: %v", err) + if !s.IsClosed() { + s.logTransportFailure("read", err) + } return } readReturned := time.Now() diff --git a/services/core/internal/runtimegateway/transport_failure.go b/services/core/internal/runtimegateway/transport_failure.go new file mode 100644 index 000000000..392efe412 --- /dev/null +++ b/services/core/internal/runtimegateway/transport_failure.go @@ -0,0 +1,38 @@ +package runtimegateway + +import ( + "errors" + "fmt" + "io" + "net" + + obslog "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log" + "github.com/gorilla/websocket" +) + +// Log transport facts without peer-controlled close text or frame contents. +func (s *Session) logTransportFailure(operation string, err error) { + kind, code := transportFailure(err) + obslog.Bg().Warn("runtime transport failed", "device_id", s.DeviceID, + "operation", operation, "error_kind", kind, "close_code", code, + "error_type", fmt.Sprintf("%T", err)) +} + +func transportFailure(err error) (string, int) { + var closeErr *websocket.CloseError + var netErr net.Error + switch { + case errors.As(err, &closeErr): + return "websocket_close", closeErr.Code + case errors.As(err, &netErr) && netErr.Timeout(): + return "timeout", 0 + case errors.Is(err, websocket.ErrReadLimit): + return "frame_too_large", 0 + case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF): + return "eof", 0 + case errors.Is(err, net.ErrClosed): + return "socket_closed", 0 + default: + return "io_error", 0 + } +} diff --git a/services/core/internal/runtimegateway/transport_failure_test.go b/services/core/internal/runtimegateway/transport_failure_test.go new file mode 100644 index 000000000..a24118d3c --- /dev/null +++ b/services/core/internal/runtimegateway/transport_failure_test.go @@ -0,0 +1,57 @@ +package runtimegateway + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "log/slog" + "net" + "os" + "strings" + "testing" + + "github.com/gorilla/websocket" +) + +func TestTransportFailureClassifiesUnderlyingCause(t *testing.T) { + for _, tc := range []struct { + name string + err error + kind string + code int + }{ + {"read timeout", &net.OpError{Op: "read", Net: "tcp", Err: os.ErrDeadlineExceeded}, "timeout", 0}, + {"abnormal close", &websocket.CloseError{Code: 1006, Text: "unexpected EOF"}, "websocket_close", 1006}, + {"peer text is not classification", &websocket.CloseError{Code: 1008, Text: "secret-peer-text"}, "websocket_close", 1008}, + {"oversized frame", websocket.ErrReadLimit, "frame_too_large", 0}, + {"wrapped EOF", fmt.Errorf("read: %w", io.EOF), "eof", 0}, + {"closed socket", net.ErrClosed, "socket_closed", 0}, + } { + t.Run(tc.name, func(t *testing.T) { + kind, code := transportFailure(tc.err) + if kind != tc.kind || code != tc.code { + t.Fatalf("got %s/%d, want %s/%d", kind, code, tc.kind, tc.code) + } + }) + } +} + +func TestTransportFailureLogsWithoutOptionalCallback(t *testing.T) { + var out bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&out, nil))) + defer slog.SetDefault(previous) + s := &Session{DeviceID: "device-fixture"} + s.logTransportFailure("read", &websocket.CloseError{Code: 1006, Text: "private-peer-text"}) + var row map[string]any + if err := json.Unmarshal(out.Bytes(), &row); err != nil { + t.Fatal(err) + } + if row["msg"] != "runtime transport failed" || row["operation"] != "read" || row["device_id"] != "device-fixture" || row["close_code"] != float64(1006) || row["error_kind"] != "websocket_close" { + t.Fatalf("missing diagnostic fields: %v", row) + } + if strings.Contains(out.String(), "private-peer-text") { + t.Fatal("peer text leaked") + } +} From ac5dd38f4b3b2789ba71fe08de265107350ff3e0 Mon Sep 17 00:00:00 2001 From: sam Date: Thu, 29 Oct 2026 21:18:40 +0800 Subject: [PATCH 2/2] test: isolate execution diagnostics from asynchronous gateway logs --- .../execution/execution_observation_test.go | 54 ++++++++++++++++++- 1 file changed, 52 insertions(+), 2 deletions(-) diff --git a/services/core/internal/execution/execution_observation_test.go b/services/core/internal/execution/execution_observation_test.go index 15bd9e814..05f1db257 100644 --- a/services/core/internal/execution/execution_observation_test.go +++ b/services/core/internal/execution/execution_observation_test.go @@ -17,9 +17,59 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimegateway" ) -func captureExecutionLogs(t *testing.T) *bytes.Buffer { +// Gateway connections can finish asynchronously after their owning test exits. +// Their lifecycle logs are verified in runtimegateway, not by execution's +// single-record observation assertions. +type executionLogBuffer struct { + mu sync.Mutex + data bytes.Buffer +} + +func (b *executionLogBuffer) Write(p []byte) (int, error) { + var row struct { + Message string `json:"msg"` + } + if json.Unmarshal(p, &row) == nil && (row.Message == "runtime transport failed" || row.Message == "runtime connection closed") { + return len(p), nil + } + b.mu.Lock() + defer b.mu.Unlock() + return b.data.Write(p) +} +func (b *executionLogBuffer) Bytes() []byte { + b.mu.Lock() + defer b.mu.Unlock() + return bytes.Clone(b.data.Bytes()) +} +func (b *executionLogBuffer) String() string { return string(b.Bytes()) } +func (b *executionLogBuffer) Len() int { b.mu.Lock(); defer b.mu.Unlock(); return b.data.Len() } +func (b *executionLogBuffer) Reset() { b.mu.Lock(); defer b.mu.Unlock(); b.data.Reset() } + +func TestExecutionLogCaptureExcludesConcurrentGatewayLifecycle(t *testing.T) { + out := captureExecutionLogs(t) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 100; i++ { + obslog.Bg().Warn("runtime transport failed", "operation", "read") + obslog.Bg().Info("runtime connection closed") + } + }() + recordInitialInputOrigin(t.Context(), "capture-fixture") + wg.Wait() + var row map[string]any + if err := json.Unmarshal(out.Bytes(), &row); err != nil { + t.Fatal(err) + } + if row["session_id"] != "capture-fixture" { + t.Fatal(row) + } +} + +func captureExecutionLogs(t *testing.T) *executionLogBuffer { t.Helper() - var out bytes.Buffer + var out executionLogBuffer previous := slog.Default() slog.SetDefault(slog.New(obslog.NewContextHandler(slog.NewJSONHandler(&out, nil)))) t.Cleanup(func() { slog.SetDefault(previous) })