Phase2 dry-run: pending A, probe, build-config client, gated task/add

Still default dry-run; ALLOW_LIVE_SUBMIT=0. Live gates closed.
Phase1 skeleton: replace demo main.py with src.server contract
2026-10-16 19:00:15 +08:00 · 2026-10-16 15:47:10 +08:00
48 changed files with 2278 additions and 106 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,4 +1,21 @@
 .git
-__pycache__/
+.gitignore
-*.pyc
+.venv
-.pytest_cache/
+venv
 __pycache__
 *.py[cod]
 *$py.class
 .pytest_cache
 .mypy_cache
 reports/*
 !reports/.gitkeep
 docs
 tests
 scripts
 config
 *.md
 !README.md
 .env
 .env.*
 *.log
 .DS_Store
--- a/.gitignore
+++ b/.gitignore
@@ -0,0 +1,31 @@
 # Secrets / local env
 .env
 .env.*
 !.env.example
 config/local.env
 # Python
 __pycache__/
 *.py[cod]
 *$py.class
 .venv/
 venv/
 .pytest_cache/
 .mypy_cache/
 *.egg-info/
 dist/
 build/
 # Dry-run / local artifacts (keep directory)
 reports/*
 !reports/.gitkeep
 # Editor / OS
 .DS_Store
 .idea/
 .vscode/
 *.swp
 *~
 # Docker
 *.tar
--- a/18
+++ b/18
@@ -1,14 +1,18 @@
-FROM modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim
+# ModelHub XC adaptation agent — Phase1 skeleton
-
+# Default base MUST stay the platform image (do not swap to public mirrors).
-ENV PYTHONUNBUFFERED=1
+ARG BASE_IMAGE=modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim
 FROM ${BASE_IMAGE}
 WORKDIR /app
-COPY requirements.txt .
+# Copy only what the runtime needs (see .dockerignore)
-RUN pip install --no-cache-dir -r requirements.txt
+COPY src/ ./src/
-COPY . .
+ENV PORT=8080
 ENV PYTHONUNBUFFERED=1
 # STRATEGY_ID and tokens are injected by the platform at run time — never bake them in.
 EXPOSE 8080
-CMD ["python", "main.py"]
+# Start HTTP service only (no register / submit side effects)
 CMD ["python", "-m", "src.server"]
--- a/44
+++ b/44
@@ -0,0 +1,44 @@
 # ModelHub XC agent contrib — Phase1
 # Allowed: test / build / smoke-health / dry-run
 # FORBIDDEN: register / submit / deploy / push targets
 .PHONY: build-config-dry probe-readonly test build smoke-health dry-run dry-run-pending help
 PYTHON ?= python3
 IMAGE  ?= modelhub-agent-contrib:local
 PORT   ?= 8080
 help:
 	@echo "build-config-dry generate YAML via API (requires token; never creates task)"
 	@echo "probe-readonly authenticated read-only task/page GET"
 	@echo "dry-run-pending read-only adapt pending A→B report"
 	@echo "test         stdlib unit tests + local SIGTERM integration (needs sockets)"
 	@echo "build        Docker build using platform base (needs Docker + TCR access)"
 	@echo "smoke-health local HTTP smoke (needs sockets; default test STRATEGY_ID)"
 	@echo "dry-run      read-only pending report (example fallback)"
 	@echo "No register/submit/deploy/push — default live off."
 test:
 	$(PYTHON) -m unittest discover -s tests -v
 build:
 	@command -v docker >/dev/null 2>&1 || { \
 	  echo "ERROR: docker not found. See docs/build-notes.md"; \
 	  exit 1; \
 	}
 	docker build -t $(IMAGE) .
 smoke-health:
 	$(PYTHON) scripts/smoke_health.py
 dry-run:
 	$(PYTHON) scripts/dry_run.py
 dry-run-pending:
 	PENDING_SUBMIT_MD=/workspace/modelhub-adapt/pending-submit.md PENDING_SUBMIT_TXT=/workspace/modelhub-adapt/pending-submit.txt $(PYTHON) scripts/dry_run.py
 probe-readonly:
 	$(PYTHON) scripts/probe_readonly.py
 build-config-dry:
 	$(PYTHON) scripts/build_config_dry.py
--- a/README.md
+++ b/README.md
@@ -1,39 +1,76 @@
-# Strategy Platform Demo Repo
+# modelhub-agent-contrib
-Minimal strategy repository for testing the platform Kaniko build and deployment flow.
+ModelHub XC（信创模盒）**适配智能体** Phase1 本地骨架。
-It satisfies the runtime contract from the parent README:
+> **不代注册、不代提交。** 发布智能体本身 **不得分**。积分来自成功适配，数字一律 **待平台核实**。  
 > 冲突裁决：**适配 W1 优先**；本仓默认 dry-run，不抢验证队列。
- Root-level `Dockerfile`
+## 契约要点
 - Listens on port `8080`
 - Exposes `GET /health`
 - Reads `EXTERNAL_SERVICE_TOKEN`
 - Handles `SIGTERM` for graceful shutdown
-## Local Run
+| 项 | 值 |
 |----|-----|
 | 端口 | 8080 |
 | 健康检查 | `GET /health` → `{"status":"ok"}` |
 | 启动 | **必须**有非空 `STRATEGY_ID`，否则 exit ≠0 |
 | Live | `ALLOW_LIVE_SUBMIT=1` **且** `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT`；Phase1 仍无真实 `task/add` 客户端 |
 | Base 镜像 | 平台 `xc_agent_platform/python:3.11-slim`（勿擅自改默认 FROM） |
 闸门以 `src/gate.py` 实现为准：仅 `ALLOW_LIVE_SUBMIT=1` 与
 `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` 同时满足才开逻辑闸门。PRD 示例
 `ALLOW_LIVE_SUBMIT=true` / `CONFIRM_LIVE=YES` 不受支持；本仓不读取确认文件。
 即使开闸，无真实客户端仍返回 blocked；请保持默认关闭。
 ## 快速开始
 ```bash
-python main.py
+cd /workspace/modelhub-agent-contrib
-curl http://localhost:8080/health
+make test
-curl http://localhost:8080/
+make smoke-health    # 自动起本地 server（需 STRATEGY_ID，脚本可注入测试值）
 make dry-run         # 写出 reports/dry-run-*.md
 make build           # 需 Docker + 可达平台 base；失败见 docs/build-notes.md
 ```
-## Build Locally
+本地起服务：
 ```bash
-docker build -t strategy-demo:v1.0.0 .
+export STRATEGY_ID=local-dev
-
+python3 -m src.server
-docker run --rm -p 8080:8080 strategy-demo:v1.0.0
+# → http://127.0.0.1:8080/health
 ```
-## Platform Test
+## 目录
-Push this directory to Gitea and keep the `v1.0.0` tag. Then create a strategy with:
+```text
-
+Dockerfile  Makefile  README.md
-```json
+src/        # server / health / strategy / gate
-{
+config/     # example.env, candidates.example.json
-  "name": "strategy-demo",
+scripts/    # smoke_health.py, dry_run.py, tag_release.py
-  "repo_url": "https://<your-gitea-host>/<user>/strategy-demo",
+reports/    # dry-run 输出（gitignore）
-  "tag": "v1.0.0"
+docs/       # 契约 / runbook / kickoff / issues
-}
+tests/      # stdlib unittest
 ```
 ## Makefile 目标
 - `test` / `build` / `smoke-health` / `dry-run`
 - **没有** `register` / `submit` / `deploy` / `push`
 ## 文档
 - [`docs/kickoff-for-victor.md`](docs/kickoff-for-victor.md) — 一页开工 + 验收勾选
 - [`docs/platform-contract-agent.md`](docs/platform-contract-agent.md) — 契约表
 - [`docs/register-runbook.md`](docs/register-runbook.md) — 仅人工注册
 - [`docs/api-probe-design.md`](docs/api-probe-design.md) — 写接口禁止代调
 - [`docs/diff-vs-demo.md`](docs/diff-vs-demo.md) — 对照官方 demo
 - [`docs/issues.md`](docs/issues.md) — M0–M4 / Phase 清单
 - [`docs/build-notes.md`](docs/build-notes.md) — 镜像构建说明
 ## 资源声明
 按平台量级设计：约 **100m–1 CPU / 256–512Mi**。
 ## 许可与边界
 本仓库用于 Victor 本地/Gitea 交付骨架；任何 ModelHub 写操作须 Victor 本人在控制台或显式批准后执行。
 Phase2：`make dry-run-pending` 只读生成 A→B 草稿，名称硬门与禁提过滤已接入；需 Victor 人工提交/确认。task/add 客户端默认不调用，必须显式注入并通过双闸门、队列及单次预算；测试仅 mock HTTP。
--- a/config/candidates.example.json
+++ b/config/candidates.example.json
@@ -0,0 +1,21 @@
 {
  "candidates": [
    {
      "model_id": "example-org/example-model",
      "modelAddress": "example-org/example-model",
      "taskType": "text-generation",
      "targetGpu": "Iluvatar_bi-100",
      "framework": "vllm",
      "notes": "Phase1 example only — not a real submit payload"
    },
    {
      "model_id": "example-org/another-model",
      "taskType": "text-generation",
      "notes": "Minimal fields: model_id required; others optional"
    }
  ],
  "_meta": {
    "description": "Minimal dry-run input schema for Phase1",
    "pending_align": "Future Phase2 may read /workspace/modelhub-adapt/pending-submit.md A→B order"
  }
 }
--- a/config/example.env
+++ b/config/example.env
@@ -0,0 +1,20 @@
 # Copy to config/local.env (gitignored) for local runs. Never commit real tokens.
 # Required for process start (Phase1): non-empty STRATEGY_ID
 STRATEGY_ID=local-dev-strategy-placeholder
 # Optional listen port (default 8080)
 PORT=8080
 # Implementation is authoritative: ALLOW_LIVE_SUBMIT=1 AND
 # LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT. No confirmation file is read.
 # PRD examples ALLOW_LIVE_SUBMIT=true / CONFIRM_LIVE=YES are unsupported.
 # Even with both set, no real client exists: still blocked. Keep defaults below.
 # Dual live gate — BOTH required for live; default is dry-run / blocked
 ALLOW_LIVE_SUBMIT=0
 LIVE_CONFIRM=
 # Platform injects EXTERNAL_SERVICE_TOKEN at runtime.
 # Community aliases (presence only; never log values):
 # XC_TOKEN_FILE / XC_TOKEN / MODELHUB_XC_TOKEN / EXTERNAL_SERVICE_TOKEN
 # EXTERNAL_SERVICE_TOKEN=
--- a/docs/api-probe-design.md
+++ b/docs/api-probe-design.md
@@ -0,0 +1,43 @@
 # API 探针设计（只读意图 · 禁止代调写接口）
 - 日期：2026-10-16 Asia/Shanghai
 - Phase2：客户端已实现但默认不调用；本轮禁止真实 task/add
 ## 管理 API `/agent_platform/strategies*`
 | 方法 | 路径 | 本仓态度 |
 |------|------|----------|
 | POST | `/agent_platform/strategies` | **禁止代调**（注册） |
 | POST | `.../{id}/deploy` | **禁止代调** |
 | POST | `.../{id}/stop` | **禁止代调** |
 | DELETE | `.../{id}` | **禁止代调** |
 | GET | `.../{id}` | 未来只读探针可选（Phase3+，需 Victor 授权） |
 | GET | `.../{id}/logs` | 未来只读探针可选 |
 鉴权（浏览器）：`Authorization: Bearer <login token>`（等级 A）。
 ## 适配任务 API
 | 方法 | 路径 | 本仓态度 |
 |------|------|----------|
 | GET | `/api/adapt/task/page` | 未来只读鉴权探针（对齐 huni-probe，`ALLOW_SUBMIT=0`） |
 | POST | `/api/adapt/task/build-config` | 设计可引用；Phase1 不实现 |
 | POST | `/api/adapt/task/add` | 显式注入客户端 + 双闸门；本轮只做 mock 测试 |
 | GET | `/api/adapt/task/calculate-credit` | 只读索引；勿写死 KPI |
 容器内社区鉴权（C）：`Xc-Token` + body `strategyId` — **待 Victor 账号核实**；60014 仍为 **D（unverified）**。
 ## Phase2 落地边界
 - `src/gate.py`：闸门逻辑 + `attempt_live_submit(..., client=Fake)` ；真实客户端只能显式注入，测试必须 mock HTTP
 - `src/task_add_client.py` 使用 urllib；import 和 dry-run 无网络调用，无自动重试
 - token 优先级：XC_TOKEN_FILE → XC_TOKEN → MODELHUB_XC_TOKEN → EXTERNAL_SERVICE_TOKEN；禁止输出 token
 - build_payload 中 GPU/framework/strategyId/configParams 占位须人工替换；真实调用要求 STRATEGY_ID 与 payload 一致
 - 队列 ≥95 阻断；无效队列提示也阻断；每进程一次调用预算在调用前保留，异常也不重试（防止响应丢失造成重复创建）
 - 名称/族硬门不能证明元数据 PASS；未跳过的 pending 行均标 needs_human
 - 仍禁止所有 agent_platform 写接口
 - Makefile **无** register / submit / deploy / push
 ## 冲突裁决提醒
 适配 W1：队列 ≥95 → 禁 live；智能体不抢验证空位。
--- a/docs/autoconfig-probe-plan.md
+++ b/docs/autoconfig-probe-plan.md
@@ -0,0 +1,25 @@
 # configParams 配置来源与容器验证
 日期：2026-10-16 Asia/Shanghai。
 结论：**API**。配置来自服务端，前端将响应 `data` 写入 `configParams`。
 依据：[autoconfig-capture-2026-10-16.md](/workspace/modelhub-adapt/logs/autoconfig-capture-2026-10-16.md)。原「公开面无 build-config」缺口关闭：**未公开文档但已抓包证实**。
 ## 已证实契约
 - `POST /api/adapt/task/build-config`
 - Query：`gpuType=Sunrise_pt-200-x1&framework=vllm&taskType=text-generation`
 - 空 body，无 JSON；模型地址不进入 query/body，也无需传 strategyId。
 - HTTP 2xx，JSON `code == 0`，非空字符串 `data` 为 YAML。
 - 当前 UI 无「自动配置」标签。新建适配弹层选择模型、GPU、文本生成后，选择框架触发 effect 并填入配置。
 - 抓包观察到 `Authorization` 头，值未记录。容器默认按社区实践发送 `Xc-Token`；`BUILD_CONFIG_USE_AUTHORIZATION=1` 改用同一 resolve_token 值作为 Authorization，不自动加 Bearer。
 ## 客户端与验证
 `src/build_config_client.py` 的 `build_config()` 成功返回 YAML，失败抛出 `BuildConfigError`，包含安全归因、HTTP 状态和平台数字码；不返回服务端消息或传输异常细节。缺失/无效凭证不联网，40100、40101、其他平台码、传输失败和无效 data 均可区分。导入无 I/O。
 默认 dry-run 使用 `templates/config_params.sample.yaml`，无网络。`CONFIG_PARAMS_SOURCE=build-config` 强制 API；或 `ALLOW_BUILD_CONFIG=1` 且可解析 token 时使用 API。调用失败直接报错，不回退 sample。调用方须传真实 GPU enum。`make build-config-dry` 仅生成配置，缺 token 非零退出，不输出 YAML 或凭证。
 ## 剩余 GAP 与边界
 尚无容器 token 鉴权实证；需运行时 token + API 调用并与只读探针对照，不能宣称鉴权就绪。sample 本身不是 production-ready。队列 ≥95 禁 live。本轮仅 mock 测试，不进行真实 API 调用，不调用 task/add，不保存秘密，不提交任务或推送 tag。
--- a/docs/build-notes.md
+++ b/docs/build-notes.md
@@ -0,0 +1,59 @@
 # Build notes
 - 日期：2026-10-16 Asia/Shanghai
 - 默认 `FROM`：`modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim`
 - **禁止**因本地拉不下而改默认 FROM 为公网镜像
 ## 本机（box）首次尝试
 | 项 | 结果 |
 |----|------|
 | `docker` CLI | **未安装 / 不可用**（`which docker` 空） |
 | `make build` | 预期失败：`docker not found`（Makefile 显式报错） |
 | 默认 FROM | **未修改** |
 ### 后续建议（Victor）
 1. 在有 Docker 且能访问腾讯云 TCR 平台镜像的环境执行 `make build`
 2. 若 `docker pull` 平台 base 失败：把完整错误贴本文件「拉取失败记录」，仍保持默认 FROM
 3. 无镜像时用本地验收：`STRATEGY_ID=test make smoke-health` / `make test`
 ## 拉取失败记录
 （尚无 Docker，无 pull 日志。有 Docker 后在此追加。）
 ## 2026-10-16 13:46 Asia/Shanghai — `make build` 实测
 ```text
 ERROR: docker not found. See docs/build-notes.md
 make: *** [Makefile:19: build] Error 1
 ```
 - 原因：box 无 Docker 引擎/CLI，**未发生**平台 base `docker pull`（因此无 TCR 拉取错误日志）
 - 默认 `FROM modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` **未改**
 - 替代验收：`make test` 绿；`STRATEGY_ID=test make smoke-health` 绿（本地 `python -m src.server`）
 ## 2026-10-16 Asia/Shanghai — Victor Mac `make build`
 - Docker Desktop：已连接
 - 结果：失败 — `modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` pull access denied / insufficient_scope
 - 默认 FROM：未修改
 ## 2026-10-16 Asia/Shanghai — Codex 沙箱复测
 - `make build` 实际退出 2：`ERROR: docker not found. See docs/build-notes.md`。
 - 无 Docker，未尝试拉取 TCR；本机结果不能证明 TCR 可达或不可达。上文 Mac 的拒绝记录为已有人工证据，未在本轮复现。
 - `STRATEGY_ID=test make smoke-health` 实际退出 2：沙箱创建 socket 被拒，`PermissionError: [Errno 1] Operation not permitted`。
 - `make test`：优化后 19 项中 18 项通过，SIGTERM 集成项在监听前同样被 socket 权限阻塞；未跳过或标绿。
 - 默认平台 base / FROM 保持原样；须在允许本地 socket、具备 Docker 与 TCR 权限的环境复验完整 DoD。
 ## 2026-10-16 ~15:20 Asia/Shanghai — tcr-then-build 探测（Mac）
 - Docker Desktop：已连接；`docker` CLI 可用
 - `~/.docker/config.json`：`auths` 空，`credsStore=desktop` → **未登录** TCR
 - `docker pull modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim`：
  `Error response from daemon: pull access denied ... may require 'docker login'`
 - 默认 FROM：**未修改**
 - 既有 `agent-contrib:local`（sha256:95a4d01e2b55…）：**不算**本轮 DoD（CMD=`src.main`，基底公网 python:3.11-slim）
 - 验收标准：`/Users/vc/modelhub-adapt-logs/agent-image-dod-2026-10-16.md`（box 同源 `/workspace/modelhub-adapt/logs/agent-image-dod-2026-10-16.md`）
 - 阻塞：需 Victor 提供 TCR 账号并对 `xc_agent_platform/python` 有 pull，执行 `docker login modelhubxc-4pd.tencentcloudcr.com` 后重试 pull + `make build`
--- a/docs/diff-vs-demo.md
+++ b/docs/diff-vs-demo.md
@@ -0,0 +1,40 @@
 # 对照官方演示仓 `luopingyi/xc_agent_platform_demo`
 - 日期：2026-10-16 Asia/Shanghai
 - 官方演示：https://dev.modelhub.org.cn/luopingyi/xc_agent_platform_demo
 - 安全探针参照：`huni/huni-probe-agent`（只读）
 - **不要照搬**：`huni/huni-adapt-agent` 默认真提交循环
 ## 对齐项（保留）
 | 项 | demo / 规划 | 本仓 |
 |----|-------------|------|
 | 根目录 Dockerfile | 是 | 是 |
 | 默认 base 镜像 | 平台 `python:3.11-slim` | 同（`BASE_IMAGE` 仅 build-arg，默认不改公网） |
 | `EXPOSE 8080` | 是 | 是 |
 | `GET /health` → 200 `{"status":"ok"}` | 是 | 是 |
 | 读 `EXTERNAL_SERVICE_TOKEN` 是否存在 | demo | `/status` 布尔；永不打值 |
 | stdlib / 轻依赖 | demo 极简 | Phase1 仅 stdlib |
 ## 本仓相对 demo 的补强
 | 项 | 说明 |
 |----|------|
 | `STRATEGY_ID` 启动校验 | Phase1：**缺失则 exit ≠0**（产品规划；相对「仅读 token 保活」的社区变体） |
 | 双重 live 闸门 | `ALLOW_LIVE_SUBMIT` + `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` |
 | `GET /status` | 仅布尔字段 |
 | dry-run 报告 | `scripts/dry_run.py` → `reports/dry-run-*.md` |
 | 单测 | health / 缺 STRATEGY_ID / 闸门默认拒绝 / SIGTERM |
 | 无默认真提交 | live 仅闸门 + **注入假客户端**单测；仓库无默认真打 `/api/adapt/task/add` 的 HTTP 客户端 |
 ## 明确不照搬
 | 来源 | 行为 | 本仓 |
 |------|------|------|
 | huni-adapt-agent | 默认真提交循环 | **禁止** |
 | CoolBoy / 社区刷分脚本 | 绕队列 / 无闸门 | **禁止** |
 | 旧对话口径「缺 STRATEGY_ID 保活 + /health 仍 200」 | 与产品规划 Phase1 冲突 | **以产品规划为准：启动失败** |
 ## 资源声明
 文档按约 **100m–1 CPU / 256–512Mi** 设计；Phase1 不做 k8s manifests。
--- a/docs/gpu-perf-order.md
+++ b/docs/gpu-perf-order.md
@@ -0,0 +1,49 @@
 # GPU 缺卡序（契约）
 - 日期：2026-10-16 Asia/Shanghai
 - 对齐：模盒产品经理 + 模盒积分
 - 正式性能表：**无** → 报告字段 `perf_order=unset`（勿写成官方 TPS 排名）
 - 缺卡补齐所用序：**经验序 / 待平台核实**（下表）— 产品要求排序器按此序；非站内正式性能榜
 ## A. 正式性能序
 | 优先级 | 展示名 | targetGpu | 依据 |
 |--------|--------|-----------|------|
 | — | — | — | 模盒积分定稿：无正式卡间性能表 → `perf_order=unset` |
 ## B. 缺卡经验序（排序器用 · 待平台核实）
 跨卡总策略优先于本序：已验证/已下载落点 → 只打**未适配卡**；跳过正在验证 / 已验证成功 / 唯一性失败 / 同卡已知失败禁重试；队列 ≥95 停交（live）。
 | 优先级（高→低） | 平台展示名 | targetGpu（已观察，待再核） | 备注 |
 |-----------------|------------|------------------------------|------|
 | 1（一梯队并列） | 壁砺166M | `Biren_166m` | ≈ 曦望 S2；社区成功量壁砺更高 |
 | 1（一梯队并列） | 曦望 S2 | `Sunrise_pt-200-x1`（抓包证实） | 与壁砺并列，勿臆造先后「性能」 |
 | 2 | 天垓150 | Iluvatar 天垓族（字符串待核） | 条件回退；饱和+唯一性碰撞多 |
 | 3 | 天垓100 | （待核） | 一般不扇出 |
 | 4 | 智铠 mrv-100 | `Iluvatar_mrv-100` | 我方唯一成功样本族；不当最快卡宣传 |
 并列一梯队时：若 pending `try-next` 已给出相对顺序则保留；否则可按「壁砺 / 曦望」与池内缺口并列候选，报告标 `order_source=experience`。
 ## C. 映射备忘（非排名）
 | 展示名 | 提交字段 | 备注 |
 |--------|----------|------|
 | 曦望 S2 | `Sunrise_pt-200-x1` | 2026-10-16 build-config query 抓包证实；非性能排名 |
 | （docs sample） | `MetaX_c-500` | 与中文悬赏卡无已核实映射 |
 ## D. 依据文件（工作流/观察，非官方性能榜）
 - `/workspace/modelhub-adapt/pending-submit.md`
 - `/workspace/modelhub-adapt/submit-plan-2026-10-15-v2.md`
 - `/workspace/modelhub-adapt/uniqueness-collision-notes-2026-10-15-eve.md`
 - `/workspace/modelhub-adapt/gpu-skip-patterns-xiwang-vs-bili-2026-10-15-eve2.md`
 ## E. configParams 缺口（产品补充）
 - 结论：**API**，未公开文档但已抓包证实，依据 [autoconfig-capture-2026-10-16.md](/workspace/modelhub-adapt/logs/autoconfig-capture-2026-10-16.md)。
 - `POST /api/adapt/task/build-config`，query 为 `gpuType` / `framework` / `taskType`，空 body；成功 JSON `code=0`，`data` 为 YAML。
 - UI 无「自动配置」标签；选择框架触发 effect 填入 configParams。模型地址不进入该请求。
 - 抓包观察 Authorization（值未记录）；容器默认 Xc-Token 社区实践，可显式切换 Authorization，尚无 token 鉴权实证。
 - 默认 dry-run 离线 sample；CONFIG_PARAMS_SOURCE=build-config 或 ALLOW_BUILD_CONFIG=1 且 token 可解析时调用 API，失败不可静默回退。
 - **live 前必须有 token + API 调用实证**；sample 单独不是 production-ready。队列 ≥95 继续禁 live，本轮无 live。
--- a/docs/issues.md
+++ b/docs/issues.md
@@ -0,0 +1,47 @@
 # Issue / Phase 清单
 对齐 PRD M0–M4 与产品规划 Phase 0–5。
 ## Phase / Milestone 映射
 | Phase（产品规划） | PRD | 状态（本仓） |
 |-------------------|-----|--------------|
 | Phase 0 调研冻结 | M0 | 文档已落 `docs/` |
 | Phase 1 本地可部署样例 | M1 | **本轮交付** |
 | Phase 2 Dry-run 闭环 | M2 | 已实现 pending A→B 解析、名称硬门/禁提过滤与 dry-run；元数据 PASS 仍需人工核实 |
 | Phase 3 人工注册 + 只读探针 | M3 | 仅 runbook / api-probe-design；**不代注册** |
 | Phase 4 一次 live 验证 | M4 | 闸门已备；客户端已实现；默认关闭，必须显式注入 |
 | Phase 5 规模化 | — | 未开 |
 ## P0 Issues
 - [x] docs: 冻结智能体平台契约（health/STRATEGY_ID/SIGTERM/资源）
 - [x] docs: 对照 xc_agent_platform_demo 写 diff-vs-demo
 - [x] chore: 初始化 Dockerfile / Makefile / example.env
 - [x] feat: GET /health 200 + STRATEGY_ID 校验（缺则启动失败）
 - [x] feat: SIGTERM 优雅退出（≤30s；测试 <5s）
 - [x] feat: dry-run 报告生成器（默认模式）
 - [x] feat: live 人工确认闸门 + 单测（默认拒绝）
 - [x] docs: 注册智能体 runbook — 仅人工
 - [x] chore: tag_release（只打本地 tag，不 push / 不注册）
 - [x] docs: api-probe-design（禁止代调写接口）
 Phase1 构建与运行验收须独立记录；环境阻塞不等于通过。
 ## P1
 - [x] 对接 `/workspace/modelhub-adapt/pending-submit.md` A→B 只读
 - [ ] 结构化 JSON 日志（禁密钥）
 - [ ] evidence-agent 模板填写
 ## P2（未做）
 - [ ] 贡献台账 CSV
 - [ ] 多 Tag 矩阵
 ## 硬边界（持续）
 - 无 register / submit / deploy / push Makefile 目标
 - 无 `/agent_platform/strategies*` 写调用
 - `/api/adapt/task/add` 客户端默认不调用；双闸门、队列≥95 阻断、每进程最多一次；本轮未授权真网
 - 适配 W1 优先；不抢队列
--- a/docs/kickoff-for-victor.md
+++ b/docs/kickoff-for-victor.md
@@ -0,0 +1,43 @@
 # 开工包（一页）— Victor
 **仓路径（本地）**：`/workspace/modelhub-agent-contrib/`  
 **远程**：Gitea 命名空间/仓名 **未定** → Phase1 以本地验收为主  
 **模式**：默认 dry-run；**不代注册、不代提交**
 ## 闸门
 | 条件 | 值 |
 |------|-----|
 | `ALLOW_LIVE_SUBMIT` | 默认 `0` |
 | `LIVE_CONFIRM` | 空；live 须 `I_CONFIRM_LIVE_SUBMIT` |
 | `STRATEGY_ID` | **启动必需**；缺失 → exit ≠0 |
 | 真实 `task/add` | Phase1 **未实现**（仅假客户端单测） |
 ## 本地命令
 ```bash
 cd /workspace/modelhub-agent-contrib
 make test
 make build          # 需 Docker + 可达平台 base；失败见 docs/build-notes.md
 make smoke-health   # 无镜像时脚本会拉起本地 server
 make dry-run        # 生成 reports/dry-run-*.md
 ```
 入口：`python -m src.server`（Dockerfile `CMD` 同）
 ## 验收勾选（Phase1）
 - [ ] `make test` 绿（health / 缺 STRATEGY_ID / 闸门拒绝 / SIGTERM）
 - [ ] `make build` 绿 **或** `docs/build-notes.md` 记录平台镜像拉取失败且 **未改默认 FROM**
 - [ ] `make smoke-health` 绿（镜像或本地 server）
 - [ ] 根目录 Dockerfile + `EXPOSE 8080`
 - [ ] README 写清：发布 = 0 分；积分待核实；不代注册/提交
 - [ ] Makefile **无** register/submit/deploy/push
 - [ ] dry-run 报告含「需 Victor 人工提交」「积分：待平台核实」
 - [ ] 未调用任何 ModelHub 写接口
 ## 人工下一步（非本 Phase 自动化）
 1. 定 Gitea `<namespace>/<repo>`
 2. push + tag（`scripts/tag_release.py` 仅本地 tag）
 3. 控制台新增智能体（见 `docs/register-runbook.md`）
--- a/docs/platform-contract-agent.md
+++ b/docs/platform-contract-agent.md
@@ -0,0 +1,62 @@
 # 平台契约 — 适配智能体（Phase1）
 - 日期：2026-10-16 Asia/Shanghai
 - 依据：产品规划 `agent-product-plan-2026-10-16.md` §1 + PRD `02-agent-contribution-prd.md`
 - **冲突裁决：适配 W1 优先**（队列 ≥95 禁 live；本骨架默认 dry-run）
 ## 核实等级
 | 等级 | 含义 |
 |------|------|
 | **A** | 官方演示仓 / 主站前端明文 |
 | **B** | PRD 已核实（docs/6 + 控制台，基线 2026-10-12） |
 | **C** | 社区样例，非平台承诺 |
 | **D** | 待核实 |
 ## 运行契约
 | 项 | 内容 | 等级 |
 |----|------|------|
 | 代码托管 | `dev.modelhub.org.cn` 用户仓 | B |
 | 注册 UI | 名称、仓库地址、**Git Tag**（不可用分支名） | B |
 | 构建 | 平台 Kaniko，`refs/tags/<tag>` | B |
 | Dockerfile | **仓库根目录**；base `modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim` | A |
 | 端口 | **8080**，`EXPOSE 8080` | A |
 | 健康检查 | `GET /health` → 200 `{"status":"ok"}` | A |
 | 可选状态 | `GET /status` → 布尔 `strategy_id_present` / `external_service_token_present`（**不回传值**） | 本仓 Phase1 |
 | 优雅退出 | `SIGTERM`，约 **≤30s** 平台窗口；本实现目标 ~1s 退出主循环 | B |
 | 平台注入 | `STRATEGY_ID`；`EXTERNAL_SERVICE_TOKEN`（及社区别名） | A/B |
 | **STRATEGY_ID 缺失** | **Phase1：进程启动失败（exit ≠0）** | 产品规划 Phase1 + PRD |
 | 资源 | 约 100m–1 CPU / 256–512Mi | B |
 | **发布智能体得分** | **0 分**（发布 ≠ 得分） | B |
 | Live 默认 | **关**；双重闸门 `ALLOW_LIVE_SUBMIT=1` + `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT` | 本仓 |
 | 积分 | 来自**成功适配**；智能体代提成功计同一成功分；公式/周报 **待平台核实** | B / D |
 ## 鉴权双轨（文档索引，Phase1 不代调）
 | 场景 | 头 | 等级 |
 |------|----|------|
 | 浏览器 / `agent_platform` | `Authorization: Bearer <login token>` | A |
 | 容器内社区实践 | `Xc-Token: <token>`；body 带 `strategyId` | C |
 ## 禁止
 - 代注册、代 live 提交、Makefile 含 register/submit/deploy/push
 - 日志/响应/镜像层写入 token 或 STRATEGY_ID **值**
 - 把「智能体已发布」写入千分进度
 ## 冲突裁决（强制）
 | 资源 | 裁决 |
 |------|------|
 | 异步验证队列 | 适配 W1 优先；≥95 全体停交（含智能体 live） |
 | Codex | hold 期扩池/归因 > 智能体脚手架 |
 | 本骨架 | 默认 dry-run；本轮无真实 `task/add` HTTP 客户端 |
 ## 闸门配置以实现为准
 `src/gate.py` 只接受 `ALLOW_LIVE_SUBMIT=1` 与
 `LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT`（去除首尾空白后精确匹配）。
 PRD 的 `ALLOW_LIVE_SUBMIT=true` / `CONFIRM_LIVE=YES` 仅为需求示例，
 本仓不支持这些值/别名，也不读取确认文件。开闸无客户端仍 blocked。
 这只是本地逻辑闸门，不代表已获 live 授权或已校验队列与候选硬门。
--- a/docs/register-runbook.md
+++ b/docs/register-runbook.md
@@ -0,0 +1,39 @@
 # 注册智能体 Runbook（仅人工 · Victor）
 > Agent / Bot **不代填、不代点、不代调写接口**。
 ## 流程
 1. **本地绿**：`make test`（及有 Docker 时 `make build` + `make smoke-health`）
 2. **Gitea 建仓**（Victor）：`dev.modelhub.org.cn/<namespace>/<repo>`  
   - 命名空间/仓名 **未定**时：仅本地验收，不阻塞 Phase1
 3. **推送代码**（Victor 本人）：本 Makefile **无 push 目标**
 4. **打 Git Tag**（可用本仓 `scripts/tag_release.py` **仅本地 tag**；push tag 由 Victor 执行）
 5. **控制台新增**：个人主页 → 我的适配智能体 → 新增  
   - 名称  
   - 仓库地址（远程 Git URL）  
   - **Git Tag**（须先出现在远程；`GET .../tags` 下拉可选后再填）
 6. **Kaniko 构建**：等待状态；证据同时记中文文案与枚举（`building` / `ready` / `failed` / `running` / `stopped`）
 7. **部署后**：确认容器 `GET /health` → 200；积分应 **无变化**（发布 = 0 分）
 ## POST `/agent_platform/strategies`（文档说明 · 禁止代调）
 | 字段 | 说明 |
 |------|------|
 | `name` | 策略显示名 |
 | `repo_url` | Gitea 仓库 URL |
 | `tag` | Git Tag（非分支） |
 成功响应含 `.id`（等级 A：前端 bundle）。  
 **本仓与 Agent 禁止代调该写接口。**
 相关写接口（同样禁止代调）：`/{id}/deploy`、`/{id}/stop`、`DELETE /{id}`。
 ## 环境变量（平台注入 / 控制台）
 | 变量 | 说明 |
 |------|------|
 | `STRATEGY_ID` | 策略 id；本仓启动必需 |
 | `EXTERNAL_SERVICE_TOKEN` | 服务凭证；只检存在，不记值 |
 | `ALLOW_LIVE_SUBMIT` | 默认 `0` |
 | `LIVE_CONFIRM` | live 时须为 `I_CONFIRM_LIVE_SUBMIT` |
--- a/docs/tag-dryrun-notes-draft.md
+++ b/docs/tag-dryrun-notes-draft.md
@@ -0,0 +1,13 @@
 # Tag 文案草稿（仍默认 dry-run）— 未授权勿 push
 - 建议 Tag 名（待 Victor 确认）：`v0.2.0-phase2-dryrun` 或 `v0.2.0-submit-ready-dryrun`
 - 相对 Phase1 `v0.1.0-phase1@0d0f3b6`：pending A 池 + 经验序排序、只读 task/page 探针、task/add 客户端仍双重闸门默认关、configParams sample 仅 dry-run
 - **不含**：live 开闸、真实自动配置、开闸环境变量默认开启
 - 推 Tag 条件：仅当 Victor **明确**说「先上线 dry-run 客户端」；产品当前：**先别 push Tag**
 - 队列满 / configParams 缺口期间：**不开闸**
 推送检查清单（人工）：
 1. `make test` 绿
 2. ALLOW_LIVE_SUBMIT 默认非 1；LIVE_CONFIRM 未设
 3. 控制台重建后日志仍为 dry-run / Phase2 格式
 4. 不代注册、不代 live
--- a/main.py
+++ b/main.py
@@ -1,70 +0,0 @@
 import json
 import os
 import signal
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 HOST = "0.0.0.0"
 PORT = 8080
 shutdown_requested = False
 def _config() -> dict:
    token = os.getenv("EXTERNAL_SERVICE_TOKEN", "")
    return {
        "external_service_token_present": bool(token),
    }
 class Handler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path == "/health":
            self._send_json({"status": "ok"})
            return
        if self.path == "/":
            self._send_json(
                {
                    "name": "strategy-demo",
                    "status": "running",
                    "config": _config(),
                }
            )
            return
        self._send_json({"error": "not found"}, status=404)
    def log_message(self, fmt: str, *args: object) -> None:
        print(f"{self.address_string()} - {fmt % args}", flush=True)
    def _send_json(self, body: dict, status: int = 200) -> None:
        payload = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)
 def _handle_signal(signum: int, _frame: object) -> None:
    global shutdown_requested
    shutdown_requested = True
    print(f"received signal {signum}, shutting down", flush=True)
 def main() -> None:
    signal.signal(signal.SIGTERM, _handle_signal)
    signal.signal(signal.SIGINT, _handle_signal)
    server = ThreadingHTTPServer((HOST, PORT), Handler)
    server.timeout = 1
    print(f"strategy-demo listening on {HOST}:{PORT}", flush=True)
    while not shutdown_requested:
        server.handle_request()
    server.server_close()
 if __name__ == "__main__":
    main()
--- a/reports/.gitkeep
+++ b/reports/.gitkeep
--- a/requirements.txt
+++ b/requirements.txt
@@ -1 +0,0 @@
--- a/scripts/build_config_dry.py
+++ b/scripts/build_config_dry.py
@@ -0,0 +1,26 @@
 #!/usr/bin/env python3
 """Generate config only; never creates a task or prints returned YAML/secrets."""
 import argparse
 import sys
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from src.build_config_client import build_config, BuildConfigError
 def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--gpu-type', default='Sunrise_pt-200-x1')
    parser.add_argument('--framework', default='vllm')
    parser.add_argument('--task-type', default='text-generation')
    args = parser.parse_args()
    try:
        yaml = build_config(args.gpu_type, args.framework, args.task_type)
    except BuildConfigError as error:
        print(str(error), file=sys.stderr)
        return 1
    print(f'configParams_source=build-config; generated YAML characters={len(yaml)}; no task submitted')
    return 0
 if __name__ == '__main__':
    sys.exit(main())
--- a/scripts/dry_run.py
+++ b/scripts/dry_run.py
@@ -0,0 +1,85 @@
 #!/usr/bin/env python3
 """Read-only Phase2 pending draft report. Offline by default; opt-in configuration generation only."""
 import json
 import os
 import sys
 from datetime import datetime
 from pathlib import Path
 from zoneinfo import ZoneInfo
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT))
 from src.pending_pool import build_ordered_candidates, input_path, experience_order
 from src.config_params import build_config_params, config_params_source, SOURCE, GAP
 from src.queue_depth import queue_depth
 from src.task_add_client import build_payload
 TZ = ZoneInfo('Asia/Shanghai')
 def _load_candidates(path):
    data = json.loads(path.read_text())
    return data if isinstance(data, list) else list(data['candidates'])
 def _render(candidates, ts, src):
    candidates = [c for c in candidates if not os.environ.get('PENDING_SECTION') or
                  c.get('section', 'B') == os.environ['PENDING_SECTION'].strip().upper()]
    depth = queue_depth()
    lines = [f'# 代提候选 / 需 Victor 确认 — Dry-run — {ts} (Asia/Shanghai)', '',
             '> 模式：dry-run；零网络写；需 Victor 人工提交/确认',
             '> 积分：待平台核实', '> perf_order=unset；order_source=experience（经验序 / 待平台核实）',
             f'> configParams source: {SOURCE}', f'> configParams_source={config_params_source()}', f'> {GAP}', '',
             '## A→B 顺序草稿', '', f'输入：{src}；候选条数：{len(candidates)}',
             f'队列深度：{depth if depth is not None else "未知，需人工确认"}',
             '队列≥95：禁 live；适配 W1 优先，不抢空位。', '']
    if depth is not None and depth >= 95:
        lines.extend(['> **禁 live / 队列≥95**', ''])
    for i, c in enumerate(candidates, 1):
        model = c.get('model_address') or c.get('modelAddress') or c.get('model_id') or '<MODEL>'
        gpus = experience_order(c.get('try_next_gpus', []))
        target = gpus[0] if gpus else c.get('targetGpu', '<TARGET_GPU:需人工确认>')
        status = c.get('status', 'needs_human')
        lines.extend([f'### {i}. Section {c.get("section", "B")} — `{model}`', '',
                      f'状态：{status}；needs_human={c.get("needs_human", True)}',
                      f'过滤：{c.get("skip_reason") or "名称硬门未命中；元数据/唯一性待人工核实"}', ''])
        lines.extend([f'未适配卡（仅 pending 缺口）：{", ".join(gpus) or "未提供，需确认"}',
                      'perf_order=unset；order_source=experience；天垓100 一般不扇出', ''])
        draft = build_payload(model, target, 'vllm', strategy_id='<STRATEGY_ID>',
                              config_params=build_config_params(target))
        # Display only; this truncated draft must never be submitted.
        draft['configParams'] = draft['configParams'][:400] + '\n… [truncated; dry-run only]'
        lines.extend(['build-config query: gpuType=<targetGpu>&framework=vllm&taskType=text-generation; POST empty body', 'POST 字段草稿（skipped 行禁止提交）：', '```json', json.dumps(draft, ensure_ascii=False, indent=2), '```',
                      '需 Victor 人工提交/确认；积分：待平台核实', ''])
    report = '\n'.join(lines)
    # Defensive redaction even if an input happens to contain an injected credential.
    secrets = [os.environ.get(k, '') for k in ('XC_TOKEN', 'MODELHUB_XC_TOKEN', 'EXTERNAL_SERVICE_TOKEN', 'Xc-Token')]
    if os.environ.get('XC_TOKEN_FILE'):
        try:
            secrets.append(Path(os.environ['XC_TOKEN_FILE']).read_text().strip())
        except OSError:
            pass
    for secret in secrets:
        if secret:
            report = report.replace(secret, '[REDACTED]')
    return report
 def main():
    md = input_path('PENDING_SUBMIT_MD', 'pending-submit.md')
    txt = input_path('PENDING_SUBMIT_TXT', 'pending-submit.txt')
    if md.is_file() or txt.is_file():
        src = md if md.is_file() else txt
        candidates = build_ordered_candidates(md, txt)
    else:
        src = Path(os.environ.get('DRY_RUN_CANDIDATES', ROOT / 'config/candidates.example.json'))
        candidates = _load_candidates(src)
    now = datetime.now(TZ)
    out = ROOT / 'reports' / f'dry-run-{now:%Y%m%d-%H%M%S}.md'
    out.parent.mkdir(exist_ok=True)
    out.write_text(_render(candidates, now.strftime('%Y-%m-%d %H:%M:%S'), src), encoding='utf-8')
    print(f'Wrote {out}')
 if __name__ == '__main__':
    main()
--- a/scripts/probe_readonly.py
+++ b/scripts/probe_readonly.py
@@ -0,0 +1,17 @@
 #!/usr/bin/env python3
 """Run only the task/page GET; print credential-free summary."""
 import json
 import sys
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from src.readonly_probe import probe_readonly
 def main():
    result = probe_readonly()
    print(json.dumps(result, ensure_ascii=False))
    return 0 if result['auth_ok'] else 1
 if __name__ == '__main__':
    sys.exit(main())
--- a/scripts/smoke_health.py
+++ b/scripts/smoke_health.py
@@ -0,0 +1,100 @@
 #!/usr/bin/env python3
 """Smoke: GET /health must return 200 {"status":"ok"}.
 Usage:
  # Against already-running server:
  SMOKE_BASE_URL=http://127.0.0.1:8080 python3 scripts/smoke_health.py
  # Or let this script start a local server (needs STRATEGY_ID):
  STRATEGY_ID=test python3 scripts/smoke_health.py
 """
 from __future__ import annotations
 import json
 import os
 import signal
 import subprocess
 import sys
 import time
 import urllib.error
 import urllib.request
 from pathlib import Path
 ROOT = Path(__file__).resolve().parents[1]
 def _get(url: str, timeout: float = 2.0) -> tuple[int, dict]:
    req = urllib.request.Request(url, method="GET")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        body = resp.read().decode("utf-8")
        return resp.status, json.loads(body)
 def _wait_health(base: str, attempts: int = 30) -> None:
    url = base.rstrip("/") + "/health"
    last_err: Exception | None = None
    for _ in range(attempts):
        try:
            code, data = _get(url)
            if code == 200 and data.get("status") == "ok":
                print(f"OK: GET {url} -> {code} {data}")
                return
            raise AssertionError(f"unexpected response: {code} {data}")
        except (urllib.error.URLError, ConnectionError, AssertionError, TimeoutError) as exc:
            last_err = exc
            time.sleep(0.2)
    raise SystemExit(f"smoke-health failed: {last_err}")
 def main() -> None:
    external = os.environ.get("SMOKE_BASE_URL", "").strip()
    if external:
        _wait_health(external)
        # optional status check
        try:
            code, data = _get(external.rstrip("/") + "/status")
            print(f"OK: GET /status -> {code} {data}")
        except Exception as exc:  # noqa: BLE001
            print(f"WARN: /status check skipped: {exc}", file=sys.stderr)
        return
    if not os.environ.get("STRATEGY_ID", "").strip():
        os.environ["STRATEGY_ID"] = "smoke-test-strategy"
    port = os.environ.get("SMOKE_PORT", "18080")
    env = os.environ.copy()
    env["PORT"] = port
    env["HOST"] = "127.0.0.1"
    env["STRATEGY_ID"] = env.get("STRATEGY_ID") or "smoke-test-strategy"
    proc = subprocess.Popen(
        [sys.executable, "-m", "src.server"],
        cwd=str(ROOT),
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
    )
    try:
        _wait_health(f"http://127.0.0.1:{port}")
        code, data = _get(f"http://127.0.0.1:{port}/status")
        print(f"OK: GET /status -> {code} {data}")
        assert "strategy_id_present" in data
        assert data["strategy_id_present"] is True
        # never assert on secret values — only booleans
    finally:
        proc.send_signal(signal.SIGTERM)
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait(timeout=2)
        if proc.returncode not in (0, -signal.SIGTERM, signal.SIGTERM):
            # 0 or terminated by SIGTERM is fine
            err = (proc.stderr.read() or b"").decode("utf-8", errors="replace")
            if proc.returncode is not None and proc.returncode > 0:
                print(f"WARN: server exit code {proc.returncode}: {err}", file=sys.stderr)
 if __name__ == "__main__":
    main()
--- a/scripts/tag_release.py
+++ b/scripts/tag_release.py
@@ -0,0 +1,43 @@
 #!/usr/bin/env python3
 """Create a local git tag only. Does NOT push. Does NOT register on ModelHub.
 Usage:
  python3 scripts/tag_release.py v0.1.0
  python3 scripts/tag_release.py v0.1.0 --message "Phase1 skeleton"
 """
 from __future__ import annotations
 import argparse
 import subprocess
 import sys
 from pathlib import Path
 ROOT = Path(__file__).resolve().parents[1]
 def main() -> None:
    parser = argparse.ArgumentParser(description="Local git tag only (no push, no register)")
    parser.add_argument("tag", help="Tag name, e.g. v0.1.0")
    parser.add_argument("--message", "-m", default="", help="Annotated tag message")
    args = parser.parse_args()
    if not (ROOT / ".git").exists():
        print(
            "WARN: not a git repository yet. Init locally if needed; "
            "this script will not push or register.",
            file=sys.stderr,
        )
        # Still attempt so the failure is clear
    msg = args.message or f"Release {args.tag} (local tag only — no push / no ModelHub register)"
    cmd = ["git", "tag", "-a", args.tag, "-m", msg]
    print(f"Running (cwd={ROOT}): {' '.join(cmd)}")
    print("NOTE: This does NOT git push and does NOT call ModelHub APIs.")
    result = subprocess.run(cmd, cwd=str(ROOT))
    if result.returncode != 0:
        raise SystemExit(result.returncode)
    print(f"OK: local tag {args.tag!r} created. Push tag yourself when ready for Kaniko.")
 if __name__ == "__main__":
    main()
--- a/src/init.py
+++ b/src/init.py
@@ -0,0 +1,3 @@
 """ModelHub XC adaptation agent — Phase1 local skeleton."""
 __version__ = "0.1.0"
--- a/src/build_config_client.py
+++ b/src/build_config_client.py
@@ -0,0 +1,60 @@
 """Explicit configuration generation; import performs no I/O.
 Returns YAML on success, otherwise raises BuildConfigError with safe attribution
 and HTTP/platform status. Server messages and transport details are never exposed.
 """
 import json
 import os
 import urllib.error
 import urllib.parse
 import urllib.request
 from src.task_add_client import resolve_token
 class BuildConfigError(RuntimeError):
    def __init__(self, reason, http_status=None, platform_code=None):
        self.error_attribution = reason
        self.http_status = http_status
        self.platform_code = platform_code
        super().__init__('build-config: ' + reason)
 def build_config(gpu_type, framework='vllm', task_type='text-generation'):
    try:
        token = resolve_token()
    except (ValueError, OSError):
        raise BuildConfigError('missing/invalid credential') from None
    if not isinstance(gpu_type, str) or not gpu_type.strip() or '<' in gpu_type:
        raise BuildConfigError('gpu_type required (platform enum)')
    query = urllib.parse.urlencode(dict(gpuType=gpu_type, framework=framework, taskType=task_type))
    header = 'Authorization' if os.environ.get('BUILD_CONFIG_USE_AUTHORIZATION') == '1' else 'Xc-Token'
    try:
        request = urllib.request.Request(
            os.environ.get('MODELHUB_API_BASE', 'https://www.modelhub.org.cn').rstrip('/')
            + '/api/adapt/task/build-config?' + query,
            headers={header: token, 'Accept': 'application/json'}, method='POST')
        with urllib.request.urlopen(request, timeout=30) as response:
            status, raw = response.status, response.read()
    except urllib.error.HTTPError as error:
        status, raw = error.code, error.read()
    except (OSError, urllib.error.URLError, ValueError):
        raise BuildConfigError('transport/timeout') from None
    try:
        body = json.loads(raw)
    except (ValueError, UnicodeError):
        raise BuildConfigError('invalid JSON response', status) from None
    if not isinstance(body, dict):
        raise BuildConfigError('invalid response object', status)
    code = body.get('code')
    # Only numeric codes are retained; arbitrary server strings may echo secrets.
    code = code if type(code) is int else None
    if code in (40100, 40101):
        raise BuildConfigError({40100: '40100 NOT_LOGIN_ERROR', 40101: '40101 NO_AUTH_ERROR'}[code], status, code)
    if not 200 <= status < 300 or code != 0:
        raise BuildConfigError('other HTTP/platform code', status, code)
    data = body.get('data')
    if not isinstance(data, str) or not data.strip():
        raise BuildConfigError('empty/non-str data', status, code)
    if token in data:
        raise BuildConfigError('credential echoed in data', status, code)
    return data
--- a/src/config_params.py
+++ b/src/config_params.py
@@ -0,0 +1,31 @@
 """Offline sample by default; explicit opt-in enables captured build-config API."""
 import os
 from pathlib import Path
 from src.build_config_client import build_config, BuildConfigError
 from src.task_add_client import resolve_token
 SOURCE = 'POST /api/adapt/task/build-config (discovered via capture); default templates/config_params.sample.yaml'
 GAP = 'GAP: build-config API discovered via capture (autoconfig-capture-2026-10-16.md); live/实锤 needs token + API call and auth proof. Sample is dry-run draft only, NOT production-ready.'
 def config_params_source():
    source = os.environ.get('CONFIG_PARAMS_SOURCE', '').strip()
    if source not in ('', 'sample', 'build-config'):
        raise BuildConfigError('unsupported CONFIG_PARAMS_SOURCE')
    if source == 'build-config':
        return 'build-config'
    if os.environ.get('ALLOW_BUILD_CONFIG') == '1':
        try:
            resolve_token()
        except (ValueError, OSError):
            pass
        else:
            return 'build-config'
    return 'sample'
 def build_config_params(gpu_type=None, framework='vllm', task_type='text-generation'):
    """Return YAML; opted-in API failures propagate, never fall back to sample."""
    if config_params_source() == 'build-config':
        return build_config(gpu_type, framework, task_type)
    return (Path(__file__).resolve().parents[1] / 'templates/config_params.sample.yaml').read_text()
--- a/src/gate.py
+++ b/src/gate.py
@@ -0,0 +1,123 @@
 """Dual live-submit gate. Default: dry-run / blocked.
 Live is allowed only when ALL of:
  - ALLOW_LIVE_SUBMIT=1
  - LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT
  - STRATEGY_ID present
 Phase2 client exists but requires explicit injection. No automatic wiring.
 Queue ≥95 blocks; at most one injected submit per process.
 """
 from __future__ import annotations
 import os
 import threading
 from .queue_depth import queue_depth
 from dataclasses import dataclass
 from typing import Any, Callable, Dict, Optional, Protocol
 from .strategy import strategy_id_present
 _submit_count = 0
 _submit_lock = threading.Lock()
 LIVE_CONFIRM_VALUE = "I_CONFIRM_LIVE_SUBMIT"
 class SubmitClient(Protocol):
    """Minimal interface for a submit client (injected fake in tests)."""
    def submit(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        ...
@dataclass(frozen=True)
 class GateDecision:
    mode: str  # "dry-run" | "live" | "blocked"
    allowed: bool
    reason: str
 def _allow_live_flag() -> bool:
    return os.environ.get("ALLOW_LIVE_SUBMIT", "0").strip() == "1"
 def _live_confirm_ok() -> bool:
    return os.environ.get("LIVE_CONFIRM", "").strip() == LIVE_CONFIRM_VALUE
 def evaluate_gate() -> GateDecision:
    """Evaluate whether live submit is permitted."""
    if not strategy_id_present():
        return GateDecision(
            mode="blocked",
            allowed=False,
            reason="STRATEGY_ID missing or empty",
        )
    if not _allow_live_flag():
        return GateDecision(
            mode="dry-run",
            allowed=False,
            reason="ALLOW_LIVE_SUBMIT is not 1 (default dry-run)",
        )
    if not _live_confirm_ok():
        return GateDecision(
            mode="blocked",
            allowed=False,
            reason=f"LIVE_CONFIRM must be exactly {LIVE_CONFIRM_VALUE!r}",
        )
    depth = queue_depth()
    if depth is None and ("ASYNC_QUEUE_DEPTH" in os.environ or "QUEUE_SNAPSHOT_JSON" in os.environ):
        return GateDecision("blocked", False, "queue depth invalid/unknown; human confirmation required")
    if depth is not None and depth >= 95:
        return GateDecision("blocked", False, "queue ≥95: 禁 live")
    return GateDecision(
        mode="live",
        allowed=True,
        reason="dual gate open (ALLOW_LIVE_SUBMIT=1 and LIVE_CONFIRM set)",
    )
 def attempt_live_submit(
    payload: Dict[str, Any],
    client: Optional[SubmitClient] = None,
 ) -> Dict[str, Any]:
    """Attempt a live submit only if the gate opens and a client is injected.
    Phase2 TaskAddClient must be explicitly injected and dual-gated.
    Tests must use fake clients or mock HTTP only.
    """
    global _submit_count
    decision = evaluate_gate()
    if not decision.allowed:
        return {
            "ok": False,
            "mode": decision.mode,
            "reason": decision.reason,
            "submitted": False,
        }
    if client is None:
        return {
            "ok": False,
            "mode": "blocked",
            "reason": (
                "Phase2 task/add client exists; explicit client injection required; "
                "no real request without injection and dual gate"
            ),
            "submitted": False,
        }
    with _submit_lock:
        if _submit_count >= 1:
            return dict(ok=False, mode="blocked", submitted=False,
                        reason="process limit: at most 1 real add")
        # Reserve before invocation: a transport failure may still have created a task.
        _submit_count += 1
    result = client.submit(payload)
    return {
        "ok": True,
        "mode": "live",
        "reason": decision.reason,
        "submitted": True,
        "result": result,
    }
--- a/src/health.py
+++ b/src/health.py
@@ -0,0 +1,33 @@
 """Health and status HTTP handlers. Never echo secrets."""
 from __future__ import annotations
 import json
 from typing import Any, Dict, Tuple
 from .strategy import external_service_token_present, strategy_id_present
 def health_payload() -> Dict[str, str]:
    return {"status": "ok"}
 def status_payload() -> Dict[str, Any]:
    """Booleans only — never include STRATEGY_ID or token values."""
    return {
        "strategy_id_present": strategy_id_present(),
        "external_service_token_present": external_service_token_present(),
    }
 def handle_get(path: str) -> Tuple[int, Dict[str, str], bytes]:
    """Return (status_code, headers, body) for GET paths."""
    headers = {"Content-Type": "application/json; charset=utf-8"}
    if path == "/health" or path.startswith("/health?"):
        body = json.dumps(health_payload(), ensure_ascii=False).encode("utf-8")
        return 200, headers, body
    if path == "/status" or path.startswith("/status?"):
        body = json.dumps(status_payload(), ensure_ascii=False).encode("utf-8")
        return 200, headers, body
    body = json.dumps({"error": "not_found"}, ensure_ascii=False).encode("utf-8")
    return 404, headers, body
--- a/src/pending_pool.py
+++ b/src/pending_pool.py
@@ -0,0 +1,106 @@
 """Read-only pending pool. Name checks cannot establish metadata/uniqueness PASS."""
 import os
 import re
 from pathlib import Path
 BASE = Path('/workspace/modelhub-adapt')
 GPUS = r'曦望(?: S2)?|壁砺(?:166M)?|天垓150|天垓100|智铠(?: mrv-100)?|Iluvatar_mrv-100|Biren_166m|Sunrise_pt-200-x1'
 FAMILIES = ('lfm2', 'olmo-2', 'olmo2', 'spark-x2.5', 'spark2_5', 'nemotron-h')
 def input_path(env, filename):
    return Path(os.environ.get(env) or BASE / filename)
 def load_do_not_resubmit(path):
    # Exact ids only; includes collision bullets and overnight tables.
    return {s for s in re.findall(r'`([^`]+)`', Path(path).read_text())
            if re.fullmatch(r'[\w.-]+/[\w.\-/]+', s)}
 def load_banned_name_tokens(path):
    text = Path(path).read_text().lower()
    return {s for s in ('abliterated', 'heretic', 'uncensored', 'unsensored') if s in text}
 def experience_order(gpus):
    """Stable tier sort of missing chips only; tier-one pending order wins.
    docs/gpu-perf-order.md Section B is experience, never official performance.
    Unknown names retain their relative order after known tiers; no chips added.
    """
    def tier(gpu):
        if any(name in gpu for name in ('壁砺', '曦望', 'Biren_166m', 'Sunrise_pt-200-x1')):
            return 1
        if '天垓150' in gpu:
            return 2
        if '天垓100' in gpu:
            return 3
        if '智铠' in gpu or 'mrv-100' in gpu:
            return 4
        return 5
    return sorted(dict.fromkeys(gpus), key=tier)
 def _candidate(model, section, gpu_text, notes):
    return dict(model_address=model, section=section,
                try_next_gpus=experience_order(re.findall(GPUS, gpu_text)),
                perf_order="unset", order_source="experience",
                notes=notes, comment=notes, skip_reason=None,
                needs_human=False, status='candidate')
 def parse_pending_md(path):
    rows, section = [], None
    for line in Path(path).read_text().splitlines():
        heading = re.match(r'##\s+Section\s+([AB])\b', line)
        if heading:
            section = heading[1]
        elif line.startswith('## '):
            section = None
        if section and re.match(r'\|\s*\d+\s*\|', line):
            cols = [s.strip() for s in line.strip('|').split('|')]
            model = re.search(r'`([^`]+)`', cols[1])
            if model:
                rows.append(_candidate(model[1], section, cols[3] if section == 'A' else cols[-1], cols[-1]))
    return rows
 def parse_pending_txt(path):
    rows = []
    for line in Path(path).read_text().splitlines():
        if not line.strip() or line.startswith('#'):
            continue
        parts = re.split(r'\t|\\t', line, maxsplit=1)
        comment = parts[1] if len(parts) == 2 else ''
        match = re.match(r'#([AB])', comment)
        section = match[1] if match else 'B'
        gpu_text = re.split(r'(?:still\s+)?try\s*:', comment, maxsplit=1)[-1] if 'try:' in comment else (comment if section == 'B' else '')
        row = _candidate(parts[0].strip(), section, gpu_text, comment)
        if not match:
            row.update(needs_human=True, status='needs_human')
        rows.append(row)
    return rows
 def build_ordered_candidates(md_path=None, txt_path=None, dnr_path=None, banned_path=None):
    md = Path(md_path) if md_path is not None else input_path('PENDING_SUBMIT_MD', 'pending-submit.md')
    txt = Path(txt_path) if txt_path is not None else input_path('PENDING_SUBMIT_TXT', 'pending-submit.txt')
    rows = parse_pending_md(md) if md.is_file() else parse_pending_txt(txt) if txt.is_file() else []
    dnr = Path(dnr_path) if dnr_path is not None else input_path('DO_NOT_RESUBMIT_MD', 'do-not-resubmit.md')
    banned = Path(banned_path) if banned_path is not None else input_path('BANNED_PATTERNS_MD', 'banned-patterns.md')
    excluded = load_do_not_resubmit(dnr) if dnr.is_file() else set()
    tokens = load_banned_name_tokens(banned) if banned.is_file() else set()
    section = os.environ.get('PENDING_SECTION', '').strip().upper()
    if section:
        rows = [row for row in rows if row['section'] == section]
    for row in rows:
        name = row['model_address']
        reason = ('do-not-resubmit' if name in excluded else
                  'hard-gate: banned name token' if any(t in name.lower() for t in tokens) else
                  'hard-gate: banned family' if any(t in name.lower() for t in FAMILIES) else None)
        row['skip_reason'] = reason
        # Metadata gates (weights, context, riskTag, uniqueness) remain unverified.
        row['needs_human'] = not bool(reason)
        row['status'] = 'skipped' if reason else 'needs_human'
    return sorted(rows, key=lambda r: r['section'] != 'A')
--- a/src/queue_depth.py
+++ b/src/queue_depth.py
@@ -0,0 +1,22 @@
 """Read queue hints without contacting the platform; unknown is conservative."""
 import json
 import os
 from pathlib import Path
 def queue_depth():
    value = os.environ.get('ASYNC_QUEUE_DEPTH')
    if value is None and os.environ.get('QUEUE_SNAPSHOT_JSON'):
        try:
            data = json.loads(Path(os.environ['QUEUE_SNAPSHOT_JSON']).read_text())
            if isinstance(data, dict):
                value = next((data[k] for k in ('depth', 'queue_depth', 'async_queue_depth', 'ASYNC_QUEUE_DEPTH') if k in data), None)
        except (OSError, ValueError):
            return None
    try:
        if isinstance(value, bool):
            return None
        depth = int(str(value))
        return depth if depth >= 0 else None
    except (TypeError, ValueError):
        return None
--- a/src/readonly_probe.py
+++ b/src/readonly_probe.py
@@ -0,0 +1,44 @@
 """Explicit read-only auth probe. Import performs no I/O; never calls task/add."""
 import json
 import os
 import urllib.error
 import urllib.request
 from src.task_add_client import resolve_token
 def probe_readonly():
    try:
        token = resolve_token()
    except (ValueError, OSError):
        return dict(http_status=None, platform_code=None, auth_ok=False,
                    record_summary={'count': 0}, error_attribution='missing/invalid credential')
    request = urllib.request.Request(
        os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/')
        + '/api/adapt/task/page?onlyMine=true&pageSize=5&current=1',
        headers={'Xc-Token': token, 'Accept': 'application/json'}, method='GET')
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            status, raw = response.status, response.read()
    except urllib.error.HTTPError as error:
        status, raw = error.code, error.read()
    except (OSError, urllib.error.URLError, ValueError):
        return dict(http_status=None, platform_code=None, auth_ok=False,
                    record_summary={'count': 0}, error_attribution='transport/timeout')
    # Never expose response strings: servers can echo credentials in any field.
    try:
        body = json.loads(raw)
    except (ValueError, UnicodeError):
        body = {}
    body = body if isinstance(body, dict) else {}
    code = body.get('code')
    code = code if type(code) is int else int(code) if isinstance(code, str) and code in ('0', '40100', '40101') else None
    ok = 200 <= status < 300 and code == 0
    data = body.get('data')
    records = data.get('records') if isinstance(data, dict) else None
    summary = {'count': len(records) if isinstance(records, list) else 0}
    if isinstance(data, dict) and type(data.get('total')) is int:
        summary['total'] = data['total']
    attribution = {40100: '40100 NOT_LOGIN_ERROR', 40101: '40101 NO_AUTH_ERROR'}.get(
        code, 'success' if ok else 'other HTTP/platform/invalid response')
    return dict(http_status=status, platform_code=code, auth_ok=ok,
                record_summary=summary, error_attribution=attribution)
--- a/src/server.py
+++ b/src/server.py
@@ -0,0 +1,83 @@
 """Minimal HTTP server for the adaptation agent (stdlib only).
 Listens on PORT (default 8080). Handles SIGTERM and exits the main loop
 within ~1s (platform window documented as ≤30s).
 Startup: require STRATEGY_ID (Phase1 product plan — fail if missing).
 """
 from __future__ import annotations
 import os
 import signal
 import sys
 import threading
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Optional
 from .health import handle_get
 from .strategy import require_strategy_id
 _shutdown = threading.Event()
 _httpd: Optional[ThreadingHTTPServer] = None
 class AgentHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    def log_message(self, fmt: str, *args) -> None:  # noqa: A003
        # Avoid noisy default logging; never log env secrets.
        sys.stderr.write("[http] request handled (details omitted)\n")
    def do_GET(self) -> None:  # noqa: N802
        code, headers, body = handle_get(self.path)
        self.send_response(code)
        for k, v in headers.items():
            self.send_header(k, v)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def do_POST(self) -> None:  # noqa: N802
        body = b'{"error":"method_not_allowed"}'
        self.send_response(405)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
 def _on_sigterm(signum: int, frame) -> None:  # noqa: ARG001
    sys.stderr.write(f"Received signal {signum}; shutting down...\n")
    _shutdown.set()
    httpd = _httpd
    if httpd is not None:
        # wake serve_forever
        threading.Thread(target=httpd.shutdown, daemon=True).start()
 def main() -> None:
    global _httpd
    require_strategy_id()  # exit 1 if missing
    port = int(os.environ.get("PORT", "8080"))
    host = os.environ.get("HOST", "0.0.0.0")
    signal.signal(signal.SIGTERM, _on_sigterm)
    signal.signal(signal.SIGINT, _on_sigterm)
    httpd = ThreadingHTTPServer((host, port), AgentHandler)
    httpd.daemon_threads = True
    _httpd = httpd
    sys.stderr.write(f"Listening on {host}:{port} (GET /health, GET /status)\n")
    try:
        httpd.serve_forever(poll_interval=0.5)
    finally:
        httpd.server_close()
        sys.stderr.write("Server stopped.\n")
 if __name__ == "__main__":
    main()
--- a/src/strategy.py
+++ b/src/strategy.py
@@ -0,0 +1,81 @@
 """STRATEGY_ID and external-token presence helpers.
 Phase1 (product plan): missing STRATEGY_ID → process must fail to start
 (exit non-zero). Values are never logged or returned in responses.
 """
 from __future__ import annotations
 import os
 import sys
 from typing import Optional
 # Community alias order (level C); we only check presence, never echo values.
 _TOKEN_ENV_KEYS = (
    "XC_TOKEN_FILE",
    "XC_TOKEN",
    "MODELHUB_XC_TOKEN",
    "EXTERNAL_SERVICE_TOKEN",
 )
 def get_strategy_id() -> Optional[str]:
    """Return STRATEGY_ID if set and non-empty after strip; else None."""
    raw = os.environ.get("STRATEGY_ID")
    if raw is None:
        return None
    value = raw.strip()
    return value if value else None
 def require_strategy_id() -> str:
    """Return STRATEGY_ID or exit the process with a non-zero status.
    Does not print the value. Safe for startup path.
    """
    sid = get_strategy_id()
    if sid is None:
        print(
            "ERROR: STRATEGY_ID is required and must be non-empty. "
            "Refusing to start (Phase1 contract).",
            file=sys.stderr,
        )
        sys.exit(1)
    return sid
 def strategy_id_present() -> bool:
    return get_strategy_id() is not None
 def _token_from_file(path: str) -> bool:
    """True if path exists and has non-empty content (value never read into logs)."""
    try:
        if not path or not os.path.isfile(path):
            return False
        with open(path, "r", encoding="utf-8") as fh:
            content = fh.read().strip()
        return bool(content)
    except OSError:
        return False
 def external_service_token_present() -> bool:
    """Whether any known token env / file alias is present (boolean only)."""
    for key in _TOKEN_ENV_KEYS:
        raw = os.environ.get(key)
        if raw is None:
            continue
        value = raw.strip()
        if not value:
            continue
        if key == "XC_TOKEN_FILE":
            if _token_from_file(value):
                return True
            continue
        # Reject obvious placeholders (community practice, level C)
        lowered = value.lower()
        if lowered in ("tmp", "placeholder", "changeme", "xxx"):
            continue
        return True
    return False
--- a/src/task_add_client.py
+++ b/src/task_add_client.py
@@ -0,0 +1,58 @@
 """Explicit task/add client; import never sends requests. No retries.
 Community Xc-Token authentication is C; platform code 60014 is D (unverified).
 GPU enums/framework/configParams require human verification. Draft placeholders
 must be replaced before a real call. Use through the dual gate with injection.
 """
 import json
 import os
 import re
 import urllib.request
 import urllib.error
 from pathlib import Path
 def build_payload(model_address, target_gpu='<TARGET_GPU>', framework='<FRAMEWORK>',
                  strategy_id=None, config_params='<CONFIG_PARAMS>'):
    return dict(modelAddress=model_address, taskType='text-generation', targetGpu=target_gpu,
                framework=framework, strategyId=strategy_id or os.environ.get('STRATEGY_ID') or '<STRATEGY_ID>',
                configParams=config_params)
 def resolve_token():
    file = os.environ.get('XC_TOKEN_FILE')
    token = Path(file).read_text().strip() if file else next(
        (os.environ[k].strip() for k in ('XC_TOKEN', 'MODELHUB_XC_TOKEN', 'EXTERNAL_SERVICE_TOKEN') if os.environ.get(k)), '')
    if not token or re.search(r'(?i)(?:^|[^a-z0-9])(placeholder|changeme|tmp)(?:$|[^a-z0-9])', token):
        raise ValueError('Missing or invalid service credential')
    return token
 class TaskAddClient:
    def submit(self, payload):
        if not os.environ.get('STRATEGY_ID', '').strip():
            raise ValueError('STRATEGY_ID required')
        required = ('modelAddress', 'taskType', 'targetGpu', 'framework', 'strategyId', 'configParams')
        if any(k not in payload or not payload[k] or '<' in str(payload[k]) for k in required):
            raise ValueError('Replace all required payload placeholders')
        if payload['strategyId'] != os.environ['STRATEGY_ID'].strip():
            raise ValueError('strategyId must match STRATEGY_ID')
        token = resolve_token()
        request = urllib.request.Request(os.environ.get('MODELHUB_API_BASE', 'https://www.modelhub.org.cn').rstrip('/') + '/api/adapt/task/add',
            data=json.dumps(payload).encode(), headers={'Xc-Token': token, 'Content-Type': 'application/json'}, method='POST')
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                status, raw = response.status, response.read()
        except urllib.error.HTTPError as error:
            status, raw = error.code, error.read()
        except (OSError, urllib.error.URLError):
            return dict(http_status=None, body=None, platform_code=None, error_attribution='transport/timeout')
        # A server may echo credentials; redact before parsing or returning anything.
        text = raw.decode('utf-8', errors='replace').replace(token, '[REDACTED]')
        try:
            body = json.loads(text)
        except ValueError:
            body = text
        code = body.get('code') if isinstance(body, dict) else None
        attribution = '60014: D (unverified)' if str(code) == '60014' else 'HTTP error' if status >= 400 else 'platform response (unverified)'
        return dict(http_status=status, body=body, platform_code=code, error_attribution=attribution)
--- a/templates/config_params.sample.yaml
+++ b/templates/config_params.sample.yaml
@@ -0,0 +1,44 @@
 framework: vllm
 api: completion
 lang: en
 max_model_len: 4096
 max_tokens: 1024
 temperature: 0.7
 repetition_penalty: 1.1
 top_p: 0.9
 sut_config:
  gpu_num: 1
  values:
    command:
    - /opt/conda/bin/vllm
    - serve
    - /model
    - --port
    - '20644'
    - --served-model-name
    - llm
    - --max-model-len
    - '4096'
    - --gpu-memory-utilization
    - '0.9'
    - -tp
    - '1'
    - --enforce-eager
    - --trust-remote-code
 ref_config:
  gpu_num: 1
  values:
    command:
    - vllm
    - serve
    - /model
    - --port
    - '80'
    - --served-model-name
    - llm
    - --max-model-len
    - '4096'
    - -tp
    - '1'
    - --enforce-eager
    - --trust-remote-code
--- a/tests/init.py
+++ b/tests/init.py
--- a/tests/test_build_config_client.py
+++ b/tests/test_build_config_client.py
@@ -0,0 +1,73 @@
 import runpy
 import io
 import json
 import os
 import unittest
 import urllib.error
 from unittest.mock import MagicMock, patch
 from urllib.parse import urlparse, parse_qs
 from src.build_config_client import build_config, BuildConfigError
 from scripts.build_config_dry import main
 class TestBuildConfigClient(unittest.TestCase):
    def response(self, body, status=200):
        response = MagicMock()
        response.__enter__.return_value = response
        response.status = status
        response.read.return_value = json.dumps(body).encode()
        return response
    def test_success_request_and_auth(self):
        for authorization in ('0', '1'):
            with patch.dict(os.environ, {'XC_TOKEN': 'test-secret', 'BUILD_CONFIG_USE_AUTHORIZATION': authorization}, clear=True), patch('urllib.request.urlopen', return_value=self.response({'code': 0, 'data': 'framework: vllm\n'})) as call:
                self.assertEqual(build_config('Sunrise_pt-200-x1'), 'framework: vllm\n')
            request = call.call_args.args[0]
            self.assertEqual(request.get_method(), 'POST')
            self.assertIsNone(request.data)
            parsed = urlparse(request.full_url)
            self.assertEqual(parsed.path, '/api/adapt/task/build-config')
            self.assertNotIn('/task/add', request.full_url)
            self.assertEqual(parse_qs(parsed.query), {'gpuType': ['Sunrise_pt-200-x1'], 'framework': ['vllm'], 'taskType': ['text-generation']})
            self.assertEqual(request.get_header('Authorization' if authorization == '1' else 'Xc-token'), 'test-secret')
            self.assertIsNone(request.get_header('Xc-token' if authorization == '1' else 'Authorization'))
    def test_missing_invalid_no_network(self):
        for env in ({}, {'XC_TOKEN': 'placeholder'}, {'XC_TOKEN_FILE': '/nonexistent/token'}):
            with patch.dict(os.environ, env, clear=True), patch('urllib.request.urlopen') as call:
                with self.assertRaisesRegex(BuildConfigError, 'missing/invalid credential'):
                    build_config('gpu')
                with patch('sys.argv', ['build_config_dry']), patch('sys.stderr', io.StringIO()):
                    self.assertEqual(main(), 1)
                call.assert_not_called()
    def test_platform_and_data_errors(self):
        for body, reason in [({'code': 40100}, '40100'), ({'code': 40101}, '40101'), ({'code': 9}, 'other HTTP/platform'), ({'code': 0, 'data': ''}, 'empty/non-str'), ({'code': 0, 'data': []}, 'empty/non-str'), ({'code': 0, 'data': 'test-secret'}, 'credential echoed')]:
            with patch.dict(os.environ, {'XC_TOKEN': 'test-secret'}, clear=True), patch('urllib.request.urlopen', return_value=self.response(body)):
                with self.assertRaisesRegex(BuildConfigError, reason):
                    build_config('gpu')
    def test_http_transport_invalid_response(self):
        for code in (40100, 40101):
            error = urllib.error.HTTPError('https://example.test', 403, '', {}, io.BytesIO(json.dumps({'code': code}).encode()))
            with patch.dict(os.environ, {'XC_TOKEN': 'test-secret'}, clear=True), patch('urllib.request.urlopen', side_effect=error):
                with self.assertRaisesRegex(BuildConfigError, str(code)):
                    build_config('gpu')
        with patch.dict(os.environ, {'XC_TOKEN': 'test-secret'}, clear=True):
            for error in (TimeoutError('secret'), urllib.error.URLError('secret')):
                with patch('urllib.request.urlopen', side_effect=error), self.assertRaisesRegex(BuildConfigError, 'transport/timeout'):
                    build_config('gpu')
            for response in (self.response([], 200), self.response({'code': 0, 'data': 'yaml'}, 500)):
                with patch('urllib.request.urlopen', return_value=response), self.assertRaises(BuildConfigError):
                    build_config('gpu')
            response = self.response({})
            response.read.return_value = b'not-json'
            with patch('urllib.request.urlopen', return_value=response), self.assertRaisesRegex(BuildConfigError, 'invalid JSON'):
                build_config('gpu')
    def test_import_no_io(self):
        import src.build_config_client
        with patch('urllib.request.urlopen') as call, patch('pathlib.Path.read_text') as read:
            runpy.run_path('src/build_config_client.py')
            call.assert_not_called()
            read.assert_not_called()
--- a/tests/test_config_params.py
+++ b/tests/test_config_params.py
@@ -0,0 +1,47 @@
 import json
 import re
 import unittest
 import os
 from unittest.mock import patch
 from pathlib import Path
 from src.config_params import build_config_params, config_params_source, GAP
 from src.build_config_client import BuildConfigError
 from scripts.dry_run import _render
 class TestConfigParams(unittest.TestCase):
    def setUp(self):
        env = patch.dict(os.environ, {}, clear=True)
        env.start()
        self.addCleanup(env.stop)
    def test_source_selection_and_failure(self):
        with patch("urllib.request.urlopen") as network:
            self.assertEqual(config_params_source(), "sample")
            build_config_params()
            network.assert_not_called()
        self.assertIn("discovered via capture", GAP)
        with patch.dict(os.environ, {"CONFIG_PARAMS_SOURCE": "build-config"}):
            with self.assertRaisesRegex(BuildConfigError, "missing/invalid credential"):
                build_config_params("gpu")
        with patch.dict(os.environ, {"ALLOW_BUILD_CONFIG": "1", "XC_TOKEN": "test-secret"}), patch("src.config_params.build_config", return_value="api: yaml") as client:
            self.assertEqual(config_params_source(), "build-config")
            self.assertEqual(build_config_params("gpu"), "api: yaml")
            client.assert_called_once_with("gpu", "vllm", "text-generation")
        with patch.dict(os.environ, {"CONFIG_PARAMS_SOURCE": "build-config"}), patch("src.config_params.build_config", side_effect=BuildConfigError("transport/timeout")):
            with self.assertRaisesRegex(BuildConfigError, "transport/timeout"):
                build_config_params("gpu")
    def test_official_sample_exact(self):
        scrape = Path('/workspace/modelhub-adapt/logs/api-docs-scrape-2026-10-16.md')
        if scrape.is_file():
            body = json.loads(re.search(r'```json\n(.*?)\n```', scrape.read_text(), re.S)[1])
            self.assertEqual(build_config_params(), body['configParams'])
        self.assertIn('max_model_len: 4096', build_config_params())
        self.assertIn('framework: vllm', build_config_params())
        self.assertIn('NOT production-ready', GAP)
    def test_report_draft_and_gap(self):
        report = _render([{'model_address': 'org/a', 'try_next_gpus': ['壁砺']}], 'test', 'fixture')
        for value in ('代提候选 / 需 Victor 确认', 'configParams source', 'configParams_source=sample', 'GAP:', 'truncated', 'max_model_len: 4096', 'perf_order=unset', 'order_source=experience'):
            self.assertIn(value, report)
--- a/tests/test_gate.py
+++ b/tests/test_gate.py
@@ -0,0 +1,159 @@
 """Gate defaults to dry-run / blocked; live only with dual confirm + STRATEGY_ID."""
 from __future__ import annotations
 import os
 import unittest
 from unittest import mock
 from src.gate import LIVE_CONFIRM_VALUE, attempt_live_submit, evaluate_gate
 class FakeClient:
    def __init__(self) -> None:
        self.calls = []
    def submit(self, payload):
        self.calls.append(payload)
        return {"task_id": "fake-1"}
 class TestGate(unittest.TestCase):
    def setUp(self):
        self.counter = mock.patch('src.gate._submit_count', 0)
        self.counter.start()
        self.addCleanup(self.counter.stop)
        self.env = mock.patch.dict(os.environ, {}, clear=True)
        self.env.start()
        self.addCleanup(self.env.stop)
    def test_queue_and_process_limit(self):
        client = FakeClient()
        with mock.patch.dict(os.environ, {
            'STRATEGY_ID': 'sid', 'ALLOW_LIVE_SUBMIT': '1',
            'LIVE_CONFIRM': LIVE_CONFIRM_VALUE, 'ASYNC_QUEUE_DEPTH': '95'
        }):
            self.assertIn('≥95', attempt_live_submit({}, client)['reason'])
            self.assertEqual(client.calls, [])
            os.environ['ASYNC_QUEUE_DEPTH'] = '94'
            self.assertTrue(attempt_live_submit({}, client)['submitted'])
            self.assertFalse(attempt_live_submit({}, client)['submitted'])
            self.assertEqual(len(client.calls), 1)
    def _clear_gate_env(self):
        return mock.patch.dict(
            os.environ,
            {
                "STRATEGY_ID": "sid-1",
                "ALLOW_LIVE_SUBMIT": "0",
                "LIVE_CONFIRM": "",
            },
            clear=False,
        )
    def test_default_dry_run(self) -> None:
        with self._clear_gate_env():
            d = evaluate_gate()
        self.assertFalse(d.allowed)
        self.assertEqual(d.mode, "dry-run")
    def test_missing_strategy_blocked(self) -> None:
        env = {
            "ALLOW_LIVE_SUBMIT": "1",
            "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
        }
        # Remove STRATEGY_ID
        base = {k: v for k, v in os.environ.items() if k != "STRATEGY_ID"}
        base.update(env)
        with mock.patch.dict(os.environ, base, clear=True):
            d = evaluate_gate()
        self.assertFalse(d.allowed)
        self.assertEqual(d.mode, "blocked")
    def test_allow_without_confirm_blocked(self) -> None:
        with mock.patch.dict(
            os.environ,
            {
                "STRATEGY_ID": "sid-1",
                "ALLOW_LIVE_SUBMIT": "1",
                "LIVE_CONFIRM": "YES",
            },
            clear=False,
        ):
            d = evaluate_gate()
        self.assertFalse(d.allowed)
        self.assertEqual(d.mode, "blocked")
    def test_dual_gate_opens(self) -> None:
        with mock.patch.dict(
            os.environ,
            {
                "STRATEGY_ID": "sid-1",
                "ALLOW_LIVE_SUBMIT": "1",
                "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
            },
            clear=False,
        ):
            d = evaluate_gate()
        self.assertTrue(d.allowed)
        self.assertEqual(d.mode, "live")
    def test_open_gate_without_client_still_blocked(self) -> None:
        """Phase2: an open gate never constructs a network client."""
        with mock.patch.dict(
            os.environ,
            {
                "STRATEGY_ID": "sid-1",
                "ALLOW_LIVE_SUBMIT": "1",
                "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
            },
            clear=False,
        ):
            result = attempt_live_submit({"modelAddress": "x/y"})
        self.assertFalse(result["submitted"])
        self.assertIn("no real", result["reason"].lower())
        self.assertEqual(result["mode"], "blocked")
        self.assertFalse(result["ok"])
    def test_prd_example_and_whitespace_id_never_submit(self) -> None:
        cases = [
            {"ALLOW_LIVE_SUBMIT": "true", "LIVE_CONFIRM": LIVE_CONFIRM_VALUE},
            {"ALLOW_LIVE_SUBMIT": "1", "LIVE_CONFIRM": "", "CONFIRM_LIVE": "YES"},
            {"ALLOW_LIVE_SUBMIT": "1", "LIVE_CONFIRM": LIVE_CONFIRM_VALUE, "STRATEGY_ID": "   "},
        ]
        for overrides in cases:
            with self.subTest(overrides=overrides):
                client = FakeClient()
                env = {"STRATEGY_ID": "sid-test", **overrides}
                with mock.patch.dict(os.environ, env, clear=True):
                    result = attempt_live_submit({}, client=client)
                self.assertFalse(result["submitted"])
                self.assertFalse(result["ok"])
                self.assertEqual(client.calls, [])
                self.assertTrue(result["reason"])
    def test_injected_fake_client(self) -> None:
        client = FakeClient()
        with mock.patch.dict(
            os.environ,
            {
                "STRATEGY_ID": "sid-1",
                "ALLOW_LIVE_SUBMIT": "1",
                "LIVE_CONFIRM": LIVE_CONFIRM_VALUE,
            },
            clear=False,
        ):
            result = attempt_live_submit({"modelAddress": "x/y"}, client=client)
        self.assertTrue(result["submitted"])
        self.assertEqual(len(client.calls), 1)
    def test_default_attempt_rejects(self) -> None:
        client = FakeClient()
        with self._clear_gate_env():
            result = attempt_live_submit({"modelAddress": "x/y"}, client=client)
        self.assertFalse(result["submitted"])
        self.assertEqual(len(client.calls), 0)
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_health.py
+++ b/tests/test_health.py
@@ -0,0 +1,60 @@
 """Tests for /health and /status handlers."""
 from __future__ import annotations
 import json
 import os
 import unittest
 from unittest import mock
 from src.health import handle_get, health_payload, status_payload
 class TestHealth(unittest.TestCase):
    def test_health_payload(self) -> None:
        self.assertEqual(health_payload(), {"status": "ok"})
    def test_handle_get_health(self) -> None:
        code, headers, body = handle_get("/health")
        self.assertEqual(code, 200)
        self.assertIn("application/json", headers["Content-Type"])
        data = json.loads(body.decode("utf-8"))
        self.assertEqual(data, {"status": "ok"})
    def test_handle_get_unknown(self) -> None:
        code, _, body = handle_get("/nope")
        self.assertEqual(code, 404)
        self.assertIn("not_found", body.decode("utf-8"))
    def test_status_booleans_only_no_secret_values(self) -> None:
        with mock.patch.dict(
            os.environ,
            {
                "STRATEGY_ID": "secret-strategy-id-xyz",
                "EXTERNAL_SERVICE_TOKEN": "secret-token-abc",
            },
            clear=False,
        ):
            payload = status_payload()
        self.assertTrue(payload["strategy_id_present"])
        self.assertTrue(payload["external_service_token_present"])
        blob = json.dumps(payload)
        self.assertNotIn("secret-strategy-id-xyz", blob)
        self.assertNotIn("secret-token-abc", blob)
    def test_status_absent(self) -> None:
        env = {k: v for k, v in os.environ.items() if k not in (
            "STRATEGY_ID",
            "EXTERNAL_SERVICE_TOKEN",
            "XC_TOKEN",
            "XC_TOKEN_FILE",
            "MODELHUB_XC_TOKEN",
        )}
        with mock.patch.dict(os.environ, env, clear=True):
            payload = status_payload()
        self.assertFalse(payload["strategy_id_present"])
        self.assertFalse(payload["external_service_token_present"])
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_pending_pool.py
+++ b/tests/test_pending_pool.py
@@ -0,0 +1,49 @@
 import tempfile
 import unittest
 from pathlib import Path
 from src.pending_pool import build_ordered_candidates, parse_pending_txt
 class TestPendingPool(unittest.TestCase):
    def test_order_filters_and_first_gpu(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            md, txt, dnr, banned = [root / n for n in ('pending.md', 'pending.txt', 'dnr.md', 'banned.md')]
            md.write_text('''## Section B
 | 1 | `org/cold` | 50 | | | 
 ## Section A
 | 1 | `org/good` | 曦望 | 壁砺, 天垓150 | — |
 | 2 | `org/exact` | 曦望 | 壁砺 | — |
 | 3 | `org/Heretic-model` | 曦望 | 壁砺 | — |
 | 4 | `org/OLMo-2-model` | 曦望 | 壁砺 | — |
 ''')
            dnr.write_text('| 1 | `org/exact` | fail |\n## Uniqueness collisions\n- `org/collision`\n## Overnight\n- `org/overnight`')
            banned.write_text('`abliterated` `heretic` `uncensored` `unsensored`')
            rows = build_ordered_candidates(md, txt, dnr, banned)
            self.assertEqual([r['section'] for r in rows], ['A'] * 4 + ['B'])
            self.assertEqual(rows[0]['try_next_gpus'][0], '壁砺')
            self.assertEqual(rows[1]['skip_reason'], 'do-not-resubmit')
            self.assertEqual(rows[2]['status'], 'skipped')
            self.assertIn('family', rows[3]['skip_reason'])
            self.assertTrue(rows[-1]['needs_human'])
            txt.write_text('org/b\t#B cold\norg/a\t#A done:曦望; still try:壁砺+天垓150\norg/c\\t#A try:天垓150\n')
            self.assertEqual(parse_pending_txt(txt)[1]['try_next_gpus'], ['壁砺', '天垓150'])
            fallback = build_ordered_candidates(root/'absent', txt, dnr, banned)
            self.assertEqual([r['model_address'] for r in fallback], ['org/a', 'org/c', 'org/b'])
            self.assertEqual(len(build_ordered_candidates(md, txt, dnr, banned)), 5)
    def test_experience_order_missing_only_and_ties(self):
        from src.pending_pool import experience_order, parse_pending_md
        self.assertEqual(experience_order(['智铠 mrv-100', '天垓100', '曦望', '天垓150', '壁砺']),
                         ['曦望', '壁砺', '天垓150', '天垓100', '智铠 mrv-100'])
        self.assertEqual(experience_order(['天垓150']), ['天垓150'])
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / 'pending.md'
            path.write_text('## Section A\n| 1 | `org/a` | 壁砺:验证失败; 曦望:正在验证 | 天垓150 | no 壁砺 retry |\n## Section B\n| 1 | `org/b` | 0 | 壁砺 |\n')
            from unittest.mock import patch
            import os
            with patch.dict(os.environ, {'PENDING_SECTION': 'A'}):
                rows = build_ordered_candidates(path, dnr_path=Path(directory)/'none', banned_path=Path(directory)/'none')
            self.assertEqual(len(rows), 1)
            self.assertEqual(rows[0]['try_next_gpus'], ['天垓150'])
            self.assertEqual(rows[0]['perf_order'], 'unset')
--- a/tests/test_readonly_probe.py
+++ b/tests/test_readonly_probe.py
@@ -0,0 +1,61 @@
 import io
 import json
 import os
 import unittest
 import urllib.error
 from unittest.mock import patch, MagicMock
 from src.readonly_probe import probe_readonly
 from scripts.probe_readonly import main
 class TestReadonlyProbe(unittest.TestCase):
    def test_missing_token_no_network(self):
        with patch.dict(os.environ, {}, clear=True), patch('urllib.request.urlopen') as call:
            self.assertFalse(probe_readonly()['auth_ok'])
            with patch('sys.stdout', io.StringIO()):
                self.assertEqual(main(), 1)
            call.assert_not_called()
    def test_success_and_exact_request(self):
        response = MagicMock()
        response.__enter__.return_value = response
        response.status = 200
        response.read.return_value = json.dumps({'code': 0, 'data': {
            'records': [{'modelId': 'secret-value'}], 'total': 10}}).encode()
        with patch.dict(os.environ, {'EXTERNAL_SERVICE_TOKEN': 'secret-value'}, clear=True), patch('urllib.request.urlopen', return_value=response) as call:
            result = probe_readonly()
        request = call.call_args.args[0]
        self.assertEqual(request.get_method(), 'GET')
        self.assertEqual(request.full_url, 'https://modelhub.org.cn/api/adapt/task/page?onlyMine=true&pageSize=5&current=1')
        self.assertEqual(request.get_header('Xc-token'), 'secret-value')
        self.assertTrue(result['auth_ok'])
        self.assertEqual(result['record_summary'], {'count': 1, 'total': 10})
        self.assertNotIn('secret-value', json.dumps(result))
    def test_errors_no_fallback_or_echo(self):
        for code in (40100, 40101, 50000):
            error = urllib.error.HTTPError('https://example.test', 403, '', {}, io.BytesIO(json.dumps({'code': code, 'message': 'secret-value'}).encode()))
            with patch.dict(os.environ, {'XC_TOKEN': 'secret-value'}, clear=True), patch('urllib.request.urlopen', side_effect=error) as call:
                result = probe_readonly()
            self.assertEqual(call.call_count, 1)
            self.assertFalse(result['auth_ok'])
            self.assertEqual(result['platform_code'], code)
            self.assertIn(str(code) if code != 50000 else 'other', result['error_attribution'])
            self.assertNotIn('secret-value', json.dumps(result))
    def test_transport_and_invalid_json(self):
        for error in (OSError('secret-value'), urllib.error.URLError('secret-value')):
            with patch.dict(os.environ, {'XC_TOKEN': 'secret-value'}, clear=True), patch('urllib.request.urlopen', side_effect=error):
                self.assertEqual(probe_readonly()['error_attribution'], 'transport/timeout')
        response = MagicMock()
        response.__enter__.return_value = response
        response.status, response.read.return_value = 200, b'invalid secret-value'
        with patch.dict(os.environ, {'XC_TOKEN': 'secret-value'}, clear=True), patch('urllib.request.urlopen', return_value=response):
            self.assertFalse(probe_readonly()['auth_ok'])
    def test_import_no_network(self):
        import importlib
        import src.readonly_probe
        with patch('urllib.request.urlopen') as call:
            importlib.reload(src.readonly_probe)
            call.assert_not_called()
--- a/tests/test_report_safety.py
+++ b/tests/test_report_safety.py
@@ -0,0 +1,46 @@
 """Local report and HTTP log regression checks; no network requests."""
 import contextlib
 import io
 import os
 import unittest
 from pathlib import Path
 from unittest import mock
 from scripts.dry_run import _render
 from src.server import AgentHandler
 class TestReportSafety(unittest.TestCase):
    def test_report_marks_pending_and_does_not_echo_gate_values(self):
        sentinel = "synthetic-sensitive-value"
        with mock.patch.dict(os.environ, {
            "ALLOW_LIVE_SUBMIT": sentinel,
            "LIVE_CONFIRM": sentinel,
            "STRATEGY_ID": sentinel,
            "EXTERNAL_SERVICE_TOKEN": sentinel,
        }, clear=True):
            report = _render([], "test-time", Path("example.json"))
        self.assertNotIn(sentinel, report)
        for required in ("A→B", "需 Victor 人工提交", "积分：待平台核实", "≥95"):
            self.assertIn(required, report)
    def test_pending_draft_queue_banner_and_skips(self):
        rows = [dict(model_address='org/a', section='A', try_next_gpus=['壁砺', '天垓150'],
                     status='skipped', skip_reason='do-not-resubmit', needs_human=False),
                dict(model_address='org/b', section='B', try_next_gpus=[],
                     status='needs_human', needs_human=True)]
        with mock.patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': '95'}, clear=True):
            report = _render(rows, 'test', Path('pending.md'))
        self.assertIn('禁 live / 队列≥95', report)
        self.assertIn('do-not-resubmit', report)
        self.assertIn('needs_human', report)
        self.assertIn('"targetGpu": "壁砺"', report)
        self.assertNotIn('"targetGpu": "天垓150"', report)
        self.assertLess(report.index('Section A'), report.index('Section B'))
    def test_http_log_omits_request_details(self):
        output = io.StringIO()
        with contextlib.redirect_stderr(output):
            AgentHandler.log_message(None, '"%s" %s', '/health?token=synthetic-sensitive-value', 200)
        self.assertNotIn("synthetic-sensitive-value", output.getvalue())
        self.assertIn("request handled", output.getvalue())
--- a/tests/test_sigterm.py
+++ b/tests/test_sigterm.py
@@ -0,0 +1,69 @@
 """SIGTERM should stop the server subprocess promptly (<< 30s platform window)."""
 from __future__ import annotations
 import json
 import os
 import signal
 import subprocess
 import sys
 import time
 import unittest
 import urllib.request
 from pathlib import Path
 ROOT = Path(__file__).resolve().parents[1]
 class TestSigterm(unittest.TestCase):
    def test_sigterm_exits_quickly(self) -> None:
        port = "19090"
        env = os.environ.copy()
        env["STRATEGY_ID"] = "sigterm-test"
        env["PORT"] = port
        env["HOST"] = "127.0.0.1"
        proc = subprocess.Popen(
            [sys.executable, "-m", "src.server"],
            cwd=str(ROOT),
            env=env,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
        )
        try:
            # wait until health is up
            url = f"http://127.0.0.1:{port}/health"
            deadline = time.time() + 5
            while time.time() < deadline:
                try:
                    with urllib.request.urlopen(url, timeout=0.5) as resp:
                        data = json.loads(resp.read().decode("utf-8"))
                        if resp.status == 200 and data.get("status") == "ok":
                            break
                except Exception:
                    time.sleep(0.1)
            else:
                if proc.poll() is not None:
                    self.fail(f"server exited before health check: {proc.stderr.read().decode(errors='replace')}")
                self.fail("server did not become healthy in time")
            t0 = time.time()
            proc.send_signal(signal.SIGTERM)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                self.fail("server did not exit within 5s after SIGTERM (platform allows ≤30s)")
            elapsed = time.time() - t0
            self.assertLess(elapsed, 5.0, f"exit took {elapsed:.2f}s")
            self.assertEqual(proc.returncode, 0, "SIGTERM must exit cleanly, not kill the process")
        finally:
            if proc.poll() is None:
                proc.kill()
                proc.wait(timeout=2)
            if proc.stderr:
                proc.stderr.close()
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_strategy_startup.py
+++ b/tests/test_strategy_startup.py
@@ -0,0 +1,57 @@
 """Missing STRATEGY_ID must cause startup failure (exit non-zero)."""
 from __future__ import annotations
 import os
 import subprocess
 import sys
 import unittest
 from pathlib import Path
 from unittest import mock
 ROOT = Path(__file__).resolve().parents[1]
 class TestStrategyStartup(unittest.TestCase):
    def test_missing_strategy_id_exits_nonzero(self) -> None:
        env = os.environ.copy()
        env["STRATEGY_ID"] = ""
        env["PORT"] = "19081"
        env["HOST"] = "127.0.0.1"
        proc = subprocess.run(
            [sys.executable, "-m", "src.server"],
            cwd=str(ROOT),
            env=env,
            capture_output=True,
            text=True,
            timeout=10,
        )
        self.assertNotEqual(proc.returncode, 0)
        self.assertIn("STRATEGY_ID", proc.stderr)
    def test_whitespace_strategy_id_exits_nonzero(self) -> None:
        env = os.environ.copy()
        env["STRATEGY_ID"] = "   "
        env["PORT"] = "19082"
        env["HOST"] = "127.0.0.1"
        proc = subprocess.run(
            [sys.executable, "-m", "src.server"],
            cwd=str(ROOT),
            env=env,
            capture_output=True,
            text=True,
            timeout=10,
        )
        self.assertNotEqual(proc.returncode, 0)
    def test_require_strategy_id_helper(self) -> None:
        from src.strategy import require_strategy_id
        with self.assertRaises(SystemExit) as ctx:
            with mock.patch.dict(os.environ, {"STRATEGY_ID": ""}, clear=False):
                require_strategy_id()
        self.assertNotEqual(ctx.exception.code, 0)
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_task_add_client.py
+++ b/tests/test_task_add_client.py
@@ -0,0 +1,69 @@
 import importlib
 import io
 import os
 import tempfile
 import unittest
 import urllib.error
 from pathlib import Path
 from unittest import mock
 from src import task_add_client as module
 class TestTaskAddClient(unittest.TestCase):
    def setUp(self):
        patch = mock.patch.dict(os.environ, {'STRATEGY_ID': 'sid', 'XC_TOKEN': 'test-secret-credential'}, clear=True)
        patch.start()
        self.addCleanup(patch.stop)
        self.payload = module.build_payload('org/model', '壁砺', 'vllm', config_params={ 'max_model_len': 4096 })
    def test_import_no_network(self):
        with mock.patch('urllib.request.urlopen') as urlopen:
            importlib.reload(module)
        urlopen.assert_not_called()
    def test_success(self):
        response = mock.MagicMock()
        response.__enter__.return_value = response
        response.status = 200
        response.read.return_value = b'{"code":0,"message":"test-secret-credential"}'
        with mock.patch('urllib.request.urlopen', return_value=response) as urlopen:
            result = module.TaskAddClient().submit(self.payload)
        self.assertEqual(result['http_status'], 200)
        self.assertNotIn(os.environ['XC_TOKEN'], str(result))
        request = urlopen.call_args.args[0]
        self.assertEqual(request.get_header('Xc-token'), os.environ['XC_TOKEN'])
        self.assertEqual(request.method, 'POST')
        urlopen.assert_called_once()
    def test_http_4xx_60014(self):
        error = urllib.error.HTTPError('https://example.invalid', 403, 'forbidden', {}, io.BytesIO(b'{"code":60014}'))
        with mock.patch('urllib.request.urlopen', side_effect=error) as urlopen:
            result = module.TaskAddClient().submit(self.payload)
        self.assertEqual(result['http_status'], 403)
        self.assertEqual(result['platform_code'], 60014)
        self.assertIn('D (unverified)', result['error_attribution'])
        urlopen.assert_called_once()
    def test_transport_no_retry(self):
        with mock.patch('urllib.request.urlopen', side_effect=TimeoutError('test-secret-credential')) as urlopen:
            result = module.TaskAddClient().submit(self.payload)
        self.assertNotIn(os.environ['XC_TOKEN'], str(result))
        self.assertEqual(result['error_attribution'], 'transport/timeout')
        urlopen.assert_called_once()
    def test_token_precedence_and_rejections(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory)/'credential'
            path.write_text(' file-credential \n')
            with mock.patch.dict(os.environ, {'XC_TOKEN_FILE': str(path)}):
                self.assertEqual(module.resolve_token(), 'file-credential')
        for token in ('', 'tmp', 'PLACEHOLDER', 'changeme', 'tmp-token'):
            with self.subTest(token=token), mock.patch.dict(os.environ, {'XC_TOKEN': token}):
                with self.assertRaises(ValueError):
                    module.resolve_token()
    def test_placeholder_never_requests(self):
        with mock.patch('urllib.request.urlopen') as urlopen:
            with self.assertRaises(ValueError):
                module.TaskAddClient().submit(module.build_payload('org/model'))
        urlopen.assert_not_called()
		`@@ -0,0 +1,3 @@`
							`"""ModelHub XC adaptation agent — Phase1 local skeleton."""`

							`__version__ = "0.1.0"`