feat(training): record append-only export delivery outcomes

app/services/hf_dataset_service.py

@@ -77,13 +77,15 @@
 import os
 import subprocess
 import threading
+import uuid
 from datetime import datetime
 from pathlib import Path
 
 from packages.ocr_contracts.integrity_ledger import (
     approve_sample,
     compute_content_hash,
     connect_ledger,
+    record_delivery_event,
     record_export_attempt,
 )
 
@@ -169,7 +171,6 @@ def _verify_remote_dataset_privacy() -> bool:
                 "make the dataset private before retrying"
             )
     else:
-        # Recheck the literal source value instead of trusting a mutable boolean.
         if _dataset_private_env is None or _dataset_private_env.strip() != "false":
             raise RuntimeError(
                 "HF export aborted: public visibility requires explicit "
@@ -230,6 +231,28 @@ def _open_integrity_ledger():
     return connect_ledger(str(_LEDGER_FILE))
 
 
+def _record_delivery_outcome(
+    *,
+    event_key: str,
+    correlation_id: str,
+    outcome: str,
+    error_code: str = "",
+) -> None:
+    """Persist a metadata-only delivery observation; never stores payloads or tokens."""
+    conn = _open_integrity_ledger()
+    try:
+        record_delivery_event(
+            conn,
+            event_key=event_key,
+            correlation_id=correlation_id,
+            destination=HF_DATASET,
+            outcome=outcome,
+            error_code=error_code,
+        )
+    finally:
+        conn.close()
+
+
 def _record_row_approval(row: dict) -> str:
     """Record local UI consent bound to the exact staged content; no network I/O."""
     payload = _integrity_payload(row)
@@ -536,18 +559,16 @@ def flush_queue() -> str:
                 "الموافقة أو بصمة المحتوى غير مطابقة. بقيت البيانات محلياً."
             )
 
-        # Verify the remote repository's actual visibility before any dataset
-        # read or write. Never infer privacy from a requested write-visibility
-        # argument or continue when Hub visibility cannot be established.
-        _verify_remote_dataset_privacy()
-
         # Dedup against existing HF dataset (drop rows whose content_hash
         # already exists remotely). This is the only network call that
         # reads the full dataset; it happens at most once per flush, not
         # once per save.
         # A failed remote read is not evidence that the dataset is empty or absent.
         # Fail closed before any write; the operator must make the target dataset
         # readable (or explicitly provision it) before export can proceed.
+        # Verify the actual Hub visibility before any dataset read/write.
+        _verify_remote_dataset_privacy()
+
         try:
             existing = load_dataset(HF_DATASET, split="train")
         except Exception as e:
@@ -641,11 +662,54 @@ def flush_queue() -> str:
                 )
             push_kwargs["token"] = HF_TOKEN
             new_ds = Dataset.from_pandas(df)
-            new_ds.push_to_hub(**push_kwargs)
+            # This ID correlates metadata-only outcome events for this one attempt.
+            # A retry gets a new ID; dedup against the verified remote snapshot remains
+            # the recovery mechanism after an ambiguous network result.
+            delivery_id = uuid.uuid4().hex
+            _record_delivery_outcome(
+                event_key=f"{delivery_id}:push_started",
+                correlation_id=delivery_id,
+                outcome="push_started",
+            )
+            try:
+                new_ds.push_to_hub(**push_kwargs)
+            except Exception as exc:
+                # A client exception cannot prove whether the Hub committed the write.
+                # Record ambiguity rather than claiming a definite remote failure.
+                _record_delivery_outcome(
+                    event_key=f"{delivery_id}:push_ambiguous",
+                    correlation_id=delivery_id,
+                    outcome="push_ambiguous",
+                    error_code=type(exc).__name__,
+                )
+                raise
 
+            _record_delivery_outcome(
+                event_key=f"{delivery_id}:push_succeeded",
+                correlation_id=delivery_id,
+                outcome="push_succeeded",
+            )
             total = len(df)
