fix(persistence): roll back failed state replacements

.github/workflows/security.yml

@@ -44,8 +44,16 @@ jobs:
         with:
           version: "0.12.5"
 
+      - name: Build locked native decision measurement
+        run: |
+          set -euo pipefail
+          rustup show active-toolchain
+          uv sync --locked --extra api --extra db --extra queue --group dev --group native-build
+          uv run --no-sync maturin develop --locked --release --features pyo3/extension-module --manifest-path rust/decision_receipt/Cargo.toml
+          uv run --no-sync maturin build --locked --release --features pyo3/extension-module --manifest-path rust/decision_receipt/Cargo.toml --out "$RUNNER_TEMP/decision-wheels"
+
       - name: Run full test suite
-        run: uv run --locked --extra api --extra db --extra queue --group dev python -m pytest -q
+        run: uv run --no-sync python -m pytest -q
 
       - name: Install hash-locked quality tools
         run: |
@@ -74,12 +82,15 @@ jobs:
           set -euo pipefail
           rm -rf dist "$RUNNER_TEMP/nim-wheel-site"
           python -m pip wheel --no-deps --no-build-isolation . --wheel-dir dist
+          python scripts/verify_decision_wheel_manifest.py dist/contextual_orchestrator-*.whl "$RUNNER_TEMP"/decision-wheels/*.whl
           python -m pip install --no-deps \
             --target "$RUNNER_TEMP/nim-wheel-site" \
-            dist/contextual_orchestrator-*.whl
+            dist/contextual_orchestrator-*.whl "$RUNNER_TEMP"/decision-wheels/*.whl
           cd "$RUNNER_TEMP"
           PYTHONPATH="$RUNNER_TEMP/nim-wheel-site" \
-            python -c "import contextual_orchestrator; import contextual_orchestrator.nim_benchmark"
+            python -c "import pathlib, contextual_orchestrator as core; import contextual_orchestrator.nim_benchmark; import contextual_orchestrator._decision_receipt as native; root = pathlib.Path('nim-wheel-site').resolve(); assert pathlib.Path(core.__file__).is_relative_to(root); assert pathlib.Path(native.__file__).is_relative_to(root)"
+          PYTHONPATH="$RUNNER_TEMP/nim-wheel-site" \
+            python -m pytest --noconftest --import-mode=importlib "$GITHUB_WORKSPACE/tests/test_decision_receipts.py" -q
 
   fuzz:
     name: Property and coverage-guided fuzzing

AGENTS.md

@@ -1,5 +1,12 @@
 # AGENTS.md
 
+Deferred batch lineage: read `docs/doctoring/batch_request_lineage.md` for the
+HTTP reproduction, atomic submission-event projection, and remote/local failure
+boundary. Do not retry a remotely submitted job after local lineage failure.
+
+Workflow origin identity and persistence limitations are documented in
+`docs/doctoring/workflow_request_link.md`; preserve origin on replacements and reload.
+
 Cross-agent conventions for `contextual-orchestrator`, readable by any coding
 agent (Claude, Codex, Cursor, opencode, …). Keep this file tool-agnostic.
 
@@ -50,6 +57,10 @@ push or open a PR.
 
 ### Code exploration
 
+- Provider logs need server-generated per-request identity, not a session hash.
+  Preserve context cleanup and validate the central collector before adoption.
+  Reproduction and exact evidence: `docs/doctoring/provider_request_correlation.md`.
+
 - This repo has **no `.codegraph/` index**, so use normal search
   (grep/ripgrep/find, file reads) to locate and understand code. If a
   `.codegraph/` directory is ever added at the repo root, prefer CodeGraph

CLAUDE.md

@@ -1,5 +1,12 @@
 # CLAUDE.md
 
+Batch request lineage evidence and unresolved registry failure semantics live
+in `docs/doctoring/batch_request_lineage.md`; HTTP 201 alone does not establish
+durable lineage. Preserve job-scoped item IDs and original submission identity.
+
+See `docs/doctoring/workflow_request_link.md` for request-to-workflow correlation
+tests, cache semantics, and the distinction between in-memory and durable outcomes.
+
 This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
 
 ## Read AGENTS.md first
@@ -27,6 +34,10 @@ This file complements AGENTS.md with commands and architecture; where they diffe
 
 ## Common commands
 
+For missing provider/error correlation, run the focused telemetry and debug-log
+tests in `docs/doctoring/provider_request_correlation.md`. A green local producer
+test does not prove the central collector preserves the new field.
+
 ```bash
 # Install (pinned, hash-locked — always this two-step form)
 python -m pip install --require-hashes -r requirements.lock

contextual_orchestrator/batch_job_registry.py

@@ -199,6 +199,11 @@ def __init__(self, client: Any = None, *, retention_seconds: int = DEFAULT_RETEN
         )
         self._local_locks_guard = threading.Lock()
 
+    @property
+    def retention_seconds(self) -> int:
+        """Return the configured registry retention for durable recovery expiry."""
+        return self._retention_seconds
+
     def lock(
         self,
         name: str,

contextual_orchestrator/batch_routing.py

@@ -218,6 +218,15 @@ class BatchJob:
     # Prompt-token fallback estimates are safe metadata, stored atomically with
     # the job handle rather than retaining submitted prompt text.
     prompt_token_estimates: Dict[str, int] = field(default_factory=dict)
+    # Finalized before the registry snapshot is written; the append-only event
+    # supplies the durable request association independently of that registry.
+    request_link_status: str = "unavailable"
+    recovery_status: str = "unavailable"
+    # Deliberately not a dataclass field: HSET may succeed before expiry fails,
+    # so an operation result must never be serialized into its own snapshot.
+    registry_persistence_status = "unavailable"
+    backend_registry_persistence_status = "unavailable"
+    recovered_request_metadata = None
 
 
 @dataclass
@@ -453,17 +462,75 @@ def __init__(
         endpoint: str = "/v1/chat/completions",
         payload_assembler: Any = None,
         job_registry: Any = None,
+        recovery_identity: str | None = None,
     ) -> None:
         self._client = client
         self._endpoint_alias = endpoint_alias
         self._endpoint = endpoint
         self._assembler = payload_assembler
+        if recovery_identity is not None and (not isinstance(recovery_identity, str) or not recovery_identity.strip()):
+            raise ValueError("recovery identity must be a nonempty operator-controlled identifier")
+        self._recovery_identity = recovery_identity
         # Tracked requests survive a restart when a Valkey-backed registry
         # is injected; a plain dict preserves the historical behavior.
         self._jobs: Dict[str, Dict[str, Any]] = (
             job_registry.mapping("pg_llm_batch_jobs") if job_registry is not None else {}
         )
 
+    @property
+    def recovery_enabled(self) -> bool:
+        """Whether the operator supplied a stable deployment/account binding."""
+        return self._recovery_identity is not None
+
+    def has_job_metadata(self, job: BatchJob) -> bool:
+        """Check whether active registry metadata supports the existing job."""
+        try:
+            document = self._jobs.get(job.job_id)
+        except Exception:
+            return False
+        return (isinstance(document, dict)
+                and document.get("endpoint_alias") == self._endpoint_alias
+                and document.get("recovery_identity") == self._recovery_identity
+                and (document.get("endpoint") == self._endpoint
+                     or ("endpoint" not in document and self._recovery_identity is None))
+                and isinstance(document.get("requests"), dict)
+                and len(document["requests"]) == job.request_count)
+
+    def recovery_descriptor(self, requests: List[BatchRequest]) -> Dict[str, Any]:
+        """Describe exact target and item metadata without submitted prompt text."""
+        from .cost_ledger import AttributionDimensions
+        return {
+            "recovery_identity": self._recovery_identity,
+            "backend_name": self.name,
+            "endpoint_alias": self._endpoint_alias,
+            "endpoint": self._endpoint,
+            "items": [{"custom_id": item.custom_id, "model": item.model,
+                       "mode": item.mode, "attribution": AttributionDimensions.from_mapping(item.attribution).as_dict()}
+                      for item in requests],
+        }
+
+    def restore_descriptor(self, job: BatchJob, descriptor: Dict[str, Any]) -> None:
+        """Restore prompt-free item identity only for this exact configured target."""
+        if (not isinstance(descriptor, dict) or descriptor.get("backend_name") != self.name
+                or self._recovery_identity is None
+                or descriptor.get("recovery_identity") != self._recovery_identity
+                or descriptor.get("endpoint_alias") != self._endpoint_alias
+                or descriptor.get("endpoint") != self._endpoint):
+            raise ValueError("batch target mismatch")
+        items = descriptor.get("items")
+        if not isinstance(items, list) or len(items) != job.request_count:
+            raise ValueError("batch item count mismatch")
+        restored = {}
+        for item in items:
+            if not isinstance(item, dict) or set(item) != {"custom_id", "model", "mode", "attribution"}:
+                raise ValueError("invalid batch item descriptor")
+            if any(not isinstance(item[field], str) or not item[field] for field in ("custom_id", "model", "mode")):
+                raise ValueError("invalid batch item identity")
+            if item["custom_id"] in restored or not isinstance(item["attribution"], dict):
+                raise ValueError("invalid batch item metadata")
+            restored[item["custom_id"]] = {**item, "messages": []}
+        job.recovered_request_metadata = restored
+
     def _assemble_payload(self, requests: List[BatchRequest]) -> str:
         if self._assembler is not None:
             return self._assembler.assemble(
@@ -496,18 +563,26 @@ async def _submit() -> Dict[str, Any]:
         # Tracked requests are stored as JSON primitives (not dataclass
         # instances) so the registry can be a JSON-backed Valkey mapping;
         # retrieve() rebuilds the dataclass view it needs.
-        self._jobs[batch_id] = {
-            "endpoint_alias": self._endpoint_alias,
-            "requests": {
-                request.custom_id: dataclasses.asdict(request) for request in requests
-            },
-        }
-        return BatchJob(
+        registry_status = "stored"
+        try:
+            self._jobs[batch_id] = {
+                "endpoint_alias": self._endpoint_alias,
+                "recovery_identity": self._recovery_identity,
+                "endpoint": self._endpoint,
+                "requests": {
+                    request.custom_id: dataclasses.asdict(request) for request in requests
+                },
+            }
+        except Exception:
+            registry_status = "write_failed"
+        job = BatchJob(
             job_id=batch_id,
             backend=self.name,
             status=job_payload.get("status", "validating"),
             request_count=len(requests),
         )
+        job.backend_registry_persistence_status = registry_status
+        return job
 
     def poll(self, job: BatchJob) -> Dict[str, Any]:
         """Poll batch status via the pg-llm-batch client."""
@@ -542,7 +617,8 @@ async def _download() -> Dict[str, Any]:
                 reason,
             )
             raise BatchDownloadError(job.job_id, reason)
-        tracked = self._jobs.get(job.job_id, {}).get("requests", {})
+        tracked = (job.recovered_request_metadata if job.recovered_request_metadata is not None
+                   else self._jobs.get(job.job_id, {}).get("requests", {}))
         responses = _validated_download_responses(
             payload,
             expected_custom_ids=set(tracked),

contextual_orchestrator/cost_router.py

@@ -20,10 +20,12 @@
 
 import hashlib
 import re
+import time
 from contextvars import ContextVar
-from dataclasses import replace
+from dataclasses import asdict, replace
 from threading import Lock
 from typing import Any, Dict, List, Optional
+from .decision_receipts import record_initial_selection
 
 from .batch_routing import (
     BatchBackend,
@@ -37,6 +39,7 @@
     LocalBatchBackend,
     LocalEmbeddingBatchBackend,
     ProviderEmbeddingBatchBackend,
+    PgLlmBatchBackend,
     RoutingHints,
     RoutingPolicy,
 )
@@ -972,6 +975,7 @@ def submit_batch(
         requests: List[BatchRequest],
         metadata: Optional[Dict[str, Any]] = None,
         owner_id: Optional[str] = None,
+        request_id: Optional[str] = None,
     ) -> BatchJob:
         """Submit a batch, resolve its targets, and bind its authenticated owner."""
         try:
@@ -995,7 +999,36 @@ def submit_batch(
         job = self.batch_backend.submit(prepared_requests, metadata=metadata)
         job.owner_id = owner_id
         job.prompt_token_estimates = prompt_token_estimates
-        self._batch_jobs[job.job_id] = job
+        if request_id is not None and self.orchestrator._store is not None:
+            try:
+                # One append-only submission envelope commits all item links
+                # together. A later retrieval never rewrites this origin.
+                self.orchestrator._store.save("batch_request_link", job.job_id, {
+                    "request_id": request_id,
+                    "batch_job_id": job.job_id,
+                    "custom_ids": [request.custom_id for request in prepared_requests],
+                    "owner_id": owner_id,
+                    "recovery_descriptor": ({
+                        "job": asdict(job),
+                        "expires_at": job.submitted_at + self._job_registry.retention_seconds,
+                        "backend": self.batch_backend.recovery_descriptor(prepared_requests),
+                    } if isinstance(self.batch_backend, PgLlmBatchBackend) else None),
+                }, durable=True)
+            except Exception:
+                # The upstream submission already happened. Preserve its handle
+                # and report incomplete lineage instead of inviting a resubmit.
+                job.request_link_status = "write_failed"
+            else:
+                job.request_link_status = "durable"
+                if isinstance(self.batch_backend, PgLlmBatchBackend) and self.batch_backend.recovery_enabled:
+                    job.recovery_status = "durable_descriptor"
+        job.registry_persistence_status = "stored"
+        try:
+            self._batch_jobs[job.job_id] = job
+        except Exception:
+            # Submission already applied remotely; an HSET/expiry failure may
+            # itself be partially applied. Return the handle without replay.
+            job.registry_persistence_status = "write_failed"
         return job
 
     def _resolve_batch_request(self, request: BatchRequest) -> BatchRequest:
@@ -1046,7 +1079,7 @@ def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict
             not self._batch_item_usage_valid(item)
             and item.custom_id not in prompt_token_estimates
             for item in items
-        )
+        ) and job.recovered_request_metadata is None
         request_by_custom_id = (
             self._legacy_batch_requests(job) if needs_legacy_lookup else {}
         )
@@ -1286,7 +1319,53 @@ def _resolve_batch_provider_model(self, item: BatchResultItem) -> tuple[str, str
         return provider, item.model
 
     def _require_job(self, job_id: str, *, owner_id: Optional[str] = None) -> BatchJob:
-        job = self._batch_jobs.get(job_id)
+        try:
+            job = self._batch_jobs.get(job_id)
+        except Exception:
+            job = None
+        if job is not None and job.owner_id != owner_id:
+            raise KeyError(f"batch job {job_id!r} not found")
+        if job is not None and isinstance(self.batch_backend, PgLlmBatchBackend):
+            if self.batch_backend.has_job_metadata(job):
+                return job
+            # Missing or differently bound metadata cannot use the ordinary
+            # retrieval path; only a validated durable descriptor may recover.
+            job = None
+        if (owner_id is not None and self.orchestrator._store is not None
+                and (job is None or (isinstance(self.batch_backend, PgLlmBatchBackend)
+                                     and self.batch_backend.recovery_enabled))):
+            record = self.orchestrator._store.load_latest_key("batch_request_link", job_id)
+            if (isinstance(record, dict) and record.get("owner_id") == owner_id
+                    and record.get("batch_job_id") == job_id
+                    and isinstance(self.batch_backend, PgLlmBatchBackend)):
+                descriptor = record.get("recovery_descriptor")
+                try:
+                    if not isinstance(descriptor, dict) or type(descriptor.get("expires_at")) is not int:
+                        raise ValueError("invalid descriptor")
+                    if descriptor["expires_at"] <= time.time():
+                        raise ValueError("expired descriptor")
+                    recovered = BatchJob(**descriptor["job"])
+                    if recovered.job_id != job_id or recovered.owner_id != owner_id or recovered.backend != self.batch_backend.name:
+                        raise ValueError("mismatched descriptor")
+                    custom_ids = record.get("custom_ids")
+                    if (type(recovered.request_count) is not int or recovered.request_count < 1
+                            or not isinstance(custom_ids, list)
+                            or any(not isinstance(item, str) or not item for item in custom_ids)
+                            or len(custom_ids) != recovered.request_count
+                            or len(set(custom_ids)) != recovered.request_count):
+                        raise ValueError("invalid recovery item identities")
+                    estimates = recovered.prompt_token_estimates
+                    if (not isinstance(estimates, dict) or not set(estimates).issubset(custom_ids)
+                            or any(type(value) is not int or value < 0 for value in estimates.values())):
+                        raise ValueError("invalid recovery estimates")
+                    self.batch_backend.restore_descriptor(recovered, descriptor["backend"])
+                    if set(recovered.recovered_request_metadata) != set(custom_ids):
+                        raise ValueError("mismatched recovery items")
+                    recovered.request_link_status = "durable"
+                    recovered.recovery_status = "durable_descriptor"
+                    job = recovered
+                except (KeyError, TypeError, ValueError):
+                    job = None
         if job is None or job.owner_id != owner_id:
             raise KeyError(f"batch job {job_id!r} not found")
         return job
@@ -1332,6 +1411,8 @@ def submit_embeddings_batch(
         if callable(reserve) and callable(start):
             job = reserve(requests, metadata=metadata)
         else:
+            if resolved_agent_id is not None:
+                record_initial_selection([resolved_agent_id], "embedding_submission")
             job = backend.submit(requests, metadata=metadata)
         self._embedding_models[job.job_id] = resolved_model
         self._embedding_owners[job.job_id] = owner_id
@@ -1341,6 +1422,8 @@ def submit_embeddings_batch(
         self._embedding_part_limits[job.job_id] = part_limits
         self._embedding_jobs[job.job_id] = job
         if callable(reserve) and callable(start):
+            if resolved_agent_id is not None:
+                record_initial_selection([resolved_agent_id], "embedding_submission")
             start(job)
         return job