Reserve media proposals across concurrent requests

README.md

@@ -54,11 +54,58 @@ factor separately does not validate their Cartesian product or timing.
   and retain their raw data elsewhere; this GP assumes a common noise scale.
 - With fewer than two observations the selection is random and reproducible.
   Otherwise it uses expected improvement with fixed GP hyperparameters.
-- Record the proposal in the pending table **before** asking again. The planner
-  is stateless and does not reserve conditions or coordinate concurrent users.
+- Without a reservation ledger, record the proposal in the pending table
+  **before** asking again. This preview mode does not reserve conditions or
+  coordinate concurrent users.
 - On completion, remove the pending ID and add its observed response.
 
 Unknown IDs, duplicate conditions, nonfinite values, out-of-box factors and an
 exhausted candidate set are rejected. Output files are created exclusively.
 `--noise` is an assumed response standard deviation, not an estimated noise
 model. The fixed GP is a baseline and its uncertainty is not calibrated on cells.
+
+## Reserve a candidate atomically
+
+Use one shared **local** SQLite ledger for all callers that need reservations:
+
+```bash
+medialoop-plan --candidates examples/candidates.csv \
+  --observations examples/observations.csv --pending examples/pending.csv \
+  --reserve-ledger artifacts/reservations.sqlite3 --request-id round-001 \
+  --seed 0 --out artifacts/round-001.json
+```
+
+Selection and reservation commit in a single transaction before the proposal
+is returned. Concurrent processes using that ledger cannot reserve the same
+condition under different request IDs. Existing observed/pending CSV entries
+are also excluded. The command reads those CSVs and writes the ledger and
+proposal JSON; it never edits the CSVs. Input hashes describe the exact bytes
+parsed for the original proposal.
+
+Choose a new `--request-id` for each new request. Retry with the **same** ID,
+candidate table, seed and noise to recover its original proposal, even after
+observations change. This also recovers a committed reservation after output
+export fails or the caller loses its connection. Retry with a new output path
+if the previous export is partial or contains different text; an existing exact
+export is accepted. A retry returns a historical proposal and does not reserve
+another condition. The Python equivalent is
+`medialoop.reservations.propose_and_reserve(..., ledger_path=..., request_id=...)`;
+`medialoop.planner.propose(...)` remains a read-only preview.
+
+The first successful reservation binds the ledger to the candidate CSV's exact
+SHA-256. Keep that table unchanged. Reservations remain recorded after their
+responses are added to the observations CSV, so completed conditions cannot
+be issued again. There is intentionally no release/reissue command. Do not
+delete, replace or copy the ledger to start another round of the same campaign;
+doing so loses coordination. If a condition was also listed in a manual pending
+CSV, remove that entry when adding its response to observations.
+
+SQLite releases locks and rolls back uncommitted writes when a process exits.
+Callers wait up to 30 seconds for a writer before returning an error; retry with
+the same request ID. Use a local filesystem with reliable SQLite locking, not
+network shares or cloud-synced copies. All reserving callers must use this API
+and ledger. Manual CSV updates are outside the transaction: pause submissions
+while updating observations/pending files, then resume with new request IDs.
+Preview mode can still display already-reserved conditions because it does not
+read the ledger. Store the ledger alongside its recovery journal files outside
+Git and use SQLite-aware backups while it is active.

src/medialoop/planner.py

@@ -9,6 +9,7 @@
 import argparse
 import csv
 import hashlib
+import io
 import json
 from pathlib import Path
 
@@ -20,7 +21,10 @@
 
 
 def _read(path, required):
-    with Path(path).open(newline="", encoding="utf-8-sig") as handle:
+    # Parse and hash the same snapshot, including when another process replaces
+    # an observations CSV while this request is being prepared.
+    raw = Path(path).read_bytes()
+    with io.StringIO(raw.decode("utf-8-sig"), newline="") as handle:
         reader = csv.DictReader(handle)
         header = reader.fieldnames or []
         if len(set(header)) != len(header) or not set(required).issubset(header):
@@ -36,13 +40,25 @@ def _read(path, required):
     ids = [row["candidate_id"] for row in rows]
     if len(ids) != len(set(ids)):
         raise ValueError(f"{path}: duplicate candidate_id (aggregate replicates explicitly)")
