diff --git a/README.md b/README.md index cfcab10..cec4852 100644 --- a/README.md +++ b/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. diff --git a/src/medialoop/planner.py b/src/medialoop/planner.py index 43171f8..20a5cf1 100644 --- a/src/medialoop/planner.py +++ b/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 diff --git a/src/medialoop/reservations.py b/src/medialoop/reservations.py new file mode 100644 index 0000000..d41258e --- /dev/null +++ b/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() diff --git a/tests/test_reservations.py b/tests/test_reservations.py new file mode 100644 index 0000000..41f3332 --- /dev/null +++ b/tests/test_reservations.py @@ -0,0 +1,174 @@ +import contextlib +import io +import json +import multiprocessing +from pathlib import Path +import sqlite3 +import tempfile +import unittest +from unittest.mock import patch + +from medialoop.planner import main, propose +from medialoop.reservations import propose_and_reserve +from medialoop.space import NAMES + + +def reserve_worker(candidates, ledger, request_id, start, messages): + messages.put(("ready", request_id)) + start.wait(15) + try: + result = propose_and_reserve(candidates, ledger_path=ledger, request_id=request_id) + messages.put(("result", result)) + except Exception as exc: + messages.put(("error", repr(exc))) + + +def interrupted_writer(ledger, ready): + connection = sqlite3.connect(ledger) + connection.execute("BEGIN IMMEDIATE") + connection.execute("INSERT INTO reservations VALUES ('interrupted', 'bogus', '{}', '{}')") + ready.set() + # The parent terminates this process with an uncommitted write and held lock. + multiprocessing.Event().wait(30) + + +class ReservationTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.candidates = self.root / "candidates.csv" + self.candidates.write_text("candidate_id," + ",".join(NAMES) + + "\na,0,0,0,0\nb,3,1,0,0\nc,6,2,0,0\nd,9,3,0,0\n") + self.ledger = self.root / "reservations.sqlite3" + + def reserve(self, request_id, **kwargs): + return propose_and_reserve(self.candidates, ledger_path=self.ledger, + request_id=request_id, **kwargs) + + def test_repeated_requests_are_unique_and_preview_remains_read_only(self): + before = self.candidates.read_bytes() + preview = propose(self.candidates) + self.assertFalse(self.ledger.exists()) + first = self.reserve("first") + self.assertEqual(first["candidate_id"], preview["candidate_id"]) + results = [first] + [self.reserve(str(i)) for i in range(3)] + self.assertEqual(len({result["candidate_id"] for result in results}), 4) + with self.assertRaisesRegex(ValueError, "no unevaluated"): + self.reserve("exhausted") + self.assertEqual(self.candidates.read_bytes(), before) + + def test_retry_recovers_original_proposal_even_when_observations_change(self): + observations = self.root / "observations.csv" + observations.write_text("candidate_id,response\na,0.1\n") + first = self.reserve("round-1", observations_path=observations) + observations.unlink() + self.assertEqual(first, self.reserve("round-1", observations_path=observations)) + with self.assertRaisesRegex(ValueError, "different seed or noise"): + self.reserve("round-1", seed=1) + + def test_pending_completed_and_reserved_conditions_are_excluded(self): + first = self.reserve("first") + remaining = sorted({"a", "b", "c", "d"} - {first["candidate_id"]}) + observations = self.root / "observations.csv" + observations.write_text(f"candidate_id,response\n{first['candidate_id']},0.2\n{remaining[0]},0.4\n") + pending = self.root / "pending.csv" + pending.write_text(f"candidate_id\n{remaining[1]}\n") + result = self.reserve("second", observations_path=observations, pending_path=pending) + self.assertEqual(result["candidate_id"], remaining[2]) + self.assertEqual(result["n_observed"], 2) + self.assertEqual(result["n_pending"], 1) + + def test_candidate_table_is_bound_to_ledger(self): + self.reserve("first") + self.candidates.write_text(self.candidates.read_text().replace("d,9", "d,10")) + for request_id in ("first", "second"): + with self.assertRaisesRegex(ValueError, "candidate table differs"): + self.reserve(request_id) + + def test_commit_failure_rolls_back_insert_and_request_id(self): + self.reserve("initial") + connect = sqlite3.connect + + class FailingCommit(sqlite3.Connection): + def commit(self): + raise sqlite3.OperationalError("simulated disk failure") + + def failing_connect(*args, **kwargs): + return connect(*args, factory=FailingCommit, **kwargs) + + with patch("medialoop.reservations.sqlite3.connect", side_effect=failing_connect): + with self.assertRaisesRegex(ValueError, "simulated disk failure"): + self.reserve("retry") + with connect(self.ledger) as connection: + self.assertEqual(connection.execute("SELECT count(*) FROM reservations").fetchone()[0], 1) + self.reserve("retry", seed=2) + + def test_concurrent_processes_reserve_distinct_candidates(self): + context = multiprocessing.get_context("spawn") + start, messages = context.Event(), context.Queue() + processes = [context.Process(target=reserve_worker, + args=(self.candidates, self.ledger, str(i), start, messages)) for i in range(4)] + try: + for process in processes: + process.start() + for _ in processes: + self.assertEqual(messages.get(timeout=15)[0], "ready") + start.set() + replies = [messages.get(timeout=20) for _ in processes] + self.assertTrue(all(kind == "result" for kind, _ in replies), replies) + self.assertEqual(len({result["candidate_id"] for _, result in replies}), 4) + for process in processes: + process.join(timeout=10) + self.assertEqual(process.exitcode, 0) + finally: + for process in processes: + if process.is_alive(): + process.terminate() + process.join(timeout=5) + messages.close() + + def test_terminated_writer_releases_lock_and_rolls_back(self): + self.reserve("first") + context = multiprocessing.get_context("spawn") + ready = context.Event() + process = context.Process(target=interrupted_writer, args=(self.ledger, ready)) + process.start() + try: + self.assertTrue(ready.wait(15)) + finally: + process.terminate() + process.join(timeout=5) + second = self.reserve("second") + self.assertIn(second["candidate_id"], {"a", "b", "c", "d"}) + with sqlite3.connect(self.ledger) as connection: + self.assertIsNone(connection.execute("SELECT * FROM reservations WHERE request_id='interrupted'").fetchone()) + + def test_cli_export_failure_keeps_recoverable_reservation(self): + output = self.root / "proposal.json" + output.write_text("existing unrelated output") + args = ["--candidates", str(self.candidates), "--reserve-ledger", str(self.ledger), + "--request-id", "round-1", "--out", str(output)] + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + main(args) + output.unlink() + self.assertEqual(main(args), 0) + result = json.loads(output.read_text()) + self.assertEqual(result, self.reserve("round-1")) + self.assertEqual(main(args), 0) + with sqlite3.connect(self.ledger) as connection: + self.assertEqual(connection.execute("SELECT count(*) FROM reservations").fetchone()[0], 1) + + def test_cli_requires_request_key_and_preserves_preview_output(self): + output = self.root / "preview.json" + args = ["--candidates", str(self.candidates), "--out", str(output)] + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + main(args + ["--reserve-ledger", str(self.ledger)]) + self.assertFalse(self.ledger.exists()) + self.assertEqual(main(args), 0) + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + main(args) + + +if __name__ == "__main__": + unittest.main()