fix: replace shared IBKR precheck dispatcher

.github/workflows/sync-cloud-run-env.yml

@@ -422,6 +422,17 @@ jobs:
           project_id: ${{ env.GCP_PROJECT_ID }}
           version: ">= 416.0.0"
 
+      - name: Require configured scheduler migration before HK verify
+        if: steps.config.outputs.env_sync_enabled == 'true' && env.WORKFLOW_TARGET == 'hk-verify'
+        run: |
+          set -euo pipefail
+          scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"
+          if gcloud scheduler jobs describe "interactive-brokers-monitor-dispatcher-scheduler" \
+            --project="${GCP_PROJECT_ID}" \
+            --location="${scheduler_location}" >/dev/null 2>&1; then
+            echo "HK verify scheduler migration requires the configured migration first." >&2
+            exit 1
+          fi
 
       - name: Build, push, and deploy Cloud Run image
         run: |
@@ -889,44 +900,6 @@ jobs:
             shared_remove_env_vars+=("STRATEGY_PLUGIN_ALERT_TELEGRAM_BODY_MAX_CHARS")
           fi
 
-          monitor_dispatch_targets_file="$(mktemp)"
-          for target_index in "${!target_services[@]}"; do
-            cloud_run_service="${target_services[${target_index}]}"
-            service_url="$(gcloud run services describe "${cloud_run_service}" \
-              --project="${GCP_PROJECT_ID}" \
-              --region="${CLOUD_RUN_REGION}" \
-              --format='value(status.url)' 2>/dev/null || true)"
-            if [ -z "${service_url}" ]; then
-              echo "Unable to resolve Cloud Run service URL for ${cloud_run_service}; cannot build monitor dispatcher targets." >&2
-              exit 1
-            fi
-            SYNC_TARGET_INDEX="${target_index}" SERVICE_URL="${service_url}" python - <<'PY' >> "${monitor_dispatch_targets_file}"
-          import json
-          import os
-
-          plan = json.loads(os.environ["SYNC_PLAN_JSON"])
-          target = plan["targets"][int(os.environ["SYNC_TARGET_INDEX"])]
-          env = target.get("env") or {}
-          scheduler = target.get("scheduler") or {}
-          payload = {
-              "service_name": target["service_name"],
-              "service_url": os.environ["SERVICE_URL"].rstrip("/"),
-              "strategy_profile": target.get("strategy_profile") or env.get("STRATEGY_PROFILE"),
-              "account_group": env.get("ACCOUNT_GROUP"),
-              "runtime_target_enabled": env.get("RUNTIME_TARGET_ENABLED", "true"),
-              # Dedicated Cloud Scheduler jobs already own /health warmups and
-              # /run execution. The shared dispatcher only owns prechecks.
-              "scheduler": {
-                  "timezone": scheduler.get("timezone"),
-                  "precheck_time": scheduler.get("precheck_time"),
-              },
-          }
-          print(json.dumps(payload, separators=(",", ":"), sort_keys=True))
-          PY
-          done
-          monitor_dispatch_targets_json="$(jq -sc '{targets:.}' "${monitor_dispatch_targets_file}")"
-          shared_env_pairs+=("IBKR_MONITOR_DISPATCH_TARGETS_JSON=${monitor_dispatch_targets_json}")
-
           for target_index in "${!target_services[@]}"; do
             cloud_run_service="${target_services[${target_index}]}"
             mapfile -t target_env_pairs < <(emit_target_env_pairs "${target_index}")
@@ -1022,6 +995,8 @@ jobs:
                           timezone,
                           str(scheduler.get("main_time") or configured_time("CLOUD_SCHEDULER_MAIN_TIME", "45 15")),
                           str(scheduler.get("probe_time") or configured_time("CLOUD_SCHEDULER_PROBE_TIME", "35 9,15")),
+                          str(scheduler.get("precheck_time") or configured_time("CLOUD_SCHEDULER_PRECHECK_TIME", "45 9")),
+                          str(env.get("RUNTIME_TARGET_ENABLED", "true")).strip().lower() or "true",
                           str(scheduler.get("attempt_deadline") or ""),
                       ]
                   )
@@ -1030,7 +1005,7 @@ jobs:
           )
 
           for update in "${scheduler_updates[@]}"; do
-            IFS=$'\t' read -r cloud_run_service market_timezone main_time warmup_time main_attempt_deadline <<< "${update}"
+            IFS=$'\t' read -r cloud_run_service market_timezone main_time warmup_time precheck_time runtime_target_enabled main_attempt_deadline <<< "${update}"
             if [ -z "${cloud_run_service}" ] || [ -z "${market_timezone}" ]; then
               continue
             fi
