diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 891f7cb..d5c6abb 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -24,6 +24,13 @@ jobs:
       - uses: actions/setup-node@v4
         with:
           node-version: '22'
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+      - name: Install the locked behavior browser
+        run: |
+          python -m pip install -r tests/requirements-cpamp-behavior.txt
+          python -m playwright install --with-deps chromium
       - uses: actions/download-artifact@v4
         with:
           name: commandcodebridge-cpamp-web
@@ -64,8 +71,20 @@ jobs:
       - name: Native panel deployment against the same CPAMP source baseline
         run: |
           chmod +x dist/cpamp-manager-test/cpa-manager-plus
-          CPA_BINARY=/tmp/cpa-test/cli-proxy-api CPAMP_BINARY="$PWD/dist/cpamp-manager-test/cpa-manager-plus" python3 tests/cpamp_native_panel.py
+          CPA_BINARY=/tmp/cpa-test/cli-proxy-api CPAMP_BINARY="$PWD/dist/cpamp-manager-test/cpa-manager-plus" python3 tests/cpamp_native_panel.py --behavior-archive dist/cpamp-web/*_cpamp-web.zip
           CPA_BINARY=/tmp/cpa-test/cli-proxy-api CPAMP_BINARY="$PWD/dist/cpamp-manager-test/cpa-manager-plus" python3 tests/cpamp_native_panel.py --environment-only
+      - name: Production Antigravity browser, snapshot and fault controls
+        run: |
+          python tests/cpamp_antigravity_behavior.py --manager dist/cpamp-manager-test/cpa-manager-plus --archive dist/cpamp-web/*_cpamp-web.zip
+          python tests/cpa_antigravity_forwarding.py --cpa /tmp/cpa-test/cli-proxy-api
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: cpamp-behavior-evidence
+          path: |
+            dist/antigravity-behavior/
+            dist/antigravity-installed/
+            dist/antigravity-forwarding.json
 
   build:
     needs: test
@@ -231,12 +250,16 @@ jobs:
           cd "$cpamp_source"
           npm ci --ignore-scripts --workspace apps/web
           npm --workspace apps/web run type-check
+          npm --workspace apps/web run test -- --config vitest.behavior.config.ts
           npm --workspace apps/web run test -- src/utils/quota/constants.test.ts src/features/accounts/components/CommandCodeQuotaCard.test.tsx src/features/accounts/components/commandCodeQuotaStore.test.ts src/features/accounts/model/commandCodeQuotaWindows.test.ts src/features/accounts/model/accountHealthEvidence.test.ts src/features/accounts/model/accountQuotaWindowDefinitions.test.ts src/features/accounts/components/QuotaWindowCard.test.tsx src/services/api/CommandCodeOAuth.test.ts src/services/api/oauth.test.ts src/features/oauth/OAuthPage.test.tsx src/features/accounts/components/accountDetail/AccountQuotaTab.test.tsx
           npm --workspace apps/web run build
           (cd apps/manager-server && go test ./internal/service/quotasnapshot ./internal/http/controller/health -run TestCommandCode -count=1)
           cd "$GITHUB_WORKSPACE"
           mkdir -p dist/cpamp-manager-test
           (cd "$cpamp_source/apps/manager-server" && go build -trimpath -o "$GITHUB_WORKSPACE/dist/cpamp-manager-test/cpa-manager-plus" ./cmd/cpa-manager-plus)
+          python -m pip install -r tests/requirements-cpamp-behavior.txt
+          python -m playwright install --with-deps chromium
+          python tests/cpamp_antigravity_mutations.py "$cpamp_source" --manager dist/cpamp-manager-test/cpa-manager-plus
           python scripts/package_cpamp.py "$cpamp_source"
           python - "$cpamp_source" <<'PY'
           import pathlib, subprocess, sys, tempfile, zipfile
@@ -247,6 +270,11 @@ jobs:
               script = pathlib.Path(temporary) / "integrations/cpamp/apply.py"
               subprocess.run([sys.executable, str(script), sys.argv[1], "--check"], check=True)
           PY
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: cpamp-mutation-evidence
+          path: dist/antigravity-mutation/
       - uses: actions/upload-artifact@v4
         with:
           name: cpamp-manager-test
diff --git a/docs/ANTIGRAVITY_BEHAVIOR_TEST_PLAN.md b/docs/ANTIGRAVITY_BEHAVIOR_TEST_PLAN.md
new file mode 100644
index 0000000..962ac96
--- /dev/null
+++ b/docs/ANTIGRAVITY_BEHAVIOR_TEST_PLAN.md
@@ -0,0 +1,185 @@
+# Antigravity 额度行为测试计划
+
+日期：2026-10-23。状态：已实现并完成本地验证；CI 门禁已接入，尚未推送触发 GitHub CI。结果见[执行记录](ANTIGRAVITY_BEHAVIOR_TEST_RESULTS.md)。
+
+## 1. 目标与验收边界
+
+围绕 PR #13 的事故补齐行为证据：执行实际请求代码，捕获发送的 UA，再验证 Gemini 和 Claude 的额度窗口经过解析、页面展示及 SQLite 持久化后仍然正确。
+
+新增测试必须证明：已知的旧 UA 回退、请求头被覆盖、窗口元信息丢失会让 CI 失败。固定响应测试无法保证 Google 将来不改变接口；真实账号是否拥有 Gemini 额度也不能由测试夹具推断。
+
+实现全部在 CommandCodeBridge 仓库维护，沿用独立临时 CPAMP checkout 应用集成文件的方式。测试未修改上游仓库或线上环境。
+
+## 2. 已核查的现状
+
+依据本地 main `032a295` 及 `dist/cpamp-reference` 中匹配的 CPAMP 源码：
+
+| 链路 | 现有能力 | 新增覆盖点 |
+| --- | --- | --- |
+| 源码与打包 | `compatibility.json` 决定基线，公共校验拒绝旧 UA scheme | 验证执行中的请求头与独立版本下限 |
+| 请求 | `providerRequests.test.ts` mock 整个 `apiCallApi.request`；UA 与同模块常量比较 | 穿过真实 `apiCallApi`、`apiClient` 和 JSON 序列化 |
+| 窗口 | 已有 builders、display windows、snapshots、matrix 的分层测试 | 将同一份请求结果贯穿整条链路 |
+| 部署 | `tests/cpamp_native_panel.py` 已验证真实 CPA 安装、真实 Manager 服务页面及字节一致性 | 浏览器执行被安装的生产页面并刷新 Antigravity 额度 |
+| CI | 显式列出 Vitest 文件；现有额度请求、builder 等测试未全部注册 | 显式注册新测试，作为 build/release 的前置条件 |
+
+两个实现细节影响测试设计：
+
+1. 应捕获 `POST /v0/management/api-call` **JSON 正文中的 `header['User-Agent']`**。浏览器访问 Manager 的 HTTP User-Agent 不代表转发给 Cloud Code 的 UA。
+2. 当前 `fetchAntigravityQuota()` 遇到第一个可用的非空 groups 就返回。因此“summary 仅有 Claude”和“summary 均不可用后 models 回退”要分别复现；不能把两者写成必然连续发生的调用链。
+
+上游 Vite 配置在 `mode === 'test'` 时启用演示模式和演示夹具。新行为测试需要明确关闭该模式，并断言本地测试服务确实收到请求。
+
+## 3. 三层行为测试
+
+### 第一层：真实请求代码 → 本地 HTTP 服务
+
+技术：Vitest + Node loopback HTTP 服务。执行真实 `fetchAntigravityQuota`、`antigravitySubscriptionApi`、`apiCallApi` 和 `apiClient`；不 mock 这些被测函数。服务模拟 CPA 管理端，读取序列化后的 JSON。
+
+服务按收到的 UA 选择响应：
+
+- 合法 `antigravity/hub/<version> <platform>` 且版本不低于独立测试契约中的 `2.23.0`：完整 Gemini / Claude 响应。
+- CLI、缺失、格式错误或低版本：裁剪响应，仅保留 Claude。
+- 根据用例指定返回 404、429、空结构或 models-only 响应。
+
+断言内容：
+
+- 正确的 `authIndex`、POST 方法、目标 endpoint 和正文 `project`；逐一检查实际调用的 summary、models、loadCodeAssist 请求。
+- UA 的 scheme、平台格式及数字版本比较；测试期望不导入生产 `ANTIGRAVITY_USER_AGENT` 常量。`2.9.0 < 2.23.0`，`2.100.0 > 2.23.0`。
+- 头名称按 HTTP 语义忽略大小写；存在多个大小写变体且值冲突时直接失败。
+- 保留管理请求与上游请求各自的 Authorization 语义；只使用虚构凭证。
+- 将返回值继续交给真实的 display windows、window definitions、snapshot write builder，检查窗口集合和字段。
+
+单独配置 `vitest.behavior.config.ts`，复用所需 alias / SCSS 设置，强制关闭演示构建及运行模式，避免普通 Vitest 配置的演示开关提前返回夹具。每个正向用例必须看到预期 HTTP 交换，未发生请求不能算通过。
+
+### 第二层：生产 ZIP → 浏览器 → 正常操作
+
+技术：Playwright Chromium；依赖独立放在本仓库测试目录并锁定版本。直接使用本次 `package_cpamp.py` 产出的 ZIP，不为浏览器测试重建演示页面。
+
+隔离 harness 启动与该 ZIP 同源构建的真实 Manager，以临时 `PANEL_PATH` 服务包内 `management.html`；Manager 的 CPA upstream 指向 loopback 假管理端。正常登录、打开凭证页、点击刷新额度，使用虚构账号和管理凭证。
+
+断言内容：
+
+1. HTTP 服务页面的 SHA256 与 ZIP、`BUILD.json` 一致；不能只验证磁盘文件。
+2. 假管理端实际收到额度请求，并按 UA 返回响应；不拦截页面内函数或填充前端额度 store。
+3. 完整响应场景下，Gemini 与 Claude 各有 5h 和周额度，比例与重置时间来自本次响应。分别使用不同数值，避免两个模型组串用仍然通过。
+4. 使用现有 `data-account-quota-matrix-row`、`data-account-quota-matrix-cell` 定位窗口，配合可访问名称触发操作；不依赖混淆变量、CSS hash 或整页截图比对。
+5. 若最终选择的列表/网格布局不渲染 matrix，使用其现有窗口卡片语义并精确限定账号；不能为了让断言通过直接注入组件。
+
+复用现有原生安装 harness 的启动、退出及页面校验逻辑：在 CI 已经安装内嵌页面的实例上增加一次主场景浏览器验收，覆盖“插件安装后的实际页面”。不重复整套安装兼容矩阵。
+
+### 第三层：页面实际写入 → Manager → SQLite → 恢复
+
+第二层 harness 保留真实 Manager 的 `/quota-snapshots` 与 `/quota-snapshots/query` 路由，禁止用浏览器 mock 响应替代。
+
+- 捕获页面发出的快照 POST；等待 Manager 写入成功后调用真实 query 接口。
+- 完整响应用例按 `authIndex`、provider、模型组、provider window ID 精确匹配四个窗口，而非仅断言总数大于零。
+- 断言 `window_kind` 为 `five_hour` / `weekly`，时长为 18,000 / 604,800 秒，模型 scope、剩余比例和重置时间对应夹具；完整用例不得产生 Gemini `unknown` 窗口。
+- 在隔离数据库中只读检查 `account_quota_windows`；数据库中的主键或聚合结构以当前 schema 为准，不依赖固定自增 ID。
+- 刷新浏览器并重启临时 Manager 后再次 query；恢复值及原始观察时间不变。恢复阶段禁止再返回新额度，防止重新抓取掩盖恢复失败。
+- 第二个虚构账号使用不同数值；相同文件名但不同 auth index 也要隔离。查询结果不能借用第一个账号的额度。
+
+这一层优先通过真实进程和实际页面验证，不再为同一个映射添加重复的 Go 单元测试。若发现后端未覆盖的独立缺陷，再追加有针对性的 Go 回归。
+
+## 4. 用例矩阵
+
+四个主窗口使用独立夹具：Gemini 5h 80%、周 60%；Claude 5h 40%、周 20%。每次运行只选取一次观察时刻 T0，之后固定使用；周期边界和重置时间相对 T0 定义，响应 Date 与浏览器测试时钟一致。
+
+| ID | 场景 | 必须成立的结果 | 执行层 |
+| --- | --- | --- | --- |
+| A01 | 合法 Hub，完整 summary | 四窗口完整；请求、UI、快照、DB 一致 | 1 / 2 / 3 |
+| A02 | CLI / 低版本 / 缺失 / 错误 UA | 假服务产生裁剪响应；同一健康场景验收必须失败 | 1；CLI 与低版本另做浏览器故障注入 |
+| A03 | 页面含 Hub 字样，但实际 header 被覆盖 | 仍捕获错误 UA；不能因页面字符串存在而通过 | 1 / 2 |
+| A04 | 合法 UA，summary 只返回 Claude | 忠实保留 Claude；健康夹具的 Gemini 完整性检查失败；不擅自合成 Gemini | 1 / 2 |
+| A05 | 第一个 summary 404，后续 summary 完整 | 按当前 URL 顺序回退；每次请求 UA 正确；四窗口正确 | 1 |
+| A06 | summary 均不可用，models-only 可用 | 保留可证实的余量；不得凭模型名捏造 5h / 周边界，shared bucket 可为 unknown | 1 / 2 / 3 |
+| A07 | summary 有组但 bucket.window 缺失 | 明确检查元信息；完整夹具验收失败，不用 label 推断掩盖 duration 丢失 | 1 |
+| A08 | 额度 endpoint 返回 429 | 停止后续额度 URL 回退；无新的成功观察写入 | 1 / 2 |
+| A09 | 订阅查询失败或返回 429，额度完整 | 保留四窗口；订阅及限流状态遵循现有契约 | 1 |
+| A10 | 空 groups 与未知成功结构 | 保留现有 inventory observed 区别；不当作完整四窗口成功 | 1 |
+| A11 | 刷新后 reload / Manager 重启 | 真实快照恢复，原观察时间不被改写，无新抓取掩盖结果 | 2 / 3 |
+| A12 | 两个账号 / 同名文件不同 auth index | 各自请求、显示、快照及恢复不串号 | 1 / 2 / 3 |
+
+“健康夹具验收失败”指测试中预先声明拥有四窗口的 fixture。正常业务允许合法账号只有部分分组，不新增“任何缺 Gemini 都是错误”的生产规则。
+
+## 5. 证明防线有效：故障注入与定向变异
+
+必须保留正向控制：相同页面、账号、project 及假服务，仅将 UA 或窗口字段改变，结果从通过变为明确的语义失败。
+
+1. **低版本源码变异**：在临时构建副本中把 Hub 版本改为 `2.22.0`，仍保留 Hub scheme。构建后让相同浏览器场景验收失败，证明仅检查 scheme 无法绕过行为防线。
+2. **实际 header 故障注入**：保持生产页面与 Hub 字符串不变，在测试传输边界覆盖实际送达假服务的 UA 为 CLI 或删除它。先证实错误请求确已送达，再检查完整窗口契约失败。
+3. **窗口元信息变异**：从完整响应中删除 Gemini 或删除 bucket.window；即使 UA 正确，仍需分别由窗口集合、duration / snapshot 断言发现。
+
+变异构建仅存在于临时目录，不进入正式 ZIP、内嵌目录或上传制品。若静态防线先拦截，也记录为静态检测，不能冒充行为层已经执行。
+
+负向用例必须匹配明确的原因码，例如 `ua_below_minimum`、`missing_gemini_window`、`missing_window_duration`；异常退出、服务没启动、超时或登录失败不能算“成功杀死变异”。不为运行变异放宽发布校验。
+
+## 6. 文件与依赖规划
+
+以下保留最初规划。实际实现使用技能要求的 Python Playwright、锁定的 requirements 文件和独立 `behavior-tests/` 目录；文件对应关系见执行记录。
+
+| 文件 | 职责 |
+| --- | --- |
+| `integrations/cpamp/antigravity-behavior-fixtures.json` | 单一测试夹具，包含响应、独立期望、相对 T0 的时间偏移及 UA 版本下限；跨 TS / Python / 浏览器共用 |
+| `integrations/cpamp/antigravityBehavior.test.ts` | 第一层真实请求和解析→快照行为测试 |
+| `integrations/cpamp/vitest.behavior.config.ts` | 非演示行为测试配置 |
+| `integrations/cpamp/apply.py` | 校验目标结构后，将测试、配置与夹具复制到独立 CPAMP checkout；保持重复应用安全 |
+| `tests/cpamp_antigravity_behavior.py` | 假 CPA 管理端、UA 敏感响应、真实 Manager 和临时 DB 生命周期 |
+| `tests/cpamp-behavior/antigravity.spec.cjs` | 浏览器正常登录、刷新、窗口、快照、恢复验收 |
+| `tests/cpamp-behavior/package.json` / `package-lock.json` | 隔离并锁定 Playwright 依赖，不改变上游依赖 |
+| `tests/cpamp_antigravity_mutations.py` | 故障注入与隔离源码变异，分类失败原因和恢复正向控制 |
+| `tests/cpamp_native_panel.py` | 暴露可复用生命周期，在已安装页面上执行一个主场景 |
+| `.github/workflows/build.yml` | 注册测试、浏览器环境、强制失败门槛与证据上传 |
+| `integrations/cpamp/README.md` | 本地执行入口、夹具及测试边界 |
+
+测试独立维护 `2.23.0` 协议下限是防止生产常量与断言同时回退。它不复制 CPAMP commit，构建基线继续只从 `compatibility.json` 读取。将来提升协议下限须有单独的契约依据和评审。
+
+## 7. CI 接入与实施顺序
+
+**步骤一：夹具与第一层。** 增加非演示 Vitest 配置及 HTTP harness，优先 A01–A07，再补限流、空响应与订阅失败。将新文件加入 `cpamp-web` 显式执行列表；现有 constants 测试保留。
+
+**步骤二：生产页面、持久化与安装后验收。** 在现有 `test` job 中，原生部署步骤后运行 Chromium 测试，复用下载的生产 ZIP、匹配 Manager 和临时 CPA 资源。所有进程使用临时目录及 loopback 随机端口。
+
+**步骤三：故障注入与源码变异。** header / 响应注入属于每次主行为测试的一部分；低版本源码变异在 `cpamp-web` 内完成隔离生产构建。明确指定正常产物路径，避免后续打包误取变异页面。全部三步完成后才满足本计划验收。
+
+CI 依赖保持：
+
+```text
+cpamp-web（源码行为 + 隔离变异 + 正式页面打包）
+    → test（现有宿主 / 安装测试 + 生产页面行为 + SQLite 恢复）
+    → build（四平台）→ release-checksums → publish-release
+```
+
+- 不使用 `continue-on-error`；断言失败不通过自动重试消除。
+- UI 通过具体状态与网络事件等待，不使用固定 sleep 充当就绪条件；重启用例有明确的进程和服务就绪检查。
+- `if: always()` 仅用于上传证据和清理资源，不能用于绕过依赖触发发布。
+- 记录新增测试耗时后再设合理超时，不预先承诺尚未测量的运行时间。
+
+## 8. 可复现性与失败证据
+
+- 浏览器只放行本次创建的 loopback 服务，意外域名/端口访问立即失败。假服务按 endpoint 返回固定数据，不向 JSON 中的 Google URL 发起网络访问。
+- 真实 Manager 的后台同步也必须通过配置或测试进程的网络限制保持本地可控；仅设置浏览器 route 不能约束后台进程。首次执行需验证进程外连边界。
+- 夹具、浏览器和假服务使用同一 T0，报告记录 T0 以便复现。真实 Manager 使用系统时间，启动与写入应落在约定容差内，重置时间留足余量；不修改宿主系统时钟。新进程、新账号上下文和新 SQLite 文件隔离每组用例。
+- 产出 JSON 报告：场景、实际 UA、请求 endpoint、窗口集合、snapshot 字段、DB 对照、页面 SHA256、CPAMP 基线及变异检测原因。
+- 浏览器失败时保存 trace、截图和相关日志；屏蔽 Authorization、Cookie、管理密钥及 access token。全部使用虚构凭证，报告也不输出原始认证头。
+- 临时进程和端口均在 finally 中回收；测试不读取用户浏览器状态或已有数据库。
+
+## 9. 独立补充：真实 CPA 转发契约
+
+上述三层证明生产页面将正确 UA 交给管理 API、Manager 保持窗口语义，假 CPA 并不能证明真实 CPA 最终向 Google 发出的 HTTP 头。
+
+已增加 `tests/cpa_antigravity_forwarding.py`：用虚构 auth fixture 调用真实 CPA `/api-call`，目标为 loopback 接收器，检查接收器读到的真实 HTTP User-Agent 及 token 占位符处理。CPA v8.0.23 实测通过；后台的外部连接尝试被本地拒绝代理截获，不访问 Google。
+
+此项单独报告覆盖状态。真实 Google 接口变化仍需未来另行安排授权的实测；本计划不引入定时监控或真实账号访问。
+
+## 10. 完成标准
+
+- [x] 完整夹具贯穿真实请求、生产页面、快照 HTTP、SQLite 和恢复，四窗口逐项匹配。
+- [x] CLI / 低版本 / 缺失 UA 的负向控制均得到预期的语义失败，正向控制通过。
+- [x] 页面含 Hub 字符串但实际 header 错误也会失败。
+- [x] 缺 Gemini 与 models-only 分别覆盖，不虚构周期元信息。
+- [x] 两账号及同名不同 auth index 不串数据，重启恢复没有重新抓取掩盖问题。
+- [x] 演示路径未介入，预期 HTTP 交换确实发生，未访问真实上游。
+- [x] 新测试接入 build/release 的必经依赖链；本地检查 YAML 与依赖通过，GitHub 运行结果待推送验证。
+- [x] 已输出请求、快照、DB、截图及变异原因证据；真实 CPA 转发契约已补齐。
+
+满足这些标准后，可表述为“已知事故模式有可执行的回归防线”；不能表述为“以后绝不会出现类似 bug”。
diff --git a/docs/ANTIGRAVITY_BEHAVIOR_TEST_RESULTS.md b/docs/ANTIGRAVITY_BEHAVIOR_TEST_RESULTS.md
new file mode 100644
index 0000000..e4a76f7
--- /dev/null
+++ b/docs/ANTIGRAVITY_BEHAVIOR_TEST_RESULTS.md
@@ -0,0 +1,76 @@
+# Antigravity 行为测试执行记录
+
+日期：2026-10-23。环境：Windows、本地隔离进程、Playwright Chromium。仓库基于 main `032a295`；CPAMP 源码基线由 `internal/bridge/compatibility.json` 读取。
+
+## 结果
+
+| 验证 | 结果 | 证据 |
+| --- | --- | --- |
+| 新增真实 HTTP 请求行为 | **19/19 通过** | `antigravityBehavior.test.ts`，Vitest 4.1.5；未 mock 请求、订阅或 API 客户端 |
+| 原有前端回归 | **11 文件、151/151 通过** | 沿用 CI 显式列表 |
+| 前端类型检查与生产构建 | **通过** | `tsc --noEmit`、Vite production、`package_cpamp.py` |
+| 原有面板身份与发布防线 | **6/6 组通过** | `python -m unittest discover -s tests -p test_cpamp_panel_validation.py -v` |
+| Manager CommandCode 快照及健康声明回归 | **通过** | `go test ./internal/service/quotasnapshot ./internal/http/controller/health -run TestCommandCode -count=1` |
+| 本次生产 ZIP 的浏览器行为 | **8/8 场景通过** | `dist/antigravity-current/summary.json`，逐场景 JSON 与截图 |
+| 源码 Hub 版本降至 2.22.0 | **成功检出** | `dist/antigravity-mutation/full.json`；实际 UA 为低版本，窗口契约报 `missing_gemini_window` |
+| 真实 CPA 安装内嵌页面后的行为 | **通过** | `dist/antigravity-installed/summary.json`；实际安装文件与公开 v0.2.11 ZIP、Manager HTTP 页面字节一致，随后相同文件通过隔离 Manager 行为测试 |
+| 真实 CPA v8.0.23 HTTP 转发 | **2/2 通过** | `dist/antigravity-forwarding.json`；分别用 `User-Agent` 和 `user-agent` 检查最终 HTTP 头和 `$TOKEN$` 替换 |
+| Workflow YAML、依赖链、Python 语法、diff 空白检查 | **通过** | 本地解析与检查 |
+
+浏览器 8 个场景：
+
+1. 完整 Hub 响应：两个账号分别匹配 Gemini / Claude 的 5h 与周额度。
+2. 实际传输边界改为 CLI UA：检出 `missing_gemini_window`。
+3. 实际传输边界改为 Hub 2.22.0：检出 `missing_gemini_window`。
+4. 删除实际 UA：检出 `missing_gemini_window`。
+5. 正确 UA 但 summary 仅有 Claude：检出 `missing_gemini_window`。
+6. 删除 bucket.window：检出 `missing_window_duration`，不能用 label 推断掩盖缺失。
+7. models-only：持久化 `unknown`，不虚构 5h / 周边界；重启恢复正确。
+8. 额度查询 429：停止回退，没有成功快照写入。
+
+负向场景的“通过”表示明确检出了指定缺陷。登录失败、超时、构建失败和普通异常不会被当作检出成功。
+
+## 三层链路的实际证据
+
+- 请求层捕获的是 `/api-call` JSON 的 `header`，并独立验证版本下限、endpoint、auth index、project、订阅请求和所有被调用的回退 URL。
+- 正式页面在正常登录后点击“刷新额度”，检查每个账号的四个窗口比例及重置提示；继续正常打开“额度”详情，触发页面实际快照 POST。
+- 快照通过真实 Manager HTTP 路由写入 SQLite。查询结果与 `account_quota_windows` 的账号、窗口 ID、窗口类型和模型 scope 一致。
+- 完整响应用例验证 Gemini 80% / 60%、Claude 40% / 20%；另一账号分别减 10 个百分点。5h / 周时长为 18,000 / 604,800 秒，周期结束时间匹配夹具。
+- Manager 重启后查询值、原始观察时间及页面恢复值一致；恢复期间假上游拒绝新抓取，并断言没有新增额度请求。
+- 两账号及同名文件不同 auth index 分别覆盖；后者没有借用已有账号的持久化窗口。
+- 低版本源码变异使用临时生产构建，真实请求发出 `antigravity/hub/2.22.0 darwin/arm64`，即使页面包含 Hub scheme，窗口完整性检查仍将其拦截。正式构建未被改写。
+
+## 实现文件
+
+| 文件 | 用途 |
+| --- | --- |
+| `integrations/cpamp/antigravity-behavior-fixtures.json` | 共享响应夹具、独立版本下限和窗口期望 |
+| `integrations/cpamp/antigravityBehavior.test.ts` | 实际 HTTP 序列化、解析、窗口和快照构造测试 |
+| `integrations/cpamp/vitest.behavior.config.ts` | 关闭演示模式，使用 production alias；测试位于 `behavior-tests/`，避免 Node 类型影响浏览器代码 |
+| `integrations/cpamp/apply.py` | 向独立 CPAMP checkout 安装测试配置和夹具，重复应用验证通过 |
+| `tests/cpamp_antigravity_behavior.py` | Python Playwright、假 CPA、真实 Manager、SQLite、恢复及故障控制 |
+| `tests/cpamp_antigravity_mutations.py` | 隔离源码降版构建并要求精确语义失败 |
+| `tests/cpa_antigravity_forwarding.py` | 真实 CPA 到 loopback 接收器的 HTTP 头与 token 替换契约 |
+| `tests/cpamp_native_panel.py` | 安装后验证页面一致，再对实际安装的页面执行行为主场景 |
+| `tests/requirements-cpamp-behavior.txt` | 锁定 Python Playwright 及依赖版本 |
+| `.github/workflows/build.yml` | 行为测试成为构建、发布依赖链中的必经步骤，始终上传诊断证据 |
+
+使用 Python Playwright 代替最初规划中的 Node spec，遵循 `webapp-testing` 技能；上游 npm 依赖未改变。诊断输出使用脱敏 JSON、失败截图与 HTML；未开启可能记录认证正文的浏览器 trace。
+
+## CI 与执行边界
+
+已接入：
+
+```text
+cpamp-web：类型检查 → 请求行为 → 原有回归 → 正式构建 → 低版本源码变异 → 打包
+    ↓
+test：原有宿主测试 → 实际安装页面行为 → 生产页面/SQLite/故障控制 → 真实 CPA 转发
+    ↓
+四平台 build → release-checksums → publish-release
+```
+
+本轮代码尚未 commit / push，因此 **没有本轮 GitHub CI 运行结果，也没有合并或发版**。本地检查通过不能表述为 GitHub CI 已绿。
+
+测试使用虚构凭证和临时数据库。浏览器仅允许本次 loopback 服务；假管理端不访问 JSON 中的 Google URL。真实 CPA 的后台外部连接尝试被本地拒绝代理截获；真正的 UA 转发目标为 loopback。未访问真实 Google 额度服务或任何 VPS。
+
+目前验证的是已知协议和事故模式；Google 将来改变接口、账户权限或客户端最低版本，仍需要新的契约依据与实测。
diff --git a/integrations/cpamp/README.md b/integrations/cpamp/README.md
index 7b3af0e..5e9ae09 100644
--- a/integrations/cpamp/README.md
+++ b/integrations/cpamp/README.md
@@ -36,6 +36,30 @@ v0.2.9 四平台插件已内嵌匹配页面，独立前端包也包含上述功
 
 CPAMP 当前源码的额度平台列表固定为内置平台，因此仅增加插件额度接口仍不会在原生凭证页渲染 CommandCode。本适配直接在 CPAMP 的原生网格、列表显示左右两区：左侧 5h、周额度上下两排，右侧月额度与 5h 顶部对齐；保留窗口标签、剩余百分比、进度条、相对重置时间和月余额。完整重置日期可悬停查看，刷新使用右侧图标；复用面板已有的 Button 和 QuotaProgressBar。详情额度页使用原生窗口历史卡片。
 
+## Antigravity 行为回归测试
+
+除页面身份校验外，CI 还执行真实 HTTP 请求代码、生产页面浏览器操作、Manager 快照与 SQLite 恢复、UA/响应故障注入及源码版本回退测试。测试仅使用 loopback 服务和虚构凭证。完整场景与边界见[行为测试计划](../../docs/ANTIGRAVITY_BEHAVIOR_TEST_PLAN.md)。
+
+在已经应用本仓库集成的独立 CPAMP checkout 中运行请求层测试：
+
+```bash
+npm --workspace apps/web run test -- --config vitest.behavior.config.ts
+```
+
+在 CommandCodeBridge 仓库根目录运行浏览器与持久化测试；Manager 必须从 `compatibility.json` 指定的同一源码基线构建，ZIP 必须是对应生产页面：
+
+```bash
+python -m pip install -r tests/requirements-cpamp-behavior.txt
+PLAYWRIGHT_SKIP_BROWSER_GC=1 python -m playwright install chromium
+python tests/cpamp_antigravity_behavior.py --manager /path/to/cpa-manager-plus --archive /path/to/commandcodebridge_cpamp-web.zip
+python tests/cpamp_antigravity_mutations.py /path/to/CPA-Manager-Plus --manager /path/to/cpa-manager-plus
+python tests/cpa_antigravity_forwarding.py --cpa /path/to/cli-proxy-api
+```
+
+浏览器报告和截图默认写入 `dist/antigravity-behavior/`；源码回退报告在 `dist/antigravity-mutation/`。回退构建位于临时目录，不进入正式打包。`cpamp_native_panel.py --behavior-archive <ZIP>` 会在插件安装及 HTTP 页面一致性验证后，将实际安装的相同页面交给隔离 Manager 执行主行为场景。
+
+行为失败必须阻止 `cpamp-web → test → build → release` 链路。真实 Google 接口和真实账号未在这些测试中访问。
+
 ## 安装
 
 1. 备份原插件动态库、CPA auth-dir、插件 data_dir 以及 CPAMP 当前管理页面。
diff --git a/integrations/cpamp/antigravity-behavior-fixtures.json b/integrations/cpamp/antigravity-behavior-fixtures.json
new file mode 100644
index 0000000..d201dc7
--- /dev/null
+++ b/integrations/cpamp/antigravity-behavior-fixtures.json
@@ -0,0 +1,22 @@
+{
+  "minimumHubVersion": [2, 23, 0],
+  "accounts": [
+    {"name": "antigravity-a.json", "auth_index": "ag-fixture-a", "type": "antigravity", "provider": "antigravity", "project_id": "fixture-project-a", "email": "a@example.test"},
+    {"name": "antigravity-b.json", "auth_index": "ag-fixture-b", "type": "antigravity", "provider": "antigravity", "project_id": "fixture-project-b", "email": "b@example.test"}
+  ],
+  "groups": [
+    {"displayName": "Gemini Models", "scope": "gemini", "buckets": [
+      {"bucketId": "gemini-5h", "displayName": "5 hour limit", "window": "5h", "remainingFraction": 0.8, "durationSeconds": 18000, "resetOffsetSeconds": 7200},
+      {"bucketId": "gemini-weekly", "displayName": "Weekly limit", "window": "weekly", "remainingFraction": 0.6, "durationSeconds": 604800, "resetOffsetSeconds": 172800}
+    ]},
+    {"displayName": "Claude and GPT models", "scope": "claude_gpt", "buckets": [
+      {"bucketId": "claude-5h", "displayName": "5 hour limit", "window": "5h", "remainingFraction": 0.4, "durationSeconds": 18000, "resetOffsetSeconds": 10800},
+      {"bucketId": "claude-weekly", "displayName": "Weekly limit", "window": "weekly", "remainingFraction": 0.2, "durationSeconds": 604800, "resetOffsetSeconds": 259200}
+    ]}
+  ],
+  "models": {
+    "gemini-fixture": {"displayName": "Gemini Fixture", "apiProvider": "API_PROVIDER_GOOGLE_GEMINI", "quotaInfo": {"remainingFraction": 0.8}},
+    "claude-fixture": {"displayName": "Claude Fixture", "apiProvider": "API_PROVIDER_ANTHROPIC_VERTEX", "quotaInfo": {"remainingFraction": 0.4}}
+  },
+  "subscription": {"currentTier": {"id": "g1-pro-tier", "name": "Antigravity Pro"}}
+}
diff --git a/integrations/cpamp/antigravityBehavior.test.ts b/integrations/cpamp/antigravityBehavior.test.ts
new file mode 100644
index 0000000..6b51593
--- /dev/null
+++ b/integrations/cpamp/antigravityBehavior.test.ts
@@ -0,0 +1,196 @@
+import { createServer, type Server } from 'node:http';
+import type { AddressInfo } from 'node:net';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import type { TFunction } from 'i18next';
+import type { AuthFileItem } from '@/types';
+import { fetchAntigravityQuota } from '@/utils/quota/providerRequests';
+import { buildAccountRows, type AccountQuotaStores } from '@/features/accounts/model/accountRows';
+import { buildAccountQuotaDisplayWindows } from '@/features/accounts/model/accountQuotaDisplayWindows';
+import { buildAccountQuotaWindowDefinitions } from '@/features/accounts/model/accountQuotaWindowDefinitions';
+import { buildAccountQuotaSnapshotWriteEntries } from '@/features/accounts/model/accountQuotaSnapshots';
+import { buildQuotaCredentialIdentity, getQuotaCredentialStoreKey } from '@/utils/quota/credentialScope';
+import { isDemoMode, setDemoMode } from '@/features/demo/demoMode';
+import fixture from './antigravity-behavior-fixtures.json';
+
+type WireRequest = { authIndex: string; method: string; url: string; header: Record<string, string>; data: string };
+const t = ((key: string) => key) as TFunction;
+const t0 = Math.floor(Date.now() / 1000) * 1000;
+let server: Server;
+let base: string;
+let scenario: string;
+let requests: WireRequest[];
+let invalidPaths: string[];
+
+function hubError(header: Record<string, string>): string | null {
+  const values = Object.entries(header).filter(([key]) => key.toLowerCase() === 'user-agent').map(([, value]) => value);
+  if (!values.length) return 'ua_missing';
+  if (new Set(values).size !== 1) return 'ua_conflicting';
+  const match = /^antigravity\/hub\/(\d+)\.(\d+)\.(\d+) [a-z0-9_-]+\/[a-z0-9_-]+$/i.exec(values[0]);
+  if (!match) return 'ua_invalid';
+  const parts = match.slice(1, 4).map(Number);
+  for (let i = 0; i < 3; i++) {
+    if (parts[i] < fixture.minimumHubVersion[i]) return 'ua_below_minimum';
+    if (parts[i] > fixture.minimumHubVersion[i]) return null;
+  }
+  return null;
+}
+
+function summary(authIndex: string, partial = false, missingWindow = false) {
+  return { groups: fixture.groups.filter((group) => !partial || group.scope !== 'gemini').map((group) => ({
+    displayName: group.displayName,
+    buckets: group.buckets.map((bucket) => ({
+      bucketId: bucket.bucketId, displayName: bucket.displayName,
+      ...(!missingWindow ? { window: bucket.window } : {}),
+      remainingFraction: bucket.remainingFraction - (authIndex === 'ag-fixture-b' ? 0.1 : 0),
+      resetTime: new Date(t0 + bucket.resetOffsetSeconds * 1000).toISOString(),
+    })),
+  })) };
+}
+
+beforeEach(async () => {
+  vi.stubEnv('NO_PROXY', '127.0.0.1,localhost');
+  vi.stubEnv('no_proxy', '127.0.0.1,localhost');
+  setDemoMode(false);
+  scenario = 'full'; requests = []; invalidPaths = [];
+  server = createServer(async (req, res) => {
+    let text = '';
+    for await (const chunk of req) text += chunk;
+    res.setHeader('Content-Type', 'application/json');
+    res.setHeader('Date', new Date(t0).toUTCString());
+    if (req.url !== '/v0/management/api-call' || req.method !== 'POST' || req.headers.authorization !== 'Bearer fixture-management') {
+      invalidPaths.push(req.url ?? ''); res.writeHead(400); res.end('{}'); return;
+    }
+    const wire: WireRequest = JSON.parse(text);
+    requests.push(wire);
+    const subscription = wire.url.endsWith(':loadCodeAssist');
+    const models = wire.url.endsWith(':fetchAvailableModels');
+    let status = 200;
+    let body: unknown;
+    const effective = { ...wire.header };
+    if (scenario.startsWith('ua:')) {
+      delete effective['User-Agent'];
+      if (scenario !== 'ua:missing') effective['User-Agent'] = scenario.slice(3);
+    }
+    if (subscription) {
+      status = scenario === 'subscription429' ? 429 : scenario === 'subscription403' ? 403 : 200;
+      body = fixture.subscription;
+    } else if (scenario === 'quota429') { status = 429; body = { error: 'rate limit' }; }
+    else if (scenario === 'empty') body = { groups: [] };
+    else if (scenario === 'unknown') body = {};
+    else if (scenario === 'models' && !models) { status = 404; body = {}; }
+    else if (scenario === 'fallback' && !models && requests.filter((item) => item.url.endsWith(':retrieveUserQuotaSummary')).length === 1) { status = 404; body = {}; }
+    else if (models) body = { models: fixture.models };
+    else body = summary(wire.authIndex, Boolean(hubError(effective)) || scenario === 'partial', scenario === 'missingWindow');
+    res.end(JSON.stringify({ status_code: status, header: { Date: new Date(t0).toUTCString() }, body: JSON.stringify(body) }));
+  });
+  await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+  base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+});
+
+afterEach(async () => {
+  await new Promise<void>((resolve, reject) => { server.close((error) => error ? reject(error) : resolve()); server.closeAllConnections(); });
+  vi.unstubAllEnvs();
+  expect(invalidPaths).toEqual([]);
+});
+
+async function run(file: AuthFileItem = fixture.accounts[0]) {
+  const result = await fetchAntigravityQuota(file, t, { apiBase: base, managementKey: 'fixture-management' });
+  const stores: AccountQuotaStores = { antigravityQuota: {}, codexQuota: {}, claudeQuota: {}, devinQuota: {}, kimiQuota: {}, metaQuota: {}, xaiQuota: {} };
+  stores.antigravityQuota[getQuotaCredentialStoreKey(file)] = { ...result, status: 'success', fetchedAtMs: t0, ...buildQuotaCredentialIdentity(file) };
+  const [row] = buildAccountRows([file], stores);
+  const windows = buildAccountQuotaDisplayWindows(row, { stores, t, translateQuotaWindowLabel: (label) => label ?? '', nowMs: t0 });
+  const definitions = buildAccountQuotaWindowDefinitions(windows, t0);
+  const entries = buildAccountQuotaSnapshotWriteEntries([row], new Map([[row.selectionKey, definitions]]), { nowMs: t0 });
+  return { result, windows, definitions, entries };
+}
+
+function requireComplete(value: Awaited<ReturnType<typeof run>>, accountOffset = 0) {
+  for (const group of fixture.groups) for (const bucket of group.buckets) {
+    const definition = value.definitions.find((item) => item.providerWindowId.endsWith(`:${bucket.bucketId}`));
+    if (!definition) throw new Error(group.scope === 'gemini' ? 'missing_gemini_window' : 'missing_claude_window');
+    if (definition.durationSeconds !== bucket.durationSeconds) throw new Error('missing_window_duration');
+    expect(definition.kind).toBe(bucket.window === '5h' ? 'five_hour' : 'weekly');
+    expect(definition.modelScope).toEqual({ kind: 'family', key: group.scope, complete: true });
+    expect(definition.remainingPercent).toBeCloseTo((bucket.remainingFraction - accountOffset) * 100);
+    expect(definition.cycleEndMs).toBe(t0 + bucket.resetOffsetSeconds * 1000);
+    const persisted = value.entries.flatMap((entry) => entry.windows).find((item) => item.provider_window_id === definition.providerWindowId);
+    expect(persisted).toMatchObject({ window_kind: definition.kind, duration_seconds: bucket.durationSeconds, model_scope_key: group.scope, cycle_end_ms: definition.cycleEndMs });
+  }
+  expect(value.definitions).toHaveLength(4);
+}
+
+describe('Antigravity real HTTP behavior (non-demo)', () => {
+  it('A01 executes quota/subscription, serialization, display and snapshot code', async () => {
+    expect(__DEMO_SITE__).toBe(false); expect(isDemoMode()).toBe(false);
+    requireComplete(await run());
+    expect(requests).toHaveLength(2);
+    for (const wire of requests) {
+      expect(hubError(wire.header)).toBeNull();
+      expect(wire).toMatchObject({ authIndex: 'ag-fixture-a', method: 'POST', header: { Authorization: 'Bearer $TOKEN$' } });
+      expect(wire.url).toMatch(/^https:\/\/(daily-cloudcode-pa|cloudcode-pa)\.(sandbox\.)?googleapis\.com\/v1internal:/);
+      if (wire.url.endsWith(':retrieveUserQuotaSummary')) expect(JSON.parse(wire.data)).toEqual({ project: 'fixture-project-a' });
+    }
+  });
+  it.each(['antigravity/cli/1.0.13', 'antigravity/hub/2.22.0 darwin/arm64', 'missing', 'malformed'])('A02/A03 kills wrong wire UA %s with missing Gemini', async (ua) => {
+    scenario = `ua:${ua}`;
+    const value = await run();
+    expect(requests.length).toBeGreaterThan(0);
+    expect(value.result.groups.some((group) => /claude/i.test(group.label))).toBe(true);
+    expect(() => requireComplete(value)).toThrow('missing_gemini_window');
+  });
+  it('A04 detects a partial summary even with a valid UA', async () => {
+    scenario = 'partial'; const value = await run();
+    expect(requests.every((wire) => !hubError(wire.header))).toBe(true);
+    expect(() => requireComplete(value)).toThrow('missing_gemini_window');
+    expect(requests.some((wire) => wire.url.endsWith(':fetchAvailableModels'))).toBe(false);
+  });
+  it('A05 retries summary endpoints while retaining the UA', async () => {
+    scenario = 'fallback'; requireComplete(await run());
+    const summaries = requests.filter((wire) => wire.url.endsWith(':retrieveUserQuotaSummary'));
+    expect(summaries).toHaveLength(2);
+    expect(summaries[0].url).toContain('daily-cloudcode-pa.googleapis.com');
+    expect(summaries[1].url).toContain('daily-cloudcode-pa.sandbox.googleapis.com');
+    requests.forEach((wire) => expect(hubError(wire.header)).toBeNull());
+  });
+  it('A06 preserves models-only observations without inventing windows', async () => {
+    scenario = 'models'; const value = await run();
+    expect(value.definitions).toHaveLength(2);
+    expect(value.definitions.every((item) => item.kind === 'unknown' && item.durationSeconds === null)).toBe(true);
+    expect(value.entries[0].windows.every((item) => item.window_kind === 'unknown')).toBe(true);
+    expect(requests.filter((wire) => wire.url.endsWith(':retrieveUserQuotaSummary'))).toHaveLength(3);
+    requests.forEach((wire) => expect(hubError(wire.header)).toBeNull());
+  });
+  it('A07 detects lost window duration even when labels infer the kind', async () => {
+    scenario = 'missingWindow';
+    const value = await run(); expect(() => requireComplete(value)).toThrow('missing_window_duration');
+  });
+  it('A08 stops quota retries on 429 and produces no successful result', async () => {
+    scenario = 'quota429'; await expect(run()).rejects.toMatchObject({ status: 429 });
+    expect(requests.filter((wire) => !wire.url.endsWith(':loadCodeAssist'))).toHaveLength(1);
+  });
+  it.each(['subscription429', 'subscription403'])('A09 retains quota after %s', async (mode) => {
+    scenario = mode; const value = await run(); requireComplete(value);
+    expect(value.result.subscription).toBeNull();
+    expect(Boolean(value.result.rateLimited)).toBe(mode === 'subscription429');
+  });
+  it.each(['empty', 'unknown'])('A10 retains inventory semantics for %s', async (mode) => {
+    scenario = mode; const value = await run();
+    expect(value.result.groups).toEqual([]);
+    expect(value.result.quotaInventoryObserved).toBe(mode === 'empty');
+    expect(value.definitions).toEqual([]);
+  });
+  it('A12 isolates different auth indices with the same filename', async () => {
+    const a = await run();
+    const b = await run({ ...fixture.accounts[1], name: fixture.accounts[0].name });
+    requireComplete(a); requireComplete(b, 0.1);
+    expect(a.entries[0].account.auth_index).toBe('ag-fixture-a');
+    expect(b.entries[0].account.auth_index).toBe('ag-fixture-b');
+    expect(a.entries[0].row_key).not.toBe(b.entries[0].row_key);
+  });
+  it.each([
+    ['antigravity/hub/2.9.0 darwin/arm64', 'ua_below_minimum'],
+    ['antigravity/hub/2.100.0 darwin/arm64', null],
+    ['antigravity/hub/2.23.0 darwin/arm64', null],
+  ])('independent policy compares version numerically: %s', (ua, error) => expect(hubError({ 'user-agent': ua })).toBe(error));
+  it('rejects conflicting case variants of User-Agent', () => expect(hubError({ 'User-Agent': 'a', 'user-agent': 'b' })).toBe('ua_conflicting'));
+});
diff --git a/integrations/cpamp/apply.py b/integrations/cpamp/apply.py
index 342d839..26e6fca 100644
--- a/integrations/cpamp/apply.py
+++ b/integrations/cpamp/apply.py
@@ -238,4 +238,9 @@
     shutil.copy2(source / 'commandcode_health_test.go.fragment', root / 'apps/manager-server/internal/http/controller/health/commandcode_health_test.go')
     shutil.copy2(source / 'CommandCodeOAuth.test.ts', root / 'apps/web/src/services/api/CommandCodeOAuth.test.ts')
     shutil.copy2(source / 'commandCodeOAuth.ts', root / 'apps/web/src/services/api/commandCodeOAuth.ts')
+    behavior_target = root / 'apps/web/behavior-tests'
+    behavior_target.mkdir(exist_ok=True)
+    for name in ['antigravityBehavior.test.ts', 'antigravity-behavior-fixtures.json']:
+        shutil.copy2(source / name, behavior_target / name)
+    shutil.copy2(source / 'vitest.behavior.config.ts', root / 'apps/web/vitest.behavior.config.ts')
 print('PASS: CPAMP native credential grid and quota detail integration' + (' compatibility check' if args.check else ' applied'))
diff --git a/integrations/cpamp/vitest.behavior.config.ts b/integrations/cpamp/vitest.behavior.config.ts
new file mode 100644
index 0000000..fda8e84
--- /dev/null
+++ b/integrations/cpamp/vitest.behavior.config.ts
@@ -0,0 +1,13 @@
+// Test real HTTP clients with production aliases. Upstream's test mode enables demo fixtures.
+import { defineConfig, mergeConfig } from 'vitest/config';
+import upstream from './vite.config';
+
+export default defineConfig(async () => {
+  const base = typeof upstream === 'function'
+    ? await upstream({ command: 'serve', mode: 'production', isSsrBuild: false, isPreview: false })
+    : upstream;
+  return mergeConfig(base, {
+    define: { __DEMO_SITE__: 'false' },
+    test: { environment: 'node', include: ['behavior-tests/antigravityBehavior.test.ts'], testTimeout: 10000 },
+  });
+});
diff --git a/tests/cpa_antigravity_forwarding.py b/tests/cpa_antigravity_forwarding.py
new file mode 100644
index 0000000..211c49e
--- /dev/null
+++ b/tests/cpa_antigravity_forwarding.py
@@ -0,0 +1,100 @@
+"""Verify real CPA /api-call HTTP UA and $TOKEN$ forwarding using a fresh fake Antigravity credential."""
+import argparse
+import json
+import os
+import pathlib
+import subprocess
+import tempfile
+import time
+
+from cpamp_antigravity_behavior import FakeCPA, LOCAL, free_port
+import urllib.request
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--cpa', type=pathlib.Path, required=True)
+    parser.add_argument('--output', type=pathlib.Path, default=pathlib.Path('dist/antigravity-forwarding.json'))
+    args = parser.parse_args()
+    token = 'fixture-antigravity-access-token'
+    headers_seen = []
+    with FakeCPA() as server, tempfile.TemporaryDirectory(prefix='antigravity-forward-') as temporary:
+        original_post = server.server.RequestHandlerClass.do_POST
+
+        def receiver(handler):
+            if handler.path == '/quota':
+                handler.rfile.read(int(handler.headers.get('Content-Length', '0')))
+                headers_seen.append({'ua': handler.headers.get('User-Agent'),
+                                     'token_matched': handler.headers.get('Authorization') == 'Bearer ' + token})
+                handler.send({'groups': []})
+            else:
+                original_post(handler)
+
+        server.server.RequestHandlerClass.do_POST = receiver
+        folder = pathlib.Path(temporary)
+        auth = folder / 'auth'
+        auth.mkdir()
+        (auth / 'antigravity-fixture.json').write_text(json.dumps({
+            'type': 'antigravity', 'email': 'fixture@example.test', 'project_id': 'fixture-project',
+            'access_token': token, 'expired': '2099-01-01T00:00:00Z',
+        }), encoding='utf-8')
+        port = free_port()
+        config = folder / 'config.yaml'
+        config.write_text(f'''host: "127.0.0.1"
+port: {port}
+auth-dir: "{auth.as_posix()}"
+remote-management:
+  allow-remote: false
+  secret-key: "fixture-management-key"
+  disable-control-panel: true
+proxy-url: "{server.url}"
+api-keys: ["fixture-client-key"]
+request-retry: 0
+logging-to-file: false
+''', encoding='utf-8')
+
+        def call(path, body=None):
+            req = urllib.request.Request(f'http://127.0.0.1:{port}/v0/management' + path,
+                                         data=None if body is None else json.dumps(body).encode(),
+                                         headers={'Authorization': 'Bearer fixture-management-key', 'Content-Type': 'application/json'})
+            with LOCAL.open(req, timeout=5) as response:
+                return json.load(response)
+
+        with (folder / 'cpa.log').open('w', encoding='utf-8') as log:
+            process = subprocess.Popen([str(args.cpa.resolve()), '-config', str(config)], cwd=folder, stdout=log, stderr=log,
+                                       creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0)
+            try:
+                deadline = time.monotonic() + 20
+                while True:
+                    assert process.poll() is None, 'CPA exited before forwarding test'
+                    try:
+                        files = call('/auth-files')['files']
+                        if files:
+                            break
+                    except OSError:
+                        pass
+                    if time.monotonic() > deadline:
+                        raise RuntimeError('cpa_auth_fixture_not_ready')
+                    time.sleep(.05)
+                index = next(item['auth_index'] for item in files if item['name'] == 'antigravity-fixture.json')
+                for header_name in ('User-Agent', 'user-agent'):
+                    response = call('/api-call', {'authIndex': index, 'method': 'POST', 'url': server.url + '/quota',
+                                                 'proxy_url': 'none', 'header': {header_name: 'antigravity/hub/2.23.0 darwin/arm64',
+                                                 'Authorization': 'Bearer $TOKEN$'}, 'data': '{"project":"fixture-project"}'})
+                    assert response['status_code'] == 200
+                assert len(headers_seen) == 2
+                assert all(item['ua'] == 'antigravity/hub/2.23.0 darwin/arm64' and item['token_matched'] for item in headers_seen)
+                args.output.parent.mkdir(parents=True, exist_ok=True)
+                args.output.write_text(json.dumps({'status': 'passed', 'headers': headers_seen, 'blocked_egress': server.egress}, indent=2))
+                print('PASS: real CPA forwards HTTP User-Agent and resolves $TOKEN$ from fixture Antigravity auth')
+            finally:
+                process.terminate()
+                try:
+                    process.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait(timeout=5)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/cpamp_antigravity_behavior.py b/tests/cpamp_antigravity_behavior.py
new file mode 100644
index 0000000..cc51ae5
--- /dev/null
+++ b/tests/cpamp_antigravity_behavior.py
@@ -0,0 +1,472 @@
+"""Production panel -> real Manager snapshots -> isolated SQLite, without cloud credentials.
+
+python tests/cpamp_antigravity_behavior.py --manager /path/to/manager --archive /path/to/cpamp-web.zip
+"""
+import argparse
+import contextlib
+import copy
+import hashlib
+import json
+import os
+import pathlib
+import re
+import socket
+import sqlite3
+import subprocess
+import tempfile
+import threading
+import time
+import urllib.parse
+import urllib.request
+import zipfile
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+FIXTURE = json.loads((ROOT / 'integrations/cpamp/antigravity-behavior-fixtures.json').read_text())
+ADMIN = 'fixture-panel-admin'
+MANAGEMENT = 'fixture-cpa-management'
+LOCAL = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+
+
+def free_port():
+    with socket.socket() as sock:
+        sock.bind(('127.0.0.1', 0))
+        return sock.getsockname()[1]
+
+
+def ua_error(headers):
+    values = [value for key, value in headers.items() if key.lower() == 'user-agent']
+    if not values:
+        return 'ua_missing'
+    if len(set(values)) != 1:
+        return 'ua_conflicting'
+    match = re.fullmatch(r'antigravity/hub/(\d+)\.(\d+)\.(\d+) [a-z0-9_-]+/[a-z0-9_-]+', values[0], re.I)
+    if not match:
+        return 'ua_invalid'
+    if tuple(map(int, match.groups())) < tuple(FIXTURE['minimumHubVersion']):
+        return 'ua_below_minimum'
+    return None
+
+
+class FakeCPA:
+    def __init__(self, scenario='full'):
+        self.scenario = scenario
+        self.t0 = int(time.time()) * 1000
+        self.calls = []
+        self.paths = []
+        self.egress = []
+        self.auth_errors = []
+        self.contract_errors = []
+        self.freeze = False
+        owner = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *_):
+                pass
+
+            def send(self, payload, status=200):
+                data = json.dumps(payload).encode()
+                self.send_response(status)
+                self.send_header('Content-Type', 'application/json')
+                self.send_header('Content-Length', str(len(data)))
+                self.send_header('X-Cpa-Version', 'v8.0.23')
+                self.end_headers()
+                self.wfile.write(data)
+
+            def do_CONNECT(self):
+                owner.egress.append(self.path)
+                self.send({'error': 'external_network_denied'}, 502)
+
+            def do_GET(self):
+                if self.path.startswith(('http://', 'https://')):
+                    owner.egress.append(self.path)
+                    self.send({'error': 'external_network_denied'}, 502)
+                    return
+                owner.paths.append(self.path)
+                path = urllib.parse.urlsplit(self.path).path
+                if path.endswith('/auth-files'):
+                    self.send({'files': FIXTURE['accounts']})
+                elif path.endswith('/config'):
+                    self.send({'debug': False, 'usage-statistics-enabled': False, 'remote-management': {}})
+                elif path.endswith('/models'):
+                    self.send({'object': 'list', 'data': []})
+                elif path.endswith('/usage'):
+                    self.send({'usage': {'apis': {}, 'total_requests': 0}})
+                elif path.endswith('/usage/queue'):
+                    self.send({'items': []})
+                else:
+                    self.send({})
+
+            def do_POST(self):
+                payload = json.loads(self.rfile.read(int(self.headers.get('Content-Length', '0'))) or '{}')
+                owner.paths.append(self.path)
+                if not self.path.endswith('/api-call'):
+                    self.send({})
+                    return
+                if self.headers.get('Authorization') != 'Bearer ' + MANAGEMENT:
+                    owner.auth_errors.append('invalid_management_authorization')
+                header = dict(payload.get('header', {}))
+                if header.get('Authorization') != 'Bearer $TOKEN$':
+                    owner.contract_errors.append('missing_upstream_token_placeholder')
+                original_ua = next((value for key, value in header.items() if key.lower() == 'user-agent'), None)
+                if owner.scenario.startswith('ua:'):
+                    header = {key: value for key, value in header.items() if key.lower() != 'user-agent'}
+                    if owner.scenario != 'ua:missing':
+                        header['User-Agent'] = owner.scenario[3:]
+                error = ua_error(header)
+                url = payload.get('url', '')
+                if not re.fullmatch(r'https://(?:daily-cloudcode-pa(?:\.sandbox)?|cloudcode-pa)\.googleapis\.com/v1internal:(?:retrieveUserQuotaSummary|fetchAvailableModels|loadCodeAssist)', url):
+                    owner.contract_errors.append('unexpected_upstream_endpoint')
+                account = next((item for item in FIXTURE['accounts'] if item['auth_index'] == payload.get('authIndex')), None)
+                if account is None or payload.get('method') != 'POST':
+                    owner.contract_errors.append('invalid_request_account_or_method')
+                if account and not url.endswith(':loadCodeAssist') and json.loads(payload.get('data', '{}')) != {'project': account['project_id']}:
+                    owner.contract_errors.append('wrong_project')
+                owner.calls.append({'authIndex': payload.get('authIndex'), 'method': payload.get('method'),
+                                    'url': url, 'original_ua': original_ua,
+                                    'effective_ua': next((v for k, v in header.items() if k.lower() == 'user-agent'), None),
+                                    'ua_error': error, 'data': payload.get('data')})
+                if owner.freeze:
+                    self.send({'status_code': 503, 'body': '{"error":"unexpected_refresh_during_restore"}'})
+                    return
+                subscription = url.endswith(':loadCodeAssist')
+                models = url.endswith(':fetchAvailableModels')
+                status = 200
+                if subscription:
+                    body = FIXTURE['subscription']
+                elif owner.scenario == 'quota429':
+                    status, body = 429, {'error': 'rate limit'}
+                elif owner.scenario == 'models' and not models:
+                    status, body = 404, {}
+                elif models:
+                    body = {'models': FIXTURE['models']}
+                else:
+                    groups = copy.deepcopy(FIXTURE['groups'])
+                    if error or owner.scenario == 'partial':
+                        groups = [group for group in groups if group['scope'] != 'gemini']
+                    for group in groups:
+                        group.pop('scope')
+                        for bucket in group['buckets']:
+                            offset = bucket.pop('resetOffsetSeconds')
+                            bucket.pop('durationSeconds')
+                            bucket['resetTime'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime((owner.t0 + offset * 1000) / 1000))
+                            if payload.get('authIndex') == FIXTURE['accounts'][1]['auth_index']:
+                                bucket['remainingFraction'] -= .1
+                            if owner.scenario == 'missingWindow':
+                                bucket.pop('window')
+                    body = {'groups': groups}
+                self.send({'status_code': status, 'header': {'Date': time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime(owner.t0 / 1000))}, 'body': json.dumps(body)})
+
+        self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.url = f'http://127.0.0.1:{self.server.server_port}'
+
+    def __enter__(self):
+        self.thread.start()
+        return self
+
+    def __exit__(self, *_):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+
+
+class Manager:
+    def __init__(self, binary, panel, folder, upstream):
+        self.binary, self.folder = binary, folder
+        self.url = f'http://127.0.0.1:{free_port()}'
+        self.db = folder / 'data/usage.sqlite'
+        self.log = (folder / 'manager.log').open('w', encoding='utf-8')
+        self.process = None
+        self.env = os.environ.copy()
+        for key in list(self.env):
+            if key.startswith(('CPA_', 'USAGE_', 'CPAMP_')) or key in ('HTTP_ADDR', 'PANEL_PATH', 'COLLECTOR_MODE', 'DATA_KEY', 'DATA_KEY_PATH'):
+                self.env.pop(key)
+        self.env.update({'HTTP_ADDR': self.url.removeprefix('http://'), 'USAGE_DATA_DIR': str(folder / 'data'),
+                         'USAGE_DB_PATH': str(self.db), 'PANEL_PATH': str(panel),
+                         'CPA_MANAGER_ADMIN_KEY': ADMIN, 'CPA_UPSTREAM_URL': upstream.url,
+                         'CPA_MANAGEMENT_KEY': MANAGEMENT, 'CPAMP_UPDATE_CHECK_ENABLED': 'false',
+                         'HTTP_PROXY': upstream.url, 'HTTPS_PROXY': upstream.url, 'ALL_PROXY': upstream.url,
+                         'NO_PROXY': '127.0.0.1,localhost', 'no_proxy': '127.0.0.1,localhost'})
+        # Avoid inherited config and secrets; all generated state stays in the temporary directory.
+        self.env['CPA_MANAGER_CONFIG'] = str(folder / 'config.json')
+
+    def start(self):
+        self.process = subprocess.Popen([str(self.binary)], cwd=self.folder, env=self.env, stdout=self.log, stderr=self.log,
+                                        creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0)
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            if self.process.poll() is not None:
+                raise RuntimeError('Manager exited: ' + (self.folder / 'manager.log').read_text())
+            try:
+                self.request('/health')
+                return
+            except OSError:
+                time.sleep(.05)
+        raise RuntimeError('manager_not_ready')
+
+    def request(self, path, body=None):
+        req = urllib.request.Request(self.url + path, data=None if body is None else json.dumps(body).encode(),
+                                     headers={'Authorization': 'Bearer ' + ADMIN, 'Content-Type': 'application/json'})
+        with LOCAL.open(req, timeout=5) as response:
+            return json.load(response)
+
+    def stop(self):
+        if self.process and self.process.poll() is None:
+            self.process.terminate()
+            try:
+                self.process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait(timeout=5)
+
+    def close(self):
+        self.stop()
+        self.log.close()
+
+
+def require_complete(windows, offset, t0):
+    for group in FIXTURE['groups']:
+        for bucket in group['buckets']:
+            window = next((item for item in windows if item['provider_window_id'].endswith(':' + bucket['bucketId'])), None)
+            if window is None:
+                raise AssertionError('missing_gemini_window' if group['scope'] == 'gemini' else 'missing_claude_window')
+            if window.get('duration_seconds') != bucket['durationSeconds']:
+                raise AssertionError('missing_window_duration')
+            assert window['window_kind'] == ('five_hour' if bucket['window'] == '5h' else 'weekly')
+            assert window['model_scope_kind'] == 'family' and window['model_scope_key'] == group['scope']
+            assert abs(window['remaining_percent'] - (bucket['remainingFraction'] - offset) * 100) < 1e-6
+            assert abs(window['used_percent'] - (1 - bucket['remainingFraction'] + offset) * 100) < 1e-6
+            assert window['cycle_end_ms'] == t0 + bucket['resetOffsetSeconds'] * 1000
+    assert len(windows) == 4
+
+
+def snapshot_identity(entry):
+    return {key: entry[key] for key in ('row_key', 'provider', 'account')}
+
+
+def stable_windows(result):
+    fields = ('provider_window_id', 'window_kind', 'duration_seconds', 'model_scope_key',
+              'remaining_percent', 'used_percent', 'observed_at_ms', 'cycle_end_ms')
+    return [[{key: window.get(key) for key in fields} for window in item['windows']] for item in result['items']]
+
+
+def run_scenario(browser, binary, content, output, scenario, expected_defect=None):
+    from playwright.sync_api import expect
+    label = re.sub(r'[^a-zA-Z0-9_-]', '_', scenario)
+    report = {'scenario': scenario, 'status': 'running', 'page_sha256': hashlib.sha256(content).hexdigest()}
+    with tempfile.TemporaryDirectory(prefix='cpamp-behavior-') as temporary:
+        folder = pathlib.Path(temporary)
+        panel = folder / 'management.html'
+        panel.write_bytes(content)
+        with FakeCPA(scenario) as upstream:
+            manager = Manager(binary, panel, folder, upstream)
+            context = browser.new_context(locale='en-US')
+            page = context.new_page()
+            blocked, errors, writes = [], [], []
+            try:
+                manager.start()
+                with LOCAL.open(manager.url + '/management.html') as response:
+                    assert response.read() == content, 'served_page_mismatch'
+
+                def route_request(route):
+                    origin = urllib.parse.urlsplit(route.request.url)
+                    if origin.scheme in ('data', 'blob') or f'{origin.scheme}://{origin.netloc}' in (manager.url, upstream.url):
+                        route.continue_()
+                    else:
+                        blocked.append(route.request.url)
+                        route.abort()
+
+                def capture(response):
+                    if response.url.endswith('/v0/management/quota-snapshots') and response.request.method == 'POST':
+                        assert response.status == 200, 'snapshot_write_failed'
+                        writes.extend(json.loads(response.request.post_data)['entries'])
+
+                context.route('**/*', route_request)
+                page.on('pageerror', lambda error: errors.append(str(error)))
+                page.on('response', capture)
+                page.goto(manager.url + '/management.html#/login')
+                page.wait_for_load_state('networkidle')
+                page.locator('input[type=password]').fill(ADMIN)
+                page.locator('input[type=password]').press('Enter')
+                page.wait_for_url(lambda url: '#/login' not in url, timeout=15000)
+                page.goto(manager.url + '/management.html#/accounts')
+                page.wait_for_load_state('networkidle')
+                accounts = FIXTURE['accounts'] if scenario == 'full' and not expected_defect else FIXTURE['accounts'][:1]
+                query_accounts = []
+                for index, account in enumerate(accounts):
+                    card = page.locator('[data-account-card]').filter(has_text=account['name'])
+                    refresh = card.get_by_role('button', name='Refresh quota', exact=True)
+                    with page.expect_response(lambda response: response.url.endswith('/api-call') and ':retrieveUserQuotaSummary' in (response.request.post_data or '')):
+                        refresh.click()
+                    expect(refresh).to_be_enabled()
+                    page.wait_for_load_state('networkidle')
+                    if scenario == 'quota429':
+                        assert not writes, 'failed_query_wrote_snapshot'
+                        expect(card).not_to_contain_text('80%')
+                        assert len([call for call in upstream.calls if not call['url'].endswith(':loadCodeAssist')]) == 1
+                        break
+                    if scenario == 'full' and not expected_defect:
+                        for group in FIXTURE['groups']:
+                            for bucket in group['buckets']:
+                                cell = card.locator(f'[data-account-quota-window$=":{bucket["bucketId"]}"]')
+                                expect(cell).to_contain_text(f'{round((bucket["remainingFraction"] - index * .1) * 100)}%')
+                                assert 'Reset:' in cell.get_attribute('title')
+                    card.get_by_role('button', name=f"Open {account['name']} detail", exact=True).click()
+                    with page.expect_response(lambda response: response.url.endswith('/v0/management/quota-snapshots') and response.request.method == 'POST', timeout=10000) as write:
+                        page.locator('#accounts-detail-tab-quota').click()
+                    assert write.value.status == 200
+                    page.wait_for_load_state('networkidle')
+                    entry = next(item for item in writes if item['account']['auth_index'] == account['auth_index'])
+                    query_accounts.append(snapshot_identity(entry))
+                    defect = expected_defect or ('missing_gemini_window' if scenario.startswith('ua:') or scenario == 'partial' else 'missing_window_duration' if scenario == 'missingWindow' else None)
+                    if defect:
+                        try:
+                            require_complete(entry['windows'], index * .1, upstream.t0)
+                        except AssertionError as error:
+                            assert str(error) == defect, f'wrong_defect: {error}'
+                            report['detected'] = defect
+                        else:
+                            raise AssertionError('mutation_survived')
+                        if scenario.startswith('ua:'):
+                            assert upstream.calls and all(call['ua_error'] for call in upstream.calls), 'wrong_ua_not_delivered'
+                    elif scenario == 'models':
+                        assert len(entry['windows']) == 2
+                        assert all(window['window_kind'] == 'unknown' and not window.get('duration_seconds') for window in entry['windows'])
+                    else:
+                        require_complete(entry['windows'], index * .1, upstream.t0)
+                    page.goto(manager.url + '/management.html#/accounts')
+                    page.wait_for_load_state('networkidle')
+
+                if query_accounts:
+                    before = manager.request('/v0/management/quota-snapshots/query', {'accounts': query_accounts})
+                    assert len(before['items']) == len(query_accounts)
+                    if scenario == 'full' and not expected_defect:
+                        for index, item in enumerate(before['items']):
+                            require_complete(item['windows'], index * .1, upstream.t0)
+                        assert before['items'][0]['account_key'] != before['items'][1]['account_key']
+                        same_name_other_index = copy.deepcopy(query_accounts[0])
+                        same_name_other_index['account']['auth_index'] = FIXTURE['accounts'][1]['auth_index']
+                        same_name_other_index['row_key'] = 'same-name-other-index'
+                        isolated = manager.request('/v0/management/quota-snapshots/query', {'accounts': [same_name_other_index]})
+                        assert not isolated['items'][0]['windows'], 'same_filename_borrowed_other_auth_index'
+                    with contextlib.closing(sqlite3.connect(f'file:{manager.db.as_posix()}?mode=ro', uri=True)) as database:
+                        database.row_factory = sqlite3.Row
+                        rows = [dict(row) for row in database.execute('SELECT account_key,provider_window_id,window_kind,model_scope_key FROM account_quota_windows')]
+                    expected_rows = {(item['account_key'], window['provider_window_id'], window['window_kind'], window.get('model_scope_key'))
+                                     for item in before['items'] for window in item['windows']}
+                    assert {(row['account_key'], row['provider_window_id'], row['window_kind'], row['model_scope_key']) for row in rows} == expected_rows
+                    report['database_windows'] = rows
+                    report['snapshots'] = stable_windows(before)
+                    if scenario in ('full', 'models') and not expected_defect:
+                        upstream.freeze = True
+                        call_count = len(upstream.calls)
+                        manager.stop()
+                        manager.start()
+                        after = manager.request('/v0/management/quota-snapshots/query', {'accounts': query_accounts})
+                        assert stable_windows(after) == stable_windows(before), 'restart_changed_snapshot'
+                        page.reload()
+                        page.wait_for_load_state('networkidle')
+                        if page.locator('input[type=password]').count():
+                            page.locator('input[type=password]').fill(ADMIN)
+                            page.locator('input[type=password]').press('Enter')
+                            page.wait_for_url(lambda url: '#/login' not in url, timeout=15000)
+                            page.goto(manager.url + '/management.html#/accounts')
+                            page.wait_for_load_state('networkidle')
+                        first = FIXTURE['accounts'][0]
+                        page.get_by_role('button', name=f"Open {first['name']} detail", exact=True).click()
+                        with page.expect_response(lambda response: response.url.endswith('/quota-snapshots/query')) as restored:
+                            page.locator('#accounts-detail-tab-quota').click()
+                        assert restored.value.status == 200
+                        page.wait_for_load_state('networkidle')
+                        assert stable_windows(restored.value.json()) == [stable_windows(before)[0]], 'page_restore_changed_snapshot'
+                        if scenario == 'full':
+                            cards = page.locator('[data-quota-card-mode="model"]')
+                            expect(cards).to_have_count(4)
+                            for group in FIXTURE['groups']:
+                                family = 'Gemini' if group['scope'] == 'gemini' else 'Claude'
+                                for bucket in group['buckets']:
+                                    matching = cards.filter(has_text=re.compile(family, re.I)).filter(
+                                        has=page.locator('strong[title]', has_text=re.compile('Weekly' if bucket['window'] == 'weekly' else '5', re.I)))
+                                    expect(matching).to_have_count(1)
+                                    expect(matching).to_contain_text(f'{round(bucket["remainingFraction"] * 100)}%')
+                        assert len(upstream.calls) == call_count, 'unexpected_refresh_during_restore'
+                        report['restart_restore'] = 'passed'
+                assert not blocked, 'browser_external_network: ' + repr(blocked)
+                assert not errors, 'browser_runtime_errors: ' + repr(errors)
+                assert not upstream.auth_errors, upstream.auth_errors
+                assert not upstream.contract_errors, upstream.contract_errors
+                # Proxy-denied background requests cannot reach the internet; record attempts separately.
+                report['blocked_manager_egress'] = upstream.egress
+                if scenario == 'full' and not expected_defect:
+                    assert upstream.calls and all(call['ua_error'] is None for call in upstream.calls), 'invalid_actual_ua'
+                report.update({'status': 'passed', 'requests': upstream.calls, 't0': upstream.t0})
+                page.screenshot(path=str(output / f'{label}.png'), full_page=True)
+                print('PASS:', scenario, report.get('detected', report.get('restart_restore', 'behavior')), flush=True)
+            except Exception as error:
+                with contextlib.suppress(Exception):
+                    page.screenshot(path=str(output / f'{label}-failure.png'), full_page=True)
+                    (output / f'{label}-failure.html').write_text(page.content(), encoding='utf-8')
+                report.update({'status': 'failed', 'error': str(error).replace(ADMIN, '[REDACTED]').replace(MANAGEMENT, '[REDACTED]'),
+                               'requests': upstream.calls, 't0': upstream.t0})
+                raise
+            finally:
+                try:
+                    context.close()
+                finally:
+                    manager.close()
+                    (output / f'{label}.json').write_text(json.dumps(report, indent=2), encoding='utf-8')
+    return report
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--manager', type=pathlib.Path, required=True)
+    parser.add_argument('--archive', type=pathlib.Path)
+    parser.add_argument('--panel', type=pathlib.Path, help='Test-only mutation page; never packaged for release')
+    parser.add_argument('--output', type=pathlib.Path, default=ROOT / 'dist/antigravity-behavior')
+    parser.add_argument('--scenario', action='append', help='Default: full positive and all browser fault controls')
+    parser.add_argument('--expected-defect', choices=['missing_gemini_window', 'missing_window_duration'])
+    args = parser.parse_args()
+    from playwright.sync_api import sync_playwright
+    args.output.mkdir(parents=True, exist_ok=True)
+    manifest = None
+    if args.panel:
+        content = args.panel.read_bytes()
+    else:
+        if not args.archive:
+            parser.error('--archive or --panel is required')
+        with zipfile.ZipFile(args.archive) as archive:
+            content = archive.read('management.html')
+            manifest = json.loads(archive.read('BUILD.json'))
+            assert hashlib.sha256(content).hexdigest() == manifest['management_html_sha256']
+    if args.panel and args.archive:
+        with zipfile.ZipFile(args.archive) as archive:
+            assert content == archive.read('management.html'), 'installed_page_differs_from_archive'
+            manifest = json.loads(archive.read('BUILD.json'))
+    if manifest:
+        compatibility = json.loads((ROOT / 'internal/bridge/compatibility.json').read_text())
+        assert manifest['cpamp_source_commit'] == compatibility['cpamp_source_commit'], 'manager_source_baseline_mismatch'
+        assert hashlib.sha256(content).hexdigest() == manifest['management_html_sha256']
+    scenarios = args.scenario or ['full', 'ua:antigravity/cli/1.0.13', 'ua:antigravity/hub/2.22.0 darwin/arm64', 'ua:missing', 'partial', 'missingWindow', 'models', 'quota429']
+    valid_scenarios = {'full', 'partial', 'missingWindow', 'models', 'quota429'}
+    if any(scenario not in valid_scenarios and not scenario.startswith('ua:') for scenario in scenarios):
+        parser.error('unknown behavior scenario')
+    reports = []
+    status = 'failed'
+    with sync_playwright() as playwright:
+        browser = playwright.chromium.launch(headless=True)
+        try:
+            for scenario in scenarios:
+                reports.append(run_scenario(browser, args.manager.resolve(), content, args.output, scenario, args.expected_defect))
+            status = 'passed'
+        finally:
+            browser.close()
+            (args.output / 'summary.json').write_text(json.dumps({
+                'status': status, 'cpamp_source_commit': (manifest or {}).get('cpamp_source_commit'),
+                'scenarios': reports,
+            }, indent=2), encoding='utf-8')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/cpamp_antigravity_mutations.py b/tests/cpamp_antigravity_mutations.py
new file mode 100644
index 0000000..dc21049
--- /dev/null
+++ b/tests/cpamp_antigravity_mutations.py
@@ -0,0 +1,53 @@
+"""Build a quarantined low-version production page and require a semantic browser failure."""
+import argparse
+import json
+import os
+import pathlib
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('checkout', type=pathlib.Path)
+    parser.add_argument('--manager', type=pathlib.Path, required=True)
+    parser.add_argument('--output', type=pathlib.Path, default=ROOT / 'dist/antigravity-mutation')
+    args = parser.parse_args()
+    checkout = args.checkout.resolve()
+    source = checkout / 'apps/web'
+    temporary_root = checkout / 'dist'
+    temporary_root.mkdir(exist_ok=True)
+    args.output.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix='antigravity-mutation-', dir=temporary_root) as temporary:
+        scratch = pathlib.Path(temporary).resolve()
+        assert scratch.is_relative_to(temporary_root.resolve())
+        # An ancestor node_modules supplies the existing locked dependencies; no second install.
+        shutil.copytree(source, scratch / 'web', ignore=shutil.ignore_patterns('node_modules', 'dist', 'dist-demo', 'behavior-tests'))
+        constants = scratch / 'web/src/utils/quota/constants.ts'
+        original = constants.read_text(encoding='utf-8')
+        mutated, count = re.subn(r"(export const ANTIGRAVITY_HUB_VERSION = )'[^']+'", r"\g<1>'2.22.0'", original)
+        assert count == 1, 'mutation_anchor_changed'
+        constants.write_text(mutated, encoding='utf-8')
+        env = os.environ.copy()
+        env.update({'VERSION': 'behavior-test-mutation', 'DEMO_SITE': 'false', 'VITE_DEMO_SITE': 'false'})
+        subprocess.run(['node', str(checkout / 'node_modules/vite/bin/vite.js'), 'build'], cwd=scratch / 'web', env=env, check=True)
+        panel = scratch / 'web/dist/index.html'
+        assert b'antigravity/hub/' in panel.read_bytes(), 'mutation_did_not_build'
+        subprocess.run([sys.executable, str(ROOT / 'tests/cpamp_antigravity_behavior.py'),
+                        '--manager', str(args.manager.resolve()), '--panel', str(panel),
+                        '--scenario', 'full', '--expected-defect', 'missing_gemini_window',
+                        '--output', str(args.output.resolve())], check=True)
+        report = json.loads((args.output / 'full.json').read_text())
+        assert report['status'] == 'passed' and report['detected'] == 'missing_gemini_window'
+        assert report['requests'] and all(call['ua_error'] == 'ua_below_minimum' for call in report['requests'])
+        assert all('hub/2.22.0 ' in call['original_ua'] for call in report['requests'])
+        print('PASS: source Hub version downgrade killed by production browser/window contract')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/cpamp_native_panel.py b/tests/cpamp_native_panel.py
index b83d33b..bd74822 100644
--- a/tests/cpamp_native_panel.py
+++ b/tests/cpamp_native_panel.py
@@ -18,6 +18,7 @@
 
 parser = argparse.ArgumentParser(description=__doc__)
 parser.add_argument('--environment-only', action='store_true', help='Reproduce CPAMP with explicit usage data/database paths and no config.json')