-            archive_path = _archive_uploaded(eligible)
-            _rewrite_pending_excluding(eligible)
+            try:
+                archive_path = _archive_uploaded(eligible)
+            except Exception as exc:
+                _record_delivery_outcome(
+                    event_key=f"{delivery_id}:archive_failed",
+                    correlation_id=delivery_id,
+                    outcome="archive_failed",
+                    error_code=type(exc).__name__,
+                )
+                raise
+            try:
+                _rewrite_pending_excluding(eligible)
+            except Exception as exc:
+                _record_delivery_outcome(
+                    event_key=f"{delivery_id}:queue_cleanup_failed",
+                    correlation_id=delivery_id,
+                    outcome="queue_cleanup_failed",
+                    error_code=type(exc).__name__,
+                )
+                raise
 
             privacy_note = "خاص (private)" if OMNI_HF_DATASET_PRIVATE else "عام — opt-in صريح"
             logger.info(

packages/ocr_contracts/integrity_ledger.py

@@ -78,6 +78,20 @@ def connect_ledger(database: str | sqlite3.Connection) -> sqlite3.Connection:
         );
         CREATE INDEX IF NOT EXISTS idx_exports_sample
             ON export_events(sample_id, created_at);
+        CREATE TABLE IF NOT EXISTS delivery_events (
+            event_id TEXT PRIMARY KEY,
+            event_key TEXT NOT NULL UNIQUE,
+            correlation_id TEXT NOT NULL,
+            destination TEXT NOT NULL,
+            outcome TEXT NOT NULL CHECK(outcome IN (
+                'push_started', 'push_succeeded', 'push_failed',
+                'push_ambiguous', 'archive_failed', 'queue_cleanup_failed'
+            )),
+            error_code TEXT NOT NULL DEFAULT '',
+            created_at TEXT NOT NULL
+        );
+        CREATE INDEX IF NOT EXISTS idx_delivery_correlation
+            ON delivery_events(correlation_id, created_at);
         CREATE TRIGGER IF NOT EXISTS approvals_no_update
         BEFORE UPDATE ON approvals BEGIN
             SELECT RAISE(ABORT, 'approvals are append-only');
@@ -94,6 +108,14 @@ def connect_ledger(database: str | sqlite3.Connection) -> sqlite3.Connection:
         BEFORE DELETE ON export_events BEGIN
             SELECT RAISE(ABORT, 'export_events are append-only');
         END;
+        CREATE TRIGGER IF NOT EXISTS delivery_events_no_update
+        BEFORE UPDATE ON delivery_events BEGIN
+            SELECT RAISE(ABORT, 'delivery_events are append-only');
+        END;
+        CREATE TRIGGER IF NOT EXISTS delivery_events_no_delete
+        BEFORE DELETE ON delivery_events BEGIN
+            SELECT RAISE(ABORT, 'delivery_events are append-only');
+        END;
     """)
     # Additive migration for ledgers created by earlier releases. Defaults preserve
     # old rows conservatively; old events cannot masquerade as fully specified requests.
@@ -228,5 +250,65 @@ def record_export_attempt(
         conn.rollback()
         raise
 
+def record_delivery_event(
+    conn: sqlite3.Connection,
+    *,
+    event_key: str,
+    correlation_id: str,
+    destination: str,
+    outcome: str,
+    error_code: str = "",
+) -> dict[str, str]:
+    """Append a delivery outcome; this records observations, not exactly-once delivery.
+
+    Reusing an event key with identical content is idempotent. Reusing it with
+    different content is rejected. No medical payload or credential is stored.
+    """
+    allowed_outcomes = {
+        "push_started", "push_succeeded", "push_failed",
+        "push_ambiguous", "archive_failed", "queue_cleanup_failed",
+    }
+    if not event_key.strip() or not correlation_id.strip() or not destination.strip():
+        raise ValueError("event_key, correlation_id, and destination are required")
+    if outcome not in allowed_outcomes:
+        raise ValueError("unsupported delivery outcome")
+    if conn.in_transaction:
+        raise ValueError("ledger connection must not have an active transaction")
+    safe_error_code = "".join(
+        char for char in str(error_code).lower()
+        if char.isalnum() or char in "_-"
+    )[:80]
+    conn.execute("BEGIN IMMEDIATE")
+    try:
+        existing = conn.execute(
+            """SELECT event_id, correlation_id, destination, outcome, error_code
+               FROM delivery_events WHERE event_key = ?""",
+            (event_key,),
+        ).fetchone()
+        requested = (correlation_id, destination, outcome, safe_error_code)
+        if existing:
+            identity = (
+                existing["correlation_id"], existing["destination"],
+                existing["outcome"], existing["error_code"],
+            )
+            if identity != requested:
+                raise ValueError("delivery_event_key_reused_for_different_event")
+            result = {"event_id": existing["event_id"], "outcome": existing["outcome"]}
+            conn.commit()
+            return result
+        event_id = str(uuid.uuid4())
+        conn.execute(
+            """INSERT INTO delivery_events
+               (event_id, event_key, correlation_id, destination, outcome, error_code, created_at)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (event_id, event_key, correlation_id, destination, outcome, safe_error_code, _now()),
+        )
+        conn.commit()
+        return {"event_id": event_id, "outcome": outcome}
+    except Exception:
+        conn.rollback()
+        raise
+
+
 def _now() -> str:
     return datetime.now(timezone.utc).isoformat(timespec="microseconds")

