feat: select open-world tuning evidence

AI_AGENT_INTEGRATION.md

@@ -241,6 +241,10 @@ Publish request fields:
 | `force_extraction` | No | `False` for normal background publishing. `True` for manual learn-now, tests, or final flushes where you intentionally want extraction to run immediately. | `false` |
 | `wait_for_response` | SDK/query option | `False` on interactive paths. `True` only when the caller is prepared to wait for extraction results. | `false` |
 
+The source contract applies to new publish and filter inputs. Read responses may
+return historical source values created before this contract; Reflexio preserves
+those values exactly because source participates in session-outcome identity.
+
 Each interaction row should resemble Reflexio's `InteractionData` shape:
 
 ```json

reflexio/models/api_schema/domain/entities.py

@@ -26,6 +26,7 @@
 from ..validators import (
     EmbeddingVector,
     NonEmptyStr,
+    PersistedSessionOutcomeSource,
     SessionOutcomeSource,
     TimeRangeValidatorMixin,
     _validate_image_url,
@@ -279,8 +280,8 @@ class Request(BaseModel):
         user_id (str): Owner of the request.
         created_at (int): Unix epoch seconds at request creation. Defaults
             to the current UTC time.
-        source (str): Non-sensitive producer/workflow label. Non-empty values
-            use the session outcome source contract.
+        source (str): Producer/workflow label. Persisted reads preserve legacy
+            values verbatim; new publish inputs use the strict source contract.
         agent_version (str): The agent version that handled this request.
         session_id (str): Non-empty session this request belongs to.
         evaluation_only (bool): Whether this request is stored for
@@ -295,7 +296,7 @@ class Request(BaseModel):
     request_id: str
     user_id: str
     created_at: int = Field(default_factory=lambda: int(datetime.now(UTC).timestamp()))
-    source: SessionOutcomeSource = ""
+    source: PersistedSessionOutcomeSource = ""
     agent_version: str = ""
     session_id: NonEmptyStr
     evaluation_only: bool = False
@@ -831,13 +832,15 @@ class DeleteSessionResponse(BaseModel):
 
 
 class SessionOutcomeRecord(BaseModel):
+    """Persisted outcome row, including its immutable historical source."""
+
     outcome_id: NonEmptyStr | None = None
     outcome_revision: int | None = Field(default=None, ge=1)
     user_id: str
     session_id: NonEmptyStr
     outcome: SessionOutcomeKind
     occurred_at: int = Field(ge=0)
-    source: SessionOutcomeSource
+    source: PersistedSessionOutcomeSource
     label: str | None = Field(default=None, max_length=128)
     value: float | None = Field(default=None, allow_inf_nan=False)
     metadata: dict[str, Any] | None = None
@@ -914,7 +917,7 @@ class SetSessionOutcomeResponse(BaseModel):
     reason: SessionOutcomeFailureReason | None = None
     message: str = ""
     user_id: str | None = None
-    source: SessionOutcomeSource | None = None
+    source: PersistedSessionOutcomeSource | None = None
     outcome_id: NonEmptyStr | None = None
     outcome_revision: int | None = Field(default=None, ge=1)
     outcome_contract_digest: Sha256Digest | None = None

reflexio/models/api_schema/validators.py

@@ -21,7 +21,7 @@
 from typing import Annotated, Any, Literal
 from urllib.parse import urlparse
 
-from pydantic import AfterValidator, HttpUrl, StringConstraints
+from pydantic import AfterValidator, HttpUrl, StringConstraints, TypeAdapter
 
 # Embedding vector dimensions — must match config_schema.EMBEDDING_DIMENSIONS.
 # Duplicated here to avoid circular imports (config_schema imports from this module).
@@ -115,6 +115,16 @@ def _check_embedding_dimensions(v: list[float]) -> list[float]:
 )
 """Outcome producer/workflow label; empty preserves the existing absent-source value."""
 
+PersistedSessionOutcomeSource = str
+"""Historical outcome source returned exactly as stored, without new-input validation."""
+
+_SESSION_OUTCOME_SOURCE_ADAPTER = TypeAdapter(SessionOutcomeSource)
+
+
+def validate_session_outcome_source(value: str) -> SessionOutcomeSource:
+    """Validate an outcome source before writing a new request."""
+    return _SESSION_OUTCOME_SOURCE_ADAPTER.validate_python(value)
+
 
 # =============================================================================
 # Security Validators — SSRF Prevention

reflexio/server/middleware.py

@@ -23,9 +23,11 @@
 # Bot protection configuration
 REQUEST_TIMEOUT_SECONDS = 60
 SYNC_REQUEST_TIMEOUT_SECONDS = (
-    600  # Longer timeout for synchronous processing (wait_for_response=true)
+    600  # Longer timeout for synchronous long-running processing.
+)
+SYNC_REQUEST_PATHS = frozenset(
+    {"/api/review_user_playbooks", "/api/run_playbook_aggregation"}
 )
-SYNC_REQUEST_PATHS = frozenset({"/api/review_user_playbooks"})
 SUSPICIOUS_USER_AGENTS = ["bot", "crawler", "spider", "scraper", "curl", "wget"]
 ALLOWED_EMPTY_UA_PATHS = ["/health", "/"]  # Paths that allow empty user agents
 DEFAULT_MAX_BODY_BYTES = 10 * 1024 * 1024

reflexio/server/services/lineage/gc_scheduler.py

@@ -118,7 +118,8 @@ def register_global_sweep(fn: Callable[[int], int]) -> None:
         fn (Callable[[int], int]): Called with the current unix epoch; returns
             the number of rows it deleted.
     """
-    _global_sweep_hooks.append(fn)
+    if fn not in _global_sweep_hooks:
+        _global_sweep_hooks.append(fn)
 
 
 def clear_global_sweeps() -> None:
@@ -142,7 +143,8 @@ def register_always_global_sweep(fn: Callable[[int], int]) -> None:
         fn (Callable[[int], int]): Called with the current unix epoch; returns
             the number of rows it processed.
     """
-    _always_global_sweep_hooks.append(fn)
+    if fn not in _always_global_sweep_hooks:
+        _always_global_sweep_hooks.append(fn)
 
 
 def clear_always_global_sweeps() -> None:
@@ -172,7 +174,8 @@ def register_per_org_sweep(fn: Callable[[str, int], int]) -> None:
         fn (Callable[[str, int], int]): Called with ``(org_id, now)`` where
             ``now`` is the current unix epoch; returns the number of rows deleted.
     """
-    _per_org_sweep_hooks.append(fn)
+    if fn not in _per_org_sweep_hooks:
+        _per_org_sweep_hooks.append(fn)
 
 
 def clear_per_org_sweeps() -> None:

reflexio/server/services/search_metering_worker.py

@@ -65,16 +65,27 @@ def start(self) -> bool:
                 return False
             self._stop_event.clear()
             self._abort_event.clear()
-            self._threads = [
+            threads = [
                 threading.Thread(
                     target=self._worker_loop,
                     name=f"search-metering-worker-{index}",
                     daemon=True,
                 )
                 for index in range(self.worker_count)
             ]
-            for thread in self._threads:
-                thread.start()
+            started_threads: list[threading.Thread] = []
+            try:
+                for thread in threads:
+                    thread.start()
+                    started_threads.append(thread)
+            except BaseException:
+                self._stop_event.set()
+                for thread in started_threads:
+                    thread.join()
+                self._stop_event.clear()
+                self._threads = []
+                raise
+            self._threads = threads
             self._started = True
         logger.info(
             "event=search_metering_worker_started workers=%d queue_capacity=%d",

reflexio/server/services/storage/retention.py

@@ -22,6 +22,7 @@ class RetentionTarget:
     id_columns: tuple[str, ...]
     priority_statuses: tuple[str, ...] = ()
     minimum_age_seconds: int = 0
+    fixed_row_limit: int | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -50,12 +51,6 @@ class OptimizationRetentionClass:
     ),
     RetentionTarget("interactions", "interactions", "created_at", ("interaction_id",)),
     RetentionTarget("requests", "requests", "created_at", ("request_id",)),
-    RetentionTarget(
-        "session_outcomes",
-        "session_outcomes",
-        "created_at",
-        ("user_id", "session_id"),
-    ),
     RetentionTarget(
         "user_playbooks",
         "user_playbooks",
@@ -144,6 +139,7 @@ class OptimizationRetentionClass:
         "ingested_at",
         ("exposure_event_id",),
         minimum_age_seconds=OPEN_WORLD_EVIDENCE_RETENTION_WINDOW_SECONDS,
+        fixed_row_limit=DEFAULT_ROW_RETENTION_LIMIT,
     ),
     RetentionTarget("skills", "skills", "created_at", ("skill_id",)),
 )
@@ -201,12 +197,16 @@ class CascadeRef:
 def get_row_retention_limits() -> dict[str, int]:
     """Return per-target row limits from env with code defaults.
 
-    ``REFLEXIO_ROW_LIMIT_<TARGET>`` takes precedence for every target.
+    ``REFLEXIO_ROW_LIMIT_<TARGET>`` takes precedence for targets without a
+    ``fixed_row_limit``. Fixed targets explicitly reject that override path.
     ``INTERACTION_CLEANUP_THRESHOLD`` remains the legacy override for
     interactions when the new variable is not present.
     """
     limits: dict[str, int] = {}
     for target in RETENTION_TARGETS:
+        if target.fixed_row_limit is not None:
+            limits[target.name] = target.fixed_row_limit
+            continue
         env_name = f"REFLEXIO_ROW_LIMIT_{target.name.upper()}"
         default = DEFAULT_ROW_RETENTION_LIMIT
         if target.name == "interactions":

reflexio/server/services/storage/retention_mixin.py

@@ -233,6 +233,8 @@ def _retention_select_oldest_keys(
             statuses (tuple[str, ...] | None): When not None, restrict the
                 select to rows whose ``status`` is one of these values. An
                 empty tuple matches nothing (never every row).
+            older_than_epoch (int | None): When set, restrict the select to
+                rows whose target ordering column is strictly older.
         """
         raise NotImplementedError