diff --git a/Cargo.lock b/Cargo.lock
index 460b7c9db..6d9169cb5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -610,6 +610,7 @@ dependencies = [
  "toml",
  "tower-http",
  "typed-path",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
diff --git a/docs/howto/parse-files-into-columns.md b/docs/howto/parse-files-into-columns.md
index 5bc294cef..f55e38f91 100644
--- a/docs/howto/parse-files-into-columns.md
+++ b/docs/howto/parse-files-into-columns.md
@@ -83,7 +83,7 @@ the fastest way to see whether a parser produces the rows you expect.
 
 The inline flag re-scans and re-parses every file on every query, defines one
 parser for the whole query, and forgets everything when the process exits. When
-you want a **watcher** that keeps the rows fresh, **persistence** across
+you want **row events** as files change, **persistence** across
 restarts, or **different parsers for different file sets**, move the parser into
 a `.dirsql.toml` [`[[table]]`](../reference/config.md#table) — and paste the
 command in verbatim:
@@ -110,7 +110,8 @@ dirsql query "SELECT title, author FROM posts ORDER BY title" -c ./.dirsql.toml
 ```
 
 Same command, same rows. The difference is what a declared table brings: it is
-indexed on build, kept fresh by the watcher, survives restarts with
+indexed on build, re-reads only the files that changed on each query, streams
+row events as files change, survives restarts with
 [`--persist`](./persist.md), and a config can declare
 [many tables](./define-tables.md) — each with its own `on-file` — where the flag
 gives every path-table one parser. In both spellings the table's columns are
diff --git a/docs/howto/react-to-changes.md b/docs/howto/react-to-changes.md
index d9889439b..189613c55 100644
--- a/docs/howto/react-to-changes.md
+++ b/docs/howto/react-to-changes.md
@@ -72,6 +72,9 @@ works — `EventSource` in a browser, an SSE client library, or plain
 
 ## Notes
 
+- Querying never waits on this stream: a query reads what is on disk when it
+  starts ([`query`](../reference/sdk.md#query)). A change a query reads first
+  still arrives here, once.
 - Events reflect *row* changes, not raw filesystem events: a file edit
   that leaves a table's rows identical emits nothing, and one file change
   can emit several events. The diffing model is part of
diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md
index bbd06ac8a..d848585a6 100644
--- a/docs/reference/sdk.md
+++ b/docs/reference/sdk.md
@@ -289,6 +289,24 @@ db.query_with(
 - Without `params`, a statement with placeholders fails the same way: a
   placeholder is never left unbound.
 
+- **Accurate to the filesystem.** A query reads what is on disk when it
+  starts: a file written, edited, renamed or deleted before the call is
+  reflected in its rows, with no watcher, wait or retry. Before reading, the
+  query walks each named table's glob and stats every match; a file whose stat
+  moved, or that changed within the last two seconds, is parsed again, and a
+  file that is gone loses its rows. The row events those re-reads produce reach
+  [`watch`](#watch) streams and the next poll, once each. If a file on disk
+  breaks a foreign key, the query fails with the error a build over that tree
+  returns. In TypeScript the re-reads run on the JavaScript thread, where
+  `onFile` runs, and the SQL then runs on the threadpool.
+
+  The check trusts the file's stat, so it shares the limits of every
+  stat-based tool (`make`, `rsync`, `git status`): a write through a shared
+  memory map can land without the file's times moving until the writer
+  unmaps or syncs it; a network filesystem that caches attributes can report
+  a stale stat; and a clock stepped backwards can stamp a new write with a
+  time older than the last read.
+
 - **Read-only.** Each statement is classified by SQLite itself
   (`sqlite3_stmt_readonly`); anything classified as a write — `INSERT`,
   `UPDATE`, `DELETE`, `DROP`, `CREATE`, `ALTER`, `VACUUM`, … — is rejected
@@ -369,7 +387,9 @@ to stop.
 The watcher is live once both `watch()` and [`ready`](#ready) have returned,
 in either order (a Rust `DirSQL` is ready when `build()` returns); a file
 written after that point produces an event and reaches the index. A write
-made while the initial scan is still running may be missed.
+made while the initial scan is still running may be missed by the stream, but
+never by a query (see [`query`](#query)). A change a query read first still
+arrives on the stream, once.
 
 An instance with no named tables cannot be watched: path-tables emit no
 watch events, so `watch()` (and the polling primitives) raise an error naming
diff --git a/packages/python/changelog.d/2026-11-03-fresh-declared-reads.md b/packages/python/changelog.d/2026-11-03-fresh-declared-reads.md
new file mode 100644
index 000000000..187636a4b
--- /dev/null
+++ b/packages/python/changelog.d/2026-11-03-fresh-declared-reads.md
@@ -0,0 +1,3 @@
+**Fixed**
+
+A query on a long-lived `DirSQL` now reads what is on disk when it starts (core change, surfaced through the Python SDK). Before reading, it walks each named table's glob, re-runs `on_file` on every file whose stat moved or that changed within the last two seconds, and drops the rows of files that are gone, so a write made just before a query is never missed. The row events those re-reads produce reach `watch()` streams once each, including a change read before the first `watch()` call.
diff --git a/packages/python/e2e-attestations/bot-1572-fresh-red-1104.json b/packages/python/e2e-attestations/bot-1572-fresh-red-1104.json
new file mode 100644
index 000000000..a85022fba
--- /dev/null
+++ b/packages/python/e2e-attestations/bot-1572-fresh-red-1104.json
@@ -0,0 +1,7 @@
+{
+  "command": "uv run python -m pytest tests/e2e/declared_table_name_test.py tests/e2e/fresh_declared_repros_test.py tests/e2e/on_file_query_test.py tests/e2e/use_case_dossier_test.py tests/e2e/use_case_join_test.py tests/e2e/use_case_notebook_test.py tests/e2e/use_case_transcripts_test.py tests/e2e/frontmatter_parse_speed_test.py tests/e2e/jsonl_parse_speed_test.py tests/e2e/fanout_test.py tests/e2e/plugin_discovery_test.py tests/e2e/plugin_extension_discovery_test.py tests/e2e/extension_package_test.py tests/e2e/server_speed_test.py tests/e2e/query_contract_test.py tests/e2e/repl_interactive_test.py tests/e2e/watch_speed_test.py tests/e2e/sigint_interrupt_test.py -q",
+  "ran_at": 1793815824,
+  "exit_code": 0,
+  "commit": "501e509a06e56f8cdd91445dae06bb55180f2787",
+  "branch": "bot/1572-fresh-red-1104"
+}
diff --git a/packages/python/migrations.d/2026-11-03-fresh-declared-reads.md b/packages/python/migrations.d/2026-11-03-fresh-declared-reads.md
new file mode 100644
index 000000000..abdffc251
--- /dev/null
+++ b/packages/python/migrations.d/2026-11-03-fresh-declared-reads.md
@@ -0,0 +1,23 @@
+### Queries read what is on disk
+
+#### Summary
+
+A query on a long-lived `DirSQL` brings its named tables in line with the files on disk before reading. No API or configuration changes are required.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+A query re-runs `on_file` on the files that changed since the index last read them, so code that waited on a watch event or slept before querying no longer needs to. A change a query reads first still arrives on `watch()`, once.
+
+Each query walks its tables' globs and stats every match, about what the same query on a path-table costs. When a file on disk breaks a foreign key, the query raises the error a build over that tree raises, until the file is fixed.
+
+#### Verification
+
+Build a `DirSQL` over a directory with a declared table, write a matching file, then `await db.query(...)` the table with no wait; the row for the new file is in the result.
diff --git a/packages/python/tests/e2e/fresh_declared_repros_test.py b/packages/python/tests/e2e/fresh_declared_repros_test.py
new file mode 100644
index 000000000..6c2513e1a
--- /dev/null
+++ b/packages/python/tests/e2e/fresh_declared_repros_test.py
@@ -0,0 +1,169 @@
+"""E2E: a long-lived ``DirSQL`` reads what is on disk when each query runs.
+
+Regressions for the reported repros, against the installed build: a declared
+table read with no watch running, a change written between ``ready()`` and
+``watch()`` starting, and a query racing a consumed ``watch()``. Every query
+runs the moment the write before it returns. No mocks: real binding, real
+filesystem, real SQLite.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+
+from dirsql import DirSQL, Table
+
+RACE_ROUNDS = 1000
+
+
+def _turns(root: Path) -> Table:
+    def on_file(path):
+        return [{"id": Path(path).parent.relative_to(root).as_posix()}]
+
+    return Table(
+        name="turns",
+        ddl="CREATE TABLE turns (id TEXT PRIMARY KEY)",
+        glob="turns/*/a.txt",
+        on_file=on_file,
+    )
+
+
+def _records() -> Table:
+    def on_file(path):
+        p = Path(path)
+        return [{"name": p.name, "v": json.loads(p.read_text())["v"]}]
+
+    return Table(
+        name="t",
+        ddl="CREATE TABLE t (name TEXT, v INTEGER)",
+        glob="*.json",
+        on_file=on_file,
+    )
+
+
+def _add_turn(root: Path, n: int) -> None:
+    (root / f"turns/{n}").mkdir()
+    (root / f"turns/{n}/a.txt").write_text("x")
+
+
+async def _turn_ids(db: DirSQL) -> list[str]:
+    return [row["id"] for row in await db.query("SELECT id FROM turns ORDER BY id")]
+
+
+async def _records_of(db: DirSQL) -> list[tuple[str, int]]:
+    rows = await db.query("SELECT name, v FROM t ORDER BY name")
+    return [(row["name"], row["v"]) for row in rows]
+
+
+async def _open_turns(root: Path) -> DirSQL:
+    db = DirSQL(str(root), no_ignore=True, tables=[_turns(root)])
+    await db.ready()
+    return db
+
+
+async def _open_records(root: Path) -> DirSQL:
+    db = DirSQL(str(root), tables=[_records()])
+    await db.ready()
+    return db
+
+
+def describe_a_declared_table_with_no_watch_running():
+    @pytest.mark.asyncio
+    async def it_reads_a_file_written_after_ready(tmp_path):
+        (tmp_path / "turns/0").mkdir(parents=True)
+        (tmp_path / "turns/0/a.txt").write_text("x")
+        db = await _open_turns(tmp_path)
+        _add_turn(tmp_path, 1)
+        assert await _turn_ids(db) == ["turns/0", "turns/1"]
+
+    @pytest.mark.asyncio
+    async def it_reads_a_create_an_edit_and_a_delete(tmp_path):
+        (tmp_path / "a.json").write_text(json.dumps({"v": 1}))
+        db = await _open_records(tmp_path)
+        assert await _records_of(db) == [("a.json", 1)]
+        (tmp_path / "b.json").write_text(json.dumps({"v": 2}))
+        (tmp_path / "a.json").write_text(json.dumps({"v": 10}))
+        assert await _records_of(db) == [("a.json", 10), ("b.json", 2)]
+        (tmp_path / "b.json").unlink()
+        assert await _records_of(db) == [("a.json", 10)]
+
+
+def describe_a_change_written_before_watch_starts():
+    @pytest.mark.asyncio
+    async def it_is_read_by_the_next_query_and_streamed_once(tmp_path):
+        (tmp_path / "turns/0").mkdir(parents=True)
+        (tmp_path / "turns/0/a.txt").write_text("x")
+        db = await _open_turns(tmp_path)
+        _add_turn(tmp_path, 1)
+        assert await _turn_ids(db) == ["turns/0", "turns/1"]
+        stream = db.watch()
+        # Unread by any query, so only the watcher reports it, after every
+        # change before it.
+        _add_turn(tmp_path, 2)
+        inserted = []
+        async for event in stream:
+            assert event.action == "insert", event
+            inserted.append(event.row["id"])
+            if event.row["id"] == "turns/2":
+                break
+        assert inserted == ["turns/1", "turns/2"]
+        assert await _turn_ids(db) == ["turns/0", "turns/1", "turns/2"]
+
+
+def describe_a_query_racing_a_consumed_watch():
+    @pytest.mark.asyncio
+    async def it_reads_every_write_before_it(tmp_path):
+        (tmp_path / "turns").mkdir()
+        db = await _open_turns(tmp_path)
+        stream = db.watch()
+
+        async def consume():
+            async for _ in stream:
+                pass
+
+        task = asyncio.create_task(consume())
+        misses = []
+        for n in range(RACE_ROUNDS):
+            _add_turn(tmp_path, n)
+            if len(await db.query("SELECT id FROM turns")) != n + 1:
+                misses.append(n)
+        task.cancel()
+        assert misses == []
+
+    @pytest.mark.asyncio
+    async def it_reads_a_create_with_a_watch_consumed(tmp_path):
+        (tmp_path / "a.json").write_text(json.dumps({"v": 1}))
+        db = await _open_records(tmp_path)
+        stream = db.watch()
+
+        async def consume():
+            async for _ in stream:
+                pass
+
+        task = asyncio.create_task(consume())
+        (tmp_path / "b.json").write_text(json.dumps({"v": 2}))
+        assert await _records_of(db) == [("a.json", 1), ("b.json", 2)]
+        task.cancel()
+
+
+def describe_a_path_table():
+    @pytest.mark.asyncio
+    async def it_reads_a_create_an_edit_and_a_delete(tmp_path):
+        (tmp_path / "a.json").write_text(json.dumps({"v": 1}))
+        db = DirSQL(str(tmp_path), no_ignore=True)
+        await db.ready()
+        sql = "SELECT path, content FROM './*.json' ORDER BY path"
+
+        async def read():
+            return [(row["path"], row["content"]) for row in await db.query(sql)]
+
+        assert await read() == [("a.json", '{"v": 1}')]
+        (tmp_path / "b.json").write_text(json.dumps({"v": 2}))
+        (tmp_path / "a.json").write_text(json.dumps({"v": 10}))
+        assert await read() == [("a.json", '{"v": 10}'), ("b.json", '{"v": 2}')]
+        (tmp_path / "b.json").unlink()
+        assert await read() == [("a.json", '{"v": 10}')]
diff --git a/packages/python/tests/integration/binding/fresh_declared_test.py b/packages/python/tests/integration/binding/fresh_declared_test.py
new file mode 100644
index 000000000..8a1013d4d
--- /dev/null
+++ b/packages/python/tests/integration/binding/fresh_declared_test.py
@@ -0,0 +1,355 @@
+"""Binding-tier tests: a declared table reads exactly what is on disk.
+
+Every query runs the moment the write before it returns, and must read what
+a ``DirSQL`` built fresh over the same tree reads.
+"""
+
+import asyncio
+import os
+import random
+import shutil
+import threading
+
+import pytest
+
+from dirsql import DirSQL, Table
+
+ROUNDS = 2000
+
+
+def _notes(root):
+    roots = [os.path.realpath(root), root]
+
+    def on_file(path):
+        rel = path
+        for base in roots:
+            if path.startswith(base + os.sep):
+                rel = path[len(base) + 1 :]
+                break
+        rel = rel.replace(os.sep, "/")
+        with open(path, encoding="utf-8") as f:
+            text = f.read()
+        return [{"file": rel, "line": line} for line in text.splitlines() if line]
+
+    return Table(
+        name="notes",
+        ddl="CREATE TABLE notes (file TEXT, line TEXT)",
+        glob="**/*.txt",
+        on_file=on_file,
+    )
+
+
+async def _open(root):
+    db = DirSQL(root, tables=[_notes(root)])
+    await db.ready()
+    return db
+
+
+def _write(root, rel, content):
+    path = os.path.join(root, rel)
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8", newline="") as f:
+        f.write(content)
+
+
+def _save_atomically(root, rel, content):
+    _write(root, rel + ".stage", content)
+    os.replace(os.path.join(root, rel + ".stage"), os.path.join(root, rel))
+
+
+def _times(root, rel):
+    st = os.stat(os.path.join(root, rel))
+    return (st.st_atime_ns, st.st_mtime_ns)
+
+
+def _restore_times(root, rel, times):
+    os.utime(os.path.join(root, rel), ns=times)
+
+
+async def _lines(db, rel):
+    rows = await db.query(f"SELECT line FROM notes WHERE file = '{rel}' ORDER BY line")
+    return [row["line"] for row in rows]
+
+
+async def _snapshot(db):
+    rows = await db.query("SELECT file, line FROM notes ORDER BY file, line")
+    return [(row["file"], row["line"]) for row in rows]
+
+
+async def _assert_matches_fresh_scan(db, root, context):
+    fresh = await _open(root)
+    assert await _snapshot(db) == await _snapshot(fresh), (
+        f"the long-lived instance disagrees with a fresh scan after {context}"
+    )
+
+
+DIRS = ["", "a/", "a/b/", "c/"]
+NAMES = ["n0", "n1", "n2", "n3"]
+EXTS = ["txt", "json", "stage"]
+IGNORES = ["n0.txt\n", "a/\n", "*.txt\n", "c/n1.*\n"]
+
+
+def _content(rng):
+    return "".join(
+        f"line-{rng.randrange(10_000):04}\n" for _ in range(rng.randrange(1, 4))
+    )
+
+
+def _random_file(rng):
+    return f"{rng.choice(DIRS)}{rng.choice(NAMES)}.{rng.choice(EXTS)}"
+
+
+def _existing(root):
+    return [
+        f"{d}{n}.{e}"
+        for d in DIRS
+        for n in NAMES
+        for e in EXTS
+        if os.path.isfile(os.path.join(root, f"{d}{n}.{e}"))
+    ]
+
+
+def _step(root, rng):
+    existing = _existing(root)
+    op = rng.randrange(9)
+    if op == 1 and existing:
+        rel = rng.choice(existing)
+        times = _times(root, rel)
+        with open(os.path.join(root, rel), encoding="utf-8") as f:
+            old = f.read()
+        flipped = "".join(str((int(c) + 1) % 10) if c.isdigit() else c for c in old)
+        _write(root, rel, flipped)
+        _restore_times(root, rel, times)
+        return f"same-size edit of {rel} with its times restored"
+    if op == 2:
+        rel = _random_file(rng)
+        _save_atomically(root, rel, _content(rng))
+        return f"atomic save of {rel}"
+    if op == 3 and existing:
+        rel = rng.choice(existing)
+        os.remove(os.path.join(root, rel))
+        return f"delete {rel}"
+    if op == 4 and existing:
+        src = rng.choice(existing)
+        dst = _random_file(rng)
+        if src == dst:
+            return "nothing"
+        os.makedirs(os.path.dirname(os.path.join(root, dst)), exist_ok=True)
+        os.replace(os.path.join(root, src), os.path.join(root, dst))
+        return f"rename {src} to {dst}"
+    if op == 5:
+        new = f"{rng.choice(DIRS)}new{rng.randrange(3)}"
+        for name in NAMES[:2]:
+            _write(root, f"{new}/{name}.{rng.choice(EXTS)}", _content(rng))
+        return f"populate new directory {new}"
+    if op == 6:
+        gone = rng.choice(DIRS[1:]).rstrip("/")
+        shutil.rmtree(os.path.join(root, gone), ignore_errors=True)
+        return f"remove directory {gone}"
+    if op == 7:
+        if rng.randrange(3) == 0:
+            if os.path.exists(os.path.join(root, ".gitignore")):
+                os.remove(os.path.join(root, ".gitignore"))
+            return "remove .gitignore"
+        rule = rng.choice(IGNORES)
+        _write(root, ".gitignore", rule)
+        return f"set .gitignore to {rule!r}"
+    rel = _random_file(rng)
+    _write(root, rel, _content(rng))
+    return f"write {rel}"
+
+
+def _seed_tree(root, rng):
+    os.makedirs(os.path.join(root, ".git"))
+    for _ in range(8):
+        _write(root, _random_file(rng), _content(rng))
+
+
+def describe_a_query_right_after_a_write():
+    @pytest.mark.asyncio
+    async def it_reads_every_create(tmp_dir):
+        db = await _open(tmp_dir)
+        for i in range(ROUNDS):
+            _write(tmp_dir, f"h/{i}.txt", f"c{i}")
+            assert await _lines(db, f"h/{i}.txt") == [f"c{i}"], f"round {i}"
+            os.remove(os.path.join(tmp_dir, f"h/{i}.txt"))
+
+    @pytest.mark.asyncio
+    async def it_reads_every_edit(tmp_dir):
+        _write(tmp_dir, "e.txt", "start")
+        db = await _open(tmp_dir)
+        for i in range(ROUNDS):
+            _write(tmp_dir, "e.txt", f"e{i}")
+            assert await _lines(db, "e.txt") == [f"e{i}"], f"round {i}"
+
+    @pytest.mark.asyncio
+    async def it_reads_every_delete(tmp_dir):
+        db = await _open(tmp_dir)
+        for i in range(ROUNDS):
+            _write(tmp_dir, "d.txt", f"d{i}")
+            assert await _lines(db, "d.txt") == [f"d{i}"], f"round {i}"
+            os.remove(os.path.join(tmp_dir, "d.txt"))
+            assert await _lines(db, "d.txt") == [], f"round {i}"
+
+    @pytest.mark.asyncio
+    async def it_reads_every_rename_into_and_out_of_the_glob(tmp_dir):
+        db = await _open(tmp_dir)
+        staged = os.path.join(tmp_dir, "r.stage")
+        final = os.path.join(tmp_dir, "r.txt")
+        for i in range(ROUNDS):
+            _write(tmp_dir, "r.stage", f"r{i}")
+            os.replace(staged, final)
+            assert await _lines(db, "r.txt") == [f"r{i}"], f"round {i}"
+            os.replace(final, staged)
+            assert await _lines(db, "r.txt") == [], f"round {i}"
+
+    @pytest.mark.asyncio
+    async def it_reads_every_atomic_save(tmp_dir):
+        _write(tmp_dir, "a.txt", "start")
+        db = await _open(tmp_dir)
+        for i in range(ROUNDS):
+            _save_atomically(tmp_dir, "a.txt", f"a{i}")
+            assert await _lines(db, "a.txt") == [f"a{i}"], f"round {i}"
+
+    @pytest.mark.asyncio
+    async def it_reads_every_same_size_edit(tmp_dir):
+        _write(tmp_dir, "s.txt", "s------")
+        db = await _open(tmp_dir)
+        for i in range(ROUNDS):
+            _write(tmp_dir, "s.txt", f"s{i:06}")
+            assert await _lines(db, "s.txt") == [f"s{i:06}"], f"round {i}"
+
+    @pytest.mark.asyncio
+    async def it_reads_every_same_size_edit_with_its_times_restored(tmp_dir):
+        _write(tmp_dir, "t.txt", "t------")
+        times = _times(tmp_dir, "t.txt")
+        db = await _open(tmp_dir)
+        for i in range(ROUNDS):
+            _write(tmp_dir, "t.txt", f"t{i:06}")
+            _restore_times(tmp_dir, "t.txt", times)
+            assert await _lines(db, "t.txt") == [f"t{i:06}"], f"round {i}"
+
+
+def describe_a_long_lived_instance():
+    @pytest.mark.asyncio
+    async def it_reads_what_a_fresh_scan_reads_after_random_operations(tmp_dir):
+        for seed in range(1, 4):
+            root = os.path.join(tmp_dir, f"seed{seed}")
+            os.makedirs(root)
+            rng = random.Random(seed)
+            _seed_tree(root, rng)
+            db = await _open(root)
+            for op in range(100):
+                done = _step(root, rng)
+                await _assert_matches_fresh_scan(
+                    db, root, f"seed {seed} op {op}: {done}"
+                )
+
+    @pytest.mark.asyncio
+    async def it_reads_what_a_fresh_scan_reads_across_idle_gaps_and_bursts(tmp_dir):
+        rng = random.Random(99)
+        _seed_tree(tmp_dir, rng)
+        db = await _open(tmp_dir)
+        for cycle in range(10):
+            await asyncio.sleep(0.25 * (cycle % 4))
+            for op in range(20):
+                done = _step(tmp_dir, rng)
+                await _assert_matches_fresh_scan(
+                    db, tmp_dir, f"cycle {cycle} op {op}: {done}"
+                )
+
+
+def describe_each_way_a_change_could_be_missed():
+    @pytest.mark.asyncio
+    async def it_reads_an_edit_whose_times_are_put_back(tmp_dir):
+        _write(tmp_dir, "touched.txt", "before")
+        times = _times(tmp_dir, "touched.txt")
+        db = await _open(tmp_dir)
+        assert await _lines(db, "touched.txt") == ["before"]
+        _write(tmp_dir, "touched.txt", "after!")
+        _restore_times(tmp_dir, "touched.txt", times)
+        assert await _lines(db, "touched.txt") == ["after!"]
+
+    @pytest.mark.asyncio
+    async def it_reads_a_file_deleted_and_recreated_under_the_same_name(tmp_dir):
+        _write(tmp_dir, "again.txt", "first")
+        db = await _open(tmp_dir)
+        assert await _lines(db, "again.txt") == ["first"]
+        os.remove(os.path.join(tmp_dir, "again.txt"))
+        _write(tmp_dir, "again.txt", "secnd")
+        assert await _lines(db, "again.txt") == ["secnd"]
+
+    @pytest.mark.asyncio
+    async def it_reads_a_new_and_a_deleted_subdirectory(tmp_dir):
+        db = await _open(tmp_dir)
+        _write(tmp_dir, "sub/deep/one.txt", "one")
+        assert await _lines(db, "sub/deep/one.txt") == ["one"]
+        shutil.rmtree(os.path.join(tmp_dir, "sub"))
+        assert await _lines(db, "sub/deep/one.txt") == []
+
+    @pytest.mark.asyncio
+    async def it_reads_a_gitignore_change(tmp_dir):
+        os.makedirs(os.path.join(tmp_dir, ".git"))
+        _write(tmp_dir, "hidden.txt", "hidden")
+        db = await _open(tmp_dir)
+        assert await _lines(db, "hidden.txt") == ["hidden"]
+        _write(tmp_dir, ".gitignore", "hidden.txt\n")
+        assert await _lines(db, "hidden.txt") == []
+        os.remove(os.path.join(tmp_dir, ".gitignore"))
+        assert await _lines(db, "hidden.txt") == ["hidden"]
+
+    @pytest.mark.asyncio
+    async def it_reads_a_retargeted_symlink(tmp_dir):
+        _write(tmp_dir, "targets/one.dat", "one")
+        _write(tmp_dir, "targets/two.dat", "two")
+        link = os.path.join(tmp_dir, "link.txt")
+        os.symlink(os.path.join(tmp_dir, "targets/one.dat"), link)
+        db = await _open(tmp_dir)
+        assert await _lines(db, "link.txt") == ["one"]
+        os.remove(link)
+        os.symlink(os.path.join(tmp_dir, "targets/two.dat"), link)
+        assert await _lines(db, "link.txt") == ["two"]
+
+    @pytest.mark.asyncio
+    async def it_never_reads_a_file_torn_while_it_is_rewritten_and_ends_exact(tmp_dir):
+        def version(k):
+            return f"v{k:06}\n" * 50
+
+        _write(tmp_dir, "busy.txt", version(0))
+        db = await _open(tmp_dir)
+        done = threading.Event()
+
+        def writer():
+            for k in range(1, 301):
+                _save_atomically(tmp_dir, "busy.txt", version(k))
+            done.set()
+
+        thread = threading.Thread(target=writer)
+        thread.start()
+        while not done.is_set():
+            lines = await _lines(db, "busy.txt")
+            assert len(lines) == 50 and len(set(lines)) == 1, lines
+        thread.join()
+        assert await _lines(db, "busy.txt") == ["v000300"] * 50
+
+
+def describe_watch_consumers():
+    @pytest.mark.asyncio
+    async def it_streams_each_change_a_query_read_first_once(tmp_dir):
+        db = await _open(tmp_dir)
+        stream = db.watch()
+        files = 100
+        for i in range(files):
+            _write(tmp_dir, f"w/{i}.txt", f"w{i}")
+            assert await _lines(db, f"w/{i}.txt") == [f"w{i}"], f"file {i}"
+        # Unread by any query, so only the watcher reports it, after every
+        # change before it.
+        _write(tmp_dir, "sentinel.txt", "end")
+        counts = {}
+        async for event in stream:
+            assert event.action == "insert", event
+            line = event.row["line"]
+            if line == "end":
+                break
+            counts[line] = counts.get(line, 0) + 1
+        assert counts == {f"w{i}": 1 for i in range(files)}
diff --git a/packages/rust/Cargo.toml b/packages/rust/Cargo.toml
index 136f73532..6473102b0 100644
--- a/packages/rust/Cargo.toml
+++ b/packages/rust/Cargo.toml
@@ -95,6 +95,9 @@ typed-path = "0.12"
 [target.'cfg(unix)'.dependencies]
 rustix = { version = "1", features = ["fs"] }
 
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem"] }
+
 [lints.rust]
 # `deny`, not `forbid`: rusqlite's API imposes five unavoidable `unsafe`
 # sites (the `load_extension` guard and four `unsafe impl VTab`/`VTabCursor`).
diff --git a/packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md b/packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md
new file mode 100644
index 000000000..677c3970a
--- /dev/null
+++ b/packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md
@@ -0,0 +1,3 @@
+**Fixed**
+
+A query on a long-lived `DirSQL` now reads what is on disk when it starts. Before reading, it walks each named table's glob, re-parses every file whose stat moved or that changed within the last two seconds, and drops the rows of files that are gone, so a write made just before a query is never missed. The row events those re-reads produce reach `watch()` streams and the next `poll_events()`, once each. A change a query reads before any watcher starts reaches the first `watch()` stream or poll. The one-shot `dirsql query` reads the index as built, unchanged.
diff --git a/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md
new file mode 100644
index 000000000..dfbfd0fd2
--- /dev/null
+++ b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md
@@ -0,0 +1,23 @@
+### Queries read what is on disk
+
+#### Summary
+
+A query on a long-lived `DirSQL` brings its named tables in line with the files on disk before reading. No API or configuration changes are required.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+A query re-parses the files that changed since the index last read them, so code that waited on a watch event or slept before querying no longer needs to. A change a query reads first still arrives on `watch()` and `poll_events()`, once.
+
+Each query walks its tables' globs and stats every match, so it costs about what the same query on a path-table costs. When a file on disk breaks a foreign key, a query fails with the same `ForeignKeyViolation` a build over that tree returns, until the file is fixed.
+
+#### Verification
+
+Build a `DirSQL` over a directory with a declared table, write a matching file, then query the table with no wait; the row for the new file is in the result.
diff --git a/packages/rust/src/cli/run.rs b/packages/rust/src/cli/run.rs
index 46a07b504..2ee0b91b7 100644
--- a/packages/rust/src/cli/run.rs
+++ b/packages/rust/src/cli/run.rs
@@ -426,6 +426,10 @@ async fn run_query(args: QueryArgs, out: &mut dyn Write) -> u8 {
         }
     };
     let state = load_state(&args.common, parser);
+    // The query runs straight after the build's own read of every file.
+    if let AppState::Ready(db) = &state {
+        db.set_fresh_reads(false);
+    }
 
     // Unbounded: the process IS the query, so `timeout(1)` expresses any cap
     // natively; only the long-lived server enforces `query_timeout` (408).
diff --git a/packages/rust/src/freshness.rs b/packages/rust/src/freshness.rs
new file mode 100644
index 000000000..778079ede
--- /dev/null
+++ b/packages/rust/src/freshness.rs
@@ -0,0 +1,268 @@
+//! Which files a query must read again so its rows match the files on disk.
+//!
+//! A file's change time moves on every write and cannot be set back by
+//! `utime`, so once it sits [`SETTLE_NS`] before the moment a read began, the
+//! read saw the file's final content and an identical stat later proves the
+//! file unchanged. Until then the file is read on every query.
+
+use std::path::Path;
+
+/// How far before a read's start a file's change time must sit for that read
+/// to have seen its final content. Covers kernels whose timestamps lag the
+/// clock by a tick and filesystems that keep whole seconds.
+pub(crate) const SETTLE_NS: i64 = 2_000_000_000;
+
+/// The stat of a file, following symlinks: what a change to it must alter.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct Fingerprint {
+    pub(crate) dev: u64,
+    pub(crate) ino: u64,
+    pub(crate) size: u64,
+    pub(crate) mtime_ns: i64,
+    /// `None` where the platform reports no change time: such a file never
+    /// settles, so every query reads it.
+    pub(crate) ctime_ns: Option<i64>,
+}
+
+/// What the index holds for one `(table, file)`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Known {
+    /// Indexed by the build and not stat'd since.
+    Built,
+    /// Last read with the file at `fp` (`None`: it could not be stat'd).
+    Read {
+        fp: Option<Fingerprint>,
+        settled: bool,
+    },
+}
+
+/// Whether a file a query found must be read for its rows to be current,
+/// given what the index holds for it and its stat now.
+pub(crate) fn must_read(known: Option<&Known>, now: Option<&Fingerprint>, built_ns: i64) -> bool {
+    match (known, now) {
+        (None, _) | (_, None) => true,
+        (Some(Known::Built), Some(now)) => !changed_before(now, built_ns),
+        (Some(Known::Read { fp, settled }), Some(now)) => !settled || fp.as_ref() != Some(now),
+    }
+}
+
+/// Whether `fp` last changed far enough before `start_ns` that a read begun
+/// then saw its final content.
+pub(crate) fn changed_before(fp: &Fingerprint, start_ns: i64) -> bool {
+    fp.ctime_ns
+        .is_some_and(|ctime| ctime < start_ns.saturating_sub(SETTLE_NS))
+}
+
+/// What the index records for a file read at `start_ns` and stat'd after.
+pub(crate) fn after_read(fp: Option<Fingerprint>, start_ns: i64) -> Known {
+    Known::Read {
+        fp,
+        settled: fp.is_some_and(|fp| changed_before(&fp, start_ns)),
+    }
+}
+
+/// What the index records for a file a query need not read: a built file
+/// counts as read when the build began; any other entry stands.
+pub(crate) fn adopted(
+    known: Option<&Known>,
+    now: Option<Fingerprint>,
+    built_ns: i64,
+) -> Option<Known> {
+    (known == Some(&Known::Built)).then(|| after_read(now, built_ns))
+}
+
+#[cfg(unix)]
+pub(crate) fn fingerprint(path: &Path) -> std::io::Result<Fingerprint> {
+    use std::os::unix::fs::MetadataExt;
+    let meta = std::fs::metadata(path)?;
+    Ok(Fingerprint {
+        dev: meta.dev(),
+        ino: meta.ino(),
+        size: meta.size(),
+        mtime_ns: unix_ns(meta.mtime(), meta.mtime_nsec()),
+        ctime_ns: Some(unix_ns(meta.ctime(), meta.ctime_nsec())),
+    })
+}
+
+#[cfg(unix)]
+fn unix_ns(secs: i64, nanos: i64) -> i64 {
+    secs.saturating_mul(1_000_000_000).saturating_add(nanos)
+}
+
+/// Windows keeps the change time (NTFS `ChangeTime`) out of `std`'s stable
+/// metadata, so it is read from the handle, with the volume serial and file
+/// index standing in for `dev` and `ino`.
+#[cfg(windows)]
+#[expect(
+    unsafe_code,
+    reason = "GetFileInformationByHandle(Ex) are the only way to read ChangeTime and the file index"
+)]
+pub(crate) fn fingerprint(path: &Path) -> std::io::Result<Fingerprint> {
+    use std::os::windows::fs::OpenOptionsExt;
+    use std::os::windows::io::AsRawHandle;
+    use windows_sys::Win32::Storage::FileSystem::{
+        BY_HANDLE_FILE_INFORMATION, FILE_BASIC_INFO, FILE_READ_ATTRIBUTES, FileBasicInfo,
+        GetFileInformationByHandle, GetFileInformationByHandleEx,
+    };
+    let file = std::fs::OpenOptions::new()
+        .access_mode(FILE_READ_ATTRIBUTES)
+        .open(path)?;
+    let handle = file.as_raw_handle();
+    // SAFETY: both structs are plain data for which all-zero is valid, and
+    // each call writes at most the size it is given into its own struct
+    // through a handle `file` keeps open.
+    let mut info: BY_HANDLE_FILE_INFORMATION = unsafe { std::mem::zeroed() };
+    if unsafe { GetFileInformationByHandle(handle, &mut info) } == 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+    let mut basic: FILE_BASIC_INFO = unsafe { std::mem::zeroed() };
+    let size = u32::try_from(std::mem::size_of::<FILE_BASIC_INFO>()).expect("small struct");
+    if unsafe { GetFileInformationByHandleEx(handle, FileBasicInfo, (&raw mut basic).cast(), size) }
+        == 0
+    {
+        return Err(std::io::Error::last_os_error());
+    }
+    Ok(Fingerprint {
+        dev: u64::from(info.dwVolumeSerialNumber),
+        ino: (u64::from(info.nFileIndexHigh) << 32) | u64::from(info.nFileIndexLow),
+        size: (u64::from(info.nFileSizeHigh) << 32) | u64::from(info.nFileSizeLow),
+        mtime_ns: windows_ns(basic.LastWriteTime),
+        ctime_ns: (basic.ChangeTime != 0).then(|| windows_ns(basic.ChangeTime)),
+    })
+}
+
+/// A Windows file time (100ns ticks since 1601) as nanoseconds since 1970.
+#[cfg(windows)]
+fn windows_ns(ticks: i64) -> i64 {
+    ticks
+        .saturating_sub(116_444_736_000_000_000)
+        .saturating_mul(100)
+}
+
+#[cfg(not(any(unix, windows)))]
+pub(crate) fn fingerprint(path: &Path) -> std::io::Result<Fingerprint> {
+    let meta = std::fs::metadata(path)?;
+    Ok(Fingerprint {
+        dev: 0,
+        ino: 0,
+        size: meta.len(),
+        mtime_ns: 0,
+        ctime_ns: None,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn fp(ctime_ns: Option<i64>) -> Fingerprint {
+        Fingerprint {
+            dev: 1,
+            ino: 2,
+            size: 3,
+            mtime_ns: 4,
+            ctime_ns,
+        }
+    }
+
+    const START: i64 = 10 * SETTLE_NS;
+
+    #[test]
+    fn a_change_time_settles_only_strictly_before_the_margin() {
+        assert!(changed_before(&fp(Some(START - SETTLE_NS - 1)), START));
+        assert!(!changed_before(&fp(Some(START - SETTLE_NS)), START));
+        assert!(!changed_before(&fp(Some(START)), START));
+    }
+
+    #[test]
+    fn a_built_file_is_adopted_as_read_when_the_build_began() {
+        let now = fp(Some(START - SETTLE_NS - 1));
+        assert_eq!(
+            adopted(Some(&Known::Built), Some(now), START),
+            Some(Known::Read {
+                fp: Some(now),
+                settled: true
+            })
+        );
+    }
+
+    #[test]
+    fn a_file_already_read_or_unknown_is_not_adopted() {
+        let read = Known::Read {
+            fp: Some(fp(Some(0))),
+            settled: true,
+        };
+        assert_eq!(adopted(Some(&read), Some(fp(Some(0))), START), None);
+        assert_eq!(adopted(None, Some(fp(Some(0))), START), None);
+    }
+
+    #[test]
+    fn a_file_without_a_change_time_never_settles() {
+        assert!(!changed_before(&fp(None), START));
+    }
+
+    #[test]
+    fn the_margin_does_not_overflow_at_the_earliest_time() {
+        assert!(!changed_before(&fp(Some(i64::MIN)), i64::MIN));
+    }
+
+    #[test]
+    fn a_file_the_index_has_never_held_is_read() {
+        assert!(must_read(None, Some(&fp(Some(0))), START));
+    }
+
+    #[test]
+    fn a_file_that_cannot_be_stat_ed_is_read() {
+        let known = after_read(Some(fp(Some(0))), START);
+        assert!(must_read(Some(&known), None, START));
+    }
+
+    #[test]
+    fn a_built_file_is_read_unless_it_changed_before_the_build_began() {
+        let old = fp(Some(START - SETTLE_NS - 1));
+        let recent = fp(Some(START - SETTLE_NS));
+        assert!(!must_read(Some(&Known::Built), Some(&old), START));
+        assert!(must_read(Some(&Known::Built), Some(&recent), START));
+    }
+
+    #[test]
+    fn a_settled_file_is_read_again_only_when_its_stat_moves() {
+        let old = fp(Some(0));
+        let known = after_read(Some(old), START);
+        assert!(!must_read(Some(&known), Some(&old), START));
+        let moved = Fingerprint { size: 9, ..old };
+        assert!(must_read(Some(&known), Some(&moved), START));
+    }
+
+    #[test]
+    fn an_unsettled_file_is_read_on_every_query() {
+        let recent = fp(Some(START));
+        let known = after_read(Some(recent), START);
+        assert!(must_read(Some(&known), Some(&recent), START));
+    }
+
+    #[test]
+    fn a_read_settles_a_file_only_when_its_change_time_is_old() {
+        assert_eq!(
+            after_read(Some(fp(Some(0))), START),
+            Known::Read {
+                fp: Some(fp(Some(0))),
+                settled: true
+            }
+        );
+        assert_eq!(
+            after_read(Some(fp(Some(START))), START),
+            Known::Read {
+                fp: Some(fp(Some(START))),
+                settled: false
+            }
+        );
+        assert_eq!(
+            after_read(None, START),
+            Known::Read {
+                fp: None,
+                settled: false
+            }
+        );
+    }
+}
diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs
index adaa71126..670ec8401 100644
--- a/packages/rust/src/lib.rs
+++ b/packages/rust/src/lib.rs
@@ -20,6 +20,7 @@ pub mod differ;
 pub mod extension_resolution;
 #[doc(hidden)]
 pub mod foreign_key;
+mod freshness;
 #[doc(hidden)]
 pub mod functions;
 mod gitignore_frame;
@@ -78,7 +79,7 @@ use std::collections::{HashMap, HashSet};
 use std::error::Error as StdError;
 use std::path::{Path, PathBuf};
 use std::sync::{
-    Arc, Mutex,
+    Arc, Mutex, MutexGuard,
     atomic::{AtomicBool, Ordering},
 };
 use std::thread;
@@ -455,6 +456,24 @@ struct DirSqlInner {
     poll_interval: Duration,
     /// Filesystem seam: [`RealFs`] in production; unit tests inject a double.
     fs: Arc<dyn FileSystem>,
+    /// Whether a query first brings the named tables in line with the files
+    /// on disk. Off where a query cannot run `on_file` on its own thread.
+    fresh_reads: AtomicBool,
+    /// When the build's walk began.
+    built_ns: i64,
+    /// Held across every change to the index, so a query's reconcile and a
+    /// watch batch never interleave and their events reach consumers in the
+    /// order they were applied.
+    fresh: Mutex<FreshState>,
+}
+
+#[derive(Default)]
+struct FreshState {
+    /// What the index holds for each `(table, file)`.
+    known: HashMap<(String, String), freshness::Known>,
+    seeded: bool,
+    /// Row events a query's reconcile produced, for the next watch batch.
+    outbox: Vec<RowEvent>,
 }
 
 fn watch_scopes(groups: &[AnchorGroup]) -> Vec<WatchScope> {
@@ -599,10 +618,148 @@ impl DirSQL {
     /// [`query_ordered`](Self::query_ordered), binding `params` as
     /// [`query_with`](Self::query_with) does.
     pub fn query_ordered_with(&self, sql: &str, params: &Params) -> Result<QueryResult> {
+        let _fresh = self.refresh()?;
         let db = self.inner.db.lock().map_err(DirSqlError::lock)?;
         db.query_ordered_with(sql, params).map_err(map_db_error)
     }
 
+    /// Turn off reading the named tables' files at query time, for a caller
+    /// that queries once, straight after the build, or that cannot run
+    /// `on_file` on the thread a query runs on.
+    #[doc(hidden)]
+    pub fn set_fresh_reads(&self, on: bool) {
+        self.inner.fresh_reads.store(on, Ordering::SeqCst);
+    }
+
+    /// Bring every named table in line with the files on disk now, whether
+    /// or not queries do so themselves: a binding whose queries run where
+    /// `on_file` cannot be called calls this first, on the thread that can.
+    #[doc(hidden)]
+    pub fn read_changes(&self) -> Result<()> {
+        let mut state = self.inner.fresh.lock().map_err(DirSqlError::lock)?;
+        self.reconcile(&mut state)
+    }
+
+    /// Bring every named table in line with the files on disk now, and hold
+    /// the index still until the returned guard drops.
+    fn refresh(&self) -> Result<Option<MutexGuard<'_, FreshState>>> {
+        if self.inner.groups.is_empty() || !self.inner.fresh_reads.load(Ordering::SeqCst) {
+            return Ok(None);
+        }
+        let mut state = self.inner.fresh.lock().map_err(DirSqlError::lock)?;
+        self.reconcile(&mut state)?;
+        Ok(Some(state))
+    }
+
+    /// Walk each anchor, read every file whose stat says it may differ from
+    /// what the index holds, and drop the rows of files that are gone, in one
+    /// transaction.
+    fn reconcile(&self, state: &mut FreshState) -> Result<()> {
+        let start_ns = now_ns();
+        if !state.seeded {
+            self.seed(state)?;
+        }
+        let mut found: Vec<(String, String, PathBuf)> = Vec::new();
+        for group in &self.inner.groups {
+            for (path, table) in
+                self.inner
+                    .fs
+                    .scan_subtree(&group.anchor, &group.anchor, &group.matcher)
+            {
+                let rel = scanner::to_slash(path.strip_prefix(&group.anchor).unwrap_or(&path));
+                found.push((table, rel, path));
+            }
+        }
+        self.with_db(|db| Ok(foreign_key::begin(db.conn())?))?;
+        let mut stats: HashMap<PathBuf, Option<freshness::Fingerprint>> = HashMap::new();
+        let mut updates: Vec<((String, String), freshness::Known)> = Vec::new();
+        let mut pending = PendingRefresh::default();
+        let mut events = Vec::new();
+        for (table, rel, path) in &found {
+            let now = *stats
+                .entry(path.clone())
+                .or_insert_with(|| self.inner.fs.fingerprint(path).ok());
+            let key = (table.clone(), rel.clone());
+            let known = state.known.get(&key);
+            if !freshness::must_read(known, now.as_ref(), self.inner.built_ns) {
+                if let Some(adopted) = freshness::adopted(known, now, self.inner.built_ns) {
+                    updates.push((key, adopted));
+                }
+                continue;
+            }
+            if self.is_per_table(table) {
+                self.track_batch_file(table, path, rel, false);
+                pending.mark(table, rel);
+            } else {
+                events.extend(self.read_file(table, path, rel));
+                updates.push((
+                    key,
+                    freshness::after_read(self.inner.fs.fingerprint(path).ok(), start_ns),
+                ));
+            }
+        }
+        let present: HashSet<(&str, &str)> = found
+            .iter()
+            .map(|(table, rel, _)| (table.as_str(), rel.as_str()))
+            .collect();
+        let gone: Vec<(String, String)> = state
+            .known
+            .keys()
+            .filter(|(table, rel)| !present.contains(&(table.as_str(), rel.as_str())))
+            .cloned()
+            .collect();
+        for (table, rel) in &gone {
+            if self.is_per_table(table) {
+                self.track_batch_file(table, Path::new(rel), rel, true);
+                pending.mark(table, rel);
+            } else {
+                events.extend(self.handle_delete(table, rel));
+            }
+        }
+        let rerun: HashSet<String> = pending.0.iter().map(|(table, _)| table.clone()).collect();
+        events.extend(self.refresh_pending(pending));
+        for (table, rel, path) in found.iter().filter(|(table, _, _)| rerun.contains(table)) {
+            updates.push((
+                (table.clone(), rel.clone()),
+                freshness::after_read(self.inner.fs.fingerprint(path).ok(), start_ns),
+            ));
+        }
+        self.with_db(|db| foreign_key::commit(db.conn()))?;
+        for key in gone {
+            state.known.remove(&key);
+        }
+        state.known.extend(updates);
+        state.outbox.extend(events);
+        Ok(())
+    }
+
+    /// Record every file the build indexed as [`freshness::Known::Built`].
+    fn seed(&self, state: &mut FreshState) -> Result<()> {
+        let owned = self.with_db(Db::owned_files)?;
+        let batch_files = self.inner.batch_files.lock().map_err(DirSqlError::lock)?;
+        let batched = batch_files
+            .iter()
+            .flat_map(|(table, files)| files.iter().map(|rel| (table.clone(), rel.clone())));
+        for key in owned.into_iter().chain(batched) {
+            state.known.insert(key, freshness::Known::Built);
+        }
+        state.seeded = true;
+        Ok(())
+    }
+
+    /// Re-read one file of a per-file table. A file whose hook fails keeps no
+    /// rows, as a fresh build would give it.
+    fn read_file(&self, table: &str, path: &Path, rel: &str) -> Vec<RowEvent> {
+        let mut events = self.handle_upsert(table, path, rel);
+        if events
+            .iter()
+            .any(|event| matches!(event, RowEvent::Error { .. }))
+        {
+            events.extend(self.handle_delete(table, rel));
+        }
+        events
+    }
+
     /// The directory the index is rooted at: where path-tables resolve.
     pub fn root(&self) -> &Path {
         &self.inner.root
@@ -636,6 +793,7 @@ impl DirSQL {
         start: impl FnOnce(&[String]) -> S,
         on_row: impl FnMut(&mut S, &[Value]),
     ) -> Result<S> {
+        let _fresh = self.refresh()?;
         let db = self.inner.db.lock().map_err(DirSqlError::lock)?;
         db.query_each_with(sql, params, start, on_row)
             .map_err(map_db_error)
@@ -650,6 +808,7 @@ impl DirSQL {
         start: impl FnOnce(&[String]) -> S,
         on_row: impl FnMut(&mut S, &[rusqlite::types::ValueRef<'_>]),
     ) -> Result<S> {
+        let _fresh = self.refresh()?;
         let db = self.inner.db.lock().map_err(DirSqlError::lock)?;
         db.query_each_ref(sql, &Params::default(), start, on_row)
             .map_err(map_db_error)
@@ -747,6 +906,21 @@ impl DirSQL {
     /// before any per-table hook runs. One write reaches the watcher as
     /// several events; a hook run for the first would run again for the rest.
     fn apply_file_events_draining(
+        &self,
+        events: Vec<FileEvent>,
+        late: impl FnMut() -> Vec<FileEvent>,
+    ) -> Vec<RowEvent> {
+        let mut fresh = self
+            .inner
+            .fresh
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut out = std::mem::take(&mut fresh.outbox);
+        out.extend(self.apply_events(events, late));
+        out
+    }
+
+    fn apply_events(
         &self,
         events: Vec<FileEvent>,
         mut late: impl FnMut() -> Vec<FileEvent>,
@@ -1425,6 +1599,7 @@ impl DirSQL {
             path_table_parser,
             no_ignore,
         } = resolved;
+        let built_ns = now_ns();
 
         let (groups, anchors, table_names) = compile_groups(&tables, &ignore, &root)?;
         let groups = groups
@@ -1548,6 +1723,7 @@ impl DirSQL {
             poll_interval,
             path_table_parser,
             no_ignore,
+            built_ns,
         })
     }
 
@@ -1600,6 +1776,7 @@ impl DirSQL {
             hint_legacy_files_table,
             path_table_parser,
             no_ignore,
+            built_ns,
         } = prepared;
 
         let (mut db, fresh_ddl, ddl_warnings, persist_ready, needs_sweep) = match persist {
@@ -1890,6 +2067,9 @@ impl DirSQL {
                 watch_thread_started: AtomicBool::new(false),
                 poll_interval,
                 fs,
+                fresh_reads: AtomicBool::new(true),
+                built_ns,
+                fresh: Mutex::new(FreshState::default()),
             }),
         })
     }
@@ -2291,6 +2471,8 @@ pub struct PreparedBuild {
     hint_legacy_files_table: bool,
     path_table_parser: Option<String>,
     no_ignore: bool,
+    /// When the walk began.
+    built_ns: i64,
 }
 
 #[doc(hidden)]
@@ -2549,6 +2731,7 @@ trait FileSystem: Send + Sync {
     fn hash(&self, path: &Path) -> std::io::Result<[u8; 32]>;
     /// Canonicalize the watch root, falling back to the literal path.
     fn canonical_root(&self, root: &Path) -> String;
+    fn fingerprint(&self, path: &Path) -> std::io::Result<freshness::Fingerprint>;
 }
 
 struct RealFs;
@@ -2579,6 +2762,10 @@ impl FileSystem for RealFs {
         hash_file(path)
     }
 
+    fn fingerprint(&self, path: &Path) -> std::io::Result<freshness::Fingerprint> {
+        freshness::fingerprint(path)
+    }
+
     fn canonical_root(&self, root: &Path) -> String {
         canonical_root(root)
     }
@@ -3521,6 +3708,16 @@ mod internal_tests {
             })
         }
 
+        fn fingerprint(&self, path: &Path) -> std::io::Result<freshness::Fingerprint> {
+            self.stat(path).map(|stat| freshness::Fingerprint {
+                dev: stat.dev.unsigned_abs(),
+                ino: stat.inode.unsigned_abs(),
+                size: stat.size.unsigned_abs(),
+                mtime_ns: stat.mtime_ns,
+                ctime_ns: Some(stat.ctime_ns),
+            })
+        }
+
         fn canonical_root(&self, root: &Path) -> String {
             self.canonical_roots
                 .get(root)
@@ -3932,6 +4129,7 @@ mod internal_tests {
             hint_legacy_files_table: false,
             path_table_parser: None,
             no_ignore: false,
+            built_ns: 0,
         };
         assert!(DirSQL::finish_build(prepared).is_err());
     }
@@ -4106,7 +4304,8 @@ mod internal_tests {
     }
 
     /// A one-table db whose fake fs stats every file in `files`, each already
-    /// upserted so the db holds a row per file.
+    /// upserted so the db holds a row per file. Queries read the index as
+    /// applied, since the fake walk lists no files.
     fn populated_db(root: &Path, glob: &str, files: &[&str]) -> DirSQL {
         let mut fake = FakeFs::default();
         for f in files {
@@ -4129,6 +4328,7 @@ mod internal_tests {
             Arc::new(fake),
         )
         .unwrap();
+        db.set_fresh_reads(false);
         for f in files {
             let events = db.handle_upsert("items", &root.join(f), f);
             assert!(
@@ -6257,6 +6457,7 @@ mod internal_tests {
             hint_legacy_files_table: false,
             path_table_parser: None,
             no_ignore: false,
+            built_ns: 0,
         }
     }
 
@@ -6779,11 +6980,13 @@ mod internal_tests {
             DirSQL::with_ignore_and_fs(&root, vec![items], Vec::<String>::new(), Arc::new(RealFs))
                 .unwrap();
 
-        let calls = calls.lock().unwrap();
-        assert_eq!(calls.len(), 1, "{calls:?}");
-        let mut seen = calls[0].clone();
-        seen.sort();
-        assert_eq!(seen, vec![root.join("a.txt"), root.join("b.txt")]);
+        {
+            let calls = calls.lock().unwrap();
+            assert_eq!(calls.len(), 1, "{calls:?}");
+            let mut seen = calls[0].clone();
+            seen.sort();
+            assert_eq!(seen, vec![root.join("a.txt"), root.join("b.txt")]);
+        }
         let mut names = row_names(&db);
         names.sort();
         assert_eq!(names, vec!["a.txt", "b.txt"]);
@@ -7332,6 +7535,7 @@ mod internal_tests {
             fake.clone(),
         )
         .unwrap();
+        db.set_fresh_reads(false);
         let events = db.apply_file_events(
             files
                 .iter()
@@ -7459,6 +7663,237 @@ mod internal_tests {
         );
         assert_eq!(held(&db), Vec::new());
     }
+
+    type Reads = Arc<Mutex<Vec<String>>>;
+
+    /// A per-file `items` table over `**/*.txt` whose hook records each file
+    /// it reads and answers one row holding the file's content, failing for
+    /// content `bad`.
+    fn content_table() -> (Table, Reads) {
+        let reads: Reads = Arc::default();
+        let seen = Arc::clone(&reads);
+        let table = Table::try_new(
+            "items",
+            "CREATE TABLE items (name TEXT)",
+            "**/*.txt",
+            move |path| {
+                seen.lock().unwrap().push(
+                    Path::new(path)
+                        .file_name()
+                        .unwrap()
+                        .to_string_lossy()
+                        .into_owned(),
+                );
+                let text = std::fs::read_to_string(path).unwrap_or_default();
+                if text == "bad" {
+                    return Err("bad content".into());
+                }
+                Ok(vec![Row::from_iter([(
+                    "name".to_string(),
+                    Value::Text(text),
+                )])])
+            },
+        );
+        (table, reads)
+    }
+
+    fn fresh_fixture(files: &[(&str, &str)]) -> (TempDir, PathBuf, DirSQL, Reads) {
+        let dir = TempDir::new().unwrap();
+        let root = dir.path().canonicalize().unwrap();
+        for (rel, text) in files {
+            std::fs::write(root.join(rel), text).unwrap();
+        }
+        let (table, reads) = content_table();
+        let db =
+            DirSQL::with_ignore_and_fs(&root, vec![table], Vec::<String>::new(), Arc::new(RealFs))
+                .unwrap();
+        (dir, root, db, reads)
+    }
+
+    #[test]
+    fn a_query_reads_a_file_written_after_the_build() {
+        let (_dir, root, db, _) = fresh_fixture(&[]);
+        std::fs::write(root.join("a.txt"), "one").unwrap();
+        assert_eq!(row_names(&db), vec!["one"]);
+    }
+
+    #[test]
+    fn a_query_reads_an_edit_to_a_file_the_build_indexed() {
+        let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]);
+        std::fs::write(root.join("a.txt"), "two").unwrap();
+        assert_eq!(row_names(&db), vec!["two"]);
+    }
+
+    #[test]
+    fn a_query_drops_the_rows_of_a_deleted_file() {
+        let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one"), ("b.txt", "two")]);
+        std::fs::remove_file(root.join("a.txt")).unwrap();
+        assert_eq!(row_names(&db), vec!["two"]);
+        assert_eq!(row_names(&db), vec!["two"]);
+    }
+
+    #[test]
+    fn a_query_with_fresh_reads_off_reads_the_index_as_built() {
+        let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]);
+        db.set_fresh_reads(false);
+        std::fs::write(root.join("b.txt"), "two").unwrap();
+        assert_eq!(row_names(&db), vec!["one"]);
+        db.set_fresh_reads(true);
+        assert_eq!(row_names(&db), vec!["one", "two"]);
+    }
+
+    #[test]
+    fn read_changes_reads_the_disk_with_fresh_reads_off() {
+        let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]);
+        db.set_fresh_reads(false);
+        std::fs::write(root.join("b.txt"), "two").unwrap();
+        db.read_changes().unwrap();
+        assert_eq!(row_names(&db), vec!["one", "two"]);
+    }
+
+    #[test]
+    fn query_ordered_and_query_each_read_fresh_rows_too() {
+        let (_dir, root, db, _) = fresh_fixture(&[]);
+        std::fs::write(root.join("a.txt"), "one").unwrap();
+        assert_eq!(
+            db.query_ordered("SELECT name FROM items")
+                .unwrap()
+                .rows
+                .len(),
+            1
+        );
+        std::fs::write(root.join("b.txt"), "two").unwrap();
+        let count = db
+            .query_each("SELECT name FROM items", |_| 0, |n, _| *n += 1)
+            .unwrap();
+        assert_eq!(count, 2);
+    }
+
+    #[test]
+    fn a_file_whose_hook_fails_at_query_time_keeps_no_rows_and_reports_the_error() {
+        let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]);
+        std::fs::write(root.join("a.txt"), "bad").unwrap();
+        assert_eq!(row_names(&db), Vec::<String>::new());
+        let events = db.apply_file_events(Vec::new());
+        assert!(
+            events
+                .iter()
+                .any(|event| matches!(event, RowEvent::Error { .. })),
+            "{events:?}"
+        );
+        assert!(
+            events
+                .iter()
+                .any(|event| matches!(event, RowEvent::Delete { .. })),
+            "{events:?}"
+        );
+    }
+
+    #[test]
+    fn a_query_hands_its_row_events_to_the_next_poll_once() {
+        let (_dir, root, db, _) = fresh_fixture(&[]);
+        std::fs::write(root.join("a.txt"), "one").unwrap();
+        row_names(&db);
+        assert_eq!(
+            rescan_summary(&db.apply_file_events(Vec::new())),
+            vec!["insert one"]
+        );
+        assert_eq!(db.apply_file_events(Vec::new()), vec![]);
+    }
+
+    #[test]
+    fn a_query_after_start_watching_hands_its_row_events_to_the_consumer() {
+        let (_dir, root, db, _) = fresh_fixture(&[]);
+        db.start_watching().unwrap();
+        std::fs::write(root.join("a.txt"), "one").unwrap();
+        row_names(&db);
+        assert_eq!(
+            rescan_summary(&db.apply_file_events(Vec::new())),
+            vec!["insert one"]
+        );
+    }
+
+    #[test]
+    fn a_query_before_any_watcher_hands_its_row_events_to_the_first_poll() {
+        let (_dir, root, db, _) = fresh_fixture(&[]);
+        std::fs::write(root.join("a.txt"), "one").unwrap();
+        row_names(&db);
+        db.start_watching().unwrap();
+        assert_eq!(
+            rescan_summary(&db.apply_file_events(Vec::new())),
+            vec!["insert one"]
+        );
+    }
+
+    #[test]
+    fn a_change_a_query_read_reaches_the_next_poll_once_when_the_watcher_reports_it_too() {
+        let (_dir, root, db, _) = fresh_fixture(&[]);
+        std::fs::write(root.join("a.txt"), "one").unwrap();
+        row_names(&db);
+        let events = db.apply_file_events(vec![FileEvent::Created(root.join("a.txt"))]);
+        assert_eq!(rescan_summary(&events), vec!["insert one"]);
+    }
+
+    /// A built instance whose filesystem double reports `found` under the
+    /// root, each file last changed long before any query.
+    fn settled_fixture(built: &[&str], found: &[&str]) -> (TempDir, DirSQL, Reads) {
+        let dir = TempDir::new().unwrap();
+        let root = dir.path().canonicalize().unwrap();
+        for rel in built {
+            std::fs::write(root.join(rel), "x").unwrap();
+        }
+        let fs = FakeFs::default().with_subtree(
+            &root,
+            found
+                .iter()
+                .map(|rel| (root.join(rel), "items".to_string()))
+                .collect(),
+        );
+        let (table, reads) = content_table();
+        let db = DirSQL::with_ignore_and_fs(&root, vec![table], Vec::<String>::new(), Arc::new(fs))
+            .unwrap();
+        reads.lock().unwrap().clear();
+        (dir, db, reads)
+    }
+
+    #[test]
+    fn a_built_file_unchanged_since_before_the_build_is_not_read_again() {
+        let (_dir, db, reads) = settled_fixture(&["a.txt"], &["a.txt"]);
+        row_names(&db);
+        row_names(&db);
+        assert_eq!(*reads.lock().unwrap(), Vec::<String>::new());
+    }
+
+    #[test]
+    fn a_settled_file_is_read_once_then_not_again() {
+        let (_dir, db, reads) = settled_fixture(&[], &["a.txt"]);
+        row_names(&db);
+        row_names(&db);
+        assert_eq!(*reads.lock().unwrap(), vec!["a.txt"]);
+    }
+
+    #[test]
+    fn a_per_table_hook_reruns_when_one_of_its_files_changes_and_when_one_goes() {
+        let dir = TempDir::new().unwrap();
+        let root = dir.path().canonicalize().unwrap();
+        std::fs::write(root.join("a.txt"), "x").unwrap();
+        let (items, calls) = recording_per_table("items", "*.txt");
+        let db =
+            DirSQL::with_ignore_and_fs(&root, vec![items], Vec::<String>::new(), Arc::new(RealFs))
+                .unwrap();
+        std::fs::write(root.join("b.txt"), "x").unwrap();
+        assert_eq!(row_names(&db), vec!["a.txt", "b.txt"]);
+        assert_eq!(
+            calls.lock().unwrap().last().unwrap(),
+            &vec![root.join("a.txt"), root.join("b.txt")]
+        );
+        std::fs::remove_file(root.join("a.txt")).unwrap();
+        assert_eq!(row_names(&db), vec!["b.txt"]);
+        assert_eq!(
+            calls.lock().unwrap().last().unwrap(),
+            &vec![root.join("b.txt")]
+        );
+    }
 }
 
 #[cfg(test)]
