Canonical SQLite 深度解析

50APPLICATION_ID = 0x41434354  # "ACCT"
51# Version 1 was the spike's unstable DDL (it drifted while the number stood
52# still); version 2 is the first checksum-pinned schema; version 3 admits the
53# 'live' store role for the writer-integration phase; version 4 adds the
54# session_observed_lineage absorb-state table. v1 candidates are rebuilt
55# from a fresh snapshot, never migrated; v2+ stores migrate forward.
56SCHEMA_VERSION = 4
57SCHEMA_BASE_NAME = "canonical-v4"
58APP_VERSION_DEFAULT = "sqlite-truth-spike"
59MINIMUM_SQLITE_VERSION = (3, 37, 0)
60TASK_PUBLIC_ID_PREFIX = "task_"
61LIVE_STORE_FILENAME = "chronicle.sqlite3"
62_MAX_SQLITE_INT = 9_223_372_036_854_775_807
63# Bare directory names, not <name>/state: live state also lives beside state/
64# (policy.yaml) and the -global root itself serves as a hooks project-dir.
65_FORBIDDEN_STATE_SEQUENCES = (
66    (".agent-sentinel",),
67    (".agent-sentinel-global",),
68    (".agent-chronicle",),
69    (".agent-chronicle-global",),
70)
71
72_GET_TASK_SQL = "SELECT * FROM rm_task_current WHERE public_task_id = ?"
73_LIST_TASKS_SQL = (
74    "SELECT * FROM rm_task_current INDEXED BY idx_rm_task_work_recent "
75    "WHERE (sort_activity_at_us, public_task_id) < (?, ?) "
76    "ORDER BY sort_activity_at_us DESC, public_task_id DESC LIMIT ?"
77)
78_LIST_SESSIONS_SQL = (
79    "SELECT * FROM sessions INDEXED BY idx_sessions_recent "
80    "WHERE (sort_activity_at_us, session_id) < (?, ?) "
81    "ORDER BY sort_activity_at_us DESC, session_id DESC LIMIT ?"
82)
83# Batched point lookups for surface labeling (phase 4.2): each resolves by
84# primary key / a pinned index over an id list bounded by the page the caller
85# just fetched. The placeholder count varies with the batch; the shape does
86# not.
87_SESSIONS_BY_IDS_SQL = "SELECT * FROM sessions WHERE session_id IN ({placeholders})"
88_SOURCES_BY_IDS_SQL = (
89    "SELECT * FROM source_instances WHERE source_instance_id IN ({placeholders})"
90)
91# At most one row per child: uq_session_edges_one_valid_parent is
92# relation-agnostic (one valid edge per child, whatever its relation), and
93# writers mint the relation from the child's session kind — 'continuation'
94# and 'retry' children carry their single validated lineage edge under those
95# relations, so filtering to relation='parent' would silently drop them.
96_VALID_LINEAGE_EDGES_SQL = (
97    "SELECT child_session_id, parent_session_id, relation "
98    "FROM session_edges INDEXED BY idx_session_edges_child "
99    "WHERE child_session_id IN ({placeholders}) "
100    "AND validation_state = 'valid'"
101)
102# Bound for every by-ids batch: one page of the widest list surface.
103_BY_IDS_BATCH_LIMIT = 2_000
104_HEALTH_STORE_SQL = (
105    "SELECT store_uuid, store_role, schema_version, canonical_sequence "
106    "FROM store_metadata WHERE singleton = 1"
107)
108# The single authority for which rm_* read models the minimal rebuild
109# maintains. The rebuild policy (canonical_live) and the health probe
110# (canonical_read) both derive from this tuple — a new read model added to
111# rebuild_minimal_read_models without updating this constant would silently
112# escape staleness detection and never_built synthesis.
113MINIMAL_READ_MODEL_NAMES = ("rm_task_current", "rm_usage_day")

