fix: keep IBKR reconciliation probes read-only (#417)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

application/runtime_broker_adapters.py

@@ -156,7 +156,14 @@ def capture_api_error(_request_id, error_code, error_message, _contract):
             ib.RaiseRequestErrors = original_raise_request_errors
             ib.RequestTimeout = original_request_timeout
 
-    def connect_ib(self):
+    def connect_ib(self, *, validate_trading_permissions: bool = True):
+        """Connect to the Gateway, optionally without an order-write permission probe.
+
+        Account validation is safe and remains mandatory for every connection.
+        The live trading permission check uses IBKR's ``whatIfOrder`` API, which
+        is non-transmitting but still exercises an order-validation endpoint.
+        Health and reconciliation probes must not invoke that endpoint.
+        """
         self.ensure_event_loop_fn()
         host = self.host_resolver()
         last_error = None
@@ -179,7 +186,8 @@ def connect_ib(self):
                 )
                 try:
                     self.validate_configured_accounts(ib)
-                    self.validate_trading_permissions(ib)
+                    if validate_trading_permissions:
+                        self.validate_trading_permissions(ib)
                 except Exception:
                     disconnect_fn = getattr(ib, "disconnect", None)
                     if callable(disconnect_fn):

main.py

@@ -882,9 +882,11 @@ def require_gateway_execution_backend():
     )
 
 
-def connect_ib():
+def connect_ib(*, validate_trading_permissions: bool = True):
     require_gateway_execution_backend()
-    return build_broker_adapters().connect_ib()
+    return build_broker_adapters().connect_ib(
+        validate_trading_permissions=validate_trading_permissions,
+    )
 
 
 def _build_health_probe_connection_error_message(exc: Exception) -> str:
@@ -1733,7 +1735,7 @@ def _handle_probe(*, response_body: str = "Probe OK"):
             http_method=request.method,
             execution_window="probe",
         )
-        ib = connect_ib()
+        ib = connect_ib(validate_trading_permissions=False)
         snapshot = build_portfolio_snapshot(ib)
         positions = tuple(getattr(snapshot, "positions", ()) or ())
         buying_power = float(getattr(snapshot, "buying_power", 0.0) or 0.0)

tests/test_request_handling.py

@@ -528,7 +528,7 @@ def fake_dry_run(*, recycle_on_timeout, timeout_state):
 
 
 def test_handle_probe_checks_account_snapshot_without_success_notification(strategy_module, monkeypatch):
-    observed = {"events": [], "disconnects": 0, "notifications": []}
+    observed = {"events": [], "disconnects": 0, "notifications": [], "connection_options": []}
 
     class FakeIB:
         def disconnect(self):
@@ -562,7 +562,11 @@ def disconnect(self):
         "attach_strategy_plugin_report",
         lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("health probe should not attach strategy plugin reports")),
     )
-    monkeypatch.setattr(strategy_module, "connect_ib", lambda: FakeIB())
+    def fake_connect_ib(*, validate_trading_permissions=True):
+        observed["connection_options"].append(validate_trading_permissions)
+        return FakeIB()
+
+    monkeypatch.setattr(strategy_module, "connect_ib", fake_connect_ib)
     monkeypatch.setattr(strategy_module, "build_portfolio_snapshot", lambda ib: snapshot)
     monkeypatch.setattr(
         strategy_module,
@@ -585,6 +589,7 @@ def disconnect(self):
     assert observed["report"]["summary"]["positions_count"] == 1
     assert observed["disconnects"] == 1
     assert observed["notifications"] == []
+    assert observed["connection_options"] == [False]
 
 
 def test_handle_probe_connect_timeout_sends_concise_connection_notification(strategy_module, monkeypatch):

tests/test_runtime_broker_adapters.py

@@ -352,3 +352,22 @@ def whatIfOrder(self, _contract, _order):
     )
 
     assert adapters.connect_ib().managedAccounts() == ["U1234567"]
+
+
+def test_connect_ib_skips_trading_permission_probe_for_explicit_read_only_connection():
+    class FakeIB:
+        def managedAccounts(self):
+            return ["U1234567"]
+
+        def whatIfOrder(self, _contract, _order):
+            pytest.fail("read-only connection must not probe trading permissions")
+
+    adapters = _build_adapters(account_ids=("U1234567",), execution_mode="live")
+    adapters = adapters.__class__(
+        **{
+            **adapters.__dict__,
+            "connect_ib_fn": lambda *_args, **_kwargs: FakeIB(),
+        }
+    )
+
+    assert adapters.connect_ib(validate_trading_permissions=False).managedAccounts() == ["U1234567"]