From cb6db3eeb897de6915c7ec75336436504ae7c3a8 Mon Sep 17 00:00:00 2001 From: guangyu-reflexio Date: Wed, 2 Sep 2026 22:55:39 -0700 Subject: [PATCH] fix(storage): resolve the SQLite file from the dataset identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two datasets sharing one SQLite file could read each other's rows. `SQLiteStorage.__init__` resolved `db_path` from LOCAL_STORAGE_PATH and never from the caller-supplied identity, so one file served every caller. Of the persistent tables only 11 carry an `org_id` column; the other 32 — profiles, requests, interactions, user_playbooks among them — have none and their reads are unscoped. Reproduced before the fix: two instances, different identities, one db_path, and tenant-b read tenant-a's profile content. After: 0 rows. The sharpest case is not two local plugins. A self-host deployment never passes base_dir, so every org it serves resolved to the same file. `_dataset_path.resolve_sqlite_db_path` now derives `reflexio_.db` and adopts an existing database rather than starting empty beside it. Adoption is first-claimer-wins, not "adopt whatever is there" — adopting on every open would let a second identity attach to the same file, leaving an already-commingled install commingled forever, and those are the only installs with the bug. The claim is a row written under BEGIN IMMEDIATE. Not a PRAGMA (application_id/user_version are 32-bit and cannot hold an identity) and not a sidecar file (it can desync from the database it describes). BEGIN IMMEDIATE takes a cross-process lock, which is required rather than defensive: the service runs multiple uvicorn workers by default and the initialization lock in _base is a threading.Lock. Identities are rejected, never rewritten — rewriting could map two identities onto one file, which is the defect being fixed. The claim also catches what validation cannot: on a case-insensitive filesystem `Acme` and `acme` derive one filename, and the second open now fails closed. Also: - The configurator's base_dir branch had the same defect and now uses the same resolver, so two orgs under one base_dir stop sharing a file. - The SQLite version guard moved above path resolution, so an unsupported SQLite fails without leaving a directory or a claim behind. - reset_db.py took --org and derives from it; it previously computed the legacy path while recreating under a hardcoded org, which after this change would rebuild a database nobody reads. Sidecar suffixes aligned to include -journal. An explicit db_path is still used verbatim: multi-tenant tests and benchmarks point several identities at one file deliberately, and that is also what keeps the residual commingling case observable. Tests: 1379 pass across storage and configurator; 30 new/updated covering separation, adoption, first-claimer-wins, idempotence, identity validation, the case-fold collision, the pre-column label scan, and a cross-process claim race. --- .../services/configurator/configurator.py | 15 +- .../services/storage/sqlite_storage/_base.py | 27 +- .../storage/sqlite_storage/_dataset_path.py | 255 ++++++++++++++++++ scripts/reset_db.py | 30 ++- .../sqlite_storage/test_dataset_path.py | 142 ++++++++++ .../services/storage/test_storage_defaults.py | 135 +++++++++- 6 files changed, 581 insertions(+), 23 deletions(-) create mode 100644 reflexio/server/services/storage/sqlite_storage/_dataset_path.py create mode 100644 tests/server/services/storage/sqlite_storage/test_dataset_path.py diff --git a/reflexio/server/services/configurator/configurator.py b/reflexio/server/services/configurator/configurator.py index ee210ff4e..a9aa64e96 100644 --- a/reflexio/server/services/configurator/configurator.py +++ b/reflexio/server/services/configurator/configurator.py @@ -2,7 +2,6 @@ import logging from collections.abc import Callable -from pathlib import Path from typing import Any from reflexio.models.config_schema import ( @@ -15,6 +14,9 @@ LocalFileConfigStorage, ) from reflexio.server.services.storage.sqlite_storage import SQLiteStorage +from reflexio.server.services.storage.sqlite_storage._dataset_path import ( + resolve_sqlite_db_path, +) from reflexio.server.services.storage.storage_base import BaseStorage logger = logging.getLogger(__name__) @@ -36,11 +38,16 @@ def _create_sqlite_storage( full_config.enable_document_expansion if full_config else False ) # When base_dir is explicitly provided (e.g. tests with temp dirs) - # and no db_path is configured, use base_dir for the SQLite DB - # so the storage is isolated from the shared default database. + # and no db_path is configured, resolve the SQLite DB under base_dir so the + # storage is isolated from the shared default database. + # + # Resolved by identity, not by base_dir alone: two orgs sharing one base_dir + # would otherwise land on one file and read each other's rows -- the same defect + # the default path had. Mirrors the config layer beside it, which already writes + # ``config_.json``. db_path = config.db_path if db_path is None and configurator.base_dir: - db_path = str(Path(configurator.base_dir) / "reflexio.db") + db_path = resolve_sqlite_db_path(configurator.base_dir, configurator.org_id) return SQLiteStorage( org_id=configurator.org_id, db_path=db_path, diff --git a/reflexio/server/services/storage/sqlite_storage/_base.py b/reflexio/server/services/storage/sqlite_storage/_base.py index 9acb53099..1be00bdbd 100644 --- a/reflexio/server/services/storage/sqlite_storage/_base.py +++ b/reflexio/server/services/storage/sqlite_storage/_base.py @@ -71,6 +71,7 @@ from reflexio.server.services.storage.storage_base import BaseStorage from reflexio.server.site_var.site_var_manager import SiteVarManager +from ._dataset_path import resolve_sqlite_db_path from ._governance import init_governance_tables from ._stall_state import init_stall_state_table @@ -963,11 +964,27 @@ def __init__( self.api_key_config = api_key_config self._enable_document_expansion = enable_document_expansion - # Resolve db_path: explicit arg > LOCAL_STORAGE_PATH env var > ~/.reflexio/data/ + # Checked before any filesystem work: resolving a derived path claims the + # dataset, and an unsupported SQLite should fail without leaving a directory + # or a claim behind. + if sqlite3.sqlite_version_info < _MINIMUM_SQLITE_VERSION: + detected_version = ".".join(map(str, sqlite3.sqlite_version_info)) + raise RuntimeError( + f"SQLite 3.35.0 or newer is required; detected {detected_version}" + ) + + # Resolve db_path: explicit arg > a file derived from org_id under + # LOCAL_STORAGE_PATH. An explicit path is used verbatim -- callers that point + # several identities at one file (multi-tenant tests, benchmarks) depend on + # that, and it is what keeps the residual commingling case observable. + # + # The import stays function-local: the test session patches + # ``reflexio.server.LOCAL_STORAGE_PATH`` on the already-imported module, which + # a module-level import would read past. if db_path is None: from reflexio.server import LOCAL_STORAGE_PATH - db_path = str(Path(LOCAL_STORAGE_PATH) / "reflexio.db") + db_path = resolve_sqlite_db_path(LOCAL_STORAGE_PATH, org_id) self.db_path = db_path self._lock = threading.RLock() @@ -978,12 +995,6 @@ def __init__( logger.info("SQLite Storage for org %s using db_path: %s", org_id, db_path) - if sqlite3.sqlite_version_info < _MINIMUM_SQLITE_VERSION: - detected_version = ".".join(map(str, sqlite3.sqlite_version_info)) - raise RuntimeError( - f"SQLite 3.35.0 or newer is required; detected {detected_version}" - ) - # Ensure parent directory exists Path(db_path).parent.mkdir(parents=True, exist_ok=True) initialization_lock = _get_sqlite_initialization_lock(db_path) diff --git a/reflexio/server/services/storage/sqlite_storage/_dataset_path.py b/reflexio/server/services/storage/sqlite_storage/_dataset_path.py new file mode 100644 index 000000000..fb99a8502 --- /dev/null +++ b/reflexio/server/services/storage/sqlite_storage/_dataset_path.py @@ -0,0 +1,255 @@ +"""Resolve which SQLite file a dataset identity owns. + +Historically every caller resolved to one file, ``/reflexio.db``, +regardless of the identity it was constructed with. Of the persistent tables only 11 +carry an ``org_id`` column; the other 32 -- ``profiles``, ``requests``, +``interactions``, ``user_playbooks`` among them -- have none, and their reads are +unscoped. Two identities sharing that file therefore read each other's rows. + +The sharpest case is not two local plugins: a self-host deployment never passes +``base_dir``, so *every* org it serves resolved to the same file. + +This module derives the file from the identity instead, and adopts an existing +database rather than starting empty next to it. + +Adoption is **first claimer wins**, not "adopt whatever is there". Adopting on every +open would let a second identity attach to the same file, leaving an already-commingled +installation commingled forever -- and those installations are the only ones with the +bug. See ``docs/superpowers/specs/2026-09-02-oss-dataset-isolation-fix-design.md``. +""" + +from __future__ import annotations + +import logging +import re +import sqlite3 +import time +from pathlib import Path + +logger = logging.getLogger(__name__) + +LEGACY_DB_FILENAME = "reflexio.db" + +IDENTITY_CLAIM_TABLE = "_dataset_identity" + +# Tables carrying an ``org_id`` column. Read only to decide whether an unclaimed +# legacy file plausibly belongs to the opening identity; a tie-breaker, not a +# decision procedure -- several are written rarely or not at all. +_ATTRIBUTED_TABLES = ( + "_agent_runs", + "_pending_tool_calls", + "audit_events", + "braintrust_connection", + "imported_score", + "learning_jobs", + "lineage_event", + "purge_operation_targets", + "purge_operations", + "share_links", + "subject_write_barriers", +) + +# A dataset identity becomes a path component, so it is constrained to characters +# that cannot traverse or escape. Rejected, never rewritten: slugifying would map +# two distinct identities onto one file, which is the defect this module exists to +# fix. +_IDENTITY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$") + + +class DatasetIdentityError(RuntimeError): + """A dataset file cannot be resolved for the requested identity.""" + + +def validate_dataset_identity(org_id: str) -> str: + """Return *org_id* if it is usable as a filename component. + + Args: + org_id (str): The caller-supplied dataset identity. + + Returns: + str: The identity, unchanged. + + Raises: + DatasetIdentityError: If it could traverse, escape, or collide. + """ + if not _IDENTITY_RE.fullmatch(org_id or ""): + raise DatasetIdentityError( + f"dataset identity {org_id!r} cannot be used as a storage filename: " + "it must match ^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$. It is rejected rather " + "than rewritten, because rewriting could map two identities onto one file." + ) + return org_id + + +def derive_db_path(root: str | Path, org_id: str) -> Path: + """Return the file *org_id* owns under *root*. + + A flat filename rather than a subdirectory, so sibling artifacts in the same + directory -- the enterprise ``sql_app.db``, the ``disk_*`` trees -- are untouched. + + Args: + root (str | Path): The storage root directory. + org_id (str): The dataset identity. + + Returns: + Path: The derived database path. + """ + validate_dataset_identity(org_id) + return Path(root) / f"reflexio_{org_id}.db" + + +def _table_exists(conn: sqlite3.Connection, table: str) -> bool: + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + return row is not None + + +def stored_identity_labels(path: Path) -> set[str] | None: + """Return the identities appearing in *path*'s attributed tables. + + Every read is guarded by ``PRAGMA table_info``: no migration adds ``org_id`` to + these tables, so one created by a build predating the column never gains it, and + an unguarded read would raise ``no such column``. + + Args: + path (Path): The database to inspect. + + Returns: + set[str] | None: The labels found, or ``None`` if the file carries no + readable attribution at all. + """ + if not path.exists(): + return None + labels: set[str] = set() + saw_column = False + try: + conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + except sqlite3.Error: + return None + try: + for table in _ATTRIBUTED_TABLES: + if not _table_exists(conn, table): + continue + columns = { + str(row[1]) for row in conn.execute(f"PRAGMA table_info({table})") + } + if "org_id" not in columns: + continue + saw_column = True + for (value,) in conn.execute( + f"SELECT DISTINCT org_id FROM {table} WHERE org_id IS NOT NULL" # noqa: S608 + ): + if value: + labels.add(str(value)) + except sqlite3.Error: + return None + finally: + conn.close() + return labels if saw_column else None + + +def claim_or_read_identity(path: Path, org_id: str) -> str: + """Claim *path* for *org_id*, or return the identity that already holds it. + + The claim is a row rather than a marker file or a PRAGMA: ``application_id`` and + ``user_version`` are 32-bit integers and cannot hold an identity string, and a + sidecar file can desync from the database it describes. + + ``BEGIN IMMEDIATE`` takes a cross-process write lock, which is required rather + than defensive -- the service runs multiple uvicorn workers by default, and the + in-process initialization lock in ``_base`` is a ``threading.Lock``. + + Args: + path (Path): The database to claim. + org_id (str): The identity claiming it. + + Returns: + str: The identity that owns *path* -- *org_id* if the claim succeeded. + """ + conn = sqlite3.connect(path) + try: + conn.execute("BEGIN IMMEDIATE") + conn.execute( + f"CREATE TABLE IF NOT EXISTS {IDENTITY_CLAIM_TABLE} (" + " k INTEGER PRIMARY KEY CHECK (k = 1)," + " org_id TEXT NOT NULL," + " claimed_at TEXT NOT NULL)" + ) + conn.execute( + f"INSERT INTO {IDENTITY_CLAIM_TABLE} (k, org_id, claimed_at)" + " VALUES (1, ?, ?) ON CONFLICT(k) DO NOTHING", + (org_id, time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())), + ) + row = conn.execute( + f"SELECT org_id FROM {IDENTITY_CLAIM_TABLE} WHERE k = 1" + ).fetchone() + conn.commit() + finally: + conn.close() + return str(row[0]) if row else org_id + + +def resolve_sqlite_db_path(root: str | Path, org_id: str) -> str: + """Return the SQLite file *org_id* should open under *root*. + + Adopts an existing unclaimed database so an installation with history never + starts on an empty file; refuses to adopt one another identity already holds. + + Args: + root (str | Path): The storage root directory. + org_id (str): The dataset identity. + + Returns: + str: The resolved database path. + + Raises: + DatasetIdentityError: If the identity is unusable, or its own derived file + is already claimed by a different identity. + """ + validate_dataset_identity(org_id) + root_path = Path(root) + root_path.mkdir(parents=True, exist_ok=True) + derived = derive_db_path(root_path, org_id) + legacy = root_path / LEGACY_DB_FILENAME + + if derived.exists(): + owner = claim_or_read_identity(derived, org_id) + if owner != org_id: + # Reachable on a case-insensitive filesystem, where two legal + # identities differing only in case derive one filename. + raise DatasetIdentityError( + f"{derived} is already claimed by dataset {owner!r}, but was opened " + f"as {org_id!r}. Refusing to share one file between two identities." + ) + return str(derived) + + if not legacy.exists(): + claim_or_read_identity(derived, org_id) + return str(derived) + + labels = stored_identity_labels(legacy) + if labels and org_id not in labels: + logger.warning( + "Not adopting %s for dataset %r: it holds rows labelled %s. Using %s instead.", + legacy, + org_id, + sorted(labels), + derived, + ) + claim_or_read_identity(derived, org_id) + return str(derived) + + owner = claim_or_read_identity(legacy, org_id) + if owner == org_id: + return str(legacy) + + logger.warning( + "%s is claimed by dataset %r; opening %s for %r instead.", + legacy, + owner, + derived, + org_id, + ) + claim_or_read_identity(derived, org_id) + return str(derived) diff --git a/scripts/reset_db.py b/scripts/reset_db.py index 1e55801a0..4cf22045d 100755 --- a/scripts/reset_db.py +++ b/scripts/reset_db.py @@ -6,6 +6,7 @@ Usage: uv run python scripts/reset_db.py + uv run python scripts/reset_db.py --org acme uv run python scripts/reset_db.py --db-path /custom/path/reflexio.db """ @@ -20,18 +21,23 @@ sys.path.insert(0, str(_PROJECT_ROOT)) +from reflexio.cli.bootstrap_config import default_org_id from reflexio.server import LOCAL_STORAGE_PATH +from reflexio.server.services.storage.sqlite_storage._dataset_path import ( + resolve_sqlite_db_path, +) -def _default_db_path() -> Path: - return Path(LOCAL_STORAGE_PATH) / "reflexio.db" +def _default_db_path(org_id: str) -> Path: + """Return the database *org_id* owns, adopting a legacy file if it holds one.""" + return Path(resolve_sqlite_db_path(LOCAL_STORAGE_PATH, org_id)) -def reset_db(db_path: Path) -> None: +def reset_db(db_path: Path, org_id: str) -> None: """Delete the SQLite database and its WAL/SHM sidecars, then recreate empty tables.""" # Remove existing files removed: list[str] = [] - for suffix in ("", "-wal", "-shm"): + for suffix in ("", "-wal", "-shm", "-journal"): p = db_path.parent / (db_path.name + suffix) if p.exists(): p.unlink() @@ -45,7 +51,9 @@ def reset_db(db_path: Path) -> None: # Re-create by importing and instantiating storage (runs DDL automatically) from reflexio.server.services.storage.sqlite_storage import SQLiteStorage - storage = SQLiteStorage(org_id="default", db_path=str(db_path)) + # Recreated under the same identity the path was resolved for -- a hardcoded + # org would rebuild a database nobody reads. + storage = SQLiteStorage(org_id=org_id, db_path=str(db_path)) storage.conn.close() print(f"Clean database created at: {db_path}") @@ -54,15 +62,21 @@ def main() -> None: parser = argparse.ArgumentParser( description="Reset local SQLite database to a clean state." ) + parser.add_argument( + "--org", + default=None, + help="Dataset identity to reset (default: the configured default org).", + ) parser.add_argument( "--db-path", type=Path, default=None, - help=f"Path to the database file (default: {_default_db_path()})", + help="Path to the database file (default: derived from --org).", ) args = parser.parse_args() - db_path: Path = args.db_path or _default_db_path() + org_id: str = args.org or default_org_id() + db_path: Path = args.db_path or _default_db_path(org_id) print(f"This will DELETE all data in: {db_path}") confirm = input("Continue? [y/N] ") @@ -70,7 +84,7 @@ def main() -> None: print("Aborted.") sys.exit(1) - reset_db(db_path) + reset_db(db_path, org_id) if __name__ == "__main__": diff --git a/tests/server/services/storage/sqlite_storage/test_dataset_path.py b/tests/server/services/storage/sqlite_storage/test_dataset_path.py new file mode 100644 index 000000000..c11bd7647 --- /dev/null +++ b/tests/server/services/storage/sqlite_storage/test_dataset_path.py @@ -0,0 +1,142 @@ +"""Unit coverage for dataset-file resolution. + +Design: ``docs/superpowers/specs/2026-09-02-oss-dataset-isolation-fix-design.md`` +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from reflexio.server.services.storage.sqlite_storage._dataset_path import ( + DatasetIdentityError, + claim_or_read_identity, + derive_db_path, + resolve_sqlite_db_path, + stored_identity_labels, + validate_dataset_identity, +) + + +@pytest.mark.parametrize( + "identity", + ["a/b", "../escape", "/absolute", "", ".hidden", "-leading", "x" * 65, "a b"], +) +def test_unusable_identities_are_rejected(identity: str) -> None: + """I-VALID: rejected, never rewritten. + + Rewriting could map two distinct identities onto one file, which is the defect + this module exists to prevent. + """ + with pytest.raises(DatasetIdentityError): + validate_dataset_identity(identity) + + +@pytest.mark.parametrize( + "identity", ["0", "acme", "self-host-org", "claude-smart", "a.b_c-1"] +) +def test_realistic_identities_are_accepted(identity: str) -> None: + assert validate_dataset_identity(identity) == identity + + +def test_derived_path_stays_inside_the_root(tmp_path: Path) -> None: + assert derive_db_path(tmp_path, "acme").parent == tmp_path + + +def test_claim_is_first_writer_wins(tmp_path: Path) -> None: + db = tmp_path / "d.db" + assert claim_or_read_identity(db, "first") == "first" + assert claim_or_read_identity(db, "second") == "first" + + +def test_derived_file_claimed_by_another_identity_fails_closed(tmp_path: Path) -> None: + """The case-fold collision: two legal identities, one filename. + + On a case-insensitive filesystem ``Acme`` and ``acme`` derive the same path, and + no charset validator can catch it -- the claim has to. + """ + derived = derive_db_path(tmp_path, "acme") + claim_or_read_identity(derived, "someone-else") + with pytest.raises(DatasetIdentityError, match="already claimed"): + resolve_sqlite_db_path(tmp_path, "acme") + + +def test_fresh_install_uses_the_derived_path(tmp_path: Path) -> None: + resolved = resolve_sqlite_db_path(tmp_path, "acme") + assert resolved == str(tmp_path / "reflexio_acme.db") + + +def test_legacy_file_with_foreign_labels_is_not_adopted(tmp_path: Path) -> None: + """I-LABEL: route around it and warn, rather than refuse to start.""" + legacy = tmp_path / "reflexio.db" + conn = sqlite3.connect(legacy) + try: + conn.execute( + "CREATE TABLE share_links (org_id TEXT NOT NULL, token TEXT NOT NULL)" + ) + conn.execute("INSERT INTO share_links VALUES ('someone-else', 'shr_x')") + conn.commit() + finally: + conn.close() + + resolved = resolve_sqlite_db_path(tmp_path, "acme") + assert resolved == str(tmp_path / "reflexio_acme.db") + + +def test_label_scan_tolerates_a_table_without_the_column(tmp_path: Path) -> None: + """A table created before ``org_id`` existed never gains it. + + No migration adds the column, so an unguarded read would raise + ``no such column`` on a sufficiently old file. + """ + legacy = tmp_path / "reflexio.db" + conn = sqlite3.connect(legacy) + try: + conn.execute("CREATE TABLE share_links (token TEXT NOT NULL)") + conn.commit() + finally: + conn.close() + + assert stored_identity_labels(legacy) is None + + +def test_label_scan_returns_none_for_a_missing_file(tmp_path: Path) -> None: + assert stored_identity_labels(tmp_path / "absent.db") is None + + +def _claim_in_subprocess(args: tuple[str, str]) -> str: + path, org_id = args + from reflexio.server.services.storage.sqlite_storage._dataset_path import ( + claim_or_read_identity, + ) + + return claim_or_read_identity(Path(path), org_id) + + +def test_concurrent_claims_across_processes_agree_on_one_winner(tmp_path: Path) -> None: + """The in-process initialization lock is a ``threading.Lock``. + + The service runs multiple uvicorn workers by default, so two *processes* can race + to claim an unclaimed file. ``BEGIN IMMEDIATE`` is what makes that safe. + """ + from concurrent.futures import ProcessPoolExecutor + + db = tmp_path / "contended.db" + identities = [f"tenant-{i}" for i in range(6)] + + with ProcessPoolExecutor(max_workers=3) as pool: + winners = set( + pool.map(_claim_in_subprocess, [(str(db), i) for i in identities]) + ) + + assert len(winners) == 1, f"claims disagreed on the owner: {winners}" + + conn = sqlite3.connect(db) + try: + rows = conn.execute("SELECT org_id FROM _dataset_identity").fetchall() + finally: + conn.close() + assert len(rows) == 1 + assert rows[0][0] in identities diff --git a/tests/server/services/storage/test_storage_defaults.py b/tests/server/services/storage/test_storage_defaults.py index 607a64807..5ed5e5690 100644 --- a/tests/server/services/storage/test_storage_defaults.py +++ b/tests/server/services/storage/test_storage_defaults.py @@ -61,15 +61,18 @@ def test_local_storage_path_empty_string_falls_back_to_default() -> None: importlib.reload(server_module) -def test_sqlite_storage_uses_local_storage_path_when_db_path_none() -> None: - """SQLiteStorage(db_path=None) resolves to LOCAL_STORAGE_PATH/reflexio.db.""" +def test_sqlite_storage_derives_db_path_from_identity_when_db_path_none() -> None: + """SQLiteStorage(db_path=None) resolves to a file named for its identity. + + Previously every identity resolved to one shared ``reflexio.db``. + """ with ( tempfile.TemporaryDirectory() as temp_dir, patch("reflexio.server.LOCAL_STORAGE_PATH", temp_dir), patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512), ): storage = SQLiteStorage(org_id="0", db_path=None) - assert storage.db_path == str(Path(temp_dir) / "reflexio.db") + assert storage.db_path == str(Path(temp_dir) / "reflexio_0.db") def test_local_storage_path_honors_reflexio_log_dir_override(tmp_path: Path) -> None: @@ -101,3 +104,129 @@ def test_sqlite_storage_explicit_db_path_overrides_env() -> None: explicit_path = str(Path(explicit_dir) / "custom.db") storage = SQLiteStorage(org_id="0", db_path=explicit_path) assert storage.db_path == explicit_path + + +# --------------------------------------------------------------------------- +# Dataset isolation: which file an identity resolves to, and when it adopts one. +# +# Design: docs/superpowers/specs/2026-09-02-oss-dataset-isolation-fix-design.md +# --------------------------------------------------------------------------- + + +def _storage(org_id: str, root: str): + with ( + patch("reflexio.server.LOCAL_STORAGE_PATH", root), + patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512), + ): + return SQLiteStorage(org_id=org_id, db_path=None) + + +def test_distinct_identities_resolve_to_distinct_files(tmp_path: Path) -> None: + """I-SEP: two identities under one root never share a file.""" + a = _storage("tenant-a", str(tmp_path)) + b = _storage("tenant-b", str(tmp_path)) + try: + assert a.db_path != b.db_path + finally: + a.conn.close() + b.conn.close() + + +def test_existing_install_adopts_its_legacy_database(tmp_path: Path) -> None: + """I-ADOPT: an installation with history never starts on an empty file.""" + legacy = tmp_path / "reflexio.db" + seeded = _storage_at("incumbent", str(legacy)) + try: + seeded.conn.execute( + "INSERT INTO profiles (profile_id, user_id, content, created_at," + " last_modified_timestamp) VALUES (?,?,?,?,?)", + ("p1", "u1", "history", 1, 1), + ) + seeded.conn.commit() + finally: + seeded.conn.close() + + adopted = _storage("incumbent", str(tmp_path)) + try: + assert adopted.db_path == str(legacy) + rows = adopted.conn.execute("SELECT content FROM profiles").fetchall() + assert [r["content"] for r in rows] == ["history"] + finally: + adopted.conn.close() + + +def test_second_identity_does_not_adopt_a_claimed_database(tmp_path: Path) -> None: + """I-CLAIM: adoption is first-claimer-wins, so it isolates rather than shares. + + The guard against the rule that would leave an already-commingled install + commingled forever. + """ + legacy = tmp_path / "reflexio.db" + seeded = _storage_at("incumbent", str(legacy)) + try: + seeded.conn.execute( + "INSERT INTO profiles (profile_id, user_id, content, created_at," + " last_modified_timestamp) VALUES (?,?,?,?,?)", + ("p1", "u1", "incumbent-only", 1, 1), + ) + seeded.conn.commit() + finally: + seeded.conn.close() + + first = _storage("incumbent", str(tmp_path)) + first.conn.close() + + second = _storage("newcomer", str(tmp_path)) + try: + assert second.db_path != str(legacy) + assert ( + second.conn.execute("SELECT count(*) AS n FROM profiles").fetchone()["n"] + == 0 + ) + finally: + second.conn.close() + + # the incumbent's file is untouched + reopened = _storage("incumbent", str(tmp_path)) + try: + assert reopened.db_path == str(legacy) + assert ( + reopened.conn.execute("SELECT count(*) AS n FROM profiles").fetchone()["n"] + == 1 + ) + finally: + reopened.conn.close() + + +def test_adoption_is_idempotent(tmp_path: Path) -> None: + """Re-opening as the incumbent re-adopts without rewriting the claim.""" + legacy = tmp_path / "reflexio.db" + _storage_at("incumbent", str(legacy)).conn.close() + + first = _storage("incumbent", str(tmp_path)) + first.conn.close() + claimed_at = _claim_row(legacy) + + second = _storage("incumbent", str(tmp_path)) + try: + assert second.db_path == str(legacy) + finally: + second.conn.close() + assert _claim_row(legacy) == claimed_at + + +def _claim_row(path: Path): + import sqlite3 + + conn = sqlite3.connect(path) + try: + return conn.execute( + "SELECT org_id, claimed_at FROM _dataset_identity WHERE k = 1" + ).fetchone() + finally: + conn.close() + + +def _storage_at(org_id: str, db_path: str): + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + return SQLiteStorage(org_id=org_id, db_path=db_path)