fix(awooop): advance parallel mainline during p0 wait

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -12,7 +12,7 @@ import re
 import urllib.error
 import urllib.parse
 import urllib.request
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
 from zoneinfo import ZoneInfo
@@ -47,6 +47,21 @@ _METRIC_LINE_RE = re.compile(
 _COMMANDER_INSERTED_REQUIREMENT_SOURCE = (
     "docs/workplans/2026-07-02-commander-inserted-requirements-priority-ledger.md"
 )
+_STOCKPLATFORM_DATA_RETRY_READBACK_TIMES = [
+    "19:12",
+    "20:12",
+    "21:12",
+    "22:12",
+    "22:47",
+    "23:31",
+]
+_PARALLEL_MAINLINE_FORBIDDEN_ACTIONS = [
+    "fake_stockplatform_freshness_marker",
+    "manual_zero_fill_for_margin_short",
+    "copy_previous_trading_day_data",
+    "db_write_or_restore_from_this_wait_lane",
+    "docker_or_host_restart_for_source_pending",
+]
 _AI_AUTOMATION_NODE_RECEIPT_SCHEMA_VERSION = "ai_automation_node_receipt_v1"
 _AI_AUTOMATION_NODE_RECEIPT_REQUIRED_FIELDS = [
     "trace_id",
@@ -3463,6 +3478,149 @@ def _refresh_rollups_after_stockplatform_overlay(
     summary["next_executable_mainline_state"] = str(
         state.get("next_executable_mainline_state") or ""
     )
+    _apply_parallel_mainline_execution_overlay(payload)
+
+
+def _next_stockplatform_data_retry_readback_at(
+    now: datetime | None = None,
+) -> str:
+    current = now or datetime.now(ZoneInfo("Asia/Taipei"))
+    if current.tzinfo is None:
+        current = current.replace(tzinfo=ZoneInfo("Asia/Taipei"))
+    for retry_time in _STOCKPLATFORM_DATA_RETRY_READBACK_TIMES:
+        hour, minute = retry_time.split(":", maxsplit=1)
+        candidate = current.replace(
+            hour=int(hour),
+            minute=int(minute),
+            second=0,
+            microsecond=0,
+        )
+        if candidate > current:
+            return candidate.isoformat()
+    next_day = current.replace(
+        hour=19,
+        minute=12,
+        second=0,
+        microsecond=0,
+    )
+    next_day = next_day.replace(day=current.day) + timedelta(days=1)
+    return next_day.isoformat()
+
+
+def _stockplatform_external_data_wait_active(state: dict[str, Any]) -> bool:
+    data_dependency = str(
+        state.get("stockplatform_public_api_controlled_recovery_data_dependency")
+        or state.get("stockplatform_public_api_data_dependency_classification")
+        or ""
+    )
+    blockers = _strings(state.get("stockplatform_freshness_blockers")) + _strings(
+        state.get("stockplatform_ingestion_blockers")
+    )
+    return (
+        state.get("stockplatform_source_contract_readback_present") is True
+        and state.get("stockplatform_source_contract_ready") is not True
+        and data_dependency == "freshness_or_ingestion_not_ready"
+        and any(
+            blocker
+            in {
+                "core_margin_short_daily_missing",
+                "ai_recommendations_stale",
+                "core.margin_short_daily_incomplete",
+            }
+            for blocker in blockers
+        )
+    )
+
+
+def _apply_parallel_mainline_execution_overlay(payload: dict[str, Any]) -> None:
+    state = _dict(payload.setdefault("mainline_execution_state", {}))
+    items = _list(payload.get("commander_inserted_requirement_work_items"))
+    if not items:
+        return
+    next_item = next(
+        (
+            _dict(item)
+            for item in items
+            if item.get("status") in {"in_progress", "pending"}
+            and str(item.get("priority") or "") == "P0"
+            and str(item.get("id") or "") != "CIR-P0-006"
+        ),
+        _dict(items[0]),
+    )
+    external_wait_active = _stockplatform_external_data_wait_active(state)
+    retry_readback_at = (
+        _next_stockplatform_data_retry_readback_at()
+        if external_wait_active
+        else ""
+    )
+    overlay = {
+        "enabled": external_wait_active,
+        "policy": str(
+            state.get("next_executable_mainline_policy")
+            or "when_p0_has_external_gate_do_not_idle"
+        ),
+        "current_p0_workplan_id": str(state.get("active_p0_workplan_id") or ""),
+        "current_p0_state": str(state.get("active_p0_state") or ""),
+        "current_p0_wait_state": (
+            "external_stockplatform_data_retry_wait"
+            if external_wait_active
+            else "active_p0_not_external_wait"
+        ),
+        "retry_readback_at": retry_readback_at,
+        "parallel_work_item": {
+            "id": str(next_item.get("id") or ""),
+            "priority": str(next_item.get("priority") or ""),
+            "lane": str(next_item.get("lane") or ""),
+            "status": str(next_item.get("status") or ""),
+            "mapped_workplan_id": str(next_item.get("mapped_workplan_id") or ""),
+            "next_action": str(next_item.get("next_action") or ""),
+        },
+        "report_contract": {
+            "current_p0": str(state.get("active_p0_workplan_id") or ""),
+            "action": "advance_parallel_mainline_work_item_while_external_data_retry_waits",
+            "evidence": "priority_readback.parallel_mainline_execution",
+            "next_action": str(next_item.get("next_action") or ""),
+        },
+        "forbidden_actions": _PARALLEL_MAINLINE_FORBIDDEN_ACTIONS,
+    }
+    payload["mainline_parallel_execution"] = overlay
+
+    state["parallel_mainline_execution_enabled"] = external_wait_active
+    state["parallel_mainline_current_p0_wait_state"] = overlay[
+        "current_p0_wait_state"
+    ]
+    state["parallel_mainline_retry_readback_at"] = retry_readback_at
+    state["parallel_mainline_work_item_id"] = overlay["parallel_work_item"]["id"]
+    state["parallel_mainline_work_item_priority"] = overlay["parallel_work_item"][
+        "priority"
+    ]
+    state["parallel_mainline_work_item_lane"] = overlay["parallel_work_item"]["lane"]
+    state["parallel_mainline_next_action"] = overlay["parallel_work_item"][
+        "next_action"
+    ]
+    state["parallel_mainline_forbidden_actions"] = _PARALLEL_MAINLINE_FORBIDDEN_ACTIONS
+
+    rollups = _dict(payload.setdefault("rollups", {}))
+    rollups["parallel_mainline_execution_enabled"] = external_wait_active
+    rollups["parallel_mainline_work_item_id"] = overlay["parallel_work_item"]["id"]
+    rollups["parallel_mainline_retry_readback_at"] = retry_readback_at
+
+    summary = _dict(payload.setdefault("summary", {}))
+    summary["parallel_mainline_execution_enabled"] = external_wait_active
+    summary["parallel_mainline_current_p0_wait_state"] = overlay[
+        "current_p0_wait_state"
+    ]
+    summary["parallel_mainline_retry_readback_at"] = retry_readback_at
+    summary["parallel_mainline_work_item_id"] = overlay["parallel_work_item"]["id"]
+    summary["parallel_mainline_work_item_priority"] = overlay["parallel_work_item"][
+        "priority"
+    ]
+    summary["parallel_mainline_work_item_lane"] = overlay["parallel_work_item"][
+        "lane"
+    ]
+    summary["parallel_mainline_next_action"] = overlay["parallel_work_item"][
+        "next_action"
+    ]
 
 
 def _set_rollups_and_summary(
@@ -3727,6 +3885,7 @@ def _apply_commander_inserted_requirement_work_items(
     summary["commander_inserted_requirement_source"] = (
         _COMMANDER_INSERTED_REQUIREMENT_SOURCE
     )
+    _apply_parallel_mainline_execution_overlay(payload)
 
 
 def apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None: