fix: recycle workers after dry-run deadline

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

application/runtime_deadline.py

@@ -3,13 +3,20 @@
 from __future__ import annotations
 
 import contextlib
+import os
 import signal
 import threading
+import time
 from collections.abc import Iterator
+from collections.abc import Callable
 
 
-class RuntimeDeadlineExceeded(TimeoutError):
-    """Raised when a synchronous runtime operation exceeds its local budget."""
+class RuntimeDeadlineExceeded(BaseException):
+    """Raised when a synchronous runtime operation exceeds its local budget.
+
+    This intentionally bypasses broad ``except Exception`` handlers in broker and
+    strategy code so an interrupted worker cannot resume with partial session state.
+    """
 
 
 @contextlib.contextmanager
@@ -46,3 +53,25 @@ def _raise_deadline(_signum, _frame) -> None:
     finally:
         signal.setitimer(signal.ITIMER_REAL, 0)
         signal.signal(signal.SIGALRM, previous_handler)
+
+
+def recycle_current_process_after_response(
+    *,
+    delay_seconds: float = 1.0,
+    process_id: int | None = None,
+    kill_fn: Callable[[int, int], None] = os.kill,
+) -> threading.Thread:
+    """Ask the process manager to replace a worker after its response can flush."""
+    target_pid = int(process_id or os.getpid())
+
+    def _terminate() -> None:
+        time.sleep(max(0.0, float(delay_seconds)))
+        kill_fn(target_pid, signal.SIGTERM)
+
+    thread = threading.Thread(
+        target=_terminate,
+        name="runtime-deadline-worker-recycle",
+        daemon=True,
+    )
+    thread.start()
+    return thread

main.py

@@ -21,7 +21,11 @@
 from application.cycle_result import coerce_strategy_cycle_result
 from application.runtime_broker_adapters import build_runtime_broker_adapters
 from application.runtime_composer import build_runtime_composer