132@dataclass(frozen=True)
133class SchemaMigration:
134    """One forward schema step from ``from_version`` to ``from_version + 1``.
135
136    Statements run one by one inside a single transaction (never via
137    ``executescript``, which would commit early) and must produce exactly the
138    DDL that compiling the target build's packaged ``schema.sql`` produces —
139    the open-time inventory comparison enforces that equivalence.
140    """
141
142    from_version: int
143    name: str
144    statements: tuple[str, ...]
145
146    @property
147    def checksum(self) -> str:
148        digest = hashlib.sha256(b"agent-chronicle-schema-migration-v1\x00")
149        for statement in self.statements:
150            payload = statement.encode("utf-8")
151            digest.update(len(payload).to_bytes(8, "big"))
152            digest.update(payload)
153        return digest.hexdigest()
154
155
156# Forward steps keyed by from_version. Version 1 was the spike's unstable
157# DDL, so v1 candidates rebuild rather than migrate. Statements are frozen
158# literals: the registered checksum is validated against every previously
159# migrated store on open, so a step may never be regenerated from the moving
160# schema.sql — the inventory comparison is what proves the literal still
161# produces the packaged DDL.
162SCHEMA_MIGRATIONS: dict[int, SchemaMigration] = {
163    2: SchemaMigration(
164        from_version=2,
165        name="allow-live-store-role",
166        statements=(
167            # SQLite cannot alter a CHECK constraint in place; rebuild the
168            # singleton table around the copied row. The final CREATE must
169            # stay byte-identical to the store_metadata block in schema.sql.
170            "CREATE TABLE store_metadata_role_migration AS "
171            "SELECT singleton, store_uuid, store_role, schema_version, "
172            "canonical_sequence, created_at_us FROM store_metadata",
173            "DROP TABLE store_metadata",
174            (
175                "CREATE TABLE store_metadata (\n"
176                "    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),\n"
177                "    store_uuid TEXT NOT NULL UNIQUE,\n"
178                "    store_role TEXT NOT NULL CHECK (store_role IN ('candidate', 'live')),\n"
179                "    schema_version INTEGER NOT NULL CHECK (schema_version >= 1),\n"
180                "    canonical_sequence INTEGER NOT NULL DEFAULT 0 CHECK (canonical_sequence >= 0),\n"
181                "    created_at_us INTEGER NOT NULL CHECK (created_at_us >= 0)\n"
182                ") STRICT"
183            ),
184            "INSERT INTO store_metadata(singleton, store_uuid, store_role, "
185            "schema_version, canonical_sequence, created_at_us) "
186            "SELECT singleton, store_uuid, store_role, schema_version, "
187            "canonical_sequence, created_at_us FROM store_metadata_role_migration",
188            "DROP TABLE store_metadata_role_migration",
189        ),
190    ),

228def _apply_schema_migrations(
229    connection: sqlite3.Connection,
230    *,
231    from_version: int,
232    app_version: str,
233) -> int:
234    journal_mode = str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower()
235    if journal_mode != "wal":
236        raise ValueError("candidate database is not in WAL mode; refusing to migrate")
237    metadata_version = connection.execute(
238        "SELECT schema_version FROM store_metadata WHERE singleton = 1"
239    ).fetchone()
240    if metadata_version is None or int(metadata_version[0]) < from_version:
241        raise ValueError("candidate metadata disagrees with the schema version; refusing to migrate")
242    current = from_version
243    while current < SCHEMA_VERSION:
244        step = SCHEMA_MIGRATIONS.get(current)
245        if step is None:
246            raise ValueError(
247                f"candidate schema version {current} has no upgrade path in "
248                f"this build; rebuild the candidate from a fresh snapshot"
249            )
250        if step.from_version != current:
251            raise ValueError(
252                f"schema migration '{step.name}' is registered for version "
253                f"{current} but declares from_version {step.from_version}"
254            )
255        target = current + 1
256        try:
257            connection.execute("BEGIN IMMEDIATE")
258        except sqlite3.OperationalError as exc:
259            raise ValueError(
260                f"schema migration '{step.name}' to version {target} could not "
261                f"acquire the write lock; another process may be migrating"
262            ) from exc
263        try:
264            # Re-read versions under the exclusive lock: a concurrent writable
265            # open may have finished this step (or the whole chain) while we
266            # waited. Adopting its committed progress keeps racing opens
267            # idempotent instead of failing on the stamping UNIQUE constraint.
268            observed = int(connection.execute("PRAGMA user_version").fetchone()[0])
269            row = connection.execute(
270                "SELECT schema_version FROM store_metadata WHERE singleton = 1"
271            ).fetchone()
272            stamped = None if row is None else int(row[0])
273            if observed != current or stamped != current:
274                if stamped == observed and current < observed <= SCHEMA_VERSION:
275                    connection.execute("ROLLBACK")
276                    current = observed
277                    continue
278                raise ValueError(
279                    "candidate metadata disagrees with the schema version; refusing to migrate"
280                )
281            for statement in step.statements:
282                connection.execute(statement)
283                if not connection.in_transaction:
284                    # A statement that is itself COMMIT/ROLLBACK/END would
285                    # make the stamping writes below autocommit one by one,
286                    # silently breaking per-step atomicity.
287                    raise ValueError(
288                        f"schema migration '{step.name}' statement ended the "
289                        f"transaction; step atomicity cannot be proved"
290                    )
291            connection.execute(
292                "INSERT INTO schema_migrations(version, name, checksum, app_version, applied_at_us) "
293                "VALUES (?, ?, ?, ?, ?)",
294                (target, step.name, step.checksum, app_version, now_us()),
295            )
296            connection.execute(
297                "UPDATE store_metadata SET schema_version = ? WHERE singleton = 1",
298                (target,),
299            )
300            connection.execute(f"PRAGMA user_version = {target}")
301            connection.execute("COMMIT")
302        except BaseException as exc:
303            if connection.in_transaction:
304                try:
305                    connection.execute("ROLLBACK")
306                except sqlite3.Error:
307                    pass
308            if isinstance(exc, sqlite3.Error):
309                # A distinct error class keeps migration failures out of the
310                # generic unreadable-DDL diagnosis in open().
311                raise ValueError(
312                    f"schema migration '{step.name}' to version {target} failed"
313                ) from exc
314            raise
315        current = target
316    return current