harden probe agent v1.0.3

安全鉴权、默认只读、显式提交开关、状态接口与测试

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+*.py[cod]

README.md

@@ -1,39 +1,69 @@
 # huni-probe-agent
 
-信创模盒 ModelHub XC 适配智能体 · **只读探针骨架 (probe-only)**。
+信创模盒 ModelHub XC 适配智能体安全探针 `v1.0.3`。
 
-对照官方样例 `luopingyi/xc_agent_platform_demo`（最小合规骨架）扩了一个零副作用探针。
+它用于验证「建仓 → Kaniko 构建 → 部署 → `/health` → 平台 API 鉴权 → 可控提交」链路。默认只读，只有显式开启提交开关后才会创建验证任务。
 
-## 用途
+## 安全行为
 
-1. 验证平台「建仓 → Kaniko 构建 → 部署 → `/health` → 运行/停止」整条链路是否通。
+- 默认只调用一次只读接口 `GET /api/adapt/task/page`。
-2. 零副作用观测平台注入的 `STRATEGY_ID` / `EXTERNAL_SERVICE_TOKEN` 是否为有效凭证：
+- 空凭证和 `tmp`、`placeholder` 等占位值会被拒绝，不会继续请求。
-   启动时用注入 token 调一次**只读** `GET /api/adapt/task/page`，看返回 `code`：
+- 任何日志和 HTTP 状态响应都不包含 token 值或长度。
-   `0` = 凭证有效；`40100` = 未登录/凭证无效。
+- 只有 `ALLOW_SUBMIT=1` 且只读鉴权成功，才会调用 `build-config` 和 `task/add`。
+- 每个进程生命周期最多调用一次 `task/add`，没有自动重试。
+- `/health` 始终只代表容器存活；链路结果请查看 `/status`。
 
-## 满足的运行时契约
+## 凭证注入
 
-- 根目录 `Dockerfile`，`EXPOSE 8080`
+按以下优先级读取凭证：
-- `GET /health` 返回 HTTP 200（K8s livenessProbe）
-- 读取 `STRATEGY_ID` 与 `EXTERNAL_SERVICE_TOKEN` 环境变量
-- 处理 `SIGTERM`，优雅停机（30s 窗口内退出）
-- 资源占用极小，符合 requests 100m/256Mi、limits 1C/512Mi
 
-## 明确不做
+1. `XC_TOKEN_FILE`：Kubernetes Secret 挂载文件路径，推荐。
+2. `XC_TOKEN`：由 Kubernetes Secret 注入的环境变量。
+3. `EXTERNAL_SERVICE_TOKEN`：兼容平台旧变量名。
 
-- **绝不提交任何验证任务**（不调 `task/add`，不调 `build-config`）。
+不要把 token 写进代码、Dockerfile、Git 仓库或智能体名称。平台侧应通过 Kubernetes Secret 注入真实凭证；如果仍注入 `tmp`，探针会进入 `blocked_missing_auth`，不会误提交。
-- 不打印 token 值、不落盘任何凭证；探针仅调只读 GET，仅记录 `http/code/计数`。
 
-## 本地运行
+## 运行时变量
+
+| 变量 | 默认值 | 说明 |
+| --- | --- | --- |
+| `STRATEGY_ID` | 无 | 平台策略 ID，必填 |
+| `ALLOW_SUBMIT` | `0` | 设置为 `1` 才允许单次提交 |
+| `MAIN_HOST` | `https://modelhub.org.cn` | 平台 API 地址 |
+| `PORT` | `8080` | HTTP 监听端口 |
+| `TEST_MODEL` | TrialSpace-1225-GGUF | 验证模型地址 |
+| `GPU` | `MetaX_c-500` | 目标 GPU |
+| `FW` | `vllm` | 推理框架 |
+| `TT` | `text-generation` | 任务类型 |
+
+## 状态接口
+
+- `GET /health`：容器存活检查，HTTP 200。
+- `GET /status`（或 `/`）：返回安全状态，不返回凭证。
+
+关键 `phase`：
+
+- `blocked_missing_strategy_id`：平台未注入策略 ID。
+- `blocked_missing_auth`：没有可用凭证，或只拿到了占位凭证。
+- `auth_failed`：只读鉴权请求未通过。
+- `read_only_complete`：鉴权通过，默认只读流程完成。
+- `build_config_failed`：配置生成失败，已阻止后续提交。
+- `submit_complete` / `submit_failed`：显式开启后，单次提交成功或失败。
+
+## 本地验证
 
 ```bash
-python main.py
+python -m unittest -v
-curl http://localhost:8080/health     # {"status":"ok"}
+
-curl http://localhost:8080/           # 查看探针结果
+PORT=8080 STRATEGY_ID=local-test python main.py
+curl http://localhost:8080/health
+curl http://localhost:8080/status
 ```
 
