Reconcile LongBridge target schedulers

scripts/build_cloud_run_env_sync_plan.py

@@ -399,6 +399,10 @@ def _build_target_plan(
             per_service_mode=per_service_mode,
             allow_shared_fallback=name in SHARED_TARGET_FALLBACK_ENV,
         )
+        if name == "RUNTIME_TARGET_ENABLED" and service_name == str(
+            env.get("CLOUD_RUN_SERVICE") or ""
+        ).strip():
+            value = _coerce_env_value(env.get(name)) or value
         # 2. runtime_target-derived
         if value is None and name == "LONGBRIDGE_DRY_RUN_ONLY":
             dry_run_value = runtime_target.get("dry_run_only")
@@ -413,6 +417,9 @@ def _build_target_plan(
             if override_value is not None:
                 value = _coerce_env_value(override_value)
 
+        if name == "RUNTIME_TARGET_ENABLED" and value is None:
+            value = "true"
+
         if value is None:
             remove_env_vars.append(name)
         else:
@@ -473,10 +480,23 @@ def _build_scheduler_plan(
             per_service_mode=per_service_mode,
             allow_shared_fallback=True,
         )
-        scheduler[key] = str(runtime_scheduler.get(key) or configured_value or SCHEDULER_TIME_DEFAULTS[key])
+        scheduler[key] = _validate_scheduler_time(
+            key,
+            str(runtime_scheduler.get(key) or configured_value or SCHEDULER_TIME_DEFAULTS[key]),
+        )
     return scheduler
 
 
+def _validate_scheduler_time(name: str, value: str) -> str:
+    fields = value.split()
+    if len(fields) == 5 and fields[2] != "*" and fields[4] != "*":
+        raise ValueError(
+            f"{name} cannot constrain both day-of-month and day-of-week; "
+            "Cloud Scheduler applies OR semantics to those fields"
+        )
+    return value
+
+
 def _validate_profile_inputs(
     *,
     service_name: str,

scripts/reconcile_cloud_runtime.py

@@ -277,7 +277,7 @@ def ensure_latest_traffic(
 
 def _legacy_jobs_for_target(platform: str, target: RuntimeTarget) -> list[str]:
     service = target.service_name
-    jobs = [f"{service}-probe-scheduler", f"{service}-precheck-scheduler"]
+    jobs: list[str] = []
     if service.endswith("-service"):
         base = service[: -len("-service")]
         jobs.extend([f"{base}-probe-scheduler", f"{base}-precheck-scheduler"])

tests/test_build_cloud_run_env_sync_plan.py

@@ -82,6 +82,34 @@ def test_build_cloud_run_env_sync_plan_legacy_mode_uses_shared_env():
     }
 
 
+def test_build_cloud_run_env_sync_plan_rejects_ambiguous_cloud_scheduler_cron():
+    env = {
+        **os.environ,
+        "CLOUD_RUN_SERVICE": "longbridge-quant-paper-service",
+        "GLOBAL_TELEGRAM_CHAT_ID": "5992562050",
+        "NOTIFY_LANG": "zh",
+        "ACCOUNT_PREFIX": "PAPER",
+        "RUNTIME_TARGET_JSON": runtime_target_json(
+            "soxl_soxx_trend_income",
+            deployment_selector="PAPER",
+            account_scope="PAPER",
+            service_name="longbridge-quant-paper-service",
+        ),
+        "LONGBRIDGE_MARKET": "US",
+        "CLOUD_SCHEDULER_PROBE_TIME": "35 9 1-7 * 1-5",
+    }
+
+    result = subprocess.run(
+        [sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"],
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+
+    assert result.returncode != 0
+    assert "cannot constrain both day-of-month and day-of-week" in result.stderr
+
+
 def test_build_cloud_run_env_sync_plan_requires_target_snapshot_in_per_service_mode():
     payload = {
         "defaults": {
@@ -219,3 +247,50 @@ def test_build_cloud_run_env_sync_plan_supports_per_service_targets():
         live_mega["env"]["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"]
         == "gs://runtime/mega/snapshot.csv.manifest.json"
     )
+
+
+def test_current_service_uses_environment_runtime_enabled_state():
+    payload = {
+        "defaults": {
+            "GLOBAL_TELEGRAM_CHAT_ID": "5992562050",
+            "NOTIFY_LANG": "zh",
+            "LONGBRIDGE_MARKET": "US",
+        },
+        "targets": [
+            {
+                "service": service,
+                "account_prefix": scope,
+                "runtime_target": json.loads(
+                    runtime_target_json(
+                        "tqqq_growth_income",
+                        deployment_selector=scope,
+                        account_scope=scope,
+                        service_name=service,
+                    )
+                ),
+            }
+            for service, scope in (
+                ("longbridge-quant-paper-service", "PAPER"),
+                ("longbridge-quant-sg-service", "SG"),
+            )
+        ],
+    }
+    env = {
+        **os.environ,
+        "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload),
+        "CLOUD_RUN_SERVICE": "longbridge-quant-sg-service",
+        "RUNTIME_TARGET_ENABLED": "false",
+    }
+
+    result = subprocess.run(
+        [sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"],
+        check=True,
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+
+    plan = json.loads(result.stdout)
+    by_service = {target["service_name"]: target for target in plan["targets"]}
+    assert by_service["longbridge-quant-sg-service"]["env"]["RUNTIME_TARGET_ENABLED"] == "false"
+    assert by_service["longbridge-quant-paper-service"]["env"]["RUNTIME_TARGET_ENABLED"] == "true"

tests/test_reconcile_cloud_runtime.py

@@ -159,16 +159,6 @@ def fake_run(args, *, json_output=False, dry_run=False):
         self.assertEqual(
             delete_commands,
             [
-                ("longbridge-quant-paper-service-probe-scheduler", "asia-east2"),
-                ("longbridge-quant-paper-service-probe-scheduler", "asia-east1"),
-                ("longbridge-quant-paper-service-probe-scheduler", "europe-west1"),
-                ("longbridge-quant-paper-service-probe-scheduler", "asia-northeast1"),
-                ("longbridge-quant-paper-service-probe-scheduler", "asia-south1"),
-                ("longbridge-quant-paper-service-precheck-scheduler", "asia-east2"),
-                ("longbridge-quant-paper-service-precheck-scheduler", "asia-east1"),
-                ("longbridge-quant-paper-service-precheck-scheduler", "europe-west1"),
-                ("longbridge-quant-paper-service-precheck-scheduler", "asia-northeast1"),
-                ("longbridge-quant-paper-service-precheck-scheduler", "asia-south1"),
                 ("longbridge-quant-paper-probe-scheduler", "asia-east2"),
                 ("longbridge-quant-paper-probe-scheduler", "asia-east1"),
                 ("longbridge-quant-paper-probe-scheduler", "europe-west1"),

tests/test_scheduler_workflow.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+
+def test_replacement_probe_and_precheck_exist_before_shared_dispatcher_cleanup() -> None:
+    workflow = Path(".github/workflows/sync-cloud-run-env.yml").read_text(encoding="utf-8")
+
+    probe = workflow.index('probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler"')
+    precheck = workflow.index('precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler"')
+    cleanup_job = workflow.index("cleanup-shared-monitor:")
+    cleanup = workflow.index('gcloud scheduler jobs delete "${monitor_job_name}"')
+
+    assert probe < cleanup_job < cleanup
+    assert precheck < cleanup_job < cleanup
+
+    cleanup_section = workflow[cleanup_job:]
+    sync_section = workflow[:cleanup_job]
+    assert "needs: sync" in cleanup_section
+    assert "environment: longbridge-sg" in cleanup_section
+    assert 'replacement_jobs=("${service_name}-probe-scheduler" "${service_name}-precheck-scheduler")' in cleanup_section
+    assert "if: steps.replacements.outputs.ready == 'true'" in cleanup_section
+    assert "MONITOR_DISPATCH_TARGETS_JSON" not in sync_section
+    assert "LONGBRIDGE_MONITOR_DISPATCH_TARGETS_JSON" not in sync_section
+
+
+def test_replacement_verifier_falls_back_to_global_scheduler_location() -> None:
+    workflow = Path(".github/workflows/sync-cloud-run-env.yml").read_text(encoding="utf-8")
+    cleanup_section = workflow[workflow.index("cleanup-shared-monitor:") :]
+
+    assert "if isinstance(payload, list):" in cleanup_section
+    assert "elif isinstance(payload, dict):" in cleanup_section
+    assert "for target in targets:" in cleanup_section
+    assert 'or os.environ.get("CLOUD_SCHEDULER_LOCATION")' in cleanup_section
+    assert 'or os.environ.get("CLOUD_RUN_REGION")' in cleanup_section
+
+
+def test_existing_scheduler_cron_rejects_ambiguous_day_fields() -> None:
+    workflow = Path(".github/workflows/sync-cloud-run-env.yml").read_text(encoding="utf-8")
+
+    assert 'if current_fields[2] != "*" and current_fields[4] != "*":' in workflow
+    assert "cannot constrain both day-of-month and day-of-week" in workflow

tests/test_sync_cloud_run_env_workflow.sh

@@ -61,7 +61,7 @@ grep -Fq 'STRATEGY_PLUGIN_ALERT_EMAIL_SENDER_PASSWORD: ${{ secrets.STRATEGY_PLUG
 grep -Fq 'STRATEGY_PLUGIN_ALERT_SMS_AUTH_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_SMS_AUTH_TOKEN }}' "$workflow_file"
 grep -Fq 'STRATEGY_PLUGIN_ALERT_PUSH_APP_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_PUSH_APP_TOKEN }}' "$workflow_file"
 grep -Fq 'STRATEGY_PLUGIN_ALERT_PUSH_ACCESS_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_PUSH_ACCESS_TOKEN }}' "$workflow_file"
-grep -Fq 'STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN: ${{ secrets.STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN }}' "$workflow_file"
+grep -Fq 'STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN: ${{ secrets.TG_TOKEN }}' "$workflow_file"
 grep -Fq 'TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}' "$workflow_file"
 grep -Fq 'LONGPORT_APP_KEY_SECRET_NAME: ${{ vars.LONGPORT_APP_KEY_SECRET_NAME }}' "$workflow_file"
 grep -Fq 'LONGPORT_APP_SECRET_SECRET_NAME: ${{ vars.LONGPORT_APP_SECRET_SECRET_NAME }}' "$workflow_file"
@@ -209,11 +209,10 @@ grep -Fq 'join_by_delimiter()' "$workflow_file"
 grep -Fq 'gcloud_args+=(--remove-secrets "$(IFS=,; echo "${remove_secret_vars[*]}")")' "$workflow_file"
 grep -Fq 'gcloud_args+=(--update-secrets "$(IFS=,; echo "${secret_pairs[*]}")")' "$workflow_file"
 grep -Fq -- '--update-env-vars "^|^$(join_by_delimiter "|" "${env_pairs[@]}")"' "$workflow_file"
-grep -Fq 'MONITOR_DISPATCH_TARGETS_JSON=${monitor_targets_json}' "$workflow_file"
-grep -Fq 'region_by_service' "$workflow_file"
-grep -Fq 'service_url_by_service' "$workflow_file"
 grep -Fq 'Sync Cloud Scheduler schedule' "$workflow_file"
 grep -Fq 'scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"' "$workflow_file"
+grep -Fq 'target_env = target.get("env") or {}' "$workflow_file"
+grep -Fq 'runtime_target_enabled="${scheduler_config[4]}"' "$workflow_file"
 grep -Fq 'scheduler_job_candidates=("${CLOUD_RUN_SERVICE}-scheduler")' "$workflow_file"
 grep -Fq 'current_schedule="$(gcloud scheduler jobs describe "${candidate_job}"' "$workflow_file"
 grep -Fq 'desired_schedule="$(CURRENT_SCHEDULE="${current_schedule}" SCHEDULE_TIME="${main_time}" python - <<' "$workflow_file"
@@ -222,15 +221,27 @@ grep -Fq 'print(" ".join(time_fields))' "$workflow_file"
 grep -Fq 'print(" ".join([*time_fields, *current_fields[2:]]))' "$workflow_file"
 grep -Fq 'gcloud scheduler jobs update http "${job_name}"' "$workflow_file"
 grep -Fq 'gcloud scheduler jobs create http "${job_name}"' "$workflow_file"
+grep -Fq 'probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler"' "$workflow_file"
+grep -Fq 'probe_uri="${service_url}/probe"' "$workflow_file"
+grep -Fq 'precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler"' "$workflow_file"
+grep -Fq 'precheck_uri="${service_url}/dry-run"' "$workflow_file"
+grep -Fq 'managed_scheduler_jobs=("${job_name}" "${probe_job_name}" "${precheck_job_name}")' "$workflow_file"
+grep -Fq 'gcloud scheduler jobs resume "${managed_job_name}"' "$workflow_file"
+grep -Fq 'gcloud scheduler jobs pause "${managed_job_name}"' "$workflow_file"
 grep -Fq 'monitor_job_name="longbridge-monitor-dispatcher-scheduler"' "$workflow_file"
-grep -Fq 'monitor_uri="${service_url}/monitor-dispatch"' "$workflow_file"
+grep -Fq 'gcloud scheduler jobs delete "${monitor_job_name}"' "$workflow_file"
 grep -Fq 'Reconcile Cloud Run traffic' "$workflow_file"
 grep -Fq 'python3 scripts/reconcile_cloud_runtime.py --platform longbridge --ensure-latest-traffic' "$workflow_file"
 grep -Fq 'Reconcile legacy Cloud Scheduler jobs' "$workflow_file"
 grep -Fq 'python3 scripts/reconcile_cloud_runtime.py --platform longbridge --delete-legacy-schedulers' "$workflow_file"
 grep -Fq -- '--schedule="${desired_schedule}"' "$workflow_file"
 grep -Fq -- '--time-zone="${market_timezone}"' "$workflow_file"
 
+if grep -Fq 'monitor_uri="${service_url}/monitor-dispatch"' "$workflow_file"; then
+  echo "unexpected shared monitor dispatcher creation still present" >&2
+  exit 1
+fi
+
 if grep -Fq 'legacy_jobs=(' "$workflow_file"; then
   echo "unexpected inline legacy scheduler deletion logic still present" >&2
   exit 1