openenv/tbench2: orphan TTL + ownership labels for per-task sandboxes

examples/experimental/openenv/tb2_sandbox_daytona.py

@@ -11,15 +11,22 @@
       the declarative path avoids the quota entirely, and repeat creates hit
       Daytona's build cache (~1min after the first build). Daytona does not
       run the image CMD, so this execs ``server_cmd()`` and waits for /health.
+      Sandboxes carry ownership labels (task / launcher / run id) and an
+      auto-stop+auto-delete TTL armed as a dead-man's switch: a keepalive
+      thread beats the activity timer while the creating process lives, so
+      a hard-killed caller's orphans are reclaimed instead of billing forever.
   bake CLI (``python tb2_sandbox_daytona.py ...``)  optionally pre-register
       named snapshots ``<prefix><task-id>`` as a warm cache.
 """
 
 import argparse
+import getpass
 import os
 import re
 import shlex
 import sys
+import threading
+import time
 from pathlib import Path
 
 from tb2_sandbox_recipe import (
@@ -61,29 +68,96 @@ def task_resources(task_dir: Path):
     )
 
 
+def sandbox_labels(task_dir: Path) -> dict[str, str]:
+    """Labels for a per-task sandbox: what it runs, and who launched it.
+
+    The Daytona API records no creator, so in a shared org labels are the only
+    attribution. ``openenv-tbench2-task`` keys sweep/cleanup tooling to exactly
+    the sandboxes this recipe created (shared orgs run other workloads).
+    ``openenv-launcher`` is OPENENV_LAUNCHER when set — do set it on shared
+    hosts, where the unix user is a generic account — else the local unix
+    user. ``openenv-run-id`` (OPENENV_RUN_ID, optional) additionally groups
+    one run's sandboxes for targeted sweeps.
+    """
+    try:
+        user = getpass.getuser()
+    except Exception:  # no passwd entry / login env on minimal hosts
+        user = "unknown"
+    labels = {
+        "openenv-tbench2-task": task_dir.name,
+        "openenv-launcher": os.environ.get("OPENENV_LAUNCHER") or user,
+    }
+    run_id = os.environ.get("OPENENV_RUN_ID")
+    if run_id:
+        labels["openenv-run-id"] = run_id
+    return labels
+
+
+# Keepalive cadence: 6 beats per 30-minute auto-stop window, and up to 3
+# consecutive failed beats (15 minutes of API blips) tolerated before the
+# thread concludes the sandbox is gone and exits.
+_KEEPALIVE_INTERVAL_S = 300.0
+_KEEPALIVE_MAX_CONSECUTIVE_FAILURES = 3
+
+
+def _start_keepalive(sandbox, task_id: str) -> None:
+    """Refresh the sandbox's activity timer for as long as THIS process lives.
+
+    Daytona's auto-stop clock counts only SDK interactions — preview-proxy
+    traffic, which is ALL of an episode's I/O, does not reset it — so without
+    a heartbeat any healthy episode longer than the auto-stop interval would
+    be stopped mid-run. A daemon thread has exactly the right lifetime: it
+    dies with the process, which is what turns auto-stop into a dead-man's
+    switch for orphans. The thread exits once refreshes fail persistently
+    (the normal case: the episode ended and the caller deleted the sandbox).
+    """
+
+    def _beat() -> None:
+        failures = 0
+        while failures < _KEEPALIVE_MAX_CONSECUTIVE_FAILURES:
+            time.sleep(_KEEPALIVE_INTERVAL_S)
+            try:
+                sandbox.refresh_activity()
+                failures = 0
+            except Exception:
+                failures += 1
+
+    threading.Thread(target=_beat, name=f"tb2-sandbox-keepalive-{task_id}", daemon=True).start()
+
+
 def create_task_sandbox(
     daytona,
     task_dir: Path,
     *,
     command_timeout_s: int = 900,
     create_timeout_s: float = 1800.0,
     ready_timeout_s: float = 300.0,
+    auto_stop_minutes: int = 30,
+    auto_delete_minutes: int = 120,
 ):
     """Create ONE per-episode sandbox for *task_dir*, declaratively (no named snapshot).
 
     Returns ``(sandbox, base_url)``. Caller must ``daytona.delete(sandbox)``
     when the episode ends. First create for a task pays the image build;
     repeat creates hit Daytona's build cache.