+本地没有凭证时，预期状态是 `blocked_missing_auth`。
+
 ## 上平台
 
-推到 `https://dev.modelhub.org.cn/<user>/<repo>` 并保留一个**真 git 标签**（如 `v1.0.0`；
+推到 `https://dev.modelhub.org.cn/<user>/<repo>` 并创建真实 Git 标签（例如 `v1.0.3`）。在「我的适配智能体 → 新增智能体」填写名称、仓库地址、标签。平台 Kaniko 按 `refs/tags/<tag>` 构建，不能用分支名代替标签。
-平台 Kaniko 按 `refs/tags/<tag>` 解析，用分支名会失败）。再在「我的适配智能体 → 新增智能体」
+
-填 名称 / 仓库地址 / 标签 提交。
+建议先保持 `ALLOW_SUBMIT=0`，确认 `/status` 为 `read_only_complete`；再由平台管理员把它改为 `1`，仅跑一次验证任务，验证后立即恢复为 `0`。

main.py

@@ -1,15 +1,10 @@
-"""ModelHub XC 适配智能体 · 只读探针 + 容器内提交测试 v1.0.2
+"""ModelHub XC 适配智能体安全探针 v1.0.3。
 
-关键实验:在容器里用注入的 EXTERNAL_SERVICE_TOKEN(=tmp) + STRATEGY_ID 真调一次 task/add,
+默认只做一次只读鉴权检查。只有显式设置 ALLOW_SUBMIT=1，且鉴权检查成功后，
-验证平台是否靠「请求来自容器 pod」放行智能体提交(浏览器/脚本提交会被 60014 挡)。
+才会调用 build-config 和 task/add；每个进程生命周期最多提交一次。
-
-诊断输出:
-  - dump 所有 env 的名字+长度(不打值)。
-  - EXTERNAL_SERVICE_TOKEN 值(短则原样、长则打码)。
-  - 容器内 build-config + task/add(真实模型) 的返回 code/msg。
-
-约束:token 值不打(短占位除外);只提交这一次测试(真实模型可能进队列);SIGTERM 优雅停机。
 """
+
+import copy
 import json
 import os
 import signal
@@ -18,100 +13,234 @@ import time
 import urllib.error
 import urllib.request
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Callable, Mapping, Optional, Tuple
 
-MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn")
+
-TOKEN = os.getenv("EXTERNAL_SERVICE_TOKEN", "")
+VERSION = "1.0.3"
-STRATEGY_ID = os.getenv("STRATEGY_ID", "")
+MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn").rstrip("/")
+STRATEGY_ID = os.getenv("STRATEGY_ID", "").strip()
 PORT = int(os.getenv("PORT", "8080"))
+ALLOW_SUBMIT = os.getenv("ALLOW_SUBMIT", "").strip().lower() in {"1", "true", "yes", "on"}
 
-TEST_MODEL = os.getenv("TEST_MODEL", "https://www.modelscope.cn/models/mradermacher/TrialSpace-1225-GGUF")
+TEST_MODEL = os.getenv(
-GPU, FW, TT = os.getenv("GPU", "MetaX_c-500"), os.getenv("FW", "vllm"), os.getenv("TT", "text-generation")
+    "TEST_MODEL",
-
+    "https://www.modelscope.cn/models/mradermacher/TrialSpace-1225-GGUF",
-HARDCODED_CFG = (
-    "framework: vllm\napi: completion\nlang: en\nmax_model_len: 4096\nmax_tokens: 1024\n"
-    "temperature: 0.7\nrepetition_penalty: 1.1\ntop_p: 0.9\nsut_config:\n  gpu_num: 1\n  values:\n"
-    "    command:\n    - /opt/conda/bin/vllm\n    - serve\n    - /model\n    - --port\n    - '20644'\n"
-    "    - --served-model-name\n    - llm\n    - --max-model-len\n    - '4096'\n"
-    "    - --gpu-memory-utilization\n    - '0.9'\n    - -tp\n    - '1'\n    - --enforce-eager\n"
-    "    - --trust-remote-code\nref_config:\n  gpu_num: 1\n  values:\n    command:\n    - vllm\n"
-    "    - serve\n    - /model\n    - --port\n    - '80'\n    - --served-model-name\n    - llm\n"
-    "    - --max-model-len\n    - '4096'\n    - -tp\n    - '1'\n    - --enforce-eager\n    - --trust-remote-code\n"
 )
+GPU = os.getenv("GPU", "MetaX_c-500")
+FW = os.getenv("FW", "vllm")
+TT = os.getenv("TT", "text-generation")
 
+_TOKEN_PLACEHOLDERS = {"tmp", "placeholder", "changeme", "change-me", "example", "test"}
 shutdown = threading.Event()
-PROBE = {"done": False, "results": []}
+STATE_LOCK = threading.Lock()
+
+
+def _initial_state() -> dict:
+    return {
+        "version": VERSION,
+        "phase": "starting",
+        "done": False,
+        "auth_ready": False,
+        "submit_enabled": ALLOW_SUBMIT,
+        "token_source": "none",
+        "results": {},
+    }
+
+
+STATE = _initial_state()
 
 
 def log(msg: str) -> None:
     print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
 
 
-def _mask(v: str) -> str:
+def _set_state(**changes) -> None:
-    return v if len(v) <= 8 else (v[:2] + "…" + v[-2:])
+    with STATE_LOCK:
+        STATE.update(changes)
 
 
-def _req(method: str, path: str, headers: dict, body=None, timeout: int = 30) -> dict:
+def _snapshot() -> dict:
-    data = body.encode() if isinstance(body, str) else body
+    with STATE_LOCK:
+        return copy.deepcopy(STATE)
+
+
+def _reset_state() -> None:
+    """Reset process state; kept separate to make the probe deterministically testable."""
+    with STATE_LOCK:
+        STATE.clear()
+        STATE.update(_initial_state())
+
+
+def _valid_token(value: str) -> bool:
+    value = value.strip()
+    return bool(value) and value.lower() not in _TOKEN_PLACEHOLDERS
+
+
+def _resolve_token(
+    env: Optional[Mapping[str, str]] = None,
+    read_text: Optional[Callable[[str], str]] = None,
+) -> Tuple[str, str, Optional[str]]:
+    """Resolve a credential without exposing its value or length.
+
+    Precedence is XC_TOKEN_FILE, XC_TOKEN, then the legacy
+    EXTERNAL_SERVICE_TOKEN. An explicitly configured but unreadable token file
+    is treated as an error instead of silently falling back.
+    """
+    values = os.environ if env is None else env
+    file_reader = read_text or (lambda path: Path(path).read_text(encoding="utf-8"))
+    token_file = values.get("XC_TOKEN_FILE", "").strip()
+    if token_file:
+        try:
+            value = file_reader(token_file).strip()
+        except (OSError, UnicodeError):
+            return "", "XC_TOKEN_FILE", "token_file_unreadable"
+        if _valid_token(value):
+            return value, "XC_TOKEN_FILE", None
+        return "", "XC_TOKEN_FILE", "token_missing_or_placeholder"
+
+    for name in ("XC_TOKEN", "EXTERNAL_SERVICE_TOKEN"):
+        value = values.get(name, "").strip()
+        if _valid_token(value):
+            return value, name, None
+        if value:
+            return "", name, "token_missing_or_placeholder"
+    return "", "none", "token_missing_or_placeholder"
+
+
+def _req(method: str, path: str, token: str, body=None, timeout: int = 30) -> dict:
+    headers = {"Accept": "application/json", "Xc-Token": token}
+    data = body.encode("utf-8") if isinstance(body, str) else body
+    if data is not None:
+        headers["Content-Type"] = "application/json"
     req = urllib.request.Request(MAIN + path, data=data, method=method, headers=headers)
     try:
-        with urllib.request.urlopen(req, timeout=timeout) as r:
+        with urllib.request.urlopen(req, timeout=timeout) as response:
-            txt = r.read().decode("utf-8", "ignore")
+            text = response.read().decode("utf-8", "ignore")
-            status = r.status
+            status = response.status
-    except urllib.error.HTTPError as e:
+    except urllib.error.HTTPError as error:
-        txt = e.read().decode("utf-8", "ignore")
+        text = error.read().decode("utf-8", "ignore")
-        status = e.code
+        status = error.code
-    except Exception as e:  # noqa: BLE001
+    except Exception as error:  # noqa: BLE001 - network diagnostics are returned safely
-        return {"http": -1, "err": str(e)[:100]}
+        return {"http": -1, "error": type(error).__name__}
+
     try:
-        j = json.loads(txt)
+        payload = json.loads(text)
-        return {"http": status, "code": j.get("code"), "msg": (j.get("message") or "")[:100],
+    except (TypeError, ValueError):
-                "data": j.get("data")}
+        return {"http": status, "code": None, "message": "non_json_response"}
-    except Exception:  # noqa: BLE001
+    return {
-        return {"http": status, "raw": txt[:100]}
+        "http": status,
+        "code": payload.get("code"),
+        "message": str(payload.get("message") or "")[:100],
+        "data": payload.get("data"),
+    }
+
+
+def _api_ok(result: dict) -> bool:
+    try:
+        status = int(result.get("http", -1))
+    except (TypeError, ValueError):
+        return False
+    return 200 <= status < 300 and result.get("code") in {0, "0"}
+
+
+def _result_summary(result: dict) -> dict:
+    summary = {"http": result.get("http"), "code": result.get("code")}
+    if result.get("message"):
+        summary["message"] = result["message"]
+    if result.get("error"):
+        summary["error"] = result["error"]
+    return summary
+
+
+def _task_id(data) -> Optional[str]:
+    if not isinstance(data, dict):
+        return None
+    for key in ("id", "taskId", "task_id"):
+        if data.get(key) is not None:
+            return str(data[key])
+    return None
 
 
 def probe() -> None:
-    env_lens = {k: len(str(os.environ.get(k, ""))) for k in sorted(os.environ)}
+    """Run one guarded probe; no retry loop and at most one task/add call."""
-    log("env keys+lens: " + json.dumps(env_lens, ensure_ascii=False))
+    token, source, token_error = _resolve_token()
-    log(f"strategy_id={STRATEGY_ID or 'MISSING'} | EXTERNAL_SERVICE_TOKEN value={_mask(TOKEN)!r} len={len(TOKEN)}")
+    _set_state(token_source=source, submit_enabled=ALLOW_SUBMIT)
 
-    # 只读:token 有效性
+    if not STRATEGY_ID:
-    page = _req("GET", "/api/adapt/task/page?current=1&pageSize=1", {"Xc-Token": TOKEN})
+        _set_state(phase="blocked_missing_strategy_id", done=True)
-    log("EST-token task/page -> " + json.dumps({k: page.get(k) for k in ("http", "code")}, ensure_ascii=False))
+        log("probe blocked: STRATEGY_ID is missing")
+        return
+    if token_error:
+        _set_state(phase="blocked_missing_auth", done=True, results={"auth": token_error})
+        log(f"probe blocked: usable credential is unavailable (source={source})")
+        return
 
-    # 关键:容器内 build-config + task/add(真实模型)
+    _set_state(phase="checking_auth")
-    bc = _req("POST", f"/api/adapt/task/build-config?gpuType={GPU}&framework={FW}&taskType={TT}", {"Xc-Token": TOKEN})
+    page = _req("GET", "/api/adapt/task/page?current=1&pageSize=1", token)
-    log("build-config(EST-token) -> " + json.dumps({k: bc.get(k) for k in ("http", "code", "msg")}, ensure_ascii=False))
+    results = {"task_page": _result_summary(page)}
-    cfg = bc.get("data") if isinstance(bc.get("data"), str) and bc.get("data") else HARDCODED_CFG
+    if not _api_ok(page):
-    body = json.dumps({"modelAddress": TEST_MODEL, "taskType": TT, "targetGpu": GPU,
+        _set_state(phase="auth_failed", done=True, results=results)
-                       "framework": FW, "strategyId": STRATEGY_ID, "configParams": cfg})
+        log("read-only auth check failed: " + json.dumps(results["task_page"], ensure_ascii=False))
-    ta = _req("POST", "/api/adapt/task/add", {"Xc-Token": TOKEN, "Content-Type": "application/json"}, body)
+        return
-    log("SUBMIT-TEST task/add(EST-token+strategyId, real model) -> " + json.dumps(ta, ensure_ascii=False))
+
-    PROBE["results"] = [{"task_page": page.get("code")}, {"build_config": bc.get("code")},
+    _set_state(auth_ready=True, results=results)
-                        {"task_add": ta.get("code"), "msg": ta.get("msg")}]
+    log(f"read-only auth check passed (source={source})")
-    PROBE["done"] = True
+    if not ALLOW_SUBMIT:
-    log("probe done | task/add code=0 => 容器提交通; 60014 => 容器来源也被判脚本; 401 => token 无效")
+        _set_state(phase="read_only_complete", done=True)
+        log("probe complete in read-only mode; task/add was not called")
+        return
+
+    _set_state(phase="preparing_submission")
+    build_path = f"/api/adapt/task/build-config?gpuType={GPU}&framework={FW}&taskType={TT}"
+    build_config = _req("POST", build_path, token)
+    results["build_config"] = _result_summary(build_config)
+    config = build_config.get("data")
+    if not _api_ok(build_config) or not isinstance(config, str) or not config:
+        _set_state(phase="build_config_failed", done=True, results=results)
+        log("build-config failed; task/add was not called")
+        return
+
+    body = json.dumps(
+        {
+            "modelAddress": TEST_MODEL,
+            "taskType": TT,
+            "targetGpu": GPU,
+            "framework": FW,
+            "strategyId": STRATEGY_ID,
+            "configParams": config,
+        }
+    )
+    task_add = _req("POST", "/api/adapt/task/add", token, body)
+    results["task_add"] = _result_summary(task_add)
+    task_id = _task_id(task_add.get("data"))
+    if task_id:
+        results["task_add"]["task_id"] = task_id
+
+    phase = "submit_complete" if _api_ok(task_add) else "submit_failed"
+    _set_state(phase=phase, done=True, results=results)
+    log("single task/add attempt finished: " + json.dumps(results["task_add"], ensure_ascii=False))
 
 
 class Handler(BaseHTTPRequestHandler):
     def _json(self, obj: dict, status: int = 200) -> None:
-        payload = json.dumps(obj, ensure_ascii=False).encode()
+        payload = json.dumps(obj, ensure_ascii=False).encode("utf-8")
         self.send_response(status)
-        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Type", "application/json; charset=utf-8")
         self.send_header("Content-Length", str(len(payload)))
         self.end_headers()
         self.wfile.write(payload)
 
     def do_GET(self) -> None:
         if self.path == "/health":
-            self._json({"status": "ok"})
+            self._json({"status": "ok", "version": VERSION})
             return
-        if self.path == "/":
+        if self.path in {"/", "/status"}:
-            self._json({"name": "huni-probe-agent", "mode": "submit-test-v1.0.2", "probe": PROBE})
+            self._json({"name": "huni-probe-agent", "probe": _snapshot()})
             return
         self._json({"error": "not found"}, 404)
 
-    def log_message(self, *_a) -> None:
+    def log_message(self, *_args) -> None:
         pass
 
 
@@ -127,7 +256,7 @@ def main() -> None:
 
     server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
     server.timeout = 1
-    log(f"probe agent v1.0.2 listening on 0.0.0.0:{PORT}")
+    log(f"probe agent v{VERSION} listening on 0.0.0.0:{PORT} (submit_enabled={ALLOW_SUBMIT})")
     while not shutdown.is_set():
         server.handle_request()
     server.server_close()

test_main.py

@@ -0,0 +1,116 @@
+import unittest
+from unittest import mock
+
+import main
+
+
+class TokenResolutionTests(unittest.TestCase):
+    def test_placeholder_legacy_token_is_rejected(self):
+        token, source, error = main._resolve_token({"EXTERNAL_SERVICE_TOKEN": "tmp"})
+        self.assertEqual(token, "")
+        self.assertEqual(source, "EXTERNAL_SERVICE_TOKEN")
+        self.assertEqual(error, "token_missing_or_placeholder")
+
+    def test_token_file_has_precedence(self):
+        token, source, error = main._resolve_token(
+            {"XC_TOKEN_FILE": "/secret/token", "XC_TOKEN": "env-token"},
+            read_text=lambda _path: "file-token\n",
+        )
+        self.assertEqual(token, "file-token")
+        self.assertEqual(source, "XC_TOKEN_FILE")
+        self.assertIsNone(error)
+
+    def test_unreadable_explicit_token_file_does_not_fall_back(self):
+        def fail(_path):
+            raise OSError("not mounted")
+
+        token, source, error = main._resolve_token(
+            {"XC_TOKEN_FILE": "/secret/token", "XC_TOKEN": "env-token"},
+            read_text=fail,
+        )
+        self.assertEqual(token, "")
+        self.assertEqual(source, "XC_TOKEN_FILE")
+        self.assertEqual(error, "token_file_unreadable")
+
+
+class ProbeTests(unittest.TestCase):
+    def setUp(self):
+        main._reset_state()
+
+    def test_missing_token_makes_no_request(self):
+        with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
+            main, "_resolve_token", return_value=("", "EXTERNAL_SERVICE_TOKEN", "token_missing_or_placeholder")
+        ), mock.patch.object(main, "_req") as request:
+            main.probe()
+        request.assert_not_called()
+        self.assertEqual(main._snapshot()["phase"], "blocked_missing_auth")
+
+    def test_read_only_mode_never_calls_post(self):
+        calls = []
+
+        def request(method, path, _token, body=None, timeout=30):
+            calls.append((method, path, body, timeout))
+            return {"http": 200, "code": 0, "data": {}}
+
+        with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
+            main, "ALLOW_SUBMIT", False
+        ), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
+            main, "_req", side_effect=request
+        ):
+            main.probe()
+
+        self.assertEqual([call[0] for call in calls], ["GET"])
+        state = main._snapshot()
+        self.assertEqual(state["phase"], "read_only_complete")
+        self.assertTrue(state["auth_ready"])
+
+    def test_failed_auth_never_calls_post(self):
+        with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
+            main, "ALLOW_SUBMIT", True
+        ), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
+            main, "_req", return_value={"http": 401, "code": 40100, "message": "unauthorized"}
+        ) as request:
+            main.probe()
+
+        self.assertEqual(request.call_count, 1)
+        self.assertEqual(request.call_args.args[0], "GET")
+        self.assertEqual(main._snapshot()["phase"], "auth_failed")
+
+    def test_submit_mode_calls_task_add_once(self):
+        responses = [
+            {"http": 200, "code": 0, "data": {}},
+            {"http": 200, "code": 0, "data": "framework: vllm\n"},
+            {"http": 200, "code": 0, "data": {"taskId": "task-1"}},
+        ]
+        with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
+            main, "ALLOW_SUBMIT", True
+        ), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
+            main, "_req", side_effect=responses
+        ) as request:
+            main.probe()
+
+        methods_and_paths = [(call.args[0], call.args[1]) for call in request.call_args_list]
+        self.assertEqual(sum(path == "/api/adapt/task/add" for _, path in methods_and_paths), 1)
+        state = main._snapshot()
+        self.assertEqual(state["phase"], "submit_complete")
+        self.assertEqual(state["results"]["task_add"]["task_id"], "task-1")
+
+    def test_failed_build_config_prevents_task_add(self):
+        responses = [
+            {"http": 200, "code": 0, "data": {}},
+            {"http": 500, "code": 50000, "message": "build failed"},
+        ]
+        with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
+            main, "ALLOW_SUBMIT", True
+        ), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
+            main, "_req", side_effect=responses
+        ) as request:
+            main.probe()
+
+        self.assertEqual(request.call_count, 2)
+        self.assertNotIn("/api/adapt/task/add", [call.args[1] for call in request.call_args_list])
+        self.assertEqual(main._snapshot()["phase"], "build_config_failed")
+
+
+if __name__ == "__main__":
+    unittest.main()