fix: name the real reason a session failed, and record it

alphoryn/memory/schema.py

@@ -46,6 +46,10 @@ class Session(Base):
     status = Column(String, nullable=False)
     # Valid statuses: COMPLETED | SKIPPED_TIMEOUT | SKIPPED_MARKET_CLOSED
     #                 | SKIPPED_DATA_UNAVAILABLE | SKIPPED_OVERRUN
+    #                 | SKIPPED_AGENT_ERROR
+    # Each names a distinct cause. SKIPPED_DATA_UNAVAILABLE means market data
+    # was unreachable and nothing else; an agent that failed to answer is
+    # SKIPPED_AGENT_ERROR, and the exception text is in `warnings`.
     html_report_path = Column(String, nullable=True)
     # JSON: {"SPY": {"strategy": "MEAN_REVERSION", "decision": "BUY", "execution_result": ...}}
     ticker_decisions = Column(Text, nullable=True)

alphoryn/scheduler/scheduler.py

@@ -13,14 +13,15 @@
 import sys
 import threading
 import time
+from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
 import typer
 from alpaca.trading.client import TradingClient
 
 from alphoryn.agents.feedback_agent import FeedbackAgent, FeedbackInput
-from alphoryn.agents.main_agent import MainAgent
+from alphoryn.agents.main_agent import MainAgent, MainAgentError
 from alphoryn.config.models import TIMEFRAME_SECONDS, AlphorynConfig
 from alphoryn.execution.agent import AssetDecision, ExecutionAgent, SessionDecision
 from alphoryn.memory.bank import MemoryBank
@@ -36,6 +37,25 @@
 _HEARTBEAT_INTERVAL_SECS = 5 * 60
 
 
+@dataclass(frozen=True)
+class SessionSkip:
+    """Why a session was not completed: the status, and the cause behind it.
+
+    The two travel together because separating them is how the memory bank
+    started lying. On 2026-08-13 two sessions failed - one on an unparseable
+    model reply, one on a Vertex 429 - and both were filed as
+    ``SKIPPED_DATA_UNAVAILABLE``. Market data was fine in both cases, and the
+    exception text existed only on stdout, so the bank sent you to check Alpaca
+    for a problem that was never there.
+
+    *detail* is the exception text. It is persisted to the session's warnings so
+    a post-mortem can read the real cause out of the bank alone.
+    """
+
+    status: str
+    detail: str | None = None
+
+
 class Scheduler:
     """Drives the candle-by-candle session loop.
 
@@ -247,13 +267,22 @@ def _run_investigation(
         session_id: str,
         candle_close_at: datetime,
         tickers: list[str],
-    ) -> "tuple[SessionDecision | None, str | None]":
+    ) -> "tuple[SessionDecision | None, SessionSkip | None]":
         """Run main_agent.decide() with investigation budget and heartbeat.
 