+
+    Orphan TTL: a caller that dies without reaching its delete (SIGKILL, OOM,
+    node loss) leaks the sandbox, and Daytona's defaults would keep it running
+    — and billing — forever. Auto-stop/auto-delete arm a backstop, and a
+    keepalive thread (see ``_start_keepalive``) beats the activity timer for
+    as long as the creating process lives: a live episode of any length is
+    safe, while a dead caller stops beating and Daytona stops the sandbox
+    within *auto_stop_minutes* of the last beat, then deletes the stopped
+    remains after *auto_delete_minutes* more.
     """
     from daytona import CreateSandboxFromImageParams
 
     params = CreateSandboxFromImageParams(
         image=build_task_image(task_dir),
         resources=task_resources(task_dir),
-        auto_stop_interval=0,
-        # Ownership marker: lets sweep/cleanup tooling target exactly the
-        # sandboxes this recipe created (shared orgs run other workloads).
-        labels={"openenv-tbench2-task": task_dir.name},
+        auto_stop_interval=auto_stop_minutes,
+        auto_delete_interval=auto_delete_minutes,
+        labels=sandbox_labels(task_dir),
     )
     sandbox = daytona.create(params, timeout=create_timeout_s)
     try:
@@ -94,6 +168,7 @@ def create_task_sandbox(
         )
         url = sandbox.create_signed_preview_url(8000, expires_in_seconds=86400).url
         wait_server_ready(url, timeout_s=ready_timeout_s)
+        _start_keepalive(sandbox, task_dir.name)
         return sandbox, url
     except Exception:
         daytona.delete(sandbox)

examples/experimental/openenv/tests/test_tb2_sandbox_daytona.py

@@ -0,0 +1,71 @@
+"""Tests for the Daytona sandbox half: ownership labels and the orphan-TTL
+keepalive lifecycle.
+
+Not collected by the repo-level pytest run (testpaths = ./tests); run manually
+when touching the recipe:
+
+    pytest examples/experimental/openenv/tests/ -q
+"""
+
+import inspect
+import sys
+import threading
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import tb2_sandbox_daytona as sandbox  # noqa: E402
+
+
+def test_sandbox_labels_default_to_unix_user(monkeypatch):
+    monkeypatch.delenv("OPENENV_LAUNCHER", raising=False)
+    monkeypatch.delenv("OPENENV_RUN_ID", raising=False)
+    labels = sandbox.sandbox_labels(Path("/opt/tb2-tasks/regex-chess"))
+    assert labels["openenv-tbench2-task"] == "regex-chess"
+    assert labels["openenv-launcher"]  # some non-empty identity, never absent
+    assert "openenv-run-id" not in labels  # omitted when unset, not ""
+
+
+def test_sandbox_labels_explicit_launcher_and_run_id(monkeypatch):
+    monkeypatch.setenv("OPENENV_LAUNCHER", "tao-lin")
+    monkeypatch.setenv("OPENENV_RUN_ID", "tb2-grpo-0717")
+    labels = sandbox.sandbox_labels(Path("/opt/tb2-tasks/regex-chess"))
+    assert labels["openenv-launcher"] == "tao-lin"
+    assert labels["openenv-run-id"] == "tb2-grpo-0717"
+
+
+def test_create_arms_ttl_by_default():
+    # The dead-man's-switch contract: creates must arm auto-stop/auto-delete,
+    # or a hard-killed caller's orphans run (and bill) forever.
+    sig = inspect.signature(sandbox.create_task_sandbox)
+    assert sig.parameters["auto_stop_minutes"].default > 0
+    assert sig.parameters["auto_delete_minutes"].default > 0
+
+
+def test_keepalive_beats_then_exits_on_persistent_failure(monkeypatch):
+    monkeypatch.setattr(sandbox, "_KEEPALIVE_INTERVAL_S", 0.02)
+
+    class Stub:
+        def __init__(self):
+            self.beats = 0
+            self.dead = False
+
+        def refresh_activity(self):
+            if self.dead:
+                raise RuntimeError("sandbox deleted")
+            self.beats += 1
+
+    stub = Stub()
+    sandbox._start_keepalive(stub, "regex-chess")
+    deadline = time.time() + 2.0
+    while stub.beats < 3 and time.time() < deadline:
+        time.sleep(0.01)
+    assert stub.beats >= 3  # beats while the sandbox is alive
+
+    stub.dead = True  # episode over, sandbox deleted -> thread must exit
+    deadline = time.time() + 2.0
+    while time.time() < deadline:
+        if not any("keepalive" in t.name for t in threading.enumerate()):
+            break
+        time.sleep(0.01)
+    assert not any("keepalive" in t.name for t in threading.enumerate())