从零搭建 Agent Harness 系列（十六）Protocol、WebSocket 与多渠道连接层

发表于 分类于 Waline：
在 Task、Runtime 和 TCP 多 Session 之后，为 go-tiny-claw 增加 JSON Protocol、结构化事件、WebSocket 接入和可视化多 Session 控制台。

上一篇我们让一个 Go 进程可以同时承载多个 TCP 对话：每个连接拥有自己的 ChannelSession、Runtime 和 Session，不同 Session 可以并行运行，而同一个 Runtime 内部仍然只允许一个 Task。

但系列十五结尾还留下了几个关键问题：

1
2
3
4
TCP 客户端和 Server 之间到底传什么格式？
浏览器为什么不能直接连接 TCP Server？
流式文本、工具调用和审批请求如何传到外部渠道？
一个前端页面如何同时管理多个 Session？

这一篇继续实现这些能力。本文对应当前 go-tiny-claw 中已经落地的连接层：

1
2
3
4
5
6
JSON Line Protocol
Structured Event
JSON Reporter
ChannelSession
WebSocket Server
React Multi-Session Console

一、从 TCP 多连接到浏览器控制台

系列十五中的 TCP 连接可以这样使用：

1
2
3
4
5
6
7
8
9
TCP Client
    ↓
TCPServer
    ↓
ChannelSession
    ↓
Runtime
    ↓
AgentEngine

TCP 是一个字节流。Server 能够读取到字节，但它并不知道这些字节代表什么业务动作。

如果客户端直接发送：

1
帮我读取 README

Server 无法可靠判断这是一条 Prompt，还是一次中断、审批响应或者关闭请求。

因此需要在传输层之上定义应用协议：

1
2
3
4
5
TCP / WebSocket
    ↓
Message Protocol
    ↓
ChannelSession

同时，浏览器还有一个限制：浏览器 JavaScript 不能直接创建原始 TCP 连接。浏览器可以使用 HTTP、WebSocket 等标准 Web 协议，但不能像 Go 客户端一样连接 net.Conn。

所以最终的连接层变成：

1
2
3
TCP Client ─────── TCP :8080 ───────┐
                                    ├── ChannelSession ─── Runtime
React Browser ─ WebSocket :8081 ───┘

TCP 和 WebSocket 使用不同的传输适配器，但进入后面的 ChannelSession、Runtime 和 AgentEngine。

二、先定义连接协议

协议实现位于：

1
internal/channel/protocol.go

当前协议采用 JSON Line 形式。每条消息是一个 JSON 对象，并以换行结束。

1. 消息类型

1
2
3
4
5
6
7
8
9
10
type MessageType string

const (
	MessagePrompt           MessageType = "prompt"
	MessageInterrupt        MessageType = "interrupt"
	MessageClose            MessageType = "close"
	MessagePing             MessageType = "ping"
	MessagePong             MessageType = "pong"
	MessageApprovalResponse MessageType = "approval_response"
)

客户端发送 Prompt：

1
{"type":"prompt","content":"请读取 README"}

客户端请求中断：

1
{"type":"interrupt"}

客户端响应审批：

1
{"type":"approval_response","request_id":"abc123","decision":"allow_once"}

这里的 type 是连接层协议类型。它和 Reporter 发出的 Agent 事件类型不是同一个概念。

1
2
3
4
5
MessageType
  表示客户端要求 Server 做什么

EventType
  表示 Agent 当前发生了什么

2. Message 结构

1
2
3
4
5
6
type Message struct {
	Type      MessageType `json:"type"`
	Content   string      `json:"content,omitempty"`
	RequestID string      `json:"request_id,omitempty"`
	Decision  string      `json:"decision,omitempty"`
}

目前的消息结构仍然比较小，但已经覆盖了连续对话所需的控制面：

1
2
3
Content   Prompt 内容
RequestID 审批请求 ID
Decision  审批决策

