ai-workspace-lab/xworkmate-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(acp): split internal server handlers and add opencode adapter

Browse Source
This commit is contained in:
Haitao Pan 2026-04-23 11:43:38 +08:00
parent 0562b99039
commit f7acb2fda5
28 changed files with 2285 additions and 2046 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,3 +1,4 @@
build/
.env
xworkmate-bridge
xworkmate-go-core-linux

204
docs/acp-public-validation-2026-04-09.md
View File

@ -1,185 +1,83 @@
# ACP Public Validation - 2026-04-09
# ACP Public Validation & Expansion Planning - 2026-04-09
This document records the post-deployment validation of the bridge public
origin at `xworkmate-bridge.svc.plus` and the independent upstream ACP ingress
at `xworkmate-bridge.svc.plus/acp-server`.
origin at `xworkmate-bridge.svc.plus` and outlines the expansion architecture
for the independent upstream ACP adapters.
For APP integration, the canonical public contract remains the bridge origin
and the `.../acp/rpc` path on that origin. The direct `xworkmate-bridge.svc.plus/acp-server`
URLs in this document are upstream validation targets, not the preferred APP
entry points.
## Expansion Modes Planning
## Verified Public Endpoints
To support a diverse set of backend providers, the bridge operates in the following expansion modes:
### Bridge root
| Mode ID | Adapter Role | Implementation Type |
| :--- | :--- | :--- |
| `acp-adapter-codex` | Codex ACP Adapter | Protocol Translator / Forwarder |
| `acp-adapter-opencode` | OpenCode ACP Adapter | JSON-RPC over stdio |
| `acp-adapter-gemini` | Gemini ACP Adapter | JSON-RPC over stdio |
| `acp-adapter-hermes` | Hermes ACP Adapter | JSON-RPC over stdio |
| `gateway-adapter-openclaw` | OpenClaw Gateway | Unified Protocol Entry |
- URL: `https://xworkmate-bridge.svc.plus/`
- Result: `200 OK`
- Body: `xworkmate-bridge is running`
## Protocol Entry Points (Public)
### ACP public ingress
The canonical entry points for external integrations are segmented by provider:
The public ACP JSON-RPC endpoint is the `.../acp/rpc` path.
* **Codex**: `https://xworkmate-bridge.svc.plus/acp-server/codex`
* **Gemini**: `https://xworkmate-bridge.svc.plus/acp-server/gemini`
* **Hermes**: `https://xworkmate-bridge.svc.plus/acp-server/hermes`
* **OpenCode**: `https://xworkmate-bridge.svc.plus/acp-server/opencode`
* **OpenClaw**: `https://xworkmate-bridge.svc.plus/gateway/openclaw`
Do not send JSON-RPC requests to `.../acp` for HTTP clients.
## Request Chain & Runtime Design
Recommended APP-facing endpoint:
### Traffic Flow
`Caddy (Ingress)` -> `xworkmate-bridge (Dispatcher)` -> `Adapter Service`
- `https://xworkmate-bridge.svc.plus/acp/rpc`
Caddy handles SSL termination and forwards requests to the `xworkmate-bridge` process, which performs path-based routing to the respective local adapter services.
Verified public HTTP JSON-RPC endpoints:
### Systemd Services & Local Mappings
- Codex: `https://xworkmate-bridge.svc.plus/acp-server/codex/acp/rpc`
- OpenCode: `https://xworkmate-bridge.svc.plus/acp-server/opencode/acp/rpc`
- Gemini: `https://xworkmate-bridge.svc.plus/acp-server/gemini/acp/rpc`
- Hermes: `https://xworkmate-bridge.svc.plus/acp-server/hermes/acp/rpc`
- OpenClaw: `https://xworkmate-bridge.svc.plus/gateway/openclaw/`
Each adapter is managed as a standalone systemd service, mapped to a specific local port/protocol:
The `.../acp` path remains reserved for WebSocket ACP.
| Service Name | Local Endpoint | Adapter Target |
| :--- | :--- | :--- |
| `acp-codex.service` | `ws://127.0.0.1:9001` | Codex Engine |
| `acp-opencode.service` | `ws://127.0.0.1:38992` | OpenCode Runtime |
| `acp-gemini.service` | `ws://127.0.0.1:8791` | Gemini Bridge |
| `acp-hermes.service` | `ws://127.0.0.1:3920` | Hermes Engine |
| `(Host Process)` | `ws://127.0.0.1:18789` | OpenClaw (Shared Runtime) |
## Auth Contract
All verified public ACP HTTP requests used:
All public ACP requests require:
- header: `Authorization: Bearer <INTERNAL_SERVICE_TOKEN>`
- header: `Content-Type: application/json`
- header: `Authorization: Bearer <INTERNAL_SERVICE_TOKEN>`
- header: `Content-Type: application/json`
Missing bearer auth returns a JSON-RPC error envelope with code `-32001`.
Missing or invalid bearer auth returns a JSON-RPC error envelope with code `-32001`.
## Public Validation Results
## Validation Results (2026-04-09)
The ingress returned `200 OK` on all public routes after re-apply, and the deployment response confirmed the active upstream mappings:
- `codex` -> `127.0.0.1:9001`
- `opencode` -> `127.0.0.1:38992`
- `gemini` -> `127.0.0.1:8791`
- `hermes` -> `127.0.0.1:3920`
- `openclaw` -> `127.0.0.1:18789` (Host process)
The ingress returned `200 OK` on all public routes after re-apply.
### Codex
Verified `acp.capabilities` over the public ingress:
```json
{
"method": "acp.capabilities",
"result": {
"providers": ["codex", "gemini", "opencode"],
"singleAgent": true,
"multiAgent": true
}
}
```
Verified end-to-end task execution over the public ingress.
Observed conversation behavior:
- `session.start` succeeded and returned `round1`
- `session.message` also succeeded and returned `round2`
- `session.message` must include the same `routing` payload as `session.start`
- omitting `routing` returns `ROUTING_REQUIRED`
- Verified `acp.capabilities`: `["codex", "gemini", "opencode"]`
- Two-turn conversation (`session.start` -> `session.message`) passed.
### OpenCode
Verified `acp.capabilities` over the public ingress:
```json
{
"method": "acp.capabilities",
"result": {
"providers": ["opencode"],
"singleAgent": true,
"multiAgent": true
}
}
```
Verified `session.start` end to end with prompt `Reply with exactly pong`.
Observed result:
```json
{
"success": true,
"provider": "opencode",
"output": "pong"
}
```
Observed conversation behavior:
- `session.start` succeeded and returned `round1`
- `session.message` also succeeded and returned `round2`
- `session.message` must include the same `routing` payload as `session.start`
- omitting `routing` returns `ROUTING_REQUIRED`
- Validated as WebSocket ACP upstream at `ws://127.0.0.1:38992/acp`.
- Two-turn conversation passed.
### Gemini
Verified `acp.capabilities` over the public ingress:
```json
{
"method": "acp.capabilities",
"result": {
"providers": ["gemini"],
"singleAgent": true,
"multiAgent": false
}
}
```
Before the compatibility layer landed, the upstream Gemini ACP returned:
```json
{
"success": false,
"error": "\"Method not found\": session.start"
}
```
The adapter has now been updated so `session.start` and `session.message` default to adapter-local prompt compatibility instead of forwarding unsupported upstream methods.
Observed conversation behavior after re-apply:
- `session.start` succeeded and returned `round1`
- `session.message` succeeded and returned `round2`
- long conversation validation passed through the public ingress
## Long Conversation Validation
All three public ACP agent entries now pass a two-turn conversation check:
1. `session.start`
2. `session.message`
Verified result summary:
- `codex` long conversation passed
- `opencode` long conversation passed
- `gemini` long conversation passed
This confirms the upstream ACP baseline. The APP-facing baseline remains
`https://xworkmate-bridge.svc.plus/acp/rpc`.
- Verified `acp.capabilities`: `["gemini"]`
- Adapter-local prompt compatibility layer enables `session.start` / `session.message` despite lack of native upstream support.
- Two-turn conversation passed.
## App Integration Notes
### Recommended request shape
For APP integration, use JSON-RPC `POST` requests against
`https://xworkmate-bridge.svc.plus/acp/rpc`.
Use JSON-RPC `POST` requests against `https://xworkmate-bridge.svc.plus/acp/rpc` for general usage, or the specific provider endpoints for targeted execution.
For capability discovery:
```json
{
"jsonrpc": "2.0",
"id": "cap-1",
"method": "acp.capabilities"
}
```
For single-agent task execution:
**Example Task Execution:**
```json
{
@ -201,8 +99,6 @@ For single-agent task execution:
```
### Provider-specific notes
- `codex`, `opencode`, `gemini`, and `hermes` are all now verified public task paths.
- `gemini` still depends on the adapter compatibility layer, not a native upstream Gemini ACP conversation method.
- For multi-turn flows, apps should preserve and resend `routing` on every `session.message`.
- `codex` and `opencode` currently require explicit `routing` on follow-up turns.
- `codex` and `opencode` require explicit `routing` on follow-up turns.
- `gemini` uses a prompt-compatibility layer for multi-turn support.
- `hermes` is verified as a public task path.

17
docs/architecture/acp-forwarding-topology.md
View File

@ -144,8 +144,7 @@ Important distinction:
`openclaw`
- `availableExecutionTargets` tells the app which first-level task dialog modes
are currently available
- for `gatewayProviderId=openclaw`, the bridge rewrites the upstream target to
`https://xworkmate-bridge.svc.plus/gateway/openclaw/`
## Production Truth
当前 production forwarding 事实（内部直连架构）：
@ -157,15 +156,15 @@ Important distinction:
### 核心真源映射 (Final Source of Truth)
为了消除冗余层（Bridge-on-Bridge）并提高就绪性响应速度，中心 Bridge 已配置为绕过旧的 9010/3910 转发层，直接对接各核心服务端口：
为了消除冗余层（Bridge-on-Bridge）并提高就绪性响应速度，中心 Bridge 作为代理和适配器，直接对接各核心服务端口：
| 服务名 | 核心端口 | 协议路径 | 角色定义 |
| :--- | :--- | :--- | :--- |
| **`acp-codex.service`** | **`9001`** | **`ws://127.0.0.1:9001/acp/rpc`** | **Codex 核心 ACP 实现** |
| **`acp-opencode.service`** | **`38992`** | **`ws//127.0.0.1:38992/acp/rpc`** | **Opencode 协议转换 (JSON-RPC over stdio) |
| `acp-gemini.service`** | **`8791`** | **`ws://127.0.0.1:8791/acp/rpc`** | **Gemini 协议转换适配器 (JSON-RPC over stdio)** |
| **`acp-hermes.service`** | **`3920`** | **`ws://127.0.0.1:3920/acp/rpc`** | **Hermes 协议转换适配器 (JSON-RPC over stdio)** |
| **`openclaw-gateway.service`** | **`18789`** | **`ws://127.0.0.1:18789/`** | **OpenClaw 独立部署网关服务（不使用 /acp）** |
| **`acp-codex.service`** | **`9001`** | **`http://127.0.0.1:9001/acp/rpc`** | **Codex 核心 ACP 实现** |
| **`acp-opencode.service`** | **`38992`** | **`http://127.0.0.1:38992/acp/rpc`** | **Opencode 核心 ACP 实现** |
| **`acp-gemini.service`** | **`8791`** | **`http://127.0.0.1:8791/acp/rpc`** | **Gemini 协议转换适配器 (Category: protocol-adapter)** |
| **`acp-hermes.service`** | **`3920`** | **`http://127.0.0.1:3920/acp/rpc`** | **Hermes 协议转换适配器 (Category: protocol-adapter)** |
对 app 而言：
@ -177,10 +176,10 @@ Important distinction:
- app traffic reaches upstream ACP and gateway services only through the bridge
- app does not call `xworkmate-bridge.svc.plus/acp-server/*` or `xworkmate-bridge.svc.plus/gateway/openclaw/` directly
- `openclaw-gateway` is an independently deployed runtime mapped to `127.0.0.1:18789`
- `openclaw-gateway` is an independently deployed runtime mapped to `ws://127.0.0.1:18789`
- internal provider routes remain bridge-owned validation targets:
- `xworkmate-bridge.svc.plus/acp-server/codex/acp/rpc`
- `xworkmate-bridge.svc.plus/acp-server/opencode/acp/rpc`
- `xworkmate-bridge.svc.plus/acp-server/opencode/acp`
- `xworkmate-bridge.svc.plus/acp-server/gemini/acp/rpc`
- `xworkmate-bridge.svc.plus/acp-server/hermes/acp/rpc`
- upstream auth stays bridge-internal:

8
example/config.yaml
View File

@ -6,11 +6,11 @@
# Upstream provider endpoints
# Priority: YAML > Environment Variable (e.g. CODEX_RPC_URL) > Default Constants
upstream:
gateway_url: "ws://127.0.0.1:18789/"
codex_url: "ws://127.0.0.1:9001/acp"
opencode_url: "http://127.0.0.1:38992"
gemini_url: "http://127.0.0.1:8791"
codex_url: "ws://127.0.0.1:9001"
gemini_url: "ws://127.0.0.1:8791"
hermes_url: "ws://127.0.0.1:3920"
gateway_url: "ws://127.0.0.1:18789/"
opencode_url: "ws://127.0.0.1:38992"
# Legacy/Reference structure (Normally managed via code constants or environment)
bridge:

20
internal/acp/execution.go
View File

