fix: make durable playbook recovery deterministic

reflexio/server/api.py

@@ -414,10 +414,11 @@ def create_app(  # noqa: C901
             logic. It is only consulted when ``REFLEXIO_DURABLE_LEARNING_QUEUE`` is
             on — when the flag is off ``maybe_start_durable_learning`` returns None
             without ever calling the provider.
-        resume_org_ids_provider: Optional zero-arg callable used only when the
-            bootstrap org is not available yet. This lets a multi-tenant deployment
-            defer the resume scheduler on an empty fleet and adopt its first real
-            org without restarting. The OSS default remains unchanged.
+        resume_org_ids_provider: Optional zero-arg callable consulted before the
+            bootstrap context on every scheduler tick. Its actionable org list is
+            authoritative, allowing a multi-tenant deployment to recover when the
+            prior bootstrap org disappears and to discover work across data refs.
+            The OSS default remains unchanged when no provider is supplied.
 
     Returns:
         Configured FastAPI application.

reflexio/server/llm/_litellm_text_generation.py

@@ -151,10 +151,26 @@ class _StructuredAttempt:
 def _is_expected_transient_llm_error(exc: BaseException) -> bool:
     """True for expected transient upstream failures (timeout / connection /
     rate-limit / overload), including our own ``LLMHardTimeoutError`` (a
-    ``TimeoutError`` subclass raised when a provider hang is killed)."""
-    if isinstance(exc, TimeoutError):  # incl. LLMHardTimeoutError
-        return True
-    return type(exc).__name__ in _TRANSIENT_LLM_ERROR_NAMES
+    ``TimeoutError`` subclass raised when a provider hang is killed).
+
+    Subprocess-isolated calls cross a pickle boundary, so their concrete
+    provider exception cannot safely be re-raised in the parent. The wrapper
+    retains ``upstream_error_type`` explicitly; checking it here prevents an
+    expected provider outage from being promoted to an application ERROR just
+    because isolation made the outer type ``LiteLLMClientError``.
+    """
+    current: BaseException | None = exc
+    seen: set[int] = set()
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        if isinstance(current, TimeoutError):  # incl. LLMHardTimeoutError
+            return True
+        if type(current).__name__ in _TRANSIENT_LLM_ERROR_NAMES:
+            return True
+        if getattr(current, "upstream_error_type", None) in _TRANSIENT_LLM_ERROR_NAMES:
+            return True
+        current = current.__cause__ or current.__context__
+    return False
 
 
 def _rung_reason(error: Exception | None) -> str:
@@ -860,7 +876,8 @@ def _completion_with_hard_timeout(
             raise LiteLLMClientError(
                 "litellm.completion failed in isolated worker: "
                 f"{payload.type_name}: {payload.message} "
-                f"({', '.join(context_parts)})"
+                f"({', '.join(context_parts)})",
+                upstream_error_type=payload.type_name,
             )
         finally:
             result_queue.close()

reflexio/server/llm/_litellm_types.py

@@ -131,9 +131,11 @@ def __init__(
         message: str,
         *,
         first_parsed_provenance: ModelProvenance | None = None,
+        upstream_error_type: str | None = None,
     ) -> None:
         super().__init__(message)
         self.first_parsed_provenance = first_parsed_provenance
+        self.upstream_error_type = upstream_error_type
 
 
 class StructuredOutputRepairError(LiteLLMClientError):

reflexio/server/services/extraction/README.md

@@ -15,7 +15,7 @@ information, and resumes outside the request path.
 | `pending_tool_call_dispatch.py` | Implements the `ask_human` and pending-info tool dispatch flow. |
 | `prior_answer_search.py` | Finds and formats previous human answers for async extraction context. |
 | `agent_run_records.py` | Builds durable extraction-agent run records and source interaction identity. |
-| `resume_scheduler.py` | Schedules due paused extraction runs in a background singleton. |
+| `resume_scheduler.py` | Discovers and schedules due paused/finalization work in a background singleton. |
 | `resume_worker.py` | Resumes paused runs, rebuilds request context, and records retry state. |
 | `outcome.py` | Provides the generic extraction outcome wrapper used by callers. |
 
@@ -29,3 +29,17 @@ information, and resumes outside the request path.
   flat package stops being easier to scan.
 - Do not recreate removed legacy modules such as `tools.py`, `plan.py`, or
   `invariants.py`; use the current focused files above.
+- New durable extraction runs require a non-empty `user_id`. Nullable stored
+  bindings remain readable for backward compatibility; playbook resume derives
+  an in-memory owner only from complete, unanimous persisted source evidence.
+
+## Resume discovery
+
+When an `org_id_provider` is installed, the scheduler calls it on every tick and
+treats its actionable org list as authoritative for cross-ref discovery. Without
+a provider, local storage discovery remains the fallback. Each provider result
+selects a current org for reading scheduler configuration, so an org retained
+from an earlier tick cannot block later discovery if it becomes stale. If the
+provider itself raises, the scheduler still attempts the last bootstrap org for
+that tick. Before draining each discovered org context, it expires pending tool
+calls on that context's storage ref; separate refs have separate queues.

reflexio/server/services/extraction/agent_run_records.py

@@ -46,7 +46,7 @@ def build_extractor_agent_run_record(
     *,
     org_id: str,
     extractor_kind: str,
-    user_id: str | None,
+    user_id: str,
     agent_version: str | None,
     source: str | None,
     request_interaction_data_models: list[RequestInteractionDataModel],
@@ -57,6 +57,19 @@ def build_extractor_agent_run_record(
     generation_request_id: str | None = None,
     request_id: str | None = None,
 ) -> AgentRunRecord:
+    user_id = user_id.strip()
+    if not user_id:
+        raise ValueError("Durable extraction runs require a non-empty user_id")
+    if any(
+        data_model.request.user_id != user_id
+        or any(
+            interaction.user_id != user_id for interaction in data_model.interactions
+        )
+        for data_model in request_interaction_data_models
+    ):
+        raise ValueError(
+            "Durable extraction run source evidence must belong to its user_id"
+        )
     if generation_request_id is not None:
         if request_id is not None and request_id != generation_request_id:
             raise TypeError(

reflexio/server/services/extraction/resumable_agent.py

@@ -191,7 +191,7 @@ def run_resumable_extraction_agent(
     request_context: RequestContext,
     client: LiteLLMClient,
     extractor_kind: str,
-    user_id: str | None,
+    user_id: str,
     request_id: str,
     agent_version: str | None,
     source: str | None,

reflexio/server/services/extraction/resume_scheduler.py

@@ -51,27 +51,46 @@ def _on_started(self) -> None:
     def _on_stopped(self) -> None:
         logger.info("event=extraction_resume_scheduler_stopped")
 
-    def _discover_org_ids(self, bootstrap_ctx: RequestContext) -> list[str]:
-        """Return every org with actionable work, always including the bootstrap org."""
+    def _discover_local_org_ids(self, bootstrap_ctx: RequestContext) -> list[str]:
+        """Return local actionable orgs plus the bootstrap org."""
         org_ids: list[str] = []
         storage = getattr(bootstrap_ctx, "storage", None)
         if storage is not None:
             try:
                 org_ids = storage.list_resumable_work_org_ids(now=datetime.now(UTC))
             except NotImplementedError:
                 org_ids = []
-        # Always sweep the bootstrap org so the maintenance loop runs even when
-        # the cross-org discovery query is empty or unsupported.
         if bootstrap_ctx.org_id not in org_ids:
             org_ids = [bootstrap_ctx.org_id, *org_ids]
-        return org_ids
+        return list(dict.fromkeys(org_ids))
 
-    def _expire_pending_tool_calls(self, bootstrap_ctx: RequestContext) -> None:
-        storage = getattr(bootstrap_ctx, "storage", None)
+    def _discover_provider_org_ids(self) -> list[str] | None:
+        """Return the provider's authoritative list, or ``None`` on failure."""
+        if self.org_id_provider is None:
+            return None
+        try:
+            return list(
+                dict.fromkeys(
+                    org_id
+                    for org_id in self.org_id_provider()
+                    if org_id != DEFAULT_ORG_ID
+                )
+            )
+        except Exception as exc:
+            with error_tags(
+                subsystem="extraction",
+                op="scheduler_org_discovery",
+                error_type=type(exc).__name__,
+            ):
+                logger.exception("event=extraction_resume_scheduler_provider_failed")
+            return None
+
+    def _expire_pending_tool_calls(self, ctx: RequestContext) -> None:
+        storage = getattr(ctx, "storage", None)
         if storage is None:
             return
