Guard explicit IBKR target creation

.github/workflows/manual-strategy-switch.yml

@@ -102,12 +102,13 @@ on:
           - enabled
           - disabled
       service_targets_mode:
-        description: "auto patches IBKR CLOUD_RUN_SERVICE_TARGETS_JSON when it exists."
+        description: "auto updates an existing IBKR target; allow_create explicitly permits a new target."
         required: true
         type: choice
         default: auto
         options:
           - auto
+          - allow_create
           - off
       apply:
         description: "Actually write GitHub variables. false = preview only."
@@ -208,7 +209,7 @@ jobs:
           fi
 
           if [ "${PLATFORM}" = "ibkr" ] \
-            && [ "${SERVICE_TARGETS_MODE}" = "auto" ] \
+            && [ "${SERVICE_TARGETS_MODE}" != "off" ] \
             && [ -z "${GH_TOKEN:-}" ]; then
             echo "RUNTIME_SETTINGS_GH_TOKEN is required for IBKR service-target preview because the workflow must read and patch CLOUD_RUN_SERVICE_TARGETS_JSON." >&2
             exit 2
@@ -222,20 +223,27 @@ jobs:
           echo "repository=${repo}" >> "$GITHUB_OUTPUT"
 
       - name: Fetch existing service targets
-        if: env.SERVICE_TARGETS_MODE == 'auto' && env.PLATFORM == 'ibkr'
+        if: env.SERVICE_TARGETS_MODE != 'off' && env.PLATFORM == 'ibkr'
         env:
           TARGET_REPOSITORY: ${{ steps.platform.outputs.repository }}
         run: |
           set -euo pipefail
           output_file="${RUNNER_TEMP}/existing-service-targets.json"
-          python - <<'PY' "${TARGET_REPOSITORY}" "${output_file}"
+          target_environment=""
+          if [ "${VARIABLE_SCOPE}" = "environment" ]; then
+            target_environment="${GITHUB_ENVIRONMENT_NAME:-${TARGET_NAME}}"
+          fi
+          python - <<'PY' "${TARGET_REPOSITORY}" "${output_file}" "${target_environment}"
           import json
           import subprocess
           import sys
 
-          repo, output_path = sys.argv[1], sys.argv[2]
+          repo, output_path, environment = sys.argv[1:4]
+          command = ["gh", "variable", "list", "--repo", repo, "--json", "name,value"]
+          if environment:
+              command.extend(["--env", environment])
           raw = subprocess.check_output(
-              ["gh", "variable", "list", "--repo", repo, "--json", "name,value"],
+              command,
               text=True,
           )
           variables = json.loads(raw)
@@ -245,7 +253,8 @@ jobs:
                   value = str(item.get("value") or "").strip()
                   break
           if not value:
-              open(output_path, "w", encoding="utf-8").close()
+              with open(output_path, "w", encoding="utf-8") as handle:
+                  handle.write("{}")
               raise SystemExit(0)
           try:
               payload = json.loads(value)
@@ -305,9 +314,12 @@ jobs:
           if [ -n "${INCOME_LAYER_MAX_RATIO:-}" ]; then
             args+=(--income-layer-max-ratio "${INCOME_LAYER_MAX_RATIO}")
           fi
-          if [ -s "${EXISTING_SERVICE_TARGETS_JSON_FILE:-}" ]; then
+          if [ -f "${EXISTING_SERVICE_TARGETS_JSON_FILE:-}" ]; then
             args+=(--existing-service-targets-json-file "${EXISTING_SERVICE_TARGETS_JSON_FILE}")
           fi
+          if [ "${SERVICE_TARGETS_MODE}" = "allow_create" ]; then
+            args+=(--allow-create-service-target)
+          fi
           python3 python/scripts/build_runtime_switch.py "${args[@]}"
           python3 python/scripts/runtime_settings.py validate "${target_file}"
           echo "TARGET_FILE=${target_file}" >> "$GITHUB_ENV"

README.md

@@ -85,7 +85,7 @@ Notes:
 - LongBridge defaults to environment-scoped variables; `target_name=sg` resolves to `longbridge-sg`.
 - Schwab defaults to repository-scoped variables.
 - Firstrade defaults to repository-scoped variables; `target_name=live` uses `firstrade-quant-service` and `account_scope=US`.
-- IBKR patches the selected service/account-scope entry inside `CLOUD_RUN_SERVICE_TARGETS_JSON` when that variable exists, so other IBKR services are preserved.
+- IBKR `service_targets_mode=auto` only patches an existing service/account-scope entry inside `CLOUD_RUN_SERVICE_TARGETS_JSON`, so other services are preserved and an unknown target fails closed. Use `allow_create` only when intentionally provisioning a new target.
 - Cross-repository variable writes and workflow dispatches require a `RUNTIME_SETTINGS_GH_TOKEN` secret in this repository with sufficient target-repository variable/workflow permissions. The workflow does not fall back to the default `github.token` for remote writes.
 - IBKR `service_targets_mode=auto` must read and patch the target repository's `CLOUD_RUN_SERVICE_TARGETS_JSON`, so even preview mode requires `RUNTIME_SETTINGS_GH_TOKEN` for IBKR.
 - The workflow is bound to the `runtime-strategy-switch` GitHub Environment. For a personal system, required reviewers are optional; prefer storing `RUNTIME_SETTINGS_GH_TOKEN` as an Environment secret and rely on preview, confirmation text, and a least-privilege token for day-to-day safety.