@ -111,7 +111,7 @@ func (s *Server) runGateway(
},
}
}
payload := asMap(result.Payload)
payload := shared.AsMap(result.Payload)
if len(payload) == 0 {
payload = map[string]any{
"success": true,
@ -162,7 +162,7 @@ func (s *Server) runSingleAgentViaExternalProvider(
if err != nil {
return nil, err
}
result := asMap(response["result"])
result := shared.AsMap(response["result"])
if len(result) == 0 {
result = response
}
@ -193,7 +193,7 @@ func (s *Server) probeExternalProvider(
if err != nil {
return nil, err
}
result := asMap(response["result"])
result := shared.AsMap(response["result"])
if len(result) == 0 {
return nil, fmt.Errorf("external provider probe missing result payload")
}
@ -301,7 +301,7 @@ func requestExternalACPHTTP(
if err := json.NewDecoder(response.Body).Decode(&decoded); err != nil {
return nil, fmt.Errorf("failed to decode external ACP response: %w", err)
}
if errPayload := asMap(decoded["error"]); len(errPayload) > 0 {
if errPayload := shared.AsMap(decoded["error"]); len(errPayload) > 0 {
return nil, fmt.Errorf(
"%s",
strings.TrimSpace(shared.StringArg(errPayload, "message", "external ACP request failed")),
@ -333,7 +333,7 @@ func (c *externalACPNotificationCollector) observe(notification map[string]any)
if method != "session.update" && method != "acp.session.update" && method != "session/update" {
return
}
params := asMap(notification["params"])
params := shared.AsMap(notification["params"])
if len(params) == 0 {
return
}
@ -410,11 +410,11 @@ func extractExternalACPNotificationText(notification map[string]any) string {
if notification == nil {
return ""
}
payload := asMap(notification["params"])
payload := shared.AsMap(notification["params"])
if len(payload) == 0 {
payload = notification
}
update := asMap(payload["update"])
update := shared.AsMap(payload["update"])
if len(update) == 0 {
update = payload
}
@ -482,14 +482,14 @@ func enrichSingleAgentResultArtifacts(result map[string]any, requestParams map[s
}
remoteWorkingDirectory := firstNonEmptyString(
shared.StringArg(result, "remoteWorkingDirectory", ""),
shared.StringArg(asMap(result["remoteExecution"]), "remoteWorkingDirectory", ""),
shared.StringArg(shared.AsMap(result["remoteExecution"]), "remoteWorkingDirectory", ""),
shared.StringArg(result, "resolvedWorkingDirectory", ""),
shared.StringArg(result, "effectiveWorkingDirectory", ""),
shared.StringArg(requestParams, "workingDirectory", ""),
)
remoteWorkspaceRefKind := firstNonEmptyString(
shared.StringArg(result, "remoteWorkspaceRefKind", ""),
shared.StringArg(asMap(result["remoteExecution"]), "remoteWorkspaceRefKind", ""),
shared.StringArg(shared.AsMap(result["remoteExecution"]), "remoteWorkspaceRefKind", ""),
"remotePath",
)
if strings.TrimSpace(shared.StringArg(result, "resultSummary", "")) == "" {
@ -694,7 +694,7 @@ func requestExternalACPWebSocket(
}
if strings.TrimSpace(shared.StringArg(payload, "id", "")) == requestID &&
(payload["result"] != nil || payload["error"] != nil) {
if errPayload := asMap(payload["error"]); len(errPayload) > 0 {
if errPayload := shared.AsMap(payload["error"]); len(errPayload) > 0 {
return nil, fmt.Errorf(
"%s",
strings.TrimSpace(shared.StringArg(errPayload, "message", "external ACP request failed")),

50
internal/acp/gateway_runtime.go
View File

@ -26,31 +26,31 @@ func handleGatewayConnect(
HasSharedAuth: parseBool(params["hasSharedAuth"]),
HasDeviceToken: parseBool(params["hasDeviceToken"]),
Endpoint: gatewayruntime.Endpoint{
Host: strings.TrimSpace(shared.StringArg(asMap(params["endpoint"]), "host", "")),
Port: parsePositiveInt(asMap(params["endpoint"])["port"]),
TLS: parseBool(asMap(params["endpoint"])["tls"]),
Host: strings.TrimSpace(shared.StringArg(shared.AsMap(params["endpoint"]), "host", "")),
Port: parsePositiveInt(shared.AsMap(params["endpoint"])["port"]),
TLS: parseBool(shared.AsMap(params["endpoint"])["tls"]),
},
PackageInfo: gatewayruntime.PackageInfo{
AppName: strings.TrimSpace(shared.StringArg(asMap(params["packageInfo"]), "appName", "")),
PackageName: strings.TrimSpace(shared.StringArg(asMap(params["packageInfo"]), "packageName", "")),
Version: strings.TrimSpace(shared.StringArg(asMap(params["packageInfo"]), "version", "")),
BuildNumber: strings.TrimSpace(shared.StringArg(asMap(params["packageInfo"]), "buildNumber", "")),
AppName: strings.TrimSpace(shared.StringArg(shared.AsMap(params["packageInfo"]), "appName", "")),
PackageName: strings.TrimSpace(shared.StringArg(shared.AsMap(params["packageInfo"]), "packageName", "")),
Version: strings.TrimSpace(shared.StringArg(shared.AsMap(params["packageInfo"]), "version", "")),
BuildNumber: strings.TrimSpace(shared.StringArg(shared.AsMap(params["packageInfo"]), "buildNumber", "")),
},
DeviceInfo: gatewayruntime.DeviceInfo{
Platform: strings.TrimSpace(shared.StringArg(asMap(params["deviceInfo"]), "platform", "")),
PlatformVersion: strings.TrimSpace(shared.StringArg(asMap(params["deviceInfo"]), "platformVersion", "")),
DeviceFamily: strings.TrimSpace(shared.StringArg(asMap(params["deviceInfo"]), "deviceFamily", "")),
ModelIdentifier: strings.TrimSpace(shared.StringArg(asMap(params["deviceInfo"]), "modelIdentifier", "")),
Platform: strings.TrimSpace(shared.StringArg(shared.AsMap(params["deviceInfo"]), "platform", "")),
PlatformVersion: strings.TrimSpace(shared.StringArg(shared.AsMap(params["deviceInfo"]), "platformVersion", "")),
DeviceFamily: strings.TrimSpace(shared.StringArg(shared.AsMap(params["deviceInfo"]), "deviceFamily", "")),
ModelIdentifier: strings.TrimSpace(shared.StringArg(shared.AsMap(params["deviceInfo"]), "modelIdentifier", "")),
},
Identity: gatewayruntime.DeviceIdentity{
DeviceID: strings.TrimSpace(shared.StringArg(asMap(params["identity"]), "deviceId", "")),
PublicKeyBase64URL: strings.TrimSpace(shared.StringArg(asMap(params["identity"]), "publicKeyBase64Url", "")),
PrivateKeyBase64URL: strings.TrimSpace(shared.StringArg(asMap(params["identity"]), "privateKeyBase64Url", "")),
DeviceID: strings.TrimSpace(shared.StringArg(shared.AsMap(params["identity"]), "deviceId", "")),
PublicKeyBase64URL: strings.TrimSpace(shared.StringArg(shared.AsMap(params["identity"]), "publicKeyBase64Url", "")),
PrivateKeyBase64URL: strings.TrimSpace(shared.StringArg(shared.AsMap(params["identity"]), "privateKeyBase64Url", "")),
},
Auth: gatewayruntime.AuthConfig{
Token: strings.TrimSpace(shared.StringArg(asMap(params["auth"]), "token", "")),
DeviceToken: strings.TrimSpace(shared.StringArg(asMap(params["auth"]), "deviceToken", "")),
Password: strings.TrimSpace(shared.StringArg(asMap(params["auth"]), "password", "")),
Token: strings.TrimSpace(shared.StringArg(shared.AsMap(params["auth"]), "token", "")),
DeviceToken: strings.TrimSpace(shared.StringArg(shared.AsMap(params["auth"]), "deviceToken", "")),
Password: strings.TrimSpace(shared.StringArg(shared.AsMap(params["auth"]), "password", "")),
},
}
if request.Mode == "" {
@ -98,7 +98,7 @@ func handleGatewayRequest(
result := server.gateway.Request(
strings.TrimSpace(shared.StringArg(params, "runtimeId", "")),
strings.TrimSpace(shared.StringArg(params, "method", "")),
asMap(params["params"]),
shared.AsMap(params["params"]),
timeout,
notify,
)
@ -121,16 +121,6 @@ func handleGatewayDisconnect(
return map[string]any{"accepted": true}
}
func asMap(value any) map[string]any {
if typed, ok := value.(map[string]any); ok {
return typed
}
if typed, ok := value.(map[string]interface{}); ok {
return typed
}
return map[string]any{}
}
func parseGatewayRuntimeStringSlice(value any) []string {
list, ok := value.([]any)
if !ok {
@ -192,8 +182,8 @@ func resolveGatewayReportedRemoteAddress(
if strings.TrimSpace(strings.ToLower(request.Mode)) != "openclaw" {
return ""
}
_ = server
return publicEndpointAddressLabel(productionGatewayEndpointURL)
gatewayURL := resolveURL(server.config.Upstream.GatewayURL, "GATEWAY_RPC_URL")
return publicEndpointAddressLabel(gatewayURL)
}
func publicEndpointAddressLabel(raw string) string {

423
internal/acp/http_handler.go Normal file
View File

@ -0,0 +1,423 @@
package acp
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"path"
"strings"
"sync"
"xworkmate-bridge/internal/shared"
)
func (s *Server) Handler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if providerID, ok := parseProviderACPRPCPath(r.URL.Path); ok {
s.HandleProviderRPC(w, r, providerID)
return
}
if providerID, ok := parseProviderBarePath(r.URL.Path); ok {
s.HandleProviderAlias(w, r, providerID)
return
}
if strings.TrimSpace(r.URL.Path) == "/gateway/openclaw" {
s.HandleGatewayAlias(w, r)
return
}
switch r.URL.Path {
case "/":
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
_, _ = w.Write([]byte("xworkmate-bridge is running"))
case "/api/ping":
info := ParseImageVersionInfo(os.Getenv("IMAGE"))
resp := map[string]any{
"status": "ok",
"image": info.ImageRef,
"tag": info.Tag,
"commit": info.Commit,
"version": info.Version,
}
body, err := json.Marshal(resp)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_, _ = w.Write(body)
case "/bridge/bootstrap/health":
s.HandleBridgeBootstrapHealth(w, r)
case "/acp/rpc":
s.HandleRPC(w, r)
case "/acp":
s.HandleWebSocket(w, r)
case "/gateway/openclaw/acp/rpc":
s.HandleGatewayRPCAlias(w, r)
default:
http.NotFound(w, r)
}
})
}
func parseProviderBarePath(pathValue string) (string, bool) {
trimmed := strings.Trim(path.Clean(strings.TrimSpace(pathValue)), "/")
parts := strings.Split(trimmed, "/")
if len(parts) != 2 {
return "", false
}
if parts[0] != "acp-server" {
return "", false
}
switch parts[1] {
case "codex", "opencode", "gemini", "hermes":
return parts[1], true
default:
return "", false
}
}
func parseProviderACPRPCPath(path string) (string, bool) {
trimmed := strings.Trim(strings.TrimSpace(path), "/")
parts := strings.Split(trimmed, "/")
if len(parts) != 4 {
return "", false
}
if parts[0] != "acp-server" || parts[2] != "acp" || parts[3] != "rpc" {
return "", false
}
switch parts[1] {
case "codex", "opencode", "gemini", "hermes":
return parts[1], true
default:
return "", false
}
}
func (s *Server) HandleProviderAlias(w http.ResponseWriter, r *http.Request, providerID string) {
if r.Method == http.MethodGet {
s.writeAliasCapabilities(w, providerID, "agent")
return
}
s.HandleProviderRPC(w, r, providerID)
}
func (s *Server) HandleGatewayAlias(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodGet {
s.writeAliasCapabilities(w, "openclaw", "gateway")
return
}
s.HandleGatewayRPCAlias(w, r)
}
func (s *Server) writeAliasCapabilities(w http.ResponseWriter, providerID, target string) {
result, rpcErr := s.handleRequest(shared.RPCRequest{
JSONRPC: "2.0",
Method: "acp.capabilities",
Params: map[string]any{
"preferredExecutionTarget": target,
"preferredProviderId": providerID,
},
}, nil)
if rpcErr != nil {
shared.WriteJSONError(w, nil, http.StatusOK, rpcErr.Code, rpcErr.Message)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(w).Encode(shared.ResultEnvelope(nil, result))
}
func (s *Server) HandleGatewayRPCAlias(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/gateway/openclaw" && r.URL.Path != "/gateway/openclaw/acp/rpc" {
http.NotFound(w, r)
return
}
if r.Method == http.MethodGet {
r = r.Clone(r.Context())
r.URL.Path = "/acp/rpc"
s.HandleRPC(w, r)
return
}
s.HandleRPC(w, r)
}
func (s *Server) HandleProviderRPC(w http.ResponseWriter, r *http.Request, providerID string) {
if r.Method == http.MethodGet {
http.NotFound(w, r)
return
}
shared.ApplyCORS(w, r, s.allowedOrigins) // ACP uses configured allowed origins
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
if r.Method != http.MethodPost {
shared.WriteJSONError(
w,
nil,
http.StatusMethodNotAllowed,
-32600,
"method not allowed",
)
return
}
origin := strings.TrimSpace(r.Header.Get("Origin"))
if !shared.OriginAllowed(origin, s.allowedOrigins) {
shared.WriteJSONError(
w,
nil,
http.StatusForbidden,
-32003,
fmt.Sprintf("origin not allowed: %s", origin),
)
return
}
payload, err := io.ReadAll(r.Body)
if err != nil {
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
return
}
r.Body = io.NopCloser(bytes.NewBuffer(payload))
if !s.authorized(r) {
var temp struct {
Method string `json:"method"`
}
_ = json.Unmarshal(payload, &temp)
method := strings.TrimSpace(temp.Method)
if method != "acp.capabilities" && method != "health" {
shared.WriteJSONError(
w,
nil,
http.StatusUnauthorized,
-32001,
"missing bearer authorization",
)
return
}
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
return
}
params := request.Params
if params == nil {
params = map[string]any{}
}
params["routing"] = map[string]any{
"routingMode": "explicit",
"explicitExecutionTarget": "singleAgent",
"explicitProviderId": providerID,
}
request.Params = injectInboundAuthorizationHeader(
params,
r.Header.Get("Authorization"),
)
response, rpcErr := s.handleRequest(request, nil)
if request.ID == nil {
return
}
if rpcErr != nil {
shared.WriteJSONError(w, request.ID, http.StatusOK, rpcErr.Code, rpcErr.Message)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(w).Encode(shared.ResultEnvelope(request.ID, response))
}
func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
origin := strings.TrimSpace(r.Header.Get("Origin"))
if !shared.OriginAllowed(origin, s.allowedOrigins) {
shared.WriteJSONError(
w,
nil,
http.StatusForbidden,
-32003,
fmt.Sprintf("origin not allowed: %s", origin),
)
return
}
if !s.authorized(r) {
shared.WriteJSONError(
w,
nil,
http.StatusUnauthorized,
-32001,
"missing bearer authorization",
)
return
}
upgrader := shared.StandardWSUpgrader
upgrader.CheckOrigin = func(req *http.Request) bool {
return shared.OriginAllowed(req.Header.Get("Origin"), s.allowedOrigins) && s.authorized(req)
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}
defer func() {
_ = conn.Close()
}()
var writeMu sync.Mutex
notify := func(message map[string]any) {
writeMu.Lock()
defer writeMu.Unlock()
_ = conn.WriteJSON(message)
}
for {
_, payload, err := conn.ReadMessage()
if err != nil {
return
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
notify(shared.ErrorEnvelope(nil, -32700, err.Error()))
continue
}
request.Params = injectInboundAuthorizationHeader(
request.Params,
r.Header.Get("Authorization"),
)
response, rpcErr := s.handleRequest(request, notify)
if request.ID == nil {
continue
}
if rpcErr != nil {
notify(shared.ErrorEnvelope(request.ID, rpcErr.Code, rpcErr.Message))
continue
}
notify(shared.ResultEnvelope(request.ID, response))
}
}
func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
shared.ApplyCORS(w, r, s.allowedOrigins)
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
if r.Method != http.MethodPost {
shared.WriteJSONError(
w,
nil,
http.StatusMethodNotAllowed,
-32600,
"method not allowed",
)
return
}
origin := strings.TrimSpace(r.Header.Get("Origin"))
if !shared.OriginAllowed(origin, s.allowedOrigins) {
shared.WriteJSONError(
w,
nil,
http.StatusForbidden,
-32003,
fmt.Sprintf("origin not allowed: %s", origin),
)
return
}
payload, err := io.ReadAll(r.Body)
if err != nil {
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
return
}
r.Body = io.NopCloser(bytes.NewBuffer(payload))
if !s.authorized(r) {
var temp struct {
Method string `json:"method"`
}
_ = json.Unmarshal(payload, &temp)
method := strings.TrimSpace(temp.Method)
if method != "acp.capabilities" && method != "health" {
shared.WriteJSONError(
w,
nil,
http.StatusUnauthorized,
-32001,
"missing bearer authorization",
)
return
}
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
return
}
request.Params = injectInboundAuthorizationHeader(
request.Params,
r.Header.Get("Authorization"),
)
accept := strings.ToLower(r.Header.Get("Accept"))
stream := strings.Contains(accept, "text/event-stream")
if stream {
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
}
flusher, _ := w.(http.Flusher)
writeNotification := func(message map[string]any) {
if !stream {
return
}
shared.WriteSSE(w, message)
if flusher != nil {
flusher.Flush()
}
}
response, rpcErr := s.handleRequest(request, writeNotification)
if request.ID == nil {
if stream {
_, _ = w.Write([]byte("data: [DONE]\n\n"))
}
return
}
if rpcErr != nil {
envelope := shared.ErrorEnvelope(request.ID, rpcErr.Code, rpcErr.Message)
if stream {
shared.WriteSSE(w, envelope)
_, _ = w.Write([]byte("data: [DONE]\n\n"))
if flusher != nil {
flusher.Flush()
}
return
}
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(w).Encode(envelope)
return
}
if stream {
shared.WriteSSE(w, shared.ResultEnvelope(request.ID, response))
_, _ = w.Write([]byte("data: [DONE]\n\n"))
if flusher != nil {
flusher.Flush()
}
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(w).Encode(shared.ResultEnvelope(request.ID, response))
}
func (s *Server) authorized(r *http.Request) bool {
if s == nil {
return false
}
if s.authService == nil {
return true
}
return s.authService.ValidateAuthorizationHeader(r.Header.Get("Authorization"))
}

60
internal/acp/provider_catalog.go
View File

@ -8,11 +8,6 @@ import (
"xworkmate-bridge/internal/shared"
)
// Default production endpoints for XWorkmate managed bridge environment.
const (
productionGatewayEndpointURL = "ws://127.0.0.1:18789/"
)
type syncedProvider struct {
ProviderID string
Label string
@ -44,7 +39,7 @@ func loadBridgeConfig() *BridgeConfig {
return config
}
func resolveURL(yamlVal string, defaultVal string, envKeys ...string) string {
func resolveURL(yamlVal string, envKeys ...string) string {
val := strings.TrimSpace(yamlVal)
if val != "" {
return val
@ -54,7 +49,7 @@ func resolveURL(yamlVal string, defaultVal string, envKeys ...string) string {
return v
}
}
return defaultVal
return ""
}
func bridgeUpstreamAuthorizationHeader() string {
@ -65,44 +60,39 @@ func bridgeUpstreamAuthorizationHeader() string {
return token
}
func newProductionProviderCatalog() (map[string]syncedProvider, []string) {
func newProductionProviderCatalog() (*BridgeConfig, map[string]syncedProvider, []string) {
config := loadBridgeConfig()
authorizationHeader := bridgeUpstreamAuthorizationHeader()
providers := []struct {
id string
label string
yaml string
envKeys []string
defaultURL string
id string
label string
yaml string
envKeys []string
}{
{
id: "codex",
label: "Codex",
yaml: config.Upstream.CodexURL,
envKeys: []string{"CODEX_RPC_URL"},
defaultURL: "ws://127.0.0.1:9001/acp",
id: "codex",
label: "Codex",
yaml: config.Upstream.CodexURL,
envKeys: []string{"CODEX_RPC_URL"},
},
{
id: "opencode",
label: "OpenCode",
yaml: config.Upstream.OpenCodeURL,
envKeys: []string{"OPENCODE_RPC_URL"},
defaultURL: "http://127.0.0.1:38992",
id: "opencode",
label: "OpenCode",
yaml: config.Upstream.OpenCodeURL,
envKeys: []string{"OPENCODE_RPC_URL"},
},
{
id: "gemini",
label: "Gemini",
yaml: config.Upstream.GeminiURL,
envKeys: []string{"GEMINI_RPC_URL"},
defaultURL: "http://127.0.0.1:8791",
id: "gemini",
label: "Gemini",
yaml: config.Upstream.GeminiURL,
envKeys: []string{"GEMINI_RPC_URL"},
},
{
id: "hermes",
label: "Hermes",
yaml: config.Upstream.HermesURL,
envKeys: []string{"HERMES_RPC_URL"},
defaultURL: "ws://127.0.0.1:3920",
id: "hermes",
label: "Hermes",
yaml: config.Upstream.HermesURL,
envKeys: []string{"HERMES_RPC_URL"},
},
}
@ -110,7 +100,7 @@ func newProductionProviderCatalog() (map[string]syncedProvider, []string) {
var order []string
for _, p := range providers {
endpoint := resolveURL(p.yaml, p.defaultURL, p.envKeys...)
endpoint := resolveURL(p.yaml, p.envKeys...)
catalog[p.id] = syncedProvider{
ProviderID: p.id,
Label: p.label,
@ -121,7 +111,7 @@ func newProductionProviderCatalog() (map[string]syncedProvider, []string) {
order = append(order, p.id)
}
return catalog, order
return config, catalog, order
}
func (s *Server) syncedProviderByID(providerID string) (syncedProvider, bool) {

11
internal/acp/routing.go
View File

@ -7,6 +7,7 @@ import (
"xworkmate-bridge/internal/memory"
"xworkmate-bridge/internal/router"
"xworkmate-bridge/internal/shared"
"xworkmate-bridge/internal/skills"
)
@ -19,11 +20,12 @@ func resolveRoutingMetadataWithProviders(
params map[string]any,
availableProviders []string,
) (router.Result, bool) {
routingParams := asMap(params["routing"])
routingParams := shared.AsMap(params["routing"])
if len(routingParams) == 0 {
return router.Result{}, false
}
installApproval := asMap(routingParams["installApproval"])
installApproval := shared.AsMap(routingParams["installApproval"])
resolver := router.NewResolver()
result := resolver.Resolve(router.Request{
@ -83,7 +85,7 @@ func recordRoutingSuccess(
result router.Result,
response map[string]any,
) error {
routingParams := asMap(params["routing"])
routingParams := shared.AsMap(params["routing"])
if len(routingParams) == 0 {
return nil
}
@ -118,8 +120,9 @@ func parseRoutingSkillCandidates(raw any) []skills.Candidate {
}
candidates := make([]skills.Candidate, 0, len(list))
for _, item := range list {
entry := asMap(item)
entry := shared.AsMap(item)
candidates = append(candidates, skills.Candidate{
ID: strings.TrimSpace(sharedString(entry, "id")),
Label: strings.TrimSpace(sharedString(entry, "label")),
Description: strings.TrimSpace(sharedString(entry, "description")),

622
internal/acp/rpc_handler.go Normal file
View File

@ -0,0 +1,622 @@
package acp
import (
"context"
"fmt"
"os"
"strings"
"time"
"xworkmate-bridge/internal/dispatch"
"xworkmate-bridge/internal/mounts"
"xworkmate-bridge/internal/router"
"xworkmate-bridge/internal/shared"
)
func (s *Server) handleRequest(
request shared.RPCRequest,
notify func(map[string]any),
) (map[string]any, *shared.RPCError) {
method := strings.TrimSpace(request.Method)
switch method {
case "health":
return map[string]any{"status": "ok", "version": "0.7.0"}, nil
case "acp.capabilities":
providerCatalog := s.availableProviderCatalog()
gatewayProviders := availableGatewayProviderCatalog()
singleAgent := len(providerCatalog) > 0
availableExecutionTargets := availableExecutionTargets(
providerCatalog,
gatewayProviders,
)
multiAgent := shared.BoolArg(
shared.EnvOrDefault("ACP_MULTI_AGENT_ENABLED", "true"),
true,
)
result := map[string]any{
"singleAgent": singleAgent,
"multiAgent": multiAgent,
"availableExecutionTargets": availableExecutionTargets,
"providerCatalog": providerCatalog,
"gatewayProviders": gatewayProviders,
"capabilities": map[string]any{
"single_agent": singleAgent,
"multi_agent": multiAgent,
"availableExecutionTargets": availableExecutionTargets,
"providerCatalog": providerCatalog,
"gatewayProviders": gatewayProviders,
},
}
return result, nil
case "session.start", "session.message":
params := request.Params
sessionID := strings.TrimSpace(shared.StringArg(params, "sessionId", ""))
if sessionID == "" {
return nil, &shared.RPCError{
Code: -32602,
Message: "sessionId is required",
}
}
threadID := strings.TrimSpace(
shared.StringArg(params, "threadId", sessionID),
)
if threadID == "" {
threadID = sessionID
}
if method == "session.start" {
s.resetSession(sessionID, threadID)
}
result, rpcErr := s.enqueue(threadID, task{
req: request,
notify: notify,
done: make(chan taskResult, 1),
})
if rpcErr != nil {
return nil, rpcErr
}
return result, nil
case "session.cancel":
params := request.Params
sessionID := strings.TrimSpace(shared.StringArg(params, "sessionId", ""))
if sessionID == "" {
return nil, &shared.RPCError{
Code: -32602,
Message: "sessionId is required",
}
}
cancelled := s.cancelSession(sessionID)
return map[string]any{"accepted": true, "cancelled": cancelled}, nil
case "session.close":
params := request.Params
sessionID := strings.TrimSpace(shared.StringArg(params, "sessionId", ""))
if sessionID == "" {
return nil, &shared.RPCError{
Code: -32602,
Message: "sessionId is required",
}
}
closed := s.closeSession(sessionID)
return map[string]any{"accepted": true, "closed": closed}, nil
case "xworkmate.dispatch.resolve":
return handleDispatchResolve(request.Params), nil
case "xworkmate.routing.resolve":
result, _ := resolveRoutingMetadataWithProviders(
request.Params,
s.availableProviders(),
)
return mergeRoutingResponse(map[string]any{"ok": true}, result), nil
case "xworkmate.provider.probe":
providerID := strings.TrimSpace(shared.StringArg(request.Params, "providerId", ""))
if providerID == "" {
return nil, &shared.RPCError{
Code: -32602,
Message: "providerId is required",
}
}
provider, ok := s.syncedProviderByID(providerID)
if !ok {
return map[string]any{
"success": false,
"providerId": providerID,
"error": "provider is not advertised by the bridge",
}, nil
}
result, err := s.probeExternalProvider(context.Background(), provider, request.Params)
if err != nil {
return map[string]any{
"success": false,
"providerId": providerID,
"error": err.Error(),
}, nil
}
return map[string]any{
"success": true,
"providerId": providerID,
"probeMethod": "acp.capabilities",
"capabilities": result,
}, nil
case "xworkmate.mounts.reconcile":
return handleMountReconcile(request.Params), nil
case "xworkmate.gateway.connect":
return handleGatewayConnect(s, request.Params, notify), nil
case "xworkmate.gateway.request":
return handleGatewayRequest(s, request.Params, notify), nil
case "xworkmate.gateway.disconnect":
return handleGatewayDisconnect(s, request.Params, notify), nil
default:
return nil, &shared.RPCError{
Code: -32601,
Message: fmt.Sprintf("unknown method: %s", method),
}
}
}
func (s *Server) executeSessionTask(task task) (map[string]any, *shared.RPCError) {
params := task.req.Params
resolvedRouting, hasResolvedRouting := resolveRoutingMetadataWithProviders(
params,
s.availableProviders(),
)
if !hasResolvedRouting {
return nil, &shared.RPCError{
Code: -32602,
Message: "ROUTING_REQUIRED",
}
}
sessionID := strings.TrimSpace(shared.StringArg(params, "sessionId", ""))
threadID := strings.TrimSpace(shared.StringArg(params, "threadId", sessionID))
if resolvedRouting.Unavailable {
response := mergeRoutingResponse(map[string]any{
"success": false,
"error": resolvedRouting.UnavailableMessage,
"unavailable": true,
"unavailableCode": resolvedRouting.UnavailableCode,
"unavailableMessage": resolvedRouting.UnavailableMessage,
}, resolvedRouting)
return response, nil
}
executionParams := buildResolvedExecutionParams(params, resolvedRouting)
mode := strings.TrimSpace(shared.StringArg(executionParams, "mode", "single-agent"))
provider := strings.TrimSpace(shared.StringArg(executionParams, "provider", ""))
session := s.getOrCreateSession(sessionID, threadID)
session.mode = mode
if provider != "" {
session.provider = provider
}
prompt := strings.TrimSpace(shared.StringArg(executionParams, "taskPrompt", ""))
if prompt != "" {
session.history = append(session.history, "USER: "+prompt)
}
turnID := fmt.Sprintf("turn-%d", time.Now().UnixNano())
ctx, cancel := context.WithCancel(context.Background())
s.setSessionCancel(sessionID, cancel)
defer s.clearSessionCancel(sessionID)
notify := task.notify
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "status",
"event": "started",
"message": "session started",
"pending": true,
"error": false,
})
if mode == router.ExecutionTargetGatewayChat || mode == router.ExecutionTargetGateway {
result := s.runGateway(
ctx,
task.req.Method,
session,
executionParams,
turnID,
notify,
)
if result.err != nil {
return nil, result.err
}
result.response = mergeRoutingResponse(result.response, resolvedRouting)
return result.response, nil
}
if mode == "multi-agent" {
result := s.runMultiAgent(ctx, session, executionParams, turnID, notify)
if result.err != nil {
return nil, result.err
}
result.response = mergeRoutingResponse(result.response, resolvedRouting)
if err := recordRoutingSuccess(params, resolvedRouting, result.response); err != nil {
return nil, &shared.RPCError{Code: -32001, Message: err.Error()}
}
return result.response, nil
}
result := s.runSingleAgent(
ctx,
task.req.Method,
session,
executionParams,
turnID,
notify,
)
if result.err != nil {
return nil, result.err
}
result.response = mergeRoutingResponse(result.response, resolvedRouting)
if err := recordRoutingSuccess(params, resolvedRouting, result.response); err != nil {
return nil, &shared.RPCError{Code: -32001, Message: err.Error()}
}
return result.response, nil
}
func (s *Server) runSingleAgent(
ctx context.Context,
method string,
session *session,
params map[string]any,
turnID string,
notify func(map[string]any),
) taskResult {
provider := session.provider
if provider == "" {
provider = strings.TrimSpace(shared.StringArg(params, "provider", "codex"))
}
workingDirectory := strings.TrimSpace(
shared.StringArg(params, "workingDirectory", ""),
)
_, effectiveWorkingDirectory := shared.NormalizeProviderWorkingDirectory(
provider,
workingDirectory,
)
if syncedProvider, ok := s.syncedProviderByID(provider); ok {
response, err := s.runSingleAgentViaExternalProvider(
ctx,
syncedProvider,
method,
params,
notify,
)
if err == nil {
result := shared.AsMap(response["result"])
if len(result) == 0 {
result = response
}
if _, exists := result["provider"]; !exists {
result["provider"] = provider
}
if _, exists := result["mode"]; !exists {
result["mode"] = "single-agent"
}
if _, exists := result["turnId"]; !exists {
result["turnId"] = turnID
}
if _, exists := result["effectiveWorkingDirectory"]; !exists && effectiveWorkingDirectory != "" {
result["effectiveWorkingDirectory"] = effectiveWorkingDirectory
}
return taskResult{response: enrichSingleAgentResultArtifacts(result, params)}
}
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "status",
"event": "completed",
"message": err.Error(),
"pending": false,
"error": true,
})
return taskResult{
response: map[string]any{
"success": false,
"error": err.Error(),
"turnId": turnID,
"mode": "single-agent",
"provider": provider,
},
}
}
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "status",
"event": "completed",
"message": "provider is not advertised by the bridge",
"pending": false,
"error": true,
})
return taskResult{
response: map[string]any{
"success": false,
"error": "provider is not advertised by the bridge",
"turnId": turnID,
"mode": "single-agent",
"provider": provider,
},
}
}
func (s *Server) runMultiAgent(
ctx context.Context,
session *session,
params map[string]any,
turnID string,
notify func(map[string]any),
) taskResult {
prompt := shared.ComposeHistoryPrompt(session.history)
if prompt == "" {
prompt = strings.TrimSpace(shared.StringArg(params, "taskPrompt", ""))
}
prompt = shared.AugmentPromptWithAttachments(prompt, params)
baseURL := shared.NormalizeBaseURL(
shared.StringArg(params, "aiGatewayBaseUrl", os.Getenv("AI_GATEWAY_BASE_URL")),
)
apiKey := strings.TrimSpace(shared.StringArg(params, "aiGatewayApiKey", os.Getenv("AI_GATEWAY_API_KEY")))
model := strings.TrimSpace(
shared.StringArg(
params,
"model",
shared.EnvOrDefault("ACP_MULTI_AGENT_MODEL", "gpt-4o"),
),
)
if model == "" {
model = "gpt-4o"
}
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "step",
"mode": "multi-agent",
"title": "Planner",
"message": "Preparing multi-agent run",
"pending": false,
"error": false,
"role": "architect",
"iteration": 1,
"score": 0,
})
if apiKey == "" {
errMsg := "aiGatewayApiKey is required for multi-agent mode"
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "status",
"mode": "multi-agent",
"message": errMsg,
"pending": false,
"error": true,
})
return taskResult{
response: map[string]any{
"success": false,
"error": errMsg,
"turnId": turnID,
"mode": "multi-agent",
},
}
}
messages := []map[string]string{
{
"role": "system",
"content": "You are a multi-agent coordinator. Return concise actionable output.",
},
{"role": "user", "content": prompt},
}
output, err := shared.CallOpenAICompatibleCtx(
ctx,
baseURL,
apiKey,
model,
messages,
)
if err != nil {
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "status",
"mode": "multi-agent",
"message": err.Error(),
"pending": false,
"error": true,
})
return taskResult{
response: map[string]any{
"success": false,
"error": err.Error(),
"turnId": turnID,
"mode": "multi-agent",
},
}
}
s.emitSessionUpdate(session, notify, turnID, map[string]any{
"type": "step",
"mode": "multi-agent",
"title": "Reviewer",
"message": output,
"pending": false,
"error": false,
"role": "tester",
"iteration": 1,
"score": 9,
})
return taskResult{
response: map[string]any{
"success": true,
"summary": output,
"finalScore": 9,
"iterations": 1,
"turnId": turnID,
"mode": "multi-agent",
},
}
}
func handleDispatchResolve(params map[string]any) map[string]any {
providers := parseDispatchProviders(params["providers"])
requiredCapabilities := parseStringSlice(params["requiredCapabilities"])
preferredProviderID := strings.TrimSpace(
shared.StringArg(params, "preferredProviderId", ""),
)
request := dispatch.Request{
Providers: providers,
PreferredProviderID: preferredProviderID,
RequiredCapabilities: requiredCapabilities,
}
if nodeState := parseDispatchNodeState(params["nodeState"]); nodeState != nil {
request.NodeState = nodeState
}
if nodeInfo := parseDispatchNodeInfo(params["nodeInfo"]); nodeInfo != nil {
request.NodeInfo = nodeInfo
}
return dispatch.ResultMap(dispatch.Resolve(request))
}
func parseDispatchProviders(raw any) []dispatch.Provider {
list, ok := raw.([]any)
if !ok {
return nil
}
providers := make([]dispatch.Provider, 0, len(list))
for _, item := range list {
entry, ok := item.(map[string]any)
if !ok {
continue
}
id := strings.TrimSpace(shared.StringArg(entry, "id", ""))
if id == "" {
continue
}
providers = append(providers, dispatch.Provider{
ID: id,
Name: strings.TrimSpace(shared.StringArg(entry, "name", "")),
DefaultArgs: parseStringSlice(entry["defaultArgs"]),
Capabilities: parseStringSlice(entry["capabilities"]),
})
}
return providers
}
func parseDispatchNodeState(raw any) *dispatch.NodeState {
entry, ok := raw.(map[string]any)
if !ok {
return nil
}
return &dispatch.NodeState{
SelectedAgentID: strings.TrimSpace(
shared.StringArg(entry, "selectedAgentId", ""),
),
GatewayConnected: shared.BoolArg(
fmt.Sprint(entry["gatewayConnected"]),
false,
),
ExecutionTarget: strings.TrimSpace(
shared.StringArg(entry, "executionTarget", ""),
),
RuntimeMode: strings.TrimSpace(shared.StringArg(entry, "runtimeMode", "")),
BridgeEnabled: shared.BoolArg(fmt.Sprint(entry["bridgeEnabled"]), false),
BridgeState: strings.TrimSpace(shared.StringArg(entry, "bridgeState", "")),
ResolvedCodexCLIPath: strings.TrimSpace(
shared.StringArg(entry, "resolvedCodexCliPath", ""),
),
ConfiguredCodexCLIPath: strings.TrimSpace(
shared.StringArg(entry, "configuredCodexCliPath", ""),
),
}
}
func parseDispatchNodeInfo(raw any) *dispatch.NodeInfo {
entry, ok := raw.(map[string]any)
if !ok {
return nil
}
return &dispatch.NodeInfo{
ID: strings.TrimSpace(shared.StringArg(entry, "id", "")),
Name: strings.TrimSpace(shared.StringArg(entry, "name", "")),
Version: strings.TrimSpace(shared.StringArg(entry, "version", "")),
}
}
func parseStringSlice(raw any) []string {
list, ok := raw.([]any)
if !ok {
return nil
}
values := make([]string, 0, len(list))
for _, item := range list {
value := strings.TrimSpace(fmt.Sprint(item))
if value == "" {
continue
}
values = append(values, value)
}
return values
}
func handleMountReconcile(params map[string]any) map[string]any {
config := parseMountConfig(params["config"])
request := mounts.Request{
Config: config,
AIGatewayURL: strings.TrimSpace(shared.StringArg(params, "aiGatewayUrl", "")),
ConfiguredCodexCLIPath: strings.TrimSpace(shared.StringArg(params, "configuredCodexCliPath", "")),
CodexHome: strings.TrimSpace(shared.StringArg(params, "codexHome", "")),
OpencodeHome: strings.TrimSpace(shared.StringArg(params, "opencodeHome", "")),
OpenClawHome: strings.TrimSpace(shared.StringArg(params, "openclawHome", "")),
Aris: parseMountArisInput(params["aris"]),
}
return mounts.ResultMap(mounts.Reconcile(request))
}
func parseMountConfig(raw any) mounts.Config {
entry, ok := raw.(map[string]any)
if !ok {
return mounts.Config{}
}
managedMCPServers := parseMountManagedServers(entry["managedMcpServers"])
return mounts.Config{
AutoSync: shared.BoolArg(fmt.Sprint(entry["autoSync"]), false),
UsesAris: shared.BoolArg(fmt.Sprint(entry["usesAris"]), false),
ManagedMCPServers: managedMCPServers,
}
}
func parseMountManagedServers(raw any) []mounts.ManagedMCPServer {
list, ok := raw.([]any)
if !ok {
return nil
}
servers := make([]mounts.ManagedMCPServer, 0, len(list))
for _, item := range list {
entry, ok := item.(map[string]any)
if !ok {
continue
}
id := strings.TrimSpace(shared.StringArg(entry, "id", ""))
if id == "" {
continue
}
servers = append(servers, mounts.ManagedMCPServer{
ID: id,
Name: strings.TrimSpace(shared.StringArg(entry, "name", "")),
Transport: strings.TrimSpace(shared.StringArg(entry, "transport", "")),
Command: strings.TrimSpace(shared.StringArg(entry, "command", "")),
URL: strings.TrimSpace(shared.StringArg(entry, "url", "")),
Args: parseStringSlice(entry["args"]),
Enabled: shared.BoolArg(fmt.Sprint(entry["enabled"]), true),
})
}
return servers
}
func parseMountArisInput(raw any) mounts.ArisInput {
entry, ok := raw.(map[string]any)
if !ok {
return mounts.ArisInput{}
}
return mounts.ArisInput{
Available: shared.BoolArg(fmt.Sprint(entry["available"]), false),
BundleVersion: strings.TrimSpace(shared.StringArg(entry, "bundleVersion", "")),
LLMChatServerPath: strings.TrimSpace(shared.StringArg(entry, "llmChatServerPath", "")),
SkillCount: shared.IntArg(fmt.Sprint(entry["skillCount"]), 0),
BridgeAvailable: shared.BoolArg(fmt.Sprint(entry["bridgeAvailable"]), false),
Error: strings.TrimSpace(shared.StringArg(entry, "error", "")),
}
}

1197
internal/acp/server.go
View File

File diff suppressed because it is too large Load Diff

133
internal/acp/session_manager.go Normal file
View File

@ -0,0 +1,133 @@
package acp
import (
"context"
"xworkmate-bridge/internal/shared"
)
func (s *Server) enqueue(threadID string, task task) (map[string]any, *shared.RPCError) {
queue := s.ensureQueue(threadID)
queue <- task
result := <-task.done
return result.response, result.err
}
func (s *Server) ensureQueue(threadID string) chan task {
s.mu.Lock()
defer s.mu.Unlock()
queue, ok := s.queues[threadID]
if ok {
return queue
}
queue = make(chan task, 32)
s.queues[threadID] = queue
go s.runQueue(queue)
return queue
}
func (s *Server) runQueue(queue chan task) {
for task := range queue {
response, err := s.executeSessionTask(task)
task.done <- taskResult{response: response, err: err}
}
}
func (s *Server) emitSessionUpdate(
session *session,
notify func(map[string]any),
turnID string,
payload map[string]any,
) {
if notify == nil {
return
}
s.mu.Lock()
session.seq++
seq := session.seq
s.mu.Unlock()
params := map[string]any{
"sessionId": session.sessionID,
"threadId": session.threadID,
"turnId": turnID,
"seq": seq,
}
for key, value := range payload {
params[key] = value
}
notify(shared.NotificationEnvelope("session.update", params))
}
func (s *Server) getOrCreateSession(sessionID, threadID string) *session {
s.mu.Lock()
defer s.mu.Unlock()
if session, ok := s.sessions[sessionID]; ok {
if threadID != "" {
session.threadID = threadID
}
session.closed = false
return session
}
session := &session{sessionID: sessionID, threadID: threadID}
s.sessions[sessionID] = session
return session
}
func (s *Server) resetSession(sessionID, threadID string) {
s.mu.Lock()
defer s.mu.Unlock()
s.sessions[sessionID] = &session{
sessionID: sessionID,
threadID: threadID,
history: []string{},
}
}
func (s *Server) setSessionCancel(sessionID string, cancel context.CancelFunc) {
s.mu.Lock()
defer s.mu.Unlock()
if session, ok := s.sessions[sessionID]; ok {
session.cancel = cancel
}
}
func (s *Server) clearSessionCancel(sessionID string) {
s.mu.Lock()
defer s.mu.Unlock()
if session, ok := s.sessions[sessionID]; ok {
session.cancel = nil
}
}
func (s *Server) cancelSession(sessionID string) bool {
s.mu.Lock()
session, ok := s.sessions[sessionID]
if !ok {
s.mu.Unlock()
return false
}
cancel := session.cancel
s.mu.Unlock()
if cancel != nil {
cancel()
return true
}
return false
}
func (s *Server) closeSession(sessionID string) bool {
s.mu.Lock()
session, ok := s.sessions[sessionID]
if !ok {
s.mu.Unlock()
return false
}
cancel := session.cancel
session.closed = true
delete(s.sessions, sessionID)
s.mu.Unlock()
if cancel != nil {
cancel()
}
return true
}

44
internal/acp/types.go Normal file
View File

@ -0,0 +1,44 @@
package acp
import (
"context"
"sync"
"xworkmate-bridge/internal/gatewayruntime"
"xworkmate-bridge/internal/service"
"xworkmate-bridge/internal/shared"
)
type session struct {
sessionID string
threadID string
mode string
provider string
history []string
seq int
cancel context.CancelFunc
closed bool
}
type task struct {
req shared.RPCRequest
notify func(map[string]any)
done chan taskResult
}
type taskResult struct {
response map[string]any
err *shared.RPCError
}
type Server struct {
mu sync.Mutex
config *BridgeConfig
sessions map[string]*session
queues map[string]chan task
gateway *gatewayruntime.Manager
providerCatalog map[string]syncedProvider
providerOrder []string
authService *service.StaticTokenAuthService
allowedOrigins []string
}

78
internal/acp/web_contract.go
View File

@ -1,78 +0,0 @@
package acp
import (
"encoding/json"
"net/http"
"strings"
"xworkmate-bridge/internal/shared"
)
func (s *Server) allowedOrigins() []string {
raw := strings.TrimSpace(shared.EnvOrDefault(
"ACP_ALLOWED_ORIGINS",
"https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*",
))
if raw == "" {
return nil
}
parts := strings.Split(raw, ",")
origins := make([]string, 0, len(parts))
for _, part := range parts {
candidate := strings.TrimSpace(part)
if candidate == "" {
continue
}
origins = append(origins, candidate)
}
return origins
}
func (s *Server) originAllowed(origin string) bool {
origin = strings.TrimSpace(origin)
if origin == "" {
return true
}
for _, allowed := range s.allowedOrigins() {
if strings.HasSuffix(allowed, ":*") {
if strings.HasPrefix(origin, strings.TrimSuffix(allowed, "*")) {
return true
}
continue
}
if origin == allowed {
return true
}
}
return false
}
func (s *Server) applyCORS(w http.ResponseWriter, r *http.Request) {
origin := strings.TrimSpace(r.Header.Get("Origin"))
if origin == "" || !s.originAllowed(origin) {
return
}
headers := w.Header()
headers.Set("Access-Control-Allow-Origin", origin)
headers.Set("Access-Control-Allow-Methods", "POST, OPTIONS")
headers.Set(
"Access-Control-Allow-Headers",
"Authorization, Content-Type, Accept",
)
headers.Set("Access-Control-Max-Age", "600")
headers.Add("Vary", "Origin")
headers.Add("Vary", "Access-Control-Request-Method")
headers.Add("Vary", "Access-Control-Request-Headers")
}
func (s *Server) writeJSONError(
w http.ResponseWriter,
requestID any,
statusCode int,
code int,
message string,
) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(statusCode)
_ = json.NewEncoder(w).Encode(shared.ErrorEnvelope(requestID, code, message))
}

10
internal/acp/web_contract_test.go
View File

@ -7,6 +7,8 @@ import (
"reflect"
"strings"
"testing"
"xworkmate-bridge/internal/shared"
)
func TestHTTPHandlerRootAndPingExposeRuntimeVersionInfo(t *testing.T) {
@ -94,7 +96,7 @@ func TestHTTPHandlerBareAliasPathsExposeCapabilities(t *testing.T) {
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatalf("decode capability alias response: %v", err)
}
result := asMap(envelope["result"])
result := shared.AsMap(envelope["result"])
if got := result["singleAgent"]; got != true {
t.Fatalf("expected singleAgent true, got %#v", got)
}
@ -167,7 +169,7 @@ func TestHTTPHandlerBareAliasPathsServeRPC(t *testing.T) {
if got := envelope["id"]; got != "rpc-1" {
t.Fatalf("expected rpc id rpc-1, got %#v", got)
}
result := asMap(envelope["result"])
result := shared.AsMap(envelope["result"])
if tc.wantMode == "gateway" {
gatewayProviders := mustObjectList(t, result["gatewayProviders"])
if len(gatewayProviders) != 1 {
@ -398,7 +400,7 @@ func TestHandleRPCCapabilitiesReturnsCanonicalProviderContract(t *testing.T) {
t.Fatalf("decode capabilities response: %v", err)
}
result := asMap(envelope["result"])
result := shared.AsMap(envelope["result"])
if got := result["singleAgent"]; got != true {
t.Fatalf("expected singleAgent true, got %v", got)
}
@ -499,7 +501,7 @@ func mustObjectList(t *testing.T, value any) []map[string]any {
}
items := make([]map[string]any, 0, len(raw))
for _, item := range raw {
items = append(items, asMap(item))
items = append(items, shared.AsMap(item))
}
return items
}

94
internal/geminiadapter/server.go
View File

@ -12,8 +12,6 @@ import (
"sync"
"time"
"github.com/gorilla/websocket"
"xworkmate-bridge/internal/service"
"xworkmate-bridge/internal/shared"
)
@ -36,14 +34,6 @@ type Server struct {
sessions map[string]*adapterSession
}
var adapterWSUpgrader = websocket.Upgrader{
ReadBufferSize: 16 * 1024,
WriteBufferSize: 16 * 1024,
CheckOrigin: func(*http.Request) bool {
return true
},
}
type adapterSession struct {
history []string
model string
@ -53,7 +43,7 @@ type adapterSession struct {
}
func Serve(args []string) error {
flags := flag.NewFlagSet("gemini-acp-adapter", flag.ExitOnError)
flags := flag.NewFlagSet("adapter gemini", flag.ExitOnError)
listen := flags.String(
"listen",
strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_LISTEN_ADDR", defaultListenAddr)),
@ -110,7 +100,7 @@ func NewServer(client rpcClient) *Server {
authService: service.NewStaticTokenAuthService(strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_AUTH_TOKEN", ""))),
providerID: strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_PROVIDER_ID", defaultProviderID)),
providerLabel: strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_PROVIDER_LABEL", defaultLabel)),
allowedOrigins: parseAllowedOrigins(strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_ALLOWED_ORIGINS", "https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*"))),
allowedOrigins: shared.ParseAllowedOrigins(strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_ALLOWED_ORIGINS", "https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*"))),
upstreamMethod: strings.TrimSpace(shared.EnvOrDefault("GEMINI_ADAPTER_UPSTREAM_METHOD", "")),
sessionRunner: func(ctx context.Context, model, prompt, workingDirectory string) (string, error) {
return shared.RunProviderCommand(
@ -126,17 +116,17 @@ func NewServer(client rpcClient) *Server {
}
func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
if !s.originAllowed(r.Header.Get("Origin")) {
s.writeJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
if !shared.OriginAllowed(r.Header.Get("Origin"), s.allowedOrigins) {
shared.WriteJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
return
}
if !s.authorized(r) {
s.writeJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
return
}
upgrader := adapterWSUpgrader
upgrader := shared.StandardWSUpgrader
upgrader.CheckOrigin = func(req *http.Request) bool {
return s.originAllowed(req.Header.Get("Origin")) && s.authorized(req)
return shared.OriginAllowed(req.Header.Get("Origin"), s.allowedOrigins) && s.authorized(req)
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
@ -172,31 +162,31 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
}
func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
s.applyCORS(w, r)
shared.ApplyCORS(w, r, s.allowedOrigins)
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
if r.Method != http.MethodPost {
s.writeJSONError(w, nil, http.StatusMethodNotAllowed, -32600, "method not allowed")
shared.WriteJSONError(w, nil, http.StatusMethodNotAllowed, -32600, "method not allowed")
return
}
if !s.originAllowed(r.Header.Get("Origin")) {
s.writeJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
if !shared.OriginAllowed(r.Header.Get("Origin"), s.allowedOrigins) {
shared.WriteJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
return
}
if !s.authorized(r) {
s.writeJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
return
}
payload, err := io.ReadAll(r.Body)
if err != nil {
s.writeJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
return
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
s.writeJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
return
}
result := s.handleRequest(request)
@ -472,56 +462,6 @@ func (s *Server) closeSession(sessionID string) bool {
return true
}
func parseAllowedOrigins(raw string) []string {
if raw == "" {
return nil
}
parts := strings.Split(raw, ",")
result := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part == "" {
continue
}
result = append(result, part)
}
return result
}
func (s *Server) originAllowed(origin string) bool {
origin = strings.TrimSpace(origin)
if origin == "" {
return true
}
for _, allowed := range s.allowedOrigins {
if strings.HasSuffix(allowed, ":*") {
if strings.HasPrefix(origin, strings.TrimSuffix(allowed, "*")) {
return true
}
continue
}
if origin == allowed {
return true
}
}
return false
}
func (s *Server) applyCORS(w http.ResponseWriter, r *http.Request) {
origin := strings.TrimSpace(r.Header.Get("Origin"))
if origin == "" || !s.originAllowed(origin) {
return
}
headers := w.Header()
headers.Set("Access-Control-Allow-Origin", origin)
headers.Set("Access-Control-Allow-Methods", "POST, OPTIONS")
headers.Set("Access-Control-Allow-Headers", "Authorization, Content-Type, Accept")
headers.Set("Access-Control-Max-Age", "600")
headers.Add("Vary", "Origin")
headers.Add("Vary", "Access-Control-Request-Method")
headers.Add("Vary", "Access-Control-Request-Headers")
}
func (s *Server) authorized(r *http.Request) bool {
if s == nil {
return false
@ -531,9 +471,3 @@ func (s *Server) authorized(r *http.Request) bool {
}
return s.authService.ValidateAuthorizationHeader(r.Header.Get("Authorization"))
}
func (s *Server) writeJSONError(w http.ResponseWriter, requestID any, statusCode int, code int, message string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(statusCode)
_ = json.NewEncoder(w).Encode(shared.ErrorEnvelope(requestID, code, message))
}

114
internal/hermesadapter/server.go
View File

@ -11,8 +11,6 @@ import (
"sync"
"time"
"github.com/gorilla/websocket"
"xworkmate-bridge/internal/service"
"xworkmate-bridge/internal/shared"
)
@ -34,14 +32,6 @@ type Server struct {
sessions map[string]*adapterSession
}
var adapterWSUpgrader = websocket.Upgrader{
ReadBufferSize: 16 * 1024,
WriteBufferSize: 16 * 1024,
CheckOrigin: func(*http.Request) bool {
return true
},
}
type adapterSession struct {
history []string
model string
@ -52,7 +42,7 @@ type adapterSession struct {
}
func Serve(args []string) error {
flags := flag.NewFlagSet("hermes-acp-adapter", flag.ExitOnError)
flags := flag.NewFlagSet("adapter hermes", flag.ExitOnError)
listen := flags.String(
"listen",
strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_LISTEN_ADDR", defaultListenAddr)),
@ -109,24 +99,24 @@ func NewServer(client rpcClient) *Server {
authService: service.NewStaticTokenAuthService(strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_AUTH_TOKEN", ""))),
providerID: strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_PROVIDER_ID", defaultProviderID)),
providerLabel: strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_PROVIDER_LABEL", defaultLabel)),
allowedOrigins: parseAllowedOrigins(strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_ALLOWED_ORIGINS", "https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*"))),
allowedOrigins: shared.ParseAllowedOrigins(strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_ALLOWED_ORIGINS", "https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*"))),
upstreamMethod: strings.TrimSpace(shared.EnvOrDefault("HERMES_ADAPTER_UPSTREAM_METHOD", "session/prompt")),
sessions: make(map[string]*adapterSession),
}
}
func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
if !s.originAllowed(r.Header.Get("Origin")) {
s.writeJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
if !shared.OriginAllowed(r.Header.Get("Origin"), s.allowedOrigins) {
shared.WriteJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
return
}
if !s.authorized(r) {
s.writeJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
return
}
upgrader := adapterWSUpgrader
upgrader := shared.StandardWSUpgrader
upgrader.CheckOrigin = func(req *http.Request) bool {
return s.originAllowed(req.Header.Get("Origin")) && s.authorized(req)
return shared.OriginAllowed(req.Header.Get("Origin"), s.allowedOrigins) && s.authorized(req)
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
@ -162,31 +152,31 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
}
func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
s.applyCORS(w, r)
shared.ApplyCORS(w, r, s.allowedOrigins)
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
if r.Method != http.MethodPost {
s.writeJSONError(w, nil, http.StatusMethodNotAllowed, -32600, "method not allowed")
shared.WriteJSONError(w, nil, http.StatusMethodNotAllowed, -32600, "method not allowed")
return
}
if !s.originAllowed(r.Header.Get("Origin")) {
s.writeJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
if !shared.OriginAllowed(r.Header.Get("Origin"), s.allowedOrigins) {
shared.WriteJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
return
}
if !s.authorized(r) {
s.writeJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
return
}
payload, err := io.ReadAll(r.Body)
if err != nil {
s.writeJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
return
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
s.writeJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
return
}
result := s.handleRequest(request)
@ -287,7 +277,7 @@ func (s *Server) handleConfiguredUpstreamSessionRequest(method, upstreamMethod s
"upstreamMethod": upstreamMethod,
}
}
result, _ := response["result"].(map[string]any)
result := shared.AsMap(response["result"])
if len(result) > 0 {
if _, ok := result["provider"]; !ok {
result["provider"] = s.providerID
@ -297,7 +287,7 @@ func (s *Server) handleConfiguredUpstreamSessionRequest(method, upstreamMethod s
}
return result
}
if errPayload, ok := response["error"].(map[string]any); ok && len(errPayload) > 0 {
if errPayload := shared.AsMap(response["error"]); len(errPayload) > 0 {
return map[string]any{
"success": false,
"provider": s.providerID,
@ -445,7 +435,7 @@ func (s *Server) handleHermesACPUpstreamSessionRequest(method string, params map
output := strings.TrimSpace(strings.Join(outputParts, ""))
if output == "" {
if resultMap, ok := response["result"].(map[string]any); ok {
if resultMap := shared.AsMap(response["result"]); resultMap != nil {
for _, key := range []string{"output", "finalResponse", "final_response", "text", "message", "response"} {
candidate := strings.TrimSpace(shared.StringArg(resultMap, key, ""))
if candidate == "" || isGenericHermesAckText(candidate) {
@ -504,7 +494,7 @@ func normalizeHermesUpstreamMethod(method string) string {
func extractHermesUpstreamSessionID(response map[string]any) string {
for _, key := range []string{"sessionId", "session_id", "id"} {
if value := strings.TrimSpace(shared.StringArg(asMap(response["result"]), key, "")); value != "" {
if value := strings.TrimSpace(shared.StringArg(shared.AsMap(response["result"]), key, "")); value != "" {
return value
}
if value := strings.TrimSpace(shared.StringArg(response, key, "")); value != "" {
@ -522,11 +512,11 @@ func extractHermesSessionUpdateText(notification map[string]any) string {
if method != "session.update" && method != "session/update" && method != "acp.session.update" {
return ""
}
payload := asMap(notification["params"])
payload := shared.AsMap(notification["params"])
if len(payload) == 0 {
payload = notification
}
update := asMap(payload["update"])
update := shared.AsMap(payload["update"])
if len(update) == 0 {
update = payload
}
@ -592,16 +582,6 @@ func isGenericHermesAckText(text string) bool {
}
}
func asMap(value any) map[string]any {
if value == nil {
return nil
}
if result, ok := value.(map[string]any); ok {
return result
}
return nil
}
func (s *Server) getOrCreateSession(sessionID string) *adapterSession {
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
@ -642,54 +622,6 @@ func (s *Server) closeSession(sessionID string) bool {
return true
}
func parseAllowedOrigins(raw string) []string {
if raw == "" {
return nil
}
parts := strings.Split(raw, ",")
result := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part == "" {
continue
}
result = append(result, part)
}
return result
}
func (s *Server) originAllowed(origin string) bool {
origin = strings.TrimSpace(origin)
if origin == "" {
return true
}
for _, allowed := range s.allowedOrigins {
if strings.HasSuffix(allowed, ":*") {
if strings.HasPrefix(origin, strings.TrimSuffix(allowed, "*")) {
return true
}
continue
}
if origin == allowed {
return true
}
}
return false
}
func (s *Server) applyCORS(w http.ResponseWriter, r *http.Request) {
origin := strings.TrimSpace(r.Header.Get("Origin"))
if origin == "" || !s.originAllowed(origin) {
return
}
headers := w.Header()
headers.Set("Access-Control-Allow-Origin", origin)
headers.Set("Access-Control-Allow-Methods", "POST, OPTIONS")
headers.Set("Access-Control-Allow-Headers", "Authorization, Content-Type, Accept")
headers.Set("Access-Control-Max-Age", "600")
headers.Add("Vary", "Origin")
}
func (s *Server) authorized(r *http.Request) bool {
if s == nil {
return false
@ -699,9 +631,3 @@ func (s *Server) authorized(r *http.Request) bool {
}
return s.authService.ValidateAuthorizationHeader(r.Header.Get("Authorization"))
}
func (s *Server) writeJSONError(w http.ResponseWriter, id any, status int, code int, message string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(shared.ErrorEnvelope(id, code, message))
}

3
internal/mounts/reconcile.go
View File

@ -292,8 +292,7 @@ func reconcileOpencode(config Config, opencodeHome string) MountTargetState {
state.DiscoveredMCPCount = discovered
state.ManagedMCPCount = len(managedServers)
state.Detail = "OpenCode public base URL: https://xworkmate-bridge.svc.plus/acp-server/opencode\n" +
"Preferred WebSocket endpoint: https://xworkmate-bridge.svc.plus/acp-server/opencode/acp\n" +
"Compatibility HTTP RPC endpoint: https://xworkmate-bridge.svc.plus/acp-server/opencode/acp/rpc"
"Preferred WebSocket endpoint: https://xworkmate-bridge.svc.plus/acp-server/opencode/acp"
return state
}

278
internal/opencodeadapter/http_client.go Normal file
View File

@ -0,0 +1,278 @@
package opencodeadapter
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os/exec"
"strings"
"sync"
"time"
"xworkmate-bridge/internal/shared"
)
type opencodeHTTPClient struct {
mu sync.Mutex
command string
args []string
cwd string
cmd *exec.Cmd
baseURL string
client *http.Client
}
func newOpenCodeHTTPClient(command string, args []string) *opencodeHTTPClient {
return &opencodeHTTPClient{
command: strings.TrimSpace(command),
args: append([]string(nil), args...),
baseURL: "http://127.0.0.1:38993",
client: &http.Client{Timeout: 5 * time.Minute},
}
}
func (c *opencodeHTTPClient) Initialize() (initializeResult, error) {
if err := c.ensureStarted(); err != nil {
return initializeResult{}, err
}
req, err := http.NewRequest(http.MethodGet, c.baseURL+"/global/health", nil)
if err != nil {
return initializeResult{}, err
}
resp, err := c.client.Do(req)
if err != nil {
return initializeResult{}, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
return initializeResult{}, fmt.Errorf("opencode health failed (%d): %s", resp.StatusCode, strings.TrimSpace(string(body)))
}
return initializeResult{
ProtocolVersion: 1,
AuthMethods: []map[string]any{
{"id": "opencode-login", "name": "Login with opencode", "description": "Run `opencode auth login` in the terminal"},
},
AgentCapabilities: map[string]any{
"loadSession": true,
"mcpCapabilities": map[string]any{"http": true, "sse": true},
"promptCapabilities": map[string]any{"embeddedContext": true, "image": true},
"sessionCapabilities": map[string]any{"fork": map[string]any{}, "list": map[string]any{}, "resume": map[string]any{}},
},
}, nil
}
func (c *opencodeHTTPClient) Call(method string, params map[string]any) (map[string]any, error) {
if err := c.ensureStarted(); err != nil {
return nil, err
}
switch strings.TrimSpace(method) {
case "session.start", "session.message":
sessionID := strings.TrimSpace(fmt.Sprint(params["sessionId"]))
prompt := strings.TrimSpace(sharedStringArg(params, "taskPrompt", ""))
return c.postSessionMessage(sessionID, prompt, params)
case "session.cancel":
return map[string]any{"accepted": true, "cancelled": false}, nil
case "session.close":
return map[string]any{"accepted": true, "closed": true}, nil
default:
return nil, fmt.Errorf("unsupported opencode method: %s", method)
}
}
func sharedStringArg(params map[string]any, key, fallback string) string {
if params == nil {
return fallback
}
if value := strings.TrimSpace(fmt.Sprint(params[key])); value != "" {
return value
}
return fallback
}
func (c *opencodeHTTPClient) CreateSession(title string) (string, error) {
if err := c.ensureStarted(); err != nil {
return "", err
}
body := map[string]any{}
if strings.TrimSpace(title) != "" {
body["title"] = strings.TrimSpace(title)
}
encoded, _ := json.Marshal(body)
req, err := http.NewRequest(http.MethodPost, c.baseURL+"/session", bytes.NewReader(encoded))
if err != nil {
return "", err
}
req.Header.Set("Content-Type", "application/json")
resp, err := c.client.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
raw, err := io.ReadAll(resp.Body)
if err != nil {
return "", err
}
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("opencode create session failed (%d): %s", resp.StatusCode, strings.TrimSpace(string(raw)))
}
var decoded map[string]any
if err := json.Unmarshal(raw, &decoded); err != nil {
return "", fmt.Errorf("decode opencode create session response: %w", err)
}
if sessionID := extractOpenCodeSessionID(decoded); sessionID != "" {
return sessionID, nil
}
if sessionID := extractOpenCodeSessionID(shared.AsMap(decoded["result"])); sessionID != "" {
return sessionID, nil
}
return "", fmt.Errorf("opencode create session returned no session id")
}
func (c *opencodeHTTPClient) SendMessage(sessionID, prompt string, params map[string]any) (map[string]any, error) {
if err := c.ensureStarted(); err != nil {
return nil, err
}
return c.postSessionMessage(sessionID, prompt, params)
}
func (c *opencodeHTTPClient) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
if c.cmd != nil && c.cmd.Process != nil {
_ = c.cmd.Process.Kill()
_, _ = c.cmd.Process.Wait()
}
c.cmd = nil
return nil
}
func (c *opencodeHTTPClient) ensureStarted() error {
c.mu.Lock()
defer c.mu.Unlock()
if c.cmd != nil {
return nil
}
if c.command == "" {
return fmt.Errorf("opencode command is empty")
}
cmd := exec.Command(c.command, c.args...)
if strings.TrimSpace(c.cwd) != "" {
cmd.Dir = strings.TrimSpace(c.cwd)
}
if err := cmd.Start(); err != nil {
return err
}
c.cmd = cmd
return c.waitReady()
}
func (c *opencodeHTTPClient) waitReady() error {
deadline := time.Now().Add(30 * time.Second)
for time.Now().Before(deadline) {
req, _ := http.NewRequest(http.MethodGet, c.baseURL+"/global/health", nil)
resp, err := c.client.Do(req)
if err == nil && resp != nil {
_ = resp.Body.Close()
if resp.StatusCode == http.StatusOK {
return nil
}
}
time.Sleep(300 * time.Millisecond)
}
return fmt.Errorf("opencode server did not become ready")
}
func (c *opencodeHTTPClient) postSessionMessage(sessionID, prompt string, params map[string]any) (map[string]any, error) {
if sessionID == "" {
return nil, fmt.Errorf("sessionId is required")
}
body := map[string]any{
"parts": []map[string]any{
{"type": "text", "text": strings.TrimSpace(prompt)},
},
}
if model := strings.TrimSpace(fmt.Sprint(params["model"])); model != "" {
body["model"] = model
}
if agent := strings.TrimSpace(fmt.Sprint(params["agent"])); agent != "" {
body["agent"] = agent
}
if system := strings.TrimSpace(fmt.Sprint(params["system"])); system != "" {
body["system"] = system
}
encoded, _ := json.Marshal(body)
req, err := http.NewRequest(http.MethodPost, c.baseURL+"/session/"+sessionID+"/message", bytes.NewReader(encoded))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
resp, err := c.client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
raw, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("opencode session message failed (%d): %s", resp.StatusCode, strings.TrimSpace(string(raw)))
}
var decoded map[string]any
if err := json.Unmarshal(raw, &decoded); err != nil {
return nil, fmt.Errorf("decode opencode response: %w", err)
}
text := extractOpenCodeText(decoded)
result := map[string]any{
"success": true,
"provider": "opencode",
"mode": "single-agent",
"sessionId": sessionID,
"output": text,
"summary": text,
"message": text,
}
return result, nil
}
func extractOpenCodeSessionID(value any) string {
switch v := value.(type) {
case string:
return strings.TrimSpace(v)
case map[string]any:
for _, key := range []string{"sessionId", "session_id", "id"} {
if sessionID := strings.TrimSpace(fmt.Sprint(v[key])); sessionID != "" {
return sessionID
}
}
}
return ""
}
func extractOpenCodeText(value any) string {
switch v := value.(type) {
case string:
return strings.TrimSpace(v)
case map[string]any:
for _, key := range []string{"text", "message", "output", "content", "summary"} {
if text := extractOpenCodeText(v[key]); text != "" {
return text
}
}
for _, child := range v {
if text := extractOpenCodeText(child); text != "" {
return text
}
}
case []any:
for _, child := range v {
if text := extractOpenCodeText(child); text != "" {
return text
}
}
}
return ""
}

363
internal/opencodeadapter/server.go Normal file
View File

@ -0,0 +1,363 @@
package opencodeadapter
import (
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"net/http"
"strings"
"sync"
"time"
"xworkmate-bridge/internal/service"
"xworkmate-bridge/internal/shared"
)
const (
defaultListenAddr = "127.0.0.1:38992"
defaultProviderID = "opencode"
defaultLabel = "OpenCode"
)
type Server struct {
client openCodeClient
authService *service.StaticTokenAuthService
providerID string
providerLabel string
allowedOrigins []string
sessions map[string]*opencodeSessionState
sessionsMu sync.Mutex
}
func Serve(args []string) error {
flags := flag.NewFlagSet("adapter opencode", flag.ExitOnError)
listen := flags.String(
"listen",
strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_LISTEN_ADDR", defaultListenAddr)),
"OpenCode ACP adapter listen address",
)
binary := flags.String(
"opencode-bin",
strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_BIN", shared.EnvOrDefault("ACP_OPENCODE_BIN", "opencode"))),
"OpenCode CLI binary path",
)
cwd := flags.String(
"cwd",
strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_CWD", "/home/ubuntu/.opencode")),
"OpenCode ACP working directory",
)
_ = flags.Parse(args)
client := newOpenCodeHTTPClient(*binary, []string{"serve", "--hostname", "127.0.0.1", "--port", "38993", "--print-logs"})
client.cwd = strings.TrimSpace(*cwd)
defer func() { _ = client.Close() }()
server := NewServer(client)
httpServer := &http.Server{
Addr: strings.TrimSpace(*listen),
Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/acp/rpc":
server.HandleRPC(w, r)
case "/acp":
server.HandleWebSocket(w, r)
default:
http.NotFound(w, r)
}
}),
ReadTimeout: 30 * time.Second,
WriteTimeout: 5 * time.Minute,
IdleTimeout: 2 * time.Minute,
}
if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
return fmt.Errorf("opencode adapter failed: %w", err)
}
return nil
}
func NewServer(client openCodeClient) *Server {
return &Server{
client: client,
authService: service.NewStaticTokenAuthService(strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_AUTH_TOKEN", ""))),
providerID: strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_PROVIDER_ID", defaultProviderID)),
providerLabel: strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_PROVIDER_LABEL", defaultLabel)),
allowedOrigins: shared.ParseAllowedOrigins(strings.TrimSpace(shared.EnvOrDefault("OPENCODE_ADAPTER_ALLOWED_ORIGINS", "https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*"))),
sessions: make(map[string]*opencodeSessionState),
}
}
func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
if !shared.OriginAllowed(r.Header.Get("Origin"), s.allowedOrigins) {
shared.WriteJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
return
}
if !s.authorized(r) {
shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
return
}
upgrader := shared.StandardWSUpgrader
upgrader.CheckOrigin = func(req *http.Request) bool {
return shared.OriginAllowed(req.Header.Get("Origin"), s.allowedOrigins) && s.authorized(req)
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}
defer func() { _ = conn.Close() }()
for {
_, payload, err := conn.ReadMessage()
if err != nil {
return
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
_ = conn.WriteJSON(shared.ErrorEnvelope(nil, -32700, err.Error()))
continue
}
response := s.handleRequest(request)
if request.ID != nil {
_ = conn.WriteJSON(shared.ResultEnvelope(request.ID, response))
}
}
}
func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
shared.ApplyCORS(w, r, s.allowedOrigins)
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
if r.Method != http.MethodPost {
shared.WriteJSONError(w, nil, http.StatusMethodNotAllowed, -32600, "method not allowed")
return
}
if !shared.OriginAllowed(r.Header.Get("Origin"), s.allowedOrigins) {
shared.WriteJSONError(w, nil, http.StatusForbidden, -32003, fmt.Sprintf("origin not allowed: %s", strings.TrimSpace(r.Header.Get("Origin"))))
return
}
if !s.authorized(r) {
shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization")
return
}
payload, err := io.ReadAll(r.Body)
if err != nil {
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32600, "invalid body")
return
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
shared.WriteJSONError(w, nil, http.StatusBadRequest, -32700, err.Error())
return
}
result := s.handleRequest(request)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(w).Encode(shared.ResultEnvelope(request.ID, result))
}
func (s *Server) handleRequest(request shared.RPCRequest) map[string]any {
switch strings.TrimSpace(request.Method) {
case "acp.capabilities":
return s.handleCapabilities()
case "session.start", "session.message":
return s.handleSessionRequest(request.Method, request.Params)
case "session.cancel":
return map[string]any{"success": true, "provider": s.providerID, "mode": "single-agent", "accepted": true, "cancelled": false}
case "session.close":
sessionID := strings.TrimSpace(shared.StringArg(request.Params, "sessionId", ""))
return map[string]any{"success": true, "provider": s.providerID, "mode": "single-agent", "accepted": true, "closed": s.closeSession(sessionID)}
default:
resp, err := s.client.Call(request.Method, request.Params)
if err != nil {
return map[string]any{"success": false, "error": err.Error()}
}
result := shared.AsMap(resp["result"])
if len(result) == 0 {
result = resp
}
return result
}
}
func (s *Server) handleCapabilities() map[string]any {
result, err := s.client.Initialize()
if err != nil {
return map[string]any{
"singleAgent": false,
"multiAgent": false,
"providers": []string{},
"capabilities": map[string]any{
"single_agent": false,
"multi_agent": false,
"providers": []string{},
},
"success": false,
"error": err.Error(),
}
}
return map[string]any{
"singleAgent": true,
"multiAgent": false,
"providers": []string{s.providerID},
"capabilities": map[string]any{
"single_agent": true,
"multi_agent": false,
"providers": []string{s.providerID},
},
"success": true,
"result": result,
}
}
type opencodeSessionState struct {
upstreamSessionID string
title string
lastOutput string
}
type initializeResult struct {
ProtocolVersion int `json:"protocolVersion"`
AuthMethods []map[string]any `json:"authMethods"`
AgentCapabilities map[string]any `json:"agentCapabilities"`
}
func (s *Server) handleSessionRequest(method string, params map[string]any) map[string]any {
if _, err := s.client.Initialize(); err != nil {
return map[string]any{
"success": false,
"provider": s.providerID,
"mode": "single-agent",
"error": err.Error(),
}
}
sessionID := strings.TrimSpace(shared.StringArg(params, "sessionId", ""))
if sessionID == "" {
return map[string]any{
"success": false,
"provider": s.providerID,
"mode": "single-agent",
"error": "sessionId is required",
}
}
taskPrompt := strings.TrimSpace(shared.StringArg(params, "taskPrompt", ""))
taskPrompt = shared.AugmentPromptWithAttachments(taskPrompt, params)
if taskPrompt == "" {
return map[string]any{
"success": false,
"provider": s.providerID,
"mode": "single-agent",
"error": "taskPrompt is required",
}
}
state := s.getOrCreateSession(sessionID)
if method == "session.start" {
state = s.resetSession(sessionID)
}
if state.upstreamSessionID == "" {
upstreamSessionID, err := s.client.CreateSession(strings.TrimSpace(shared.StringArg(params, "title", sessionID)))
if err != nil {
return map[string]any{"success": false, "provider": s.providerID, "mode": "single-agent", "error": err.Error()}
}
state.upstreamSessionID = upstreamSessionID
s.setSession(sessionID, state)
}
response, err := s.client.SendMessage(state.upstreamSessionID, taskPrompt, params)
if err != nil {
return map[string]any{"success": false, "provider": s.providerID, "mode": "single-agent", "error": err.Error(), "upstreamSessionId": state.upstreamSessionID}
}
output := strings.TrimSpace(shared.StringArg(response, "message", ""))
if output == "" {
output = strings.TrimSpace(shared.StringArg(response, "output", ""))
}
if output == "" {
output = strings.TrimSpace(shared.StringArg(response, "summary", ""))
}
if output == "" {
output = strings.TrimSpace(shared.StringArg(response, "text", ""))
}
if output == "" {
if result := shared.AsMap(response["result"]); len(result) > 0 {
output = strings.TrimSpace(shared.StringArg(result, "message", ""))
if output == "" {
output = strings.TrimSpace(shared.StringArg(result, "output", ""))
}
}
}
if output == "" {
return map[string]any{"success": false, "provider": s.providerID, "mode": "single-agent", "error": "opencode returned empty response", "upstreamSessionId": state.upstreamSessionID, "upstream": response}
}
state.lastOutput = output
s.setSession(sessionID, state)
result := map[string]any{
"success": true,
"provider": s.providerID,
"mode": "single-agent",
"sessionId": sessionID,
"upstreamSessionId": state.upstreamSessionID,
"output": output,
"summary": output,
"message": output,
}
if method == "session.start" {
result["started"] = true
}
return result
}
func (s *Server) closeSession(sessionID string) bool {
if sessionID == "" {
return false
}
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
if _, ok := s.sessions[sessionID]; !ok {
return false
}
delete(s.sessions, sessionID)
return true
}
func (s *Server) getOrCreateSession(sessionID string) *opencodeSessionState {
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
state := s.sessions[sessionID]
if state == nil {
state = &opencodeSessionState{}
s.sessions[sessionID] = state
}
return state
}
func (s *Server) setSession(sessionID string, state *opencodeSessionState) {
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
s.sessions[sessionID] = state
}
func (s *Server) resetSession(sessionID string) *opencodeSessionState {
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
state := &opencodeSessionState{}
s.sessions[sessionID] = state
return state
}
func (s *Server) authorized(r *http.Request) bool {
if s == nil {
return false
}
if s.authService == nil {
return true
}
return s.authService.ValidateAuthorizationHeader(r.Header.Get("Authorization"))
}
type openCodeClient interface {
Initialize() (initializeResult, error)
Call(method string, params map[string]any) (map[string]any, error)
CreateSession(title string) (string, error)
SendMessage(sessionID, prompt string, params map[string]any) (map[string]any, error)
Close() error
}

87
internal/opencodeadapter/server_test.go Normal file
View File

@ -0,0 +1,87 @@
package opencodeadapter
import (
"testing"
"xworkmate-bridge/internal/shared"
)
type stubOpenCodeClient struct {
initializeCalled int
createTitle string
sendSessionID string
sendPrompt string
sendParams map[string]any
}
func (s *stubOpenCodeClient) Initialize() (initializeResult, error) {
s.initializeCalled++
return initializeResult{ProtocolVersion: 1}, nil
}
func (s *stubOpenCodeClient) Call(method string, params map[string]any) (map[string]any, error) {
return map[string]any{}, nil
}
func (s *stubOpenCodeClient) CreateSession(title string) (string, error) {
s.createTitle = title
return "upstream-1", nil
}
func (s *stubOpenCodeClient) SendMessage(sessionID, prompt string, params map[string]any) (map[string]any, error) {
s.sendSessionID = sessionID
s.sendPrompt = prompt
s.sendParams = params
return map[string]any{"message": "pong"}, nil
}
func (s *stubOpenCodeClient) Close() error { return nil }
func TestHandleSessionStartUsesCreateSessionAndSendMessage(t *testing.T) {
client := &stubOpenCodeClient{}
server := NewServer(client)
result := server.handleRequest(sharedRequest("session.start", map[string]any{
"sessionId": "thread-1",
"taskPrompt": "hello",
"workingDirectory": t.TempDir(),
"title": "demo",
}))
if got := result["sessionId"]; got != "thread-1" {
t.Fatalf("expected bridge session id thread-1, got %v", got)
}
if client.createTitle != "demo" {
t.Fatalf("expected create title demo, got %q", client.createTitle)
}
if client.sendSessionID != "upstream-1" {
t.Fatalf("expected upstream session id upstream-1, got %q", client.sendSessionID)
}
if client.sendPrompt != "hello" {
t.Fatalf("expected prompt hello, got %q", client.sendPrompt)
}
}
func TestHandleSessionMessageReusesSession(t *testing.T) {
client := &stubOpenCodeClient{}
server := NewServer(client)
server.handleRequest(sharedRequest("session.start", map[string]any{
"sessionId": "thread-1",
"taskPrompt": "hello",
}))
result := server.handleRequest(sharedRequest("session.message", map[string]any{
"sessionId": "thread-1",
"taskPrompt": "follow-up",
}))
if got := result["message"]; got != "pong" {
t.Fatalf("expected message pong, got %v", got)
}
if client.sendPrompt != "follow-up" {
t.Fatalf("expected follow-up prompt, got %q", client.sendPrompt)
}
}
func sharedRequest(method string, params map[string]any) shared.RPCRequest {
return shared.RPCRequest{
Method: method,
Params: params,
}
}

13
internal/shared/helpers.go
View File

@ -79,3 +79,16 @@ func BoolArg(raw string, fallback bool) bool {
return fallback
}
}
func AsMap(value any) map[string]any {
if value == nil {
return nil
}
if typed, ok := value.(map[string]any); ok {
return typed
}
if typed, ok := value.(map[string]interface{}); ok {
return typed
}
return nil
}

76
internal/shared/http.go Normal file
View File

@ -0,0 +1,76 @@
package shared
import (
"encoding/json"
"net/http"
"strings"
"github.com/gorilla/websocket"
)
var StandardWSUpgrader = websocket.Upgrader{
ReadBufferSize: 16 * 1024,
WriteBufferSize: 16 * 1024,
CheckOrigin: func(*http.Request) bool {
return true
},
}
func ApplyCORS(w http.ResponseWriter, r *http.Request, allowedOrigins []string) {
origin := strings.TrimSpace(r.Header.Get("Origin"))
if origin == "" || !OriginAllowed(origin, allowedOrigins) {
return
}
headers := w.Header()
headers.Set("Access-Control-Allow-Origin", origin)
headers.Set("Access-Control-Allow-Methods", "POST, OPTIONS, GET")
headers.Set("Access-Control-Allow-Headers", "Authorization, Content-Type, Accept")
headers.Set("Access-Control-Max-Age", "600")
headers.Add("Vary", "Origin")
headers.Add("Vary", "Access-Control-Request-Method")
headers.Add("Vary", "Access-Control-Request-Headers")
}
func OriginAllowed(origin string, allowedOrigins []string) bool {
origin = strings.TrimSpace(origin)
if origin == "" {
return true
}
if len(allowedOrigins) == 0 {
return true
}
for _, allowed := range allowedOrigins {
if strings.HasSuffix(allowed, ":*") {
if strings.HasPrefix(origin, strings.TrimSuffix(allowed, "*")) {
return true
}
continue
}
if origin == allowed {
return true
}
}
return false
}
func WriteJSONError(w http.ResponseWriter, requestID any, statusCode int, code int, message string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(statusCode)
_ = json.NewEncoder(w).Encode(ErrorEnvelope(requestID, code, message))
}
func ParseAllowedOrigins(raw string) []string {
if raw == "" {
return nil
}
parts := strings.Split(raw, ",")
result := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part == "" {
continue
}
result = append(result, part)
}
return result
}

32
internal/skills/command_io.go
View File

@ -123,7 +123,7 @@ func parseCandidates(raw any) []Candidate {
case []any:
result := make([]Candidate, 0, len(typed))
for _, item := range typed {
entry := toMap(item)
entry := shared.AsMap(item)
if len(entry) == 0 {
continue
}
@ -131,7 +131,7 @@ func parseCandidates(raw any) []Candidate {
ID: strings.TrimSpace(stringValue(entry["id"])),
Label: strings.TrimSpace(stringValue(entry["label"])),
Description: strings.TrimSpace(stringValue(entry["description"])),
Installed: boolValue(entry["installed"]),
Installed: shared.BoolArg(stringValue(entry["installed"]), false),
})
}
return dedupeCandidates(result)
@ -146,7 +146,7 @@ func parseCandidates(raw any) []Candidate {
ID: strings.TrimSpace(stringValue(typed["id"])),
Label: strings.TrimSpace(stringValue(typed["label"])),
Description: strings.TrimSpace(stringValue(typed["description"])),
Installed: boolValue(typed["installed"]),
Installed: shared.BoolArg(stringValue(typed["installed"]), false),
}
if entry.ID == "" && entry.Label == "" {
return nil
@ -170,16 +170,6 @@ func routingCandidatesPayload(candidates []Candidate) []map[string]any {
return result
}
func toMap(value any) map[string]any {
if typed, ok := value.(map[string]any); ok {
return typed
}
if typed, ok := value.(map[string]interface{}); ok {
return typed
}
return nil
}
func stringValue(value any) string {
if value == nil {
return ""
@ -191,19 +181,3 @@ func stringValue(value any) string {
return strings.TrimSpace(fmt.Sprint(value))
}
}
func boolValue(value any) bool {
switch typed := value.(type) {
case bool:
return typed
case string:
normalized := strings.ToLower(strings.TrimSpace(typed))
return normalized == "true" || normalized == "1" || normalized == "yes"
case float64:
return typed != 0
case int:
return typed != 0
default:
return false
}
}

194
internal/toolbridge/runner.go
View File

@ -1,194 +0,0 @@
package toolbridge
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"io"
"strings"
"xworkmate-bridge/internal/shared"
)
func Run(input io.Reader, output io.Writer) {
reader := bufio.NewReader(input)
for {
payload, err := readMessage(reader)
if err != nil {
if errors.Is(err, io.EOF) {
return
}
writeError(output, nil, -32700, err.Error())
continue
}
if len(strings.TrimSpace(string(payload))) == 0 {
continue
}
request, err := shared.DecodeRPCRequest(payload)
if err != nil {
writeError(output, nil, -32700, err.Error())
continue
}
response := handleRequest(request)
if response != nil {
writeMessage(output, response)
}
}
}
func readMessage(reader *bufio.Reader) ([]byte, error) {
line, err := reader.ReadString('\n')
if err != nil {
return nil, err
}
line = strings.TrimSpace(line)
if line == "" {
return nil, nil
}
if strings.HasPrefix(strings.ToLower(line), "content-length:") {
var contentLength int
if _, err := fmt.Sscanf(line, "Content-Length: %d", &contentLength); err != nil {
if _, err2 := fmt.Sscanf(line, "content-length: %d", &contentLength); err2 != nil {
return nil, fmt.Errorf("invalid content-length header")
}
}
for {
headerLine, err := reader.ReadString('\n')
if err != nil {
return nil, err
}
if strings.TrimSpace(headerLine) == "" {
break
}
}
body := make([]byte, contentLength)
if _, err := io.ReadFull(reader, body); err != nil {
return nil, err
}
return body, nil
}
return []byte(line), nil
}
func writeMessage(output io.Writer, message map[string]any) {
payload, _ := json.Marshal(message)
_, _ = output.Write(append(payload, '\n'))
}
func writeError(output io.Writer, id any, code int, message string) {
writeMessage(output, shared.ErrorEnvelope(id, code, message))
}
func handleRequest(request shared.RPCRequest) map[string]any {
if request.ID == nil {
return nil
}
switch request.Method {
case "initialize":
return shared.ResultEnvelope(request.ID, map[string]any{
"protocolVersion": "2024-11-05",
"capabilities": map[string]any{
"tools": map[string]any{},
},
"serverInfo": map[string]any{
"name": "xworkmate-go-core",
"version": "0.2.0",
},
})
case "ping":
return shared.ResultEnvelope(request.ID, map[string]any{})
case "tools/list":
return shared.ResultEnvelope(request.ID, map[string]any{
"tools": []map[string]any{
{
"name": "chat",
"description": "OpenAI-compatible reviewer chat bridge",
"inputSchema": map[string]any{
"type": "object",
"properties": map[string]any{
"prompt": map[string]any{"type": "string"},
"model": map[string]any{"type": "string"},
"system": map[string]any{"type": "string"},
},
"required": []string{"prompt"},
},
},
{
"name": "claude_review",
"description": "Review-only bridge over Claude CLI",
"inputSchema": map[string]any{
"type": "object",
"properties": map[string]any{
"prompt": map[string]any{"type": "string"},
"model": map[string]any{"type": "string"},
"system": map[string]any{"type": "string"},
"tools": map[string]any{"type": "string"},
},
"required": []string{"prompt"},
},
},
{
"name": "vault_kv",
"description": "HashiCorp Vault K/V v2 bridge",
"inputSchema": map[string]any{
"type": "object",
"properties": map[string]any{
"operation": map[string]any{"type": "string"},
"mount": map[string]any{"type": "string"},
"path": map[string]any{"type": "string"},
"data": map[string]any{"type": "object"},
"cas": map[string]any{"type": "number"},
},
"required": []string{"operation", "path"},
},
},
},
})
case "tools/call":
var params shared.ToolCallParams
raw, _ := json.Marshal(request.Params)
if err := json.Unmarshal(raw, &params); err != nil {
return shared.ErrorResponse(
request.ID,
-32602,
fmt.Sprintf("invalid tool params: %v", err),
)
}
switch params.Name {
case "chat":
content, err := shared.HandleChatTool(params.Arguments)
if err != nil {
return shared.ToolErrorResult(request.ID, err)
}
return shared.ToolTextResult(request.ID, content)
case "claude_review":
content, err := shared.HandleClaudeReviewTool(params.Arguments)
if err != nil {
return shared.ToolErrorResult(request.ID, err)
}
return shared.ToolTextResult(request.ID, content)
case "vault_kv":
content, err := shared.HandleVaultKVTool(params.Arguments)
if err != nil {
return shared.ToolErrorResult(request.ID, err)
}
return shared.ToolTextResult(request.ID, content)
default:
return shared.ErrorResponse(
request.ID,
-32601,
fmt.Sprintf("unknown tool: %s", params.Name),
)
}
default:
return shared.ErrorResponse(
request.ID,
-32601,
fmt.Sprintf("unknown method: %s", request.Method),
)
}
}

80
internal/toolbridge/runner_test.go
View File

@ -1,80 +0,0 @@
package toolbridge
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"xworkmate-bridge/internal/shared"
)
func TestHandleRequestListsVaultKVTool(t *testing.T) {
t.Parallel()
response := handleRequest(sharedRequest("tools/list", nil))
result := mapStringAny(response["result"])
tools := result["tools"].([]map[string]any)
found := false
for _, tool := range tools {
if tool["name"] == "vault_kv" {
found = true
break
}
}
if !found {
t.Fatalf("expected vault_kv tool in %v", tools)
}
}
func TestHandleRequestCallsVaultKVTool(t *testing.T) {
var requestPath string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requestPath = r.URL.Path
_ = json.NewEncoder(w).Encode(map[string]any{
"data": map[string]any{
"data": map[string]any{
"demo": "value",
},
},
})
}))
defer server.Close()
t.Setenv("VAULT_SERVER_URL", server.URL)
t.Setenv("VAULT_SERVER_ROOT_ACCESS_TOKEN", "root-token")
response := handleRequest(sharedRequest("tools/call", map[string]any{
"name": "vault_kv",
"arguments": map[string]any{
"operation": "read",
"path": "apps/demo",
},
}))
result := mapStringAny(response["result"])
content := result["content"].([]map[string]any)
text := strings.TrimSpace(content[0]["text"].(string))
if !strings.Contains(text, `"demo": "value"`) {
t.Fatalf("unexpected tool output: %s", text)
}
if requestPath != "/v1/secret/data/apps/demo" {
t.Fatalf("unexpected request path: %s", requestPath)
}
}
func sharedRequest(method string, params map[string]any) shared.RPCRequest {
return shared.RPCRequest{
JSONRPC: "2.0",
ID: 1,
Method: method,
Params: params,
}
}
func mapStringAny(raw any) map[string]any {
if typed, ok := raw.(map[string]any); ok {
return typed
}
return map[string]any{}
}

113
main.go
View File

@ -8,66 +8,95 @@ import (
"xworkmate-bridge/internal/acp"
"xworkmate-bridge/internal/geminiadapter"
"xworkmate-bridge/internal/hermesadapter"
"xworkmate-bridge/internal/toolbridge"
"xworkmate-bridge/internal/opencodeadapter"
)
var (
buildCommit = ""
buildVersion = "v1.0-beta2"
buildVersion = "v1.1.0"
buildDate = ""
)
func main() {
if len(os.Args) > 1 {
switch os.Args[1] {
case "-v", "--version":
if err := printBridgeVersionInfo(); err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
return
}
}
if len(os.Args) > 1 && os.Args[1] == "serve" {
if err := acp.Serve(os.Args[2:]); err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
return
}
if len(os.Args) > 1 && os.Args[1] == "acp-stdio" {
acp.RunStdio(os.Stdin, os.Stdout)
return
}
if len(os.Args) > 1 && os.Args[1] == "gemini-acp-adapter" {
if err := geminiadapter.Serve(os.Args[2:]); err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
return
}
if len(os.Args) > 1 && os.Args[1] == "hermes-acp-adapter" {
if err := hermesadapter.Serve(os.Args[2:]); err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)
}
return
if len(os.Args) < 2 {
printUsage()
os.Exit(1)
}
toolbridge.Run(os.Stdin, os.Stdout)
cmd := os.Args[1]
args := os.Args[2:]
switch cmd {
case "serve":
if err := acp.Serve(args); err != nil {
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
os.Exit(1)
}
case "adapter":
handleAdapterCommand(args)
case "stdio":
acp.RunStdio(os.Stdin, os.Stdout)
case "version", "-v", "--version":
printBridgeVersionInfo()
default:
// Backward compatibility for old subcommands (optional, but we said no backward compatibility)
// However, for the transition, we can be nice or just fail.
// The user said "彻底清理陈旧代码", so I'll just fail with a help message.
fmt.Fprintf(os.Stderr, "Unknown command: %s\n", cmd)
printUsage()
os.Exit(1)
}
}
func handleAdapterCommand(args []string) {
if len(args) < 1 {
fmt.Fprintf(os.Stderr, "Usage: xworkmate-bridge adapter <type> [options]\n")
fmt.Fprintf(os.Stderr, "Supported types: gemini, hermes, opencode\n")
os.Exit(1)
}
adapterType := args[0]
adapterArgs := args[1:]
var err error
switch adapterType {
case "gemini":
err = geminiadapter.Serve(adapterArgs)
case "hermes":
err = hermesadapter.Serve(adapterArgs)
case "opencode":
err = opencodeadapter.Serve(adapterArgs)
default:
fmt.Fprintf(os.Stderr, "Unknown adapter type: %s\n", adapterType)
os.Exit(1)
}
if err != nil {
fmt.Fprintf(os.Stderr, "Adapter error: %v\n", err)
os.Exit(1)
}
}
func printUsage() {
fmt.Printf("xworkmate-bridge %s\n\n", buildVersion)
fmt.Println("Usage:")
fmt.Println(" xworkmate-bridge serve [options] Start the main ACP bridge server")
fmt.Println(" xworkmate-bridge adapter <type> [options] Start a specific adapter (gemini, hermes, opencode)")
fmt.Println(" xworkmate-bridge stdio Run the bridge over stdio")
fmt.Println(" xworkmate-bridge version Print version info")
}
func printBridgeVersionInfo() error {
payload := map[string]any{
"status": "ok",
"commit": buildCommit,
"version": buildVersion,
"status": "ok",
"commit": buildCommit,
"version": buildVersion,
"build-date": buildDate,
}
encoded, err := json.Marshal(payload)
if err != nil {
return err
}
_, err = os.Stdout.Write(append(encoded, '\n'))
return err
fmt.Println(string(encoded))
return nil
}

6
tests/bridge_functional_tests.sh
View File

@ -97,11 +97,11 @@ assert_success "$CONNECT_RES" "Gateway 连接"
log_step "3. Gemini 流式对话测试"
log_info "正在向 Gemini 发起对话..."
START_GEMINI=$(call_api "/acp/rpc" "session.start" "{\"sessionId\":\"$SESSION_ID_GEMINI\",\"taskPrompt\":\"你好，请记住我叫 Gemini-Tester。\",\"routing\":{\"explicitProviderId\":\"gemini\"}}" "true" | tail -n 2 | head -n 1 | sed 's/^data: //')
START_GEMINI=$(call_api "/acp/rpc" "session.start" "{\"sessionId\":\"$SESSION_ID_GEMINI\",\"taskPrompt\":\"你好，请记住我叫 Gemini-Tester。\",\"routing\":{\"explicitProviderId\":\"gemini\"}}" "true" | grep "^data: {" | tail -n 1 | sed 's/^data: //')
assert_success "$START_GEMINI" "Gemini 会话启动"
log_info "验证 Gemini 上下文..."
MSG_GEMINI=$(call_api "/acp/rpc" "session.message" "{\"sessionId\":\"$SESSION_ID_GEMINI\",\"taskPrompt\":\"我刚才说我叫什么？\",\"routing\":{\"explicitProviderId\":\"gemini\"}}" "true" | tail -n 2 | head -n 1 | sed 's/^data: //')
MSG_GEMINI=$(call_api "/acp/rpc" "session.message" "{\"sessionId\":\"$SESSION_ID_GEMINI\",\"taskPrompt\":\"我刚才说我叫什么？\",\"routing\":{\"explicitProviderId\":\"gemini\"}}" "true" | grep "^data: {" | tail -n 1 | sed 's/^data: //')
assert_success "$MSG_GEMINI" "Gemini 上下文验证"
log_info "Gemini 回复: $(echo "$MSG_GEMINI" | jq -r '.result.output // .payload.output')"
@ -110,7 +110,7 @@ log_info "Gemini 回复: $(echo "$MSG_GEMINI" | jq -r '.result.output // .payloa
log_step "4. OpenCode 深度测试 (通过 Gateway)"
log_info "正在通过 Gateway 向 OpenCode 发起对话..."
# 注意：OpenCode 需要通过已连接的 gateway 执行
START_OPENCODE=$(call_api "/acp/rpc" "session.start" "{\"sessionId\":\"$SESSION_ID_OPENCODE\",\"taskPrompt\":\"你好 OpenCode，请问你能做什么？\",\"routing\":{\"explicitProviderId\":\"opencode\"}}" "true" | tail -n 2 | head -n 1 | sed 's/^data: //')
START_OPENCODE=$(call_api "/acp/rpc" "session.start" "{\"sessionId\":\"$SESSION_ID_OPENCODE\",\"taskPrompt\":\"你好 OpenCode，请问你能做什么？\",\"routing\":{\"explicitProviderId\":\"opencode\"}}" "true" | grep "^data: {" | tail -n 1 | sed 's/^data: //')
# 如果 gateway 依然报错，可能是环境限制，此处做容错处理
if echo "$START_OPENCODE" | jq -e '.ok == true or .result.success == true' > /dev/null; then