From 0355af7833283c2dc761a9bccf59be89f1671448 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E9=93=AD=E8=BD=A9?= Date: Thu, 8 Oct 2026 17:33:48 +0800 Subject: [PATCH] fix: preserve live tasks during daemon recovery --- .env.example | 2 + apps/admin_console/server.py | 44 ++++--- .../services/automation_service.py | 73 +++++++++++- .../services/automation_store.py | 6 + apps/admin_console/services/report_export.py | 4 + artemis/config/settings.py | 1 + artemis/runtime/daemon_client.py | 17 ++- docs/automation-workspace.md | 4 +- tests/unit/admin_console/test_automation.py | 110 ++++++++++++++++++ .../admin_console/test_server_lifecycle.py | 46 ++++++++ tests/unit/runtime/test_daemon_client.py | 37 ++++++ 11 files changed, 327 insertions(+), 17 deletions(-) diff --git a/.env.example b/.env.example index 6a7a36d..2cce48c 100644 --- a/.env.example +++ b/.env.example @@ -50,6 +50,8 @@ ARTEMIS_CHECKPOINT_TIMEOUT_SECONDS=180 ARTEMIS_SETTLEMENT_TIMEOUT_SECONDS=120 ARTEMIS_CREDENTIAL_TIMEOUT_SECONDS=12 ARTEMIS_MODEL_CACHE_TTL_SECONDS=10 +# 守护服务轻量探测超时；探测超时且端口仍占用时，不重复启动服务。 +ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS=2 # 模型重试耗尽后的暂停等待上限（秒）；0 表示无限等待 LLM_PAUSE_TIMEOUT_SECONDS=900 diff --git a/apps/admin_console/server.py b/apps/admin_console/server.py index 47d7c52..f5cb9c8 100644 --- a/apps/admin_console/server.py +++ b/apps/admin_console/server.py @@ -24,6 +24,7 @@ import os from pathlib import Path import secrets import signal +import socket import sys from types import FrameType @@ -421,6 +422,14 @@ queue_goals = state.queue_goals class ArtemisUvicornServer(uvicorn.Server): """Notify application streams before Uvicorn waits for them to close.""" + async def startup(self, sockets: list[socket.socket] | None = None) -> None: + # Reload workers start in a fresh interpreter; restore the actual bind + # address there before lifespan publishes runtime metadata. + state.host = self.config.host + state.port = self.config.port + app.state.uvicorn_server = self + await super().startup(sockets=sockets) + @staticmethod def _task_is_active() -> bool: """Check task state without doing process I/O from a signal handler.""" @@ -470,31 +479,40 @@ class ArtemisUvicornServer(uvicorn.Server): def run_ui_server(host: str, port: int, reload: bool = False) -> None: """Run the UI server with bounded, signal-aware graceful shutdown.""" + # Reserve a listening socket before publishing metadata or running lifespan. + # Uvicorn normally runs application startup *before* binding its port; a + # duplicate daemon would otherwise recover jobs owned by the real server. + family = socket.AF_INET6 if ":" in host else socket.AF_INET + try: + listener = socket.create_server((host, port), family=family) + except OSError as exc: + print(f"无法启动服务：{host}:{port} 端口不可用（{exc}）。现有服务和任务保持不变。") + raise SystemExit(1) from exc + listener.set_inheritable(True) + port = listener.getsockname()[1] state.host = host state.port = port - write_server_info(port=port, host=host, lifecycle_token=LIFECYCLE_TOKEN) try: - if reload: - uvicorn.run( - "apps.admin_console.server:app", - host=host, - port=port, - reload=True, - timeout_graceful_shutdown=5, - ) - return - config = uvicorn.Config( - app, + "apps.admin_console.server:app" if reload else app, host=host, port=port, + reload=reload, timeout_graceful_shutdown=5, ) + if reload: + from uvicorn.supervisors import ChangeReload + + server = ArtemisUvicornServer(config) + ChangeReload(config, target=server.run, sockets=[listener]).run() + return + server = ArtemisUvicornServer(config) app.state.uvicorn_server = server - server.run() + server.run(sockets=[listener]) finally: + listener.close() app.state.uvicorn_server = None clear_server_info(port=port, lifecycle_token=LIFECYCLE_TOKEN) diff --git a/apps/admin_console/services/automation_service.py b/apps/admin_console/services/automation_service.py index 6d32584..9f32558 100644 --- a/apps/admin_console/services/automation_service.py +++ b/apps/admin_console/services/automation_service.py @@ -8,7 +8,7 @@ import time from uuid import UUID from apps.admin_console.core.state import state -from apps.admin_console.services.automation_store import store, new_id +from apps.admin_console.services.automation_store import RECOVERY_ERRORS, store, new_id from apps.admin_console.services.task_import import case_goal, case_title from artemis.config.settings import settings from artemis.runtime.task_limits import task_timeout @@ -42,9 +42,46 @@ def register_submission(item: dict, case: dict | None = None, formats=None, **ex ) +def resolve_recovery_error(job: dict) -> bool: + """Archive a superseded recovery warning without erasing execution failures.""" + if job.get("error") not in RECOVERY_ERRORS: + return False + job.setdefault("recovery_events", []).append( + {"message": job.pop("error"), "resolved_at": time.time()} + ) + return True + + def recover_batches(): """Recovery never silently repeats a potentially side-effecting interrupted task.""" + from apps.admin_console.database.repositories.session_repository import SessionRepository + from artemis.runtime import DeviceExecutionLock + + repository = SessionRepository(store.path) + live_sessions = { + str(owner.session_id) + for owner in DeviceExecutionLock.get_active_owners().values() + if owner.session_id + } + live_sessions.update( + str(row["session_id"]) + for row in repository.get_all_sessions() + if row.get("status") not in TERMINAL and repository.process_is_alive(row.get("pid")) + ) + live_sessions.update( + str(item["session_id"]) + for item in state.queue_items + if isinstance(item, dict) and item.get("session_id") and item.get("status") not in TERMINAL + ) for job in store.jobs(): + if job["status"] == "completed" and resolve_recovery_error(job): + store.save_job(job) + if str(job["session_id"]) in live_sessions: + if job["status"] not in TERMINAL and str(job["session_id"]) not in state.active_runs: + job["recovery_waiting"] = True + store.save_job(job) + logger.info("Recovery preserved live session %s", job["session_id"]) + continue if job["status"] in {"running", "queued", "finalizing", "task_paused", "resuming"}: job.update( status=job.get("terminal_status", "interrupted"), @@ -52,6 +89,9 @@ def recover_batches(): or ("" if job.get("terminal_status") else "服务中断；未完成任务需手动重跑"), ended_at=job.get("ended_at") or time.time(), ) + if job["status"] == "completed": + resolve_recovery_error(job) + logger.info("Recovery resolved session %s to %s", job["session_id"], job["status"]) store.save_job(job) for batch in store.records("batch"): if has_legacy_replay(batch["id"]): @@ -64,6 +104,8 @@ def recover_batches(): store.put("batch", batch["id"], batch) continue if batch["status"] in {"running", "pausing", "paused"}: + if any(str(job["session_id"]) in live_sessions for job in store.jobs(batch["id"])): + continue batch.update( status="paused", reason="服务已重启；请检查中断任务，点击继续执行未开始任务" ) @@ -88,6 +130,9 @@ async def on_terminal(item: dict, status: str): raise RuntimeError("任务元数据写入失败") if job.get("timed_out"): status = "failed" + job.pop("recovery_waiting", None) + if status == "completed": + resolve_recovery_error(job) job.update(status="finalizing", terminal_status=status, ended_at=time.time()) store.save_job(job) try: @@ -117,11 +162,37 @@ async def on_terminal(item: dict, status: str): store.save_job(job) +async def reconcile_recovered_jobs(): + """Finalize surviving workers once they exit; never resubmit their goals.""" + from apps.admin_console.database.repositories.session_repository import SessionRepository + from artemis.runtime import DeviceExecutionLock + from artemis.runtime.task_result import read_worker_outcome + + waiting = [job for job in store.jobs() if job.get("recovery_waiting")] + if not waiting: + return + repository = SessionRepository(store.path) + owners = {str(owner.session_id) for owner in DeviceExecutionLock.get_active_owners().values()} + for job in waiting: + sid = str(job["session_id"]) + row = repository.get_session_by_id(sid) or {} + if sid in owners or repository.process_is_alive(row.get("pid")): + continue + outcome = read_worker_outcome(sid) + status = job.get("terminal_status") or outcome.get("task_status") or row.get("status") + if status not in TERMINAL: + status = "interrupted" + job["error"] = job.get("error") or "服务中断；未完成任务需手动重跑" + store.save_job(job) + await on_terminal({"session_id": sid}, status) + + async def tick_batches(): from apps.admin_console.server import task_queue_service from artemis.runtime import device_pool from apps.admin_console.services.report_export import export_batch + await reconcile_recovered_jobs() for batch in store.records("batch"): if batch["status"] not in {"running", "paused"}: continue diff --git a/apps/admin_console/services/automation_store.py b/apps/admin_console/services/automation_store.py index 40b0318..73182b1 100644 --- a/apps/admin_console/services/automation_store.py +++ b/apps/admin_console/services/automation_store.py @@ -12,6 +12,12 @@ import uuid from artemis.config import DB_PATH +RECOVERY_ERRORS = { + "服务中断；未完成任务需手动重跑", + "服务中断，需手动重跑", +} + + def identifier(value: str) -> str: if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", value): raise ValueError("无效的记录编号") diff --git a/apps/admin_console/services/report_export.py b/apps/admin_console/services/report_export.py index 50d595f..1267845 100644 --- a/apps/admin_console/services/report_export.py +++ b/apps/admin_console/services/report_export.py @@ -88,6 +88,10 @@ def failure_reason(directory: Path, status: str, recorded: str | None) -> str: ) return safe[:500] + from apps.admin_console.services.automation_store import RECOVERY_ERRORS + + if status == "completed" and recorded in RECOVERY_ERRORS: + return "" if recorded: return redact(recorded) if status not in {"failed", "interrupted"}: diff --git a/artemis/config/settings.py b/artemis/config/settings.py index ea49c13..f1670ab 100644 --- a/artemis/config/settings.py +++ b/artemis/config/settings.py @@ -108,6 +108,7 @@ class Settings(BaseSettings): OPENAI_REASONING_EFFORT: str = "high" ARTEMIS_CREDENTIAL_TIMEOUT_SECONDS: float = Field(default=12.0, gt=0) ARTEMIS_MODEL_CACHE_TTL_SECONDS: float = Field(default=10.0, ge=0) + ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS: float = Field(default=2.0, gt=0) ARTEMIS_VIDEO_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0) ARTEMIS_VIDEO_MAX_RETRIES: int = Field(default=1, ge=0) ARTEMIS_VIDEO_CHUNK_SECONDS: float = Field(default=60.0, gt=0) diff --git a/artemis/runtime/daemon_client.py b/artemis/runtime/daemon_client.py index 4f8d289..a226f39 100644 --- a/artemis/runtime/daemon_client.py +++ b/artemis/runtime/daemon_client.py @@ -28,6 +28,9 @@ import urllib.error import urllib.request from artemis.config.paths import ROOT_DIR +from artemis.config.settings import settings +from artemis.runtime.process_probe import pid_is_alive +from artemis.runtime.server_lifecycle import is_port_in_use, read_server_info from artemis.utils.logger import get_logger logger = get_logger(__name__) @@ -44,10 +47,11 @@ def is_standalone_forced() -> bool: def is_daemon_running( host: str = DEFAULT_DAEMON_HOST, port: int = DEFAULT_DAEMON_PORT, - timeout: float = 0.3, + timeout: float | None = None, ) -> bool: """Fast probe to check if the Artemis Daemon HTTP server is responding.""" - url = f"http://{host}:{port}/api/system/readiness" + timeout = timeout if timeout is not None else settings.ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS + url = f"http://{host}:{port}/api/system/server-status" try: req = urllib.request.Request(url, headers={"User-Agent": "Artemis-Daemon-Probe"}) with urllib.request.urlopen(req, timeout=timeout) as response: @@ -190,6 +194,15 @@ def ensure_daemon_running( if is_daemon_running(host, port): return True, f"http://{host}:{port}" + if is_port_in_use(port, host): + # A slow/busy HTTP loop is not a stopped daemon. Only trust persisted + # metadata belonging to a live process on the requested endpoint. + info = read_server_info() or {} + if info.get("port") == port and info.get("host") == host and pid_is_alive(info.get("pid")): + return True, f"http://{host}:{port}" + logger.warning(f"Port {host}:{port} is occupied; refusing to spawn another daemon") + return False, None + # Auto-spawn Daemon in background spawn_daemon(host, port) diff --git a/docs/automation-workspace.md b/docs/automation-workspace.md index 3def2bc..5901d7a 100644 --- a/docs/automation-workspace.md +++ b/docs/automation-workspace.md @@ -22,7 +22,7 @@ 同一批次按顺序派发。默认失败后继续且不自动重试，也可以选择失败即停止。当前任务因设备异常或超时失败后，默认继续后续用例；设置失败即停止时才停止后续派发。暂停批次只停止后续派发。单个任务可在安全边界暂停并释放设备；批次下一项可继续，暂停的任务待提交指导或点击继续后在设备队列末尾恢复。运行中提交指导会中断当前模型思考或只读诊断，重新观察画面；已经开始的设备动作会安全结束。网页流程会显示指导的提交与领取状态，仍应检查后续动作是否符合指导。暂停等待和重新排队时间不计入该任务执行时限。 -服务重启后保留批次，等待用户继续未开始的用例。正在执行或暂停的任务标记中断，不自动重做；“复制失败项为新批次”保留原记录并产生新任务编号。 +服务启动前先取得监听端口，重复启动不会修改现有任务或服务信息。恢复时核对原任务进程和设备锁：仍存活的任务继续等待其真实结果，退出后补齐报告；已经退出且没有终态的任务标记中断，不自动重做，批次等待用户继续未开始的用例。“复制失败项为新批次”保留原记录并产生新任务编号。任务正常完成后，已失效的“服务中断”提示归档为恢复事件，不再显示为当前失败原因；启动时也会纠正已有完成记录中的此类旧提示。 每个任务结束后生成所选格式报告。批次 ZIP 包含汇总、每项任务的报告和已保存的有序步骤截图；已开启录屏且文件存在时，还会按任务收录录屏片段。失败、跳过和中断任务也会列入。暂停时下载的是当前快照。录屏与 AI 视频分析分别配置，录屏默认关闭；仅开启分析时，录制内容用于本次分析，结束后不保留下载文件。暂停期间不录制其他任务的设备画面。 @@ -36,6 +36,8 @@ `.env` 可配置 `ARTEMIS_TASK_TIMEOUT_SECONDS`（单任务总执行时限，默认 1800 秒）、`ARTEMIS_IMPORT_MAX_ROWS`（默认 1000 行）和 `ARTEMIS_IMPORT_MAX_BYTES`（默认 10 MiB）。修改后重启服务。网页的单批次超时设置优先于环境默认值，每行 JSON 的 `timeout_seconds` 优先于批次。单任务、批次和 MCP 均受总执行时限保护，不计排队和用户暂停时间；即使执行进程卡住，也由父进程终止并保存失败原因。 +守护服务通过轻量状态接口探测，`ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS` 默认 2 秒。短暂超时且端口仍被存活的原服务占用时复用原服务，不重复启动；未知进程占用端口时明确拒绝覆盖。 + 网页“功能配置 → 执行保护”及 MCP `mobile_config(section="capabilities")` 可配置总时限、设备观测时限（默认 60 秒）、设备观测次数（默认 3 次）、ADB 命令超时（默认 10 秒）、连续动作失败次数（默认 3 次）。任务 JSON 可用 `observation_timeout_seconds`、`observation_max_attempts`、`device_command_timeout_seconds`、`action_failure_attempts` 覆盖。截图失效或收到占位图时先有限重取，耗尽后结束任务，不把旧画面发给模型；有效截图即使没有 XML 也可按坐标定位。连续动作失败时先刷新画面再尝试兜底，兜底耗尽后结束，不循环重提用例。 批次、任务元数据和报告偏好保存在现有 SQLite 数据库的 `automation_records`、`automation_jobs` 表。全局“清空任务历史”会清除批次与执行元数据；它与上文的单批次删除不同，已导出的文件会保留。 diff --git a/tests/unit/admin_console/test_automation.py b/tests/unit/admin_console/test_automation.py index 03140c2..0dd1675 100644 --- a/tests/unit/admin_console/test_automation.py +++ b/tests/unit/admin_console/test_automation.py @@ -300,6 +300,116 @@ def test_restart_never_repeats_started_work(isolated): assert isolated.jobs(retry["id"])[0]["session_id"] != job["session_id"] +@pytest.mark.parametrize("liveness_source", ["pid", "device_lock"]) +def test_recovery_preserves_live_runner_and_batch(isolated, monkeypatch, liveness_source): + import os + from types import SimpleNamespace + from artemis.runtime import DeviceExecutionLock + + batch = service.create_batch(payload()) + service.batch_action(batch["id"], "resume") + job = isolated.jobs(batch["id"])[0] + isolated.save_job({**job, "status": "running"}) + if liveness_source == "pid": + with sqlite3.connect(isolated.path) as conn: + conn.execute( + "INSERT INTO sessions(session_id,initial_goal,start_time,status,device_info,pid) " + "VALUES (?,?,?,?,?,?)", + (job["session_id"], "goal", 1, "running", "{}", os.getpid()), + ) + monkeypatch.setattr( + DeviceExecutionLock, + "get_active_owners", + lambda: ( + {"device": SimpleNamespace(session_id=job["session_id"])} + if liveness_source == "device_lock" + else {} + ), + ) + service.recover_batches() + assert isolated.job(job["session_id"])["status"] == "running" + assert not isolated.job(job["session_id"]).get("error") + assert isolated.get("batch", batch["id"])["status"] == "running" + + +@pytest.mark.asyncio +async def test_natural_completion_resolves_old_recovery_error_in_job_report_and_history( + client, isolated +): + job = isolated.jobs(service.create_batch(payload())["id"])[0] + message = "服务中断；未完成任务需手动重跑" + isolated.save_job({**job, "status": "interrupted", "error": message}) + await service.on_terminal({"session_id": job["session_id"]}, "completed") + saved = isolated.job(job["session_id"]) + assert saved["status"] == "completed" + assert not saved.get("error") + assert saved["recovery_events"][0]["message"] == message + assert reports.session_report(job["session_id"])["error"] == "" + assert not client.get("/api/automation/history").json()["items"][0].get("error") + + +def test_recovery_repairs_completed_legacy_warning_but_preserves_real_failure(isolated): + batch = service.create_batch(payload()) + first, second = isolated.jobs(batch["id"]) + isolated.save_job({**first, "status": "completed", "error": "服务中断，需手动重跑"}) + isolated.save_job({**second, "status": "failed", "error": "任务执行超时（30 秒）"}) + assert reports.session_report(first["session_id"])["error"] == "" + service.recover_batches() + assert not isolated.job(first["session_id"]).get("error") + assert "服务中断" in isolated.job(first["session_id"])["recovery_events"][0]["message"] + assert "超时" in isolated.job(second["session_id"])["error"] + + +@pytest.mark.asyncio +async def test_surviving_worker_is_finalized_when_it_exits_without_resubmission( + isolated, monkeypatch +): + import os + from artemis.runtime import DeviceExecutionLock + + batch = service.create_batch(payload()) + service.batch_action(batch["id"], "resume") + job = isolated.jobs(batch["id"])[0] + isolated.save_job({**job, "status": "running"}) + monkeypatch.setattr(DeviceExecutionLock, "get_active_owners", lambda: {}) + with sqlite3.connect(isolated.path) as conn: + conn.execute( + "INSERT INTO sessions(session_id,initial_goal,start_time,status,device_info,pid) " + "VALUES (?,?,?,?,?,?)", + (job["session_id"], "goal", time.time(), "running", "{}", os.getpid()), + ) + service.recover_batches() + await service.reconcile_recovered_jobs() + assert isolated.job(job["session_id"])["status"] == "running" + with sqlite3.connect(isolated.path) as conn: + conn.execute( + "UPDATE sessions SET pid=NULL,status='completed' WHERE session_id=?", + (job["session_id"],), + ) + await service.reconcile_recovered_jobs() + saved = isolated.job(job["session_id"]) + assert saved["status"] == "completed" + assert saved["report_ready"] + assert not saved.get("recovery_waiting") + assert not saved.get("error") + assert len(isolated.jobs(batch["id"])) == 2 + assert isolated.jobs(batch["id"])[1]["status"] == "pending" + + +@pytest.mark.asyncio +async def test_surviving_worker_that_dies_without_result_is_interrupted(isolated, monkeypatch): + from artemis.runtime import DeviceExecutionLock + + job = isolated.jobs(service.create_batch(payload())["id"])[0] + isolated.save_job({**job, "status": "running", "recovery_waiting": True}) + monkeypatch.setattr(DeviceExecutionLock, "get_active_owners", lambda: {}) + await service.reconcile_recovered_jobs() + saved = isolated.job(job["session_id"]) + assert saved["status"] == "interrupted" + assert "服务中断" in saved["error"] + assert not saved.get("recovery_waiting") + + @pytest.mark.asyncio async def test_batch_admission_idempotent_and_bound_device(isolated, monkeypatch): from apps.admin_console import server diff --git a/tests/unit/admin_console/test_server_lifecycle.py b/tests/unit/admin_console/test_server_lifecycle.py index e0171c8..669866d 100644 --- a/tests/unit/admin_console/test_server_lifecycle.py +++ b/tests/unit/admin_console/test_server_lifecycle.py @@ -15,6 +15,7 @@ import asyncio from contextlib import suppress import signal +import socket from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -23,10 +24,55 @@ import uvicorn from apps.admin_console.core.state import state from apps.admin_console.routers.tasks import stream_events from apps.admin_console.server import ArtemisUvicornServer, app, on_shutdown +from apps.admin_console import server as server_module WINDOWS_FORCE_SIGNAL = getattr(signal, "SIGBREAK", signal.SIGTERM) +@pytest.mark.parametrize("reload", [False, True]) +def test_occupied_port_never_initializes_or_overwrites_server_metadata( + tmp_path, monkeypatch, reload +): + from artemis.runtime import server_lifecycle + + metadata = tmp_path / "server.json" + metadata.write_text('{"pid":123,"lifecycle_token":"original"}', encoding="utf-8") + monkeypatch.setattr(server_lifecycle, "get_server_info_file", lambda: metadata) + before = metadata.read_bytes() + with ( + socket.create_server(("127.0.0.1", 0)) as original, + patch.object(server_module.ArtemisUvicornServer, "run") as run, + patch.object(server_module, "write_server_info") as publish, + ): + with pytest.raises(SystemExit, match="1"): + server_module.run_ui_server("127.0.0.1", original.getsockname()[1], reload=reload) + run.assert_not_called() + publish.assert_not_called() + assert metadata.read_bytes() == before + + +def test_server_holds_port_before_running_lifespan(monkeypatch): + monkeypatch.setattr(state, "port", 0, raising=False) + monkeypatch.setattr(state, "host", "127.0.0.1", raising=False) + observed = [] + + def run(_server, sockets): + listener = sockets[0] + port = listener.getsockname()[1] + observed.append(listener) + assert state.port == port + with pytest.raises(OSError), socket.create_server(("127.0.0.1", port)): + pass + + monkeypatch.setattr(server_module.ArtemisUvicornServer, "run", run) + monkeypatch.setattr(server_module, "clear_server_info", lambda **kwargs: None) + with patch.object(server_module, "write_server_info") as publish: + server_module.run_ui_server("127.0.0.1", 0) + publish.assert_not_called() # Publication belongs to the later lifespan. + assert len(observed) == 1 + assert observed[0].fileno() == -1 + + def _pending_queue_get_tasks(): return { task diff --git a/tests/unit/runtime/test_daemon_client.py b/tests/unit/runtime/test_daemon_client.py index 67a90d3..f1c3484 100644 --- a/tests/unit/runtime/test_daemon_client.py +++ b/tests/unit/runtime/test_daemon_client.py @@ -93,6 +93,7 @@ def test_ensure_daemon_running_standalone_override(monkeypatch): def test_ensure_daemon_running_auto_spawns(): with ( patch("artemis.runtime.daemon_client.is_daemon_running", side_effect=[False, True]), + patch("artemis.runtime.daemon_client.is_port_in_use", return_value=False), patch("artemis.runtime.daemon_client.spawn_daemon") as mock_spawn, ): ok, base_url = ensure_daemon_running(timeout=1.0, wait_ready=True) @@ -101,6 +102,42 @@ def test_ensure_daemon_running_auto_spawns(): mock_spawn.assert_called_once() +def test_busy_live_daemon_is_reused_without_spawning(): + with ( + patch("artemis.runtime.daemon_client.is_daemon_running", return_value=False), + patch("artemis.runtime.daemon_client.is_port_in_use", return_value=True), + patch( + "artemis.runtime.daemon_client.read_server_info", + return_value={"host": "127.0.0.1", "port": 8000, "pid": 12345}, + ), + patch("artemis.runtime.daemon_client.pid_is_alive", return_value=True), + patch("artemis.runtime.daemon_client.spawn_daemon") as spawn, + ): + assert ensure_daemon_running() == (True, "http://127.0.0.1:8000") + spawn.assert_not_called() + + +def test_unknown_port_owner_is_not_overwritten(): + with ( + patch("artemis.runtime.daemon_client.is_daemon_running", return_value=False), + patch("artemis.runtime.daemon_client.is_port_in_use", return_value=True), + patch("artemis.runtime.daemon_client.read_server_info", return_value=None), + patch("artemis.runtime.daemon_client.spawn_daemon") as spawn, + ): + assert ensure_daemon_running() == (False, None) + spawn.assert_not_called() + + +def test_liveness_probe_uses_lightweight_endpoint_and_configured_timeout(monkeypatch): + from artemis.config.settings import settings + + monkeypatch.setattr(settings, "ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS", 7.0) + with patch("urllib.request.urlopen", return_value=_daemon_ok_response(b"{}")) as probe: + assert is_daemon_running() + assert probe.call_args.args[0].full_url.endswith("/api/system/server-status") + assert probe.call_args.kwargs["timeout"] == 7.0 + + def test_submit_task_to_daemon(): mock_resp = MagicMock() mock_resp.status = 200