[feat] Reuse Daytona sandboxes across turns instead of deleting them every turn

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

@@ -0,0 +1,28 @@
+"""add session state sandbox fingerprint
+
+Revision ID: oss000000011
+Revises: oss000000010
+Create Date: 2026-07-11 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "oss000000011"
+down_revision: Union[str, None] = "oss000000010"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "session_states",
+        sa.Column("sandbox_fingerprint", sa.String(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("session_states", "sandbox_fingerprint")

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

@@ -37,7 +37,7 @@ class SessionStreamCommandRequestModel(BaseModel):
 class SessionHeartbeatRequestModel(BaseModel):
     # project scope comes from the caller's credential, never the body
     session_id: str
-    replica_id: str
+    replica_id: str = Field(min_length=1)
     turn_id: Optional[str] = None
     is_running: bool = True
 
@@ -96,6 +96,17 @@ class SessionStateUpsertRequest(BaseModel):
         default=None,
         description="Remote sandbox id to record alongside the continuity state.",
     )
+    sandbox_fingerprint: Optional[str] = Field(
+        default=None,
+        description="Fingerprint of the sandbox create specification.",
+    )
+    sandbox_turn_index: Optional[int] = Field(
+        default=None,
+        description=(
+            "the writer's conversation turn index; the pointer write is applied only "
+            "when it is >= the row's data.latest_turn_index."
+        ),
+    )
 
 
 # ---------------------------------------------------------------------------

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

@@ -68,6 +68,10 @@ class SessionState(Lifecycle):
         default=None,
         description="Remote sandbox id — the single source of truth resume pointer.",
     )
+    sandbox_fingerprint: Optional[str] = Field(
+        default=None,
+        description="Fingerprint of the sandbox create specification.",
+    )
     flags: SessionStateFlags = Field(default_factory=SessionStateFlags)
     tags: Optional[Dict[str, Any]] = Field(default=None)
     meta: Optional[Dict[str, Any]] = Field(default=None)
@@ -85,3 +89,14 @@ class SessionStateUpsert(BaseModel):
         default=None,
         description="Remote sandbox id to record alongside the continuity state.",
     )
+    sandbox_fingerprint: Optional[str] = Field(
+        default=None,
+        description="Fingerprint of the sandbox create specification.",
+    )
+    sandbox_turn_index: Optional[int] = Field(
+        default=None,
+        description=(
+            "the writer's conversation turn index; the pointer write is applied only "
+            "when it is >= the row's data.latest_turn_index."
+        ),
+    )

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

@@ -4,7 +4,7 @@
 
 import uuid_utils.compat as uuid_utils
 
-from sqlalchemy import select
+from sqlalchemy import Integer, case, cast, func, select
 from sqlalchemy.dialects.postgresql import insert
 
 from oss.src.utils.logging import get_module_logger
@@ -78,6 +78,7 @@ async def set_session_state(
             "session_id": session_id,
             "data": data_json,
             "sandbox_id": upsert.sandbox_id,
+            "sandbox_fingerprint": upsert.sandbox_fingerprint,
             "flags": SessionStateFlags().model_dump(mode="json"),
             "created_at": now,
             "updated_at": None,
@@ -94,8 +95,33 @@ async def set_session_state(
         }
         if "data" in upsert.model_fields_set:
             update_values["data"] = stmt.excluded.data
-        if "sandbox_id" in upsert.model_fields_set:
+        guarded_pointer_write = (
+            "sandbox_id" in upsert.model_fields_set
+            and upsert.sandbox_turn_index is not None
+        )
+        if guarded_pointer_write:
+            pointer_write_allowed = (
+                func.coalesce(
+                    cast(SessionStateDBE.data["latest_turn_index"].astext, Integer),
+                    -1,
+                )
+                <= upsert.sandbox_turn_index
+            )
+            update_values["sandbox_id"] = case(
+                (pointer_write_allowed, stmt.excluded.sandbox_id),
+                else_=SessionStateDBE.sandbox_id,
+            )
+        elif "sandbox_id" in upsert.model_fields_set:
             update_values["sandbox_id"] = stmt.excluded.sandbox_id
+        if "sandbox_fingerprint" in upsert.model_fields_set:
+            update_values["sandbox_fingerprint"] = (
+                case(
+                    (pointer_write_allowed, stmt.excluded.sandbox_fingerprint),
+                    else_=SessionStateDBE.sandbox_fingerprint,
+                )
+                if guarded_pointer_write
+                else stmt.excluded.sandbox_fingerprint
+            )
 
         stmt = stmt.on_conflict_do_update(
             constraint="uq_session_states_project_session_id",

api/oss/src/dbs/postgres/sessions/states/dbes.py

@@ -51,3 +51,4 @@ class SessionStateDBE(
 
     # resume pointer: which sandbox to reconnect (null = no live sandbox)
     sandbox_id = Column(String, nullable=True)
+    sandbox_fingerprint = Column(String, nullable=True)

api/oss/src/dbs/postgres/sessions/states/mappings.py

@@ -1,19 +1,44 @@
+from typing import Optional
+
+from pydantic import ValidationError
+
+from oss.src.utils.logging import get_module_logger
 from oss.src.core.sessions.states.dtos import (
     SessionState,
     SessionStateData,
     SessionStateFlags,
 )
 from oss.src.dbs.postgres.sessions.states.dbes import SessionStateDBE
 
+log = get_module_logger(__name__)
+
+
+def _parse_state_data(dbe: SessionStateDBE) -> Optional[SessionStateData]:
+    if not dbe.data:
+        return None
+
+    try:
+        return SessionStateData.model_validate(dbe.data)
+    except ValidationError as e:
+        log.error(
+            "[SESSION_STATES] Corrupt session_states.data; degrading to None",
+            session_id=dbe.session_id,
+            project_id=dbe.project_id,
+            raw_data=dbe.data,
+            error=str(e),
+        )
+        return None
+
 
 def dbe_to_dto(dbe: SessionStateDBE) -> SessionState:
-    data = SessionStateData.model_validate(dbe.data) if dbe.data else None
+    data = _parse_state_data(dbe)
     return SessionState(
         id=dbe.id,
         project_id=dbe.project_id,
         session_id=dbe.session_id,
         data=data,
         sandbox_id=dbe.sandbox_id,
+        sandbox_fingerprint=dbe.sandbox_fingerprint,
         flags=SessionStateFlags.model_validate(dbe.flags)
         if dbe.flags
         else SessionStateFlags(),

api/oss/tests/pytest/acceptance/session_states/test_session_states_basics.py

@@ -202,3 +202,128 @@ def test_sandbox_id_accepts_runner_post(self, authed_api):
         state = response.json()["session_state"]
         assert state["session_id"] == session_id
         assert state["sandbox_id"] == "sbx-runner"
+
+    def test_guarded_pointer_write_applies_at_latest_turn(self, authed_api):
+        session_id = str(uuid.uuid4())
+        authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "data": {"latest_turn_index": 2},
+                "sandbox_id": "sbx-old",
+                "sandbox_fingerprint": "fingerprint-old",
+            },
+        )
+
+        response = authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "sandbox_id": "sbx-new",
+                "sandbox_fingerprint": "fingerprint-new",
+                "sandbox_turn_index": 2,
+            },
+        )
+
+        assert response.status_code == 200
+        state = response.json()["session_state"]
+        assert state["sandbox_id"] == "sbx-new"
+        assert state["sandbox_fingerprint"] == "fingerprint-new"
+
+    def test_stale_guarded_pointer_write_returns_unchanged_row(self, authed_api):
+        session_id = str(uuid.uuid4())
+        authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "data": {"latest_turn_index": 3},
+                "sandbox_id": "sbx-current",
+                "sandbox_fingerprint": "fingerprint-current",
+            },
+        )
+
+        response = authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "sandbox_id": "sbx-stale",
+                "sandbox_fingerprint": "fingerprint-stale",
+                "sandbox_turn_index": 2,
+            },
+        )
+
+        assert response.status_code == 200
+        state = response.json()["session_state"]
+        assert state["sandbox_id"] == "sbx-current"
+        assert state["sandbox_fingerprint"] == "fingerprint-current"
+        assert state["data"]["latest_turn_index"] == 3
+
+    def test_tokenless_pointer_write_remains_unconditional(self, authed_api):
+        session_id = str(uuid.uuid4())
+        authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "data": {"latest_turn_index": 4},
+                "sandbox_id": "sbx-old",
+                "sandbox_fingerprint": "fingerprint-old",
+            },
+        )
+
+        response = authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "sandbox_id": "sbx-tokenless",
+                "sandbox_fingerprint": "fingerprint-tokenless",
+            },
+        )
+
+        state = response.json()["session_state"]
+        assert state["sandbox_id"] == "sbx-tokenless"
+        assert state["sandbox_fingerprint"] == "fingerprint-tokenless"
+
+    def test_sandbox_fingerprint_round_trips_with_sandbox_id(self, authed_api):
+        session_id = str(uuid.uuid4())
+        authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "sandbox_id": "sbx-fingerprint",
+                "sandbox_fingerprint": "fingerprint-123",
+            },
+        )
+
+        response = authed_api(
+            "GET", "/sessions/states/", params={"session_id": session_id}
+        )
+
+        state = response.json()["session_state"]
+        assert state["sandbox_id"] == "sbx-fingerprint"
+        assert state["sandbox_fingerprint"] == "fingerprint-123"
+
+    def test_guarded_pointer_write_creates_missing_row(self, authed_api):
+        session_id = str(uuid.uuid4())
+        response = authed_api(
+            "PUT",
+            "/sessions/states/",
+            params={"session_id": session_id},
+            json={
+                "sandbox_id": "sbx-first",
+                "sandbox_fingerprint": "fingerprint-first",
+                "sandbox_turn_index": 7,
+            },
+        )
+
+        assert response.status_code == 200
+        state = response.json()["session_state"]
+        assert state["sandbox_id"] == "sbx-first"
+        assert state["sandbox_fingerprint"] == "fingerprint-first"
+        assert state.get("data") is None