Compare commits
No commits in common. "main" and "release/v1.1.4" have entirely different histories.
main
...
release/v1
38
.github/workflows/pipeline.yml
vendored
38
.github/workflows/pipeline.yml
vendored
@ -21,7 +21,7 @@ on:
|
||||
required: true
|
||||
default: true
|
||||
type: boolean
|
||||
ai_workspace_auth_token:
|
||||
internal_service_token:
|
||||
description: "Optional ACP auth token for deploy"
|
||||
required: false
|
||||
default: ""
|
||||
@ -56,7 +56,6 @@ jobs:
|
||||
- name: Load Vault secrets
|
||||
id: vault
|
||||
if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }}
|
||||
continue-on-error: true
|
||||
uses: hashicorp/vault-action@v2
|
||||
with:
|
||||
url: ${{ env.VAULT_ADDR }}
|
||||
@ -65,28 +64,17 @@ jobs:
|
||||
jwtGithubAudience: vault
|
||||
ignoreNotFound: true
|
||||
secrets: |
|
||||
kv/data/github-actions/xworkmate-bridge AI_WORKSPACE_AUTH_TOKEN | AI_WORKSPACE_AUTH_TOKEN
|
||||
kv/data/github-actions/xworkmate-bridge INTERNAL_SERVICE_TOKEN | INTERNAL_SERVICE_TOKEN
|
||||
|
||||
- name: Export bridge auth token
|
||||
if: ${{ steps.vault.outcome == 'success' && steps.vault.outputs.AI_WORKSPACE_AUTH_TOKEN != '' }}
|
||||
run: echo "AI_WORKSPACE_AUTH_TOKEN=${{ steps.vault.outputs.AI_WORKSPACE_AUTH_TOKEN }}" >> "$GITHUB_ENV"
|
||||
if: ${{ steps.vault.outcome == 'success' }}
|
||||
run: echo "BRIDGE_AUTH_TOKEN=${{ steps.vault.outputs.INTERNAL_SERVICE_TOKEN }}" >> "$GITHUB_ENV"
|
||||
|
||||
- name: Probe current production bridge
|
||||
id: production_state
|
||||
env:
|
||||
BRIDGE_SERVER_URL: https://xworkmate-bridge.svc.plus
|
||||
run: |
|
||||
if [[ -z "${AI_WORKSPACE_AUTH_TOKEN:-}" ]]; then
|
||||
echo "::notice title=Production state skipped::AI_WORKSPACE_AUTH_TOKEN is unavailable from Vault; continuing without production bridge metadata."
|
||||
{
|
||||
echo "production_image="
|
||||
echo "production_tag="
|
||||
echo "production_commit="
|
||||
echo "production_version="
|
||||
} >> "$GITHUB_OUTPUT"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
while IFS='=' read -r key value; do
|
||||
echo "${key}=${value}" >> "$GITHUB_OUTPUT"
|
||||
done < <(bash ./scripts/github-actions/report-production-state.sh "${BRIDGE_SERVER_URL}")
|
||||
@ -232,7 +220,7 @@ jobs:
|
||||
jwtGithubAudience: vault
|
||||
ignoreNotFound: true
|
||||
secrets: |
|
||||
kv/data/github-actions/xworkmate-bridge AI_WORKSPACE_AUTH_TOKEN | AI_WORKSPACE_AUTH_TOKEN ;
|
||||
kv/data/github-actions/xworkmate-bridge INTERNAL_SERVICE_TOKEN | INTERNAL_SERVICE_TOKEN ;
|
||||
kv/data/github-actions/xworkmate-bridge WORKSPACE_REPO_TOKEN | WORKSPACE_REPO_TOKEN ;
|
||||
kv/data/github-actions/xworkmate-bridge SINGLE_NODE_VPS_SSH_PRIVATE_KEY | SINGLE_NODE_VPS_SSH_PRIVATE_KEY ;
|
||||
kv/data/github-actions/xworkmate-bridge SINGLE_NODE_VPS_SSH_PRIVATE_KEY_B64 | SINGLE_NODE_VPS_SSH_PRIVATE_KEY_B64 ;
|
||||
@ -241,20 +229,20 @@ jobs:
|
||||
- name: Export deploy secrets
|
||||
run: |
|
||||
{
|
||||
if [[ "${{ github.event_name }}" == "workflow_dispatch" && -n "${{ inputs.ai_workspace_auth_token }}" ]]; then
|
||||
echo "AI_WORKSPACE_AUTH_TOKEN=${{ inputs.ai_workspace_auth_token }}"
|
||||
if [[ "${{ github.event_name }}" == "workflow_dispatch" && -n "${{ inputs.internal_service_token }}" ]]; then
|
||||
echo "BRIDGE_AUTH_TOKEN=${{ inputs.internal_service_token }}"
|
||||
else
|
||||
echo "AI_WORKSPACE_AUTH_TOKEN=${{ steps.vault.outputs.AI_WORKSPACE_AUTH_TOKEN }}"
|
||||
echo "BRIDGE_AUTH_TOKEN=${{ steps.vault.outputs.INTERNAL_SERVICE_TOKEN }}"
|
||||
fi
|
||||
} >> "$GITHUB_ENV"
|
||||
|
||||
- name: Validate deploy secrets
|
||||
run: |
|
||||
if [[ -z "${AI_WORKSPACE_AUTH_TOKEN}" ]]; then
|
||||
echo "::error::AI_WORKSPACE_AUTH_TOKEN is empty. Provide it via the workflow_dispatch input, or ensure kv/data/github-actions/xworkmate-bridge AI_WORKSPACE_AUTH_TOKEN is readable from Vault."
|
||||
if [[ -z "${BRIDGE_AUTH_TOKEN}" ]]; then
|
||||
echo "::error::BRIDGE_AUTH_TOKEN is empty. Provide it via the workflow_dispatch input, or ensure kv/data/github-actions/xworkmate-bridge INTERNAL_SERVICE_TOKEN is readable from Vault."
|
||||
exit 1
|
||||
fi
|
||||
echo "AI_WORKSPACE_AUTH_TOKEN length=${#AI_WORKSPACE_AUTH_TOKEN}"
|
||||
echo "BRIDGE_AUTH_TOKEN length=${#BRIDGE_AUTH_TOKEN}"
|
||||
|
||||
- name: Checkout playbooks repository
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
@ -396,10 +384,10 @@ jobs:
|
||||
jwtGithubAudience: vault
|
||||
ignoreNotFound: true
|
||||
secrets: |
|
||||
kv/data/github-actions/xworkmate-bridge AI_WORKSPACE_AUTH_TOKEN | AI_WORKSPACE_AUTH_TOKEN
|
||||
kv/data/github-actions/xworkmate-bridge INTERNAL_SERVICE_TOKEN | INTERNAL_SERVICE_TOKEN
|
||||
|
||||
- name: Export bridge auth token
|
||||
run: echo "AI_WORKSPACE_AUTH_TOKEN=${{ steps.vault.outputs.AI_WORKSPACE_AUTH_TOKEN }}" >> "$GITHUB_ENV"
|
||||
run: echo "BRIDGE_AUTH_TOKEN=${{ steps.vault.outputs.INTERNAL_SERVICE_TOKEN }}" >> "$GITHUB_ENV"
|
||||
|
||||
- name: Validate deployed endpoints
|
||||
run: bash ./scripts/github-actions/validate-deploy.sh "$(git rev-parse --short HEAD)" "${BRIDGE_SERVER_URL}"
|
||||
|
||||
24
.github/workflows/runtime-release.yml
vendored
24
.github/workflows/runtime-release.yml
vendored
@ -19,12 +19,11 @@ concurrency:
|
||||
|
||||
jobs:
|
||||
build:
|
||||
name: Build ${{ matrix.os }}-${{ matrix.arch }}
|
||||
name: Build linux-${{ matrix.arch }}
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
os: [linux, darwin]
|
||||
arch: [amd64, arm64]
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
@ -39,13 +38,12 @@ jobs:
|
||||
|
||||
- name: Build runtime asset
|
||||
env:
|
||||
TARGET_OS: ${{ matrix.os }}
|
||||
TARGET_ARCH: ${{ matrix.arch }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
root="dist/runtime/xworkmate-bridge"
|
||||
mkdir -p "${root}/bin" dist/assets
|
||||
CGO_ENABLED=0 GOOS="${TARGET_OS}" GOARCH="${TARGET_ARCH}" \
|
||||
CGO_ENABLED=0 GOOS=linux GOARCH="${TARGET_ARCH}" \
|
||||
go build -buildvcs=false -trimpath \
|
||||
-ldflags "-X main.buildCommit=${GITHUB_SHA}" \
|
||||
-o "${root}/bin/xworkmate-go-core" .
|
||||
@ -53,27 +51,21 @@ jobs:
|
||||
{
|
||||
"component": "xworkmate-bridge",
|
||||
"commit": "${GITHUB_SHA}",
|
||||
"os": "${TARGET_OS}",
|
||||
"os": "linux",
|
||||
"arch": "${TARGET_ARCH}",
|
||||
"binary": "bin/xworkmate-go-core"
|
||||
}
|
||||
JSON
|
||||
tar -czf "dist/assets/xworkmate-bridge-${TARGET_OS}-${TARGET_ARCH}.tar.gz" \
|
||||
tar -czf "dist/assets/xworkmate-bridge-linux-${TARGET_ARCH}.tar.gz" \
|
||||
-C dist/runtime xworkmate-bridge
|
||||
(
|
||||
cd dist/assets
|
||||
# Name the per-job checksum file by OS *and* ARCH. Keying on ARCH
|
||||
# alone makes the linux/darwin jobs of the same arch both emit
|
||||
# SHA256SUMS-<arch>, which then clobber each other under the
|
||||
# publish job's merge-multiple download — leaving SHA256SUMS with
|
||||
# only 2 of the 4 platforms and breaking arm64 (and darwin-amd64)
|
||||
# consumers with "missing checksum".
|
||||
sha256sum -- ./*.tar.gz | sed 's# \./# #' > "SHA256SUMS-${TARGET_OS}-${TARGET_ARCH}"
|
||||
sha256sum -- ./*.tar.gz | sed 's# \./# #' > "SHA256SUMS-${TARGET_ARCH}"
|
||||
)
|
||||
|
||||
- uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: xworkmate-bridge-${{ matrix.os }}-${{ matrix.arch }}
|
||||
name: xworkmate-bridge-linux-${{ matrix.arch }}
|
||||
path: |
|
||||
dist/assets/*.tar.gz
|
||||
dist/assets/SHA256SUMS-*
|
||||
@ -85,7 +77,7 @@ jobs:
|
||||
steps:
|
||||
- uses: actions/download-artifact@v4
|
||||
with:
|
||||
pattern: xworkmate-bridge-*
|
||||
pattern: xworkmate-bridge-linux-*
|
||||
path: dist
|
||||
merge-multiple: true
|
||||
|
||||
@ -105,5 +97,5 @@ jobs:
|
||||
--repo "${GITHUB_REPOSITORY}" \
|
||||
--target "${GITHUB_SHA}" \
|
||||
--title "XWorkmate Bridge runtime ${GITHUB_SHA::12}" \
|
||||
--notes "Prebuilt bridge binaries. No target-host Go build is required."
|
||||
--notes "Prebuilt Linux bridge binaries. No target-host Go build is required."
|
||||
fi
|
||||
|
||||
44
.github/workflows/validate-release-pr.yml
vendored
44
.github/workflows/validate-release-pr.yml
vendored
@ -1,44 +0,0 @@
|
||||
name: Validate Release PR
|
||||
|
||||
# release/* 分支的发布策略门禁:仅接受 hotfix/* 或带 cherry-pick/backport 标签的 PR。
|
||||
# 详见 iac_modules/docs/tldr-github-branch-model.md
|
||||
on:
|
||||
pull_request_target:
|
||||
types: [opened, synchronize, reopened, labeled, unlabeled]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
pull-requests: read
|
||||
|
||||
jobs:
|
||||
validate-release-source:
|
||||
runs-on: ubuntu-latest
|
||||
if: startsWith(github.base_ref, 'release/')
|
||||
steps:
|
||||
- name: Check PR source branch
|
||||
run: |
|
||||
SRC="${{ github.head_ref }}"
|
||||
TGT="${{ github.base_ref }}"
|
||||
LABELS="${{ join(github.event.pull_request.labels.*.name, ',') }}"
|
||||
|
||||
echo "🔍 Validating PR into release branch"
|
||||
echo " source: $SRC"
|
||||
echo " target: $TGT"
|
||||
echo " labels: $LABELS"
|
||||
|
||||
if [[ "$SRC" =~ ^hotfix/ ]]; then
|
||||
echo "✅ Allowed: hotfix/* branch"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "$LABELS" =~ (^|,)(cherry-pick|backport)(,|$) ]]; then
|
||||
echo "✅ Allowed: cherry-pick/backport labeled PR"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
echo "❌ Rejected."
|
||||
echo "release/* 仅接受:"
|
||||
echo " - 来自 hotfix/* 的 PR"
|
||||
echo " - 带 cherry-pick 或 backport 标签的 PR(已验证 feature 的 backport/cherry-pick)"
|
||||
echo "禁止从 main / develop / feature/* 直接合并到 release/*。"
|
||||
exit 1
|
||||
@ -94,7 +94,7 @@ Optional GitHub secrets:
|
||||
|
||||
Optional workflow input:
|
||||
|
||||
- `ai_workspace_auth_token`: manual dispatch input that is forwarded as `AI_WORKSPACE_AUTH_TOKEN`
|
||||
- `internal_service_token`: manual dispatch input that is forwarded to Ansible as `INTERNAL_SERVICE_TOKEN`
|
||||
|
||||
## Environment
|
||||
|
||||
|
||||
@ -39,8 +39,7 @@
|
||||
|
||||
环境变量:
|
||||
|
||||
- `AI_WORKSPACE_AUTH_TOKEN`:主共享 token,用于 bridge 入站鉴权、上游 provider 转发、OpenClaw Gateway 重签发与任务转发 fallback。
|
||||
- `BRIDGE_AUTH_TOKEN`:旧主 token。没有 `AI_WORKSPACE_AUTH_TOKEN` 时继续生效并参与上游转发,用于存量租户兼容,直到 `AI_WORKSPACE_AUTH_TOKEN` 完成彻底替代后下线。
|
||||
- `BRIDGE_AUTH_TOKEN`
|
||||
- `BRIDGE_REVIEW_AUTH_TOKEN`(可选):Apple review / beta 工测专用临时 token。清空该环境变量并重启/reload bridge 即可单独关停,不影响主 token。
|
||||
- `ACP_ALLOWED_ORIGINS`
|
||||
|
||||
@ -49,9 +48,9 @@
|
||||
- `/acp` 与 `/acp/rpc` 都做 origin allowlist 校验
|
||||
- 空 `Origin` 默认允许
|
||||
- `/api/ping`、`/acp`、`/acp/rpc` 在任一 bridge token 非空时都要求 bearer header
|
||||
- `AI_WORKSPACE_AUTH_TOKEN`、`BRIDGE_AUTH_TOKEN` 与 `BRIDGE_REVIEW_AUTH_TOKEN` 都为空时默认放行
|
||||
- `BRIDGE_AUTH_TOKEN` 与 `BRIDGE_REVIEW_AUTH_TOKEN` 都为空时默认放行
|
||||
- token 非空时,接受裸 token 或 `Bearer <token>`
|
||||
- 线上 Caddy 入口必须与 bridge origin 保持同一 token set:主 `AI_WORKSPACE_AUTH_TOKEN`、兼容 `BRIDGE_AUTH_TOKEN` 与可选 `BRIDGE_REVIEW_AUTH_TOKEN` 都应放行;无 token 仍返回 `401`
|
||||
- 线上 Caddy 入口必须与 bridge origin 保持同一 token set:主 `BRIDGE_AUTH_TOKEN` 与可选 `BRIDGE_REVIEW_AUTH_TOKEN` 都应放行;无 token 仍返回 `401`
|
||||
- `xworkmate-app` 生产 Origin 固定为 `https://xworkmate.svc.plus`
|
||||
|
||||
## 3.1 Lightweight Distributed Task Forwarding
|
||||
@ -140,7 +139,7 @@ distributed:
|
||||
- `bridge_endpoint` 是 peer bridge base URL,bridge 会按当前请求路径拼接 `/acp/rpc` 或 `/gateway/openclaw`
|
||||
- 同步消息不能走公网;`bridge_endpoint` 必须是 loopback、private、link-local 这类本机或 VPN 内网地址,用于 WireGuard over VLESS 等隧道已经提供加密的场景
|
||||
- 只要求本机网络能路由到 endpoint;bridge 不依赖 config center 或额外注册中心
|
||||
- `task_forward_token` 为空时复用本机 `AI_WORKSPACE_AUTH_TOKEN`;未配置时兼容复用 `BRIDGE_AUTH_TOKEN`
|
||||
- `task_forward_token` 为空时复用本机 `BRIDGE_AUTH_TOKEN`
|
||||
- 转发请求会带 `X-XWorkmate-Bridge-Forwarded: 1`
|
||||
- `X-XWorkmate-Forward-Source` 是源节点,`X-XWorkmate-Forward-Target` 是最终目标节点
|
||||
- `X-XWorkmate-Forward-Hop` 逐跳递增,超过 `forwarding.hop_limit` 时拒绝转发,避免循环
|
||||
@ -158,7 +157,7 @@ distributed:
|
||||
BRIDGE_SERVER_URL=https://xworkmate-bridge.svc.plus
|
||||
BRIDGE_WS_URL=wss://xworkmate-bridge.svc.plus/acp
|
||||
BRIDGE_HTTP_RPC_URL=https://xworkmate-bridge.svc.plus/acp/rpc
|
||||
Authorization: Bearer $AI_WORKSPACE_AUTH_TOKEN
|
||||
Authorization: Bearer $BRIDGE_AUTH_TOKEN
|
||||
Origin: https://xworkmate.svc.plus
|
||||
```
|
||||
|
||||
|
||||
@ -109,7 +109,7 @@ Gateway access remains bridge-owned via JSON-RPC methods:
|
||||
|
||||
Upstream authentication is unified for both ACP and gateway routes:
|
||||
|
||||
- `Authorization: Bearer $AI_WORKSPACE_AUTH_TOKEN`
|
||||
- `Authorization: Bearer $INTERNAL_SERVICE_TOKEN`
|
||||
|
||||
## Consequences
|
||||
|
||||
|
||||
@ -109,43 +109,15 @@ func resolveURL(yamlVal string, envKeys ...string) string {
|
||||
}
|
||||
|
||||
func bridgeUpstreamAuthorizationHeader() string {
|
||||
token := bridgePublicAuthToken()
|
||||
token := bridgeSharedAuthToken()
|
||||
if token != "" && !strings.HasPrefix(strings.ToLower(token), "bearer ") {
|
||||
return "Bearer " + token
|
||||
}
|
||||
return token
|
||||
}
|
||||
|
||||
func bridgePublicAuthToken() string {
|
||||
if token := strings.TrimSpace(os.Getenv("AI_WORKSPACE_AUTH_TOKEN")); token != "" {
|
||||
return token
|
||||
}
|
||||
return strings.TrimSpace(shared.EnvOrDefault("BRIDGE_AUTH_TOKEN", ""))
|
||||
}
|
||||
|
||||
func bridgeSharedAuthToken() string {
|
||||
return bridgePublicAuthToken()
|
||||
}
|
||||
|
||||
func bridgeInboundAuthTokens() []string {
|
||||
var tokens []string
|
||||
seen := map[string]struct{}{}
|
||||
for _, token := range []string{
|
||||
os.Getenv("AI_WORKSPACE_AUTH_TOKEN"),
|
||||
os.Getenv("BRIDGE_AUTH_TOKEN"),
|
||||
os.Getenv("BRIDGE_REVIEW_AUTH_TOKEN"),
|
||||
} {
|
||||
trimmed := strings.TrimSpace(token)
|
||||
if trimmed == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[trimmed]; ok {
|
||||
continue
|
||||
}
|
||||
seen[trimmed] = struct{}{}
|
||||
tokens = append(tokens, trimmed)
|
||||
}
|
||||
return tokens
|
||||
return strings.TrimSpace(shared.EnvOrDefault("BRIDGE_AUTH_TOKEN", ""))
|
||||
}
|
||||
|
||||
func resolveDistributedTaskForwardToken(config *BridgeConfig) string {
|
||||
|
||||
@ -92,7 +92,7 @@ func handleGatewayConnect(
|
||||
server.gateway = gatewayruntime.NewManager()
|
||||
}
|
||||
|
||||
result := connectOpenClawGateway(server.gateway, request, notify, usesBridgeIdentity)
|
||||
result := server.gateway.Connect(request, notify)
|
||||
if result.OK && usesBridgeIdentity {
|
||||
saveBridgeGatewayDeviceToken(result.ReturnedDeviceToken)
|
||||
}
|
||||
@ -284,7 +284,7 @@ func ensureProductionGatewayConnected(
|
||||
request.Auth.DeviceToken = deviceToken
|
||||
request.HasDeviceToken = deviceToken != ""
|
||||
request.ReportedRemoteAddress = resolveGatewayReportedRemoteAddress(server, request)
|
||||
result := connectOpenClawGateway(server.gateway, request, notify, true)
|
||||
result := server.gateway.Connect(request, notify)
|
||||
if result.OK {
|
||||
saveBridgeGatewayDeviceToken(result.ReturnedDeviceToken)
|
||||
return nil
|
||||
@ -297,43 +297,6 @@ func ensureProductionGatewayConnected(
|
||||
return &shared.RPCError{Code: -32002, Message: "GATEWAY_CONNECT_FAILED: " + message}
|
||||
}
|
||||
|
||||
func connectOpenClawGateway(
|
||||
manager *gatewayruntime.Manager,
|
||||
request gatewayruntime.ConnectRequest,
|
||||
notify func(map[string]any),
|
||||
usesBridgeIdentity bool,
|
||||
) gatewayruntime.ConnectResult {
|
||||
result := manager.Connect(request, notify)
|
||||
if !usesBridgeIdentity || !shouldRetryOpenClawGatewayWithSharedToken(result) {
|
||||
return result
|
||||
}
|
||||
|
||||
clearBridgeGatewayDeviceToken()
|
||||
request.Auth.DeviceToken = ""
|
||||
request.HasDeviceToken = false
|
||||
request.Auth.Token = bridgeSharedAuthToken()
|
||||
request.HasSharedAuth = true
|
||||
request.ConnectAuthMode = "shared-token"
|
||||
request.ConnectAuthFields = []string{"token"}
|
||||
request.ConnectAuthSources = []string{"bridge:device-token-reissue"}
|
||||
return manager.Connect(request, notify)
|
||||
}
|
||||
|
||||
func shouldRetryOpenClawGatewayWithSharedToken(result gatewayruntime.ConnectResult) bool {
|
||||
if result.OK || strings.TrimSpace(bridgeSharedAuthToken()) == "" {
|
||||
return false
|
||||
}
|
||||
code := strings.ToUpper(strings.TrimSpace(shared.StringArg(result.Error, "code", "")))
|
||||
message := strings.ToLower(strings.TrimSpace(shared.StringArg(result.Error, "message", "")))
|
||||
details := shared.AsMap(result.Error["details"])
|
||||
detailCode := strings.ToUpper(strings.TrimSpace(shared.StringArg(details, "code", "")))
|
||||
return detailCode == "AUTH_DEVICE_TOKEN_MISMATCH" ||
|
||||
detailCode == "PAIRING_REQUIRED" ||
|
||||
code == "NOT_PAIRED" ||
|
||||
strings.Contains(message, "device token mismatch") ||
|
||||
strings.Contains(message, "rotate/reissue device token")
|
||||
}
|
||||
|
||||
func configureProductionOpenClawGatewayRuntime(manager *gatewayruntime.Manager) {
|
||||
if manager == nil {
|
||||
return
|
||||
|
||||
@ -138,20 +138,6 @@ func saveBridgeGatewayDeviceToken(deviceToken string) {
|
||||
)
|
||||
}
|
||||
|
||||
func clearBridgeGatewayDeviceToken() {
|
||||
bridgeGatewayIdentity.Lock()
|
||||
defer bridgeGatewayIdentity.Unlock()
|
||||
if strings.TrimSpace(bridgeGatewayIdentity.value.DeviceID) == "" {
|
||||
return
|
||||
}
|
||||
bridgeGatewayIdentity.deviceToken = ""
|
||||
_ = persistBridgeGatewayIdentity(
|
||||
bridgeGatewayIdentityPath(),
|
||||
bridgeGatewayIdentity.value,
|
||||
"",
|
||||
)
|
||||
}
|
||||
|
||||
func persistBridgeGatewayIdentity(
|
||||
path string,
|
||||
identity gatewayruntime.DeviceIdentity,
|
||||
|
||||
@ -144,26 +144,6 @@ func TestBridgeGatewayIdentityPersistsReturnedDeviceToken(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBridgeGatewayIdentityClearsStoredDeviceToken(t *testing.T) {
|
||||
identityPath := filepath.Join(t.TempDir(), "openclaw-device.json")
|
||||
t.Setenv("XWORKMATE_BRIDGE_OPENCLAW_IDENTITY_PATH", identityPath)
|
||||
resetBridgeGatewayIdentityForTest()
|
||||
t.Cleanup(resetBridgeGatewayIdentityForTest)
|
||||
|
||||
identity := newBridgeGatewayIdentity()
|
||||
saveBridgeGatewayDeviceToken("device-token-2")
|
||||
clearBridgeGatewayDeviceToken()
|
||||
resetBridgeGatewayIdentityForTest()
|
||||
|
||||
reloaded, token := bridgeGatewayOpenClawCredentials()
|
||||
if reloaded.DeviceID != identity.DeviceID {
|
||||
t.Fatalf("reloaded identity = %q, want %q", reloaded.DeviceID, identity.DeviceID)
|
||||
}
|
||||
if token != "" {
|
||||
t.Fatalf("device token should be cleared, got %q", token)
|
||||
}
|
||||
}
|
||||
|
||||
func resetBridgeGatewayIdentityForTest() {
|
||||
bridgeGatewayIdentity.Lock()
|
||||
defer bridgeGatewayIdentity.Unlock()
|
||||
|
||||
@ -108,117 +108,3 @@ func TestSystemLogsConnectsProductionGatewayForStatus(t *testing.T) {
|
||||
t.Fatalf("expected connected status to reuse gateway session, got %d connect attempts", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProductionGatewayReconnectsWithSharedTokenAfterDeviceTokenMismatch(t *testing.T) {
|
||||
gateway := newAcpFakeOpenClawGateway(t)
|
||||
defer gateway.Close()
|
||||
gateway.rejectDeviceTokenOnce.Store(true)
|
||||
|
||||
identityPath := filepath.Join(t.TempDir(), "openclaw-device.json")
|
||||
t.Setenv("GATEWAY_RPC_URL", gateway.URL())
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", filepath.Join(t.TempDir(), "missing-config.yaml"))
|
||||
t.Setenv("XWORKMATE_BRIDGE_OPENCLAW_IDENTITY_PATH", identityPath)
|
||||
resetBridgeGatewayIdentityForTest()
|
||||
t.Cleanup(resetBridgeGatewayIdentityForTest)
|
||||
|
||||
_ = newBridgeGatewayIdentity()
|
||||
saveBridgeGatewayDeviceToken("stale-device-token")
|
||||
server := NewServer()
|
||||
|
||||
result, rpcErr := server.handleRequest(
|
||||
shared.RPCRequest{
|
||||
ID: "status",
|
||||
Method: "system.logs",
|
||||
Params: map[string]any{},
|
||||
},
|
||||
func(map[string]any) {},
|
||||
)
|
||||
if rpcErr != nil {
|
||||
t.Fatalf("system.logs returned rpc error: %#v", rpcErr)
|
||||
}
|
||||
if got := result["gatewayStatus"]; got != "connected" {
|
||||
t.Fatalf("expected gatewayStatus connected after repair, got %#v", result)
|
||||
}
|
||||
if got := gateway.ConnectCount(); got != 2 {
|
||||
t.Fatalf("expected stale device token retry with shared token, got %d connects", got)
|
||||
}
|
||||
resetBridgeGatewayIdentityForTest()
|
||||
_, token := bridgeGatewayOpenClawCredentials()
|
||||
if token != "device-token-1" {
|
||||
t.Fatalf("expected repaired device token to be persisted, got %q", token)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProductionGatewayReconnectPrefersAIWorkspaceToken(t *testing.T) {
|
||||
gateway := newAcpFakeOpenClawGateway(t)
|
||||
defer gateway.Close()
|
||||
gateway.rejectDeviceTokenOnce.Store(true)
|
||||
|
||||
t.Setenv("GATEWAY_RPC_URL", gateway.URL())
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "ai-workspace-token")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", filepath.Join(t.TempDir(), "missing-config.yaml"))
|
||||
t.Setenv("XWORKMATE_BRIDGE_OPENCLAW_IDENTITY_PATH", filepath.Join(t.TempDir(), "openclaw-device.json"))
|
||||
resetBridgeGatewayIdentityForTest()
|
||||
t.Cleanup(resetBridgeGatewayIdentityForTest)
|
||||
|
||||
_ = newBridgeGatewayIdentity()
|
||||
saveBridgeGatewayDeviceToken("stale-device-token")
|
||||
server := NewServer()
|
||||
|
||||
result, rpcErr := server.handleRequest(
|
||||
shared.RPCRequest{
|
||||
ID: "status",
|
||||
Method: "system.logs",
|
||||
Params: map[string]any{},
|
||||
},
|
||||
func(map[string]any) {},
|
||||
)
|
||||
if rpcErr != nil {
|
||||
t.Fatalf("system.logs returned rpc error: %#v", rpcErr)
|
||||
}
|
||||
if got := result["gatewayStatus"]; got != "connected" {
|
||||
t.Fatalf("expected gatewayStatus connected after AI workspace token repair, got %#v", result)
|
||||
}
|
||||
if got := gateway.ConnectCount(); got != 2 {
|
||||
t.Fatalf("expected stale device token retry with AI workspace token, got %d connects", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProductionGatewayDoesNotUseInternalServiceTokenFallback(t *testing.T) {
|
||||
gateway := newAcpFakeOpenClawGateway(t)
|
||||
defer gateway.Close()
|
||||
gateway.rejectDeviceTokenOnce.Store(true)
|
||||
|
||||
t.Setenv("GATEWAY_RPC_URL", gateway.URL())
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "")
|
||||
t.Setenv("INTERNAL_SERVICE_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", filepath.Join(t.TempDir(), "missing-config.yaml"))
|
||||
t.Setenv("XWORKMATE_BRIDGE_OPENCLAW_IDENTITY_PATH", filepath.Join(t.TempDir(), "openclaw-device.json"))
|
||||
resetBridgeGatewayIdentityForTest()
|
||||
t.Cleanup(resetBridgeGatewayIdentityForTest)
|
||||
|
||||
_ = newBridgeGatewayIdentity()
|
||||
saveBridgeGatewayDeviceToken("stale-device-token")
|
||||
server := NewServer()
|
||||
|
||||
result, rpcErr := server.handleRequest(
|
||||
shared.RPCRequest{
|
||||
ID: "status",
|
||||
Method: "system.logs",
|
||||
Params: map[string]any{},
|
||||
},
|
||||
func(map[string]any) {},
|
||||
)
|
||||
if rpcErr != nil {
|
||||
t.Fatalf("system.logs returned rpc error: %#v", rpcErr)
|
||||
}
|
||||
if got := result["gatewayStatus"]; got != "disconnected" {
|
||||
t.Fatalf("expected gatewayStatus disconnected without AI workspace token, got %#v", result)
|
||||
}
|
||||
if got := gateway.ConnectCount(); got != 1 {
|
||||
t.Fatalf("expected no retry with internal token, got %d connects", got)
|
||||
}
|
||||
}
|
||||
|
||||
@ -37,7 +37,6 @@ func (s *Server) Handler() http.Handler {
|
||||
"commit": info.Commit,
|
||||
"version": info.Version,
|
||||
"buildDate": info.BuildDate,
|
||||
"metrics": bridgeStabilityMetricsSnapshot(), // T12
|
||||
}
|
||||
body, _ := json.Marshal(resp)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
@ -1,29 +0,0 @@
|
||||
package acp
|
||||
|
||||
import "sync/atomic"
|
||||
|
||||
// 关键稳定性指标(T12,docs/cases/06 §5)。
|
||||
//
|
||||
// 进程内累计计数,经 /api/ping 暴露,用于把「网关抖动 / run 超时」从靠用户截图
|
||||
// 变为可监控。三个计数对应三类已知的不稳定来源:
|
||||
// - gatewaySocketClosed : gatewayRPCError 命中 OPENCLAW_GATEWAY_SOCKET_CLOSED(连接断)
|
||||
// - taskGetUnconfirmedFallback: tasks.get 走持久 run 仓兜底(gateway 无法确认 run,T7)
|
||||
// - runDeadlineInterrupt : run 超过 DeadlineAt 且 gateway 无法确认,回 interrupted(T9)
|
||||
var bridgeStabilityMetrics struct {
|
||||
gatewaySocketClosed atomic.Int64
|
||||
taskGetUnconfirmedFallback atomic.Int64
|
||||
runDeadlineInterrupt atomic.Int64
|
||||
}
|
||||
|
||||
func metricGatewaySocketClosedInc() { bridgeStabilityMetrics.gatewaySocketClosed.Add(1) }
|
||||
func metricTaskGetUnconfirmedFallbackInc() { bridgeStabilityMetrics.taskGetUnconfirmedFallback.Add(1) }
|
||||
func metricRunDeadlineInterruptInc() { bridgeStabilityMetrics.runDeadlineInterrupt.Add(1) }
|
||||
|
||||
// bridgeStabilityMetricsSnapshot 返回当前计数快照,供 /api/ping 输出。
|
||||
func bridgeStabilityMetricsSnapshot() map[string]any {
|
||||
return map[string]any{
|
||||
"gatewaySocketClosed": bridgeStabilityMetrics.gatewaySocketClosed.Load(),
|
||||
"taskGetUnconfirmedFallback": bridgeStabilityMetrics.taskGetUnconfirmedFallback.Load(),
|
||||
"runDeadlineInterrupt": bridgeStabilityMetrics.runDeadlineInterrupt.Load(),
|
||||
}
|
||||
}
|
||||
@ -1,197 +0,0 @@
|
||||
package acp
|
||||
|
||||
import (
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"xworkmate-bridge/internal/shared"
|
||||
)
|
||||
|
||||
// 持久 run 仓 / run 关联与 WS 解耦(T7/T8/T9)。
|
||||
//
|
||||
// 背景:OpenClaw gateway turn 采用异步模型——chat.send 快速返回 runId,bridge 把
|
||||
// run 记录(sess.openClaw)、预算(sess.task.DeadlineAt)、运行句柄(sess.lastResult)
|
||||
// 落在「按 sessionID 维度」的 per-session store 里(s.sessions),其生命周期独立于
|
||||
// bridge↔gateway 的 WebSocket 连接。客户端随后轮询 tasks.get。
|
||||
//
|
||||
// 此前 tasks.get 每次都强依赖 gateway 应答:一旦 WS 抖动 / 重连后 run 内存态丢失,
|
||||
// tasks.get 回 not_found 或 socket_closed,已完成的结果就此丢失,客户端要么硬失败、
|
||||
// 要么(修复前)无限轮询。下列辅助把 tasks.get 改造为「优先用持久 run 仓兜底」:
|
||||
//
|
||||
// T8 已观察到的终态结果缓存进 sess.lastResult,gateway 之后查不到也不丢;
|
||||
// T7 gateway 暂时无法确认(unavailable / socket closed / not_found)但 run 仍在预算内时,
|
||||
// 合成一个 running 句柄让客户端继续轮询,跨越瞬时抖动(与 WS 生命周期解耦);
|
||||
// T9 run 超过 DeadlineAt 且 gateway 仍无法确认时,回确定性的 interrupted 终态。
|
||||
|
||||
// openClawTaskGetResultIsTerminal 判断一个 tasks.get 结果是否表示 run 已结束。
|
||||
// 注意:仍在 artifact 同步中的结果会被 normalizeOpenClawTaskGetResult 重写为 status=running,
|
||||
// 因此这里只认显式终态,不会把「同步中」误判为终态。
|
||||
func openClawTaskGetResultIsTerminal(payload map[string]any) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(shared.StringArg(payload, "status", ""))) {
|
||||
case string(TaskStateCompleted), string(TaskStateFailed), string(TaskStateCancelled),
|
||||
"interrupted", "partially_delivered":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// cacheOpenClawTaskGetResultIfTerminal 把一次 gateway 确认的终态结果落进 per-session 持久 run 仓(T8)。
|
||||
func (s *Server) cacheOpenClawTaskGetResultIfTerminal(params map[string]any, payload map[string]any) {
|
||||
if len(payload) == 0 || !openClawTaskGetResultIsTerminal(payload) {
|
||||
return
|
||||
}
|
||||
sess := s.findTaskSession(params)
|
||||
if sess == nil {
|
||||
return
|
||||
}
|
||||
sess.mu.Lock()
|
||||
defer sess.mu.Unlock()
|
||||
switch strings.ToLower(strings.TrimSpace(shared.StringArg(payload, "status", ""))) {
|
||||
case string(TaskStateFailed):
|
||||
sess.task.State = TaskStateFailed
|
||||
case string(TaskStateCancelled):
|
||||
sess.task.State = TaskStateCancelled
|
||||
default:
|
||||
sess.task.State = TaskStateCompleted
|
||||
}
|
||||
sess.task.ProgressTerminal = true
|
||||
sess.task.ProgressStage = strings.ToLower(strings.TrimSpace(shared.StringArg(payload, "status", "")))
|
||||
sess.task.UpdatedAt = time.Now()
|
||||
sess.lastResult = cloneMap(payload)
|
||||
}
|
||||
|
||||
// cachedTerminalOpenClawResult 返回某 run 此前已观察到的终态结果(若有)(T7/T8)。
|
||||
func (s *Server) cachedTerminalOpenClawResult(params map[string]any) (map[string]any, bool) {
|
||||
sess := s.findTaskSession(params)
|
||||
if sess == nil {
|
||||
return nil, false
|
||||
}
|
||||
sess.mu.Lock()
|
||||
defer sess.mu.Unlock()
|
||||
return cachedTerminalForRunLocked(sess, params)
|
||||
}
|
||||
|
||||
// cachedTerminalForRunLocked 仅当缓存终态确实属于「本次请求的 runId」时才命中,
|
||||
// 防止同一 session 复用后把旧 run 的终态错配给新 run。调用方须持有 sess.mu。
|
||||
func cachedTerminalForRunLocked(sess *session, params map[string]any) (map[string]any, bool) {
|
||||
if !sess.task.ProgressTerminal || len(sess.lastResult) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
if !openClawTaskGetResultIsTerminal(sess.lastResult) {
|
||||
return nil, false
|
||||
}
|
||||
requestedRun := strings.TrimSpace(shared.StringArg(params, "runId", ""))
|
||||
if requestedRun == "" {
|
||||
requestedRun = strings.TrimSpace(shared.StringArg(params, "taskId", ""))
|
||||
}
|
||||
if requestedRun != "" {
|
||||
cachedRun := firstNonEmptyString(sess.lastResult, "runId", "taskId")
|
||||
if cachedRun != "" && !strings.EqualFold(cachedRun, requestedRun) {
|
||||
return nil, false
|
||||
}
|
||||
}
|
||||
return cloneMap(sess.lastResult), true
|
||||
}
|
||||
|
||||
// openClawTaskGetGatewayUnconfirmedFallback 在 gateway 无法确认 run 时,用持久 run 仓兜底(T7/T9):
|
||||
// - 已有缓存终态 -> 直接返回;
|
||||
// - run 仍在预算内 -> 合成 running 句柄,客户端继续轮询,跨越瞬时抖动;
|
||||
// - run 超过 deadline -> 回确定性 interrupted 终态。
|
||||
//
|
||||
// 没有任何 per-session 记录时退回旧行为(not_found),不改变无状态查询的语义。
|
||||
func (s *Server) openClawTaskGetGatewayUnconfirmedFallback(params map[string]any, code string, message string) map[string]any {
|
||||
notFound := func() map[string]any {
|
||||
return map[string]any{
|
||||
"ok": false,
|
||||
"status": "not_found",
|
||||
"code": fallbackString(code, "TASK_LOOKUP_FAILED"),
|
||||
"message": fallbackString(message, "openclaw native task lookup failed"),
|
||||
}
|
||||
}
|
||||
sess := s.findTaskSession(params)
|
||||
if sess == nil {
|
||||
return notFound()
|
||||
}
|
||||
sess.mu.Lock()
|
||||
defer sess.mu.Unlock()
|
||||
if cached, ok := cachedTerminalForRunLocked(sess, params); ok {
|
||||
return cached
|
||||
}
|
||||
if sess.openClaw == nil {
|
||||
return notFound()
|
||||
}
|
||||
now := time.Now()
|
||||
if !sess.task.DeadlineAt.IsZero() && now.After(sess.task.DeadlineAt) {
|
||||
return s.markOpenClawRunDeadlineInterruptedLocked(sess, code, message)
|
||||
}
|
||||
// 仍在预算内:合成 running 句柄让客户端继续轮询,不因一次瞬时抖动硬失败。
|
||||
metricTaskGetUnconfirmedFallbackInc() // T12
|
||||
running := openClawRunningTaskResult(sess.openClaw)
|
||||
running["transportDegraded"] = true
|
||||
if strings.TrimSpace(code) != "" {
|
||||
running["transportDegradedCode"] = strings.TrimSpace(code)
|
||||
}
|
||||
// T11:带 runId 的日志,便于与 App / 插件 / 网关四层按 runId 串联。
|
||||
log.Printf("level=warn component=openclaw_run_registry event=tasks_get_unconfirmed_fallback runId=%q openclawSessionKey=%q code=%q",
|
||||
sess.openClaw.RunID, sess.openClaw.SessionKey, strings.TrimSpace(code))
|
||||
sess.lastResult = cloneMap(running)
|
||||
return running
|
||||
}
|
||||
|
||||
// markOpenClawRunDeadlineInterruptedLocked 为「超过预算且 gateway 无法确认」的 run 生成确定性
|
||||
// interrupted 终态(T9)。调用方须持有 sess.mu。
|
||||
func (s *Server) markOpenClawRunDeadlineInterruptedLocked(sess *session, code string, message string) map[string]any {
|
||||
now := time.Now()
|
||||
sess.task.State = TaskStateFailed
|
||||
sess.task.ProgressTerminal = true
|
||||
sess.task.ProgressStage = "interrupted"
|
||||
sess.task.ProgressMessage = "OpenClaw run exceeded its budget and could not be confirmed"
|
||||
sess.task.UpdatedAt = now
|
||||
metricRunDeadlineInterruptInc() // T12
|
||||
// T11:带 runId 的终态日志。
|
||||
if sess.openClaw != nil {
|
||||
log.Printf("level=warn component=openclaw_run_registry event=run_deadline_interrupt runId=%q openclawSessionKey=%q deadlineAt=%q code=%q",
|
||||
sess.openClaw.RunID, sess.openClaw.SessionKey,
|
||||
sess.openClaw.DeadlineAt.UTC().Format(time.RFC3339Nano), strings.TrimSpace(code))
|
||||
}
|
||||
|
||||
result := map[string]any{
|
||||
"ok": true,
|
||||
"success": false,
|
||||
"status": "interrupted",
|
||||
"event": "interrupted",
|
||||
"pending": false,
|
||||
"code": "OPENCLAW_RUN_DEADLINE_EXCEEDED",
|
||||
"artifactSyncStatus": "interrupted",
|
||||
"message": "OpenClaw 任务超过预算上限且网关无法确认结果,已结束本轮等待。任务可能已在后台完成,请重新发送请求以拿回结果。",
|
||||
"artifacts": []any{},
|
||||
}
|
||||
if strings.TrimSpace(code) != "" {
|
||||
result["gatewayUnconfirmedCode"] = strings.TrimSpace(code)
|
||||
}
|
||||
if strings.TrimSpace(message) != "" {
|
||||
result["gatewayUnconfirmedMessage"] = strings.TrimSpace(message)
|
||||
}
|
||||
if record := sess.openClaw; record != nil {
|
||||
result["runId"] = record.RunID
|
||||
result["taskId"] = record.RunID
|
||||
result["turnId"] = record.TurnID
|
||||
result["sessionId"] = record.SessionID
|
||||
result["threadId"] = record.ThreadID
|
||||
result["appThreadKey"] = record.ThreadID
|
||||
result["openclawSessionKey"] = record.SessionKey
|
||||
result["resolvedGatewayProviderId"] = record.GatewayProviderID
|
||||
result["startedAt"] = record.StartedAt.UTC().Format(time.RFC3339Nano)
|
||||
result["deadlineAt"] = record.DeadlineAt.UTC().Format(time.RFC3339Nano)
|
||||
}
|
||||
sess.lastResult = cloneMap(result)
|
||||
return result
|
||||
}
|
||||
|
||||
func fallbackString(value string, fallback string) string {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return fallback
|
||||
}
|
||||
return strings.TrimSpace(value)
|
||||
}
|
||||
@ -1,154 +0,0 @@
|
||||
package acp
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"xworkmate-bridge/internal/shared"
|
||||
)
|
||||
|
||||
func newRunRegistryTestServer(deadline time.Time) (*Server, map[string]any) {
|
||||
sess := &session{sessionID: "s1", threadID: "t1"}
|
||||
sess.task.RunID = "run-1"
|
||||
sess.task.SessionKey = "sk"
|
||||
sess.task.GatewayProviderID = "openclaw"
|
||||
sess.task.DeadlineAt = deadline
|
||||
sess.openClaw = &OpenClawTaskRecord{
|
||||
SessionID: "s1",
|
||||
ThreadID: "t1",
|
||||
TurnID: "turn-1",
|
||||
RunID: "run-1",
|
||||
SessionKey: "sk",
|
||||
GatewayProviderID: "openclaw",
|
||||
StartedAt: time.Now().Add(-time.Minute),
|
||||
DeadlineAt: deadline,
|
||||
}
|
||||
srv := &Server{sessions: map[string]*session{"s1": sess}}
|
||||
params := map[string]any{"sessionId": "s1", "runId": "run-1"}
|
||||
return srv, params
|
||||
}
|
||||
|
||||
func TestOpenClawTaskGetResultIsTerminal(t *testing.T) {
|
||||
cases := []struct {
|
||||
status string
|
||||
want bool
|
||||
}{
|
||||
{"completed", true},
|
||||
{"failed", true},
|
||||
{"cancelled", true},
|
||||
{"interrupted", true},
|
||||
{"partially_delivered", true},
|
||||
{"running", false},
|
||||
{"syncing-artifacts", false},
|
||||
{"queued", false},
|
||||
{"", false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
if got := openClawTaskGetResultIsTerminal(map[string]any{"status": tc.status}); got != tc.want {
|
||||
t.Errorf("status=%q: got %v, want %v", tc.status, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// T7: gateway 无法确认但 run 仍在预算内 -> 合成 running 句柄续轮询。
|
||||
func TestGatewayUnconfirmedFallbackWithinBudgetKeepsPolling(t *testing.T) {
|
||||
srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute))
|
||||
got := srv.openClawTaskGetGatewayUnconfirmedFallback(params, "SOCKET_CLOSED", "socket closed")
|
||||
if status := shared.StringArg(got, "status", ""); status != string(TaskStateRunning) {
|
||||
t.Fatalf("status = %q, want running", status)
|
||||
}
|
||||
if !parseBool(got["transportDegraded"]) {
|
||||
t.Fatalf("transportDegraded not set: %v", got)
|
||||
}
|
||||
if shared.StringArg(got, "runId", "") != "run-1" {
|
||||
t.Fatalf("runId mismatch: %v", got["runId"])
|
||||
}
|
||||
}
|
||||
|
||||
// T9: run 超过 deadline 且 gateway 无法确认 -> 确定性 interrupted 终态。
|
||||
func TestGatewayUnconfirmedFallbackPastDeadlineInterrupts(t *testing.T) {
|
||||
srv, params := newRunRegistryTestServer(time.Now().Add(-time.Minute))
|
||||
got := srv.openClawTaskGetGatewayUnconfirmedFallback(params, "SOCKET_CLOSED", "socket closed")
|
||||
if status := shared.StringArg(got, "status", ""); status != "interrupted" {
|
||||
t.Fatalf("status = %q, want interrupted", status)
|
||||
}
|
||||
if code := shared.StringArg(got, "code", ""); code != "OPENCLAW_RUN_DEADLINE_EXCEEDED" {
|
||||
t.Fatalf("code = %q, want OPENCLAW_RUN_DEADLINE_EXCEEDED", code)
|
||||
}
|
||||
if parseBool(got["success"]) {
|
||||
t.Fatalf("interrupted result must not be success")
|
||||
}
|
||||
sess := srv.findTaskSession(params)
|
||||
if sess == nil || !sess.task.ProgressTerminal || sess.task.State != TaskStateFailed {
|
||||
t.Fatalf("session terminal state not recorded: %+v", sess)
|
||||
}
|
||||
}
|
||||
|
||||
// T8: 已观察到的终态被缓存,且即使之后 gateway 不可达也优先返回缓存终态。
|
||||
func TestTerminalResultCachedAndServedAfterGatewayLoss(t *testing.T) {
|
||||
srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute))
|
||||
terminal := map[string]any{
|
||||
"ok": true,
|
||||
"success": true,
|
||||
"status": "completed",
|
||||
"runId": "run-1",
|
||||
"message": "done",
|
||||
}
|
||||
srv.cacheOpenClawTaskGetResultIfTerminal(params, terminal)
|
||||
|
||||
cached, ok := srv.cachedTerminalOpenClawResult(params)
|
||||
if !ok {
|
||||
t.Fatalf("expected cached terminal result")
|
||||
}
|
||||
if shared.StringArg(cached, "status", "") != "completed" {
|
||||
t.Fatalf("cached status = %q, want completed", cached["status"])
|
||||
}
|
||||
|
||||
// 即使 run 已过 deadline + gateway 丢失,也应优先返回缓存终态而非 interrupted。
|
||||
sess := srv.findTaskSession(params)
|
||||
sess.mu.Lock()
|
||||
sess.task.DeadlineAt = time.Now().Add(-time.Hour)
|
||||
sess.mu.Unlock()
|
||||
got := srv.openClawTaskGetGatewayUnconfirmedFallback(params, "SOCKET_CLOSED", "socket closed")
|
||||
if shared.StringArg(got, "status", "") != "completed" {
|
||||
t.Fatalf("expected cached completed to win over deadline interrupt, got %v", got["status"])
|
||||
}
|
||||
}
|
||||
|
||||
// 同一 session 复用后,旧 run 的终态不得错配给新 runId 的查询。
|
||||
func TestCachedTerminalNotServedForDifferentRunId(t *testing.T) {
|
||||
srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute))
|
||||
srv.cacheOpenClawTaskGetResultIfTerminal(params, map[string]any{
|
||||
"status": "completed", "success": true, "runId": "run-1",
|
||||
})
|
||||
// 新一轮查询带不同 runId -> 不应命中旧缓存。
|
||||
newParams := map[string]any{"sessionId": "s1", "runId": "run-2"}
|
||||
if _, ok := srv.cachedTerminalOpenClawResult(newParams); ok {
|
||||
t.Fatalf("stale terminal for run-1 must not be served for run-2")
|
||||
}
|
||||
// 原 runId 仍应命中。
|
||||
if _, ok := srv.cachedTerminalOpenClawResult(params); !ok {
|
||||
t.Fatalf("terminal for run-1 should still be served for run-1")
|
||||
}
|
||||
}
|
||||
|
||||
// running 结果不应被当作终态缓存。
|
||||
func TestRunningResultNotCachedAsTerminal(t *testing.T) {
|
||||
srv, params := newRunRegistryTestServer(time.Now().Add(30 * time.Minute))
|
||||
srv.cacheOpenClawTaskGetResultIfTerminal(params, map[string]any{"status": "running", "runId": "run-1"})
|
||||
if _, ok := srv.cachedTerminalOpenClawResult(params); ok {
|
||||
t.Fatalf("running result must not be cached as terminal")
|
||||
}
|
||||
}
|
||||
|
||||
// 无 per-session 记录时退回旧的 not_found 行为。
|
||||
func TestGatewayUnconfirmedFallbackWithoutSessionReturnsNotFound(t *testing.T) {
|
||||
srv := &Server{sessions: map[string]*session{}}
|
||||
got := srv.openClawTaskGetGatewayUnconfirmedFallback(map[string]any{"sessionId": "missing"}, "X", "y")
|
||||
if parseBool(got["ok"]) {
|
||||
t.Fatalf("expected ok=false not_found, got %v", got)
|
||||
}
|
||||
if shared.StringArg(got, "status", "") != "not_found" {
|
||||
t.Fatalf("status = %q, want not_found", got["status"])
|
||||
}
|
||||
}
|
||||
@ -342,13 +342,12 @@ func (o *SessionOrchestrator) startOpenClawGatewayTask(
|
||||
return nil, rpcErr
|
||||
}
|
||||
applyOpenClawPreparedArtifactToChatParams(chatParams, preparedArtifact, sessionKey, turnID, artifactContract)
|
||||
chatSendTimeout := openClawAgentWaitTimeout(params, chatParams)
|
||||
sendStarted := time.Now()
|
||||
sendResult := o.openClawGatewayRequestWithRetry(
|
||||
gatewayProvider,
|
||||
"chat.send",
|
||||
chatParams,
|
||||
chatSendTimeout,
|
||||
2*time.Minute,
|
||||
notifyWithCollection,
|
||||
)
|
||||
logOpenClawGatewayTiming(
|
||||
@ -417,10 +416,6 @@ func (o *SessionOrchestrator) startOpenClawGatewayTask(
|
||||
sess.task.DeadlineAt = record.DeadlineAt
|
||||
sess.task.ProgressStage = "running"
|
||||
sess.task.ProgressMessage = "OpenClaw task accepted"
|
||||
// 新一轮 turn 复用同一 session 时,必须重置上一轮可能留下的终态标记,
|
||||
// 否则持久 run 仓(T8)会把旧 runId 的终态错配给新 run。
|
||||
sess.task.State = TaskStateRunning
|
||||
sess.task.ProgressTerminal = false
|
||||
sess.openClaw = record
|
||||
running := openClawRunningTaskResult(record)
|
||||
sess.lastResult = cloneMap(running)
|
||||
@ -603,9 +598,6 @@ func (o *SessionOrchestrator) openClawArtifactPrepare(
|
||||
notify,
|
||||
)
|
||||
if !prepareResult.OK {
|
||||
if isOpenClawUnknownMethodError(prepareResult.Error, "xworkmate.session.prepare") {
|
||||
return openClawPreparedArtifactScopeFromPayload(openClawFallbackSessionPreparePayload(prepareParams)), nil
|
||||
}
|
||||
return nil, gatewayRPCError(prepareResult.Error, "openclaw artifact prepare failed")
|
||||
}
|
||||
prepared := openClawPreparedArtifactScopeFromPayload(shared.AsMap(prepareResult.Payload))
|
||||
@ -615,60 +607,6 @@ func (o *SessionOrchestrator) openClawArtifactPrepare(
|
||||
return prepared, nil
|
||||
}
|
||||
|
||||
func isOpenClawUnknownMethodError(errorPayload map[string]any, method string) bool {
|
||||
message := strings.ToLower(strings.TrimSpace(shared.StringArg(errorPayload, "message", "")))
|
||||
if message == "" {
|
||||
return false
|
||||
}
|
||||
// 消息形如「unknown method: <method>」已明确指向「网关不认识该方法」,足以判定,
|
||||
// 据此走 graceful fallback(如 openClawFallbackSessionPreparePayload)。
|
||||
//
|
||||
// 注意:不能再用严格的 code 白名单来 gate。真实网关常以数字 JSON-RPC code
|
||||
// (-32601 method not found / -32600 invalid request / -32002 等) 回传,
|
||||
// 经 shared.StringArg(fmt.Sprint) 会被字符串化为 "-32601"/"-32002",
|
||||
// 旧实现只接受 {"", INVALID_REQUEST, METHOD_NOT_FOUND},导致 fallback 失效、
|
||||
// session.prepare 直接以 -32002 硬失败整轮任务。
|
||||
return strings.Contains(message, "unknown method") &&
|
||||
strings.Contains(message, strings.ToLower(strings.TrimSpace(method)))
|
||||
}
|
||||
|
||||
func openClawFallbackSessionPreparePayload(params map[string]any) map[string]any {
|
||||
sessionKey := strings.TrimSpace(shared.StringArg(params, "openclawSessionKey", ""))
|
||||
if sessionKey == "" {
|
||||
sessionKey = strings.TrimSpace(shared.StringArg(params, "sessionKey", ""))
|
||||
}
|
||||
if sessionKey == "" {
|
||||
sessionKey = "main"
|
||||
}
|
||||
runID := strings.TrimSpace(shared.StringArg(params, "runId", ""))
|
||||
if runID == "" {
|
||||
runID = strings.TrimSpace(shared.StringArg(params, "taskId", ""))
|
||||
}
|
||||
if runID == "" {
|
||||
runID = strings.TrimSpace(shared.StringArg(params, "requestId", ""))
|
||||
}
|
||||
if runID == "" {
|
||||
runID = "default"
|
||||
}
|
||||
relativeArtifactDirectory := filepath.Join("tasks", sessionKey, runID)
|
||||
workspaceDir := openClawArtifactWorkspaceDir(params)
|
||||
artifactDirectory := filepath.Join(workspaceDir, relativeArtifactDirectory)
|
||||
return map[string]any{
|
||||
"ok": true,
|
||||
"fallback": true,
|
||||
"compatibilityMode": "local-session-prepare",
|
||||
"runId": runID,
|
||||
"sessionKey": sessionKey,
|
||||
"openclawSessionKey": sessionKey,
|
||||
"remoteWorkingDirectory": workspaceDir,
|
||||
"remoteWorkspaceRefKind": "path",
|
||||
"artifactScope": relativeArtifactDirectory,
|
||||
"artifactDirectory": artifactDirectory,
|
||||
"relativeArtifactDirectory": relativeArtifactDirectory,
|
||||
"scopeKind": "task",
|
||||
}
|
||||
}
|
||||
|
||||
func openClawSessionPrepareParams(params map[string]any, openClawSessionKey string, runID string, artifactContract openClawArtifactContract) map[string]any {
|
||||
appThreadKey := openClawAppThreadKey(params)
|
||||
result := map[string]any{
|
||||
@ -846,7 +784,6 @@ type openClawArtifactContract struct {
|
||||
RequiresArtifactExport bool
|
||||
ExpectedArtifactDirs []string
|
||||
RequiredArtifactExts []string
|
||||
ExpectedFileCounts map[string]int
|
||||
SourceMessage string
|
||||
}
|
||||
|
||||
@ -869,66 +806,16 @@ func openClawArtifactContractForParams(params map[string]any, chatParams map[str
|
||||
if len(requiredExts) == 0 {
|
||||
requiredExts = inferOpenClawRequiredArtifactExts(lowerMessage)
|
||||
}
|
||||
expectedFileCounts := normalizeOpenClawArtifactExtCountMap(shared.AsMap(contract["expectedFileCountByExtension"]))
|
||||
if len(expectedFileCounts) == 0 {
|
||||
expectedFileCounts = normalizeOpenClawArtifactExtCountMap(shared.AsMap(metadata["expectedFileCountByExtension"]))
|
||||
}
|
||||
if len(expectedFileCounts) == 0 {
|
||||
expectedFileCounts = normalizeOpenClawArtifactExtCountMap(shared.AsMap(shared.AsMap(metadata["xworkmateArtifactConstraints"])["expectedFileCountByExtension"]))
|
||||
}
|
||||
return openClawArtifactContract{
|
||||
TaskLoadClass: taskLoadClass,
|
||||
ComplexLongChain: complex,
|
||||
RequiresArtifactExport: requiresExport,
|
||||
ExpectedArtifactDirs: expectedDirs,
|
||||
RequiredArtifactExts: requiredExts,
|
||||
ExpectedFileCounts: expectedFileCounts,
|
||||
SourceMessage: message,
|
||||
}
|
||||
}
|
||||
|
||||
func normalizeOpenClawArtifactExtCountMap(values map[string]any) map[string]int {
|
||||
if len(values) == 0 {
|
||||
return nil
|
||||
}
|
||||
result := map[string]int{}
|
||||
for key, raw := range values {
|
||||
ext := strings.ToLower(strings.TrimSpace(key))
|
||||
ext = strings.TrimPrefix(ext, ".")
|
||||
if ext == "" || strings.Contains(ext, "/") || strings.Contains(ext, "\\") {
|
||||
continue
|
||||
}
|
||||
count := openClawPositiveInt(raw)
|
||||
if count <= 0 {
|
||||
continue
|
||||
}
|
||||
result[ext] = count
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return nil
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func openClawPositiveInt(value any) int {
|
||||
switch v := value.(type) {
|
||||
case int:
|
||||
return v
|
||||
case int64:
|
||||
return int(v)
|
||||
case float64:
|
||||
return int(v)
|
||||
case float32:
|
||||
return int(v)
|
||||
case string:
|
||||
var parsed int
|
||||
if _, err := fmt.Sscanf(strings.TrimSpace(v), "%d", &parsed); err == nil {
|
||||
return parsed
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func normalizeOpenClawDirList(values []any) []string {
|
||||
if len(values) == 0 {
|
||||
return nil
|
||||
@ -1542,13 +1429,6 @@ func (o *SessionOrchestrator) openClawArtifactExport(
|
||||
if len(artifactContract.RequiredArtifactExts) > 0 {
|
||||
exportParams["requiredArtifactExtensions"] = append([]string(nil), artifactContract.RequiredArtifactExts...)
|
||||
}
|
||||
if len(artifactContract.ExpectedFileCounts) > 0 {
|
||||
counts := map[string]int{}
|
||||
for ext, count := range artifactContract.ExpectedFileCounts {
|
||||
counts[ext] = count
|
||||
}
|
||||
exportParams["expectedFileCountByExtension"] = counts
|
||||
}
|
||||
payload := o.openClawArtifactExportRequest(gatewayProvider, exportParams, notify)
|
||||
return payload
|
||||
}
|
||||
@ -1617,9 +1497,6 @@ func mergeOpenClawArtifactPayload(result map[string]any, source map[string]any)
|
||||
if _, ok := source["missingRequiredExtensions"]; ok {
|
||||
result["missingRequiredExtensions"] = appendStringList(result["missingRequiredExtensions"], source["missingRequiredExtensions"])
|
||||
}
|
||||
if value, ok := source["missingRequiredFileCounts"]; ok {
|
||||
result["missingRequiredFileCounts"] = value
|
||||
}
|
||||
}
|
||||
|
||||
func appendStringList(existing any, incoming any) []any {
|
||||
@ -1687,16 +1564,11 @@ func applyOpenClawConstraintDeliveryStatus(result map[string]any) {
|
||||
|
||||
func gatewayRPCError(errorPayload map[string]any, fallback string) *shared.RPCError {
|
||||
if isOpenClawRetryableGatewayError(errorPayload) {
|
||||
metricGatewaySocketClosedInc() // T12
|
||||
// T10:连接断属「可重试 / run 可能仍在后台、可续轮询」语义,而非 run 确实失败。
|
||||
// 带 retryable/poll 提示,客户端据此降级为「后台续跑·重连中」(T5) 续轮询 tasks.get,而非硬失败。
|
||||
return &shared.RPCError{
|
||||
Code: -32002,
|
||||
Message: "OPENCLAW_GATEWAY_SOCKET_CLOSED: OpenClaw gateway connection closed during task execution",
|
||||
Data: map[string]any{
|
||||
"code": "OPENCLAW_GATEWAY_SOCKET_CLOSED",
|
||||
"retryable": true,
|
||||
"poll": true,
|
||||
"originalCode": strings.TrimSpace(shared.StringArg(errorPayload, "code", "")),
|
||||
"originalError": strings.TrimSpace(shared.StringArg(errorPayload, "message", "")),
|
||||
},
|
||||
|
||||
@ -132,49 +132,6 @@ func TestNormalizeOpenClawTaskGetUnknownArtifactEvidenceKeepsActiveRecordRunning
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpectedArtifactDirectoriesDoNotBlockTerminalTaskState(t *testing.T) {
|
||||
params := map[string]any{"expectedArtifactDirs": []any{"reports/", "artifacts/"}}
|
||||
payload := map[string]any{
|
||||
"success": true,
|
||||
"status": string(TaskStateCompleted),
|
||||
"artifactScope": "tasks/session/run",
|
||||
"artifactDirectory": "/remote/openclaw/workspace/tasks/session/run",
|
||||
"expectedArtifactDirs": []any{
|
||||
"reports/",
|
||||
"artifacts/",
|
||||
},
|
||||
}
|
||||
|
||||
if openClawTaskGetRequiresArtifactExport(params, payload) {
|
||||
t.Fatal("expectedArtifactDirs must remain non-blocking scan hints")
|
||||
}
|
||||
got := normalizeOpenClawTaskGetResult(params, payload, "openclaw", nil)
|
||||
if status := shared.StringArg(got, "status", ""); status != string(TaskStateCompleted) {
|
||||
t.Fatalf("expected terminal status to remain completed, got %#v", got)
|
||||
}
|
||||
if parseBool(got["pending"]) {
|
||||
t.Fatalf("expected terminal payload not to become pending, got %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequiredArtifactExtensionsStillBlockUntilVerified(t *testing.T) {
|
||||
params := map[string]any{"requiredArtifactExtensions": []any{"md"}}
|
||||
payload := map[string]any{
|
||||
"success": true,
|
||||
"status": string(TaskStateCompleted),
|
||||
"artifactScope": "tasks/session/run",
|
||||
"artifactDirectory": "/remote/openclaw/workspace/tasks/session/run",
|
||||
}
|
||||
|
||||
if !openClawTaskGetRequiresArtifactExport(params, payload) {
|
||||
t.Fatal("requiredArtifactExtensions must remain a blocking delivery contract")
|
||||
}
|
||||
got := normalizeOpenClawTaskGetResult(params, payload, "openclaw", nil)
|
||||
if status := shared.StringArg(got, "status", ""); status != string(TaskStateRunning) {
|
||||
t.Fatalf("expected missing required artifact to remain syncing, got %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeOpenClawTaskGetUnknownArtifactEvidenceFailsAfterDeadlineWithoutRequiredArtifacts(t *testing.T) {
|
||||
payload := map[string]any{
|
||||
"success": false,
|
||||
@ -306,9 +263,6 @@ func TestTaskGetArtifactExportReceivesRequiredArtifactExtensions(t *testing.T) {
|
||||
"gatewayProviderId": shared.StringArg(start, "resolvedGatewayProviderId", ""),
|
||||
"requiresArtifactExport": true,
|
||||
"requiredArtifactExtensions": []any{"pdf"},
|
||||
"expectedFileCountByExtension": map[string]any{
|
||||
"pdf": 1,
|
||||
},
|
||||
},
|
||||
}, nil)
|
||||
if rpcErr != nil {
|
||||
@ -321,54 +275,4 @@ func TestTaskGetArtifactExportReceivesRequiredArtifactExtensions(t *testing.T) {
|
||||
if got := shared.ListArg(exportParams, "requiredArtifactExtensions"); len(got) != 1 || got[0] != "pdf" {
|
||||
t.Fatalf("expected requiredArtifactExtensions to reach export, got %#v", exportParams)
|
||||
}
|
||||
if got := shared.AsMap(exportParams["expectedFileCountByExtension"]); openClawPositiveInt(got["pdf"]) != 1 {
|
||||
t.Fatalf("expected expectedFileCountByExtension to reach export, got %#v", exportParams)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsOpenClawUnknownMethodErrorAcceptsNumericGatewayCodes(t *testing.T) {
|
||||
const method = "xworkmate.session.prepare"
|
||||
cases := []struct {
|
||||
name string
|
||||
payload map[string]any
|
||||
want bool
|
||||
}{
|
||||
{
|
||||
name: "string invalid_request code",
|
||||
payload: map[string]any{"code": "INVALID_REQUEST", "message": "unknown method: xworkmate.session.prepare"},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "numeric -32002 (real gateway shape that previously hard-failed)",
|
||||
payload: map[string]any{"code": float64(-32002), "message": "unknown method: xworkmate.session.prepare"},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "numeric -32601 method not found",
|
||||
payload: map[string]any{"code": float64(-32601), "message": "Unknown method: xworkmate.session.prepare"},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "empty code",
|
||||
payload: map[string]any{"message": "unknown method: xworkmate.session.prepare"},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "unrelated error must not be swallowed",
|
||||
payload: map[string]any{"code": float64(-32002), "message": "gateway socket closed"},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "unknown method for a different method name",
|
||||
payload: map[string]any{"code": float64(-32601), "message": "unknown method: chat.send"},
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := isOpenClawUnknownMethodError(tc.payload, method); got != tc.want {
|
||||
t.Fatalf("isOpenClawUnknownMethodError(%v) = %v, want %v", tc.payload, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@ -95,8 +95,8 @@ func TestCapabilitiesExposeBuiltInProductionProviderCatalog(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestProductionProviderCatalogFallsBackToBridgeAuthToken(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-token")
|
||||
t.Setenv("INTERNAL_SERVICE_TOKEN", "")
|
||||
|
||||
_, catalog, _ := newProductionProviderCatalog()
|
||||
p, ok := catalog["codex"]
|
||||
@ -109,24 +109,9 @@ func TestProductionProviderCatalogFallsBackToBridgeAuthToken(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestProductionProviderCatalogPrefersAIWorkspaceAuthToken(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "ai-workspace-token")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-token")
|
||||
|
||||
_, catalog, _ := newProductionProviderCatalog()
|
||||
p, ok := catalog["codex"]
|
||||
if !ok {
|
||||
t.Fatal("missing codex")
|
||||
}
|
||||
|
||||
if got := p.AuthorizationHeader; got != "Bearer ai-workspace-token" {
|
||||
t.Fatalf("expected AI workspace bearer header, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProductionProviderCatalogPrefersDedicatedBridgeAuthToken(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "dedicated-token")
|
||||
t.Setenv("INTERNAL_SERVICE_TOKEN", "legacy-token")
|
||||
|
||||
_, catalog, _ := newProductionProviderCatalog()
|
||||
p, ok := catalog["codex"]
|
||||
@ -140,7 +125,6 @@ func TestProductionProviderCatalogPrefersDedicatedBridgeAuthToken(t *testing.T)
|
||||
}
|
||||
|
||||
func TestProductionProviderCatalogIgnoresInternalServiceToken(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "")
|
||||
t.Setenv("INTERNAL_SERVICE_TOKEN", "legacy-token")
|
||||
|
||||
|
||||
@ -833,7 +833,7 @@ func TestGatewayRequestSkillsStatusAutoConnectsOpenClaw(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteSessionTaskGatewayFallsBackWhenPrepareUnsupported(t *testing.T) {
|
||||
func TestExecuteSessionTaskGatewayFailsWhenPrepareUnsupported(t *testing.T) {
|
||||
gateway := newAcpFakeOpenClawGateway(t)
|
||||
gateway.unsupportedSessionPrepare.Store(true)
|
||||
defer gateway.Close()
|
||||
@ -858,48 +858,15 @@ func TestExecuteSessionTaskGatewayFallsBackWhenPrepareUnsupported(t *testing.T)
|
||||
},
|
||||
},
|
||||
})
|
||||
if rpcErr != nil {
|
||||
t.Fatalf("expected prepare compatibility fallback, got error: %#v", rpcErr)
|
||||
if rpcErr == nil {
|
||||
t.Fatalf("expected prepare error without legacy fallback, got response: %#v", response)
|
||||
return
|
||||
}
|
||||
if response["success"] != true {
|
||||
t.Fatalf("expected successful gateway task with prepare fallback, got %#v", response)
|
||||
}
|
||||
if got := gateway.Methods(); !sameMethods(got, []string{"connect", "xworkmate.session.prepare", "chat.send", "xworkmate.tasks.get"}) {
|
||||
t.Fatalf("expected bridge to continue to chat.send when prepare is unsupported, got %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleSessionPrepareFallsBackWhenGatewayMethodUnsupported(t *testing.T) {
|
||||
gateway := newAcpFakeOpenClawGateway(t)
|
||||
gateway.unsupportedSessionPrepare.Store(true)
|
||||
defer gateway.Close()
|
||||
|
||||
t.Setenv("GATEWAY_RPC_URL", gateway.URL())
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-token")
|
||||
|
||||
server := NewServer()
|
||||
response, rpcErr := server.handleRequest(
|
||||
shared.RPCRequest{
|
||||
Method: "xworkmate.session.prepare",
|
||||
Params: map[string]any{
|
||||
"openclawSessionKey": "thread-prepare",
|
||||
"runId": "run-prepare",
|
||||
"workspaceDir": "/remote/openclaw/workspace",
|
||||
},
|
||||
},
|
||||
func(map[string]any) {},
|
||||
)
|
||||
if rpcErr != nil {
|
||||
t.Fatalf("expected fallback prepare response, got error: %#v", rpcErr)
|
||||
}
|
||||
if response["fallback"] != true {
|
||||
t.Fatalf("expected fallback marker, got %#v", response)
|
||||
}
|
||||
if response["artifactScope"] != "tasks/thread-prepare/run-prepare" {
|
||||
t.Fatalf("expected fallback task artifact scope, got %#v", response)
|
||||
if rpcErr.Code != -32002 || !strings.Contains(rpcErr.Message, "unknown method: xworkmate.session.prepare") {
|
||||
t.Fatalf("expected surfaced prepare unsupported error, got %#v", rpcErr)
|
||||
}
|
||||
if got := gateway.Methods(); !sameMethods(got, []string{"connect", "xworkmate.session.prepare"}) {
|
||||
t.Fatalf("expected bridge to try gateway prepare before fallback, got %#v", got)
|
||||
t.Fatalf("expected bridge to stop before chat.send when prepare is unsupported, got %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
@ -3089,7 +3056,6 @@ type acpFakeOpenClawGateway struct {
|
||||
artifactCount atomic.Int32
|
||||
artifactReadCount atomic.Int32
|
||||
artifactReadFailures atomic.Int32
|
||||
rejectDeviceTokenOnce atomic.Bool
|
||||
closeNextChatSend atomic.Bool
|
||||
alwaysCloseChatSend atomic.Bool
|
||||
agentWaitDelayMs atomic.Int64
|
||||
@ -3175,23 +3141,7 @@ func newAcpFakeOpenClawGateway(t *testing.T) *acpFakeOpenClawGateway {
|
||||
})
|
||||
return
|
||||
}
|
||||
auth := shared.AsMap(params["auth"])
|
||||
if fake.rejectDeviceTokenOnce.Swap(false) && strings.TrimSpace(shared.StringArg(auth, "deviceToken", "")) != "" {
|
||||
_ = conn.WriteJSON(map[string]any{
|
||||
"type": "res",
|
||||
"id": id,
|
||||
"ok": false,
|
||||
"error": map[string]any{
|
||||
"code": "INVALID_REQUEST",
|
||||
"message": "unauthorized: device token mismatch (rotate/reissue device token)",
|
||||
"details": map[string]any{
|
||||
"code": "AUTH_DEVICE_TOKEN_MISMATCH",
|
||||
},
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
if got, want := shared.StringArg(auth, "token", ""), bridgeSharedAuthToken(); got != want {
|
||||
if got, want := shared.StringArg(shared.AsMap(params["auth"]), "token", ""), os.Getenv("BRIDGE_AUTH_TOKEN"); got != want {
|
||||
_ = conn.WriteJSON(map[string]any{
|
||||
"type": "res",
|
||||
"id": id,
|
||||
|
||||
@ -76,9 +76,6 @@ func (s *Server) handleRequest(request shared.RPCRequest, notify func(map[string
|
||||
case "xworkmate.tasks.get":
|
||||
return s.handleTaskGet(ctx, request.Params, notify), nil
|
||||
|
||||
case "xworkmate.session.prepare":
|
||||
return s.handleSessionPrepare(ctx, request.Params, notify)
|
||||
|
||||
case "xworkmate.tasks.cancel":
|
||||
return s.handleTaskCancel(ctx, request.Params, notify), nil
|
||||
|
||||
@ -93,36 +90,6 @@ func (s *Server) handleRequest(request shared.RPCRequest, notify func(map[string
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleSessionPrepare(ctx context.Context, params map[string]any, notify func(map[string]any)) (map[string]any, *shared.RPCError) {
|
||||
gatewayProvider := strings.TrimSpace(shared.StringArg(params, "gatewayProviderId", ""))
|
||||
if gatewayProvider == "" {
|
||||
gatewayProvider = strings.TrimSpace(shared.StringArg(params, "resolvedGatewayProviderId", ""))
|
||||
}
|
||||
if gatewayProvider == "" {
|
||||
gatewayProvider = "openclaw"
|
||||
}
|
||||
if rpcErr := ensureProductionGatewayConnected(s, gatewayProvider, notify); rpcErr != nil {
|
||||
return openClawFallbackSessionPreparePayload(params), nil
|
||||
}
|
||||
result := s.gateway.RequestByMode(
|
||||
gatewayProvider,
|
||||
"xworkmate.session.prepare",
|
||||
params,
|
||||
30*time.Second,
|
||||
notify,
|
||||
)
|
||||
if result.OK {
|
||||
payload := shared.AsMap(result.Payload)
|
||||
if openClawPreparedArtifactScopeFromPayload(payload) != nil {
|
||||
return payload, nil
|
||||
}
|
||||
}
|
||||
if !result.OK && !isOpenClawUnknownMethodError(result.Error, "xworkmate.session.prepare") {
|
||||
return nil, gatewayRPCError(result.Error, "openclaw artifact prepare failed")
|
||||
}
|
||||
return openClawFallbackSessionPreparePayload(params), nil
|
||||
}
|
||||
|
||||
func (s *Server) handleTaskGet(ctx context.Context, params map[string]any, notify func(map[string]any)) map[string]any {
|
||||
params = s.taskGetParamsWithSessionScope(params)
|
||||
gatewayProvider := strings.TrimSpace(shared.StringArg(params, "gatewayProviderId", ""))
|
||||
@ -132,13 +99,13 @@ func (s *Server) handleTaskGet(ctx context.Context, params map[string]any, notif
|
||||
if gatewayProvider == "" {
|
||||
gatewayProvider = "openclaw"
|
||||
}
|
||||
// T7/T8: 一旦观察到终态就从持久 run 仓返回,避免之后 gateway 查不到导致结果丢失。
|
||||
if cached, ok := s.cachedTerminalOpenClawResult(params); ok {
|
||||
return cached
|
||||
}
|
||||
if rpcErr := ensureProductionGatewayConnected(s, gatewayProvider, notify); rpcErr != nil {
|
||||
// T7/T9: gateway 不可达时按持久 run 仓兜底(续轮询 / deadline 终态),而非裸 not_found。
|
||||
return s.openClawTaskGetGatewayUnconfirmedFallback(params, "GATEWAY_UNAVAILABLE", rpcErr.Message)
|
||||
return map[string]any{
|
||||
"ok": false,
|
||||
"status": "not_found",
|
||||
"code": "GATEWAY_UNAVAILABLE",
|
||||
"message": rpcErr.Message,
|
||||
}
|
||||
}
|
||||
result := s.gateway.RequestByMode(
|
||||
gatewayProvider,
|
||||
@ -162,15 +129,16 @@ func (s *Server) handleTaskGet(ctx context.Context, params map[string]any, notif
|
||||
}
|
||||
s.decorateOpenClawArtifactDownloadURLs(payload, sessionKey, runID)
|
||||
stripOpenClawArtifactInlineContent(payload)
|
||||
// T8: 缓存「最终客户端可见形态」(已 decorate 下载 URL + strip 内联内容),
|
||||
// 这样从缓存回放时与正常路径完全一致。
|
||||
s.cacheOpenClawTaskGetResultIfTerminal(params, payload)
|
||||
return payload
|
||||
}
|
||||
// T7/T9: gateway 返回错误(socket closed / not_found / lookup failed)时同样走持久 run 仓兜底。
|
||||
message := strings.TrimSpace(shared.StringArg(result.Error, "message", "openclaw native task lookup failed"))
|
||||
code := strings.TrimSpace(shared.StringArg(result.Error, "code", "TASK_LOOKUP_FAILED"))
|
||||
return s.openClawTaskGetGatewayUnconfirmedFallback(params, code, message)
|
||||
return map[string]any{
|
||||
"ok": false,
|
||||
"status": "not_found",
|
||||
"code": code,
|
||||
"message": message,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) taskGetParamsWithSessionScope(params map[string]any) map[string]any {
|
||||
@ -277,9 +245,6 @@ func (s *Server) mergeOpenClawTaskGetArtifactExport(payload map[string]any, para
|
||||
if requiredExts := openClawTaskGetRequiredArtifactExtensions(params, payload); len(requiredExts) > 0 {
|
||||
exportParams["requiredArtifactExtensions"] = append([]string(nil), requiredExts...)
|
||||
}
|
||||
if expectedCounts := openClawTaskGetExpectedFileCounts(params, payload); len(expectedCounts) > 0 {
|
||||
exportParams["expectedFileCountByExtension"] = expectedCounts
|
||||
}
|
||||
exportPayload := s.orchestrator.openClawArtifactExportRequest(gatewayProvider, exportParams, notify)
|
||||
if openClawArtifactExportPayloadAuthoritative(exportPayload) {
|
||||
replaceOpenClawArtifactPayload(payload, exportPayload)
|
||||
@ -471,11 +436,9 @@ func openClawTaskGetRequiresArtifactExport(params map[string]any, payload map[st
|
||||
if parseBool(params["requiresExportBeforeFinalResponse"]) || parseBool(payload["requiresExportBeforeFinalResponse"]) {
|
||||
return true
|
||||
}
|
||||
// expectedArtifactDirs are discovery hints for the plugin's workspace-root
|
||||
// scan. They do not prove that the caller requires a file before the run can
|
||||
// reach a terminal state. Treating them as a blocking contract turns a
|
||||
// failed/no-output agent run into an endless "syncing-artifacts" loop.
|
||||
return len(shared.ListArg(params, "requiredArtifactExtensions")) > 0 ||
|
||||
return len(shared.ListArg(params, "expectedArtifactDirs")) > 0 ||
|
||||
len(shared.ListArg(payload, "expectedArtifactDirs")) > 0 ||
|
||||
len(shared.ListArg(params, "requiredArtifactExtensions")) > 0 ||
|
||||
len(shared.ListArg(payload, "requiredArtifactExtensions")) > 0
|
||||
}
|
||||
|
||||
@ -528,20 +491,6 @@ func openClawTaskGetRequiredArtifactExtensions(params map[string]any, payload ma
|
||||
return normalizeOpenClawArtifactExtList(openClawTaskGetMergedList(params, payload, "requiredArtifactExtensions"))
|
||||
}
|
||||
|
||||
func openClawTaskGetExpectedFileCounts(params map[string]any, payload map[string]any) map[string]int {
|
||||
result := normalizeOpenClawArtifactExtCountMap(shared.AsMap(payload["expectedFileCountByExtension"]))
|
||||
for ext, count := range normalizeOpenClawArtifactExtCountMap(shared.AsMap(params["expectedFileCountByExtension"])) {
|
||||
if result == nil {
|
||||
result = map[string]int{}
|
||||
}
|
||||
result[ext] = count
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return nil
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func openClawTaskGetMergedList(params map[string]any, payload map[string]any, key string) []any {
|
||||
seen := map[string]bool{}
|
||||
result := []any{}
|
||||
|
||||
@ -43,20 +43,13 @@ func newHTTPServer(addr string, handler http.Handler) *http.Server {
|
||||
|
||||
func NewServer() *Server {
|
||||
config := loadBridgeConfig()
|
||||
authTokens := bridgeInboundAuthTokens()
|
||||
authToken := ""
|
||||
authExtraTokens := []string(nil)
|
||||
if len(authTokens) > 0 {
|
||||
authToken = authTokens[0]
|
||||
authExtraTokens = authTokens[1:]
|
||||
}
|
||||
s := &Server{
|
||||
sessions: make(map[string]*session),
|
||||
config: config,
|
||||
allowedOrigins: shared.ParseAllowedOrigins(shared.EnvOrDefault("ACP_ALLOWED_ORIGINS", "https://xworkmate.svc.plus,http://localhost:*,http://127.0.0.1:*")),
|
||||
authService: service.NewStaticTokenAuthService(
|
||||
authToken,
|
||||
authExtraTokens...,
|
||||
shared.EnvOrDefault("BRIDGE_AUTH_TOKEN", ""),
|
||||
shared.EnvOrDefault("BRIDGE_REVIEW_AUTH_TOKEN", ""),
|
||||
),
|
||||
openClawGate: newOpenClawGatewayAdmissionGate(config),
|
||||
taskRouter: newDistributedTaskRouter(distributedTaskRouterConfig{
|
||||
|
||||
@ -394,11 +394,11 @@ func TestHTTPHandlerGatewayOpenClawHandlesFiveConcurrentE2ECases(t *testing.T) {
|
||||
defer httpServer.Close()
|
||||
|
||||
prompts := []string{
|
||||
"采集最新AI资讯,保存在md文件",
|
||||
"围绕\n从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进 右侧是当下 \n测试制作视频,附件带有图片",
|
||||
"从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进 \n制作 使用codex 制作连续制作 7张的一些列图片",
|
||||
"围绕\n从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进\n输出Markdown格式文件, 微信公众号短图文 400-600字 插入关键词的软文\n输出Markdown格式文件, 小红书风格 600-800字 插入钩子话题的软文\n输出Markdown格式文件, X文案串 小于144字的英语 鲜明的观点\n输出Markdown格式文件, 微信公众号文章 800-1200字左右\n输出Markdown格式文件, 头条号长文 800-1200字左右",
|
||||
"围绕\n\n从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进 \n\n拆章节 -> 每章调用 Codex -> 每章 GPT images2 生成图 -> 汇总排版 -> 输出 PDF",
|
||||
"参考附件模版制作 ,围绕\n从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进 \n连续制作 7张的一些列图片",
|
||||
"拆章节 -> 每章调用 Codex -> 每章 GPT images2 生成图 -> 汇总排版 -> 输出 PDF\n\n右侧 artifact栏 显示的陈旧文件 make artifact",
|
||||
"围绕\n从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进 右侧是当下 \n测试制作视频",
|
||||
"围绕\n\n从单机权限 → 网络边界 → Web安全 → 云身份 → Zero Trust → AI Agent 身份 → AI模型与知识保护 演进 \n\n拆章节 -> 每章调用 Codex -> 每章 GPT images2 生成图 -> 汇总排版 -> 制作视频",
|
||||
}
|
||||
type result struct {
|
||||
body string
|
||||
@ -875,7 +875,6 @@ func (w *panicSSEWriter) Write(payload []byte) (int, error) {
|
||||
func (w *panicSSEWriter) WriteHeader(int) {}
|
||||
|
||||
func TestHTTPHandlerPingRequiresBearerAuthorizationWhenBridgeAuthTokenConfigured(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
server := NewServer()
|
||||
@ -890,27 +889,7 @@ func TestHTTPHandlerPingRequiresBearerAuthorizationWhenBridgeAuthTokenConfigured
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPHandlerPingAcceptsAIWorkspaceBearerAuthorization(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "ai-workspace-test-token")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_REVIEW_AUTH_TOKEN", "")
|
||||
t.Setenv("INTERNAL_SERVICE_TOKEN", "")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
server := NewServer()
|
||||
handler := server.Handler()
|
||||
|
||||
recorder := httptest.NewRecorder()
|
||||
request := httptest.NewRequest(http.MethodGet, "http://127.0.0.1/api/ping", nil)
|
||||
request.Header.Set("Authorization", "Bearer ai-workspace-test-token")
|
||||
handler.ServeHTTP(recorder, request)
|
||||
|
||||
if recorder.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200 for AI workspace token, got %d", recorder.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPHandlerPingAllowsReviewBearerAuthorizationWhenConfigured(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "ai-workspace-test-token")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_REVIEW_AUTH_TOKEN", "review-bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
@ -971,10 +950,8 @@ func TestHandleRPCAllowsPreflightForConfiguredOrigin(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestHandleRPCAllowsUnauthenticatedRequestsWhenBridgeAuthTokenUnset(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_REVIEW_AUTH_TOKEN", "")
|
||||
t.Setenv("INTERNAL_SERVICE_TOKEN", "")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
server := NewServer()
|
||||
recorder := httptest.NewRecorder()
|
||||
@ -993,7 +970,6 @@ func TestHandleRPCAllowsUnauthenticatedRequestsWhenBridgeAuthTokenUnset(t *testi
|
||||
}
|
||||
|
||||
func TestHandleRPCRequiresBearerAuthorizationWhenBridgeAuthTokenConfigured(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
server := NewServer()
|
||||
@ -1014,7 +990,6 @@ func TestHandleRPCRequiresBearerAuthorizationWhenBridgeAuthTokenConfigured(t *te
|
||||
}
|
||||
|
||||
func TestHandleRPCCapabilitiesRequiresBearerAuthorizationWhenBridgeAuthTokenConfigured(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
server := NewServer()
|
||||
@ -1034,7 +1009,6 @@ func TestHandleRPCCapabilitiesRequiresBearerAuthorizationWhenBridgeAuthTokenConf
|
||||
}
|
||||
|
||||
func TestHandleRPCAllowsReviewBearerAuthorizationWhenConfigured(t *testing.T) {
|
||||
t.Setenv("AI_WORKSPACE_AUTH_TOKEN", "ai-workspace-test-token")
|
||||
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
|
||||
t.Setenv("BRIDGE_REVIEW_AUTH_TOKEN", "review-bridge-test-token")
|
||||
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")
|
||||
|
||||
@ -360,11 +360,7 @@ func sameConnectTarget(current ConnectRequest, next ConnectRequest) bool {
|
||||
strings.TrimSpace(current.Endpoint.Host) == strings.TrimSpace(next.Endpoint.Host) &&
|
||||
current.Endpoint.Port == next.Endpoint.Port &&
|
||||
current.Endpoint.TLS == next.Endpoint.TLS &&
|
||||
normalizeEndpointPath(current.Endpoint.Path) == normalizeEndpointPath(next.Endpoint.Path) &&
|
||||
strings.TrimSpace(current.Identity.DeviceID) == strings.TrimSpace(next.Identity.DeviceID) &&
|
||||
strings.TrimSpace(current.Auth.Token) == strings.TrimSpace(next.Auth.Token) &&
|
||||
strings.TrimSpace(current.Auth.DeviceToken) == strings.TrimSpace(next.Auth.DeviceToken) &&
|
||||
strings.TrimSpace(current.Auth.Password) == strings.TrimSpace(next.Auth.Password)
|
||||
normalizeEndpointPath(current.Endpoint.Path) == normalizeEndpointPath(next.Endpoint.Path)
|
||||
}
|
||||
|
||||
func (s *session) connectAttempt() (ConnectResult, *GatewayError) {
|
||||
@ -416,6 +412,11 @@ func (s *session) connectAttempt() (ConnectResult, *GatewayError) {
|
||||
snapshotPayload := asMap(payload["snapshot"])
|
||||
sessionDefaults := asMap(snapshotPayload["sessionDefaults"])
|
||||
returnedDeviceToken := strings.TrimSpace(stringValue(auth["deviceToken"]))
|
||||
if returnedDeviceToken != "" {
|
||||
s.mu.Lock()
|
||||
s.config.Auth.DeviceToken = returnedDeviceToken
|
||||
s.mu.Unlock()
|
||||
}
|
||||
negotiatedScopes := stringSlice(auth["scopes"])
|
||||
negotiatedRole := strings.TrimSpace(stringValue(auth["role"]))
|
||||
if negotiatedRole == "" {
|
||||
|
||||
@ -2,10 +2,10 @@
|
||||
set -euo pipefail
|
||||
|
||||
BRIDGE_SERVER_URL="${BRIDGE_SERVER_URL:-https://xworkmate-bridge.svc.plus}"
|
||||
BRIDGE_AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
BRIDGE_AUTH_TOKEN="${BRIDGE_AUTH_TOKEN:-}"
|
||||
|
||||
if [[ -z "${BRIDGE_AUTH_TOKEN}" ]]; then
|
||||
echo "Error: AI_WORKSPACE_AUTH_TOKEN or BRIDGE_AUTH_TOKEN is required" >&2
|
||||
echo "Error: BRIDGE_AUTH_TOKEN is required" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
@ -3,10 +3,10 @@
|
||||
set -euo pipefail
|
||||
|
||||
BRIDGE_SERVER_URL="${BRIDGE_SERVER_URL:-https://xworkmate-bridge.svc.plus}"
|
||||
BRIDGE_AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
BRIDGE_AUTH_TOKEN="${BRIDGE_AUTH_TOKEN:-}"
|
||||
|
||||
if [[ -z "${BRIDGE_AUTH_TOKEN}" ]]; then
|
||||
echo "Error: AI_WORKSPACE_AUTH_TOKEN or BRIDGE_AUTH_TOKEN is required" >&2
|
||||
echo "Error: BRIDGE_AUTH_TOKEN is required" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
@ -2,11 +2,11 @@
|
||||
set -euo pipefail
|
||||
|
||||
BRIDGE_SERVER_URL="${BRIDGE_SERVER_URL:-https://xworkmate-bridge.svc.plus}"
|
||||
BRIDGE_AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
BRIDGE_AUTH_TOKEN="${BRIDGE_AUTH_TOKEN:-}"
|
||||
HERMES_RPC_URL="${HERMES_RPC_URL:-${BRIDGE_SERVER_URL%/}/acp/rpc}"
|
||||
|
||||
if [[ -z "${BRIDGE_AUTH_TOKEN}" ]]; then
|
||||
echo "Error: AI_WORKSPACE_AUTH_TOKEN or BRIDGE_AUTH_TOKEN is required" >&2
|
||||
echo "Error: BRIDGE_AUTH_TOKEN is required" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
@ -57,22 +57,15 @@ resolve_token_from_unit() {
|
||||
|
||||
REMOTE_SYSTEM_SERVICE_UNIT_CONTENT="$(ssh -o BatchMode=yes "${SYSTEM_MIGRATION_USER}@${TARGET_HOST}" "cat '${SYSTEM_SERVICE_UNIT_PATH}' 2>/dev/null || true" 2>/dev/null || true)"
|
||||
|
||||
if [[ -z "${AI_WORKSPACE_AUTH_TOKEN:-}" && -n "${REMOTE_SYSTEM_SERVICE_UNIT_CONTENT}" ]]; then
|
||||
AI_WORKSPACE_AUTH_TOKEN="$(printf '%s\n' "${REMOTE_SYSTEM_SERVICE_UNIT_CONTENT}" | resolve_token_from_unit /dev/stdin "AI_WORKSPACE_AUTH_TOKEN")"
|
||||
if [[ -n "${AI_WORKSPACE_AUTH_TOKEN}" ]]; then
|
||||
echo "recovered AI_WORKSPACE_AUTH_TOKEN from ${SYSTEM_SERVICE_UNIT_PATH} on ${TARGET_HOST}" >&2
|
||||
fi
|
||||
fi
|
||||
|
||||
if [[ -z "${AI_WORKSPACE_AUTH_TOKEN:-}" && -z "${BRIDGE_AUTH_TOKEN:-}" && -n "${REMOTE_SYSTEM_SERVICE_UNIT_CONTENT}" ]]; then
|
||||
if [[ -z "${BRIDGE_AUTH_TOKEN:-}" && -n "${REMOTE_SYSTEM_SERVICE_UNIT_CONTENT}" ]]; then
|
||||
BRIDGE_AUTH_TOKEN="$(printf '%s\n' "${REMOTE_SYSTEM_SERVICE_UNIT_CONTENT}" | resolve_token_from_unit /dev/stdin "BRIDGE_AUTH_TOKEN")"
|
||||
if [[ -n "${BRIDGE_AUTH_TOKEN}" ]]; then
|
||||
echo "recovered BRIDGE_AUTH_TOKEN from ${SYSTEM_SERVICE_UNIT_PATH} on ${TARGET_HOST}" >&2
|
||||
fi
|
||||
fi
|
||||
|
||||
if [[ -z "${AI_WORKSPACE_AUTH_TOKEN:-}" && -z "${BRIDGE_AUTH_TOKEN:-}" ]]; then
|
||||
echo "::error::AI_WORKSPACE_AUTH_TOKEN is required: pass it via env, -e ai_workspace_auth_token=, or keep AI_WORKSPACE_AUTH_TOKEN/BRIDGE_AUTH_TOKEN in the existing system service unit at ${SYSTEM_SERVICE_UNIT_PATH}" >&2
|
||||
if [[ -z "${BRIDGE_AUTH_TOKEN:-}" ]]; then
|
||||
echo "::error::BRIDGE_AUTH_TOKEN is required: pass it via env, -e xworkmate_bridge_auth_token=, or keep the existing system service unit at ${SYSTEM_SERVICE_UNIT_PATH}" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
@ -80,12 +73,7 @@ if [[ -z "${BRIDGE_REVIEW_AUTH_TOKEN:-}" && -n "${REMOTE_SYSTEM_SERVICE_UNIT_CON
|
||||
BRIDGE_REVIEW_AUTH_TOKEN="$(printf '%s\n' "${REMOTE_SYSTEM_SERVICE_UNIT_CONTENT}" | resolve_token_from_unit /dev/stdin "BRIDGE_REVIEW_AUTH_TOKEN")"
|
||||
fi
|
||||
|
||||
AUTH_TOKEN_LINE=""
|
||||
if [[ -n "${AI_WORKSPACE_AUTH_TOKEN:-}" ]]; then
|
||||
AUTH_TOKEN_LINE="Environment=\"AI_WORKSPACE_AUTH_TOKEN=$(escape_systemd_env "${AI_WORKSPACE_AUTH_TOKEN}")\""
|
||||
else
|
||||
AUTH_TOKEN_LINE="Environment=\"BRIDGE_AUTH_TOKEN=$(escape_systemd_env "${BRIDGE_AUTH_TOKEN}")\""
|
||||
fi
|
||||
AUTH_TOKEN_LINE="Environment=\"BRIDGE_AUTH_TOKEN=$(escape_systemd_env "${BRIDGE_AUTH_TOKEN}")\""
|
||||
|
||||
REVIEW_TOKEN_LINE=""
|
||||
if [[ -n "${BRIDGE_REVIEW_AUTH_TOKEN:-}" ]]; then
|
||||
@ -139,7 +127,7 @@ existing_env="$(
|
||||
systemctl --user show -p Environment --value "${SERVICE_NAME}" 2>/dev/null || true
|
||||
systemctl show -p Environment --value "${SYSTEM_SERVICE_NAME}" 2>/dev/null || true
|
||||
if [[ -f "${SYSTEM_SERVICE_UNIT_PATH}" ]]; then
|
||||
sed -n 's/^Environment="\(AI_WORKSPACE_AUTH_TOKEN=[^"]*\|BRIDGE_AUTH_TOKEN=[^"]*\|BRIDGE_REVIEW_AUTH_TOKEN=[^"]*\)"$/\1/p' "${SYSTEM_SERVICE_UNIT_PATH}"
|
||||
sed -n 's/^Environment="\(BRIDGE_AUTH_TOKEN=[^"]*\|BRIDGE_REVIEW_AUTH_TOKEN=[^"]*\)"$/\1/p' "${SYSTEM_SERVICE_UNIT_PATH}"
|
||||
fi
|
||||
} | sed '/^$/d' | head -n 1
|
||||
)"
|
||||
@ -158,7 +146,7 @@ for line in lines:
|
||||
|
||||
for item in shlex.split(os.environ.get("EXISTING_ENV", "")):
|
||||
key, sep, value = item.partition("=")
|
||||
if sep and key in {"AI_WORKSPACE_AUTH_TOKEN", "BRIDGE_AUTH_TOKEN", "BRIDGE_REVIEW_AUTH_TOKEN"} and key not in present:
|
||||
if sep and key in {"BRIDGE_AUTH_TOKEN", "BRIDGE_REVIEW_AUTH_TOKEN"} and key not in present:
|
||||
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
|
||||
lines.append(f'Environment="{key}={escaped}"')
|
||||
present.add(key)
|
||||
|
||||
@ -20,6 +20,6 @@ if [[ "${RUN_APPLY}" != "true" ]]; then
|
||||
fi
|
||||
|
||||
ANSIBLE_CONFIG="${PWD}/ansible.cfg" \
|
||||
BRIDGE_AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}" \
|
||||
BRIDGE_AUTH_TOKEN="${INTERNAL_SERVICE_TOKEN:-}" \
|
||||
BRIDGE_REVIEW_AUTH_TOKEN="${BRIDGE_REVIEW_AUTH_TOKEN:-}" \
|
||||
"${args[@]}"
|
||||
|
||||
@ -31,9 +31,8 @@ curl_args=(
|
||||
--max-time 20
|
||||
)
|
||||
|
||||
AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
if [[ -n "${AUTH_TOKEN}" ]]; then
|
||||
curl_args+=(-H "Authorization: Bearer ${AUTH_TOKEN}")
|
||||
if [[ -n "${BRIDGE_AUTH_TOKEN:-}" ]]; then
|
||||
curl_args+=(-H "Authorization: Bearer ${BRIDGE_AUTH_TOKEN}")
|
||||
fi
|
||||
|
||||
for ((attempt = 1; attempt <= attempts; attempt += 1)); do
|
||||
@ -84,3 +83,4 @@ print(f"production_tag={deployed_tag}")
|
||||
print(f"production_commit={deployed_commit}")
|
||||
print(f"production_version={deployed_version}")
|
||||
PY
|
||||
|
||||
|
||||
@ -149,7 +149,7 @@ run_deploy() {
|
||||
|
||||
run_env_token_case() {
|
||||
local tmp_dir
|
||||
tmp_dir="$(setup_test_env "case: AI_WORKSPACE_AUTH_TOKEN env var drives the unit file")"
|
||||
tmp_dir="$(setup_test_env "case: BRIDGE_AUTH_TOKEN env var drives the unit file")"
|
||||
|
||||
local log_file="${tmp_dir}/deploy.log"
|
||||
local unit_file="${tmp_dir}/remote/home/ubuntu/.config/systemd/user/xworkmate-bridge.service"
|
||||
@ -160,7 +160,7 @@ run_env_token_case() {
|
||||
BRIDGE_CONFIG_PATH="${tmp_dir}/remote/opt/cloud-neutral/xworkmate-bridge/config.yaml" \
|
||||
USER_SYSTEMD_DIR="${tmp_dir}/remote/home/ubuntu/.config/systemd/user" \
|
||||
DEPLOY_NATIVE_SKIP_PROC_CHECK=true \
|
||||
AI_WORKSPACE_AUTH_TOKEN="test-token"
|
||||
BRIDGE_AUTH_TOKEN="test-token"
|
||||
|
||||
local log_output
|
||||
log_output="$(cat "${log_file}")"
|
||||
@ -168,7 +168,7 @@ run_env_token_case() {
|
||||
assert_contains "${log_output}" "scp ubuntu@example.test:"
|
||||
assert_contains "${log_output}" "ssh ubuntu@example.test"
|
||||
assert_contains "${log_output}" "systemctl --user restart xworkmate-bridge.service"
|
||||
assert_file_contains "${unit_file}" 'Environment="AI_WORKSPACE_AUTH_TOKEN=test-token"'
|
||||
assert_file_contains "${unit_file}" 'Environment="BRIDGE_AUTH_TOKEN=test-token"'
|
||||
assert_file_contains "${unit_file}" "WantedBy=default.target"
|
||||
|
||||
rm -rf "${tmp_dir}"
|
||||
@ -176,7 +176,7 @@ run_env_token_case() {
|
||||
|
||||
run_unit_fallback_case() {
|
||||
local tmp_dir
|
||||
tmp_dir="$(setup_test_env "case: AI_WORKSPACE_AUTH_TOKEN recovered from system service unit file")"
|
||||
tmp_dir="$(setup_test_env "case: BRIDGE_AUTH_TOKEN recovered from system service unit file")"
|
||||
|
||||
local system_unit_dir="${tmp_dir}/remote/etc/systemd/system"
|
||||
mkdir -p "${system_unit_dir}"
|
||||
@ -185,7 +185,7 @@ run_unit_fallback_case() {
|
||||
[Unit]
|
||||
Description=Stale system service
|
||||
[Service]
|
||||
Environment="AI_WORKSPACE_AUTH_TOKEN=recovered-from-systemd"
|
||||
Environment="BRIDGE_AUTH_TOKEN=recovered-from-systemd"
|
||||
Environment="BRIDGE_REVIEW_AUTH_TOKEN=recovered-review-token"
|
||||
ExecStart=/bin/true
|
||||
EOF
|
||||
@ -201,7 +201,7 @@ EOF
|
||||
SYSTEM_SERVICE_UNIT_PATH="${system_unit_file}" \
|
||||
DEPLOY_NATIVE_SKIP_PROC_CHECK=true
|
||||
|
||||
assert_file_contains "${unit_file}" 'Environment="AI_WORKSPACE_AUTH_TOKEN=recovered-from-systemd"'
|
||||
assert_file_contains "${unit_file}" 'Environment="BRIDGE_AUTH_TOKEN=recovered-from-systemd"'
|
||||
assert_file_contains "${unit_file}" 'Environment="BRIDGE_REVIEW_AUTH_TOKEN=recovered-review-token"'
|
||||
|
||||
rm -rf "${tmp_dir}"
|
||||
@ -209,7 +209,7 @@ EOF
|
||||
|
||||
run_fail_fast_case() {
|
||||
local tmp_dir
|
||||
tmp_dir="$(setup_test_env "case: missing AI_WORKSPACE_AUTH_TOKEN fails fast with clear error")"
|
||||
tmp_dir="$(setup_test_env "case: missing BRIDGE_AUTH_TOKEN fails fast with clear error")"
|
||||
|
||||
local log_file="${tmp_dir}/deploy.log"
|
||||
local stderr_file="${tmp_dir}/deploy.stderr"
|
||||
@ -226,9 +226,9 @@ run_fail_fast_case() {
|
||||
set -e
|
||||
|
||||
if [[ "${exit_code}" == "0" ]]; then
|
||||
fail "expected deploy to fail when AI_WORKSPACE_AUTH_TOKEN is empty and no system service unit exists"
|
||||
fail "expected deploy to fail when BRIDGE_AUTH_TOKEN is empty and no system service unit exists"
|
||||
fi
|
||||
assert_contains "$(cat "${stderr_file}")" "AI_WORKSPACE_AUTH_TOKEN is required"
|
||||
assert_contains "$(cat "${stderr_file}")" "BRIDGE_AUTH_TOKEN is required"
|
||||
|
||||
rm -rf "${tmp_dir}"
|
||||
}
|
||||
|
||||
@ -28,11 +28,7 @@ fi
|
||||
|
||||
BASE_URL="$(normalize_url "${BRIDGE_SERVER_URL:-${2:-https://xworkmate-bridge.svc.plus}}")"
|
||||
RPC_URL="${BASE_URL%/}/acp/rpc"
|
||||
AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
if [[ -z "${AUTH_TOKEN}" ]]; then
|
||||
echo "AI_WORKSPACE_AUTH_TOKEN or BRIDGE_AUTH_TOKEN is required" >&2
|
||||
exit 1
|
||||
fi
|
||||
AUTH_TOKEN="${BRIDGE_AUTH_TOKEN:?BRIDGE_AUTH_TOKEN is required}"
|
||||
|
||||
fast_http_curl_common=(
|
||||
--silent
|
||||
|
||||
@ -2,14 +2,14 @@
|
||||
set -euo pipefail
|
||||
|
||||
BASE_URL="${BRIDGE_SERVER_URL:-https://xworkmate-bridge.svc.plus}"
|
||||
AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
AUTH_TOKEN="${BRIDGE_AUTH_TOKEN:-}"
|
||||
REQUEST_ORIGIN="${OPENCLAW_SMOKE_ORIGIN:-https://xworkmate.svc.plus}"
|
||||
RPC_TIMEOUT_SECONDS="${OPENCLAW_SMOKE_RPC_TIMEOUT_SECONDS:-180}"
|
||||
POLL_TIMEOUT_SECONDS="${OPENCLAW_SMOKE_POLL_TIMEOUT_SECONDS:-120}"
|
||||
POLL_INTERVAL_SECONDS="${OPENCLAW_SMOKE_POLL_INTERVAL_SECONDS:-2}"
|
||||
|
||||
if [[ -z "${AUTH_TOKEN}" ]]; then
|
||||
echo "AI_WORKSPACE_AUTH_TOKEN or BRIDGE_AUTH_TOKEN is required" >&2
|
||||
echo "BRIDGE_AUTH_TOKEN is required" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
@ -2,12 +2,12 @@
|
||||
set -euo pipefail
|
||||
|
||||
BASE_URL="${BRIDGE_SERVER_URL:-https://xworkmate-bridge.svc.plus}"
|
||||
AUTH_TOKEN="${AI_WORKSPACE_AUTH_TOKEN:-${BRIDGE_AUTH_TOKEN:-}}"
|
||||
AUTH_TOKEN="${BRIDGE_AUTH_TOKEN:-}"
|
||||
HTTP_TIMEOUT_SECONDS="${HTTP_TIMEOUT_SECONDS:-30}"
|
||||
RPC_TIMEOUT_SECONDS="${RPC_TIMEOUT_SECONDS:-90}"
|
||||
|
||||
if [[ -z "${AUTH_TOKEN}" ]]; then
|
||||
echo "AI_WORKSPACE_AUTH_TOKEN or BRIDGE_AUTH_TOKEN is required" >&2
|
||||
echo "BRIDGE_AUTH_TOKEN is required" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user