未来如果需要协议版本、客户端 ID、任务 ID 和请求追踪，可以继续扩展这个结构，或者引入统一的 Envelope。

3. MessageReader 为什么接收 io.Reader

MessageReader 的构造函数不再要求调用方传入 *bufio.Reader：

1
2
3
4
5
6
7
func NewMessageReader(input io.Reader) (*MessageReader, error) {
	if input == nil {
		return nil, errors.New("消息读取器输入不能为空")
	}

	return &MessageReader{reader: bufio.NewReader(input)}, nil
}

这个边界很重要：

1
2
外部依赖：io.Reader
内部实现：bufio.Reader

因此以下输入都可以被协议层使用：

1
2
3
4
net.Conn
WebSocket Adapter
bytes.Buffer
测试输入

调用方不需要知道协议层是否使用缓冲。

4. Reader 负责协议校验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
func (r *MessageReader) Read() (Message, error) {
	line, err := r.reader.ReadBytes('\n')

	if len(line) > MaxMessageSize {
		return Message{}, errors.New("消息长度超过限制")
	}

	if err != nil {
		return Message{}, err
	}

	line = bytes.TrimSpace(line)

	var message Message
	if err := json.Unmarshal(line, &message); err != nil {
		return Message{}, errors.New("消息格式错误")
	}

	switch message.Type {
	case MessagePrompt:
		if message.Content == "" {
			return Message{}, errors.New("提示不能为空")
		}
	case MessageApprovalResponse:
		if message.RequestID == "" {
			return Message{}, errors.New("审批响应缺少请求 ID")
		}
		if message.Decision == "" {
			return Message{}, errors.New("审批响应缺少决策")
		}
	default:
		return Message{}, errors.New("不支持的消息类型")
	}

	return message, nil
}

协议层至少应该负责：

1
2
3
4
限制单条消息大小
校验 JSON
校验消息类型
校验必需字段

它不应该负责启动 Agent 或执行工具。协议层只回答“收到了一条什么消息”。

三、MessageWriter 解决并发输出

Server 中一个 Session 可能同时产生多种输出：

1
2
3
4
5
6
模型文本 Delta
工具调用事件
工具结果事件
审批请求
任务完成事件
Ping 响应

这些事件可能来自不同 Goroutine。因此不能让多个 Goroutine 直接对同一个连接调用 json.Encoder。

当前 Writer 使用一个互斥锁：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
type MessageWriter struct {
	encoder *json.Encoder
	mu      sync.Mutex
}

func NewMessageWriter(output io.Writer) (*MessageWriter, error) {
	if output == nil {
		return nil, errors.New("消息写入器输出不能为空")
	}

	return &MessageWriter{encoder: json.NewEncoder(output)}, nil
}

func (w *MessageWriter) Write(value any) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	return w.encoder.Encode(value)
}

这里有两个设计点。

第一，Writer 接收 io.Writer，不和 TCP、WebSocket 绑定。

第二，Write 接收 any，因为它既可能写入连接控制消息：

1
Message{Type: MessagePong}

也可能写入 Agent 事件：

1
reporter.Event{Type: reporter.EventToolCall}

它的真实职责不是“只写 Message”，而是“串行写入 JSON 对象”。

同一个连接内，Writer 保证 JSON 字节不会交错；不同连接之间则因为底层连接不同，天然隔离。
同一个连接内，Writer 保证 JSON 字节不会交错；不同连接之间则因为底层连接不同，天然隔离。

四、从 Reporter 回调到结构化事件

AgentEngine 只依赖 Reporter 接口：

1
2
AgentEngine
    └── reporter.Reporter

它不会直接调用 fmt.Printf，也不会直接写 TCP。

1. Event 定义

文件：

1
internal/reporter/event.go

当前事件类型包括：

