feat(robinhood): wire agentic-trading OAuth connect flow end-to-end

packages/agents/curiosity.py

@@ -85,7 +85,7 @@ class CuriosityAction:
     payload: dict[str, Any] = field(default_factory=dict)
     ts: str = ""
 
-    def with_ts(self, now: datetime | None = None) -> "CuriosityAction":
+    def with_ts(self, now: datetime | None = None) -> CuriosityAction:
         stamp = (now or datetime.now(UTC)).isoformat(timespec="seconds")
         return CuriosityAction(
             kind=self.kind, rationale=self.rationale, payload=self.payload, ts=stamp

packages/agents/narration.py

@@ -96,7 +96,7 @@ class AgentStatus:
     ts: str = ""
     """ISO-8601 UTC timestamp. Auto-filled on emit if blank."""
 
-    def with_ts(self, now: datetime | None = None) -> "AgentStatus":
+    def with_ts(self, now: datetime | None = None) -> AgentStatus:
         """Return a copy with ``ts`` set, leaving the original frozen."""
         stamp = (now or datetime.now(UTC)).isoformat(timespec="seconds")
         return AgentStatus(

packages/agents/reddit_trust/tests/test_fetcher_oauth.py

@@ -118,10 +118,6 @@ def handler(request):
         await client.aclose()
 
     assert len(out) == 1
-    # First /hot request must be against oauth.reddit.com.
-    hot_hosts = [h for h, p in zip(hits, [str(x) for x in hits]) if True]
-    # Filter to only hot-listing requests (exclude about lookups).
-    hot_only = [h for h in hits if h != "www.reddit.com" or True]
     # The very first request should be to oauth.reddit.com
     assert hits[0] == "oauth.reddit.com"
     assert auth_headers[0] == "bearer BEARER_OK"

packages/agents/reddit_trust/tests/test_oauth.py

@@ -24,9 +24,9 @@
 import pytest
 
 from packages.agents.reddit_trust.oauth import (
+    _EXPIRY_SLACK_S,
     REDDIT_TOKEN_URL,
     RedditOAuthClient,
-    _EXPIRY_SLACK_S,
     reset_oauth_for_tests,
 )
 

packages/agents/tests/test_curiosity.py

@@ -11,7 +11,6 @@
 """
 from __future__ import annotations
 
-import json
 import random
 from pathlib import Path
 
@@ -29,21 +28,20 @@
     read_recent_actions,
 )
 
-
 # --- Helpers -----------------------------------------------------------------
 
 
 def _baseline(**overrides) -> CuriosityInput:
     """A 'quiet bot, no stall' state. Tests bend exactly one knob."""
-    base = dict(
-        idle_streak=0,
-        watchlist_age_s=0.0,
-        cumulative_relaxation=0.0,
-        dominant_rejection="",
-        universe=("AAPL", "MSFT"),
-        wildcard_pool=("NVDA", "TSLA", "AMD", "META", "GOOG", "AMZN"),
-        last_reflection_age_s=0.0,
-    )
+    base = {
+        "idle_streak": 0,
+        "watchlist_age_s": 0.0,
+        "cumulative_relaxation": 0.0,
+        "dominant_rejection": "",
+        "universe": ("AAPL", "MSFT"),
+        "wildcard_pool": ("NVDA", "TSLA", "AMD", "META", "GOOG", "AMZN"),
+        "last_reflection_age_s": 0.0,
+    }
     base.update(overrides)
     return CuriosityInput(**base)
 
@@ -84,7 +82,7 @@ def test_wildcard_skipped_if_pool_exhausted_by_universe() -> None:
 
 def test_wildcard_skipped_if_pool_empty() -> None:
     state = _baseline(
-        watchlist_age_s=WATCHLIST_STALE_S + 1, wildcard_pool=tuple()
+        watchlist_age_s=WATCHLIST_STALE_S + 1, wildcard_pool=()
     )
     action = decide(state, rng=random.Random(0))
     assert action.kind == "noop"

packages/agents/tests/test_narration.py

@@ -26,7 +26,6 @@
     read_latest,
 )
 
-
 # --- Schema pin --------------------------------------------------------------
 
 
@@ -46,7 +45,7 @@ def test_actors_list_is_stable() -> None:
 def test_agent_status_frozen() -> None:
     """Frozen dataclass: mutation must raise. This prevents agents from\n    rewriting each other's status objects after emit."""
     s = AgentStatus(actor="research", working_on="x", waiting_on="y")
-    with pytest.raises(Exception):  # FrozenInstanceError or AttributeError
+    with pytest.raises(AttributeError):  # FrozenInstanceError subclasses AttributeError
         s.working_on = "z"  # type: ignore[misc]
 
 

packages/backtests/intraday_walk_forward.py

@@ -35,10 +35,7 @@
 
 import logging
 from dataclasses import dataclass, field
-from datetime import datetime
-from typing import Any
 
-import numpy as np
 import pandas as pd
 
 from packages.backtests.champion_challenger import (

packages/backtests/tests/test_intraday_walk_forward.py

@@ -20,14 +20,13 @@
 """
 from __future__ import annotations
 
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 
 import numpy as np
 import pandas as pd
 import pytest
 
 from packages.backtests.intraday_walk_forward import (
-    DEFAULT_INTRADAY_COST_MODEL,
     DEFAULT_INTRADAY_GRID,
     IntradayCostModel,
     IntradayParamSet,
@@ -40,7 +39,6 @@
 )
 from packages.strategies.intraday_trend import IntradayTrendFollowing
 
-
 # ---------------------------------------------------------------------------
 # Bar helpers
 # ---------------------------------------------------------------------------
@@ -59,11 +57,9 @@ def _session_bars(
     timestamps so the audit's tz logic exercises real conversion.
     """
     # 09:30 ET == 14:30 UTC (no DST math; tests use fixed dates).
-    start = datetime(day.year, day.month, day.day, 14, 30, tzinfo=timezone.utc)
+    start = datetime(day.year, day.month, day.day, 14, 30, tzinfo=UTC)
     rows = []
-    price = 100.0
     for i in range(n_bars):
-        ts = start + timedelta(minutes=5 * i)
         if flat:
             close = 100.0
         elif breakout and i >= 8:
@@ -81,7 +77,6 @@ def _session_bars(
                 "volume": 10_000.0,
             }
         )
-        price = close
     idx = pd.DatetimeIndex(
         [start + timedelta(minutes=5 * i) for i in range(n_bars)],
         name="ts",

packages/cockpit/onboarding.py

@@ -55,10 +55,44 @@
 RhMode = Literal["shadow", "live"]
 VALID_RH_MODES: tuple[RhMode, ...] = ("shadow", "live")
 
+# ---------------------------------------------------------------------------
+# Active broker backend. Selects which broker the autonomy loop trades
+# through. Defaults to the existing Alpaca paper path so unset / fresh
+# installs keep the current behavior (and all current tests pass). Only an
+# explicit ``robinhood`` selection routes orders through the Robinhood
+# agentic broker -- and even then SHADOW stays the default unless the
+# resolve_mode promotion gate + ENABLE_LIVE_TRADING authorize live.
+# ---------------------------------------------------------------------------
+BrokerBackend = Literal["alpaca_paper", "robinhood"]
+VALID_BROKER_BACKENDS: tuple[BrokerBackend, ...] = ("alpaca_paper", "robinhood")
+
 # Defensive default: $300 first-float cap per the user's stated comfort.
 # Can be raised after 14 days of positive shadow-trading PnL (Phase 6).
 DEFAULT_FLOAT_CAP_USD = 300.0
 
+# Hard upper bound on the user-configurable float cap. Mirrors
+# ``packages.execution.robinhood.ABSOLUTE_MAX_FLOAT_USD`` -- duplicated here
+# (not imported) to keep the onboarding layer free of an execution-layer
+# dependency. Any value the user sets is clamped into ``[0, this]``.
+ABSOLUTE_MAX_FLOAT_USD = 10_000.0
+
+
+def clamp_float_cap(value: float) -> float:
+    """Clamp a requested float cap into ``[0, ABSOLUTE_MAX_FLOAT_USD]``.
+
+    Rejects non-finite input (NaN / inf) by falling back to the safe
+    default -- a NaN cap would otherwise make every comparison False and
+    silently disable the ceiling."""
+    import math
+
+    try:
+        v = float(value)
+    except (TypeError, ValueError):
+        return DEFAULT_FLOAT_CAP_USD
+    if not math.isfinite(v):
+        return DEFAULT_FLOAT_CAP_USD
+    return max(0.0, min(v, ABSOLUTE_MAX_FLOAT_USD))
+
 
 @dataclass
 class OnboardingState:
@@ -85,6 +119,18 @@ class OnboardingState:
     # Robinhood broker mode (shadow vs live). Defaults to shadow.
     rh_mode: RhMode = "shadow"
 
+    # Which broker the autonomy loop trades through. Defaults to the
+    # existing Alpaca paper path; ``robinhood`` opts into the Robinhood
+    # agentic broker (still shadow unless the live gate authorizes).
+    broker_backend: BrokerBackend = "alpaca_paper"
+
+    # Robinhood agentic account number to target for reads + orders. Empty
+    # until discovered (the only account with agentic_allowed=true). Stored
+    # so the broker doesn't re-discover on every call; refreshed by the
+    # auto-select helper. Robinhood rejects trades on non-agentic accounts
+    # at the API level, so this MUST point at the agentic account.
+    rh_account_number: str = ""
+
     # Wizard lifecycle timestamps. Useful for telemetry & support.
     wizard_started_at: str = ""
     wizard_completed_at: str = ""
@@ -125,6 +171,11 @@ def load_onboarding(path: Path | None = None) -> OnboardingState:
     mode_raw = raw.get("rh_mode", "shadow")
     mode: RhMode = mode_raw if mode_raw in VALID_RH_MODES else "shadow"
 
+    backend_raw = raw.get("broker_backend", "alpaca_paper")
+    backend: BrokerBackend = (
+        backend_raw if backend_raw in VALID_BROKER_BACKENDS else "alpaca_paper"
+    )
+
     # Float cap is clamped to non-negative; a corrupted negative value
     # would otherwise nuke risk gating downstream.
     try:
@@ -140,6 +191,8 @@ def load_onboarding(path: Path | None = None) -> OnboardingState:
         live_float_cap_usd=cap,
         accepted_disclaimer_at=str(raw.get("accepted_disclaimer_at", "")),
         rh_mode=mode,
+        broker_backend=backend,
+        rh_account_number=str(raw.get("rh_account_number", "")),
         wizard_started_at=str(raw.get("wizard_started_at", "")),
         wizard_completed_at=str(raw.get("wizard_completed_at", "")),
         display_name=str(raw.get("display_name", "")),

packages/cockpit/robinhood_access.py

@@ -92,16 +92,96 @@ def _probe_discovery(timeout_s: float = RH_PROBE_TIMEOUT_S) -> ProbeResult:
 
 
 def _check_granted_via_token() -> ProbeResult | None:
-    """Phase 2 stub: when a Robinhood refresh token is present in the OS
-    keychain we'll introspect it to confirm the sub-account is active.
-
-    Returns ``None`` until Phase 2 lands. Wizard treats ``None`` as
-    'fall through to the discovery probe'.
+    """Confirm 'granted' by exercising the stored token against the MCP
+    server.
+
+    Logic:
+      1. No token in keychain -> return ``None`` (fall through to the
+         public discovery probe; the user is still on the waitlist).
+      2. Token present (refreshing first if stale) -> run an authenticated
+         MCP ``initialize`` + ``tools/list`` handshake. If the server
+         answers with a tool catalog, the sub-account is live -> ``granted``.
+      3. Auth rejected (401/403/refresh failure) -> ``waitlist`` with a
+         detail explaining the token isn't accepted yet.
+      4. Any other transport hiccup -> ``None`` so we fall through to the
+         reachability probe rather than mislabeling the user.
+
+    This is read-only: ``initialize`` + ``tools/list`` never submit a
+    trade. Imports live inside the function so a fresh install without the
+    keyring backend never breaks cockpit boot at import time.
     """
-    # Intentional: Phase 2 (RobinhoodAgenticBroker) will add the keyring
-    # lookup + introspection call here. Stubbed so callers don't crash
-    # before that work lands.
-    return None
+    import asyncio
+
+    try:
+        from packages.execution.broker import BrokerError
+        from packages.execution.robinhood import is_connected
+        from packages.execution.robinhood_mcp import (
+            McpError,
+            RobinhoodMcpClient,
+        )
+    except Exception as exc:  # pragma: no cover - import-time safety net
+        logger.debug("rh token check unavailable: %s", exc.__class__.__name__)
+        return None
+
+    if not is_connected():
+        return None  # no usable token -> still waitlist
+
+    async def _handshake() -> ProbeResult:
+        # Reuse the broker's token-resolution + refresh logic so a stale
+        # access token is silently refreshed before we probe.
+        from packages.execution.modes import ExecutionMode
+        from packages.execution.robinhood import RobinhoodAgenticBroker
+
+        broker = RobinhoodAgenticBroker(mode=ExecutionMode.SHADOW)
+        try:
+            tokens = broker._require_token()  # refreshes if stale
+        except BrokerError as exc:
+            return ProbeResult(
+                outcome="waitlist",
+                detail=f"token present but not usable: {exc}",
+            )
+        client = RobinhoodMcpClient(
+            bearer_token=tokens.access_token,
+            timeout_s=RH_PROBE_TIMEOUT_S,
+        )
+        try:
+            await client.initialize()
+            tools = await client.list_tools()
+            return ProbeResult(
+                outcome="granted",
+                detail=f"authenticated MCP handshake ok ({len(tools)} tools)",
+                http_status=200,
+            )
+        except McpError as exc:
+            msg = str(exc)
+            # 401/403 means the token isn't authorized for the agentic
+            # account yet -- the user is approved for OAuth but the
+            # sub-account isn't live. Treat as waitlist, not granted.
+            if "401" in msg or "403" in msg:
+                return ProbeResult(
+                    outcome="waitlist",
+                    detail=f"MCP rejected token (not yet provisioned): {msg}",
+                )
+            # Other MCP errors are ambiguous -> fall through.
+            return ProbeResult(outcome="unknown", detail=f"mcp error: {msg}")
+        finally:
+            await client.aclose()
+
+    try:
+        result = asyncio.run(_handshake())
+    except RuntimeError:
+        # Already inside an event loop (e.g. called from async cockpit
+        # context). Run in a dedicated loop on a thread to stay safe.
+        import concurrent.futures
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            result = pool.submit(lambda: asyncio.run(_handshake())).result()
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.warning("rh token handshake failed: %s", exc.__class__.__name__)
+        return None
+
+    # 'unknown' from the handshake means fall through to discovery probe.
+    return None if result.outcome == "unknown" else result
 
 
 def detect_access(