perf(history): serve long threads from stored turn slices and keep agent state to the window

migrations/versions/064_turn_slices.py

@@ -0,0 +1,73 @@
+"""Migration 064: per-turn and per-subagent-run slices of checkpoint history.
+
+A slice is stored once when a turn or run settles, so replay reads rows
+instead of materializing checkpoint state. It is derived data: deleting rows
+loses nothing, the next read rebuilds them. A slice is valid while
+``slice_key`` names the code that cut it; ``lines`` is the projected replay
+of the turn, wire-ready, valid while ``lines_key`` matches. ``run_claims``
+pins each subagent launch without a ledger stamp to the run it started, which
+only a pass over the whole thread can work out.
+
+Rollback: ``downgrade`` drops both tables; the previous build never reads them.
+"""
+
+from alembic import op
+
+revision = "064"
+down_revision = "063"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Each foreign key takes a lock on the table it references, which live
+    # turns write; fail fast rather than queue every writer behind it.
+    op.execute("SET LOCAL lock_timeout = '5s'")
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS turn_slices (
+            conversation_response_id UUID PRIMARY KEY
+                REFERENCES conversation_responses(conversation_response_id)
+                ON DELETE CASCADE,
+            conversation_thread_id UUID NOT NULL,
+            input_checkpoint_id TEXT NOT NULL,
+            tail_checkpoint_id TEXT NOT NULL,
+            slice_key TEXT NOT NULL,
+            slice_codec TEXT NOT NULL,
+            slice BYTEA NOT NULL,
+            lines_key TEXT,
+            lines TEXT,
+            run_claims JSONB,
+            built_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+    """)
+    # Every read by thread filters on the code that cut the slice too.
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS idx_turn_slices_thread "
+        "ON turn_slices (conversation_thread_id, slice_key)"
+    )
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS task_run_slices (
+            conversation_thread_id UUID NOT NULL
+                REFERENCES conversation_threads(conversation_thread_id)
+                ON DELETE CASCADE,
+            task_id TEXT NOT NULL,
+            input_checkpoint_id TEXT NOT NULL,
+            task_run_id UUID REFERENCES subagent_runs(task_run_id)
+                ON DELETE CASCADE,
+            tail_checkpoint_id TEXT NOT NULL,
+            slice_key TEXT NOT NULL,
+            slice_codec TEXT NOT NULL,
+            slice BYTEA NOT NULL,
+            built_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            PRIMARY KEY (conversation_thread_id, task_id, input_checkpoint_id)
+        )
+    """)
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS idx_task_run_slices_run "
+        "ON task_run_slices (task_run_id) WHERE task_run_id IS NOT NULL"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS task_run_slices")
+    op.execute("DROP TABLE IF EXISTS turn_slices")

migrations/versions/065_replay_facts.py

@@ -0,0 +1,39 @@
+"""Migration 065: replay facts on settled runs.
+
+Some of what a turn's replay shows was only ever in the stream it emitted:
+how long each reasoning block took, which sandbox images were captured to
+which URLs, and the steering a subagent run handed back unread. ``replay_facts``
+keeps those few values on the run row itself, so replay reads them next to the
+checkpoint instead of from the stored event log. NULL means a run settled
+before the column existed, and replay keeps reading the stored events for it.
+
+Rollback: ``downgrade`` drops both columns; the previous build never reads them.
+"""
+
+from alembic import op
+
+revision = "065"
+down_revision = "064"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Each ALTER waits for an ACCESS EXCLUSIVE lock on a table live turns
+    # write; fail fast rather than queue every writer behind it.
+    op.execute("SET LOCAL lock_timeout = '5s'")
+    op.execute(
+        "ALTER TABLE conversation_responses "
+        "ADD COLUMN IF NOT EXISTS replay_facts JSONB"
+    )
+    op.execute(
+        "ALTER TABLE subagent_runs ADD COLUMN IF NOT EXISTS replay_facts JSONB"
+    )
+
+
+def downgrade() -> None:
+    op.execute("SET LOCAL lock_timeout = '5s'")
+    op.execute("ALTER TABLE subagent_runs DROP COLUMN IF EXISTS replay_facts")
+    op.execute(
+        "ALTER TABLE conversation_responses DROP COLUMN IF EXISTS replay_facts"
+    )

scripts/_errors.py

@@ -0,0 +1,14 @@
+"""How the operator scripts under ``scripts/`` name a failure in their output."""
+
+from __future__ import annotations
+
+import traceback
+from pathlib import Path
+
+
+def where(exc: BaseException) -> str:
+    """An exception's type and innermost frame. Its message can quote message
+    content (a validation error echoes its input), so it is not printed."""
+    frames = traceback.extract_tb(exc.__traceback__)
+    at = f" at {Path(frames[-1].filename).name}:{frames[-1].lineno}" if frames else ""
+    return f"{type(exc).__name__}{at}"

scripts/ops/_db.py

@@ -10,6 +10,8 @@
 import os
 from urllib.parse import quote_plus
 
+from psycopg.conninfo import conninfo_to_dict
+
 
 def build_db_uri(prefix: str = "DB_") -> str:
     """Assemble ``<prefix>HOST``/``PORT``/``NAME``/``USER``/``PASSWORD`` into a URI.
@@ -28,3 +30,20 @@ def build_db_uri(prefix: str = "DB_") -> str:
         f"postgresql://{quote_plus(user)}:{quote_plus(password)}"
         f"@{host}:{port}/{name}?sslmode={sslmode}"
     )
+
+
+def point_app_pool_at(conninfo: str) -> None:
+    """Make the app's pool, which builds its target from DB_*, connect where
+    the script writes. Must run before the app is imported, whose dotenv
+    load fills only what is unset."""
+    params = conninfo_to_dict(conninfo)
+    for key, param in (
+        ("DB_HOST", "host"),
+        ("DB_PORT", "port"),
+        ("DB_NAME", "dbname"),
+        ("DB_USER", "user"),
+        ("DB_PASSWORD", "password"),
+        ("DB_SSLMODE", "sslmode"),
+    ):
+        if params.get(param):
+            os.environ[key] = str(params[param])