Daytona label-list vs per-id endpoint divergence probe (lathe issue #62)

daytona_ghost_probe.py

# /// script
# requires-python = ">=3.11"
# dependencies = ["httpx"]
# ///
"""
Probe Daytona's label-list vs per-id endpoint divergence during teardown.

Two runs, each a fresh sandbox lifecycle:
  Run A: DELETE ?force=true
  Run B: DELETE (no force)

Each run polls both endpoints at ~400ms for 30s after DELETE, logging
(t_ms, endpoint, http_status, state). Prints an aligned timeline per run
so list-lag and per-id monotonicity are directly visible.

Reads DAYTONA_API_KEY from env. Uses a unique label per run so the list
filter is unambiguous.
"""
import asyncio
import json
import os
import sys
import time
import uuid

import httpx

API = "https://app.daytona.io/api"
POLL_INTERVAL = 0.4
POLL_BUDGET = 30.0


def headers(key: str) -> dict:
    return {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}


async def create_sandbox(client, key, label_key, email):
    name = f"{label_key}/{email}"
    resp = await client.post(
        f"{API}/sandbox",
        headers=headers(key),
        json={
            "name": name,
            "labels": {label_key: email},
            "autoStopInterval": 0,
            "autoArchiveInterval": 60,
            "autoDeleteInterval": -1,
        },
        timeout=30.0,
    )
    resp.raise_for_status()
    sb = resp.json()
    sid = sb["id"]
    # Wait for started
    for _ in range(120):
        await asyncio.sleep(1)
        r = await client.get(f"{API}/sandbox/{sid}", headers=headers(key), timeout=15.0)
        if r.status_code == 404:
            raise RuntimeError(f"sandbox {sid} went 404 during start wait")
        r.raise_for_status()
        st = r.json().get("state")
        if st == "started":
            return sid
        if st == "error":
            raise RuntimeError(f"sandbox {sid} entered error during start")
    raise RuntimeError(f"sandbox {sid} never reached started")


async def poll_pair(client, key, sid, label_key, email, t0):
    """One poll iteration: per-id then list. Returns (t_ms, per_id, list_entry)."""
    t_ms = int((time.time() - t0) * 1000)

    # per-id
    r = await client.get(f"{API}/sandbox/{sid}", headers=headers(key), timeout=15.0)
    if r.status_code == 404:
        per_id = (404, None)
    else:
        r.raise_for_status()
        per_id = (200, r.json().get("state"))

    # list
    labels_filter = json.dumps({label_key: email})
    r = await client.get(
        f"{API}/sandbox",
        params={"labels": labels_filter},
        headers=headers(key),
        timeout=15.0,
    )
    r.raise_for_status()
    payload = r.json()
    items = payload if isinstance(payload, list) else (payload.get("items") or [])
    matches = [s for s in items if s.get("labels", {}).get(label_key) == email]
    if not matches:
        list_entry = ("absent", None)
    else:
        # track the matching entry's state (should be our sid)
        m = matches[0]
        list_entry = ("present", m.get("state"))

    return t_ms, per_id, list_entry


def fmt_row(t_ms, per_id, list_entry):
    pid_status, pid_state = per_id
    list_status, list_state = list_entry
    pid_s = f"{pid_status} {pid_state or '-'}"
    list_s = f"{list_status} {list_state or '-'}"
    return f"  t={t_ms:>6}ms  per-id={pid_s:<22}  list={list_s:<22}"


async def run_probe(client, key, label_key, email, force: bool):
    tag = "force=true" if force else "no-force"
    print(f"\n=== Run [{tag}]  label={label_key} email={email} ===")
    print("  creating sandbox...")
    sid = await create_sandbox(client, key, label_key, email)
    print(f"  started: {sid}")

    # Fire DELETE
    params = {"force": "true"} if force else None
    t0 = time.time()
    r = await client.delete(
        f"{API}/sandbox/{sid}",
        headers=headers(key),
        params=params,
        timeout=30.0,
    )
    del_status = r.status_code
    print(f"  DELETE {tag} -> HTTP {del_status}  (t0 set)")
    if del_status not in (200, 202, 204):
        print(f"  DELETE body: {r.text[:300]}")

    # Poll
    rows = []
    per_id_404_at = None
    list_absent_at = None
    list_flap_log = []
    prev_list_present = True

    while time.time() - t0 < POLL_BUDGET:
        t_ms, per_id, list_entry = await poll_pair(client, key, sid, label_key, email, t0)
        rows.append((t_ms, per_id, list_entry))
        print(fmt_row(t_ms, per_id, list_entry))

        if per_id[0] == 404 and per_id_404_at is None:
            per_id_404_at = t_ms
        if list_entry[0] == "absent" and list_absent_at is None:
            list_absent_at = t_ms

        # flap detection: present -> absent -> present
        if prev_list_present and list_entry[0] == "absent":
            prev_list_present = False
        elif not prev_list_present and list_entry[0] == "present":
            list_flap_log.append(t_ms)
            prev_list_present = True

        # No early break: the list flaps absent->present, so a single
        # "absent" reading is not proof of completion. Capture the full
        # 30s timeline to see when (if ever) the list stabilizes.

        await asyncio.sleep(POLL_INTERVAL)

    # Summary
    print(f"\n  --- summary [{tag}] ---")
    print(f"  per-id first 404 at: {per_id_404_at}ms")
    print(f"  list first absent at: {list_absent_at}ms")
    if per_id_404_at is not None and list_absent_at is not None:
        lag = list_absent_at - per_id_404_at
        print(f"  list-lag after per-id 404: {lag}ms")
    else:
        print(f"  list-lag after per-id 404: (incomplete — one or both not observed in {POLL_BUDGET}s)")
    if list_flap_log:
        print(f"  LIST FLAP detected (absent->present) at: {list_flap_log}ms")
    else:
        print(f"  list flap (absent->present): none observed")
    return rows


async def main():
    key = os.environ.get("DAYTONA_API_KEY")
    if not key:
        sys.exit("DAYTONA_API_KEY not set in env")
    stamp = uuid.uuid4().hex[:8]
    async with httpx.AsyncClient() as client:
        # Run A: force=true
        await run_probe(client, key, f"lathe-probe-{stamp}-a", f"a-{stamp}@probe.local", force=True)
        # Run B: no force
        await run_probe(client, key, f"lathe-probe-{stamp}-b", f"b-{stamp}@probe.local", force=False)
    print("\nDone.")


if __name__ == "__main__":
    asyncio.run(main())