python/scripts/build_runtime_switch.py

@@ -802,6 +802,7 @@ def _patch_service_targets(
     mounts_variable: str,
     mounts: list[dict[str, Any]],
     extra_variables: dict[str, Any],
+    allow_create: bool,
 ) -> dict[str, Any]:
     payload = dict(current_payload)
     raw_entries = payload.get("targets") if isinstance(payload.get("targets"), list) else []
@@ -836,6 +837,11 @@ def _patch_service_targets(
             replaced = True
             break
 
+    if not replaced and not allow_create:
+        raise ValueError(
+            "existing IBKR service target was not found; "
+            "use --allow-create-service-target to append a new target"
+        )
     if not replaced:
         entries.append(replacement)
     payload["targets"] = entries
@@ -886,20 +892,21 @@ def build_switch_target(args: argparse.Namespace) -> dict[str, Any]:
         )
     )
 
-    service_targets = _load_json_from_file(
-        args.existing_service_targets_json_file,
-        field_name="existing_service_targets_json_file",
-    )
     top_level_mounts = mounts
     plugin_mounts_variable: str | None = mounts_variable
-    if service_targets:
+    if args.existing_service_targets_json_file:
+        service_targets = _load_json_from_file(
+            args.existing_service_targets_json_file,
+            field_name="existing_service_targets_json_file",
+        )
         patched_service_targets = _patch_service_targets(
             current_payload=service_targets,
             platform=platform,
             runtime_target=runtime_target,
             mounts_variable=mounts_variable,
             mounts=mounts,
             extra_variables=extra_variables,
+            allow_create=args.allow_create_service_target,
         )
         extra_variables = {"CLOUD_RUN_SERVICE_TARGETS_JSON": patched_service_targets}
         top_level_mounts = []
@@ -957,6 +964,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--dca-mode", default="")
     parser.add_argument("--dca-base-investment-usd", default="")
     parser.add_argument("--existing-service-targets-json-file", default="")
+    parser.add_argument("--allow-create-service-target", action="store_true")
     parser.add_argument("--no-platform-dry-run-variable", dest="set_platform_dry_run_variable", action="store_false")
     parser.set_defaults(set_platform_dry_run_variable=True)
     parser.add_argument("--output", default="-", help="output path, or '-' for stdout")

python/tests/test_runtime_settings.py

@@ -92,6 +92,14 @@ def test_manual_switch_platform_choices_cover_supported_platforms(self):
 
         self.assertEqual(set(platform_choices), set(runtime_settings.SUPPORTED_PLATFORMS))
 
+    def test_manual_switch_reads_ibkr_targets_from_selected_environment_scope(self):
+        workflow = (ROOT / ".github/workflows/manual-strategy-switch.yml").read_text(encoding="utf-8")
+
+        assert 'if [ "${VARIABLE_SCOPE}" = "environment" ]; then' in workflow
+        assert 'target_environment="${GITHUB_ENVIRONMENT_NAME:-${TARGET_NAME}}"' in workflow
+        assert 'command.extend(["--env", environment])' in workflow
+        assert 'handle.write("{}")' in workflow
+
     def test_live_candidate_queue_lists_profiles_needing_promotion_review(self):
         catalog = [
             {
@@ -1658,6 +1666,53 @@ def test_build_switch_target_patches_ibkr_service_targets_with_soxl_plugin_mount
             "soxl_soxx_trend_income/plugins/market_regime_control/latest_signal.json",
         )
 
+    def test_build_switch_target_rejects_unknown_ibkr_service_target_by_default(self):
+        path = ROOT / ".pytest_runtime_service_targets_unknown.json"
+        path.write_text('{"targets":[]}', encoding="utf-8")
+        self.addCleanup(lambda: path.unlink(missing_ok=True))
+        parser = build_runtime_switch.build_parser()
+        args = parser.parse_args(
+            [
+                "--platform",
+                "ibkr",
+                "--target-name",
+                "new-account",
+                "--strategy-profile",
+                "tqqq_growth_income",
+                "--existing-service-targets-json-file",
+                str(path),
+            ]
+        )
+
+        with self.assertRaisesRegex(ValueError, "existing IBKR service target was not found"):
+            build_runtime_switch.build_switch_target(args)
+
+    def test_build_switch_target_can_explicitly_append_ibkr_service_target(self):
+        path = ROOT / ".pytest_runtime_service_targets_create.json"
+        path.write_text("{}", encoding="utf-8")
+        self.addCleanup(lambda: path.unlink(missing_ok=True))
+        parser = build_runtime_switch.build_parser()
+        args = parser.parse_args(
+            [
+                "--platform",
+                "ibkr",
+                "--target-name",
+                "new-account",
+                "--strategy-profile",
+                "tqqq_growth_income",
+                "--existing-service-targets-json-file",
+                str(path),
+                "--allow-create-service-target",
+            ]
+        )
+
+        target = build_runtime_switch.build_switch_target(args)
+        assignments = {item.name: item.value for item in runtime_settings.build_assignments(target)}
+        patched = json.loads(assignments["CLOUD_RUN_SERVICE_TARGETS_JSON"])
+
+        self.assertEqual(len(patched["targets"]), 1)
+        self.assertEqual(patched["targets"][0]["runtime_target"]["account_scope"], "new-account")
+
 
 if __name__ == "__main__":
     unittest.main()