A5-R3-R3A dormant dual-path snapshot identity

core/generation_runner.py

@@ -221,6 +221,7 @@ class GenerationRunner:
         executor_instance_id: str = "",
         confirmed_checkpoint_seq: int | None = None,
         confirmed_checkpoint_stage: str = "",
+        trusted_task_policy_snapshot: dict[str, Any] | None = None,
         progress_callback: ProgressCallback | None = None,
         cancel_checker: Callable[[], bool] | None = None,
     ):
@@ -232,6 +233,9 @@ class GenerationRunner:
             None if confirmed_checkpoint_seq is None else max(0, int(confirmed_checkpoint_seq))
         )
         self.confirmed_checkpoint_stage = str(confirmed_checkpoint_stage or "")[:120]
+        self.trusted_task_policy_snapshot = copy.deepcopy(
+            dict(trusted_task_policy_snapshot or {})
+        )
         self.progress_callback = progress_callback or (lambda **_: None)
         self.cancel_checker = cancel_checker or (lambda: False)
         expected_hash = str(self.request.get("request_hash") or "")
@@ -2809,6 +2813,87 @@ class GenerationRunner:
                 output.append(str(path))
         return output
 
+    def _validate_trusted_task_policy_for_execution(self, mode: str, target_words: int) -> dict[str, Any] | None:
+        snapshot = copy.deepcopy(self.trusted_task_policy_snapshot)
+        if not snapshot:
+            return None
+        resolved = snapshot.get("resolved_policy")
+        if not isinstance(resolved, dict):
+            raise RuntimeError("trusted_task_policy_snapshot_invalid")
+
+        task_id = str(snapshot.get("task_id") or "").strip()
+        request_task_id = str(self.request.get("task_id") or "").strip()
+        if not task_id or task_id != request_task_id:
+            raise RuntimeError("trusted_task_policy_task_mismatch")
+
+        tier = str(snapshot.get("product_tier") or "").strip().upper()
+        entitlement_class = str(snapshot.get("entitlement_class") or "").strip().upper()
+        if tier not in {"STANDARD", "FLAGSHIP"}:
+            raise RuntimeError("trusted_task_policy_product_tier_invalid")
+        if entitlement_class not in {"STANDARD_TRIAL", "STANDARD_PAID", "FLAGSHIP_PAID"}:
+            raise RuntimeError("trusted_task_policy_entitlement_invalid")
+        if tier == "STANDARD" and entitlement_class == "FLAGSHIP_PAID":
+            raise RuntimeError("trusted_task_policy_entitlement_mismatch")
+        if tier == "FLAGSHIP" and entitlement_class != "FLAGSHIP_PAID":
+            raise RuntimeError("trusted_task_policy_entitlement_mismatch")
+        if str(resolved.get("product_tier") or "").strip().upper() != tier:
+            raise RuntimeError("trusted_task_policy_product_tier_mismatch")
+        if str(resolved.get("snapshot_schema_version") or "") != str(snapshot.get("snapshot_schema_version") or ""):
+            raise RuntimeError("trusted_task_policy_schema_mismatch")
+        for field in ("product_policy_version", "word_count_policy_version", "model_policy_version", "cost_policy_version"):
+            if str(resolved.get(field) or "") != str(snapshot.get(field) or ""):
+                raise RuntimeError("trusted_task_policy_version_mismatch")
+        for field in ("billing_policy_version", "trial_policy_version"):
+            resolved_value = str(resolved.get(field) or "").strip() or None
+            snapshot_value = str(snapshot.get(field) or "").strip() or None
+            if resolved_value != snapshot_value:
+                raise RuntimeError("trusted_task_policy_version_mismatch")
+        if str(resolved.get("manuscript_mode") or "").strip().lower() != mode:
+            raise RuntimeError("trusted_task_policy_mode_mismatch")
+        minimum = int(snapshot.get("min_body_words") or 0)
+        maximum = int(snapshot.get("max_body_words") or 0)
+        if minimum <= 0 or maximum < minimum or not minimum <= target_words <= maximum:
+            raise RuntimeError("trusted_task_policy_word_range_mismatch")
+        word_count = resolved.get("word_count") if isinstance(resolved.get("word_count"), dict) else {}
+        if (
+            int(word_count.get("min_body_words") or 0) != minimum
+            or int(word_count.get("max_body_words") or 0) != maximum
+            or str(word_count.get("counting_contract_version") or "") != str(snapshot.get("counting_contract_version") or "")
+        ):
+            raise RuntimeError("trusted_task_policy_word_policy_mismatch")
+
+        envelope = {
+            "task_id": task_id,
+            "snapshot_schema_version": str(snapshot.get("snapshot_schema_version") or ""),
+            "product_tier": str(snapshot.get("product_tier") or ""),
+            "product_policy_version": str(snapshot.get("product_policy_version") or ""),
+            "word_count_policy_version": str(snapshot.get("word_count_policy_version") or ""),
+            "model_policy_version": str(snapshot.get("model_policy_version") or ""),
+            "cost_policy_version": str(snapshot.get("cost_policy_version") or ""),
+            "billing_policy_version": str(snapshot.get("billing_policy_version") or "").strip() or None,
+            "trial_policy_version": str(snapshot.get("trial_policy_version") or "").strip() or None,
+            "entitlement_class": str(snapshot.get("entitlement_class") or "").strip() or None,
+            "min_body_words": minimum,
+            "max_body_words": maximum,
+            "counting_contract_version": str(snapshot.get("counting_contract_version") or ""),
+            "resolved_policy": resolved,
+        }
+        observed_sha = hashlib.sha256(
+            json.dumps(envelope, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        ).hexdigest()
+        if observed_sha != str(snapshot.get("snapshot_sha256") or "").strip().lower():
+            raise RuntimeError("trusted_task_policy_snapshot_sha_mismatch")
+
+        model = resolved.get("model") if isinstance(resolved.get("model"), dict) else {}
+        routes = model.get("routes") if isinstance(model.get("routes"), dict) else {}
+        if not routes:
+            raise RuntimeError("trusted_task_policy_model_routes_missing")
+        # A5-R3-R3A closes snapshot identity only. It deliberately does not
+        # reinterpret fine-grained logical-node routes as the legacy coarse
+        # capability roles. Product execution remains fail-closed until the
+        # separate logical-node execution contract is implemented.
+        raise RuntimeError("trusted_task_policy_model_route_contract_not_ready")
+
     def _run_impl(self) -> dict[str, Any]:
         req = self.request
         mode = str(req.get("mode") or "").lower()
@@ -2817,6 +2902,9 @@ class GenerationRunner:
         target_words = int(req.get("target_words") or 0)
         if not APP_CONFIG["target_words_min"] <= target_words <= APP_CONFIG["target_words_max"]:
             raise RuntimeError("target_words_out_of_range")
+        trusted_task_policy_snapshot = self._validate_trusted_task_policy_for_execution(
+            mode, target_words
+        )
 
         # A2: provider credentials are adapter-owned. DeepSeek remains the
         # current default adapter, but the runner no longer hard-codes it as the
@@ -2851,6 +2939,11 @@ class GenerationRunner:
         quality_diagnostics = list(state.get("quality_diagnostics") or [])
         quality_notes = list(state.get("quality_notes") or [])
         state.setdefault("plan_policy_snapshot", copy.deepcopy(plan_policy_snapshot))
+        if trusted_task_policy_snapshot is not None:
+            state.setdefault(
+                "trusted_task_policy_snapshot",
+                copy.deepcopy(trusted_task_policy_snapshot),
+            )
         self._evidence_coverage_events = list(state.get("evidence_coverage_events") or [])
         if str(state.get("evidence_status") or "") == "limited":
             self._preview_evidence_status = "limited"

managed_auth.py

@@ -342,24 +342,26 @@ def start_generation_task(
         raise RuntimeError("idempotency_key_required")
     projection = get_product_ui_projection()
     normalized_tier = str(product_tier or "").strip().upper()
-    if projection.get("mode") == "a5_product":
+    product_mode = projection.get("mode") == "a5_product"
+    if product_mode:
         if normalized_tier not in projection.get("products", {}):
             raise RuntimeError("product_tier_required")
-        # A5-R3-R1 is projection-only. Until entitlement reservation/task policy
-        # snapshot ownership is separately authorized, never fall through to the
-        # legacy quota task path.
-        raise RuntimeError("a5_product_task_path_not_authorized")
-    if normalized_tier:
+    elif normalized_tier:
         raise RuntimeError("product_tier_not_available_in_legacy_mode")
+    payload = {
+        "idempotency_key": normalized,
+        "mode": str(mode or "").strip().lower(),
+        "interface_language": str(interface_language or "zh").strip(),
+        "target_words": int(target_words),
+    }
+    if product_mode:
+        # Product tier is user intent only. Worker/D1 resolves entitlement and
+        # freezes the trusted TaskPolicySnapshot; browser state is never authority.
+        payload["product_tier"] = normalized_tier
     result = _edge_call(
         "task_start",
         normalized,
-        {
-            "idempotency_key": normalized,
-            "mode": str(mode or "").strip().lower(),
-            "interface_language": str(interface_language or "zh").strip(),
-            "target_words": int(target_words),
-        },
+        payload,
         key_prefix="sp_task_start",
     )
     if result is None:

task_executor.py

@@ -11,6 +11,7 @@ Worker so quota is never left reserved by an unowned task.
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import logging
 import os
@@ -409,6 +410,106 @@ def _cleanup_user_cancelled_task_artifacts(task_dir: Path, request_data: dict) -
         logging.warning("[SP_EXECUTOR] cancel_task_cleanup_error task_id=%s error=%s", task_dir.name, _safe(exc, 160))
 
 
+def _snapshot_nullable_version(value):
+    text = str(value or "").strip()
+    return text or None
+
+
+def _trusted_task_policy_snapshot_envelope(snapshot: dict, resolved: dict) -> dict:
+    return {
+        "task_id": str(snapshot.get("task_id") or ""),
+        "snapshot_schema_version": str(snapshot.get("snapshot_schema_version") or ""),
+        "product_tier": str(snapshot.get("product_tier") or ""),
+        "product_policy_version": str(snapshot.get("product_policy_version") or ""),
+        "word_count_policy_version": str(snapshot.get("word_count_policy_version") or ""),
+        "model_policy_version": str(snapshot.get("model_policy_version") or ""),
+        "cost_policy_version": str(snapshot.get("cost_policy_version") or ""),
+        "billing_policy_version": _snapshot_nullable_version(snapshot.get("billing_policy_version")),
+        "trial_policy_version": _snapshot_nullable_version(snapshot.get("trial_policy_version")),
+        "entitlement_class": _snapshot_nullable_version(snapshot.get("entitlement_class")),
+        "min_body_words": int(snapshot.get("min_body_words") or 0),
+        "max_body_words": int(snapshot.get("max_body_words") or 0),
+        "counting_contract_version": str(snapshot.get("counting_contract_version") or ""),
+        "resolved_policy": resolved,
+    }
+
+
+def _validate_trusted_task_policy_snapshot(claim: dict, task: dict, request_data: dict) -> dict | None:
+    raw = claim.get("task_policy_snapshot")
+    if raw is None:
+        return None
+    if not isinstance(raw, dict):
+        raise RuntimeError("task_policy_snapshot_invalid")
+    snapshot = dict(raw)
+    resolved = snapshot.get("resolved_policy")
+    if not isinstance(resolved, dict):
+        raise RuntimeError("task_policy_snapshot_resolved_policy_missing")
+
+    snapshot_task_id = str(snapshot.get("task_id") or "").strip()
+    claimed_task_id = str(task.get("task_id") or "").strip()
+    request_task_id = str(request_data.get("task_id") or "").strip()
+    if not snapshot_task_id or snapshot_task_id != claimed_task_id or snapshot_task_id != request_task_id:
+        raise RuntimeError("task_policy_snapshot_task_mismatch")
+
+    tier = str(snapshot.get("product_tier") or "").strip().upper()
+    entitlement_class = str(snapshot.get("entitlement_class") or "").strip().upper()
+    if tier not in {"STANDARD", "FLAGSHIP"}:
+        raise RuntimeError("task_policy_snapshot_product_tier_invalid")
+    if entitlement_class not in {"STANDARD_TRIAL", "STANDARD_PAID", "FLAGSHIP_PAID"}:
+        raise RuntimeError("task_policy_snapshot_entitlement_invalid")
+    if tier == "STANDARD" and entitlement_class == "FLAGSHIP_PAID":
+        raise RuntimeError("task_policy_snapshot_entitlement_mismatch")
+    if tier == "FLAGSHIP" and entitlement_class != "FLAGSHIP_PAID":
+        raise RuntimeError("task_policy_snapshot_entitlement_mismatch")
+
+    if str(resolved.get("snapshot_schema_version") or "") != str(snapshot.get("snapshot_schema_version") or ""):
+        raise RuntimeError("task_policy_snapshot_schema_mismatch")
+    if str(resolved.get("product_tier") or "").strip().upper() != tier:
+        raise RuntimeError("task_policy_snapshot_product_tier_mismatch")
+    version_fields = (
+        "product_policy_version",
+        "word_count_policy_version",
+        "model_policy_version",
+        "cost_policy_version",
+    )
+    for field in version_fields:
+        if str(resolved.get(field) or "") != str(snapshot.get(field) or ""):
+            raise RuntimeError("task_policy_snapshot_policy_version_mismatch")
+    for field in ("billing_policy_version", "trial_policy_version"):
+        if _snapshot_nullable_version(resolved.get(field)) != _snapshot_nullable_version(snapshot.get(field)):
+            raise RuntimeError("task_policy_snapshot_policy_version_mismatch")
+
+    if str(resolved.get("manuscript_mode") or "").strip().lower() != str(task.get("mode") or "").strip().lower():
+        raise RuntimeError("task_policy_snapshot_mode_mismatch")
+    target_words = int(task.get("target_words") or 0)
+    if target_words != int(request_data.get("target_words") or 0):
+        raise RuntimeError("target_words_claim_mismatch")
+    minimum = int(snapshot.get("min_body_words") or 0)
+    maximum = int(snapshot.get("max_body_words") or 0)
+    if minimum <= 0 or maximum < minimum or not minimum <= target_words <= maximum:
+        raise RuntimeError("task_policy_snapshot_word_range_mismatch")
+    word_count = resolved.get("word_count") if isinstance(resolved.get("word_count"), dict) else {}
+    if (
+        int(word_count.get("min_body_words") or 0) != minimum
+        or int(word_count.get("max_body_words") or 0) != maximum
+        or str(word_count.get("counting_contract_version") or "") != str(snapshot.get("counting_contract_version") or "")
+    ):
+        raise RuntimeError("task_policy_snapshot_word_policy_mismatch")
+
+    observed = hashlib.sha256(
+        json.dumps(
+            _trusted_task_policy_snapshot_envelope(snapshot, resolved),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    ).hexdigest()
+    expected = str(snapshot.get("snapshot_sha256") or "").strip().lower()
+    if not re.fullmatch(r"[a-f0-9]{64}", expected) or observed != expected:
+        raise RuntimeError("task_policy_snapshot_sha_mismatch")
+    return snapshot
+
+
 def _load_request(task_dir: Path) -> dict:
     path = task_dir / "request.json"
     if not path.is_file():
@@ -441,6 +542,11 @@ def _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data
         raise RuntimeError("executor_claim_identity_mismatch")
     if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
         raise RuntimeError("target_words_claim_mismatch")
+    trusted_task_policy_snapshot = _validate_trusted_task_policy_snapshot(
+        claim, task, request_data
+    )
+    if trusted_task_policy_snapshot is not None:
+        task["_trusted_task_policy_snapshot"] = trusted_task_policy_snapshot
     try:
         with checkpoint_authority_lock(task_dir):
             _atomic_json(
@@ -454,6 +560,12 @@ def _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data
                     "executor_instance_id": instance_id,
                     "worker_checkpoint_seq": int(task.get("checkpoint_seq") or 0),
                     "worker_checkpoint_stage": str(task.get("checkpoint_stage") or "")[:120],
+                    "task_policy_snapshot_sha256": str(
+                        (trusted_task_policy_snapshot or {}).get("snapshot_sha256") or ""
+                    ),
+                    "task_policy_product_tier": str(
+                        (trusted_task_policy_snapshot or {}).get("product_tier") or ""
+                    ),
                     "claimed_at": datetime.now(timezone.utc).isoformat(),
                 },
             )
@@ -569,6 +681,7 @@ def main() -> int:
             executor_instance_id=instance_id,
             confirmed_checkpoint_seq=remote_checkpoint_seq,
             confirmed_checkpoint_stage=remote_checkpoint_stage,
+            trusted_task_policy_snapshot=task.get("_trusted_task_policy_snapshot"),
             progress_callback=state.touch,
             cancel_checker=state.cancelled,
         )