feat(runner): reliable multi-gate approvals: truthful results, persisted answers, per-card dispatch

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

@@ -1,8 +1,8 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Literal, Optional
 from uuid import UUID
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from oss.src.core.sessions.streams.dtos import (
     SessionStream,
@@ -101,16 +101,34 @@ class SessionInteractionCreateRequest(BaseModel):
     meta: Optional[Dict[str, Any]] = None
 
 
+class SessionInteractionResolution(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    verdict: Literal["approved", "denied"]
+    tool_call_id: str
+
+
 class SessionInteractionTransitionRequest(BaseModel):
     # No project_id: scope comes from the caller's credential (request.state).
     session_id: str
     token: str
     status: SessionInteractionStatus
+    resolution: Optional[SessionInteractionResolution] = None
+
+    @model_validator(mode="after")
+    def validate_resolution_status(self) -> "SessionInteractionTransitionRequest":
+        # Resolution is answer data, so it is valid only on the resolved lifecycle edge.
+        if (
+            self.resolution is not None
+            and self.status != SessionInteractionStatus.resolved
+        ):
+            raise ValueError("resolution is only valid when status is resolved")
+        return self
 
 
 class SessionInteractionCancelStaleRequest(BaseModel):
-    # Cancels prior turns' pending gates, sparing this turn's own (`turn_id`) and any gates
-    # the current turn answers in-band (`tokens` — a resume must resolve them, not cancel).
+    # Cancels prior turns' pending gates, sparing this turn's own (`turn_id`) and every prior
+    # gate still owned by a live partial resume (`tokens`, including carried gates).
     session_id: str
     turn_id: str
     tokens: Optional[List[str]] = None

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

@@ -54,6 +54,8 @@
 from oss.src.core.sessions.records.streaming import publish_record
 from oss.src.core.sessions.interactions.dtos import (
     SessionInteractionCreate,
+    SessionInteractionKind,
+    SessionInteractionQuery,
     SessionInteractionStatus,
     SessionInteractionTransition,
 )
@@ -640,13 +642,41 @@ async def transition_interaction(
         ):
             raise FORBIDDEN_EXCEPTION
 
+        resolution = (
+            body.resolution.model_dump() if body.resolution is not None else None
+        )
+        if resolution is not None:
+            interactions = await self.interactions_service.query_interactions(
+                project_id=project_id,
+                query=SessionInteractionQuery(session_id=body.session_id),
+            )
+            source = next(
+                (
+                    interaction
+                    for interaction in interactions
+                    if interaction.token == body.token
+                ),
+                None,
+            )
+            if source is None:
+                raise HTTPException(
+                    status_code=status.HTTP_404_NOT_FOUND,
+                    detail="Interaction not found or already terminal",
+                )
+            if source.kind != SessionInteractionKind.user_approval:
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail="Resolution is only valid for user approval interactions",
+                )
+
         try:
             interaction = await self.interactions_service.transition_interaction(
                 transition=SessionInteractionTransition(
                     project_id=project_id,
                     session_id=body.session_id,
                     token=body.token,
                     status=body.status,
+                    resolution=resolution,
                 ),
             )
         except InteractionNotFound:

api/oss/src/core/sessions/interactions/dtos.py

@@ -68,6 +68,7 @@ class SessionInteractionTransition(BaseModel):
     session_id: str
     token: str
     status: SessionInteractionStatus
+    resolution: Optional[Dict[str, Any]] = None
 
 
 class SessionInteractionQuery(BaseModel):

api/oss/src/dbs/postgres/sessions/interactions/dao.py

@@ -2,7 +2,8 @@
 from typing import List, Optional
 from uuid import UUID
 
-from sqlalchemy import delete as sa_delete, func, select, update as sa_update
+from sqlalchemy import cast, delete as sa_delete, func, select, update as sa_update
+from sqlalchemy.dialects.postgresql import JSON, JSONB, array
 from sqlalchemy.exc import IntegrityError
 
 from oss.src.core.sessions.interactions.dtos import (
@@ -97,6 +98,24 @@ async def transition_interaction(
             # Only non-terminal interactions transition: pending (responded|resolved|
             # cancelled) and responded (resolved, when the runner consumes an API-plane
             # answer). resolved/cancelled are terminal.
+            transition_values = {
+                "status": transition.status.value,
+                "updated_at": datetime.now(timezone.utc),
+            }
+            if transition.resolution is not None:
+                # Preserve request/references while atomically adding the answer under the guard.
+                transition_values["data"] = cast(
+                    func.jsonb_set(
+                        func.coalesce(
+                            cast(SessionInteractionDBE.data, JSONB),
+                            cast({}, JSONB),
+                        ),
+                        array(["resolution"]),
+                        cast(transition.resolution, JSONB),
+                        True,
+                    ),
+                    JSON,
+                )
             stmt = (
                 sa_update(SessionInteractionDBE)
                 .where(
@@ -105,10 +124,7 @@ async def transition_interaction(
                     SessionInteractionDBE.token == transition.token,
                     SessionInteractionDBE.status.in_(("pending", "responded")),
                 )
-                .values(
-                    status=transition.status.value,
-                    updated_at=datetime.now(timezone.utc),
-                )
+                .values(**transition_values)
                 .returning(SessionInteractionDBE)
             )
             result = await session.execute(stmt)

api/oss/tests/pytest/unit/sessions/test_interactions_dispatcher.py

@@ -5,6 +5,9 @@
 
 from unittest.mock import AsyncMock, MagicMock
 
+from oss.src.apis.fastapi.sessions.models import SessionInteractionCreateRequest
+from oss.src.core.sessions.interactions.dtos import SessionInteractionKind
+
 from oss.src.tasks.asyncio.sessions.interactions_dispatcher import (
     InteractionsDispatcher,
 )
@@ -31,6 +34,19 @@ def _make_interaction(*, with_refs=True):
     )
 
 
+def test_create_request_accepts_omitted_workflow_references():
+    request = SessionInteractionCreateRequest(
+        session_id="sess-no-refs",
+        turn_id="turn-no-refs",
+        token="token-no-refs",
+        kind=SessionInteractionKind.user_approval,
+        data={"request": {"tool": "bash", "args": {"command": "pwd"}}},
+    )
+
+    assert request.data is not None
+    assert request.data.references is None
+
+
 async def test_respond_fallback_calls_invoke_when_no_dispatch_fn():
     interaction = _make_interaction()
     project_id = uuid4()
@@ -60,6 +76,35 @@ async def test_respond_fallback_calls_invoke_when_no_dispatch_fn():
     assert invoke_kwargs["user_id"] == user_id
 
 
+async def test_respond_without_references_builds_a_safe_reference_less_request():
+    interaction = _make_interaction(with_refs=False)
+    project_id = uuid4()
+    user_id = uuid4()
+
+    interactions_service = MagicMock()
+    interactions_service.fetch_interaction = AsyncMock(return_value=interaction)
+
+    workflows_service = MagicMock()
+    workflows_service.invoke_workflow = AsyncMock(return_value=SimpleNamespace())
+
+    worker = InteractionsDispatcher(
+        workflows_service=workflows_service,
+        interactions_service=interactions_service,
+    )
+
+    await worker.respond(
+        project_id=project_id,
+        user_id=user_id,
+        interaction_id=interaction.id,
+        answer={"approved": True},
+    )
+
+    workflows_service.invoke_workflow.assert_awaited_once()
+    invoke_request = workflows_service.invoke_workflow.await_args.kwargs["request"]
+    assert invoke_request.references is None
+    assert invoke_request.session_id == interaction.session_id
+
+
 async def test_respond_detached_calls_dispatch_fn_not_invoke():
     interaction = _make_interaction()
     project_id = uuid4()

api/oss/tests/pytest/unit/sessions/test_interactions_transition_update.py

@@ -0,0 +1,80 @@
+from uuid import uuid4
+
+from sqlalchemy.dialects import postgresql
+
+from oss.src.core.sessions.interactions.dtos import (
+    SessionInteractionStatus,
+    SessionInteractionTransition,
+)
+from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO
+
+
+class _Result:
+    def scalar_one_or_none(self):
+        return None
+
+
+class _Session:
+    def __init__(self):
+        self.statement = None
+
+    async def execute(self, statement):
+        self.statement = statement
+        return _Result()
+
+    async def commit(self):
+        return None
+
+
+class _SessionContext:
+    def __init__(self, session):
+        self.session = session
+
+    async def __aenter__(self):
+        return self.session
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+class _Engine:
+    def __init__(self, session):
+        self._session = session
+
+    def session(self):
+        return _SessionContext(self._session)
+
+
+async def _transition_statement(resolution=None):
+    session = _Session()
+    dao = SessionInteractionsDAO(engine=_Engine(session))
+    await dao.transition_interaction(
+        transition=SessionInteractionTransition(
+            project_id=uuid4(),
+            session_id="session-1",
+            token="token-1",
+            status=SessionInteractionStatus.resolved,
+            resolution=resolution,
+        )
+    )
+    return session.statement
+
+
+async def test_resolution_update_preserves_existing_data_with_jsonb_set():
+    resolution = {"verdict": "approved", "tool_call_id": "tool-1"}
+    statement = await _transition_statement(resolution)
+    compiled = statement.compile(dialect=postgresql.dialect())
+    set_clause = str(compiled).split(" WHERE ", maxsplit=1)[0]
+
+    assert "jsonb_set" in set_clause
+    assert "CAST(session_interactions.data AS JSONB)" in set_clause
+    assert resolution in compiled.params.values()
+
+
+async def test_transition_without_resolution_does_not_write_data():
+    statement = await _transition_statement()
+    set_clause = str(statement.compile(dialect=postgresql.dialect())).split(
+        " WHERE ", maxsplit=1
+    )[0]
+
+    assert "data=" not in set_clause