feat(sessions): deliver durable Stop directly to the runner

api/entrypoints/routers.py

@@ -182,6 +182,10 @@
 from oss.src.dbs.postgres.sessions.streams.dbes import SessionStreamDBE  # noqa: F401
 from oss.src.dbs.postgres.sessions.streams.dao import SessionStreamsDAO
 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.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
 from oss.src.dbs.redis.shared.engine import get_lock_engine
 
@@ -838,6 +842,7 @@ async def lifespan(*args, **kwargs):
 interactions_service = SessionInteractionsService(
     interactions_dao=interactions_dao,
     watch_publisher=_sessions_watch_publisher,
+    records_service=records_service,
 )
 
 triggers_service = TriggersService(
@@ -1115,6 +1120,26 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     records_service=records_service,
 )
 
+# Durable session commands (Stop). The control-delivery adapter is chosen by one setting.
+# `direct` posts the command to the runner's own /cancel over the hop that already carries hard
+# kill; `long_poll` is not built yet, and naming it fails at boot rather than silently falling
+# back to a transport the operator did not choose.
+_control_adapter = (env.agenta.sessions.commands.adapter or "direct").strip().lower()
+if _control_adapter != "direct":
+    raise RuntimeError(
+        f"AGENTA_SESSIONS_CONTROL_ADAPTER={_control_adapter!r} is not available in this build. "
+        "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(),
+)
+
 sessions = SessionsRouter(
     streams_service=session_streams_service,
     records_service=records_service,
@@ -1125,6 +1150,7 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     mounts_service=mounts_service,
     turns_service=session_turns_service,
     sessions_service=sessions_service,
+    commands_service=session_commands_service,
     respond_task=_interactions_worker.respond_interaction,
     interactions_dispatcher=_interactions_dispatcher,
 )
@@ -1599,6 +1625,12 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     tags=["Sessions"],
 )
 
+# After `root`, so the literal /sessions/<verb> routes always win a path match.
+app.include_router(
+    router=sessions.control.router,
+    tags=["Sessions"],
+)
+
 
 @app.get("/health", operation_id="health_check", tags=["Status"])
 async def health_check():

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

