"""真实 Node worker、HTTP 网关与官方 MCP SDK 的控制面验收。"""
from __future__ import annotations

import asyncio
import io
import gzip
from functools import lru_cache
import json
import logging
import os
import socket
import sys
import tarfile
import tempfile
import unittest
from pathlib import Path
from dataclasses import replace
from unittest.mock import patch

import httpx
import uvicorn
from fastapi.testclient import TestClient
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "apps/market-api"))
from app.native.application import create_app
from app.native.config import Settings
from app.native.supervisor import loopback_http_client, Supervisor

ADMIN = "fixture-admin-token-not-a-secret"
TOKEN_A = "fixture-profile-a-token-not-a-secret"
TOKEN_B = "fixture-profile-b-token-not-a-secret"


def archive(files):
    stream = io.BytesIO()
    with tarfile.open(fileobj=stream, mode="w") as tar:
        for name, data in files.items():
            raw = data.encode() if isinstance(data, str) else data
            info = tarfile.TarInfo("package/" + name)
            info.size = len(raw)
            tar.addfile(info, io.BytesIO(raw))
    return gzip.compress(stream.getvalue(), mtime=0)


@lru_cache
def local_bundle(name):
    directory = ROOT / "plugins" / name
    return archive({str(path.relative_to(directory)): path.read_bytes() for path in directory.rglob("*") if path.is_file() and "node_modules" not in path.parts and "tests" not in path.parts})


def fixture(version="1.0.0", value="v1"):
    manifest = dict(name="@message/api-fixture", version=version, type="module", exports="./index.mjs", peerDependencies={"@deepseek-ai/dsh-tools": "0.2.0-rc.2", "@deepseek-ai/cordis": "4.0.4"}, dsh={"bundle": {"patch": "./cordis.patch.yml"}})
    source = """import { defineTool } from '@deepseek-ai/dsh-tools';
export const inject=['tools'];
export function apply(ctx){
 ctx.tools.register(defineTool({name:'api_fixture',description:'验收 fixture',parameters:{},output:{schema:{type:'string'},render:(_a,v)=>[{type:'text',text:v}]},async execute(){return VALUE;}}));
}
""".replace("VALUE", json.dumps(value))
    return archive({"package.json": json.dumps(manifest), "index.mjs": source, "cordis.patch.yml": '- insert:\n    - id: api-fixture\n      name: "@message/api-fixture"\n'})


def diagnostic_fixture(mode):
    name = "@message/diagnostic-" + mode
    manifest = dict(name=name, version="1.0.0", type="module", exports="./index.mjs", peerDependencies={"@deepseek-ai/dsh-tools": "0.2.0-rc.2", "@deepseek-ai/cordis": "4.0.4"}, dsh={"bundle": {"patch": "./cordis.patch.yml"}})
    source = r"""
import { writeSync } from 'node:fs';
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
async function writePieces(stream, text) {
  // 原生写入绕过 CLI write 过滤，独立验证 API 的第二层保护。
  writeSync(stream.fd, text.slice(0, 5)); await sleep(40); writeSync(stream.fd, text.slice(5) + '\n');
}
export async function apply() {
  const fail = async () => {
    await writePieces(process.stderr, process.env.SERVICE_A_TOKEN);
    for (const token of PLATFORM_TOKENS) await writePieces(process.stdout, token);
    process.exit(23);
  };
  if (LATE) setTimeout(fail, 2000); else await fail();
}
""".replace("PLATFORM_TOKENS", json.dumps([ADMIN, TOKEN_A, TOKEN_B])).replace("LATE", "true" if mode == "late" else "false")
    return archive({"package.json": json.dumps(manifest), "index.mjs": source, "cordis.patch.yml": f'- insert:\n    - id: diagnostic-{mode}\n      name: "{name}"\n'})