1
2
3
4
5
6
7
8
9
10
11
12
13
const (
	EventThinking         EventType = "thinking"
	EventTextDelta       EventType = "text_delta"
	EventTextCompleted   EventType = "text_completed"
	EventToolCall        EventType = "tool_call"
	EventToolResult      EventType = "tool_result"
	EventTaskCompleted   EventType = "task_completed"
	EventTaskCanceled    EventType = "task_canceled"
	EventTaskFailed      EventType = "task_failed"
	EventError           EventType = "error"
	EventPong            EventType = "pong"
	EventApprovalRequest EventType = "approval_request"
)

事件结构：

1
2
3
4
5
6
7
8
9
10
11
12
type Event struct {
	Type      EventType `json:"type"`
	Content   string    `json:"content,omitempty"`
	ToolName  string    `json:"tool_name,omitempty"`
	Result    string    `json:"result,omitempty"`
	RequestID string    `json:"request_id,omitempty"`
	Decision  string    `json:"decision,omitempty"`
	Risk      string    `json:"risk,omitempty"`
	Reason    string    `json:"reason,omitempty"`
	IsError   bool      `json:"is_error,omitempty"`
	Error     string    `json:"error,omitempty"`
}

2. JSONReporter

文件：

1
internal/reporter/json_reporter.go

JSONReporter 把 Engine 的回调转换成 Event：

1
2
3
4
5
6
7
8
9
10
11
func (r *JSONReporter) OnToolCall(
	ctx context.Context,
	toolName string,
	args string,
) {
	r.publish(ctx, Event{
		Type:     EventToolCall,
		ToolName: toolName,
		Content:  args,
	})
}

最终发送给浏览器的是：

1
2
3
4
5
{
  "type":"tool_call",
  "tool_name":"write_file",
  "content":"{...}"
}

同一个 Engine 可以继续使用 TerminalReporter：

1
2
CLI → TerminalReporter → stdout
WebSocket → JSONReporter → EventSink → MessageWriter

Reporter 的抽象让输出渠道可以替换，而不需要修改 Agent Loop。

3. EventSink

文件：

1
internal/channel/event_sink.go
1
2
3
type EventSink interface {
	Publish(ctx context.Context, event Event) error
}

JSONReporter 只依赖 EventSink：

1
2
3
4
5
6
7
JSONReporter
    ↓
EventSink
    ↓
MessageWriter
    ↓
TCP / WebSocket

这样 Reporter 不需要知道底层是字节流还是 WebSocket Frame。

五、ChannelSession：连接和 Runtime 的绑定层

文件：

1
internal/channel/channel_session.go

一个 ChannelSession 的结构是：

1
2
3
4
5
6
外部连接
  ├── MessageReader
  ├── MessageWriter
  ├── JSONReporter
  ├── ChannelApprovalHandler
  └── Runtime

构造时，所有输出共享同一个 MessageWriter：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
writer, err := NewMessageWriter(conn)
if err != nil {
	return nil, err
}

eventSink, err := NewJSONEventSinkWithWriter(writer)
if err != nil {
	return nil, err
}

bundle, err := manager.Create(id, runtimepkg.RuntimeOptions{
	ApprovalHandler: channelApproval,
	Reporter:        reporter.NewJSONReporter(eventSink),
})

这里不能分别创建两个 Writer：

1
2
3
4
5
错误做法：
Reporter → Writer A
Ping     → Writer B
             ↓
        同一个连接

两个 Writer 各自持有锁，无法保护彼此，可能导致输出交错。

正确做法是：

1
2
3
Reporter ─┐
          ├── 同一个 MessageWriter
Ping ─────┘

1. 消息循环

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
func (s *ChannelSession) Run(ctx context.Context) error {
	for {
		message, err := s.reader.Read()
		if err != nil {
			return err
		}

		switch message.Type {
		case MessagePrompt:
			if err := s.startTask(ctx, message.Content); err != nil {
				s.publishError(ctx, err)
			}
		case MessageInterrupt:
			s.runtime.Cancel()
		case MessageApprovalResponse:
			decision := approval.Decision(message.Decision)
			if err := s.approval.Respond(message.RequestID, decision); err != nil {
				s.publishError(ctx, err)
			}
		case MessagePing:
			if err := s.writer.Write(Message{Type: MessagePong}); err != nil {
				return err
			}
		case MessageClose:
			return nil
		}
	}
}

