harden probe agent v1.0.3

安全鉴权、默认只读、显式提交开关、状态接口与测试
probe v1.0.2 container submit test
2026-08-26 17:16:02 +08:00 · 2026-08-24 13:13:11 +08:00
4 changed files with 368 additions and 101 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -0,0 +1,2 @@
+__pycache__/
+*.py[cod]
--- a/README.md
+++ b/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` → 运行/停止」整条链路是否通。
-2. 零副作用观测平台注入的 `STRATEGY_ID` / `EXTERNAL_SERVICE_TOKEN` 是否为有效凭证：
-   启动时用注入 token 调一次**只读** `GET /api/adapt/task/page`，看返回 `code`：
-   `0` = 凭证有效；`40100` = 未登录/凭证无效。
+- 默认只调用一次只读接口 `GET /api/adapt/task/page`。
+- 空凭证和 `tmp`、`placeholder` 等占位值会被拒绝，不会继续请求。
+- 任何日志和 HTTP 状态响应都不包含 token 值或长度。
+- 只有 `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 值、不落盘任何凭证；探针仅调只读 GET，仅记录 `http/code/计数`。
+不要把 token 写进代码、Dockerfile、Git 仓库或智能体名称。平台侧应通过 Kubernetes Secret 注入真实凭证；如果仍注入 `tmp`，探针会进入 `blocked_missing_auth`，不会误提交。

-## 本地运行
+## 运行时变量
+
+| 变量 | 默认值 | 说明 |
+| --- | --- | --- |
+| `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
-curl http://localhost:8080/health     # {"status":"ok"}
-curl http://localhost:8080/           # 查看探针结果
+python -m unittest -v
+
+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`；
-平台 Kaniko 按 `refs/tags/<tag>` 解析，用分支名会失败）。再在「我的适配智能体 → 新增智能体」
-填 名称 / 仓库地址 / 标签 提交。
+推到 `https://dev.modelhub.org.cn/<user>/<repo>` 并创建真实 Git 标签（例如 `v1.0.3`）。在「我的适配智能体 → 新增智能体」填写名称、仓库地址、标签。平台 Kaniko 按 `refs/tags/<tag>` 构建，不能用分支名代替标签。
+
+建议先保持 `ALLOW_SUBMIT=0`，确认 `/status` 为 `read_only_complete`；再由平台管理员把它改为 `1`，仅跑一次验证任务，验证后立即恢复为 `0`。
--- a/main.py
+++ b/main.py
@@ -1,18 +1,10 @@
-"""ModelHub XC 适配智能体 · 只读探针骨架 (probe-only) v1.0.1
+"""ModelHub XC 适配智能体安全探针 v1.0.3。