-        Returns ``(decision, skip_status)``. On success ``skip_status`` is None.
-        A budget overrun yields ``SKIPPED_TIMEOUT``; any failure reaching market
-        data or the investigation agent yields ``SKIPPED_DATA_UNAVAILABLE``.
-        Both leave ``decision`` as None. Emits BUDGET_TIMEOUT on timeout.
+        Returns ``(decision, skip)``. On success *skip* is None.
+
+        A budget overrun yields ``SKIPPED_TIMEOUT``. A failure *inside the
+        agent* - an unparseable reply, or no reply at all because the model
+        provider refused - yields ``SKIPPED_AGENT_ERROR``. Only a failure
+        reaching market data yields ``SKIPPED_DATA_UNAVAILABLE``.
+
+        Those are three different problems with three different fixes, and
+        filing one as another sends you to check Alpaca when the real story is
+        the model. The same reasoning already separates ``SKIPPED_OVERRUN``
+        (see ``_handle_overrun_candles``).
+
+        All three leave ``decision`` as None. Emits BUDGET_TIMEOUT on timeout.
         """
         stop_heartbeat = threading.Event()
         heartbeat_thread = threading.Thread(
@@ -283,15 +312,23 @@ def _run_investigation(
                             {"phase": "investigation", "budget_secs": self._investigation_budget},
                             session_id=session_id,
                         )
-                    return None, "SKIPPED_TIMEOUT"
+                    return None, SessionSkip("SKIPPED_TIMEOUT")
+                except MainAgentError as exc:
+                    # The agent was reached and failed to produce a usable
+                    # answer. Nothing to do with market data.
+                    typer.echo(
+                        f"[{session_id}] Investigation failed (agent): {exc}",
+                        err=True,
+                    )
+                    return None, SessionSkip("SKIPPED_AGENT_ERROR", str(exc))
                 except Exception as exc:
-                    # Market data or the agent itself was unreachable. Skipping
-                    # the session is correct; crashing the run is not.
+                    # Market data was unreachable. Skipping the session is
+                    # correct; crashing the run is not.
                     typer.echo(
                         f"[{session_id}] Investigation failed: {exc}",
                         err=True,
                     )
-                    return None, "SKIPPED_DATA_UNAVAILABLE"
+                    return None, SessionSkip("SKIPPED_DATA_UNAVAILABLE", str(exc))
         finally:
             stop_heartbeat.set()
             heartbeat_thread.join(timeout=1.0)
@@ -460,19 +497,21 @@ def _process_session(
 
         if active:
             typer.echo(f"[{session_id}] Investigating market snapshot …")
-            decision, skip_status = self._run_investigation(session_id, candle_close_at, active)
+            decision, skip = self._run_investigation(session_id, candle_close_at, active)
         else:
             # FR-005: no Investigation Agent call is made when every ticker is blocked.
             typer.echo(f"[{session_id}] All tickers feedback-blocked — skipping investigation")
-            decision, skip_status = SessionDecision(session_id=session_id, decisions=[]), None
+            decision, skip = SessionDecision(session_id=session_id, decisions=[]), None
 
         if decision is not None:
             decision, merge_warnings = self._merge_blocked_holds(decision, blocked)
             warnings.extend(merge_warnings)
 
-        if decision is None:
-            typer.echo(f"[{session_id}] SKIPPED  {skip_status}")
-            warnings.append(f"Session not completed: {skip_status}.")
+        if skip is not None:
+            typer.echo(f"[{session_id}] SKIPPED  {skip.status}")
+            warnings.append(f"Session not completed: {skip.status}.")
+            if skip.detail is not None:
+                warnings.append(f"Cause: {skip.detail}")
         else:
             decision_str = "  |  ".join(
                 f"{d.ticker}: {d.action} ({d.strategy})" for d in decision.decisions
@@ -526,7 +565,7 @@ def _process_session(
                 }
             )
 
-        session_status = "COMPLETED" if decision is not None else skip_status
+        session_status = skip.status if skip is not None else "COMPLETED"
         session_record = Session(
             id=session_id,
             run_id=run_id,

tests/unit/test_scheduler.py

@@ -8,10 +8,11 @@
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+from alphoryn.agents.main_agent import MainAgentError
 from alphoryn.config.models import AlphorynConfig
 from alphoryn.execution.agent import AssetDecision, SessionDecision
 from alphoryn.monitor.monitor import PositionMonitor
-from alphoryn.scheduler.scheduler import Scheduler
+from alphoryn.scheduler.scheduler import Scheduler, SessionSkip
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -499,13 +500,24 @@ def test_process_session_writes_data_unavailable_status() -> None:
     """Issue #136: SKIPPED_DATA_UNAVAILABLE is reachable, not just declared."""
     sched = _full_scheduler()
     with patch.object(
-        sched, "_run_investigation", return_value=(None, "SKIPPED_DATA_UNAVAILABLE")
+        sched, "_run_investigation", return_value=(None, SessionSkip("SKIPPED_DATA_UNAVAILABLE"))
     ):
         sched._process_session(1, "run-1/session-0001", 1, datetime.now(UTC))
 
     assert sched._bank.write_session.call_args.args[0].status == "SKIPPED_DATA_UNAVAILABLE"
 
 
