refactor(database): make transaction boundaries explicit

cueweaver/adapters/sqlite_term_maps.py

@@ -35,7 +35,7 @@ def __init__(self, database: SqliteDatabase) -> None:
 
     def list(self) -> list[TermMapSummary]:
         try:
-            with self._database.session() as session:
+            with self._database.read_session() as session:
                 rows = session.execute(
                     select(TermMapRow, func.count(TermMapEntryRow.position))
                     .outerjoin(
@@ -53,7 +53,7 @@ def list(self) -> list[TermMapSummary]:
 
     def get(self, term_map_id: str) -> TermMapDetail:
         try:
-            with self._database.session() as session:
+            with self._database.read_session() as session:
                 row = _require_row(session, term_map_id)
                 entries = session.scalars(
                     select(TermMapEntryRow)
@@ -78,8 +78,7 @@ def get(self, term_map_id: str) -> TermMapDetail:
     def create(self, name: str, content: Mapping[str, str]) -> TermMapSummary:
         timestamp = _utc_timestamp()
         try:
-            with self._database.session() as session:
-                session.connection().exec_driver_sql("BEGIN IMMEDIATE")
+            with self._database.write_transaction(immediate=True) as session:
                 sequence = session.scalar(select(func.max(TermMapRow.sequence)))
                 row = TermMapRow(
                     id=_new_id(),
@@ -91,7 +90,6 @@ def create(self, name: str, content: Mapping[str, str]) -> TermMapSummary:
                 session.add(row)
                 _replace_entries(session, row.id, content)
                 count = _entry_count(session, row.id)
-                session.commit()
                 return _summary(row, count)
         except (
             DatabaseOpenError,
@@ -103,13 +101,12 @@ def create(self, name: str, content: Mapping[str, str]) -> TermMapSummary:
 
     def rename(self, term_map_id: str, name: str) -> TermMapSummary:
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 row = _require_row(session, term_map_id)
                 row.name = name
                 row.name_folded = name.casefold()
                 row.updated_at = _utc_timestamp()
                 count = _entry_count(session, term_map_id)
-                session.commit()
                 return _summary(row, int(count or 0))
         except ServiceError:
             raise
@@ -123,12 +120,11 @@ def rename(self, term_map_id: str, name: str) -> TermMapSummary:
 
     def replace(self, term_map_id: str, content: Mapping[str, str]) -> TermMapSummary:
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 row = _require_row(session, term_map_id)
                 _replace_entries(session, term_map_id, content)
                 row.updated_at = _utc_timestamp()
                 count = _entry_count(session, term_map_id)
-                session.commit()
                 return _summary(row, count)
         except ServiceError:
             raise
@@ -143,7 +139,7 @@ def replace(self, term_map_id: str, content: Mapping[str, str]) -> TermMapSummar
 
     def delete(self, term_map_id: str, name: str) -> TermMapSummary:
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 row = _require_row(session, term_map_id)
                 if name != row.name:
                     raise ServiceError(
@@ -154,7 +150,6 @@ def delete(self, term_map_id: str, name: str) -> TermMapSummary:
                 count = _entry_count(session, term_map_id)
                 summary = _summary(row, int(count or 0))
                 session.delete(row)
-                session.commit()
                 return summary
         except ServiceError:
             raise
@@ -176,7 +171,7 @@ def __init__(self, database: SqliteDatabase) -> None:
 
     def snapshot_bindings(self) -> dict[str, str]:
         try:
-            with self._database.session() as session:
+            with self._database.read_session() as session:
                 rows = session.scalars(
                     select(DirectoryTermMapBindingRow).order_by(
                         DirectoryTermMapBindingRow.directory
@@ -195,7 +190,7 @@ def bind(
         term_map_id: str,
     ) -> None:
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 row = session.get(DirectoryTermMapBindingRow, directory)
                 if row is None:
                     session.add(
@@ -205,7 +200,6 @@ def bind(
                     )
                 else:
                     row.term_map_id = term_map_id
-                session.commit()
         except ServiceError:
             raise
         except (DatabaseOpenError, DatabasePathError) as error:
@@ -225,13 +219,12 @@ def bind(
 
     def remove(self, directory: str) -> None:
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 session.execute(
                     delete(DirectoryTermMapBindingRow).where(
                         DirectoryTermMapBindingRow.directory == directory
                     )
                 )
-                session.commit()
         except (DatabaseOpenError, DatabasePathError) as error:
             raise ServiceError(
                 "directory_term_maps_unavailable",

cueweaver/application/database.py

@@ -125,7 +125,7 @@ class DatabaseOpenError(sqlite3.Error):
 
 
 class SqliteDatabase:
-    """Run migrations once and provide short-lived ORM sessions."""
+    """Run migrations once and provide explicit ORM read/write scopes."""
 
     def __init__(self, path: Path) -> None:
         self.path = path
@@ -139,12 +139,25 @@ def initialize(self) -> None:
         self._initialize()
 
     @contextmanager
-    def session(self) -> Iterator[Session]:
+    def read_session(self) -> Iterator[Session]:
         self.initialize()
         assert self._session_factory is not None
         session = self._session_factory()
         try:
             yield session
+        finally:
+            session.close()
+
+    @contextmanager
+    def write_transaction(self, *, immediate: bool = False) -> Iterator[Session]:
+        self.initialize()
+        assert self._session_factory is not None
+        session = self._session_factory()
+        try:
+            if immediate:
+                session.connection().exec_driver_sql("BEGIN IMMEDIATE")
+            yield session
+            session.commit()
         except Exception:
             session.rollback()
             raise

cueweaver/application/jobs/store.py

@@ -45,7 +45,7 @@ def __init__(self, database: SqliteDatabase) -> None:
 
     def load(self) -> list[JobRecord]:
         try:
-            with self._database.session() as session:
+            with self._database.read_session() as session:
                 rows = session.scalars(
                     select(JobRow).order_by(
                         JobRow.queue_sequence, JobRow.created_at, JobRow.id
@@ -63,11 +63,10 @@ def write(self, record: JobRecord) -> None:
     def remove(self, job_id: str) -> None:
         _require_valid_job_id(job_id)
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 row = session.get(JobRow, job_id)
                 if row is not None:
                     session.delete(row)
-                session.commit()
         except (sqlite3.Error, SQLAlchemyError) as error:
             raise ServiceError(
                 "database_unavailable", "Job record could not be deleted"
@@ -76,9 +75,8 @@ def remove(self, job_id: str) -> None:
     def _upsert(self, record: JobRecord) -> None:
         prepared = self._prepare_record(record)
         try:
-            with self._database.session() as session:
+            with self._database.write_transaction() as session:
                 _upsert_row(session, prepared)
-                session.commit()
         except (sqlite3.Error, SQLAlchemyError) as error:
             raise ServiceError(
                 "database_unavailable", "Job record could not be persisted"

tests/test_database.py

@@ -7,7 +7,13 @@
 from alembic.config import Config
 from sqlalchemy import URL, create_engine
 
-from cueweaver.application.database import SqliteDatabase, _migration_config
+from cueweaver.application.database import (
+    JobRow,
+    JobStatusHistoryRow,
+    JobTermMapSnapshotRow,
+    SqliteDatabase,
+    _migration_config,
+)
 from cueweaver.application.jobs.store import SqliteJobRecordStore
 
 
@@ -42,10 +48,98 @@ def test_sqlite_database_bootstraps_the_application_schema(tmp_path: Path):
             row[1] for row in connection.execute("PRAGMA table_info(jobs)")
         }.isdisjoint({"record_json", "content_json"})
 
-    with database.session() as session:
+    with database.read_session() as session:
         assert session.connection().exec_driver_sql("PRAGMA foreign_keys").scalar() == 1
 
 
+def test_write_transaction_commits_a_multi_table_change(tmp_path: Path):
+    database = SqliteDatabase(tmp_path / "cueweaver.sqlite3")
+
+    with database.write_transaction() as session:
+        _add_job_rows(session)
+
+    with database.read_session() as session:
+        assert session.get(JobRow, "job-1") is not None
+        assert session.get(JobStatusHistoryRow, {"job_id": "job-1", "sequence": 0})
+        assert session.get(JobTermMapSnapshotRow, {"job_id": "job-1", "position": 0})
+
+
+def test_write_transaction_rolls_back_when_the_scope_fails(tmp_path: Path):
+    database = SqliteDatabase(tmp_path / "cueweaver.sqlite3")
+
+    with (
+        pytest.raises(RuntimeError, match="abort"),
+        database.write_transaction() as session,
+    ):
+        _add_job_rows(session)
+        raise RuntimeError("abort")
+
+    with database.read_session() as session:
+        assert session.get(JobRow, "job-1") is None
+        assert (
+            session.get(JobStatusHistoryRow, {"job_id": "job-1", "sequence": 0}) is None
+        )
+        assert (
+            session.get(JobTermMapSnapshotRow, {"job_id": "job-1", "position": 0})
+            is None
+        )
+
+
+def test_sqlite_database_can_close_and_reinitialize_between_scopes(tmp_path: Path):
+    database = SqliteDatabase(tmp_path / "cueweaver.sqlite3")
+
+    with database.write_transaction() as session:
+        _add_job_rows(session)
+    database.close()
+
+    with database.read_session() as session:
+        assert session.get(JobRow, "job-1") is not None
+    database.close()
+
+
+def _job_row() -> JobRow:
+    return JobRow(
+        id="job-1",
+        schema_version=1,
+        status="Queued",
+        attempt=1,
+        created_at="2026-08-24T00:00:00Z",
+        queue_sequence=0,
+        media_path="Movie.mkv",
+        subtitle_path="Movie.en.srt",
+        target_language_code="zh-Hans",
+        term_map_mode="none",
+        output_path="Movie.zh-Hans.srt",
+        source_format="srt",
+        dynamic_terminology_enabled=True,
+        subtitle_terminology_filter_enabled=True,
+        output_suffix="zh-Hans",
+        output_conflict_policy="skip",
+    )
+
+
+def _add_job_rows(session) -> None:
+    session.add(_job_row())
+    session.add(
+        JobStatusHistoryRow(
+            job_id="job-1",
+            sequence=0,
+            status="Queued",
+            attempt=1,
+            started_at="2026-08-24T00:00:00Z",
+        )
+    )
+    session.add(
+        JobTermMapSnapshotRow(
+            job_id="job-1",
+            position=0,
+            source="Captain",
+            source_folded="captain",
+            target="队长",
+        )
+    )
+
+
 def test_migration_discards_issue_193_application_data(tmp_path: Path):
     path = tmp_path / "cueweaver.sqlite3"
     record = {

tests/test_jobs.py

@@ -217,6 +217,45 @@ def test_sqlite_record_store_persists_records_and_uses_a_transactional_database(
     assert store.load() == []
 
 
+def test_sqlite_job_write_rolls_back_related_rows_on_database_failure(tmp_path: Path):
+    database_path = tmp_path / "cueweaver.sqlite3"
+    database = SqliteDatabase(database_path)
+    database.initialize()
+    with sqlite3.connect(database_path) as connection:
+        connection.execute(
+            """
+            CREATE TRIGGER fail_job_snapshot_insert
+            BEFORE INSERT ON job_term_map_snapshots
+            BEGIN
+                SELECT RAISE(ABORT, 'forced snapshot failure');
+            END
+            """
+        )
+
+    record = persisted_job_record("atomic-job")
+    request = record["request"]
+    assert isinstance(request, dict)
+    request["term_map_mode"] = "selected"
+    request["term_map"] = {
+        "id": "map-1",
+        "name": "Characters",
+        "content": {"Captain": "队长"},
+    }
+    store = SqliteJobRecordStore(database)
+
+    with pytest.raises(ServiceError, match="could not be persisted"):
+        store.write(record)
+
+    with sqlite3.connect(database_path) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM jobs").fetchone() == (0,)
+        assert connection.execute(
+            "SELECT COUNT(*) FROM job_status_history"
+        ).fetchone() == (0,)
+        assert connection.execute(
+            "SELECT COUNT(*) FROM job_term_map_snapshots"
+        ).fetchone() == (0,)
+
+
 def test_sqlite_record_store_rejects_a_future_schema_version(tmp_path: Path):
     store = SqliteJobRecordStore(SqliteDatabase(tmp_path / "cueweaver.sqlite3"))
     store.write(persisted_job_record("future-schema"))