diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index a4ffa92..9f8b94e 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -341,12 +341,26 @@ jobs: targets = plan.get("targets") or [] if not targets: raise SystemExit("Cloud Run env sync did not resolve any targets") - for target in targets: - service_name = str(target.get("service_name") or "").strip() - if not service_name: - raise SystemExit("Cloud Run sync target is missing service_name") - if not isinstance(target.get("env"), dict): - raise SystemExit(f"Cloud Run sync target {service_name} is missing env") + matching_targets = [ + candidate + for candidate in targets + if str(candidate.get("service_name") or "").strip() + == os.environ["CLOUD_RUN_SERVICE"] + ] + if len(matching_targets) != 1: + raise SystemExit( + f"Expected exactly one Cloud Run sync target for {os.environ['CLOUD_RUN_SERVICE']!r}" + ) + target = matching_targets[0] + service_name = str(target.get("service_name") or "").strip() + if not service_name: + raise SystemExit("Cloud Run sync target is missing service_name") + if service_name != os.environ["CLOUD_RUN_SERVICE"]: + raise SystemExit( + f"Cloud Run sync target {service_name!r} does not match CLOUD_RUN_SERVICE" + ) + if not isinstance(target.get("env"), dict): + raise SystemExit(f"Cloud Run sync target {service_name} is missing env") PY if [ "${#missing_vars[@]}" -gt 0 ]; then @@ -391,7 +405,17 @@ jobs: import os plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + matching_targets = [ + candidate + for candidate in plan.get("targets") or [] + if str(candidate.get("service_name") or "").strip() + == os.environ["CLOUD_RUN_SERVICE"] + ] + if len(matching_targets) != 1: + raise SystemExit( + f"Expected exactly one Cloud Run sync target for {os.environ['CLOUD_RUN_SERVICE']!r}" + ) + target = matching_targets[0] for key, value in sorted(target["env"].items()): print(f"{key}={value}") PY @@ -401,7 +425,17 @@ jobs: import os plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + matching_targets = [ + candidate + for candidate in plan.get("targets") or [] + if str(candidate.get("service_name") or "").strip() + == os.environ["CLOUD_RUN_SERVICE"] + ] + if len(matching_targets) != 1: + raise SystemExit( + f"Expected exactly one Cloud Run sync target for {os.environ['CLOUD_RUN_SERVICE']!r}" + ) + target = matching_targets[0] for key in sorted(target.get("remove_env_vars") or []): print(key) PY @@ -533,7 +567,17 @@ jobs: import os plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + matching_targets = [ + candidate + for candidate in plan.get("targets") or [] + if str(candidate.get("service_name") or "").strip() + == os.environ["CLOUD_RUN_SERVICE"] + ] + if len(matching_targets) != 1: + raise SystemExit( + f"Expected exactly one Cloud Run sync target for {os.environ['CLOUD_RUN_SERVICE']!r}" + ) + target = matching_targets[0] env = target.get("env") or {} runtime_target = json.loads(env.get("RUNTIME_TARGET_JSON") or "{}") payload = { @@ -575,9 +619,11 @@ jobs: python scripts/reconcile_cloud_runtime.py traffic - name: Sync Cloud Scheduler schedule + id: scheduler_sync if: steps.env_sync_config.outputs.enabled == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} + DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }} run: | set -euo pipefail @@ -592,20 +638,31 @@ jobs: import os plan = json.loads(os.environ["SYNC_PLAN_JSON"]) - target = plan["targets"][0] + targets = plan.get("targets") or [] + matching_targets = [ + candidate + for candidate in targets + if str(candidate.get("service_name") or "").strip() == os.environ["CLOUD_RUN_SERVICE"] + ] + if len(matching_targets) != 1: + raise SystemExit( + f"Expected exactly one scheduler target for {os.environ['CLOUD_RUN_SERVICE']!r}" + ) + target = matching_targets[0] scheduler = target.get("scheduler") or {} + target_env = target.get("env") or {} print(str(scheduler.get("timezone") or "America/New_York").strip()) print(str(scheduler.get("main_time") or os.environ.get("CLOUD_SCHEDULER_MAIN_TIME", "").strip() or "45 15")) print(str(scheduler.get("probe_time") or os.environ.get("CLOUD_SCHEDULER_PROBE_TIME", "").strip() or "35 9,15")) print(str(scheduler.get("precheck_time") or os.environ.get("CLOUD_SCHEDULER_PRECHECK_TIME", "").strip() or "45 9")) + print(str(target_env.get("RUNTIME_TARGET_ENABLED") or "true").strip().lower()) PY ) market_timezone="${scheduler_config[0]}" main_time="${scheduler_config[1]}" probe_time="${scheduler_config[2]}" precheck_time="${scheduler_config[3]}" - # Keep probe/precheck parsed for runtime-target schema parity; this step only syncs the main scheduler. - : "${probe_time}" "${precheck_time}" + runtime_target_enabled="${scheduler_config[4]}" service_url="$(gcloud run services describe "${CLOUD_RUN_SERVICE}" \ --project="${GCP_PROJECT_ID}" \ @@ -710,81 +767,230 @@ jobs: --quiet fi - monitor_job_name="firstrade-monitor-dispatcher-scheduler" - monitor_uri="${service_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}" \ - --project="${GCP_PROJECT_ID}" \ - --location="${scheduler_location}" \ - --uri="${monitor_uri}" \ - --http-method=POST \ - --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ - --oidc-token-audience="${service_url}" \ - --schedule="*/5 * * * *" \ - --time-zone="UTC" \ - --attempt-deadline=180s \ - --quiet - else - echo "Creating Cloud Scheduler job ${monitor_job_name} at ${monitor_uri}." - gcloud scheduler jobs create http "${monitor_job_name}" \ + managed_scheduler_jobs=("${job_name}") + probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler" + probe_uri="${service_url}/probe" + precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler" + precheck_uri="${service_url}/dry-run" + if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then + 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: + 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_probe_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${probe_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 + )" + + if gcloud scheduler jobs describe "${probe_job_name}" \ --project="${GCP_PROJECT_ID}" \ - --location="${scheduler_location}" \ - --uri="${monitor_uri}" \ - --http-method=POST \ - --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ - --oidc-token-audience="${service_url}" \ - --schedule="*/5 * * * *" \ - --time-zone="UTC" \ - --attempt-deadline=180s \ - --quiet - fi + --location="${scheduler_location}" >/dev/null 2>&1; then + echo "Updating Cloud Scheduler probe ${probe_job_name} to ${desired_probe_schedule}." + gcloud scheduler jobs update http "${probe_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${probe_uri}" \ + --schedule="${desired_probe_schedule}" \ + --time-zone="${market_timezone}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + else + echo "Creating Cloud Scheduler probe ${probe_job_name} at ${desired_probe_schedule}." + gcloud scheduler jobs create http "${probe_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${probe_uri}" \ + --schedule="${desired_probe_schedule}" \ + --time-zone="${market_timezone}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + fi - invoke_bridge_jobs=( - "${CLOUD_RUN_SERVICE}-probe-scheduler|${service_url}/probe" - "${CLOUD_RUN_SERVICE}-precheck-scheduler|${service_url}/dry-run" - ) - if [[ "${CLOUD_RUN_SERVICE}" == *-service ]]; then - invoke_bridge_jobs+=( - "${CLOUD_RUN_SERVICE%-service}-probe-scheduler|${service_url}/probe" - "${CLOUD_RUN_SERVICE%-service}-precheck-scheduler|${service_url}/dry-run" - ) - fi - for bridge_entry in "${invoke_bridge_jobs[@]}"; do - bridge_job="${bridge_entry%%|*}" - bridge_uri="${bridge_entry#*|}" - if gcloud scheduler jobs describe "${bridge_job}" \ + if gcloud scheduler jobs describe "${precheck_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" >/dev/null 2>&1; 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=180s \ + --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=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ + --quiet + fi + + managed_scheduler_jobs+=("${probe_job_name}" "${precheck_job_name}") + else + monitor_job_name="firstrade-monitor-dispatcher-scheduler" + monitor_uri="${service_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 invoke-bridge Cloud Scheduler job ${bridge_job} to ${bridge_uri}." - gcloud scheduler jobs update http "${bridge_job}" \ + echo "Updating Cloud Scheduler job ${monitor_job_name} to ${monitor_uri}." + gcloud scheduler jobs update http "${monitor_job_name}" \ --project="${GCP_PROJECT_ID}" \ --location="${scheduler_location}" \ - --uri="${bridge_uri}" \ + --uri="${monitor_uri}" \ --http-method=POST \ --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ --oidc-token-audience="${service_url}" \ + --schedule="*/5 * * * *" \ + --time-zone="UTC" \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ --quiet else - echo "Creating invoke-bridge Cloud Scheduler job ${bridge_job} at ${bridge_uri}." - gcloud scheduler jobs create http "${bridge_job}" \ + 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="${bridge_uri}" \ + --uri="${monitor_uri}" \ --http-method=POST \ --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ --oidc-token-audience="${service_url}" \ - --schedule="0 0 1 1 *" \ + --schedule="*/5 * * * *" \ --time-zone="UTC" \ - --attempt-deadline=600s \ + --attempt-deadline=180s \ + --max-retry-attempts=0 \ + --max-retry-duration=0s \ --quiet fi + managed_scheduler_jobs+=("${monitor_job_name}") + + invoke_bridge_jobs=( + "${probe_job_name}|${probe_uri}" + "${precheck_job_name}|${precheck_uri}" + ) + for bridge_entry in "${invoke_bridge_jobs[@]}"; do + bridge_job="${bridge_entry%%|*}" + bridge_uri="${bridge_entry#*|}" + if gcloud scheduler jobs describe "${bridge_job}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" >/dev/null 2>&1; then + echo "Updating dormant invoke-bridge Cloud Scheduler job ${bridge_job}." + gcloud scheduler jobs update http "${bridge_job}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${bridge_uri}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --schedule="0 0 1 1 *" \ + --time-zone="UTC" \ + --attempt-deadline=600s \ + --quiet + else + echo "Creating dormant invoke-bridge Cloud Scheduler job ${bridge_job}." + gcloud scheduler jobs create http "${bridge_job}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --uri="${bridge_uri}" \ + --http-method=POST \ + --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \ + --oidc-token-audience="${service_url}" \ + --schedule="0 0 1 1 *" \ + --time-zone="UTC" \ + --attempt-deadline=600s \ + --quiet + fi + managed_scheduler_jobs+=("${bridge_job}") + done + fi + for managed_job_name in "${managed_scheduler_jobs[@]}"; do + managed_job_state="$(gcloud scheduler jobs describe "${managed_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --format='value(state)')" + case "${runtime_target_enabled}" in + 1|true|yes|on) + if [ "${managed_job_state}" = "PAUSED" ]; then + echo "Resuming Cloud Scheduler job ${managed_job_name} because ${CLOUD_RUN_SERVICE} is enabled." + gcloud scheduler jobs resume "${managed_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --quiet + fi + ;; + *) + if [ "${managed_job_state}" != "PAUSED" ]; then + echo "Pausing Cloud Scheduler job ${managed_job_name} because ${CLOUD_RUN_SERVICE} is disabled." + gcloud scheduler jobs pause "${managed_job_name}" \ + --project="${GCP_PROJECT_ID}" \ + --location="${scheduler_location}" \ + --quiet + fi + ;; + esac done + if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then + echo "direct_monitors_reconciled=true" >> "${GITHUB_OUTPUT}" + fi - name: Reconcile legacy Cloud Scheduler jobs if: steps.env_sync_config.outputs.enabled == 'true' + env: + SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} + DIRECT_MONITOR_MIGRATION_COMPLETE: ${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }} + DIRECT_MONITOR_SCHEDULERS_RECONCILED: ${{ steps.scheduler_sync.outputs.direct_monitors_reconciled }} run: | set -euo pipefail python scripts/reconcile_cloud_runtime.py scheduler-cleanup diff --git a/application/execution_service.py b/application/execution_service.py index 50cba00..839c913 100644 --- a/application/execution_service.py +++ b/application/execution_service.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass +from math import floor from typing import Any from quant_platform_kit.common.order_status import compute_confirmed_sell_release_value @@ -139,6 +140,7 @@ class ExecutionCycleResult: DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0 SMALL_ACCOUNT_SAFE_HAVEN_CASH_SUBSTITUTE_LIMIT_USD = 2000.0 MIN_NOTIONAL_BUY_USD = 1.0 +NOTIONAL_BUY_CASH_UTILIZATION_RATIO = 0.98 _ACCEPTED_ORDER_STATUSES = frozenset( {"accepted", "filled", "partiallyfilled", "previewed", "submitted"} ) @@ -567,6 +569,19 @@ def _submit_notional_buy_order( } +def _apply_notional_cash_buffer(*, buy_budget: float, investable_cash: float) -> float: + budget = max(0.0, float(buy_budget or 0.0)) + available = max(0.0, float(investable_cash or 0.0)) + if budget + 0.005 < available: + return budget + buffered_available = floor( + available * NOTIONAL_BUY_CASH_UTILIZATION_RATIO * 100 + ) / 100 + if budget >= MIN_NOTIONAL_BUY_USD and available >= MIN_NOTIONAL_BUY_USD: + buffered_available = max(buffered_available, MIN_NOTIONAL_BUY_USD) + return min(budget, buffered_available) + + def _order_submission_accepted(order: dict[str, Any]) -> bool: status = "".join( ch for ch in str(order.get("status") or "").strip().lower() if ch.isalnum() @@ -770,6 +785,10 @@ def execute_value_target_plan( if order_notional_cap is not None: buy_budget = min(buy_budget, order_notional_cap) if notional_buy_execution: + buy_budget = _apply_notional_cash_buffer( + buy_budget=buy_budget, + investable_cash=investable_cash, + ) if buy_budget >= MIN_NOTIONAL_BUY_USD: estimated_buy_cost += buy_budget elif float(delta_value) >= MIN_NOTIONAL_BUY_USD: @@ -820,6 +839,10 @@ def execute_value_target_plan( if order_notional_cap is not None: buy_budget = min(buy_budget, order_notional_cap) if notional_buy_execution: + buy_budget = _apply_notional_cash_buffer( + buy_budget=buy_budget, + investable_cash=investable_cash, + ) if buy_budget < MIN_NOTIONAL_BUY_USD: skipped.append( { diff --git a/application/runtime_broker_adapters.py b/application/runtime_broker_adapters.py index cb5772d..e77ce84 100644 --- a/application/runtime_broker_adapters.py +++ b/application/runtime_broker_adapters.py @@ -110,6 +110,21 @@ def _positive_or_none(value: float | None) -> float | None: return resolved if resolved > 0.0 else None +def _resolve_buying_power( + *, + cash_balance: float | None, + reported_buying_power: float | None, + cash_only_execution: bool, +) -> float | None: + if not cash_only_execution: + return reported_buying_power if reported_buying_power is not None else cash_balance + if cash_balance is None: + return None + if reported_buying_power is None: + return cash_balance + return max(0.0, min(float(cash_balance), float(reported_buying_power))) + + def _resolve_total_equity( *, balances, @@ -281,8 +296,10 @@ def build_portfolio_snapshot(self) -> PortfolioSnapshot: ) cash_balance = _first_numeric_by_keyword_groups(balances, _CASH_BALANCE_KEYWORD_GROUPS) reported_buying_power = _first_numeric_by_keyword_groups(balances, _BUYING_POWER_KEYWORD_GROUPS) - buying_power = cash_balance if self.cash_only_execution else ( - reported_buying_power if reported_buying_power is not None else cash_balance + buying_power = _resolve_buying_power( + cash_balance=cash_balance, + reported_buying_power=reported_buying_power, + cash_only_execution=self.cash_only_execution, ) position_market_value = sum(position.market_value for position in positions) total_equity, total_equity_source = _resolve_total_equity( diff --git a/application/strategy_run_persistence.py b/application/strategy_run_persistence.py index 3d800b2..0143f74 100644 --- a/application/strategy_run_persistence.py +++ b/application/strategy_run_persistence.py @@ -85,9 +85,16 @@ def resolve_strategy_run_period( return f"{now.year:04d}-{now.month:02d}" -def strategy_run_state_key(*, account: str, strategy_profile: str, run_period: str) -> str: +def strategy_run_state_key( + *, + account: str, + strategy_profile: str, + run_period: str, + dry_run_only: bool = False, +) -> str: + prefix = "strategy-runs/dry-run" if dry_run_only else "strategy-runs" return ( - f"strategy-runs/{safe_key(account)}/{safe_key(strategy_profile)}/" + f"{prefix}/{safe_key(account)}/{safe_key(strategy_profile)}/" f"{safe_key(run_period)}/latest.json" ) @@ -99,10 +106,12 @@ def strategy_run_history_key( run_period: str, stage: str, now: datetime, + dry_run_only: bool = False, ) -> str: stamp = now.strftime("%Y%m%dT%H%M%SZ") + prefix = "strategy-runs/dry-run" if dry_run_only else "strategy-runs" return ( - f"strategy-runs/{safe_key(account)}/{safe_key(strategy_profile)}/" + f"{prefix}/{safe_key(account)}/{safe_key(strategy_profile)}/" f"{safe_key(run_period)}/history/{now:%Y/%m/%d}/{stamp}-{safe_key(stage)}.json" ) @@ -189,11 +198,13 @@ def persist_strategy_run_state( strategy_profile = str(state.get("strategy_profile") or "unknown") run_period = str(state.get("run_period") or f"{as_of.year:04d}-{as_of.month:02d}") stage = str(state.get("stage") or "UNKNOWN") + dry_run_only = bool(state.get("dry_run_only")) store.write_json( strategy_run_state_key( account=account, strategy_profile=strategy_profile, run_period=run_period, + dry_run_only=dry_run_only, ), dict(state), ) @@ -204,6 +215,7 @@ def persist_strategy_run_state( run_period=run_period, stage=stage, now=as_of, + dry_run_only=dry_run_only, ), dict(state), ) diff --git a/entrypoints/cloud_run.py b/entrypoints/cloud_run.py index ec86977..67e6f93 100644 --- a/entrypoints/cloud_run.py +++ b/entrypoints/cloud_run.py @@ -17,6 +17,11 @@ def is_market_open_now(*, calendar_name="NYSE", timezone_name="America/New_York" schedule = calendar.schedule(start_date=now_market.date(), end_date=now_market.date()) if schedule.empty: return False, None - return calendar.open_at_time(schedule, now_market), None + try: + return calendar.open_at_time(schedule, now_market), None + except ValueError as exc: + if "not covered by the schedule" not in str(exc): + raise + return False, None except Exception as exc: return False, exc diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 8301a25..f82ec28 100644 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -3,9 +3,7 @@ This script keeps the runtime logic minimal and explicit: - reconcile Cloud Run traffic to the latest ready revision and verify commit-sha -- delete only explicit legacy session-check Cloud Scheduler jobs - -It deliberately does not touch probe/precheck bridge jobs or unknown schedulers. +- delete only explicit legacy Cloud Scheduler jobs """ from __future__ import annotations @@ -52,9 +50,28 @@ def _first_target(plan: Mapping[str, Any]) -> Mapping[str, Any]: def _resolve_context(env: Mapping[str, str] = os.environ) -> tuple[RuntimeContext, dict[str, Any]]: plan = _parse_sync_plan(str(env.get("SYNC_PLAN_JSON", "") or "")) target = _first_target(plan) + configured_service = str(env.get("CLOUD_RUN_SERVICE", "") or "").strip() + targets = plan.get("targets") + if configured_service and isinstance(targets, list) and targets: + matching_targets = [ + candidate + for candidate in targets + if isinstance(candidate, Mapping) + and configured_service + in { + str(candidate.get("service_name") or "").strip(), + str(candidate.get("service") or "").strip(), + str(candidate.get("cloud_run_service") or "").strip(), + } + ] + if len(matching_targets) != 1: + raise ValueError( + f"CLOUD_RUN_SERVICE {configured_service} does not match any sync-plan target" + ) + target = matching_targets[0] service_name = ( - str(env.get("CLOUD_RUN_SERVICE", "") or "").strip() + configured_service or str(target.get("service_name") or "").strip() or str(target.get("service") or "").strip() or str(target.get("cloud_run_service") or "").strip() @@ -256,11 +273,18 @@ def reconcile_traffic( time.sleep(5) -def _legacy_session_check_jobs(service_name: str) -> list[str]: +def _legacy_scheduler_jobs(service_name: str) -> list[str]: candidates = [f"{service_name}-session-check-scheduler"] alias = service_name.removesuffix("-service") if alias and alias != service_name: - candidates.append(f"{alias}-session-check-scheduler") + candidates.extend( + [ + f"{alias}-session-check-scheduler", + f"{alias}-probe-scheduler", + f"{alias}-precheck-scheduler", + ] + ) + candidates.append("firstrade-monitor-dispatcher-scheduler") seen: list[str] = [] for candidate in candidates: if candidate not in seen: @@ -273,9 +297,53 @@ def cleanup_legacy_scheduler_jobs( *, run_gcloud: RunGcloud = _run_gcloud, ) -> None: - ctx, _plan = _resolve_context(env) + ctx, plan = _resolve_context(env) deleted: list[str] = [] - for job_name in _legacy_session_check_jobs(ctx.service_name): + legacy_jobs = _legacy_scheduler_jobs(ctx.service_name) + dispatcher_job = "firstrade-monitor-dispatcher-scheduler" + direct_jobs = ( + f"{ctx.service_name}-probe-scheduler", + f"{ctx.service_name}-precheck-scheduler", + ) + targets = plan.get("targets") + has_single_sync_target = not str(env.get("SYNC_PLAN_JSON", "") or "").strip() or ( + isinstance(targets, list) and len(targets) == 1 + ) + migration_confirmed = ( + str(env.get("DIRECT_MONITOR_MIGRATION_COMPLETE") or "").strip() == "true" + ) + current_sync_confirmed = ( + str(env.get("DIRECT_MONITOR_SCHEDULERS_RECONCILED") or "").strip().lower() + == "true" + ) + direct_jobs_exist = ( + migration_confirmed + and current_sync_confirmed + and has_single_sync_target + and all( + run_gcloud( + [ + "scheduler", + "jobs", + "describe", + job_name, + "--project", + ctx.project_id, + "--location", + ctx.scheduler_location, + ] + ).returncode + == 0 + for job_name in direct_jobs + ) + ) + if dispatcher_job in legacy_jobs and not direct_jobs_exist: + legacy_jobs.remove(dispatcher_job) + print( + f"Keeping legacy Cloud Scheduler job {dispatcher_job} until direct monitor jobs exist." + ) + + for job_name in legacy_jobs: result = run_gcloud( [ "scheduler", @@ -317,7 +385,7 @@ def build_parser() -> argparse.ArgumentParser: subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("traffic", help="reconcile Cloud Run traffic") subparsers.add_parser( - "scheduler-cleanup", help="delete explicit legacy session-check scheduler jobs" + "scheduler-cleanup", help="delete explicit legacy scheduler jobs" ) return parser diff --git a/tests/test_cloud_run_entrypoint.py b/tests/test_cloud_run_entrypoint.py new file mode 100644 index 0000000..f0bfb3f --- /dev/null +++ b/tests/test_cloud_run_entrypoint.py @@ -0,0 +1,17 @@ +from entrypoints import cloud_run + + +def test_market_hours_after_close_is_closed_without_calendar_error(monkeypatch): + class FakeSchedule: + empty = False + + class FakeCalendar: + def schedule(self, **_kwargs): + return FakeSchedule() + + def open_at_time(self, _schedule, _now): + raise ValueError("The provided timestamp is not covered by the schedule") + + monkeypatch.setattr(cloud_run.mcal, "get_calendar", lambda _name: FakeCalendar()) + + assert cloud_run.is_market_open_now() == (False, None) diff --git a/tests/test_execution_service.py b/tests/test_execution_service.py index a20ad27..975e0b8 100644 --- a/tests/test_execution_service.py +++ b/tests/test_execution_service.py @@ -4,6 +4,7 @@ from datetime import datetime, timezone from application.execution_service import ( + _apply_notional_cash_buffer, execute_value_target_plan, substitute_small_safe_haven_targets_with_cash, ) @@ -582,6 +583,33 @@ def test_execute_value_target_plan_uses_notional_buy_when_enabled(): assert result.execution_notes == () +def test_notional_buy_keeps_cash_buffer_when_order_would_use_all_available_cash(): + execution_port = FakeExecutionPort() + result = execute_value_target_plan( + plan={ + "allocation": {"targets": {"IBIT": 150.0}}, + "portfolio": { + "market_values": {"IBIT": 70.0}, + "quantities": {"IBIT": 2.0}, + "liquid_cash": 80.0, + "total_equity": 150.0, + }, + "execution": {"current_min_trade": 1.0, "investable_cash": 80.0}, + }, + market_data_port=FakeMarketDataPort({"IBIT": 35.0}), + execution_port=execution_port, + dry_run_only=False, + notional_buy_execution=True, + ) + + assert result.action_done is True + assert execution_port.orders[0].metadata["notional_usd"] == 78.4 + + +def test_notional_cash_buffer_preserves_minimum_eligible_order(): + assert _apply_notional_cash_buffer(buy_budget=1.02, investable_cash=1.02) == 1.0 + + def test_execute_value_target_plan_routes_rejected_notional_buy_to_skipped_orders(): class RejectedExecutionPort(FakeExecutionPort): def submit_order(self, order_intent) -> ExecutionReport: @@ -622,7 +650,7 @@ def submit_order(self, order_intent) -> ExecutionReport: assert result.action_done is False assert result.submitted_orders == () assert result.skipped_orders[0]["reason"] == "fractional_trading_disclosure_required" - assert result.skipped_orders[0]["notional_usd"] == 80.0 + assert result.skipped_orders[0]["notional_usd"] == 78.4 assert result.execution_notes == () diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index f19b423..a603ed9 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -578,13 +578,24 @@ def test_run_strategy_cycle_strategy_plugin_load_error_is_non_blocking(monkeypat assert result["strategy_plugin_alert_sms_sent_count"] == 0 -def test_run_strategy_cycle_persists_strategy_run_state(monkeypatch): +def test_run_strategy_cycle_dry_run_preserves_live_strategy_run_state(monkeypatch): store = FakeStateStore() + key = "strategy-runs/____5678/tqqq_growth_income/2026_05/latest.json" + existing_state = { + "stage": "SUBMITTED", + "as_of": "2026-05-01T01:02:03+00:00", + "dry_run_only": False, + } + store.payloads[key] = dict(existing_state) monkeypatch.setattr( "application.rebalance_service.load_strategy_runtime", lambda *_args, **_kwargs: FakeStrategyRuntime(), ) + monkeypatch.setattr( + "application.rebalance_service._utcnow", + lambda: datetime(2026, 5, 15, tzinfo=timezone.utc), + ) result = run_strategy_cycle( runtime_settings=_runtime_settings_with_persistence(persist_strategy_runs=True), @@ -594,15 +605,17 @@ def test_run_strategy_cycle_persists_strategy_run_state(monkeypatch): env_reader=lambda _name, default=None: default, ) - stages = [payload["stage"] for _key, payload in store.writes if _key.endswith("latest.json")] - assert stages == ["ORDERS_PLANNED", "DRY_RUN_COMPLETED"] + assert store.payloads[key] == existing_state + dry_run_key = "strategy-runs/dry-run/____5678/tqqq_growth_income/2026_05/latest.json" + dry_run_state = store.payloads[dry_run_key] + assert dry_run_state["stage"] == "DRY_RUN_COMPLETED" + assert dry_run_state["dry_run_only"] is True + assert dry_run_state["submitted_orders"][0]["symbol"] == "AAA" + assert dry_run_state["plan"]["allocation"]["targets"]["AAA"] == 50.0 assert result["strategy_run_persisted"] is True - assert result["strategy_run_period"] + assert result["strategy_run_period"] == "2026-05" assert result["strategy_run_stage"] == "DRY_RUN_COMPLETED" - latest_payload = store.writes[-2][1] - assert latest_payload["stage"] == "DRY_RUN_COMPLETED" - assert latest_payload["submitted_orders"][0]["symbol"] == "AAA" - assert latest_payload["plan"]["allocation"]["targets"]["AAA"] == 50.0 + assert result["submitted_orders"][0]["symbol"] == "AAA" def test_run_strategy_cycle_skips_duplicate_live_monthly_run(monkeypatch): diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index c456659..63e0ffe 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -37,6 +37,19 @@ def test_resolve_context_prefers_cloud_run_service_and_first_target(self): self.assertEqual(ctx.scheduler_location, "us-central1") self.assertEqual(plan["targets"][0]["region"], "asia-east1") + def test_resolve_context_rejects_service_missing_from_sync_plan(self): + with self.assertRaisesRegex(ValueError, "does not match any sync-plan target"): + reconciler._resolve_context( + { + "GCP_PROJECT_ID": "firstradequant", + "CLOUD_RUN_SERVICE": "firstrade-missing-service", + "CLOUD_RUN_REGION": "us-central1", + "SYNC_PLAN_JSON": json.dumps( + {"targets": [{"service_name": "firstrade-platform-service"}]} + ), + } + ) + def test_reconcile_traffic_updates_latest_ready_revision_and_checks_commit_sha(self): env = { "GCP_PROJECT_ID": "firstradequant", @@ -113,12 +126,14 @@ def fake_run_gcloud(command: list[str]): with self.assertRaisesRegex(RuntimeError, "commit-sha"): reconciler.reconcile_traffic(env, run_gcloud=fake_run_gcloud) - def test_cleanup_legacy_scheduler_jobs_only_targets_explicit_session_check_jobs(self): + def test_cleanup_legacy_scheduler_jobs_preserves_canonical_direct_jobs(self): env = { "GCP_PROJECT_ID": "firstradequant", "CLOUD_RUN_SERVICE": "firstrade-platform-service", "CLOUD_RUN_REGION": "us-central1", "CLOUD_SCHEDULER_LOCATION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", "SYNC_PLAN_JSON": json.dumps( {"targets": [{"service_name": "firstrade-platform-service"}]} ), @@ -127,19 +142,23 @@ def test_cleanup_legacy_scheduler_jobs_only_targets_explicit_session_check_jobs( existing_jobs = { "firstrade-platform-service-session-check-scheduler", "firstrade-platform-session-check-scheduler", + "firstrade-platform-service-probe-scheduler", + "firstrade-platform-service-precheck-scheduler", + "firstrade-platform-probe-scheduler", + "firstrade-platform-precheck-scheduler", + "firstrade-monitor-dispatcher-scheduler", } def fake_run_gcloud(command: list[str]): calls.append(command) - if command[:4] == ["scheduler", "jobs", "describe", "firstrade-platform-service-session-check-scheduler"]: - return _completed(command, stdout="{}") if command[3] in existing_jobs else _completed(command, returncode=1) - if command[:4] == ["scheduler", "jobs", "describe", "firstrade-platform-session-check-scheduler"]: - return _completed(command, stdout="{}") - if command[:4] == ["scheduler", "jobs", "delete", "firstrade-platform-service-session-check-scheduler"]: - existing_jobs.discard("firstrade-platform-service-session-check-scheduler") - return _completed(command, stdout="") - if command[:4] == ["scheduler", "jobs", "delete", "firstrade-platform-session-check-scheduler"]: - existing_jobs.discard("firstrade-platform-session-check-scheduler") + if command[:3] == ["scheduler", "jobs", "describe"]: + return ( + _completed(command, stdout="{}") + if command[3] in existing_jobs + else _completed(command, returncode=1) + ) + if command[:3] == ["scheduler", "jobs", "delete"]: + existing_jobs.discard(command[3]) return _completed(command, stdout="") raise AssertionError(f"Unexpected gcloud command: {command}") @@ -161,7 +180,192 @@ def fake_run_gcloud(command: list[str]): ["scheduler", "jobs", "delete", "firstrade-platform-session-check-scheduler", "--project", "firstradequant", "--location", "us-central1", "--quiet"], calls, ) - self.assertFalse(any("probe" in " ".join(command) or "precheck" in " ".join(command) for command in calls)) + deleted_jobs = [ + command[3] + for command in calls + if command[:3] == ["scheduler", "jobs", "delete"] + ] + self.assertEqual( + deleted_jobs, + [ + "firstrade-platform-service-session-check-scheduler", + "firstrade-platform-session-check-scheduler", + "firstrade-platform-probe-scheduler", + "firstrade-platform-precheck-scheduler", + "firstrade-monitor-dispatcher-scheduler", + ], + ) + self.assertNotIn("firstrade-platform-service-probe-scheduler", deleted_jobs) + + def test_cleanup_keeps_dispatcher_until_direct_jobs_exist(self): + env = { + "GCP_PROJECT_ID": "firstradequant", + "CLOUD_RUN_SERVICE": "firstrade-platform-service", + "CLOUD_RUN_REGION": "us-central1", + "CLOUD_SCHEDULER_LOCATION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + "SYNC_PLAN_JSON": json.dumps( + {"targets": [{"service_name": "firstrade-platform-service"}]} + ), + } + existing_jobs = { + "firstrade-platform-service-probe-scheduler", + "firstrade-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run_gcloud(command: list[str]): + if command[:3] == ["scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[3] in existing_jobs else 1, + ) + if command[:3] == ["scheduler", "jobs", "delete"]: + deleted_jobs.append(command[3]) + return _completed(command) + raise AssertionError(f"Unexpected gcloud command: {command}") + + reconciler.cleanup_legacy_scheduler_jobs(env, run_gcloud=fake_run_gcloud) + + self.assertNotIn("firstrade-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_keeps_dispatcher_without_current_sync_proof(self): + env = { + "GCP_PROJECT_ID": "firstradequant", + "CLOUD_RUN_SERVICE": "firstrade-platform-service", + "CLOUD_RUN_REGION": "us-central1", + "CLOUD_SCHEDULER_LOCATION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "SYNC_PLAN_JSON": json.dumps( + {"targets": [{"service_name": "firstrade-platform-service"}]} + ), + } + existing_jobs = { + "firstrade-platform-service-probe-scheduler", + "firstrade-platform-service-precheck-scheduler", + "firstrade-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run_gcloud(command: list[str]): + if command[:3] == ["scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[3] in existing_jobs else 1, + ) + if command[:3] == ["scheduler", "jobs", "delete"]: + deleted_jobs.append(command[3]) + return _completed(command) + raise AssertionError(f"Unexpected gcloud command: {command}") + + reconciler.cleanup_legacy_scheduler_jobs(env, run_gcloud=fake_run_gcloud) + + self.assertNotIn("firstrade-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_keeps_dispatcher_without_migration_confirmation(self): + env = { + "GCP_PROJECT_ID": "firstradequant", + "CLOUD_RUN_SERVICE": "firstrade-platform-service", + "CLOUD_RUN_REGION": "us-central1", + "CLOUD_SCHEDULER_LOCATION": "us-central1", + "SYNC_PLAN_JSON": json.dumps( + {"targets": [{"service_name": "firstrade-platform-service"}]} + ), + } + existing_jobs = { + "firstrade-platform-service-probe-scheduler", + "firstrade-platform-service-precheck-scheduler", + "firstrade-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run_gcloud(command: list[str]): + if command[:3] == ["scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[3] in existing_jobs else 1, + ) + if command[:3] == ["scheduler", "jobs", "delete"]: + deleted_jobs.append(command[3]) + return _completed(command) + raise AssertionError(f"Unexpected gcloud command: {command}") + + reconciler.cleanup_legacy_scheduler_jobs(env, run_gcloud=fake_run_gcloud) + + self.assertNotIn("firstrade-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_requires_exact_lowercase_migration_confirmation(self): + env = { + "GCP_PROJECT_ID": "firstradequant", + "CLOUD_RUN_SERVICE": "firstrade-platform-service", + "CLOUD_RUN_REGION": "us-central1", + "CLOUD_SCHEDULER_LOCATION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "TRUE", + "DIRECT_MONITOR_SCHEDULERS_RECONCILED": "true", + "SYNC_PLAN_JSON": json.dumps( + {"targets": [{"service_name": "firstrade-platform-service"}]} + ), + } + existing_jobs = { + "firstrade-platform-service-probe-scheduler", + "firstrade-platform-service-precheck-scheduler", + "firstrade-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run_gcloud(command: list[str]): + if command[:3] == ["scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[3] in existing_jobs else 1, + ) + if command[:3] == ["scheduler", "jobs", "delete"]: + deleted_jobs.append(command[3]) + return _completed(command) + raise AssertionError(f"Unexpected gcloud command: {command}") + + reconciler.cleanup_legacy_scheduler_jobs(env, run_gcloud=fake_run_gcloud) + + self.assertNotIn("firstrade-monitor-dispatcher-scheduler", deleted_jobs) + + def test_cleanup_keeps_dispatcher_for_multi_target_plan(self): + env = { + "GCP_PROJECT_ID": "firstradequant", + "CLOUD_RUN_SERVICE": "firstrade-platform-service", + "CLOUD_RUN_REGION": "us-central1", + "CLOUD_SCHEDULER_LOCATION": "us-central1", + "DIRECT_MONITOR_MIGRATION_COMPLETE": "true", + "SYNC_PLAN_JSON": json.dumps( + { + "targets": [ + {"service_name": "firstrade-platform-service"}, + {"service_name": "firstrade-secondary-service"}, + ] + } + ), + } + existing_jobs = { + "firstrade-platform-service-probe-scheduler", + "firstrade-platform-service-precheck-scheduler", + "firstrade-monitor-dispatcher-scheduler", + } + deleted_jobs: list[str] = [] + + def fake_run_gcloud(command: list[str]): + if command[:3] == ["scheduler", "jobs", "describe"]: + return _completed( + command, + returncode=0 if command[3] in existing_jobs else 1, + ) + if command[:3] == ["scheduler", "jobs", "delete"]: + deleted_jobs.append(command[3]) + return _completed(command) + raise AssertionError(f"Unexpected gcloud command: {command}") + + reconciler.cleanup_legacy_scheduler_jobs(env, run_gcloud=fake_run_gcloud) + + self.assertNotIn("firstrade-monitor-dispatcher-scheduler", deleted_jobs) if __name__ == "__main__": diff --git a/tests/test_runtime_broker_adapters.py b/tests/test_runtime_broker_adapters.py index e8aaba3..7c2ac0e 100644 --- a/tests/test_runtime_broker_adapters.py +++ b/tests/test_runtime_broker_adapters.py @@ -118,6 +118,27 @@ def get_positions(self, _account): assert portfolio.metadata["total_equity_source"] == "cash_plus_positions" +def test_cash_only_portfolio_uses_lower_reported_buying_power(): + class UnsettledCashClient(FakeClient): + def get_balances(self, _account): + return {"cash_balance": "$80.00", "non_margin_buying_power": "$60.00"} + + def get_positions(self, _account): + return {"items": []} + + adapters = build_runtime_broker_adapters( + client=UnsettledCashClient(), + account="12345678", + strategy_symbols=("SPY",), + cash_only_execution=True, + ) + + portfolio = adapters.build_portfolio_port().get_portfolio_snapshot() + + assert portfolio.cash_balance == 80.0 + assert portfolio.buying_power == 60.0 + + def test_price_series_appends_live_quote_when_history_lags_today(): adapters = build_runtime_broker_adapters( client=FakeClient( diff --git a/tests/test_sync_cloud_run_env_workflow.py b/tests/test_sync_cloud_run_env_workflow.py index 39f14f5..b4ae6bd 100644 --- a/tests/test_sync_cloud_run_env_workflow.py +++ b/tests/test_sync_cloud_run_env_workflow.py @@ -145,6 +145,12 @@ def test_sync_cloud_run_env_workflow_uses_sync_plan_script(): assert "add_optional_env " not in workflow assert "requires_snapshot_artifacts=" not in workflow assert "Resolve selected strategy runtime requirements" not in workflow + assert "Cloud Run env sync currently supports exactly one target" not in workflow + assert workflow.count("matching_targets = [") >= 5 + assert workflow.count("if len(matching_targets) != 1:") >= 5 + assert workflow.index("name: Validate env sync inputs") < workflow.index( + "name: Sync Cloud Run environment" + ) def test_sync_cloud_run_env_workflow_syncs_scheduler_from_sync_plan(): @@ -156,6 +162,10 @@ def test_sync_cloud_run_env_workflow_syncs_scheduler_from_sync_plan(): assert "MONITOR_DISPATCH_TARGETS_JSON=${monitor_targets_json}" in workflow assert 'scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"' in workflow assert 'plan = json.loads(os.environ["SYNC_PLAN_JSON"])' in workflow + assert ( + 'str(candidate.get("service_name") or "").strip() == os.environ["CLOUD_RUN_SERVICE"]' + in workflow + ) assert 'scheduler = target.get("scheduler") or {}' in workflow assert 'print(str(scheduler.get("timezone") or "America/New_York").strip())' in workflow assert 'scheduler.get("main_time") or os.environ.get("CLOUD_SCHEDULER_MAIN_TIME"' in workflow @@ -168,12 +178,43 @@ def test_sync_cloud_run_env_workflow_syncs_scheduler_from_sync_plan(): assert 'print(" ".join([*time_fields, *current_fields[2:]]))' in workflow assert 'gcloud scheduler jobs update http "${job_name}"' in workflow assert 'gcloud scheduler jobs create http "${job_name}"' in workflow + assert 'runtime_target_enabled="${scheduler_config[4]}"' in workflow + assert 'desired_probe_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${probe_time}" python - <<' in workflow + assert 'desired_precheck_schedule="$(CURRENT_SCHEDULE="${desired_schedule}" SCHEDULE_TIME="${precheck_time}" python - <<' in workflow + assert 'probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler"' in workflow + assert 'probe_uri="${service_url}/probe"' in workflow + assert 'precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler"' in workflow + assert 'precheck_uri="${service_url}/dry-run"' in workflow + assert 'managed_scheduler_jobs=("${job_name}")' in workflow + assert 'if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then' in workflow + assert 'managed_scheduler_jobs+=("${probe_job_name}" "${precheck_job_name}")' in workflow + assert "id: scheduler_sync" in workflow assert 'monitor_job_name="firstrade-monitor-dispatcher-scheduler"' in workflow assert 'monitor_uri="${service_url}/monitor-dispatch"' in workflow - assert 'invoke_bridge_jobs=(' in workflow - assert '"${CLOUD_RUN_SERVICE}-probe-scheduler|${service_url}/probe"' in workflow - assert '"${CLOUD_RUN_SERVICE}-precheck-scheduler|${service_url}/dry-run"' in workflow - assert 'Creating invoke-bridge Cloud Scheduler job ${bridge_job} at ${bridge_uri}.' in workflow + assert '--schedule="*/5 * * * *"' in workflow + assert "invoke_bridge_jobs=(" in workflow + assert '--schedule="0 0 1 1 *"' in workflow + assert 'echo "direct_monitors_reconciled=true" >> "${GITHUB_OUTPUT}"' in workflow + assert ( + "DIRECT_MONITOR_SCHEDULERS_RECONCILED: " + "${{ steps.scheduler_sync.outputs.direct_monitors_reconciled }}" + ) in workflow + assert 'gcloud scheduler jobs resume "${managed_job_name}"' in workflow + assert 'gcloud scheduler jobs pause "${managed_job_name}"' in workflow + assert ( + "DIRECT_MONITOR_MIGRATION_COMPLETE: " + "${{ vars.DIRECT_MONITOR_MIGRATION_COMPLETE }}" + ) in workflow + assert 'DIRECT_MONITOR_MIGRATION_COMPLETE: "true"' not in workflow assert '--schedule="${desired_schedule}"' in workflow assert '--time-zone="${market_timezone}"' in workflow assert "legacy_jobs=(" not in workflow + direct_gate = workflow.index( + 'if [ "${DIRECT_MONITOR_MIGRATION_COMPLETE:-}" = "true" ]; then' + ) + assert direct_gate < workflow.index( + 'desired_probe_schedule="$(CURRENT_SCHEDULE="${desired_schedule}"' + ) + assert direct_gate < workflow.index( + 'desired_precheck_schedule="$(CURRENT_SCHEDULE="${desired_schedule}"' + )