diff --git a/openspace/codex_session_scenarios.py b/openspace/codex_session_scenarios.py new file mode 100644 index 0000000..9d9ff00 --- /dev/null +++ b/openspace/codex_session_scenarios.py @@ -0,0 +1,627 @@ +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import signal +import subprocess +import tempfile +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + + +OPENING_PREFIX = "先做一次 OpenSpace 预检，再开始当前任务。" +SESSION_STATUS_RE = re.compile(r"OpenSpace session:\s*(\S+)") +MACHINE_STATUS_RE = re.compile(r"OpenSpace machine:\s*(\S+)") +FALLBACK_LINE = "当前线程不依赖 OpenSpace 自动沉淀；我会先按本地文档、脚本或手动收尾路径继续。" + + +@dataclass(frozen=True) +class OpeningStatus: + session_status: str + machine_status: str + fallback_present: bool + text: str + + +@dataclass(frozen=True) +class ExecOutput: + thread_id: str | None + events: list[dict[str, Any]] + agent_messages: list[str] + opening: OpeningStatus | None + + +@dataclass(frozen=True) +class SessionFamily: + thread_ids: set[str] + session_files: set[Path] + + +@dataclass(frozen=True) +class RunResult: + label: str + exit_code: int + thread_id: str | None + opening: dict[str, Any] | None + agent_messages: list[str] + mcp_tool_calls: list[dict[str, Any]] + session_files: list[str] + thread_ids: list[str] + stdout_path: str + stderr_path: str + + +@dataclass(frozen=True) +class ScenarioResult: + name: str + ok: bool + details: dict[str, Any] + + +def _iter_json_events(output: str) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + for raw_line in output.splitlines(): + line = raw_line.strip() + if not line.startswith("{"): + continue + try: + payload = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(payload, dict): + events.append(payload) + return events + + +def _extract_agent_message(event: dict[str, Any]) -> str | None: + if event.get("type") != "item.completed": + return None + item = event.get("item") + if not isinstance(item, dict): + return None + if item.get("type") != "agent_message": + return None + text = item.get("text") + return text if isinstance(text, str) else None + + +def _extract_mcp_tool_call(event: dict[str, Any]) -> dict[str, Any] | None: + if event.get("type") != "item.completed": + return None + item = event.get("item") + if not isinstance(item, dict): + return None + if item.get("type") != "mcp_tool_call": + return None + return { + "server": item.get("server"), + "tool": item.get("tool"), + "status": item.get("status"), + "error": item.get("error"), + } + + +def _parse_opening(text: str) -> OpeningStatus | None: + if not text.startswith(OPENING_PREFIX): + return None + session_match = SESSION_STATUS_RE.search(text) + machine_match = MACHINE_STATUS_RE.search(text) + if session_match is None or machine_match is None: + return None + return OpeningStatus( + session_status=session_match.group(1), + machine_status=machine_match.group(1), + fallback_present=FALLBACK_LINE in text, + text=text, + ) + + +def parse_exec_output(output: str) -> ExecOutput: + events = _iter_json_events(output) + thread_id: str | None = None + agent_messages: list[str] = [] + opening: OpeningStatus | None = None + + for event in events: + if event.get("type") == "thread.started" and thread_id is None: + candidate = event.get("thread_id") + if isinstance(candidate, str): + thread_id = candidate + agent_message = _extract_agent_message(event) + if agent_message is not None: + agent_messages.append(agent_message) + if opening is None: + opening = _parse_opening(agent_message) + + return ExecOutput( + thread_id=thread_id, + events=events, + agent_messages=agent_messages, + opening=opening, + ) + + +def _read_session_meta(path: Path) -> dict[str, Any] | None: + try: + first_line = path.read_text(encoding="utf-8").splitlines()[0] + except Exception: + return None + try: + payload = json.loads(first_line) + except json.JSONDecodeError: + return None + if payload.get("type") != "session_meta": + return None + session_payload = payload.get("payload") + return session_payload if isinstance(session_payload, dict) else None + + +def collect_session_family(thread_id: str, sessions_root: Path) -> SessionFamily: + metas: list[tuple[Path, dict[str, Any]]] = [] + for path in sessions_root.glob("**/*.jsonl"): + meta = _read_session_meta(path) + if meta is not None: + metas.append((path, meta)) + + thread_ids: set[str] = {thread_id} + session_files: set[Path] = set() + changed = True + + while changed: + changed = False + for path, meta in metas: + current_id = meta.get("id") + if isinstance(current_id, str) and current_id in thread_ids: + if path not in session_files: + session_files.add(path) + changed = True + + source = meta.get("source") + if not isinstance(source, dict): + continue + subagent = source.get("subagent") + if not isinstance(subagent, dict): + continue + thread_spawn = subagent.get("thread_spawn") + if not isinstance(thread_spawn, dict): + continue + parent_thread_id = thread_spawn.get("parent_thread_id") + if ( + isinstance(parent_thread_id, str) + and parent_thread_id in thread_ids + and isinstance(current_id, str) + and current_id not in thread_ids + ): + thread_ids.add(current_id) + session_files.add(path) + changed = True + + return SessionFamily(thread_ids=thread_ids, session_files=session_files) + + +def cleanup_session_artifacts( + *, + thread_ids: set[str], + session_files: set[Path], + session_index_path: Path, +) -> None: + for path in session_files: + path.unlink(missing_ok=True) + + if not session_index_path.exists(): + return + + kept_lines: list[str] = [] + for line in session_index_path.read_text(encoding="utf-8").splitlines(): + stripped = line.strip() + if not stripped: + continue + try: + payload = json.loads(stripped) + except json.JSONDecodeError: + kept_lines.append(line) + continue + + payload_thread_id = payload.get("thread_id") + payload_session_file = payload.get("session_file") + if payload_thread_id in thread_ids: + continue + if isinstance(payload_session_file, str) and Path(payload_session_file) in session_files: + continue + kept_lines.append(line) + + new_content = "\n".join(kept_lines) + if new_content: + new_content += "\n" + session_index_path.write_text(new_content, encoding="utf-8") + + +def snapshot_daemons(state_dir: Path, workspace: str) -> dict[str, dict[str, Any]]: + snapshot: dict[str, dict[str, Any]] = {} + if not state_dir.exists(): + return snapshot + + for path in state_dir.glob("*.json"): + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except Exception: + continue + if payload.get("workspace") != workspace: + continue + server_kind = payload.get("server_kind") + if not isinstance(server_kind, str): + continue + snapshot[server_kind] = payload + return snapshot + + +def kill_daemons_in_state_dir(state_dir: Path) -> None: + if not state_dir.exists(): + return + for path in state_dir.glob("*.json"): + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except Exception: + continue + pid = payload.get("pid") + if not isinstance(pid, int): + continue + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + pass + except PermissionError: + pass + else: + deadline = time.time() + 2.0 + while time.time() < deadline: + try: + os.kill(pid, 0) + except ProcessLookupError: + break + time.sleep(0.05) + else: + try: + os.kill(pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, AttributeError): + pass + shutil.rmtree(state_dir, ignore_errors=True) + + +def run_exec_session( + *, + label: str, + prompt: str, + cwd: Path, + extra_configs: list[str] | None = None, + timeout_seconds: int = 180, + codex_binary: str = "codex", + sessions_root: Path | None = None, + session_index_path: Path | None = None, + output_dir: Path | None = None, +) -> RunResult: + sessions_root = sessions_root or (Path.home() / ".codex" / "sessions") + session_index_path = session_index_path or (Path.home() / ".codex" / "session_index.jsonl") + output_dir = output_dir or Path(tempfile.mkdtemp(prefix="openspace-codex-session-")) + output_dir.mkdir(parents=True, exist_ok=True) + stdout_path = output_dir / f"{label}.stdout.log" + stderr_path = output_dir / f"{label}.stderr.log" + + command = [ + codex_binary, + "exec", + "--json", + "--skip-git-repo-check", + "--dangerously-bypass-approvals-and-sandbox", + "-C", + str(cwd), + ] + for item in extra_configs or []: + command.extend(["-c", item]) + command.append(prompt) + + completed = subprocess.run( + command, + capture_output=True, + text=True, + errors="replace", + timeout=timeout_seconds, + check=False, + ) + stdout_path.write_text(completed.stdout, encoding="utf-8") + stderr_path.write_text(completed.stderr, encoding="utf-8") + + parsed = parse_exec_output(completed.stdout) + session_family = ( + collect_session_family(parsed.thread_id, sessions_root) + if parsed.thread_id + else SessionFamily(thread_ids=set(), session_files=set()) + ) + mcp_tool_calls = [ + call + for event in parsed.events + if (call := _extract_mcp_tool_call(event)) is not None + ] + + return RunResult( + label=label, + exit_code=completed.returncode, + thread_id=parsed.thread_id, + opening=asdict(parsed.opening) if parsed.opening else None, + agent_messages=parsed.agent_messages, + mcp_tool_calls=mcp_tool_calls, + session_files=sorted(str(path) for path in session_family.session_files), + thread_ids=sorted(session_family.thread_ids), + stdout_path=str(stdout_path), + stderr_path=str(stderr_path), + ) + + +def _session_env_overrides(state_dir: Path, workspace: Path) -> list[str]: + return [ + f'mcp_servers.openspace.env.OPENSPACE_MCP_DAEMON_STATE_DIR="{state_dir}"', + f'mcp_servers.openspace_evolution.env.OPENSPACE_MCP_DAEMON_STATE_DIR="{state_dir}"', + 'mcp_servers.openspace.env.OPENSPACE_MCP_PROXY_MODE="daemon"', + 'mcp_servers.openspace_evolution.env.OPENSPACE_MCP_PROXY_MODE="daemon"', + f'mcp_servers.openspace.env.OPENSPACE_WORKSPACE="{workspace}"', + f'mcp_servers.openspace_evolution.env.OPENSPACE_WORKSPACE="{workspace}"', + ] + + +def _assert(condition: bool, message: str, errors: list[str]) -> None: + if not condition: + errors.append(message) + + +def _scenario_cold_start(base_dir: Path, cwd: Path) -> ScenarioResult: + state_dir = base_dir / "cold-start-state" + run = run_exec_session( + label="cold-start", + prompt=( + "这是 OpenSpace 冷启动预检测试。严格按 TMP 的 AGENTS.md 做开场预检，" + "然后只输出一句“cold-start done”。不要修改任何文件，也不要使用子代理。" + ), + cwd=cwd, + extra_configs=_session_env_overrides(state_dir, cwd), + output_dir=base_dir / "cold-start-output", + ) + + errors: list[str] = [] + opening = run.opening or {} + _assert(run.exit_code == 0, f"cold-start exit_code={run.exit_code}", errors) + _assert(opening.get("session_status") == "ready", f"cold-start session_status={opening.get('session_status')}", errors) + _assert(opening.get("machine_status") == "ready", f"cold-start machine_status={opening.get('machine_status')}", errors) + _assert(opening.get("fallback_present") is False, "cold-start unexpectedly printed fallback line", errors) + _assert(run.agent_messages[-1:] == ["cold-start done"], f"cold-start final message={run.agent_messages[-1:]}", errors) + + return ScenarioResult( + name="cold-start-preflight", + ok=not errors, + details={"errors": errors, "run": asdict(run), "state_dirs": [str(state_dir)]}, + ) + + +def _scenario_warm_reuse(base_dir: Path, cwd: Path) -> ScenarioResult: + state_dir = base_dir / "warm-reuse-state" + extra_configs = _session_env_overrides(state_dir, cwd) + prompt = ( + "这是 OpenSpace warm-session reuse 测试。先完成 TMP 的开场预检，" + "然后调用 openspace 的 search_skills 工具，参数用 query='OpenSpace MCP 健康检查'、" + "source='local'、limit=1、auto_import=false。最后只输出一句“warm-session done”。" + "不要修改任何文件，也不要使用子代理。" + ) + + first = run_exec_session( + label="warm-reuse-first", + prompt=prompt, + cwd=cwd, + extra_configs=extra_configs, + output_dir=base_dir / "warm-reuse-output", + ) + first_snapshot = snapshot_daemons(state_dir, str(cwd)) + + second = run_exec_session( + label="warm-reuse-second", + prompt=prompt, + cwd=cwd, + extra_configs=extra_configs, + output_dir=base_dir / "warm-reuse-output", + ) + second_snapshot = snapshot_daemons(state_dir, str(cwd)) + + errors: list[str] = [] + for label, run in (("first", first), ("second", second)): + opening = run.opening or {} + _assert(run.exit_code == 0, f"warm-reuse {label} exit_code={run.exit_code}", errors) + _assert(opening.get("session_status") == "ready", f"warm-reuse {label} session_status={opening.get('session_status')}", errors) + _assert(any(call.get("server") == "openspace" and call.get("tool") == "search_skills" for call in run.mcp_tool_calls), f"warm-reuse {label} missing openspace.search_skills call", errors) + _assert(run.agent_messages[-1:] == ["warm-session done"], f"warm-reuse {label} final message={run.agent_messages[-1:]}", errors) + + _assert("main" in first_snapshot, "warm-reuse first run did not create main daemon metadata", errors) + _assert("main" in second_snapshot, "warm-reuse second run did not create main daemon metadata", errors) + if "main" in first_snapshot and "main" in second_snapshot: + _assert( + first_snapshot["main"].get("pid") == second_snapshot["main"].get("pid"), + f"warm-reuse main pid changed: {first_snapshot['main'].get('pid')} -> {second_snapshot['main'].get('pid')}", + errors, + ) + _assert( + first_snapshot["main"].get("port") == second_snapshot["main"].get("port"), + f"warm-reuse main port changed: {first_snapshot['main'].get('port')} -> {second_snapshot['main'].get('port')}", + errors, + ) + + return ScenarioResult( + name="warm-session-reuse", + ok=not errors, + details={ + "errors": errors, + "first": asdict(first), + "second": asdict(second), + "first_snapshot": first_snapshot, + "second_snapshot": second_snapshot, + "state_dirs": [str(state_dir)], + }, + ) + + +def _scenario_unhealthy_fallback(base_dir: Path, cwd: Path) -> ScenarioResult: + state_dir = base_dir / "unhealthy-state" + missing_command = base_dir / "missing-openspace-command" + run = run_exec_session( + label="unhealthy-fallback", + prompt=( + "这是 OpenSpace unhealthy-session fallback 测试。严格按 TMP 的 AGENTS.md 做开场预检，" + "然后只输出一句“unhealthy-session done”。不要修改任何文件，也不要使用子代理。" + ), + cwd=cwd, + extra_configs=_session_env_overrides(state_dir, cwd) + + [ + f'mcp_servers.openspace.command="{missing_command}"', + f'mcp_servers.openspace_evolution.command="{missing_command}"', + ], + output_dir=base_dir / "unhealthy-output", + ) + + errors: list[str] = [] + opening = run.opening or {} + _assert(run.exit_code == 0, f"unhealthy fallback exit_code={run.exit_code}", errors) + _assert( + opening.get("session_status") in {"exposed-but-unhealthy", "unknown"}, + f"unhealthy fallback session_status={opening.get('session_status')}", + errors, + ) + _assert(opening.get("machine_status") == "ready", f"unhealthy fallback machine_status={opening.get('machine_status')}", errors) + _assert(opening.get("fallback_present") is True, "unhealthy fallback missing fallback line", errors) + _assert(run.agent_messages[-1:] == ["unhealthy-session done"], f"unhealthy fallback final message={run.agent_messages[-1:]}", errors) + + return ScenarioResult( + name="unhealthy-session-fallback", + ok=not errors, + details={"errors": errors, "run": asdict(run), "state_dirs": [str(state_dir)]}, + ) + + +def _scenario_agent_team(base_dir: Path, cwd: Path) -> ScenarioResult: + state_dir = base_dir / "agent-team-state" + run = run_exec_session( + label="agent-team", + prompt=( + "这是 OpenSpace agent-team 测试。先完成 TMP 的开场预检，然后使用 agent team，" + "至少启动两个只读子代理，分别检查 TMP 仓库里的 AGENTS.md 和 " + "scripts/check_openspace_mcp_preflight.py 与 OpenSpace 预检相关的内容。" + "父线程最后只输出一句“agent-team done”。不要修改任何文件。" + ), + cwd=cwd, + extra_configs=_session_env_overrides(state_dir, cwd), + timeout_seconds=240, + output_dir=base_dir / "agent-team-output", + ) + + errors: list[str] = [] + opening = run.opening or {} + _assert(run.exit_code == 0, f"agent-team exit_code={run.exit_code}", errors) + _assert(opening.get("session_status") == "ready", f"agent-team session_status={opening.get('session_status')}", errors) + _assert(len(run.thread_ids) >= 3, f"agent-team expected parent + >=2 child threads, got {run.thread_ids}", errors) + _assert(run.agent_messages[-1:] == ["agent-team done"], f"agent-team final message={run.agent_messages[-1:]}", errors) + + return ScenarioResult( + name="agent-team-split", + ok=not errors, + details={"errors": errors, "run": asdict(run), "state_dirs": [str(state_dir)]}, + ) + + +def _cleanup_run_artifacts(result: ScenarioResult, session_index_path: Path) -> None: + details = result.details + candidate_runs: list[dict[str, Any]] = [] + if "run" in details and isinstance(details["run"], dict): + candidate_runs.append(details["run"]) + for key in ("first", "second"): + if key in details and isinstance(details[key], dict): + candidate_runs.append(details[key]) + + session_files: set[Path] = set() + thread_ids: set[str] = set() + for run in candidate_runs: + for item in run.get("session_files", []): + session_files.add(Path(item)) + for item in run.get("thread_ids", []): + thread_ids.add(item) + + cleanup_session_artifacts( + thread_ids=thread_ids, + session_files=session_files, + session_index_path=session_index_path, + ) + + +def run_scenarios(*, cwd: Path, cleanup: bool = True) -> dict[str, Any]: + base_dir = Path(tempfile.mkdtemp(prefix="openspace-codex-scenarios-")) + session_index_path = Path.home() / ".codex" / "session_index.jsonl" + started_at = time.time() + + scenario_steps = [ + ("cold-start-preflight", _scenario_cold_start), + ("warm-session-reuse", _scenario_warm_reuse), + ("unhealthy-session-fallback", _scenario_unhealthy_fallback), + ("agent-team-split", _scenario_agent_team), + ] + results: list[ScenarioResult] = [] + + try: + for scenario_name, scenario_fn in scenario_steps: + try: + results.append(scenario_fn(base_dir, cwd)) + except Exception as exc: # noqa: BLE001 + results.append( + ScenarioResult( + name=scenario_name, + ok=False, + details={"errors": [f"{type(exc).__name__}: {exc}"]}, + ) + ) + finally: + if cleanup: + for result in results: + _cleanup_run_artifacts(result, session_index_path) + for state_dir in base_dir.glob("*-state"): + kill_daemons_in_state_dir(state_dir) + shutil.rmtree(base_dir, ignore_errors=True) + + return { + "cwd": str(cwd), + "started_at": started_at, + "cleanup": cleanup, + "all_ok": all(item.ok for item in results), + "results": [asdict(item) for item in results], + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="Run real Codex OpenSpace session scenarios") + parser.add_argument( + "--cwd", + type=Path, + default=Path("/Users/admin/PycharmProjects/TMP"), + help="Working directory for real session scenarios", + ) + parser.add_argument( + "--keep-artifacts", + action="store_true", + help="Keep generated session files, daemon state, and command logs", + ) + args = parser.parse_args() + + summary = run_scenarios(cwd=args.cwd.resolve(), cleanup=not args.keep_artifacts) + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return 0 if summary["all_ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/openspace/evolution_mcp_server.py b/openspace/evolution_mcp_server.py index b97bcac..2587f04 100644 --- a/openspace/evolution_mcp_server.py +++ b/openspace/evolution_mcp_server.py @@ -13,6 +13,7 @@ import json import logging import os import subprocess +import signal import sys import threading import time @@ -21,6 +22,7 @@ from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional +from openspace.mcp_stdio import maybe_redirect_stderr_to_file from openspace.mcp_tool_registration import register_evolution_tools class _MCPSafeStdout: @@ -82,11 +84,7 @@ _LOG_DIR = Path(__file__).resolve().parent.parent / "logs" _LOG_DIR.mkdir(parents=True, exist_ok=True) _real_stdout = sys.stdout -if os.name == "nt": - _stderr_file = open( - _LOG_DIR / "evolution_mcp_stderr.log", "a", encoding="utf-8", buffering=1 - ) - sys.stderr = _stderr_file +maybe_redirect_stderr_to_file(_LOG_DIR, "evolution_mcp_stderr.log") sys.stdout = _MCPSafeStdout(_real_stdout, sys.stderr) @@ -117,6 +115,8 @@ _idle_watchdog_started = False _activity_lock = threading.Lock() _active_request_count = 0 _last_activity_at = time.monotonic() +_shutdown_started = False +_shutdown_lock = threading.Lock() def _json_ok(data: Any) -> str: @@ -132,6 +132,9 @@ def _mark_request_start() -> None: with _activity_lock: _active_request_count += 1 _last_activity_at = time.monotonic() + from openspace.shared_mcp_runtime import update_current_daemon_status + + update_current_daemon_status("evolution", touch=True, active_delta=1) def _mark_request_end() -> None: @@ -139,6 +142,51 @@ def _mark_request_end() -> None: with _activity_lock: _active_request_count = max(0, _active_request_count - 1) _last_activity_at = time.monotonic() + from openspace.shared_mcp_runtime import update_current_daemon_status + + update_current_daemon_status("evolution", touch=True, active_delta=-1) + + +def _shutdown_worker(reason: str) -> None: + logger.info("Shutting down OpenSpace evolution daemon: %s", reason) + instance = _openspace_instance + if instance is not None and instance.is_initialized(): + try: + asyncio.run(asyncio.wait_for(instance.cleanup(), timeout=10.0)) + except Exception as exc: + logger.warning("OpenSpace evolution cleanup during shutdown failed: %s", exc) + logging.shutdown() + os._exit(0) + + +def _begin_shutdown(reason: str) -> None: + global _shutdown_started + with _shutdown_lock: + if _shutdown_started: + return + _shutdown_started = True + + threading.Thread( + target=_shutdown_worker, + args=(reason,), + name="openspace-evolution-shutdown", + daemon=True, + ).start() + + +def _install_signal_handlers() -> None: + def _handle(signum, _frame) -> None: + try: + signame = signal.Signals(signum).name + except Exception: + signame = str(signum) + _begin_shutdown(f"signal {signame}") + + for signum in (signal.SIGTERM, signal.SIGINT): + try: + signal.signal(signum, _handle) + except Exception: + continue def _idle_watchdog_loop(idle_timeout_seconds: int) -> None: @@ -154,8 +202,8 @@ def _idle_watchdog_loop(idle_timeout_seconds: int) -> None: "Evolution MCP idle watchdog exiting process after %.1fs idle with no active requests", idle_for, ) - logging.shutdown() - os._exit(0) + _begin_shutdown(f"idle timeout after {idle_for:.1f}s") + return def _maybe_start_idle_watchdog() -> None: @@ -700,6 +748,7 @@ def run_mcp_server() -> None: args = parser.parse_args() if args.transport == "stdio" or os.environ.get("OPENSPACE_MCP_DAEMON") == "1": + _install_signal_handlers() _maybe_start_idle_watchdog() mcp.settings.port = args.port diff --git a/openspace/mcp_proxy.py b/openspace/mcp_proxy.py index e990bb3..814d317 100644 --- a/openspace/mcp_proxy.py +++ b/openspace/mcp_proxy.py @@ -1,14 +1,17 @@ from __future__ import annotations +import asyncio import argparse import inspect import json import os +from pathlib import Path from typing import Any from mcp.server.fastmcp import FastMCP from mcp.types import TextContent +from openspace.mcp_stdio import maybe_redirect_stderr_to_file from openspace.grounding.backends.mcp.client import MCPClient from openspace.mcp_tool_registration import ( register_evolution_tools, @@ -17,6 +20,10 @@ from openspace.mcp_tool_registration import ( from openspace.shared_mcp_runtime import ServerKind, ensure_daemon +_LOG_DIR = Path(__file__).resolve().parent.parent / "logs" +maybe_redirect_stderr_to_file(_LOG_DIR, "mcp_proxy_stderr.log") + + def _proxy_mode_for(server_kind: ServerKind) -> str: raw = os.environ.get("OPENSPACE_MCP_PROXY_MODE", "").strip().lower() if raw in {"daemon", "direct"}: @@ -45,45 +52,38 @@ def _extract_text_payload(result: Any) -> str: class _RemoteProxyBase: def __init__(self, server_kind: ServerKind): self._server_kind = server_kind - self._client: MCPClient | None = None - self._current_url: str | None = None - async def _get_client(self) -> MCPClient: + async def _call_remote_tool_once(self, tool_name: str, args: dict[str, Any]) -> str: record = await ensure_daemon(self._server_kind) - if self._client is not None and self._current_url == record.url: - return self._client - - await self._reset_client() - - self._client = MCPClient( + client = MCPClient( config={"mcpServers": {"daemon": {"url": record.url}}}, timeout=10.0, sse_read_timeout=60 * 60.0, check_dependencies=False, ) - self._current_url = record.url - return self._client + try: + session = await client.create_session("daemon", auto_initialize=True) + if session is None: + raise RuntimeError("Failed to create daemon MCP session") + result = await session.connector.call_tool(tool_name, args) + return _extract_text_payload(result) + finally: + await client.close_all_sessions() - async def _reset_client(self) -> None: - if self._client is not None: - await self._client.close_all_sessions() - self._client = None - self._current_url = None + def _call_remote_tool_blocking(self, tool_name: str, args: dict[str, Any]) -> str: + return asyncio.run(self._call_remote_tool_once(tool_name, args)) async def _call_remote_tool(self, tool_name: str, args: dict[str, Any]) -> str: for attempt in range(2): try: - client = await self._get_client() - session = await client.create_session("daemon", auto_initialize=True) - if session is None: - raise RuntimeError("Failed to create daemon MCP session") - result = await session.connector.call_tool(tool_name, args) - return _extract_text_payload(result) + return await asyncio.to_thread( + self._call_remote_tool_blocking, + tool_name, + args, + ) except Exception as exc: - if attempt == 0: - await self._reset_client() - continue - return _json_error(exc, status="error") + if attempt == 1: + return _json_error(exc, status="error") return _json_error("Unreachable proxy retry path", status="error") diff --git a/openspace/mcp_server.py b/openspace/mcp_server.py index b1f665b..06b1a02 100644 --- a/openspace/mcp_server.py +++ b/openspace/mcp_server.py @@ -21,12 +21,14 @@ import inspect import json import logging import os +import signal import sys import threading import time from pathlib import Path from typing import Any, Dict, List, Optional +from openspace.mcp_stdio import maybe_redirect_stderr_to_file from openspace.mcp_tool_registration import register_main_tools from openspace.shared_mcp_runtime import update_current_daemon_status @@ -89,17 +91,7 @@ _LOG_DIR = Path(__file__).resolve().parent.parent / "logs" _LOG_DIR.mkdir(parents=True, exist_ok=True) _real_stdout = sys.stdout - -# Windows pipe buffers are small. When using stdio MCP transport, -# the parent process only reads stdout for MCP messages and does NOT -# drain stderr. Heavy log/print output during execute_task fills the stderr -# pipe buffer, blocking this process on write() → deadlock → timeout. -# Redirect stderr to a log file on Windows to prevent this. -if os.name == "nt": - _stderr_file = open( - _LOG_DIR / "mcp_stderr.log", "a", encoding="utf-8", buffering=1 - ) - sys.stderr = _stderr_file +maybe_redirect_stderr_to_file(_LOG_DIR, "mcp_stderr.log") sys.stdout = _MCPSafeStdout(_real_stdout, sys.stderr) @@ -132,6 +124,8 @@ _active_request_count = 0 _last_activity_at = time.monotonic() _embedding_prewarm_started = False _embedding_prewarm_lock = threading.Lock() +_shutdown_started = False +_shutdown_lock = threading.Lock() # Internal state: tracks bot skill directories already registered this session. _registered_skill_dirs: set = set() @@ -632,6 +626,7 @@ def _mark_request_start() -> None: with _activity_lock: _active_request_count += 1 _last_activity_at = time.monotonic() + update_current_daemon_status("main", touch=True, active_delta=1) def _mark_request_end() -> None: @@ -639,6 +634,49 @@ def _mark_request_end() -> None: with _activity_lock: _active_request_count = max(0, _active_request_count - 1) _last_activity_at = time.monotonic() + update_current_daemon_status("main", touch=True, active_delta=-1) + + +def _shutdown_worker(reason: str) -> None: + logger.info("Shutting down OpenSpace MCP daemon: %s", reason) + instance = _openspace_instance + if instance is not None and instance.is_initialized(): + try: + asyncio.run(asyncio.wait_for(instance.cleanup(), timeout=10.0)) + except Exception as exc: + logger.warning("OpenSpace MCP cleanup during shutdown failed: %s", exc) + logging.shutdown() + os._exit(0) + + +def _begin_shutdown(reason: str) -> None: + global _shutdown_started + with _shutdown_lock: + if _shutdown_started: + return + _shutdown_started = True + + threading.Thread( + target=_shutdown_worker, + args=(reason,), + name="openspace-mcp-shutdown", + daemon=True, + ).start() + + +def _install_signal_handlers() -> None: + def _handle(signum, _frame) -> None: + try: + signame = signal.Signals(signum).name + except Exception: + signame = str(signum) + _begin_shutdown(f"signal {signame}") + + for signum in (signal.SIGTERM, signal.SIGINT): + try: + signal.signal(signum, _handle) + except Exception: + continue def _idle_watchdog_loop(idle_timeout_seconds: int) -> None: @@ -654,8 +692,8 @@ def _idle_watchdog_loop(idle_timeout_seconds: int) -> None: "MCP idle watchdog exiting process after %.1fs idle with no active requests", idle_for, ) - logging.shutdown() - os._exit(0) + _begin_shutdown(f"idle timeout after {idle_for:.1f}s") + return def _maybe_start_idle_watchdog() -> None: @@ -1142,6 +1180,7 @@ def run_mcp_server() -> None: args = parser.parse_args() if args.transport == "stdio" or os.environ.get("OPENSPACE_MCP_DAEMON") == "1": + _install_signal_handlers() _maybe_start_idle_watchdog() if args.transport == "streamable-http": _maybe_start_main_daemon_embedding_prewarm() diff --git a/openspace/mcp_stdio.py b/openspace/mcp_stdio.py new file mode 100644 index 0000000..f7a4ac5 --- /dev/null +++ b/openspace/mcp_stdio.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import os +import sys +from pathlib import Path +from typing import TextIO + + +_STDERR_CAPTURE_HANDLE: TextIO | None = None + + +def maybe_redirect_stderr_to_file(log_dir: Path, filename: str) -> TextIO | None: + """Redirect stderr to a log file when running as a non-interactive MCP child. + + Codex and similar MCP hosts typically do not surface or continuously drain + child-process stderr. Leaving verbose transport logs attached to a pipe can + back up the buffer and stall stdio tool calls. For interactive terminals we + keep stderr unchanged so local debugging still behaves normally. + """ + global _STDERR_CAPTURE_HANDLE + + if os.environ.get("OPENSPACE_MCP_CAPTURE_STDERR", "").strip().lower() in { + "0", + "false", + "no", + }: + return None + + if _STDERR_CAPTURE_HANDLE is not None: + return _STDERR_CAPTURE_HANDLE + + try: + if sys.stderr is not None and sys.stderr.isatty(): + return None + except Exception: + pass + + log_dir.mkdir(parents=True, exist_ok=True) + handle = (log_dir / filename).open("a", encoding="utf-8", buffering=1) + sys.stderr = handle + _STDERR_CAPTURE_HANDLE = handle + return handle diff --git a/openspace/shared_mcp_runtime.py b/openspace/shared_mcp_runtime.py index aa69d9d..971b6b2 100644 --- a/openspace/shared_mcp_runtime.py +++ b/openspace/shared_mcp_runtime.py @@ -82,6 +82,8 @@ class MCPDaemonRecord: ready_at: float | None = None warmed_at: float | None = None warmup_error: str | None = None + last_used_at: float | None = None + active_requests: int = 0 @property def url(self) -> str: @@ -260,6 +262,82 @@ def _write_record(path: Path, record: MCPDaemonRecord) -> None: tmp_path.replace(path) +def _record_last_used_at(record: MCPDaemonRecord) -> float: + return record.last_used_at or record.warmed_at or record.ready_at or record.started_at + + +def _max_daemons_per_kind() -> int: + raw = os.environ.get("OPENSPACE_MCP_MAX_DAEMONS_PER_KIND", "").strip() + if not raw: + return 8 + try: + return int(raw) + except ValueError: + logger.warning("Invalid OPENSPACE_MCP_MAX_DAEMONS_PER_KIND=%r", raw) + return 8 + + +def _unlink_record_artifacts(record: MCPDaemonRecord, state_dir: str) -> None: + metadata_path, lock_path = _metadata_paths(record.server_kind, record.instance_key, state_dir) + with contextlib.suppress(FileNotFoundError): + metadata_path.unlink() + with contextlib.suppress(FileNotFoundError): + lock_path.unlink() + + +def _collect_records(state_dir: str, server_kind: ServerKind) -> list[MCPDaemonRecord]: + state_path = Path(state_dir) + if not state_path.is_dir(): + return [] + + records: list[MCPDaemonRecord] = [] + for metadata_path in sorted(state_path.glob(f"{server_kind}-*.json")): + record = _read_record(metadata_path) + if record is not None: + records.append(record) + return records + + +def _reap_state_dir_records( + state_dir: str, + server_kind: ServerKind, + *, + keep_instance_key: str | None = None, +) -> None: + max_records = _max_daemons_per_kind() + live_records: list[MCPDaemonRecord] = [] + + for record in _collect_records(state_dir, server_kind): + if not _pid_exists(record.pid) or not _pid_matches_server(record): + _unlink_record_artifacts(record, state_dir) + continue + live_records.append(record) + + if max_records <= 0 or len(live_records) <= max_records: + return + + remaining = len(live_records) + for record in sorted(live_records, key=_record_last_used_at): + if remaining <= max_records: + break + if keep_instance_key and record.instance_key == keep_instance_key: + continue + if record.active_requests > 0: + continue + + logger.info( + "Reaping %s daemon pid=%s instance_key=%s last_used_at=%.3f to keep fleet <= %s", + record.server_kind, + record.pid, + record.instance_key, + _record_last_used_at(record), + max_records, + ) + _terminate_record_process(record) + _unlink_record_artifacts(record, state_dir) + remaining -= 1 + + def _metadata_paths( server_kind: ServerKind, instance_key: str, @@ -278,6 +356,8 @@ def update_current_daemon_status( ready: bool | None = None, warmed: bool | None = None, warmup_error: str | None = None, + touch: bool = False, + active_delta: int = 0, ) -> MCPDaemonRecord | None: instance_key = os.environ.get("OPENSPACE_MCP_INSTANCE_KEY", "").strip() state_dir = os.environ.get("OPENSPACE_MCP_DAEMON_STATE_DIR", "").strip() @@ -303,6 +383,12 @@ def update_current_daemon_status( if warmup_error is not None: updates["warmup_error"] = warmup_error + if active_delta: + updates["active_requests"] = max(0, record.active_requests + active_delta) + touch = True + if touch: + updates["last_used_at"] = now + if not updates: return record @@ -412,6 +498,7 @@ def _spawn_daemon(identity: MCPDaemonIdentity, port: int) -> MCPDaemonRecord: **popen_kwargs, ) log_handle.close() + started_at = time.time() return MCPDaemonRecord( server_kind=identity.server_kind, instance_key=identity.instance_key, @@ -423,8 +510,9 @@ def _spawn_daemon(identity: MCPDaemonIdentity, port: int) -> MCPDaemonRecord: backend_scope=list(identity.backend_scope), host_skill_dirs=list(identity.host_skill_dirs), grounding_config_fingerprint=identity.grounding_config_fingerprint, - started_at=time.time(), + started_at=started_at, log_path=str(log_path), + last_used_at=started_at, ) @@ -466,10 +554,15 @@ async def ensure_daemon(server_kind: ServerKind) -> MCPDaemonRecord: identity.metadata_path.parent.mkdir(parents=True, exist_ok=True) with _FileLock(identity.lock_path): + _reap_state_dir_records( + identity.state_dir, + server_kind, + keep_instance_key=identity.instance_key, + ) existing = _read_record(identity.metadata_path) if existing and _pid_exists(existing.pid) and await _probe_record(existing): + now = time.time() if not existing.ready or (server_kind != "main" and not existing.warmed): - now = time.time() refreshed = MCPDaemonRecord( **{ **asdict(existing), @@ -480,16 +573,23 @@ async def ensure_daemon(server_kind: ServerKind) -> MCPDaemonRecord: existing.warmed_at or (now if (existing.warmed or server_kind != "main") else None) ), + "last_used_at": now, } ) _write_record(identity.metadata_path, refreshed) return refreshed or existing - return existing + refreshed = MCPDaemonRecord( + **{ + **asdict(existing), + "last_used_at": now, + } + ) + _write_record(identity.metadata_path, refreshed) + return refreshed if existing: _terminate_record_process(existing) - with contextlib.suppress(FileNotFoundError): - identity.metadata_path.unlink() + _unlink_record_artifacts(existing, identity.state_dir) last_error: Exception | None = None for _ in range(3): @@ -504,6 +604,7 @@ async def ensure_daemon(server_kind: ServerKind) -> MCPDaemonRecord: "ready_at": now, "warmed": (server_kind != "main"), "warmed_at": (now if server_kind != "main" else None), + "last_used_at": now, } ) _write_record(identity.metadata_path, updated) @@ -513,7 +614,6 @@ async def ensure_daemon(server_kind: ServerKind) -> MCPDaemonRecord: f"Daemon for key={identity.instance_key} did not become ready" ) _terminate_record_process(record) - with contextlib.suppress(FileNotFoundError): - identity.metadata_path.unlink() + _unlink_record_artifacts(record, identity.state_dir) raise last_error or RuntimeError("Failed to start daemon") diff --git a/openspace/utils/logging.py b/openspace/utils/logging.py index 7382374..f164b7e 100644 --- a/openspace/utils/logging.py +++ b/openspace/utils/logging.py @@ -212,7 +212,9 @@ class Logger: # Console Handler if log_to_console: - ch = logging.StreamHandler(sys.stdout) + # Logs should never share stdout with structured program output + # such as MCP stdio frames. + ch = logging.StreamHandler(sys.stderr) ch.setLevel(resolved_level) ch.setFormatter(console_formatter) target_logger.addHandler(ch) @@ -229,7 +231,8 @@ class Logger: # Record log file location if not cls._configured: - print(f"Log file enabled: {actual_log_file}") + sys.stderr.write(f"Log file enabled: {actual_log_file}\n") + sys.stderr.flush() cls._configured = True @@ -293,7 +296,7 @@ class Logger: @staticmethod def _stdout_supports_color() -> bool: - return sys.stdout.isatty() and not os.getenv("NO_COLOR") + return sys.stderr.isatty() and not os.getenv("NO_COLOR") @classmethod def _resolve_level(cls, level: Optional[int]) -> int: diff --git a/scripts/cleanup_openspace_daemons.py b/scripts/cleanup_openspace_daemons.py new file mode 100755 index 0000000..94d3709 --- /dev/null +++ b/scripts/cleanup_openspace_daemons.py @@ -0,0 +1,324 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import signal +import subprocess +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Iterable + + +PROCESS_MARKERS = ( + "openspace.mcp_server", + "openspace.evolution_mcp_server", +) +STATE_FILE_PREFIXES = ("main-", "evolution-") +STATE_FILE_SUFFIXES = (".json", ".lock", ".log") + + +@dataclass +class ManagedProcess: + pid: int + command: str + source: str + state_dir: str | None = None + record_file: str | None = None + + +def default_state_dirs() -> list[Path]: + env_override = os.environ.get("OPENSPACE_MCP_DAEMON_STATE_DIR", "").strip() + candidates = [ + Path(env_override).expanduser().resolve() if env_override else None, + Path.home() / ".codex" / "state" / "openspace", + Path.home() / ".codex-openspace" / "state" / "openspace", + Path.home() / "Library" / "Application Support" / "openspace" / "mcp-daemons", + ] + + result: list[Path] = [] + seen: set[str] = set() + for path in candidates: + if path is None: + continue + key = str(path) + if key in seen: + continue + seen.add(key) + result.append(path) + return result + + +def is_target_command(command: str) -> bool: + return any(marker in command for marker in PROCESS_MARKERS) + + +def pid_exists(pid: int) -> bool: + if pid <= 0: + return False + try: + os.kill(pid, 0) + return True + except OSError: + return False + + +def command_for_pid(pid: int) -> str: + if pid <= 0: + return "" + proc = subprocess.run( + ["ps", "-o", "command=", "-p", str(pid)], + check=False, + capture_output=True, + text=True, + ) + return proc.stdout.strip() + + +def collect_metadata_processes(state_dir: Path) -> tuple[list[ManagedProcess], list[dict[str, object]], list[str]]: + managed: list[ManagedProcess] = [] + metadata_rows: list[dict[str, object]] = [] + warnings: list[str] = [] + + if not state_dir.is_dir(): + return managed, metadata_rows, warnings + + for metadata_path in sorted(state_dir.glob("*.json")): + try: + payload = json.loads(metadata_path.read_text(encoding="utf-8")) + except Exception as exc: # noqa: BLE001 + warnings.append(f"failed to read {metadata_path}: {exc}") + continue + + pid = int(payload.get("pid") or 0) + command = command_for_pid(pid) if pid_exists(pid) else "" + live_target = bool(command and is_target_command(command)) + metadata_rows.append( + { + "state_dir": str(state_dir), + "record_file": metadata_path.name, + "pid": pid, + "port": payload.get("port"), + "workspace": payload.get("workspace"), + "server_kind": payload.get("server_kind"), + "live_target": live_target, + "command": command, + } + ) + if live_target: + managed.append( + ManagedProcess( + pid=pid, + command=command, + source="metadata", + state_dir=str(state_dir), + record_file=metadata_path.name, + ) + ) + + return managed, metadata_rows, warnings + + +def collect_orphan_processes(excluded_pids: Iterable[int]) -> list[ManagedProcess]: + excluded = set(excluded_pids) + proc = subprocess.run( + ["ps", "-axo", "pid=,command="], + check=False, + capture_output=True, + text=True, + ) + + results: list[ManagedProcess] = [] + for line in proc.stdout.splitlines(): + stripped = line.strip() + if not stripped: + continue + try: + pid_text, command = stripped.split(None, 1) + except ValueError: + continue + pid = int(pid_text) + if pid in excluded: + continue + if not is_target_command(command): + continue + results.append(ManagedProcess(pid=pid, command=command, source="orphan-scan")) + return results + + +def terminate_process(pid: int, timeout_seconds: float) -> str: + if not pid_exists(pid): + return "already-exited" + + os.kill(pid, signal.SIGTERM) + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if not pid_exists(pid): + return "terminated" + time.sleep(0.1) + + if pid_exists(pid): + os.kill(pid, signal.SIGKILL) + deadline = time.monotonic() + 1.0 + while time.monotonic() < deadline: + if not pid_exists(pid): + return "killed" + time.sleep(0.05) + return "kill-sent" + + +def removable_state_files(state_dir: Path, keep_logs: bool) -> list[Path]: + if not state_dir.is_dir(): + return [] + + removable: list[Path] = [] + for path in sorted(state_dir.iterdir()): + if not path.is_file(): + continue + if not path.name.startswith(STATE_FILE_PREFIXES): + continue + if not path.name.endswith(STATE_FILE_SUFFIXES): + continue + if keep_logs and path.suffix == ".log": + continue + removable.append(path) + return removable + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Clean OpenSpace MCP daemon processes and state files") + parser.add_argument( + "--state-dir", + action="append", + default=[], + help="Additional state directory to clean. Can be passed multiple times.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be terminated/removed without changing anything.", + ) + parser.add_argument( + "--keep-logs", + action="store_true", + help="Keep *.log files in state dirs while removing json/lock artifacts.", + ) + parser.add_argument( + "--timeout-seconds", + type=float, + default=3.0, + help="How long to wait after SIGTERM before SIGKILL.", + ) + parser.add_argument( + "--json", + action="store_true", + help="Print the final report as JSON.", + ) + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + state_dirs = default_state_dirs() + for raw in args.state_dir: + state_dirs.append(Path(raw).expanduser().resolve()) + + deduped_state_dirs: list[Path] = [] + seen_state_dirs: set[str] = set() + for path in state_dirs: + key = str(path) + if key in seen_state_dirs: + continue + seen_state_dirs.add(key) + deduped_state_dirs.append(path) + + metadata_processes: list[ManagedProcess] = [] + metadata_rows: list[dict[str, object]] = [] + warnings: list[str] = [] + for state_dir in deduped_state_dirs: + managed, rows, row_warnings = collect_metadata_processes(state_dir) + metadata_processes.extend(managed) + metadata_rows.extend(rows) + warnings.extend(row_warnings) + + orphan_processes = collect_orphan_processes(proc.pid for proc in metadata_processes) + + unique_processes: dict[int, ManagedProcess] = {} + for proc in [*metadata_processes, *orphan_processes]: + unique_processes.setdefault(proc.pid, proc) + + process_actions: list[dict[str, object]] = [] + for proc in sorted(unique_processes.values(), key=lambda item: item.pid): + action = "would-terminate" if args.dry_run else terminate_process(proc.pid, args.timeout_seconds) + process_actions.append( + { + **asdict(proc), + "action": action, + } + ) + + file_actions: list[dict[str, object]] = [] + for state_dir in deduped_state_dirs: + for path in removable_state_files(state_dir, keep_logs=args.keep_logs): + action = "would-remove" + if not args.dry_run: + path.unlink(missing_ok=True) + action = "removed" + file_actions.append( + { + "state_dir": str(state_dir), + "path": str(path), + "action": action, + } + ) + + report = { + "state_dirs": [str(path) for path in deduped_state_dirs], + "metadata_records": metadata_rows, + "process_actions": process_actions, + "file_actions": file_actions, + "warnings": warnings, + "dry_run": args.dry_run, + } + + if args.json: + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + + print("OpenSpace daemon cleanup") + print(f"dry run: {'yes' if args.dry_run else 'no'}") + print("state dirs:") + for path in report["state_dirs"]: + print(f"- {path}") + + print("\nprocesses:") + if process_actions: + for item in process_actions: + print( + f"- pid={item['pid']} source={item['source']} action={item['action']} " + f"record={item.get('record_file') or '-'}" + ) + else: + print("- none") + + print("\nfiles:") + if file_actions: + for item in file_actions: + print(f"- {item['action']}: {item['path']}") + else: + print("- none") + + if warnings: + print("\nwarnings:") + for warning in warnings: + print(f"- {warning}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/test_codex_openspace_sessions.py b/scripts/test_codex_openspace_sessions.py new file mode 100644 index 0000000..b10103e --- /dev/null +++ b/scripts/test_codex_openspace_sessions.py @@ -0,0 +1,11 @@ +#!/usr/bin/env python3 +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from openspace.codex_session_scenarios import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_codex_session_scenarios.py b/tests/test_codex_session_scenarios.py new file mode 100644 index 0000000..2929579 --- /dev/null +++ b/tests/test_codex_session_scenarios.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from openspace.codex_session_scenarios import ( + cleanup_session_artifacts, + collect_session_family, + parse_exec_output, + snapshot_daemons, +) + + +def _write_session(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"type": "session_meta", "payload": payload}, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + + +def test_parse_exec_output_extracts_thread_and_opening_status() -> None: + output = """ +2026-04-13T00:00:00Z WARN unrelated noise +{"type":"thread.started","thread_id":"thread-parent"} +{"type":"item.completed","item":{"id":"item_1","type":"agent_message","text":"先做一次 OpenSpace 预检，再开始当前任务。\\n\\nOpenSpace session: ready\\nOpenSpace machine: ready"}} +{"type":"item.completed","item":{"id":"item_2","type":"agent_message","text":"cold-start done"}} +""".strip() + + result = parse_exec_output(output) + + assert result.thread_id == "thread-parent" + assert result.opening is not None + assert result.opening.session_status == "ready" + assert result.opening.machine_status == "ready" + assert result.opening.fallback_present is False + assert result.agent_messages[-1] == "cold-start done" + + +def test_collect_session_family_discovers_descendants(tmp_path: Path) -> None: + sessions_root = tmp_path / "sessions" + parent = sessions_root / "2026/04/13/parent.jsonl" + child = sessions_root / "2026/04/13/child.jsonl" + grandchild = sessions_root / "2026/04/13/grandchild.jsonl" + unrelated = sessions_root / "2026/04/13/unrelated.jsonl" + + _write_session(parent, {"id": "parent-thread"}) + _write_session( + child, + { + "id": "child-thread", + "source": { + "subagent": { + "thread_spawn": { + "parent_thread_id": "parent-thread", + } + } + }, + }, + ) + _write_session( + grandchild, + { + "id": "grandchild-thread", + "source": { + "subagent": { + "thread_spawn": { + "parent_thread_id": "child-thread", + } + } + }, + }, + ) + _write_session(unrelated, {"id": "other-thread"}) + + family = collect_session_family("parent-thread", sessions_root) + + assert family.thread_ids == {"parent-thread", "child-thread", "grandchild-thread"} + assert family.session_files == {parent, child, grandchild} + + +def test_cleanup_session_artifacts_removes_files_and_index_rows(tmp_path: Path) -> None: + sessions_root = tmp_path / "sessions" + session_file = sessions_root / "2026/04/13/parent.jsonl" + _write_session(session_file, {"id": "parent-thread"}) + + session_index = tmp_path / "session_index.jsonl" + session_index.write_text( + "\n".join( + [ + json.dumps({"thread_id": "parent-thread", "session_file": str(session_file)}), + json.dumps({"thread_id": "keep-thread", "session_file": "/tmp/keep.jsonl"}), + ] + ) + + "\n", + encoding="utf-8", + ) + + cleanup_session_artifacts( + thread_ids={"parent-thread"}, + session_files={session_file}, + session_index_path=session_index, + ) + + assert not session_file.exists() + remaining = session_index.read_text(encoding="utf-8") + assert "parent-thread" not in remaining + assert "keep-thread" in remaining + + +def test_snapshot_daemons_filters_by_workspace(tmp_path: Path) -> None: + state_dir = tmp_path / "state" + state_dir.mkdir() + matching = state_dir / "main-match.json" + other = state_dir / "main-other.json" + + matching.write_text( + json.dumps( + { + "server_kind": "main", + "instance_key": "match", + "pid": 111, + "port": 9001, + "workspace": "/tmp/workspace-a", + "ready": True, + } + ), + encoding="utf-8", + ) + other.write_text( + json.dumps( + { + "server_kind": "main", + "instance_key": "other", + "pid": 222, + "port": 9002, + "workspace": "/tmp/workspace-b", + "ready": True, + } + ), + encoding="utf-8", + ) + + snapshot = snapshot_daemons(state_dir, "/tmp/workspace-a") + + assert set(snapshot) == {"main"} + assert snapshot["main"]["pid"] == 111 diff --git a/tests/test_logging_stdout_safety.py b/tests/test_logging_stdout_safety.py new file mode 100644 index 0000000..1f028cf --- /dev/null +++ b/tests/test_logging_stdout_safety.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import io +import logging +from pathlib import Path + +from openspace.utils.logging import Logger + + +def test_log_file_enable_announcement_avoids_stdout(monkeypatch, tmp_path: Path) -> None: + stdout = io.StringIO() + stderr = io.StringIO() + + monkeypatch.setattr("sys.stdout", stdout) + monkeypatch.setattr("sys.stderr", stderr) + + Logger.reset_configuration() + Logger.configure( + level=logging.INFO, + log_to_console=False, + log_to_file=str(tmp_path / "unit.log"), + attach_to_root=True, + ) + + assert stdout.getvalue() == "" + assert "Log file enabled:" in stderr.getvalue() + + Logger.reset_configuration() + + +def test_console_logging_avoids_stdout(monkeypatch, tmp_path: Path) -> None: + stdout = io.StringIO() + stderr = io.StringIO() + + monkeypatch.setattr("sys.stdout", stdout) + monkeypatch.setattr("sys.stderr", stderr) + + Logger.reset_configuration() + Logger.configure( + level=logging.INFO, + log_to_console=True, + log_to_file=str(tmp_path / "unit.log"), + attach_to_root=True, + ) + Logger.get_logger("openspace.test").info("console log should stay off stdout") + + assert stdout.getvalue() == "" + assert "console log should stay off stdout" in stderr.getvalue() + + Logger.reset_configuration() diff --git a/tests/test_mcp_entrypoints.py b/tests/test_mcp_entrypoints.py index 4fe6f63..ba03e4a 100644 --- a/tests/test_mcp_entrypoints.py +++ b/tests/test_mcp_entrypoints.py @@ -34,6 +34,11 @@ def test_stdio_entrypoint_uses_stdio_transport(module_name, monkeypatch) -> None "_maybe_start_idle_watchdog", lambda: watchdog_calls.append(True), ) + monkeypatch.setattr( + module, + "_install_signal_handlers", + lambda: None, + ) module.run_mcp_server() @@ -63,6 +68,11 @@ def test_sse_entrypoint_does_not_forward_sse_params(module_name, monkeypatch) -> "_maybe_start_idle_watchdog", lambda: watchdog_calls.append(True), ) + monkeypatch.setattr( + module, + "_install_signal_handlers", + lambda: None, + ) module.run_mcp_server() @@ -95,6 +105,11 @@ def test_streamable_http_entrypoint_uses_watchdog_for_daemon( "_maybe_start_idle_watchdog", lambda: watchdog_calls.append(True), ) + monkeypatch.setattr( + module, + "_install_signal_handlers", + lambda: None, + ) monkeypatch.setenv("OPENSPACE_MCP_DAEMON", "1") module.run_mcp_server() diff --git a/tests/test_mcp_stdio.py b/tests/test_mcp_stdio.py new file mode 100644 index 0000000..fecd473 --- /dev/null +++ b/tests/test_mcp_stdio.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import io +from pathlib import Path + +from openspace import mcp_stdio + + +class _FakeStream(io.StringIO): + def __init__(self, *, tty: bool) -> None: + super().__init__() + self._tty = tty + + def isatty(self) -> bool: + return self._tty + + +def _reset_capture() -> None: + handle = mcp_stdio._STDERR_CAPTURE_HANDLE + if handle is not None: + handle.close() + mcp_stdio._STDERR_CAPTURE_HANDLE = None + + +def test_redirects_stderr_for_non_interactive_sessions(monkeypatch, tmp_path: Path) -> None: + _reset_capture() + monkeypatch.setattr("sys.stderr", _FakeStream(tty=False)) + + handle = mcp_stdio.maybe_redirect_stderr_to_file(tmp_path, "stderr.log") + + assert handle is not None + assert Path(handle.name) == tmp_path / "stderr.log" + assert Path(handle.name).exists() + assert handle is mcp_stdio._STDERR_CAPTURE_HANDLE + + _reset_capture() + + +def test_keeps_stderr_for_interactive_sessions(monkeypatch, tmp_path: Path) -> None: + _reset_capture() + original = _FakeStream(tty=True) + monkeypatch.setattr("sys.stderr", original) + + handle = mcp_stdio.maybe_redirect_stderr_to_file(tmp_path, "stderr.log") + + assert handle is None + assert mcp_stdio._STDERR_CAPTURE_HANDLE is None + assert original is not None diff --git a/tests/test_shared_mcp_runtime_metadata.py b/tests/test_shared_mcp_runtime_metadata.py index bb2f4fc..0e0405d 100644 --- a/tests/test_shared_mcp_runtime_metadata.py +++ b/tests/test_shared_mcp_runtime_metadata.py @@ -146,3 +146,90 @@ def test_update_current_daemon_status_timestamps_after_lock_wait(monkeypatch, tm assert updated is not None assert updated.warmed is True assert updated.warmed_at == 107.5 + + +def test_update_current_daemon_status_tracks_last_used_and_active_requests(monkeypatch, tmp_path) -> None: + identity = _build_identity(tmp_path) + metadata_path = identity.metadata_path + metadata_path.parent.mkdir(parents=True, exist_ok=True) + shared_mcp_runtime._write_record(metadata_path, _build_record(identity)) + + monkeypatch.setenv("OPENSPACE_MCP_INSTANCE_KEY", identity.instance_key) + monkeypatch.setenv("OPENSPACE_MCP_DAEMON_STATE_DIR", identity.state_dir) + + monkeypatch.setattr(shared_mcp_runtime.time, "time", lambda: 111.0) + started = shared_mcp_runtime.update_current_daemon_status( + "main", + touch=True, + active_delta=1, + ) + assert started is not None + assert started.active_requests == 1 + assert started.last_used_at == 111.0 + + monkeypatch.setattr(shared_mcp_runtime.time, "time", lambda: 114.5) + finished = shared_mcp_runtime.update_current_daemon_status( + "main", + touch=True, + active_delta=-1, + ) + assert finished is not None + assert finished.active_requests == 0 + assert finished.last_used_at == 114.5 + + +def test_reap_state_dir_records_limits_live_daemons_per_kind(monkeypatch, tmp_path) -> None: + records = [] + for index, last_used_at in enumerate((10.0, 20.0, 30.0), start=1): + identity = shared_mcp_runtime.MCPDaemonIdentity( + server_kind="main", + workspace=f"/tmp/workspace-{index}", + resolved_model="unit-model", + llm_kwargs_fingerprint=f"llm-{index}", + backend_scope=("shell",), + host_skill_dirs=(f"/tmp/skills-{index}",), + grounding_config_fingerprint=f"cfg-{index}", + instance_key=f"key-{index}", + state_dir=str(tmp_path), + ) + record = shared_mcp_runtime.MCPDaemonRecord( + server_kind="main", + instance_key=identity.instance_key, + pid=4000 + index, + port=56000 + index, + workspace=identity.workspace, + resolved_model=identity.resolved_model, + llm_kwargs_fingerprint=identity.llm_kwargs_fingerprint, + backend_scope=list(identity.backend_scope), + host_skill_dirs=list(identity.host_skill_dirs), + grounding_config_fingerprint=identity.grounding_config_fingerprint, + started_at=1.0, + log_path=str(tmp_path / f"{identity.instance_key}.log"), + ready=True, + warmed=True, + last_used_at=last_used_at, + active_requests=0, + ) + shared_mcp_runtime._write_record(identity.metadata_path, record) + records.append(record) + + reaped: list[str] = [] + monkeypatch.setattr(shared_mcp_runtime, "_max_daemons_per_kind", lambda: 2) + monkeypatch.setattr(shared_mcp_runtime, "_pid_exists", lambda pid: True) + monkeypatch.setattr(shared_mcp_runtime, "_pid_matches_server", lambda record: True) + monkeypatch.setattr( + shared_mcp_runtime, + "_terminate_record_process", + lambda record: reaped.append(record.instance_key), + ) + + shared_mcp_runtime._reap_state_dir_records( + str(tmp_path), + "main", + keep_instance_key="key-3", + ) + + assert reaped == ["key-1"] + assert not (tmp_path / "main-key-1.json").exists() + assert (tmp_path / "main-key-2.json").exists() + assert (tmp_path / "main-key-3.json").exists()