2. Prompt 必须异步启动

如果 MessagePrompt 直接同步等待：

1
2
task, _ := s.runtime.Start(ctx, prompt, s.reporter)
task.Wait()

那么消息循环会被阻塞。在 Agent 运行期间，下面这些消息都无法处理：

1
2
3
4
interrupt
approval_response
ping
close

当前实现启动 Task 后立即返回消息循环：

1
2
3
4
5
6
7
8
9
task, err := s.runtime.Start(ctx, prompt, s.reporter)
if err != nil {
	return err
}

go func() {
	err := task.Wait()
	// 发布 task_completed、task_canceled 或 task_failed
}()

这就是 Server 能够在 Agent 执行期间响应中断和审批的原因。
这就是 Server 能够在 Agent 执行期间响应中断和审批的原因。

六、通道审批如何工作

终端审批可以直接读取 stdin，但 TCP 或 WebSocket 审批不能再启动一个 Reader 去读取同一连接。

否则会变成：

1
2
3
ChannelSession.Reader ──┐
                        ├── 同一个连接
ApprovalHandler.Reader ─┘

两个 Reader 会竞争输入，导致 Prompt 或审批响应被错误消费。

当前 ChannelApprovalHandler 使用一个 pending Map：

1
2
3
4
5
6
type ChannelApprovalHandler struct {
	sink reporter.EventSink

	mu      sync.Mutex
	pending map[string]chan approval.Decision
}

审批开始时：

1
2
3
4
5
6
7
8
9
Engine
  ↓
ApprovalGate.Check
  ↓
ChannelApprovalHandler.Approve
  ↓
发送 approval_request 事件
  ↓
等待 pending[requestID]

前端收到：

1
2
3
4
5
6
{
  "type":"approval_request",
  "request_id":"request-1",
  "tool_name":"write_file",
  "risk":"mutating"
}

点击允许一次后发送：

1
2
3
4
5
{
  "type":"approval_response",
  "request_id":"request-1",
  "decision":"allow_once"
}

ChannelSession 的唯一输入循环收到响应后，调用：

1
s.approval.Respond(message.RequestID, decision)

pending Channel 被唤醒，ApprovalGate 才会把工具调用交给 Engine 后续执行。

审批响应和 Prompt 使用同一个输入循环，这是多渠道审批能够正确工作的关键。

七、为什么需要 WebSocket Adapter

浏览器使用 WebSocket Frame，而当前 MessageReader 使用 JSON Line：

1
2
MessageReader 期待：{"type":"ping"}\n
WebSocket 实际：一个 Text Frame

因此不能直接把 *websocket.Conn 传给 MessageReader。需要一个适配器，把 Frame 转换成 Reader/Writer 认识的字节流。

文件：

1
internal/server/websocket_conn.go

读取时，适配器从下一条 WebSocket 消息读取内容，并补充换行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
func (c *websocketConn) Read(p []byte) (int, error) {
	for c.readOffset >= len(c.readBuffer) {
		messageType, reader, err := c.conn.NextReader()
		if err != nil {
			return 0, err
		}

		if messageType != websocket.TextMessage && messageType != websocket.BinaryMessage {
			return 0, errors.New("只支持文本或二进制消息")
		}

		message, err := io.ReadAll(reader)
		if err != nil {
			return 0, err
		}

		c.readBuffer = append(message[:0:0], message...)
		c.readBuffer = append(c.readBuffer, '\n')
		c.readOffset = 0
	}

	count := copy(p, c.readBuffer[c.readOffset:])
	c.readOffset += count
	return count, nil
}

