fix: Start untagged warm candidates with an approved micro split

.github/workflows/deploy-pilot-no-traffic-protected.yml

@@ -82,6 +82,7 @@ jobs:
           PILOT_PROFILE: ${{ inputs.profile }}
           PILOT_SOURCE_COMMIT: ${{ inputs.source_commit }}
           PILOT_CLEANUP_RUN_ID: ${{ inputs.isolated_cleanup_run_id }}
+          PILOT_NETWORK_PREFLIGHT: ${{ inputs.isolated_network_preflight }}
           PILOT_JOB_NETWORK_PREFLIGHT: ${{ inputs.isolated_job_network_preflight }}
           PILOT_JOB_PHASE: ${{ inputs.isolated_job_phase }}
           PILOT_TEMPLATE_DIAGNOSTIC: ${{ inputs.isolated_template_diagnostic }}

docs/ibkr-runtime-rollout.md

@@ -402,14 +402,24 @@ It changes no Scheduler, environment, Secret, IAM or runtime execution switch.
 
 The candidate is deployed with `--no-traffic --tag=cand-<source-sha8>
 --min-instances=1` when the service name plus the candidate tag fits Cloud Run's
-46-character combined limit. Longer names omit `--tag` entirely; promotion still
-uses the exact qualified revision name, and cleanup skips tag removal. Short
+46-character combined limit. When the preferred tag does not fit, the shared
+policy allocates a shorter operation-bound tag if at least one character fits.
+Only tag-impossible service names omit `--tag` entirely. After verifying the
+unchanged no-traffic baseline, the exact-target traffic-approved lifecycle
+assigns 99% to the previous revision
+and 1% to the exact warm candidate, asserting the split readback before health
+qualification. Promotion uses the exact qualified revision name. On failure
+before promotion, cleanup recognizes only this exact split, restores the previous
+revision to 100% and verifies it before resetting the minimum. Unexpected traffic
+drift blocks cleanup mutations. Cleanup skips tag removal for untagged services. Short
 configured service names use the tagged path. This is a **revision-level**
 minimum, never service-level `--min`. Google Cloud's
 **Revisions and minimum instances** section says revision minimums start when referenced in a traffic split or assigned a tag; an untagged
 zero-percent revision alone does not establish startup. The untagged path retains
-`--deploy-health-check` and all existing fresh health and baseline checks; missing
-startup evidence stops promotion rather than routing traffic to warm the revision.
+`--deploy-health-check` and strict fresh startup health checks after the approved
+micro-split starts the revision. A Retired revision with no authentic instance
+startup evidence cannot pass qualification. The tagged path retains zero-percent
+serving traffic during qualification and verifies its baseline after health.
 One minimum instance is billed during this warm-up window, including idle time. Missing ContainerHealthy
 remains a hard denial, with bounded fresh readbacks. Raw condition readbacks
 separate absent startup evidence from an observed ContainerHealthy=False probe
@@ -487,3 +497,28 @@ the candidate. Missing controls or invalid targets fail validation. Notification
 runtime-switch and initial-state recovery modes require explicit `dry_run=false`.
 Only a separately authorized execution may set `dry_run=false`; all existing
 main-ref, admission and approval gates still apply.
+
+### Shared revision-tag preflight
+
+PR #976 is source-only. Peer4 is excluded from its service scope. Merging this
+change authorizes no deployment or redeployment; any later deployment requires
+a separate owner card naming the exact service, revision and permitted actions.
+
+Operator entry points use `scripts/cloud_run_tag_policy.py`, a local Option B
+implementation of `cloud-run-tags/v1`; QuantPlatformKit is the intended canonical
+owner. The policy validates the exact resolved service name and the combined
+46-character service/tag budget. It keeps a fitting preferred tag, otherwise
+uses a deterministic operation-bound shortened tag. Occupied tags are refused,
+including tags on zero-percent routes; a short name alone proves no ownership.
+The sync plan reports `tag_preflight` without changing service identity.
+
+Only the configured candidate warmup lifecycle accepts an exact-service warmup
+traffic approval for the temporary 99/1 split when no nonempty tag fits. TCP and
+maintenance-cash probes remain strictly no-traffic and refuse such a target.
+Image-only and notification-only entry points validate names without acquiring
+traffic authority. Existing workflow concurrency, immutable image/identity
+checks, baseline readbacks and cleanup remain required. Single-use local claims
+are retained after interruption and success: reconcile the existing attempt
+rather than deleting a claim or replaying an uncertain mutation. Local claims
+do not provide a distributed lock; provider resource-version checks and the
+protected workflow concurrency boundary remain required for cloud execution.

