diff --git a/.github/workflows/invoke-cloud-run.yml b/.github/workflows/invoke-cloud-run.yml
new file mode 100644
index 0000000..abee9d6
--- /dev/null
+++ b/.github/workflows/invoke-cloud-run.yml
@@ -0,0 +1,301 @@
+name: Invoke Cloud Run
+
+on:
+  workflow_dispatch:
+    inputs:
+      path:
+        description: "HTTP path to call"
+        required: false
+        default: "/dry-run"
+        type: string
+      allow_live_execution:
+        description: "Allow calling /run live execution entrypoint"
+        required: false
+        default: false
+        type: boolean
+
+env:
+  GCP_PROJECT_ID: firstradequant
+  GCP_WORKLOAD_IDENTITY_PROVIDER: projects/1088907247379/locations/global/workloadIdentityPools/github-actions/providers/github-main
+  GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: firstrade-platform-deploy@firstradequant.iam.gserviceaccount.com
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: false
+
+jobs:
+  invoke:
+    name: Invoke Firstrade Cloud Run
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
+      CLOUD_RUN_SERVICE: ${{ vars.CLOUD_RUN_SERVICE }}
+      CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }}
+    steps:
+      - name: Validate inputs
+        run: |
+          set -euo pipefail
+
+          if [ -z "${CLOUD_RUN_REGION:-}" ] || [ -z "${CLOUD_RUN_SERVICE:-}" ]; then
+            echo "CLOUD_RUN_REGION and CLOUD_RUN_SERVICE are required repository variables." >&2
+            exit 1
+          fi
+
+      - name: Authenticate to Google Cloud
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+
+      - name: Set up gcloud
+        uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: ${{ env.GCP_PROJECT_ID }}
+          version: ">= 416.0.0"
+
+      - name: Resolve service URL
+        id: service
+        run: |
+          set -euo pipefail
+
+          raw_path="${{ inputs.path }}"
+          if [ -z "${raw_path}" ]; then
+            raw_path="/dry-run"
+          fi
+          if [[ "${raw_path}" != /* ]]; then
+            raw_path="/${raw_path}"
+          fi
+          if [ "${raw_path}" = "/run" ] && [ "${{ inputs.allow_live_execution }}" != "true" ]; then
+            echo "Calling ${raw_path} can trigger the live execution entrypoint. Re-run with allow_live_execution=true if this is intentional." >&2
+            exit 1
+          fi
+
+          service_json="$(
+            gcloud run services describe "${CLOUD_RUN_SERVICE}" \
+              --region "${CLOUD_RUN_REGION}" \
+              --format=json
+          )"
+          service_url="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
+          import json
+          import os
+
+          service = json.loads(os.environ["SERVICE_JSON"])
+          print((service.get("status") or {}).get("url") or "")
+          PY
+          )"
+          if [ -z "${service_url}" ]; then
+            echo "Unable to resolve Cloud Run service URL." >&2
+            exit 1
+          fi
+
+          service_ingress="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
+          import json
+          import os
+
+          service = json.loads(os.environ["SERVICE_JSON"])
+          annotations = (service.get("metadata") or {}).get("annotations") or {}
+          print(annotations.get("run.googleapis.com/ingress-status") or annotations.get("run.googleapis.com/ingress") or "")
+          PY
+          )"
+          latest_ready_revision="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
+          import json
+          import os
+
+          service = json.loads(os.environ["SERVICE_JSON"])
+          print((service.get("status") or {}).get("latestReadyRevisionName") or "")
+          PY
+          )"
+          deployed_commit="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
+          import json
+          import os
+
+          service = json.loads(os.environ["SERVICE_JSON"])
+          template = ((service.get("spec") or {}).get("template") or {}).get("metadata") or {}
+          print((template.get("labels") or {}).get("commit-sha") or "")
+          PY
+          )"
+
+          invoke_method="direct"
+          scheduler_job=""
+          scheduler_location=""
+          scheduler_expected_path=""
+          if [ "${service_ingress}" = "internal" ]; then
+            scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"
+            case "${raw_path}" in
+              /run)
+                scheduler_job="${CLOUD_RUN_SERVICE}-scheduler"
+                scheduler_expected_path="/run"
+                ;;
+              /probe)
+                scheduler_job="${CLOUD_RUN_SERVICE}-probe-scheduler"
+                scheduler_expected_path="/probe"
+                ;;
+              /dry-run)
+                scheduler_job="${CLOUD_RUN_SERVICE}-precheck-scheduler"
+                scheduler_expected_path="/dry-run"
+                ;;
+              *)
+                echo "Cloud Run service ${CLOUD_RUN_SERVICE} has internal ingress, so GitHub-hosted runners cannot curl ${raw_path} directly." >&2
+                echo "Use one of the scheduler-backed paths: /run, /probe, /dry-run." >&2
+                exit 1
+                ;;
+            esac
+
+            scheduler_uri="$(
+              gcloud scheduler jobs describe "${scheduler_job}" \
+                --location="${scheduler_location}" \
+                --format='value(httpTarget.uri)' 2>/dev/null || true
+            )"
+            if [ -z "${scheduler_uri}" ]; then
+              echo "Cloud Scheduler job ${scheduler_job} was not found in ${scheduler_location}." >&2
+              exit 1
+            fi
+            scheduler_path="$(SCHEDULER_URI="${scheduler_uri}" python3 - <<'PY'
+          import os
+          from urllib.parse import urlparse
+
+          def normalize(path: str) -> str:
+              clean = (path or "/").rstrip("/")
+              return clean or "/"
+
+          print(normalize(urlparse(os.environ["SCHEDULER_URI"]).path))
+          PY
+            )"
+            requested_path="$(RAW_PATH="${scheduler_expected_path}" python3 - <<'PY'
+          import os
+
+          clean = (os.environ["RAW_PATH"] or "/").rstrip("/")
+          print(clean or "/")
+          PY
+            )"
+            if [ "${scheduler_path}" != "${requested_path}" ]; then
+              echo "Cloud Scheduler job ${scheduler_job} targets ${scheduler_uri}, not ${scheduler_expected_path}." >&2
+              exit 1
+            fi
+            invoke_method="scheduler"
+          fi
+
+          echo "Cloud Run service: ${CLOUD_RUN_SERVICE}"
+          echo "Cloud Run region: ${CLOUD_RUN_REGION}"
+          echo "Cloud Run URL: ${service_url}"
+          echo "Cloud Run ingress: ${service_ingress:-<empty>}"
+          echo "Latest ready revision: ${latest_ready_revision:-<empty>}"
+          echo "Deployed commit: ${deployed_commit:-<empty>}"
+          echo "Invoke method: ${invoke_method}"
+          if [ -n "${scheduler_job}" ]; then
+            echo "Cloud Scheduler job: ${scheduler_job}"
+            echo "Cloud Scheduler location: ${scheduler_location}"
+          fi
+
+          {
+            echo "url=${service_url}"
+            echo "path=${raw_path}"
+            echo "invoke_method=${invoke_method}"
+            echo "scheduler_job=${scheduler_job}"
+            echo "scheduler_location=${scheduler_location}"
+          } >> "$GITHUB_OUTPUT"
+
+      - name: Authenticate for service invocation
+        if: steps.service.outputs.invoke_method == 'direct'
+        id: invoke-auth
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: ${{ steps.service.outputs.url }}
+          id_token_include_email: true
+
+      - name: Invoke service
+        if: steps.service.outputs.invoke_method == 'direct'
+        run: |
+          set -euo pipefail
+
+          curl --fail-with-body --show-error --silent \
+            --request POST \
+            --header "Authorization: Bearer ${{ steps.invoke-auth.outputs.id_token }}" \
+            "${{ steps.service.outputs.url }}${{ steps.service.outputs.path }}"
+
+      - name: Invoke internal service through Cloud Scheduler
+        if: steps.service.outputs.invoke_method == 'scheduler'
+        run: |
+          set -euo pipefail
+
+          started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+          scheduler_job="${{ steps.service.outputs.scheduler_job }}"
+          scheduler_location="${{ steps.service.outputs.scheduler_location }}"
+
+          echo "Triggering ${scheduler_job} at ${started_at}."
+          gcloud scheduler jobs run "${scheduler_job}" \
+            --location="${scheduler_location}" \
+            --quiet
+
+          deadline=$((SECONDS + 180))
+          while true; do
+            job_json="$(
+              gcloud scheduler jobs describe "${scheduler_job}" \
+                --location="${scheduler_location}" \
+                --format=json
+            )"
+            attempt_seen="$(JOB_JSON="${job_json}" STARTED_AT="${started_at}" python3 - <<'PY'
+          import datetime as dt
+          import json
+          import os
+
+          def parse_timestamp(value: str) -> dt.datetime | None:
+              if not value:
+                  return None
+              text = value.replace("Z", "+00:00")
+              return dt.datetime.fromisoformat(text)
+
+          job = json.loads(os.environ["JOB_JSON"])
+          last_attempt = parse_timestamp(job.get("lastAttemptTime") or "")
+          started_at = parse_timestamp(os.environ["STARTED_AT"])
+          print("true" if last_attempt and started_at and last_attempt >= started_at else "false")
+          PY
+            )"
+            status_code="$(JOB_JSON="${job_json}" python3 - <<'PY'
+          import json
+          import os
+
+          status = (json.loads(os.environ["JOB_JSON"]).get("status") or {})
+          print(status.get("code") or "")
+          PY
+            )"
+            status_message="$(JOB_JSON="${job_json}" python3 - <<'PY'
+          import json
+          import os
+
+          status = (json.loads(os.environ["JOB_JSON"]).get("status") or {})
+          print(status.get("message") or "")
+          PY
+            )"
+            last_attempt_time="$(JOB_JSON="${job_json}" python3 - <<'PY'
+          import json
+          import os
+
+          print(json.loads(os.environ["JOB_JSON"]).get("lastAttemptTime") or "")
+          PY
+            )"
+
+            if [ "${attempt_seen}" = "true" ]; then
+              if [ -n "${status_code}" ] && [ "${status_code}" != "0" ]; then
+                echo "Cloud Scheduler job ${scheduler_job} failed with status ${status_code}: ${status_message}" >&2
+                exit 1
+              fi
+              echo "Cloud Scheduler job ${scheduler_job} ran at ${last_attempt_time}."
+              break
+            fi
+
+            if [ "${SECONDS}" -ge "${deadline}" ]; then
+              echo "Timed out waiting for Cloud Scheduler job ${scheduler_job} to record a new attempt." >&2
+              exit 1
+            fi
+
+            echo "Waiting for Cloud Scheduler job ${scheduler_job} attempt..."
+            sleep 10
+          done