[feat] Approve, deny, and stop a run from your phone (8/12)

api/entrypoints/routers.py

@@ -854,6 +854,7 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
 _interactions_dispatcher = InteractionsDispatcher(
     workflows_service=workflows_service,
     interactions_service=interactions_service,
+    records_service=records_service,
     dispatch_fn=_dispatch_detached_run,
 )
 
@@ -1098,6 +1099,7 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     turns_service=session_turns_service,
     sessions_service=sessions_service,
     respond_task=_interactions_worker.respond_interaction,
+    interactions_dispatcher=_interactions_dispatcher,
 )
 
 # PLATFORM ADMIN ---------------------------------------------------------------

api/entrypoints/worker_queues.py

@@ -49,6 +49,7 @@
 from oss.src.core.evaluators.service import EvaluatorsService, SimpleEvaluatorsService
 from oss.src.core.queries.service import QueriesService
 from oss.src.core.sessions.interactions.service import SessionInteractionsService
+from oss.src.core.sessions.records.service import RecordsService
 from oss.src.core.testcases.service import TestcasesService
 from oss.src.core.testsets.service import SimpleTestsetsService, TestsetsService
 from oss.src.core.tracing.service import TracingService
@@ -67,7 +68,11 @@
     QueryVariantDBE,
 )
 from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO
-from oss.src.dbs.postgres.shared.engine import get_transactions_engine
+from oss.src.dbs.postgres.sessions.records.dao import RecordsDAO
+from oss.src.dbs.postgres.shared.engine import (
+    get_analytics_engine,
+    get_transactions_engine,
+)
 from oss.src.dbs.postgres.testcases.dbes import TestcaseBlobDBE
 from oss.src.dbs.postgres.testsets.dbes import (
     TestsetArtifactDBE,
@@ -214,6 +219,11 @@ def _build_interactions_broker() -> tuple[AsyncBroker, int]:
     environments_service.embeds_service = embeds_service
 
     interactions_service = SessionInteractionsService(interactions_dao=interactions_dao)
+    # Approval answers replay the session's durable records into the resume conversation;
+    # records live on the analytics engine (same as the API composition in routers.py).
+    records_service = RecordsService(
+        records_dao=RecordsDAO(engine=get_analytics_engine()),
+    )
 
     async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
         result = await workflows_service.invoke_workflow_detached(
@@ -226,6 +236,7 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     interactions_dispatcher = InteractionsDispatcher(
         workflows_service=workflows_service,
         interactions_service=interactions_service,
+        records_service=records_service,
         dispatch_fn=_dispatch_detached_run,
     )
     InteractionsWorker(broker=broker, dispatcher=interactions_dispatcher)

api/oss/src/apis/fastapi/sessions/models.py

@@ -159,6 +159,9 @@ class SessionInteractionsResponse(BaseModel):
 
 
 class SessionInteractionRespondRequest(BaseModel):
+    # For a user_approval interaction the answer is {approved: bool, tool_call_id?: str,
+    # message?: str} — the dispatcher composes the full resume conversation server-side
+    # (interactions_dispatcher.compose_approval_messages). Other kinds pass through as-is.
     answer: Optional[Dict[str, Any]] = None
 
 

api/oss/src/apis/fastapi/sessions/router.py

@@ -639,10 +639,15 @@ def __init__(
         interactions_service: SessionInteractionsService,
         workflows_service: WorkflowsService,
         respond_task: Optional[Any] = None,
+        # InteractionsDispatcher (typed loosely, like respond_task: the API layer does not
+        # import the tasks layer). When present, the no-worker respond fallback goes through
+        # it so both paths share ONE answer-composition implementation.
+        interactions_dispatcher: Optional[Any] = None,
     ) -> None:
         self.interactions_service = interactions_service
         self.workflows_service = workflows_service
         self.respond_task = respond_task
+        self.interactions_dispatcher = interactions_dispatcher
 
         self.router = APIRouter()
 
@@ -910,15 +915,23 @@ async def respond_interaction(
                 detail="Interaction is no longer pending",
             )
 
-        # Enqueue onto the interactions worker when wired; otherwise fall back to an
-        # inline blocking invoke (keeps the route usable in minimal/test compositions).
+        # Enqueue onto the interactions worker when wired; otherwise fall back to the
+        # dispatcher directly (same answer composition, fired in-process), or as a last
+        # resort an inline blocking invoke (keeps minimal/test compositions usable).
         if self.respond_task is not None:
             await self.respond_task.kiq(
                 project_id=str(project_id),
                 user_id=str(user_id),
                 interaction_id=str(interaction_id),
                 answer=answer,
             )
+        elif self.interactions_dispatcher is not None:
+            await self.interactions_dispatcher.respond(
+                project_id=UUID(str(project_id)),
+                user_id=UUID(str(user_id)),
+                interaction_id=interaction_id,
+                answer=answer,
+            )
         else:
             references = (
                 {
@@ -1672,6 +1685,7 @@ def __init__(
         turns_service: SessionTurnsService,
         sessions_service: SessionsService,
         respond_task: Optional[Any] = None,
+        interactions_dispatcher: Optional[Any] = None,
     ) -> None:
         self.streams = SessionStreamsRouter(
             service=streams_service,
@@ -1682,6 +1696,7 @@ def __init__(
             interactions_service=interactions_service,
             workflows_service=workflows_service,
             respond_task=respond_task,
+            interactions_dispatcher=interactions_dispatcher,
         )
         self.attachments = SessionAttachmentsRouter(
             attachments_service=attachments_service,