diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml
index 5d3cdf7..e5e0869 100644
--- a/.github/workflows/sync-cloud-run-env.yml
+++ b/.github/workflows/sync-cloud-run-env.yml
@@ -563,6 +563,14 @@ jobs:
 
           gcloud "${gcloud_args[@]}"
 
+      - name: Reconcile Cloud Run traffic
+        if: steps.env_sync_config.outputs.enabled == 'true'
+        env:
+          SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }}
+        run: |
+          set -euo pipefail
+          python scripts/reconcile_cloud_runtime.py traffic
+
       - name: Sync Cloud Scheduler schedule
         if: steps.env_sync_config.outputs.enabled == 'true'
         env:
@@ -772,25 +780,11 @@ jobs:
             fi
           done
 
-          legacy_jobs=(
-            "${CLOUD_RUN_SERVICE}-session-check-scheduler"
-          )
-          if [[ "${CLOUD_RUN_SERVICE}" == *-service ]]; then
-            legacy_jobs+=(
-              "${CLOUD_RUN_SERVICE%-service}-session-check-scheduler"
-            )
-          fi
-          for legacy_job in "${legacy_jobs[@]}"; do
-            if gcloud scheduler jobs describe "${legacy_job}" \
-              --project="${GCP_PROJECT_ID}" \
-              --location="${scheduler_location}" >/dev/null 2>&1; then
-              echo "Deleting legacy Cloud Scheduler job ${legacy_job}; session checks are handled elsewhere."
-              gcloud scheduler jobs delete "${legacy_job}" \
-                --project="${GCP_PROJECT_ID}" \
-                --location="${scheduler_location}" \
-                --quiet
-            fi
-          done
+      - name: Reconcile legacy Cloud Scheduler jobs
+        if: steps.env_sync_config.outputs.enabled == 'true'
+        run: |
+          set -euo pipefail
+          python scripts/reconcile_cloud_runtime.py scheduler-cleanup
 
       - name: Clean up old Cloud Run revisions and images
         if: steps.deploy_config.outputs.enabled == 'true'
diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py
new file mode 100644
index 0000000..8301a25
--- /dev/null
+++ b/scripts/reconcile_cloud_runtime.py
@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+"""Reconcile Cloud Run runtime state after deploy/env sync.
+
+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.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from typing import Any, Callable, Mapping
+
+
+RunGcloud = Callable[[list[str]], subprocess.CompletedProcess[str]]
+
+
+@dataclass(frozen=True)
+class RuntimeContext:
+    project_id: str
+    service_name: str
+    region: str
+    scheduler_location: str
+
+def _parse_sync_plan(raw: str) -> dict[str, Any]:
+    raw = raw.strip()
+    if not raw:
+        return {}
+    payload = json.loads(raw)
+    if isinstance(payload, dict):
+        return payload
+    if isinstance(payload, list):
+        return {"targets": payload}
+    raise ValueError("SYNC_PLAN_JSON must be a JSON object or array")
+
+
+def _first_target(plan: Mapping[str, Any]) -> Mapping[str, Any]:
+    targets = plan.get("targets") or []
+    if not isinstance(targets, list) or not targets:
+        return {}
+    first = targets[0]
+    return first if isinstance(first, Mapping) else {}
+
+
+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)
+
+    service_name = (
+        str(env.get("CLOUD_RUN_SERVICE", "") or "").strip()
+        or str(target.get("service_name") or "").strip()
+        or str(target.get("service") or "").strip()
+        or str(target.get("cloud_run_service") or "").strip()
+    )
+    region = (
+        str(env.get("CLOUD_RUN_REGION", "") or "").strip()
+        or str(target.get("region") or "").strip()
+    )
+    project_id = str(env.get("GCP_PROJECT_ID", "") or "").strip()
+    scheduler_location = (
+        str(env.get("CLOUD_SCHEDULER_LOCATION", "") or "").strip()
+        or region
+    )
+
+    missing = [
+        name
+        for name, value in (
+            ("GCP_PROJECT_ID", project_id),
+            ("CLOUD_RUN_SERVICE", service_name),
+            ("CLOUD_RUN_REGION", region),
+            ("CLOUD_SCHEDULER_LOCATION", scheduler_location),
+        )
+        if not value
+    ]
+    if missing:
+        raise ValueError(f"Missing required runtime context: {', '.join(missing)}")
+
+    return (
+        RuntimeContext(
+            project_id=project_id,
+            service_name=service_name,
+            region=region,
+            scheduler_location=scheduler_location,
+        ),
+        plan,
+    )
+
+
+def _run_gcloud(command: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["gcloud", *command],
+        check=False,
+        text=True,
+        capture_output=True,
+    )
+
+
+def _run_gcloud_json(command: list[str], run_gcloud: RunGcloud = _run_gcloud) -> dict[str, Any]:
+    result = run_gcloud(command)
+    if result.returncode != 0:
+        stderr = (result.stderr or "").strip()
+        raise RuntimeError(
+            f"gcloud {' '.join(command)} failed with exit code {result.returncode}"
+            + (f": {stderr}" if stderr else "")
+        )
+    raw = (result.stdout or "").strip()
+    return json.loads(raw) if raw else {}
+
+
+def _run_gcloud_ok(command: list[str], run_gcloud: RunGcloud = _run_gcloud) -> None:
+    result = run_gcloud(command)
+    if result.returncode != 0:
+        stderr = (result.stderr or "").strip()
+        raise RuntimeError(
+            f"gcloud {' '.join(command)} failed with exit code {result.returncode}"
+            + (f": {stderr}" if stderr else "")
+        )
+
+
+def _describe_service(ctx: RuntimeContext, run_gcloud: RunGcloud = _run_gcloud) -> dict[str, Any]:
+    return _run_gcloud_json(
+        [
+            "run",
+            "services",
+            "describe",
+            ctx.service_name,
+            "--project",
+            ctx.project_id,
+            "--region",
+            ctx.region,
+            "--format=json",
+        ],
+        run_gcloud=run_gcloud,
+    )
+
+
+def _describe_revision(
+    ctx: RuntimeContext,
+    revision_name: str,
+    run_gcloud: RunGcloud = _run_gcloud,
+) -> dict[str, Any]:
+    return _run_gcloud_json(
+        [
+            "run",
+            "revisions",
+            "describe",
+            revision_name,
+            "--project",
+            ctx.project_id,
+            "--region",
+            ctx.region,
+            "--format=json",
+        ],
+        run_gcloud=run_gcloud,
+    )
+
+
+def _wait_for_latest_ready_revision(
+    ctx: RuntimeContext,
+    *,
+    run_gcloud: RunGcloud = _run_gcloud,
+    timeout_seconds: int = 1800,
+    poll_seconds: int = 10,
+) -> tuple[dict[str, Any], str]:
+    deadline = time.monotonic() + timeout_seconds
+    last_revision = ""
+    while True:
+        service = _describe_service(ctx, run_gcloud=run_gcloud)
+        latest_revision = str(
+            service.get("status", {}).get("latestReadyRevisionName") or ""
+        ).strip()
+        if latest_revision:
+            return service, latest_revision
+        last_revision = latest_revision
+        if time.monotonic() >= deadline:
+            raise RuntimeError(
+                f"Timed out waiting for latest ready revision on {ctx.service_name}"
+                + (f"; last seen: {last_revision or '<none>'}" if last_revision else "")
+            )
+        time.sleep(poll_seconds)
+
+
+def _traffic_is_reconciled(service: Mapping[str, Any], latest_revision: str) -> bool:
+    traffic = service.get("status", {}).get("traffic", []) or []
+    positive = [
+        entry
+        for entry in traffic
+        if isinstance(entry, Mapping) and int(entry.get("percent", 0) or 0) > 0
+    ]
+    if len(positive) != 1:
+        return False
+    entry = positive[0]
+    return int(entry.get("percent", 0) or 0) == 100 and (
+        entry.get("latestRevision") is True or str(entry.get("revisionName") or "") == latest_revision
+    )
+
+
+def reconcile_traffic(
+    env: Mapping[str, str] = os.environ,
+    *,
+    run_gcloud: RunGcloud = _run_gcloud,
+) -> None:
+    ctx, _plan = _resolve_context(env)
+    target_sha = str(env.get("GITHUB_SHA", "") or "").strip()
+    if not target_sha:
+        raise ValueError("GITHUB_SHA is required for traffic reconciliation")
+
+    service, latest_revision = _wait_for_latest_ready_revision(ctx, run_gcloud=run_gcloud)
+    revision = _describe_revision(ctx, latest_revision, run_gcloud=run_gcloud)
+    revision_sha = str(
+        revision.get("metadata", {}).get("labels", {}).get("commit-sha") or ""
+    ).strip()
+    if revision_sha != target_sha:
+        raise RuntimeError(
+            f"Latest ready revision {latest_revision} on {ctx.service_name} has commit-sha "
+            f"{revision_sha or '<missing>'}, expected {target_sha}"
+        )
+
+    if not _traffic_is_reconciled(service, latest_revision):
+        _run_gcloud_ok(
+            [
+                "run",
+                "services",
+                "update-traffic",
+                ctx.service_name,
+                "--project",
+                ctx.project_id,
+                "--region",
+                ctx.region,
+                "--to-revisions",
+                f"{latest_revision}=100",
+                "--quiet",
+            ],
+            run_gcloud=run_gcloud,
+        )
+
+    deadline = time.monotonic() + 300
+    while True:
+        service = _describe_service(ctx, run_gcloud=run_gcloud)
+        if _traffic_is_reconciled(service, latest_revision):
+            print(
+                f"Cloud Run service {ctx.service_name} traffic reconciled to {latest_revision}."
+            )
+            return
+        if time.monotonic() >= deadline:
+            raise RuntimeError(
+                f"Timed out waiting for {ctx.service_name} traffic to converge on {latest_revision}"
+            )
+        time.sleep(5)
+
+
+def _legacy_session_check_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")
+    seen: list[str] = []
+    for candidate in candidates:
+        if candidate not in seen:
+            seen.append(candidate)
+    return seen
+
+
+def cleanup_legacy_scheduler_jobs(
+    env: Mapping[str, str] = os.environ,
+    *,
+    run_gcloud: RunGcloud = _run_gcloud,
+) -> None:
+    ctx, _plan = _resolve_context(env)
+    deleted: list[str] = []
+    for job_name in _legacy_session_check_jobs(ctx.service_name):
+        result = run_gcloud(
+            [
+                "scheduler",
+                "jobs",
+                "describe",
+                job_name,
+                "--project",
+                ctx.project_id,
+                "--location",
+                ctx.scheduler_location,
+            ]
+        )
+        if result.returncode != 0:
+            continue
+        _run_gcloud_ok(
+            [
+                "scheduler",
+                "jobs",
+                "delete",
+                job_name,
+                "--project",
+                ctx.project_id,
+                "--location",
+                ctx.scheduler_location,
+                "--quiet",
+            ],
+            run_gcloud=run_gcloud,
+        )
+        deleted.append(job_name)
+
+    if deleted:
+        print(
+            "Deleted legacy Cloud Scheduler job(s): " + ", ".join(deleted)
+        )
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    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"
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        if args.command == "traffic":
+            reconcile_traffic()
+        elif args.command == "scheduler-cleanup":
+            cleanup_legacy_scheduler_jobs()
+        else:  # pragma: no cover - argparse enforces subcommands
+            raise ValueError(f"Unknown command: {args.command}")
+    except (ValueError, RuntimeError, json.JSONDecodeError) as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py
new file mode 100644
index 0000000..c456659
--- /dev/null
+++ b/tests/test_reconcile_cloud_runtime.py
@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import unittest
+
+from scripts import reconcile_cloud_runtime as reconciler
+
+
+def _completed(command: list[str], stdout: str = "", stderr: str = "", returncode: int = 0):
+    return subprocess.CompletedProcess(command, returncode, stdout=stdout, stderr=stderr)
+
+
+class ReconcileCloudRuntimeTest(unittest.TestCase):
+    def test_resolve_context_prefers_cloud_run_service_and_first_target(self):
+        ctx, plan = reconciler._resolve_context(
+            {
+                "GCP_PROJECT_ID": "firstradequant",
+                "CLOUD_RUN_SERVICE": "firstrade-platform-service",
+                "CLOUD_RUN_REGION": "us-central1",
+                "SYNC_PLAN_JSON": json.dumps(
+                    {
+                        "targets": [
+                            {
+                                "service_name": "firstrade-platform-service",
+                                "region": "asia-east1",
+                            }
+                        ]
+                    }
+                ),
+            }
+        )
+
+        self.assertEqual(ctx.project_id, "firstradequant")
+        self.assertEqual(ctx.service_name, "firstrade-platform-service")
+        self.assertEqual(ctx.region, "us-central1")
+        self.assertEqual(ctx.scheduler_location, "us-central1")
+        self.assertEqual(plan["targets"][0]["region"], "asia-east1")
+
+    def test_reconcile_traffic_updates_latest_ready_revision_and_checks_commit_sha(self):
+        env = {
+            "GCP_PROJECT_ID": "firstradequant",
+            "CLOUD_RUN_SERVICE": "firstrade-platform-service",
+            "CLOUD_RUN_REGION": "us-central1",
+            "GITHUB_SHA": "abc123",
+            "SYNC_PLAN_JSON": json.dumps(
+                {"targets": [{"service_name": "firstrade-platform-service"}]}
+            ),
+        }
+        service_state = {
+            "status": {
+                "latestReadyRevisionName": "firstrade-platform-service-00002",
+                "traffic": [{"revisionName": "firstrade-platform-service-00001", "percent": 100}],
+            }
+        }
+        revision_state = {
+            "metadata": {"labels": {"commit-sha": "abc123"}},
+        }
+        calls: list[list[str]] = []
+
+        def fake_run_gcloud(command: list[str]):
+            calls.append(command)
+            if command[:4] == ["run", "services", "describe", "firstrade-platform-service"]:
+                return _completed(command, stdout=json.dumps(service_state))
+            if command[:4] == ["run", "revisions", "describe", "firstrade-platform-service-00002"]:
+                return _completed(command, stdout=json.dumps(revision_state))
+            if command[:4] == ["run", "services", "update-traffic", "firstrade-platform-service"]:
+                service_state["status"]["traffic"] = [
+                    {"revisionName": "firstrade-platform-service-00002", "percent": 100}
+                ]
+                return _completed(command, stdout="")
+            raise AssertionError(f"Unexpected gcloud command: {command}")
+
+        reconciler.reconcile_traffic(env, run_gcloud=fake_run_gcloud)
+
+        self.assertEqual(
+            calls,
+            [
+                ["run", "services", "describe", "firstrade-platform-service", "--project", "firstradequant", "--region", "us-central1", "--format=json"],
+                ["run", "revisions", "describe", "firstrade-platform-service-00002", "--project", "firstradequant", "--region", "us-central1", "--format=json"],
+                ["run", "services", "update-traffic", "firstrade-platform-service", "--project", "firstradequant", "--region", "us-central1", "--to-revisions", "firstrade-platform-service-00002=100", "--quiet"],
+                ["run", "services", "describe", "firstrade-platform-service", "--project", "firstradequant", "--region", "us-central1", "--format=json"],
+            ],
+        )
+
+    def test_reconcile_traffic_rejects_revision_with_wrong_commit_sha(self):
+        env = {
+            "GCP_PROJECT_ID": "firstradequant",
+            "CLOUD_RUN_SERVICE": "firstrade-platform-service",
+            "CLOUD_RUN_REGION": "us-central1",
+            "GITHUB_SHA": "abc123",
+            "SYNC_PLAN_JSON": json.dumps(
+                {"targets": [{"service_name": "firstrade-platform-service"}]}
+            ),
+        }
+        service_state = {
+            "status": {
+                "latestReadyRevisionName": "firstrade-platform-service-00002",
+                "traffic": [{"revisionName": "firstrade-platform-service-00002", "percent": 100}],
+            }
+        }
+        revision_state = {
+            "metadata": {"labels": {"commit-sha": "deadbeef"}},
+        }
+
+        def fake_run_gcloud(command: list[str]):
+            if command[:4] == ["run", "services", "describe", "firstrade-platform-service"]:
+                return _completed(command, stdout=json.dumps(service_state))
+            if command[:4] == ["run", "revisions", "describe", "firstrade-platform-service-00002"]:
+                return _completed(command, stdout=json.dumps(revision_state))
+            raise AssertionError(f"Unexpected gcloud command: {command}")
+
+        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):
+        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"}]}
+            ),
+        }
+        calls: list[list[str]] = []
+        existing_jobs = {
+            "firstrade-platform-service-session-check-scheduler",
+            "firstrade-platform-session-check-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")
+                return _completed(command, stdout="")
+            raise AssertionError(f"Unexpected gcloud command: {command}")
+
+        reconciler.cleanup_legacy_scheduler_jobs(env, run_gcloud=fake_run_gcloud)
+
+        self.assertIn(
+            ["scheduler", "jobs", "describe", "firstrade-platform-service-session-check-scheduler", "--project", "firstradequant", "--location", "us-central1"],
+            calls,
+        )
+        self.assertIn(
+            ["scheduler", "jobs", "describe", "firstrade-platform-session-check-scheduler", "--project", "firstradequant", "--location", "us-central1"],
+            calls,
+        )
+        self.assertIn(
+            ["scheduler", "jobs", "delete", "firstrade-platform-service-session-check-scheduler", "--project", "firstradequant", "--location", "us-central1", "--quiet"],
+            calls,
+        )
+        self.assertIn(
+            ["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))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_sync_cloud_run_env_workflow.py b/tests/test_sync_cloud_run_env_workflow.py
index 66e35fa..da75f18 100644
--- a/tests/test_sync_cloud_run_env_workflow.py
+++ b/tests/test_sync_cloud_run_env_workflow.py
@@ -138,6 +138,10 @@ def test_sync_cloud_run_env_workflow_uses_sync_plan_script():
     assert "for key, value in sorted(target[\"env\"].items()):" in workflow
     assert "target.get(\"remove_env_vars\")" in workflow
 
+    assert "Reconcile Cloud Run traffic" in workflow
+    assert "python scripts/reconcile_cloud_runtime.py traffic" in workflow
+    assert "Reconcile legacy Cloud Scheduler jobs" in workflow
+    assert "python scripts/reconcile_cloud_runtime.py scheduler-cleanup" in workflow
     assert "add_optional_env " not in workflow
     assert "requires_snapshot_artifacts=" not in workflow
     assert "Resolve selected strategy runtime requirements" not in workflow
@@ -170,7 +174,6 @@ def test_sync_cloud_run_env_workflow_syncs_scheduler_from_sync_plan():
     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 '"${CLOUD_RUN_SERVICE}-session-check-scheduler"' in workflow
-    assert 'gcloud scheduler jobs delete "${legacy_job}"' in workflow
     assert '--schedule="${desired_schedule}"' in workflow
     assert '--time-zone="${market_timezone}"' in workflow
+    assert "legacy_jobs=(" not in workflow