diff --git a/README.md b/README.md index 2eaf59ab3c..0bcf7b927b 100644 --- a/README.md +++ b/README.md @@ -1955,7 +1955,7 @@ values nested in lists or structs, before opening the destination. Filter them or replace them with `NULL` in SQL; the CLI reports an error and preserves any existing output rather than writing invalid JSON. -The **data-analysis** skill preserves each CSV file or Excel sheet when names collide across inputs. [Prepare its Python environment](skills/public/data-analysis/SKILL.md#environment-setup) explicitly with DuckDB and openpyxl; missing packages report setup instructions instead of installing themselves during analysis. Use the SQL names printed by `inspect` (such as `report`, `report_1`) for queries and summaries. Its cache follows ordered input basenames and contents, so renaming or reordering inputs rebuilds their table map instead of silently selecting a different source. Quoted original names also resolve without rewriting query text; apostrophes in upload paths and double quotes in column names are preserved through loading, inspection, queries, and summaries. CSV, JSON, and Markdown exports accept bare filenames such as `report.csv` in the current working directory as well as nested relative and absolute paths. +The **data-analysis** skill preserves each CSV file or Excel sheet when names collide across inputs. [Prepare its Python environment](skills/public/data-analysis/SKILL.md#environment-setup) explicitly with DuckDB and openpyxl; missing packages report setup instructions instead of installing themselves during analysis. Use the SQL names printed by `inspect` (such as `report`, `report_1`) for queries and summaries. Its cache follows ordered input basenames and contents, so renaming or reordering inputs rebuilds their table map instead of silently selecting a different source. Cache publication uses complete immutable generations: import writers close before their database and table map become visible together, so overlapping processes can read the same snapshot. Interrupted private builds are removed and damaged entries rebuild; legacy flat caches rebuild once. Complete generations remain available to active readers, so stop analyses before manually clearing `.data-analysis-cache/`. Automatic cache garbage collection is not provided. Quoted original names also resolve without rewriting query text; apostrophes in upload paths and double quotes in column names are preserved through loading, inspection, queries, and summaries. CSV, JSON, and Markdown exports accept bare filenames such as `report.csv` in the current working directory as well as nested relative and absolute paths. After an answer loads skills, its toolbar includes **Skills used**. Hover over or click the icon to see the skills and their sources; hovering a skill name underlines it. Select a skill to inspect its `SKILL.md` in the resizable side panel (a drawer on mobile). The view uses snapshots captured during successful configured read-tool loads or explicit slash activation, so later edits or removal of a skill do not rewrite its history. Repeated loads appear once per run, in first-load order. Range reads and bounded snapshots are labeled as partial; older conversations without captured evidence do not show this menu. Copy returns the captured Markdown, including YAML frontmatter. Package-relative links and images remain readable references rather than navigating away from the conversation. Skill loads performed through other tools, such as shell commands, are not inferred from answer text. diff --git a/skills/public/data-analysis/AGENTS.md b/skills/public/data-analysis/AGENTS.md index 1453814e79..1bc4142e78 100644 --- a/skills/public/data-analysis/AGENTS.md +++ b/skills/public/data-analysis/AGENTS.md @@ -8,6 +8,23 @@ Cache identity includes input order, filesystem-encoded basenames, per-file cont digests, and a namespace version. Directory relocation alone does not change a readable input's identity. Bump the namespace if table-mapping semantics change. +Cache publication uses `open_analysis_cache()`: build in a private UUID directory +under .data-analysis-cache/generations-v1/, close the DuckDB writer, reopen +read-only, then atomically replace one manifest containing its generation and map. +Never publish metadata while an import writer still owns the database. Validate +the manifest, allocator invariants and physical main-schema tables before reuse; +invalid entries rebuild. Flat legacy caches are ignored, not migrated or deleted. +Remove private builds if manifest preparation fails or an unsuccessful +replacement leaves its unique temporary source in place. Retain a generation +when publication may have happened because another process may be reading it. +Concurrent cold builders can repeat work but never share a writable database. +Stop analyses before manual cache cleanup. There is no automatic garbage +collection or power-loss durability guarantee. +Keep real-DuckDB process-barrier and failure-cleanup regressions in +tests/skills/test_data_analysis_cache_lifecycle.py. Use production CSV loading +to avoid optional spatial setup. The reader at the publication barrier must +reuse the published generation without loading the source again. + Regressions live under the repository's root tests directory. They use real DuckDB. Excel and CLI integration additionally run with an already installed official spatial extension via `DEERFLOW_TEST_SPATIAL_DIRECTORY`; default tests diff --git a/skills/public/data-analysis/SKILL.md b/skills/public/data-analysis/SKILL.md index 90a398c683..6d61449855 100644 --- a/skills/public/data-analysis/SKILL.md +++ b/skills/public/data-analysis/SKILL.md @@ -260,11 +260,18 @@ After analysis: The script automatically caches loaded data to avoid re-parsing files on every call: -- On first load, files are parsed and stored in a persistent DuckDB database under the system temporary directory (`.data-analysis-cache/`) -- The cache key hashes the ordered input basenames and a separate SHA256 digest of each file. Renaming, reordering, or changing inputs creates a new cache; moving unchanged, identically named inputs to another directory preserves reuse. The updated key namespace ignores older cached table maps, which are rebuilt on first use -- Subsequent calls with the same files will use the cached database directly (near-instant startup) +- On first load, files are parsed into a private DuckDB generation under the system temporary directory (.data-analysis-cache/generations-v1/). Its writer closes before a complete generation and table map become visible together through an atomic manifest replacement +- The cache key hashes the ordered input basenames and a separate SHA256 digest of each file. Renaming, reordering, or changing inputs creates a new cache; moving unchanged, identically named inputs to another directory preserves reuse. Older flat caches are ignored and rebuilt from source without deleting the old files +- Subsequent calls open a complete generation read-only. Overlapping readers can reuse it while another analysis is still running; invalid manifests or damaged databases rebuild from source - Cache is transparent — no extra parameters needed +Concurrent cold calls may build separate complete generations. Failed private +builds are removed, while complete generations are retained so active readers +remain valid. Stop analyses before manually clearing `.data-analysis-cache/`; +automatic cache garbage collection is not implemented. These are rebuildable +local caches, not snapshots of changing input files or a power-loss guarantee. +Analysis connections are read-only; export query results with `--output-file`. + This is especially useful when running multiple queries against the same data files (inspect → query → summary). ## Notes diff --git a/skills/public/data-analysis/scripts/analyze.py b/skills/public/data-analysis/scripts/analyze.py index acc46e9662..200cbba2a3 100644 --- a/skills/public/data-analysis/scripts/analyze.py +++ b/skills/public/data-analysis/scripts/analyze.py @@ -11,9 +11,14 @@ import json import logging import os import re +import shutil import sys import tempfile +import uuid from collections import Counter +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger(__name__) @@ -34,7 +39,6 @@ except ImportError: # Cache directory for persistent DuckDB databases CACHE_DIR = os.path.join(tempfile.gettempdir(), ".data-analysis-cache") -TABLE_MAP_SUFFIX = ".table_map.json" def compute_files_hash(files: list[str]) -> str: @@ -61,34 +65,145 @@ def compute_files_hash(files: list[str]) -> str: return hasher.hexdigest() -def get_cache_db_path(files_hash: str) -> str: - """Get the path to the cached DuckDB database file.""" - os.makedirs(CACHE_DIR, exist_ok=True) - return os.path.join(CACHE_DIR, f"{files_hash}.duckdb") - - -def get_table_map_path(files_hash: str) -> str: - """Get the path to the cached table map JSON file.""" - return os.path.join(CACHE_DIR, f"{files_hash}{TABLE_MAP_SUFFIX}") - - -def save_table_map(files_hash: str, table_map: dict[str, str]) -> None: - """Save table map to a JSON file alongside the cached DB.""" - path = get_table_map_path(files_hash) - with open(path, "w", encoding="utf-8") as f: - json.dump(table_map, f, ensure_ascii=False) - - -def load_table_map(files_hash: str) -> dict[str, str] | None: - """Load table map from cache. Returns None if not found.""" - path = get_table_map_path(files_hash) - if not os.path.exists(path): - return None +def _open_cached_generation(root: Path, files_hash: str): + """Open one complete generation; invalid cache entries are ordinary misses.""" + con = None try: - with open(path, "r", encoding="utf-8") as f: - return json.load(f) - except Exception: + manifest = json.loads((root / f"{files_hash}.json").read_text(encoding="utf-8")) + if ( + not isinstance(manifest, dict) + or type(manifest.get("format")) is not int + or manifest["format"] != 1 + ): + return None + generation = manifest.get("generation") + table_map = manifest.get("table_map") + if not isinstance(generation, str) or not re.fullmatch( + rf"{files_hash}\.[0-9a-f]{{32}}", generation + ): + return None + if ( + not isinstance(table_map, dict) + or not table_map + or not all( + isinstance(k, str) and isinstance(v, str) and k and v + for k, v in table_map.items() + ) + ): + return None + aliases = {name.casefold() for name in table_map} + targets = {name.casefold() for name in table_map.values()} + if len(aliases) != len(table_map) or len(targets) != len(table_map): + return None + if any( + alias.casefold() in targets and alias.casefold() != target.casefold() + for alias, target in table_map.items() + ): + return None + directory = root / generation + database = directory / "data.duckdb" + if directory.is_symlink() or database.is_symlink(): + return None + con = duckdb.connect(str(database), read_only=True) + physical_tables = { + row[0].casefold() + for row in con.execute( + "SELECT table_name FROM information_schema.tables " + "WHERE table_schema = 'main' AND table_type = 'BASE TABLE' " + "AND table_catalog = current_database()" + ).fetchall() + } + if targets != physical_tables: + raise ValueError("Cached table map does not match its database") + for table in table_map.values(): + quoted = '"' + table.replace('"', '""') + '"' + con.execute(f"SELECT * FROM main.{quoted} LIMIT 0") + return con, table_map + except (OSError, ValueError, duckdb.Error): + if con is not None: + con.close() + logger.debug("Ignoring unreadable analysis cache", exc_info=True) return None + except BaseException: + if con is not None: + con.close() + raise + + +@contextmanager +def _prepare_cache_manifest( + root: Path, generation: str, table_map: dict[str, str] +) -> Iterator[str]: + """Write and flush a private manifest before its atomic publication.""" + fd, temporary = tempfile.mkstemp(prefix=".manifest-", suffix=".tmp", dir=root) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump( + {"format": 1, "generation": generation, "table_map": table_map}, + handle, + ensure_ascii=False, + ) + handle.flush() + os.fsync(handle.fileno()) + yield temporary + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +@contextmanager +def open_analysis_cache(files: list[str]): + """Own a read-only analysis connection backed by an immutable generation.""" + files_hash = compute_files_hash(files) + root = Path(CACHE_DIR) / "generations-v1" + root.mkdir(parents=True, exist_ok=True) + cached = _open_cached_generation(root, files_hash) + if cached is not None: + con, table_map = cached + logger.info("Cache hit! Using complete cached generation") + try: + yield con, table_map + finally: + con.close() + return + + generation = f"{files_hash}.{uuid.uuid4().hex}" + directory = root / generation + directory.mkdir(mode=0o700) + database = directory / "data.duckdb" + con = None + possibly_published = False + try: + logger.info("Loading files into a private cache generation...") + con = duckdb.connect(str(database)) + table_map = load_files(con, files) + if not table_map: + logger.error("No tables were loaded. Check file paths and formats.") + raise SystemExit(1) + con.close() + con = None + # Never expose a database still owned by a writer, including the first + # invocation's analysis action. Other processes may already be readers. + con = duckdb.connect(str(database), read_only=True) + destination = root / f"{files_hash}.json" + with _prepare_cache_manifest(root, generation, table_map) as temporary: + # Replacement may succeed before an interruption reaches Python. + possibly_published = True + try: + os.replace(temporary, destination) + except BaseException: + # A successful replacement consumes this unique source even + # if a concurrent builder has since replaced the destination. + if os.path.exists(temporary): + possibly_published = False + raise + logger.info("Loaded %s table(s): %s", len(table_map), ", ".join(table_map)) + yield con, table_map + finally: + if con is not None: + con.close() + if not possibly_published: + shutil.rmtree(directory) def sanitize_table_name(name: str) -> str: @@ -564,51 +679,16 @@ def main(): if args.action == "summary" and not args.table: parser.error("--table is required for 'summary' action") - # Compute file hash for caching - files_hash = compute_files_hash(args.files) - db_path = get_cache_db_path(files_hash) - cached_table_map = load_table_map(files_hash) - - if cached_table_map and os.path.exists(db_path): - # Cache hit: connect to existing DB - logger.info(f"Cache hit! Using cached database: {db_path}") - con = duckdb.connect(db_path, read_only=True) - table_map = cached_table_map - logger.info( - f"Loaded {len(table_map)} table(s) from cache: {', '.join(table_map.keys())}" - ) - else: - # Cache miss: load files and persist to DB - logger.info("Loading files (first time, will cache for future use)...") - con = duckdb.connect(db_path) - table_map = load_files(con, args.files) - - if not table_map: - logger.error("No tables were loaded. Check file paths and formats.") - # Clean up empty DB file - con.close() - if os.path.exists(db_path): - os.remove(db_path) - sys.exit(1) - - # Save table map for future cache lookups - save_table_map(files_hash, table_map) - logger.info( - f"\nLoaded {len(table_map)} table(s): {', '.join(table_map.keys())}" - ) - logger.info(f"Cached database saved to: {db_path}") - try: - if args.action == "inspect": - action_inspect(con, table_map) - elif args.action == "query": - action_query(con, args.sql, table_map, args.output_file) - elif args.action == "summary": - action_summary(con, args.table, table_map) + with open_analysis_cache(args.files) as (con, table_map): + if args.action == "inspect": + action_inspect(con, table_map) + elif args.action == "query": + action_query(con, args.sql, table_map, args.output_file) + elif args.action == "summary": + action_summary(con, args.table, table_map) except ValueError as exc: sys.exit(f"Error: {exc}") - finally: - con.close() if __name__ == "__main__": diff --git a/tests/skills/test_data_analysis_cache_lifecycle.py b/tests/skills/test_data_analysis_cache_lifecycle.py new file mode 100644 index 0000000000..2b2a9948e6 --- /dev/null +++ b/tests/skills/test_data_analysis_cache_lifecycle.py @@ -0,0 +1,523 @@ +"""Real DuckDB cache lifetimes, using the production CSV loader without spatial.""" + +import errno +import json +import multiprocessing +import runpy +import sys +from pathlib import Path + +import pytest + +duckdb = pytest.importorskip("duckdb") +pytest.importorskip("openpyxl") +SCRIPT = ( + Path(__file__).resolve().parents[2] + / "skills/public/data-analysis/scripts/analyze.py" +) + + +def configure(cache, source): + module = runpy.run_path(str(SCRIPT)) + scope = module["main"].__globals__ + scope["CACHE_DIR"] = str(cache) + + def load_csv(con, files): + # Replace load_files dispatch with production CSV loading, avoiding the + # optional INSTALL spatial setup. Cache, SQL and filesystem stay real. + tables = {} + for path in files: + scope["_load_csv"](con, path, tables) + return tables + + scope["load_files"] = load_csv + sys.argv = [str(SCRIPT), "--files", str(source), "--action", "inspect"] + return scope + + +def worker( + cache, + source, + started, + finish, + results, + hold, + load_barrier=None, + *, + require_cache_hit=False, +): + try: + scope = configure(cache, source) + if require_cache_hit: + scope["load_files"] = lambda *_: pytest.fail( + "reader must reuse the published cache" + ) + if load_barrier is not None: + loader = scope["load_files"] + + def gated_load(con, files): + load_barrier.wait(10) + return loader(con, files) + + scope["load_files"] = gated_load + + def inspect(con, tables): + if hold: + started.set() + if not finish.wait(10): + raise TimeoutError("parent did not release analysis") + value = con.execute('SELECT value FROM "report"').fetchall() + results.put(("ok", value)) + + scope["action_inspect"] = inspect + scope["main"]() + except BaseException as error: + results.put(("error", type(error).__name__, str(error))) + + +@pytest.fixture +def input_file(tmp_path): + source = tmp_path / "report.csv" + source.write_text("value\n7\n", encoding="utf-8") + return source + + +def test_second_process_reads_while_first_action_is_running(tmp_path, input_file): + ctx = multiprocessing.get_context("spawn") + started, finish, results = ctx.Event(), ctx.Event(), ctx.Queue() + cache = tmp_path / "cache" + first = ctx.Process( + target=worker, args=(cache, input_file, started, finish, results, True) + ) + second = ctx.Process( + target=worker, args=(cache, input_file, started, finish, results, False) + ) + first.start() + try: + assert started.wait(10), "first analysis must reach the action boundary" + second.start() + result = results.get(timeout=10) + assert result == ("ok", [(7,)]), "cache reader overlapped a live writer" + finish.set() + assert results.get(timeout=10) == ("ok", [(7,)]) + finally: + finish.set() + for process in (first, second): + if process.pid is not None: + process.join(10) + if process.is_alive(): + process.terminate() + process.join(5) + results.close() + + +def test_failed_loader_leaves_no_database_for_retry(tmp_path, input_file): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + loader = scope["load_files"] + + def fail(con, files): + loader(con, files) + raise RuntimeError("loader interrupted after writing tables") + + scope["load_files"] = fail + with pytest.raises(RuntimeError, match="loader interrupted"): + scope["main"]() + assert list(cache.rglob("*.duckdb")) == [] + + +def publication_worker(cache, source, published, release, results): + """Pause after real manifest publication, before its caller continues.""" + try: + scope = configure(cache, source) + + # The regression also runs against the flat-cache implementation. Its + # publication point exposes the map before closing the DuckDB writer. + def pause(): + published.set() + if not release.wait(10): + raise TimeoutError("parent did not release publication") + + if "save_table_map" in scope: + save = scope["save_table_map"] + + def pause_after_map(*args): + save(*args) + pause() + + scope["save_table_map"] = pause_after_map + else: + replace = scope["os"].replace + + def pause_after_replace(source_path, destination): + replace(source_path, destination) + if Path(destination).parent == cache / "generations-v1": + pause() + + scope["os"].replace = pause_after_replace + scope["action_inspect"] = lambda con, _: con.execute( + "SELECT value FROM report" + ).fetchall() + scope["main"]() + results.put(("publisher-ok",)) + except BaseException as error: + results.put(("error", type(error).__name__, str(error))) + + +def test_reader_can_open_cache_at_publication_barrier(tmp_path, input_file): + """Publishing metadata must imply an independently readable database.""" + ctx = multiprocessing.get_context("spawn") + published, release, results = ctx.Event(), ctx.Event(), ctx.Queue() + cache = tmp_path / "cache" + publisher = ctx.Process( + target=publication_worker, args=(cache, input_file, published, release, results) + ) + reader = ctx.Process( + target=worker, + args=(cache, input_file, published, release, results, False), + kwargs={"require_cache_hit": True}, + ) + publisher.start() + try: + assert published.wait(10), "publisher must reach its real metadata publication" + reader.start() + assert results.get(timeout=10) == ("ok", [(7,)]), ( + "published cache is still owned by an import writer" + ) + release.set() + assert results.get(timeout=10) == ("publisher-ok",) + finally: + release.set() + for process in (publisher, reader): + if process.pid is not None: + process.join(10) + if process.is_alive(): + process.terminate() + process.join(5) + results.close() + + +def test_action_failure_closes_connection(tmp_path, input_file, monkeypatch): + scope = configure(tmp_path / "cache", input_file) + connect = duckdb.connect + opened = [] + + def recording_connect(*args, **kwargs): + con = connect(*args, **kwargs) + opened.append(con) + return con + + def fail(*args): + raise RuntimeError("action failed") + + monkeypatch.setattr(duckdb, "connect", recording_connect) + scope["action_inspect"] = fail + with pytest.raises(RuntimeError, match="action failed"): + scope["main"]() + assert opened + for con in opened: + with pytest.raises(duckdb.ConnectionException): + con.execute("SELECT 1") + + +def test_corrupt_table_map_rebuilds_before_action(tmp_path, input_file): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + seen = [] + loader = scope["load_files"] + + def record_load(con, files): + seen.append("load") + return loader(con, files) + + scope["load_files"] = record_load + scope["action_inspect"] = lambda con, tables: con.execute( + 'SELECT value FROM "' + tables["report"] + '"' + ).fetchall() + scope["main"]() + manifests = list(cache.rglob("*.json")) + assert len(manifests) == 1 + data = json.loads(manifests[0].read_text(encoding="utf-8")) + if "table_map" in data: + data["table_map"] = {"report": "missing"} + else: + data = {"report": "missing"} + manifests[0].write_text(json.dumps(data), encoding="utf-8") + scope["main"]() + assert seen == ["load", "load"] + + +def test_simultaneous_cold_processes_publish_complete_generations(tmp_path, input_file): + ctx = multiprocessing.get_context("spawn") + started, finish, results, barrier = ( + ctx.Event(), + ctx.Event(), + ctx.Queue(), + ctx.Barrier(2), + ) + cache = tmp_path / "cache" + processes = [ + ctx.Process( + target=worker, + args=(cache, input_file, started, finish, results, False, barrier), + ) + for _ in range(2) + ] + for process in processes: + process.start() + try: + assert [results.get(timeout=15) for _ in processes] == [ + ("ok", [(7,)]), + ("ok", [(7,)]), + ] + scope = configure(cache, input_file) + scope["load_files"] = lambda *_: pytest.fail("published cache must be reusable") + with scope["open_analysis_cache"]([str(input_file)]) as (con, tables): + assert con.execute('SELECT value FROM "report"').fetchall() == [(7,)] + assert tables == {"report": "report"} + finally: + for process in processes: + process.join(10) + if process.is_alive(): + process.terminate() + process.join(5) + results.close() + + +def manifest(cache): + paths = list(cache.rglob("*.json")) + assert len(paths) == 1 + return paths[0], json.loads(paths[0].read_text(encoding="utf-8")) + + +@pytest.mark.parametrize( + "damaged_map", + [ + {"report": "report", "report_1": "report"}, + {"report": "report_1", "other": "report"}, + {"report": "report"}, + {"Report": "report", "report": "report_1"}, + ], +) +def test_inconsistent_manifest_rebuilds_without_redirecting_sources( + tmp_path, input_file, damaged_map +): + second = tmp_path / "other" / "report.csv" + second.parent.mkdir() + second.write_text("value\n99\n", encoding="utf-8") + files = [str(input_file), str(second)] + cache = tmp_path / "cache" + scope = configure(cache, input_file) + with scope["open_analysis_cache"](files): + pass + path, data = manifest(cache) + data["table_map"] = damaged_map + path.write_text(json.dumps(data), encoding="utf-8") + loads = [] + loader = scope["load_files"] + + def record_load(con, paths): + loads.append(True) + return loader(con, paths) + + scope["load_files"] = record_load + with scope["open_analysis_cache"](files) as (con, tables): + assert tables == {"report": "report", "report_1": "report_1"} + assert con.execute("SELECT value FROM report").fetchall() == [(7,)] + assert con.execute("SELECT value FROM report_1").fetchall() == [(99,)] + assert loads == [True] + + +def test_reader_keeps_old_generation_during_repair(tmp_path, input_file): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + with scope["open_analysis_cache"]([str(input_file)]) as (old, _): + path, data = manifest(cache) + data["format"] = 0 + path.write_text(json.dumps(data), encoding="utf-8") + with scope["open_analysis_cache"]([str(input_file)]) as (new, tables): + assert new.execute('SELECT value FROM "report"').fetchall() == [(7,)] + assert tables == {"report": "report"} + assert old.execute('SELECT value FROM "report"').fetchall() == [(7,)] + assert old.execute('SELECT value FROM "report"').fetchall() == [(7,)] + + +@pytest.mark.parametrize( + "bad", + [ + None, + [], + {}, + {"format": True}, + {"format": 2}, + {"format": 1, "generation": "../../outside", "table_map": {"report": "report"}}, + ], +) +def test_invalid_manifest_is_rebuilt(tmp_path, input_file, bad): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + scope["main"]() + path, _ = manifest(cache) + path.write_text(json.dumps(bad), encoding="utf-8") + rebuilt = [] + loader = scope["load_files"] + scope["load_files"] = lambda con, files: (rebuilt.append(True), loader(con, files))[ + 1 + ] + scope["main"]() + assert rebuilt == [True] + + +@pytest.mark.parametrize("damage", ["missing", "truncated"]) +def test_missing_or_corrupt_database_is_rebuilt(tmp_path, input_file, damage): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + scope["main"]() + _, data = manifest(cache) + path = cache / "generations-v1" / data["generation"] / "data.duckdb" + if damage == "missing": + path.unlink() + else: + path.write_bytes(b"not a database") + scope["main"]() + with scope["open_analysis_cache"]([str(input_file)]) as (con, _): + assert con.execute('SELECT value FROM "report"').fetchall() == [(7,)] + + +@pytest.mark.parametrize( + "failure", [RuntimeError("failed"), KeyboardInterrupt(), SystemExit(2)] +) +def test_load_interruption_cleans_private_build(tmp_path, input_file, failure): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + loader = scope["load_files"] + + def fail(con, files): + loader(con, files) + raise failure + + scope["load_files"] = fail + with pytest.raises(type(failure)): + scope["main"]() + assert list(cache.rglob("*.duckdb")) == [] + assert list(cache.rglob("*.json")) == [] + scope["load_files"] = loader + scope["main"]() + + +@pytest.mark.parametrize( + ("module", "operation"), + [("tempfile", "mkstemp"), ("json", "dump"), ("os", "fsync")], +) +def test_manifest_preparation_failure_cleans_private_generation( + tmp_path, input_file, monkeypatch, module, operation +): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + + def fail(*args, **kwargs): + raise OSError(errno.ENOSPC, "manifest preparation ran out of space") + + with monkeypatch.context() as patch: + patch.setattr(scope[module], operation, fail) + with pytest.raises(OSError) as error: + scope["main"]() + assert error.value.errno == errno.ENOSPC + assert list((cache / "generations-v1").iterdir()) == [] + + with scope["open_analysis_cache"]([str(input_file)]) as (con, _): + assert con.execute("SELECT value FROM report").fetchall() == [(7,)] + + +def test_publication_failure_preserves_prior_manifest( + tmp_path, input_file, monkeypatch +): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + scope["main"]() + path, data = manifest(cache) + data["format"] = 0 # Force repair, retaining the original publication. + path.write_text(json.dumps(data), encoding="utf-8") + before = path.read_bytes() + original_generations = set(cache.rglob("*.duckdb")) + + def fail(*args): + raise OSError("publication failed") + + with monkeypatch.context() as patch: + patch.setattr(scope["os"], "replace", fail) + with pytest.raises(OSError, match="publication failed"): + scope["main"]() + assert path.read_bytes() == before + assert set(cache.rglob("*.duckdb")) == original_generations + assert list(cache.rglob("*.tmp")) == [] + scope["main"]() + + +def test_failed_replacement_retries_remove_private_generations(tmp_path, input_file): + cache = tmp_path / "cache" + scope = configure(cache, input_file) + root = cache / "generations-v1" + manifest_path = root / f"{scope['compute_files_hash']([str(input_file)])}.json" + manifest_path.mkdir(parents=True) + + # A directory at the destination makes the real atomic replacement fail. + for _ in range(3): + with pytest.raises(OSError) as error: + scope["main"]() + assert Path(error.value.filename2) == manifest_path + assert list(root.iterdir()) == [manifest_path] + + manifest_path.rmdir() + with scope["open_analysis_cache"]([str(input_file)]) as (con, _): + assert con.execute("SELECT value FROM report").fetchall() == [(7,)] + + +def test_interrupted_cache_validation_closes_reader(tmp_path, input_file, monkeypatch): + scope = configure(tmp_path / "cache", input_file) + scope["main"]() + connect = duckdb.connect + opened = [] + + class InterruptedConnection: + def __init__(self, connection): + self.connection = connection + + def execute(self, *_): + raise KeyboardInterrupt() + + def close(self): + self.connection.close() + + def wrapped_connect(*args, **kwargs): + con = connect(*args, **kwargs) + opened.append(con) + return InterruptedConnection(con) + + monkeypatch.setattr(duckdb, "connect", wrapped_connect) + with pytest.raises(KeyboardInterrupt): + scope["main"]() + assert opened + with pytest.raises(duckdb.ConnectionException): + opened[0].execute("SELECT 1") + + +def test_interruption_after_replace_keeps_published_generation( + tmp_path, input_file, monkeypatch +): + scope = configure(tmp_path / "cache", input_file) + replace = scope["os"].replace + + def publish_then_interrupt(*args): + replace(*args) + raise KeyboardInterrupt() + + with monkeypatch.context() as patch: + patch.setattr(scope["os"], "replace", publish_then_interrupt) + with pytest.raises(KeyboardInterrupt): + scope["main"]() + scope["load_files"] = lambda *_: pytest.fail( + "a published generation must remain readable" + ) + with scope["open_analysis_cache"]([str(input_file)]) as (con, _): + assert con.execute("SELECT value FROM report").fetchall() == [(7,)] diff --git a/tests/skills/test_data_analysis_export_paths.py b/tests/skills/test_data_analysis_export_paths.py index 3c29eaf0f2..ce7b1f4ec3 100644 --- a/tests/skills/test_data_analysis_export_paths.py +++ b/tests/skills/test_data_analysis_export_paths.py @@ -66,13 +66,18 @@ def test_cli_exports_basename_from_real_read_only_cache(tmp_path, extension): source = tmp_path / "source.csv" source.write_text("answer,label\n42,分析\n", encoding="utf-8") analysis = runpy.run_path(str(SCRIPT)) - key = analysis["compute_files_hash"]([str(source)]) - table_map = {} - with duckdb.connect(str(cache_dir / f"{key}.duckdb")) as con: - analysis["_load_csv"](con, str(source), table_map) - (cache_dir / f"{key}.table_map.json").write_text( - json.dumps(table_map), encoding="utf-8" - ) + scope = analysis["main"].__globals__ + scope["CACHE_DIR"] = str(cache_dir) + + def load_csv(con, files): + table_map = {} + for path in files: + analysis["_load_csv"](con, path, table_map) + return table_map + + scope["load_files"] = load_csv + with analysis["open_analysis_cache"]([str(source)]): + pass output = f"report.{extension}" completed = subprocess.run( [ diff --git a/tests/skills/test_data_analysis_input_identity.py b/tests/skills/test_data_analysis_input_identity.py index cfcf747ce7..33e0a11752 100644 --- a/tests/skills/test_data_analysis_input_identity.py +++ b/tests/skills/test_data_analysis_input_identity.py @@ -61,22 +61,27 @@ def test_each_input_keeps_a_distinct_queryable_table( def test_cache_reopen_keeps_all_duplicate_inputs(analysis, tmp_path): files = [csv_file(tmp_path / str(i), "report.csv", i) for i in (1, 2)] - db = tmp_path / "loaded.duckdb" - table_map = {} - with duckdb.connect(str(db)) as con: - for file in files: + scope = analysis["main"].__globals__ + scope["CACHE_DIR"] = str(tmp_path / "cache") + + def load_csv(con, paths): + table_map = {} + for file in paths: analysis["_load_csv"](con, file, table_map) - # Exercise the actual JSON persistence used on a cache hit. - analysis["save_table_map"].__globals__["CACHE_DIR"] = str(tmp_path) - key = analysis["compute_files_hash"](files) - analysis["save_table_map"](key, table_map) - restored = analysis["load_table_map"](key) - assert restored == {"report": "report", "report_1": "report_1"} - with duckdb.connect(str(db), read_only=True) as con: - output = analysis["action_query"]( - con, "SELECT a.value, b.value FROM report a CROSS JOIN report_1 b", restored - ) - assert output.splitlines()[2].replace(" ", "") == "1|2" + return table_map + + # Exercise the production generation manifest and read-only reopening. + scope["load_files"] = load_csv + for _ in range(2): + with analysis["open_analysis_cache"](files) as (con, restored): + assert restored == {"report": "report", "report_1": "report_1"} + output = analysis["action_query"]( + con, + "SELECT a.value, b.value FROM report a CROSS JOIN report_1 b", + restored, + ) + assert output.splitlines()[2].replace(" ", "") == "1|2" + scope["load_files"] = lambda *_: pytest.fail("published map must be reused") def test_cache_identity_changes_with_duplicate_input_order(analysis, tmp_path): diff --git a/tests/skills/test_data_analysis_json_export.py b/tests/skills/test_data_analysis_json_export.py index 30fa5183aa..b6a5947652 100644 --- a/tests/skills/test_data_analysis_json_export.py +++ b/tests/skills/test_data_analysis_json_export.py @@ -4,6 +4,7 @@ import csv import json import os import runpy +import shutil import subprocess import sys from pathlib import Path @@ -30,14 +31,19 @@ def cached_source(analysis, tmp_path): source = tmp_path / "input.csv" source.write_text("id\n1\n", encoding="utf-8") temp_root = tmp_path / "temp" - analysis["get_cache_db_path"].__globals__["CACHE_DIR"] = str( - temp_root / ".data-analysis-cache" - ) - key = analysis["compute_files_hash"]([str(source)]) - table_map = {} - with duckdb.connect(analysis["get_cache_db_path"](key)) as con: - analysis["_load_csv"](con, str(source), table_map) - analysis["save_table_map"](key, table_map) + scope = analysis["main"].__globals__ + scope["CACHE_DIR"] = str(temp_root / ".data-analysis-cache") + + def load_csv(con, files): + table_map = {} + for path in files: + analysis["_load_csv"](con, path, table_map) + return table_map + + # Seed through real generation publication, bypassing only spatial setup. + scope["load_files"] = load_csv + with analysis["open_analysis_cache"]([str(source)]): + pass return source, temp_root @@ -51,7 +57,9 @@ def opened_connections(analysis, monkeypatch): return con monkeypatch.setitem( - analysis["main"].__globals__, "duckdb", SimpleNamespace(connect=connect) + analysis["main"].__globals__, + "duckdb", + SimpleNamespace(connect=connect, Error=duckdb.Error), ) yield connections for con in connections: @@ -351,17 +359,19 @@ def test_empty_cache_is_closed_before_removal( monkeypatch.setitem(namespace, "load_files", lambda con, files: {}) removed = [] - def remove_closed_database(path): + rmtree = shutil.rmtree + + def remove_closed_generation(path): assert len(opened_connections) == 1 with pytest.raises(duckdb.ConnectionException, match="closed"): opened_connections[0].execute("SELECT 1") removed.append(path) - os.remove(path) + rmtree(path) monkeypatch.setitem( namespace, - "os", - SimpleNamespace(**{**vars(os), "remove": remove_closed_database}), + "shutil", + SimpleNamespace(rmtree=remove_closed_generation), ) monkeypatch.setattr( sys,