-        # ``expire_pending_tool_calls`` is not org-scoped, so a single call
-        # sweeps every tenant's overdue pending rows for this tick.
+        # One storage ref can contain several tenants, but another ref is an
+        # independent queue. Sweep every discovered ref before draining it.
         try:
             expired = storage.expire_pending_tool_calls(now=datetime.now(UTC))
         except NotImplementedError:
@@ -84,6 +103,7 @@ def _drain_org(self, org_id: str) -> None:
             ctx = self.request_context_factory(org_id)
             if not pending_tool_calls_enabled(ctx):
                 return
+            self._expire_pending_tool_calls(ctx)
             resumed = ExtractionResumeWorker(request_context=ctx).drain(
                 max_runs=self.max_runs_per_tick
             )
@@ -108,23 +128,26 @@ def _drain_org(self, org_id: str) -> None:
     def _run_once(self) -> float:
         poll_interval = _DEFAULT_POLL_INTERVAL_SECONDS
         try:
-            if (
-                self.bootstrap_org_id == DEFAULT_ORG_ID
-                and self.org_id_provider is not None
-            ):
-                org_ids = [
-                    org_id
-                    for org_id in self.org_id_provider()
-                    if org_id != DEFAULT_ORG_ID
-                ]
-                if not org_ids:
+            provider_org_ids = self._discover_provider_org_ids()
+            if self.org_id_provider is not None and provider_org_ids is not None:
+                if not provider_org_ids:
                     return poll_interval