-from application.runtime_deadline import RuntimeDeadlineExceeded, enforce_runtime_deadline
+from application.runtime_deadline import (
+    RuntimeDeadlineExceeded,
+    enforce_runtime_deadline,
+    recycle_current_process_after_response,
+)
 from application.runtime_strategy_adapters import (
     build_runtime_strategy_adapters,
     fetch_yfinance_historical_candles,
@@ -1113,6 +1117,7 @@ def _handle_request(
     )
     execution_window = "dry_run" if dry_run_only_override else "execution"
     lock_acquired = STRATEGY_RUN_LOCK.acquire(blocking=False)
+    deadline_exceeded = False
     try:
         log_runtime_event(
             log_context,
@@ -1200,18 +1205,14 @@ def _handle_request(
         if dry_run_only_override is None:
             publish_strategy_plugin_alerts(strategy_plugin_signals, report=report)
         notification_delivery_events: list[dict] = []
-        with enforce_runtime_deadline(
-            IBKR_DRY_RUN_DEADLINE_SECONDS if dry_run_only_override is True else 0,
-            operation="IBKR strategy dry-run",
-        ):
-            cycle_result = coerce_strategy_cycle_result(
-                run_strategy_core(
-                    strategy_plugin_signals=strategy_plugin_signals,
-                    strategy_plugin_error=strategy_plugin_error,
-                    dry_run_only_override=dry_run_only_override,
-                    notification_delivery_events=notification_delivery_events,
-                )
+        cycle_result = coerce_strategy_cycle_result(
+            run_strategy_core(
+                strategy_plugin_signals=strategy_plugin_signals,
+                strategy_plugin_error=strategy_plugin_error,
+                dry_run_only_override=dry_run_only_override,
+                notification_delivery_events=notification_delivery_events,
             )
+        )
         execution_summary = dict(cycle_result.execution_summary or {})
         reconciliation_record = dict(cycle_result.reconciliation_record or {})
         signal_metadata = dict(cycle_result.signal_metadata or {})
@@ -1283,30 +1284,9 @@ def _handle_request(
             result=cycle_result.result,
         )
         return (response_body if dry_run_only_override else cycle_result.result), 200
-    except RuntimeDeadlineExceeded as exc:
-        append_runtime_report_error(
-            report,
-            stage="strategy_deadline",
-            message=str(exc),
-            error_type=type(exc).__name__,
-        )
-        finalize_runtime_report(report, status="error")
-        log_runtime_event(
-            log_context,
-            "strategy_cycle_deadline_exceeded",
-            message="Strategy dry-run exceeded its application deadline",
-            severity="ERROR",
-            execution_window=execution_window,
-            timeout_seconds=IBKR_DRY_RUN_DEADLINE_SECONDS,
-            error_message=str(exc),
-        )
-        error_msg = f"🚨 【IBKR 运行超时】\n{str(exc)}"
-        _publish_runtime_failure_notification(
-            detailed_text=error_msg,
-            compact_text=error_msg,
-            exc=exc,
-        )
-        return "Error", 503
+    except RuntimeDeadlineExceeded:
+        deadline_exceeded = True
+        raise
     except TimeoutError as exc:
         append_runtime_report_error(
             report,
@@ -1356,14 +1336,15 @@ def _handle_request(
     finally:
         if lock_acquired:
             STRATEGY_RUN_LOCK.release()
-        try:
-            if dry_run_only_override is None:
-                report_path = persist_execution_report(report)
-            else:
-                report_path = persist_execution_report(report, dry_run_only_override=dry_run_only_override)
-            print(f"execution_report {report_path}", flush=True)
-        except Exception as persist_exc:
-            print(f"failed to persist execution report: {persist_exc}", flush=True)
+        if not deadline_exceeded:
+            try:
+                if dry_run_only_override is None:
+                    report_path = persist_execution_report(report)
+                else:
+                    report_path = persist_execution_report(report, dry_run_only_override=dry_run_only_override)
+                print(f"execution_report {report_path}", flush=True)
+            except Exception as persist_exc:
+                print(f"failed to persist execution report: {persist_exc}", flush=True)
 
 
 def _handle_probe(*, response_body: str = "Probe OK"):
@@ -1507,11 +1488,7 @@ def _dispatch_local_monitor(dispatch):
     if window == "probe":
         _body, status_code = _handle_probe()
     elif window == "precheck":
-        _body, status_code = _handle_request(
-            dry_run_only_override=True,
-            response_body="Dry Run OK",
-            dry_run_label="strategy dry-run",
-        )
+        _body, status_code = _handle_dry_run_with_deadline()
     else:
         raise ValueError(f"Unsupported local monitor window: {window!r}")
     return {"status_code": int(status_code)}
@@ -1522,14 +1499,42 @@ def handle_request():
     return _route_with_runtime_error_fallback(_handle_request)
 
 
+def _handle_dry_run_with_deadline():
+    try:
+        with enforce_runtime_deadline(
+            IBKR_DRY_RUN_DEADLINE_SECONDS,
+            operation="IBKR strategy dry-run request",
+        ):
+            return _route_with_runtime_error_fallback(
+                _handle_request,
+                dry_run_only_override=True,
+                response_body="Dry Run OK",
+                dry_run_label="strategy dry-run",
+            )
+    except RuntimeDeadlineExceeded as exc:
+        print(
+            json.dumps(
+                {
+                    "severity": "ERROR",
+                    "event": "strategy_cycle_deadline_exceeded",
+                    "message": str(exc),
+                    "service_name": SERVICE_NAME or os.getenv("K_SERVICE"),
+                    "strategy_profile": STRATEGY_PROFILE,
+                    "timeout_seconds": IBKR_DRY_RUN_DEADLINE_SECONDS,
+                    "worker_recycle_scheduled": True,
+                },
+                ensure_ascii=False,
+                separators=(",", ":"),
+            ),
+            flush=True,
+        )
+        recycle_current_process_after_response()
+        return "Error", 503
+
+
 @app.route("/dry-run", methods=["POST", "GET"])
 def handle_dry_run():
-    return _route_with_runtime_error_fallback(
-        _handle_request,
-        dry_run_only_override=True,
-        response_body="Dry Run OK",
-        dry_run_label="strategy dry-run",
-    )
+    return _handle_dry_run_with_deadline()
 
 
 @app.route("/probe", methods=["POST", "GET"])

tests/test_request_handling.py

@@ -154,7 +154,7 @@ def fake_run_strategy_core(**kwargs):
 
 
 def test_handle_dry_run_returns_service_unavailable_before_platform_timeout(strategy_module, monkeypatch):
-    observed = {"events": [], "notifications": []}
+    observed = {"recycles": 0}
 
     class ExpiredDeadline:
         def __enter__(self):
@@ -163,36 +163,55 @@ def __enter__(self):
         def __exit__(self, *_args):
             return False
 
-    monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001"))
-    monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending", "errors": []})
+    monkeypatch.setattr(strategy_module, "enforce_runtime_deadline", lambda *_args, **_kwargs: ExpiredDeadline())
     monkeypatch.setattr(
         strategy_module,
-        "persist_execution_report",
-        lambda report, **_kwargs: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json",
+        "recycle_current_process_after_response",
+        lambda: observed.__setitem__("recycles", observed["recycles"] + 1),
     )
+
+    with strategy_module.app.test_request_context("/dry-run", method="POST"):
+        body, status = strategy_module.handle_dry_run()
+
+    assert status == 503
+    assert body == "Error"
+    assert observed["recycles"] == 1
+
+
+def test_handle_dry_run_deadline_from_strategy_skips_persistence_and_recycles(strategy_module, monkeypatch):
+    observed = {"persist_calls": 0, "recycles": 0}
+
+    monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001"))
+    monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending"})
     monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True)
     monkeypatch.setattr(strategy_module, "load_strategy_plugin_signals", lambda: ((), None))
     monkeypatch.setattr(strategy_module, "attach_strategy_plugin_report", lambda *_args, **_kwargs: None)
-    monkeypatch.setattr(strategy_module, "enforce_runtime_deadline", lambda *_args, **_kwargs: ExpiredDeadline())
+    monkeypatch.setattr(strategy_module, "log_runtime_event", lambda *_args, **_kwargs: None)
     monkeypatch.setattr(
         strategy_module,
-        "log_runtime_event",
-        lambda _context, event, **fields: observed["events"].append((event, fields)),
+        "run_strategy_core",
+        lambda **_kwargs: (_ for _ in ()).throw(
+            strategy_module.RuntimeDeadlineExceeded("strategy deadline exceeded")
+        ),
+    )
+    monkeypatch.setattr(
+        strategy_module,
+        "persist_execution_report",
+        lambda *_args, **_kwargs: observed.__setitem__("persist_calls", observed["persist_calls"] + 1),
     )
     monkeypatch.setattr(
         strategy_module,
-        "_publish_runtime_failure_notification",
-        lambda **kwargs: observed["notifications"].append(kwargs) or True,
+        "recycle_current_process_after_response",
+        lambda: observed.__setitem__("recycles", observed["recycles"] + 1),
     )
 
     with strategy_module.app.test_request_context("/dry-run", method="POST"):
         body, status = strategy_module.handle_dry_run()
 
     assert status == 503
     assert body == "Error"
-    assert observed["report"]["errors"][0]["stage"] == "strategy_deadline"
-    assert observed["events"][-1][0] == "strategy_cycle_deadline_exceeded"
-    assert len(observed["notifications"]) == 1
+    assert observed["persist_calls"] == 0
+    assert observed["recycles"] == 1
 
 
 def test_dry_run_composer_wires_dry_run_override_to_order_execution(strategy_module, monkeypatch):

tests/test_runtime_deadline.py

@@ -4,7 +4,11 @@
 
 import pytest
 
-from application.runtime_deadline import RuntimeDeadlineExceeded, enforce_runtime_deadline
+from application.runtime_deadline import (
+    RuntimeDeadlineExceeded,
+    enforce_runtime_deadline,
+    recycle_current_process_after_response,
+)
 
 
 def test_runtime_deadline_interrupts_stalled_main_thread_work() -> None:
@@ -20,3 +24,16 @@ def test_runtime_deadline_interrupts_stalled_main_thread_work() -> None:
 def test_disabled_runtime_deadline_does_not_interrupt() -> None:
     with enforce_runtime_deadline(0, operation="disabled operation"):
         pass
+
+
+def test_recycle_current_process_sends_sigterm_after_delay() -> None:
+    calls = []
+
+    thread = recycle_current_process_after_response(
+        delay_seconds=0,
+        process_id=1234,
+        kill_fn=lambda process_id, signum: calls.append((process_id, signum)),
+    )
+    thread.join(timeout=1)
+
+    assert calls == [(1234, 15)]