feat: add replay-safe playbook optimization foundations

docs/playbook_optimizer_assistant_backends.md

@@ -55,7 +55,9 @@ The "assistant backend" is the thing that turns *(messages, playbooks)* into the
 │       )                                                               │              │
 │                                                                       │              │
 │  7.  if best passes commit thresholds:                                │              │
-│         _commit_if_allowed → archive incumbent, save successor        │              │
+│         agent target → _commit_if_allowed                             │              │
+│         user target  → stage proof + projection, then atomically      │              │
+│                        replace incumbent with one successor            │              │
 └───────────────────────────────────────────────────────────────────────┼──────────────┘
                                                                         │
                                                                         │ for each
@@ -262,7 +264,15 @@ Useful as a smoke test — confirms the optimizer end-to-end without needing an
 
 ## 5. How a backend is selected at runtime
 
-Selection happens once per `optimize()` call inside `PlaybookOptimizer._create_assistant`:
+Before selecting a backend, `optimize()` applies the feature, target-kind, and
+adoption kill switches. For user-playbook targets it then attempts to recover
+any expired staged publication. Recovery uses the already durable candidate,
+evaluation digests, decision proof, and search projection; it does not rerun
+GEPA or require an assistant backend. A disabled kill switch prevents both a
+new run and publication recovery.
+
+For a new optimization run, backend selection happens once per `optimize()`
+call inside `PlaybookOptimizer._create_assistant`:
 
 ```python
 def _create_assistant(self, config) -> AssistantCallable | None:
@@ -280,7 +290,11 @@ def _create_assistant(self, config) -> AssistantCallable | None:
 | Both set | **Rejected at config load time** by a Pydantic validator (`ValueError: Configure only one playbook optimizer assistant backend...`) |
 | Neither set | Optimizer logs `Skipping playbook optimization: no assistant backend configured` and returns without creating a job |
 
-The backend check happens *before* loading the incumbent or resolving scenario windows, so an unconfigured optimizer short-circuits cheaply. After source-window resolution, the optimizer also skips before job creation when the validation holdout is smaller than `min_commit_windows` or the target kind has no enabled auto-update path.
+The backend check happens *before* loading the incumbent or resolving scenario
+windows, so an unconfigured optimizer short-circuits cheaply when there is no
+recoverable publication. After source-window resolution, the optimizer also
+skips before job creation when the validation holdout is smaller than
+`min_commit_windows`.
 
 ---
 
@@ -362,13 +376,13 @@ playbook_optimizer_config:
 
 ## 7. End-to-end workflow
 
-What happens, in order, when a new agent playbook is generated and the optimizer is enabled:
+What happens, in order, when a playbook is scheduled and the optimizer is enabled:
 
 | Step | Component | What it does |
 |---|---|---|
 | 1 | `PlaybookAggregator._enqueue_playbook_optimization` | After saving new agent playbooks, enqueues each PENDING playbook with the scheduler. |
 | 2 | `PlaybookOptimizationScheduler` | Debounces by `(org_id, kind, target_id)`, fires after a small jitter, spawns a daemon thread. |
-| 3 | `PlaybookOptimizer.optimize(target)` | Loads config, calls `_create_assistant` → backend instance (or returns early). |
+| 3 | `PlaybookOptimizer.optimize(target)` | Applies feature, target, and adoption gates. For user targets, resumes an expired staged publication before starting new search work. |
 | 4 | `ScenarioResolver` | Builds `ScenarioWindow`s from snapshotted agent source windows or a user playbook's `source_interaction_ids`. |
 | 5 | `gepa.api.optimize(...)` | Runs the candidate-search loop. |
 | 6 | `ReflexioPlaybookGEPAAdapter.evaluate` | For each `(candidate, window)`: |
@@ -378,9 +392,12 @@ What happens, in order, when a new agent playbook is generated and the optimizer
 |   |   | • Persists `PlaybookOptimizationEvaluation` row |
 | 7 | Failure handling | Any `AssistantFailedError` → `verdict="aborted"`, `score=0.0`, GEPA continues. |
 | 8 | `PlaybookOptimizer._passes_commit_thresholds` | Checks score / likert / per-window verdict counts. |
-| 9 | `PlaybookOptimizer._commit_if_allowed` | If gates pass → archive incumbent, save successor playbook. |
+| 9a | Agent-playbook publication | `_commit_if_allowed` archives the pending incumbent and saves its successor. |
+| 9b | User-playbook publication | Builds a fixed decision proof and search projection, stages them durably, then atomically replaces the current user playbook with exactly one successor. The compact proof binds each full durable evaluation through a canonical digest rather than duplicating rollout documents. |
+| 10 | User-playbook recovery | A later enabled run may reclaim an expired publication lease and finish from the staged bytes. Incumbent changes produce `incumbent_changed`; retries do not rerun GEPA or regenerate the projection. |
 
-The same flow runs for user playbooks, gated by `optimize_user_playbooks`.
+User-playbook recovery and new runs both require `enabled`,
+`optimize_user_playbooks`, and `auto_update_user_playbooks`.
 
 ---
 

reflexio/cli/commands/status_cmd.py

@@ -50,15 +50,18 @@ def check(
         render_error(err, json_mode=json_mode)
         raise SystemExit(1) from exc
     except requests.HTTPError as exc:
+        response = exc.response
         if json_mode:
             envelope = {
                 "ok": False,
                 "error": {"type": "unhealthy", "message": str(exc), "url": url},
             }
             print(json.dumps(envelope, indent=2), file=sys.stderr)
+        elif response is None:
+            print(f"Error: Server at {url} returned an HTTP error", file=sys.stderr)
         else:
             print(
-                f"Error: Server at {url} returned {exc.response.status_code}",
+                f"Error: Server at {url} returned {response.status_code}",
                 file=sys.stderr,
             )
         raise SystemExit(1) from exc

reflexio/cli/errors.py

@@ -63,8 +63,9 @@ def _classify_http_error(exc: requests.HTTPError) -> CliError:
         )
     if status == 422:
         detail = ""
-        with contextlib.suppress(ValueError, AttributeError):
-            detail = exc.response.json().get("detail", "")
+        if exc.response is not None:
+            with contextlib.suppress(ValueError, AttributeError):
+                detail = exc.response.json().get("detail", "")
         return CliError(
             error_type="validation",
             message=f"Invalid request: {detail or exc}",

reflexio/cli/utils.py

@@ -71,21 +71,23 @@ def _pids_from_lsof(port: int) -> list[int]:
             check=False,
         )
     except FileNotFoundError:
-        return []
-    if result.returncode != 0 or not result.stdout.strip():
-        return []
+        result = None
 
     pids: set[int] = set()
-    current_pid: int | None = None
-    suffix = f":{port}"
-    for line in result.stdout.splitlines():
-        if line.startswith("p") and line[1:].isdigit():
-            current_pid = int(line[1:])
-        elif line.startswith("n") and current_pid is not None:
-            name = line[1:]
-            if "->" not in name and name.endswith(suffix):
-                pids.add(current_pid)
-    return sorted(pids)
+    if result is not None and result.returncode == 0:
+        current_pid: int | None = None
+        suffix = f":{port}"
+        for line in result.stdout.splitlines():
+            if line.startswith("p") and line[1:].isdigit():
+                current_pid = int(line[1:])
+            elif line.startswith("n") and current_pid is not None:
+                name = line[1:]
+                if "->" not in name and name.endswith(suffix):
+                    pids.add(current_pid)
+    if pids:
+        return sorted(pids)
+
+    return []
 
 
 def _pids_from_ss(port: int) -> list[int]:

reflexio/client/__init__.py

@@ -1,3 +1,3 @@
-from .client import ReflexioClient
+from .client import ConfigResponse, OfflineTunerConfigResponse, ReflexioClient
 
-__all__ = ["ReflexioClient"]
+__all__ = ["ConfigResponse", "OfflineTunerConfigResponse", "ReflexioClient"]

reflexio/client/client.py

@@ -12,7 +12,7 @@
 
 import aiohttp
 import requests
-from pydantic import ConfigDict
+from pydantic import BaseModel, ConfigDict
 
 from reflexio.defaults import DEFAULT_AGENT_VERSION
 from reflexio.models.api_schema.eval_overview_schema import (
@@ -135,7 +135,12 @@
 from .cache import InMemoryCache
 
 
-class _ClientConfigPayload(Config):
+class OfflineTunerConfigResponse(BaseModel):
+    enabled: bool
+
+
+class ConfigResponse(Config):
+    offline_tuner_config: OfflineTunerConfigResponse | None = None
     model_config = ConfigDict(extra="allow")
 
 
@@ -1467,10 +1472,16 @@ def set_config(self, config: Config | dict) -> dict:
         Returns:
             dict: Response containing success status and message
         """
-        config = self._convert_to_model(  # type: ignore[reportAssignmentType]
-            config,
-            _ClientConfigPayload,
-        )
+        if isinstance(config, Config):
+            config = Config.model_validate(
+                config.model_dump(mode="python"),
+                extra="ignore",
+            )
+        else:
+            config = self._convert_to_model(  # type: ignore[reportAssignmentType]
+                config,
+                Config,
+            )
         return self._make_request(
             "POST",
             "/api/set_config",
@@ -1506,17 +1517,17 @@ def update_config(self, partial: dict) -> dict:
             )
         return self._make_request("POST", "/api/update_config", json=partial)
 
-    def get_config(self) -> Config:
+    def get_config(self) -> ConfigResponse:
         """Get configuration for the organization.
 
         Returns:
-            Config: The current configuration
+            ConfigResponse: The current configuration, including response overlays.
         """
         response = self._make_request(
             "GET",
             "/api/get_config",
         )
-        return _ClientConfigPayload(**response)
+        return ConfigResponse(**response)
 
     def invalidate_cache(self, org_id: str | None = None) -> dict:
         """Explicitly evict the server-side per-org Reflexio cache entry.

reflexio/integrations/openclaw/plugin/tests/test_publish.py

@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Sequence
+from typing import Any
 
 import pytest
 from openclaw_smart import publish, state
+from openclaw_smart.reflexio_adapter import Adapter
 
 
 @pytest.fixture(autouse=True)
@@ -15,11 +18,21 @@ def isolate_state_dir(monkeypatch, tmp_path):
     return sessions
 
 
-class _Adapter:
+class _Adapter(Adapter):
     def __init__(self) -> None:
+        super().__init__()
         self.calls = 0
 
-    def publish(self, **_kwargs) -> bool:  # noqa: ANN003
+    def publish(
+        self,
+        *,
+        session_id: str,
+        project_id: str,
+        interactions: Sequence[dict[str, Any]],
+        force_extraction: bool = False,
+        skip_aggregation: bool = False,
+    ) -> bool:
+        del session_id, project_id, interactions, force_extraction, skip_aggregation
         self.calls += 1
         return True
 

reflexio/lib/_base.py

@@ -14,10 +14,7 @@
 from reflexio.models.config_schema import SearchMode
 from reflexio.server.api_endpoints.request_context import RequestContext
 from reflexio.server.llm.litellm_client import LiteLLMClient, LiteLLMConfig
-from reflexio.server.llm.model_defaults import (
-    ModelRole,
-    resolve_model_name,
-)
+from reflexio.server.llm.model_defaults import ModelRole, resolve_model_name
 from reflexio.server.services.configurator.base_configurator import BaseConfigurator
 from reflexio.server.services.storage.storage_base import BaseStorage
 from reflexio.server.site_var.site_var_manager import SiteVarManager
@@ -31,6 +28,31 @@
 )
 
 
+def create_generation_litellm_client(
+    request_context: RequestContext,
+) -> LiteLLMClient:
+    """Build the tenant client used by normal generation and aggregation."""
+    model_setting = SiteVarManager().get_site_var("llm_model_setting")
+    site_var = model_setting if isinstance(model_setting, dict) else {}
+    config = request_context.configurator.get_config()
+    api_key_config = config.api_key_config if config else None
+    config_llm_config = config.llm_config if config else None
+    generation_model_name = resolve_model_name(
+        ModelRole.GENERATION,
+        site_var_value=site_var.get("default_generation_model_name"),
+        config_override=(
+            config_llm_config.generation_model_name if config_llm_config else None
+        ),
+        api_key_config=api_key_config,
+    )
+    return LiteLLMClient(
+        LiteLLMConfig(
+            model=generation_model_name,
+            api_key_config=api_key_config,
+        )
+    )
+
+
 def _require_storage[T: BaseModel](
     response_type: type[T], *, msg_field: str = "message"
 ) -> Callable[..., Callable[..., T]]:
@@ -88,29 +110,7 @@ def __init__(
             org_id=org_id, storage_base_dir=storage_base_dir, configurator=configurator
         )
 
-        # Create single LLM client for all services
-        model_setting = SiteVarManager().get_site_var("llm_model_setting")
-        site_var = model_setting if isinstance(model_setting, dict) else {}
-
-        # Get API key config and LLM config from configuration if available
-        config = self.request_context.configurator.get_config()
-        api_key_config = config.api_key_config if config else None
-        config_llm_config = config.llm_config if config else None
-
-        generation_model_name = resolve_model_name(
-            ModelRole.GENERATION,
-            site_var_value=site_var.get("default_generation_model_name"),
-            config_override=config_llm_config.generation_model_name
-            if config_llm_config
-            else None,
-            api_key_config=api_key_config,
-        )
-
-        llm_config = LiteLLMConfig(
-            model=generation_model_name,
-            api_key_config=api_key_config,
-        )
-        self.llm_client = LiteLLMClient(llm_config)
+        self.llm_client = create_generation_litellm_client(self.request_context)
 
     def _is_storage_configured(self) -> bool:
         """Check if storage is configured and available.

reflexio/lib/_config.py

@@ -3,6 +3,9 @@
 from reflexio.lib._base import ReflexioBase
 from reflexio.models.api_schema.retriever_schema import SetConfigResponse
 from reflexio.models.config_schema import Config, StorageConfigManagedSupabase
+from reflexio.server.services.configurator.config_storage import (
+    ConfigWriteConflictError,
+)
 
 
 class ConfigMixin(ReflexioBase):
@@ -18,8 +21,12 @@ def set_config(self, config: Config | dict) -> SetConfigResponse:
         try:
             configurator = self.request_context.configurator
             if isinstance(config, dict):
-                config = configurator.normalize_config_payload(config)
-                config = Config(**config)
+                normalized = configurator.normalize_config_payload(config)
+                config = (
+                    normalized
+                    if isinstance(normalized, Config)
+                    else Config(**normalized)
+                )
 
             # Validate storage connection before setting config.
             # If no storage_config provided, or the caller round-tripped the
@@ -65,6 +72,8 @@ def set_config(self, config: Config | dict) -> SetConfigResponse:
             configurator.set_config(config)
 
             return SetConfigResponse(success=True, msg="Configuration set successfully")
+        except ConfigWriteConflictError:
+            raise
         except Exception as e:
             return SetConfigResponse(
                 success=False, msg=f"Failed to set configuration: {str(e)}"

reflexio/lib/generation_client.py

@@ -0,0 +1,5 @@
+"""Public factory for the tenant generation client."""
+
+from reflexio.lib._base import create_generation_litellm_client
+
+__all__ = ["create_generation_litellm_client"]