.env.example

@@ -36,7 +36,12 @@ ALPHA_VANTAGE_API_KEY=
 DOLT_BINARY=dolt
 DOLT_DATA_DIR=dolt-data
 DOLT_EARNINGS_SUBDIR=earnings
-DOLT_MIN_FREE_DISK_GB=2.0
+# Free-space floor checked before a pull (clone is ~1.7 GB and grows). 5 GB is a
+# safe production default; lower only on a space-constrained dev box.
+DOLT_MIN_FREE_DISK_GB=5.0
+# Hard timeout (s) on each dolt subprocess so a hung pull/sql can't pin the
+# import connection + advisory lock.
+DOLT_COMMAND_TIMEOUT_SECONDS=600.0
 
 # Regime Monitor — FRED (VIX + HY credit spreads). Free key: https://fred.stlouisfed.org/docs/api/api_key.html
 # Optional: without it the volatility (V1) and credit (C1) pillars show as n/a.

app/config.py

@@ -45,7 +45,12 @@ class Settings(BaseSettings):
     dolt_binary: str = "dolt"
     dolt_data_dir: str = "dolt-data"
     dolt_earnings_subdir: str = "earnings"
-    dolt_min_free_disk_gb: float = 2.0
+    # Headroom above the ~1.7 GB earnings clone (grows with pulls); 5 GB is a safe
+    # production floor — override lower only in a space-constrained dev box.
+    dolt_min_free_disk_gb: float = 5.0
+    # Bound every dolt subprocess so a hung pull/sql can't pin the import's
+    # connection + advisory lock indefinitely.
+    dolt_command_timeout_seconds: float = 600.0
 
     # Regime Monitor — FRED (VIX level + HY credit spreads). Optional: without it
     # the volatility (P5) and credit-spread (F2) signals are reported as n/a.

app/services/dolt_client.py

@@ -23,12 +23,19 @@ from pathlib import Path
 
 logger = logging.getLogger(__name__)
 
+# Default subprocess timeout. A hung `dolt pull`/`sql` would otherwise pin the
+# import's connection and its advisory lock indefinitely, so every call is
+# bounded; callers may override per operation.
+DEFAULT_TIMEOUT = 600.0
+
 
 class DoltError(RuntimeError):