@@ -1103,6 +1078,24 @@ jobs:
             desired_warmup_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${warmup_time}" python - <<'PY'
           import os
 
+          current_fields = os.environ["CURRENT_SCHEDULE"].split()
+          time_fields = os.environ["SCHEDULE_TIME"].split()
+          if len(current_fields) != 5:
+              raise SystemExit(f"Cloud Scheduler schedule must have 5 fields: {os.environ['CURRENT_SCHEDULE']!r}")
+          if len(time_fields) == 5:
+              print(" ".join(time_fields))
+          elif len(time_fields) == 2:
+              print(" ".join([*time_fields, *current_fields[2:]]))
+          else:
+              raise SystemExit(
+                  f"Cloud Scheduler override must have 2 time fields or 5 cron fields: {os.environ['SCHEDULE_TIME']!r}"
+              )
+          PY
+            )"
+
+            desired_precheck_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${precheck_time}" python - <<'PY'
+          import os
+
           current_fields = os.environ["CURRENT_SCHEDULE"].split()
           time_fields = os.environ["SCHEDULE_TIME"].split()
           if len(current_fields) != 5:
@@ -1187,56 +1180,90 @@ jobs:
                 --max-doublings=2 \
                 --quiet
             fi
-          done
 
-          monitor_dispatch_service="$(printf '%s\n' "${scheduler_updates[@]}" | head -n 1 | cut -f1)"
-          if [ -z "${monitor_dispatch_service}" ]; then
-            echo "Unable to resolve monitor dispatch service." >&2
-            exit 1
-          fi
-          monitor_dispatch_base_url="$(gcloud run services describe "${monitor_dispatch_service}" \
-            --project="${GCP_PROJECT_ID}" \
-            --region="${CLOUD_RUN_REGION}" \
-            --format='value(status.url)' 2>/dev/null || true)"
-          if [ -z "${monitor_dispatch_base_url}" ]; then
-            echo "Unable to resolve monitor dispatch URL for ${monitor_dispatch_service}." >&2
-            exit 1
-          fi
-          monitor_job_name="interactive-brokers-monitor-dispatcher-scheduler"
-          monitor_uri="${monitor_dispatch_base_url}/monitor-dispatch"
-          if gcloud scheduler jobs describe "${monitor_job_name}" \
-            --project="${GCP_PROJECT_ID}" \
-            --location="${scheduler_location}" >/dev/null 2>&1; then
-            echo "Updating Cloud Scheduler job ${monitor_job_name} to ${monitor_uri}."
-            gcloud scheduler jobs update http "${monitor_job_name}" \
+            precheck_job_name="${cloud_run_service%-service}-precheck-scheduler"
+            precheck_uri="${service_url}/dry-run"
+            precheck_state="$(gcloud scheduler jobs describe "${precheck_job_name}" \
               --project="${GCP_PROJECT_ID}" \
               --location="${scheduler_location}" \
-              --uri="${monitor_uri}" \
-              --schedule="*/5 * * * *" \
-              --time-zone="UTC" \
-              --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
-              --oidc-token-audience="${monitor_dispatch_base_url}" \
-              --quiet
-          else
-            echo "Creating Cloud Scheduler job ${monitor_job_name} at ${monitor_uri}."
-            gcloud scheduler jobs create http "${monitor_job_name}" \
-              --project="${GCP_PROJECT_ID}" \
-              --location="${scheduler_location}" \
-              --uri="${monitor_uri}" \
-              --schedule="*/5 * * * *" \
-              --time-zone="UTC" \
-              --http-method=POST \
-              --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
-              --oidc-token-audience="${monitor_dispatch_base_url}" \
-              --quiet
-          fi
+              --format='value(state)' 2>/dev/null || true)"
+            if [ -n "${precheck_state}" ]; then
+              echo "Updating Cloud Scheduler precheck ${precheck_job_name} to ${desired_precheck_schedule}."
+              gcloud scheduler jobs update http "${precheck_job_name}" \
+                --project="${GCP_PROJECT_ID}" \
+                --location="${scheduler_location}" \
+                --uri="${precheck_uri}" \
+                --schedule="${desired_precheck_schedule}" \
+                --time-zone="${market_timezone}" \
+                --http-method=POST \
+                --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
+                --oidc-token-audience="${service_url}" \
+                --attempt-deadline=120s \
+                --max-retry-attempts=0 \
+                --max-retry-duration=0s \
+                --quiet
+            else
+              echo "Creating Cloud Scheduler precheck ${precheck_job_name} at ${desired_precheck_schedule}."
+              gcloud scheduler jobs create http "${precheck_job_name}" \
+                --project="${GCP_PROJECT_ID}" \
+                --location="${scheduler_location}" \
+                --uri="${precheck_uri}" \
+                --schedule="${desired_precheck_schedule}" \
+                --time-zone="${market_timezone}" \
+                --http-method=POST \
+                --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
+                --oidc-token-audience="${service_url}" \
+                --attempt-deadline=120s \
+                --max-retry-attempts=0 \
+                --max-retry-duration=0s \
+                --quiet
+            fi
+            case "${runtime_target_enabled}" in
+              1|true|yes|on)
+                if [ "${precheck_state}" = "PAUSED" ]; then
+                  gcloud scheduler jobs resume "${precheck_job_name}" \
+                    --project="${GCP_PROJECT_ID}" \
+                    --location="${scheduler_location}" \
+                    --quiet
+                fi
+                ;;
+              *)
+                if [ "${precheck_state}" != "PAUSED" ]; then
+                  echo "Pausing Cloud Scheduler precheck ${precheck_job_name} because ${cloud_run_service} is disabled."
+                  gcloud scheduler jobs pause "${precheck_job_name}" \
+                    --project="${GCP_PROJECT_ID}" \
+                    --location="${scheduler_location}" \
+                    --quiet
+                fi
+                ;;
+            esac
+          done
 
