Skip to content

Commit 0f68ec3

Browse files
hiidkaboutyou-spechiidkaboutyou-spec
committed
feat: add Phase 2 per-source completeness ledger
1 parent be51bf8 commit 0f68ec3

4 files changed

Lines changed: 568 additions & 6 deletions

File tree

‎app/__init__.py‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,15 +24,16 @@
2424
from . import configured_source_runtime as _configured_source_runtime # noqa: F401,E402
2525
from . import configured_source_complete_windows as _configured_source_complete_windows # noqa: F401,E402
2626

27-
# Phase 3 extends the already-authoritative source collector with bounded retries and
28-
# durable provider-cursor checkpoints. Install/harden it before Phase 2 so lifecycle
29-
# and Zero-Silent-Miss observability wrap the final resumable retrieval behavior.
27+
# Existing recovery remains responsible for provider pagination/checkpoints/retries.
3028
from . import phase3_recovery as _phase3_recovery # noqa: F401,E402
3129
from . import phase3_recovery_hardening as _phase3_recovery_hardening # noqa: F401,E402
3230

33-
# Phase 2 is intentionally installed last so it observes the exact Phase 1/Phase 3
34-
# source-authority/runtime behavior rather than replacing it. It adds lifecycle,
35-
# correlation, failure visibility, quarantine records and privacy-safe telemetry only.
31+
# Product Roadmap Phase 2 records the final completeness result source-by-source.
32+
# Import after recovery so the ledger wraps the effective resumable collector rather
33+
# than an obsolete method. COMPLETE advances only that source's durable watermark.
34+
from . import source_ledger_runtime as _source_ledger_runtime # noqa: F401,E402
35+
36+
# Legacy observability stays downstream and remains compatibility-only.
3637
from . import zero_silent_miss as _zero_silent_miss # noqa: F401,E402
3738
from . import phase2_runtime_compat as _phase2_runtime_compat # noqa: F401,E402
3839
from . import phase2_final_visibility as _phase2_final_visibility # noqa: F401,E402

‎app/source_ledger.py‎