-                self.bootstrap_org_id = org_ids[0]
+                # Resolve config through an org that the provider proved is
+                # actionable on this tick. The previous bootstrap may have
+                # been deleted or moved and must not gate future discovery.
+                self.bootstrap_org_id = provider_org_ids[0]
             bootstrap_ctx = self.request_context_factory(self.bootstrap_org_id)
             config = bootstrap_ctx.configurator.get_config()
             poll_interval = config.pending_tool_call_config.resume_poll_interval_seconds
-            self._expire_pending_tool_calls(bootstrap_ctx)
-            for org_id in self._discover_org_ids(bootstrap_ctx):
+            if provider_org_ids is not None:
+                org_ids = provider_org_ids
+            elif self.org_id_provider is not None:
+                # A raised provider cannot authoritatively replace the list;
+                # preserve the last known bootstrap as a one-org fallback.
+                org_ids = [bootstrap_ctx.org_id]
+            else:
+                org_ids = self._discover_local_org_ids(bootstrap_ctx)
+            for org_id in org_ids:
                 if self._stop_event.is_set():
                     break
                 self._drain_org(org_id)

reflexio/server/services/extraction/resume_worker.py

@@ -6,6 +6,7 @@
 import uuid
 from collections import defaultdict
 from collections.abc import Callable, Sequence
+from dataclasses import replace
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
@@ -46,6 +47,9 @@
     has_expert_content,
     uses_evidence_grounded_extraction,
 )
+from reflexio.server.services.playbook.review_window import (
+    infer_playbook_review_user_id,
+)
 from reflexio.server.services.playbook.service import (
     PlaybookGenerationService,
     PlaybookGenerationServiceConfig,
@@ -256,6 +260,19 @@ def drain(self, *, max_runs: int = 10) -> int:
             resumed += 1
         return resumed
 
+    def _with_resolved_playbook_user_id(self, run: AgentRunRecord) -> AgentRunRecord:
+        """Return a playbook run with a proven owner for legacy nullable rows."""
+        if run.binding.extractor_kind != "playbook":
+            return run
+        if run.binding.user_id and run.binding.user_id.strip():
+            return run
+        user_id = infer_playbook_review_user_id(
+            storage=self.storage,
+            source_interaction_ids=run.binding.source_interaction_ids,
+            subject=f"Playbook extraction run {run.id}",
+        )
+        return replace(run, binding=replace(run.binding, user_id=user_id))
+
     def run_once(self) -> AgentRunRecord | None:
         config = self.request_context.configurator.get_config()
         pending_config = config.pending_tool_call_config
@@ -283,6 +300,7 @@ def run_once(self) -> AgentRunRecord | None:
             )
 
         try:
