feat(monkeyai): 实现单进程端点 WebSocket 桥接 (#1295)

Author: yokowu

monkeyai/.env.example

@@ -5,6 +5,8 @@ MONKEYAI_INITIAL_ADMIN_PASSWORD=change-this-admin-password
 # 修改对外端口时，同时修改统一对外访问地址。
 # MONKEYAI_ADMIN_PORT=8080
 # MONKEYAI_PUBLIC_URL=http://localhost:8080
+# 桥接仅支持一个后端进程；按内存预算设置连接上限（默认 1000）。
+# MONKEYAI_ENDPOINT_MAX_CONNECTIONS=1000
 
 # 自动域名证书：默认关闭；启用表示同意所配置 CA 的服务条款。
 # 域名须指向当前服务器，公网 80 和 443 端口可达；仅支持单域名 HTTP-01。

monkeyai/admin/nginx/app.conf

@@ -1,6 +1,19 @@
 root /usr/share/nginx/html;
 index index.html;
 
+location = /api/v1/endpoints/connect {
+    proxy_pass http://monkeyai_backend;
+    proxy_http_version 1.1;
+    proxy_set_header Upgrade $http_upgrade;
+    proxy_set_header Connection "upgrade";
+    proxy_set_header Host $http_host;
+    proxy_set_header X-Real-IP $remote_addr;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header X-Forwarded-Host $http_host;
+    proxy_set_header X-Forwarded-Proto $scheme;
+    proxy_buffering off;
+}
+
 location ~ ^/(?:api|oauth|v1|mcp)(?:/|$) {
     proxy_pass http://monkeyai_backend;
     proxy_http_version 1.1;

monkeyai/admin/vite.config.ts

@@ -10,6 +10,7 @@ export default defineConfig({
   plugins: [react(), tailwindcss()],
   server: {
     proxy: {
+      "^/api/v1/endpoints/connect$": { target: backend, ws: true },
       "/api": backend,
       "/oauth": backend,
       "/mcp": backend,

monkeyai/backend/README.md

@@ -144,6 +144,20 @@ go run ./cmd/server
 
 服务提供 `/healthz` 存活检查和 `/readyz` 数据库及 RustFS Bucket 就绪检查。pprof 默认单独监听 `127.0.0.1:6060`，入口为 `/debug/pprof/`，不对业务端口暴露。
 
+## 端点桥接
+
+`/api/v1/endpoints` 提供当前用户的端点管理，`/api/v1/endpoints/connect` 使用原生客户端 OAuth Bearer 升级为 WebSocket；不接受 Cookie、调用密钥或 URL token。完整协议见 [单进程端点桥接设计](../design/endpoint-bridge-design.md)，HTTP 契约见 `api/agent.yaml`。
+
+只支持一个后端进程：PostgreSQL 保存端点资料，在线连接、队列与路由仅在内存中；不引入 Redis。发布须先停止旧进程再启动新进程，禁止双进程滚动接入。进程重启不恢复在线状态和消息，客户端重新握手并查询业务状态；超时或断线的请求结果未知，不自动重发。
+
+生产使用 WSS，后端端口只对可信入口开放。非空 Origin 必须匹配 `MONKEYAI_PUBLIC_URL`，原生客户端可以省略。Nginx 和 Vite 已对桥接入口启用 Upgrade，代理空闲超时需大于心跳及等待窗口，默认 90 秒。应用退出时主动关闭桥接连接并使 `/readyz` 返回不可用。
+
+`MONKEYAI_ENDPOINT_MAX_CONNECTIONS`（或 `-endpoint-max-connections`）默认 1000，包含等待 hello 的连接；每连接业务队列最多 64 条/2 MiB，需按内存预算下调连接数。用户最多 20 个 active 端点，离线仍计入。停用端点不是撤销设备登录授权；持有有效账号凭据可显式恢复。
+
+access token 刷新后旧连接凭据失效，客户端应通过统一 OAuth 管理器轮换并重新连接。服务端每 30 秒复验一次，超时 5 秒；已知到期时间直接限制路由，数据库撤销或用户停用最多约 35 秒收敛。桥接不定义会话控制、文件操作等 Agent 业务方法。
+
+纯协议测试运行 `go test ./internal/endpoint/...`；设置隔离的 `MONKEYAI_TEST_DATABASE_URL` 后，同一命令还覆盖真实数据库与 WebSocket，并可加 `-race` 验证并发路径。
+
 ## 资源功能验证
 
 `go test ./...` 和 `go vet ./...` 运行常规检查。真实 PostgreSQL / RustFS 集成验证需要：

monkeyai/backend/api/agent.yaml

@@ -7,6 +7,112 @@ servers:
 security:
   - OAuthAccessToken: []
 paths:
+  /api/v1/endpoints:
+    get:
+      summary: 分页读取当前用户端点
+      description: 包含已停用端点；online 仅代表当前单进程连接。响应不缓存。
+      parameters:
+        - {name: page, in: query, schema: {type: integer, minimum: 1, maximum: 1000000, default: 1}}
+        - {name: page_size, in: query, schema: {type: integer, minimum: 1, maximum: 100, default: 20}}
+      responses:
+        '200':
+          description: 端点分页
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [items, total, page, page_size]
+                properties:
+                  items: {type: array, items: {$ref: '#/components/schemas/Endpoint'}}
+                  total: {type: integer}
+                  page: {type: integer}
+                  page_size: {type: integer}
+        '400': {$ref: '#/components/responses/ResourceError400'}
+        '401': {$ref: '#/components/responses/Unauthorized'}
+        '503': {$ref: '#/components/responses/EndpointUnavailable'}
+  /api/v1/endpoints/connect:
+    get:
+      summary: 建立原生客户端端点桥接 WebSocket
+      description: >-
+        仅接受 OAuth Bearer access token，不接受 Cookie 或 URL token。
+        非空 Origin 必须匹配 MONKEYAI_PUBLIC_URL；生产使用 WSS。
+        Upgrade 后 5 秒内发送 hello，服务端依次发送 welcome、directory.snapshot。
+        使用协议主版本 1，UTF-8 JSON 文本消息最大 256 KiB（含分片累计），禁用压缩。
+        request/response/event 信封及关闭码详见 design/endpoint-bridge-design.md。
+        4003 需恢复 OAuth 凭据，4001/4002 不自动重连；断线不得自动重发业务消息。
+      responses:
+        '101': {description: WebSocket 升级成功}
+        '400': {$ref: '#/components/responses/ResourceError400'}
+        '401': {$ref: '#/components/responses/Unauthorized'}
+        '403': {$ref: '#/components/responses/ResourceError403'}
+        '429':
+          description: 当前用户建连过于频繁，遵循 Retry-After 退避
+          headers:
+            Retry-After: {schema: {type: integer}}
+          content:
+            application/json:
+              schema: {$ref: '#/components/schemas/ResourceError'}
+        '503': {$ref: '#/components/responses/EndpointUnavailable'}
+  /api/v1/endpoints/{machine_id}:
+    parameters:
+      - {$ref: '#/components/parameters/EndpointMachine'}
+    get:
+      summary: 读取当前用户的单个端点
+      responses:
+        '200': {$ref: '#/components/responses/EndpointResult'}
+        '400': {$ref: '#/components/responses/ResourceError400'}
+        '401': {$ref: '#/components/responses/Unauthorized'}
+        '404': {$ref: '#/components/responses/ResourceError404'}
+        '503': {$ref: '#/components/responses/EndpointUnavailable'}
+    patch:
+      summary: 修改端点别名
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: false
+              required: [alias]
+              properties:
+                alias:
+                  type: [string, 'null']
+                  description: null 清除别名；非空白文本最多 128 UTF-8 字节，拒绝控制字符
+      responses:
+        '200': {$ref: '#/components/responses/EndpointResult'}
+        '400': {$ref: '#/components/responses/ResourceError400'}
+        '401': {$ref: '#/components/responses/Unauthorized'}
+        '404': {$ref: '#/components/responses/ResourceError404'}
+        '503': {$ref: '#/components/responses/EndpointUnavailable'}
+  /api/v1/endpoints/{machine_id}/revoke:
+    parameters:
+      - {$ref: '#/components/parameters/EndpointMachine'}
+    post:
+      summary: 停用端点并断开连接
+      description: 幂等，不接受请求体。不等于撤销设备登录授权，持有有效账号凭据仍可显式恢复。
+      responses:
+        '200': {$ref: '#/components/responses/EndpointResult'}
+        '400': {$ref: '#/components/responses/ResourceError400'}
+        '401': {$ref: '#/components/responses/Unauthorized'}
+        '404': {$ref: '#/components/responses/ResourceError404'}
+        '503': {$ref: '#/components/responses/EndpointUnavailable'}
+  /api/v1/endpoints/{machine_id}/restore:
+    parameters:
+      - {$ref: '#/components/parameters/EndpointMachine'}
+    post:
+      summary: 恢复已停用端点
+      description: 幂等，不接受请求体。不主动建立连接；active 端点最多 20 个，离线也计入。
+      responses:
+        '200': {$ref: '#/components/responses/EndpointResult'}
+        '400': {$ref: '#/components/responses/ResourceError400'}
+        '401': {$ref: '#/components/responses/Unauthorized'}
+        '404': {$ref: '#/components/responses/ResourceError404'}
+        '409':
+          description: active 端点达到上限，错误码 endpoint_limit_exceeded
+          content:
+            application/json:
+              schema: {$ref: '#/components/schemas/ResourceError'}
+        '503': {$ref: '#/components/responses/EndpointUnavailable'}
   /oauth/connectors/{id}/callback:
     get:
       summary: 处理指定 MCP 的 OAuth 回调
@@ -2540,6 +2646,12 @@ components:
       bearerFormat: OAuth 2.0 access token
       description: Agent 登录授权流程签发的短期 access token。
   parameters:
+    EndpointMachine:
+      name: machine_id
+      in: path
+      required: true
+      description: 规范小写、带连字符的 UUIDv4，仅限当前用户端点
+      schema: {type: string, format: uuid}
     ResourceID:
       name: id
       in: path
@@ -2583,6 +2695,18 @@ components:
         type: string
         const: private, no-cache
   responses:
+    EndpointResult:
+      description: 当前端点资料，时间为 Unix 毫秒
+      headers:
+        Cache-Control: {schema: {type: string, const: 'private, no-store'}}
+      content:
+        application/json:
+          schema: {$ref: '#/components/schemas/Endpoint'}
+    EndpointUnavailable:
+      description: 认证或存储暂不可用、连接数超限，或服务正在停止
+      content:
+        application/json:
+          schema: {$ref: '#/components/schemas/ResourceError'}
     NotModified:
       description: 当前接口内容未变化，无响应体。
       headers:
@@ -2669,6 +2793,25 @@ components:
           schema:
             $ref: "#/components/schemas/ErrorResponse"
   schemas:
+    Endpoint:
+      type: object
+      required: [machine_id, device_name, alias, display_name, platform, os_version, arch, client_version, protocol_version, status, online, created_at, updated_at, last_seen_at, revoked_at]
+      properties:
+        machine_id: {type: string, format: uuid}
+        device_name: {type: string}
+        alias: {type: [string, 'null']}
+        display_name: {type: string}
+        platform: {type: string, enum: [macos, windows, linux, ios, android]}
+        os_version: {type: string}
+        arch: {type: string}
+        client_version: {type: string}
+        protocol_version: {type: integer, const: 1}
+        status: {type: string, enum: [active, revoked]}
+        online: {type: boolean}
+        created_at: {type: integer, format: int64}
+        updated_at: {type: integer, format: int64}
+        last_seen_at: {type: [integer, 'null'], format: int64}
+        revoked_at: {type: [integer, 'null'], format: int64}
     UserRuleInput:
       type: object
       properties:

monkeyai/backend/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.33.3
 	github.com/aws/aws-sdk-go-v2/credentials v1.20.3
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.111.0
+	github.com/coder/websocket v1.8.14
 	github.com/go-chi/chi/v5 v5.3.2
 	github.com/jackc/pgx/v5 v5.10.0
 	gopkg.in/yaml.v3 v3.0.1

monkeyai/backend/go.sum

@@ -36,6 +36,8 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.49.0 h1:N7Ey8obY3uSui+cxl0OUzFlFmkxS
 github.com/aws/aws-sdk-go-v2/service/sts v1.49.0/go.mod h1:zMBwjSf4Pt8a1OHYiZ5rPD0PJRK1kQrUaxhS/Dbld8E=
 github.com/aws/smithy-go v1.28.1 h1:R/nXH00c8qcfCzQVELtRw+eLQWtzv+VAIEFJ1/xxXlQ=
 github.com/aws/smithy-go v1.28.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
+github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
+github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

monkeyai/backend/internal/app/app.go

@@ -16,6 +16,7 @@ import (
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/billing"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/config"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/database"
+	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/endpoint"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/expert"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/group"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/httpapi"
@@ -39,6 +40,7 @@ type App struct {
 	shutdownTimeout time.Duration
 	billing         *billing.Service
 	proxy           *proxy.Proxy
+	endpoints       *endpoint.Service
 }
 
 func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*App, error) {
@@ -73,6 +75,7 @@ func New(ctx context.Context, cfg config.Config, logger *slog.Logger) (*App, err
 		},
 		billing:         handler.(*applicationHandler).billing,
 		proxy:           handler.(*applicationHandler).proxy,
+		endpoints:       handler.(*applicationHandler).endpoints,
 		database:        pool,
 		shutdownTimeout: cfg.ShutdownTimeout,
 	}, nil
@@ -145,6 +148,8 @@ func newApplicationHandler(ctx context.Context, logger *slog.Logger, pool *pgxpo
 
 	agent := chi.NewRouter()
 	agent.Use(identities.RequireAgent)
+	endpoints := endpoint.NewService(endpoint.NewPostgres(pool), endpointAuth{identities}, logger, cfg.PublicURL).WithMaxConnections(cfg.EndpointMaxConnections)
+	endpoints.RegisterAgent(agent)
 	identities.RegisterAgent(agent)
 	keys.RegisterAgent(agent)
 	models.RegisterAgent(agent)
@@ -177,8 +182,8 @@ func newApplicationHandler(ctx context.Context, logger *slog.Logger, pool *pgxpo
 			return audit.Actor{}
 		}
 	})(identities.AuthRouter())
-	router.Mount("/", httpapi.New(logger, readiness{pool: pool, storage: storage}, admin, agent, auth))
-	return &applicationHandler{Handler: router, billing: charges, proxy: modelProxy}, nil
+	router.Mount("/", httpapi.New(logger, readiness{pool: pool, storage: storage, endpoints: endpoints}, admin, agent, auth))
+	return &applicationHandler{Handler: router, billing: charges, proxy: modelProxy, endpoints: endpoints}, nil
 }
 
 type modelResolver struct {
@@ -205,7 +210,9 @@ func (a *App) Run(ctx context.Context) error {
 	workerCtx, stopWorker := context.WithCancel(context.Background())
 	workerDone := make(chan struct{})
 	go func() { defer close(workerDone); a.billing.Run(workerCtx) }()
-	defer func() { stopWorker(); <-workerDone }()
+	observeDone := make(chan struct{})
+	go func() { defer close(observeDone); a.endpoints.Observe(workerCtx) }()
+	defer func() { stopWorker(); <-workerDone; <-observeDone }()
 
 	result := make(chan error, len(a.servers))
 	for _, server := range a.servers {
@@ -231,6 +238,11 @@ func (a *App) Run(ctx context.Context) error {
 
 	shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), a.shutdownTimeout)
 	defer cancel()
+	bridgeCtx, bridgeCancel := context.WithTimeout(shutdownCtx, 10*time.Second)
+	if err := a.endpoints.Shutdown(bridgeCtx); err != nil {
+		runErrors = append(runErrors, fmt.Errorf("关闭端点桥接: %w", err))
+	}
+	bridgeCancel()
 	for _, server := range a.servers {
 		if err := server.Shutdown(shutdownCtx); err != nil {
 			runErrors = append(runErrors, fmt.Errorf("关闭 %s: %w", server.Addr, err))
@@ -248,11 +260,15 @@ func (a *App) Run(ctx context.Context) error {
 }
 
 type readiness struct {
-	pool    *pgxpool.Pool
-	storage resource.Storage
+	pool      *pgxpool.Pool
+	storage   resource.Storage
+	endpoints *endpoint.Service
 }
 
 func (r readiness) Ping(ctx context.Context) error {
+	if r.endpoints != nil && !r.endpoints.Ready() {
+		return errors.New("端点桥接正在停止")
+	}
 	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 	if err := r.pool.Ping(ctx); err != nil {

monkeyai/backend/internal/app/billing.go

@@ -7,15 +7,17 @@ import (
 	"net/http"
 
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/billing"
+	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/endpoint"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/mcp"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/proxy"
 	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resource"
 )
 
 type applicationHandler struct {
 	http.Handler
-	billing *billing.Service
-	proxy   *proxy.Proxy
+	billing   *billing.Service
+	proxy     *proxy.Proxy
+	endpoints *endpoint.Service
 }
 type modelBilling struct{ service *billing.Service }
 

monkeyai/backend/internal/app/endpoint.go

@@ -0,0 +1,20 @@
+package app
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/endpoint"
+	"github.com/chaitin/MonkeyCode/monkeyai/backend/internal/identity"
+)
+
+type endpointAuth struct{ service *identity.Service }
+
+func (a endpointAuth) Credential(r *http.Request) (endpoint.Credential, bool) {
+	c, ok := identity.CredentialFromContext(r.Context())
+	return endpoint.Credential{UserID: c.UserID, Reference: c.Reference, ExpiresAt: c.ExpiresAt}, ok
+}
+
+func (a endpointAuth) Valid(ctx context.Context, c endpoint.Credential) (bool, error) {
+	return a.service.ValidCredential(ctx, identity.AccessCredential{UserID: c.UserID, Reference: c.Reference, ExpiresAt: c.ExpiresAt})
+}