diff --git a/experiments/physical-architecture/candidate_a/maintenance.py b/experiments/physical-architecture/candidate_a/maintenance.py index aa17ea90..281954df 100644 --- a/experiments/physical-architecture/candidate_a/maintenance.py +++ b/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( diff --git a/tests/experiments/physical-architecture/candidate_a/test_candidate_a_tail_hardening.py b/tests/experiments/physical-architecture/candidate_a/test_candidate_a_tail_hardening.py index 0463a128..8a0b5690 100644 --- a/tests/experiments/physical-architecture/candidate_a/test_candidate_a_tail_hardening.py +++ b/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"):