feat: 添加 cpa-ext 插件及相关配置和构建脚本

2026-08-13 00:41:15 +08:00
parent 24b6d9bf5f
commit a6e764936d
18 changed files with 705 additions and 0 deletions
@@ -0,0 +1,10 @@
+/bin/
+/dist/
+*.dll
+*.so
+*.dylib
+*.h
+*.test
+coverage.out
+.cache/
+/.runtime/
@@ -0,0 +1,52 @@
+# cpa-ext
+
+`cpa-ext` 是 CLIProxyAPI 的收敛式核心扩展。当前基础里程碑只接入 Usage 能力，用来验证动态库 ABI、RPC 注册、配置热更新和并发用量入口；持久化、统计 API、管理面板和多用户 Key 管理将在此边界后逐步加入。
+
+## 当前兼容目标
+
+- CLIProxyAPI 源码：`CLIProxyAPI/`，检查时 revision 为 `f43aad7637ad813745bf7d341acb5663617570c5`
+- Native ABI：`1`
+- RPC schema：最高 `3`，注册时按宿主版本向下协商
+- 插件 ID / 动态库文件名：`cpa-ext`
+- 已声明能力：`usage_plugin`
+
+契约来源以本仓库内 `CLIProxyAPI/sdk/pluginabi/types.go`、`CLIProxyAPI/sdk/pluginapi/types.go`、`CLIProxyAPI/internal/pluginhost/rpc_schema.go` 为准。
+
+## 环境与构建（WSL/Linux）
+
+需要 Go 1.24+ 和 GCC。Ubuntu/Debian 可先准备 C 工具链：
+
+```bash
+sudo apt-get update
+sudo apt-get install -y build-essential
+./scripts/check-env.sh
+./scripts/build.sh
+```
+
+从 PowerShell 也可调用 WSL 构建入口：
+
+```powershell
+./scripts/build.ps1
+```
+
+产物为 `bin/cpa-ext.so`，适用于运行在 WSL/Linux 的 CLIProxyAPI。普通开发测试不需要 CGO：
+
+```powershell
+go test ./...
+```
+
+## 加载到 CLIProxyAPI
+
+1. 将 `bin/cpa-ext.so` 放进 Linux CLIProxyAPI 配置的插件目录（默认 `plugins/`）。
+2. 合并 `config.example.yaml` 中的 `plugins` 配置。
+3. 重启宿主，确认日志中成功加载插件 `cpa-ext`。
+4. 发起一个 Codex 请求，确认宿主调用 `usage.handle`。
+
+`config_yaml` 中包含宿主补充的 `enabled` 和 `priority`；插件会解析自己的 `codex_only` 配置。不要在日志、测试快照或管理接口中暴露 API Key、Auth ID、Bearer Token 或原始请求体。
+
+## 工程布局
+
+- `cmd/cpa-ext`：仅负责 C ABI、请求字节复制和 C 内存释放。
+- `internal/plugin`：RPC dispatcher、契约 DTO、原子配置与 Usage 接入。
+- `scripts`：环境检查与可复现构建。
+- `CLIProxyAPI`：上游契约参考，不属于插件实现。
@@ -0,0 +1,110 @@
+//go:build cshared
+
+// Command cpa-ext is the thin C ABI entry point for CLIProxyAPI.
+package main
+
+/*
+#include <stdint.h>
+#include <stdlib.h>
+
+typedef struct {
+	void* ptr;
+	size_t len;
+} cliproxy_buffer;
+
+typedef int (*cliproxy_host_call_fn)(void*, const char*, const uint8_t*, size_t, cliproxy_buffer*);
+typedef void (*cliproxy_host_free_fn)(void*, size_t);
+
+typedef struct {
+	uint32_t abi_version;
+	void* host_ctx;
+	cliproxy_host_call_fn call;
+	cliproxy_host_free_fn free_buffer;
+} cliproxy_host_api;
+
+typedef int (*cliproxy_plugin_call_fn)(char*, uint8_t*, size_t, cliproxy_buffer*);
+typedef void (*cliproxy_plugin_free_fn)(void*, size_t);
+typedef void (*cliproxy_plugin_shutdown_fn)(void);
+
+typedef struct {
+	uint32_t abi_version;
+	cliproxy_plugin_call_fn call;
+	cliproxy_plugin_free_fn free_buffer;
+	cliproxy_plugin_shutdown_fn shutdown;
+} cliproxy_plugin_api;
+
+extern int cliproxyPluginCall(char*, uint8_t*, size_t, cliproxy_buffer*);
+extern void cliproxyPluginFree(void*, size_t);
+extern void cliproxyPluginShutdown(void);
+*/
+import "C"
+
+import (
+	"net/http"
+	"unsafe"
+
+	"cpa-ext/internal/plugin"
+)
+
+var app = plugin.NewApp()
+
+func main() {}
+
+//export cliproxy_plugin_init
+func cliproxy_plugin_init(_ *C.cliproxy_host_api, api *C.cliproxy_plugin_api) C.int {
+	if api == nil {
+		return 1
+	}
+	api.abi_version = C.uint32_t(plugin.ABIVersion)
+	api.call = C.cliproxy_plugin_call_fn(C.cliproxyPluginCall)
+	api.free_buffer = C.cliproxy_plugin_free_fn(C.cliproxyPluginFree)
+	api.shutdown = C.cliproxy_plugin_shutdown_fn(C.cliproxyPluginShutdown)
+	return 0
+}
+
+//export cliproxyPluginCall
+func cliproxyPluginCall(method *C.char, request *C.uint8_t, requestLen C.size_t, response *C.cliproxy_buffer) C.int {
+	if response != nil {
+		response.ptr = nil
+		response.len = 0
+	}
+	if method == nil {
+		writeResponse(response, plugin.ErrorEnvelope("invalid_method", "缺少插件方法", http.StatusBadRequest))
+		return 1
+	}
+	var requestBytes []byte
+	if request != nil && requestLen > 0 {
+		requestBytes = C.GoBytes(unsafe.Pointer(request), C.int(requestLen))
+	}
+	raw, err := app.HandleMethod(C.GoString(method), requestBytes)
+	if err != nil {
+		writeResponse(response, plugin.ErrorEnvelope("plugin_error", err.Error(), http.StatusInternalServerError))
+		return 1
+	}
+	writeResponse(response, raw)
+	return 0
+}
+
+//export cliproxyPluginFree
+func cliproxyPluginFree(ptr unsafe.Pointer, _ C.size_t) {
+	if ptr != nil {
+		C.free(ptr)
+	}
+}
+
+//export cliproxyPluginShutdown
+func cliproxyPluginShutdown() {
+	app.Shutdown()
+}
+
+func writeResponse(response *C.cliproxy_buffer, raw []byte) {
+	if response == nil || len(raw) == 0 {
+		return
+	}
+	ptr := C.CBytes(raw)
+	if ptr == nil {
+		return
+	}
+	response.ptr = ptr
+	response.len = C.size_t(len(raw))
+}
@@ -0,0 +1,5 @@
+//go:build !cshared
+
+package main
+
+func main() {}
@@ -0,0 +1,9 @@
+plugins:
+  enabled: true
+  dir: plugins
+  configs:
+    cpa-ext:
+      enabled: true
+      priority: 100
+      codex_only: true
+
@@ -0,0 +1,5 @@
+module cpa-ext
+
+go 1.24
+
+require gopkg.in/yaml.v3 v3.0.1
@@ -0,0 +1,4 @@
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
@@ -0,0 +1,110 @@
+package plugin
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"sync"
+	"sync/atomic"
+)
+
+type App struct {
+	config atomic.Pointer[Config]
+	mu     sync.Mutex
+	closed bool
+	seen   atomic.Uint64
+}
+
+func NewApp() *App {
+	a := &App{}
+	cfg := defaultConfig()
+	a.config.Store(&cfg)
+	return a
+}
+
+func (a *App) HandleMethod(method string, request []byte) (response []byte, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			response = nil
+			err = fmt.Errorf("插件处理 %s 时发生异常: %v", method, recovered)
+		}
+	}()
+
+	switch method {
+	case MethodPluginRegister, MethodPluginReconfigure:
+		return a.configure(request)
+	case MethodUsageHandle:
+		return a.handleUsage(request)
+	case MethodPluginShutdown:
+		a.Shutdown()
+		return OKEnvelope(struct{}{})
+	default:
+		return ErrorEnvelope("unknown_method", "不支持的插件方法: "+method, http.StatusNotFound), nil
+	}
+}
+
+func (a *App) configure(raw []byte) ([]byte, error) {
+	var req LifecycleRequest
+	if len(raw) > 0 {
+		if err := json.Unmarshal(raw, &req); err != nil {
+			return nil, fmt.Errorf("解析生命周期请求: %w", err)
+		}
+	}
+	if req.SchemaVersion == 0 {
+		req.SchemaVersion = 1
+	}
+	negotiated := min(req.SchemaVersion, SchemaVersion)
+	cfg, err := decodeConfig(req.ConfigYAML)
+	if err != nil {
+		return nil, err
+	}
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if a.closed {
+		return nil, fmt.Errorf("插件已经关闭")
+	}
+	a.config.Store(&cfg)
+	return OKEnvelope(registration(negotiated))
+}
+
+func registration(schemaVersion uint32) Registration {
+	return Registration{
+		SchemaVersion: schemaVersion,
+		Metadata: Metadata{
+			Name:             PluginName,
+			Version:          Version,
+			Author:           "cpa-ext",
+			GitHubRepository: "https://git.pchuan.top/agent/cpa-plugin",
+			ConfigFields: []ConfigField{
+				{Name: "enabled", Type: "boolean", Description: "启用 CPA 扩展。"},
+				{Name: "codex_only", Type: "boolean", Description: "只接收 Codex/OpenAI 模型的用量事件。"},
+			},
+		},
+		Capabilities: Capabilities{UsagePlugin: true},
+	}
+}
+
+func (a *App) handleUsage(raw []byte) ([]byte, error) {
+	var record UsageRecord
+	if err := json.Unmarshal(raw, &record); err != nil {
+		return nil, fmt.Errorf("解析用量事件: %w", err)
+	}
+	cfg := a.config.Load()
+	if cfg != nil && cfg.accepts(record) {
+		// The first milestone only proves ingestion. Persistence and aggregation
+		// belong in a separate package added behind this boundary.
+		a.seen.Add(1)
+	}
+	return OKEnvelope(struct{}{})
+}
+
+func (a *App) Seen() uint64 {
+	return a.seen.Load()
+}
+
+func (a *App) Shutdown() {
+	a.mu.Lock()
+	a.closed = true
+	a.mu.Unlock()
+}
@@ -0,0 +1,109 @@
+package plugin
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"sync"
+	"testing"
+)
+
+func lifecycleRequest(t *testing.T, schema uint32, config string) []byte {
+	t.Helper()
+	raw, err := json.Marshal(LifecycleRequest{ConfigYAML: []byte(config), SchemaVersion: schema})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+func TestRegisterNegotiatesSchemaAndDeclaresOnlyUsage(t *testing.T) {
+	app := NewApp()
+	raw, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, 99, "enabled: true\ncodex_only: true\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var env Envelope
+	if err := json.Unmarshal(raw, &env); err != nil {
+		t.Fatal(err)
+	}
+	if !env.OK {
+		t.Fatalf("register failed: %s", raw)
+	}
+	var got Registration
+	if err := json.Unmarshal(env.Result, &got); err != nil {
+		t.Fatal(err)
+	}
+	if got.SchemaVersion != SchemaVersion || !got.Capabilities.UsagePlugin {
+		t.Fatalf("unexpected registration: %+v", got)
+	}
+}
+
+func TestLifecycleConfigYAMLUsesBase64WireEncoding(t *testing.T) {
+	raw := lifecycleRequest(t, SchemaVersion, "enabled: true\n")
+	if !json.Valid(raw) {
+		t.Fatal("invalid JSON")
+	}
+	var wire map[string]any
+	_ = json.Unmarshal(raw, &wire)
+	want := base64.StdEncoding.EncodeToString([]byte("enabled: true\n"))
+	if wire["config_yaml"] != want {
+		t.Fatalf("config_yaml = %v, want %q", wire["config_yaml"], want)
+	}
+}
+
+func TestReconfigureIsAtomicAndFiltersUsage(t *testing.T) {
+	app := NewApp()
+	if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, "enabled: true\ncodex_only: true\n")); err != nil {
+		t.Fatal(err)
+	}
+	for _, record := range []UsageRecord{{Provider: "codex", Model: "gpt-5.5"}, {Provider: "gemini", Model: "gemini-pro"}} {
+		raw, _ := json.Marshal(record)
+		if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if got := app.Seen(); got != 1 {
+		t.Fatalf("seen = %d, want 1", got)
+	}
+	if _, err := app.HandleMethod(MethodPluginReconfigure, lifecycleRequest(t, SchemaVersion, "enabled: [invalid")); err == nil {
+		t.Fatal("invalid reconfiguration unexpectedly succeeded")
+	}
+	raw, _ := json.Marshal(UsageRecord{Provider: "codex", Model: "gpt-5.5"})
+	_, _ = app.HandleMethod(MethodUsageHandle, raw)
+	if got := app.Seen(); got != 2 {
+		t.Fatalf("last valid config was not retained: seen = %d", got)
+	}
+}
+
+func TestConcurrentUsage(t *testing.T) {
+	app := NewApp()
+	raw, _ := json.Marshal(UsageRecord{Provider: "codex", Model: "gpt-5.5"})
+	var wg sync.WaitGroup
+	for range 100 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
+				t.Errorf("handle usage: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+	if got := app.Seen(); got != 100 {
+		t.Fatalf("seen = %d, want 100", got)
+	}
+}
+
+func TestUnknownMethodReturnsErrorEnvelope(t *testing.T) {
+	raw, err := NewApp().HandleMethod("missing", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var env Envelope
+	if err := json.Unmarshal(raw, &env); err != nil {
+		t.Fatal(err)
+	}
+	if env.OK || env.Error == nil || env.Error.Code != "unknown_method" {
+		t.Fatalf("unexpected envelope: %s", raw)
+	}
+}
@@ -0,0 +1,40 @@
+package plugin
+
+import (
+	"fmt"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+)
+
+type Config struct {
+	Enabled   bool `yaml:"enabled"`
+	CodexOnly bool `yaml:"codex_only"`
+}
+
+func defaultConfig() Config {
+	return Config{Enabled: true, CodexOnly: true}
+}
+
+func decodeConfig(raw []byte) (Config, error) {
+	cfg := defaultConfig()
+	if len(raw) == 0 {
+		return cfg, nil
+	}
+	if err := yaml.Unmarshal(raw, &cfg); err != nil {
+		return Config{}, fmt.Errorf("解析插件配置: %w", err)
+	}
+	return cfg, nil
+}
+
+func (c Config) accepts(record UsageRecord) bool {
+	if !c.Enabled {
+		return false
+	}
+	if !c.CodexOnly {
+		return true
+	}
+	provider := strings.ToLower(strings.TrimSpace(record.Provider))
+	model := strings.ToLower(strings.TrimSpace(record.Model))
+	return provider == "codex" || strings.Contains(model, "codex") || strings.HasPrefix(model, "gpt-")
+}
@@ -0,0 +1,26 @@
+package plugin
+
+import (
+	"encoding/json"
+	"strings"
+)
+
+func OKEnvelope(value any) ([]byte, error) {
+	raw, err := json.Marshal(value)
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(Envelope{OK: true, Result: raw})
+}
+
+func ErrorEnvelope(code, message string, status int) []byte {
+	raw, _ := json.Marshal(Envelope{
+		OK: false,
+		Error: &EnvelopeError{
+			Code:       strings.TrimSpace(code),
+			Message:    strings.TrimSpace(message),
+			HTTPStatus: status,
+		},
+	})
+	return raw
+}
@@ -0,0 +1,104 @@
+package plugin
+
+import (
+	"encoding/json"
+	"net/http"
+	"time"
+)
+
+const (
+	ABIVersion    uint32 = 1
+	SchemaVersion uint32 = 3
+	PluginName           = "cpa-ext"
+	Version              = "0.1.0-dev"
+)
+
+const (
+	MethodPluginRegister    = "plugin.register"
+	MethodPluginReconfigure = "plugin.reconfigure"
+	MethodPluginShutdown    = "plugin.shutdown"
+	MethodUsageHandle       = "usage.handle"
+)
+
+type Envelope struct {
+	OK     bool            `json:"ok"`
+	Result json.RawMessage `json:"result,omitempty"`
+	Error  *EnvelopeError  `json:"error,omitempty"`
+}
+
+type EnvelopeError struct {
+	Code       string `json:"code"`
+	Message    string `json:"message"`
+	Retryable  bool   `json:"retryable,omitempty"`
+	HTTPStatus int    `json:"http_status,omitempty"`
+}
+
+type LifecycleRequest struct {
+	ConfigYAML    []byte `json:"config_yaml"`
+	SchemaVersion uint32 `json:"schema_version"`
+}
+
+type Registration struct {
+	SchemaVersion uint32       `json:"schema_version"`
+	Metadata      Metadata     `json:"metadata"`
+	Capabilities  Capabilities `json:"capabilities"`
+}
+
+type Metadata struct {
+	Name             string
+	Version          string
+	Author           string
+	GitHubRepository string
+	Logo             string
+	ConfigFields     []ConfigField
+}
+
+type ConfigField struct {
+	Name        string
+	Type        string
+	EnumValues  []string
+	Description string
+}
+
+type Capabilities struct {
+	UsagePlugin bool `json:"usage_plugin"`
+}
+
+// UsageRecord mirrors CLIProxyAPI sdk/pluginapi. Keep it in sync with the
+// target host because exported field names are part of the JSON wire contract.
+type UsageRecord struct {
+	Provider        string
+	ExecutorType    string
+	Model           string
+	Alias           string
+	APIKey          string
+	AuthID          string
+	AuthIndex       string
+	AuthType        string
+	Source          string
+	ReasoningEffort string
+	ServiceTier     string
+	Generate        bool
+	RequestedAt     time.Time
+	Latency         time.Duration
+	TTFT            time.Duration
+	Failed          bool
+	Failure         UsageFailure
+	Detail          UsageDetail
+	ResponseHeaders http.Header
+}
+
+type UsageFailure struct {
+	StatusCode int
+	Body       string
+}
+
+type UsageDetail struct {
+	InputTokens         int64
+	OutputTokens        int64
+	ReasoningTokens     int64
+	CachedTokens        int64
+	CacheReadTokens     int64
+	CacheCreationTokens int64
+	TotalTokens         int64
+}
@@ -0,0 +1,13 @@
+[CmdletBinding()]
+param()
+
+$ErrorActionPreference = 'Stop'
+Push-Location (Split-Path -Parent $PSScriptRoot)
+try {
+    & wsl.exe bash ./scripts/build.sh
+} finally {
+    Pop-Location
+}
+if ($LASTEXITCODE -ne 0) {
+    throw "WSL 构建失败，退出码 $LASTEXITCODE。"
+}
@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+bash "$root/scripts/check-env.sh"
+
+cd "$root"
+mkdir -p bin
+gofmt -w cmd internal
+go mod tidy
+go test ./...
+CGO_ENABLED=1 go build -tags cshared -buildmode=c-shared -o bin/cpa-ext.so ./cmd/cpa-ext
+
+echo "Built: $root/bin/cpa-ext.so"
@@ -0,0 +1,17 @@
+[CmdletBinding()]
+param()
+
+$ErrorActionPreference = 'Stop'
+
+function Resolve-Tool([string]$Name) {
+    $tool = Get-Command $Name -ErrorAction SilentlyContinue
+    if ($null -eq $tool) {
+        throw "缺少 $Name，请先安装并加入 PATH。"
+    }
+    return $tool.Source
+}
+
+$go = Resolve-Tool 'go'
+Write-Host "Go:  $(& $go version)"
+Write-Host 'CLIProxyAPI target: ABI 1 / RPC schema 3'
+Write-Warning '本项目默认在 WSL/Linux 构建；请优先运行 wsl bash ./scripts/build.sh。'
@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+command -v go >/dev/null 2>&1 || {
+  echo '缺少 Go，请先在 WSL 中安装 Go 1.24+ 并加入 PATH。' >&2
+  exit 1
+}
+command -v gcc >/dev/null 2>&1 || {
+  echo '缺少 gcc，请安装 build-essential（CGO 构建动态库需要）。' >&2
+  exit 1
+}
+
+echo "Go:  $(go version)"
+echo "GCC: $(gcc --version | head -n 1)"
+echo 'CLIProxyAPI target: ABI 1 / RPC schema 3'
+
@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+runtime="$root/.runtime"
+pid_file="$runtime/cliproxy.pid"
+log_file="$runtime/cliproxy.log"
+
+if [[ -f "$pid_file" ]]; then
+  old_pid="$(cat "$pid_file" || true)"
+  if [[ -n "$old_pid" ]] && kill -0 "$old_pid" 2>/dev/null; then
+    echo "CPA test host is already running (PID $old_pid)."
+    exit 0
+  fi
+fi
+
+mkdir -p "$runtime/auths" "$runtime/plugins"
+: >"$log_file"
+nohup "$runtime/cliproxy" -config "$runtime/config.yaml" \
+  >"$log_file" 2>&1 </dev/null &
+pid=$!
+echo "$pid" >"$pid_file"
+echo "Started CPA test host (PID $pid)."
+
+for _ in $(seq 1 30); do
+  if curl -fsS http://127.0.0.1:8317/healthz >/dev/null 2>&1; then
+    echo 'CPA test host is ready at http://127.0.0.1:8317'
+    exit 0
+  fi
+  if ! kill -0 "$pid" 2>/dev/null; then
+    echo 'CPA test host exited during startup.' >&2
+    tail -n 100 "$log_file" >&2
+    exit 1
+  fi
+  sleep 1
+done
+
+echo 'Timed out waiting for CPA test host.' >&2
+tail -n 100 "$log_file" >&2
+exit 1
@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+pid_file="$root/.runtime/cliproxy.pid"
+
+if [[ ! -f "$pid_file" ]]; then
+  echo 'CPA test host is not running.'
+  exit 0
+fi
+
+pid="$(cat "$pid_file")"
+if [[ -n "$pid" ]] && kill -0 "$pid" 2>/dev/null; then
+  kill "$pid"
+  for _ in $(seq 1 20); do
+    kill -0 "$pid" 2>/dev/null || break
+    sleep 0.25
+  done
+fi
+rm -f "$pid_file"
+echo 'CPA test host stopped.'