fix(sessions): settle executions after runner or sandbox loss

api/entrypoints/routers.py

@@ -184,6 +184,7 @@
 from oss.src.core.sessions.streams.service import SessionStreamsService
 from oss.src.dbs.postgres.sessions.commands.dbes import SessionCommandDBE  # noqa: F401
 from oss.src.dbs.postgres.sessions.commands.dao import SessionCommandsDAO
+from oss.src.dbs.postgres.sessions.executions.dao import SessionExecutionsDAO
 from oss.src.core.sessions.commands.service import SessionCommandsService
 from oss.src.dbs.http.sessions.control_delivery_direct import DirectControlDelivery
 from oss.src.tasks.asyncio.sessions.orphan_sweep import orphan_sweep_loop
@@ -283,8 +284,16 @@ async def lifespan(*args, **kwargs):
         except Exception as e:  # noqa: BLE001
             log.warning("Store bucket ensure failed at startup: %s", e)
 
+    # The execution watchdog. It needs the records plane to write the terminal outcome a
+    # dead runner owed, and the watch publisher so an open browser sees the turn close.
     _orphan_sweep_task = asyncio.create_task(
-        orphan_sweep_loop(_transactions_engine, _lock_engine)
+        orphan_sweep_loop(
+            _transactions_engine,
+            _lock_engine,
+            records_service=records_service,
+            watch_publisher=_sessions_watch_publisher,
+            commands_service=session_commands_service,
+        )
     )
 
     _attachment_sweep_task = asyncio.create_task(
@@ -591,6 +600,8 @@ async def lifespan(*args, **kwargs):
 folders_dao = FoldersDAO(engine=_transactions_engine)
 session_streams_dao = SessionStreamsDAO(engine=_transactions_engine)
 session_turns_dao = SessionTurnsDAO(engine=_transactions_engine)
+session_commands_dao = SessionCommandsDAO(engine=_transactions_engine)
+session_executions_dao = SessionExecutionsDAO(engine=_transactions_engine)
 
 connections_dao = ConnectionsDAO(engine=_transactions_engine)
 mounts_dao = MountsDAO(engine=_transactions_engine)
@@ -625,6 +636,7 @@ async def lifespan(*args, **kwargs):
 
 records_service = RecordsService(
     records_dao=records_dao,
+    executions_dao=session_executions_dao,
 )
 
 
@@ -1131,13 +1143,13 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
         "Only 'direct' is implemented; the long-poll adapter is a later change."
     )
 