-    return rows
+    return rows, hashlib.sha256(raw).hexdigest()
+
+
+def _inputs(candidates_path, observations_path, pending_path):
+    rows, candidates_hash = _read(candidates_path, ["candidate_id", *NAMES])
+    observations, observations_hash = _read(observations_path, ["candidate_id", "response"]) if observations_path else ([], None)
+    pending, pending_hash = _read(pending_path, ["candidate_id"]) if pending_path else ([], None)
+    return rows, observations, pending, {"candidates": candidates_hash,
+                                        "observations": observations_hash, "pending": pending_hash}
 
 
 def propose(candidates_path, observations_path=None, pending_path=None, *, seed=0, noise=0.02):
+    """Preview one candidate without reserving it or modifying input files."""
+    return _propose(*_inputs(candidates_path, observations_path, pending_path), seed=seed, noise=noise)
+
+
+def _propose(rows, observations, pending, hashes, *, seed, noise, reserved_ids=()):
     if not np.isfinite(noise) or noise < 0:
         raise ValueError("noise must be finite and nonnegative")
-    rows = _read(candidates_path, ["candidate_id", *NAMES])
     if not rows:
         raise ValueError("empty candidate list")
     x = np.array([[float(row[name]) for name in NAMES] for row in rows])
@@ -52,14 +68,13 @@ def propose(candidates_path, observations_path=None, pending_path=None, *, seed=
         raise ValueError("duplicate factor combinations; give each condition one candidate_id")
     ids = [row["candidate_id"] for row in rows]
     lookup = {value: i for i, value in enumerate(ids)}
-    observations = _read(observations_path, ["candidate_id", "response"]) if observations_path else []
-    pending = _read(pending_path, ["candidate_id"]) if pending_path else []
     observed_ids = {row["candidate_id"] for row in observations}
     pending_ids = {row["candidate_id"] for row in pending}
     if (observed_ids | pending_ids) - set(ids):
         raise ValueError("ledger contains an unknown candidate_id")
     if observed_ids & pending_ids:
         raise ValueError("candidate cannot be both observed and pending")
+    pending_ids |= set(reserved_ids) - observed_ids
     y = np.array([float(row["response"]) for row in observations])
     if not np.isfinite(y).all():
         raise ValueError("responses must be finite")
@@ -81,13 +96,10 @@ def propose(candidates_path, observations_path=None, pending_path=None, *, seed=
         chosen = available[local]
         mu, sigma, acquisition = float(means[local]), float(stds[local]), float(ei[local])
         method = "expected_improvement"
-    hashes = {name: None if path is None else hashlib.sha256(Path(path).read_bytes()).hexdigest()
-              for name, path in [("candidates", candidates_path), ("observations", observations_path),
-                                 ("pending", pending_path)]}
     return {"schema_version": 1, "candidate_id": ids[chosen], "factors": dict(zip(NAMES, x[chosen].tolist())),
             "method": method, "predicted_mean": mu, "predicted_std": sigma,
             "expected_improvement": acquisition, "seed": seed, "noise": noise,
-            "n_observed": len(observations), "n_pending": len(pending), "input_sha256": hashes,
+            "n_observed": len(observations), "n_pending": len(pending_ids), "input_sha256": hashes,
             "note": "Maximizes the supplied response. A proposal is not an executable protocol. Record it as pending before asking again."}
 
 
@@ -96,17 +108,33 @@ def main(argv=None):
     parser.add_argument("--candidates", required=True)
     parser.add_argument("--observations")
     parser.add_argument("--pending")
+    parser.add_argument("--reserve-ledger", help="atomically reserve in this local SQLite ledger")
+    parser.add_argument("--request-id", help="unique request key; reuse it to recover the same reservation")
     parser.add_argument("--seed", type=int, default=0)
     parser.add_argument("--noise", type=float, default=0.02)
     parser.add_argument("--out", required=True)
     args = parser.parse_args(argv)
+    if bool(args.reserve_ledger) != bool(args.request_id):
+        parser.error("--reserve-ledger and --request-id must be supplied together")
     try:
-        result = propose(args.candidates, args.observations, args.pending, seed=args.seed, noise=args.noise)
+        if args.reserve_ledger:
+            from medialoop.reservations import propose_and_reserve
+            result = propose_and_reserve(args.candidates, args.observations, args.pending,
+                                         ledger_path=args.reserve_ledger, request_id=args.request_id,
+                                         seed=args.seed, noise=args.noise)
+        else:
+            result = propose(args.candidates, args.observations, args.pending, seed=args.seed, noise=args.noise)
         out = Path(args.out)
         out.parent.mkdir(parents=True, exist_ok=True)
-        with out.open("x") as handle:
-            json.dump(result, handle, indent=2, allow_nan=False)
-            handle.write("\n")
+        text = json.dumps(result, indent=2, allow_nan=False) + "\n"
+        try:
+            with out.open("x", encoding="utf-8") as handle:
+                handle.write(text)
+        except FileExistsError:
+            # A successful reservation retry may safely reuse its exact export.
+            # Partial/different exports still require a fresh destination.
+            if not args.reserve_ledger or out.read_text(encoding="utf-8") != text:
+                raise
     except (ValueError, OSError) as exc:
         parser.error(str(exc))
     return 0

src/medialoop/reservations.py

@@ -0,0 +1,69 @@
+"""Transactional local reservations and idempotent proposal recovery."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sqlite3
+
+from medialoop.planner import _read, _propose
+from medialoop.space import NAMES
+
+
+def propose_and_reserve(candidates_path, observations_path=None, pending_path=None, *,
+                        ledger_path, request_id, seed=0, noise=0.02):
+    """Reserve before returning; retry a request ID to retrieve its saved result.
+
+    All cooperating callers must use the same local SQLite file. Reservations
+    persist even after completion, so a condition cannot be issued twice. Input
+    CSVs are snapshots, not transactionally managed experimental records.
+    """
+    if (not isinstance(request_id, str) or not 1 <= len(request_id) <= 128
+            or any(not 33 <= ord(char) <= 126 for char in request_id)):
+        raise ValueError("request_id must contain 1-128 printable ASCII characters without spaces")
+    ledger = Path(ledger_path)
+    ledger.parent.mkdir(parents=True, exist_ok=True)
+    connection = None
+    try:
+        connection = sqlite3.connect(ledger, timeout=30, isolation_level=None)
+        connection.execute("PRAGMA synchronous=FULL")
+        # SQLite releases this lock on process exit; no stale lock-file removal
+        # is required. Selection and insertion are serialized by one transaction.
+        connection.execute("BEGIN IMMEDIATE")
+        connection.execute("CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
+        connection.execute("CREATE TABLE IF NOT EXISTS reservations ("
+                           "request_id TEXT PRIMARY KEY, candidate_id TEXT NOT NULL UNIQUE, "
+                           "configuration TEXT NOT NULL, proposal TEXT NOT NULL)")
+        rows, candidate_hash = _read(candidates_path, ["candidate_id", *NAMES])
+        bound = connection.execute("SELECT value FROM metadata WHERE key='candidates_sha256'").fetchone()
+        if bound is not None and bound[0] != candidate_hash:
+            raise ValueError("candidate table differs from this ledger; keep its original candidate table")
+        configuration = json.dumps({"seed": seed, "noise": noise}, sort_keys=True, allow_nan=False)
+        saved = connection.execute("SELECT configuration, proposal FROM reservations WHERE request_id=?",
+                                   (request_id,)).fetchone()
+        if saved is not None:
+            if saved[0] != configuration:
+                raise ValueError("request_id already exists with different seed or noise")
+            result = json.loads(saved[1])
+        else:
+            observations, observations_hash = _read(observations_path, ["candidate_id", "response"]) if observations_path else ([], None)
+            pending, pending_hash = _read(pending_path, ["candidate_id"]) if pending_path else ([], None)
+            hashes = {"candidates": candidate_hash, "observations": observations_hash, "pending": pending_hash}
+            reserved = [row[0] for row in connection.execute("SELECT candidate_id FROM reservations")]
+            result = _propose(rows, observations, pending, hashes, seed=seed, noise=noise, reserved_ids=reserved)
+            result["reservation"] = {"request_id": request_id, "candidates_sha256": candidate_hash}
+            result["note"] = ("Maximizes the supplied response. A proposal is not an executable protocol. "
+                              "Reserved in the ledger; reuse this request ID to recover the original proposal.")
+            if bound is None:
+                connection.execute("INSERT INTO metadata VALUES ('candidates_sha256', ?)", (candidate_hash,))
+            connection.execute("INSERT INTO reservations VALUES (?, ?, ?, ?)",
+                               (request_id, result["candidate_id"], configuration,
+                                json.dumps(result, allow_nan=False)))
+        connection.commit()
+        return result
+    except sqlite3.Error as exc:
+        raise ValueError(f"reservation ledger could not be updated: {exc}") from exc
+    finally:
+        if connection is not None:
+            # An uncommitted insertion is rolled back, including on interruption.
+            connection.close()