Spaces:
Running
Running
| """End-to-end: the whole governance chain, against live market data. | |
| trigger → supervisor → parallel workers → writer → verify → human gate → deliver | |
| Every assertion here is about the *chain*, not about a particular market value: | |
| that a figure reconciles, that a fault is caught, that nothing ships without a | |
| human. Those hold whatever the market did today. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| from typing import Any | |
| import pytest | |
| from app.core.settings import Settings | |
| from app.models.run import HumanDecision, RunMode, RunStatus | |
| from app.services.repository import ActiveRunExistsError | |
| from app.services.runner import RunService | |
| SETTLE_TIMEOUT_S = 180.0 | |
| async def _settle(service: RunService, run_id: str) -> dict[str, Any]: | |
| """Wait until the run leaves RUNNING.""" | |
| deadline = asyncio.get_running_loop().time() + SETTLE_TIMEOUT_S | |
| while asyncio.get_running_loop().time() < deadline: | |
| record = await service.get_run(run_id) | |
| if record and record["status"] not in ("RUNNING", "QUEUED"): | |
| return record | |
| await asyncio.sleep(0.25) | |
| pytest.fail(f"run {run_id} did not settle within {SETTLE_TIMEOUT_S}s") | |
| def service(settings: Settings) -> RunService: | |
| return RunService(settings=settings) | |
| class TestHappyPath: | |
| async def test_full_chain_from_trigger_to_delivery(self, service: RunService) -> None: | |
| run_id = await service.start_run(tickers=["AAPL"], trigger="test") | |
| record = await _settle(service, run_id) | |
| # 1. It paused for a human rather than delivering itself. | |
| assert record["status"] == RunStatus.AWAITING_APPROVAL | |
| gate = await service.gate_payload(run_id) | |
| assert gate is not None | |
| report = gate["verification"] | |
| # 2. Every numeric claim was recomputed and matched. | |
| assert gate["verified"] is True | |
| assert report["checked_claims"] > 0 | |
| assert report["matched"] == report["checked_claims"] | |
| assert report["mismatched"] == 0 | |
| assert report["coverage"] == 1.0 | |
| # 3. The brief is well-formed and renders. | |
| assert gate["brief"]["snapshot"] | |
| assert gate["markdown"].startswith("# AlphaBrief") | |
| # 4. Approval releases it. | |
| result = await service.submit_decision( | |
| run_id, HumanDecision(action="approve", reviewer="tester") | |
| ) | |
| assert result["status"] == RunStatus.DELIVERED | |
| final = await service.get_run(run_id) | |
| assert final is not None | |
| assert final["status"] == RunStatus.DELIVERED | |
| assert final["verified"] is True | |
| assert final["iterations"] >= 1 | |
| async def test_telemetry_covers_the_whole_chain(self, service: RunService) -> None: | |
| run_id = await service.start_run(tickers=["MSFT"], trigger="test") | |
| await _settle(service, run_id) | |
| kinds = {str(event.kind) for event in await service.bus.history(run_id)} | |
| for expected in ( | |
| "run.started", | |
| "supervisor.plan", | |
| "agent.started", | |
| "agent.completed", | |
| "mcp.tool_call", | |
| "writer.started", | |
| "verify.started", | |
| "verify.claim", | |
| "verify.completed", | |
| "gate.awaiting", | |
| ): | |
| assert expected in kinds, f"missing telemetry: {expected}" | |
| await service.submit_decision(run_id, HumanDecision(action="approve")) | |
| async def test_tool_calls_carry_arguments_and_timing(self, service: RunService) -> None: | |
| run_id = await service.start_run(tickers=["NVDA"], trigger="test") | |
| await _settle(service, run_id) | |
| events = [e for e in await service.bus.history(run_id) if str(e.kind) == "mcp.tool_call"] | |
| assert events | |
| for event in events: | |
| assert "tool" in event.payload | |
| assert "arguments" in event.payload | |
| assert isinstance(event.payload["duration_ms"], float) | |
| await service.submit_decision(run_id, HumanDecision(action="approve")) | |
| class TestHumanGate: | |
| async def test_rejection_blocks_delivery(self, service: RunService) -> None: | |
| run_id = await service.start_run(tickers=["TSLA"], trigger="test") | |
| await _settle(service, run_id) | |
| result = await service.submit_decision( | |
| run_id, HumanDecision(action="reject", note="not today") | |
| ) | |
| assert result["status"] == RunStatus.REJECTED | |
| record = await service.get_run(run_id) | |
| assert record is not None | |
| assert record["status"] == RunStatus.REJECTED | |
| async def test_edit_applies_narrative_only(self, service: RunService) -> None: | |
| run_id = await service.start_run(tickers=["AMZN"], trigger="test") | |
| await _settle(service, run_id) | |
| await service.submit_decision( | |
| run_id, | |
| HumanDecision( | |
| action="edit", | |
| edited_headline="Reviewer rewrote this headline", | |
| reviewer="tester", | |
| ), | |
| ) | |
| stored = await service.repository.latest_brief(run_id) | |
| assert stored is not None | |
| assert stored["brief"]["headline"] == "Reviewer rewrote this headline" | |
| async def test_an_edit_containing_a_numeral_is_refused(self, service: RunService) -> None: | |
| """Reviewers may reword; they may not introduce an unverified figure.""" | |
| run_id = await service.start_run(tickers=["META"], trigger="test") | |
| await _settle(service, run_id) | |
| gate_before = await service.gate_payload(run_id) | |
| assert gate_before is not None | |
| original = gate_before["brief"]["headline"] | |
| await service.submit_decision( | |
| run_id, | |
| HumanDecision(action="edit", edited_headline="Stock jumped 42 percent today"), | |
| ) | |
| stored = await service.repository.latest_brief(run_id) | |
| assert stored is not None | |
| assert stored["brief"]["headline"] == original | |
| async def test_decision_on_an_unknown_run_is_refused(self, service: RunService) -> None: | |
| from app.services.runner import RunNotFoundError | |
| with pytest.raises(RunNotFoundError): | |
| await service.submit_decision("run_does_not_exist", HumanDecision(action="approve")) | |
| async def test_a_second_decision_is_refused(self, service: RunService) -> None: | |
| from app.services.runner import RunNotFoundError | |
| run_id = await service.start_run(tickers=["GOOGL"], trigger="test") | |
| await _settle(service, run_id) | |
| await service.submit_decision(run_id, HumanDecision(action="approve")) | |
| with pytest.raises(RunNotFoundError): | |
| await service.submit_decision(run_id, HumanDecision(action="approve")) | |
| class TestFaultInjectionModes: | |
| async def test_demo_fault_degrades_without_crashing(self, service: RunService) -> None: | |
| """The interview kill shot: a fake ticker must not break the run.""" | |
| run_id = await service.start_run(tickers=["AAPL"], mode=RunMode.DEMO_FAULT) | |
| record = await _settle(service, run_id) | |
| assert record["status"] == RunStatus.AWAITING_APPROVAL | |
| assert record["status"] != RunStatus.FAILED | |
| assert record["partial"] is True | |
| assert record["error_count"] > 0 | |
| gate = await service.gate_payload(run_id) | |
| assert gate is not None | |
| brief = gate["brief"] | |
| assert brief["partial"] is True | |
| assert brief["data_gaps"] | |
| # The dead ticker is in the watchlist but not in the snapshot. | |
| assert "ZZZZQQQQ" in brief["watchlist"] | |
| assert "ZZZZQQQQ" not in [row["ticker"] for row in brief["snapshot"]] | |
| # Everything that *did* resolve still verified cleanly. | |
| assert gate["verified"] is True | |
| await service.submit_decision(run_id, HumanDecision(action="approve")) | |
| async def test_demo_mismatch_is_caught_and_escalated(self, service: RunService) -> None: | |
| """Injected bad figure: one regeneration, then HUMAN_REVIEW. Never delivered.""" | |
| run_id = await service.start_run(tickers=["MSFT"], mode=RunMode.DEMO_MISMATCH) | |
| record = await _settle(service, run_id) | |
| assert record["status"] == RunStatus.HUMAN_REVIEW | |
| gate = await service.gate_payload(run_id) | |
| assert gate is not None | |
| assert gate["verified"] is False | |
| assert gate["requires_review"] is True | |
| assert gate["verification"]["mismatched"] >= 1 | |
| events = await service.bus.history(run_id) | |
| # Exactly one regeneration was attempted before escalating. | |
| assert sum(1 for e in events if str(e.kind) == "writer.started") == 2 | |
| assert sum(1 for e in events if str(e.kind) == "verify.completed") == 2 | |
| await service.submit_decision(run_id, HumanDecision(action="reject", note="bad numbers")) | |
| final = await service.get_run(run_id) | |
| assert final is not None | |
| assert final["status"] == RunStatus.REJECTED | |
| class TestConcurrencyGuard: | |
| async def test_one_active_run_per_watchlist(self, service: RunService) -> None: | |
| run_id = await service.start_run(tickers=["INTC"], trigger="test") | |
| with pytest.raises(ActiveRunExistsError): | |
| await service.start_run(tickers=["intc"], trigger="test") | |
| await _settle(service, run_id) | |
| await service.submit_decision(run_id, HumanDecision(action="approve")) | |
| # Slot released. | |
| second = await service.start_run(tickers=["INTC"], trigger="test") | |
| await _settle(service, second) | |
| await service.submit_decision(second, HumanDecision(action="approve")) | |