From 6eb7ded9840ad48a4cae1548070ddab2a98a299c Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Fri, 21 Aug 2026 03:38:06 +0800 Subject: [PATCH] fix: keep agent alive across worker crashes --- README.md | 11 ++- main.py | 135 ++++++++++++++++++++++++---- modelhub_submmit_api/poll_runner.py | 15 +++- modelhub_submmit_api/state_sync.py | 11 ++- modelhub_submmit_api/version.py | 2 +- tests/test_agent_entrypoint.py | 25 ++++++ tests/test_super_agent.py | 10 +++ 7 files changed, 183 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index 6f59dcd4..1cfe1d09 100644 --- a/README.md +++ b/README.md @@ -362,19 +362,24 @@ counts fail closed for older candidates, and no startup or periodic cleanup can cancel a task by date. It also keeps unclassified/ambiguous historical failures neutral in GPU/framework success feedback while preserving deterministic OOM and architecture cleanup. -Version `2026.08.15.3` replaces the 70/30 quota with hierarchical success-first +Version `2026.08.21.1` replaces the 70/30 quota with hierarchical success-first routing, dynamically gates models through the official GPU/task/framework/config APIs, enriches ModelScope metadata and model lineage, learns only proven safe config vectors, and adds crash-safe write-ahead state synchronization to the `agent-state` branch. It also exposes `/ready` and extends deterministic cleanup to officially removed waiting GPU/framework routes. State recovery uses the pure-Python Dulwich client, avoiding slow OS package installation during builds. +The HTTP process now supervises the submission worker with exponential restart +backoff instead of exiting the container, while `/health` remains a liveness +probe and `/ready` reports worker availability. Poll and cleanup summaries are +bounded in memory, and the last 200 worker crash records are synchronized with +the durable state branch for post-restart diagnosis. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-v27 -m "ModelHub agent 2026.08.15.3" -git push origin main agent-v27 +git tag -a agent-v28 -m "ModelHub agent 2026.08.21.1" +git push origin main agent-v28 ``` diff --git a/main.py b/main.py index 10b5b421..86d17956 100644 --- a/main.py +++ b/main.py @@ -6,6 +6,7 @@ import signal import subprocess import sys import time +from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -18,10 +19,18 @@ PORT = int(os.getenv("PORT", "8080")) ROOT = Path(__file__).resolve().parent WORKER_SCRIPT = ROOT / "modelhub_submmit_api" / "poll_runner.py" READINESS_PATH = ROOT / ".modelhub_state" / "readiness.json" +WORKER_CRASH_PATH = ROOT / ".modelhub_state" / "worker_crashes.jsonl" +WORKER_STABLE_SECONDS = max(60, int(os.getenv("MODELHUB_AGENT_WORKER_STABLE_SECONDS", "600"))) +WORKER_RESTART_MAX_SECONDS = max(30, int(os.getenv("MODELHUB_AGENT_WORKER_RESTART_MAX_SECONDS", "300"))) shutdown_requested = False worker: subprocess.Popen | None = None config_error: str | None = None +worker_start_enabled = False +worker_started_at: float | None = None +worker_next_restart_at = 0.0 +worker_restart_count = 0 +worker_last_exit_code: int | None = None def _has_modelhub_auth() -> bool: @@ -84,7 +93,14 @@ def _config() -> dict[str, object]: "modelhub_auth_present": _has_modelhub_auth(), "config_error": config_error, "worker_running": worker is not None and worker.poll() is None, - "worker_return_code": None if worker is None else worker.poll(), + "worker_return_code": worker_last_exit_code if worker is None else worker.poll(), + "worker_restart_count": worker_restart_count, + "worker_restart_pending": bool( + worker_start_enabled + and worker is None + and not shutdown_requested + and config_error is None + ), } @@ -99,13 +115,12 @@ def _readiness() -> dict[str, object]: class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: if self.path == "/health": + # This endpoint is a liveness probe for the supervisor. Worker + # readiness is reported separately by /ready. + status = "ok" if worker is not None and worker.poll() is None else "degraded" if config_error: - self._send_json({"status": "config_error", "config": _config()}, status=500) - return - if worker is not None and worker.poll() is not None: - self._send_json({"status": "worker_exited", "config": _config()}, status=500) - return - self._send_json({"status": "ok", "config": _config()}) + status = "config_error" + self._send_json({"status": status, "config": _config()}) return if self.path == "/ready": @@ -152,6 +167,96 @@ def _stop_worker() -> None: worker.wait(timeout=5) +def _restart_delay(restart_count: int) -> int: + return min(WORKER_RESTART_MAX_SECONDS, 5 * (2 ** min(max(0, restart_count - 1), 6))) + + +def _write_supervisor_readiness(reason: str, *, exit_code: int | None = None) -> None: + READINESS_PATH.parent.mkdir(parents=True, exist_ok=True) + payload = { + "ready": False, + "reason": reason, + "updatedAt": datetime.now(timezone.utc).isoformat(), + "supervisor": { + "workerRestartCount": worker_restart_count, + "lastExitCode": exit_code, + }, + } + temporary = READINESS_PATH.with_name(f".{READINESS_PATH.name}.supervisor-{os.getpid()}") + temporary.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + os.replace(temporary, READINESS_PATH) + + +def _record_worker_crash(*, exit_code: int | None, uptime_seconds: float, reason: str) -> None: + WORKER_CRASH_PATH.parent.mkdir(parents=True, exist_ok=True) + event = { + "at": datetime.now(timezone.utc).isoformat(), + "exitCode": exit_code, + "uptimeSeconds": round(max(0.0, uptime_seconds), 3), + "restartCount": worker_restart_count, + "reason": reason[:500], + } + with WORKER_CRASH_PATH.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n") + + +def _launch_worker(*, now: float | None = None) -> bool: + global worker, worker_started_at, worker_last_exit_code + now = time.monotonic() if now is None else now + cmd = _worker_command() + print("starting submission worker: " + " ".join(cmd), flush=True) + try: + worker = subprocess.Popen(cmd, cwd=str(ROOT)) + except OSError as exc: + worker = None + worker_started_at = None + worker_last_exit_code = None + print(f"submission worker spawn failed: {type(exc).__name__}: {exc}", flush=True) + return False + worker_started_at = now + worker_last_exit_code = None + return True + + +def _supervise_worker(*, now: float | None = None) -> None: + global worker, worker_started_at, worker_next_restart_at + global worker_restart_count, worker_last_exit_code + if not worker_start_enabled or shutdown_requested or config_error: + return + now = time.monotonic() if now is None else now + if worker is not None: + return_code = worker.poll() + if return_code is None: + if worker_started_at is not None and now - worker_started_at >= WORKER_STABLE_SECONDS: + worker_restart_count = 0 + return + uptime = max(0.0, now - worker_started_at) if worker_started_at is not None else 0.0 + if uptime >= WORKER_STABLE_SECONDS: + worker_restart_count = 0 + worker_restart_count += 1 + worker_last_exit_code = int(return_code) + delay = _restart_delay(worker_restart_count) + worker_next_restart_at = now + delay + print( + f"submission worker exited code={return_code} uptime={uptime:.1f}s " + f"restart_in={delay}s restart_count={worker_restart_count}", + flush=True, + ) + _record_worker_crash(exit_code=return_code, uptime_seconds=uptime, reason="worker_exited") + _write_supervisor_readiness("worker_restarting", exit_code=return_code) + worker = None + worker_started_at = None + return + if now < worker_next_restart_at: + return + if not _launch_worker(now=now): + worker_restart_count += 1 + delay = _restart_delay(worker_restart_count) + worker_next_restart_at = now + delay + _record_worker_crash(exit_code=None, uptime_seconds=0.0, reason="worker_spawn_failed") + _write_supervisor_readiness("worker_spawn_failed") + + def _handle_signal(signum: int, _frame: object) -> None: global shutdown_requested shutdown_requested = True @@ -160,19 +265,17 @@ def _handle_signal(signum: int, _frame: object) -> None: def main() -> int: - global config_error, worker + global config_error, worker_start_enabled print(f"modelhub-submmit-agent version={AGENT_VERSION}", flush=True) signal.signal(signal.SIGTERM, _handle_signal) signal.signal(signal.SIGINT, _handle_signal) - start_worker = os.getenv("MODELHUB_AGENT_START_WORKER", "1").strip().lower() not in {"0", "false", "no"} - if start_worker and not _has_modelhub_auth(): + worker_start_enabled = os.getenv("MODELHUB_AGENT_START_WORKER", "1").strip().lower() not in {"0", "false", "no"} + if worker_start_enabled and not _has_modelhub_auth(): config_error = "missing ModelHub auth: set MODELHUB_XC_TOKEN/XC_TOKEN or MODELHUB_JWT_TOKEN/JWT_TOKEN" print(config_error, flush=True) - elif start_worker: - cmd = _worker_command() - print("starting submission worker: " + " ".join(cmd), flush=True) - worker = subprocess.Popen(cmd, cwd=str(ROOT)) + elif worker_start_enabled: + _launch_worker() else: print("submission worker disabled by MODELHUB_AGENT_START_WORKER", flush=True) @@ -183,9 +286,7 @@ def main() -> int: try: while not shutdown_requested: server.handle_request() - if worker is not None and worker.poll() is not None: - print(f"submission worker exited with code {worker.returncode}", flush=True) - return worker.returncode or 1 + _supervise_worker() time.sleep(0.1) finally: server.server_close() diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 683a3c39..5b294752 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -5,6 +5,7 @@ import json import os import sys import time +from collections import deque from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import timedelta from pathlib import Path @@ -600,8 +601,10 @@ def run_poll_loop( else: architecture_bootstrap_summary["reason"] = "outcome_sync_disabled" - cycle_summaries: list[dict[str, Any]] = [] - queue_cleanup_runs: list[dict[str, Any]] = [] + retained_cycles = max(10, int(os.getenv("MODELHUB_AGENT_RETAINED_CYCLE_SUMMARIES", "50"))) + retained_cleanups = max(5, int(os.getenv("MODELHUB_AGENT_RETAINED_CLEANUP_SUMMARIES", "20"))) + cycle_summaries: deque[dict[str, Any]] = deque(maxlen=retained_cycles) + queue_cleanup_runs: deque[dict[str, Any]] = deque(maxlen=retained_cleanups) submitted_total = 0 cycles = 0 stopped_reason = "max_cycles_reached" @@ -920,8 +923,12 @@ def run_poll_loop( "submittedTotal": submitted_total, "stoppedReason": stopped_reason, "pollRunDir": str(poll_run_dir), - "cycleSummaries": cycle_summaries, - "queueCleanupRuns": queue_cleanup_runs, + "cycleSummaries": list(cycle_summaries), + "cycleSummariesRetained": len(cycle_summaries), + "cycleSummariesRetentionLimit": retained_cycles, + "queueCleanupRuns": list(queue_cleanup_runs), + "queueCleanupRunsRetained": len(queue_cleanup_runs), + "queueCleanupRunsRetentionLimit": retained_cleanups, "architectureBootstrap": architecture_bootstrap_summary, "outcomeStats": stats_report, } diff --git a/modelhub_submmit_api/state_sync.py b/modelhub_submmit_api/state_sync.py index 33c25380..ac276d77 100644 --- a/modelhub_submmit_api/state_sync.py +++ b/modelhub_submmit_api/state_sync.py @@ -42,6 +42,7 @@ STATE_ALLOWLIST = ( ".modelhub_state/recovery_intents.jsonl", ".modelhub_state/routing_intelligence.json", ".modelhub_state/submission_exclusions.jsonl", + ".modelhub_state/worker_crashes.jsonl", "ledger/submissions.jsonl", "outcomes/submissions.jsonl", ) @@ -276,7 +277,15 @@ class StateGitSync: destination = self.project_root / relative destination.parent.mkdir(parents=True, exist_ok=True) temporary = destination.with_name(f".{destination.name}.restore-{uuid.uuid4().hex}") - shutil.copy2(source, temporary) + if relative == ".modelhub_state/worker_crashes.jsonl": + merged: dict[str, dict[str, Any]] = {} + for row in [*read_jsonl(source), *read_jsonl(destination)]: + key = json.dumps(row, ensure_ascii=False, sort_keys=True) + merged[key] = row + rows = sorted(merged.values(), key=lambda row: str(row.get("at") or ""))[-200:] + write_jsonl(temporary, rows) + else: + shutil.copy2(source, temporary) os.replace(temporary, destination) self.log( f"[state-recovery] generation={self.generation} source=remote branch={self.branch} status=ok" diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 2cfc3d4c..86a18f17 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.15.3" +AGENT_VERSION = "2026.08.21.1" diff --git a/tests/test_agent_entrypoint.py b/tests/test_agent_entrypoint.py index d06e637a..e865e53c 100644 --- a/tests/test_agent_entrypoint.py +++ b/tests/test_agent_entrypoint.py @@ -46,6 +46,31 @@ class HostedAgentEntrypointTests(unittest.TestCase): self.assertNotIn("apt-get", dockerfile) self.assertIn("dulwich", requirements.casefold()) + def test_exited_worker_is_backed_off_without_exiting_supervisor(self) -> None: + class ExitedWorker: + def poll(self) -> int: + return 137 + + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + with ( + patch.object(ENTRYPOINT, "worker", ExitedWorker()), + patch.object(ENTRYPOINT, "worker_start_enabled", True), + patch.object(ENTRYPOINT, "shutdown_requested", False), + patch.object(ENTRYPOINT, "config_error", None), + patch.object(ENTRYPOINT, "worker_started_at", 90.0), + patch.object(ENTRYPOINT, "worker_restart_count", 0), + patch.object(ENTRYPOINT, "worker_next_restart_at", 0.0), + patch.object(ENTRYPOINT, "WORKER_CRASH_PATH", root / "worker_crashes.jsonl"), + patch.object(ENTRYPOINT, "READINESS_PATH", root / "readiness.json"), + ): + ENTRYPOINT._supervise_worker(now=100.0) + self.assertIsNone(ENTRYPOINT.worker) + self.assertEqual(1, ENTRYPOINT.worker_restart_count) + self.assertEqual(105.0, ENTRYPOINT.worker_next_restart_at) + self.assertEqual("worker_restarting", ENTRYPOINT._readiness()["reason"]) + self.assertIn('"exitCode":137', (root / "worker_crashes.jsonl").read_text(encoding="utf-8")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_super_agent.py b/tests/test_super_agent.py index c9215ef9..9e9d2f4a 100644 --- a/tests/test_super_agent.py +++ b/tests/test_super_agent.py @@ -141,6 +141,14 @@ class SuperAgentTests(unittest.TestCase): restored_project.mkdir() porcelain.init(remote, bare=True) write_json(project / ".modelhub_state" / "account_capacity.json", {"version": 1}) + write_jsonl( + project / ".modelhub_state" / "worker_crashes.jsonl", + [{"at": "2026-08-21T00:00:00+00:00", "exitCode": 137}], + ) + write_jsonl( + restored_project / ".modelhub_state" / "worker_crashes.jsonl", + [{"at": "2026-08-21T01:00:00+00:00", "exitCode": 1}], + ) credentials = {"username": "tester", "email": "tester@example.com", "password": "secret-value"} manager = StateGitSync( project_root=project, @@ -175,6 +183,8 @@ class SuperAgentTests(unittest.TestCase): self.assertTrue(restored.restore()) intents = read_jsonl(restored_project / ".modelhub_state" / "recovery_intents.jsonl") self.assertEqual("owner/model", intents[0]["repoId"]) + crashes = read_jsonl(restored_project / ".modelhub_state" / "worker_crashes.jsonl") + self.assertEqual([137, 1], [row["exitCode"] for row in crashes]) state_text = "\n".join( path.read_text(encoding="utf-8") for path in restored._workspace.rglob("*")