Files
Agent-System-Dev/tests/test_svc_recover.py
T

555 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""svc.sh 自恢复回归:autostart 期望态 + recover + watchdog(2026-10-11)。
验证崩溃(SIGKILL,模拟 OOM 被杀)后 recover 按持久化登记拉回服务、显式 stop
不被复活、并发生命周期锁防抢跑、watchdog 单元生成正确。全部使用临时目录 +
桩二进制(bash + python http.server 模拟 agent-system;HTTP 子进程随父进程
退出——真实产物是单进程二进制,被杀即端口释放,桩须对齐该语义)。
"""
import fcntl
import os
import shutil
import socket
import subprocess
import tempfile
import time
import unittest
from pathlib import Path
SOURCE = Path(__file__).resolve().parents[1]
SVC_SCRIPT = SOURCE / "scripts" / "svc.sh"
# 桩 agent-system:version / config explain(环境层 DSN 感知)/ serve(记录参数后
# 用 http.server 应答健康探测)。TERM 转发给子进程保证 stop 干净;bash 自身不
# exec——pidcheck(二进制路径)须留在 cmdline。HTTP 子进程轮询 ppid:父进程被
# SIGKILL 后 0.5s 内自行退出(对齐单进程产物“进程死=端口释放”的语义)。
STUB = r"""#!/usr/bin/env bash
set -u
trace_dir="${SVC_STUB_TRACE:-$PWD/.svc-stub-trace}"
mkdir -p "$trace_dir"
case "${1:-}" in
version)
if [ "${2:-}" = "--json" ]; then
printf '%s\n' '{"protocol":1,"product":"agent-system","version":"stub","identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},"cli":{"unknownSubcommand":"reject"}}'
exit 0
fi
echo "stub-agent-system 1.0"
exit 0 ;;
config)
if [ "${2:-}" = "explain" ]; then
echo "config file: (none)"
echo "field value source"
if [ -n "${AGENT_SYSTEM_DSN:-}" ]; then
printf 'dsn postgresql://%s env\n' "$AGENT_SYSTEM_DSN"
else
echo "dsn - default"
fi
fi
exit 0 ;;
identity)
# U1 I3:与真实二进制同表契约(SQLite);PG DSN 在桩上拒绝(真 PG 专项另跑)。
shift
sub="${1:-}"; shift || true
iid=""; dbref=""; data="./data"; dsn=""
while [ $# -gt 0 ]; do
case "$1" in
--instance-id) iid="$2"; shift 2 ;;
--dbref) dbref="$2"; shift 2 ;;
--data) data="$2"; shift 2 ;;
--dsn) dsn="$2"; shift 2 ;;
-h|--help)
echo "用法:agent-system identity <ensure|verify> --instance-id ID [--dbref REF] [--data DIR] [--dsn DSN]"
exit 0 ;;
*) shift ;;
esac
done
if [ -z "$sub" ] || [ -z "$iid" ]; then
echo "用法:agent-system identity <ensure|verify> --instance-id ID [--dbref REF] [--data DIR] [--dsn DSN]" >&2
exit 2
fi
if [ -n "$dsn" ]; then
echo "identity: stub 不支持 --dsn(真 PG 专项用真实二进制)" >&2
exit 1
fi
[ -n "$dbref" ] || dbref=sqlite
mkdir -p "$data"
python3 - "$sub" "$iid" "$dbref" "$data" <<'PYID' || exit $?
import sqlite3, sys, time
sub, iid, dbref, data = sys.argv[1:5]
path = data.rstrip("/") + "/agent-system.db"
con = sqlite3.connect(path)
ddl = (
"CREATE TABLE IF NOT EXISTS agent_system_identity ("
"instance_id TEXT NOT NULL PRIMARY KEY,"
"dbref TEXT NOT NULL,"
"schema_version INTEGER NOT NULL,"
"created_at TEXT NOT NULL)"
)
con.execute(ddl)
rows = list(con.execute("SELECT instance_id, dbref FROM agent_system_identity"))
if sub == "ensure":
if len(rows) > 1:
print("identity ensure: too many rows", file=sys.stderr); sys.exit(1)
if len(rows) == 1:
if rows[0][0] != iid or rows[0][1] != dbref:
print("identity ensure: mismatch", file=sys.stderr); sys.exit(1)
print("ensured instance_id=%s dbref=%s" % (iid, dbref)); sys.exit(0)
con.execute(
"INSERT INTO agent_system_identity(instance_id,dbref,schema_version,created_at) VALUES(?,?,?,?)",
(iid, dbref, 26, time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())))
con.commit()
print("ensured instance_id=%s dbref=%s" % (iid, dbref)); sys.exit(0)
if sub == "verify":
if not rows:
print("identity verify: missing", file=sys.stderr); sys.exit(1)
if len(rows) > 1 or rows[0][0] != iid or rows[0][1] != dbref:
print("identity verify: mismatch", file=sys.stderr); sys.exit(1)
print("verified instance_id=%s dbref=%s" % (iid, dbref)); sys.exit(0)
print("identity: unknown subcommand", sub, file=sys.stderr); sys.exit(2)
PYID
exit $? ;;
serve)
shift
case "${1:-}" in
-h|--help)
echo "Usage: agent-system serve -addr H:P -data DIR -plugins DIR -auth MODE -config FILE"
exit 0 ;;
esac
printf 'args: %s\n' "$*" > "$trace_dir/serve.txt"
# 身份环境记录(U1 批量隔离回归观测点):serve 实际拿到谁的实例身份。
if [ -n "${AGENT_SYSTEM_INSTANCE_ID:-}" ]; then
printf 'AGENT_SYSTEM_INSTANCE_ID=%s\nAGENT_SYSTEM_DBREF=%s\n' \
"$AGENT_SYSTEM_INSTANCE_ID" "${AGENT_SYSTEM_DBREF:-sqlite}" >> "$trace_dir/serve.txt"
fi
addr=""; data="./data"
while [ $# -gt 0 ]; do
case "$1" in --addr) addr="$2"; shift 2 ;; --data) data="$2"; shift 2 ;; *) shift ;; esac
done
# serve 二次门禁:与真实二进制 AGENT_SYSTEM_INSTANCE_ID 对齐。
if [ -n "${AGENT_SYSTEM_INSTANCE_ID:-}" ]; then
"$0" identity verify --instance-id "$AGENT_SYSTEM_INSTANCE_ID" \
--dbref "${AGENT_SYSTEM_DBREF:-sqlite}" --data "$data" || exit 1
fi
port="${addr##*:}"; host="${addr%:*}"
trap 'kill "$child" 2>/dev/null; exit 0' TERM INT
python3 - "$port" "$host" <<'PYSRV' >/dev/null 2>&1 &
import os, sys, threading, time
from http.server import BaseHTTPRequestHandler, HTTPServer
port, host = int(sys.argv[1]), sys.argv[2]
class H(BaseHTTPRequestHandler):
def _ok(self, body=b'{"ok":true}'):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
if self.command != 'HEAD':
self.wfile.write(body)
def do_GET(self):
self._ok()
def do_HEAD(self):
self._ok(b'')
def do_PUT(self):
n = int(self.headers.get('Content-Length') or 0)
self.rfile.read(n)
self._ok()
def log_message(self, *a):
pass
ppid = os.getppid()
def watch():
while True:
try:
if os.getppid() != ppid:
os._exit(0)
except Exception:
os._exit(0)
time.sleep(0.3)
threading.Thread(target=watch, daemon=True).start()
try:
HTTPServer((host, port), H).serve_forever()
except Exception:
os._exit(1)
PYSRV
child=$!
wait "$child"
;;
user)
printf 'user %s\n' "$*" >> "$trace_dir/cli.txt"
case "${2:-}" in
list) echo '[{"owner":"op","name":"op","isAdmin":true,"mustChangePassword":false}]' ;;
create) echo '{"owner":"new","issued":true}' ;;
esac
exit 0 ;;
*)
exit 0 ;;
esac
"""
def free_fixture_port():
with socket.socket() as reservation:
reservation.bind(("127.0.0.1", 0))
return reservation.getsockname()[1]
class SvcAutoRecoverTests(unittest.TestCase):
def setUp(self):
tmp = tempfile.mkdtemp(prefix="svc-autorecover-")
self.addCleanup(shutil.rmtree, tmp, True)
self.root = Path(tmp)
(self.root / "scripts").mkdir(parents=True)
shutil.copy2(SVC_SCRIPT, self.root / "scripts" / "svc.sh")
self.trace = self.root / "trace"
self.stub = self.root / "stub-agent-system"
self.stub.write_text(STUB)
self.stub.chmod(0o755)
(self.root / "plugins").mkdir()
self.env = os.environ.copy()
self.env.update({
"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
"GIT_TERMINAL_PROMPT": "0",
"GIT_AUTHOR_NAME": "svc recover", "GIT_COMMITTER_NAME": "svc recover",
"GIT_AUTHOR_EMAIL": "svc-test@example.invalid",
"GIT_COMMITTER_EMAIL": "svc-test@example.invalid",
"SVC_STUB_TRACE": str(self.trace),
"SVC_SRC_DIR": str(self.root / "srccache"),
# 隔离 watchdog 单元目录(--print 不落盘,但断言要在受控目录下做)。
"XDG_CONFIG_HOME": str(self.root / "xdg"),
})
for key in list(self.env):
if key.startswith("SVC_") and key not in ("SVC_STUB_TRACE", "SVC_SRC_DIR"):
self.env.pop(key)
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE"):
self.env.pop(key, None)
def tearDown(self):
# 兜底清理:停掉本测试遗留的服务进程,避免端口跨用例泄漏。
runtime = self.root / "runtime"
if runtime.is_dir():
for d in sorted(runtime.iterdir()):
if (d / "svc.pid").exists():
try:
subprocess.run(
["bash", str(self.root / "scripts" / "svc.sh"), d.name, "stop"],
cwd=self.root, env=self.env, timeout=30,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except subprocess.TimeoutExpired:
pass # 兜底强杀(见下)
# 直接 SIGKILL 兜底:确保桩进程(及其锁 fd/端口)不跨用例泄漏。
try:
pidf = d / "svc.pid"
if pidf.exists():
pid = int(pidf.read_text().strip() or 0)
if pid:
os.kill(pid, 9)
except (ValueError, ProcessLookupError, PermissionError, FileNotFoundError):
pass
# ---- 夹具 ----
def svc(self, *args, check=True, env=None):
proc = subprocess.run(
["bash", str(self.root / "scripts" / "svc.sh"), *map(str, args)],
cwd=self.root, env=env if env is not None else self.env, text=True,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=120)
if check:
self.assertEqual(proc.returncode, 0, proc.stdout)
return proc
def runtime(self, name):
return self.root / "runtime" / name
def kv(self, name, key):
envfile = self.runtime(name) / "svc.env"
if not envfile.exists():
return None
for line in envfile.read_text().splitlines():
if line.startswith(key + "="):
return line[len(key) + 1:]
return None
def seed_agent(self, name, port):
d = self.runtime(name)
(d / "bin").mkdir(parents=True, exist_ok=True)
(d / "data").mkdir(exist_ok=True)
(d / "plugins").mkdir(exist_ok=True)
shutil.copy2(self.stub, d / "bin" / "agent-system")
(d / "svc.env").write_text("".join("%s=%s\n" % kv for kv in [
("kind", "agent"), ("env", name), ("pid", ""),
("addr", "127.0.0.1:%d" % port), ("tags", ""),
("source", "local@deadbee"), ("version", "stub"), ("started", ""),
("pidcheck", str(d / "bin" / "agent-system")),
("exec", "--addr 127.0.0.1:%d --data %s --plugins %s"
% (port, d / "data", d / "plugins")),
]))
def start_ok(self, name):
result = self.svc(name, "start")
self.assertIn("已启动", result.stdout)
return result
def pid_of(self, name):
pidf = self.runtime(name) / "svc.pid"
if not pidf.exists():
return None
text = pidf.read_text().strip()
return int(text) if text else None
def pid_alive(self, pid):
if not pid:
return False
try:
os.kill(pid, 0)
except (ProcessLookupError, PermissionError):
return False
# SIGKILL 后的孤儿在 CI 容器里可能成僵尸(PID 1 不收割)——僵尸按死处理,
# 与 svc.sh pid_alive_mark 的口径一致(僵尸 cmdline 为空 = 不匹配 = 已死)。
try:
state = Path("/proc/%d/stat" % pid).read_text().rsplit(")", 1)[1].split()[0]
if state == "Z":
return False
except (OSError, IndexError):
pass
return True
def kill9(self, name):
"""SIGKILL 服务主进程(模拟 OOM killer),保留 pidfile 与登记。"""
pid = self.pid_of(name)
self.assertTrue(self.pid_alive(pid), "服务应先在运行")
os.kill(pid, 9)
for _ in range(100): # 等 bash 主进程消失(HTTP 子进程由 ppid 看护自杀)
if not self.pid_alive(pid):
break
time.sleep(0.05)
self.assertFalse(self.pid_alive(pid))
time.sleep(0.6)
return pid
def history_events(self, name, event):
f = self.runtime(name) / "history.jsonl"
if not f.exists():
return []
out = []
for line in f.read_text().splitlines():
if ('"event":"%s"' % event) in line:
out.append(line)
return out
# ---- 1. 期望态:start 成功置位 / 显式 stop 清除(含进程已死的 stop)----
def test_start_records_autostart_and_stop_clears(self):
self.seed_agent("dev", free_fixture_port())
self.start_ok("dev")
self.assertEqual(self.kv("dev", "autostart"), "1")
self.svc("dev", "stop")
self.assertIsNone(self.kv("dev", "autostart"))
def test_stop_on_dead_process_clears_autostart(self):
self.seed_agent("dev", free_fixture_port())
self.start_ok("dev")
self.kill9("dev")
# 进程已死但 autostart 保留(等待 recover);显式 stop 表示"保持停止"。
self.assertEqual(self.kv("dev", "autostart"), "1")
self.svc("dev", "stop")
self.assertIsNone(self.kv("dev", "autostart"))
# ---- 2. recover:崩溃拉回 / 运行中不动 / 显式停止不复活 / 未启动不碰 ----
def test_recover_restarts_sigkilled_service(self):
port = free_fixture_port()
self.seed_agent("dev", port)
self.start_ok("dev")
old_pid = self.kill9("dev")
result = self.svc("recover", "--quiet")
self.assertEqual(result.returncode, 0, result.stdout)
self.assertIn("recover 汇总:拉回 1", result.stdout) # c_info 在 --quiet 下仍输出
new_pid = self.pid_of("dev")
self.assertTrue(self.pid_alive(new_pid))
self.assertNotEqual(new_pid, old_pid)
self.assertEqual(self.kv("dev", "autostart"), "1")
ok_events = self.history_events("dev", "recover")
self.assertTrue(any('"result":"ok"' in l for l in ok_events), ok_events)
# 恢复用的仍是持久化 exec=(svc.env 登记未漂移)。
self.assertIn("--addr 127.0.0.1:%d" % port, self.kv("dev", "exec"))
def test_recover_noop_while_running(self):
self.seed_agent("dev", free_fixture_port())
self.start_ok("dev")
pid = self.pid_of("dev")
result = self.svc("recover")
self.assertEqual(result.returncode, 0, result.stdout)
self.assertEqual(self.pid_of("dev"), pid) # 未重启
self.assertEqual(self.history_events("dev", "recover"), [])
def test_recover_respects_explicit_stop(self):
self.seed_agent("dev", free_fixture_port())
self.start_ok("dev")
self.svc("dev", "stop")
result = self.svc("recover")
self.assertEqual(result.returncode, 0, result.stdout)
self.assertFalse(self.pid_alive(self.pid_of("dev")))
self.assertEqual(self.history_events("dev", "recover"), [])
def test_recover_ignores_never_started_service(self):
self.seed_agent("dev", free_fixture_port()) # 从未 start:无 autostart 键
result = self.svc("recover")
self.assertEqual(result.returncode, 0, result.stdout)
self.assertFalse(self.pid_alive(self.pid_of("dev")))
self.assertEqual(self.history_events("dev", "recover"), [])
# ---- 3. 并发防护:生命周期锁被持有时 recover 跳过 ----
def test_recover_skips_when_lock_held(self):
self.seed_agent("dev", free_fixture_port())
self.start_ok("dev")
self.kill9("dev")
lock = self.runtime("dev") / ".start.lock"
with open(lock, "w") as fh:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
result = self.svc("recover")
self.assertEqual(result.returncode, 0, result.stdout)
self.assertIn("锁被占用", result.stdout)
self.assertFalse(self.pid_alive(self.pid_of("dev")))
self.assertEqual(self.history_events("dev", "recover"), [])
fcntl.flock(fh, fcntl.LOCK_UN)
# 锁释放后同一命令即可拉回。
result = self.svc("recover")
self.assertEqual(result.returncode, 0, result.stdout)
self.assertTrue(self.pid_alive(self.pid_of("dev")))
# ---- 4. 恢复失败路径:端口被外部进程占用 → fail 事件 + 非零退出 ----
def test_recover_fails_when_port_stolen(self):
port = free_fixture_port()
self.seed_agent("dev", port)
self.start_ok("dev")
self.kill9("dev")
thief = subprocess.Popen(
["python3", "-m", "http.server", str(port), "--bind", "127.0.0.1"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.addCleanup(thief.terminate)
# 等 thief 真正完成 bind 再 recover——慢环境里 Popen 返回时端口尚未监听,
# recover 会抢先起服务,把“端口被抢”测成竞态(CI 首轮实测)。
deadline = time.monotonic() + 5.0
bound = False
while time.monotonic() < deadline:
s = socket.socket()
s.settimeout(0.2)
try:
s.connect(("127.0.0.1", port))
bound = True
except OSError:
pass
finally:
s.close()
if bound:
break
time.sleep(0.05)
self.assertTrue(bound, "thief 未能在 5s 内监听 18808")
try:
result = self.svc("recover", check=False)
self.assertEqual(result.returncode, 1, result.stdout)
fail_events = self.history_events("dev", "recover")
self.assertTrue(any('"result":"fail"' in l for l in fail_events), fail_events)
self.assertFalse(self.pid_alive(self.pid_of("dev"))) # 未强抢端口
finally:
thief.terminate()
thief.wait(timeout=10)
# ---- 4b. 本服务孤儿占口(无 pidfile)→ start 回收后成功;外部占口仍拒 ----
def _wait_port(self, port, timeout=5.0):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
s = socket.socket()
s.settimeout(0.2)
try:
s.connect(("127.0.0.1", port))
return True
except OSError:
time.sleep(0.05)
finally:
s.close()
return False
def _ss_can_attribute_pid(self):
"""reclaim_own_port /「非本服务」归因依赖 ss -ltnp 的 pid= 字段。
CI 容器常无 iproute2 或 ss 不带进程信息——此时只能探测占用、不能归因,
相关用例跳过(拒绝启动的兜底路径仍由「未能识别占用者」覆盖)。
"""
try:
out = subprocess.run(
["ss", "-ltnp"], capture_output=True, text=True, timeout=5)
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
return False
return out.returncode == 0 and "pid=" in out.stdout
def test_start_reclaims_own_orphan_without_pidfile(self):
"""模拟 update 误清 pidfile 后本 bin 仍占口:裸 start 应回收并拉起新进程。"""
if not self._ss_can_attribute_pid():
self.skipTest("ss -ltnp 无法归因 PID(reclaim_own_port 需此能力)")
port = free_fixture_port()
self.seed_agent("dev", port)
self.start_ok("dev")
old = self.pid_of("dev")
self.assertTrue(self.pid_alive(old))
# 只删 pidfile、不杀进程——对齐 2026-10-12 genui 现场(list=未运行、端口仍监听)。
(self.runtime("dev") / "svc.pid").unlink()
self.assertIsNone(self.pid_of("dev"))
self.assertTrue(self.pid_alive(old))
self.assertTrue(self._wait_port(port))
result = self.start_ok("dev")
self.assertIn("残留", result.stdout)
new = self.pid_of("dev")
self.assertTrue(self.pid_alive(new), result.stdout)
self.assertNotEqual(new, old)
self.assertFalse(self.pid_alive(old))
stop_events = self.history_events("dev", "stop")
self.assertTrue(any("reclaim_own_port" in l for l in stop_events), stop_events)
def test_start_rejects_foreign_port_occupant(self):
"""外部 python http.server 占口:start 必须失败并提示非本服务。"""
port = free_fixture_port()
self.seed_agent("dev", port)
thief = subprocess.Popen(
["python3", "-m", "http.server", str(port), "--bind", "127.0.0.1"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.addCleanup(thief.terminate)
self.assertTrue(self._wait_port(port), "thief 未能监听")
try:
result = self.svc("dev", "start", check=False)
self.assertEqual(result.returncode, 1, result.stdout)
# 有 ss 时归因到外部进程 →「非本服务」;无 ss 时仅能探测 →「未能识别占用者」。
# 两种路径都拒绝启动,不拉起本服务。
self.assertTrue(
("非本服务" in result.stdout) or ("未能识别占用者" in result.stdout),
result.stdout)
self.assertFalse(self.pid_alive(self.pid_of("dev")))
finally:
thief.terminate()
thief.wait(timeout=10)
# ---- 5. watchdog:单元文件生成(--print 走 main 分发全链路)----
def test_watchdog_print_plan(self):
result = self.svc("watchdog", "install", "--print")
self.assertIn(
"ExecStart=%s/scripts/svc.sh recover --quiet" % self.root, result.stdout)
self.assertIn("OnUnitActiveSec=60s", result.stdout)
self.assertIn("OnBootSec=1min", result.stdout)
self.assertIn("WantedBy=timers.target", result.stdout)
# oneshot 结束不得连带杀掉刚拉起的服务(cgroup 陷阱)。
self.assertIn("KillMode=process", result.stdout)
# 只打印,不落盘(受控 XDG_CONFIG_HOME 下无单元文件)。
unit_dir = self.root / "xdg" / "systemd" / "user"
self.assertFalse((unit_dir / "agent-system-svc-recover.timer").exists())
self.assertFalse((unit_dir / "agent-system-svc-recover.service").exists())
def test_recover_reserved_name_rejected(self):
self.seed_agent("recover", free_fixture_port()) # 保留字不可作服务名
result = self.svc("recover", "recover", check=False)
self.assertEqual(result.returncode, 2, result.stdout)
if __name__ == "__main__":
unittest.main()