feat(operators): add fail-closed Cloud Run tag policy

docs/README.md

@@ -12,6 +12,8 @@ Read documents at the library revision used by your consumer. These are source-l
 
 ## Runtime contracts
 
+- [Cloud Run revision tags](cloud-run-tags.md): pinned offline allocation, exact-target warmup policy and caller restore requirements.
+
 - [One-time setup intent](strategy-release-runtime-contract.zh-CN.md#一次确认的自动化-setup-intent): immutable, non-executing owner scope and Decisions handoff; no runtime adoption or activation authority.
 
 - [Platform notification outcomes](platform-notification-outcomes.md): credential-free intents, receipts and log redaction.

docs/cloud-run-tags.md

@@ -0,0 +1,107 @@
+# Cloud Run revision tags
+
+`quant_platform_kit.operators` exports the pinned offline operator contract
+`cloud-run-tags/v1`. It performs no network calls, deployment, traffic updates,
+renames, IAM changes or credential access. Runtime trading controls are outside
+this module. Existing callers acquire no additional authority by importing it.
+
+## Allocation
+
+Call `headroom(service_name)` on the exact original service name, before display
+redaction. The tagged URL budget is `len(service_name) + len(tag) <= 46`.
+`allocate_tag` takes the exact service name, preferred tag, project, region,
+candidate revision, operation/run identity, a complete fresh `existing_tags`
+mapping, and a private job-local `state_path`.
+
+- A fitting preferred tag is selected only when free or bound to this exact
+  operation and candidate in the journal. A cloud revision match alone does
+  not prove ownership. Foreign or unknown occupancy returns `TAG_OCCUPIED`.
+  The first occupancy refusal durably initializes the journal without reserving
+  a tag, so unrelated allocations can continue using the same journal.
+- With headroom 1–12, an oversized preferred tag becomes one deterministic
+  hash-derived legal tag, no longer than headroom. Collisions return
+  `TAG_RESERVED` or `TAG_OCCUPIED`; there is no retry search or overwrite.
+- Headroom at or below zero returns `Refusal('TAG_IMPOSSIBLE')`. Invalid input
+  and unreadable, corrupt, incompatible or replaced state also return explicit
+  refusals. Callers must branch on the result type and stop on every refusal;
+  truthiness is not a success check.
+
+Use an absent journal path in an existing private directory for the first
+allocation. Only this module creates the mode-0600 file. Every process in the
+same job must share it; preserve it across restarts. Never unlink, replace,
+pre-create, truncate, copy to Git or reset it to recover from a failure. Missing
+state after a restart requires operator reconciliation, not a fresh allocation.
+The allocator holds a Linux `flock`, reads the current journal after locking,
+checks the locked file's identity, then flushes/fsyncs the reservation before
+returning. An interrupted in-place write can invalidate the journal and must
+stop subsequent allocation. The private directory and its parent must remain
+under exclusive caller control; this is not a cross-host lock or cloud lease.
+
+The journal binds project/region/service, operation, candidate and preferred tag
+using full hashes, retaining only the operational tag as plaintext. Hashes are
+private correlation data, not anonymisation. Returned tags are operational
+values; do not log them. Dataclass representations suppress operational fields;
+refusal messages contain static codes and never raw exceptions. For any caller
+logs or API output, use the existing shared masking helper, including
+`quant_platform_kit.notifications._redaction.mask_identifier` for resource
+identifiers. Never mask an identifier before budgeting or identity comparisons.
+
+The caller's `existing_tags` maps **every** tag to its exact revision, including
+zero-percent entries. An unknown revision is `None` and remains occupied.
+Allocation leaves this mapping untouched. An allocation recovered after restart
+is the same reservation, not an execution or replay permit. No release, expiry
+or ownership-stealing operation is provided.
+
+## Impossible-tag lifecycle
+
+`lifecycle_for_impossible_tag` defaults to `STOP` and never claims an untagged
+zero-percent candidate was warmed. A `MICRO_SPLIT` result proposes exactly
+previous=99 / candidate=1, only when all these conditions hold:
+
+- The service has no nonempty tag headroom.
+- The caller verified an authentic owner decision and supplied a
+  `WarmupApproval` with `warmup_traffic_approved=True` (a literal boolean).
+- Project, region, service, action=`warmup`, and operation/run identity match
+  exactly. An approval for promotion, another run or another service cannot
+  widen the scope.
+- `no_traffic` is explicitly `False` and `image_only` remains `False`; truthy
+  strings and unknown values cannot satisfy either gate.
+
+An approval object is policy input, not authentication. Callers must verify its
+owner, validity, exact candidate and scope at their existing protected entry
+point. Strict no-traffic diagnostics and image-only callers always stop on
+impossible tags; this helper cannot grant them micro-traffic authority.
+
+## Caller mutation and restore contract
+
+Before any cloud write, the caller must hold a current exclusive execution
+claim covering the exact target/run/candidate, durably record a single-use
+mutation intent, and re-read the current service generation, image/source,
+revision identity, traffic distribution and complete tags. The local allocation
+journal does not serialize cloud writes across different jobs or hosts. Bind
+approval to the exact candidate; do not reuse an operation ID for another image
+or revision. Authenticate evidence using the caller's existing gates.
+
+Preserve all pre-existing tags and traffic entries, including zero-percent tags,
+and compare the current cloud state against the approved before-image before
+every stage. Use the caller's existing conditional-write/ownership mechanism;
+a preflight read alone cannot prevent cloud races. A lost claim, state drift,
+unknown write result or timeout requires readback and reconciliation, never an
+automatic replay. A matching allocated tag is not permission to overwrite it.
+
+For an explicitly approved micro-split, verify exactly the permitted 99/1 split
+and candidate startup/health. Restore the approved previous distribution on
+failure, preserving every existing tag, then read it back before cleanup or a
+subsequent stage. If ownership is lost or restoration cannot be verified, stop
+all further writes and report `RESTORE_UNVERIFIED`. The policy supplies neither
+gcloud mutation commands nor a claim that restore is atomic or always possible.
+
+## Adoption
+
+IBKR, Firstrade, LongBridge and QRS adopters must pin the reviewed QPK revision
+or vendor this module with its contract version and offline tests. Replace
+local formulas only within separately authorized caller changes, retaining the
+entry point's existing traffic restrictions and identity gates. Confirm actual
+caller use and pinned version before claiming adoption. This module alone
+changes no deployed service or runtime behavior. Long-name service migration
+remains deferred.

src/quant_platform_kit/operators/__init__.py

@@ -0,0 +1,17 @@
+"""Pinned operator policies; no cloud mutations or runtime trading imports."""
+from .cloud_run_tags import (
+    CONTRACT_VERSION,
+    AllocatedTag,
+    Lifecycle,
+    Refusal,
+    Target,
+    WarmupApproval,
+    allocate_tag,
+    headroom,
+    lifecycle_for_impossible_tag,
+)
+
+__all__ = [
+    'CONTRACT_VERSION', 'AllocatedTag', 'Lifecycle', 'Refusal', 'Target',
+    'WarmupApproval', 'allocate_tag', 'headroom', 'lifecycle_for_impossible_tag',
+]

src/quant_platform_kit/operators/cloud_run_tags.py

@@ -0,0 +1,230 @@
+"""Versioned offline Cloud Run tag policy. This module never calls a cloud API.
+
+Allocation is an exclusive local reservation, NOT permission to mutate a service.
+See docs/cloud-run-tags.md for inventory, ownership and restore requirements.
+"""
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import stat
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+
+CONTRACT_VERSION = 'cloud-run-tags/v1'
+TAGGED_URL_BUDGET = 46
+_NAME = re.compile(r'[a-z](?:[a-z0-9-]*[a-z0-9])?\Z')
+_DIGEST = re.compile(r'[0-9a-f]{64}\Z')
+
+
+@dataclass(frozen=True)
+class Refusal:
+    """Explicit failure with a static, identifier-free reason code."""
+    reason: str
+
+
+@dataclass(frozen=True)
+class AllocatedTag:
+    """Operational value: do not log the tag or serialize it into public output."""
+    tag: str = field(repr=False)
+
+
+@dataclass(frozen=True)
+class Target:
+    project: str = field(repr=False)
+    region: str = field(repr=False)
+    service_name: str = field(repr=False)
+
+
+@dataclass(frozen=True)
+class WarmupApproval:
+    """Caller-verified owner decision; this value cannot authenticate an approval."""
+    target: Target = field(repr=False)
+    action: str = field(repr=False)
+    operation_id: str = field(repr=False)
+    warmup_traffic_approved: bool = field(repr=False)
+
+
+@dataclass(frozen=True)
+class Lifecycle:
+    kind: str
+    reason: str
+    previous_percent: int = 0
+    candidate_percent: int = 0
+
+
+def headroom(service_name: str) -> int:
+    """Measure the exact original name, before any display masking."""
+    return TAGGED_URL_BUDGET - len(service_name)
+
+
+def _nonempty(value: object) -> bool:
+    return isinstance(value, str) and bool(value.strip()) and value == value.strip()
+
+
+def _name(value: object) -> bool:
+    return isinstance(value, str) and bool(_NAME.fullmatch(value))
+
+
+def _target_valid(target: object) -> bool:
+    return (isinstance(target, Target) and _nonempty(target.project)
+            and _nonempty(target.region) and _name(target.service_name))
+
+
+def lifecycle_for_impossible_tag(
+    target: Target, action: str, operation_id: str, *,
+    approval: WarmupApproval | None = None, no_traffic: bool = True,
+    image_only: bool = False,
+) -> Lifecycle:
+    """Default STOP; permit only an exact, explicitly approved warmup 99/1 plan.
+
+    A returned plan still requires the caller's exclusive, single-use execution
+    claim and fresh cloud readback. It never grants deployment/traffic authority.
+    """
+    if not _target_valid(target) or not _nonempty(operation_id):
+        return Lifecycle('STOP', 'INVALID_INPUT')
+    if no_traffic is not False or image_only is not False:
+        return Lifecycle('STOP', 'NO_TRAFFIC_REQUIRED')
+    if headroom(target.service_name) > 0:
+        return Lifecycle('STOP', 'TAG_NOT_IMPOSSIBLE')
+    if (action != 'warmup' or not isinstance(approval, WarmupApproval)
+            or approval.warmup_traffic_approved is not True
+            or approval.target != target or approval.action != action
+            or approval.operation_id != operation_id):
+        return Lifecycle('STOP', 'EXACT_WARMUP_APPROVAL_REQUIRED')
+    return Lifecycle('MICRO_SPLIT', 'EXACT_WARMUP_APPROVAL', 99, 1)
+
+
+def _digest(*parts: str) -> str:
+    return hashlib.sha256(json.dumps(parts, separators=(',', ':')).encode()).hexdigest()
+
+
+def _short_tag(scope: str, operation: str, candidate: str, size: int) -> str:
+    digest = _digest(CONTRACT_VERSION, scope, operation, candidate)
+    # Always begin with a letter, including the one-character boundary.
+    return (chr(ord('a') + int(digest[:2], 16) % 26) + digest[2:])[:size]
+
+
+def _journal_valid(data: object) -> bool:
+    if (not isinstance(data, dict) or set(data) != {'contract_version', 'scopes'}
+            or data['contract_version'] != CONTRACT_VERSION
+            or not isinstance(data['scopes'], dict)):
+        return False
+    for scope, records in data['scopes'].items():
+        if (not isinstance(scope, str) or not _DIGEST.fullmatch(scope)
+                or not isinstance(records, dict)):
+            return False
+        tags = set()
+        for operation, record in records.items():
+            if (not isinstance(operation, str) or not _DIGEST.fullmatch(operation)
+                    or not isinstance(record, dict)
+                    or set(record) != {'candidate', 'preferred', 'tag'}
+                    or any(not isinstance(record[k], str) or not _DIGEST.fullmatch(record[k])
+                           for k in ('candidate', 'preferred'))
+                    or not _name(record['tag']) or record['tag'] in tags):
+                return False
+            tags.add(record['tag'])
+    return True
+
+
+def _same_file(path: Path, fd: int) -> bool:
+    current = os.stat(path, follow_symlinks=False)
+    held = os.fstat(fd)
+    return (stat.S_ISREG(current.st_mode) and current.st_nlink == 1
+            and (current.st_dev, current.st_ino) == (held.st_dev, held.st_ino)
+            and stat.S_IMODE(current.st_mode) == 0o600)
+
+
+def _reserve(data: dict, scope: str, operation: str, candidate: str,
+             preferred: str, tag: str, existing: Mapping) -> AllocatedTag | Refusal:
+    records = data['scopes'].setdefault(scope, {})
+    record = {'candidate': candidate, 'preferred': preferred, 'tag': tag}
+    previous = records.get(operation)
+    if previous is not None and previous != record:
+        return Refusal('OPERATION_REBOUND')
+    if any(key != operation and value['tag'] == tag for key, value in records.items()):
+        return Refusal('TAG_RESERVED')
+    # A cloud revision match alone does not prove operation ownership.
+    if tag in existing and (previous != record or _digest(existing[tag]) != candidate):
+        return Refusal('TAG_OCCUPIED')
+    records[operation] = record
+    return AllocatedTag(tag)
+
+
+def allocate_tag(
+    service_name: str, preferred_tag: str, *, existing_tags: Mapping[str, str | None],
+    operation_id: str, candidate_id: str, state_path: str | Path,
+    project: str, region: str,
+) -> AllocatedTag | Refusal:
+    """Reserve preferred or one deterministic shortened tag; never overwrite.
+
+    existing_tags must come from a fresh, complete inventory (including zero-
+    percent entries); None as a value is occupied with unknown ownership. Use
+    one private, shared job-local state_path for all processes/restarts. Only
+    the allocator creates this file. Missing evidence or state I/O fails closed.
+    """
+    if (not _target_valid(Target(project, region, service_name))
+            or not _name(preferred_tag) or not _nonempty(operation_id)
+            or not _nonempty(candidate_id) or not isinstance(existing_tags, Mapping)
+            or any(not _name(k) or (v is not None and not _nonempty(v))
+                   for k, v in existing_tags.items())):
+        return Refusal('INVALID_INPUT')
+    room = headroom(service_name)
+    if room <= 0:
+        return Refusal('TAG_IMPOSSIBLE')
+    scope = _digest(project, region, service_name)
+    operation, candidate, preferred = map(_digest, (operation_id, candidate_id, preferred_tag))
+    tag = preferred_tag if len(preferred_tag) <= room else _short_tag(scope, operation, candidate, room)
+    # Snapshot caller input without ever mutating it. Cloud freshness remains a
+    # caller obligation; the reservation ledger is read only AFTER flock.
+    existing = dict(existing_tags)
+    fd = None
+    try:
+        path = Path(state_path)
+        try:
+            fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+            created = True
+        except FileExistsError:
+            fd = os.open(path, os.O_RDWR | os.O_NOFOLLOW)
+            created = False
+        with os.fdopen(fd, 'r+', encoding='utf-8') as stream:
+            fd = None
+            fcntl.flock(stream.fileno(), fcntl.LOCK_EX)
+            if not _same_file(path, stream.fileno()):
+                return Refusal('STATE_UNAVAILABLE')
+            text = stream.read()
+            # Another process may have initialized a just-created file first.
+            data = json.loads(text) if text else ({'contract_version': CONTRACT_VERSION, 'scopes': {}} if created else None)
+            if not _journal_valid(data):
+                return Refusal('STATE_UNAVAILABLE')
+            result = _reserve(data, scope, operation, candidate, preferred, tag, existing)
+            # Persist a newly initialized journal even on occupancy refusal.
+            # An empty existing journal still fails validation above.
+            if isinstance(result, Refusal) and not (created and not text):
+                return result
+            if not _same_file(path, stream.fileno()):
+                return Refusal('STATE_UNAVAILABLE')
+            stream.seek(0)
+            json.dump(data, stream, sort_keys=True, separators=(',', ':'))
+            stream.truncate()
+            stream.flush()
+            os.fsync(stream.fileno())
+            # Persist the directory entry before returning a usable allocation.
+            directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
+            if not _same_file(path, stream.fileno()):
+                return Refusal('STATE_UNAVAILABLE')
+            return result
+    except (OSError, ValueError, TypeError, RecursionError):
+        # Do not include exception text: paths/resources may identify accounts.
+        return Refusal('STATE_UNAVAILABLE')
+    finally:
+        if fd is not None:
+            os.close(fd)