Lines changed: 327 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,327 @@
1+
from __future__ import annotations
2+
3+
import sqlite3
4+
from dataclasses import dataclass
5+
from datetime import datetime, timezone
6+
from enum import Enum
7+
from pathlib import Path
8+
from typing import Any
9+
10+
SCHEMA_VERSION = 1
11+
12+
13+
class SourceWindowStatus(str, Enum):
14+
ATTEMPTING = "attempting"
15+
COMPLETE = "complete"
16+
PARTIAL = "partial"
17+
UNPROVEN = "unproven"
18+
19+
20+
class SourceLedgerError(RuntimeError):
21+
"""Per-source completeness truth could not be persisted safely."""
22+
23+
24+
def _utc_now() -> str:
25+
return datetime.now(timezone.utc).isoformat()
26+
27+
28+
def _norm_handle(value: str) -> str:
29+
return str(value or "").lstrip("@").strip().casefold()
30+
31+
32+
@dataclass(frozen=True, slots=True)
33+
class SourceWindowResult:
34+
source_handle: str
35+
window_start: str
36+
window_end: str
37+
status: SourceWindowStatus
38+
attempt_id: str = ""
39+
raw_observation_count: int = 0
40+
retained_count: int = 0
41+
retry_count: int = 0
42+
error_class: str = ""
43+
error_summary: str = ""
44+
provider_cursor: str = ""
45+
proof_kind: str = ""
46+
47+
def __post_init__(self) -> None:
48+
handle = _norm_handle(self.source_handle)
49+
if not handle or not self.window_start or not self.window_end:
50+
raise SourceLedgerError("Source ledger rows require source and window bounds.")
51+
object.__setattr__(self, "source_handle", handle)
52+
53+
54+
class SourceLedgerStore:
55+
"""Durable source-by-source completeness, retry and cursor ledger.
56+
57+
A COMPLETE row may advance only that source's durable watermark. PARTIAL and
58+
UNPROVEN rows are recorded but never advance it. This makes one failing source
59+
independent from all other configured sources.
60+
"""
61+
62+
def __init__(self, path: Path) -> None:
63+
self.path = Path(path)
64+
self.path.parent.mkdir(parents=True, exist_ok=True)
65+
try:
66+
self.conn = sqlite3.connect(self.path, timeout=15)
67+
self.conn.row_factory = sqlite3.Row
68+
self.conn.execute("PRAGMA journal_mode=WAL")
69+
self.conn.execute("PRAGMA foreign_keys=ON")
70+
self._init_schema()
71+
except sqlite3.Error as exc:
72+
raise SourceLedgerError(
73+
f"Could not initialize source ledger: {type(exc).__name__}"
74+
) from exc
75+
76+
def _init_schema(self) -> None:
77+
try:
78+
self.conn.executescript(
79+
"""
80+
CREATE TABLE IF NOT EXISTS source_ledger_meta (
81+
key TEXT PRIMARY KEY,
82+
value TEXT NOT NULL
83+
);
84+
85+
CREATE TABLE IF NOT EXISTS source_windows (
86+
source_handle TEXT NOT NULL,
87+
window_start TEXT NOT NULL,
88+
window_end TEXT NOT NULL,
89+
status TEXT NOT NULL,
90+
attempt_id TEXT NOT NULL DEFAULT '',
91+
raw_observation_count INTEGER NOT NULL DEFAULT 0,
92+
retained_count INTEGER NOT NULL DEFAULT 0,
93+
retry_count INTEGER NOT NULL DEFAULT 0,
94+
error_class TEXT NOT NULL DEFAULT '',
95+
error_summary TEXT NOT NULL DEFAULT '',
96+
provider_cursor TEXT NOT NULL DEFAULT '',
97+
proof_kind TEXT NOT NULL DEFAULT '',
98+
first_attempted_at TEXT NOT NULL,
99+
last_attempted_at TEXT NOT NULL,
100+
completed_at TEXT NOT NULL DEFAULT '',
101+
attempt_count INTEGER NOT NULL DEFAULT 1,
102+
PRIMARY KEY(source_handle, window_start, window_end)
103+
);
104+
105+
CREATE INDEX IF NOT EXISTS source_windows_status_idx
106+
ON source_windows(status, last_attempted_at);
107+
108+
CREATE INDEX IF NOT EXISTS source_windows_source_idx
109+
ON source_windows(source_handle, window_end);
110+
111+
CREATE TABLE IF NOT EXISTS source_cursors (
112+
source_handle TEXT PRIMARY KEY,
113+
complete_through TEXT NOT NULL DEFAULT '',
114+
provider_cursor TEXT NOT NULL DEFAULT '',
115+
last_complete_window_start TEXT NOT NULL DEFAULT '',
116+
last_complete_window_end TEXT NOT NULL DEFAULT '',
117+
last_status TEXT NOT NULL DEFAULT '',
118+
last_attempt_id TEXT NOT NULL DEFAULT '',
119+
last_error_class TEXT NOT NULL DEFAULT '',
120+
last_error_summary TEXT NOT NULL DEFAULT '',
121+
updated_at TEXT NOT NULL
122+
);
123+
"""
124+
)
125+
self.conn.execute(
126+
"""
127+
INSERT INTO source_ledger_meta(key, value)
128+
VALUES('schema_version', ?)
129+
ON CONFLICT(key) DO UPDATE SET value=excluded.value
130+
""",
131+
(str(SCHEMA_VERSION),),
132+
)
133+
self.conn.commit()
134+
except sqlite3.Error as exc:
135+
raise SourceLedgerError(
136+
f"Could not create source ledger schema: {type(exc).__name__}"
137+
) from exc
138+
139+
def start_attempt(
140+
self,
141+
*,
142+
source_handle: str,
143+
window_start: str,
144+
window_end: str,
145+
attempt_id: str = "",
146+
) -> None:
147+
now = _utc_now()
148+
handle = _norm_handle(source_handle)
149+
if not handle:
150+
raise SourceLedgerError("Source handle is required.")
151+
try:
152+
with self.conn:
153+
self.conn.execute(
154+
"""
155+
INSERT INTO source_windows(
156+
source_handle, window_start, window_end, status,
157+
attempt_id, first_attempted_at, last_attempted_at
158+
) VALUES(?,?,?,?,?,?,?)
159+
ON CONFLICT(source_handle, window_start, window_end) DO UPDATE SET
160+
status=excluded.status,
161+
attempt_id=excluded.attempt_id,
162+
last_attempted_at=excluded.last_attempted_at,
163+
attempt_count=source_windows.attempt_count + 1,
164+
retry_count=source_windows.retry_count + 1,
165+
error_class='',
166+
error_summary=''
167+
""",
168+
(
169+
handle,
170+
str(window_start),
171+
str(window_end),
172+
SourceWindowStatus.ATTEMPTING.value,
173+
str(attempt_id or ""),
174+
now,
175+
now,
176+
),
177+
)
178+
self.conn.execute(
179+
"""
180+
INSERT INTO source_cursors(
181+
source_handle, last_status, last_attempt_id, updated_at
182+
) VALUES(?,?,?,?)
183+
ON CONFLICT(source_handle) DO UPDATE SET
184+
last_status=excluded.last_status,
185+
last_attempt_id=excluded.last_attempt_id,
186+
updated_at=excluded.updated_at
187+
""",
188+
(handle, SourceWindowStatus.ATTEMPTING.value, str(attempt_id or ""), now),
189+
)
190+
except sqlite3.Error as exc:
191+
raise SourceLedgerError(
192+
f"Could not start source ledger attempt: {type(exc).__name__}"
193+
) from exc
194+
195+
def finish(self, result: SourceWindowResult) -> None:
196+
now = _utc_now()
197+
complete = result.status is SourceWindowStatus.COMPLETE
198+
completed_at = now if complete else ""
199+
try:
200+
with self.conn:
201+
self.conn.execute(
202+
"""
203+
INSERT INTO source_windows(
204+
source_handle, window_start, window_end, status, attempt_id,
205+
raw_observation_count, retained_count, retry_count,
206+
error_class, error_summary, provider_cursor, proof_kind,
207+
first_attempted_at, last_attempted_at, completed_at
208+
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
209+
ON CONFLICT(source_handle, window_start, window_end) DO UPDATE SET
210+
status=excluded.status,
211+
attempt_id=excluded.attempt_id,
212+
raw_observation_count=excluded.raw_observation_count,
213+
retained_count=excluded.retained_count,
214+
retry_count=MAX(source_windows.retry_count, excluded.retry_count),
215+
error_class=excluded.error_class,
216+
error_summary=excluded.error_summary,
217+
provider_cursor=excluded.provider_cursor,
218+
proof_kind=excluded.proof_kind,
219+
last_attempted_at=excluded.last_attempted_at,
220+
completed_at=excluded.completed_at
221+
""",
222+
(
223+
result.source_handle,
224+
result.window_start,
225+
result.window_end,
226+
result.status.value,
227+
result.attempt_id,
228+
max(0, int(result.raw_observation_count)),
229+
max(0, int(result.retained_count)),
230+
max(0, int(result.retry_count)),
231+
str(result.error_class or "")[:160],
232+
str(result.error_summary or "")[:1000],
233+
str(result.provider_cursor or "")[:4096],
234+
str(result.proof_kind or "")[:120],
235+
now,
236+
now,
237+
completed_at,
238+
),
239+
)
240+
241+
if complete:
242+
self.conn.execute(
243+
"""
244+
INSERT INTO source_cursors(
245+
source_handle, complete_through, provider_cursor,
246+
last_complete_window_start, last_complete_window_end,
247+
last_status, last_attempt_id, last_error_class,
248+
last_error_summary, updated_at
249+
) VALUES(?,?,?,?,?,?,?,?,?,?)
250+
ON CONFLICT(source_handle) DO UPDATE SET
251+
complete_through=CASE
252+
WHEN source_cursors.complete_through='' OR excluded.complete_through > source_cursors.complete_through
253+
THEN excluded.complete_through ELSE source_cursors.complete_through END,
254+
provider_cursor=excluded.provider_cursor,
255+
last_complete_window_start=excluded.last_complete_window_start,
256+
last_complete_window_end=excluded.last_complete_window_end,
257+
last_status=excluded.last_status,
258+
last_attempt_id=excluded.last_attempt_id,
259+
last_error_class='',
260+
last_error_summary='',
261+
updated_at=excluded.updated_at
262+
""",
263+
(
264+
result.source_handle,
265+
result.window_end,
266+
result.provider_cursor,
267+
result.window_start,
268+
result.window_end,
269+
result.status.value,
270+
result.attempt_id,
271+
"",
272+
"",
273+
now,
274+
),
275+
)
276+
else:
277+
self.conn.execute(
278+
"""
279+
INSERT INTO source_cursors(
280+
source_handle, last_status, last_attempt_id,
281+
last_error_class, last_error_summary, updated_at
282+
) VALUES(?,?,?,?,?,?)
283+
ON CONFLICT(source_handle) DO UPDATE SET
284+
last_status=excluded.last_status,
285+
last_attempt_id=excluded.last_attempt_id,
286+
last_error_class=excluded.last_error_class,
287+
last_error_summary=excluded.last_error_summary,
288+
updated_at=excluded.updated_at
289+
""",
290+
(
291+
result.source_handle,
292+
result.status.value,
293+
result.attempt_id,
294+
str(result.error_class or "")[:160],
295+
str(result.error_summary or "")[:1000],
296+
now,
297+
),
298+
)
299+
except sqlite3.Error as exc:
300+
raise SourceLedgerError(
301+
f"Could not finish source ledger result: {type(exc).__name__}"
302+
) from exc
303+
304+
def cursor(self, source_handle: str) -> dict[str, Any] | None:
305+
row = self.conn.execute(
306+
"SELECT * FROM source_cursors WHERE source_handle=?",
307+
(_norm_handle(source_handle),),
308+
).fetchone()
309+
return dict(row) if row is not None else None
310+
311+
def window(self, source_handle: str, window_start: str, window_end: str) -> dict[str, Any] | None:
312+
row = self.conn.execute(
313+
"""
314+
SELECT * FROM source_windows
315+
WHERE source_handle=? AND window_start=? AND window_end=?
316+
""",
317+
(_norm_handle(source_handle), str(window_start), str(window_end)),
318+
).fetchone()
319+
return dict(row) if row is not None else None
320+
321+
def source_statuses(self) -> list[dict[str, Any]]:
322+
return [dict(row) for row in self.conn.execute(
323+
"SELECT * FROM source_cursors ORDER BY source_handle"
324+
).fetchall()]
325+
326+
def close(self) -> None:
327+
self.conn.close()

0 commit comments

Comments
 (0)