+            run = self._with_resolved_playbook_user_id(run)
             resolved_calls = self._load_resolved_tool_calls(run)
             if not resolved_calls:
                 raise ResumeWorkerError(
@@ -360,6 +378,7 @@ def _retry_finalization(self, run: AgentRunRecord) -> AgentRunRecord | None:
         config = self.request_context.configurator.get_config()
         pending_config = config.pending_tool_call_config
         try:
+            run = self._with_resolved_playbook_user_id(run)
             items, pending_tool_call_ids, model_provenance = (
                 self._items_from_committed_output(run)
             )
@@ -557,12 +576,15 @@ def _resume_playbook(
     ) -> tuple[list[Any], list[str], ModelProvenance | None]:
         if not isinstance(extractor_config, PlaybookConfig):
             raise ResumeWorkerError("Expected playbook extractor config")
+        user_id = run.binding.user_id
+        if not user_id:
+            raise ResumeWorkerError("Playbook resume requires user_id")
 
         agent_context = self.request_context.configurator.get_agent_context()
         service_config = PlaybookGenerationServiceConfig(
             request_id=run.binding.request_id,
             agent_version=run.binding.agent_version or "",
-            user_id=run.binding.user_id,
+            user_id=user_id,
             source=run.binding.source,
             auto_run=False,
             force_extraction=True,
@@ -802,6 +824,9 @@ def _playbook_items_from_output(
     ) -> tuple[list[Any], list[str]]:
         if not isinstance(extractor_config, PlaybookConfig):
             raise ResumeWorkerError("Expected playbook extractor config")
+        user_id = run.binding.user_id
+        if not user_id:
+            raise ResumeWorkerError("Playbook finalization retry requires user_id")
         expert_mode = has_expert_content(
             extract_interactions_from_request_interaction_data_models(
                 request_interaction_data_models
@@ -840,7 +865,7 @@ def _playbook_items_from_output(
         service_config = PlaybookGenerationServiceConfig(
             request_id=run.binding.request_id,
             agent_version=run.binding.agent_version or "",
-            user_id=run.binding.user_id,
+            user_id=user_id,
             source=run.binding.source,
             auto_run=False,
             force_extraction=True,
@@ -892,20 +917,25 @@ def _finalize_items(
             )
             return
         if run.binding.extractor_kind == "playbook":
+            user_id = run.binding.user_id
+            if not user_id:
+                raise ResumeWorkerError("Playbook finalization requires user_id")
             service = PlaybookGenerationService(
                 llm_client=self.client,
                 request_context=self.request_context,
             )
             service.service_config = PlaybookGenerationServiceConfig(
                 request_id=run.binding.request_id,
                 agent_version=run.binding.agent_version or "",
-                user_id=run.binding.user_id,
+                user_id=user_id,
                 source=run.binding.source,
                 auto_run=False,
                 force_extraction=True,
             )
             persisted_items = service._finalize_extracted_items(
-                items, model_provenance=model_provenance
+                items,
+                model_provenance=model_provenance,
+                extraction_run=run,
             )
             self._record_finalized_learnings(
                 run, persisted_items or [], entity_type="user_playbook"

reflexio/server/services/playbook/README.md

@@ -20,6 +20,7 @@ Description: Evidence-grounded playbook extraction, candidate review, aggregatio
 | `playbook_service_constants.py` | Prompt IDs for all playbook operations |
 | `playbook_service_utils.py` | Request dataclasses, Pydantic output schemas, message construction utilities |
 | `playbook_evidence.py` | Strict evidence validation, call-local reference checks, and persisted provenance helpers |
+| `review_window.py` | Shared fail-closed reconstruction of persisted review chronology for automatic and manual review |
 | `review_service.py` | Time-window selection, persisted evidence reconstruction, reporting, and newest-first per-playbook apply |
 | `aggregation_trigger.py` | Converts post-generation activity into an idempotent durable scheduling signal |
 | `aggregation_scheduler.py` | Polling, fleet claim/lease handling, retries, and structured aggregation progress telemetry |
@@ -66,7 +67,10 @@ Every candidate must be accounted for exactly once as `accept`, `revise`, or
 `reject`. Revisions may narrow unsupported wording but cannot add evidence or
 create a lesson that extraction missed. Reviewer output receives one bounded
 repair attempt and otherwise fails closed. Expert and legacy extraction paths
-do not use this reviewer.
+do not use this reviewer. New user-playbook generation requires a non-empty
+`user_id`. A legacy durable run whose stored binding is null may resume only
+when every persisted source interaction and request proves the same owner;
+missing or mixed-owner evidence fails closed before review or persistence.
 
 ### Persisted Review (`review_service.py`)
 
@@ -77,9 +81,11 @@ generation window or cited evidence can no longer be reconstructed yields a
 
 Manual review reconstructs context from the full interaction window persisted on
 the row's finalized playbook-extraction run, plus any extra cited interactions
-retained through consolidation. It never substitutes the current extractor
-window or silently falls back to the smaller cited-evidence subset. Automatic
-post-generation review continues to use the generation call's configured window.
+retained through consolidation. Automatic review uses the exact
+`source_interaction_ids` on the extraction agent run, including when finalization
+is retried later. Neither path re-runs the sliding last-K query or silently falls
+back to a smaller evidence subset, so interactions published after extraction
+cannot change the review chronology.
 Only the playbook row's cited interaction IDs become candidate evidence units;
 the rest of the generation window is ancillary chronology. Evidence spans are
 rebuilt from those exact stored interactions instead of requiring every cited