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):