mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-18 13:42:36 +08:00
infstellar
3df9f7878c
Add authenticated in-container execution, shared runtime activity summaries, strict result-file validation, and OpenCode integration tests. 引入带认证的容器内执行器、共享 runtime 活动摘要、严格结果文件校验和 OpenCode 集成测试。
39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
"""Agent executor 容器内的无泄密健康探针。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from executor.token import ExecutorTokenError, read_token_file
|
|
|
|
|
|
def main() -> int:
|
|
"""读取共享 token 并请求本地健康接口，返回 Compose 探针退出码。
|
|
|
|
探针只输出错误码对应的退出状态，不打印 token 或响应中的敏感字段；调用
|
|
场景是 Agent 容器 healthcheck 和运维脚本的运行时验证。
|
|
"""
|
|
token_path = os.getenv("MEMEMEOW_AGENT_EXECUTOR_TOKEN_FILE", "")
|
|
try:
|
|
token = read_token_file(Path(token_path)) if token_path else os.getenv("MEMEMEOW_AGENT_EXECUTOR_TOKEN", "")
|
|
if not token:
|
|
return 1
|
|
request = urllib.request.Request(
|
|
"http://127.0.0.1:8277/health",
|
|
headers={"Accept": "application/json", "Authorization": f"Bearer {token}"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
payload = json.load(response)
|
|
return 0 if payload.get("ready") is True and payload.get("docker_socket_absent") is True else 1
|
|
except (ExecutorTokenError, OSError, ValueError, urllib.error.URLError, json.JSONDecodeError):
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|