-用途:
-  1) 验证平台「建仓 -> Kaniko 构建 -> 部署 -> /health -> 运行/停止」整条链路。
-  2) 零副作用诊断平台注入的凭证:
-     - dump 所有环境变量的「名字 + 值长度」(不打值),看真 token 是否藏在别的变量名里。
-     - 打印 EXTERNAL_SERVICE_TOKEN 的值(短值必是废值、直接看;长值打码,避免泄露真 token)。
-     - 对每个「长得像 token」(len>=16) 的环境变量,试当 Xc-Token 调一次只读接口,看哪个能通(code=0)。
-     - 仍用 EXTERNAL_SERVICE_TOKEN 三种头各探一次,作对照。
-
-严格约束:
-  - 只服务 GET /health 与 GET /,绝不提交任何验证任务 (不调 task/add, 不调 build-config)。
-  - 只读 GET;只记录 http/code/计数,不打印任何接口返回体内容(防 PII);不打印任何长凭证的完整值。
-  - 纯 stdlib 零依赖;正确处理 SIGTERM 优雅停机。
+默认只做一次只读鉴权检查。只有显式设置 ALLOW_SUBMIT=1，且鉴权检查成功后，
+才会调用 build-config 和 task/add；每个进程生命周期最多提交一次。
 """
+
+import copy
 import json
 import os
 import signal
@@ -21,107 +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", "")
-STRATEGY_ID = os.getenv("STRATEGY_ID", "")
+
+VERSION = "1.0.3"
+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",
+)
+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:
-    """短值(必是废值)原样看;长值只留头尾,避免泄露可能的真 token。"""
-    return v if len(v) <= 8 else (v[:2] + "…" + v[-2:])
+def _set_state(**changes) -> None:
+    with STATE_LOCK:
+        STATE.update(changes)


-def _readonly_get(path: str, headers: dict, timeout: int = 20) -> dict:
-    req = urllib.request.Request(MAIN + path, method="GET", headers=headers)
+def _snapshot() -> dict:
+    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:
-            body = r.read().decode("utf-8", "ignore")
-            status = r.status
-    except urllib.error.HTTPError as e:
-        body = e.read().decode("utf-8", "ignore")
-        status = e.code
-    except Exception as e:  # noqa: BLE001
-        return {"http": -1, "err": str(e)[:120]}
-    code = count = None
+        with urllib.request.urlopen(req, timeout=timeout) as response:
+            text = response.read().decode("utf-8", "ignore")
+            status = response.status
+    except urllib.error.HTTPError as error:
+        text = error.read().decode("utf-8", "ignore")
+        status = error.code
+    except Exception as error:  # noqa: BLE001 - network diagnostics are returned safely
+        return {"http": -1, "error": type(error).__name__}
+
    try:
-        j = json.loads(body)
-        code = j.get("code")
-        d = j.get("data")
-        if isinstance(d, dict):
-            count = d.get("total")
-            if count is None and isinstance(d.get("records"), list):
-                count = len(d["records"])
-        elif isinstance(d, list):
-            count = len(d)
-    except Exception:  # noqa: BLE001
-        pass
-    return {"http": status, "code": code, "count": count}
+        payload = json.loads(text)
+    except (TypeError, ValueError):
+        return {"http": status, "code": None, "message": "non_json_response"}
+    return {
+        "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:
-    # 1) dump 所有 env 的 key + 值长度(不打值),找隐藏的真 token
-    env_lens = {k: len(str(os.environ.get(k, ""))) for k in sorted(os.environ)}
-    log("env keys+lens: " + json.dumps(env_lens, ensure_ascii=False))
+    """Run one guarded probe; no retry loop and at most one task/add call."""
+    token, source, token_error = _resolve_token()
+    _set_state(token_source=source, submit_enabled=ALLOW_SUBMIT)

-    # 2) EXTERNAL_SERVICE_TOKEN 的值(短则全打,长则打码)+ STRATEGY_ID
-    log(f"strategy_id={STRATEGY_ID or 'MISSING'} | "
-        f"EXTERNAL_SERVICE_TOKEN value={_mask(TOKEN)!r} len={len(TOKEN)}")
+    if not STRATEGY_ID:
+        _set_state(phase="blocked_missing_strategy_id", done=True)
+        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

-    path = "/api/adapt/task/page?current=1&pageSize=1"
+    _set_state(phase="checking_auth")
+    page = _req("GET", "/api/adapt/task/page?current=1&pageSize=1", token)
+    results = {"task_page": _result_summary(page)}
+    if not _api_ok(page):
+        _set_state(phase="auth_failed", done=True, results=results)
+        log("read-only auth check failed: " + json.dumps(results["task_page"], ensure_ascii=False))
+        return

-    # 3) 对每个「长得像 token」(len>=16) 的 env 值,试当 Xc-Token,看哪个能通
-    candidates = [k for k in sorted(os.environ) if len(str(os.environ.get(k, ""))) >= 16]
-    log("token-like env candidates (len>=16): " + json.dumps(candidates, ensure_ascii=False))
-    for k in candidates:
-        res = _readonly_get(path, {"Xc-Token": os.environ[k]})
-        PROBE["results"].append({"try_env": k, "as": "Xc-Token", **res})
-        log(f"try Xc-Token from ${k} -> {json.dumps(res, ensure_ascii=False)}")
+    _set_state(auth_ready=True, results=results)
+    log(f"read-only auth check passed (source={source})")
+    if not ALLOW_SUBMIT:
+        _set_state(phase="read_only_complete", done=True)
+        log("probe complete in read-only mode; task/add was not called")
+        return

-    # 4) EXTERNAL_SERVICE_TOKEN 三种头对照
-    trials = [
-        ("Xc-Token", {"Xc-Token": TOKEN}),
-        ("Authorization-Bearer", {"Authorization": "Bearer " + TOKEN}),
-        ("Authorization-raw", {"Authorization": TOKEN}),
-    ]
-    for name, headers in trials:
-        res = {"skip": "no-token"} if not TOKEN else _readonly_get(path, headers)
-        PROBE["results"].append({"est_token_hdr": name, **res})
-        log(f"EST-token {name} -> {json.dumps(res, ensure_ascii=False)}")
+    _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

-    PROBE["done"] = True
-    log("probe done | 判读: 任一 code=0 => 该来源凭证有效; 全 401/40100 => 无有效凭证")
+    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 == "/":
-            self._json({"name": "huni-probe-agent", "mode": "probe-only-v1.0.1",
-                        "strategy_id_present": bool(STRATEGY_ID),
-                        "token_present": bool(TOKEN), "probe": PROBE})
+        if self.path in {"/", "/status"}:
+            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


@@ -137,7 +256,7 @@ def main() -> None:

    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    server.timeout = 1
-    log(f"probe-only agent v1.0.1 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()
--- a/test_main.py
+++ b/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()
Author	SHA1	Message	Date
huni	f18eeaf334	harden probe agent v1.0.3 安全鉴权、默认只读、显式提交开关、状态接口与测试	2026-08-26 17:16:02 +08:00
huni	5db73110a5	probe v1.0.2 container submit test	2026-08-24 13:13:11 +08:00