fix: detect unresponsive runtime connections and expose transport failures

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 {

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))
+	})
+}

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")
+	}
+}

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
 		}
 	}

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.
 

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 绑定。