写入时，适配器把一条 JSON Line 转换成一个 Text Frame：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func (c *websocketConn) Write(p []byte) (int, error) {
	c.writeMu.Lock()
	defer c.writeMu.Unlock()

	payload := bytes.TrimSpace(p)
	if len(payload) == 0 {
		return len(p), nil
	}

	if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil {
		return 0, err
	}

	return len(p), nil
}

适配完成后，ChannelSession 不需要知道自己面对的是 TCP 还是 WebSocket：

1
2
3
4
5
TCP Conn ────────────────┐
                         ├── io.ReadWriteCloser
WebSocket Stream Adapter ┘
                         ↓
                   ChannelSession

八、WebSocket Server 如何承载多个 Session

文件：

1
internal/server/websocket_server.go

TCP Server 和 WebSocket Server 的职责类似：

1
2
3
4
5
6
7
监听连接
生成 Session ID
创建 ChannelSession
注册连接
运行消息循环
连接结束后销毁 Runtime
服务关闭时清理所有连接

WebSocket Session ID 使用独立前缀：

1
2
ws-channel-session-1
ws-channel-session-2

TCP 使用：

1
2
tcp-channel-session-1
tcp-channel-session-2

两个 Server 共享同一个 RuntimeManager，但每个连接仍然拥有独立 Runtime：

1
2
3
4
5
RuntimeManager
  ├── tcp-channel-session-1 → Runtime A
  ├── tcp-channel-session-2 → Runtime B
  ├── ws-channel-session-1  → Runtime C
  └── ws-channel-session-2  → Runtime D

启动时，当前 Server 使用两个端口：

1
2
TCP       :8080
WebSocket :8081/ws

这里没有强行把 TCP 和 WebSocket 复用到同一个端口，因为二者的连接握手不同。生产环境可以使用反向代理统一域名，也可以使用连接复用器，但学习阶段使用两个端口更清晰。

九、前端多 Session 控制台

前端目录：

1
web-console

技术栈：

1
2
3
4
Vite
React
TypeScript
WebSocket

1. 一个 Session 一条 WebSocket

前端维护一个 Socket Map：

1
const sockets = useRef<Record<string, WebSocket>>({})

创建 Session 时：

1
2
3
4
5
6
7
8
9
10
11
12
const id = `session-${sequence.current++}`

setSessions((current) => [...current, {
  id,
  status: 'connecting',
  items: [],
  draft: '',
  createdAt: Date.now(),
}])

setActiveID(id)
window.setTimeout(() => connectSession(id), 0)

每个连接的消息事件都带着自己的 Session ID 回到状态更新函数：

1
2
3
socket.onmessage = (message) => {
  handleEvent(id, JSON.parse(message.data))
}

因此连接 A 的事件只会更新 Session A 的消息列表。

2. 流式文本聚合

Server 会连续发送：

1
2
3
{"type":"text_delta","content":"第一段"}
{"type":"text_delta","content":"第二段"}
{"type":"text_completed"}

前端在 Session 内记录当前的 streamItemId：

1
2
3
4
5
6
7
8
if (session.streamItemId) {
  return {
    ...session,
    items: session.items.map((item) => item.id === session.streamItemId
      ? { ...item, content: item.content + (event.content ?? '') }
      : item),
  }
}

这样每个 Delta 都会追加到当前 Agent 消息，而不是生成很多气泡。

3. 审批按钮

收到 approval_request 后，前端生成审批卡片：

1
2
3
4
需要审批 · write_file
risk: mutating

[允许一次] [会话允许] [拒绝]

按钮最终发送：

1
2
3
4
5
socket.send(JSON.stringify({
  type: 'approval_response',
  request_id: requestID,
  decision,
}))

前端不直接执行工具，也不决定工具权限。它只是把用户选择传回 Server，真正的 Policy、GrantStore 和 Gate 仍然在 Runtime 内部。

4. 固定底部输入框

连续对话的输入框不能随着消息增长被推到页面之外。这里不是简单地给输入框设置 position: fixed，而是让工作区成为一个高度受控的 Flex 容器：消息区滚动，输入区不参与滚动。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
html,
#root {
  height: 100%;
}