+def test_process_session_writes_agent_error_status() -> None:
+    """An agent failure is its own status, not a market-data one."""
+    sched = _full_scheduler()
+    with patch.object(
+        sched, "_run_investigation", return_value=(None, SessionSkip("SKIPPED_AGENT_ERROR", "boom"))
+    ):
+        sched._process_session(1, "run-1/session-0001", 1, datetime.now(UTC))
+
+    assert sched._bank.write_session.call_args.args[0].status == "SKIPPED_AGENT_ERROR"
+
+
 # ---------------------------------------------------------------------------
 # Budget timeout (T029)
 # ---------------------------------------------------------------------------
@@ -622,20 +634,41 @@ def test_run_investigation_returns_timeout_status_on_timeout() -> None:
     sched = _full_scheduler(_investigation_budget_secs=0)
     sched._main_agent.decide.side_effect = lambda *a, **kw: time.sleep(0.5) or _FIXTURE_DECISION
     result = sched._run_investigation("sess-001", datetime.now(UTC), ["SPY", "QQQ"])
-    assert result == (None, "SKIPPED_TIMEOUT")
+    assert result == (None, SessionSkip("SKIPPED_TIMEOUT"))
 
 
-def test_run_investigation_returns_data_unavailable_when_the_agent_raises() -> None:
+def test_run_investigation_returns_data_unavailable_when_market_data_raises() -> None:
     """Issue #136: a market-data failure skips the session, it does not crash the run."""
     sched = _full_scheduler()
     sched._main_agent.decide.side_effect = RuntimeError("alpaca down")
     buf = StringIO()
     with patch("sys.stderr", buf):
         result = sched._run_investigation("sess-001", datetime.now(UTC), ["SPY", "QQQ"])
-    assert result == (None, "SKIPPED_DATA_UNAVAILABLE")
+    assert result == (None, SessionSkip("SKIPPED_DATA_UNAVAILABLE", "alpaca down"))
     assert "alpaca down" in buf.getvalue()
 
 
+def test_run_investigation_separates_an_agent_failure_from_a_data_failure() -> None:
+    """2026-08-13: two sessions died on the model and were filed as data outages.
+
+    A MainAgentError means the agent was reached and gave an unusable answer -
+    an unparseable reply, or none at all because the provider returned 429.
+    Market data was never involved.
+    """
+    sched = _full_scheduler()
+    sched._main_agent.decide.side_effect = MainAgentError(
+        "main_agent response is not valid JSON: Expecting value: line 1 column 1 (char 0)"
+    )
+    buf = StringIO()
+    with patch("sys.stderr", buf):
+        decision, skip = sched._run_investigation("sess-001", datetime.now(UTC), ["SPY"])
+
+    assert decision is None
+    assert skip.status == "SKIPPED_AGENT_ERROR"
+    assert "not valid JSON" in skip.detail
+    assert "Investigation failed (agent)" in buf.getvalue()
+
+
 # ---------------------------------------------------------------------------
 # _run_execute — direct tests
 # ---------------------------------------------------------------------------
@@ -657,7 +690,9 @@ def test_process_session_with_none_decision_writes_skipped_session() -> None:
     sched._main_agent = None  # force decision = None via direct override
 
     # Manually patch _run_investigation to return None
