From ca39a698d5c86adefe648b4d9475921e3a7e992b Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 2 Jul 2026 20:37:07 +0800 Subject: [PATCH] fix(awooop): advance parallel mainline during p0 wait --- .../awoooi_priority_work_order_readback.py | 161 +++++++++++++++++- ...awoooi_priority_work_order_readback_api.py | 97 +++++++++++ apps/web/messages/en.json | 8 + apps/web/messages/zh-TW.json | 8 + .../app/[locale]/awooop/work-items/page.tsx | 29 +++- 5 files changed, 300 insertions(+), 3 deletions(-) diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py index 6537d67da..889e18f0f 100644 --- a/apps/api/src/services/awoooi_priority_work_order_readback.py +++ b/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: diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py index 9dbace3c1..2775b9efa 100644 --- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py +++ b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py @@ -1368,6 +1368,37 @@ def test_awoooi_priority_work_order_readback_routes_stockplatform_data_dependenc ][0] +def test_awoooi_priority_work_order_readback_allows_parallel_mainline_on_stockplatform_external_wait(): + payload = load_latest_awoooi_priority_work_order_readback() + runtime = _stockplatform_runtime_source_contract_blocked() + preflight = load_latest_stockplatform_public_api_controlled_recovery_preflight( + runtime_readback=runtime + ) + + apply_stockplatform_public_api_runtime_readback(payload, runtime) + apply_stockplatform_public_api_controlled_recovery_preflight(payload, preflight) + + parallel = payload["mainline_parallel_execution"] + state = payload["mainline_execution_state"] + assert parallel["enabled"] is True + assert parallel["current_p0_wait_state"] == ( + "external_stockplatform_data_retry_wait" + ) + assert parallel["retry_readback_at"] + assert parallel["parallel_work_item"]["id"] == "CIR-P0-001" + assert parallel["parallel_work_item"]["priority"] == "P0" + assert parallel["report_contract"]["current_p0"] == "P0-006" + assert "fake_stockplatform_freshness_marker" in parallel["forbidden_actions"] + assert "db_write_or_restore_from_this_wait_lane" in parallel["forbidden_actions"] + assert state["parallel_mainline_execution_enabled"] is True + assert state["parallel_mainline_work_item_id"] == "CIR-P0-001" + assert payload["summary"]["parallel_mainline_execution_enabled"] is True + assert payload["summary"]["parallel_mainline_work_item_id"] == "CIR-P0-001" + assert payload["summary"]["parallel_mainline_work_item_priority"] == "P0" + assert payload["rollups"]["parallel_mainline_execution_enabled"] is True + assert payload["rollups"]["parallel_mainline_work_item_id"] == "CIR-P0-001" + + def test_awoooi_priority_work_order_readback_normalizes_runtime_source_truth( monkeypatch: pytest.MonkeyPatch, ): @@ -1780,6 +1811,72 @@ def _stockplatform_runtime_data_dependency_blocked() -> dict: } +def _stockplatform_runtime_source_contract_blocked() -> dict: + payload = _stockplatform_runtime_data_dependency_blocked() + payload["safe_next_step"] = ( + "run_stockplatform_data_dependency_source_contract_readback_then_" + "rerun_public_api_verifier" + ) + payload["active_blockers"] = [ + "stockplatform_freshness_core_margin_short_daily_missing", + "stockplatform_freshness_ai_recommendations_stale", + "stockplatform_ingestion_core.margin_short_daily_incomplete", + "stockplatform_post_apply_public_api_verifier_not_green", + ] + payload["readback"]["data_dependency_classification"] = ( + "freshness_or_ingestion_not_ready" + ) + payload["readback"]["postgres_not_ready"] = False + payload["rollups"]["data_dependency_classification"] = ( + "freshness_or_ingestion_not_ready" + ) + payload["rollups"]["postgres_not_ready"] = False + payload["data_readiness_control_readback"] = { + "schema_version": "stockplatform_postgres_readiness_or_data_source_readback_v1", + "receipt_key": "stockplatform_postgres_readiness_or_data_source_readback", + "status": "blocked_stockplatform_data_dependency_not_ready", + "readback_present": True, + "receipt_provided": True, + "postgres_ready": False, + "postgres_not_ready": False, + } + payload["source_contract_control_readback"] = { + "schema_version": "stockplatform_source_contract_control_readback_v1", + "receipt_key": "stockplatform_source_contract_readback", + "status": "blocked_stockplatform_source_contract_not_ready", + "readback_present": True, + "receipt_provided": True, + "source_contract_ready": False, + "non_ok_source_count": 3, + "active_blockers": [ + "stockplatform_freshness_core_margin_short_daily_missing", + "stockplatform_freshness_ai_recommendations_stale", + "stockplatform_ingestion_core.margin_short_daily_incomplete", + ], + "blocking_sources": [ + { + "source": "raw.source_quality_observations.unit_sanity", + "category": "freshness", + "status": "warning", + "classification": "freshness_source_non_ok", + }, + { + "source": "core.margin_short_daily", + "category": "freshness", + "status": "missing", + "classification": "freshness_source_non_ok", + }, + { + "source": "ai.recommendations", + "category": "freshness", + "status": "stale", + "classification": "freshness_source_non_ok", + }, + ], + } + return payload + + def _harbor_registry_ready() -> dict: return { "schema_version": "harbor_registry_controlled_recovery_preflight_v1", diff --git a/apps/web/messages/en.json b/apps/web/messages/en.json index 73f59ecb5..e34805f3c 100644 --- a/apps/web/messages/en.json +++ b/apps/web/messages/en.json @@ -9188,6 +9188,7 @@ "head": "Deployed source", "decision": "Decision", "decisionValue": "Fix P0, no side lane", + "decisionParallel": "P0 waits for data; advance {id}", "blockingSources": "Blocking sources", "noSourceNames": "Waiting for source contract readback", "states": { @@ -9217,6 +9218,13 @@ "label": "Next step", "value": "Repair source contract", "detail": "Run verifier immediately after repair." + }, + "parallel": { + "label": "Parallel mainline", + "value": "Advance {id}", + "detail": "{action}", + "idle": "Waiting for P0 readback", + "wait": "Next readback {time}" } }, "chips": { diff --git a/apps/web/messages/zh-TW.json b/apps/web/messages/zh-TW.json index 8ac074f47..362d8ef45 100644 --- a/apps/web/messages/zh-TW.json +++ b/apps/web/messages/zh-TW.json @@ -9188,6 +9188,7 @@ "head": "正式部署", "decision": "決策", "decisionValue": "先修 P0，不開支線", + "decisionParallel": "P0 等資料；並行推進 {id}", "blockingSources": "阻塞來源", "noSourceNames": "等待 source contract readback", "states": { @@ -9217,6 +9218,13 @@ "label": "下一步", "value": "補來源合約", "detail": "修完直接跑 verifier" + }, + "parallel": { + "label": "並行主線", + "value": "推進 {id}", + "detail": "{action}", + "idle": "等待 P0 readback", + "wait": "下一次讀回 {time}" } }, "chips": { diff --git a/apps/web/src/app/[locale]/awooop/work-items/page.tsx b/apps/web/src/app/[locale]/awooop/work-items/page.tsx index f030d9498..f5e5dadcd 100644 --- a/apps/web/src/app/[locale]/awooop/work-items/page.tsx +++ b/apps/web/src/app/[locale]/awooop/work-items/page.tsx @@ -1076,6 +1076,13 @@ type PriorityWorkOrderResponse = { active_p0_state?: string | null; next_executable_mainline_workplan_id?: string | null; next_executable_mainline_state?: string | null; + parallel_mainline_execution_enabled?: boolean | null; + parallel_mainline_current_p0_wait_state?: string | null; + parallel_mainline_retry_readback_at?: string | null; + parallel_mainline_work_item_id?: string | null; + parallel_mainline_work_item_priority?: string | null; + parallel_mainline_work_item_lane?: string | null; + parallel_mainline_next_action?: string | null; commander_inserted_requirement_work_item_count?: number | null; commander_inserted_requirement_p0_count?: number | null; commander_inserted_requirement_p1_count?: number | null; @@ -8783,6 +8790,10 @@ function ManagerSituationBoard({ const activeWorkplan = summary?.next_executable_mainline_workplan_id ?? "P0-006-STOCKPLATFORM-PUBLIC-API-CONTROLLED-RECOVERY-PREFLIGHT"; + const parallelEnabled = summary?.parallel_mainline_execution_enabled === true; + const parallelWorkItem = summary?.parallel_mainline_work_item_id ?? "--"; + const parallelNextAction = summary?.parallel_mainline_next_action ?? "--"; + const parallelRetryAt = summary?.parallel_mainline_retry_readback_at ?? "--"; const activeState = summary?.active_p0_state ?? "blocked_stockplatform_public_api_controlled_recovery_preflight"; @@ -8846,6 +8857,18 @@ function ManagerSituationBoard({ detail: t("cards.next.detail"), tone: "border-[#cbd7bf] bg-[#f4faef] text-[#3d6b24]", }, + { + key: "parallel", + icon: GitBranch, + label: t("cards.parallel.label"), + value: parallelEnabled + ? t("cards.parallel.value", { id: parallelWorkItem }) + : t("cards.parallel.idle"), + detail: parallelEnabled + ? t("cards.parallel.detail", { action: parallelNextAction }) + : t("cards.parallel.wait", { time: parallelRetryAt }), + tone: "border-[#dbcdf0] bg-[#f7f3ff] text-[#62439b]", + }, ]; return ( @@ -8879,14 +8902,16 @@ function ManagerSituationBoard({ {t("decision")}
- {t("decisionValue")} + {parallelEnabled + ? t("decisionParallel", { id: parallelWorkItem }) + : t("decisionValue")}
-
+
{managerCards.map((card) => { const Icon = card.icon; return (