diff --git a/.github/workflows/short-name-migration.yml b/.github/workflows/short-name-migration.yml
new file mode 100644
index 00000000..62da735b
--- /dev/null
+++ b/.github/workflows/short-name-migration.yml
@@ -0,0 +1,89 @@
+name: Observe or initialize four disabled short-name destinations
+
+on:
+  workflow_dispatch:
+    inputs:
+      mode:
+        description: Observe, validate only, or separately authorized inert initialization
+        required: true
+        default: observe
+        type: choice
+        options: [observe, validate, initialize]
+      mapping_ref:
+        description: Exact restricted four-target mapping reference
+        required: true
+        type: string
+      source_revision:
+        description: Exact approved source revision
+        required: true
+        type: string
+      plan_sha256:
+        description: SHA-256 of the canonical protected migration plan
+        required: true
+        type: string
+
+permissions:
+  contents: read
+
+# All four destinations share one lock. Never cancel an in-flight cloud request.
+concurrency:
+  group: ibkr-four-target-short-name-migration
+  cancel-in-progress: false
+
+jobs:
+  migration:
+    if: github.event_name == 'workflow_dispatch' && github.ref == 'refs/heads/main' && github.run_attempt == 1
+    environment: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      MIGRATION_ENVIRONMENT: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }}
+      MIGRATION_PLAN_JSON: ${{ secrets.IBKR_SHORT_NAME_MIGRATION_PLAN_JSON }}
+      MIGRATION_AUTHORIZATION_JSON: ${{ secrets.IBKR_SHORT_NAME_MIGRATION_AUTHORIZATION_JSON }}
+      GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}
+      INPUT_MODE: ${{ inputs.mode }}
+      INPUT_MAPPING_REF: ${{ inputs.mapping_ref }}
+      INPUT_SOURCE_REVISION: ${{ inputs.source_revision }}
+      INPUT_PLAN_SHA256: ${{ inputs.plan_sha256 }}
+      CLOUDSDK_CORE_DISABLE_FILE_LOGGING: 'true'
+      CLOUDSDK_CORE_LOG_HTTP: 'false'
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          ref: ${{ github.sha }}
+          persist-credentials: false
+      - name: Mask private bindings and admit exact scope before authentication
+        shell: bash
+        run: |
+          set -euo pipefail
+          python3 -m scripts.mask_sync_identifiers
+          python3 - <<'PY'
+          import os
+          from scripts.short_name_migration import admit
+          try:
+              admit(os.environ['INPUT_MODE'], os.environ)
+          except Exception:
+              raise SystemExit('Migration admission blocked; private diagnostics suppressed') from None
+          PY
+      - uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ secrets.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ secrets.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+      - uses: google-github-actions/setup-gcloud@v3
+        with:
+          version: '>= 416.0.0'
+      - name: Observe or initialize inert destinations without cutover
+        shell: bash
+        run: |
+          set -euo pipefail
+          python3 -m scripts.short_name_migration --mode "$INPUT_MODE" --report "$RUNNER_TEMP/short-name-migration.safe.json"
+      - name: Retain anonymous reference-only result
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: short-name-migration-${{ github.run_id }}
+          path: ${{ runner.temp }}/short-name-migration.safe.json
+          if-no-files-found: error
diff --git a/docs/short-name-migration.md b/docs/short-name-migration.md
new file mode 100644
index 00000000..9df62046
--- /dev/null
+++ b/docs/short-name-migration.md
@@ -0,0 +1,111 @@
+# Protected short-name destination migration
+
+The existing Deploy workflow admits only inventory services that already exist.
+`short-name-migration.yml` supplies a separate destination observation and inert
+initialization channel for the four references `S07-C01` through `S07-C04`.
+It never updates an existing service, inventory, traffic allocation, Scheduler,
+IAM policy, Secret value or trading switch. Source naming remains governed by
+[the naming contract](naming-trading-diff.zh-CN.md).
+
+## Protection and inputs
+
+Run only from `main`, attempt 1, through the existing protected Environment named
+by `IBKR_DEPLOY_ENVIRONMENT`, using its existing WIF credentials. Workflow
+permissions are `contents: read` and job-scoped `id-token: write`. The cloud
+identity needs project observation, destination service observation and operation
+observation; validation and initialization additionally need service creation and
+`iam.serviceAccounts.actAs` for the approved disabled identity. This change grants
+none of those permissions. A missing binding stops the run. Use a disabled
+identity without broker, account, Secret or trading authority; its exact identity
+belongs only in protected configuration.
+
+The Environment secret `IBKR_SHORT_NAME_MIGRATION_PLAN_JSON` is the complete
+approved plan, not a free-form dispatch payload. Its object has exactly:
+
+- `mapping_ref`: `authority.values.s07-ibkr-rename-draft-r1-mapping`.
+- `source_revision`: the exact source revision approved for this migration.
+- `project`, `region`, `disabled_service_account`: protected infrastructure bindings.
+- `targets`: exactly four ordered objects, each with `reference`, `old`, `new`,
+  `image` and `expected_uid`. Names must be distinct across both sets; each new
+  name must fit the existing service-plus-tag limit. `image` is an immutable
+  regional Artifact Registry digest in the approved project. `expected_uid` is
+  empty for an absent service, or the exact previously observed service UID.
+
+The script pins the source revision from the merged naming change. Operators must
+resolve the mapping in the restricted authority store and compare all four pairs
+before preparing the Environment plan. A matching reference string alone does not
+prove the names were resolved correctly. Dispatch must repeat the exact mapping
+reference and source revision and supply `plan_sha256`, computed over UTF-8 JSON
+with sorted keys and compact separators. No binding or raw resource name belongs
+in dispatch inputs, Git, public logs or artifacts.
+
+Environment configuration is a separate owner operation. This Draft does not
+create or populate any secret, identity or Environment. A future deployment card
+must identify the accepted workflow revision and exact plan; the historical
+four-target deployment scope is retained, not consumed by this implementation.
+
+## Modes and disabled behavior
+
+`observe` performs read-only project and destination GETs. A service 404 means
+absent only after successful active-project readback. Authentication, transport,
+malformed JSON, redirects and other HTTP failures block. Existing destinations require an
+explicit matching UID and the exact inert configuration, ready revision and
+traffic readback; an unknown existing service is a collision. No prior inventory
+admission is required for new names.
+
+`validate` requires a separate current authorization and sends service creation
+requests with the API's `validateOnly=true` for absent destinations. It creates
+nothing. Each validation operation must finish successfully before proceeding;
+a successful validate-only operation need not return a service. Pending operations
+are polled, and errors or unknown outcomes block all creation. Operation names
+may use the exact project ID or numeric name from the authenticated active-project
+readback, within the approved region only. `initialize` first completes those
+validations for all four, then
+re-reads each destination immediately before an atomic create. It never patches
+or replaces an existing service. An existing accepted destination is checked
+again and left unchanged. A concurrent creator, collision or failure stops the
+run; no automatic retry, deletion or rollback follows.
+
+A first Cloud Run service cannot use an existing service's no-traffic deployment
+baseline. Initialization therefore creates an **inert disabled service**, with
+internal-only ingress, IAM invocation checks, zero minimum instances and a fixed
+Python HTTP process that returns a fixed disabled response with status 503. It does not launch the application entry point, mount a
+volume, inject an account/Secret/environment variable or connect to IB Gateway.
+Its initial revision receives 100% of that new service's traffic; that revision
+runs only the inert process. Old services and their traffic remain untouched.
+This is disabled-target initialization, not a no-traffic application release or
+cutover. Later application deployment and inventory adoption require a separate
+approved path; this entry cannot perform them.
+
+Every validation/create needs the Environment secret
+`IBKR_SHORT_NAME_MIGRATION_AUTHORIZATION_JSON` with `action` equal to
+`initialize-disabled`, an `authority_ref` for the separate owner card, and exact
+`run_id`, `workflow_sha`, `plan_sha256`, `source_revision`, `mapping_ref` and
+`expires_at` (timezone-aware ISO timestamp, at most one hour ahead). The Draft card
+and historical deployment approval alone are rejected as run authority. Approval
+is checked before authentication and again before each POST, after obtaining its
+access token. Environment protection, restricted secret maintenance and the
+single workflow concurrency group remain required operational protections.
+
+## Readback, restart and evidence
+
+Creation waits on the returned operation, validates its service UID, then GETs
+the service again. The ready/created revision must agree; observed generation
+must match current generation; reconciliation must be finished; traffic must be
+exactly 100% to that inert revision with no tag. The service's identity, image
+digest, inert command, empty environment, service account, ingress and scaling
+must match the protected plan. Both service-level and revision-template minimum
+instance counts must be zero. Proto3 JSON omission of default false/zero fields
+is interpreted using the documented API defaults; required identities and readiness
+fields still need explicit evidence. These values stay in runner memory. The uploaded
+result contains only four reference states and revision/traffic verification
+booleans; it does not expose revisions, UIDs, names, image paths or private hashes.
+
+Polling reads may repeat, but mutations never retry. Timeout, process loss or
+ambiguous POST outcome leaves `BLOCKED_OR_OUTCOME_UNKNOWN`; partial creation is
+not rolled back. Observe and reconcile cloud state before a separately authorized
+continuation; record exact UIDs privately and prepare a fresh plan/run card.
+Workflow reruns are blocked. No automatic action is authorized by an artifact or
+an observation result. The source revision binding and approved digest are intent
+constraints, not an attestation of image build provenance or production adoption.
+Live, PAPER, Kelly and every stop-loss switch stay outside this channel.
diff --git a/scripts/short_name_migration.py b/scripts/short_name_migration.py
new file mode 100644
index 00000000..68e902ec
--- /dev/null
+++ b/scripts/short_name_migration.py
@@ -0,0 +1,308 @@
+"""Protected four-target destination observation and inert initialization.
+
+Never changes an existing service, traffic, inventory, Scheduler, IAM or secrets.
+Provider responses and resource bindings remain in memory; reports use refs only.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime, timezone
+from pathlib import Path
+
+SOURCE_REVISION = '097c16d9a5dc703f682658b189d5df80bef6670b'
+MAPPING_REF = 'authority.values.s07-ibkr-rename-draft-r1-mapping'
+REFERENCES = [f'S07-C0{i}' for i in range(1, 5)]
+RUN_API = 'https://run.googleapis.com/v2'
+PROJECT_API = 'https://cloudresourcemanager.googleapis.com/v3/projects/'
+NAME = re.compile(r'[a-z][a-z0-9-]{0,47}[a-z0-9]')
+DISABLED_SERVER = (
+    'from http.server import BaseHTTPRequestHandler, HTTPServer\n'
+    'class Disabled(BaseHTTPRequestHandler):\n'
+    ' def do_GET(self):\n'
+    '  self.send_response(503); self.end_headers(); self.wfile.write(b"disabled")\n'
+    ' do_POST = do_GET\n'
+    ' def log_message(self, *args): pass\n'
+    'HTTPServer(("0.0.0.0", 8080), Disabled).serve_forever()'
+)
+
+
+class Blocked(Exception):
+    """No provider or private input detail may be included in this exception."""
+
+
+def require(condition):
+    if not condition:
+        raise Blocked('Migration contract not satisfied')
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(',', ':'))
+
+
+def admit(mode, env):
+    require(mode in {'observe', 'validate', 'initialize'})
+    require(env.get('GITHUB_EVENT_NAME') == 'workflow_dispatch')
+    require(env.get('GITHUB_REF') == 'refs/heads/main' and env.get('GITHUB_RUN_ATTEMPT') == '1')
+    require(bool(env.get('MIGRATION_ENVIRONMENT')))
+    p = json.loads(env['MIGRATION_PLAN_JSON'])
+    require(set(p) == {'mapping_ref', 'source_revision', 'project', 'region',
+                       'disabled_service_account', 'targets'})
+    require(p['source_revision'] == env.get('INPUT_SOURCE_REVISION') == SOURCE_REVISION)
+    require(p['mapping_ref'] == env.get('INPUT_MAPPING_REF') == MAPPING_REF)
+    require(hashlib.sha256(canonical(p).encode()).hexdigest() == env.get('INPUT_PLAN_SHA256'))
+    require(isinstance(p['project'], str) and re.fullmatch(r'[a-z][a-z0-9-]{4,61}[a-z0-9]', p['project']))
+    require(p['project'] == env.get('GCP_PROJECT_ID'))
+    require(isinstance(p['region'], str) and re.fullmatch(r'[a-z]+-[a-z]+[0-9]', p['region']))
+    require(re.fullmatch(r'[a-z][a-z0-9-]{4,28}[a-z0-9]@' + re.escape(p['project']) +
+                         r'\.iam\.gserviceaccount\.com', p['disabled_service_account']))
+    require(isinstance(p['targets'], list) and len(p['targets']) == 4)
+    require([t['reference'] for t in p['targets']] == REFERENCES)
+    for t in p['targets']:
+        require(set(t) == {'reference', 'old', 'new', 'image', 'expected_uid'})
+        require(all(isinstance(t[k], str) for k in t))
+        require(NAME.fullmatch(t['old']) and NAME.fullmatch(t['new']))
+        require(len(t['new']) + 13 <= 46)
+        require(re.fullmatch(re.escape(p['region'] + '-docker.pkg.dev/' + p['project']) +
+                             r'/[a-z0-9-]+/[a-z0-9/-]+@sha256:[0-9a-f]{64}', t['image']))
+        require(not t['expected_uid'] or re.fullmatch(r'[a-zA-Z0-9-]{1,128}', t['expected_uid']))
+    require(len({t['old'] for t in p['targets']}) == 4)
+    require(len({t['new'] for t in p['targets']}) == 4)
+    require(not {t['old'] for t in p['targets']} & {t['new'] for t in p['targets']})
+    if mode != 'observe':
+        authorize(env)
+    return p
+
+
+def authorize(env):
+    """A future card, bound to this run and tree, must be installed in Environment."""
+    card = json.loads(env.get('MIGRATION_AUTHORIZATION_JSON', '{}'))
+    for key, expected in {
+        'action': 'initialize-disabled', 'run_id': env.get('GITHUB_RUN_ID'),
+        'workflow_sha': env.get('GITHUB_SHA'), 'plan_sha256': env.get('INPUT_PLAN_SHA256'),
+        'source_revision': SOURCE_REVISION, 'mapping_ref': MAPPING_REF,
+    }.items():
+        require(bool(expected) and card.get(key) == expected)
+    require(re.fullmatch(r'[0-9a-f]{40}', card['workflow_sha']))
+    require(re.fullmatch(r'[0-9a-f]{64}', card['plan_sha256']))
+    require(isinstance(card.get('authority_ref'), str) and bool(card['authority_ref'].strip()))
+    # The draft card and retained historical deployment scope are not a new run card.
+    require(card['authority_ref'] not in {'t1320s7739', 't1320s7726'})
+    expiry = datetime.fromisoformat(card['expires_at'])
+    require(expiry.tzinfo is not None)
+    seconds = (expiry - datetime.now(timezone.utc)).total_seconds()
+    require(0 < seconds <= 3600)
+
+
+def service_body(p, t):
+    return {
+        'name': f"projects/{p['project']}/locations/{p['region']}/services/{t['new']}",
+        'description': 'S07 inert destination ' + t['reference'],
+        'labels': {'s07-reference': t['reference'].lower(), 's07-source': SOURCE_REVISION},
+        'ingress': 'INGRESS_TRAFFIC_INTERNAL_ONLY', 'invokerIamDisabled': False,
+        'template': {
+            'serviceAccount': p['disabled_service_account'],
+            'scaling': {'minInstanceCount': 0, 'maxInstanceCount': 1},
+            'timeout': '30s', 'maxInstanceRequestConcurrency': 1,
+            'containers': [{
+                'image': t['image'], 'command': ['python', '-c'],
+                'args': [DISABLED_SERVER], 'env': [],
+                'ports': [{'name': 'http1', 'containerPort': 8080}],
+                'resources': {'limits': {'cpu': '1', 'memory': '128Mi'},
+                              'cpuIdle': True, 'startupCpuBoost': False},
+            }],
+        },
+        'traffic': [{'type': 'TRAFFIC_TARGET_ALLOCATION_TYPE_LATEST', 'percent': 100}],
+    }
+
+
+def verify_service(p, t, service, uid):
+    """Require exact inert configuration and explicit ready revision/traffic."""
+    require(bool(uid) and service.get('uid') == uid)
+    body = service_body(p, t)
+    for key in ('name', 'description', 'labels', 'ingress'):
+        require(service.get(key) == body[key])
+    # Proto3 JSON omits default false/zero fields; these are defined API values.
+    require(service.get('invokerIamDisabled', False) is False)
+    require(service.get('scaling', {}).get('minInstanceCount', 0) == 0)
+    template = service.get('template', {})
+    expected = body['template']
+    for key in ('serviceAccount', 'timeout', 'maxInstanceRequestConcurrency'):
+        require(template.get(key) == expected[key])
+    scaling = template.get('scaling', {})
+    require(scaling.get('minInstanceCount', 0) == 0 and scaling.get('maxInstanceCount') == 1)
+    containers = template.get('containers')
+    require(isinstance(containers, list) and len(containers) == 1)
+    for key, value in expected['containers'][0].items():
+        if key == 'resources':
+            resources = containers[0].get('resources', {})
+            require(resources.get('limits') == value['limits'])
+            require(resources.get('cpuIdle') is True)
+            require(resources.get('startupCpuBoost', False) is False)
+        else:
+            require(containers[0].get(key, [] if key == 'env' else None) == value)
+    # Never accept volumes, sidecars, secret env or additional container inputs.
+    require(not template.get('volumes'))
+    require(set(containers[0]) <= set(expected['containers'][0]) | {'name', 'startupProbe', 'livenessProbe'})
+    require(service.get('reconciling', False) is False)
+    require(str(service.get('generation')) == str(service.get('observedGeneration')) != 'None')
+    require(service.get('terminalCondition', {}).get('state') == 'CONDITION_SUCCEEDED')
+    revision = service.get('latestCreatedRevision')
+    require(isinstance(revision, str) and revision.startswith(body['name'] + '/revisions/'))
+    require(service.get('latestReadyRevision') == revision)
+    traffic = service.get('trafficStatuses')
+    require(isinstance(traffic, list) and len(traffic) == 1)
+    require(traffic[0].get('revision') == revision and traffic[0].get('percent') == 100)
+    require(not traffic[0].get('tag'))
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
+
+
+def open_request(request, timeout):
+    return urllib.request.build_opener(NoRedirect()).open(request, timeout=timeout)
+
+
+class Cloud:
+    """Bounded REST client. No diagnostic body, token or raw exception escapes."""
+    def __init__(self, env, mode):
+        self.env, self.mode = env, mode
+
+    def __call__(self, method, url, body=None):
+        require(url.startswith(RUN_API + '/projects/') or
+                method == 'GET' and url.startswith(PROJECT_API))
+        require(method in {'GET', 'POST'})
+        try:
+            token = subprocess.run(['gcloud', 'auth', 'print-access-token'], capture_output=True,
+                                   text=True, check=True, timeout=30).stdout.strip()
+            require(bool(token))
+            if method == 'POST':
+                authorize(self.env)  # Token acquisition can outlive the approval.
+            request = urllib.request.Request(
+                url, data=canonical(body).encode() if body is not None else None,
+                method=method, headers={'Authorization': 'Bearer ' + token,
+                                        'Content-Type': 'application/json'})
+            with open_request(request, timeout=30) as response:
+                return json.load(response)
+        except urllib.error.HTTPError as exc:
+            if method == 'GET' and '/services/' in url and exc.code == 404:
+                return None
+            raise Blocked('Cloud request failed; no automatic retry') from None
+        except Blocked:
+            raise
+        except Exception:
+            raise Blocked('Cloud transport failed; outcome may be unknown') from None
+
+
+def wait_operation(cloud, operation, parents, *, validation=False):
+    """Confirm completion; validate-only success need not contain a service."""
+    require(isinstance(operation, dict) and 'error' not in operation)
+    if validation and operation.get('done') is True and 'name' not in operation:
+        return None
+    name = operation.get('name', '')
+    require(isinstance(name, str))
+    if name.startswith(RUN_API + '/'):
+        name = name[len(RUN_API) + 1:]
+    parents = (parents,) if isinstance(parents, str) else parents
+    require(any(name.startswith(parent + '/operations/') for parent in parents) and
+            re.fullmatch(r'[A-Za-z0-9-]+', name.split('/')[-1]))
+    result = operation
+    for attempt in range(31):
+        require(isinstance(result, dict) and 'error' not in result)
+        require(isinstance(result.get('done', False), bool))
+        if result.get('done') is True:
+            if validation:
+                return None
+            require(isinstance(result.get('response'), dict))
+            return result['response']
+        if attempt == 30:
+            break
+        if attempt:
+            time.sleep(2)
+        result = cloud('GET', RUN_API + '/' + name)
+    raise Blocked('Operation pending; inspect before any separately authorized continuation')
+
+
+def run(mode, env, cloud):
+    p = admit(mode, env)
+    project = cloud('GET', PROJECT_API + p['project'])
+    require(isinstance(project, dict) and project.get('projectId') == p['project'] and project.get('state') == 'ACTIVE')
+    parent = f"projects/{p['project']}/locations/{p['region']}"
+    project_name = project.get('name', '')
+    require(isinstance(project_name, str) and re.fullmatch(r'projects/[0-9]+', project_name))
+    operation_parents = (parent, project_name + '/locations/' + p['region'])
+    rows = []
+    # Complete all four read-only checks before the first validation or creation.
+    for t in p['targets']:
+        url = RUN_API + '/' + service_body(p, t)['name']
+        current = cloud('GET', url)
+        if current is not None:
+            verify_service(p, t, current, t['expected_uid'])
+        else:
+            require(not t['expected_uid'])  # A formerly known service disappearing is drift.
+        rows.append({'reference': t['reference'], 'state': 'ABSENT' if current is None else 'OWNED_DISABLED',
+                     'revision_verified': current is not None, 'traffic_verified': current is not None})
+    if mode == 'observe':
+        return {'mode': mode, 'targets': rows}
+    for t, row in zip(p['targets'], rows):
+        if row['state'] == 'ABSENT':
+            authorize(env)
+            operation = cloud('POST', RUN_API + '/' + parent + '/services?serviceId=' +
+                              urllib.parse.quote(t['new'], safe='') + '&validateOnly=true', service_body(p, t))
+            wait_operation(cloud, operation, operation_parents, validation=True)
+    if mode == 'validate':
+        return {'mode': mode, 'targets': rows}
+    for t, row in zip(p['targets'], rows):
+        url = RUN_API + '/' + service_body(p, t)['name']
+        current = cloud('GET', url)  # Never use the earlier preflight snapshot to create.
+        if row['state'] == 'OWNED_DISABLED':
+            require(current is not None)
+            verify_service(p, t, current, t['expected_uid'])
+            continue
+        require(current is None)
+        authorize(env)
+        operation = cloud('POST', RUN_API + '/' + parent + '/services?serviceId=' +
+                          urllib.parse.quote(t['new'], safe=''), service_body(p, t))
+        created = wait_operation(cloud, operation, operation_parents)
+        verify_service(p, t, created, created.get('uid'))
+        readback = cloud('GET', url)
+        require(isinstance(readback, dict))
+        verify_service(p, t, readback, created['uid'])
+        row.update(state='INITIALIZED_DISABLED', revision_verified=True, traffic_verified=True)
+    return {'mode': mode, 'targets': rows}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--mode', choices=['observe', 'validate', 'initialize'], required=True)
+    parser.add_argument('--report', type=Path)
+    args = parser.parse_args(argv)
+    try:
+        result = run(args.mode, os.environ, Cloud(os.environ, args.mode))
+    except Exception:
+        # Fixed diagnostic includes no environment, provider response or resource identity.
+        result = {'mode': args.mode, 'state': 'BLOCKED_OR_OUTCOME_UNKNOWN',
+                  'automatic_retry': False, 'revision_and_traffic': 'NOT_PROVEN'}
+        status = 1
+    else:
+        status = 0
+    safe = canonical(result)
+    if args.report:
+        args.report.write_text(safe + '\n', encoding='utf-8')
+    print(safe, file=sys.stderr if status else sys.stdout)
+    return status
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/tests/test_short_name_migration.py b/tests/test_short_name_migration.py
new file mode 100644
index 00000000..7a8204c9
--- /dev/null
+++ b/tests/test_short_name_migration.py
@@ -0,0 +1,484 @@
+"""Fake-only adverse contracts for the four-target disabled migration channel."""
+import copy
+import hashlib
+import json
+from datetime import datetime, timedelta, timezone
+from unittest.mock import Mock
+
+import pytest
+
+from scripts import short_name_migration as migration
+
+
+def plan():
+    return {
+        'mapping_ref': migration.MAPPING_REF,
+        'source_revision': migration.SOURCE_REVISION,
+        'project': 'fake-project', 'region': 'asia-southeast1',
+        'disabled_service_account': 'fake-disabled@fake-project.iam.gserviceaccount.com',
+        'targets': [
+            {'reference': f'S07-C0{i}', 'old': f'fake-old-{i}', 'new': f'fake-new-{i}',
+             'image': 'asia-southeast1-docker.pkg.dev/fake-project/fake/image@sha256:' + 'a' * 64,
+             'expected_uid': ''}
+            for i in range(1, 5)
+        ],
+    }
+
+
+def environment(candidate=None):
+    candidate = candidate or plan()
+    digest = hashlib.sha256(migration.canonical(candidate).encode()).hexdigest()
+    env = {
+        'GITHUB_EVENT_NAME': 'workflow_dispatch', 'GITHUB_REF': 'refs/heads/main',
+        'GITHUB_RUN_ATTEMPT': '1', 'GITHUB_RUN_ID': '123', 'GITHUB_SHA': 'b' * 40,
+        'MIGRATION_ENVIRONMENT': 'fake-protected', 'MIGRATION_PLAN_JSON': json.dumps(candidate),
+        'INPUT_PLAN_SHA256': digest, 'INPUT_SOURCE_REVISION': migration.SOURCE_REVISION,
+        'INPUT_MAPPING_REF': migration.MAPPING_REF,
+        'GCP_PROJECT_ID': 'fake-project',
+    }
+    card = {
+        'action': 'initialize-disabled', 'run_id': '123', 'workflow_sha': 'b' * 40,
+        'plan_sha256': digest, 'source_revision': migration.SOURCE_REVISION,
+        'mapping_ref': migration.MAPPING_REF,
+        'expires_at': (datetime.now(timezone.utc) + timedelta(minutes=20)).isoformat(),
+        'authority_ref': 'fake-separate-owner-card',
+    }
+    env['MIGRATION_AUTHORIZATION_JSON'] = json.dumps(card)
+    return env
+
+
+def owned(candidate, target, uid='fake-uid'):
+    body = migration.service_body(candidate, target)
+    body.update(uid=uid, generation='1', observedGeneration='1', reconciling=False,
+                terminalCondition={'state': 'CONDITION_SUCCEEDED'},
+                latestCreatedRevision=body['name'] + '/revisions/fake-revision',
+                latestReadyRevision=body['name'] + '/revisions/fake-revision',
+                trafficStatuses=[{'revision': body['name'] + '/revisions/fake-revision',
+                                  'percent': 100, 'type': 'TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION'}])
+    return body
+
+
+class Cloud:
+    def __init__(self, candidate):
+        self.plan = candidate
+        self.services = {}
+        self.calls = []
+
+    def __call__(self, method, path, body=None):
+        self.calls.append((method, path, body))
+        if path.startswith('https://cloudresourcemanager.googleapis.com/'):
+            return {'projectId': self.plan['project'], 'name': 'projects/123456789012', 'state': 'ACTIVE'}
+        if '/operations/' in path:
+            return {'done': True, 'response': self.created}
+        if method == 'GET':
+            return copy.deepcopy(self.services.get(path))
+        if 'validateOnly=true' in path:
+            return {'done': True}
+        self.created = owned(self.plan, next(t for t in self.plan['targets'] if body['name'].endswith('/' + t['new'])))
+        self.services[migration.RUN_API + '/' + body['name']] = self.created
+        return {'name': migration.RUN_API + '/projects/fake-project/locations/asia-southeast1/operations/fake'}
+
+
+@pytest.mark.parametrize('mutation', [
+    lambda p: p['targets'].pop(),
+    lambda p: p['targets'][0].update(reference='S07-C05'),
+    lambda p: p['targets'][1].update(new=p['targets'][0]['new']),
+    lambda p: p['targets'][0].update(new=p['targets'][1]['old']),
+    lambda p: p.update(source_revision='c' * 40),
+    lambda p: p['targets'][0].update(image='fake:latest'),
+    lambda p: p.update(project='../fake-project'),
+    lambda p: p['targets'][0].update(enable_live=True),
+])
+def test_invalid_scope_fails_before_cloud(mutation):
+    p = plan()
+    mutation(p)
+    cloud = Mock()
+    with pytest.raises(migration.Blocked):
+        migration.run('observe', environment(p), cloud)
+    cloud.assert_not_called()
+
+
+@pytest.mark.parametrize('key,value', [('GITHUB_REF', 'refs/heads/fake'), ('GITHUB_RUN_ATTEMPT', '2'),
+                                     ('MIGRATION_ENVIRONMENT', ''), ('INPUT_PLAN_SHA256', 'a' * 64),
+                                     ('MIGRATION_AUTHORIZATION_JSON', '{}')])
+def test_dispatch_or_authority_failure_blocks_before_cloud(key, value):
+    env = environment()
+    env[key] = value
+    cloud = Mock()
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', env, cloud)
+    cloud.assert_not_called()
+
+
+def test_absent_destination_observation_needs_no_inventory_or_mutation():
+    p = plan()
+    cloud = Cloud(p)
+    report = migration.run('observe', environment(p), cloud)
+    assert [r['state'] for r in report['targets']] == ['ABSENT'] * 4
+    assert all(c[0] == 'GET' for c in cloud.calls)
+    assert 'fake-project' not in json.dumps(report)
+
+
+def test_collision_never_writes_even_if_other_destinations_absent():
+    p = plan()
+    cloud = Cloud(p)
+    cloud.services[migration.RUN_API + '/' + migration.service_body(p, p['targets'][3])['name']] = owned(p, p['targets'][3])
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', environment(p), cloud)
+    assert all(c[0] == 'GET' for c in cloud.calls)
+
+
+def test_initialize_creates_only_fixed_disabled_containers_and_reads_back():
+    p = plan()
+    cloud = Cloud(p)
+    report = migration.run('initialize', environment(p), cloud)
+    assert [r['state'] for r in report['targets']] == ['INITIALIZED_DISABLED'] * 4
+    writes = [c for c in cloud.calls if c[0] == 'POST' and 'validateOnly=true' not in c[1]]
+    assert len(writes) == 4
+    for _, _, body in writes:
+        assert body['template']['containers'][0]['command'] == ['python', '-c']
+        assert body['template']['containers'][0]['args'] == [migration.DISABLED_SERVER]
+        assert body['ingress'] == 'INGRESS_TRAFFIC_INTERNAL_ONLY'
+        assert body['invokerIamDisabled'] is False
+        assert body['template']['containers'][0]['env'] == []
+    assert all(r['revision_verified'] and r['traffic_verified'] for r in report['targets'])
+
+
+def test_concurrent_creator_is_detected_by_fresh_read_before_post():
+    p = plan()
+    cloud = Cloud(p)
+    original = cloud.__call__
+    reads = 0
+    def race(method, path, body=None):
+        nonlocal reads
+        if method == 'GET' and '/services/' in path:
+            reads += 1
+            if reads == 5:
+                return owned(p, p['targets'][0])
+        return original(method, path, body)
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', environment(p), race)
+    assert not any(c[0] == 'POST' and 'validateOnly' not in c[1] for c in cloud.calls)
+
+
+def test_expiry_rechecked_before_each_mutation():
+    env = environment()
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    def expire(method, path, body=None):
+        result = original(method, path, body)
+        if 'validateOnly' in path:
+            card = json.loads(env['MIGRATION_AUTHORIZATION_JSON'])
+            card['expires_at'] = '2000-01-01T00:00:00+00:00'
+            env['MIGRATION_AUTHORIZATION_JSON'] = json.dumps(card)
+        return result
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', env, expire)
+    assert not any(c[0] == 'POST' and 'validateOnly' not in c[1] for c in cloud.calls)
+
+
+@pytest.mark.parametrize('field,value', [('trafficStatuses', []), ('reconciling', True),
+                                        ('latestReadyRevision', ''), ('uid', 'different')])
+def test_bad_or_stale_readback_blocks(field, value):
+    p = plan()
+    target = p['targets'][0]
+    service = owned(p, target)
+    service[field] = value
+    with pytest.raises(migration.Blocked):
+        migration.verify_service(p, target, service, 'fake-uid')
+
+
+def test_transport_and_provider_details_never_escape(capsys, monkeypatch):
+    monkeypatch.setenv('MIGRATION_PLAN_JSON', '{invalid fake-sensitive-value')
+    assert migration.main(['--mode', 'observe']) == 1
+    assert 'fake-sensitive-value' not in capsys.readouterr().err
+
+
+def test_validate_is_not_creation():
+    cloud = Cloud(plan())
+    report = migration.run('validate', environment(), cloud)
+    assert report['mode'] == 'validate'
+    posts = [c for c in cloud.calls if c[0] == 'POST']
+    assert len(posts) == 4
+    assert all('validateOnly=true' in c[1] for c in posts)
+    assert not cloud.services
+
+
+def test_existing_owned_destinations_are_not_changed():
+    p = plan()
+    for t in p['targets']:
+        t['expected_uid'] = 'fake-uid'
+    cloud = Cloud(p)
+    for t in p['targets']:
+        cloud.services[migration.RUN_API + '/' + migration.service_body(p, t)['name']] = owned(p, t)
+    report = migration.run('initialize', environment(p), cloud)
+    assert all(row['state'] == 'OWNED_DISABLED' for row in report['targets'])
+    assert all(c[0] == 'GET' for c in cloud.calls)
+
+
+@pytest.mark.parametrize('status,absent', [(404, True), (403, False), (429, False), (500, False)])
+def test_only_service_not_found_means_absent(monkeypatch, status, absent):
+    import urllib.error
+    monkeypatch.setattr(migration.subprocess, 'run', lambda *a, **kw: Mock(stdout='fake-token'))
+    def fail(*a, **kw):
+        raise urllib.error.HTTPError('fake-private-url', status, 'fake-private-diagnostic', {}, None)
+    monkeypatch.setattr(migration, 'open_request', fail)
+    cloud = migration.Cloud(environment(), 'observe')
+    if absent:
+        assert cloud('GET', migration.RUN_API + '/projects/fake-project/locations/fake/services/fake') is None
+    else:
+        with pytest.raises(migration.Blocked) as caught:
+            cloud('GET', migration.RUN_API + '/projects/fake-project/locations/fake/services/fake')
+        assert 'fake-private' not in str(caught.value)
+
+
+def test_token_acquisition_expiry_prevents_post(monkeypatch):
+    env = environment()
+    def token(*a, **kw):
+        card = json.loads(env['MIGRATION_AUTHORIZATION_JSON'])
+        card['expires_at'] = '2000-01-01T00:00:00+00:00'
+        env['MIGRATION_AUTHORIZATION_JSON'] = json.dumps(card)
+        return Mock(stdout='fake-token')
+    monkeypatch.setattr(migration.subprocess, 'run', token)
+    request = Mock()
+    monkeypatch.setattr(migration, 'open_request', request)
+    with pytest.raises(migration.Blocked):
+        migration.Cloud(env, 'initialize')('POST', migration.RUN_API + '/projects/fake-project/locations/fake/services', {})
+    request.assert_not_called()
+
+
+@pytest.mark.parametrize('status', [409, 500])
+def test_creation_error_is_never_retried_or_assumed_absent(monkeypatch, status):
+    import urllib.error
+    monkeypatch.setattr(migration.subprocess, 'run', lambda *a, **kw: Mock(stdout='fake-token'))
+    request = Mock(side_effect=urllib.error.HTTPError('fake-url', status, 'fake-private-error', {}, None))
+    monkeypatch.setattr(migration, 'open_request', request)
+    with pytest.raises(migration.Blocked):
+        migration.Cloud(environment(), 'initialize')('POST', migration.RUN_API + '/projects/fake-project/locations/fake/services', {})
+    assert request.call_count == 1
+
+
+def test_operation_timeout_does_not_repeat_mutation(monkeypatch):
+    monkeypatch.setattr(migration.time, 'sleep', lambda _: None)
+    cloud = Mock(return_value={'done': False})
+    with pytest.raises(migration.Blocked):
+        migration.wait_operation(cloud, {'name': 'projects/fake-project/locations/fake/operations/fake'},
+                                 'projects/fake-project/locations/fake')
+    assert all(call.args[0] == 'GET' for call in cloud.call_args_list)
+
+
+def test_workflow_protection_contract():
+    from pathlib import Path
+    workflow = Path('.github/workflows/short-name-migration.yml').read_text()
+    assert 'on:\n  workflow_dispatch:' in workflow
+    assert 'workflow_call:' not in workflow and 'push:' not in workflow
+    assert 'permissions:\n  contents: read\n' in workflow
+    assert 'cancel-in-progress: false' in workflow
+    assert 'environment: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }}' in workflow
+    assert "github.ref == 'refs/heads/main'" in workflow
+    assert 'github.run_attempt == 1' in workflow
+    assert workflow.index('admit exact scope') < workflow.index('google-github-actions/auth@v3')
+    assert 'persist-credentials: false' in workflow
+    assert 'ref: ${{ github.sha }}' in workflow
+    assert 'path: ${{ runner.temp }}/short-name-migration.safe.json' in workflow
+
+
+def test_provider_proto3_default_omission_preserves_disabled_semantics():
+    p = plan()
+    t = p['targets'][0]
+    service = owned(p, t)
+    del service['invokerIamDisabled']
+    del service['reconciling']
+    del service['template']['scaling']['minInstanceCount']
+    del service['template']['containers'][0]['env']
+    del service['template']['containers'][0]['resources']['startupCpuBoost']
+    migration.verify_service(p, t, service, 'fake-uid')
+
+
+@pytest.mark.parametrize('change', [
+    lambda s: s.update(invokerIamDisabled=True),
+    lambda s: s['template']['containers'][0].update(command=['python', 'main.py']),
+    lambda s: s['template']['containers'][0].update(env=[{'name': 'FAKE_SECRET', 'value': 'fake'}]),
+    lambda s: s['template'].update(volumes=[{'secret': {'secret': 'fake'}}]),
+    lambda s: s['template']['scaling'].update(minInstanceCount=1),
+    lambda s: s['trafficStatuses'][0].update(tag='fake-tag'),
+])
+def test_active_or_divergent_service_is_not_owned_disabled(change):
+    p = plan()
+    t = p['targets'][0]
+    service = owned(p, t)
+    change(service)
+    with pytest.raises(migration.Blocked):
+        migration.verify_service(p, t, service, 'fake-uid')
+
+
+def test_redirect_does_not_forward_bearer_token():
+    assert migration.NoRedirect().redirect_request(None, None, 302, 'fake', {}, 'https://fake.invalid') is None
+
+
+def test_inert_container_command_is_valid_python():
+    compile(migration.DISABLED_SERVER, '<disabled-server>', 'exec')
+
+
+@pytest.mark.parametrize('mode', ['validate', 'initialize'])
+@pytest.mark.parametrize('result', [{'done': True, 'error': {'message': 'fake-private-error'}},
+                                   {'done': True, 'error': {}}, None, {}])
+def test_failed_or_unknown_validation_blocks_all_creates(mode, result):
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    validations = 0
+    def fail_last(method, path, body=None):
+        nonlocal validations
+        response = original(method, path, body)
+        if 'validateOnly=true' in path:
+            validations += 1
+            if validations == 4:
+                return result
+        return response
+    with pytest.raises(migration.Blocked):
+        migration.run(mode, environment(), fail_last)
+    assert validations == 4
+    assert not cloud.services
+
+
+def test_pending_validation_finishes_before_creation(monkeypatch):
+    monkeypatch.setattr(migration.time, 'sleep', lambda _: None)
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    pending = polls = 0
+    def pending_validation(method, path, body=None):
+        nonlocal pending, polls
+        if 'validateOnly=true' in path:
+            original(method, path, body)
+            pending += 1
+            return {'name': 'projects/123456789012/locations/asia-southeast1/operations/validation'}
+        if path.endswith('/operations/validation'):
+            polls += 1
+            return {'done': polls % 2 == 0}
+        if method == 'POST':
+            assert polls == 8 and pending == 4
+        return original(method, path, body)
+    report = migration.run('initialize', environment(), pending_validation)
+    assert polls == 8
+    assert all(r['state'] == 'INITIALIZED_DISABLED' for r in report['targets'])
+
+
+@pytest.mark.parametrize('completion', [{'done': True, 'error': {'message': 'fake-private-error'}},
+                                       {'done': False}])
+def test_pending_validation_failure_or_timeout_never_creates(monkeypatch, completion):
+    monkeypatch.setattr(migration.time, 'sleep', lambda _: None)
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    def pending(method, path, body=None):
+        if '/operations/' in path:
+            return completion
+        response = original(method, path, body)
+        if 'validateOnly=true' in path:
+            return {'name': 'projects/fake-project/locations/asia-southeast1/operations/validation'}
+        return response
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', environment(), pending)
+    assert not cloud.services
+
+
+def test_validation_poll_expiry_blocks_creation(monkeypatch):
+    monkeypatch.setattr(migration.time, 'sleep', lambda _: None)
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    env = environment()
+    def expire(method, path, body=None):
+        if '/operations/' in path:
+            card = json.loads(env['MIGRATION_AUTHORIZATION_JSON'])
+            card['expires_at'] = '2000-01-01T00:00:00+00:00'
+            env['MIGRATION_AUTHORIZATION_JSON'] = json.dumps(card)
+            return {'done': True}
+        response = original(method, path, body)
+        if 'validateOnly=true' in path:
+            return {'name': 'projects/fake-project/locations/asia-southeast1/operations/validation'}
+        return response
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', env, expire)
+    assert not cloud.services
+
+
+@pytest.mark.parametrize('project', ['fake-project', '123456789012'])
+def test_creation_accepts_only_authenticated_project_operation_names(project):
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    def numeric(method, path, body=None):
+        response = original(method, path, body)
+        if method == 'POST' and 'validateOnly' not in path:
+            response['name'] = response['name'].replace('projects/fake-project/', 'projects/' + project + '/')
+        return response
+    report = migration.run('initialize', environment(), numeric)
+    assert all(r['state'] == 'INITIALIZED_DISABLED' for r in report['targets'])
+
+
+@pytest.mark.parametrize('prefix', ['projects/999999999999/locations/asia-southeast1',
+                                    'projects/123456789012/locations/other-region1',
+                                    'projects/fake-other/locations/asia-southeast1'])
+def test_validation_operation_wrong_project_or_region_blocks(prefix):
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    def foreign(method, path, body=None):
+        response = original(method, path, body)
+        if 'validateOnly=true' in path:
+            return {'name': prefix + '/operations/fake'}
+        return response
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', environment(), foreign)
+    assert not cloud.services
+
+
+@pytest.mark.parametrize('minimum', [1, 4, '0', None])
+def test_service_level_minimum_drift_blocks_before_any_post(minimum):
+    p = plan()
+    p['targets'][3]['expected_uid'] = 'fake-uid'
+    cloud = Cloud(p)
+    target = p['targets'][3]
+    service = owned(p, target)
+    service['scaling'] = {'minInstanceCount': minimum}
+    cloud.services[migration.RUN_API + '/' + service['name']] = service
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', environment(p), cloud)
+    assert all(c[0] == 'GET' for c in cloud.calls)
+
+
+def test_validation_error_stays_redacted_at_cli(monkeypatch, capsys):
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    def failure(method, path, body=None):
+        if 'validateOnly=true' in path:
+            return {'done': True, 'error': {'message': 'fake-private-validation-detail'}}
+        return original(method, path, body)
+    monkeypatch.setattr(migration.os, 'environ', environment())
+    monkeypatch.setattr(migration, 'Cloud', lambda *a: failure)
+    assert migration.main(['--mode', 'initialize']) == 1
+    captured = capsys.readouterr()
+    assert 'fake-private' not in captured.err + captured.out
+    assert 'BLOCKED_OR_OUTCOME_UNKNOWN' in captured.err
+    assert not cloud.services
+
+
+@pytest.mark.parametrize('project_name', [None, '', 'projects/fake-project',
+                                         'projects/123/locations/fake'])
+def test_missing_or_invalid_authenticated_project_number_blocks(project_name):
+    cloud = Cloud(plan())
+    original = cloud.__call__
+    def invalid_project(method, path, body=None):
+        response = original(method, path, body)
+        if path.startswith(migration.PROJECT_API):
+            response['name'] = project_name
+        return response
+    with pytest.raises(migration.Blocked):
+        migration.run('initialize', environment(), invalid_project)
+    assert len(cloud.calls) == 1
+
+
+@pytest.mark.parametrize('scaling', [{}, {'minInstanceCount': 0}])
+def test_explicit_default_service_minimum_remains_disabled(scaling):
+    p = plan()
+    t = p['targets'][0]
+    service = owned(p, t)
+    service['scaling'] = scaling
+    migration.verify_service(p, t, service, 'fake-uid')