perf: correct hosted lifecycle tail

experiments/physical-architecture/candidate_a/maintenance.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import sqlite3
 import time
@@ -425,17 +426,29 @@ def _insert_calls(
     else:
         event_at = int(tail_state["maximum_event_at_us"]) + 1
     source_order = int(tail_state["maximum_source_order"]) + 1
+    change_name = "late" if late else "tail"
+    session_json = json.dumps(
+        session_id,
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
+    identity_prefix = (
+        f'{{"candidate":"A","change":"{change_name}","event_at_us":'.encode()
+    )
+    identity_middle = b',"ordinal":'
+    identity_suffix = f',"session":{session_json}}}\n'.encode()
     rows = []
     for ordinal in range(count):
-        digest = shared.canonical_sha256(
-            {
-                "candidate": "A",
-                "change": "late" if late else "tail",
-                "session": session_id,
-                "ordinal": ordinal,
-                "event_at_us": event_at + ordinal,
-            }
+        # Preserve the exact sorted-key canonical JSON identity without paying
+        # for a full JSON encoder invocation for every row in the timed tail.
+        identity = (
+            identity_prefix
+            + str(event_at + ordinal).encode()
+            + identity_middle
+            + str(ordinal).encode()
+            + identity_suffix
         )
+        digest = hashlib.sha256(identity).hexdigest()
         rows.append(
             (
                 f"call:candidate-a:{digest}",
@@ -458,19 +471,29 @@ def _insert_calls(
                 int(source["byte_count"]),
             )
         )
+    width = len(rows[0])
+    sqlite_variable_limit = 999
+    getlimit = getattr(connection, "getlimit", None)
+    if callable(getlimit):
+        runtime_limit = getlimit(9)  # SQLITE_LIMIT_VARIABLE_NUMBER
+        if isinstance(runtime_limit, int):
+            sqlite_variable_limit = runtime_limit
+    per_statement = max(1, min(sqlite_variable_limit, 30_000) // width)
     row_placeholders = "(" + ", ".join("?" for _ in rows[0]) + ")"
-    connection.execute(
-        f"""
-        INSERT INTO model_call_tail(
-            call_id, session_id, turn_id, model, reasoning_effort,
-            context_window_tokens, uncached_input_tokens, cached_input_tokens,
-            reasoning_tokens, output_tokens, event_at_us, source_rank,
-            occurrence_source_key, source_order, event_kind_order,
-            record_ordinal, byte_start, byte_end
-        ) VALUES {", ".join(row_placeholders for _ in rows)}
-        """,
-        tuple(value for row in rows for value in row),
-    )
+    for start in range(0, len(rows), per_statement):
+        batch = rows[start : start + per_statement]
+        connection.execute(
+            f"""
+            INSERT INTO model_call_tail(
+                call_id, session_id, turn_id, model, reasoning_effort,
+                context_window_tokens, uncached_input_tokens, cached_input_tokens,
+                reasoning_tokens, output_tokens, event_at_us, source_rank,
+                occurrence_source_key, source_order, event_kind_order,
+                record_ordinal, byte_start, byte_end
+            ) VALUES {", ".join(row_placeholders for _ in batch)}
+            """,
+            tuple(value for row in batch for value in row),
+        )
     return (
         len(rows),
         _UsageDelta(

tests/experiments/physical-architecture/candidate_a/test_candidate_a_tail_hardening.py

@@ -166,37 +166,51 @@ def test_pending_tool_lookup_stays_indexed_with_closed_history(
     # The contract is that closed history is never scanned.
 
 
-def test_bulk_call_tail_uses_one_set_sized_bounded_tail_insert(
+def test_bulk_call_tail_uses_bounded_set_inserts(
     fixture: Any,
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     artifact = candidate_a.build_artifact(fixture, tmp_path / "bulk-tail.sqlite")
-    statements: list[str] = []
     original_open_database = maintenance_module.open_database
 
-    def traced_open_database(*args: Any, **kwargs: Any) -> Any:
+    def bounded_open_database(*args: Any, **kwargs: Any) -> Any:
         connection = original_open_database(*args, **kwargs)
+        setlimit = getattr(connection, "setlimit", None)
+        if callable(setlimit):
+            setlimit(9, 30_000)  # SQLITE_LIMIT_VARIABLE_NUMBER
+        return connection
 
-        def capture_bulk_insert(statement: str) -> None:
-            if not statements and "INSERT INTO MODEL_CALL_TAIL" in " ".join(
-                statement.upper().split()
-            ):
-                statements.append(statement)
-                connection.set_trace_callback(None)
+    monkeypatch.setattr(maintenance_module, "open_database", bounded_open_database)
+    stats = apply_ordinary_change(artifact.path, "2000_call_tail")
+
+    assert stats.facts_inserted == 2_000
+    assert stats.dirty_keys == 6
+
+
+def test_bulk_call_tail_respects_runtime_sql_variable_limit(
+    fixture: Any,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    artifact = candidate_a.build_artifact(fixture, tmp_path / "bounded-variables.sqlite")
+    original_open_database = maintenance_module.open_database
 
-        connection.set_trace_callback(capture_bulk_insert)
+    def constrained_open_database(*args: Any, **kwargs: Any) -> Any:
+        connection = original_open_database(*args, **kwargs)
+        setlimit = getattr(connection, "setlimit", None)
+        if callable(setlimit):
+            setlimit(9, 999)  # SQLITE_LIMIT_VARIABLE_NUMBER
         return connection
 
-    monkeypatch.setattr(maintenance_module, "open_database", traced_open_database)
+    monkeypatch.setattr(
+        maintenance_module,
+        "open_database",
+        constrained_open_database,
+    )
+
     stats = apply_ordinary_change(artifact.path, "2000_call_tail")
-    model_call_inserts = [
-        statement
-        for statement in statements
-        if "INSERT INTO MODEL_CALL_TAIL" in " ".join(statement.upper().split())
-    ]
 
-    assert len(model_call_inserts) == 1
     assert stats.facts_inserted == 2_000
     assert stats.dirty_keys == 6
 
@@ -228,6 +242,28 @@ def test_model_call_tail_is_append_only_and_cross_table_unique(
             ).fetchone()[0]
             == 2_000
         )
+        for ordinal in (0, 1_999):
+            row = connection.execute(
+                """
+                SELECT call_id, session_id, event_at_us
+                FROM model_call_tail
+                ORDER BY event_at_us, source_order
+                LIMIT 1 OFFSET ?
+                """,
+                (ordinal,),
+            ).fetchone()
+            assert row["call_id"] == (
+                "call:candidate-a:"
+                + shared.canonical_sha256(
+                    {
+                        "candidate": "A",
+                        "change": "tail",
+                        "session": row["session_id"],
+                        "ordinal": ordinal,
+                        "event_at_us": row["event_at_us"],
+                    }
+                )
+            )
         with pytest.raises(sqlite3.IntegrityError, match="cross-table"):
             connection.execute("INSERT INTO model_call_tail SELECT * FROM model_calls LIMIT 1")
         with pytest.raises(sqlite3.IntegrityError, match="append-only"):