fix(history): fence context commits and resolve stored run identity

tests/test_execution_context_database.py

@@ -1,6 +1,8 @@
 """Immutable scalar capture, deduplication, and conservative navigation hints."""
 
 from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from uuid import uuid4
 from dataclasses import replace
 from pathlib import Path
 import subprocess
@@ -70,6 +72,70 @@ class ContextDatabaseTests(unittest.TestCase):
             )
         )
 
+    def test_finish_capture_uses_stored_parent_identity(self):
+        db = self.database()
+        start = replace(self.start(db), run_id=str(uuid4()))
+        db.write_execution_start(start)
+        finish = self.finish(start)
+        self.assertTrue(db.write_execution_contexts_for_finish(finish, bundle()))
+        self.assertEqual(db.get_execution_contexts(start.run_id)["state"], "complete")
+        missing = replace(finish, queue_item_id=999)
+        self.assertFalse(db.write_execution_contexts_for_finish(missing, bundle()))
+        self.assertEqual(self.counts(db), (1, 1, 2, 1))
+
+    def test_finish_generation_guard_encloses_commit_and_rollback(self):
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+        finish = self.finish(start)
+        observed = []
+
+        @contextmanager
+        def guard():
+            observed.append(db.model.get_connection().in_transaction)
+            try:
+                yield True
+            finally:
+                observed.append(db.model.get_connection().in_transaction)
+
+        self.assertTrue(
+            db.write_execution_contexts_for_finish(
+                finish, bundle(), generation_guard=guard
+            )
+        )
+        self.assertEqual(observed, [True, False])
+        observed.clear()
+        other = replace(self.start(db, 1), run_id=str(uuid4()))
+        db.write_execution_start(other)
+        with patch(
+            "database.execution_contexts._insert_exact",
+            side_effect=ValueError("collision"),
+        ):
+            with self.assertRaises(ValueError):
+                db.write_execution_contexts_for_finish(
+                    self.finish(other), bundle(), generation_guard=guard
+                )
+        self.assertEqual(observed, [True, False])
+        self.assertEqual(
+            db.get_execution_contexts(other.run_id)["state"], "not_captured"
+        )
+
+    def test_finish_generation_guard_rejects_without_capture(self):
+        db = self.database()
+        start = self.start(db)
+        db.write_execution_start(start)
+
+        @contextmanager
+        def stale():
+            yield False
+
+        self.assertFalse(
+            db.write_execution_contexts_for_finish(
+                self.finish(start), bundle(), generation_guard=stale
+            )
+        )
+        self.assertEqual(self.counts(db), (1, 0, 0, 0))
+
     def test_a_a_b_cardinality(self):
         db = self.database()
         self.assertTrue(

tests/test_execution_history_writer.py

@@ -186,12 +186,14 @@ class HistoryWriterTests(unittest.TestCase):
                 sealed = self.sealed(writer)
                 entered, release = self.block()
 
-                def fail_context(*args):
+                def fail_context(*args, **kwargs):
                     entered.set()
                     release.wait(3)
                     raise error
 
-                with patch.object(self.db, "write_execution_contexts", fail_context):
+                with patch.object(
+                    self.db, "write_execution_contexts_for_finish", fail_context
+                ):
                     self.assertTrue(
                         writer.submit_finalization(self.finish_command(start), sealed)
                     )
@@ -309,16 +311,16 @@ class HistoryWriterTests(unittest.TestCase):
         start = self.start_command()
         self.db.write_execution_start(start)
         entered, release = self.block()
-        original = self.db.write_execution_contexts
+        original = self.db.write_execution_contexts_for_finish
         first = self.sealed(writer)
         charge = first.lease.byte_count
 
-        def write(*args):
+        def write(*args, **kwargs):
             entered.set()
             release.wait(3)
-            return original(*args)
+            return original(*args, **kwargs)
 
-        with patch.object(self.db, "write_execution_contexts", write):
+        with patch.object(self.db, "write_execution_contexts_for_finish", write):
             self.assertTrue(
                 writer.submit_finalization(self.finish_command(start), first)
             )
@@ -412,6 +414,118 @@ class HistoryWriterTests(unittest.TestCase):
         self.assertEqual(foreign.charged_bytes, 0)
         self.assertEqual(self.db.get_execution_contexts(start.run_id)["contexts"], [])
 
+    def test_custom_run_identity_receives_finalized_context(self):
+        writer = self.writer()
+        start = replace(self.start_command(), run_id=str(uuid4()))
+        self.assertTrue(writer.submit(start))
+        self.assertTrue(
+            writer.submit_finalization(self.finish_command(start), self.sealed(writer))
+        )
+        self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+        )
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
+    def test_invalidation_during_context_preparation_fences_actual_transaction(self):
+        import database.execution_contexts as contexts_db
+
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        entered, release = self.block()
+        original = contexts_db._prepare
+
+        def prepare(bundle):
+            prepared = original(bundle)
+            entered.set()
+            release.wait(3)
+            return prepared
+
+        with patch.object(contexts_db, "_prepare", prepare):
+            self.assertTrue(
+                writer.submit_finalization(
+                    self.finish_command(start), self.sealed(writer)
+                )
+            )
+            self.assertTrue(entered.wait(1))
+            self.assertEqual(
+                self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+            )
+            generation = writer.status().capture_generation
+            writer.invalidate()
+            self.assertGreater(writer.status().capture_generation, generation)
+            release.set()
+            self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+        self.clock.advance(5)
+        self.wait_for(lambda: writer.status().accepting)
+        self.assertEqual(writer.status().epoch, start.storage_epoch)
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "not_captured"
+        )
+
+    def test_context_commit_serializes_invalidation_without_blocking_queue_or_close(
+        self,
+    ):
+        import database.execution_contexts as contexts_db
+
+        writer = self.writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        entered, release = self.block()
+        invalidating, invalidated = threading.Event(), threading.Event()
+        original = contexts_db._write_capture
+
+        def capture(*args, **kwargs):
+            entered.set()
+            release.wait(3)
+            return original(*args, **kwargs)
+
+        def invalidate():
+            invalidating.set()
+            writer.invalidate()
+            invalidated.set()
+
+        with patch.object(contexts_db, "_write_capture", capture):
+            self.assertTrue(
+                writer.submit_finalization(
+                    self.finish_command(start), self.sealed(writer)
+                )
+            )
+            self.assertTrue(entered.wait(1))
+            invalidator = threading.Thread(target=invalidate)
+            invalidator.start()
+            try:
+                self.assertTrue(invalidating.wait(1))
+                self.assertFalse(invalidated.wait(0.03))
+                before = time.monotonic()
+                self.assertTrue(writer.status().accepting)
+                self.assertTrue(writer.submit(self.finish_command(start)))
+                self.assertLess(time.monotonic() - before, 0.1)
+                self.assertFalse(writer.close(0.02))
+                self.assertGreater(writer.context_budget.charged_bytes, 0)
+                self.assertEqual(writer.status().pending, 1)
+            finally:
+                release.set()
+                invalidator.join(3)
+            self.assertFalse(invalidator.is_alive())
+            self.assertTrue(invalidated.is_set())
+            self.wait_for(lambda: not writer.status().alive)
+        self.assertEqual(
+            self.db.get_execution_run(start.run_id)["outcome"], "succeeded"
+        )
+        self.assertEqual(
+            self.db.get_execution_contexts(start.run_id)["state"], "complete"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+
     def test_fifo_single_worker_and_thread_local_connection_cleanup(self):
         caller_conn = self.db.model.get_connection()
         threads, timeouts = [], []