From 5f163e23cb51c52c37cb3b769d48c6fb1c4ecbb2 Mon Sep 17 00:00:00 2001 From: Mohammad Date: Thu, 13 Aug 2026 20:18:28 +0200 Subject: [PATCH] fix: name the real reason a session failed, and record it Two sessions in the 2026-08-13 live run failed for two unrelated reasons and both were filed as SKIPPED_DATA_UNAVAILABLE: session-0014 main_agent reply was not JSON session-0019 Vertex AI 429 RESOURCE_EXHAUSTED Market data was fine in both cases. The exception text existed only on stdout, so anyone reading the memory bank afterwards would go and check Alpaca for an outage that never happened. An agent that was reached and gave an unusable answer is now SKIPPED_AGENT_ERROR, and SKIPPED_DATA_UNAVAILABLE means only what its name says. This is the same distinction _handle_overrun_candles already draws for SKIPPED_OVERRUN. The cause travels with the status as SessionSkip(status, detail) rather than a bare string, because keeping them apart is how they drifted in the first place. The detail is persisted to the session's warnings, so a post-mortem can read the real cause out of the bank alone. --- alphoryn/memory/schema.py | 4 ++ alphoryn/scheduler/scheduler.py | 71 +++++++++++++++++++++++------- tests/unit/test_scheduler.py | 77 ++++++++++++++++++++++++++++----- 3 files changed, 125 insertions(+), 27 deletions(-) diff --git a/alphoryn/memory/schema.py b/alphoryn/memory/schema.py index b6d0c18..df8f12f 100644 --- a/alphoryn/memory/schema.py +++ b/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) diff --git a/alphoryn/scheduler/scheduler.py b/alphoryn/scheduler/scheduler.py index 1c8730f..9cfc62c 100644 --- a/alphoryn/scheduler/scheduler.py +++ b/alphoryn/scheduler/scheduler.py @@ -13,6 +13,7 @@ import sys import threading import time +from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any @@ -20,7 +21,7 @@ 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, diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 83da817..3616c51 100644 --- a/tests/unit/test_scheduler.py +++ b/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,7 +1289,7 @@ 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)) @@ -1262,6 +1297,24 @@ def test_a_skipped_session_records_why_as_a_warning() -> None: 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):