diff --git a/packages/rust/tests/common/fresh.rs b/packages/rust/tests/common/fresh.rs
new file mode 100644
index 000000000..0efabac92
--- /dev/null
+++ b/packages/rust/tests/common/fresh.rs
@@ -0,0 +1,238 @@
+//! A tree, the instance over it, and the readings the freshness tests assert:
+//! a declared table must read exactly what a fresh scan of the disk reads.
+
+#![allow(dead_code)]
+
+use dirsql::{DirSQL, Row, Table, Value};
+use std::collections::HashMap;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::time::SystemTime;
+use tempfile::TempDir;
+
+/// `recs` is a config table, so its hook runs once over every matched file.
+const CONFIG: &str = r#"
+[[table]]
+name    = "recs"
+glob    = "**/*.json"
+on-file = "cat"
+ddl     = "CREATE TABLE recs (k TEXT, v TEXT)"
+"#;
+
+pub struct Tree {
+    dir: TempDir,
+}
+
+impl Tree {
+    pub fn new() -> Self {
+        let dir = TempDir::new().unwrap();
+        fs::write(dir.path().join(".dirsql.toml"), CONFIG).unwrap();
+        Self { dir }
+    }
+
+    pub fn root(&self) -> &Path {
+        self.dir.path()
+    }
+
+    pub fn path(&self, rel: &str) -> PathBuf {
+        self.root().join(rel)
+    }
+
+    /// The instance under test: `notes`, a per-file table over `**/*.txt`,
+    /// plus the config's `recs`.
+    pub fn open(&self) -> DirSQL {
+        DirSQL::builder()
+            .root(self.root())
+            .config(self.path(".dirsql.toml"))
+            .table(notes_table(self.root()))
+            .build()
+            .unwrap()
+    }
+
+    pub fn write(&self, rel: &str, content: &str) {
+        let path = self.path(rel);
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent).unwrap();
+        }
+        fs::write(path, content).unwrap();
+    }
+
+    /// Write a sibling outside every glob, then rename it over `rel`.
+    pub fn save_atomically(&self, rel: &str, content: &str) {
+        let staged = self.path(&format!("{rel}.stage"));
+        if let Some(parent) = staged.parent() {
+            fs::create_dir_all(parent).unwrap();
+        }
+        fs::write(&staged, content).unwrap();
+        fs::rename(staged, self.path(rel)).unwrap();
+    }
+
+    pub fn remove(&self, rel: &str) {
+        fs::remove_file(self.path(rel)).unwrap();
+    }
+
+    pub fn rename(&self, from: &str, to: &str) {
+        let to = self.path(to);
+        if let Some(parent) = to.parent() {
+            fs::create_dir_all(parent).unwrap();
+        }
+        fs::rename(self.path(from), to).unwrap();
+    }
+
+    pub fn times(&self, rel: &str) -> (SystemTime, SystemTime) {
+        let meta = fs::metadata(self.path(rel)).unwrap();
+        (meta.accessed().unwrap(), meta.modified().unwrap())
+    }
+
+    /// `touch -r`: put back the access and modification times `times` holds.
+    pub fn restore_times(&self, rel: &str, times: (SystemTime, SystemTime)) {
+        let file = fs::File::options()
+            .write(true)
+            .open(self.path(rel))
+            .unwrap();
+        file.set_times(
+            fs::FileTimes::new()
+                .set_accessed(times.0)
+                .set_modified(times.1),
+        )
+        .unwrap();
+    }
+
+    /// Point the symlink `rel` at `target`, replacing any link already there.
+    pub fn link(&self, rel: &str, target: &str) {
+        let link = self.path(rel);
+        if fs::symlink_metadata(&link).is_ok() {
+            fs::remove_file(&link).unwrap();
+        }
+        symlink_file(&self.path(target), &link);
+    }
+}
+
+#[cfg(unix)]
+fn symlink_file(target: &Path, link: &Path) {
+    std::os::unix::fs::symlink(target, link).unwrap();
+}
+
+#[cfg(windows)]
+fn symlink_file(target: &Path, link: &Path) {
+    std::os::windows::fs::symlink_file(target, link).unwrap();
+}
+
+/// One row per non-empty line, keyed by the file's root-relative path.
+pub fn notes_table(root: &Path) -> Table {
+    let roots = [
+        root.to_path_buf(),
+        root.canonicalize().unwrap_or_else(|_| root.to_path_buf()),
+    ];
+    Table::new(
+        "notes",
+        "CREATE TABLE notes (file TEXT, line TEXT)",
+        "**/*.txt",
+        move |path| {
+            let path = Path::new(path);
+            let rel = roots
+                .iter()
+                .find_map(|root| path.strip_prefix(root).ok())
+                .unwrap_or(path)
+                .to_string_lossy()
+                .replace('\\', "/");
+            let content = fs::read_to_string(path).unwrap_or_default();
+            content
+                .lines()
+                .filter(|line| !line.is_empty())
+                .map(|line| {
+                    HashMap::from([
+                        ("file".to_string(), Value::Text(rel.clone())),
+                        ("line".to_string(), Value::Text(line.to_string())),
+                    ])
+                })
+                .collect::<Vec<Row>>()
+        },
+    )
+}
+
+/// Every declared table's rows, sorted.
+#[derive(Debug, PartialEq, Eq)]
+pub struct Snapshot {
+    pub notes: Vec<(String, String)>,
+    pub recs: Vec<(String, String)>,
+}
+
+pub fn snapshot(db: &DirSQL) -> Snapshot {
+    Snapshot {
+        notes: pairs(db, "SELECT file AS a, line AS b FROM notes ORDER BY a, b"),
+        recs: pairs(db, "SELECT k AS a, v AS b FROM recs ORDER BY a, b"),
+    }
+}
+
+fn pairs(db: &DirSQL, sql: &str) -> Vec<(String, String)> {
+    db.query(sql)
+        .unwrap()
+        .iter()
+        .map(|row| (text(&row["a"]), text(&row["b"])))
+        .collect()
+}
+
+pub fn text(value: &Value) -> String {
+    match value {
+        Value::Text(s) => s.clone(),
+        Value::Null => "<null>".to_string(),
+        other => format!("{other:?}"),
+    }
+}
+
+/// The `notes` lines read for `file`, in line order.
+pub fn lines_of(db: &DirSQL, file: &str) -> Vec<String> {
+    db.query(&format!(
+        "SELECT line FROM notes WHERE file = '{file}' ORDER BY line"
+    ))
+    .unwrap()
+    .iter()
+    .map(|row| text(&row["line"]))
+    .collect()
+}
+
+/// The `recs` values read for key `k`.
+pub fn recs_of(db: &DirSQL, k: &str) -> Vec<String> {
+    db.query(&format!("SELECT v FROM recs WHERE k = '{k}' ORDER BY v"))
+        .unwrap()
+        .iter()
+        .map(|row| text(&row["v"]))
+        .collect()
+}
+
+/// The oracle: what a fresh scan of the disk reads right now.
+pub fn assert_matches_fresh_scan(db: &DirSQL, tree: &Tree, context: &str) {
+    let fresh = tree.open();
+    assert_eq!(
+        snapshot(db),
+        snapshot(&fresh),
+        "the long-lived instance disagrees with a fresh scan after {context}"
+    );
+}
+
+/// A small deterministic generator, so a failing sequence replays.
+pub struct Rng(u64);
+
+impl Rng {
+    pub fn new(seed: u64) -> Self {
+        Self(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1)
+    }
+
+    pub fn next(&mut self) -> u64 {
+        let mut x = self.0;
+        x ^= x << 13;
+        x ^= x >> 7;
+        x ^= x << 17;
+        self.0 = x;
+        x
+    }
+
+    pub fn below(&mut self, n: usize) -> usize {
+        usize::try_from(self.next() % n as u64).unwrap()
+    }
+
+    pub fn pick<'a, T>(&mut self, items: &'a [T]) -> &'a T {
+        &items[self.below(items.len())]
+    }
+}
diff --git a/packages/rust/tests/foreign_keys.rs b/packages/rust/tests/foreign_keys.rs
index 0f97b8734..f2f6cc7f5 100644
--- a/packages/rust/tests/foreign_keys.rs
+++ b/packages/rust/tests/foreign_keys.rs
@@ -173,6 +173,7 @@ fn a_parent_and_its_child_arriving_in_one_batch_both_land() {
 #[test]
 fn orphaning_a_child_in_watch_reports_it_and_keeps_the_previous_state() {
     let (_dir, root, db) = family(&[("1", "alpha"), ("2", "beta")], &[("kid", "1")]);
+    db.set_fresh_reads(false);
     let parent = root.join("parents").join("1.txt");
     fs::remove_file(&parent).unwrap();
 
@@ -211,6 +212,23 @@ fn orphaning_a_child_in_watch_reports_it_and_keeps_the_previous_state() {
     );
 }
 
+#[test]
+fn a_query_over_an_orphaned_child_fails_as_the_build_would() {
+    let (_dir, root, db) = family(&[("1", "alpha")], &[("kid", "1")]);
+    fs::remove_file(root.join("parents").join("1.txt")).unwrap();
+
+    let message = db.query("SELECT name FROM parent").unwrap_err().to_string();
+
+    for needle in ["child", "parent", "children/kid.txt", "parent_id = 1"] {
+        assert!(
+            message.contains(needle),
+            "{needle:?} missing from: {message}"
+        );
+    }
+    fs::write(root.join("parents").join("1.txt"), "alpha again").unwrap();
+    assert_eq!(names(&db, "SELECT name FROM parent"), vec!["alpha again"]);
+}
+
 #[test]
 fn a_rejected_batch_lands_with_the_batch_that_resolves_it() {
     let (_dir, root, db) = family(&[("1", "alpha"), ("2", "beta")], &[("kid", "1")]);
diff --git a/packages/rust/tests/fresh_declared_cli_e2e.rs b/packages/rust/tests/fresh_declared_cli_e2e.rs
new file mode 100644
index 000000000..1f1a81c1b
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_cli_e2e.rs
@@ -0,0 +1,106 @@
+//! A long-lived `dirsql` process answers a query issued the moment a write
+//! returns with that write: over the server's HTTP query endpoint, and in the
+//! REPL. Real binary, real filesystem, nothing mocked.
+
+#![cfg(feature = "cli")]
+
+#[path = "common/fresh.rs"]
+mod fresh;
+#[path = "common/server.rs"]
+mod server;
+
+use assert_cmd::prelude::*;
+use fresh::Tree;
+use reqwest::blocking::Client;
+use serde_json::{Value, json};
+use std::io::{BufRead, BufReader, Write};
+use std::net::TcpListener;
+use std::process::{Command, Stdio};
+
+const ROUNDS: usize = 300;
+const SQL: &str = "SELECT v FROM recs WHERE k = 'h' ORDER BY v";
+
+fn record(v: &str) -> String {
+    format!("{{\"k\":\"h\",\"v\":\"{v}\"}}")
+}
+
+fn values(rows: &Value) -> Vec<String> {
+    rows.as_array()
+        .unwrap()
+        .iter()
+        .map(|row| row["v"].as_str().unwrap().to_string())
+        .collect()
+}
+
+/// Create, edit and delete a file each round, asking `query` straight after
+/// every write.
+fn create_edit_delete(tree: &Tree, mut query: impl FnMut() -> Vec<String>) {
+    for i in 0..ROUNDS {
+        let rel = format!("h/{i}.json");
+        tree.write(&rel, &record(&format!("c{i}")));
+        assert_eq!(query(), [format!("c{i}")], "create {i}");
+        tree.write(&rel, &record(&format!("e{i}")));
+        assert_eq!(query(), [format!("e{i}")], "edit {i}");
+        tree.remove(&rel);
+        assert!(query().is_empty(), "delete {i}");
+    }
+}
+
+#[test]
+fn the_server_answers_every_query_with_the_write_before_it() {
+    let tree = Tree::new();
+    let port = TcpListener::bind("localhost:0")
+        .unwrap()
+        .local_addr()
+        .unwrap()
+        .port();
+    let config = tree.path(".dirsql.toml");
+    let mut server = server::spawn_server(
+        tree.root(),
+        port,
+        &["--config".as_ref(), config.as_os_str()],
+    );
+    let mut banner = String::new();
+    BufReader::new(server.stdout.take().unwrap())
+        .read_line(&mut banner)
+        .unwrap();
+    assert!(banner.starts_with("Running at"), "{banner:?}");
+    let client = Client::new();
+    let url = format!("http://localhost:{port}/query");
+    create_edit_delete(&tree, || {
+        let rows: Value = client
+            .post(&url)
+            .json(&json!({ "sql": SQL }))
+            .send()
+            .unwrap()
+            .json()
+            .unwrap();
+        values(&rows)
+    });
+}
+
+#[test]
+fn the_repl_answers_every_statement_with_the_write_before_it() {
+    let tree = Tree::new();
+    let mut repl = Command::cargo_bin("dirsql")
+        .unwrap()
+        .arg("--config")
+        .arg(tree.path(".dirsql.toml"))
+        .current_dir(tree.root())
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit())
+        .spawn()
+        .unwrap();
+    let mut stdin = repl.stdin.take().unwrap();
+    let mut stdout = BufReader::new(repl.stdout.take().unwrap());
+    create_edit_delete(&tree, || {
+        writeln!(stdin, "{SQL}").unwrap();
+        stdin.flush().unwrap();
+        let mut line = String::new();
+        stdout.read_line(&mut line).unwrap();
+        values(&serde_json::from_str(&line).unwrap())
+    });
+    drop(stdin);
+    assert!(repl.wait().unwrap().success());
+}
diff --git a/packages/rust/tests/fresh_declared_concurrent.rs b/packages/rust/tests/fresh_declared_concurrent.rs
new file mode 100644
index 000000000..f3d77d9e0
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_concurrent.rs
@@ -0,0 +1,90 @@
+//! A query issued the moment a write returns reads that write while another
+//! thread consumes the instance's events, and a change read before `watch()`
+//! starts still reaches the stream, once.
+
+#[path = "common/fresh.rs"]
+mod fresh;
+
+use dirsql::{DirSQL, RowEvent};
+use fresh::{Tree, lines_of, text};
+use futures_executor::block_on;
+use futures_util::StreamExt;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread;
+use std::time::Duration;
+
+const ROUNDS: usize = 1000;
+const FILES: usize = 200;
+
+fn create_edit_delete(tree: &Tree, db: &DirSQL) {
+    for i in 0..ROUNDS {
+        let rel = format!("h/{i}.txt");
+        tree.write(&rel, &format!("c{i}"));
+        assert_eq!(lines_of(db, &rel), [format!("c{i}")], "create {i}");
+        tree.write(&rel, &format!("e{i}"));
+        assert_eq!(lines_of(db, &rel), [format!("e{i}")], "edit {i}");
+        tree.remove(&rel);
+        assert!(lines_of(db, &rel).is_empty(), "delete {i}");
+    }
+}
+
+#[test]
+fn every_write_is_read_by_the_next_query_while_another_thread_consumes_watch() {
+    let tree = Tree::new();
+    let db = tree.open();
+    let mut stream = db.watch().unwrap();
+    thread::spawn(move || block_on(async { while stream.next().await.is_some() {} }));
+    create_edit_delete(&tree, &db);
+}
+
+#[test]
+fn every_write_is_read_by_the_next_query_while_another_thread_polls_events() {
+    let tree = Tree::new();
+    let db = tree.open();
+    db.start_watching().unwrap();
+    let stop = Arc::new(AtomicBool::new(false));
+    let poller = {
+        let db = db.clone();
+        let stop = Arc::clone(&stop);
+        thread::spawn(move || {
+            while !stop.load(Ordering::SeqCst) {
+                db.poll_events(Duration::from_millis(5)).unwrap();
+            }
+        })
+    };
+    create_edit_delete(&tree, &db);
+    stop.store(true, Ordering::SeqCst);
+    poller.join().unwrap();
+}
+
+#[test]
+fn a_change_read_before_watch_starts_reaches_the_stream_once() {
+    let tree = Tree::new();
+    let db = tree.open();
+    for i in 0..FILES {
+        let rel = format!("g/{i}.txt");
+        tree.write(&rel, &format!("g{i}"));
+        assert_eq!(lines_of(&db, &rel), [format!("g{i}")], "file {i}");
+    }
+    let mut stream = db.watch().unwrap();
+    // Unread by any query, so only the watcher reports it, after every
+    // change before it.
+    tree.write("sentinel.txt", "end");
+    let mut counts = BTreeMap::new();
+    loop {
+        match block_on(stream.next()).unwrap() {
+            RowEvent::Insert { table, row, .. } if table == "notes" => {
+                let line = text(&row["line"]);
+                if line == "end" {
+                    break;
+                }
+                *counts.entry(line).or_insert(0) += 1;
+            }
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+    let expected: BTreeMap<String, usize> = (0..FILES).map(|i| (format!("g{i}"), 1)).collect();
+    assert_eq!(counts, expected);
+}
diff --git a/packages/rust/tests/fresh_declared_edges.rs b/packages/rust/tests/fresh_declared_edges.rs
new file mode 100644
index 000000000..d6dc614f3
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_edges.rs
@@ -0,0 +1,213 @@
+//! Each way a change could slip past a freshness check, read straight after
+//! the change returns.
+
+#[path = "common/fresh.rs"]
+mod fresh;
+
+use fresh::{Tree, assert_matches_fresh_scan, lines_of, snapshot};
+use std::fs;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+#[test]
+fn a_same_size_edit_within_one_mtime_tick_is_read() {
+    let tree = Tree::new();
+    tree.write("tick.txt", "before");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "tick.txt"), ["before"]);
+    let mut same_tick = 0;
+    for i in 0..2000 {
+        let (_, mtime) = tree.times("tick.txt");
+        let content = format!("x{i:05}");
+        tree.write("tick.txt", &content);
+        if tree.times("tick.txt").1 == mtime {
+            same_tick += 1;
+        }
+        assert_eq!(lines_of(&db, "tick.txt"), [content], "round {i}");
+    }
+    eprintln!("{same_tick} of 2000 edits kept the previous mtime");
+}
+
+#[test]
+fn an_edit_whose_times_are_put_back_is_read() {
+    let tree = Tree::new();
+    tree.write("touched.txt", "before");
+    let times = tree.times("touched.txt");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "touched.txt"), ["before"]);
+    tree.write("touched.txt", "after!");
+    tree.restore_times("touched.txt", times);
+    assert_eq!(tree.times("touched.txt").1, times.1);
+    assert_eq!(lines_of(&db, "touched.txt"), ["after!"]);
+}
+
+#[test]
+fn an_atomic_save_over_the_original_is_read() {
+    let tree = Tree::new();
+    tree.write("doc.txt", "one");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "doc.txt"), ["one"]);
+    tree.save_atomically("doc.txt", "two");
+    assert_eq!(lines_of(&db, "doc.txt"), ["two"]);
+}
+
+#[test]
+fn a_file_deleted_and_recreated_under_the_same_name_is_read() {
+    let tree = Tree::new();
+    tree.write("again.txt", "first");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "again.txt"), ["first"]);
+    tree.remove("again.txt");
+    tree.write("again.txt", "secnd");
+    assert_eq!(lines_of(&db, "again.txt"), ["secnd"]);
+    tree.remove("again.txt");
+    assert!(lines_of(&db, "again.txt").is_empty());
+    tree.write("again.txt", "third");
+    assert_eq!(lines_of(&db, "again.txt"), ["third"]);
+}
+
+#[test]
+fn a_file_renamed_into_the_glob_is_read() {
+    let tree = Tree::new();
+    tree.write("incoming.stage", "arrived");
+    let db = tree.open();
+    assert!(lines_of(&db, "incoming.txt").is_empty());
+    tree.rename("incoming.stage", "incoming.txt");
+    assert_eq!(lines_of(&db, "incoming.txt"), ["arrived"]);
+}
+
+#[test]
+fn a_file_renamed_out_of_the_glob_is_dropped() {
+    let tree = Tree::new();
+    tree.write("outgoing.txt", "leaving");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "outgoing.txt"), ["leaving"]);
+    tree.rename("outgoing.txt", "outgoing.stage");
+    assert!(lines_of(&db, "outgoing.txt").is_empty());
+}
+
+#[test]
+fn a_new_subdirectory_is_read() {
+    let tree = Tree::new();
+    let db = tree.open();
+    tree.write("fresh/deep/one.txt", "one");
+    tree.write("fresh/two.txt", "two");
+    assert_eq!(lines_of(&db, "fresh/deep/one.txt"), ["one"]);
+    assert_eq!(lines_of(&db, "fresh/two.txt"), ["two"]);
+    assert_matches_fresh_scan(&db, &tree, "a new subdirectory");
+}
+
+#[test]
+fn a_deleted_subdirectory_is_dropped() {
+    let tree = Tree::new();
+    tree.write("old/deep/one.txt", "one");
+    tree.write("old/two.txt", "two");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "old/two.txt"), ["two"]);
+    fs::remove_dir_all(tree.path("old")).unwrap();
+    assert!(lines_of(&db, "old/deep/one.txt").is_empty());
+    assert!(lines_of(&db, "old/two.txt").is_empty());
+    assert_matches_fresh_scan(&db, &tree, "a deleted subdirectory");
+}
+
+#[test]
+fn a_gitignore_change_changes_which_files_are_read() {
+    let tree = Tree::new();
+    fs::create_dir(tree.path(".git")).unwrap();
+    tree.write("kept.txt", "kept");
+    tree.write("hidden.txt", "hidden");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "hidden.txt"), ["hidden"]);
+    tree.write(".gitignore", "hidden.txt\n");
+    assert!(lines_of(&db, "hidden.txt").is_empty());
+    assert_matches_fresh_scan(&db, &tree, "adding a .gitignore rule");
+    tree.remove(".gitignore");
+    assert_eq!(lines_of(&db, "hidden.txt"), ["hidden"]);
+    assert_matches_fresh_scan(&db, &tree, "removing a .gitignore rule");
+}
+
+#[test]
+fn a_retargeted_symlink_is_read() {
+    let tree = Tree::new();
+    tree.write("targets/one.dat", "one");
+    tree.write("targets/two.dat", "two");
+    tree.link("link.txt", "targets/one.dat");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "link.txt"), ["one"]);
+    tree.link("link.txt", "targets/two.dat");
+    assert_eq!(lines_of(&db, "link.txt"), ["two"]);
+    assert_matches_fresh_scan(&db, &tree, "retargeting a symlink");
+}
+
+#[test]
+fn an_edit_to_a_symlink_target_is_read() {
+    let tree = Tree::new();
+    tree.write("targets/one.dat", "one");
+    tree.link("link.txt", "targets/one.dat");
+    let db = tree.open();
+    assert_eq!(lines_of(&db, "link.txt"), ["one"]);
+    tree.write("targets/one.dat", "uno");
+    assert_eq!(lines_of(&db, "link.txt"), ["uno"]);
+}
+
+fn version(k: usize) -> String {
+    (0..50).map(|_| format!("v{k:06}\n")).collect()
+}
+
+/// A file always holds one whole version on disk, so a query that mixes two
+/// versions, or reads part of one, tore it.
+#[test]
+fn a_file_rewritten_while_queries_run_never_reads_torn_and_ends_exact() {
+    let tree = Arc::new(Tree::new());
+    tree.write("busy.txt", &version(0));
+    let db = tree.open();
+    let done = Arc::new(AtomicBool::new(false));
+    let writer = {
+        let tree = Arc::clone(&tree);
+        let done = Arc::clone(&done);
+        std::thread::spawn(move || {
+            for k in 1..=500 {
+                tree.save_atomically("busy.txt", &version(k));
+            }
+            done.store(true, Ordering::SeqCst);
+        })
+    };
+    while !done.load(Ordering::SeqCst) {
+        let lines = lines_of(&db, "busy.txt");
+        assert_eq!(lines.len(), 50, "a partial version: {lines:?}");
+        assert!(
+            lines.iter().all(|line| line == &lines[0]),
+            "mixed versions: {lines:?}"
+        );
+    }
+    writer.join().unwrap();
+    assert_eq!(lines_of(&db, "busy.txt"), vec!["v000500".to_string(); 50]);
+    assert_matches_fresh_scan(&db, &tree, "the concurrent writer finished");
+}
+
+#[test]
+fn a_file_written_in_place_while_queries_run_ends_exact() {
+    let tree = Arc::new(Tree::new());
+    tree.write("inplace.txt", &version(0));
+    let db = tree.open();
+    let done = Arc::new(AtomicBool::new(false));
+    let writer = {
+        let tree = Arc::clone(&tree);
+        let done = Arc::clone(&done);
+        std::thread::spawn(move || {
+            for k in 1..=500 {
+                tree.write("inplace.txt", &version(k));
+            }
+            done.store(true, Ordering::SeqCst);
+        })
+    };
+    while !done.load(Ordering::SeqCst) {
+        snapshot(&db);
+    }
+    writer.join().unwrap();
+    assert_eq!(
+        lines_of(&db, "inplace.txt"),
+        vec!["v000500".to_string(); 50]
+    );
+    assert_matches_fresh_scan(&db, &tree, "the in-place writer finished");
+}
diff --git a/packages/rust/tests/fresh_declared_events.rs b/packages/rust/tests/fresh_declared_events.rs
new file mode 100644
index 000000000..aa86ea156
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_events.rs
@@ -0,0 +1,77 @@
+//! A change a query reads first still reaches `watch()` and `poll_events()`
+//! consumers, exactly once.
+
+#[path = "common/fresh.rs"]
+mod fresh;
+
+use dirsql::{DirSQL, RowEvent};
+use fresh::{Tree, lines_of, text};
+use std::collections::BTreeMap;
+use std::time::{Duration, Instant};
+
+const FILES: usize = 200;
+
+/// How many inserts each `notes` line got, after events stop arriving.
+fn tally(mut next: impl FnMut() -> Vec<RowEvent>) -> BTreeMap<String, usize> {
+    let mut counts = BTreeMap::new();
+    let deadline = Instant::now() + Duration::from_secs(20);
+    let mut quiet_since = Instant::now();
+    while Instant::now() < deadline {
+        let batch = next();
+        if batch.is_empty() {
+            if counts.len() >= FILES && quiet_since.elapsed() > Duration::from_secs(1) {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+            continue;
+        }
+        quiet_since = Instant::now();
+        for event in batch {
+            match event {
+                RowEvent::Insert { table, row, .. } if table == "notes" => {
+                    *counts.entry(text(&row["line"])).or_insert(0) += 1;
+                }
+                other => panic!("unexpected event: {other:?}"),
+            }
+        }
+    }
+    counts
+}
+
+fn expected() -> BTreeMap<String, usize> {
+    (0..FILES).map(|i| (format!("w{i}"), 1)).collect()
+}
+
+fn write_and_read_each(tree: &Tree, db: &DirSQL) {
+    for i in 0..FILES {
+        let rel = format!("w/{i}.txt");
+        tree.write(&rel, &format!("w{i}"));
+        assert_eq!(lines_of(db, &rel), [format!("w{i}")], "file {i}");
+    }
+}
+
+#[test]
+fn a_watch_stream_gets_each_change_a_query_read_first_once() {
+    let tree = Tree::new();
+    let db = tree.open();
+    let mut stream = db.watch().unwrap();
+    write_and_read_each(&tree, &db);
+    let counts = tally(|| {
+        let mut batch = Vec::new();
+        while let Ok(event) = stream.try_recv() {
+            batch.push(event);
+        }
+        batch
+    });
+    assert_eq!(counts, expected());
+}
+
+#[test]
+fn poll_events_returns_each_change_a_query_read_first_once() {
+    let tree = Tree::new();
+    let db = tree.open();
+    assert!(db.poll_events(Duration::ZERO).unwrap().is_empty());
+    write_and_read_each(&tree, &db);
+    let counts = tally(|| db.poll_events(Duration::from_millis(50)).unwrap());
+    assert_eq!(counts, expected());
+}
diff --git a/packages/rust/tests/fresh_declared_hammer.rs b/packages/rust/tests/fresh_declared_hammer.rs
new file mode 100644
index 000000000..bc8a21fda
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_hammer.rs
@@ -0,0 +1,104 @@
+//! A query issued the moment a write returns reads that write, every time.
+
+#[path = "common/fresh.rs"]
+mod fresh;
+
+use fresh::{Tree, lines_of, recs_of};
+
+const ROUNDS: usize = 2000;
+
+#[test]
+fn every_create_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        let rel = format!("h/{i}.txt");
+        tree.write(&rel, &format!("c{i}"));
+        assert_eq!(lines_of(&db, &rel), [format!("c{i}")], "round {i}");
+        tree.remove(&rel);
+    }
+}
+
+#[test]
+fn every_edit_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    tree.write("e.txt", "start");
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        tree.write("e.txt", &format!("e{i}"));
+        assert_eq!(lines_of(&db, "e.txt"), [format!("e{i}")], "round {i}");
+    }
+}
+
+#[test]
+fn every_delete_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        tree.write("d.txt", &format!("d{i}"));
+        assert_eq!(lines_of(&db, "d.txt"), [format!("d{i}")], "round {i}");
+        tree.remove("d.txt");
+        assert!(lines_of(&db, "d.txt").is_empty(), "round {i}");
+    }
+}
+
+#[test]
+fn every_rename_into_and_out_of_the_glob_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        tree.write("r.stage", &format!("r{i}"));
+        tree.rename("r.stage", "r.txt");
+        assert_eq!(lines_of(&db, "r.txt"), [format!("r{i}")], "round {i}");
+        tree.rename("r.txt", "r.stage");
+        assert!(lines_of(&db, "r.txt").is_empty(), "round {i}");
+    }
+}
+
+#[test]
+fn every_atomic_save_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    tree.write("a.txt", "start");
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        tree.save_atomically("a.txt", &format!("a{i}"));
+        assert_eq!(lines_of(&db, "a.txt"), [format!("a{i}")], "round {i}");
+    }
+}
+
+#[test]
+fn every_same_size_edit_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    tree.write("s.txt", "s------");
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        let content = format!("s{i:06}");
+        tree.write("s.txt", &content);
+        assert_eq!(lines_of(&db, "s.txt"), [content], "round {i}");
+    }
+}
+
+#[test]
+fn every_same_size_edit_with_its_times_restored_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    tree.write("t.txt", "t------");
+    let times = tree.times("t.txt");
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        let content = format!("t{i:06}");
+        tree.write("t.txt", &content);
+        tree.restore_times("t.txt", times);
+        assert_eq!(lines_of(&db, "t.txt"), [content], "round {i}");
+    }
+}
+
+#[test]
+fn every_edit_to_a_table_hook_file_is_read_by_the_next_query() {
+    let tree = Tree::new();
+    tree.write("rec.json", "{\"k\":\"rec\",\"v\":\"start\"}\n");
+    let db = tree.open();
+    for i in 0..ROUNDS {
+        tree.write("rec.json", &format!("{{\"k\":\"rec\",\"v\":\"{i}\"}}\n"));
+        assert_eq!(recs_of(&db, "rec"), [i.to_string()], "round {i}");
+    }
+}
diff --git a/packages/rust/tests/fresh_declared_open_handle.rs b/packages/rust/tests/fresh_declared_open_handle.rs
new file mode 100644
index 000000000..16d68e879
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_open_handle.rs
@@ -0,0 +1,98 @@
+//! A write through a handle that is still open is read by the next query,
+//! whether or not the writer flushed, even when the file had settled and its
+//! last read was trusted.
+
+#[path = "common/fresh.rs"]
+mod fresh;
+
+use fresh::{Tree, lines_of};
+use std::fs::{File, OpenOptions};
+use std::io::{Seek, SeekFrom, Write};
+use std::thread::sleep;
+use std::time::Duration;
+
+const ROUNDS: usize = 300;
+
+/// Past the settle window, so every file's last read is trusted on its stat
+/// alone when the write through the open handle lands.
+const SETTLED: Duration = Duration::from_millis(2500);
+
+/// A same-size rewrite never truncates, so the length never moves.
+fn overwrite(file: &mut File, content: &str, resize: bool, flush: bool) {
+    file.seek(SeekFrom::Start(0)).unwrap();
+    file.write_all(content.as_bytes()).unwrap();
+    if resize {
+        file.set_len(content.len() as u64).unwrap();
+    }
+    if flush {
+        file.sync_all().unwrap();
+    }
+}
+
+fn hammer(
+    resize: bool,
+    flush: bool,
+    before: impl Fn(usize) -> String,
+    after: impl Fn(usize) -> String,
+) {
+    let tree = Tree::new();
+    let rel = |i: usize| format!("o/{i}.txt");
+    for i in 0..ROUNDS {
+        tree.write(&rel(i), &before(i));
+    }
+    let db = tree.open();
+    let mut files: Vec<File> = (0..ROUNDS)
+        .map(|i| {
+            OpenOptions::new()
+                .write(true)
+                .open(tree.path(&rel(i)))
+                .unwrap()
+        })
+        .collect();
+    sleep(SETTLED);
+    for i in 0..ROUNDS {
+        assert_eq!(lines_of(&db, &rel(i)), [before(i)], "file {i} before");
+    }
+    for (i, file) in files.iter_mut().enumerate() {
+        let expected = after(i);
+        overwrite(file, &expected, resize, flush);
+        assert_eq!(lines_of(&db, &rel(i)), [expected], "file {i}");
+    }
+    drop(files);
+}
+
+fn same_size_before(i: usize) -> String {
+    format!("a{i:08}")
+}
+
+fn same_size_after(i: usize) -> String {
+    format!("b{i:08}")
+}
+
+fn short(i: usize) -> String {
+    format!("{i}")
+}
+
+fn long(i: usize) -> String {
+    format!("{i}-{}", "z".repeat(1 + i % 37))
+}
+
+#[test]
+fn a_same_size_rewrite_through_an_open_handle_is_read_by_the_next_query() {
+    hammer(false, false, same_size_before, same_size_after);
+}
+
+#[test]
+fn a_flushed_same_size_rewrite_through_an_open_handle_is_read_by_the_next_query() {
+    hammer(false, true, same_size_before, same_size_after);
+}
+
+#[test]
+fn a_resize_through_an_open_handle_is_read_by_the_next_query() {
+    hammer(true, false, short, long);
+}
+
+#[test]
+fn a_flushed_resize_through_an_open_handle_is_read_by_the_next_query() {
+    hammer(true, true, short, long);
+}
diff --git a/packages/rust/tests/fresh_declared_oracle.rs b/packages/rust/tests/fresh_declared_oracle.rs
new file mode 100644
index 000000000..6052dcb95
--- /dev/null
+++ b/packages/rust/tests/fresh_declared_oracle.rs
@@ -0,0 +1,164 @@
+//! After any sequence of file operations, a long-lived instance reads exactly
+//! what an instance built fresh over the same tree reads.
+
+#[path = "common/fresh.rs"]
+mod fresh;
+
+use fresh::{Rng, Tree, assert_matches_fresh_scan};
+use std::fs;
+use std::time::Duration;
+
+const DIRS: &[&str] = &["", "a/", "a/b/", "c/"];
+const NAMES: &[&str] = &["n0", "n1", "n2", "n3"];
+const EXTS: &[&str] = &["txt", "json", "stage"];
+const IGNORES: &[&str] = &["n0.txt\n", "a/\n", "*.json\n", "c/n1.*\n"];
+
+/// Every file holds JSON lines, so a rename across globs leaves both tables
+/// readable.
+fn content(rng: &mut Rng, rel: &str) -> String {
+    let mut text = String::new();
+    for line in 0..=rng.below(3) {
+        text.push_str(&format!(
+            "{{\"k\":\"{rel}-{line}\",\"v\":\"{:04}\"}}\n",
+            rng.below(10_000)
+        ));
+    }
+    text
+}
+
+fn random_file(rng: &mut Rng) -> String {
+    format!("{}{}.{}", rng.pick(DIRS), rng.pick(NAMES), rng.pick(EXTS))
+}
+
+fn existing_files(tree: &Tree) -> Vec<String> {
+    let mut files = Vec::new();
+    for dir in DIRS {
+        for name in NAMES {
+            for ext in EXTS {
+                let rel = format!("{dir}{name}.{ext}");
+                if tree.path(&rel).is_file() {
+                    files.push(rel);
+                }
+            }
+        }
+    }
+    files
+}
+
+/// Apply one random operation and say what it was.
+fn step(tree: &Tree, rng: &mut Rng) -> String {
+    let existing = existing_files(tree);
+    match rng.below(10) {
+        0 | 1 => {
+            let rel = random_file(rng);
+            let text = content(rng, &rel);
+            tree.write(&rel, &text);
+            format!("write {rel}")
+        }
+        2 if !existing.is_empty() => {
+            let rel = rng.pick(&existing).clone();
+            let old = fs::read_to_string(tree.path(&rel)).unwrap();
+            let times = tree.times(&rel);
+            let flipped: String = old
+                .chars()
+                .map(|c| match c.to_digit(10) {
+                    Some(d) => char::from_digit((d + 1) % 10, 10).unwrap(),
+                    None => c,
+                })
+                .collect();
+            tree.write(&rel, &flipped);
+            tree.restore_times(&rel, times);
+            format!("same-size edit of {rel} with its times restored")
+        }
+        3 => {
+            let rel = random_file(rng);
+            let text = content(rng, &rel);
+            tree.save_atomically(&rel, &text);
+            format!("atomic save of {rel}")
+        }
+        4 if !existing.is_empty() => {
+            let rel = rng.pick(&existing).clone();
+            tree.remove(&rel);
+            format!("delete {rel}")
+        }
+        5 if !existing.is_empty() => {
+            let from = rng.pick(&existing).clone();
+            let to = random_file(rng);
+            if from == to {
+                return "nothing".to_string();
+            }
+            tree.rename(&from, &to);
+            format!("rename {from} to {to}")
+        }
+        6 => {
+            let dir = format!("{}new{}", rng.pick(DIRS), rng.below(3));
+            for name in &NAMES[..2] {
+                let rel = format!("{dir}/{name}.{}", rng.pick(EXTS));
+                let text = content(rng, &rel);
+                tree.write(&rel, &text);
+            }
+            format!("populate new directory {dir}")
+        }
+        7 => {
+            let dir = rng.pick(&DIRS[1..]).trim_end_matches('/').to_string();
+            if tree.path(&dir).is_dir() {
+                fs::remove_dir_all(tree.path(&dir)).unwrap();
+            }
+            format!("remove directory {dir}")
+        }
+        8 => {
+            if rng.below(3) == 0 {
+                let _ = fs::remove_file(tree.path(".gitignore"));
+                "remove .gitignore".to_string()
+            } else {
+                let rule = rng.pick(IGNORES);
+                tree.write(".gitignore", rule);
+                format!("set .gitignore to {rule:?}")
+            }
+        }
+        _ => {
+            let rel = random_file(rng);
+            let text = content(rng, &rel);
+            tree.write(&rel, &text);
+            format!("write {rel}")
+        }
+    }
+}
+
+fn seeded_tree(rng: &mut Rng) -> Tree {
+    let tree = Tree::new();
+    fs::create_dir(tree.path(".git")).unwrap();
+    for _ in 0..8 {
+        let rel = random_file(rng);
+        let text = content(rng, &rel);
+        tree.write(&rel, &text);
+    }
+    tree
+}
+
+#[test]
+fn random_operations_read_what_a_fresh_scan_reads() {
+    for seed in 1..=4 {
+        let mut rng = Rng::new(seed);
+        let tree = seeded_tree(&mut rng);
+        let db = tree.open();
+        for op in 0..150 {
+            let done = step(&tree, &mut rng);
+            assert_matches_fresh_scan(&db, &tree, &format!("seed {seed} op {op}: {done}"));
+        }
+    }
+}
+
+#[test]
+fn a_long_lived_instance_reads_what_a_fresh_scan_reads_across_idle_gaps_and_bursts() {
+    let mut rng = Rng::new(99);
+    let tree = seeded_tree(&mut rng);
+    let db = tree.open();
+    for cycle in 0..12 {
+        std::thread::sleep(Duration::from_millis(250 * (cycle % 4)));
+        for op in 0..25 {
+            let done = step(&tree, &mut rng);
+            assert_matches_fresh_scan(&db, &tree, &format!("cycle {cycle} op {op}: {done}"));
+        }
+    }
+}
diff --git a/packages/rust/tests/persist.rs b/packages/rust/tests/persist.rs
index ae4f1ca59..153aca23b 100644
--- a/packages/rust/tests/persist.rs
+++ b/packages/rust/tests/persist.rs
@@ -378,6 +378,8 @@ fn persist_fans_out_and_composite_key_round_trips() {
     let cb = Arc::new(AtomicUsize::new(0));
     {
         let db = open_two(root.path(), ca.clone(), cb.clone());
+        assert_eq!(ca.swap(0, Ordering::SeqCst), 1, "ta extracted once");
+        assert_eq!(cb.swap(0, Ordering::SeqCst), 1, "tb extracted once");
         assert_eq!(db.query("SELECT col_a FROM ta").unwrap().len(), 1);
         assert_eq!(
             db.query("SELECT col_b FROM tb").unwrap().len(),
@@ -385,8 +387,8 @@ fn persist_fans_out_and_composite_key_round_trips() {
             "second-declared table populated on cold start"
         );
     }
-    assert_eq!(ca.swap(0, Ordering::SeqCst), 1, "ta extracted once");
-    assert_eq!(cb.swap(0, Ordering::SeqCst), 1, "tb extracted once");
+    ca.store(0, Ordering::SeqCst);
+    cb.store(0, Ordering::SeqCst);
 
     // Warm start over the same root/cache: both (rel_path, table) entries are
     // trusted, so neither table re-extracts, yet both still serve the row.
diff --git a/packages/rust/tests/watch_scan_correctness.rs b/packages/rust/tests/watch_scan_correctness.rs
index c7e158e6a..037844144 100644
--- a/packages/rust/tests/watch_scan_correctness.rs
+++ b/packages/rust/tests/watch_scan_correctness.rs
@@ -98,7 +98,7 @@ fn rename_out_deletes_rows() {
 
     let mut saw_delete = false;
     let deadline = Instant::now() + Duration::from_secs(5);
-    while Instant::now() < deadline && paths(&db).iter().any(|p| p == "a.txt") {
+    while Instant::now() < deadline && !saw_delete {
         for e in db.poll_events(Duration::from_millis(200)).unwrap() {
             if matches!(e, RowEvent::Delete { .. }) {
                 saw_delete = true;
@@ -129,6 +129,7 @@ fn dir_moved_into_root_indexes_its_files() {
     fs::write(stage.path().join("moved/a.txt"), "alpha").unwrap();
 
     let db = DirSQL::new(dir.path(), vec![files_table(dir.path())]).unwrap();
+    db.set_fresh_reads(false);
     db.start_watching().unwrap();
     std::thread::sleep(Duration::from_millis(250));
 
@@ -163,6 +164,7 @@ fn dir_moved_into_root_indexes_nested_files() {
     fs::write(stage.path().join("moved/one/two/deep.txt"), "deep").unwrap();
 
     let db = DirSQL::new(dir.path(), vec![files_table(dir.path())]).unwrap();
+    db.set_fresh_reads(false);
     db.start_watching().unwrap();
     std::thread::sleep(Duration::from_millis(250));
 
@@ -227,7 +229,7 @@ fn dir_moved_out_of_root_deletes_its_rows() {
 
     let mut deleted = Vec::new();
     let deadline = Instant::now() + Duration::from_secs(5);
-    while Instant::now() < deadline && !under_leaving(&db).is_empty() {
+    while Instant::now() < deadline && deleted.len() < 2 {
         for e in db.poll_events(Duration::from_millis(200)).unwrap() {
             if let RowEvent::Delete { file_path, .. } = e {
                 deleted.push(file_path);
@@ -279,7 +281,7 @@ fn dir_renamed_within_root_deletes_rows_under_old_name() {
 
     let mut saw_delete = false;
     let deadline = Instant::now() + Duration::from_secs(5);
-    while Instant::now() < deadline && paths(&db).iter().any(|p| p.starts_with("a/")) {
+    while Instant::now() < deadline && !saw_delete {
         for e in db.poll_events(Duration::from_millis(200)).unwrap() {
             if matches!(&e, RowEvent::Delete { file_path, .. } if file_path == "a/x.txt") {
                 saw_delete = true;
diff --git a/packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md b/packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md
new file mode 100644
index 000000000..973df2024
--- /dev/null
+++ b/packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md
@@ -0,0 +1,3 @@
+**Fixed**
+
+`db.query()` and `db.queryOrdered()` now read what is on disk when they are called. Before handing the SQL to the threadpool, the call walks each named table's glob on the JavaScript thread, runs `onFile` on every file whose stat moved or that changed within the last two seconds, and drops the rows of files that are gone, so a write made just before a query is never missed. The row events those re-reads produce reach `watch()` streams once each, including a change read before the first `watch()` call.
diff --git a/packages/ts/e2e-attestations/bot-1572-fresh-red-1104.json b/packages/ts/e2e-attestations/bot-1572-fresh-red-1104.json
new file mode 100644
index 000000000..c5030311a
--- /dev/null
+++ b/packages/ts/e2e-attestations/bot-1572-fresh-red-1104.json
@@ -0,0 +1,7 @@
+{
+  "command": "pnpm test:e2e",
+  "ran_at": 1793815844,
+  "exit_code": 0,
+  "commit": "4127e53ead1544e246125d4dffaff4f79432d2d1",
+  "branch": "bot/1572-fresh-red-1104"
+}
diff --git a/packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md b/packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md
new file mode 100644
index 000000000..f617a9c37
--- /dev/null
+++ b/packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md
@@ -0,0 +1,23 @@
+### Queries read what is on disk
+
+#### Summary
+
+`db.query()` and `db.queryOrdered()` bring the named tables in line with the files on disk before running. No API or configuration changes are required.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+A query re-runs `onFile` on the files that changed since the index last read them, so code that waited on a watch event or slept before querying no longer needs to. A change a query reads first still arrives on `watch()`, once.
+
+The re-reads run on the JavaScript thread, because `onFile` can only run there: each query walks its tables' globs and stats every match before the SQL leaves for the threadpool, about what the same query on a path-table costs. When a file on disk breaks a foreign key, the query rejects with the error a build over that tree returns, until the file is fixed.
+
+#### Verification
+
+Build a `DirSQL` over a directory with a declared table, write a matching file, then `await db.query(...)` the table with no wait; the row for the new file is in the result.
diff --git a/packages/ts/napi/src/lib.rs b/packages/ts/napi/src/lib.rs
index 332a1a2af..4fa3479ca 100644
--- a/packages/ts/napi/src/lib.rs
+++ b/packages/ts/napi/src/lib.rs
@@ -319,7 +319,8 @@ impl DirSQL {
 
     /// Execute a SQL query and return results as an array of objects.
     ///
-    /// Runs on the libuv threadpool so queries don't block the JS event loop.
+    /// Reads changed files on the JS thread, then runs the SQL on the libuv
+    /// threadpool so the event loop stays free while it runs.
     /// Returns a `Promise` in JS. `params` binds to the statement's
     /// placeholders: an array fills `?`, a plain object fills `:name` /
     /// `@name` / `$name`.
@@ -334,7 +335,7 @@ impl DirSQL {
         params: Option<Unknown<'_>>,
     ) -> Result<AsyncTask<QueryTask>> {
         let params = js_to_params(&env, params)?;
-        let inner = self.inner.borrow().as_ref().cloned();
+        let inner = self.read_changes()?;
         Ok(AsyncTask::new(QueryTask { inner, sql, params }))
     }
 
@@ -351,10 +352,21 @@ impl DirSQL {
         params: Option<Unknown<'_>>,
     ) -> Result<AsyncTask<QueryOrderedTask>> {
         let params = js_to_params(&env, params)?;
-        let inner = self.inner.borrow().as_ref().cloned();
+        let inner = self.read_changes()?;
         Ok(AsyncTask::new(QueryOrderedTask { inner, sql, params }))
     }
 
+    /// Reads the files changed since the last query here, on the JS thread,
+    /// because `onFile` can only be called there; the SQL then runs on the
+    /// threadpool.
+    fn read_changes(&self) -> Result<Option<CoreDirSQL>> {
+        let inner = self.inner.borrow().as_ref().cloned();
+        if let Some(db) = &inner {
+            db.read_changes().map_err(to_napi_err)?;
+        }
+        Ok(inner)
+    }
+
     /// Start the file watcher. Must be called before pollEvents. Idempotent.
     ///
     /// Synchronous, so the watcher is live when the call returns: a stream
@@ -468,6 +480,8 @@ impl Task for OpenTask {
 
     fn resolve(&mut self, _env: Env, output: Self::Output) -> Result<Self::JsValue> {
         let inner = CoreDirSQL::finish_build(output).map_err(to_napi_err)?;
+        // A query runs on a worker thread, where `on_file` cannot be called.
+        inner.set_fresh_reads(false);
         Ok(DirSQL {
             inner: RefCell::new(Some(inner)),
         })
@@ -630,8 +644,9 @@ fn required<V: FromNapiValue>(def: &Object<'_>, name: &str) -> Result<V> {
 /// A persistent reference to a JS function, safe to store across calls.
 ///
 /// SAFETY: All access happens on the JS main thread via `#[napi]` methods.
-/// `DirSQL::new` and `DirSQL::pollEvents` both run on that thread, and the
-/// onFile closure is only invoked synchronously within those methods.
+/// `DirSQL::new`, `query`, `queryOrdered`, and `pollEvents`'s resolve run on
+/// that thread, and the onFile closure is only invoked synchronously within
+/// them.
 struct FnRef {
     raw_env: napi::sys::napi_env,
     raw_ref: napi::sys::napi_ref,
diff --git a/packages/ts/src/dirsql.ts b/packages/ts/src/dirsql.ts
index c2b78b1ab..de5adf10f 100644
--- a/packages/ts/src/dirsql.ts
+++ b/packages/ts/src/dirsql.ts
@@ -245,9 +245,10 @@ export class DirSQL {
   /**
    * Execute a SQL query and return results as an array of row objects.
    *
-   * Awaits the initial scan if it has not yet finished, then runs the
-   * query on the libuv threadpool, so the JS event loop stays responsive
-   * even for large result sets or long-running queries.
+   * Awaits the initial scan if it has not yet finished, then reads the
+   * files that changed since the last query, so the rows match the disk.
+   * The query itself runs on the libuv threadpool, so the JS event loop
+   * stays responsive even for large result sets or long-running queries.
    *
    * `params` binds to the statement's placeholders through SQLite, so the
    * values never enter the SQL text:
diff --git a/packages/ts/tests/e2e/fresh-declared-repros.test.ts b/packages/ts/tests/e2e/fresh-declared-repros.test.ts
new file mode 100644
index 000000000..5650ee639
--- /dev/null
+++ b/packages/ts/tests/e2e/fresh-declared-repros.test.ts
@@ -0,0 +1,209 @@
+// E2E: a long-lived `DirSQL` reads what is on disk when each query runs.
+// Regressions for the reported repros: a declared table read with no watch
+// running, a change written between `ready` and `watch()` starting, and a
+// query racing a consumed `watch()`. Every query runs the moment the write
+// before it returns. No mocks: real addon, real filesystem, real SQLite.
+
+import {
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  rmSync,
+  unlinkSync,
+  writeFileSync,
+} from "node:fs";
+import { tmpdir } from "node:os";
+import { basename, dirname, join, relative, sep } from "node:path";
+import { DirSQL } from "dirsql";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+const RACE_ROUNDS = 1000;
+const TEST_TIMEOUT = 120_000;
+
+function turnsDb(root: string): DirSQL {
+  return new DirSQL({
+    root,
+    noIgnore: true,
+    tables: [
+      {
+        name: "turns",
+        ddl: "CREATE TABLE turns (id TEXT PRIMARY KEY)",
+        glob: "turns/*/a.txt",
+        onFile: (filePath: string) => [
+          { id: relative(root, dirname(filePath)).split(sep).join("/") },
+        ],
+      },
+    ],
+  });
+}
+
+function recordsDb(root: string): DirSQL {
+  return new DirSQL({
+    root,
+    tables: [
+      {
+        name: "t",
+        ddl: "CREATE TABLE t (name TEXT, v INTEGER)",
+        glob: "*.json",
+        onFile: (filePath: string) => [
+          {
+            name: basename(filePath),
+            v: JSON.parse(readFileSync(filePath, "utf8")).v,
+          },
+        ],
+      },
+    ],
+  });
+}
+
+function addTurn(root: string, n: number): void {
+  mkdirSync(join(root, "turns", String(n)), { recursive: true });
+  writeFileSync(join(root, "turns", String(n), "a.txt"), "x");
+}
+
+async function turnIds(db: DirSQL): Promise<string[]> {
+  const rows = await db.query("SELECT id FROM turns ORDER BY id");
+  return rows.map((row) => String(row.id));
+}
+
+async function records(db: DirSQL): Promise<[string, number][]> {
+  const rows = await db.query("SELECT name, v FROM t ORDER BY name");
+  return rows.map((row) => [String(row.name), Number(row.v)]);
+}
+
+// Closing the instance ends the stream by rejecting its pending `next()`.
+function consume(db: DirSQL): void {
+  (async () => {
+    for await (const _ of db.watch()) {
+      // drained
+    }
+  })().catch((error: Error) => {
+    if (!error.message.includes("instance closed")) {
+      throw error;
+    }
+  });
+}
+
+describe("a long-lived DirSQL reads what is on disk", () => {
+  let dir: string;
+  let db: DirSQL | undefined;
+
+  beforeEach(() => {
+    dir = mkdtempSync(join(tmpdir(), "dirsql-fresh-repro-"));
+  });
+
+  afterEach(() => {
+    db?.close();
+    db = undefined;
+    rmSync(dir, { recursive: true, force: true });
+  });
+
+  describe("with no watch running", () => {
+    it("reads a file written after ready", async () => {
+      addTurn(dir, 0);
+      db = turnsDb(dir);
+      await db.ready;
+      addTurn(dir, 1);
+      expect(await turnIds(db)).toEqual(["turns/0", "turns/1"]);
+    });
+
+    it("reads a create, an edit and a delete", async () => {
+      writeFileSync(join(dir, "a.json"), JSON.stringify({ v: 1 }));
+      db = recordsDb(dir);
+      await db.ready;
+      expect(await records(db)).toEqual([["a.json", 1]]);
+      writeFileSync(join(dir, "b.json"), JSON.stringify({ v: 2 }));
+      writeFileSync(join(dir, "a.json"), JSON.stringify({ v: 10 }));
+      expect(await records(db)).toEqual([
+        ["a.json", 10],
+        ["b.json", 2],
+      ]);
+      unlinkSync(join(dir, "b.json"));
+      expect(await records(db)).toEqual([["a.json", 10]]);
+    });
+  });
+
+  describe("a change written before watch starts", () => {
+    it(
+      "is read by the next query and streamed once",
+      async () => {
+        addTurn(dir, 0);
+        db = turnsDb(dir);
+        await db.ready;
+        addTurn(dir, 1);
+        expect(await turnIds(db)).toEqual(["turns/0", "turns/1"]);
+        const stream = db.watch()[Symbol.asyncIterator]();
+        // Unread by any query, so only the watcher reports it, after every
+        // change before it.
+        addTurn(dir, 2);
+        const inserted: string[] = [];
+        for (;;) {
+          const next = await stream.next();
+          expect(next.done).toBe(false);
+          expect(next.value.action).toBe("insert");
+          const id = String(next.value.row?.id);
+          inserted.push(id);
+          if (id === "turns/2") {
+            break;
+          }
+        }
+        expect(inserted).toEqual(["turns/1", "turns/2"]);
+        expect(await turnIds(db)).toEqual(["turns/0", "turns/1", "turns/2"]);
+      },
+      TEST_TIMEOUT,
+    );
+  });
+
+  describe("a query racing a consumed watch", () => {
+    it(
+      "reads every write before it",
+      async () => {
+        mkdirSync(join(dir, "turns"));
+        db = turnsDb(dir);
+        await db.ready;
+        consume(db);
+        const misses: number[] = [];
+        for (let n = 0; n < RACE_ROUNDS; n++) {
+          addTurn(dir, n);
+          if ((await db.query("SELECT id FROM turns")).length !== n + 1) {
+            misses.push(n);
+          }
+        }
+        expect(misses).toEqual([]);
+      },
+      TEST_TIMEOUT,
+    );
+
+    it("reads a create", async () => {
+      writeFileSync(join(dir, "a.json"), JSON.stringify({ v: 1 }));
+      db = recordsDb(dir);
+      await db.ready;
+      consume(db);
+      writeFileSync(join(dir, "b.json"), JSON.stringify({ v: 2 }));
+      expect(await records(db)).toEqual([
+        ["a.json", 1],
+        ["b.json", 2],
+      ]);
+    });
+  });
+
+  describe("a path table", () => {
+    it("reads a create, an edit and a delete", async () => {
+      writeFileSync(join(dir, "a.json"), JSON.stringify({ v: 1 }));
+      db = new DirSQL({ root: dir, noIgnore: true });
+      await db.ready;
+      const sql = "SELECT path, content FROM './*.json' ORDER BY path";
+      const read = async () =>
+        (await db?.query(sql))?.map((row) => [row.path, row.content]);
+      expect(await read()).toEqual([["a.json", '{"v":1}']]);
+      writeFileSync(join(dir, "b.json"), JSON.stringify({ v: 2 }));
+      writeFileSync(join(dir, "a.json"), JSON.stringify({ v: 10 }));
+      expect(await read()).toEqual([
+        ["a.json", '{"v":10}'],
+        ["b.json", '{"v":2}'],
+      ]);
+      unlinkSync(join(dir, "b.json"));
+      expect(await read()).toEqual([["a.json", '{"v":10}']]);
+    });
+  });
+});
diff --git a/packages/ts/tests/integration/binding/fresh-declared.test.ts b/packages/ts/tests/integration/binding/fresh-declared.test.ts
new file mode 100644
index 000000000..6470d2e75
--- /dev/null
+++ b/packages/ts/tests/integration/binding/fresh-declared.test.ts
@@ -0,0 +1,511 @@
+import {
+  existsSync,
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  realpathSync,
+  renameSync,
+  rmSync,
+  statSync,
+  symlinkSync,
+  unlinkSync,
+  utimesSync,
+  writeFileSync,
+} from "node:fs";
+import { tmpdir } from "node:os";
+import { dirname, join, sep } from "node:path";
+import { Worker } from "node:worker_threads";
+import { DirSQL, type RowEvent } from "dirsql";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+const ROUNDS = 2000;
+const TEST_TIMEOUT = 120_000;
+
+function notesDb(root: string): DirSQL {
+  const bases = [realpathSync(root), root];
+  return new DirSQL({
+    root,
+    tables: [
+      {
+        name: "notes",
+        ddl: "CREATE TABLE notes (file TEXT, line TEXT)",
+        glob: "**/*.txt",
+        onFile: (filePath: string) => {
+          let rel = filePath;
+          for (const base of bases) {
+            if (filePath.startsWith(base + sep)) {
+              rel = filePath.slice(base.length + 1);
+              break;
+            }
+          }
+          rel = rel.split(sep).join("/");
+          return readFileSync(filePath, "utf8")
+            .split("\n")
+            .filter((line) => line.length > 0)
+            .map((line) => ({ file: rel, line }));
+        },
+      },
+    ],
+  });
+}
+
+async function open(root: string): Promise<DirSQL> {
+  const db = notesDb(root);
+  await db.ready;
+  return db;
+}
+
+function write(root: string, rel: string, content: string): void {
+  const path = join(root, rel);
+  mkdirSync(dirname(path), { recursive: true });
+  writeFileSync(path, content);
+}
+
+function saveAtomically(root: string, rel: string, content: string): void {
+  write(root, `${rel}.stage`, content);
+  renameSync(join(root, `${rel}.stage`), join(root, rel));
+}
+
+function times(root: string, rel: string): [Date, Date] {
+  const st = statSync(join(root, rel));
+  return [st.atime, st.mtime];
+}
+
+function restoreTimes(root: string, rel: string, [atime, mtime]: [Date, Date]) {
+  utimesSync(join(root, rel), atime, mtime);
+}
+
+async function lines(db: DirSQL, rel: string): Promise<string[]> {
+  const rows = await db.query(
+    `SELECT line FROM notes WHERE file = '${rel}' ORDER BY line`,
+  );
+  return rows.map((row) => String(row.line));
+}
+
+async function snapshot(db: DirSQL): Promise<[string, string][]> {
+  const rows = await db.query(
+    "SELECT file, line FROM notes ORDER BY file, line",
+  );
+  return rows.map((row) => [String(row.file), String(row.line)]);
+}
+
+async function expectMatchesFreshScan(
+  db: DirSQL,
+  root: string,
+  context: string,
+): Promise<void> {
+  const fresh = await open(root);
+  try {
+    expect(await snapshot(db), context).toEqual(await snapshot(fresh));
+  } finally {
+    fresh.close();
+  }
+}
+
+function rng(seed: number): (n: number) => number {
+  let state = seed >>> 0;
+  return (n) => {
+    state = (state + 0x6d2b79f5) >>> 0;
+    let t = state;
+    t = Math.imul(t ^ (t >>> 15), t | 1);
+    t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
+    return (((t ^ (t >>> 14)) >>> 0) / 4294967296) * n;
+  };
+}
+
+const DIRS = ["", "a/", "a/b/", "c/"];
+const NAMES = ["n0", "n1", "n2", "n3"];
+const EXTS = ["txt", "json", "stage"];
+const IGNORES = ["n0.txt\n", "a/\n", "*.txt\n", "c/n1.*\n"];
+
+function pick<T>(below: (n: number) => number, items: T[]): T {
+  return items[Math.floor(below(items.length))];
+}
+
+function content(below: (n: number) => number): string {
+  let text = "";
+  const count = 1 + Math.floor(below(3));
+  for (let i = 0; i < count; i++) {
+    text += `line-${String(Math.floor(below(10_000))).padStart(4, "0")}\n`;
+  }
+  return text;
+}
+
+function randomFile(below: (n: number) => number): string {
+  return `${pick(below, DIRS)}${pick(below, NAMES)}.${pick(below, EXTS)}`;
+}
+
+function existing(root: string): string[] {
+  const files: string[] = [];
+  for (const d of DIRS) {
+    for (const n of NAMES) {
+      for (const e of EXTS) {
+        const rel = `${d}${n}.${e}`;
+        if (existsSync(join(root, rel)) && statSync(join(root, rel)).isFile()) {
+          files.push(rel);
+        }
+      }
+    }
+  }
+  return files;
+}
+
+function step(root: string, below: (n: number) => number): string {
+  const files = existing(root);
+  const op = Math.floor(below(9));
+  if (op === 1 && files.length > 0) {
+    const rel = pick(below, files);
+    const before = times(root, rel);
+    const flipped = readFileSync(join(root, rel), "utf8").replace(/\d/g, (d) =>
+      String((Number(d) + 1) % 10),
+    );
+    write(root, rel, flipped);
+    restoreTimes(root, rel, before);
+    return `same-size edit of ${rel} with its times restored`;
+  }
+  if (op === 2) {
+    const rel = randomFile(below);
+    saveAtomically(root, rel, content(below));
+    return `atomic save of ${rel}`;
+  }
+  if (op === 3 && files.length > 0) {
+    const rel = pick(below, files);
+    unlinkSync(join(root, rel));
+    return `delete ${rel}`;
+  }
+  if (op === 4 && files.length > 0) {
+    const from = pick(below, files);
+    const to = randomFile(below);
+    if (from === to) {
+      return "nothing";
+    }
+    mkdirSync(dirname(join(root, to)), { recursive: true });
+    renameSync(join(root, from), join(root, to));
+    return `rename ${from} to ${to}`;
+  }
+  if (op === 5) {
+    const dir = `${pick(below, DIRS)}new${Math.floor(below(3))}`;
+    for (const name of NAMES.slice(0, 2)) {
+      write(root, `${dir}/${name}.${pick(below, EXTS)}`, content(below));
+    }
+    return `populate new directory ${dir}`;
+  }
+  if (op === 6) {
+    const dir = pick(below, DIRS.slice(1)).replace(/\/$/, "");
+    rmSync(join(root, dir), { recursive: true, force: true });
+    return `remove directory ${dir}`;
+  }
+  if (op === 7) {
+    if (Math.floor(below(3)) === 0) {
+      rmSync(join(root, ".gitignore"), { force: true });
+      return "remove .gitignore";
+    }
+    const rule = pick(below, IGNORES);
+    write(root, ".gitignore", rule);
+    return `set .gitignore to ${JSON.stringify(rule)}`;
+  }
+  const rel = randomFile(below);
+  write(root, rel, content(below));
+  return `write ${rel}`;
+}
+
+function seedTree(root: string, below: (n: number) => number): void {
+  mkdirSync(join(root, ".git"));
+  for (let i = 0; i < 8; i++) {
+    write(root, randomFile(below), content(below));
+  }
+}
+
+describe("a declared table reads exactly what is on disk", () => {
+  let dir: string;
+  let db: DirSQL | undefined;
+
+  beforeEach(() => {
+    dir = mkdtempSync(join(tmpdir(), "dirsql-fresh-"));
+    db = undefined;
+  });
+
+  afterEach(() => {
+    db?.close();
+    rmSync(dir, { recursive: true, force: true });
+  });
+
+  describe("a query right after a write", () => {
+    it(
+      "reads every create",
+      async () => {
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          write(dir, `h/${i}.txt`, `c${i}`);
+          expect(await lines(db, `h/${i}.txt`), `round ${i}`).toEqual([
+            `c${i}`,
+          ]);
+          unlinkSync(join(dir, `h/${i}.txt`));
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads every edit",
+      async () => {
+        write(dir, "e.txt", "start");
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          write(dir, "e.txt", `e${i}`);
+          expect(await lines(db, "e.txt"), `round ${i}`).toEqual([`e${i}`]);
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads every delete",
+      async () => {
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          write(dir, "d.txt", `d${i}`);
+          expect(await lines(db, "d.txt"), `round ${i}`).toEqual([`d${i}`]);
+          unlinkSync(join(dir, "d.txt"));
+          expect(await lines(db, "d.txt"), `round ${i}`).toEqual([]);
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads every rename into and out of the glob",
+      async () => {
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          write(dir, "r.stage", `r${i}`);
+          renameSync(join(dir, "r.stage"), join(dir, "r.txt"));
+          expect(await lines(db, "r.txt"), `round ${i}`).toEqual([`r${i}`]);
+          renameSync(join(dir, "r.txt"), join(dir, "r.stage"));
+          expect(await lines(db, "r.txt"), `round ${i}`).toEqual([]);
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads every atomic save",
+      async () => {
+        write(dir, "a.txt", "start");
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          saveAtomically(dir, "a.txt", `a${i}`);
+          expect(await lines(db, "a.txt"), `round ${i}`).toEqual([`a${i}`]);
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads every same-size edit",
+      async () => {
+        write(dir, "s.txt", "s------");
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          const text = `s${String(i).padStart(6, "0")}`;
+          write(dir, "s.txt", text);
+          expect(await lines(db, "s.txt"), `round ${i}`).toEqual([text]);
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads every same-size edit with its times restored",
+      async () => {
+        write(dir, "t.txt", "t------");
+        const before = times(dir, "t.txt");
+        db = await open(dir);
+        for (let i = 0; i < ROUNDS; i++) {
+          const text = `t${String(i).padStart(6, "0")}`;
+          write(dir, "t.txt", text);
+          restoreTimes(dir, "t.txt", before);
+          expect(await lines(db, "t.txt"), `round ${i}`).toEqual([text]);
+        }
+      },
+      TEST_TIMEOUT,
+    );
+  });
+
+  describe("a long-lived instance", () => {
+    it(
+      "reads what a fresh scan reads after random operations",
+      async () => {
+        for (let seed = 1; seed <= 3; seed++) {
+          const root = join(dir, `seed${seed}`);
+          mkdirSync(root);
+          const below = rng(seed);
+          seedTree(root, below);
+          db?.close();
+          db = await open(root);
+          for (let op = 0; op < 100; op++) {
+            const done = step(root, below);
+            await expectMatchesFreshScan(
+              db,
+              root,
+              `seed ${seed} op ${op}: ${done}`,
+            );
+          }
+        }
+      },
+      TEST_TIMEOUT,
+    );
+
+    it(
+      "reads what a fresh scan reads across idle gaps and bursts",
+      async () => {
+        const below = rng(99);
+        seedTree(dir, below);
+        db = await open(dir);
+        for (let cycle = 0; cycle < 10; cycle++) {
+          await new Promise((resolve) =>
+            setTimeout(resolve, 250 * (cycle % 4)),
+          );
+          for (let op = 0; op < 20; op++) {
+            const done = step(dir, below);
+            await expectMatchesFreshScan(
+              db,
+              dir,
+              `cycle ${cycle} op ${op}: ${done}`,
+            );
+          }
+        }
+      },
+      TEST_TIMEOUT,
+    );
+  });
+
+  describe("each way a change could be missed", () => {
+    it("reads an edit whose times are put back", async () => {
+      write(dir, "touched.txt", "before");
+      const before = times(dir, "touched.txt");
+      db = await open(dir);
+      expect(await lines(db, "touched.txt")).toEqual(["before"]);
+      write(dir, "touched.txt", "after!");
+      restoreTimes(dir, "touched.txt", before);
+      expect(await lines(db, "touched.txt")).toEqual(["after!"]);
+    });
+
+    it("reads a file deleted and recreated under the same name", async () => {
+      write(dir, "again.txt", "first");
+      db = await open(dir);
+      expect(await lines(db, "again.txt")).toEqual(["first"]);
+      unlinkSync(join(dir, "again.txt"));
+      write(dir, "again.txt", "secnd");
+      expect(await lines(db, "again.txt")).toEqual(["secnd"]);
+    });
+
+    it("reads a new and a deleted subdirectory", async () => {
+      db = await open(dir);
+      write(dir, "sub/deep/one.txt", "one");
+      expect(await lines(db, "sub/deep/one.txt")).toEqual(["one"]);
+      rmSync(join(dir, "sub"), { recursive: true });
+      expect(await lines(db, "sub/deep/one.txt")).toEqual([]);
+    });
+
+    it("reads a .gitignore change", async () => {
+      mkdirSync(join(dir, ".git"));
+      write(dir, "hidden.txt", "hidden");
+      db = await open(dir);
+      expect(await lines(db, "hidden.txt")).toEqual(["hidden"]);
+      write(dir, ".gitignore", "hidden.txt\n");
+      expect(await lines(db, "hidden.txt")).toEqual([]);
+      unlinkSync(join(dir, ".gitignore"));
+      expect(await lines(db, "hidden.txt")).toEqual(["hidden"]);
+    });
+
+    it("reads a retargeted symlink", async () => {
+      write(dir, "targets/one.dat", "one");
+      write(dir, "targets/two.dat", "two");
+      const link = join(dir, "link.txt");
+      symlinkSync(join(dir, "targets/one.dat"), link);
+      db = await open(dir);
+      expect(await lines(db, "link.txt")).toEqual(["one"]);
+      unlinkSync(link);
+      symlinkSync(join(dir, "targets/two.dat"), link);
+      expect(await lines(db, "link.txt")).toEqual(["two"]);
+    });
+
+    it(
+      "never reads a file torn while it is rewritten, and ends exact",
+      async () => {
+        const version = (k: number) =>
+          `v${String(k).padStart(6, "0")}\n`.repeat(50);
+        write(dir, "busy.txt", version(0));
+        db = await open(dir);
+        const writer = new Worker(
+          `const { renameSync, writeFileSync } = require("node:fs");
+          const { join } = require("node:path");
+          const { workerData } = require("node:worker_threads");
+          for (let k = 1; k <= 300; k++) {
+            const text = ("v" + String(k).padStart(6, "0") + "\\n").repeat(50);
+            writeFileSync(join(workerData, "busy.txt.stage"), text);
+            renameSync(join(workerData, "busy.txt.stage"), join(workerData, "busy.txt"));
+          }`,
+          { eval: true, workerData: dir },
+        );
+        let finished = false;
+        const exited = new Promise<void>((resolve, reject) => {
+          writer.on("error", reject);
+          writer.on("exit", () => {
+            finished = true;
+            resolve();
+          });
+        });
+        while (!finished) {
+          const seen = await lines(db, "busy.txt");
+          expect(seen.length, "a partial version").toBe(50);
+          expect(new Set(seen).size, "mixed versions").toBe(1);
+        }
+        await exited;
+        expect(await lines(db, "busy.txt")).toEqual(
+          version(300)
+            .split("\n")
+            .filter((line) => line.length > 0),
+        );
+      },
+      TEST_TIMEOUT,
+    );
+  });
+
+  describe("watch consumers", () => {
+    it(
+      "stream each change a query read first, once",
+      async () => {
+        db = await open(dir);
+        const stream = db.watch()[Symbol.asyncIterator]();
+        const files = 100;
+        for (let i = 0; i < files; i++) {
+          write(dir, `w/${i}.txt`, `w${i}`);
+          expect(await lines(db, `w/${i}.txt`), `file ${i}`).toEqual([`w${i}`]);
+        }
+        // Unread by any query, so only the watcher reports it, after every
+        // change before it.
+        write(dir, "sentinel.txt", "end");
+        const counts = new Map<string, number>();
+        for (;;) {
+          const next = await stream.next();
+          expect(next.done).toBe(false);
+          const event: RowEvent = next.value;
+          expect(event.action).toBe("insert");
+          const line = String(event.row?.line);
+          if (line === "end") {
+            break;
+          }
+          counts.set(line, (counts.get(line) ?? 0) + 1);
+        }
+        expect(Object.fromEntries(counts)).toEqual(
+          Object.fromEntries(
+            Array.from({ length: files }, (_, i) => [`w${i}`, 1]),
+          ),
+        );
+      },
+      TEST_TIMEOUT,
+    );
+  });
+});
diff --git a/testing-conventions.toml b/testing-conventions.toml
index c4e81d3fa..4d1b28aad 100644
--- a/testing-conventions.toml
+++ b/testing-conventions.toml
@@ -107,6 +107,12 @@ rules = ["mutation"]
 lines = [513, 520]
 reason = "set_terminated_text writes through a raw sqlite3_context, which only a real Connection supplies and unit-lint isolation forbids; tests/vtab.rs (stat_columns_carry_real_values) asserts the full path, basename and ext text"
 
+[[rust.exempt]]
+path = "src/freshness.rs"
+rules = ["mutation"]
+lines = [101, 115, 121, 127, 128, 130, 137, 144]
+reason = "the cfg(windows) and non-unix fingerprint bodies are compiled out on the Linux mutation runner, so no Linux test can reach them; the Windows CI job runs tests/fresh_declared_*.rs, whose t+0 asserts fail if the Windows fingerprint misses a change"
+
 # Unit-only coverage floor: measured by `cargo llvm-cov --lib --features cli`
 # (the `--lib` scoping keeps the integration tier under packages/rust/tests/ out
 # of the number -- testing-conventions #269). `branch` runs under `--branch`,