#!/usr/bin/env python3 """Transactionally truncate one acknowledged session heart-rate capture.""" from __future__ import annotations import argparse import hashlib import json import sqlite3 from pathlib import Path import heart_rate_correction_exchange as correction_exchange from trainlog_sqlite import connect_database class RecoveryError(ValueError): """The requested recovery does not match the persisted factual session.""" def _digest(value: object) -> str: raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") return hashlib.sha256(raw).hexdigest() def _rows(db: sqlite3.Connection, query: str, parameters: tuple = ()) -> list[list]: return [list(row) for row in db.execute(query, parameters)] def protected_snapshot(db: sqlite3.Connection, session_id: str, capture_id: str) -> dict: session = db.execute( "SELECT id FROM sessions WHERE session_id=?", (session_id,), ).fetchone() if session is None: raise RecoveryError("session not found") session_row = session[0] occurrence_rows = [row[0] for row in db.execute( "SELECT id FROM session_exercises WHERE session_row_id=? ORDER BY id", (session_row,), )] occurrence_marks = ",".join("?" for _ in occurrence_rows) or "NULL" protected = { "session": _rows(db, "SELECT * FROM sessions WHERE id=?", (session_row,)), "occurrences": _rows( db, "SELECT * FROM session_exercises WHERE session_row_id=? ORDER BY id", (session_row,), ), "sets": _rows( db, f"SELECT * FROM performed_sets WHERE session_exercise_row_id IN " f"({occurrence_marks}) ORDER BY id", tuple(occurrence_rows), ), "continuous": _rows( db, f"SELECT * FROM continuous_activity WHERE session_exercise_row_id IN " f"({occurrence_marks}) ORDER BY session_exercise_row_id", tuple(occurrence_rows), ), "maxima": _rows( db, f"SELECT * FROM max_results WHERE session_exercise_row_id IN " f"({occurrence_marks}) ORDER BY session_exercise_row_id", tuple(occurrence_rows), ), "other_captures": _rows( db, "SELECT capture_id,context_kind,context_id,started_at,ended_at,sensor_name " "FROM heart_rate_captures WHERE capture_id<>? ORDER BY capture_id", (capture_id,), ), "other_sample_counts": _rows( db, "SELECT capture_id,COUNT(*),MIN(observed_at),MAX(observed_at) " "FROM heart_rate_samples WHERE capture_id<>? GROUP BY capture_id " "ORDER BY capture_id", (capture_id,), ), "other_samples": _rows( db, "SELECT * FROM heart_rate_samples WHERE capture_id<>? " "ORDER BY capture_id,sequence", (capture_id,), ), "other_rr": _rows( db, "SELECT * FROM heart_rate_rr_intervals WHERE capture_id<>? " "ORDER BY capture_id,sample_sequence,rr_index", (capture_id,), ), } if db.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='exercise_feedback'", ).fetchone(): feedback_ids = [row[0] for row in db.execute( f"SELECT feedback_id FROM exercise_feedback WHERE session_exercise_row_id IN " f"({occurrence_marks}) ORDER BY feedback_id", tuple(occurrence_rows), )] feedback_marks = ",".join("?" for _ in feedback_ids) or "NULL" protected["feedback"] = _rows( db, f"SELECT * FROM exercise_feedback WHERE feedback_id IN ({feedback_marks}) " "ORDER BY feedback_id", tuple(feedback_ids), ) protected["feedback_revisions"] = _rows( db, f"SELECT * FROM exercise_feedback_revisions WHERE feedback_id IN " f"({feedback_marks}) ORDER BY feedback_id,revision_id", tuple(feedback_ids), ) if db.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='session_followups'", ).fetchone(): followup_ids = [row[0] for row in db.execute( "SELECT followup_id FROM session_followups WHERE session_row_id=? " "ORDER BY followup_id", (session_row,), )] followup_marks = ",".join("?" for _ in followup_ids) or "NULL" protected["session_followups"] = _rows( db, f"SELECT * FROM session_followups WHERE followup_id IN " f"({followup_marks}) ORDER BY followup_id", tuple(followup_ids), ) protected["session_followup_revisions"] = _rows( db, f"SELECT * FROM session_followup_revisions WHERE followup_id IN " f"({followup_marks}) ORDER BY followup_id,revision_id", tuple(followup_ids), ) if db.execute( "SELECT 1 FROM sqlite_master WHERE type='table' " "AND name='session_timeline_sessions'", ).fetchone(): protected["timeline_session"] = _rows( db, "SELECT * FROM session_timeline_sessions WHERE session_id=?", (session_id,), ) protected["timeline_exercises"] = _rows( db, "SELECT * FROM session_timeline_exercises WHERE session_id=? " "ORDER BY entry_id", (session_id,), ) else: protected["timeline_exercises"] = _rows( db, "SELECT * FROM session_exercise_timeline WHERE session_id=? " "ORDER BY entry_id", (session_id,), ) for table in ("sleep_diary_entries", "sleep_diary_revisions", "sleep_diary_events", "sleep_medications", "sleep_medication_intakes"): if db.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,), ).fetchone(): protected[table] = _rows(db, f"SELECT * FROM {table} ORDER BY rowid") return protected def capture_report(db: sqlite3.Connection, capture_id: str, cutoff: str) -> dict: cutoff_key = correction_exchange.exact_timestamp(cutoff) capture = db.execute( "SELECT context_kind,context_id,started_at,ended_at,sensor_name " "FROM heart_rate_captures WHERE capture_id=?", (capture_id,), ).fetchone() if capture is None: raise RecoveryError("capture not found") samples = db.execute( "SELECT sequence,observed_at,bpm,exercise_entry_id FROM heart_rate_samples " "WHERE capture_id=? ORDER BY sequence", (capture_id,), ).fetchall() keep = [row for row in samples if correction_exchange.exact_timestamp(row[1]) <= cutoff_key] remove = [row for row in samples if correction_exchange.exact_timestamp(row[1]) > cutoff_key] rr = db.execute( "SELECT s.observed_at FROM heart_rate_rr_intervals r " "JOIN heart_rate_samples s ON s.capture_id=r.capture_id " "AND s.sequence=r.sample_sequence WHERE r.capture_id=?", (capture_id,), ).fetchall() rr_keep = sum(correction_exchange.exact_timestamp(row[0]) <= cutoff_key for row in rr) values = [row[2] for row in samples] kept_values = [row[2] for row in keep] def stats(items: list[int]) -> dict: return { "minimum": min(items) if items else None, "average": sum(items) / len(items) if items else None, "maximum": max(items) if items else None, } return { "capture_id": capture_id, "context_kind": capture[0], "context_id": capture[1], "started_at": capture[2], "ended_at": capture[3], "sensor_name": capture[4], "bpm_total": len(samples), "bpm_keep": len(keep), "bpm_remove": len(remove), "rr_total": len(rr), "rr_keep": rr_keep, "rr_remove": len(rr) - rr_keep, "first_removed_at": remove[0][1] if remove else None, "last_removed_at": remove[-1][1] if remove else None, "removed_exercise_entry_ids": sorted({row[3] for row in remove}, key=lambda value: value or ""), "current_stats": stats(values), "kept_stats": stats(kept_values), } def load_operation(path: Path) -> dict: value = json.loads(path.read_text(encoding="utf-8")) if isinstance(value, dict) and set(value) == {"format", "version", "generated_at", "corrections"}: correction_exchange.load(path) if len(value["corrections"]) != 1: raise RecoveryError("correction artifact must contain exactly one operation") return value["corrections"][0] return correction_exchange.validate_correction(value) def recover(db: sqlite3.Connection, session_id: str, capture_id: str, cutoff: str, creator_id: str, apply: bool, operation: dict | None = None, fail_after_effect: bool = False) -> tuple[dict, dict | None]: """Apply or simulate one exact correction without touching sibling facts. CONTRACT: an emitted operation is the immutable content of the separate heart-rate-corrections V1 companion. INVARIANT: every mutation commits together or rolls back. """ db.execute("PRAGMA foreign_keys=ON") correction_exchange.require_schema(db) db.execute("BEGIN IMMEDIATE") emitted = operation try: session = db.execute( "SELECT started_at,ended_at FROM sessions WHERE session_id=?", (session_id,), ).fetchone() if session is None: raise RecoveryError("session not found") if session[1] != cutoff: raise RecoveryError("cutoff does not equal the factual session end") capture = db.execute( "SELECT context_id FROM heart_rate_captures WHERE capture_id=?", (capture_id,), ).fetchone() if capture is None or capture[0] != session_id: raise RecoveryError("capture does not belong to the requested session") before = capture_report(db, capture_id, cutoff) protected_before = protected_snapshot(db, session_id, capture_id) if operation is not None: if operation.get("capture_id") != capture_id or operation.get("cutoff") != cutoff: raise RecoveryError("operation does not match requested capture and cutoff") outcome = correction_exchange.apply_correction(db, operation) elif before["bpm_remove"] == 0 and before["ended_at"] == cutoff: outcome = "unchanged" emitted = None else: emitted, outcome = correction_exchange.local_correction( db, capture_id, cutoff, creator_id, ) if fail_after_effect: raise RecoveryError("injected failure after correction effect") after = capture_report(db, capture_id, cutoff) protected_after = protected_snapshot(db, session_id, capture_id) if protected_before != protected_after: raise RecoveryError("recovery modified protected session or unrelated data") if after["ended_at"] != cutoff or after["bpm_remove"] != 0 or after["rr_remove"] != 0: raise RecoveryError("post-cutoff heart-rate data remain after correction") if db.execute("PRAGMA foreign_key_check").fetchone() is not None: raise RecoveryError("foreign key check failed after correction") if apply: db.commit() else: db.rollback() return { "mode": "apply" if apply else "dry-run", "outcome": outcome, "session_id": session_id, "cutoff": cutoff, "before": before, "after": after, "protected_sha256": _digest(protected_before), "protected_unchanged": True, }, emitted except Exception: db.rollback() raise def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--database", required=True, type=Path) parser.add_argument("--session-id", required=True) parser.add_argument("--capture-id", required=True) parser.add_argument("--cutoff", required=True) parser.add_argument("--creator-id", default="peer_android_recovery") parser.add_argument("--operation-input", type=Path) parser.add_argument("--operation-output", type=Path) parser.add_argument("--report-output", type=Path) parser.add_argument("--apply", action="store_true") args = parser.parse_args() operation = load_operation(args.operation_input) if args.operation_input else None db = connect_database(args.database) try: db.execute("PRAGMA foreign_keys=ON") report, emitted = recover( db, args.session_id, args.capture_id, args.cutoff, args.creator_id, args.apply, operation, ) finally: db.close() if args.operation_output and emitted is not None: args.operation_output.write_text( json.dumps(emitted, ensure_ascii=False, sort_keys=True, separators=(",", ":")), encoding="utf-8", ) rendered = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) if args.report_output: args.report_output.write_text(rendered + "\n", encoding="utf-8") print(rendered) if __name__ == "__main__": try: main() except Exception as error: print(f"HEART_RATE_RECOVERY=FAIL {error}") raise SystemExit(1)