scripts/build_cloud_run_env_sync_plan.py

@@ -60,6 +60,7 @@ def _should_add_local_src(candidate: Path) -> bool:
 from quant_platform_kit.common.runtime_heartbeat_policy import (  # noqa: E402
     runtime_target_permits_standard_execution,
 )
+from scripts import cloud_run_tag_policy as tag_policy  # noqa: E402
 from scripts.reconciliation_recovery_state_ledger import (  # noqa: E402
     apply_recovery_state_ledger_from_env,
 )
@@ -539,6 +540,9 @@ def _build_target_plan(
     if not service_name:
         raise ValueError("Each Cloud Run sync target requires service/service_name")
 
+    # Validate the exact configured name; budget classification is not authority
+    # to rename a target or start an untagged no-traffic candidate.
+    tag_preflight = tag_policy.preflight(service_name, "cand-" + "0" * 8)
     runtime_target_service = str(runtime_target.get("service_name") or "").strip()
     if runtime_target_service and runtime_target_service != service_name:
         raise ValueError(
@@ -568,6 +572,7 @@ def _build_target_plan(
         permitted = enabled and runtime_target_permits_standard_execution(runtime_target)
         return {
             "service_name": service_name,
+            "tag_preflight": tag_preflight,
             "strategy_profile": raw_profile,
             "env": {"RUNTIME_TARGET_ENABLED": "true" if enabled else "false",
                     "RUNTIME_TARGET_JSON": json.dumps(runtime_target, separators=(",", ":"), sort_keys=True)},
@@ -740,6 +745,7 @@ def _build_target_plan(
 
     return {
         "service_name": service_name,
+        "tag_preflight": tag_preflight,
         "strategy_profile": canonical_profile,
         "env": env_values,
         "scheduler": scheduler,

scripts/cloud_run_candidate_warmup.py

@@ -18,6 +18,7 @@
 
 from notifications.redaction import redact_notification_diagnostics
 from scripts import reconcile_cloud_runtime as reconcile
+from scripts import cloud_run_tag_policy as tag_policy
 from scripts.runtime_target_continuity import verify_traffic_revision
 
 MINIMUM = 'autoscaling.knative.dev/minScale'
@@ -66,26 +67,32 @@ def active(service: dict) -> list:
 class Release:
     def __init__(self, *, project: str, region: str, service: str, commit: str, digest: str,
                  run_id: str, deploy_args: list[str], state_path: Path,
-                 cloud: Callable[[list[str]], dict] = cloud, no_traffic_verify: Callable[[], None] | None = None):
+                 cloud: Callable[[list[str]], dict] = cloud, no_traffic_verify: Callable[[], None] | None = None,
+                 warmup_traffic_approved: bool = False, approved_service: str | None = None):
         self.project, self.region, self.service = project, region, service
         self.commit, self.digest, self.run_id = commit, digest, run_id
         self.deploy_args, self.state_path, self.cloud = deploy_args, state_path, cloud
         self.no_traffic_verify = no_traffic_verify
-        tag = 'cand-' + commit[:8]
-        # Cloud Run limits the combined tag and service name to 46 characters.
-        # Long exact targets use revision identity without a traffic tag.
-        self.tag = tag if len(service) + len(tag) <= 46 else None
+        self.warmup_traffic_approved = warmup_traffic_approved
+        self.approved_service = approved_service
+        self.operation_id = run_id + ':' + commit + ':' + digest
+        self.allocation = tag_policy.allocate_tag(service, 'cand-' + commit[:8],
+            existing_tags=[], operation_id=self.operation_id)
+        self.tag = self.allocation.tag
         self.common = [f'--project={project}', f'--region={region}', '--format=json']
         self.warm = None
         self.reset = None
         self.expected = None
         self.before = None
         self.cleaning = False
+        self.micro_traffic = None
 
     def record(self, status: str) -> None:
         # No resource names, bindings, env or provider diagnostics in this result.
         self.state_path.write_text(json.dumps({'status': status, 'tag': self.tag,
-                                              'reset_creates_revision': True}) + '\n')
+                                              'reset_creates_revision': True,
+                                              'contract_version': tag_policy.CONTRACT_VERSION,
+                                              'operation_fingerprint': self.allocation.operation_fingerprint}) + '\n')
 
     def service_read(self) -> dict:
         return self.cloud(['gcloud', 'run', 'services', 'describe', self.service, *self.common])
@@ -128,7 +135,9 @@ def readback(self, stage: str) -> None:
 
     def stable(self) -> dict:
         current = self.service_read()
-        if self.expected is None or current != self.expected:
+        if (self.expected is None or current != self.expected
+                or (self.before is not None and current.get('metadata', {}).get('uid')
+                    != self.before.get('metadata', {}).get('uid'))):
             raise RuntimeError('candidate service changed outside this attempt')
         return current
 
@@ -151,7 +160,9 @@ def health(self, value: dict) -> None:
         reconcile._verify_deploy_candidate_with_wait(
             value, project=self.project, region=self.region, service_name=self.service,
             service=current, revision=value['metadata']['name'], commit=self.commit,
-            require_generation=True, wait_for_start=True)
+            require_generation=True, wait_for_start=True,
+            **({'allow_warmup_split': True}
+               if self.micro_traffic is not None and value['metadata']['name'] == self.warm else {}))
         self.stable()
         # A fresh read after the wait prevents a cached healthy value from
         # authorizing a later update. Existing continuity checks stay strict.
@@ -161,6 +172,25 @@ def health(self, value: dict) -> None:
         verify_traffic_revision(fresh, expected_revision=value['metadata']['name'],
                                 expected_commit=self.commit, require_startup_health=True)
 
+    def start_untagged(self) -> None:
+        if tag_policy.lifecycle_for_impossible_tag(service_name=self.service,
+                approved_service=self.approved_service,
+                warmup_traffic_approved=self.warmup_traffic_approved) != 'MICRO_SPLIT':
+            raise RuntimeError('TAG_IMPOSSIBLE: exact warmup approval required')
+        self.stable()
+        previous = active(self.before)[0][0]
+        self.micro_traffic = sorted([(previous, 99), (self.warm, 1)])
+        try:
+            self.cloud(['gcloud', 'run', 'services', 'update-traffic', self.service, *self.common,
+                        '--to-revisions=' + previous + '=99,' + self.warm + '=1', '--quiet'])
+        finally:
+            # A timeout may follow an applied split. Read once, never replay;
+            # cleanup accepts only the exact original or intended distribution.
+            self.expected = self.service_read()
+        if (self.expected.get('metadata', {}).get('uid') != self.before['metadata']['uid']
+                or active(self.expected) != self.micro_traffic):
+            raise RuntimeError('candidate micro traffic split readback mismatch')
+
     def promote(self, value: dict) -> None:
         self.stable()
         # Baseline validation can outlast the earlier health observation.
@@ -186,7 +216,18 @@ def restore(self) -> str | None:
         self.warm = warm['metadata']['name']
         serving = active(current) == [(self.warm, 100)]
         if not serving and active(current) != active(self.before):
-            raise RuntimeError('candidate cleanup traffic identity changed')
+            if self.micro_traffic is None or active(current) != self.micro_traffic:
+                raise RuntimeError('candidate cleanup traffic identity changed')
+            self.stable()
+            previous = active(self.before)[0][0]
+            self.cloud(['gcloud', 'run', 'services', 'update-traffic', self.service, *self.common,
+                        '--to-revisions=' + previous + '=100', '--quiet'])
+            self.expected = self.service_read()
+            if (self.expected.get('metadata', {}).get('uid') != current.get('metadata', {}).get('uid')
+                    or active(self.expected) != active(self.before)):
+                raise RuntimeError('candidate cleanup prior traffic restore mismatch')
+            self.candidate(self.expected, '1')
+            current = self.expected
         tag_removed = True
         if self.tag:
             rows = current['status']['traffic']
@@ -243,6 +284,10 @@ def execute(self) -> str | None:
                 raise ValueError('candidate deploy target binding mismatch')
         if '--no-traffic' not in self.deploy_args or '--deploy-health-check' not in self.deploy_args:
             raise ValueError('candidate deploy requires no-traffic and health check')
+        if self.tag is None and tag_policy.lifecycle_for_impossible_tag(
+                service_name=self.service, approved_service=self.approved_service,
+                warmup_traffic_approved=self.warmup_traffic_approved) != 'MICRO_SPLIT':
+            raise RuntimeError('TAG_IMPOSSIBLE: exact warmup approval required')
         # Exclusive create persists across crashes/restarts; no automatic replay.
         try:
             fd = os.open(self.state_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
@@ -259,7 +304,10 @@ def execute(self) -> str | None:
         previous = self.revision_read(initial_traffic[0][0])
         if str(previous.get('metadata', {}).get('annotations', {}).get(MINIMUM, '0')) != '0':
             raise RuntimeError('candidate previous serving revision minimum is not zero')
-        if self.tag and any(row.get('tag') == self.tag for row in self.before['status']['traffic']):
+        allocation = tag_policy.allocate_tag(self.service, 'cand-' + self.commit[:8],
+            existing_tags=[row['tag'] for row in self.before['status']['traffic'] if row.get('tag')],
+            operation_id=self.operation_id)
+        if allocation.status == 'REFUSE' or allocation != self.allocation:
             raise RuntimeError('candidate tag collision; no mutation')
         if str(self.before.get('spec', {}).get('template', {}).get('metadata', {}).get('annotations', {}).get(MINIMUM, '0')) != '0':
             raise RuntimeError('candidate requires previous revision minimum zero')
@@ -280,11 +328,17 @@ def execute(self) -> str | None:
             self.warm = warm['metadata']['name']
             if active(self.expected) != active(self.before):
                 raise RuntimeError('candidate deploy unexpectedly moved traffic')
-            self.readback('set')
-            self.health(warm)
             if self.no_traffic_verify is None:
                 raise RuntimeError('candidate original no-traffic verification missing')
-            self.no_traffic_verify()
+            if self.tag is None:
+                # Verify the unchanged baseline before temporarily routing 1%.
+                self.no_traffic_verify()
+                self.start_untagged()
+                warm = self.candidate(self.expected, '1')
+            self.readback('set')
+            self.health(warm)
+            if self.tag is not None:
+                self.no_traffic_verify()
             self.stable()
             self.promote(warm)
             self.record('promoted')
@@ -334,7 +388,8 @@ def main(argv=None) -> int:
                           service=os.environ['INPUT_CONFIGURED_SERVICE'], commit=os.environ['SOURCE_COMMIT'],
                           digest=args.image.split('@')[-1], run_id=os.environ['GITHUB_RUN_ID'],
                           deploy_args=args.deploy_args[1:] if args.deploy_args[:1] == ['--'] else args.deploy_args,
-                          state_path=args.state_path, no_traffic_verify=lambda: _verify(baseline_args))
+                          state_path=args.state_path, no_traffic_verify=lambda: _verify(baseline_args),
+                          warmup_traffic_approved=True, approved_service=baseline_args.service)
         def interrupted(signum, frame):
             # A catchable abort enters finally. Once cleanup starts, finish its
             # bounded attempts instead of interrupting the restoration itself.