fix: judge short sessions on completion or inactivity (#659)

Short sessions can be saved without reaching the first extraction window
or stride. Add optional `session_finished` publishing and
`finalize_session` operations, plus a default six-hour session
inactivity fallback. Both reuse the existing judge → extractor →
consolidation → conditional aggregation pipeline. Normal window/stride
and explicit manual-force behavior stay intact.

Boundaries survive late messages, respect project/user scope, ignore
covered or erased prefixes, and retain durable wakeups across restart.
Existing pinned work remains immutable. Timeout configuration is
captured prospectively; zero disables fallback for new publications.
Legacy admissions support explicit completion without inferred idle
backfill.

Validation: 443 backend/SDK tests passed after rebasing onto current
main, covering real HTTP/provider seams, positive/negative judges,
retries, rollback, late arrivals, erasure, index use and scheduler
restart. Ruff/Pyright passed. Follow-up review fix invalidates learning
caches on successful sync/async completion; 90 SDK/cache tests passed.
All four current-head installed-wheel CI jobs passed. GitHub Codex
reviewed the current head with no major issues and no P0/P1 findings.

Related PRs:
- PostgreSQL integration and public documentation:
ReflexioAI/reflexio-enterprise#1765
- Plugin: ReflexioAI/claude-smart#189


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **New Features**
* Partial extraction windows can now be evaluated after a
session-completion hint or a configurable inactivity period, while
retaining the usefulness check.
* Added session-finalization methods for standard and session-scoped
clients.
* Short messages can now reach the usefulness check instead of being
rejected by a length filter.
* **Documentation**
* Updated guidance on completion hints, idle timeouts, partial windows,
and manual forced extraction.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
Completion acceptance also survives post-commit scheduler failures. Five
HTTP/library short-session regression cases passed; Ruff and Pyright
passed. GitHub Codex review completed on the final head with no major
issues and no P0/P1 findings.

Author: guangyu-reflexio

README.md

@@ -471,7 +471,28 @@ Publishing commits interactions before returning success. Background workers pro
 configured sliding windows in admission order, regardless of request boundaries.
 Different users can extract concurrently within `REFLEXIO_PUBLISH_LEARNING_WORKERS`
 (default 4 per process); each user's profile/playbook cursors retry independently.
-A partial window waits for more input or an eligible `force_extraction` request.
+A partial window is evaluated after an optional completion hint or six hours of
+session inactivity by default. Both paths retain the usefulness judge; a negative
+judge result covers the input without generating learnings. Normal full windows
+and strides run as before. Later messages in a finalized session remain valid.
+
+```python
+client.publish_interaction(user_id="u", session_id="s", interactions=messages,
+                           session_finished=True)
+# Or signal completion after messages were already saved:
+client.finalize_session(user_id="u", session_id="s")
+```
+
+Completion returns acceptance, not extraction completion. The async and
+`client.for_session("s")` APIs support the same operation. Set Config's
+`extraction_idle_timeout_seconds` (default `21600`; `0` disables fallback for new
+publications). This is a configurable low-urgency policy, not proof that a session
+ended. Activity is scoped by organization/project/user/session; other sessions do
+not postpone its deadline. Each publication captures its timeout, so edits apply
+to subsequent publications. Existing pre-upgrade admissions have no inferred
+idle deadline; explicitly finalize those surviving sessions when needed.
+The library scheduler needs a live process and reachable `Reflexio` handle;
+reopening resumes durable deadlines without another publish.
 Provider fallback remains enabled. Existing extracted data is preserved, and
 interactions stored before stream admission are not automatically replayed.
 
@@ -489,9 +510,9 @@ or waiting for input, reporting `learning_status="deferred"` with
 `learning_reason="waiting_for_window"`. Poll `GET /api/learning_status` with the
 returned `request_id` to see coverage complete later. Note the practical
 consequence of the default `window_size` of 10: a brand-new user's first window
-needs ten eligible interactions before anything is extracted, so a single publish
-legitimately returns uncovered. Pass `force_extraction` when you need that publish
-extracted on its own.
+normally needs ten eligible interactions, so an incomplete first publish returns
+uncovered until more input, an optional completion hint, or its idle deadline.
+Reserve `force_extraction` for an explicit manual bypass of the judge.
 
 ### Nomic dependency compatibility
 

reflexio/__init__.py

@@ -57,6 +57,8 @@
         DeleteUserPlaybooksByIdsRequest,
         DeleteUserProfileRequest,
         DeleteUserProfileResponse,
+        FinalizeSessionRequest,
+        FinalizeSessionResponse,
         Interaction,
         InteractionData,
         PlaybookStatus,
@@ -149,6 +151,8 @@
             "InteractionData",
             "PlaybookStatus",
             "ProfileTimeToLive",
+            "FinalizeSessionRequest",
+            "FinalizeSessionResponse",
             "PublishUserInteractionRequest",
             "PublishUserInteractionResponse",
             "RerunPlaybookGenerationRequest",
@@ -214,6 +218,8 @@
     "EvaluationResultView",
     "ProfileChangeLogView",
     # Request types
+    "FinalizeSessionRequest",
+    "FinalizeSessionResponse",
     "PublishUserInteractionRequest",
     "DeleteUserProfileRequest",
     "DeleteUserInteractionRequest",

reflexio/client/client.py

@@ -124,6 +124,8 @@
     DeleteUserPlaybooksByIdsRequest,
     DeleteUserProfileRequest,
     DeleteUserProfileResponse,
+    FinalizeSessionRequest,
+    FinalizeSessionResponse,
     GetOperationStatusResponse,
     GetSessionOutcomesRequest,
     GetSessionOutcomesResponse,
@@ -457,6 +459,40 @@ def _make_request(
                 "Is REFLEXIO_URL pointing at the API host?"
             ) from exc
 
+    def finalize_session(
+        self, user_id: str, session_id: str
+    ) -> FinalizeSessionResponse:
+        """Accept an optional completion hint for saved messages.
+
+        Returns acceptance, not extraction completion. The judge remains
+        enabled, and the session can receive later publications.
+        """
+        request = FinalizeSessionRequest(user_id=user_id, session_id=session_id)
+        result = FinalizeSessionResponse(
+            **self._make_request(
+                "POST", "/api/finalize_session", json=request.model_dump(mode="json")
+            )
+        )
+        if result.success:
+            self._cache.invalidate("get_profiles")
+            self._cache.invalidate("get_agent_playbooks")
+        return result
+
+    async def finalize_session_async(
+        self, user_id: str, session_id: str
+    ) -> FinalizeSessionResponse:
+        """Native-async counterpart to :meth:`finalize_session`."""
+        request = FinalizeSessionRequest(user_id=user_id, session_id=session_id)
+        result = FinalizeSessionResponse(
+            **await self._make_async_request(
+                "POST", "/api/finalize_session", json=request.model_dump(mode="json")
+            )
+        )
+        if result.success:
+            self._cache.invalidate("get_profiles")
+            self._cache.invalidate("get_agent_playbooks")
+        return result
+
     def _publish_interaction_sync(
         self,
         request: PublishUserInteractionRequest,
@@ -511,6 +547,7 @@ def _build_publish_interaction_request(
         override_learning_stall: bool,
         retrieval_experiment_id: str | None,
         retrieval_experiment_arm: Literal["treatment", "holdout"] | None,
+        session_finished: bool = False,
     ) -> PublishUserInteractionRequest:
         """Validate and build the request shared by sync and async publishing."""
         if session_id is None or not session_id.strip():
@@ -531,6 +568,7 @@ def _build_publish_interaction_request(
             agent_version=agent_version,
             skip_aggregation=skip_aggregation,
             force_extraction=force_extraction,
+            session_finished=session_finished,
             evaluation_only=evaluation_only,
             override_learning_stall=override_learning_stall,
             retrieval_experiment_id=retrieval_experiment_id,
@@ -562,6 +600,7 @@ def publish_interaction(
         override_learning_stall: bool = False,
         retrieval_experiment_id: str | None = None,
         retrieval_experiment_arm: Literal["treatment", "holdout"] | None = None,
+        session_finished: bool = False,
     ) -> PublishUserInteractionResponse:
         """Publish user interactions.
 
@@ -598,6 +637,8 @@ def publish_interaction(
             force_extraction: If True, bypass all extraction gates
                 (stride_size, cheap pre-filter, LLM should_run) and
                 always run extractors.
+            session_finished: Optional completion hint; evaluates the saved prefix
+                through the judge without waiting for a full window. Later messages remain valid.
             evaluation_only: If True, store the interaction for
                 session-level evaluation only and permanently exclude it
                 from profile/playbook extraction.
@@ -639,6 +680,7 @@ def publish_interaction(
             session_id=session_id,
             skip_aggregation=skip_aggregation,
             force_extraction=force_extraction,
+            session_finished=session_finished,
             evaluation_only=evaluation_only,
             override_learning_stall=override_learning_stall,
             retrieval_experiment_id=retrieval_experiment_id,
@@ -669,6 +711,7 @@ async def publish_interaction_async(
         override_learning_stall: bool = False,
         retrieval_experiment_id: str | None = None,
         retrieval_experiment_arm: Literal["treatment", "holdout"] | None = None,
+        session_finished: bool = False,
     ) -> PublishUserInteractionResponse:
         """Native-async counterpart to :meth:`publish_interaction`."""
         request = self._build_publish_interaction_request(
@@ -679,6 +722,7 @@ async def publish_interaction_async(
             session_id=session_id,
             skip_aggregation=skip_aggregation,
             force_extraction=force_extraction,
+            session_finished=session_finished,
             evaluation_only=evaluation_only,
             override_learning_stall=override_learning_stall,
             retrieval_experiment_id=retrieval_experiment_id,
@@ -3724,6 +3768,20 @@ def __getattr__(self, name: str) -> Any:
             )
         return attribute
 
+    def finalize_session(
+        self, user_id: str, session_id: str | None = None
+    ) -> FinalizeSessionResponse:
+        """Finalize the bound session's saved prefix, retaining the judge."""
+        return self._client.finalize_session(user_id, self._resolve(session_id))
+
+    async def finalize_session_async(
+        self, user_id: str, session_id: str | None = None
+    ) -> FinalizeSessionResponse:
+        """Native-async completion hint for the bound session."""
+        return await self._client.finalize_session_async(
+            user_id, self._resolve(session_id)
+        )
+
     # -- publishing -------------------------------------------------------
 
     def publish_interaction(
@@ -3740,6 +3798,7 @@ def publish_interaction(
         override_learning_stall: bool = False,
         retrieval_experiment_id: str | None = None,
         retrieval_experiment_arm: Literal["treatment", "holdout"] | None = None,
+        session_finished: bool = False,
     ) -> PublishUserInteractionResponse:
         """Publish interactions into the bound session.
 
@@ -3755,6 +3814,7 @@ def publish_interaction(
             wait_for_response=wait_for_response,
             skip_aggregation=skip_aggregation,
             force_extraction=force_extraction,
+            session_finished=session_finished,
             evaluation_only=evaluation_only,
             override_learning_stall=override_learning_stall,
             retrieval_experiment_id=retrieval_experiment_id,
@@ -3775,6 +3835,7 @@ async def publish_interaction_async(
         override_learning_stall: bool = False,
         retrieval_experiment_id: str | None = None,
         retrieval_experiment_arm: Literal["treatment", "holdout"] | None = None,
+        session_finished: bool = False,
     ) -> PublishUserInteractionResponse:
         """Native-async counterpart to :meth:`publish_interaction`."""
         return await self._client.publish_interaction_async(
@@ -3786,6 +3847,7 @@ async def publish_interaction_async(
             wait_for_response=wait_for_response,
             skip_aggregation=skip_aggregation,
             force_extraction=force_extraction,
+            session_finished=session_finished,
             evaluation_only=evaluation_only,
             override_learning_stall=override_learning_stall,
             retrieval_experiment_id=retrieval_experiment_id,

reflexio/lib/_interactions.py

@@ -25,6 +25,8 @@
     DeleteSessionResponse,
     DeleteUserInteractionRequest,
     DeleteUserInteractionResponse,
+    FinalizeSessionRequest,
+    FinalizeSessionResponse,
     PublishUserInteractionRequest,
     PublishUserInteractionResponse,
 )
@@ -181,6 +183,46 @@ def publish_interaction(
         except Exception as e:
             return PublishUserInteractionResponse(success=False, message=str(e))
 
+    def finalize_session(
+        self,
+        request: FinalizeSessionRequest | dict,
+    ) -> FinalizeSessionResponse:
+        """Accept an optional completion hint without bypassing the judge.
+
+        The saved prefix is processed asynchronously. Later publications are
+        accepted. A missing session is a successful no-op.
+        """
+        if not self._is_storage_configured():
+            return FinalizeSessionResponse(
+                success=False, message=STORAGE_NOT_CONFIGURED_MSG
+            )
+        if isinstance(request, dict):
+            request = FinalizeSessionRequest(**request)
+        accepted = self._get_storage().finalize_session(
+            request.user_id, request.session_id
+        )
+        from reflexio.server.services.durable_learning.local import (
+            ensure_local_extraction,
+        )
+
+        if accepted:
+            try:
+                ensure_local_extraction(self.request_context)
+            except Exception as error:
+                # The marker is durable; scheduler wakeup is best-effort.
+                logger.warning(
+                    "event=session_finalization_wakeup_failed error_type=%s",
+                    type(error).__name__,
+                    exc_info=True,
+                )
+        return FinalizeSessionResponse(
+            success=True,
+            accepted=accepted,
+            message="Completion hint accepted; learning is asynchronous."
+            if accepted
+            else "No saved session to finalize.",
+        )
+
     def search_interactions(
         self,
         request: SearchInteractionRequest | dict,

reflexio/models/api_schema/domain/entities.py

@@ -88,6 +88,8 @@
     "ClearUserDataRequest",
     "ClearUserDataResponse",
     "InteractionData",
+    "FinalizeSessionRequest",
+    "FinalizeSessionResponse",
     "PublishUserInteractionRequest",
     "PublishUserInteractionResponse",
     "WhoamiResponse",
@@ -1611,6 +1613,9 @@ class PublishUserInteractionRequest(CapturesUnknownFields):
     # this is used for aggregating interactions for generating agent playbooks
     agent_version: str = Field(default="", max_length=1_000)
     session_id: NonEmptyStr  # used for grouping requests together
+    session_finished: bool = (
+        False  # Finalize the saved prefix without bypassing the judge.
+    )
     skip_aggregation: bool = (
         False  # when True, extract profiles/playbooks but skip aggregation
     )
@@ -1771,6 +1776,21 @@ def validate_retrieved_learnings_total(self) -> Self:
 
 
 # publish user interaction response
+class FinalizeSessionRequest(BaseModel):
+    """Optional completion hint for an authenticated user's saved session."""
+
+    user_id: NonEmptyStr
+    session_id: NonEmptyStr
+
+
+class FinalizeSessionResponse(BaseModel):
+    """Acceptance only; extraction continues asynchronously through its judge."""
+
+    success: bool
+    accepted: bool = False
+    message: str = ""
+
+
 class PublishUserInteractionResponse(BaseModel):
     success: bool
     message: str = ""

reflexio/models/config_schema.py

@@ -924,6 +924,8 @@ class Config(BaseModel):
     # extraction parameters
     window_size: int = Field(default=DEFAULT_WINDOW_SIZE, gt=0)
     stride_size: int = Field(default=DEFAULT_STRIDE_SIZE, gt=0)
+    # Captured at admission; 0 disables idle finalization for new publications.
+    extraction_idle_timeout_seconds: int = Field(default=21600, ge=0)
     # API key configuration for LLM providers
     api_key_config: APIKeyConfig | None = None
     # LLM model configuration overrides

reflexio/server/routes/interactions.py

@@ -33,6 +33,8 @@
     DeleteSessionResponse,
     DeleteUserInteractionRequest,
     DeleteUserInteractionResponse,
+    FinalizeSessionRequest,
+    FinalizeSessionResponse,
     GetSessionOutcomesRequest,
     GetSessionOutcomesResponse,
     LearningStatusResponse,
@@ -60,6 +62,18 @@
 PUBLISH_REQUEST_TIMEOUT_SECONDS = 240.0
 
 
+@router.post("/api/finalize_session", response_model=FinalizeSessionResponse)
+@limiter.limit("60/minute")
+def finalize_session(
+    request: Request,
+    payload: FinalizeSessionRequest,
+    org_id: str = Depends(default_get_org_id),
+    _gate: None = Depends(default_billing_gate("learnings_generated")),  # noqa: B008
+) -> FinalizeSessionResponse:
+    """Accept a completion hint; this does not wait for extraction."""
+    return publisher_api.get_reflexio(org_id=org_id).finalize_session(payload)
+
+
 @router.post(
     "/api/session_outcome",
     response_model=SetSessionOutcomeResponse,

reflexio/server/services/base_generation/_should_run.py

@@ -159,7 +159,7 @@ def _should_run_before_extraction(self, extractor_config: TExtractorConfig) -> b
             )
 
         # Cheap pre-filter: apply user-text-only heuristics (slash-commands,
-        # short turns, extractor-prompt echoes) without burning a 5–7s LLM
+        # extractor-prompt echoes) without burning a 5–7s LLM
         # call. Batches with no non-empty user text bypass these heuristics;
         # stride/window eligibility was already evaluated independently.
         from reflexio.server.services.base_generation_service import (
@@ -176,7 +176,7 @@ def _should_run_before_extraction(self, extractor_config: TExtractorConfig) -> b
             )
             # Report WHICH heuristic rejected, not just that we skipped: the
             # caller's bare ``should_skip`` cannot distinguish "the cheap
-            # pre-filter rejected on short/slash-command turns" from "the LLM
+            # pre-filter rejected on slash-command turns" from "the LLM
             # gate voted no", and the cheap path never reaches the LLM so there
             # is no token spend to infer it from either.
             #