@@ -0,0 +1,164 @@
+"""add session commands, and the two session_streams columns a Stop needs
+
+A user Stop reached the runner only through the absence of a Redis lock, discovered on the next
+heartbeat up to 30 seconds later. Nothing recorded that a Stop had been asked for, so a Stop
+against an unreachable runner was simply lost and no execution ever reached a terminal outcome
+anyone could read.
+
+`session_commands` is that record. One row per durable request to change an execution. `state`
+is where the COMMAND is (pending, claimed, applied, obsolete); `outcome` is what happened to the
+EXECUTION (stopped, not_running, superseded_by_newer_turn, failed, lost). The two are separate
+columns because they answer different questions and settle at different times.
+
+Two columns join `session_streams`:
+
+  * `stopping_turn_id` names the execution an accepted Stop is waiting on, written in the same
+    transaction as the command insert and cleared at settlement.
+  * `turn_started_at` records when the row's current `turn_id` started. Nothing else could serve
+    the stale-Stop guard: `updated_at` is the heartbeat timestamp and moves every 30 seconds,
+    runner-minted turn ids are uuid4 and carry no time, the Redis lock value is a bare turn id
+    that a Lua compare reads whole, and the `session_turns` append is fire-and-forget so a
+    running turn may have no row at all.
+
+Both are nullable and backfill to NULL. A row written before this migration yields no
+comparison, and the guard then does not fire — deliberately, because a guard that refused every
+Stop it could not verify would break the common case to protect a rare one.
+
+Revision ID: oss000000022
+Revises: oss000000021
+Create Date: 2026-09-02 23:30:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "oss000000022"
+down_revision: Union[str, None] = "oss000000021"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "session_commands",
+        sa.Column("id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("project_id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("kind", sa.String(), nullable=False),
+        sa.Column("target_turn_id", sa.String(), nullable=True),
+        sa.Column("expected_turn_id", sa.String(), nullable=True),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("claimed_by", sa.String(), nullable=True),
+        sa.Column("claim_expires_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column(
+            "claim_count",
+            sa.Integer(),
+            server_default="0",
+            nullable=False,
+        ),
+        sa.Column("outcome", sa.String(), nullable=True),
+        sa.Column("idempotency_key", sa.String(), nullable=True),
+        sa.Column("settled_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("data", sa.JSON(), nullable=True),
+        sa.Column(
+            "flags",
+            postgresql.JSONB(none_as_null=True),
+            nullable=True,
+        ),
+        sa.Column(
+            "tags",
+            postgresql.JSONB(none_as_null=True),
+            nullable=True,
+        ),
+        sa.Column("meta", sa.JSON(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            server_default=sa.func.current_timestamp(),
+            nullable=True,
+        ),
+        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("created_by_id", sa.UUID(as_uuid=True), nullable=True),
+        sa.Column("updated_by_id", sa.UUID(as_uuid=True), nullable=True),
+        sa.Column("deleted_by_id", sa.UUID(as_uuid=True), nullable=True),
+        sa.CheckConstraint("kind IN ('cancel')", name="ck_session_commands_kind"),
+        sa.CheckConstraint(
+            "state IN ('pending', 'claimed', 'applied', 'obsolete')",
+            name="ck_session_commands_state",
+        ),
+        sa.ForeignKeyConstraint(
+            ["project_id"],
+            ["projects.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("project_id", "id"),
+        sa.UniqueConstraint(
+            "project_id",
+            "session_id",
+            "idempotency_key",
+            name="uq_session_commands_idempotency",
+        ),
+    )
+    # One open command per target execution, enforced by the database because admission's
+    # read-then-insert races itself: two Stops in the same instant both find no open command.
+    op.create_index(
+        "uq_session_commands_open_target",
+        "session_commands",
+        ["project_id", "session_id", "kind", "target_turn_id"],
+        unique=True,
+        postgresql_where=sa.text(
+            "state IN ('pending', 'claimed') AND deleted_at IS NULL"
+        ),
+    )
+    op.create_index(
+        "ix_session_commands_open",
+        "session_commands",
+        ["project_id", "session_id", "created_at"],
+        postgresql_where=sa.text(
+            "state IN ('pending', 'claimed') AND deleted_at IS NULL"
+        ),
+    )
+    op.create_index(
+        "ix_session_commands_claims",
+        "session_commands",
+        ["claim_expires_at"],
+        postgresql_where=sa.text("state = 'claimed' AND deleted_at IS NULL"),
+    )
+    op.create_index(
+        "ix_session_commands_project_session",
+        "session_commands",
+        ["project_id", "session_id", "created_at"],
+    )
+    # The runner reports an outcome with the command id alone; it holds no project credential,
+    # so that read cannot use the primary key's leading column.
+    op.create_index(
+        "ix_session_commands_id",
+        "session_commands",
+        ["id"],
+    )
+
+    op.add_column(
+        "session_streams",
+        sa.Column("stopping_turn_id", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "session_streams",
+        sa.Column("turn_started_at", sa.TIMESTAMP(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("session_streams", "turn_started_at")
+    op.drop_column("session_streams", "stopping_turn_id")
+    op.drop_index("ix_session_commands_id", table_name="session_commands")
+    op.drop_index("ix_session_commands_project_session", table_name="session_commands")
+    op.drop_index("ix_session_commands_claims", table_name="session_commands")
+    op.drop_index("ix_session_commands_open", table_name="session_commands")
+    op.drop_index("uq_session_commands_open_target", table_name="session_commands")
+    op.drop_table("session_commands")

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

@@ -346,3 +346,77 @@ class SessionRecordIngestRequest(BaseModel):
     # Both forward-fill only (tracing-DB rule) — absent on producers that predate this.
     turn_id: Optional[str] = None
     span_id: Optional[OTelSpanId] = None
+
+
+# ---------------------------------------------------------------------------
+# Session control: durable commands (Stop)
+# ---------------------------------------------------------------------------
+
+
+class SessionCancelRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    # Optional stale-request guard. When present, the API cancels only this execution and
+    # 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
+
+
+class SessionCommandRef(BaseModel):
+    """The durable command an accepted request created. Identity and DELIVERY state only.
+
+    A client must not read execution state from it. `state` says where the command is; the
+    session's own state says what the execution is doing.
+    """
+
+    id: UUID
+    state: Literal["pending", "claimed", "applied", "obsolete"]
+
+
+class SessionExecutionRef(BaseModel):
+    """What the caller should render. `id` is null when the session was idle."""
+
+    id: Optional[str] = None
+    state: Literal["stopping", "idle"]
+
+
+class SessionCancelResponse(BaseModel):
+    command: SessionCommandRef
+    execution: SessionExecutionRef
+
+
+class SessionExecutionOutcome(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    # The execution the runner acted on. Null when it held none.
+    id: Optional[str] = None
+    # stopped: cancelled as asked. not_running: no such execution on this runner.
+    # superseded_by_newer_turn: the held execution started after the command arrived.
+    # failed: the cancel itself failed.
+    state: Literal["stopped", "failed", "not_running", "superseded_by_newer_turn"]
+    # Short and human-readable, present only when `state` is "failed".
+    error: Optional[str] = Field(default=None, max_length=2000)
+
+
+class SessionControlOutcomeRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    replica_id: str = Field(min_length=1, max_length=128)
+    # The command's terminal state. `applied` means the runner did the work; `obsolete` means
+    # there was nothing to do.
+    result: Literal["applied", "obsolete"]
+    execution: SessionExecutionOutcome
+
+
+class SessionCommandSettlement(BaseModel):
+    id: UUID
+    state: Literal["applied", "obsolete"]
+    outcome: Literal[
+        "stopped", "not_running", "superseded_by_newer_turn", "failed", "lost"
+    ]
+    settled_at: Optional[datetime] = None
+
+
+class SessionControlOutcomeResponse(BaseModel):
+    command: SessionCommandSettlement