diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md
index 6db977a3..4fbb9b97 100644
--- a/docs/resource-budgets.md
+++ b/docs/resource-budgets.md
@@ -1957,3 +1957,53 @@ activation remain separate requirements. Ordinary edits within the same epoch
 remain the source adapter's responsibility. Local tests use mocked file handles
 and in-memory SQLite; the synthetic file-backed worker test is opt-in for GPUBox.
 No native-memory reduction or resolution of the original Mac incident is claimed.
+
+### Selected tier job: durable binding through the accepted connection
+
+The additive `select_tier_job`, `check_selected_tier_job`,
+`open_selected_tier_job` and `end_selected_tier_job` APIs bind a background job
+to a committed marker written through the accepted clean connection. The original
+capture/check/open functions retain their previous behavior. The descriptor still
+contains no source corpus, connection, model or ownership token. No public manager
+or worker caller adopts these selected-job functions in this checkpoint.
+
+Selection owns one `BEGIN IMMEDIATE` transaction. It validates source identity in
+that writer snapshot, creates or checks the exact reserved singleton schema,
+writes the generated job ID with the explicit target and tracker/epoch/schema,
+rechecks identity, and commits. It never installs source tracking. Creating the
+marker leaves the captured messages-schema signature unchanged, but changes the
+global SQLite schema version: select the job before preparing a source plan.
+Borrowed transactions are refused untouched. A live selection cannot be replaced.
+Replacing a failed/cancelled selection requires its exact previous job ID.
+
+Selected reopen retains worker-thread maintenance ownership and the existing
+file/path/source checks, then verifies the committed job marker before yielding
+a clean connection. Marker checks read a coherent source/selection snapshot and
+make no writes, so they do not invalidate a source plan's total_changes fence.
+Ending a job accepts only `cancelled` or `failed` and compares the complete job,
+target and source identity before updating its state. A stale job cannot end a
+replacement. Terminal writes belong after abandoning the source plan; there are
+no per-page marker writes or automatically inferred generation associations.
+Successful completion and activation require a later certified publication API.
+
+The marker stores explicit typed columns, not JSON or source content. Text bounds
+are 32 job-ID characters, 6 tier, 512 model ID, 7 group, 30 tracker, 128 epoch,
+64 schema signature and 9 state. Before payload materialization, SQL probes byte
+lengths using BLOB casts, including bytes after embedded NUL. The conservative
+UTF-8 admission bound is `4 * (32+6+512+7+30+128+64+9) = 3152` payload bytes,
+plus one integer dimension and object/SQLite overhead. Character lengths and
+exact identities are checked after that bounded read. Noncanonical schema,
+unexpected marker indexes/triggers, TEMP objects, invalid values, mismatched
+selection and uncertain rollback fail closed. Errors do not include path,
+target or source details. Commit errors report failure without assuming that an
+ambiguous completed commit can safely be undone or retried.
+
+A clone made before selection lacks the newly committed marker (or retains an
+older job ID), so reopening it is rejected even if it shares the source epoch.
+A clone or hostile marker copy after that commit is outside this guarantee.
+The stable-path and known-current-connection preconditions remain; Python SQLite
+still exposes no native database descriptor identity. Current-source coverage,
+exact pair generation, final writer ownership and activation remain separate.
+Synthetic tests use in-memory databases and mocked file observations only. This
+binding neither removes the public manager's retained source list nor proves the
+original Mac memory incident fixed.
diff --git a/tests/test-tier-job-binding-795.py b/tests/test-tier-job-binding-795.py
new file mode 100644
index 00000000..c5c77e98
--- /dev/null
+++ b/tests/test-tier-job-binding-795.py
@@ -0,0 +1,462 @@
+"""Durable selected-job admission using only synthetic in-memory databases."""
+
+from __future__ import annotations
+
+import dataclasses
+import runpy
+import sqlite3
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+BASE = runpy.run_path(str(Path(__file__).with_name("test-tier-job-identity-795.py")))
+CONNECT = BASE["CONNECT"]
+CONNECTION = BASE["MockFileConnection"]
+Cancelled = BASE["Cancelled"]
+
+
+class FaultConnection(CONNECTION):
+    fault_sql: str | None = None
+    fault: BaseException | None = None
+    fail_commit = False
+    retain_commit = False
+    selection_payload_reads = 0
+
+    def execute(self, sql: str, parameters: tuple = ()):
+        if sql.startswith("SELECT job_id, tier"):
+            self.selection_payload_reads += 1
+        if self.fault_sql is not None and self.fault_sql in sql:
+            raise self.fault
+        return super().execute(sql, parameters)
+
+    def commit(self) -> None:
+        if self.fail_commit:
+            raise sqlite3.OperationalError("synthetic marker commit failure")
+        if not self.retain_commit:
+            super().commit()
+
+
+class TestTierJobBinding(unittest.TestCase):
+    def setUp(self) -> None:
+        self.fixture = BASE["TestTierJobIdentity"]()
+        self.fixture.setUp()
+        self.addCleanup(self.fixture.doCleanups)
+        self.api = self.fixture.api
+        self.conn = self.fixture.conn
+        self.target = self.fixture.target
+        self.table = self.api._SELECTION_TABLE
+
+    def select(self, **kwargs):
+        return self.api.select_tier_job(self.conn, self.target, **kwargs)
+
+    def row(self, conn=None):
+        return (conn or self.conn).execute(f"SELECT * FROM {self.table}").fetchone()
+
+    def clone(self, *, factory=CONNECTION):
+        conn = CONNECT(":memory:", factory=factory)
+        self.conn.backup(conn)
+        self.addCleanup(conn.close)
+        return conn
+
+    def test_selection_commits_exact_marker_without_changing_source_identity(self) -> None:
+        previous = self.fixture.capture()
+        before = self.conn.total_changes
+        job = self.select()
+        self.assertEqual(self.conn.total_changes - before, 1)
+        self.assertFalse(self.conn.in_transaction)
+        self.assertEqual(job.target, self.target)
+        self.assertEqual(job.source_schema_signature, previous.source_schema_signature)
+        self.assertEqual(job.source_epoch, previous.source_epoch)
+        self.assertEqual(self.row(), (1, job.job_id, "selected", "base", "qwen3_256", 256,
+                                      "basepro", job.tracker, job.source_epoch, job.source_schema_signature))
+        self.api.check_tier_job(self.conn, job)
+        self.api.check_selected_tier_job(self.conn, job)
+        self.assertFalse(self.conn.in_transaction)
+        self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0], 0)
+
+    def test_repeated_read_checks_do_not_change_source_plan_write_fence(self) -> None:
+        job = self.select()
+        changes, version = self.conn.total_changes, self.conn.execute("PRAGMA data_version").fetchone()[0]
+        for _ in range(3):
+            self.api.check_selected_tier_job(self.conn, job)
+        self.assertEqual(self.conn.total_changes, changes)
+        self.assertEqual(self.conn.execute("PRAGMA data_version").fetchone()[0], version)
+        self.assertFalse(self.conn.in_transaction)
+
+    def test_live_job_cannot_be_replaced_even_with_its_id(self) -> None:
+        job = self.select()
+        before = self.row()
+        for kwargs in ({}, {"previous_job_id": job.job_id}, {"previous_job_id": "0" * 32}):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.select(**kwargs)
+            self.assertEqual(self.row(), before)
+            self.assertFalse(self.conn.in_transaction)
+
+    def test_terminal_replacement_requires_exact_previous_id(self) -> None:
+        old = self.select()
+        self.assertTrue(self.api.end_selected_tier_job(self.conn, old, outcome="cancelled"))
+        for kwargs in ({}, {"previous_job_id": "0" * 32}):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.select(**kwargs)
+        new = self.select(previous_job_id=old.job_id)
+        self.assertNotEqual(old.job_id, new.job_id)
+        self.api.check_selected_tier_job(self.conn, new)
+        self.assertFalse(self.api.end_selected_tier_job(self.conn, old, outcome="failed"))
+        self.assertEqual(self.row()[1:3], (new.job_id, "selected"))
+
+    def test_termination_compares_full_identity_and_is_not_completion(self) -> None:
+        job = self.select()
+        mismatched = dataclasses.replace(job, target=self.fixture.modules["embedding_target"].EmbeddingTarget(
+            "pro", "qwen3_256", 256, "basepro",
+        ))
+        self.assertFalse(self.api.end_selected_tier_job(self.conn, mismatched, outcome="failed"))
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.end_selected_tier_job(self.conn, job, outcome="completed")
+        self.assertEqual(self.row()[2], "selected")
+        self.assertTrue(self.api.end_selected_tier_job(self.conn, job, outcome="failed"))
+        self.assertFalse(self.api.end_selected_tier_job(self.conn, job, outcome="cancelled"))
+        self.assertEqual(self.row()[2], "failed")
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.check_selected_tier_job(self.conn, job)
+
+    def test_preselection_clone_reopen_is_rejected_after_stale_connection_selection(self) -> None:
+        # The clone shares the source epoch but predates the marker write through
+        # the accepted connection. File observations intentionally cannot tell.
+        clone = self.clone()
+        job = self.select()
+        self.fixture.reopen_hooks()
+        with patch.object(self.api.sqlite3, "connect", return_value=clone):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                with self.api.open_selected_tier_job(job):
+                    self.fail("Pre-selection clone must not reach model admission")
+        self.assertTrue(clone.closed)
+        self.assertEqual(self.fixture.events[-1], "release")
+        self.api.check_selected_tier_job(self.conn, job)
+
+    def test_clone_with_a_previously_selected_id_cannot_impersonate_new_selection(self) -> None:
+        old = self.select()
+        clone = self.clone()
+        self.api.end_selected_tier_job(self.conn, old, outcome="cancelled")
+        job = self.select(previous_job_id=old.job_id)
+        self.fixture.reopen_hooks()
+        with patch.object(self.api.sqlite3, "connect", return_value=clone):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                with self.api.open_selected_tier_job(job):
+                    self.fail("Old cloned selection must not yield")
+        self.assertTrue(clone.closed)
+
+    def test_postselection_marker_copy_is_explicitly_outside_identity_guarantee(self) -> None:
+        job = self.select()
+        clone = self.clone()
+        self.fixture.reopen_hooks()
+        with patch.object(self.api.sqlite3, "connect", return_value=clone):
+            with self.api.open_selected_tier_job(job) as opened:
+                self.assertIs(opened, clone)
+                self.assertFalse(opened.in_transaction)
+        self.assertTrue(clone.closed)
+
+    def test_owned_selected_reopen_validates_before_yield_and_releases_on_cancellation(self) -> None:
+        job = self.select()
+        reopened = self.fixture.reopen_hooks()
+        sentinel = Cancelled("synthetic cancellation")
+        with self.assertRaises(Cancelled) as caught:
+            with self.api.open_selected_tier_job(job) as conn:
+                self.assertFalse(conn.in_transaction)
+                self.api.check_selected_tier_job(conn, job)
+                raise sentinel
+        self.assertIs(caught.exception, sentinel)
+        self.assertTrue(reopened[0].closed)
+        self.assertEqual(self.fixture.events[-1], "release")
+
+    def test_borrowed_transaction_never_installs_or_mutates_marker(self) -> None:
+        job = self.fixture.capture()
+        self.conn.execute("BEGIN")
+        before = self.conn.total_changes
+        for call in (lambda: self.select(), lambda: self.api.check_selected_tier_job(self.conn, job),
+                     lambda: self.api.end_selected_tier_job(self.conn, job, outcome="failed")):
+            with self.assertRaises(self.api.TierJobIdentityError):
+                call()
+            self.assertTrue(self.conn.in_transaction)
+            self.assertEqual(self.conn.total_changes, before)
+        self.conn.rollback()
+        self.assertIsNone(self.conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_wrong_schema_view_temp_shadow_and_side_effects_refuse(self) -> None:
+        for ddl in (f"CREATE TABLE {self.table}(singleton INTEGER PRIMARY KEY, job_id TEXT)",
+                    f"CREATE VIEW {self.table} AS SELECT 1 AS singleton",
+                    f"CREATE TEMP TABLE {self.table.upper()}(value TEXT)"):
+            conn = self.clone()
+            conn.execute(ddl)
+            with self.assertRaises(self.api.TierJobIdentityError):
+                self.api.select_tier_job(conn, self.target)
+            self.assertFalse(conn.in_transaction)
+        job = self.select()
+        self.conn.execute(f"CREATE TRIGGER synthetic_marker_side_effect AFTER UPDATE ON {self.table} BEGIN SELECT 1; END")
+        before = self.row()
+        for call in (lambda: self.api.check_selected_tier_job(self.conn, job),
+                     lambda: self.api.end_selected_tier_job(self.conn, job, outcome="failed")):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                call()
+            self.assertEqual(self.row(), before)
+
+    def test_invalid_and_oversized_stored_values_fail_before_value_materialization(self) -> None:
+        job = self.select()
+        for field, value in (("model_id", "x" * 513), ("source_epoch", "x" * 129),
+                             ("dimension", "not-an-integer"), ("job_id", "z" * 32),
+                             ("state", "completed"), ("source_schema_signature", "z" * 64)):
+            conn = self.clone()
+            conn.execute("PRAGMA ignore_check_constraints=ON")
+            conn.execute(f"UPDATE {self.table} SET {field}=?", (value,))
+            conn.commit()
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.api.check_selected_tier_job(conn, job)
+            self.assertFalse(conn.in_transaction)
+
+    def test_extra_singleton_and_changed_target_refuse(self) -> None:
+        job = self.select()
+        conn = self.clone()
+        conn.execute("PRAGMA ignore_check_constraints=ON")
+        columns = ", ".join((*self.api._SELECTION_FIELDS, "state"))
+        conn.execute(f"INSERT INTO {self.table} SELECT 2, {columns} FROM {self.table}")
+        conn.commit()
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.check_selected_tier_job(conn, job)
+        changed = dataclasses.replace(job, target=self.fixture.modules["embedding_target"].EmbeddingTarget(
+            "custom", "synthetic/encoder", 384, "custom",
+        ))
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.check_selected_tier_job(self.conn, changed)
+
+    def test_embedded_nul_cannot_hide_an_oversized_payload_from_admission(self) -> None:
+        job = self.select()
+        for field, bound in (("model_id", 512), ("source_epoch", 128), ("source_schema_signature", 64)):
+            conn = self.clone(factory=FaultConnection)
+            conn.execute("PRAGMA ignore_check_constraints=ON")
+            conn.execute(f"UPDATE {self.table} SET {field}=?", ("a\0" + "x" * (4 * bound),))
+            conn.commit()
+            self.assertEqual(conn.execute(f"SELECT length({field}) FROM {self.table}").fetchone()[0], 1)
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.api.check_selected_tier_job(conn, job)
+            self.assertEqual(conn.selection_payload_reads, 0)
+            self.assertFalse(conn.in_transaction)
+
+    def test_marker_byte_limit_accepts_multibyte_model_identity_at_character_bound(self) -> None:
+        target = self.fixture.modules["embedding_target"].EmbeddingTarget("custom", "\U00010400" * 512, 384, "custom")
+        job = self.api.select_tier_job(self.conn, target)
+        self.api.check_selected_tier_job(self.conn, job)
+        self.assertEqual(len(job.target.model_id.encode("utf-8")), 2048)
+        self.assertEqual(len(job.target.model_id), 512)
+
+    def test_source_epoch_changed_between_capture_and_writer_refuses_without_marker(self) -> None:
+        original = self.api.capture_tier_job
+        def replace_after_capture(conn, target):
+            job = original(conn, target)
+            conn.execute("UPDATE truememory_rebuild_source_v1 SET epoch='synthetic-replacement'")
+            conn.commit()
+            return job
+        with patch.object(self.api, "capture_tier_job", replace_after_capture):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.select()
+        self.assertIsNone(self.conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+        self.assertFalse(self.conn.in_transaction)
+
+    def test_temp_or_path_changed_after_capture_is_rechecked_inside_writer(self) -> None:
+        original = self.api.capture_tier_job
+        for change in ("temp", "path"):
+            conn = self.clone()
+            def change_after_capture(database, target):
+                job = original(database, target)
+                if change == "temp":
+                    database.execute("CREATE TEMP TABLE MESSAGES(id INTEGER PRIMARY KEY, content TEXT)")
+                else:
+                    database.fake_path = str(BASE["SYNTHETIC_PATH"].with_name("other.db"))
+                return job
+            with patch.object(self.api, "capture_tier_job", change_after_capture):
+                with self.assertRaises(self.api.TierJobSelectionError):
+                    self.api.select_tier_job(conn, self.target)
+            self.assertFalse(conn.in_transaction)
+            self.assertIsNone(conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_invalid_previous_id_and_oversized_identity_make_no_marker_writes(self) -> None:
+        for value in (True, 1, "invalid", "0" * 33):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.select(previous_job_id=value)
+        target = self.fixture.modules["embedding_target"].EmbeddingTarget("custom", "x" * 513, 384, "custom")
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.select_tier_job(self.conn, target)
+        self.assertIsNone(self.conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+        self.assertFalse(self.conn.in_transaction)
+
+    def test_marker_state_change_between_admission_and_read_snapshot_refuses(self) -> None:
+        job = self.select()
+        original = self.api.check_tier_job
+        def cancel_after_identity(database, captured):
+            original(database, captured)
+            database.execute(f"UPDATE {self.table} SET state='cancelled'")
+            database.commit()
+        with patch.object(self.api, "check_tier_job", cancel_after_identity):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.api.check_selected_tier_job(self.conn, job)
+        self.assertFalse(self.conn.in_transaction)
+
+    def test_retained_commit_rolls_back_and_never_returns_selected_descriptor(self) -> None:
+        conn = self.clone(factory=FaultConnection)
+        conn.retain_commit = True
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.select_tier_job(conn, self.target)
+        self.assertFalse(conn.in_transaction)
+        self.assertIsNone(conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_source_change_after_marker_insert_rolls_back_entire_selection(self) -> None:
+        original = self.api._selection_source
+        calls = []
+        def replace_before_final_check(database, job):
+            calls.append(1)
+            if len(calls) == 2:
+                database.execute("UPDATE truememory_rebuild_source_v1 SET epoch='synthetic-raced-epoch'")
+            original(database, job)
+        old_epoch = self.fixture.capture().source_epoch
+        with patch.object(self.api, "_selection_source", replace_before_final_check):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.select()
+        self.assertEqual(self.fixture.capture().source_epoch, old_epoch)
+        self.assertIsNone(self.conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_selection_errors_do_not_include_source_or_target_details(self) -> None:
+        conn = self.clone(factory=FaultConnection)
+        conn.fault_sql = "INSERT INTO main.truememory_tier_selected"
+        conn.fault = sqlite3.OperationalError("synthetic sensitive-path target text")
+        with self.assertRaises(self.api.TierJobSelectionError) as caught:
+            self.api.select_tier_job(conn, self.target)
+        self.assertNotIn("sensitive", str(caught.exception))
+        self.assertTrue(caught.exception.__suppress_context__)
+
+    def test_current_source_changes_do_not_rebind_the_marker_to_a_generation(self) -> None:
+        job = self.select()
+        before = self.row()
+        self.conn.execute("INSERT INTO messages VALUES (1,'synthetic appended content')")
+        self.conn.commit()
+        self.api.check_selected_tier_job(self.conn, job)
+        self.assertEqual(self.row(), before)
+        fields = {row[1] for row in self.conn.execute(f"PRAGMA table_info({self.table})")}
+        self.assertNotIn("generation", fields)
+        self.assertNotIn("last_message_id", fields)
+        self.assertNotIn("count", fields)
+
+    def test_schema_creation_insert_commit_and_cancellation_rollback_together(self) -> None:
+        for stage in ("CREATE TABLE truememory_tier_selected", "INSERT INTO main.truememory_tier_selected", "commit", "cancel"):
+            conn = self.clone(factory=FaultConnection)
+            if stage == "commit":
+                conn.fail_commit = True
+            else:
+                conn.fault_sql = "INSERT INTO main.truememory_tier_selected" if stage == "cancel" else stage
+                conn.fault = Cancelled("synthetic") if stage == "cancel" else sqlite3.OperationalError("synthetic failure")
+            expected = Cancelled if stage == "cancel" else self.api.TierJobSelectionError
+            with self.assertRaises(expected):
+                self.api.select_tier_job(conn, self.target)
+            self.assertFalse(conn.in_transaction)
+            self.assertIsNone(conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_terminal_commit_failure_preserves_selected_marker(self) -> None:
+        job = self.select()
+        conn = self.clone(factory=FaultConnection)
+        conn.fail_commit = True
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.end_selected_tier_job(conn, job, outcome="failed")
+        self.assertFalse(conn.in_transaction)
+        self.assertEqual(self.row(conn)[2], "selected")
+
+    def test_replacement_commit_failure_retains_previous_terminal_selection(self) -> None:
+        old = self.select()
+        self.api.end_selected_tier_job(self.conn, old, outcome="cancelled")
+        conn = self.clone(factory=FaultConnection)
+        conn.fail_commit = True
+        before = self.row(conn)
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.select_tier_job(conn, self.target, previous_job_id=old.job_id)
+        self.assertEqual(self.row(conn), before)
+        self.assertFalse(conn.in_transaction)
+
+    def test_cancellation_after_insert_rolls_back_marker_and_preserves_exception(self) -> None:
+        original = self.api._selection_source
+        sentinel = Cancelled("synthetic after-insert cancellation")
+        checks = []
+        def cancel_after_insert(database, job):
+            checks.append(1)
+            if len(checks) == 2:
+                raise sentinel
+            original(database, job)
+        with patch.object(self.api, "_selection_source", cancel_after_insert):
+            with self.assertRaises(Cancelled) as caught:
+                self.select()
+        self.assertIs(caught.exception, sentinel)
+        self.assertFalse(self.conn.in_transaction)
+        self.assertIsNone(self.conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_selected_read_rollback_uncertainty_closes_owned_connection(self) -> None:
+        job = self.select()
+        reopened = self.fixture.reopen_hooks()
+        original = self.api._selection_row
+        def retain_read_snapshot(database):
+            row = original(database)
+            database.rollback_failure = "retained"
+            return row
+        with patch.object(self.api, "_selection_row", retain_read_snapshot):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                with self.api.open_selected_tier_job(job):
+                    self.fail("Uncertain selected read must not yield")
+        self.assertTrue(reopened[0].closed)
+        self.assertEqual(self.fixture.events[-1], "release")
+
+    def test_uncertain_rollback_stops_without_additional_status_write(self) -> None:
+        job = self.select()
+        for failure in ("retained", sqlite3.OperationalError("synthetic rollback failure")):
+            conn = self.clone(factory=FaultConnection)
+            conn.rollback_failure = failure
+            # The identity read itself cannot release its snapshot. No terminal
+            # write may follow this uncertainty.
+            before = conn.total_changes
+            with self.assertRaises(self.api.TierJobIdentityError):
+                self.api.end_selected_tier_job(conn, job, outcome="failed")
+            self.assertTrue(conn.in_transaction)
+            self.assertEqual(conn.total_changes, before)
+            conn.rollback_failure = None
+            conn.rollback()
+
+    def test_writer_rollback_failure_does_not_retry_or_publish_another_status(self) -> None:
+        conn = self.clone(factory=FaultConnection)
+        original = self.api.capture_tier_job
+        def fail_writer_after_capture(database, target):
+            job = original(database, target)
+            database.rollback_failure = "retained"
+            database.fault_sql = "INSERT INTO main.truememory_tier_selected"
+            database.fault = sqlite3.OperationalError("synthetic write failure")
+            return job
+        with patch.object(self.api, "capture_tier_job", fail_writer_after_capture):
+            with self.assertRaises(self.api.TierJobSelectionError):
+                self.api.select_tier_job(conn, self.target)
+        self.assertTrue(conn.in_transaction)
+        self.assertEqual(conn.total_changes, 0)
+        conn.rollback_failure = None
+        conn.rollback()
+        self.assertIsNone(conn.execute("SELECT 1 FROM sqlite_master WHERE name=?", (self.table,)).fetchone())
+
+    def test_selections_on_shared_memory_connection_serialize_without_stealing(self) -> None:
+        uri = "file:synthetic-tier-binding?mode=memory&cache=shared"
+        first = CONNECT(uri, uri=True, factory=CONNECTION)
+        second = CONNECT(uri, uri=True, factory=CONNECTION)
+        self.addCleanup(first.close)
+        self.addCleanup(second.close)
+        self.conn.backup(first)
+        job = self.api.select_tier_job(first, self.target)
+        with self.assertRaises(self.api.TierJobSelectionError):
+            self.api.select_tier_job(second, self.target)
+        self.api.check_selected_tier_job(second, job)
+        self.assertTrue(self.api.end_selected_tier_job(second, job, outcome="cancelled"))
+        next_job = self.api.select_tier_job(first, self.target, previous_job_id=job.job_id)
+        self.assertFalse(self.api.end_selected_tier_job(second, job, outcome="failed"))
+        self.api.check_selected_tier_job(first, next_job)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/truememory/tier_switch/job.py b/truememory/tier_switch/job.py
index 66e888d1..5bdc95cd 100644
--- a/truememory/tier_switch/job.py
+++ b/truememory/tier_switch/job.py
@@ -16,6 +16,7 @@
 from contextlib import ExitStack, contextmanager
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Literal
 
 from truememory.embedding_target import EmbeddingTarget
 from truememory.maintenance import MaintenanceUnavailableError, maintenance_owner
@@ -159,3 +160,227 @@ def open_tier_job(job: TierJob) -> Iterator[sqlite3.Connection]:
         except (OSError, ValueError, sqlite3.Error, MaintenanceUnavailableError):
             raise TierJobIdentityError("Background database could not be reopened") from None
         yield conn
+
+
+_SELECTION_TABLE = "truememory_tier_selected_job_v1"
+_SELECTION_SQL = f"""CREATE TABLE {_SELECTION_TABLE} (
+    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+    job_id TEXT NOT NULL CHECK (typeof(job_id) = 'text' AND length(job_id) = 32),
+    state TEXT NOT NULL CHECK (state IN ('selected', 'cancelled', 'failed')),
+    tier TEXT NOT NULL CHECK (tier IN ('edge', 'base', 'pro', 'custom')),
+    model_id TEXT NOT NULL CHECK (typeof(model_id) = 'text' AND length(model_id) BETWEEN 1 AND 512 AND length(CAST(model_id AS BLOB)) <= 2048),
+    dimension INTEGER NOT NULL CHECK (typeof(dimension) = 'integer' AND dimension BETWEEN 1 AND 4096),
+    tier_group TEXT NOT NULL CHECK (tier_group IN ('edge', 'basepro', 'custom')),
+    tracker TEXT NOT NULL CHECK (tracker IN ('canonical-v1', 'rebuild-bridge-v1', 'rebuild-bridge-conservative-v1')),
+    source_epoch TEXT NOT NULL CHECK (typeof(source_epoch) = 'text' AND length(source_epoch) BETWEEN 1 AND 128 AND length(CAST(source_epoch AS BLOB)) <= 512),
+    source_schema_signature TEXT NOT NULL CHECK (typeof(source_schema_signature) = 'text' AND length(source_schema_signature) = 64)
+)"""
+_SELECTION_FIELDS = ("job_id", "tier", "model_id", "dimension", "tier_group", "tracker",
+                     "source_epoch", "source_schema_signature")
+_SELECTION_BOUNDS = (32, 6, 512, None, 7, 30, 128, 64)
+
+
+class TierJobSelectionError(TierJobIdentityError):
+    """The durable selected job is unavailable or does not match this request."""
+
+
+def _selection_values(job: TierJob) -> tuple[str | int, ...]:
+    values = (job.job_id, job.target.tier, job.target.model_id, job.target.dimension,
+              job.target.tier_group, job.tracker, job.source_epoch, job.source_schema_signature)
+    for value, bound in zip(values, _SELECTION_BOUNDS):
+        if bound is None:
+            if type(value) is not int or not 1 <= value <= 4096:
+                raise TierJobSelectionError("Invalid selected job identity")
+        elif type(value) is not str or not 1 <= len(value) <= bound:
+            raise TierJobSelectionError("Selected job identity exceeds its protocol bounds")
+    return values
+
+
+def _selection_schema(conn: sqlite3.Connection, *, create: bool = False) -> None:
+    row = conn.execute(
+        "SELECT type, sql FROM main.sqlite_master WHERE name = ? COLLATE NOCASE", (_SELECTION_TABLE,),
+    ).fetchone()
+    if row is None and create:
+        conn.execute(_SELECTION_SQL)
+    elif (row is None or row[0] != "table" or type(row[1]) is not str
+          or " ".join(row[1].strip().rstrip(";").split()) != " ".join(_SELECTION_SQL.split())):
+        raise TierJobSelectionError("Selected job storage has an unsupported schema")
+    if conn.execute(
+        "SELECT 1 FROM main.sqlite_master WHERE tbl_name = ? COLLATE NOCASE "
+        "AND type IN ('index', 'trigger') LIMIT 1", (_SELECTION_TABLE,),
+    ).fetchone() is not None:
+        raise TierJobSelectionError("Selected job storage has unsupported side effects")
+
+
+def _selection_row(conn: sqlite3.Connection) -> tuple[str | int, ...] | None:
+    identifiers = [tuple(row) for row in conn.execute(f"SELECT singleton FROM main.{_SELECTION_TABLE} LIMIT 2")]
+    if not identifiers:
+        return None
+    if identifiers != [(1,)]:
+        raise TierJobSelectionError("Selected job storage contains an invalid singleton")
+    # Read only sizes/types before copying bounded marker values into Python.
+    probes = ", ".join(f"typeof({name}), length(CAST({name} AS BLOB))" for name in (*_SELECTION_FIELDS, "state"))
+    metadata = conn.execute(f"SELECT {probes} FROM main.{_SELECTION_TABLE} WHERE singleton = 1").fetchone()
+    if metadata is None:
+        raise TierJobSelectionError("Selected job storage changed during validation")
+    for index, bound in enumerate((*_SELECTION_BOUNDS, 9)):
+        kind, length = metadata[index * 2:index * 2 + 2]
+        if ((bound is None and kind != "integer")
+                or (bound is not None and (kind != "text" or type(length) is not int or not 1 <= length <= 4 * bound))):
+            raise TierJobSelectionError("Selected job storage contains invalid values")
+    fields = ", ".join((*_SELECTION_FIELDS, "state"))
+    row = tuple(conn.execute(f"SELECT {fields} FROM main.{_SELECTION_TABLE} WHERE singleton = 1").fetchone())
+    if any(bound is not None and (type(value) is not str or not 1 <= len(value) <= bound)
+           for value, bound in zip(row, (*_SELECTION_BOUNDS, 9))):
+        raise TierJobSelectionError("Selected job storage contains oversized values")
+    if (re.fullmatch(r"[0-9a-f]{32}", row[0]) is None
+            or re.fullmatch(r"[0-9a-f]{64}", row[7]) is None
+            or row[5] not in {"canonical-v1", "rebuild-bridge-v1", "rebuild-bridge-conservative-v1"}
+            or row[8] not in {"selected", "cancelled", "failed"}):
+        raise TierJobSelectionError("Selected job storage contains an invalid identity")
+    EmbeddingTarget(row[1], row[2], row[3], row[4])
+    return row
+
+
+def _rollback_selection(conn: sqlite3.Connection) -> None:
+    try:
+        conn.rollback()
+    except (OSError, sqlite3.Error):
+        raise TierJobSelectionError("Selected job transaction rollback is uncertain") from None
+    if conn.in_transaction:
+        raise TierJobSelectionError("Selected job transaction rollback is uncertain")
+
+
+@contextmanager
+def _selection_transaction(conn: sqlite3.Connection, *, write: bool) -> Iterator[None]:
+    if conn.in_transaction:
+        raise TierJobSelectionError("Selected job admission requires a clean connection")
+    conn.execute("BEGIN IMMEDIATE" if write else "BEGIN")
+    try:
+        yield
+        if write:
+            conn.commit()
+            if conn.in_transaction:
+                raise TierJobSelectionError("Selected job transaction did not commit")
+    except BaseException:
+        if conn.in_transaction:
+            _rollback_selection(conn)
+        raise
+    else:
+        if not write:
+            _rollback_selection(conn)
+
+
+def _selection_source(conn: sqlite3.Connection, job: TierJob) -> None:
+    databases = tuple(conn.execute("PRAGMA database_list"))
+    mains = [row[2] for row in databases if row[1] == "main"]
+    if (len(mains) != 1 or not mains[0]
+            or any(row[1] not in {"main", "temp"} for row in databases)
+            or Path(mains[0]).expanduser().resolve(strict=True) != job.database_path):
+        raise TierJobSelectionError("Selected job database path changed")
+    if conn.execute("SELECT 1 FROM sqlite_temp_master LIMIT 1").fetchone() is not None:
+        raise TierJobSelectionError("Temporary database objects make selected job identity ambiguous")
+    tracker, revision, signature = read_rebuild_revision(conn)
+    if (tracker, revision.epoch, signature) != (job.tracker, job.source_epoch, job.source_schema_signature):
+        raise TierJobSelectionError("Selected job source identity changed")
+    with _observe_file(job.database_path) as identity:
+        if identity != job.file_identity:
+            raise TierJobSelectionError("Selected job database file changed")
+
+
+def select_tier_job(
+    conn: sqlite3.Connection, target: EmbeddingTarget, *, previous_job_id: str | None = None,
+) -> TierJob:
+    """Commit a new selection before planning, without replacing a live job.
+
+    A terminal marker can be replaced only by explicitly naming its previous ID.
+    The new ID is written through the accepted connection, not a pathname reopen.
+    """
+    if previous_job_id is not None and (type(previous_job_id) is not str
+            or re.fullmatch(r"[0-9a-f]{32}", previous_job_id) is None):
+        raise TierJobSelectionError("Invalid previous selected job identity")
+    job = capture_tier_job(conn, target)
+    values = _selection_values(job)
+    try:
+        with _selection_transaction(conn, write=True):
+            _selection_source(conn, job)
+            _selection_schema(conn, create=True)
+            previous = _selection_row(conn)
+            if previous is None:
+                if previous_job_id is not None:
+                    raise TierJobSelectionError("Previous selected job is no longer present")
+                fields = ", ".join(_SELECTION_FIELDS)
+                conn.execute(
+                    f"INSERT INTO main.{_SELECTION_TABLE}(singleton, {fields}, state) "
+                    "VALUES (1, ?, ?, ?, ?, ?, ?, ?, ?, 'selected')", values,
+                )
+            else:
+                if previous[0] != previous_job_id or previous[8] == "selected":
+                    raise TierJobSelectionError("Another job already owns the selection")
+                assignments = ", ".join(f"{name} = ?" for name in _SELECTION_FIELDS)
+                changed = conn.execute(
+                    f"UPDATE main.{_SELECTION_TABLE} SET {assignments}, state = 'selected' "
+                    "WHERE singleton = 1 AND job_id = ? AND state IN ('cancelled', 'failed')",
+                    (*values, previous_job_id),
+                ).rowcount
+                if changed != 1:
+                    raise TierJobSelectionError("Selected job changed during replacement")
+            _selection_source(conn, job)
+        return job
+    except (OSError, ValueError, sqlite3.Error, MaintenanceUnavailableError):
+        raise TierJobSelectionError("Background job could not be selected") from None
+
+
+def check_selected_tier_job(conn: sqlite3.Connection, job: TierJob) -> None:
+    """Read a coherent source/marker snapshot; return with no transaction open."""
+    values = _selection_values(job)
+    try:
+        check_tier_job(conn, job)
+        with _selection_transaction(conn, write=False):
+            _selection_source(conn, job)
+            _selection_schema(conn)
+            if _selection_row(conn) != (*values, "selected"):
+                raise TierJobSelectionError("Background job is no longer selected")
+            _selection_source(conn, job)
+    except (OSError, ValueError, sqlite3.Error, MaintenanceUnavailableError):
+        raise TierJobSelectionError("Selected background job could not be validated") from None
+
+
+@contextmanager
+def open_selected_tier_job(job: TierJob) -> Iterator[sqlite3.Connection]:
+    """Reopen under worker ownership and validate the committed selection."""
+    with open_tier_job(job) as conn:
+        check_selected_tier_job(conn, job)
+        yield conn
+
+
+def end_selected_tier_job(
+    conn: sqlite3.Connection, job: TierJob, *, outcome: Literal["cancelled", "failed"],
+) -> bool:
+    """End only this exact selected job; never modify a replacement's marker.
+
+    This is a terminal write after abandoning any source plan, not page progress
+    or successful completion/activation. False means the expected job is absent.
+    """
+    if outcome not in {"cancelled", "failed"}:
+        raise TierJobSelectionError("Unsupported terminal selected job outcome")
+    values = _selection_values(job)
+    try:
+        check_tier_job(conn, job)
+        with _selection_transaction(conn, write=True):
+            _selection_source(conn, job)
+            _selection_schema(conn)
+            row = _selection_row(conn)
+            if row != (*values, "selected"):
+                return False
+            conditions = " AND ".join(f"{name} = ?" for name in _SELECTION_FIELDS)
+            changed = conn.execute(
+                f"UPDATE main.{_SELECTION_TABLE} SET state = ? "
+                f"WHERE singleton = 1 AND {conditions} AND state = 'selected'", (outcome, *values),
+            ).rowcount
+            if changed != 1:
+                raise TierJobSelectionError("Selected job changed during termination")
+            _selection_source(conn, job)
+        return True
+    except (OSError, ValueError, sqlite3.Error, MaintenanceUnavailableError):
+        raise TierJobSelectionError("Selected background job could not be ended") from None