-    """A dolt subprocess exited non-zero."""
+    """A dolt subprocess failed, timed out, or exited non-zero."""
 
 
-async def _run(binary: str, args: list[str], *, cwd: Path) -> str:
+async def _run(
+    binary: str, args: list[str], *, cwd: Path, timeout: float = DEFAULT_TIMEOUT
+) -> str:
     proc = await asyncio.create_subprocess_exec(
         binary,
         *args,
@@ -36,7 +43,15 @@ async def _run(binary: str, args: list[str], *, cwd: Path) -> str:
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
-    stdout, stderr = await proc.communicate()
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+    except asyncio.TimeoutError:
+        proc.kill()
+        try:
+            await proc.wait()
+        except ProcessLookupError:
+            pass
+        raise DoltError(f"dolt {args[0] if args else ''} timed out after {timeout:.0f}s")
     if proc.returncode != 0:
         raise DoltError(
             f"dolt {' '.join(args)} failed ({proc.returncode}): "
@@ -60,24 +75,31 @@ def ensure_free_disk(path: Path, min_free_gb: float) -> None:
         )
 
 
-async def pull(repo_dir: Path, *, binary: str) -> None:
+async def pull(repo_dir: Path, *, binary: str, timeout: float = DEFAULT_TIMEOUT) -> None:
     """`dolt pull` the persistent clone to the latest upstream revision."""
-    await _run(binary, ["pull"], cwd=repo_dir)
+    await _run(binary, ["pull"], cwd=repo_dir, timeout=timeout)
 
 
-async def current_commit(repo_dir: Path, *, binary: str) -> str:
-    """The HEAD commit hash of the clone — used as the import revision."""
+async def current_commit(
+    repo_dir: Path, *, binary: str, timeout: float = DEFAULT_TIMEOUT
+) -> str:
+    """The HEAD commit hash of the clone — used as the import revision.
+
+    Uses ``DOLT_HASHOF('HEAD')`` (which formally identifies HEAD) rather than
+    ordering ``dolt_log`` by timestamp."""
     rows = await query_csv(
-        repo_dir, "SELECT commit_hash FROM dolt_log ORDER BY date DESC LIMIT 1", binary=binary
+        repo_dir, "SELECT DOLT_HASHOF('HEAD') AS commit_hash", binary=binary, timeout=timeout
     )
     if not rows or not rows[0].get("commit_hash"):
-        raise DoltError("could not read HEAD commit hash from dolt_log")
+        raise DoltError("could not read HEAD commit hash")
     return rows[0]["commit_hash"]
 
 
-async def query_csv(repo_dir: Path, sql: str, *, binary: str) -> list[dict[str, str]]:
+async def query_csv(
+    repo_dir: Path, sql: str, *, binary: str, timeout: float = DEFAULT_TIMEOUT
+) -> list[dict[str, str]]:
     """Run a read query and parse the CSV result into a list of dict rows."""
-    out = await _run(binary, ["sql", "-q", sql, "-r", "csv"], cwd=repo_dir)
+    out = await _run(binary, ["sql", "-q", sql, "-r", "csv"], cwd=repo_dir, timeout=timeout)
     if not out.strip():
         return []
     return list(csv.DictReader(io.StringIO(out)))

app/services/dolt_earnings_importer.py

@@ -49,6 +49,14 @@ MAX_LEAD_DAYS = 14
 # currently-loaded forward calendar (guards the destructive re-insert against a
 # partial parse / symbol-mapping regression).
 MIN_FUTURE_RATIO = 0.5
+# Initial-load gates (when nothing is loaded yet — the ratio gate has no baseline).
+# The source publishes a forward calendar; require a real horizon, not one stray
+# future row. 21 days is a conservative floor under the ~35d horizon observed on
+# the live clone.
+MIN_FORWARD_HORIZON_DAYS = 21
+# ...and require the symbol join to reach most of the tracked universe, so a
+# broken/normalization-dropped join can't seed a hollow calendar.
+MIN_INITIAL_COVERAGE = 0.5
 
 _CAL_SQL = (
     "SELECT act_symbol, `date`, `when` FROM earnings_calendar "
@@ -96,16 +104,24 @@ class DoltEarningsImporter:
     # -- SourceImporter protocol -------------------------------------------
 
     async def detect_revision(self, db) -> str | None:
+        timeout = settings.dolt_command_timeout_seconds
         if self.do_pull:
             dolt_client.ensure_free_disk(self.repo_dir, settings.dolt_min_free_disk_gb)
-            await self._dolt.pull(self.repo_dir, binary=self.binary)
-        return await self._dolt.current_commit(self.repo_dir, binary=self.binary)
+            await self._dolt.pull(self.repo_dir, binary=self.binary, timeout=timeout)
+        return await self._dolt.current_commit(
+            self.repo_dir, binary=self.binary, timeout=timeout
+        )
 
     async def stage(self, db) -> StagedEarnings:
         universe = await self._load_universe(db)  # {normalised symbol: ticker_id}
 
-        cal_raw = await self._dolt.query_csv(self.repo_dir, _CAL_SQL, binary=self.binary)
-        hist_raw = await self._dolt.query_csv(self.repo_dir, _HIST_SQL, binary=self.binary)
+        timeout = settings.dolt_command_timeout_seconds
+        cal_raw = await self._dolt.query_csv(
+            self.repo_dir, _CAL_SQL, binary=self.binary, timeout=timeout
+        )
+        hist_raw = await self._dolt.query_csv(
+            self.repo_dir, _HIST_SQL, binary=self.binary, timeout=timeout
+        )
         _require_columns(cal_raw, {"act_symbol", "date", "when"}, "earnings_calendar")
         _require_columns(
             hist_raw, {"act_symbol", "period_end_date", "reported", "estimate"}, "eps_history"
@@ -164,13 +180,36 @@ class DoltEarningsImporter:
         )
 
     async def validate(self, db, staged: StagedEarnings) -> ValidationResult:
+        # Promote deletes+reinserts the forward calendar, so this gate is
+        # fail-closed. The forward calendar is the project's acceptance gate.
         messages: list[str] = []
+        current_future = await self._current_future_count(db)
+        universe_size = int(staged.stats.get("universe_size", 0) or 0)
+        coverage = (
+            staged.stats.get("symbols_with_calendar", 0) / universe_size
+            if universe_size
+            else 0.0
+        )
+        horizon_days = (
+            (staged.max_announce_date - self.today).days if staged.max_announce_date else 0
+        )
 
-        # Fail-closed forward-calendar protection (promote deletes+reinserts it).
         if staged.future_count == 0:
             messages.append("no future-dated earnings rows staged")
-        current_future = await self._current_future_count(db)
-        if current_future > 0 and staged.future_count < current_future * MIN_FUTURE_RATIO:
+        elif current_future == 0:
+            # Initial load: no baseline for the ratio gate, so require a real
+            # forward horizon and broad universe coverage instead of one stray row.
+            if horizon_days < MIN_FORWARD_HORIZON_DAYS:
+                messages.append(
+                    f"forward horizon only {horizon_days}d < {MIN_FORWARD_HORIZON_DAYS}d "
+                    "on initial load"
+                )
+            if coverage < MIN_INITIAL_COVERAGE:
+                messages.append(
+                    f"initial universe coverage {coverage:.0%} "
+                    f"< {MIN_INITIAL_COVERAGE:.0%} — symbol join likely broken"
+                )
+        elif staged.future_count < current_future * MIN_FUTURE_RATIO:
             messages.append(
                 f"forward calendar collapsed: staged {staged.future_count} future rows "
                 f"< {MIN_FUTURE_RATIO:.0%} of current {current_future}"
@@ -185,6 +224,8 @@ class DoltEarningsImporter:
             "staged_rows": len(staged.rows),
             "future_rows": staged.future_count,
             "current_future_rows": current_future,
+            "forward_horizon_days": horizon_days,
+            "universe_coverage": round(coverage, 3),
         }
         return ValidationResult(
             ok=not messages,

tests/unit/test_dolt_client.py

@@ -0,0 +1,47 @@
+"""Tests for the async dolt subprocess wrapper's failure handling.
+
+Uses the Python interpreter as a stand-in subprocess (cross-platform, no dolt
+needed) to prove a non-zero exit and a hung command both raise DoltError — the
+latter is what stops a hung pull from pinning the import connection + advisory
+lock forever.
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+from app.services import dolt_client
+from app.services.dolt_client import DoltError
+
+
+async def test_run_raises_on_nonzero_exit():
+    with pytest.raises(DoltError) as exc:
+        await dolt_client._run(
+            sys.executable, ["-c", "import sys; sys.exit(3)"], cwd=Path.cwd(), timeout=30
+        )
+    assert "3" in str(exc.value)
+
+
+async def test_run_times_out_and_kills():
+    start = time.monotonic()
+    with pytest.raises(DoltError) as exc:
+        await dolt_client._run(
+            sys.executable,
+            ["-c", "import time; time.sleep(30)"],
+            cwd=Path.cwd(),
+            timeout=0.5,
+        )
+    elapsed = time.monotonic() - start
+    assert "timed out" in str(exc.value)
+    assert elapsed < 10  # killed promptly, not waited out
+
+
+async def test_run_returns_stdout_on_success():
+    out = await dolt_client._run(
+        sys.executable, ["-c", "print('hello')"], cwd=Path.cwd(), timeout=30
+    )
+    assert out.strip() == "hello"

tests/unit/test_dolt_earnings_importer.py

@@ -57,13 +57,13 @@ class FakeDolt:
         self.commit = commit
         self.pulled = False
 
-    async def pull(self, repo_dir, *, binary):
+    async def pull(self, repo_dir, *, binary, timeout=None):
         self.pulled = True
 
-    async def current_commit(self, repo_dir, *, binary):
+    async def current_commit(self, repo_dir, *, binary, timeout=None):
         return self.commit
 
-    async def query_csv(self, repo_dir, sql, *, binary):
+    async def query_csv(self, repo_dir, sql, *, binary, timeout=None):
         if "earnings_calendar" in sql:
             return self.calendar
         if "eps_history" in sql:
@@ -111,18 +111,18 @@ async def test_stage_and_promote_basic(engine):
     factory = _factory(engine)
     await _seed_tickers(factory, ["AAPL"])
     fake = FakeDolt(
-        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")],
+        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-20")],
         history=[_hist("AAPL", "2026-03-31", 1.5, 1.4)],  # only the reported quarter
     )
     run = await run_import(_importer(fake), engine=engine)
 
     assert run.status == STATUS_PROMOTED
-    assert run.source_max_date == date(2026, 8, 1)
+    assert run.source_max_date == date(2026, 8, 20)
 
     events = await _events(factory)
     assert len(events) == 2
     past = next(e for e in events if e.announce_date == date(2026, 5, 1))
-    future = next(e for e in events if e.announce_date == date(2026, 8, 1))
+    future = next(e for e in events if e.announce_date == date(2026, 8, 20))
     # past announcement paired to the reported quarter
     assert past.eps_actual == 1.5 and past.eps_estimate == 1.4
     assert past.period_end == date(2026, 3, 31) and past.session == "amc"
@@ -136,7 +136,7 @@ async def test_symbol_normalisation_join(engine):
     factory = _factory(engine)
     ids = await _seed_tickers(factory, ["AAPL", "BRK.B"])
     fake = FakeDolt(
-        calendar=[_cal("AAPL", "2026-08-01"), _cal("BRK.B", "2026-08-05")],  # dotted source symbol
+        calendar=[_cal("AAPL", "2026-08-20"), _cal("BRK.B", "2026-08-25")],  # dotted source symbol
         history=[],
     )
     run = await run_import(_importer(fake), engine=engine)
@@ -151,15 +151,15 @@ async def test_reschedule_moves_future_row(engine):
     factory = _factory(engine)
     await _seed_tickers(factory, ["AAPL"])
 
-    fake1 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c1")
+    fake1 = FakeDolt(calendar=[_cal("AAPL", "2026-08-20")], history=[], commit="c1")
     await run_import(_importer(fake1), engine=engine)
 
-    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-08")], history=[], commit="c2")  # moved
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-27")], history=[], commit="c2")  # moved
     run2 = await run_import(_importer(fake2), engine=engine)
     assert run2.status == STATUS_PROMOTED
 
     dates = {e.announce_date for e in await _events(factory)}
-    assert dates == {date(2026, 8, 8)}  # old future date gone, new one present
+    assert dates == {date(2026, 8, 27)}  # old future date gone, new one present
 
 
 async def test_cancellation_removes_future_row(engine):
@@ -183,17 +183,17 @@ async def test_past_row_never_deleted(engine):
     await _seed_tickers(factory, ["AAPL"])
 
     fake1 = FakeDolt(
-        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")], history=[], commit="c1"
+        calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-20")], history=[], commit="c1"
     )
     await run_import(_importer(fake1), engine=engine)
 
     # Second import's calendar omits the past date but keeps a future one.
-    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2")
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-20")], history=[], commit="c2")
     await run_import(_importer(fake2), engine=engine)
 
     dates = {e.announce_date for e in await _events(factory)}
     assert date(2026, 5, 1) in dates  # past result survived
-    assert date(2026, 8, 1) in dates
+    assert date(2026, 8, 20) in dates
 
 
 async def test_validate_fails_when_no_future(engine):
@@ -214,10 +214,10 @@ async def test_validate_fails_on_forward_collapse(engine):
 
     fake1 = FakeDolt(
         calendar=[
-            _cal("AAPL", "2026-08-01"),
-            _cal("MSFT", "2026-08-02"),
-            _cal("NVDA", "2026-08-03"),
-            _cal("AMZN", "2026-08-04"),
+            _cal("AAPL", "2026-08-20"),
+            _cal("MSFT", "2026-08-21"),
+            _cal("NVDA", "2026-08-22"),
+            _cal("AMZN", "2026-08-23"),
         ],
         history=[],
         commit="c1",
@@ -226,7 +226,7 @@ async def test_validate_fails_on_forward_collapse(engine):
     assert len([e for e in await _events(factory) if e.announce_date > TODAY]) == 4
 
     # Only one future row now → 1 < 50% of 4 → fail-closed, no destructive wipe.
-    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2")
+    fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-20")], history=[], commit="c2")
     run2 = await run_import(_importer(fake2), engine=engine)
 
     assert run2.status == STATUS_FAILED
@@ -249,7 +249,9 @@ async def test_real_clone_smoke(engine):
     from app.services import dolt_client
 
     factory = _factory(engine)
-    await _seed_tickers(factory, ["AAPL", "MSFT", "JPM"])
+    # A few tickers spanning near + further-out reporters so the initial-load
+    # forward-horizon gate (>= 21d) is satisfied on the fixed clone.
+    await _seed_tickers(factory, ["AAPL", "MSFT", "NVDA", "JPM", "BRK.B"])
     imp = DoltEarningsImporter(
         repo_dir=_CLONE_DIR, binary=_DOLT_BIN, today=date.today(), do_pull=False, dolt=dolt_client
     )