-session_commands_dao = SessionCommandsDAO()
 session_commands_service = SessionCommandsService(
     commands_dao=session_commands_dao,
     streams_service=session_streams_service,
     interactions_service=interactions_service,
     lock_engine=_lock_engine,
     delivery=DirectControlDelivery(),
+    executions_dao=session_executions_dao,
 )
 
 sessions = SessionsRouter(

api/oss/databases/postgres/migrations/core_oss/versions/oss000000023_add_session_executions.py

@@ -0,0 +1,43 @@
+"""add authoritative session execution terminal outcomes
+
+Revision ID: oss000000023
+Revises: oss000000022
+Create Date: 2026-09-03 22:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "oss000000023"
+down_revision: Union[str, None] = "oss000000022"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "session_executions",
+        sa.Column("project_id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("execution_id", sa.String(), nullable=False),
+        sa.Column("terminal_outcome", sa.String(), nullable=False),
+        sa.Column("settled_by", sa.String(), nullable=False),
+        sa.Column("settled_at", sa.TIMESTAMP(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["project_id"], ["projects.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("project_id", "session_id", "execution_id"),
+    )
+    op.create_index(
+        "ix_session_executions_project_session",
+        "session_executions",
+        ["project_id", "session_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_session_executions_project_session", table_name="session_executions"
+    )
+    op.drop_table("session_executions")

api/oss/databases/postgres/migrations/core_oss/versions/oss000000024_add_execution_redis_reconciliation.py

@@ -0,0 +1,41 @@
+"""track execution Redis reconciliation
+
+Revision ID: oss000000024
+Revises: oss000000023
+Create Date: 2026-09-03 22:30:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "oss000000024"
+down_revision: Union[str, None] = "oss000000023"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "session_executions",
+        sa.Column("redis_reconciled_at", sa.TIMESTAMP(timezone=True), nullable=True),
+    )
+    op.create_index(
+        "ix_session_executions_redis_unreconciled",
+        "session_executions",
+        ["settled_at"],
+        postgresql_where=sa.text(
+            "settled_by = 'runner' AND terminal_outcome = 'stopped' "
+            "AND redis_reconciled_at IS NULL"
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_session_executions_redis_unreconciled",
+        table_name="session_executions",
+    )
+    op.drop_column("session_executions", "redis_reconciled_at")

api/oss/databases/postgres/migrations/core_oss/versions/oss000000026_add_execution_ending_marker.py

@@ -0,0 +1,38 @@
+"""add session execution ending marker
+
+Revision ID: oss000000026
+Revises: oss000000024
+Create Date: 2026-09-04 12:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "oss000000026"
+down_revision: Union[str, None] = "oss000000024"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "session_executions",
+        sa.Column("ending_written_at", sa.TIMESTAMP(timezone=True), nullable=True),
+    )
+    op.create_index(
+        "ix_session_executions_ending_unwritten",
+        "session_executions",
+        ["settled_at"],
+        postgresql_where=sa.text("ending_written_at IS NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_session_executions_ending_unwritten",
+        table_name="session_executions",
+    )
+    op.drop_column("session_executions", "ending_written_at")

api/oss/databases/postgres/migrations/tracing_oss/versions/oss000000005_add_records_quarantined_at.py

@@ -0,0 +1,35 @@
+"""add_records_quarantined_at
+
+Revision ID: oss000000005
+Revises: oss000000004
+Create Date: 2026-09-03 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "oss000000005"
+down_revision: Union[str, None] = "oss000000004"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # A record that reached ingest for a turn the execution watchdog had already ended.
+    # Nullable and forward-fill only, like every other column on this table: the tracing DB
+    # is never backfilled, and no existing row can be classified retroactively anyway.
+    #
+    # No index. Every read that filters on it is already scoped to one project and one
+    # session by an existing index, and the column is null on all but a handful of rows.
+    op.add_column(
+        "records",
+        sa.Column("quarantined_at", sa.TIMESTAMP(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("records", "quarantined_at")

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

@@ -360,7 +360,13 @@ class SessionCancelRequest(BaseModel):
     # refuses the request if another one is running. When absent, it cancels whichever
     # execution is active when the request is applied. A person never types this: the browser
     # fills it from the session's own state, and a first-party client always sends it.
-    expected_execution_id: Optional[str] = None
+    expected_execution_id: Optional[str] = Field(
+        default=None,
+        description=(
+            "Optional stale-request guard honored only in cancel mode; ignored for send, "
+            "steer, and attach."
+        ),
+    )
 
 
 class SessionCommandRef(BaseModel):

api/oss/src/core/sessions/commands/interfaces.py

@@ -7,7 +7,7 @@
 
 from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import List, NamedTuple, Optional
+from typing import Any, AsyncContextManager, List, NamedTuple, Optional
 from uuid import UUID
 
 from pydantic import BaseModel
@@ -69,6 +69,10 @@ async def acknowledge(self, *, command_id: UUID, replica_id: str) -> None:
 
 
 class SessionCommandsDAOInterface(ABC):
+    @abstractmethod
+    def transaction(self) -> AsyncContextManager[Any]:
+        """Open a transaction that sibling session DAOs can share."""
+
     @abstractmethod
     async def create_command(
         self,
@@ -147,11 +151,23 @@ async def claim_for_delivery(
         a direct call. The long-poll adapter reaches the same transition through
         `claim_commands`; both exist so the outcome route's guard reads the same either way."""
 
+    @abstractmethod
+    async def record_delivery_attempt(
+        self,
+        *,
+        project_id: UUID,
+        command_id: UUID,
+        now: datetime,
+        max_deliveries: int,
+    ) -> Optional[SessionCommand]:
+        """Reserve one bounded delivery attempt and return the updated command."""
+
     @abstractmethod
     async def settle_command(
         self,
         *,
         settle: SessionCommandSettle,
+        transaction: Optional[Any] = None,
     ) -> Optional[SessionCommand]:
         """Terminal transition, guarded on `state='claimed' AND claimed_by=:replica_id`.
         None means the claim had expired or somebody else settled it first."""
@@ -173,6 +189,6 @@ async def expire_claims(
         *,
         now: datetime,
         max_deliveries: int,
+        pending_before: Optional[datetime] = None,
     ) -> List[SessionCommand]:
-        """Commands whose claim lease has passed. The settlement sweep reads this. Not called
-        in this slice; the execution watchdog owns settlement (see the slice document)."""
+        """Pending or claimed commands old enough for recovery."""