Files
ai-exchange/tests/unit/test_run_until_blocked.py

1459 lines
73 KiB
Python

import contextlib
import io
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
from scripts import run_until_blocked
class RunUntilBlockedTests(unittest.TestCase):
def setUp(self) -> None:
self._status_tmp = tempfile.TemporaryDirectory()
self._original_runner_status_path = run_until_blocked.RUNNER_STATUS_PATH
self._original_monitor_history_path = run_until_blocked.ANALYSIS_DECISION_MONITOR_HISTORY_PATH
self._original_monitor_trend_path = run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH
self._original_monitor_quality_gate_path = run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH
self._original_monitor_maturity_rollup_path = run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH
self._original_monitor_remote_drift_check_path = run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_CHECK_PATH
self._original_support_quality_review_path = run_until_blocked.ANALYSIS_DECISION_SUPPORT_QUALITY_REVIEW_PATH
self._original_monitor_p7_selector_path = run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_PATH
run_until_blocked.RUNNER_STATUS_PATH = Path(self._status_tmp.name) / "automation_status.json"
run_until_blocked.ANALYSIS_DECISION_MONITOR_HISTORY_PATH = Path(self._status_tmp.name) / "analysis_decision_monitor_history.jsonl"
run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH = Path(self._status_tmp.name) / "analysis_decision_monitor_trend.json"
run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH = Path(self._status_tmp.name) / "analysis_decision_monitor_quality_gate.json"
run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH = Path(self._status_tmp.name) / "analysis_decision_monitor_maturity_rollup.json"
run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_CHECK_PATH = Path(self._status_tmp.name) / "analysis_decision_monitor_remote_drift_check.json"
run_until_blocked.ANALYSIS_DECISION_SUPPORT_QUALITY_REVIEW_PATH = Path(self._status_tmp.name) / "analysis_decision_support_quality_review.json"
run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_PATH = Path(self._status_tmp.name) / "analysis_decision_monitor_p7_slice_selector.json"
def tearDown(self) -> None:
run_until_blocked.RUNNER_STATUS_PATH = self._original_runner_status_path
run_until_blocked.ANALYSIS_DECISION_MONITOR_HISTORY_PATH = self._original_monitor_history_path
run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH = self._original_monitor_trend_path
run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH = self._original_monitor_quality_gate_path
run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH = self._original_monitor_maturity_rollup_path
run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_CHECK_PATH = self._original_monitor_remote_drift_check_path
run_until_blocked.ANALYSIS_DECISION_SUPPORT_QUALITY_REVIEW_PATH = self._original_support_quality_review_path
run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_PATH = self._original_monitor_p7_selector_path
self._status_tmp.cleanup()
def test_selector_picks_first_unfinished_execution_layer_slice(self) -> None:
selection = run_until_blocked.select_automation_execution_slice(
anchor_state={
"missing_required_anchors": [],
"completed_execution_slices": ["automation_execution_layer_p0_governance_gate"],
"in_progress_execution_slices": ["automation_execution_layer_p1_anchor_slice_selector"],
},
tick_payload={
"status": "no_op_healthy",
"recommended_actions": ["continue_periodic_optimization_cycle"],
},
)
self.assertEqual(selection["status"], "ready")
self.assertEqual(selection["slice"], "automation_execution_layer_p1_anchor_slice_selector")
self.assertTrue(selection["requires_code"])
self.assertTrue(selection["already_in_progress"])
def test_required_automation_anchor_paths_match_new_kickoff_prompt(self) -> None:
anchors = run_until_blocked.get_automation_required_anchor_paths()
self.assertEqual(
[path.name for path in anchors],
[
"AI_ICT_实盘交易分析决策助手下一阶段开发方案_v1.md",
"SESSION.md",
"TASKS.md",
],
)
def test_render_session_uses_new_kickoff_anchors(self) -> None:
text = run_until_blocked.render_session(
completed_stages=[],
current_stage="P0-2",
last_result={},
blocked_reason=None,
)
self.assertIn("AI_ICT_实盘交易分析决策助手下一阶段开发方案_v1.md", text)
self.assertIn("SESSION.md", text)
self.assertIn("TASKS.md", text)
self.assertNotIn("AI_ICT_实盘辅助决策系统开发方案_v2.md", text)
def test_selector_blocks_when_required_anchors_are_missing(self) -> None:
selection = run_until_blocked.select_automation_execution_slice(
anchor_state={
"missing_required_anchors": ["docs/missing.md"],
"completed_execution_slices": [],
"in_progress_execution_slices": [],
},
tick_payload={"status": "no_op_healthy"},
)
self.assertEqual(selection["status"], "blocked")
self.assertEqual(selection["slice"], "repair_missing_automation_anchors")
self.assertEqual(selection["blocked_reason"], "missing_required_anchors:docs/missing.md")
self.assertFalse(selection["requires_code"])
def test_validation_plan_for_execution_layer_slice_uses_focused_checks(self) -> None:
plan = run_until_blocked.build_validation_plan(
selected_slice={
"slice": "automation_execution_layer_p2_validation_quality_router",
"requires_code": True,
},
changed_files=["scripts/run_until_blocked.py", "tests/unit/test_run_until_blocked.py"],
)
self.assertIn("D:\\ai-exchange\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_run_until_blocked", plan["focused_commands"])
self.assertIn("D:\\ai-exchange\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py", plan["quality_gates"])
self.assertIn("npm run lint", plan["quality_gates"])
self.assertFalse(plan["deploy_required"])
def test_validation_plan_requires_deploy_for_declared_runtime_api_changes(self) -> None:
plan = run_until_blocked.build_validation_plan(
selected_slice={
"slice": "automation_execution_layer_p3_deploy_smoke_router",
"requires_code": True,
},
changed_files=["src/api/research_council.py"],
)
self.assertTrue(plan["deploy_required"])
self.assertEqual(plan["deploy_reason"], "declared_runtime_api_db_or_workbench_change")
self.assertIn("deploy_smoke_required_by_declared_runtime_api_db_or_workbench_change", plan["notes"])
self.assertTrue(plan["deploy_smoke_plan"]["deploy_required"])
self.assertIn("/health", plan["deploy_smoke_plan"]["smoke_targets"])
self.assertIn("POST /trading/research-council/governance-tick", plan["deploy_smoke_plan"]["smoke_targets"])
def test_deploy_smoke_plan_stays_local_for_script_docs_state_changes(self) -> None:
plan = run_until_blocked.build_validation_plan(
selected_slice={
"slice": "automation_execution_layer_p3_deploy_smoke_router",
"requires_code": True,
},
changed_files=[
"scripts/run_until_blocked.py",
"tests/unit/test_run_until_blocked.py",
"docs/archive/AI_ICT_自动化流程使用教程_v1.md",
"SESSION.md",
"TASKS.md",
],
)
self.assertFalse(plan["deploy_required"])
self.assertFalse(plan["deploy_smoke_plan"]["deploy_required"])
self.assertEqual(plan["deploy_smoke_plan"]["smoke_targets"], [])
def test_state_writeback_plan_tracks_status_and_next_slice(self) -> None:
summary = run_until_blocked.build_automation_execution_summary(
started_ts="2026-05-11T00:00:00+00:00",
tick_payload={
"status": "no_op_healthy",
"effective_health_state": "healthy",
"safe_to_auto_continue": True,
"recommended_actions": ["continue_periodic_optimization_cycle"],
"boundary_notes": [],
},
anchor_state={
"missing_required_anchors": [],
"completed_execution_slices": [
"automation_execution_layer_p0_governance_gate",
"automation_execution_layer_p1_anchor_slice_selector",
"automation_execution_layer_p2_validation_quality_router",
"automation_execution_layer_p3_deploy_smoke_router",
],
"in_progress_execution_slices": ["automation_execution_layer_p4_state_writeback_renderer"],
},
changed_files=["scripts/run_until_blocked.py"],
)
plan = summary["state_writeback_plan"]
self.assertEqual(summary["next_slice"], "automation_execution_layer_p4_state_writeback_renderer")
self.assertEqual(plan["state"], "automation_execution_layer_p3_deploy_smoke_router_closed")
self.assertIn("SESSION.md", plan["files_to_write"])
self.assertIn("TASKS.md", plan["files_to_write"])
self.assertTrue(plan["no_new_phase_created"])
self.assertIn("execution_allowed", plan["guards_unchanged"])
def test_automation_execution_layer_noops_on_healthy_governance_tick(self) -> None:
class _HealthyGovernanceTickService:
def run_tick(self, **kwargs):
return {
"status": "no_op_healthy",
"passed": True,
"instrument_id": kwargs["instrument_id"],
"effective_health_state": "healthy",
"safe_to_auto_continue": True,
"requires_cycle_run": False,
"requires_bounded_slice": False,
"recommended_actions": ["continue_periodic_optimization_cycle"],
"boundary_notes": [
"governance_tick_does_not_change_execution_gate",
"governance_tick_does_not_change_live_readiness",
"governance_tick_does_not_change_trust_bucket",
],
}
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n- state: automation_flow_p0_p6_review_closed\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n- [ ] Start automation execution-layer development\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_service = run_until_blocked.CouncilGovernanceTickService
original_rehearsal = run_until_blocked.run_trading_delivery_rehearsal
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.CouncilGovernanceTickService = _HealthyGovernanceTickService
run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: (_ for _ in ()).throw(
AssertionError("legacy stage runner should not be called")
)
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.CouncilGovernanceTickService = original_service
run_until_blocked.run_trading_delivery_rehearsal = original_rehearsal
self.assertEqual(summary["status"], "completed")
self.assertEqual(summary["mode"], "automation_execution_layer")
self.assertEqual(summary["tick_status"], "no_op_healthy")
self.assertEqual(summary["next_slice"], "automation_execution_layer_p1_anchor_slice_selector")
self.assertEqual(summary["selected_slice"]["reason"], "first_unfinished_execution_layer_slice")
self.assertEqual(summary["anchor_state"]["state"], "automation_flow_p0_p6_review_closed")
self.assertIn("npm run lint", summary["validation_plan"]["quality_gates"])
self.assertTrue(summary["no_new_phase_created"])
self.assertIn("automation_execution_layer_p0_governance_gate_closed", session_text)
self.assertIn("Automation Execution-Layer P0", tasks_text)
self.assertNotIn("### P7", session_text)
self.assertNotIn("- [ ] `P7", tasks_text)
self.assertEqual(status_payload["state"], "completed")
self.assertEqual(status_payload["mode"], "automation_execution_layer")
self.assertEqual(status_payload["tick_status"], "no_op_healthy")
def test_analysis_decision_monitor_phase_runs_monitor_instead_of_legacy_stages(self) -> None:
monitor_payload = {
"passed": True,
"generated_ts": "2026-05-11T04:45:00+00:00",
"phase": "analysis_decision_only",
"blockers": [],
"warnings": [],
"observed": {
"broker_status": "broker_state_ok",
"broker_fills": 0,
"runtime_health": "ok",
"market_stream": "market_stream_connected",
"market_freshness": "market_data_fresh",
"market_gap": "market_data_gap_clear",
"real_trade_sample_gate_required": False,
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
"p6_governance_tick_status": "no_op_healthy",
},
}
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
"- state: analysis_decision_monitor_runner_deployed\n"
"- current_phase: Restricted Decision-Support Soft Rollout\n"
"- repair_next_action: run_analysis_decision_monitor_periodically_without_fill_gate\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
"- [ ] `run_analysis_decision_monitor_periodically_without_fill_gate`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_monitor = run_until_blocked.run_analysis_decision_monitor.run_monitor
original_rehearsal = run_until_blocked.run_trading_delivery_rehearsal
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.run_analysis_decision_monitor.run_monitor = lambda **kwargs: monitor_payload
run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: (_ for _ in ()).throw(
AssertionError("legacy stage runner should not be called")
)
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.run_analysis_decision_monitor.run_monitor = original_monitor
run_until_blocked.run_trading_delivery_rehearsal = original_rehearsal
self.assertEqual(summary["status"], "completed")
self.assertEqual(summary["mode"], "analysis_decision_monitor")
self.assertTrue(summary["monitor_passed"])
self.assertEqual(summary["next_slice"], "continue_periodic_analysis_decision_monitoring")
self.assertIsNone(summary["blocked_reason"])
self.assertIn("analysis_decision_monitor_runner_automation_wired", session_text)
self.assertIn("latest_analysis_decision_monitor_status: passed", session_text)
self.assertIn("latest_real_trade_sample_gate_status: deferred_analysis_decision_only", session_text)
self.assertIn("continue_periodic_analysis_decision_monitoring", tasks_text)
self.assertNotIn("wait_for_and_verify_real_fill_sample", session_text)
self.assertEqual(status_payload["state"], "completed")
self.assertEqual(status_payload["mode"], "analysis_decision_monitor")
self.assertEqual(status_payload["monitor_blockers"], [])
self.assertEqual(summary["monitor_history"]["record_count"], 1)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_MONITOR_HISTORY_PATH.exists())
history_record = json.loads(run_until_blocked.ANALYSIS_DECISION_MONITOR_HISTORY_PATH.read_text(encoding="utf-8").strip())
self.assertTrue(history_record["monitor_passed"])
self.assertEqual(history_record["observed"]["broker_fills"], 0)
self.assertIn("latest_analysis_decision_monitor_history_records: 1", session_text)
self.assertIn("latest_analysis_decision_monitor_history_records: 1", tasks_text)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH.exists())
trend_payload = json.loads(run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH.read_text(encoding="utf-8"))
self.assertEqual(trend_payload["record_count"], 1)
self.assertEqual(trend_payload["consecutive_pass_count"], 1)
self.assertFalse(trend_payload["stable_monitoring_window"])
self.assertIn("latest_analysis_decision_monitor_consecutive_passes: 1", session_text)
self.assertIn("latest_analysis_decision_monitor_stable_window: false", tasks_text)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH.exists())
quality_gate_payload = json.loads(run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH.read_text(encoding="utf-8"))
self.assertEqual(quality_gate_payload["status"], "attention_required")
self.assertIn("stable_monitoring_window_not_reached", quality_gate_payload["failure_reasons"])
self.assertIn("latest_analysis_decision_monitor_quality_gate_status: attention_required", session_text)
def test_analysis_decision_monitor_routes_to_p7_after_stable_quality_gate(self) -> None:
observed = {
"broker_status": "broker_state_ok",
"broker_fills": 0,
"runtime_health": "ok",
"market_stream": "market_stream_connected",
"market_freshness": "market_data_fresh",
"market_gap": "market_data_gap_clear",
"real_trade_sample_gate_required": False,
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
"p6_governance_tick_status": "no_op_healthy",
}
existing_records = [
{
"recorded_ts": "2026-05-11T04:35:00+00:00",
"monitor_generated_ts": "2026-05-11T04:35:00+00:00",
"status": "completed",
"monitor_passed": True,
"blockers": [],
"warnings": [],
"observed": observed,
"next_slice": "continue_periodic_analysis_decision_monitoring",
"guards_unchanged": ["execution_allowed"],
},
{
"recorded_ts": "2026-05-11T04:40:00+00:00",
"monitor_generated_ts": "2026-05-11T04:40:00+00:00",
"status": "completed",
"monitor_passed": True,
"blockers": [],
"warnings": [],
"observed": observed,
"next_slice": "continue_periodic_analysis_decision_monitoring",
"guards_unchanged": ["execution_allowed"],
},
]
run_until_blocked.ANALYSIS_DECISION_MONITOR_HISTORY_PATH.write_text(
"\n".join(json.dumps(record, sort_keys=True) for record in existing_records) + "\n",
encoding="utf-8",
)
monitor_payload = {
"passed": True,
"generated_ts": "2026-05-11T04:45:00+00:00",
"phase": "analysis_decision_only",
"blockers": [],
"warnings": [],
"observed": observed,
}
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
"- state: analysis_decision_monitor_runner_deployed\n"
"- current_phase: Restricted Decision-Support Soft Rollout\n"
"- repair_next_action: continue_periodic_analysis_decision_monitoring\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
"- [ ] `continue_periodic_analysis_decision_monitoring`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_monitor = run_until_blocked.run_analysis_decision_monitor.run_monitor
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.run_analysis_decision_monitor.run_monitor = lambda **kwargs: monitor_payload
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.run_analysis_decision_monitor.run_monitor = original_monitor
self.assertEqual(summary["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE)
self.assertEqual(summary["monitor_history"]["record_count"], 3)
self.assertEqual(summary["monitor_history"]["quality_gate"]["status"], "passed")
self.assertEqual(
summary["monitor_history"]["quality_gate"]["next_slice"],
run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE,
)
self.assertIn(run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE, session_text)
self.assertIn("automation execution-layer P7", session_text)
self.assertIn("Stable monitor quality may enter automation execution-layer P7", tasks_text)
def test_p7_monitor_maturity_rollup_closes_and_routes_to_remote_drift_check(self) -> None:
observed = {
"broker_status": "broker_state_ok",
"broker_fills": 0,
"runtime_health": "ok",
"market_gap": "market_data_gap_clear",
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
}
trend = {
"record_count": 9,
"consecutive_pass_count": 9,
"stable_sample_threshold": 3,
"stable_monitoring_window": True,
"latest_monitor_passed": True,
"latest_monitor_generated_ts": "2026-05-11T08:47:56+00:00",
"latest_blockers": [],
"latest_warnings": [],
"latest_observed": observed,
"next_slice": run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE,
"guards_unchanged": [
"execution_allowed",
"real_live_ready",
"risk_governor",
"trust_bucket",
"operator_confirmation_protocol",
],
}
quality_gate = {
"status": "passed",
"passed": True,
"failure_reasons": [],
"record_count": 9,
"consecutive_pass_count": 9,
"stable_monitoring_window": True,
"latest_monitor_generated_ts": "2026-05-11T08:47:56+00:00",
"latest_observed": observed,
"next_slice": run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE,
"guards_unchanged": trend["guards_unchanged"],
}
run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH.write_text(
json.dumps(trend),
encoding="utf-8",
)
run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH.write_text(
json.dumps(quality_gate),
encoding="utf-8",
)
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
"### automation_execution_layer_p7_monitor_maturity_rollup\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
f"- [ ] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE}`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
self.assertEqual(summary["mode"], "analysis_decision_monitor_maturity_rollup")
self.assertEqual(summary["status"], "completed")
self.assertEqual(summary["rollup"]["state"], "automation_execution_layer_p7_monitor_maturity_rollup_closed")
self.assertEqual(summary["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_REMOTE_DRIFT_SLICE)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH.exists())
rollup_payload = json.loads(run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH.read_text(encoding="utf-8"))
self.assertEqual(rollup_payload["maturity_status"], "mature")
self.assertEqual(rollup_payload["remote_drift_check_status"], "pending")
self.assertIn("automation_execution_layer_p7_monitor_maturity_rollup_closed", session_text)
self.assertIn(run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_REMOTE_DRIFT_SLICE, tasks_text)
self.assertEqual(status_payload["mode"], "analysis_decision_monitor_maturity_rollup")
def test_p7_remote_drift_check_closes_and_routes_to_quality_review(self) -> None:
observed = {
"broker_status": "broker_state_ok",
"broker_fills": 0,
"runtime_health": "ok",
"market_gap": "market_data_gap_clear",
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
"p6_governance_tick_status": "no_op_healthy",
}
guards = [
"execution_allowed",
"real_live_ready",
"risk_governor",
"trust_bucket",
"operator_confirmation_protocol",
]
local_rollup = {
"maturity_status": "mature",
"local": {
"quality_gate_status": "passed",
"record_count": 9,
"consecutive_pass_count": 9,
"stable_monitoring_window": True,
},
"latest_observed": observed,
"guards_unchanged": guards,
}
remote_quality = {
"status": "passed",
"passed": True,
"record_count": 12,
"consecutive_pass_count": 4,
"stable_monitoring_window": True,
"latest_observed": observed,
"guards_unchanged": guards,
}
remote_state = {
"status": "available",
"host": "192.168.3.90",
"container_quality_gate": remote_quality,
"container_trend": {"stable_sample_threshold": 3, "latest_observed": observed},
"host_quality_gate": remote_quality,
}
run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH.write_text(
json.dumps(local_rollup),
encoding="utf-8",
)
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
"### automation_execution_layer_p7_remote_drift_check\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
f"- [ ] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_REMOTE_DRIFT_SLICE}`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_fetch_remote = run_until_blocked.fetch_remote_analysis_decision_monitor_state
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.fetch_remote_analysis_decision_monitor_state = lambda *, env: remote_state
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.fetch_remote_analysis_decision_monitor_state = original_fetch_remote
self.assertEqual(summary["mode"], "analysis_decision_monitor_remote_drift_check")
self.assertEqual(summary["status"], "completed")
self.assertEqual(summary["drift_check"]["drift_status"], "clear")
self.assertEqual(summary["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_QUALITY_REVIEW_SLICE)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_CHECK_PATH.exists())
self.assertIn("automation_execution_layer_p7_remote_drift_check_closed", session_text)
self.assertIn(run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_QUALITY_REVIEW_SLICE, tasks_text)
self.assertEqual(status_payload["mode"], "analysis_decision_monitor_remote_drift_check")
def test_p7_remote_drift_check_blocks_when_remote_quality_not_passed(self) -> None:
local_rollup = {
"maturity_status": "mature",
"local": {"quality_gate_status": "passed"},
"latest_observed": {"runtime_health": "ok"},
"guards_unchanged": ["execution_allowed"],
}
remote_state = {
"status": "available",
"container_quality_gate": {"status": "attention_required", "latest_observed": {"runtime_health": "noisy"}},
"container_trend": {"stable_sample_threshold": 3, "consecutive_pass_count": 0},
"host_quality_gate": {"status": "attention_required"},
}
drift = run_until_blocked.build_analysis_decision_monitor_remote_drift_check(
local_rollup=local_rollup,
remote_state=remote_state,
)
self.assertEqual(drift["status"], "blocked")
self.assertEqual(drift["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_REPAIR_SLICE)
self.assertIn("remote_container_quality_gate_not_passed", drift["failure_reasons"])
def test_p7_selectors_ignore_completed_milestone_mentions(self) -> None:
session_text = (
"# ai-exchange Session\n\n"
"- current_phase: Restricted Decision-Support Soft Rollout\n"
f"### {run_until_blocked.ANALYSIS_DECISION_MONITOR_WITH_MATURITY_SLICE}\n"
)
tasks_text = (
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
f"- [ ] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_WITH_MATURITY_SLICE}`\n\n"
"## Major Done Milestones\n\n"
f"- [x] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE}`\n"
f"- [x] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_REMOTE_DRIFT_SLICE}`\n"
f"- [x] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_QUALITY_REVIEW_SLICE}`\n"
f"- [x] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE}`\n"
)
self.assertFalse(
run_until_blocked.should_use_analysis_decision_monitor_p7_selector(
session_text=session_text,
tasks_text=tasks_text,
)
)
self.assertFalse(
run_until_blocked.should_use_analysis_decision_monitor_maturity_rollup(
session_text=session_text,
tasks_text=tasks_text,
)
)
self.assertFalse(
run_until_blocked.should_use_analysis_decision_monitor_remote_drift_check(
session_text=session_text,
tasks_text=tasks_text,
)
)
self.assertFalse(
run_until_blocked.should_use_analysis_decision_support_quality_review(
session_text=session_text,
tasks_text=tasks_text,
)
)
self.assertTrue(
run_until_blocked.should_use_analysis_decision_monitor(
session_text=session_text,
tasks_text=tasks_text,
)
)
def test_p7_quality_review_closes_and_routes_to_p7_selector(self) -> None:
observed = {
"broker_status": "broker_state_ok",
"broker_fills": 0,
"runtime_health": "ok",
"market_gap": "market_data_gap_clear",
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
}
monitor_payload = {
"passed": True,
"generated_ts": "2026-05-11T09:15:00+00:00",
"phase": "analysis_decision_only",
"blockers": [],
"warnings": [],
"observed": observed,
"checks": [
{"code": "decision_restricted_support", "passed": True},
{"code": "decision_observe_only", "passed": True},
{"code": "decision_execution_not_allowed", "passed": True},
{"code": "review_packet_present", "passed": True},
{"code": "review_trust_score_present", "passed": True},
{"code": "review_research_summary_present", "passed": True},
{"code": "real_trade_sample_gate_deferred", "passed": True},
],
}
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
"### automation_execution_layer_p7_alert_decision_support_quality_review\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
f"- [ ] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_QUALITY_REVIEW_SLICE}`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_monitor = run_until_blocked.run_analysis_decision_monitor.run_monitor
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.run_analysis_decision_monitor.run_monitor = lambda **kwargs: monitor_payload
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.run_analysis_decision_monitor.run_monitor = original_monitor
self.assertEqual(summary["mode"], "analysis_decision_support_quality_review")
self.assertEqual(summary["status"], "completed")
self.assertEqual(summary["quality_review"]["quality_status"], "healthy")
self.assertEqual(summary["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_SUPPORT_QUALITY_REVIEW_PATH.exists())
review_payload = json.loads(run_until_blocked.ANALYSIS_DECISION_SUPPORT_QUALITY_REVIEW_PATH.read_text(encoding="utf-8"))
self.assertEqual(review_payload["state"], "automation_execution_layer_p7_alert_decision_support_quality_review_closed")
self.assertIn("automation_execution_layer_p7_alert_decision_support_quality_review_closed", session_text)
self.assertIn(run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE, tasks_text)
self.assertEqual(status_payload["mode"], "analysis_decision_support_quality_review")
def test_p7_quality_review_blocks_when_required_research_summary_missing(self) -> None:
monitor_payload = {
"passed": True,
"blockers": [],
"warnings": [],
"observed": {
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
},
"checks": [
{"code": "decision_restricted_support", "passed": True},
{"code": "decision_observe_only", "passed": True},
{"code": "decision_execution_not_allowed", "passed": True},
{"code": "review_packet_present", "passed": True},
{"code": "review_trust_score_present", "passed": True},
{"code": "real_trade_sample_gate_deferred", "passed": True},
],
}
review = run_until_blocked.build_analysis_decision_support_quality_review(
monitor_payload=monitor_payload,
)
self.assertEqual(review["status"], "blocked")
self.assertEqual(review["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE)
self.assertIn("required_quality_checks_missing", review["failure_reasons"])
self.assertIn("review_research_summary_present", review["missing_required_checks"])
def test_p7_selector_stage_closes_and_routes_to_maturity_rollup(self) -> None:
guards = ["execution_allowed"]
run_until_blocked.ANALYSIS_DECISION_MONITOR_TREND_PATH.write_text(
json.dumps(
{
"record_count": 10,
"consecutive_pass_count": 10,
"stable_sample_threshold": 3,
"stable_monitoring_window": True,
"latest_monitor_passed": True,
"latest_blockers": [],
"guards_unchanged": guards,
}
),
encoding="utf-8",
)
run_until_blocked.ANALYSIS_DECISION_MONITOR_QUALITY_GATE_PATH.write_text(
json.dumps(
{
"status": "passed",
"passed": True,
"stable_monitoring_window": True,
"failure_reasons": [],
"guards_unchanged": guards,
}
),
encoding="utf-8",
)
run_until_blocked.ANALYSIS_DECISION_MONITOR_MATURITY_ROLLUP_PATH.write_text(
json.dumps(
{
"state": "automation_execution_layer_p7_monitor_maturity_rollup_closed",
"status": "completed",
"maturity_status": "mature",
"guards_unchanged": guards,
}
),
encoding="utf-8",
)
run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_CHECK_PATH.write_text(
json.dumps(
{
"state": "automation_execution_layer_p7_remote_drift_check_closed",
"status": "completed",
"drift_status": "clear",
"failure_reasons": [],
"guards_unchanged": guards,
}
),
encoding="utf-8",
)
run_until_blocked.ANALYSIS_DECISION_SUPPORT_QUALITY_REVIEW_PATH.write_text(
json.dumps(
{
"state": "automation_execution_layer_p7_alert_decision_support_quality_review_closed",
"status": "completed",
"quality_status": "healthy",
"failure_reasons": [],
"guards_unchanged": guards,
}
),
encoding="utf-8",
)
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
f"### {run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE}\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"## In Progress\n\n"
f"- [ ] `{run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE}`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "0"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
self.assertEqual(summary["mode"], "analysis_decision_monitor_p7_selector")
self.assertEqual(summary["status"], "completed")
self.assertEqual(summary["selector"]["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_WITH_MATURITY_SLICE)
self.assertTrue(run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_PATH.exists())
self.assertIn("automation_execution_layer_p7_bounded_automation_slice_selector_closed", session_text)
self.assertIn(run_until_blocked.ANALYSIS_DECISION_MONITOR_WITH_MATURITY_SLICE, tasks_text)
self.assertEqual(status_payload["mode"], "analysis_decision_monitor_p7_selector")
def test_p7_selector_chooses_maturity_rollup_when_all_inputs_are_healthy(self) -> None:
observed = {
"broker_status": "broker_state_ok",
"broker_fills": 0,
"runtime_health": "ok",
"market_gap": "market_data_gap_clear",
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"decision_trust_classification": "blacklist",
"p6_governance_tick_status": "no_op_healthy",
}
trend = {
"record_count": 10,
"consecutive_pass_count": 10,
"stable_sample_threshold": 3,
"stable_monitoring_window": True,
"latest_monitor_passed": True,
"latest_monitor_generated_ts": "2026-05-11T10:13:03+00:00",
"latest_blockers": [],
"latest_warnings": [],
"latest_observed": observed,
"next_slice": run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE,
"guards_unchanged": [
"execution_allowed",
"real_live_ready",
"risk_governor",
"trust_bucket",
"operator_confirmation_protocol",
],
}
quality_gate = {
"status": "passed",
"passed": True,
"failure_reasons": [],
"record_count": 10,
"consecutive_pass_count": 10,
"stable_monitoring_window": True,
"latest_monitor_generated_ts": "2026-05-11T10:13:03+00:00",
"latest_observed": observed,
"next_slice": run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_NEXT_SLICE,
"guards_unchanged": trend["guards_unchanged"],
}
maturity_rollup = {
"state": "automation_execution_layer_p7_monitor_maturity_rollup_closed",
"status": "completed",
"maturity_status": "mature",
"local": {
"quality_gate_status": "passed",
"record_count": 10,
"consecutive_pass_count": 10,
"stable_monitoring_window": True,
"quality_gate_passed": True,
},
"latest_observed": observed,
"remote_drift_check_status": "clear",
"guards_unchanged": trend["guards_unchanged"],
}
remote_drift_check = {
"state": "automation_execution_layer_p7_remote_drift_check_closed",
"status": "completed",
"drift_status": "clear",
"local": {
"maturity_status": "mature",
"quality_gate_status": "passed",
"record_count": 10,
"consecutive_pass_count": 10,
"stable_monitoring_window": True,
},
"remote": {
"status": "available",
"host": "192.168.3.90",
"container_quality_gate_status": "passed",
"container_record_count": 12,
"container_consecutive_pass_count": 4,
"container_stable_monitoring_window": True,
"host_quality_gate_status": "passed",
"host_record_count": 12,
"host_consecutive_pass_count": 4,
"host_stable_monitoring_window": True,
},
"observed_diffs": {},
"latest_observed": {"local": observed, "remote": observed},
"guards_unchanged": trend["guards_unchanged"],
}
quality_review = {
"state": "automation_execution_layer_p7_alert_decision_support_quality_review_closed",
"status": "completed",
"quality_status": "healthy",
"current_phase": "Restricted Decision-Support Soft Rollout",
"current_slice": run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_QUALITY_REVIEW_SLICE,
"next_slice": run_until_blocked.ANALYSIS_DECISION_MONITOR_P7_SELECTOR_SLICE,
"failure_reasons": [],
"monitor_blockers": [],
"monitor_warnings": [],
"latest_observed": observed,
"operator_mode": "observe_only",
"decision_action": "restricted_decision_support",
"trust_classification": "blacklist",
"guards_unchanged": trend["guards_unchanged"],
}
selector = run_until_blocked.build_analysis_decision_monitor_p7_slice_selector(
trend=trend,
quality_gate=quality_gate,
maturity_rollup=maturity_rollup,
remote_drift_check=remote_drift_check,
quality_review=quality_review,
)
self.assertEqual(selector["status"], "completed")
self.assertEqual(selector["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_WITH_MATURITY_SLICE)
self.assertEqual(selector["selection_reason"], "all_monitoring_artifacts_healthy")
def test_p7_selector_routes_to_remote_drift_repair_when_drift_detected(self) -> None:
trend = {
"record_count": 10,
"consecutive_pass_count": 10,
"stable_sample_threshold": 3,
"stable_monitoring_window": True,
"latest_monitor_passed": True,
"latest_blockers": [],
"guards_unchanged": ["execution_allowed"],
}
quality_gate = {
"status": "passed",
"passed": True,
"stable_monitoring_window": True,
"failure_reasons": [],
"guards_unchanged": ["execution_allowed"],
}
maturity_rollup = {
"state": "automation_execution_layer_p7_monitor_maturity_rollup_closed",
"status": "completed",
"maturity_status": "mature",
"local": {"quality_gate_status": "passed"},
"guards_unchanged": ["execution_allowed"],
}
remote_drift_check = {
"state": "automation_execution_layer_p7_remote_drift_check_blocked",
"status": "blocked",
"drift_status": "drift_detected",
"failure_reasons": ["local_remote_observed_fields_drifted"],
"guards_unchanged": ["execution_allowed"],
}
quality_review = {
"state": "automation_execution_layer_p7_alert_decision_support_quality_review_closed",
"status": "completed",
"quality_status": "healthy",
"failure_reasons": [],
"guards_unchanged": ["execution_allowed"],
}
selector = run_until_blocked.build_analysis_decision_monitor_p7_slice_selector(
trend=trend,
quality_gate=quality_gate,
maturity_rollup=maturity_rollup,
remote_drift_check=remote_drift_check,
quality_review=quality_review,
)
self.assertEqual(selector["next_slice"], run_until_blocked.ANALYSIS_DECISION_MONITOR_REMOTE_DRIFT_REPAIR_SLICE)
self.assertEqual(selector["selection_reason"], "remote_drift_detected")
def test_analysis_decision_monitor_phase_blocks_on_monitor_blockers(self) -> None:
monitor_payload = {
"passed": False,
"generated_ts": "2026-05-11T04:46:00+00:00",
"phase": "analysis_decision_only",
"blockers": ["market_data_fresh"],
"warnings": [],
"observed": {
"real_trade_sample_gate_required": False,
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
},
}
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text(
"# ai-exchange Session\n\n"
"- state: analysis_decision_monitor_runner_deployed\n"
"- repair_next_action: run_analysis_decision_monitor_periodically_without_fill_gate\n",
encoding="utf-8",
)
tasks_path.write_text(
"# ai-exchange Tasks\n\n"
"- [ ] `run_analysis_decision_monitor_periodically_without_fill_gate`\n",
encoding="utf-8",
)
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_monitor = run_until_blocked.run_analysis_decision_monitor.run_monitor
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.run_analysis_decision_monitor.run_monitor = lambda **kwargs: monitor_payload
summary = run_until_blocked.run_until_blocked(env={})
tasks_text = tasks_path.read_text(encoding="utf-8")
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.run_analysis_decision_monitor.run_monitor = original_monitor
self.assertEqual(summary["status"], "blocked")
self.assertEqual(summary["next_slice"], "repair_analysis_decision_monitor_blockers")
self.assertEqual(summary["blocked_reason"], "analysis_decision_monitor_blockers:market_data_fresh")
self.assertIn("analysis_decision_monitor_blockers:market_data_fresh", tasks_text)
self.assertEqual(status_payload["state"], "blocked")
self.assertEqual(status_payload["blocked_reason"], "analysis_decision_monitor_blockers:market_data_fresh")
self.assertEqual(summary["monitor_history"]["record_count"], 1)
def test_analysis_decision_monitor_trend_marks_stable_window_after_three_passes(self) -> None:
records = [
{
"recorded_ts": "2026-05-11T00:00:00+00:00",
"monitor_generated_ts": "2026-05-11T00:00:00+00:00",
"status": "completed",
"monitor_passed": True,
"blockers": [],
"warnings": [],
"observed": {"broker_status": "broker_state_ok"},
"next_slice": "continue_periodic_analysis_decision_monitoring",
"guards_unchanged": ["execution_allowed"],
},
{
"recorded_ts": "2026-05-11T00:05:00+00:00",
"monitor_generated_ts": "2026-05-11T00:05:00+00:00",
"status": "completed",
"monitor_passed": True,
"blockers": [],
"warnings": [],
"observed": {"broker_status": "broker_state_ok"},
"next_slice": "continue_periodic_analysis_decision_monitoring",
"guards_unchanged": ["execution_allowed"],
},
{
"recorded_ts": "2026-05-11T00:10:00+00:00",
"monitor_generated_ts": "2026-05-11T00:10:00+00:00",
"status": "completed",
"monitor_passed": True,
"blockers": [],
"warnings": [],
"observed": {"broker_status": "broker_state_ok"},
"next_slice": "continue_periodic_analysis_decision_monitoring",
"guards_unchanged": ["execution_allowed"],
},
]
trend = run_until_blocked.build_analysis_decision_monitor_trend(records=records)
self.assertEqual(trend["record_count"], 3)
self.assertEqual(trend["consecutive_pass_count"], 3)
self.assertTrue(trend["stable_monitoring_window"])
self.assertEqual(trend["latest_observed"]["broker_status"], "broker_state_ok")
def test_analysis_decision_monitor_quality_gate_passes_for_stable_observe_only_window(self) -> None:
trend = {
"record_count": 3,
"consecutive_pass_count": 3,
"stable_monitoring_window": True,
"latest_monitor_passed": True,
"latest_blockers": [],
"latest_warnings": [],
"latest_monitor_generated_ts": "2026-05-11T00:10:00+00:00",
"latest_observed": {
"decision_action": "restricted_decision_support",
"decision_operator_mode": "observe_only",
"real_trade_sample_gate_required": False,
"real_trade_sample_gate_status": "deferred_analysis_decision_only",
"runtime_health": "ok",
"market_gap": "market_data_gap_clear",
"p6_governance_tick_status": "no_op_healthy",
},
"next_slice": "continue_periodic_analysis_decision_monitoring",
"guards_unchanged": [
"execution_allowed",
"real_live_ready",
"risk_governor",
"trust_bucket",
"operator_confirmation_protocol",
],
}
gate = run_until_blocked.build_analysis_decision_monitor_quality_gate(trend=trend)
self.assertTrue(gate["passed"])
self.assertEqual(gate["status"], "passed")
self.assertEqual(gate["failure_reasons"], [])
self.assertIn("execution_gate_unchanged", gate["notes"])
def test_advances_to_next_stage_and_updates_anchors(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n- [x] `P0-1` 实时市场真值层\n", encoding="utf-8")
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_runner = run_until_blocked.run_trading_delivery_rehearsal
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
"mode": "rehearsal_completed",
"passed": True,
"continuation_decision": "advance",
"promotion_decision": "promote_candidate",
"blockers": [],
}
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_MAX_ROUNDS": "1"})
session_text = session_path.read_text(encoding="utf-8")
tasks_text = tasks_path.read_text(encoding="utf-8")
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.run_trading_delivery_rehearsal = original_runner
self.assertEqual(summary["status"], "stopped")
self.assertEqual(summary["completed_stages"], ["P0-1", "P0-2"])
self.assertEqual(summary["current_stage"], "P0-3")
self.assertIn("### `P0-3` 盘中持续调度与运行监督", session_text)
self.assertIn("- [x] `P0-2` 真实账户与订单只读同步", tasks_text)
self.assertIn("- [ ] `P0-3` 盘中持续调度与运行监督", tasks_text)
def test_stops_when_rehearsal_reports_blocked(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n", encoding="utf-8")
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_runner = run_until_blocked.run_trading_delivery_rehearsal
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
"mode": "autonomous_cycle_failed",
"passed": False,
"continuation_decision": "blocked",
"promotion_decision": "hold",
"blockers": ["缺凭据:broker_readonly_api_key"],
}
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_MAX_ROUNDS": "2"})
tasks_text = tasks_path.read_text(encoding="utf-8")
session_text = session_path.read_text(encoding="utf-8")
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.run_trading_delivery_rehearsal = original_runner
self.assertEqual(summary["status"], "blocked")
self.assertEqual(summary["blocked_reason"], "缺凭据:broker_readonly_api_key")
self.assertEqual(summary["current_stage"], "P0-2")
self.assertIn("缺凭据:broker_readonly_api_key", tasks_text)
self.assertIn("当前自动化已阻塞:缺凭据:broker_readonly_api_key", session_text)
def test_json_output_is_machine_readable(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n", encoding="utf-8")
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_runner = run_until_blocked.run_trading_delivery_rehearsal
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
"mode": "autonomous_cycle_failed",
"passed": False,
"continuation_decision": "blocked",
"promotion_decision": "hold",
"blockers": ["缺凭据:broker_readonly_api_key"],
}
output = io.StringIO()
with contextlib.redirect_stdout(output):
summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "1"})
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.run_trading_delivery_rehearsal = original_runner
payload = json.loads(output.getvalue())
self.assertEqual(payload["status"], "blocked")
self.assertEqual(payload["blocked_reason"], "缺凭据:broker_readonly_api_key")
self.assertEqual(summary["status"], "blocked")
def test_run_until_blocked_writes_runner_status_file(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
session_path = Path(tmp) / "SESSION.md"
tasks_path = Path(tmp) / "TASKS.md"
status_path = Path(tmp) / "automation_status.json"
session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n", encoding="utf-8")
original_session = run_until_blocked.SESSION_PATH
original_tasks = run_until_blocked.TASKS_PATH
original_status = run_until_blocked.RUNNER_STATUS_PATH
original_runner = run_until_blocked.run_trading_delivery_rehearsal
try:
run_until_blocked.SESSION_PATH = session_path
run_until_blocked.TASKS_PATH = tasks_path
run_until_blocked.RUNNER_STATUS_PATH = status_path
run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
"mode": "autonomous_cycle_failed",
"passed": False,
"continuation_decision": "blocked",
"promotion_decision": "hold",
"blockers": ["缺凭据:broker_readonly_api_key"],
}
run_until_blocked.run_until_blocked(env={})
payload = json.loads(status_path.read_text(encoding="utf-8"))
finally:
run_until_blocked.SESSION_PATH = original_session
run_until_blocked.TASKS_PATH = original_tasks
run_until_blocked.RUNNER_STATUS_PATH = original_status
run_until_blocked.run_trading_delivery_rehearsal = original_runner
self.assertEqual(payload["state"], "blocked")
self.assertEqual(payload["current_stage"], "P0-2")
self.assertEqual(payload["blocked_reason"], "缺凭据:broker_readonly_api_key")
def test_supervisor_pauses_on_code_fix_blocker(self) -> None:
sleeper = Mock()
original_run_until_blocked = run_until_blocked.run_until_blocked
original_fingerprint = run_until_blocked.compute_resume_fingerprint
original_status = run_until_blocked.RUNNER_STATUS_PATH
try:
with tempfile.TemporaryDirectory() as tmp:
run_until_blocked.RUNNER_STATUS_PATH = Path(tmp) / "automation_status.json"
run_until_blocked.run_until_blocked = lambda **kwargs: {
"status": "blocked",
"started_ts": "2026-05-07T00:00:00+00:00",
"current_stage": "P0-2",
"completed_stages": ["P0-1"],
"blocked_reason": "autonomous_mode:pipeline_failed",
"rounds": [{"round": 1}],
}
run_until_blocked.compute_resume_fingerprint = lambda: "same"
result = run_until_blocked.run_supervised_until_stopped(
env={"RUN_UNTIL_BLOCKED_LOOP_SLEEP_SECONDS": "1", "RUN_UNTIL_BLOCKED_SUPERVISOR_CYCLES": "2"},
sleeper=sleeper,
)
finally:
run_until_blocked.run_until_blocked = original_run_until_blocked
run_until_blocked.compute_resume_fingerprint = original_fingerprint
run_until_blocked.RUNNER_STATUS_PATH = original_status
sleeper.assert_called_once_with(1)
self.assertEqual(result["status"], "blocked")
def test_supervisor_resumes_after_fingerprint_change(self) -> None:
summaries = [
{
"status": "blocked",
"started_ts": "2026-05-07T00:00:00+00:00",
"current_stage": "P0-2",
"completed_stages": ["P0-1"],
"blocked_reason": "autonomous_mode:pipeline_failed",
"rounds": [{"round": 1}],
},
{
"status": "completed",
"started_ts": "2026-05-07T00:10:00+00:00",
"current_stage": None,
"completed_stages": ["P0-1", "P0-2"],
"blocked_reason": None,
"rounds": [{"round": 1}],
},
]
fingerprints = ["a", "a", "b", "b"]
sleeper = Mock()
original_run_until_blocked = run_until_blocked.run_until_blocked
original_fingerprint = run_until_blocked.compute_resume_fingerprint
original_status = run_until_blocked.RUNNER_STATUS_PATH
try:
with tempfile.TemporaryDirectory() as tmp:
run_until_blocked.RUNNER_STATUS_PATH = Path(tmp) / "automation_status.json"
run_until_blocked.run_until_blocked = lambda **kwargs: summaries.pop(0)
run_until_blocked.compute_resume_fingerprint = lambda: fingerprints.pop(0)
result = run_until_blocked.run_supervised_until_stopped(
env={"RUN_UNTIL_BLOCKED_LOOP_SLEEP_SECONDS": "1", "RUN_UNTIL_BLOCKED_SUPERVISOR_CYCLES": "2"},
sleeper=sleeper,
)
finally:
run_until_blocked.run_until_blocked = original_run_until_blocked
run_until_blocked.compute_resume_fingerprint = original_fingerprint
run_until_blocked.RUNNER_STATUS_PATH = original_status
sleeper.assert_called_once_with(1)
self.assertEqual(result["status"], "completed")
self.assertEqual(result["completed_stages"], ["P0-1", "P0-2"])
def test_main_uses_supervisor_mode_when_enabled(self) -> None:
original_run_until_blocked = run_until_blocked.run_until_blocked
original_supervisor = run_until_blocked.run_supervised_until_stopped
original_environ = dict(os.environ)
try:
os.environ["RUN_UNTIL_BLOCKED_SUPERVISOR"] = "1"
run_until_blocked.run_until_blocked = lambda **kwargs: (_ for _ in ()).throw(AssertionError("should not call one-shot runner"))
run_until_blocked.run_supervised_until_stopped = lambda **kwargs: {"status": "completed"}
exit_code = run_until_blocked.main()
finally:
os.environ.clear()
os.environ.update(original_environ)
run_until_blocked.run_until_blocked = original_run_until_blocked
run_until_blocked.run_supervised_until_stopped = original_supervisor
self.assertEqual(exit_code, 0)
if __name__ == "__main__":
unittest.main()