class ConfigurationTests(unittest.TestCase):
    def test_fail_closed_and_outside_repository(self):
        common = dict(data_dir=Path("/tmp/native-config-test"), admin_token=ADMIN, mcp_tokens={"a": TOKEN_A}, allowed_hosts=("localhost",), allowed_origins=("http://localhost",))
        Settings(**common).validate()
        for changed in ({"admin_token": ""}, {"mcp_tokens": {"a": ADMIN}}, {"data_dir": ROOT / "profiles/test"}, {"allowed_hosts": ("*",)}, {"allowed_origins": ("null",)}):
            with self.subTest(changed=changed), self.assertRaises(ValueError):
                Settings(**(common | changed)).validate()

    def test_worker_environment_does_not_inherit_unrelated_credentials(self):
        settings = Settings(data_dir=Path("/tmp/native-config-test"), admin_token=ADMIN, mcp_tokens={}, allowed_hosts=("localhost",), allowed_origins=("http://localhost",))
        with patch.dict(os.environ, {"MESSAGE_ADMIN_TOKEN": ADMIN, "OPENAI_API_KEY": "fixture-must-not-leak", "AWS_SECRET_ACCESS_KEY": "fixture-must-not-leak", "GITHUB_TOKEN": "fixture-must-not-leak"}):
            env = Supervisor(settings, None).env()
        self.assertNotIn("MESSAGE_ADMIN_TOKEN", env)
        self.assertNotIn("OPENAI_API_KEY", env)
        self.assertNotIn("AWS_SECRET_ACCESS_KEY", env)
        self.assertNotIn("GITHUB_TOKEN", env)
        self.assertIn("PATH", env)

    def test_plugin_environment_values_are_profile_and_worker_scoped(self):
        settings = Settings(data_dir=Path("/tmp/native-config-test"), admin_token=ADMIN, mcp_tokens={}, allowed_hosts=("localhost",), allowed_origins=("http://localhost",), plugin_env_allowlist={"a": ["SERVICE_A_TOKEN"], "b": ["SERVICE_B_TOKEN"]})
        settings.validate()
        with patch.dict(os.environ, {"SERVICE_A_TOKEN": "fixture-value-a", "SERVICE_B_TOKEN": "fixture-value-b"}):
            supervisor = Supervisor(settings, None)
            self.assertEqual(supervisor.env("a", "http")["SERVICE_A_TOKEN"], "fixture-value-a")
            self.assertNotIn("SERVICE_B_TOKEN", supervisor.env("a", "http"))
            self.assertNotIn("SERVICE_A_TOKEN", supervisor.env("b", "http"))
            self.assertNotIn("SERVICE_A_TOKEN", supervisor.env("a", "configure"))
            self.assertEqual(json.loads(supervisor.env("a", "list")["MESSAGE_PLUGIN_ENV_ALLOWLIST"]), ["SERVICE_A_TOKEN"])
        for invalid in ({"a": ["MESSAGE_ADMIN_TOKEN"]}, {"a": ["NODE_OPTIONS"]}, {"a": ["LD_PRELOAD"]}, {"a": ["HTTP_PROXY"]}, {"a": ["bad-name"]}, {"a": "SERVICE_A_TOKEN"}):
            with self.assertRaises(ValueError):
                replace(settings, plugin_env_allowlist=invalid).validate()

    def test_native_import_does_not_load_legacy(self):
        self.assertNotIn("app.plugins.manager", sys.modules)
        self.assertNotIn("app.main", sys.modules)


class GatewayPolicyTests(unittest.TestCase):
    def test_authentication_precedes_body_parsing_and_origin_is_exact(self):
        with tempfile.TemporaryDirectory(prefix="message-policy-") as temporary:
            settings = Settings(data_dir=Path(temporary), admin_token=ADMIN, mcp_tokens={"a": TOKEN_A}, allowed_hosts=("testserver",), allowed_origins=("http://testserver",))
            consumed = []
            def body():
                consumed.append(True)
                yield b"untrusted multipart body"
            with TestClient(create_app(settings)) as client:
                rejected = client.post("/api/native/catalog", content=body(), headers={"Content-Type": "multipart/form-data; boundary=fixture"})
                self.assertEqual(rejected.status_code, 401)
                self.assertFalse(consumed, "鉴权前不得解析或读取上传请求体")
                self.assertEqual(client.get("/api/native/health", headers={"Origin": "null"}).status_code, 403)
                self.assertEqual(client.get("/api/native/health", headers={"Origin": "http://testserver.evil"}).status_code, 403)
                self.assertEqual(client.get("/api/native/health", headers={"Host": "testserver:invalid"}).status_code, 400)
                self.assertEqual(client.get("/api/native/catalog", headers={"Authorization": f"Bearer {TOKEN_A}"}).status_code, 401)
                self.assertEqual(client.get("/mcp/a", headers={"Authorization": f"Bearer {ADMIN}"}).status_code, 401)
                self.assertEqual(client.get("/api/native/readiness").json(), {"status": "ready"})
                self.assertEqual(client.get("/api/plugins", headers={"Authorization": f"Bearer {ADMIN}"}).status_code, 404)


class NativeIntegrationTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="message-native-api-")
        self.root = Path(self.temp.name)
        self.settings = Settings(data_dir=self.root, admin_token=ADMIN, mcp_tokens={"a": TOKEN_A, "b": TOKEN_B}, plugin_env_allowlist={"a": ["SERVICE_A_TOKEN", "SERVICE_MISSING_TOKEN"], "b": ["SERVICE_B_TOKEN"]}, allowed_hosts=("127.0.0.1",), allowed_origins=("http://127.0.0.1",))
        self.running = []
        await self.start_server()
        self.client = httpx.AsyncClient(trust_env=False, base_url=self.url, headers={"Authorization": f"Bearer {ADMIN}"}, timeout=90)

    async def start_server(self):
        self.app = create_app(self.settings)
        sock = socket.socket()
        sock.bind(("127.0.0.1", 0))
        self.url = f"http://127.0.0.1:{sock.getsockname()[1]}"
        server = uvicorn.Server(uvicorn.Config(self.app, log_level="error", lifespan="on"))
        task = asyncio.create_task(server.serve(sockets=[sock]))
        self.running.append((server, task, sock))
        while not server.started:
            if task.done():
                await task
                self.fail("API 未启动")
            await asyncio.sleep(.02)

    async def stop_server(self):
        server, task, sock = self.running.pop()
        server.should_exit = True
        await asyncio.wait_for(task, 45)
        sock.close()

    async def asyncTearDown(self):
        await self.client.aclose()
        while self.running:
            await self.stop_server()
        self.temp.cleanup()

    async def post(self, path, data=None, **kwargs):
        response = await self.client.post("/api/native" + path, json=data, **kwargs)
        self.assertLess(response.status_code, 300, response.text)
        return response.json()

    async def operation(self, profile_id, kind, payload=None, **kwargs):
        operation = await self.post(f"/profiles/{profile_id}/{kind}", payload, **kwargs)
        return await self.wait_operation(operation["operation_id"])

    async def wait_operation(self, operation_id):
        for _ in range(3000):
            response = await self.client.get("/api/native/operations/" + operation_id)
            value = response.json()
            if value["status"] in {"succeeded", "failed", "cancelled"}:
                return value
            await asyncio.sleep(.05)
        self.fail("操作未在验收窗口完成")

    async def upload(self, data):
        response = await self.client.post("/api/native/catalog", files={"file": ("bundle.tgz", data, "application/gzip")})
        self.assertEqual(response.status_code, 201, response.text)
        return response.json()

    async def install(self, profile_id, data):
        item = await self.upload(data)
        operation = await self.operation(profile_id, "install", {"artifact_id": item["id"]})
        self.assertEqual(operation["status"], "succeeded", operation)
        return item

    async def call(self, profile_id="a", name="api_fixture"):
        return await self.post(f"/profiles/{profile_id}/call", {"name": name, "arguments": {}})

    async def test_complete_real_lifecycle_gateway_persistence_and_isolation(self):
        for profile_id in ("a", "b"):
            await self.post("/profiles", {"id": profile_id})
        self.assertEqual((await self.client.get("/api/native/catalog", headers={"Authorization": ""})).status_code, 401)
        self.assertEqual((await self.client.get("/api/native/catalog", headers={"Authorization": f"Bearer {TOKEN_A}"})).status_code, 401)
        self.assertEqual((await self.client.get("/api/native/health", headers={"Authorization": ""})).json(), {"status": "healthy"})
        self.assertEqual((await self.client.get("/api/native/health", headers={"Host": "evil.example"})).status_code, 400)
        self.assertEqual((await self.client.get("/api/native/health", headers={"Origin": "https://evil.example"})).status_code, 403)
        failed = await self.operation("a", "restart")
        self.assertEqual(failed["status"], "failed", failed)
        readiness = await self.client.get("/api/native/readiness", headers={"Authorization": ""})
        self.assertEqual(readiness.status_code, 503)
        self.assertEqual(readiness.json(), {"status": "degraded"})
        self.assertIn("profiles", (await self.client.get("/api/native/readiness")).json())
        for profile_id in ("a", "b"):
            await self.install(profile_id, local_bundle("mcp-export"))
            await self.install(profile_id, fixture())
            pending = (await self.client.get(f"/api/native/profiles/{profile_id}")).json()
            self.assertTrue(pending["restart_required"])
            restarted = await self.operation(profile_id, "restart")
            self.assertEqual(restarted["status"], "succeeded", restarted)
        self.assertEqual((await self.call())["content"][0]["text"], "v1")
        self.assertEqual((await self.client.get("/api/native/readiness")).status_code, 200)
        self.assertEqual((await self.client.get("/mcp/b", headers={"Authorization": f"Bearer {TOKEN_A}"})).status_code, 401)
        self.assertEqual((await self.client.get("/mcp/a", headers={"Authorization": f"Bearer {ADMIN}"})).status_code, 401)
        await self.install("a", local_bundle("text-tools"))
        configured_state = (await self.client.get("/api/native/profiles/a")).json()
        configurable = next(plugin for plugin in configured_state["plugins"] if plugin["moduleName"] == "@message/text-tools")
        changed = await self.operation("a", "configure", {"entry_id": configurable["entry_id"], "config": {"prefix": "API配置："}})
        self.assertEqual(changed["status"], "succeeded", changed)
        self.assertTrue((await self.client.get("/api/native/profiles/a")).json()["restart_required"])
        self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
        rendered = await self.post("/profiles/a/call", {"name": "message_text_uppercase", "arguments": {"text": "hello"}})
        self.assertEqual(rendered["content"][0]["text"], "API配置：HELLO")
        rejected_config = await self.operation("a", "configure", {"entry_id": configurable["entry_id"], "config": {"prefix": 5}})
        self.assertEqual(rejected_config["status"], "failed", rejected_config)
        config_after = (await self.client.get("/api/native/profiles/a")).json()
        self.assertEqual(next(plugin["config"]["prefix"] for plugin in config_after["plugins"] if plugin["moduleName"] == "@message/text-tools"), "API配置：")
        unauthorized = await self.operation("a", "configure", {"entry_id": configurable["entry_id"], "config": {"prefix": {"$env": "SERVICE_B_TOKEN"}}})
        self.assertEqual(unauthorized["status"], "failed")
        with patch.dict(os.environ, {"SERVICE_A_TOKEN": "fixture-env-secret-alpha", "SERVICE_B_TOKEN": "fixture-env-secret-beta", "SERVICE_MISSING_TOKEN": ""}):
            missing = await self.operation("a", "configure", {"entry_id": configurable["entry_id"], "config": {"prefix": {"$env": "SERVICE_MISSING_TOKEN"}}})
            self.assertEqual(missing["status"], "succeeded", missing)
            refused = await self.operation("a", "restart")
            self.assertEqual(refused["status"], "failed")
            self.assertIn("SERVICE_MISSING_TOKEN", refused["error"])
            reference = await self.operation("a", "configure", {"entry_id": configurable["entry_id"], "config": {"prefix": {"$env": "SERVICE_A_TOKEN"}}})
            self.assertEqual(reference["status"], "succeeded", reference)
            self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
            redacted = await self.post("/profiles/a/call", {"name": "message_text_uppercase", "arguments": {"text": "hello"}})
            self.assertEqual(redacted["content"][0]["text"], "[REDACTED]HELLO")
            state_text = (await self.client.get("/api/native/profiles/a")).text
            self.assertNotIn("fixture-env-secret-alpha", state_text)
            self.assertNotIn("fixture-env-secret-beta", state_text)
            self.assertIn("SERVICE_A_TOKEN", state_text)
            with patch.dict(os.environ, {"SERVICE_A_TOKEN": ADMIN}):
                reused = await self.operation("a", "restart")
                self.assertEqual(reused["status"], "failed")
                self.assertNotIn(ADMIN, json.dumps(reused))
            await self.operation("a", "configure", {"entry_id": configurable["entry_id"], "config": {"prefix": "API配置："}})
            self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
        # 两个真实 SDK 会话同时存在，DELETE 一个之后另一个和新会话仍可用。
        async with streamablehttp_client(self.url + "/mcp/a", httpx_client_factory=loopback_http_client, headers={"Authorization": f"Bearer {TOKEN_A}"}) as (r1, w1, session1):
            async with ClientSession(r1, w1) as first:
                await first.initialize()
                async with streamablehttp_client(self.url + "/mcp/a", httpx_client_factory=loopback_http_client, headers={"Authorization": f"Bearer {TOKEN_A}"}) as (r2, w2, session2):
                    async with ClientSession(r2, w2) as second:
                        await second.initialize()
                        self.assertNotEqual(session1(), session2())
                        self.assertEqual((await second.call_tool("api_fixture", {})).content[0].text, "v1")
                        # 会话 ID 不能替代访问令牌，且不能跨 profile 重用。
                        self.assertEqual((await self.client.get("/mcp/a", headers={"Authorization": "", "Mcp-Session-Id": session1()})).status_code, 401)
                        self.assertEqual((await self.client.get("/mcp/b", headers={"Authorization": f"Bearer {TOKEN_B}", "Mcp-Session-Id": session1(), "Accept": "text/event-stream"})).status_code, 404)
                self.assertTrue((await first.list_tools()).tools)
        disabled = await self.operation("a", "disable", {"name": "@message/api-fixture"}, headers={"Idempotency-Key": "disable-once"})
        repeated = await self.post("/profiles/a/disable", {"name": "@message/api-fixture"}, headers={"Idempotency-Key": "disable-once"})
        self.assertEqual(repeated["operation_id"], disabled["id"])
        conflict = await self.client.post("/api/native/profiles/a/enable", json={"name": "@message/api-fixture"}, headers={"Idempotency-Key": "disable-once"})
        self.assertEqual(conflict.status_code, 409)
        self.assertEqual((await self.call())["content"][0]["text"], "v1")
        self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
        self.assertTrue((await self.call())["isError"])
        self.assertEqual((await self.call("b"))["content"][0]["text"], "v1")
        await self.operation("a", "enable", {"name": "@message/api-fixture"})
        upgraded_data = fixture("1.0.1", "v2")
        upgrade = await self.install("a", upgraded_data)
        self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
        self.assertEqual((await self.call())["content"][0]["text"], "v2")
        download = await self.client.get(f"/api/native/catalog/{upgrade['id']}/download")
        self.assertEqual(download.status_code, 200)
        self.assertEqual(download.content, upgraded_data)
        conflict = await self.client.post("/api/native/catalog", files={"file": ("conflict.tgz", fixture("1.0.1", "altered"))})
        self.assertEqual(conflict.status_code, 409)
        pids = [worker.process.pid for worker in self.app.state.supervisor.workers.values()]
        await self.stop_server()
        for pid in pids:
            with self.assertRaises(ProcessLookupError):
                os.kill(pid, 0)
        await self.start_server()
        await self.client.aclose()
        self.client = httpx.AsyncClient(trust_env=False, base_url=self.url, headers={"Authorization": f"Bearer {ADMIN}"}, timeout=90)
        self.assertEqual((await self.call())["content"][0]["text"], "v2")
        self.assertEqual((await self.call("b"))["content"][0]["text"], "v1")
        self.assertEqual((await self.client.get(f"/api/native/operations/{disabled['id']}")).json()["status"], "succeeded")
        removed = await self.operation("a", "remove", {"name": "@message/api-fixture"})
        self.assertEqual(removed["status"], "succeeded", removed)
        state = (await self.client.get("/api/native/profiles/a")).json()
        self.assertEqual(state["status"], "stopped")
        self.assertFalse(any(bundle["name"] == "@message/api-fixture" for bundle in state["bundles"]))
        self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
        self.assertTrue((await self.call())["isError"])

    async def test_split_worker_diagnostics_never_persist_secrets(self):
        secret = "fixture-cross-chunk-environment-secret-2026"
        await self.post("/profiles", {"id": "a"})
        await self.install("a", local_bundle("mcp-export"))
        captured = io.StringIO()
        logger = logging.getLogger("uvicorn.error")
        handler = logging.StreamHandler(captured)
        logger.addHandler(handler)
        try:
            with patch.dict(os.environ, {"SERVICE_A_TOKEN": secret}):
                await self.install("a", diagnostic_fixture("startup"))
                failed = await self.operation("a", "restart")
                self.assertEqual(failed["status"], "failed", failed)
                self.assertIn("[REDACTED]", failed["error"])
                await self.operation("a", "disable", {"name": "@message/diagnostic-startup"})
                await self.install("a", diagnostic_fixture("late"))
                restarted = await self.operation("a", "restart")
                self.assertEqual(restarted["status"], "succeeded", restarted)
                for _ in range(300):
                    state = (await self.client.get("/api/native/profiles/a")).json()
                    if state["status"] == "error":
                        break
                    await asyncio.sleep(.05)
                self.assertEqual(state["status"], "error")
                self.assertIn("[REDACTED]", state["last_error"])
                # 覆盖 Uvicorn 异常日志的独立出口，防止安全响应之外泄漏 traceback。
                try:
                    raise RuntimeError("fixture exception " + secret)
                except RuntimeError:
                    logger.exception("fixture diagnostic %s", ADMIN)
            # 环境值离开作用域后，旧 generation 的诊断仍应保持脱敏。
            snapshots = [(await self.client.get("/api/native/profiles/a")).text, (await self.client.get("/api/native/operations")).text, (await self.client.get("/api/native/readiness")).text, captured.getvalue()]
            self.assertIn("[REDACTED]", captured.getvalue())
            for forbidden in (secret, ADMIN, TOKEN_A, TOKEN_B):
                for snapshot in snapshots:
                    self.assertNotIn(forbidden, snapshot)
                for database in self.root.glob("control.sqlite3*"):
                    self.assertNotIn(forbidden.encode(), database.read_bytes())
        finally:
            logger.removeHandler(handler)

    async def test_api_sigkill_recovers_only_owned_worker(self):
        await self.stop_server()
        children = []
        log_path = self.root / "api-crash.log"
        log = log_path.open("wb")
        async def start_child():
            with socket.socket() as probe:
                probe.bind(("127.0.0.1", 0))
                port = probe.getsockname()[1]
            env = Supervisor(self.settings, None).env()
            env.update(MESSAGE_AUTH_MODE="legacy_token", PYTHONPATH=str(ROOT / "apps/market-api"), MESSAGE_DATA_DIR=str(self.root), MESSAGE_ADMIN_TOKEN=ADMIN, MESSAGE_MCP_TOKENS_JSON=json.dumps({"a": TOKEN_A, "b": TOKEN_B}), MESSAGE_ALLOWED_HOSTS="127.0.0.1", MESSAGE_ALLOWED_ORIGINS="http://127.0.0.1")
            child = await asyncio.create_subprocess_exec(sys.executable, "-m", "uvicorn", "app.cloud_main:app", "--host", "127.0.0.1", "--port", str(port), "--log-level", "error", "--timeout-graceful-shutdown", "2", cwd=ROOT, env=env, stdout=log, stderr=log)
            children.append(child)
            await self.client.aclose()
            self.client = httpx.AsyncClient(base_url=f"http://127.0.0.1:{port}", headers={"Authorization": f"Bearer {ADMIN}"}, timeout=90, trust_env=False)
            for _ in range(1000):
                self.assertIsNone(child.returncode, log_path.read_text())
                try:
                    if (await self.client.get("/api/native/health")).status_code == 200:
                        return child
                except httpx.ConnectError:
                    pass
                await asyncio.sleep(.05)
            self.fail("强杀恢复测试 API 启动超时：" + log_path.read_text())
        try:
            first = await start_child()
            await self.post("/profiles", {"id": "a"})
            await self.install("a", local_bundle("mcp-export"))
            await self.install("a", fixture())
            self.assertEqual((await self.operation("a", "restart"))["status"], "succeeded")
            before = (await self.client.get("/api/native/profiles/a")).json()
            marker = self.root / "profiles/a/.message-worker.json"
            old_pid = json.loads(marker.read_text())["pid"]
            first.kill()
            await first.wait()
            # 强杀控制面不应让冷恢复把旧 worker 报作新 generation。
            await start_child()
            after = (await self.client.get("/api/native/profiles/a")).json()
            self.assertEqual(after["status"], "running", after)
            self.assertGreater(after["generation"], before["generation"])
            self.assertNotEqual(json.loads(marker.read_text())["pid"], old_pid)
            self.assertEqual((await self.call())["content"][0]["text"], "v1")
            if Path(f"/proc/{old_pid}/stat").exists():
                fields = Path(f"/proc/{old_pid}/stat").read_text().split(")", 1)[1].split()
                self.assertEqual(fields[0], "Z", "旧 worker 必须退出；孤儿僵尸由容器 init 回收")
        finally:
            for child in children:
                if child.returncode is None:
                    child.terminate()
                    try:
                        await asyncio.wait_for(child.wait(), 30)
                    except asyncio.TimeoutError:
                        child.kill()
                        await child.wait()
            log.close()

    async def test_recovery_refuses_unverified_reused_pid(self):
        await self.post("/profiles", {"id": "a"})
        # 记录故意指向本验收进程；命令并非 Node worker，绝不能收到信号。
        identity = self.app.state.supervisor.process_identity(os.getpid())
        self.app.state.store.own_process(dict(pid=os.getpid(), profile_id="a", command="http", **identity))
        await self.stop_server()
        await self.start_server()
        await self.client.aclose()
        self.client = httpx.AsyncClient(base_url=self.url, headers={"Authorization": f"Bearer {ADMIN}"}, timeout=90, trust_env=False)
        state = (await self.client.get("/api/native/profiles/a")).json()
        self.assertEqual(state["status"], "error")
        self.assertIn("无法验证", state["last_error"])
        self.assertEqual((await self.client.get("/api/native/readiness")).status_code, 503)
        failed = await self.operation("a", "restart")
        self.assertEqual(failed["status"], "failed")
        self.assertFalse(self.app.state.supervisor.workers)
        os.kill(os.getpid(), 0)
        self.app.state.store.forget_process(os.getpid())

    async def test_running_install_cancel_and_integrity_recovery(self):
        await self.post("/profiles", {"id": "a"})
        data = local_bundle("mcp-export")
        artifact = await self.upload(data)
        queued = await self.post("/profiles/a/install", {"artifact_id": artifact["id"]})
        run_file = self.root / "profiles/a/.plugin-manager/run.json"
        for _ in range(1000):
            if run_file.exists():
                break
            await asyncio.sleep(.01)
        self.assertTrue(run_file.exists(), "应观察到真实 pnpm 安装后再取消")
        await self.post(f"/operations/{queued['operation_id']}/cancel")
        cancelled = await self.wait_operation(queued["operation_id"])
        self.assertEqual(cancelled["status"], "cancelled", cancelled)
        self.assertFalse(run_file.exists())
        profile = (await self.client.get("/api/native/profiles/a")).json()
        self.assertEqual(profile["desired_revision"], 0)
        self.assertFalse(profile["bundles"])
        stored = self.root / "catalog" / (artifact["id"] + ".tgz")
        stored.write_bytes(b"tampered fixture")
        self.assertEqual((await self.client.get(f"/api/native/catalog/{artifact['id']}/download")).status_code, 409)
        failed = await self.operation("a", "install", {"artifact_id": artifact["id"]})
        self.assertEqual(failed["status"], "failed")
        self.assertIn("SHA256", failed["error"])
        restored = await self.upload(data)
        self.assertEqual(restored["id"], artifact["id"])
        self.assertEqual(stored.read_bytes(), data)

    async def test_queued_cancel_single_instance_and_interrupted_recovery(self):
        await self.post("/profiles", {"id": "a"})
        for _ in range(2):
            empty_key = await self.client.post("/api/native/profiles/a/restart", headers={"Idempotency-Key": ""})
            self.assertEqual(empty_key.status_code, 409)
        self.assertFalse(self.app.state.store.operations("a"))
        lock = self.app.state.supervisor.lock("a")
        await lock.acquire()
        try:
            queued = await self.post("/profiles/a/restart")
            cancelled = await self.post(f"/operations/{queued['operation_id']}/cancel")
            self.assertEqual(cancelled["status"], "cancelled")
            self.assertEqual((await self.client.post(f"/api/native/operations/{queued['operation_id']}/cancel")).status_code, 409)
        finally:
            lock.release()
        other = create_app(self.settings)
        with self.assertRaisesRegex(RuntimeError, "一个 worker"):
            async with other.router.lifespan_context(other):
                pass
        while self.app.state.operations.tasks:
            await asyncio.sleep(.01)
        value = self.app.state.store.operation(queued["operation_id"])
        value.update(status="running", cancel_requested=False)
        self.app.state.store.save_operation(value)
        await self.stop_server()
        await self.start_server()
        async with httpx.AsyncClient(trust_env=False, base_url=self.url, headers={"Authorization": f"Bearer {ADMIN}"}) as client:
            recovered = (await client.get(f"/api/native/operations/{value['id']}")).json()
        self.assertEqual(recovered["status"], "failed")
        self.assertIn("interrupted", recovered["error"])


if __name__ == "__main__":
    unittest.main()