tests/test_hf_dataset_integrity_ledger.py

@@ -16,8 +16,7 @@ def ledger_paths(tmp_path, monkeypatch):
     monkeypatch.setattr(service, "_PENDING_FILE", stage / "pending.jsonl")
     stage.mkdir(parents=True, exist_ok=True)
 
-    # Existing tests must never make live Hub calls. Default fixture models a
-    # verified private dataset; individual privacy tests override this stub.
+    # Keep CI fully offline: model a verified private remote by default.
     class PrivateHubInfo:
         private = True
 
@@ -178,6 +177,16 @@ def push_to_hub(self, **kwargs):
     assert service._PENDING_FILE.exists()
     assert service._read_pending() == [row]
     assert not list(service._UPLOADED_DIR.glob("uploaded-*.jsonl"))
+    conn = connect_ledger(str(ledger_paths))
+    try:
+        outcomes = [
+            item["outcome"] for item in conn.execute(
+                "SELECT outcome FROM delivery_events ORDER BY created_at, rowid"
+            )
+        ]
+        assert outcomes == ["push_started", "push_ambiguous"]
+    finally:
+        conn.close()
 
 
 def test_remote_push_then_archive_failure_recovers_by_remote_dedup(ledger_paths, monkeypatch):
@@ -243,6 +252,16 @@ def fail_archive(rows):
     assert row["content_hash"] in remote_hashes
     assert service._PENDING_FILE.exists()
     assert service._read_pending() == [row]
+    conn = connect_ledger(str(ledger_paths))
+    try:
+        outcomes = [
+            item["outcome"] for item in conn.execute(
+                "SELECT outcome FROM delivery_events ORDER BY created_at, rowid"
+            )
+        ]
+        assert outcomes == ["push_started", "push_succeeded", "archive_failed"]
+    finally:
+        conn.close()
 
     # On restart/retry, the remote hash check deduplicates the already-pushed row
     # and allows archive/queue cleanup without a second remote push.

tests/test_training_integrity_ledger.py

@@ -7,6 +7,7 @@
     approve_sample,
     compute_content_hash,
     connect_ledger,
+    record_delivery_event,
     record_export_attempt,
 )
 
@@ -284,3 +285,29 @@ def test_legacy_export_event_schema_migrates_additively_and_fails_closed(tmp_pat
         assert conn.execute("SELECT COUNT(*) FROM export_events").fetchone()[0] == 2
     finally:
         conn.close()
+
+
+
+def test_delivery_events_are_append_only_and_idempotent():
+    conn = connect_ledger(":memory:")
+    first = record_delivery_event(
+        conn, event_key="attempt-1:push-started", correlation_id="attempt-1",
+        destination="remote-hub", outcome="push_started",
+    )
+    again = record_delivery_event(
+        conn, event_key="attempt-1:push-started", correlation_id="attempt-1",
+        destination="remote-hub", outcome="push_started",
+    )
+    assert first == again
+    assert conn.execute("SELECT COUNT(*) FROM delivery_events").fetchone()[0] == 1
+
+    with pytest.raises(ValueError, match="delivery_event_key_reused_for_different_event"):
+        record_delivery_event(
+            conn, event_key="attempt-1:push-started", correlation_id="attempt-1",
+            destination="remote-hub", outcome="push_succeeded",
+        )
+    with pytest.raises(sqlite3.IntegrityError, match="append-only"):
+        conn.execute("UPDATE delivery_events SET outcome='push_succeeded'")
+    with pytest.raises(sqlite3.IntegrityError, match="append-only"):
+        conn.execute("DELETE FROM delivery_events")
+    conn.close()