-    with patch.object(sched, "_run_investigation", return_value=(None, "SKIPPED_TIMEOUT")):
+    with patch.object(
+        sched, "_run_investigation", return_value=(None, SessionSkip("SKIPPED_TIMEOUT"))
+    ):
         sched._process_session(
             run_id=1,
             session_id="run-1/session-0001",
@@ -738,7 +773,7 @@ def test_investigation_timeout_no_logger_returns_none() -> None:
     sched._logger = None
     sched._main_agent.decide.side_effect = lambda *a, **kw: time.sleep(0.5) or _FIXTURE_DECISION
     result = sched._run_investigation("sess-001", datetime.now(UTC), ["SPY", "QQQ"])
-    assert result == (None, "SKIPPED_TIMEOUT")
+    assert result == (None, SessionSkip("SKIPPED_TIMEOUT"))
 
 
 def test_execute_timeout_no_logger_does_not_raise() -> None:
@@ -1254,14 +1289,32 @@ def test_a_clean_session_records_no_warnings() -> None:
 def test_a_skipped_session_records_why_as_a_warning() -> None:
     sched = _blocking_scheduler(set())
     with patch.object(
-        sched, "_run_investigation", return_value=(None, "SKIPPED_DATA_UNAVAILABLE")
+        sched, "_run_investigation", return_value=(None, SessionSkip("SKIPPED_DATA_UNAVAILABLE"))
     ):
         sched._process_session(1, "run-1/session-0001", 1, datetime.now(UTC))
 
     warnings = json.loads(sched._bank.write_session.call_args.args[0].warnings)
     assert warnings == ["Session not completed: SKIPPED_DATA_UNAVAILABLE."]
 
 
+def test_a_skipped_session_records_the_exception_text_in_the_bank() -> None:
+    """The cause must be readable from the bank alone, not only from stdout.
+
+    Without this the 2026-08-13 post-mortem had to go to the run log to learn
+    that a 'data unavailable' session was really a model failure.
+    """
+    sched = _blocking_scheduler(set())
+    skip = SessionSkip("SKIPPED_AGENT_ERROR", "main_agent produced no final response")
+    with patch.object(sched, "_run_investigation", return_value=(None, skip)):
+        sched._process_session(1, "run-1/session-0001", 1, datetime.now(UTC))
+
+    warnings = json.loads(sched._bank.write_session.call_args.args[0].warnings)
+    assert warnings == [
+        "Session not completed: SKIPPED_AGENT_ERROR.",
+        "Cause: main_agent produced no final response",
+    ]
+
+
 def test_a_ticker_the_investigation_dropped_is_recorded_as_a_warning() -> None:
     sched = _blocking_scheduler(set())
     with patch.object(sched, "_run_investigation", return_value=(_decision_for("SPY"), None)):
@@ -1360,7 +1413,9 @@ def test_merge_keeps_a_ticker_the_agent_invented() -> None:
 def test_timed_out_investigation_is_not_merged() -> None:
     """A budget timeout must still record SKIPPED_TIMEOUT, not a wall of Holds."""
     sched = _blocking_scheduler({"SPY"})
-    with patch.object(sched, "_run_investigation", return_value=(None, "SKIPPED_TIMEOUT")):
+    with patch.object(
+        sched, "_run_investigation", return_value=(None, SessionSkip("SKIPPED_TIMEOUT"))
+    ):
         sched._process_session(1, "run-1/session-0001", 1, datetime.now(UTC))
 
     written = sched._bank.write_session.call_args.args[0]
@@ -1470,11 +1525,11 @@ def test_skipped_sessions_do_not_count_against_session_budget() -> None:
     sched = _full_scheduler()
     call_count = 0
 
-    def mock_investigation(*args: Any, **kwargs: Any) -> tuple[Any, str | None]:
+    def mock_investigation(*args: Any, **kwargs: Any) -> tuple[Any, SessionSkip | None]:
         nonlocal call_count
         call_count += 1
         if call_count == 1:
-            return None, "SKIPPED_TIMEOUT"
+            return None, SessionSkip("SKIPPED_TIMEOUT")
         return _decision_for("SPY"), None
 
     with patch.object(sched, "_run_investigation", side_effect=mock_investigation):