-          python3 scripts/reconcile_cloud_runtime.py \
-            --platform=ibkr \
-            --project="${GCP_PROJECT_ID}" \
-            --region="${CLOUD_RUN_REGION}" \
-            --scheduler-location="${scheduler_location}" \
+          reconcile_args=(
+            --platform=ibkr
+            --project="${GCP_PROJECT_ID}"
+            --region="${CLOUD_RUN_REGION}"
+            --scheduler-location="${scheduler_location}"
             --delete-legacy-schedulers
+          )
+          python3 scripts/reconcile_cloud_runtime.py "${reconcile_args[@]}"
+
+          for update in "${scheduler_updates[@]}"; do
+            IFS=$'\t' read -r cloud_run_service _ <<< "${update}"
+            legacy_env_names="$(gcloud run services describe "${cloud_run_service}" \
+              --project="${GCP_PROJECT_ID}" \
+              --region="${CLOUD_RUN_REGION}" \
+              --format='value(spec.template.spec.containers[0].env[].name)')"
+            if ! printf '%s\n' "${legacy_env_names}" | tr ';' '\n' | grep -Fxq 'IBKR_MONITOR_DISPATCH_TARGETS_JSON'; then
+              continue
+            fi
+            echo "Removing retired monitor dispatcher configuration from ${cloud_run_service}."
+            gcloud run services update "${cloud_run_service}" \
+              --project="${GCP_PROJECT_ID}" \
+              --region="${CLOUD_RUN_REGION}" \
+              --remove-env-vars=IBKR_MONITOR_DISPATCH_TARGETS_JSON \
+              --quiet
+          done
 
       - name: Prune old Cloud Run revisions
         if: steps.config.outputs.enabled == 'true'

scripts/reconcile_cloud_runtime.py

@@ -291,6 +291,9 @@ def _legacy_jobs_for_target(platform: str, target: RuntimeTarget) -> list[str]:
         if scope and scope != service:
             jobs.append(f"lb-{scope}-backup-execution")
     elif platform == "ibkr":
+        canonical_precheck = f"{service.removesuffix('-service')}-precheck-scheduler"
+        jobs.remove(canonical_precheck)
+        jobs.append("interactive-brokers-monitor-dispatcher-scheduler")
         prefix = "interactive-brokers-quant-live-"
         suffix = service
         if suffix.startswith(prefix):

tests/test_reconcile_cloud_runtime.py

@@ -60,9 +60,9 @@ def test_legacy_jobs_for_ibkr_service_include_only_explicit_candidates(self) ->
                 "interactive-brokers-quant-live-u1234-service-probe-scheduler",
                 "interactive-brokers-quant-live-u1234-service-precheck-scheduler",
                 "interactive-brokers-quant-live-u1234-probe-scheduler",
-                "interactive-brokers-quant-live-u1234-precheck-scheduler",
                 "ibkr-u1234-backup-execution",
                 "ibkr-u1234-pre-market-dry-run",
+                "interactive-brokers-monitor-dispatcher-scheduler",
             },
         )
 
@@ -181,9 +181,9 @@ def test_delete_legacy_schedulers_deletes_only_known_jobs(self) -> None:
             "interactive-brokers-quant-live-u1234-service-probe-scheduler",
             "interactive-brokers-quant-live-u1234-service-precheck-scheduler",
             "interactive-brokers-quant-live-u1234-probe-scheduler",
-            "interactive-brokers-quant-live-u1234-precheck-scheduler",
             "ibkr-u1234-backup-execution",
             "ibkr-u1234-pre-market-dry-run",