.workspace {
  display: flex;
  flex: 1;
  flex-direction: column;
  min-height: 0;
  height: 100vh;
  overflow: hidden;
}

.conversation {
  flex: 1 1 auto;
  min-height: 0;
  overflow-x: hidden;
  overflow-y: auto;
}

.composer-wrap {
  flex: 0 0 auto;
}

min-height: 0 很重要。Flex 子元素默认可能按照内容的最小高度撑开父容器，如果省略它，消息列表会把整个页面撑长，输入框仍然会被推走。

十、一次请求的完整时序

把各层连起来后，一条用户消息的生命周期如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
React Session A
  -> WebSocket Text Frame
  -> WebSocketServer
  -> websocketConn.Read
  -> MessageReader
  -> ChannelSession.handleMessage
  -> Runtime.Start
  -> Task goroutine
  -> AgentEngine.Run
  -> JSONReporter.OnTextDelta
  -> EventSink
  -> MessageWriter
  -> WebSocket Text Frame
  -> task_completed

这里有两个容易混淆的异步边界。

第一，ChannelSession 的读循环不能直接同步执行 Agent。否则 Agent 调用模型、执行工具时，连接就无法继续读取 interrupt 或 approval_response。因此收到 prompt 后要启动任务，读循环继续工作。

第二，Agent 的输出不能直接写 WebSocket。Agent 只依赖 reporter.Reporter，由 JSONReporter 把领域事件转换成协议消息，再由 MessageWriter 串行写入连接。这样 Runtime 不知道当前连接是 TCP、WebSocket 还是未来的飞书。

审批的时序则是：

1
2
3
4
5
6
Agent -> Policy/Gate -> approval_request -> Client
Client -> approval_response -> ChannelApprovalHandler
ChannelApprovalHandler -> pending[requestID]
Gate <- decision
Gate -> Tool.Execute
Agent -> tool_result -> Client

request_id 是审批闭环的关联键。不能只依赖工具名，因为同一个 Session 中可能同时存在多个工具调用，多个连接也可能并行审批。

十一、如何验证这一阶段

先启动 Server：

1
2
3
cd /Users/smsun/Documents/github/go-tiny-claw
export ZHIPU_API_KEY="你的 API Key"
go run ./cmd/claw_server

启动 Web 控制台：

1
2
3
cd /Users/smsun/Documents/github/go-tiny-claw/web-console
npm install
npm run dev

浏览器打开 http://localhost:5173。开发服务器会把 WebSocket 请求转发到 ws://127.0.0.1:8081/ws。打开两个会话后分别发送消息，可以验证：

  1. 两个会话能够同时运行任务。
  2. 会话 A 的流式事件不会显示到会话 B。
  3. 会话 A 按下中断时，只取消会话 A 的 Task。
  4. 会话 A 等待审批时，会话 B 仍然可以继续对话。
  5. 审批响应只唤醒对应的 request_id。
  6. Task 结束后会收到且只收到一个终态事件。

已有的 Go 测试和前端构建可以这样运行：

1
2
3
4
5
6
cd /Users/smsun/Documents/github/go-tiny-claw
GOCACHE=/tmp/go-tiny-claw-cache go test ./...
GOCACHE=/tmp/go-tiny-claw-cache go test -race ./test/channel ./test/runtime

cd web-console
npm run build

测试重点不是只验证“能不能返回文本”，而是验证连接边界和并发边界：协议读写、审批等待、Session 隔离、Task 终止、事件顺序和竞态安全。

十二、距离生产级还缺什么

这一阶段解决的是“多连接可用”和“浏览器可接入”，还不能直接当作公网生产服务。下一步至少要补齐以下能力。

1. 身份认证与授权

当前连接建立后还没有可靠的用户身份。生产环境需要在握手或首条消息中完成认证，并把 UserID、TenantID、ProjectID 绑定到 Session。Policy 和 GrantStore 也要按用户、项目或租户隔离，不能让一个连接查询到另一个用户的授权。

2. TLS、Origin 和网络保护

WebSocket 生产部署应使用 TLS。服务端不能永久允许任意 Origin，要配置允许的前端域名。同时还需要限制连接数、单条消息大小、并发 Task 数和请求频率，避免一个客户端耗尽进程资源。

3. 超时、心跳和连接清理

需要分别设置握手超时、读超时、写超时、模型调用超时和工具执行超时。ping/pong 不能只作为协议示例，还要用来发现断开的连接。连接断开后必须取消当前 Task，并从 Manager 中清理 Session，避免 goroutine 和状态泄漏。

4. 协议版本和错误模型

协议消息应增加版本字段，并定义稳定的错误结构，例如错误码、可重试标记、关联的 request_id 和 task_id。不能只返回一段无法被前端稳定解析的中文错误字符串。

5. 事件关联与恢复

生产事件至少需要 session_id、task_id、event_id、时间戳和序号。客户端重连后才能根据序号补事件，而不是只能重新发起任务。长任务还需要持久化 Task 状态、审批状态和必要的事件日志。

6. 持久化和水平扩展

现在 Manager、Session 和 GrantStore 主要是进程内内存实现。单进程可以支持多个 TCP/WebSocket 连接，但服务重启后状态会丢失，也无法让多个实例共同管理同一个 Session。生产化需要把 Session、Task、授权和事件分别抽象出持久化接口，必要时使用 Redis、数据库和消息总线。

7. 可观测性和可靠性

日志要携带结构化的 session_id、task_id 和 request_id，指标要覆盖连接数、活跃 Task、模型延迟、工具延迟、审批等待时间、错误率和取消率。还需要补充 TCP/WebSocket 集成测试、断线测试、重复审批测试、并发压力测试和故障恢复测试。

8. 前端生产能力

控制台还需要登录、自动重连、历史消息加载、断线提示、重复发送保护、审批超时提示和权限展示。开发环境的 Vite 代理只解决本地调试问题，生产环境应由反向代理统一提供 HTTPS、WebSocket 转发和静态资源服务。

十三、这一阶段的架构总结

到这里，连接层可以整理成下面的依赖关系：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
TCP / WebSocket
      |
      v
MessageReader / MessageWriter
      |
      v
ChannelSession / ChannelApprovalHandler
      |
      +--> Reporter: TerminalReporter / JSONReporter
      |
      v
Runtime / Session / Task / AgentEngine
      |
      v
Tools / Policy / GrantStore

这套分层的核心价值是：

  • Transport 只负责连接，不负责 Agent 业务。
  • Protocol 只负责消息格式，不负责工具权限。
  • Channel 只负责把外部消息翻译成 Runtime 调用，把内部事件翻译成外部消息。
  • Runtime 负责 Session、Task、审批和 Agent 生命周期。
  • Reporter 负责不同渠道的输出表达。
  • Web Console 只是一个客户端，不会反向侵入核心引擎。

因此，增加 WebSocket 并不是复制一套 Agent 逻辑，而是增加一个新的 Transport 和一个新的入口适配器。未来接入飞书、HTTP API、MCP 或 A2A 时，也应沿着同样的方向扩展，而不是把渠道判断散落到 Engine、Tool 和 Task 中。

结语

从最初的单次命令行调用，到现在的 Task、Runtime、Manager、TCP 多连接、JSON Protocol、结构化事件、审批闭环和 WebSocket 控制台，go-tiny-claw 已经从一个 Agent Demo 变成了一个具备连接层和运行时边界的 Harness 原型。

但“能连接”不等于“能生产”。后续工作应优先补齐身份、协议安全、超时和恢复，再做 MCP 工具生态与 A2A 多 Agent 协作。只有先把连接、状态、事件和授权的边界稳定下来，外部协议越多，系统才不会越混乱。