+parser.add_argument('--behavior-archive', type=pathlib.Path, help='Execute the installed panel with the Antigravity browser/snapshot harness and compare it to this ZIP')
 args = parser.parse_args()
 manager_source = pathlib.Path(os.environ['CPAMP_BINARY']).resolve()
 local = urllib.request.build_opener(urllib.request.ProxyHandler({}))
@@ -105,6 +106,10 @@ def stop():
                     f'http://127.0.0.1:{port}/management.html', '--expected', saved['panelPath']], capture_output=True, text=True)
                 assert verification.returncode == 0, verification.stdout + verification.stderr
                 assert host.call(API + '/cpamp/panel/install', {}).get('backups', []) == []
+                if args.behavior_archive:
+                    subprocess.run([os.sys.executable, str(pathlib.Path(__file__).with_name('cpamp_antigravity_behavior.py')),
+                        '--manager', str(manager_source), '--panel', saved['panelPath'], '--archive', str(args.behavior_archive.resolve()),
+                        '--scenario', 'full', '--output', str(pathlib.Path(__file__).resolve().parents[1] / 'dist/antigravity-installed')], check=True)
                 # A configured JSON panelPath also reads the file per request.
                 # Simulate an older external native page, then update in place.
                 manager_pid = process.pid
diff --git a/tests/requirements-cpamp-behavior.txt b/tests/requirements-cpamp-behavior.txt
new file mode 100644
index 0000000..b8b152a
--- /dev/null
+++ b/tests/requirements-cpamp-behavior.txt
@@ -0,0 +1,4 @@
+playwright==1.58.0
+pyee==13.0.1
+greenlet==3.5.6
+typing_extensions==4.16.0