+            "interactive-brokers-monitor-dispatcher-scheduler",
         }
         describe_calls: list[list[str]] = []
         delete_calls: list[list[str]] = []
@@ -221,6 +221,5 @@ def fake_run(args, *, json_output=False, dry_run=False):
         self.assertEqual({call[4] for call in describe_calls}, expected_jobs)
         self.assertEqual({call[4] for call in delete_calls}, expected_jobs)
 
-
 if __name__ == "__main__":
     unittest.main()

tests/test_scheduler_deadline_contract.py

@@ -8,6 +8,7 @@
 
 ROOT = Path(__file__).resolve().parents[1]
 WORKFLOW = ROOT / ".github" / "workflows" / "sync-cloud-run-env.yml"
+MAIN = ROOT / "main.py"
 
 
 def test_live_gateway_run_deadline_exceeds_cloud_run_timeout() -> None:
@@ -41,3 +42,51 @@ def test_run_deadline_is_plan_scoped_and_warmup_remains_60_seconds() -> None:
     assert 'main_attempt_deadline_args+=(--attempt-deadline="${main_attempt_deadline}")' in workflow
     assert workflow.count('"${main_attempt_deadline_args[@]}"') == 2
     assert workflow.count("--attempt-deadline=60s") == 2
+
+
+def test_precheck_uses_per_service_scheduler_with_bounded_deadline() -> None:
+    workflow = WORKFLOW.read_text(encoding="utf-8")
+    main = MAIN.read_text(encoding="utf-8")
+
+    assert 'precheck_job_name="${cloud_run_service%-service}-precheck-scheduler"' in workflow
+    assert 'precheck_uri="${service_url}/dry-run"' in workflow
+    assert workflow.count("--attempt-deadline=120s") == 2
+    assert workflow.count("--max-retry-attempts=0") == 2
+    assert workflow.count("--max-retry-duration=0s") == 2
+    assert 'gcloud scheduler jobs resume "${precheck_job_name}"' in workflow
+    assert 'gcloud scheduler jobs pause "${precheck_job_name}"' in workflow
+    assert 'monitor_job_name="interactive-brokers-monitor-dispatcher-scheduler"' not in workflow
+    assert 'shared_env_pairs+=("IBKR_MONITOR_DISPATCH_TARGETS_JSON=' not in workflow
+
+    strategy_deadline = re.search(r'IBKR_DRY_RUN_DEADLINE_SECONDS = get_positive_int_env\([^,]+, (\d+)\)', main)
+    report_grace = re.search(r'IBKR_DEADLINE_REPORT_GRACE_SECONDS = get_positive_int_env\([^,]+, (\d+)\)', main)
+    assert strategy_deadline is not None
+    assert report_grace is not None
+    assert 120 > int(strategy_deadline.group(1)) + int(report_grace.group(1))
+
+
+def test_disabled_target_gets_a_paused_canonical_precheck_before_legacy_cleanup() -> None:
+    workflow = WORKFLOW.read_text(encoding="utf-8")
+
+    ensure_precheck = workflow.index('if [ -n "${precheck_state}" ]; then')
+    enabled_state = workflow.index('case "${runtime_target_enabled}" in')
+    pause_precheck = workflow.index('gcloud scheduler jobs pause "${precheck_job_name}"')
+    retire_legacy = workflow.index('python3 scripts/reconcile_cloud_runtime.py "${reconcile_args[@]}"')
+
+    assert ensure_precheck < enabled_state < pause_precheck < retire_legacy
+
+
+def test_legacy_dispatcher_is_retired_after_replacement_jobs_are_ready() -> None:
+    workflow = WORKFLOW.read_text(encoding="utf-8")
+
+    env_sync_start = workflow.index("shared_remove_env_vars=(")
+    env_sync_end = workflow.index("shared_remove_secret_vars=(")
+    hk_migration_guard = workflow.index("HK verify scheduler migration requires the configured migration first.")
+    replacement_jobs = workflow.index('precheck_job_name="${cloud_run_service%-service}-precheck-scheduler"')
+    retire_legacy = workflow.index('python3 scripts/reconcile_cloud_runtime.py "${reconcile_args[@]}"')
+    remove_targets = workflow.index("--remove-env-vars=IBKR_MONITOR_DISPATCH_TARGETS_JSON")
+
+    assert "IBKR_MONITOR_DISPATCH_TARGETS_JSON" not in workflow[env_sync_start:env_sync_end]
+    assert hk_migration_guard < replacement_jobs < retire_legacy < remove_targets
+    assert 'Skipping precheck scheduler migration for the isolated HK verify target.' not in workflow
+    assert 'reconcile_args+=(--preserve-shared-monitor-dispatcher)' not in workflow